IBR-DTNSuite  0.10
NeighborRoutingExtension.cpp
Go to the documentation of this file.
1 /*
2  * NeighborRoutingExtension.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 "config.h"
23 #include "routing/NeighborRoutingExtension.h"
24 #include "routing/QueueBundleEvent.h"
25 #include "core/BundleCore.h"
26 #include "core/TimeEvent.h"
27 #include "net/TransferCompletedEvent.h"
28 #include "net/TransferAbortedEvent.h"
29 #include "net/ConnectionEvent.h"
30 #include "core/NodeEvent.h"
31 #include "core/Node.h"
32 #include "net/ConnectionManager.h"
33 #include "ibrcommon/thread/MutexLock.h"
34 #include "storage/BundleStorage.h"
35 #include "core/BundleEvent.h"
36 #include <ibrcommon/Logger.h>
37 
38 #ifdef HAVE_SQLITE
39 #include "storage/SQLiteBundleStorage.h"
40 #endif
41 
42 #include <functional>
43 #include <list>
44 #include <algorithm>
45 #include <typeinfo>
46 #include <memory>
47 
48 namespace dtn
49 {
50  namespace routing
51  {
52  const std::string NeighborRoutingExtension::TAG = "NeighborRoutingExtension";
53 
54  NeighborRoutingExtension::NeighborRoutingExtension()
55  {
56  }
57 
58  NeighborRoutingExtension::~NeighborRoutingExtension()
59  {
60  join();
61  }
62 
63  void NeighborRoutingExtension::__cancellation() throw ()
64  {
65  _taskqueue.abort();
66  }
67 
68  void NeighborRoutingExtension::run() throw ()
69  {
70 #ifdef HAVE_SQLITE
71  class BundleFilter : public dtn::storage::BundleSelector, public dtn::storage::SQLiteDatabase::SQLBundleQuery
72 #else
73  class BundleFilter : public dtn::storage::BundleSelector
74 #endif
75  {
76  public:
77  BundleFilter(NeighborRoutingExtension &e, const NeighborDatabase::NeighborEntry &entry)
78  : _extension(e), _entry(entry)
79  {};
80 
81  virtual ~BundleFilter() {};
82 
83  virtual dtn::data::Size limit() const throw () { return _entry.getFreeTransferSlots(); };
84 
85  virtual bool shouldAdd(const dtn::data::MetaBundle &meta) const throw (dtn::storage::BundleSelectorException)
86  {
87  return _extension.shouldRouteTo(meta, _entry);
88  };
89 
90 #ifdef HAVE_SQLITE
91  const std::string getWhere() const throw ()
92  {
93  return "destination LIKE ?";
94  };
95 
96  int bind(sqlite3_stmt *st, int offset) const throw ()
97  {
98  const std::string d = _entry.eid.getNode().getString() + "%";
99  sqlite3_bind_text(st, offset, d.c_str(), static_cast<int>(d.size()), SQLITE_TRANSIENT);
100  return offset + 1;
101  }
102 #endif
103 
104  private:
105  NeighborRoutingExtension &_extension;
106  const NeighborDatabase::NeighborEntry &_entry;
107  };
108 
109  dtn::storage::BundleResultList list;
110 
111  while (true)
112  {
113  NeighborDatabase &db = (**this).getNeighborDB();
114 
115  try {
116  Task *t = _taskqueue.getnpop(true);
117  std::auto_ptr<Task> killer(t);
118 
119  IBRCOMMON_LOGGER_DEBUG_TAG(NeighborRoutingExtension::TAG, 5) << "processing task " << t->toString() << IBRCOMMON_LOGGER_ENDL;
120 
126  try {
127  SearchNextBundleTask &task = dynamic_cast<SearchNextBundleTask&>(*t);
128 
129  // lock the neighbor database while searching for bundles
130  {
131  // this destination is not handles by any static route
132  ibrcommon::MutexLock l(db);
133  NeighborDatabase::NeighborEntry &entry = db.get(task.eid);
134 
135  // check if enough transfer slots available (threshold reached)
136  if (!entry.isTransferThresholdReached())
137  throw NeighborDatabase::NoMoreTransfersAvailable();
138 
139  // create a new bundle filter
140  BundleFilter filter(*this, entry);
141 
142  // query an unknown bundle from the storage, the list contains max. 10 items.
143  list.clear();
144  (**this).getSeeker().get(filter, list);
145  }
146 
147  IBRCOMMON_LOGGER_DEBUG_TAG(NeighborRoutingExtension::TAG, 5) << "got " << list.size() << " items to transfer to " << task.eid.getString() << IBRCOMMON_LOGGER_ENDL;
148 
149  // send the bundles as long as we have resources
150  for (std::list<dtn::data::MetaBundle>::const_iterator iter = list.begin(); iter != list.end(); ++iter)
151  {
152  try {
153  // transfer the bundle to the neighbor
154  transferTo(task.eid, *iter);
155  } catch (const NeighborDatabase::AlreadyInTransitException&) { };
156  }
157  } catch (const NeighborDatabase::NoMoreTransfersAvailable &ex) {
158  IBRCOMMON_LOGGER_DEBUG_TAG(NeighborRoutingExtension::TAG, 10) << "task " << t->toString() << " aborted: " << ex.what() << IBRCOMMON_LOGGER_ENDL;
159  } catch (const NeighborDatabase::NeighborNotAvailableException &ex) {
160  IBRCOMMON_LOGGER_DEBUG_TAG(NeighborRoutingExtension::TAG, 10) << "task " << t->toString() << " aborted: " << ex.what() << IBRCOMMON_LOGGER_ENDL;
161  } catch (const dtn::storage::NoBundleFoundException &ex) {
162  IBRCOMMON_LOGGER_DEBUG_TAG(NeighborRoutingExtension::TAG, 10) << "task " << t->toString() << " aborted: " << ex.what() << IBRCOMMON_LOGGER_ENDL;
163  } catch (const std::bad_cast&) { };
164 
168  try {
169  const ProcessBundleTask &task = dynamic_cast<ProcessBundleTask&>(*t);
170 
171  // lock the neighbor database while searching for bundles
172  {
173  // this destination is not handles by any static route
174  ibrcommon::MutexLock l(db);
175  NeighborDatabase::NeighborEntry &entry = db.get(task.nexthop);
176 
177  if (!shouldRouteTo(task.bundle, entry))
178  throw NeighborDatabase::NoRouteKnownException();
179  }
180 
181  // transfer the bundle to the neighbor
182  transferTo(task.nexthop, task.bundle);
183  } catch (const NeighborDatabase::AlreadyInTransitException &ex) {
184  IBRCOMMON_LOGGER_DEBUG_TAG(TAG, 10) << "task " << t->toString() << " aborted: " << ex.what() << IBRCOMMON_LOGGER_ENDL;
185  } catch (const NeighborDatabase::NoMoreTransfersAvailable &ex) {
186  IBRCOMMON_LOGGER_DEBUG_TAG(TAG, 10) << "task " << t->toString() << " aborted: " << ex.what() << IBRCOMMON_LOGGER_ENDL;
187  } catch (const NeighborDatabase::NeighborNotAvailableException &ex) {
188  IBRCOMMON_LOGGER_DEBUG_TAG(TAG, 10) << "task " << t->toString() << " aborted: " << ex.what() << IBRCOMMON_LOGGER_ENDL;
189  } catch (const NeighborDatabase::NoRouteKnownException &ex) {
190  // nothing to do here.
191  } catch (const std::bad_cast&) { };
192  } catch (const std::exception &ex) {
193  IBRCOMMON_LOGGER_DEBUG_TAG(NeighborRoutingExtension::TAG, 15) << "terminated due to " << ex.what() << IBRCOMMON_LOGGER_ENDL;
194  return;
195  }
196 
197  yield();
198  }
199  }
200 
201  bool NeighborRoutingExtension::shouldRouteTo(const dtn::data::MetaBundle &meta, const NeighborDatabase::NeighborEntry &n) const
202  {
203  // check Scope Control Block - do not forward bundles with hop limit == 0
204  if (meta.hopcount == 0)
205  {
206  return false;
207  }
208 
209  if (meta.get(dtn::data::PrimaryBlock::DESTINATION_IS_SINGLETON))
210  {
211  // do not forward local bundles
212  if (meta.destination.getNode() == dtn::core::BundleCore::local)
213  {
214  return false;
215  }
216 
217  // do not forward bundles for other nodes
218  if (n.eid != meta.destination.getNode())
219  {
220  return false;
221  }
222  }
223  else
224  {
225  // do not forward non-singleton bundles
226  return false;
227  }
228 
229  // do not forward bundles already known by the destination
230  if (n.has(meta))
231  {
232  return false;
233  }
234 
235  return true;
236  }
237 
238  void NeighborRoutingExtension::notify(const dtn::core::Event *evt) throw ()
239  {
240  try {
241  const QueueBundleEvent &queued = dynamic_cast<const QueueBundleEvent&>(*evt);
242 
243  // try to deliver new bundles to all neighbors
244  const std::set<dtn::core::Node> nl = dtn::core::BundleCore::getInstance().getConnectionManager().getNeighbors();
245 
246  for (std::set<dtn::core::Node>::const_iterator iter = nl.begin(); iter != nl.end(); ++iter)
247  {
248  const dtn::core::Node &n = (*iter);
249 
250  if (n.getEID() != queued.origin) {
251  // transfer the next bundle to this destination
252  _taskqueue.push( new ProcessBundleTask(queued.bundle, queued.origin, n.getEID()) );
253  }
254  }
255 
256  return;
257  } catch (const std::bad_cast&) { };
258 
259  try {
260  const dtn::net::TransferCompletedEvent &completed = dynamic_cast<const dtn::net::TransferCompletedEvent&>(*evt);
261  // transfer the next bundle to this destination
262  _taskqueue.push( new SearchNextBundleTask( completed.getPeer() ) );
263  return;
264  } catch (const std::bad_cast&) { };
265 
266  try {
267  const dtn::net::TransferAbortedEvent &aborted = dynamic_cast<const dtn::net::TransferAbortedEvent&>(*evt);
268  // transfer the next bundle to this destination
269  _taskqueue.push( new SearchNextBundleTask( aborted.getPeer() ) );
270 
271  return;
272  } catch (const std::bad_cast&) { };
273 
274  try {
275  const dtn::core::NodeEvent &nodeevent = dynamic_cast<const dtn::core::NodeEvent&>(*evt);
276  const dtn::core::Node &n = nodeevent.getNode();
277 
278  if (nodeevent.getAction() == NODE_AVAILABLE)
279  {
280  _taskqueue.push( new SearchNextBundleTask( n.getEID() ) );
281  }
282  else if (nodeevent.getAction() == NODE_DATA_ADDED)
283  {
284  _taskqueue.push( new SearchNextBundleTask( n.getEID() ) );
285  }
286 
287  return;
288  } catch (const std::bad_cast&) { };
289 
290  try {
291  const dtn::net::ConnectionEvent &ce = dynamic_cast<const dtn::net::ConnectionEvent&>(*evt);
292 
293  if (ce.state == dtn::net::ConnectionEvent::CONNECTION_UP)
294  {
295  // send all (multi-hop) bundles in the storage to the neighbor
296  _taskqueue.push( new SearchNextBundleTask(ce.peer) );
297  }
298  return;
299  } catch (const std::bad_cast&) { };
300  }
301 
302  void NeighborRoutingExtension::componentUp() throw ()
303  {
304  // reset the task queue
305  _taskqueue.reset();
306 
307  // routine checked for throw() on 15.02.2013
308  try {
309  // run the thread
310  start();
311  } catch (const ibrcommon::ThreadException &ex) {
312  IBRCOMMON_LOGGER_TAG(NeighborRoutingExtension::TAG, error) << "componentUp failed: " << ex.what() << IBRCOMMON_LOGGER_ENDL;
313  }
314  }
315 
316  void NeighborRoutingExtension::componentDown() throw ()
317  {
318  // routine checked for throw() on 15.02.2013
319  try {
320  // stop the thread
321  stop();
322  join();
323  } catch (const ibrcommon::ThreadException &ex) {
324  IBRCOMMON_LOGGER_TAG(NeighborRoutingExtension::TAG, error) << "componentDown failed: " << ex.what() << IBRCOMMON_LOGGER_ENDL;
325  }
326  }
327 
328  /****************************************/
329 
330  NeighborRoutingExtension::SearchNextBundleTask::SearchNextBundleTask(const dtn::data::EID &e)
331  : eid(e)
332  { }
333 
334  NeighborRoutingExtension::SearchNextBundleTask::~SearchNextBundleTask()
335  { }
336 
337  std::string NeighborRoutingExtension::SearchNextBundleTask::toString()
338  {
339  return "SearchNextBundleTask: " + eid.getString();
340  }
341 
342  /****************************************/
343 
344  NeighborRoutingExtension::ProcessBundleTask::ProcessBundleTask(const dtn::data::MetaBundle &meta, const dtn::data::EID &o, const dtn::data::EID &n)
345  : bundle(meta), origin(o), nexthop(n)
346  { }
347 
348  NeighborRoutingExtension::ProcessBundleTask::~ProcessBundleTask()
349  { }
350 
351  std::string NeighborRoutingExtension::ProcessBundleTask::toString()
352  {
353  return "ProcessBundleTask: " + bundle.toString();
354  }
355  }
356 }