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

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

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