349 lines
12 KiB
C++
349 lines
12 KiB
C++
/*
|
|
* File: UnicastTransfer.cc
|
|
* Author: jgaebler
|
|
*
|
|
* Created on October 25, 2012, 2:30 PM
|
|
*/
|
|
|
|
#include "UnicastTransfer.h"
|
|
|
|
#include "Moversight.h"
|
|
#include "Dispatcher.h"
|
|
|
|
#include "MessagReferenceCircularBuffer.h"
|
|
|
|
#include "common/transport/MessageReference.h"
|
|
#include "msg/UnicastMessage.h"
|
|
#include "msg/UnicastMessageConfirm.h"
|
|
#include "timer/UnicastMessageRetransmitTimer.h"
|
|
|
|
#include "ms/events/JoinGroupDoneEvent.h"
|
|
#include "ut/events/UnicastMessageTransferDoneEvent.h"
|
|
#include "ut/events/UnicastMessageTransferFailedEvent.h"
|
|
|
|
namespace ubeeme {
|
|
namespace moversight {
|
|
|
|
#undef DEBUG
|
|
#define DEBUG(msg) if( module.isPrintDebugUT()) MOV_DEBUG << "UT@" << getLocalID() << " " << msg << endl;
|
|
|
|
/**
|
|
* @brief Constructor
|
|
* @param dis A reference to the dispatcher instance
|
|
*/
|
|
UnicastTransfer::UnicastTransfer(Dispatcher & dis) : MoversightService(dis, "UnicastTransfer") {
|
|
}
|
|
|
|
/**
|
|
* @brief Destructor
|
|
*/
|
|
UnicastTransfer::~UnicastTransfer() {
|
|
}
|
|
|
|
/**
|
|
* @brief Initialize the service
|
|
*/
|
|
void
|
|
UnicastTransfer::initialise() {
|
|
dispatcher.subscribe<JoinGroupDoneEvent>(this);
|
|
|
|
//init the message reference counter
|
|
currentMessageReference.setPeerID(getLocalID());
|
|
}
|
|
|
|
/**
|
|
* @brief Finalise the service.
|
|
*/
|
|
void
|
|
UnicastTransfer::finalise() {
|
|
}
|
|
|
|
/**
|
|
* @brief Assignment operator
|
|
* @param other The instance to assign
|
|
* @return A reference to the local instance
|
|
*/
|
|
UnicastTransfer &
|
|
UnicastTransfer::operator=(const UnicastTransfer & other) {
|
|
if (this != &other) { // protect against invalid self-assignment
|
|
|
|
MoversightService::operator =(other);
|
|
|
|
}//End if
|
|
|
|
return *this;
|
|
}
|
|
|
|
/**
|
|
* @brief Handles the joinGroupDone event fired by the membership service
|
|
* @param localPeer A reference of the local peer.
|
|
*/
|
|
void
|
|
UnicastTransfer::handleEvent( const JoinGroupDoneEvent& e) {
|
|
// update the peer id of the message reference
|
|
currentMessageReference.setPeerID( e.getPeer().getPeerID());
|
|
}
|
|
|
|
/**
|
|
* @brief Handle a received unicast message.
|
|
* @param pdu The received message.
|
|
*
|
|
* A received message will be confirmed by a confirm message, pointing to the message
|
|
* reference of the received message. If the message has not been seen before, it will be
|
|
* delivered locally.
|
|
*/
|
|
void
|
|
UnicastTransfer::handleMessage( UnicastMessage & pdu) {
|
|
|
|
DEBUG("handleMessage - receive message");
|
|
|
|
// did we see this message before?
|
|
// aka. is this a confirmation?
|
|
MessageReference mRef = pdu.getMessageReference();
|
|
|
|
if( lastSeenMessageReferences.contains(mRef) == false) {
|
|
// no
|
|
// => remember and deliver the message
|
|
lastSeenMessageReferences.add(mRef);
|
|
deliverMessage(pdu);
|
|
}//end if
|
|
|
|
// send an ack and deliver the message
|
|
confirmReceivedMessage( pdu);
|
|
}
|
|
|
|
/**
|
|
* @brief Confirms the given message.
|
|
* @param pdu The message to confirm.
|
|
*
|
|
* Send a confirmation back to the sender, which contains the message
|
|
* reference of the received pdu.
|
|
*/
|
|
void
|
|
UnicastTransfer::confirmReceivedMessage( UnicastMessage & pdu) {
|
|
// extract the origin of this message
|
|
TransportAddress dest = pdu.getSourceTA();
|
|
// create the answer and insert the message reference of the received pdu.
|
|
UnicastMessageConfirm ack( pdu.getMessageReference());
|
|
// deliver the confirmation
|
|
sendTo( ack, dest);
|
|
}
|
|
|
|
/**
|
|
* @brief Handle an incoming confirmation.
|
|
* @param pdu Confirmation PDU.
|
|
*/
|
|
void
|
|
UnicastTransfer::handleMessage( UnicastMessageConfirm & pdu) {
|
|
DEBUG("handleMessage - received message confirm.");
|
|
|
|
// get the stored message reference
|
|
MessageReference mRef = pdu.getMessageReference();
|
|
|
|
// check if the message was sent by the local peer
|
|
if( retransmitMessageQueue.contains(mRef)) {
|
|
|
|
// stop and delete the retransmit timer
|
|
stopAndDeleteTimer( mRef);
|
|
|
|
// signal the confirmation to the correspondent upper layer
|
|
// Note that all services that registered to the emitted event type
|
|
// will receive the notification.
|
|
dispatcher.signal( new UnicastMessageTransferDoneEvent(mRef));
|
|
|
|
// delete the stored message from the retransmit queue
|
|
deleteMessage(mRef);
|
|
}//End if
|
|
}
|
|
|
|
/**
|
|
* @brief Handles the timeout of an unicast message retransmit timer.
|
|
* @param timer The timer to handle.
|
|
*
|
|
* The corresponding message is resent, unless the retry counter falls
|
|
* below 0. If so, the transmission is signaled to the upper layer as failed,
|
|
* thus deleting the message reference and timer.
|
|
*/
|
|
void
|
|
UnicastTransfer::handleTimeout( UnicastMessageRetransmitTimer * timer) {
|
|
// get the corresponding message reference.
|
|
MessageReference mRef = timer->getReference();
|
|
|
|
// try to find the message among those that are to be retransmitted.
|
|
try {
|
|
UnicastMessage* pdu = retransmitMessageQueue.find(mRef);
|
|
|
|
// the number of retries is actually decreased with each retransmission.
|
|
if( timer->getNumberOfRetries() > 0) {
|
|
// retransmit the pdu.
|
|
TransportAddress dest = timer->getSource();
|
|
sendTo( *pdu, dest);
|
|
|
|
// restart the timer.
|
|
timer->restart();
|
|
return;
|
|
}//End if
|
|
|
|
// the maximum number of retries is reached, so stop
|
|
// retransmission and clean up
|
|
deleteMessage( mRef);
|
|
}//End try
|
|
catch( EntryNotFoundException& /* e */) {
|
|
//message not found clean up timer
|
|
}//End catch
|
|
|
|
stopAndDeleteTimer(mRef);
|
|
|
|
// signal the failing upwards
|
|
// Note that every registered service will receive this event.
|
|
dispatcher.signal( new UnicastMessageTransferFailedEvent( mRef));
|
|
}
|
|
|
|
/**
|
|
* @brief Send a message either reliable or unreliable to a destination host.
|
|
* @param pdu The message to send.
|
|
* @param dest Address of the destination host.
|
|
* @param reliable Whether to send this message reliable or unreliable.
|
|
*/
|
|
void
|
|
UnicastTransfer::send( const MoversightMessage& pdu, const TransportAddress& dest, const bool reliable) {
|
|
if( reliable == true) {
|
|
// do something
|
|
}
|
|
|
|
dispatcher.getMoversight().sendToPeer( pdu, dest);
|
|
}
|
|
|
|
/**
|
|
* @brief Send an unicast message to the given destination.
|
|
* @param msg The message to send.
|
|
* @param destination The destination of the message.
|
|
* @return The message reference of the sent message, used to monitor its delivery.
|
|
*/
|
|
MessageReference
|
|
UnicastTransfer::send( UnicastMessage& msg, const PeerID& destination) {
|
|
// get the transport address of the destination peer.
|
|
TransportAddress ta = dispatcher.getMembershipService().getPeer(destination).getLocalAddress();
|
|
return send( msg, ta);
|
|
}
|
|
|
|
/**
|
|
* @brief Send an unicast message to the given destination.
|
|
* @param msg The message to send.
|
|
* @param destination The destination of the message.
|
|
* @return The message reference of the sent message, used to monitor its delivery.
|
|
*/
|
|
MessageReference
|
|
UnicastTransfer::send( UnicastMessage& msg, const TransportAddress& destination) {
|
|
|
|
// create a new message reference
|
|
MessageReference mRef = createMesssageReference();
|
|
|
|
// set the source and message reference fields of the message
|
|
msg.setSourceTA( getLocalAddress());
|
|
msg.setMessageReference( mRef);
|
|
|
|
std::stringstream buf;
|
|
buf << "send - send message to peer " << destination;
|
|
DEBUG( buf.str().c_str());
|
|
|
|
sendTo( msg, destination);
|
|
|
|
// monitor the confirmation of the message
|
|
createAndStartTimer( mRef);
|
|
|
|
// store the message for retransmission and local delivery
|
|
storeMessage( msg);
|
|
|
|
return mRef;
|
|
}
|
|
|
|
/**
|
|
* @brief Create and start an unicast message retransmission timer.
|
|
* @param mRef The message reference of the message to monitor.
|
|
*
|
|
* Set the timer's fields and start the timer. It is afterwards added
|
|
* to the retransmission timer queue and can be accessed via the given
|
|
* message reference.
|
|
*/
|
|
void
|
|
UnicastTransfer::createAndStartTimer( MessageReference& mRef) {
|
|
UnicastMessageRetransmitTimer* timer = new UnicastMessageRetransmitTimer( *this);
|
|
|
|
timer->setReference(mRef);
|
|
timer->start();
|
|
|
|
retransmitTimerQueue.add(timer);
|
|
}
|
|
|
|
/**
|
|
* @brief Stop and delete the retransmission timer for the message,
|
|
* identified by the given message reference.
|
|
* @param mRef The message reference to identify the message and the
|
|
* corresponding timer.
|
|
*/
|
|
void
|
|
UnicastTransfer::stopAndDeleteTimer( MessageReference const& mRef) {
|
|
UnicastMessageRetransmitTimer* timer = retransmitTimerQueue.find(mRef);
|
|
|
|
if( timer != NULL) {
|
|
timer->stop();
|
|
retransmitTimerQueue.remove(mRef);
|
|
delete timer;
|
|
}//End if
|
|
}
|
|
|
|
/**
|
|
* @brief Store the given message in the retransmitMessageQueue, if the message is not already stored.
|
|
* @param msg The message to store.
|
|
*/
|
|
void
|
|
UnicastTransfer::storeMessage(UnicastMessage & msg) {
|
|
|
|
// not already stored?
|
|
if( !retransmitMessageQueue.contains( msg.getMessageReference())) {
|
|
// store the message
|
|
UnicastMessage* copyMsg = msg.dup();
|
|
retransmitMessageQueue.add( copyMsg);
|
|
}//end if
|
|
}
|
|
|
|
/**
|
|
* @brief Remove the message from the retransmission message queue,
|
|
* identified by the message reference.
|
|
* @param mRef The message reference to identify the message.
|
|
*/
|
|
void
|
|
UnicastTransfer::deleteMessage( MessageReference const & mRef) {
|
|
if( retransmitMessageQueue.contains(mRef)) {
|
|
retransmitMessageQueue.removeByMessageReference(mRef);
|
|
}//End if
|
|
}
|
|
|
|
/**
|
|
* @brief Deliver a message to the higher layer.
|
|
* @param msg The message to deliver.
|
|
*/
|
|
void
|
|
UnicastTransfer::deliverMessage( UnicastMessage& msg) {
|
|
// deliver the message independent of message ordering
|
|
msg.handleDeliver( dispatcher);
|
|
}
|
|
|
|
/**
|
|
* @brief Creates a new message reference, used to identify a send or
|
|
* received message.
|
|
* @return The created message.
|
|
*/
|
|
MessageReference
|
|
UnicastTransfer::createMesssageReference() {
|
|
|
|
MessageReference mRef = currentMessageReference;
|
|
currentMessageReference++;
|
|
|
|
return mRef;
|
|
}
|
|
|
|
}
|
|
}
|