lp-reassembler.cpp
Go to the documentation of this file.
1 /* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
26 #include "lp-reassembler.hpp"
27 #include "link-service.hpp"
28 #include <numeric>
29 
30 namespace nfd {
31 namespace face {
32 
33 NFD_LOG_INIT("LpReassembler");
34 
36  : nMaxFragments(400)
37  , reassemblyTimeout(time::milliseconds(500))
38 {
39 }
40 
42  : m_options(options)
43  , m_linkService(linkService)
44 {
45 }
46 
47 std::tuple<bool, Block, lp::Packet>
48 LpReassembler::receiveFragment(Transport::EndpointId remoteEndpoint, const lp::Packet& packet)
49 {
50  BOOST_ASSERT(packet.has<lp::FragmentField>());
51 
52  static auto FALSE_RETURN = std::make_tuple(false, Block(), lp::Packet());
53 
54  // read and check FragIndex and FragCount
55  uint64_t fragIndex = 0;
56  uint64_t fragCount = 1;
57  if (packet.has<lp::FragIndexField>()) {
58  fragIndex = packet.get<lp::FragIndexField>();
59  }
60  if (packet.has<lp::FragCountField>()) {
61  fragCount = packet.get<lp::FragCountField>();
62  }
63 
64  if (fragIndex >= fragCount) {
65  NFD_LOG_FACE_WARN("reassembly error, FragIndex>=FragCount: DROP");
66  return FALSE_RETURN;
67  }
68 
69  if (fragCount > m_options.nMaxFragments) {
70  NFD_LOG_FACE_WARN("reassembly error, FragCount over limit: DROP");
71  return FALSE_RETURN;
72  }
73 
74  // check for fast path
75  if (fragIndex == 0 && fragCount == 1) {
76  ndn::Buffer::const_iterator fragBegin, fragEnd;
77  std::tie(fragBegin, fragEnd) = packet.get<lp::FragmentField>();
78  Block netPkt(&*fragBegin, std::distance(fragBegin, fragEnd));
79  return std::make_tuple(true, netPkt, packet);
80  }
81 
82  // check Sequence and compute message identifier
83  if (!packet.has<lp::SequenceField>()) {
84  NFD_LOG_FACE_WARN("reassembly error, Sequence missing: DROP");
85  return FALSE_RETURN;
86  }
87  lp::Sequence messageIdentifier = packet.get<lp::SequenceField>() - fragIndex;
88  Key key = std::make_tuple(remoteEndpoint, messageIdentifier);
89 
90  // add to PartialPacket
91  PartialPacket& pp = m_partialPackets[key];
92  if (pp.fragCount == 0) { // new PartialPacket
93  pp.fragCount = fragCount;
94  pp.nReceivedFragments = 0;
95  pp.fragments.resize(fragCount);
96  }
97  else {
98  if (fragCount != pp.fragCount) {
99  NFD_LOG_FACE_WARN("reassembly error, FragCount changed: DROP");
100  return FALSE_RETURN;
101  }
102  }
103 
104  if (pp.fragments[fragIndex].has<lp::SequenceField>()) {
105  NFD_LOG_FACE_TRACE("fragment already received: DROP");
106  return FALSE_RETURN;
107  }
108 
109  pp.fragments[fragIndex] = packet;
110  ++pp.nReceivedFragments;
111 
112  // check complete condition
113  if (pp.nReceivedFragments == pp.fragCount) {
114  Block reassembled = doReassembly(key);
115  lp::Packet firstFrag(std::move(pp.fragments[0]));
116  m_partialPackets.erase(key);
117  return std::make_tuple(true, reassembled, firstFrag);
118  }
119 
120  // set drop timer
121  pp.dropTimer = scheduler::schedule(m_options.reassemblyTimeout,
122  bind(&LpReassembler::timeoutPartialPacket, this, key));
123 
124  return FALSE_RETURN;
125 }
126 
127 Block
128 LpReassembler::doReassembly(const Key& key)
129 {
130  PartialPacket& pp = m_partialPackets[key];
131 
132  size_t payloadSize = std::accumulate(pp.fragments.begin(), pp.fragments.end(), 0,
133  [&] (size_t sum, const lp::Packet& pkt) -> size_t {
134  ndn::Buffer::const_iterator fragBegin, fragEnd;
135  std::tie(fragBegin, fragEnd) = pkt.get<lp::FragmentField>();
136  return sum + std::distance(fragBegin, fragEnd);
137  });
138 
139  ndn::Buffer fragBuffer(payloadSize);
140  ndn::Buffer::iterator it = fragBuffer.begin();
141 
142  for (const lp::Packet& frag : pp.fragments) {
143  ndn::Buffer::const_iterator fragBegin, fragEnd;
144  std::tie(fragBegin, fragEnd) = frag.get<lp::FragmentField>();
145  it = std::copy(fragBegin, fragEnd, it);
146  }
147 
148  return Block(&*(fragBuffer.cbegin()), std::distance(fragBuffer.cbegin(), fragBuffer.cend()));
149 }
150 
151 void
152 LpReassembler::timeoutPartialPacket(const Key& key)
153 {
154  auto it = m_partialPackets.find(key);
155  if (it == m_partialPackets.end()) {
156  return;
157  }
158 
159  this->beforeTimeout(std::get<0>(key), it->second.nReceivedFragments);
160  m_partialPackets.erase(it);
161 }
162 
163 std::ostream&
164 operator<<(std::ostream& os, const FaceLogHelper<LpReassembler>& flh)
165 {
166  if (flh.obj.getLinkService() == nullptr) {
167  os << "[id=0,local=unknown,remote=unknown] ";
168  }
169  else {
170  os << FaceLogHelper<LinkService>(*flh.obj.getLinkService());
171  }
172  return os;
173 }
174 
175 } // namespace face
176 } // namespace nfd
time::nanoseconds reassemblyTimeout
timeout before a partially reassembled packet is dropped
LpReassembler(const Options &options=Options(), const LinkService *linkService=nullptr)
#define NFD_LOG_FACE_TRACE(msg)
Log a message at TRACE level.
Definition: face-log.hpp:79
size_t nMaxFragments
maximum number of fragments in a packet
Table::const_iterator iterator
Definition: cs-internal.hpp:41
#define NFD_LOG_FACE_WARN(msg)
Log a message at WARN level.
Definition: face-log.hpp:88
Options that control the behavior of LpReassembler.
Copyright (c) 2014-2015, Regents of the University of California, Arizona Board of Regents...
Definition: algorithm.hpp:32
signal::Signal< LpReassembler, Transport::EndpointId, size_t > beforeTimeout
signals before a partial packet is dropped due to timeout
uint64_t EndpointId
identifies an endpoint on the link
Definition: transport.hpp:109
#define NFD_LOG_INIT(name)
Definition: logger.hpp:122
std::tuple< bool, Block, lp::Packet > receiveFragment(Transport::EndpointId remoteEndpoint, const lp::Packet &packet)
adds received fragment to buffer
EventId schedule(time::nanoseconds after, const EventCallback &event)
schedule an event
Definition: scheduler.cpp:47