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

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

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