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

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

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