IBR-DTNSuite  0.10
FloodRoutingExtension.cpp
Go to the documentation of this file.
1 /*
2  * FloodRoutingExtension.cpp
3  *
4  * Copyright (C) 2011 IBR, TU Braunschweig
5  *
6  * Written-by: Johannes Morgenroth <morgenroth@ibr.cs.tu-bs.de>
7  *
8  * Licensed under the Apache License, Version 2.0 (the "License");
9  * you may not use this file except in compliance with the License.
10  * You may obtain a copy of the License at
11  *
12  * http://www.apache.org/licenses/LICENSE-2.0
13  *
14  * Unless required by applicable law or agreed to in writing, software
15  * distributed under the License is distributed on an "AS IS" BASIS,
16  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17  * See the License for the specific language governing permissions and
18  * limitations under the License.
19  *
20  */
21 
22 #include "routing/flooding/FloodRoutingExtension.h"
23 #include "routing/QueueBundleEvent.h"
24 #include "core/NodeEvent.h"
25 #include "net/TransferCompletedEvent.h"
26 #include "net/TransferAbortedEvent.h"
27 #include "net/ConnectionEvent.h"
28 #include "core/Node.h"
29 #include "net/ConnectionManager.h"
30 #include "Configuration.h"
31 #include "core/BundleCore.h"
32 
33 #include <ibrdtn/data/MetaBundle.h>
34 #include <ibrcommon/thread/MutexLock.h>
35 #include <ibrcommon/Logger.h>
36 
37 #include <functional>
38 #include <list>
39 #include <algorithm>
40 #include <iomanip>
41 #include <ios>
42 #include <iostream>
43 #include <memory>
44 
45 #include <stdlib.h>
46 #include <typeinfo>
47 
48 namespace dtn
49 {
50  namespace routing
51  {
52  const std::string FloodRoutingExtension::TAG = "FloodRoutingExtension";
53 
54  FloodRoutingExtension::FloodRoutingExtension()
55  {
56  // write something to the syslog
57  IBRCOMMON_LOGGER_TAG(FloodRoutingExtension::TAG, info) << "Initializing flooding routing module" << IBRCOMMON_LOGGER_ENDL;
58  }
59 
60  FloodRoutingExtension::~FloodRoutingExtension()
61  {
62  join();
63  }
64 
65  void FloodRoutingExtension::notify(const dtn::core::Event *evt) throw ()
66  {
67  try {
68  const QueueBundleEvent &queued = dynamic_cast<const QueueBundleEvent&>(*evt);
69 
70  // new bundles are forwarded to all neighbors
71  const std::set<dtn::core::Node> nl = dtn::core::BundleCore::getInstance().getConnectionManager().getNeighbors();
72 
73  for (std::set<dtn::core::Node>::const_iterator iter = nl.begin(); iter != nl.end(); ++iter)
74  {
75  const dtn::core::Node &n = (*iter);
76 
77  if (n.getEID() != queued.origin) {
78  // transfer the next bundle to this destination
79  _taskqueue.push( new SearchNextBundleTask( n.getEID() ) );
80  }
81  }
82 
83  return;
84  } catch (const std::bad_cast&) { };
85 
86  try {
87  const dtn::core::NodeEvent &nodeevent = dynamic_cast<const dtn::core::NodeEvent&>(*evt);
88  const dtn::core::Node &n = nodeevent.getNode();
89 
90  if (nodeevent.getAction() == NODE_AVAILABLE)
91  {
92  const dtn::data::EID &eid = n.getEID();
93 
94  // send all (multi-hop) bundles in the storage to the neighbor
95  _taskqueue.push( new SearchNextBundleTask(eid) );
96  }
97  else if (nodeevent.getAction() == NODE_DATA_ADDED)
98  {
99  const dtn::data::EID &eid = n.getEID();
100 
101  // send all (multi-hop) bundles in the storage to the neighbor
102  _taskqueue.push( new SearchNextBundleTask(eid) );
103  }
104 
105  return;
106  } catch (const std::bad_cast&) { };
107 
108  try {
109  const dtn::net::ConnectionEvent &ce = dynamic_cast<const dtn::net::ConnectionEvent&>(*evt);
110 
111  if (ce.state == dtn::net::ConnectionEvent::CONNECTION_UP)
112  {
113  // send all (multi-hop) bundles in the storage to the neighbor
114  _taskqueue.push( new SearchNextBundleTask(ce.peer) );
115  }
116  return;
117  } catch (const std::bad_cast&) { };
118 
119  // The bundle transfer has been aborted
120  try {
121  const dtn::net::TransferAbortedEvent &aborted = dynamic_cast<const dtn::net::TransferAbortedEvent&>(*evt);
122 
123  // transfer the next bundle to this destination
124  _taskqueue.push( new SearchNextBundleTask( aborted.getPeer() ) );
125  return;
126  } catch (const std::bad_cast&) { };
127 
128  // A bundle transfer was successful
129  try {
130  const dtn::net::TransferCompletedEvent &completed = dynamic_cast<const dtn::net::TransferCompletedEvent&>(*evt);
131 
132  // transfer the next bundle to this destination
133  _taskqueue.push( new SearchNextBundleTask( completed.getPeer() ) );
134  return;
135  } catch (const std::bad_cast&) { };
136  }
137 
138  void FloodRoutingExtension::componentUp() throw ()
139  {
140  // reset the task queue
141  _taskqueue.reset();
142 
143  // routine checked for throw() on 15.02.2013
144  try {
145  // run the thread
146  start();
147  } catch (const ibrcommon::ThreadException &ex) {
148  IBRCOMMON_LOGGER_TAG(FloodRoutingExtension::TAG, error) << "componentUp failed: " << ex.what() << IBRCOMMON_LOGGER_ENDL;
149  }
150  }
151 
152  void FloodRoutingExtension::componentDown() throw ()
153  {
154  try {
155  // stop the thread
156  stop();
157  join();
158  } catch (const ibrcommon::ThreadException &ex) {
159  IBRCOMMON_LOGGER_TAG(FloodRoutingExtension::TAG, error) << "componentDown failed: " << ex.what() << IBRCOMMON_LOGGER_ENDL;
160  }
161  }
162 
163  void FloodRoutingExtension::__cancellation() throw ()
164  {
165  _taskqueue.abort();
166  }
167 
168  void FloodRoutingExtension::run() throw ()
169  {
170  class BundleFilter : public dtn::storage::BundleSelector
171  {
172  public:
173  BundleFilter(const NeighborDatabase::NeighborEntry &entry)
174  : _entry(entry)
175  {};
176 
177  virtual ~BundleFilter() {};
178 
179  virtual dtn::data::Size limit() const throw () { return _entry.getFreeTransferSlots(); };
180 
181  virtual bool shouldAdd(const dtn::data::MetaBundle &meta) const throw (dtn::storage::BundleSelectorException)
182  {
183  // check Scope Control Block - do not forward bundles with hop limit == 0
184  if (meta.hopcount == 0)
185  {
186  return false;
187  }
188 
189  // do not forward any routing control message
190  // this is done by the neighbor routing module
191  if (isRouting(meta.source))
192  {
193  return false;
194  }
195 
196  // do not forward local bundles
197  if ((meta.destination.getNode() == dtn::core::BundleCore::local)
198  && meta.get(dtn::data::PrimaryBlock::DESTINATION_IS_SINGLETON)
199  )
200  {
201  return false;
202  }
203 
204  // check Scope Control Block - do not forward non-group bundles with hop limit <= 1
205  if ((meta.hopcount <= 1) && (meta.get(dtn::data::PrimaryBlock::DESTINATION_IS_SINGLETON)))
206  {
207  return false;
208  }
209 
210  // do not forward bundles addressed to this neighbor,
211  // because this is handled by neighbor routing extension
212  if (_entry.eid == meta.destination.getNode())
213  {
214  return false;
215  }
216 
217  // do not forward bundles already known by the destination
218  if (_entry.has(meta))
219  {
220  return false;
221  }
222 
223  return true;
224  };
225 
226  private:
227  const NeighborDatabase::NeighborEntry &_entry;
228  };
229 
230  dtn::storage::BundleResultList list;
231 
232  while (true)
233  {
234  try {
235  Task *t = _taskqueue.getnpop(true);
236  std::auto_ptr<Task> killer(t);
237 
238  IBRCOMMON_LOGGER_DEBUG_TAG(FloodRoutingExtension::TAG, 50) << "processing task " << t->toString() << IBRCOMMON_LOGGER_ENDL;
239 
240  try {
241  try {
242  SearchNextBundleTask &task = dynamic_cast<SearchNextBundleTask&>(*t);
243 
244  // lock the neighbor database while searching for bundles
245  {
246  NeighborDatabase &db = (**this).getNeighborDB();
247 
248  ibrcommon::MutexLock l(db);
249  NeighborDatabase::NeighborEntry &entry = db.get(task.eid);
250 
251  // check if enough transfer slots available (threshold reached)
252  if (!entry.isTransferThresholdReached())
253  throw NeighborDatabase::NoMoreTransfersAvailable();
254 
255  // get the bundle filter of the neighbor
256  BundleFilter filter(entry);
257 
258  // some debug
259  IBRCOMMON_LOGGER_DEBUG_TAG(FloodRoutingExtension::TAG, 40) << "search some bundles not known by " << task.eid.getString() << IBRCOMMON_LOGGER_ENDL;
260 
261  // query all bundles from the storage
262  list.clear();
263  (**this).getSeeker().get(filter, list);
264  }
265 
266  // send the bundles as long as we have resources
267  for (std::list<dtn::data::MetaBundle>::const_iterator iter = list.begin(); iter != list.end(); ++iter)
268  {
269  try {
270  // transfer the bundle to the neighbor
271  transferTo(task.eid, *iter);
272  } catch (const NeighborDatabase::AlreadyInTransitException&) { };
273  }
274  } catch (const NeighborDatabase::NoMoreTransfersAvailable&) {
275  } catch (const NeighborDatabase::NeighborNotAvailableException&) {
276  } catch (const dtn::storage::NoBundleFoundException&) {
277  } catch (const std::bad_cast&) { };
278  } catch (const ibrcommon::Exception &ex) {
279  IBRCOMMON_LOGGER_DEBUG_TAG(FloodRoutingExtension::TAG, 20) << "task failed: " << ex.what() << IBRCOMMON_LOGGER_ENDL;
280  }
281  } catch (const std::exception &ex) {
282  IBRCOMMON_LOGGER_DEBUG_TAG(FloodRoutingExtension::TAG, 15) << "terminated due to " << ex.what() << IBRCOMMON_LOGGER_ENDL;
283  return;
284  }
285 
286  yield();
287  }
288  }
289 
290  /****************************************/
291 
292  FloodRoutingExtension::SearchNextBundleTask::SearchNextBundleTask(const dtn::data::EID &e)
293  : eid(e)
294  { }
295 
296  FloodRoutingExtension::SearchNextBundleTask::~SearchNextBundleTask()
297  { }
298 
299  std::string FloodRoutingExtension::SearchNextBundleTask::toString()
300  {
301  return "SearchNextBundleTask: " + eid.getString();
302  }
303  }
304 }