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

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