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

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

Last change on this file since 4836 was 3718, checked in by Christoph Mayer, 15 years ago

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