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

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

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