source: source/ariba/overlay/modules/chord/Chord.cpp@ 10572

Last change on this file since 10572 was 10572, checked in by Michael Tänzer, 12 years ago

Fix DHT: messages got lost if not communicating over a direct link.

Also:

  • Fix mem leak
  • Code clean up
File size: 15.5 KB
Line 
1// [License]
2// The Ariba-Underlay Copyright
3//
4// Copyright (c) 2008-2009, Institute of Telematics, UniversitÀt Karlsruhe (TH)
5//
6// Institute of Telematics
7// UniversitÀt Karlsruhe (TH)
8// Zirkel 2, 76128 Karlsruhe
9// Germany
10//
11// Redistribution and use in source and binary forms, with or without
12// modification, are permitted provided that the following conditions are
13// met:
14//
15// 1. Redistributions of source code must retain the above copyright
16// notice, this list of conditions and the following disclaimer.
17// 2. Redistributions in binary form must reproduce the above copyright
18// notice, this list of conditions and the following disclaimer in the
19// documentation and/or other materials provided with the distribution.
20//
21// THIS SOFTWARE IS PROVIDED BY THE INSTITUTE OF TELEMATICS ``AS IS'' AND
22// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
23// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
24// PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE ARIBA PROJECT OR
25// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
26// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
27// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
28// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
29// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
30// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
31// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
32//
33// The views and conclusions contained in the software and documentation
34// are those of the authors and should not be interpreted as representing
35// official policies, either expressed or implied, of the Institute of
36// Telematics.
37// [License]
38
39#include "ariba/overlay/BaseOverlay.h"
40#include "ariba/overlay/messages/OverlayMsg.h"
41
42#include "Chord.h"
43#include "detail/chord_routing_table.hpp"
44
45#include "messages/Discovery.h"
46
47namespace ariba {
48namespace overlay {
49
50enum signalMessageTypes {
51 typeDiscovery = OverlayMsg::typeSignalingStart + 0x01,
52 typeLeave = OverlayMsg::typeSignalingStart + 0x02,
53};
54
55typedef chord_routing_table::item route_item;
56
57use_logging_cpp( Chord );
58
59Chord::Chord(BaseOverlay& _baseoverlay, const NodeID& _nodeid,
60 OverlayStructureEvents* _eventsReceiver, const OverlayParameterSet& param) :
61 OverlayInterface(_baseoverlay, _nodeid, _eventsReceiver, param) {
62
63 // create routing table
64 this->table = new chord_routing_table(_nodeid, 4);
65 orphan_removal_counter = 0;
66 discovery_count = 0;
67 stabilize_counter = 0;
68 stabilize_finger = 0;
69}
70
71Chord::~Chord() {
72
73 // delete routing table
74 delete table;
75}
76
77/// helper: sets up a link using the base overlay
78LinkID Chord::setup(const EndpointDescriptor& endpoint, const NodeID& remote ) {
79
80 // check if we already have a connection
81 for (size_t i=0; i<table->size(); i++)
82 if ((*table)[i]->ref_count > 0 && (*table)[i]->id == remote && !((*table)[i]->info.isUnspecified()))
83 return LinkID::UNSPECIFIED;
84
85 // check if we are already trying to establish a link
86 for (size_t i=0; i<pending.size(); i++)
87 if ( pending[i] == remote ) {
88 logging_debug("Already trying to establish a link to node "
89 << remote.toString() );
90 return LinkID::UNSPECIFIED;
91 }
92
93 // adding node to list of pending connections
94 pending.push_back( remote );
95
96 logging_info("Request to setup link to " << endpoint.toString() );
97
98 // establish link via base overlay
99 return baseoverlay.establishLink( endpoint, remote,
100 OverlayInterface::OVERLAY_SERVICE_ID );
101}
102
103/// helper: sends a message using the "base overlay"
104seqnum_t Chord::send( OverlayMsg* msg, const LinkID& link ) {
105 if (link.isUnspecified()) return 0;
106 return baseoverlay.send_link( msg, link );
107}
108
109/// sends a discovery message
110void Chord::send_discovery_to(const NodeID& remote, int ttl) {
111 LinkID link = getNextLinkId(remote);
112 if ( remote == nodeid || link.isUnspecified()) return;
113 if ( table->size() == 0 ) return;
114 ttl = 2;
115
116 OverlayMsg msg( typeDiscovery );
117 msg.setRegisterRelay(true);
118 Discovery dmsg( Discovery::normal, (uint8_t)ttl, baseoverlay.getEndpointDescriptor() );
119 msg.encapsulate(&dmsg);
120
121 // send to node
122 baseoverlay.send_node( &msg, remote );
123}
124
125void Chord::discover_neighbors( const LinkID& link ) {
126 uint8_t ttl = 1;
127 {
128 // send predecessor discovery
129 OverlayMsg msg( typeDiscovery );
130 msg.setRegisterRelay(true);
131 Discovery dmsg( Discovery::predecessor, ttl,
132 baseoverlay.getEndpointDescriptor() );
133 msg.encapsulate(&dmsg);
134 send(&msg, link);
135 }
136 {
137 // send successor discovery
138 OverlayMsg msg( typeDiscovery );
139 msg.setSourceEndpoint( baseoverlay.getEndpointDescriptor() );
140 msg.setRegisterRelay(true);
141 Discovery dmsg( Discovery::successor, ttl,
142 baseoverlay.getEndpointDescriptor() );
143 msg.encapsulate(&dmsg);
144 send(&msg, link);
145 }
146}
147
148
149void Chord::createOverlay() {
150}
151
152void Chord::deleteOverlay() {
153
154}
155
156void Chord::joinOverlay(const EndpointDescriptor& boot) {
157 logging_info( "joining Chord overlay structure through end-point " <<
158 (boot.isUnspecified() ? "local" : boot.toString()) );
159
160 // initiator? no->setup first link
161 if (!boot.isUnspecified())
162 bootstrapLinks.push_back( setup(boot) );
163
164 // timer for stabilization management
165 Timer::setInterval(1000);
166 Timer::start();
167}
168
169void Chord::leaveOverlay() {
170 Timer::stop();
171 for (size_t i = 0; i < table->size(); i++) {
172 route_item* it = (*table)[i];
173 OverlayMsg msg( typeLeave );
174 send( &msg, it->info );
175 }
176}
177
178/// @see OverlayInterface.h
179const EndpointDescriptor& Chord::resolveNode(const NodeID& node) {
180 const route_item* item = table->get(node);
181 if (item == NULL || item->info.isUnspecified()) return EndpointDescriptor::UNSPECIFIED();
182 return baseoverlay.getEndpointDescriptor(item->info);
183}
184
185/// @see OverlayInterface.h
186bool Chord::isClosestNodeTo( const NodeID& node ) {
187 return table->is_closest_to(node);
188}
189
190/// @see OverlayInterface.h
191const LinkID& Chord::getNextLinkId( const NodeID& id ) const {
192 // get next hop
193 const route_item* item = table->get_next_hop(id);
194
195 // returns a unspecified id when this is itself
196 if (item == NULL || item->id == nodeid)
197 return LinkID::UNSPECIFIED;
198
199 /// return routing info
200 return item->info;
201}
202
203/// @see OverlayInterface.h
204const NodeID& Chord::getNextNodeId( const NodeID& id ) const {
205 // get next hop
206 const route_item* item = table->get_next_hop(id);
207
208 // return unspecified if no next hop could be found
209 if (item == NULL) {
210 return NodeID::UNSPECIFIED;
211 }
212
213 return item->id;
214}
215
216OverlayInterface::NodeList Chord::getKnownNodes(bool deep) const {
217 OverlayInterface::NodeList nodelist;
218
219 if( deep ){
220 // all nodes that I know, fingers, succ/pred
221 for (size_t i = 0; i < table->size(); i++){
222 if ((*table)[i]->ref_count != 0
223 && !(*table)[i]->info.isUnspecified())
224 nodelist.push_back((*table)[i]->id);
225 }
226 } else {
227 // only succ and pred
228 if( table->get_predesessor() != NULL ){
229 nodelist.push_back( *(table->get_predesessor()) );
230 }
231 if( table->get_successor() != NULL ){
232 OverlayInterface::NodeList::iterator i =
233 std::find( nodelist.begin(), nodelist.end(), *(table->get_successor()) );
234 if( i == nodelist.end() )
235 nodelist.push_back( *(table->get_successor()) );
236 }
237 }
238
239 return nodelist;
240}
241
242/// @see CommunicationListener.h
243/// @see OverlayInterface.h
244void Chord::onLinkUp(const LinkID& lnk, const NodeID& remote) {
245 logging_info("link_up: link=" << lnk.toString() << " remote=" <<
246 remote.toString() );
247 for (vector<NodeID>::iterator i=pending.begin(); i!=pending.end(); i++)
248 if (*i == remote) {
249 pending.erase(i);
250 break;
251 }
252
253 if (remote==nodeid) {
254 logging_warn("dropping link that has been established to myself (nodes have same nodeid?)");
255 baseoverlay.dropLink(lnk);
256 return;
257 }
258
259 route_item* item = table->insert(remote);
260
261 // item added to routing table?
262 if (item != NULL) { // yes-> add to routing table
263 logging_info("new routing neighbor: " << remote.toString()
264 << " with link " << lnk.toString());
265
266 // replace with new link if link is "better"
267 if (item->info!=lnk && item->info.isUnspecified()==false) {
268 if (baseoverlay.compare( item->info, lnk ) == 1) {
269 logging_info("Replacing link due to concurrent link establishment.");
270 baseoverlay.dropLink(item->info);
271 item->info = lnk;
272 }
273 } else {
274 item->info = lnk;
275 }
276
277 // discover neighbors of new overlay neighbor
278 showLinks();
279 } else { // no-> add orphan entry to routing table
280 logging_info("new orphan: " << remote.toString()
281 << " with link " << lnk.toString());
282 table->insert_orphan(remote)->info = lnk;
283 }
284
285 // erase bootstrap link
286 vector<LinkID>::iterator it = std::find(bootstrapLinks.begin(), bootstrapLinks.end(), lnk);
287 if( it != bootstrapLinks.end() ) bootstrapLinks.erase( it );
288}
289
290/// @see CommunicationListener.h or @see OverlayInterface.h
291void Chord::onLinkDown(const LinkID& lnk, const NodeID& remote) {
292 logging_debug("link_down: link=" << lnk.toString() << " remote=" <<
293 remote.toString() );
294
295 // remove link from routing table
296 route_item* item = table->get(remote);
297 if (item!=NULL && item->info==lnk) {
298 item->info = LinkID::UNSPECIFIED;
299 table->remove(remote);
300 }
301}
302
303/// @see CommunicationListener.h
304/// @see OverlayInterface.h
305void Chord::onMessage(const DataMessage& msg, const NodeID& remote,
306 const LinkID& link) {
307
308 // decode message
309 OverlayMsg* m = dynamic_cast<OverlayMsg*>(msg.getMessage());
310 if (m == NULL) return;
311
312 // handle messages
313 switch ((signalMessageTypes)m->getType()) {
314
315 // discovery request
316 case typeDiscovery: {
317 // decapsulate message
318 Discovery* dmsg = m->decapsulate<Discovery> ();
319 logging_debug("Received discovery message with"
320 << " src=" << m->getSourceNode().toString()
321 << " dst=" << m->getDestinationNode().toString()
322 << " ttl=" << (int)dmsg->getTTL()
323 << " type=" << (int)dmsg->getType()
324 );
325
326 // add discovery node id
327 bool found = false;
328 BOOST_FOREACH( NodeID& value, discovery )
329 if (value == m->getSourceNode()) {
330 found = true;
331 break;
332 }
333 if (!found) discovery.push_back(m->getSourceNode());
334
335 // check if source node can be added to routing table and setup link
336 if (m->getSourceNode() != nodeid)
337 setup( dmsg->getEndpoint(), m->getSourceNode() );
338
339 // process discovery message -------------------------- switch start --
340 switch (dmsg->getType()) {
341
342 // normal: route discovery message like every other message
343 case Discovery::normal: {
344 // closest node? yes-> split to follow successor and predecessor
345 if ( table->is_closest_to(m->getDestinationNode()) ) {
346 logging_debug("Discovery split:");
347 if (!table->get_successor()->isUnspecified()) {
348 OverlayMsg omsg(*m);
349 dmsg->setType(Discovery::successor);
350 omsg.encapsulate(dmsg);
351 logging_debug("* Routing to successor "
352 << table->get_successor()->toString() );
353 baseoverlay.send( &omsg, *table->get_successor() );
354 }
355
356 // send predecessor message
357 if (!table->get_predesessor()->isUnspecified()) {
358 OverlayMsg omsg(*m);
359 dmsg->setType(Discovery::predecessor);
360 omsg.encapsulate(dmsg);
361 logging_debug("* Routing to predecessor "
362 << table->get_predesessor()->toString() );
363 baseoverlay.send( &omsg, *table->get_predesessor() );
364 }
365 }
366 // no-> route message
367 else {
368 baseoverlay.route( m );
369 }
370 break;
371 }
372
373 // successor mode: follow the successor until TTL is zero
374 case Discovery::successor:
375 case Discovery::predecessor: {
376 // reached destination? no->forward!
377 if (m->getDestinationNode() != nodeid) {
378 OverlayMsg omsg(*m);
379 omsg.encapsulate(dmsg);
380 omsg.setService(OverlayInterface::OVERLAY_SERVICE_ID);
381 baseoverlay.route( &omsg );
382 break;
383 }
384
385 // time to live ended? yes-> stop routing
386 if (dmsg->getTTL() == 0 || dmsg->getTTL() > 10) break;
387
388 // decrease time-to-live
389 dmsg->setTTL(dmsg->getTTL() - 1);
390
391 const route_item* item = NULL;
392 if (dmsg->getType() == Discovery::successor &&
393 table->get_successor() != NULL) {
394 item = table->get(*table->get_successor());
395 } else {
396 if (table->get_predesessor()!=NULL)
397 item = table->get(*table->get_predesessor());
398 }
399 if (item == NULL)
400 break;
401
402 logging_debug("Routing discovery message to succ/pred "
403 << item->id.toString() );
404 OverlayMsg omsg(*m);
405 omsg.encapsulate(dmsg);
406 omsg.setDestinationNode(item->id);
407 omsg.setService(OverlayInterface::OVERLAY_SERVICE_ID);
408 baseoverlay.send(&omsg, omsg.getDestinationNode());
409 break;
410 }
411 case Discovery::invalid:
412 break;
413
414 default:
415 break;
416 }
417 // process discovery message ---------------------------- switch end --
418
419 delete dmsg;
420 break;
421 }
422
423 // leave
424 case typeLeave: {
425 if (link!=LinkID::UNSPECIFIED) {
426 route_item* item = table->get(remote);
427 if (item!=NULL) item->info = LinkID::UNSPECIFIED;
428 table->remove(remote);
429 baseoverlay.dropLink(link);
430 }
431 break;
432 }}
433}
434
435void Chord::eventFunction() {
436 stabilize_counter++;
437 if (stabilize_counter < 0 || stabilize_counter == 2) {
438
439 // reset counter
440 stabilize_counter = 0;
441
442 // clear pending connections
443 pending.clear();
444
445 // get number of real neighbors
446 size_t numNeighbors = 0;
447 for (size_t i = 0; i < table->size(); i++) {
448 route_item* it = (*table)[i];
449 if (it->ref_count != 0 && !it->info.isUnspecified()) numNeighbors++;
450 }
451 logging_info("Running stabilization: #links="
452 << table->size() << " #neighbors=" << numNeighbors );
453
454 // updating neighbors
455 logging_debug("Discover new ring neighbors");
456 for (size_t i=0; i<table->size(); i++) {
457 LinkID id = (*table)[i]->info;
458 if (!id.isUnspecified()) discover_neighbors(id);
459 }
460
461 // sending discovery
462 logging_debug("Sending discovery message to my neighbors and fingers");
463 stabilize_finger = ((stabilize_finger+1) % table->get_finger_table_size() );
464 const NodeID disc = table->get_finger_table(stabilize_finger).get_compare().get_center();
465 if (disc != nodeid)
466 send_discovery_to(disc);
467
468 // remove orphan links
469 orphan_removal_counter++;
470 if (orphan_removal_counter <0 || orphan_removal_counter >= 2) {
471 logging_info("Discovered nodes: ");
472 BOOST_FOREACH( NodeID& id, discovery )
473 logging_info("* " << id.toString());
474 discovery.clear();
475 logging_info("Running orphan removal");
476 orphan_removal_counter = 0;
477 for (size_t i = 0; i < table->size(); i++) {
478 route_item* it = (*table)[i];
479 if (it->ref_count == 0 && !it->info.isUnspecified()) {
480 logging_info("Dropping orphaned link " << it->info.toString() << " to " << it->id.toString());
481 table->insert(it->id);
482 if (it->ref_count==0) {
483 LinkID id = it->info;
484 it->info = LinkID::UNSPECIFIED;
485 baseoverlay.dropLink(id);
486 }
487 }
488 }
489 }
490 }
491}
492
493void Chord::showLinks() {
494 logging_info("--- chord routing information ----------------------------------");
495 logging_info("predecessor: " << (table->get_predesessor()==NULL? "<none>" :
496 table->get_predesessor()->toString()) );
497 logging_info("node_id : " << nodeid.toString() );
498 logging_info("successor : " << (table->get_successor()==NULL? "<none>" :
499 table->get_successor()->toString()));
500 logging_info("----------------------------------------------------------------");
501}
502
503/// @see OverlayInterface.h
504std::string Chord::debugInformation() const {
505 std::ostringstream s;
506 s << "protocol : Chord" << endl;
507 s << "node_id : " << nodeid.toString() << endl;
508 s << "predecessor: " << (table->get_predesessor()==NULL? "<none>" :
509 table->get_predesessor()->toString()) << endl;
510 s << "successor : " << (table->get_successor()==NULL? "<none>" :
511 table->get_successor()->toString()) << endl;
512 s << "nodes: " << endl;
513 for (size_t i = 0; i < table->size(); i++) {
514 route_item* it = (*table)[i];
515 if (it->ref_count != 0 && !it->info.isUnspecified()) {
516 s << it->id.toString().substr(0,6)
517 << " using " << it->info.toString().substr(0,6) << endl;
518 }
519 }
520 return s.str();
521}
522
523
524
525}} // namespace ariba, overlay
Note: See TracBrowser for help on using the repository browser.