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

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