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

source: source/ariba/communication/BaseCommunication.cpp@ 5539

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

static initialization workaround
File size: 20.5 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 "BaseCommunication.h"
40
41#ifdef UNDERLAY_OMNET
42 #include "ariba/communication/modules/transport/omnet/AribaOmnetModule.h"
43 #include "ariba/communication/modules/network/omnet/OmnetNetworkProtocol.h"
44 #include "ariba/utility/system/StartupWrapper.h"
45
46 using ariba::communication::AribaOmnetModule;
47 using ariba::communication::OmnetNetworkProtocol;
48 using ariba::utility::StartupWrapper;
49#endif
50
51namespace ariba {
52namespace communication {
53
54#include "networkinfo/AddressDiscovery.hpp"
55
56use_logging_cpp(BaseCommunication);
57
58BaseCommunication::BaseCommunication() {
59 this->transport = NULL;
60 this->started = false;
61}
62
63BaseCommunication::~BaseCommunication(){
64}
65
66void BaseCommunication::start() {
67 logging_info( "Starting up ..." );
68 currentSeqnum = 0;
69
70 // creating transports
71 logging_info( "Creating transports ..." );
72
73#ifdef UNDERLAY_OMNET
74 AribaOmnetModule* module = StartupWrapper::getCurrentModule();
75 module->setServerPort( listenport );
76
77 transport = module;
78 network = new OmnetNetworkProtocol( module );
79#else
80 transport = new transport_peer( localDescriptor.getEndpoints() );
81#endif
82
83 logging_info( "Searching for local locators ..." );
84 discover_endpoints(localDescriptor.getEndpoints());
85 logging_info( "Done. Local endpoints = " << localDescriptor.toString() );
86
87 transport->register_listener( this );
88 transport->start();
89
90#ifndef UNDERLAY_OMNET
91 // bind to the network change detection
92 networkMonitor.registerNotification( this );
93#endif
94
95 // base comm startup done
96 started = true;
97 logging_info( "Started up." );
98}
99
100void BaseCommunication::stop() {
101 logging_info( "Stopping transports ..." );
102
103 transport->stop();
104 delete transport;
105 started = false;
106
107 logging_info( "Stopped." );
108}
109
110bool BaseCommunication::isStarted(){
111 return started;
112}
113
114/// Sets the endpoints
115void BaseCommunication::setEndpoints( string& _endpoints ) {
116 localDescriptor.getEndpoints().assign(_endpoints);
117 logging_info("Setting local end-points: "
118 << localDescriptor.getEndpoints().to_string());
119}
120
121const LinkID BaseCommunication::establishLink(
122 const EndpointDescriptor& descriptor,
123 const LinkID& link_id,
124 const QoSParameterSet& qos,
125 const SecurityParameterSet& sec) {
126
127 // copy link id
128 LinkID linkid = link_id;
129
130 // debug
131 logging_debug( "Request to establish link" );
132
133 // create link identifier
134 if (linkid.isUnspecified()) linkid = LinkID::create();
135
136 // create link descriptor
137 logging_debug( "Creating new descriptor entry with local link id=" << linkid.toString() );
138 LinkDescriptor* ld = new LinkDescriptor();
139 ld->localLink = linkid;
140 addLink( ld );
141
142 // send a message to request new link to remote
143 logging_debug( "Send messages with request to open link to " << descriptor.toString() );
144 AribaBaseMsg baseMsg( AribaBaseMsg::typeLinkRequest, linkid );
145 baseMsg.getLocalDescriptor() = localDescriptor;
146
147 // serialize and send message
148 send( &baseMsg, descriptor );
149
150 return linkid;
151}
152
153void BaseCommunication::dropLink(const LinkID link) {
154
155 logging_debug( "Starting to drop link " + link.toString() );
156
157 // see if we have the link
158 LinkDescriptor& ld = queryLocalLink( link );
159 if( ld.isUnspecified() ) {
160 logging_error( "Don't know the link you want to drop "+ link.toString() );
161 return;
162 }
163
164 // tell the registered listeners
165 BOOST_FOREACH( CommunicationEvents* i, eventListener ) {
166 i->onLinkDown( link, ld.localLocator, ld.remoteLocator );
167 }
168
169 // create message to drop the link
170 logging_debug( "Sending out link close request. for us, the link is closed now" );
171 AribaBaseMsg msg( AribaBaseMsg::typeLinkClose, ld.localLink, ld.remoteLink );
172
173 // send message to drop the link
174 send( &msg, ld );
175
176 // remove from map
177 removeLink(link);
178}
179
180seqnum_t BaseCommunication::sendMessage( const LinkID lid, const Message* message) {
181
182 logging_debug( "Sending out message to link " << lid.toString() );
183
184 // query local link info
185 LinkDescriptor& ld = queryLocalLink(lid);
186 if( ld.isUnspecified() ){
187 logging_error( "Don't know the link with id " << lid.toString() );
188 return -1;
189 }
190
191 // link not up-> error
192 if( !ld.up ) {
193 logging_error("Can not send on link " << lid.toString() << ": link not up");
194 return -1;
195 }
196
197 // create message
198 AribaBaseMsg msg( AribaBaseMsg::typeData, ld.localLink, ld.remoteLink );
199
200 // encapsulate the payload message
201 msg.encapsulate( const_cast<Message*>(message) );
202
203 // send message
204 send( &msg, ld );
205
206 // return sequence number
207 return ++currentSeqnum;
208}
209
210const EndpointDescriptor& BaseCommunication::getEndpointDescriptor(const LinkID link) const {
211 if( link == LinkID::UNSPECIFIED){
212 return localDescriptor;
213 } else {
214 LinkDescriptor& linkDesc = queryLocalLink(link);
215 if (linkDesc.isUnspecified()) return EndpointDescriptor::UNSPECIFIED;
216 return linkDesc.remoteEndpoint;
217 }
218}
219
220void BaseCommunication::registerEventListener(CommunicationEvents* _events){
221 if( eventListener.find( _events ) == eventListener.end() )
222 eventListener.insert( _events );
223}
224
225void BaseCommunication::unregisterEventListener(CommunicationEvents* _events){
226 EventListenerSet::iterator i = eventListener.find( _events );
227 if( i != eventListener.end() )
228 eventListener.erase( i );
229}
230
231SystemEventType TransportEvent("Transport");
232SystemEventType MessageDispatchEvent("MessageDispatchEvent", TransportEvent );
233
234class DispatchMsg {
235public:
236 address_v* local;
237 address_v* remote;
238 Message* message;
239};
240
241/// called when a system event is emitted by system queue
242void BaseCommunication::handleSystemEvent(const SystemEvent& event) {
243
244 // dispatch received messages
245 if ( event.getType() == MessageDispatchEvent ){
246 logging_debug( "Forwarding message receiver" );
247 DispatchMsg* dmsg = event.getData<DispatchMsg>();
248 Message* msg = dmsg->message;
249 receiveMessage(msg, dmsg->local, dmsg->remote);
250 msg->dropPayload();
251 delete dmsg;
252 delete msg;
253 }
254}
255
256/// called when a message is received form transport_peer
257void BaseCommunication::receive_message(transport_protocol* transport,
258 const address_vf local, const address_vf remote, const uint8_t* data,
259 size_t size) {
260
261// logging_debug( "Dispatching message" );
262
263 // convert data
264 Data data_( const_cast<uint8_t*>(data), size * 8 );
265 DispatchMsg* dmsg = new DispatchMsg();
266
267 Message* msg = new Message(data_);
268 dmsg->local = local->clone();
269 dmsg->remote = remote->clone();
270 dmsg->message = msg;
271
272 SystemQueue::instance().scheduleEvent(
273 SystemEvent( this, MessageDispatchEvent, dmsg )
274 );
275}
276
277/// handles a message from the underlay transport
278void BaseCommunication::receiveMessage(const Message* message,
279 const address_v* local, const address_v* remote ){
280
281 /// decapsulate message
282 AribaBaseMsg* msg = ((Message*)message)->decapsulate<AribaBaseMsg>();
283 logging_debug( "Receiving message of type " << msg->getTypeString() );
284
285 // handle message
286 switch (msg->getType()) {
287
288 // ---------------------------------------------------------------------
289 // data message
290 // ---------------------------------------------------------------------
291 case AribaBaseMsg::typeData: {
292 logging_debug( "Received data message, forwarding to overlay" );
293 if( messageReceiver != NULL ) {
294 messageReceiver->receiveMessage(
295 msg, msg->getRemoteLink(), NodeID::UNSPECIFIED
296 );
297 }
298 break;
299 }
300
301 // ---------------------------------------------------------------------
302 // handle link request from remote
303 // ---------------------------------------------------------------------
304 case AribaBaseMsg::typeLinkRequest: {
305 logging_debug( "Received link open request" );
306
307 /// only answer the first request
308 if (!queryRemoteLink(msg->getLocalLink()).isUnspecified()) {
309 logging_debug("Link request already received. Ignore!");
310 break;
311 }
312
313 /// create link ids
314 LinkID localLink = LinkID::create();
315 LinkID remoteLink = msg->getLocalLink();
316 logging_debug( "local=" << local->to_string()
317 << " remote=" << remote->to_string()
318 );
319
320 // check if link creation is allowed by ALL listeners
321 bool allowlink = true;
322 BOOST_FOREACH( CommunicationEvents* i, eventListener ){
323 allowlink &= i->onLinkRequest( localLink, local, remote );
324 }
325
326 // not allowed-> warn
327 if( !allowlink ){
328 logging_warn( "Overlay denied creation of link" );
329 return;
330 }
331
332 // create descriptor
333 LinkDescriptor* ld = new LinkDescriptor();
334 ld->localLink = localLink;
335 ld->remoteLink = remoteLink;
336 ld->localLocator = local->clone();
337 ld->remoteLocator = remote->clone();
338 ld->remoteEndpoint = msg->getLocalDescriptor();
339
340 // add layer 1-3 addresses
341 ld->remoteEndpoint.getEndpoints().add(
342 ld->remoteLocator, endpoint_set::Layer1_3);
343 localDescriptor.getEndpoints().add(
344 local, endpoint_set::Layer1_3
345 );
346
347 // link is now up-> add it
348 ld->up = true;
349 addLink(ld);
350
351 // link is up!
352 logging_debug( "Link (initiated from remote) is up with "
353 << "local(id=" << ld->localLink.toString() << ","
354 << "locator=" << ld->localLocator->to_string() << ") "
355 << "remote(id=" << ld->remoteLink.toString() << ", "
356 << "locator=" << ld->remoteLocator->to_string() << ")"
357 );
358
359 // sending link request reply
360 logging_debug( "Sending link request reply with ids "
361 << "local=" << localLink.toString() << ", "
362 << "remote=" << remoteLink.toString() );
363 AribaBaseMsg reply( AribaBaseMsg::typeLinkReply, localLink, remoteLink );
364 reply.getLocalDescriptor() = localDescriptor;
365 reply.getRemoteDescriptor() = ld->remoteEndpoint;
366
367 send( &reply, *ld );
368
369 // inform listeners about new open link
370 BOOST_FOREACH( CommunicationEvents* i, eventListener ) {
371 i->onLinkUp( localLink, ld->localLocator, ld->remoteLocator);
372 }
373
374 // done
375 break;
376 }
377
378 // ---------------------------------------------------------------------
379 // handle link request reply
380 // ---------------------------------------------------------------------
381 case AribaBaseMsg::typeLinkReply: {
382 logging_debug( "Received link open reply for a link we initiated" );
383
384 // this is a reply to a link open request, so we have already
385 // a link mapping and can now set the remote link to valid
386 LinkDescriptor& ld = queryLocalLink( msg->getRemoteLink() );
387
388 // no link found-> warn!
389 if (ld.isUnspecified()) {
390 logging_warn("Failed to find local link " << msg->getRemoteLink().toString());
391 return;
392 }
393
394 // set remote locator and link id
395 ld.remoteLink = msg->getLocalLink();
396 ld.remoteLocator = remote->clone();
397 localDescriptor.getEndpoints().add(
398 msg->getRemoteDescriptor().getEndpoints(),
399 endpoint_set::Layer1_3
400 );
401 ld.up = true;
402
403 logging_debug( "Link is now up with local id "
404 << ld.localLink.toString() << " and remote id "
405 << ld.remoteLink.toString() );
406
407
408 // inform lisneters about link up event
409 BOOST_FOREACH( CommunicationEvents* i, eventListener ){
410 i->onLinkUp( ld.localLink, ld.localLocator, ld.remoteLocator );
411 }
412
413 // done
414 break;
415 }
416
417 // ---------------------------------------------------------------------
418 // handle link close requests
419 // ---------------------------------------------------------------------
420 case AribaBaseMsg::typeLinkClose: {
421 // get remote link
422 const LinkID& localLink = msg->getRemoteLink();
423 logging_debug( "Received link close request for link " << localLink.toString() );
424
425 // searching for link, not found-> warn
426 LinkDescriptor& linkDesc = queryLocalLink( localLink );
427 if (linkDesc.isUnspecified()) {
428 logging_warn("Failed to find local link " << localLink.toString());
429 return;
430 }
431
432 // inform listeners
433 BOOST_FOREACH( CommunicationEvents* i, eventListener ){
434 i->onLinkDown( linkDesc.localLink,
435 linkDesc.localLocator, linkDesc.remoteLocator );
436 }
437
438 // remove the link descriptor
439 removeLink( localLink );
440
441 // done
442 break;
443 }
444
445 // ---------------------------------------------------------------------
446 // handle link locator changes
447 // ---------------------------------------------------------------------
448 case AribaBaseMsg::typeLinkUpdate: {
449 const LinkID& localLink = msg->getRemoteLink();
450 logging_debug( "Received link update for link "
451 << localLink.toString() );
452
453 // find the link description
454 LinkDescriptor& linkDesc = queryLocalLink( localLink );
455 if (linkDesc.isUnspecified()) {
456 logging_warn("Failed to update local link "
457 << localLink.toString());
458 return;
459 }
460
461 // update the remote locator
462 const address_v* oldremote = linkDesc.remoteLocator;
463 linkDesc.remoteLocator = remote->clone();
464
465 // inform the listeners (local link has _not_ changed!)
466 BOOST_FOREACH( CommunicationEvents* i, eventListener ){
467 i->onLinkChanged(
468 linkDesc.localLink, // linkid
469 linkDesc.localLocator, // old local
470 linkDesc.localLocator, // new local
471 oldremote, // old remote
472 linkDesc.remoteLocator // new remote
473 );
474 }
475
476 // done
477 break;
478 }
479 }
480}
481
482/// add a newly allocated link to the set of links
483void BaseCommunication::addLink( LinkDescriptor* link ) {
484 linkSet.push_back( link );
485}
486
487/// remove a link from set
488void BaseCommunication::removeLink( const LinkID& localLink ) {
489 for(LinkSet::iterator i=linkSet.begin(); i != linkSet.end(); i++){
490 if( (*i)->localLink != localLink) continue;
491 delete *i;
492 linkSet.erase( i );
493 break;
494 }
495}
496
497/// query a descriptor by local link id
498BaseCommunication::LinkDescriptor& BaseCommunication::queryLocalLink( const LinkID& link ) const {
499 for (int i=0; i<linkSet.size();i++)
500 if (linkSet[i]->localLink == link) return (LinkDescriptor&)*linkSet[i];
501
502 LinkDescriptor ret;
503 ret.unspecified = true;
504 return ret;
505}
506
507/// query a descriptor by remote link id
508BaseCommunication::LinkDescriptor& BaseCommunication::queryRemoteLink( const LinkID& link ) const {
509 for (int i=0; i<linkSet.size();i++)
510 if (linkSet[i]->remoteLink == link) return (LinkDescriptor&)*linkSet[i];
511
512 LinkDescriptor ret;
513 ret.unspecified = true;
514 return ret;
515}
516
517LinkIDs BaseCommunication::getLocalLinks( const address_v* addr ) const {
518 LinkIDs ids;
519 for (int i=0; i<linkSet.size(); i++){
520 if( addr == NULL ){
521 ids.push_back( linkSet[i]->localLink );
522 } else {
523 if ( *linkSet[i]->remoteLocator == *addr )
524 ids.push_back( linkSet[i]->localLink );
525 }
526 }
527 return ids;
528}
529
530void BaseCommunication::onNetworkChange(const NetworkChangeInterface::NetworkChangeInfo& info){
531
532#ifdef UNDERLAY_OMNET
533
534 // we have no mobility support for simulations
535 return
536
537#endif // UNDERLAY_OMNET
538
539/*- disabled!
540
541 // we only care about address changes, not about interface changes
542 // as address changes are triggered by interface changes, we are safe here
543 if( info.type != NetworkChangeInterface::EventTypeAddressNew &&
544 info.type != NetworkChangeInterface::EventTypeAddressDelete ) return;
545
546 logging_info( "base communication is handling network address changes" );
547
548 // get all now available addresses
549 NetworkInformation networkInformation;
550 AddressInformation addressInformation;
551
552 NetworkInterfaceList interfaces = networkInformation.getInterfaces();
553 AddressList addresses;
554
555 for( NetworkInterfaceList::iterator i = interfaces.begin(); i != interfaces.end(); i++ ){
556 AddressList newaddr = addressInformation.getAddresses(*i);
557 addresses.insert( addresses.end(), newaddr.begin(), newaddr.end() );
558 }
559
560 //
561 // get current locators for the local endpoint
562 // TODO: this code is dublicate of the ctor code!!! cleanup!
563 //
564
565 NetworkProtocol::NetworkLocatorSet locators = network->getAddresses();
566 NetworkProtocol::NetworkLocatorSet::iterator i = locators.begin();
567 NetworkProtocol::NetworkLocatorSet::iterator iend = locators.end();
568
569 //
570 // remember the old local endpoint, in case it changes
571 //
572
573 EndpointDescriptor oldLocalDescriptor( localDescriptor );
574
575 //
576 // look for local locators that we can use in communication
577 //
578 // choose the first locator that is not localhost
579 //
580
581 bool foundLocator = false;
582 bool changedLocator = false;
583
584 for( ; i != iend; i++){
585 logging_debug( "local locator found " << (*i)->toString() );
586 IPv4Locator* ipv4locator = dynamic_cast<IPv4Locator*>(*i);
587
588 if( *ipv4locator != IPv4Locator::LOCALHOST &&
589 *ipv4locator != IPv4Locator::ANY &&
590 *ipv4locator != IPv4Locator::BROADCAST ){
591
592 ipv4locator->setPort( listenport );
593 changedLocator = *localDescriptor.locator != *ipv4locator;
594 localDescriptor.locator = ipv4locator;
595 logging_info( "binding to addr = " << ipv4locator->toString() );
596 foundLocator = true;
597 break;
598 }
599 } // for( ; i != iend; i++)
600
601 //
602 // if we found no locator, bind to localhost
603 //
604
605 if( !foundLocator ){
606 changedLocator = *localDescriptor.locator != IPv4Locator::LOCALHOST;
607 localDescriptor.locator = new IPv4Locator( IPv4Locator::LOCALHOST );
608 ((IPv4Locator*)(localDescriptor.locator))->setPort( listenport );
609 logging_info( "found no good local lcoator, binding to addr = " <<
610 localDescriptor.locator->toString() );
611 }
612
613 //
614 // if we have connections that have no more longer endpoints
615 // close these. they will be automatically built up again.
616 // also update the local locator in the linkset mapping
617 //
618
619 if( changedLocator ){
620
621 logging_debug( "local endp locator has changed to " << localDescriptor.toString() <<
622 ", resettings connections that end at old locator " <<
623 oldLocalDescriptor.toString());
624
625 LinkSet::iterator i = linkSet.begin();
626 LinkSet::iterator iend = linkSet.end();
627
628 for( ; i != iend; i++ ){
629
630 logging_debug( "checking connection for locator change: " <<
631 " local " << (*i).localLocator->toString() <<
632 " old " << oldLocalDescriptor.locator->toString() );
633
634 if( *((*i).localLocator) == *(oldLocalDescriptor.locator) ){
635
636 logging_debug("terminating connection to " << (*i).remoteLocator->toString() );
637 transport->terminate( oldLocalDescriptor.locator, (*i).remoteLocator );
638
639 (*i).localLocator = localDescriptor.locator;
640 }
641 } // for( ; i != iend; i++ )
642
643 // wait 500ms to give the sockets time to shut down
644 usleep( 500000 );
645
646 } else {
647
648 logging_debug( "locator has not changed, not resetting connections" );
649
650 }
651
652 //
653 // handle the connections that have no longer any
654 // valid locator. send update messages with the new
655 // locator, so the remote node updates its locator/link mapping
656 //
657
658 LinkSet::iterator iAffected = linkSet.begin();
659 LinkSet::iterator endAffected = linkSet.end();
660
661 for( ; iAffected != endAffected; iAffected++ ){
662 LinkDescriptor descr = *iAffected;
663 logging_debug( "sending out link locator update to " << descr.remoteLocator->toString() );
664
665 AribaBaseMsg updateMsg( descr.remoteLocator,
666 AribaBaseMsg::LINK_STATE_UPDATE,
667 descr.localLink, descr.remoteLink );
668
669 transport->sendMessage( &updateMsg );
670 }
671*/
672}
673
674/// sends a message to all end-points in the end-point descriptor
675void BaseCommunication::send(Message* message, const EndpointDescriptor& endpoint) {
676 Data data = data_serialize( message, DEFAULT_V );
677 transport->send( endpoint.getEndpoints(), data.getBuffer(), data.getLength() / 8);
678}
679
680/// sends a message to the remote locator inside the link descriptor
681void BaseCommunication::send(Message* message, const LinkDescriptor& desc) {
682 Data data = data_serialize( message, DEFAULT_V );
683 transport->send( desc.remoteLocator, data.getBuffer(), data.getLength() / 8);
684}
685
686}} // namespace ariba, communication
Note: See TracBrowser for help on using the repository browser.