Changeset 16769 for trunk

Timestamp:

06/08/13 12:22:15 (11 years ago)

Author:

tbretz

Message:

There is no need to make the scheduling of the async write asynchronously, so I schedule the async write now directly; this also removed the send queue, which was primarily introduced to make the tranmission to the FTM serial, but this is not needed anymore; to keep track of the send queue size and for debugging, a counter for the messages in the send buffer has been introduced

Location:

trunk/FACT++/src

Files:

• Connection.cc (modified) (7 diffs)
• Connection.h (modified) (3 diffs)

trunk/FACT++/src/Connection.cc

@@ -103 +103 @@
 
     // Reset the connection status
+    fQueueSize = 0;
     fConnectionStatus = kDisconnected;
 
@@ -108 +109 @@
     fInTimeout.cancel();
     fOutTimeout.cancel();
-
-    // Empty output queue
-    fOutQueue.clear();
 
     if (!restart || IsConnecting())
@@ -168 +166 @@
 void Connection::HandleSentData(const bs::error_code& error, size_t n)
 {
+    if (error==ba::error::basic_errors::operation_aborted)
+        return;
+
     if (error && error != ba::error::not_connected)
     {
@@ -183 +184 @@
         msg << n << " bytes could not be sent to " << URL() << " due to missing connection.";
         Warn(msg);
-    }
-    else
-    {
-        if (fDebugTx)
-        {
-            ostringstream msg;
-            msg << n << " bytes successfully sent to " << URL();
-            Debug(msg);
-        }
-    }
-
-    // This is "thread" safe because SendMessage and HandleSentMessage
-    // are serialized in the EventQueue. Note: Do not call these
-    // functions directly from any other place then Handlers, use
-    // PostMessage instead
-    if (!fOutQueue.empty())
-        fOutQueue.pop_front();
-
-    if (fOutQueue.empty())
-    {
-        // Queue went empty, remove deadline
+
+        return;
+    }
+
+    if (--fQueueSize==0)
         fOutTimeout.cancel();
-        return;
-    }
-
-    // AsyncWrite + Deadline
-    AsyncWrite(ba::const_buffers_1(fOutQueue.front().data(), fOutQueue.front().size())/*, &Connection::HandleSentData*/);
+
+    if (fDebugTx)
+    {
+        ostringstream msg;
+        msg << n << " bytes successfully sent to " << URL();
+        Debug(msg);
+    }
+}
+
+void Connection::PostMessage(const void *ptr, size_t sz)
+{
+    // This function can be called from a different thread...
+    if (!is_open())
+        return;
+
+    // ... this is why we have to increase fQueueSize first
+    fQueueSize++;
+
+    // ... and shift the deadline timer
+    // This is not ideal, because if we are continously
+    // filling the buffer, it will never timeout
     AsyncWait(fOutTimeout, 5000, &Connection::HandleWriteTimeout);
-}
-
-// It is important that when SendMessageImp is called, or to be more
-// precise boost::bind is called, teh data is copied!
-void Connection::SendMessageImp(const vector<char> msg)
-{
-    /*
-    if (!fConnectionEstablished)
-    {
-        UpdateWarn("SendMessageImp, but no connection to "+fAddress+":"+fPort+".");
-        return;
-    }*/
-
-    const bool first_message_in_queue = fOutQueue.empty();
-
-    // This is "thread" safe because SendMessage and HandleSentMessage
-    // are serialized in the EventQueue. Note: Do not call these
-    // functions directly from any other place then Handlers, use
-    // PostMessage instead
-    fOutQueue.push_back(msg);
-
-    if (!first_message_in_queue)
-        return;
-
-    // AsyncWrite + Deadline
-    AsyncWrite(ba::const_buffers_1(fOutQueue.front().data(), fOutQueue.front().size())/*, &Connection::HandleSentData*/);
-    AsyncWait(fOutTimeout, 5000, &Connection::HandleWriteTimeout);
-}
-
-void Connection::PostMessage(const void *ptr, size_t max)
-{
-    const vector<char> msg(reinterpret_cast<const char*>(ptr),
-                           reinterpret_cast<const char*>(ptr)+max);
-
-    get_io_service().post(boost::bind(&Connection::SendMessageImp, this, msg));
+
+    // Now we can schedule the buffer to be sent
+    AsyncWrite(ba::const_buffers_1(ptr, sz));
+
+    // If a socket is closed, all pending asynchronous
+    // operation will be aborted.
 }
 
@@ -253 +225 @@
         max = cmd.length()+1;
 
-    vector <char>msg(max);
-
-    copy(cmd.begin(), cmd.begin()+min(cmd.length()+1, max), msg.begin());
-
-    PostMessage(msg);
+    PostMessage(cmd.c_str(), min(cmd.length()+1, max));
 }
 
@@ -306 +274 @@
             Info("Connection established to "+host+"...");
 
+        fQueueSize = 0;
         fConnectionStatus = kConnected;
 
@@ -487 +456 @@
 fLog(0), fVerbose(true), fDebugTx(false),
 fInTimeout(ioservice), fOutTimeout(ioservice), fConnectionTimer(ioservice),
-fConnectionStatus(kDisconnected)
-{
-}
+fQueueSize(0), fConnectionStatus(kDisconnected)
+{
+}

trunk/FACT++/src/Connection.h

@@ -39 +39 @@
     boost::asio::deadline_timer   fOutTimeout;
     boost::asio::deadline_timer   fConnectionTimer;
-    std::list<std::vector<char>> fOutQueue;
+
+    size_t fQueueSize;
 
     ConnectionStatus_t fConnectionStatus;
@@ -116 +117 @@
 
     // ------------------------ write --------------------------
-    void SendMessageImp(const std::vector<char> msg);
     void PostMessage(const void *msg, size_t s=0);
     void PostMessage(const std::string &cmd, size_t s=-1);
@@ -137 +137 @@
     virtual void HandleReadTimeout(const boost::system::error_code&) { }
 
-    bool IsTxQueueEmpty() const { return fOutQueue.empty(); }
+    bool IsTxQueueEmpty() const { return fQueueSize==0; /*fOutQueue.empty();*/ }
 
     int IsClosed() const { return !is_open(); }