Changeset 16768

Timestamp:

06/08/13 12:21:51 (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:

• ConnectionUSB.cc (modified) (8 diffs)
• ConnectionUSB.h (modified) (2 diffs)

trunk/FACT++/src/ConnectionUSB.cc

@@ -103 +103 @@
 
     // Reset the connection status
+    fQueueSize = 0;
     fConnectionStatus = kDisconnected;
-
-    // Empty output queue
-    fOutQueue.clear();
 
 #ifdef DEBUG
@@ -198 +196 @@
         return;
 
-    Error("fOutTimeout has expired, writing data to "+URL()+" ["+to_string(fOutQueue.size())+"]");
+    Error("fOutTimeout has expired, writing data to "+URL());
 
     CloseImp(-1);
@@ -205 +203 @@
 void ConnectionUSB::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)
     {
@@ -220 +221 @@
         msg << n << " bytes could not be sent to " << URL() << " due to missing connection.";
         Warn(msg);
-    }
-    else
-    {
+        return;
+    }
+
+    if (--fQueueSize==0)
+        fOutTimeout.cancel();
+
 #ifdef DEBUG_TX
-        ostringstream msg;
-        msg << n << " bytes successfully sent to " << URL();
-        Message(msg);
+    ostringstream msg;
+    msg << n << " bytes successfully sent to " << URL();
+    Message(msg);
 #endif
-    }
 
 #ifdef DEBUG
@@ -239 +242 @@
 
     HandleTransmittedData(n);
-
-    // 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
-        fOutTimeout.cancel();
-        return;
-    }
-
-     // AsyncWrite + Deadline
-    AsyncWrite(ba::const_buffers_1(fOutQueue.front().data(), fOutQueue.front().size())/*, &ConnectionUSB::HandleSentData*/);
+}
+
+void ConnectionUSB::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, &ConnectionUSB::HandleWriteTimeout);
-}
-
-// It is important that when SendMessageImp is called, or to be more
-// precise boost::bind is called, teh data is copied!
-void ConnectionUSB::SendMessageImp(const vector<uint8_t> 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;
-
-#ifdef DEBUG
-    ofstream fout("send.txt", ios::app);
-    fout << Time() << ": ";
-    for (unsigned int i=0; i<msg.size(); i++)
-        fout << hex << setfill('0') << setw(2) << (uint32_t)msg[i];
-    fout << endl;
-#endif
-
-    // AsyncWrite + Deadline
-    AsyncWrite(ba::const_buffers_1(fOutQueue.front().data(), fOutQueue.front().size())/*, &ConnectionUSB::HandleSentData*/);
-    AsyncWait(fOutTimeout, 5000, &ConnectionUSB::HandleWriteTimeout);
-}
-
-void ConnectionUSB::PostMessage(const void *ptr, size_t max)
-{
-    const vector<uint8_t> msg(reinterpret_cast<const uint8_t*>(ptr),
-                              reinterpret_cast<const uint8_t*>(ptr)+max);
-
-    get_io_service().post(boost::bind(&ConnectionUSB::SendMessageImp, this, msg));
+
+    // Now we can schedule the buffer to be sent
+    AsyncWrite(ba::const_buffers_1(ptr, sz));
 }
 
@@ -307 +267 @@
         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));
 }
 
@@ -349 +305 @@
     }
 
+    fQueueSize = 0;
     fConnectionStatus = kConnected;
 
@@ -369 +326 @@
 fFlowControl(flow_control::hardware),
 fInTimeout(ioservice), fOutTimeout(ioservice), fConnectTimeout(ioservice),
-fConnectionStatus(kDisconnected)
-{
-}
+fQueueSize(0), fConnectionStatus(kDisconnected)
+{
+}

trunk/FACT++/src/ConnectionUSB.h

@@ -38 +38 @@
     boost::asio::deadline_timer   fOutTimeout;
     boost::asio::deadline_timer   fConnectTimeout;
-    std::list<std::vector<uint8_t>> fOutQueue;
+
+    size_t fQueueSize;
 
     ConnectionStatus_t fConnectionStatus;
@@ -81 +82 @@
 
     // ------------------------ write --------------------------
-    void SendMessageImp(const std::vector<uint8_t> msg);
     void PostMessage(const void *msg, size_t s=0);
     void PostMessage(const std::string &cmd, size_t s=-1);