00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018
00019
00020
00021
00022
00023
00024
00025
00026
00027
00028
00029
00030
00031
00032
00033
00034
00035
00036
00037
00038
00039 #include "OneHop.h"
00040 #include "ariba/overlay/BaseOverlay.h"
00041
00042 #include "ariba/overlay/modules/onehop/messages/OneHopMessage.h"
00043 #include "ariba/overlay/modules/onehop/messages/NodeListingRequest.h"
00044 #include "ariba/overlay/modules/onehop/messages/NodeListingReply.h"
00045
00046 namespace ariba {
00047 namespace overlay {
00048
00049 use_logging_cpp( OneHop );
00050
00051 OneHop::OneHop(BaseOverlay& _baseoverlay, const NodeID& _nodeid,
00052 OverlayStructureEvents* _eventsReceiver, const OverlayParameterSet& param)
00053 : OverlayInterface( _baseoverlay, _nodeid, _eventsReceiver, param ),
00054 state ( OneHopStateInvalid ),
00055 bootstrapLink ( LinkID::UNSPECIFIED ),
00056 pendingLinks ( 0 ) {
00057
00058
00059
00060
00061 overlayNodes.insert( make_pair(_nodeid, LinkID::UNSPECIFIED) );
00062
00063 Timer::setInterval(5000);
00064 Timer::start();
00065 }
00066
00067 OneHop::~OneHop(){
00068 Timer::stop();
00069 deleteOverlay();
00070 }
00071
00072 const EndpointDescriptor& OneHop::resolveNode(const NodeID& node){
00073
00074 OverlayNodeMapping::const_iterator i = overlayNodes.find( node );
00075 if (i == overlayNodes.end()) return EndpointDescriptor::UNSPECIFIED;
00076
00077 const EndpointDescriptor& ep = baseoverlay.getEndpointDescriptor( i->second );
00078
00079 logging_debug( "resolved node " << node.toString() << " to endpoint " << ep.toString() );
00080 return ep;
00081 }
00082
00083 void OneHop::routeMessage(const NodeID& destnode, Message* msg){
00084
00085
00086
00087
00088 logging_debug( "routing message to node " << destnode.toString() );
00089
00090 OverlayNodeMapping::const_iterator i = overlayNodes.find( destnode );
00091 if (i == overlayNodes.end()) {
00092 logging_error( "not able to route message to node " << destnode.toString() );
00093 return;
00094 }
00095 OneHopMessage onehopRoute( OneHopMessage::OneHopMessageTypeRoute );
00096 onehopRoute.encapsulate(msg);
00097
00098 baseoverlay.sendMessage( &onehopRoute, i->second );
00099 }
00100
00101 void OneHop::createOverlay() {
00102
00103
00104 logging_info( "creating onehop overlay structure" );
00105 state = OneHopStateCompleted;
00106 }
00107
00108 void OneHop::deleteOverlay(){
00109
00110 logging_info( "deleting onehop overlay structure" );
00111 state = OneHopStateInvalid;
00112 pendingLinks = 0;
00113 }
00114
00115 OverlayInterface::NodeList OneHop::getKnownNodes() const {
00116
00117 OverlayInterface::NodeList retlist;
00118
00119 OverlayNodeMapping::const_iterator i = overlayNodes.begin();
00120 OverlayNodeMapping::const_iterator iend = overlayNodes.end();
00121
00122 for( ; i != iend; i++ )
00123 retlist.push_back( i->first );
00124
00125 return retlist;
00126 }
00127
00128 void OneHop::joinOverlay(const EndpointDescriptor& bootstrapEp){
00129
00130 logging_info( "joining onehop overlay structure through end-point " <<
00131 (bootstrapEp == EndpointDescriptor::UNSPECIFIED ?
00132 "local" : bootstrapEp.toString()) );
00133
00134 state = OneHopStateJoinInitiated;
00135 pendingLinks = 0;
00136
00137 if( bootstrapEp == EndpointDescriptor::UNSPECIFIED ){
00138
00139
00140
00141
00142
00143 state = OneHopStateCompleted;
00144 } else {
00145 bootstrapLink = baseoverlay.establishLink( bootstrapEp,
00146 OverlayInterface::OVERLAY_SERVICE_ID );
00147 }
00148 }
00149
00150 void OneHop::leaveOverlay(){
00151
00152 logging_info( "leaving onehop overlay structure" );
00153
00154
00155
00156
00157
00158
00159 state = OneHopStateInvalid;
00160
00161
00162
00163
00164
00165
00166 OverlayNodeMapping::iterator i = overlayNodes.begin();
00167 OverlayNodeMapping::iterator iend = overlayNodes.end();
00168
00169 for( ; i != iend; i++){
00170 if( i->first != nodeid && i->second != LinkID::UNSPECIFIED ){
00171
00172 OneHopMessage msg (OneHopMessage::OneHopMessageTypeLeave);
00173 baseoverlay.sendMessage( &msg, i->second );
00174 }
00175 }
00176
00177 pendingLinks = 0;
00178 }
00179
00180
00181 void OneHop::onLinkDown(const LinkID& lnk, const NodeID& remote){
00182
00183
00184
00185 if( state == OneHopStateInvalid ) return;
00186
00187
00188 logging_debug( "link " << lnk.toString() << " to node " << remote.toString() << " went down, removing node" );
00189
00190 OverlayNodeMapping::iterator i = overlayNodes.begin();
00191 OverlayNodeMapping::iterator iend = overlayNodes.end();
00192
00193 for( ; i != iend; i++ ){
00194 if( i->second == lnk ){
00195 overlayNodes.erase( i );
00196 break;
00197 }
00198 }
00199 }
00200
00201 void OneHop::onLinkUp(const LinkID& lnk, const NodeID& remote){
00202
00203
00204
00205
00206
00207
00208 if( lnk != bootstrapLink ){
00209 if( pendingLinks > 0 ) pendingLinks--;
00210 if( pendingLinks == 0 ) state = OneHopStateCompleted;
00211 }
00212
00213 logging_debug( "link is up, sending out node listing request" );
00214
00215 NodeListingRequest requestmsg;
00216 OneHopMessage onemsg( OneHopMessage::OneHopMessageTypeListingRequest );
00217 onemsg.encapsulate( &requestmsg );
00218
00219 state = OneHopStateJoinListingRequested;
00220 baseoverlay.sendMessage( &onemsg, lnk );
00221 }
00222
00223 void OneHop::onMessage(const DataMessage& msg, const NodeID& remote, const LinkID& lnk){
00224
00225 OneHopMessage* onemsg = msg.getMessage()->convert<OneHopMessage>();
00226 if( onemsg == NULL ) return;
00227
00228
00229
00230
00231
00232 if( onemsg->isType( OneHopMessage::OneHopMessageTypeListingRequest ) ){
00233
00234 NodeListingRequest* request = onemsg->decapsulate<NodeListingRequest>();
00235
00236 logging_info( "onehop received node listing request from node " << remote.toString() );
00237
00238
00239
00240
00241
00242 overlayNodes.insert( make_pair(remote, lnk) );
00243
00244
00245
00246
00247
00248
00249 OneHopMessage onehopReply( OneHopMessage::OneHopMessageTypeListingReply );
00250 NodeListingReply listingReply;
00251
00252 OverlayNodeMapping::iterator i = overlayNodes.begin();
00253 OverlayNodeMapping::iterator iend = overlayNodes.end();
00254
00255 logging_debug( "sending out node listing reply with the following items" );
00256
00257 for( ; i != iend; i++ ){
00258
00259 const NodeID node = i->first;
00260 const LinkID link = i->second;
00261 const EndpointDescriptor& endpoint = baseoverlay.getEndpointDescriptor( link );
00262
00263 logging_debug( "node: " + node.toString() + ", endp: " + endpoint.toString());
00264 listingReply.add( node, const_cast<EndpointDescriptor*>(new EndpointDescriptor(endpoint)) );
00265 }
00266
00267 onehopReply.encapsulate( &listingReply );
00268 baseoverlay.sendMessage( &onehopReply, lnk );
00269
00270
00271
00272
00273
00274
00275 eventsReceiver->onNodeJoin( remote );
00276
00277 }
00278
00279
00280
00281
00282
00283 if( onemsg->isType( OneHopMessage::OneHopMessageTypeListingReply) ){
00284
00285 NodeListingReply* reply = onemsg->decapsulate<NodeListingReply>();
00286
00287 logging_debug( "received node listing reply from node " << remote.toString()
00288 << " with all overlay nodes. connecting to all of them" );
00289
00290
00291
00292
00293
00294
00295
00296 const NodeListingReply::NodeEndpointList& endpoints = reply->getList();
00297 logging_debug( "received " << endpoints.size() << " nodes in listing" );
00298 pendingLinks = 0;
00299
00300 NodeListingReply::NodeEndpointList::const_iterator i = endpoints.begin();
00301 NodeListingReply::NodeEndpointList::const_iterator iend = endpoints.end();
00302
00303 for( ; i != iend; i++ ){
00304
00305
00306
00307
00308
00309
00310 const NodeID& node = (*i).first;
00311 if( overlayNodes.find(node) != overlayNodes.end() ) continue;
00312 if( node == nodeid ) continue;
00313
00314 logging_debug( "building up link to node in overlay " << node.toString() );
00315 const LinkID link = baseoverlay.establishLink( *((*i).second),
00316 OverlayInterface::OVERLAY_SERVICE_ID );
00317
00318 overlayNodes.insert( make_pair(node, link) );
00319 pendingLinks++;
00320
00321 }
00322
00323 }
00324
00325
00326
00327
00328
00329 if( onemsg->isType(OneHopMessage::OneHopMessageTypeLeave) ){
00330
00331 logging_debug("received leave message from " <<
00332 remote.toString() << " on link " << lnk.toString());
00333
00334
00335 baseoverlay.dropLink( lnk );
00336
00337 }
00338
00339
00340
00341
00342
00343 if( onemsg->isType( OneHopMessage::OneHopMessageTypeRoute) ){
00344 logging_debug( "Route message arrived at destination node -> delegate to BaseOverlay" );
00345 baseoverlay.incomingRouteMessage( onemsg );
00346 }
00347
00348 }
00349
00350 void OneHop::eventFunction(){
00351
00352 logging_debug("<<<<<<<<<<<<<<<<onehop-table<<<<<<<<<<<<<<<<<<<");
00353
00354 OverlayNodeMapping::iterator i = overlayNodes.begin();
00355 OverlayNodeMapping::iterator iend = overlayNodes.end();
00356
00357 for( ; i != iend; i++ ){
00358
00359 const NodeID node = i->first;
00360 const LinkID link = i->second;
00361 const EndpointDescriptor& endpoint = baseoverlay.getEndpointDescriptor( link );
00362
00363 logging_debug( "node: " << node.toString() <<
00364 ", link_: " << link.toString() << ", endp: " << endpoint.toString());
00365 }
00366
00367 logging_debug(">>>>>>>>>>>>>>>>>onehop-table>>>>>>>>>>>>>>>>>>>>>");
00368
00369 }
00370
00371 }}