close Warning: Can't use blame annotator: 
No changeset 2259 in the repository

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

Last change on this file since 5781 was 5780, checked in by mies, 15 years ago
File size: 15.8 KB
RevLine 
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
41#include "Chord.h"
42#include "messages/ChordMessage.h"
43#include "messages/Discovery.h"
44
45#include "detail/chord_routing_table.hpp"
46
47namespace ariba {
48namespace overlay {
49
50typedef chord_routing_table::item route_item;
51
52use_logging_cpp( Chord );
53
54Chord::Chord(BaseOverlay& _baseoverlay, const NodeID& _nodeid,
55 OverlayStructureEvents* _eventsReceiver, const OverlayParameterSet& param) :
56 OverlayInterface(_baseoverlay, _nodeid, _eventsReceiver, param) {
57
58 // create routing table
59 this->table = new chord_routing_table(_nodeid, 2);
60
61 // init counters
62 orphan_removal_counter = 0;
63 discovery_count = 0;
64 stabilize_counter = 0;
65 stabilize_finger = 0;
66}
67
68Chord::~Chord() {
69
70 // delete routing table
71 delete table;
72}
73
74/// helper: sets up a link using the base overlay
75LinkID Chord::setup(const EndpointDescriptor& endp, const NodeID& node, const NodeID& remoteRelay ) {
76 logging_debug("Request to setup link to " << endp.toString() );
77
78 // check if we already have a connection
79 for (int i=0; i<table->size(); i++)
80 if ((*table)[i]->id == node && !((*table)[i]->info.isUnspecified()))
81 return LinkID::UNSPECIFIED;
82
83 // check if we are already trying to establish a link
84 for (size_t i=0; i<pending.size(); i++)
85 if ( pending[i] == node ) {
86 logging_debug("Already trying to establish a link to node " << node.toString() );
87 return LinkID::UNSPECIFIED;
88 }
89
90 // adding node to list of pending connections
91 pending.push_back( node );
92
93 // establish link via base overlay
94 return baseoverlay.establishLink(endp, node, OverlayInterface::OVERLAY_SERVICE_ID,
95 remoteRelay );
96}
97
98/// helper: sends a message using the "base overlay"
99seqnum_t Chord::send(Message* msg, const LinkID& link) {
100 if (link.isUnspecified()) return 0;
101 return baseoverlay.sendMessage(msg, link);
102}
103
104const LinkID& Chord::get_next_hop( const NodeID& id ) const {
105 BOOST_FOREACH( const back_route& br, back_routes )
106 if (br.id==id) return br.link;
107 const route_item* item = table->get_next_hop(id);
108 if (item==NULL) return LinkID::UNSPECIFIED;
109 return item->info;
110}
111
112/// sends a discovery message
113void Chord::send_discovery_to(const NodeID& destination, int ttl) {
114 if ( table->size() == 0 ) return;
115
116 ChordMessage cmsg(ChordMessage::discovery, nodeid, destination);
117 Discovery dmsg;
118 dmsg.setSourceEndpoint(&baseoverlay.getEndpointDescriptor());
119 dmsg.setSourceRelay(baseoverlay.getRelayNode(destination));
120 dmsg.setFollowType(Discovery::normal);
121 dmsg.setTTL((uint8_t) ttl);
122 cmsg.encapsulate(&dmsg);
123
124 // get next hop
125 LinkID next_link = get_next_hop(destination);
126 if (!next_link.isUnspecified()) send(&cmsg, next_link);
127}
128
129void Chord::discover_neighbors( const LinkID& lnk ) {
130 {
131 // send successor discovery
132 ChordMessage cmsg(ChordMessage::discovery, nodeid, nodeid);
133 Discovery dmsg;
134 dmsg.setSourceEndpoint(&baseoverlay.getEndpointDescriptor());
135 dmsg.setSourceRelay(baseoverlay.getRelayNode(nodeid));
136 dmsg.setFollowType(Discovery::successor);
137 dmsg.setTTL((uint8_t)4);
138 cmsg.encapsulate(&dmsg);
139 send(&cmsg, lnk);
140 }
141 {
142 // send predecessor discovery
143 ChordMessage cmsg(ChordMessage::discovery, nodeid, nodeid);
144 Discovery dmsg;
145 dmsg.setSourceEndpoint(&baseoverlay.getEndpointDescriptor());
146 dmsg.setSourceRelay(baseoverlay.getRelayNode(nodeid));
147 dmsg.setFollowType(Discovery::predecessor);
148 dmsg.setTTL((uint8_t)4);
149 cmsg.encapsulate(&dmsg);
150 send(&cmsg, lnk);
151 }
152}
153
154
155void Chord::createOverlay() {
156}
157
158void Chord::deleteOverlay() {
159
160}
161
162void Chord::joinOverlay(const EndpointDescriptor& boot) {
163 logging_info( "joining Chord overlay structure through end-point " <<
164 (boot.isUnspecified() ? "local" : boot.toString()) );
165
166 // initiator? no->setup first link
167 if (!boot.isUnspecified())
168 bootstrapLinks.push_back( setup(boot) );
169
170 // timer for stabilization management
171 Timer::setInterval(1000);
172 Timer::start();
173}
174
175void Chord::leaveOverlay() {
176 Timer::stop();
177 for (size_t i = 0; i < table->size(); i++) {
178 route_item* it = (*table)[i];
179 ChordMessage msg(ChordMessage::leave, nodeid, it->id);
180 send(&msg,it->info);
181 }
182}
183
184const EndpointDescriptor& Chord::resolveNode(const NodeID& node) {
185 const route_item* item = table->get(node);
186 if (item == NULL || item->info.isUnspecified()) return EndpointDescriptor::UNSPECIFIED();
187 return baseoverlay.getEndpointDescriptor(item->info);
188}
189
190void Chord::routeMessage(const NodeID& destnode, Message* msg) {
191 // get next hop
192 LinkID next_link = get_next_hop(destnode);
193
194 // message for this node? yes-> delegate to base overlay
195 if (destnode == nodeid)
196 baseoverlay.incomingRouteMessage( msg, LinkID::UNSPECIFIED, nodeid );
197
198 else { // no-> send to next hop
199 ChordMessage cmsg(ChordMessage::route, nodeid, destnode);
200 cmsg.encapsulate(msg);
201 send(&cmsg, next_link);
202 }
203}
204
205/// @see OverlayInterface.h
206void Chord::routeMessage(const NodeID& node, const LinkID& link, Message* msg) {
207 logging_debug("Redirect over Chord to node id=" << node.toString()
208 << " link id=" << link.toString() );
209 ChordMessage cmsg(ChordMessage::route, nodeid, node);
210 cmsg.encapsulate(msg);
211 send(&cmsg, link);
212}
213
214/// @see OverlayInterface.h
215const LinkID& Chord::getNextLinkId( const NodeID& id ) const {
216 return get_next_hop(id);
217}
218
219OverlayInterface::NodeList Chord::getKnownNodes(bool deep) const {
220 OverlayInterface::NodeList nodelist;
221
222 if( deep ){
223 // all nodes that I know, fingers, succ/pred
224 for (size_t i = 0; i < table->size(); i++){
225 if (/*(*table)[i]->ref_count != 0
226 &&*/ !(*table)[i]->info.isUnspecified())
227 nodelist.push_back((*table)[i]->id);
228 }
229 } else {
230 // only succ and pred
231 if( table->get_predesessor() != NULL ){
232 nodelist.push_back( *(table->get_predesessor()) );
233 }
234 if( table->get_successor() != NULL ){
235 OverlayInterface::NodeList::iterator i =
236 std::find( nodelist.begin(), nodelist.end(), *(table->get_successor()) );
237 if( i == nodelist.end() )
238 nodelist.push_back( *(table->get_successor()) );
239 }
240 }
241
242 return nodelist;
243}
244
245/// @see CommunicationListener.h
246/// @see OverlayInterface.h
247void Chord::onLinkUp(const LinkID& lnk, const NodeID& remote) {
248 logging_debug("link_up: link=" << lnk.toString() << " remote=" <<
249 remote.toString() );
250 for (vector<NodeID>::iterator i=pending.begin(); i!=pending.end(); i++)
251 if (*i == remote) {
252 pending.erase(i);
253 break;
254 }
255 route_item* item = table->insert(remote);
256
257 // item added to routing table?
258 if (item != NULL) { // yes-> add to routing table
259 logging_info("new routing neighbor: " << remote.toString()
260 << " with link " << lnk.toString());
261 // replace with new link
262 if (!item->info.isUnspecified())
263 baseoverlay.dropLink(item->info);
264 item->info = lnk;
265 } else { // no-> add orphan entry to routing table
266 logging_info("new orphan: " << remote.toString()
267 << " with link " << lnk.toString());
268 table->insert_orphan(remote)->info = lnk;
269 }
270
271 discover_neighbors( lnk );
272
273 vector<LinkID>::iterator it = std::find(bootstrapLinks.begin(), bootstrapLinks.end(), lnk);
274 if( it != bootstrapLinks.end() ) {
275 discover_neighbors( lnk );
276 bootstrapLinks.erase( it );
277 }
278}
279
280/// @see CommunicationListener.h or @see OverlayInterface.h
281void Chord::onLinkDown(const LinkID& lnk, const NodeID& remote) {
282 logging_debug("link_down: link=" << lnk.toString() << " remote=" <<
283 remote.toString() );
284
285 // remove link from routing table
286 route_item* item = table->get(remote);
287 if (item!=NULL) item->info = LinkID::UNSPECIFIED;
288 table->remove(remote);
289}
290
291/// @see CommunicationListener.h
292/// @see OverlayInterface.h
293void Chord::onMessage(const DataMessage& msg, const NodeID& remote,
294 const LinkID& link) {
295
296 // decode message
297 typedef ChordMessage M;
298 M* m = msg.getMessage()->convert<ChordMessage> ();
299 if (m == NULL) return;
300
301 // handle messages
302 switch (m->getType()) {
303
304 // invalid message
305 case M::invalid:
306 break;
307
308 // route message with payload
309 case M::route: {
310 // find next hop
311 LinkID next_link = get_next_hop(m->getDestination());
312
313 // add to back-routes
314 bool found = false;
315 BOOST_FOREACH( back_route br, back_routes) {
316 if (br.id == m->getSource()) {
317 br.link = link;
318 br.lastseen = time(NULL);
319 found = true;
320 break;
321 }
322 }
323 if (!found) {
324 back_route br;
325 br.id = m->getSource();
326 br.link = link;
327 br.lastseen = time(NULL);
328 back_routes.push_back(br);
329 }
330
331 // next hop == myself?
332 if (m->getDestination() == nodeid) { // yes-> route to base overlay
333 logging_debug("Send message to baseoverlay");
334 baseoverlay.incomingRouteMessage( m, next_link, remote );
335 }
336 // no-> route to next hop
337 else {
338// logging_debug("Route chord message to "
339// << item->id.toString() << " (destination=" << m->getDestination() << ")");
340 send(m, next_link);
341 }
342 break;
343 }
344
345 // discovery request
346 case M::discovery: {
347 // decapsulate message
348 Discovery* dmsg = m->decapsulate<Discovery> ();
349 logging_debug("received discovery message with"
350 << " src=" << m->getSource().toString()
351 << " dst=" << m->getDestination().toString()
352 << " ttl=" << (int)dmsg->getTTL()
353 << " type=" << (int)dmsg->getFollowType()
354 );
355
356 // check if source node can be added to routing table and setup link
357 if (m->getSource() != nodeid && table->is_insertable(m->getSource()))
358 setup(*dmsg->getSourceEndpoint(), m->getSource(), dmsg->getSourceRelay() );
359
360 // delegate discovery message
361 switch (dmsg->getFollowType()) {
362
363 // normal: route discovery message like every other message
364 case Discovery::normal: {
365 // closest node? yes-> split to follow successor and predecessor
366 if (table->is_closest_to(m->getDestination())
367 || (table->get_successor()!=NULL && *table->get_successor() == m->getDestination())
368 || (table->get_predesessor()!=NULL && *table->get_predesessor() == m->getDestination())
369 ) {
370
371 if (table->get_successor() != NULL) {
372 // send successor message
373 ChordMessage cmsg_s(*m);
374 Discovery dmsg_s(*dmsg);
375 dmsg_s.setFollowType(Discovery::successor);
376 cmsg_s.encapsulate(&dmsg_s);
377 route_item* succ_item = table->get(*table->get_successor());
378 logging_debug("Discovery split: routing discovery message to successor "
379 << succ_item->id.toString() );
380 cmsg_s.setDestination(succ_item->id);
381 send(&cmsg_s, succ_item->info);
382 }
383
384 // send predecessor message
385 if (table->get_predesessor() != NULL) {
386 ChordMessage cmsg_p(*m);
387 Discovery dmsg_p(*dmsg);
388 dmsg_p.setFollowType(Discovery::predecessor);
389 cmsg_p.encapsulate(&dmsg_p);
390 route_item* pred_item = table->get(
391 *table->get_predesessor());
392 logging_debug("Discovery split: routing discovery message to predecessor "
393 << pred_item->id.toString() );
394 cmsg_p.setDestination(pred_item->id);
395 send(&cmsg_p, pred_item->info);
396 }
397 }
398 // no-> route message
399 else {
400 // find next hop
401 LinkID next_link = get_next_hop(m->getDestination());
402 // logging_debug("routing discovery message to " <<
403 // linkitem->id.toString() );
404 send(m, next_link);
405 }
406 break;
407 }
408
409 // successor mode: follow the successor until TTL is zero
410 case Discovery::successor:
411 case Discovery::predecessor: {
412 // time to live ended? yes-> stop routing
413 if (dmsg->getTTL() == 0 || dmsg->getTTL() > 10) break;
414
415 // decrease time-to-live
416 dmsg->setTTL(dmsg->getTTL() - 1);
417
418 const route_item* item = NULL;
419 if (dmsg->getFollowType() == Discovery::successor &&
420 table->get_successor() != NULL) {
421 item = table->get(*table->get_successor());
422 } else {
423 if (table->get_predesessor()!=NULL)
424 item = table->get(*table->get_predesessor());
425 }
426 if (item == NULL) break;
427 logging_debug("routing discovery message to succ/pred "
428 << item->id.toString() );
429 ChordMessage cmsg(*m);
430 Discovery dmsg_p(*dmsg);
431 cmsg.encapsulate(&dmsg_p);
432 send(&cmsg, item->info);
433 break;
434 }}
435 delete dmsg;
436 break;
437 }
438
439 // leave
440 case M::leave: {
441 if (link!=LinkID::UNSPECIFIED) {
442 route_item* item = table->get(remote);
443 if (item!=NULL) item->info = LinkID::UNSPECIFIED;
444 table->remove(remote);
445 baseoverlay.dropLink(link);
446 }
447 break;
448 }
449 }
450 delete m;
451}
452
453void Chord::eventFunction() {
454 stabilize_counter++;
455 if (stabilize_counter < 0 || stabilize_counter >= 2) {
456
457 // reset counter
458 stabilize_counter = 0;
459
460 // clear pending connections
461 pending.clear();
462
463 // get number of real neighbors
464 size_t numNeighbors = 0;
465 for (size_t i = 0; i < table->size(); i++) {
466 route_item* it = (*table)[i];
467 if (it->ref_count != 0 && !it->info.isUnspecified()) numNeighbors++;
468 }
469 logging_info("Running stabilization: #links="
470 << table->size() << " #neighbors=" << numNeighbors );
471
472 // sending discovery
473 logging_debug("Sending discovery message to my neighbors and fingers");
474 stabilize_finger = ((stabilize_finger+1) % table->get_finger_table_size() );
475 const NodeID disc1 = nodeid;
476 const NodeID disc2 = table->get_finger_table(stabilize_finger).get_compare().get_center();
477 send_discovery_to(disc1);
478 if (disc1 != disc2)
479 send_discovery_to(disc2);
480
481 for (int i=0; i<table->size(); i++) {
482 LinkID id = (*table)[i]->info;
483 if (!id.isUnspecified()) discover_neighbors(id);
484 }
485
486 // remove orphan links
487 orphan_removal_counter++;
488 if (orphan_removal_counter <0 || orphan_removal_counter >= 4) {
489 back_routes.clear();
490/* for (vector<back_route>::iterator i = back_routes.begin();
491 i!=back_routes.end(); i++) {
492 back_route& br = *i;
493 if (difftime(br.lastseen,time(NULL))>5) i =
494 back_routes.erase(i)-1;
495 }
496*/
497 logging_info("Running orphan removal");
498 orphan_removal_counter = 0;
499 for (size_t i = 0; i < table->size(); i++) {
500 route_item* it = (*table)[i];
501 if (it->ref_count == 0 && !it->info.isUnspecified()) {
502 logging_info("Dropping orphaned link " << it->info.toString() << " to " << it->id.toString());
503 table->insert(it->id);
504 if (it->ref_count==0) {
505 baseoverlay.dropLink(it->info);
506 it->info = LinkID::UNSPECIFIED;
507 }
508 }
509 }
510 }
511 }
512 logging_info("--- chord routing information ----------------------------------");
513 logging_info("predecessor: " << (table->get_predesessor()==NULL? "<none>" :
514 table->get_predesessor()->toString()) );
515 logging_info("node_id : " << nodeid.toString() );
516 logging_info("successor : " << (table->get_successor()==NULL? "<none>" :
517 table->get_successor()->toString()));
518 logging_info("----------------------------------------------------------------");
519}
520
521}} // namespace ariba, overlay
Note: See TracBrowser for help on using the repository browser.