ProtocolTCP.cpp

Go to the documentation of this file.

/*
---------------------------------------------------------------------------

    This file is part of uHAL.

    uHAL is a hardware access library and programming framework
    originally developed for upgrades of the Level-1 trigger of the CMS
    experiment at CERN.

    uHAL is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    uHAL is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with uHAL.  If not, see <http://www.gnu.org/licenses/>.


      Andrew Rose, Imperial College, London
      email: awr01 <AT> imperial.ac.uk

      Marc Magrans de Abril, CERN
      email: marc.magrans.de.abril <AT> cern.ch

---------------------------------------------------------------------------
*/

#include "uhal/ProtocolTCP.hpp"


#include <sys/time.h>

#include <boost/bind/bind.hpp>
#include <boost/lambda/lambda.hpp>
#include <boost/asio/connect.hpp>
#include <boost/asio/write.hpp>
#include <boost/asio/read.hpp>
#include <boost/asio/placeholders.hpp>
#include "boost/date_time/posix_time/posix_time.hpp"
#include <boost/chrono/chrono_io.hpp>

#include "uhal/Buffers.hpp"
#include "uhal/grammars/URI.hpp"
#include "uhal/IPbusInspector.hpp"
#include "uhal/log/LogLevels.hpp"
#include "uhal/log/log.hpp"
#include "uhal/log/log_inserters.integer.hpp"
#include "uhal/log/log_inserters.location.hpp"
#include "uhal/log/log_inserters.quote.hpp"
#include "uhal/ProtocolIPbus.hpp"
#include "uhal/ProtocolControlHub.hpp"


namespace uhal
{

  //--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
  //--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

  template < typename InnerProtocol , std::size_t nr_buffers_per_send >
  TCP< InnerProtocol, nr_buffers_per_send >::TCP ( const std::string& aId, const URI& aUri ) :
    InnerProtocol ( aId , aUri ),
    mIOservice ( ),
    mSocket ( mIOservice ),
    mEndpoint ( boost::asio::ip::tcp::resolver ( mIOservice ).resolve ( boost::asio::ip::tcp::resolver::query ( aUri.mHostname , aUri.mPort ) ) ),
    mDeadlineTimer ( mIOservice ),
    //     mReplyMemory ( 65536 , 0x00000000 ),
    mIOserviceWork ( mIOservice ),
    mDispatchThread ( boost::bind ( &boost::asio::io_service::run , & ( mIOservice ) ) ),
    mDispatchQueue(),
    mReplyQueue(),
    mPacketsInFlight ( 0 ),
    mFlushStarted ( false ),
    mFlushDone ( true ),
    mAsynchronousException ( NULL )
  {
    mDeadlineTimer.async_wait ( boost::bind ( &TCP::CheckDeadline, this ) );
  }


  template < typename InnerProtocol , std::size_t nr_buffers_per_send >
  TCP< InnerProtocol , nr_buffers_per_send >::~TCP()
  {
    try
    {
      mSocket.close();

      while ( mSocket.is_open() )
        {}

      mIOservice.stop();
      mDispatchThread.join();
      ClientInterface::returnBufferToPool ( mDispatchQueue );
      for (size_t i = 0; i < mReplyQueue.size(); i++)
        ClientInterface::returnBufferToPool ( mReplyQueue.at(i).first );
      ClientInterface::returnBufferToPool ( mDispatchBuffers );
      ClientInterface::returnBufferToPool ( mReplyBuffers.first );
    }
    catch ( const std::exception& aExc )
    {
      log ( Error() , "Exception " , Quote ( aExc.what() ) , " caught at " , ThisLocation() );
    }
  }


  template < typename InnerProtocol , std::size_t nr_buffers_per_send >
  void TCP< InnerProtocol , nr_buffers_per_send >::implementDispatch ( boost::shared_ptr< Buffers > aBuffers )
  {
    boost::lock_guard<boost::mutex> lLock ( mTransportLayerMutex );

    if ( mAsynchronousException )
    {
      log ( *mAsynchronousException , "Rethrowing Asynchronous Exception from " , ThisLocation() );
      mAsynchronousException->ThrowAsDerivedType();
    }

    if ( ! mSocket.is_open() )
    {
      connect();
    }

    mFlushStarted = false;
    mDispatchQueue.push_back ( aBuffers );

    if ( mDispatchBuffers.empty() && ( mDispatchQueue.size() >= nr_buffers_per_send ) && ( mPacketsInFlight < this->getMaxNumberOfBuffers() ) )
    {
      write ( );
    }
  }


  template < typename InnerProtocol , std::size_t nr_buffers_per_send >
  uint32_t TCP< InnerProtocol , nr_buffers_per_send >::getMaxSendSize()
  {
    return (350 * 4);
  }

 
  template < typename InnerProtocol , std::size_t nr_buffers_per_send >
  uint32_t TCP< InnerProtocol , nr_buffers_per_send >::getMaxReplySize()
  {
    return (350 * 4);
  }


  template < typename InnerProtocol , std::size_t nr_buffers_per_send >
  void TCP< InnerProtocol , nr_buffers_per_send >::connect()
  {
    log ( Info() , "Attempting to create TCP connection to '" , mEndpoint->host_name() , "' port " , mEndpoint->service_name() , "." );
    mDeadlineTimer.expires_from_now ( this->getBoostTimeoutPeriod() );
    boost::system::error_code lErrorCode;
    boost::asio::connect ( mSocket , mEndpoint , lErrorCode );

    if ( lErrorCode )
    {
      mSocket.close();

      while ( mSocket.is_open() )
        {}

      exception::TcpConnectionFailure lExc;
      std::ostringstream oss;

      if ( mDeadlineTimer.expires_at () == boost::posix_time::pos_infin )
      {
        oss << "Timeout (" << this->getBoostTimeoutPeriod().total_milliseconds() << " milliseconds) occurred when connecting to ";
      }
      else if ( lErrorCode == boost::asio::error::connection_refused )
      {
        oss << "Connection refused for ";
      }
      else
      {
        oss << "Error \"" << lErrorCode.message() << "\" encountered when connecting to ";
      }

      log ( lExc , oss.str() , ( this->uri().find ( "chtcp-" ) == 0 ? "ControlHub" : "TCP server" ) ,
            " at ", Quote ( this->mUri.mHostname + ":" + this->mUri.mPort ) , ". URI=" , Quote ( this->uri() ) );
      throw lExc;
    }

    mSocket.set_option ( boost::asio::ip::tcp::no_delay ( true ) );
    //    boost::asio::socket_base::non_blocking_io lNonBlocking ( true );
    //    mSocket.io_control ( lNonBlocking );
    log ( Info() , "TCP connection succeeded" );
  }



  template < typename InnerProtocol , std::size_t nr_buffers_per_send >
  void TCP< InnerProtocol , nr_buffers_per_send >::write ( )
  {
    NotifyConditionalVariable ( false );

    std::vector< boost::asio::const_buffer > lAsioSendBuffer;
    lAsioSendBuffer.push_back ( boost::asio::const_buffer ( &mSendByteCounter , 4 ) );
    mSendByteCounter = 0;
    std::size_t lNrBuffersToSend = std::min ( mDispatchQueue.size(), nr_buffers_per_send );
    mDispatchBuffers.reserve ( lNrBuffersToSend );

    for ( std::size_t i = 0; i < lNrBuffersToSend; i++ )
    {
      mDispatchBuffers.push_back ( mDispatchQueue.front() );
      mDispatchQueue.pop_front();
      const boost::shared_ptr<Buffers>& lBuffer = mDispatchBuffers.back();
      mSendByteCounter += lBuffer->sendCounter();
      lAsioSendBuffer.push_back ( boost::asio::const_buffer ( lBuffer->getSendBuffer() , lBuffer->sendCounter() ) );
    }

    log ( Debug() , "Sending " , Integer ( mSendByteCounter ) , " bytes from ", Integer ( mDispatchBuffers.size() ), " buffers" );
    mSendByteCounter = htonl ( mSendByteCounter );

    mDeadlineTimer.expires_from_now ( this->getBoostTimeoutPeriod() );

    // Patch for suspected bug in using boost asio with boost python; see https://svnweb.cern.ch/trac/cactus/ticket/323#comment:7
    while ( mDeadlineTimer.expires_from_now() < boost::posix_time::microseconds ( 600 ) )
    {
      log ( Debug() , "Resetting deadline timer since it just got set to strange value, likely due to a bug within boost (expires_from_now was: ", mDeadlineTimer.expires_from_now() , ")." );
      mDeadlineTimer.expires_from_now ( this->getBoostTimeoutPeriod() );
    }

    boost::asio::async_write ( mSocket , lAsioSendBuffer , boost::bind ( &TCP< InnerProtocol , nr_buffers_per_send >::write_callback, this, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred ) );
    mPacketsInFlight += mDispatchBuffers.size();

    SteadyClock_t::time_point lNow = SteadyClock_t::now();
    if (mLastSendQueued > SteadyClock_t::time_point())
      mInterSendTimeStats.add(mLastSendQueued, lNow);
    mLastSendQueued = lNow;
  }



  template < typename InnerProtocol , std::size_t nr_buffers_per_send >
  void TCP< InnerProtocol , nr_buffers_per_send >::write_callback ( const boost::system::error_code& aErrorCode , std::size_t aBytesTransferred )
  {
    boost::lock_guard<boost::mutex> lLock ( mTransportLayerMutex );
    mSendByteCounter = ntohl ( mSendByteCounter );

    if ( mAsynchronousException )
    {
      NotifyConditionalVariable ( true );
      return;
    }

    if ( mDeadlineTimer.expires_at () == boost::posix_time::pos_infin )
    {
      mAsynchronousException = new exception::TcpTimeout();
      log ( *mAsynchronousException , "Timeout (" , Integer ( this->getBoostTimeoutPeriod().total_milliseconds() ) , " milliseconds) occurred for send to ",
            ( this->uri().find ( "chtcp-" ) == 0 ? "ControlHub" : "TCP server" ) , " with URI: ", this->uri() );

      if ( aErrorCode && aErrorCode != boost::asio::error::operation_aborted )
      {
        log ( *mAsynchronousException , "ASIO reported an error: " , Quote ( aErrorCode.message() ) );
      }

      NotifyConditionalVariable ( true );
      return;
    }

    if ( ( aErrorCode && ( aErrorCode != boost::asio::error::eof ) ) || ( aBytesTransferred != ( mSendByteCounter+4 ) ) )
    {
      mAsynchronousException = new exception::ASIOTcpError();
      log ( *mAsynchronousException , "Error ", Quote ( aErrorCode.message() ) , " encountered during send to ",
            ( this->uri().find ( "chtcp-" ) == 0 ? "ControlHub" : "TCP server" ) , " with URI: " , this->uri() );

      try
      {
        mSocket.close();

        while ( mSocket.is_open() )
          {}
      }
      catch ( const std::exception& aExc )
      {
        log ( *mAsynchronousException , "Error closing TCP socket following the ASIO send error" );
      }

      if ( aBytesTransferred != ( mSendByteCounter + 4 ) )
      {
        log ( *mAsynchronousException , "Attempted to send " , Integer ( mSendByteCounter ) , " bytes to ",
             ( this->uri().find ( "chtcp-" ) == 0 ? "ControlHub" : "TCP server" ) ,
             " with URI "  , Quote ( this->uri() ) , ", but only sent " , Integer ( aBytesTransferred ) , " bytes" );
      }

      NotifyConditionalVariable ( true );
      return;
    }

    if ( ! mReplyBuffers.first.empty() )
    {
      mReplyQueue.push_back ( std::make_pair(mDispatchBuffers, mLastSendQueued) );
    }
    else
    {
      mReplyBuffers = std::make_pair(mDispatchBuffers, mLastSendQueued);
      read ( );
    }

    mDispatchBuffers.clear();

    if ( ( mDispatchQueue.size() >= (mFlushStarted ? 1 : nr_buffers_per_send) ) && ( mPacketsInFlight < this->getMaxNumberOfBuffers() ) )
    {
      write();
    }

  }


  template < typename InnerProtocol , std::size_t nr_buffers_per_send >
  void TCP< InnerProtocol , nr_buffers_per_send >::read ( )
  {
    //     std::deque< std::pair< uint8_t* , uint32_t > >& lReplyBuffers ( mReplyBuffers->getReplyBuffer() );
    //     std::vector< boost::asio::mutable_buffer > lAsioReplyBuffer;
    //     lAsioReplyBuffer.reserve ( lReplyBuffers.size() +1 );
    //     for ( std::deque< std::pair< uint8_t* , uint32_t > >::iterator lIt = lReplyBuffers.begin() ; lIt != lReplyBuffers.end() ; ++lIt )
    //     {
    //       lAsioReplyBuffer.push_back ( boost::asio::mutable_buffer ( lIt->first , lIt->second ) );
    //     }
    std::vector< boost::asio::mutable_buffer > lAsioReplyBuffer;
    lAsioReplyBuffer.push_back ( boost::asio::mutable_buffer ( &mReplyByteCounter , 4 ) );
    log ( Debug() , "Getting reply byte counter" );
    boost::asio::ip::tcp::endpoint lEndpoint;
    mDeadlineTimer.expires_from_now ( this->getBoostTimeoutPeriod() );

    // Patch for suspected bug in using boost asio with boost python; see https://svnweb.cern.ch/trac/cactus/ticket/323#comment:7
    while ( mDeadlineTimer.expires_from_now() < boost::posix_time::microseconds ( 600 ) )
    {
      log ( Debug() , "Resetting deadline timer since it just got set to strange value, likely due to a bug within boost (expires_from_now was: ", mDeadlineTimer.expires_from_now() , ")." );
      mDeadlineTimer.expires_from_now ( this->getBoostTimeoutPeriod() );
    }

    boost::asio::async_read ( mSocket , lAsioReplyBuffer ,  boost::asio::transfer_exactly ( 4 ), boost::bind ( &TCP< InnerProtocol , nr_buffers_per_send >::read_callback, this, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred ) );

    SteadyClock_t::time_point lNow = SteadyClock_t::now();
    if (mLastRecvQueued > SteadyClock_t::time_point())
      mInterRecvTimeStats.add(mLastRecvQueued, lNow);
    mLastRecvQueued = lNow;
  }





  template < typename InnerProtocol , std::size_t nr_buffers_per_send >
  void TCP< InnerProtocol , nr_buffers_per_send >::read_callback ( const boost::system::error_code& aErrorCode , std::size_t aBytesTransferred )
  {
    std::size_t lNrReplyBuffers = 1;
    uint32_t lRequestBytes = 0;
    uint32_t lExpectedReplyBytes = 0;

    for ( std::vector< boost::shared_ptr<Buffers> >::const_iterator lBufIt = mReplyBuffers.first.begin(); lBufIt != mReplyBuffers.first.end(); lBufIt++ )
    {
      lNrReplyBuffers += ( *lBufIt )->getReplyBuffer().size();
      lRequestBytes += ( *lBufIt )->sendCounter();
      lExpectedReplyBytes += ( *lBufIt )->replyCounter();
    }

    {
      boost::lock_guard<boost::mutex> lLock ( mTransportLayerMutex );
      if ( mAsynchronousException )
      {
        NotifyConditionalVariable ( true );
        return;
      }

      SteadyClock_t::time_point lNow = SteadyClock_t::now();

      if ( mDeadlineTimer.expires_at () == boost::posix_time::pos_infin )
      {
        exception::TcpTimeout* lExc = new exception::TcpTimeout();
        log ( *lExc , "Timeout (" , Integer ( this->getBoostTimeoutPeriod().total_milliseconds() ) , " ms) occurred for receive (header) from ",
              ( this->uri().find ( "chtcp-" ) == 0 ? "ControlHub" : "TCP server" ) , " with URI '", this->uri(), "'. ",
              Integer(mPacketsInFlight), " packets in flight, ", Integer(mReplyBuffers.first.size()), " in this chunk (",
              Integer(lRequestBytes), "/", Integer(lExpectedReplyBytes), " bytes sent/expected). Last send / receive queued ",
              boost::chrono::duration<float, boost::milli>(lNow - mLastSendQueued).count(), " / ",
              boost::chrono::duration<float, boost::milli>(lNow - mLastRecvQueued), " ago.");

        log ( Error(), "Extra timeout-related info - round-trip times: ", mRTTStats);
        log ( Error(), "Extra timeout-related info - send-recv  times: ", mLSTStats);
        log ( Error(), "Extra timeout-related info - inter-send times: ", mInterSendTimeStats);
        log ( Error(), "Extra timeout-related info - inter-recv times: ", mInterRecvTimeStats);

        if ( aErrorCode && aErrorCode != boost::asio::error::operation_aborted )
        {
          log ( *lExc , "ASIO reported an error: " , Quote ( aErrorCode.message() ) );
        }

        mAsynchronousException = lExc;
        NotifyConditionalVariable ( true );
        return;
      }
    }

    if ( ( aErrorCode && ( aErrorCode != boost::asio::error::eof ) ) || ( aBytesTransferred != 4 ) )
    {
      boost::lock_guard<boost::mutex> lLock ( mTransportLayerMutex );
      mAsynchronousException = new exception::ASIOTcpError();

      if ( aErrorCode )
      {
        log ( *mAsynchronousException , "Error ", Quote ( aErrorCode.message() ) , " encountered during receive from ",
              ( this->uri().find ( "chtcp-" ) == 0 ? "ControlHub" : "TCP server" ) , " with URI: " , this->uri() );
      }

      if ( aBytesTransferred != 4 )
      {
        log ( *mAsynchronousException, "Expected to receive 4-byte header in async read from ",
             ( this->uri().find ( "chtcp-" ) == 0 ? "ControlHub" : "TCP server" ) ,
             " with URI "  , Quote ( this->uri() ) , ", but only received " , Integer ( aBytesTransferred ) , " bytes" );
      }

      try
      {
        mSocket.close();
        while ( mSocket.is_open() )
          {}
      }
      catch ( const std::exception& aExc )
      {
        log ( *mAsynchronousException , "Error closing socket following ASIO read error" );
      }

      NotifyConditionalVariable ( true );
      return;
    }

    SteadyClock_t::time_point lReadHeaderTimestamp = SteadyClock_t::now();
    mRTTStats.add(mReplyBuffers.second, lReadHeaderTimestamp);
    mLSTStats.add(mLastSendQueued, lReadHeaderTimestamp);

    mReplyByteCounter = ntohl ( mReplyByteCounter );
    log ( Debug() , "Byte Counter says " , Integer ( mReplyByteCounter ) , " bytes are coming" );
    std::vector< boost::asio::mutable_buffer > lAsioReplyBuffer;

    lAsioReplyBuffer.reserve ( lNrReplyBuffers );
    log ( Debug() , "Expecting " , Integer ( lExpectedReplyBytes ) , " bytes in reply, for ", Integer ( mReplyBuffers.first.size() ), " buffers" );

    for ( std::vector< boost::shared_ptr<Buffers> >::const_iterator lBufIt = mReplyBuffers.first.begin(); lBufIt != mReplyBuffers.first.end(); lBufIt++ )
    {
      std::deque< std::pair< uint8_t* , uint32_t > >& lReplyBuffers ( ( *lBufIt )->getReplyBuffer() );

      for ( std::deque< std::pair< uint8_t* , uint32_t > >::iterator lIt = lReplyBuffers.begin() ; lIt != lReplyBuffers.end() ; ++lIt )
      {
        lAsioReplyBuffer.push_back ( boost::asio::mutable_buffer ( lIt->first , lIt->second ) );
      }
    }

    boost::system::error_code lErrorCode = boost::asio::error::would_block;
    std::size_t lBytesTransferred = boost::asio::read ( mSocket , lAsioReplyBuffer ,  boost::asio::transfer_exactly ( mReplyByteCounter ), lErrorCode );

    if ( ( lErrorCode && ( lErrorCode != boost::asio::error::eof ) ) || ( lBytesTransferred != mReplyByteCounter ) )
    {
      boost::lock_guard<boost::mutex> lLock ( mTransportLayerMutex );
      mSocket.close();

      if ( mDeadlineTimer.expires_at () == boost::posix_time::pos_infin )
      {
        mAsynchronousException = new exception::TcpTimeout();
        log ( *mAsynchronousException , "Timeout (" , Integer ( this->getBoostTimeoutPeriod().total_milliseconds() ) , " milliseconds) occurred for receive (chunk) from ",
              ( this->uri().find ( "chtcp-" ) == 0 ? "ControlHub" : "TCP server" ) , " with URI: ", this->uri() );
      }
      else
      {
        mAsynchronousException = new exception::ASIOTcpError();
        log ( *mAsynchronousException , "Error ", Quote ( aErrorCode.message() ) , " encountered during receive from ",
              ( this->uri().find ( "chtcp-" ) == 0 ? "ControlHub" : "TCP server" ) , " with URI: " , this->uri() );
      }

      if ( lBytesTransferred != mReplyByteCounter )
      {
        log ( *mAsynchronousException, "Expected to receive " , Integer ( mReplyByteCounter ) , " bytes in read from ",
             ( this->uri().find ( "chtcp-" ) == 0 ? "ControlHub" : "TCP server" ) ,
             " with URI "  , Quote ( this->uri() ) , ", but only received " , Integer ( lBytesTransferred ) , " bytes" );
      }

      NotifyConditionalVariable ( true );
      return;
    }


    for ( std::vector< boost::shared_ptr<Buffers> >::const_iterator lBufIt = mReplyBuffers.first.begin(); lBufIt != mReplyBuffers.first.end(); lBufIt++ )
    {
      try
      {
        if ( uhal::exception::exception* lExc =  ClientInterface::validate ( *lBufIt ) ) //Control of the pointer has been passed back to the client interface
        {
          boost::lock_guard<boost::mutex> lLock ( mTransportLayerMutex );
          mAsynchronousException = lExc;
        }
      }
      catch ( exception::exception& aExc )
      {
        boost::lock_guard<boost::mutex> lLock ( mTransportLayerMutex );
        mAsynchronousException = new exception::ValidationError ();
        log ( *mAsynchronousException , "Exception caught during reply validation for TCP device with URI " , Quote ( this->uri() ) , "; what returned: " , Quote ( aExc.what() ) );
      }

      if ( mAsynchronousException )
      {
        NotifyConditionalVariable ( true );
        return;
      }
    }

    boost::lock_guard<boost::mutex> lLock ( mTransportLayerMutex );
    mPacketsInFlight -= mReplyBuffers.first.size();

    if ( mReplyQueue.size() )
    {
      mReplyBuffers = mReplyQueue.front();
      mReplyQueue.pop_front();
      read();
    }
    else
    {
      mReplyBuffers.first.clear();
    }

    if ( mDispatchBuffers.empty() && ( mDispatchQueue.size() >= (mFlushStarted ? 1 : nr_buffers_per_send) ) && ( mPacketsInFlight < this->getMaxNumberOfBuffers() ) )
    {
      write();
    }

    if ( mDispatchBuffers.empty() && mReplyBuffers.first.empty() )
    {
      mDeadlineTimer.expires_from_now ( boost::posix_time::seconds(60) );
      NotifyConditionalVariable ( true );
    }
  }



  template < typename InnerProtocol , std::size_t nr_buffers_per_send >
  void TCP< InnerProtocol , nr_buffers_per_send >::CheckDeadline()
  {
    boost::lock_guard<boost::mutex> lLock ( this->mTransportLayerMutex );

    // Check whether the deadline has passed. We compare the deadline against the current time since a new asynchronous operation may have moved the deadline before this actor had a chance to run.
    if ( mDeadlineTimer.expires_at() <= boost::asio::deadline_timer::traits_type::now() )
    {
      // SETTING THE EXCEPTION HERE CAN APPEAR AS A TIMEOUT WHEN NONE ACTUALLY EXISTS
      if (  mDispatchBuffers.size() || mReplyBuffers.first.size() )
      {
        log ( Warning() , "Closing TCP socket for device with URI " , Quote ( this->uri() ) , " since deadline has passed" );
      }
      else
      {
        log ( Debug() , "Closing TCP socket for device with URI " , Quote ( this->uri() ) , " since no communication in 60 seconds" );
      }

      // The deadline has passed. The socket is closed so that any outstanding asynchronous operations are cancelled.
      mSocket.close();
      // There is no longer an active deadline. The expiry is set to positive infinity so that the actor takes no action until a new deadline is set.
      mDeadlineTimer.expires_at ( boost::posix_time::pos_infin );
    }

    // Put the actor back to sleep.
    mDeadlineTimer.async_wait ( boost::bind ( &TCP::CheckDeadline, this ) );
  }


  template < typename InnerProtocol , std::size_t nr_buffers_per_send >
  void TCP< InnerProtocol , nr_buffers_per_send >::Flush( )
  {
    {
      boost::lock_guard<boost::mutex> lLock ( mTransportLayerMutex );
      mFlushStarted = true;

      if ( mDispatchQueue.size() && mDispatchBuffers.empty() )
      {
        write();
      }
    }

    WaitOnConditionalVariable();

    boost::lock_guard<boost::mutex> lLock ( mTransportLayerMutex ); 

    if ( mAsynchronousException )
    {
      mAsynchronousException->ThrowAsDerivedType();
    }
  }



  template < typename InnerProtocol , std::size_t nr_buffers_per_send >
  void TCP< InnerProtocol , nr_buffers_per_send >::dispatchExceptionHandler()
  {
    if ( mSocket.is_open() )
    {
      log ( Warning() , "Closing TCP socket for device with URI " , Quote ( this->uri() ) , " since exception detected." );

      try
      {
        mSocket.close();

        while ( mSocket.is_open() )
          {}
      }
      catch ( ... )
        {}
    }

    NotifyConditionalVariable ( true );

    if ( mAsynchronousException )
    {
      delete mAsynchronousException;
      mAsynchronousException = NULL;
    }

    ClientInterface::returnBufferToPool ( mDispatchQueue );
      for (size_t i = 0; i < mReplyQueue.size(); i++)
        ClientInterface::returnBufferToPool ( mReplyQueue.at(i).first );
    mPacketsInFlight = 0;
    ClientInterface::returnBufferToPool ( mDispatchBuffers );
    ClientInterface::returnBufferToPool ( mReplyBuffers.first );

    mSendByteCounter = 0;
    mReplyByteCounter = 0;

    InnerProtocol::dispatchExceptionHandler();
  }


  //--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
  //--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------


  template < typename InnerProtocol , std::size_t nr_buffers_per_send >
  void TCP< InnerProtocol , nr_buffers_per_send >::NotifyConditionalVariable ( const bool& aValue )
  {
    {
      boost::lock_guard<boost::mutex> lLock ( mConditionalVariableMutex );
      mFlushDone = aValue;
    }
    mConditionalVariable.notify_one();
  }

  template < typename InnerProtocol , std::size_t nr_buffers_per_send >
  void TCP< InnerProtocol , nr_buffers_per_send >::WaitOnConditionalVariable()
  {
    boost::unique_lock<boost::mutex> lLock ( mConditionalVariableMutex );

    while ( !mFlushDone )
    {
      mConditionalVariable.wait ( lLock );
    }
  }


  template class TCP< IPbus< 1 , 3 > , 1 >;
  template class TCP< IPbus< 2 , 0 > , 1 >;

  template class TCP< ControlHub < IPbus< 1 , 3 > > , 3 >;
  template class TCP< ControlHub < IPbus< 2 , 0 > > , 3 >;
}