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

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

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