Add Read Thread to GDBRemoteCommunication

EwanC · EwanC · commit fab40d3911d3 · 2015-06-16T15:50:18.000Z
In order to support asynchronous notifications for non-stop mode this patch adds a packet read thread. This is done by implementing AppendBytesToCache() from the communications class, which continually reads packets into a packet queue. To initialize this thread StartReadThread() must be called by the client, so since llgs and platform tools use the GBDRemoteCommunicatos code they must also call this function as well as ProcessGDBRemote. When the read thread detects an async notify packet it broadcasts this event, where the matching listener will be added in the next non-stop patch. Packets are now accessed by calling ReadPacket() which pops a packet from the queue, instead of using WaitForPacketWithTimeoutMicroSecondsNoLock() Reviewers: vharron, clayborg Subscribers: lldb-commits, labath, ted, domipheus, deepak2427 Differential Revision: http://reviews.llvm.org/D10085 llvm-svn: 239824
diff --git a/lldb/source/Plugins/Process/gdb-remote/GDBRemoteCommunication.cpp b/lldb/source/Plugins/Process/gdb-remote/GDBRemoteCommunication.cpp
@@ -171,6 +171,12 @@ GDBRemoteCommunication::~GDBRemoteCommunication()
     {
         Disconnect();
     }
+
+    // Stop the communications read thread which is used to parse all
+    // incoming packets.  This function will block until the read
+    // thread returns.
+    if (m_read_thread_enabled)
+        StopReadThread();
 }
 
 char
@@ -295,7 +301,7 @@ GDBRemoteCommunication::PacketResult
 GDBRemoteCommunication::GetAck ()
 {
     StringExtractorGDBRemote packet;
-    PacketResult result = WaitForPacketWithTimeoutMicroSecondsNoLock (packet, GetPacketTimeoutInMicroSeconds (), false);
+    PacketResult result = ReadPacket (packet, GetPacketTimeoutInMicroSeconds (), false);
     if (result == PacketResult::Success)
     {
         if (packet.GetResponseType() == StringExtractorGDBRemote::ResponseType::eAck)
@@ -323,6 +329,62 @@ GDBRemoteCommunication::WaitForNotRunningPrivate (const TimeValue *timeout_ptr)
     return m_private_is_running.WaitForValueEqualTo (false, timeout_ptr, NULL);
 }
 
+GDBRemoteCommunication::PacketResult
+GDBRemoteCommunication::ReadPacket (StringExtractorGDBRemote &response, uint32_t timeout_usec, bool sync_on_timeout)
+{
+   if (m_read_thread_enabled)
+       return PopPacketFromQueue (response, timeout_usec);
+   else
+       return WaitForPacketWithTimeoutMicroSecondsNoLock (response, timeout_usec, sync_on_timeout);
+}
+
+
+// This function is called when a packet is requested.
+// A whole packet is popped from the packet queue and returned to the caller.
+// Packets are placed into this queue from the communication read thread.
+// See GDBRemoteCommunication::AppendBytesToCache.
+GDBRemoteCommunication::PacketResult
+GDBRemoteCommunication::PopPacketFromQueue (StringExtractorGDBRemote &response, uint32_t timeout_usec)
+{
+    // Calculate absolute timeout value
+    TimeValue timeout = TimeValue::Now();
+    timeout.OffsetWithMicroSeconds(timeout_usec);
+
+    do
+    {
+        // scope for the mutex
+        {
+            // lock down the packet queue
+            Mutex::Locker locker(m_packet_queue_mutex);
+
+            // Wait on condition variable.
+            if (m_packet_queue.size() == 0)
+                m_condition_queue_not_empty.Wait(m_packet_queue_mutex, &timeout);
+
+            if (m_packet_queue.size() > 0)
+            {
+                // get the front element of the queue
+                response = m_packet_queue.front();
+
+                // remove the front element
+                m_packet_queue.pop();
+
+                // we got a packet
+                return PacketResult::Success;
+            }
+         }
+
+         // Disconnected
+         if (!IsConnected())
+             return PacketResult::ErrorDisconnected;
+
+      // Loop while not timed out
+    } while (TimeValue::Now() < timeout);
+
+    return PacketResult::ErrorReplyTimeout;
+}
+
+
 GDBRemoteCommunication::PacketResult
 GDBRemoteCommunication::WaitForPacketWithTimeoutMicroSecondsNoLock (StringExtractorGDBRemote &packet, uint32_t timeout_usec, bool sync_on_timeout)
 {
@@ -1094,3 +1156,53 @@ GDBRemoteCommunication::ScopedTimeout::~ScopedTimeout ()
 {
     m_gdb_comm.SetPacketTimeout (m_saved_timeout);
 }
+
+// This function is called via the Communications class read thread when bytes become available
+// for this connection. This function will consume all incoming bytes and try to parse whole
+// packets as they become available. Full packets are placed in a queue, so that all packet
+// requests can simply pop from this queue. Async notification packets will be dispatched
+// immediately to the ProcessGDBRemote Async thread via an event.
+void GDBRemoteCommunication::AppendBytesToCache (const uint8_t * bytes, size_t len, bool broadcast, lldb::ConnectionStatus status)
+{
+    StringExtractorGDBRemote packet;
+
+    while (true)
+    {
+        PacketType type = CheckForPacket(bytes, len, packet);
+
+        // scrub the data so we do not pass it back to CheckForPacket
+        // on future passes of the loop
+        bytes = nullptr;
+        len = 0;
+
+        // we may have received no packet so lets bail out
+        if (type == PacketType::Invalid)
+            break;
+
+        if (type == PacketType::Standard)
+        {
+            // scope for the mutex
+            {
+                // lock down the packet queue
+                Mutex::Locker locker(m_packet_queue_mutex);
+                // push a new packet into the queue
+                m_packet_queue.push(packet);
+                // Signal condition variable that we have a packet
+                m_condition_queue_not_empty.Signal();
+
+            }
+        }
+
+        if (type == PacketType::Notify)
+        {
+            // put this packet into an event
+            const char *pdata = packet.GetStringRef().c_str();
+
+            // as the communication class, we are a broadcaster and the
+            // async thread is tuned to listen to us
+            BroadcastEvent(
+                eBroadcastBitGdbReadThreadGotNotify,
+                new EventDataBytes(pdata));
+        }
+    }
+}
diff --git a/lldb/source/Plugins/Process/gdb-remote/GDBRemoteCommunication.h b/lldb/source/Plugins/Process/gdb-remote/GDBRemoteCommunication.h
@@ -14,6 +14,7 @@
 // C++ Includes
 #include <list>
 #include <string>
+#include <queue>
 
 // Other libraries and framework includes
 // Project includes
@@ -47,7 +48,8 @@ class GDBRemoteCommunication : public Communication
 public:
     enum
     {
-        eBroadcastBitRunPacketSent = kLoUserBroadcastBit
+        eBroadcastBitRunPacketSent = kLoUserBroadcastBit,
+        eBroadcastBitGdbReadThreadGotNotify = kLoUserBroadcastBit << 1 // Sent when we received a notify packet.
     };
 
     enum class PacketType
@@ -279,6 +281,13 @@ class GDBRemoteCommunication : public Communication
     SendPacketNoLock (const char *payload, 
                       size_t payload_length);
 
+    PacketResult
+    ReadPacket (StringExtractorGDBRemote &response, uint32_t timeout_usec, bool sync_on_timeout);
+
+    // Pop a packet from the queue in a thread safe manner
+    PacketResult
+    PopPacketFromQueue (StringExtractorGDBRemote &response, uint32_t timeout_usec);
+
     PacketResult
     WaitForPacketWithTimeoutMicroSecondsNoLock (StringExtractorGDBRemote &response, 
                                                 uint32_t timeout_usec,
@@ -316,7 +325,24 @@ class GDBRemoteCommunication : public Communication
     static lldb::thread_result_t
     ListenThread (lldb::thread_arg_t arg);
 
+    // GDB-Remote read thread
+    //  . this thread constantly tries to read from the communication
+    //    class and stores all packets received in a queue.  The usual
+    //    threads read requests simply pop packets off the queue in the
+    //    usual order.
+    //    This setup allows us to intercept and handle async packets, such
+    //    as the notify packet.
+
+    // This method is defined as part of communication.h
+    // when the read thread gets any bytes it will pass them on to this function
+    virtual void AppendBytesToCache (const uint8_t * bytes, size_t len, bool broadcast, lldb::ConnectionStatus status);
+
 private:
+
+    std::queue<StringExtractorGDBRemote> m_packet_queue; // The packet queue
+    lldb_private::Mutex m_packet_queue_mutex;            // Mutex for accessing queue
+    Condition m_condition_queue_not_empty;               // Condition variable to wait for packets
+
     HostThread m_listen_thread;
     std::string m_listen_url;
 
diff --git a/lldb/source/Plugins/Process/gdb-remote/GDBRemoteCommunicationClient.cpp b/lldb/source/Plugins/Process/gdb-remote/GDBRemoteCommunicationClient.cpp
@@ -145,7 +145,7 @@ GDBRemoteCommunicationClient::HandshakeWithServer (Error *error_ptr)
         PacketResult packet_result = PacketResult::Success;
         const uint32_t timeout_usec = 10 * 1000; // Wait for 10 ms for a response
         while (packet_result == PacketResult::Success)
-            packet_result = WaitForPacketWithTimeoutMicroSecondsNoLock (response, timeout_usec, false);
+            packet_result = ReadPacket (response, timeout_usec, false);
 
         // The return value from QueryNoAckModeSupported() is true if the packet
         // was sent and _any_ response (including UNIMPLEMENTED) was received),
@@ -654,7 +654,7 @@ GDBRemoteCommunicationClient::SendPacketAndWaitForResponseNoLock (const char *pa
 {
     PacketResult packet_result = SendPacketNoLock (payload, payload_length);
     if (packet_result == PacketResult::Success)
-        packet_result = WaitForPacketWithTimeoutMicroSecondsNoLock (response, GetPacketTimeoutInMicroSeconds (), true);
+        packet_result = ReadPacket (response, GetPacketTimeoutInMicroSeconds (), true);
     return packet_result;
 }
 
@@ -670,6 +670,12 @@ GDBRemoteCommunicationClient::SendPacketAndWaitForResponse
     PacketResult packet_result = PacketResult::ErrorSendFailed;
     Mutex::Locker locker;
     Log *log (ProcessGDBRemoteLog::GetLogIfAllCategoriesSet (GDBR_LOG_PROCESS));
+
+    // In order to stop async notifications from being processed in the middle of the
+    // send/recieve sequence Hijack the broadcast. Then rebroadcast any events when we are done.
+    static Listener hijack_listener("lldb.NotifyHijacker");
+    HijackBroadcaster(&hijack_listener, eBroadcastBitGdbReadThreadGotNotify);    
+
     if (GetSequenceMutex (locker))
     {
         packet_result = SendPacketAndWaitForResponseNoLock (payload, payload_length, response);
@@ -761,6 +767,15 @@ GDBRemoteCommunicationClient::SendPacketAndWaitForResponse
                 log->Printf("error: failed to get packet sequence mutex, not sending packet '%*s'", (int) payload_length, payload);
         }
     }
+
+    // Remove our Hijacking listner from the broadcast.
+    RestoreBroadcaster();
+
+    // If a notification event occured, rebroadcast since it can now be processed safely.  
+    EventSP event_sp;
+    if (hijack_listener.GetNextEvent(event_sp))
+        BroadcastEvent(event_sp);
+
     return packet_result;
 }
 
@@ -902,9 +917,9 @@ GDBRemoteCommunicationClient::SendContinuePacketAndWaitForResponse
         got_async_packet = false;
 
         if (log)
-            log->Printf ("GDBRemoteCommunicationClient::%s () WaitForPacket(%s)", __FUNCTION__, continue_packet.c_str());
+            log->Printf ("GDBRemoteCommunicationClient::%s () ReadPacket(%s)", __FUNCTION__, continue_packet.c_str());
 
-        if (WaitForPacketWithTimeoutMicroSecondsNoLock(response, UINT32_MAX, false) == PacketResult::Success)
+        if (ReadPacket(response, UINT32_MAX, false) == PacketResult::Success)
         {
             if (response.Empty())
                 state = eStateInvalid;
@@ -961,7 +976,7 @@ GDBRemoteCommunicationClient::SendContinuePacketAndWaitForResponse
                                 // packet to make sure it doesn't get in the way
                                 StringExtractorGDBRemote extra_stop_reply_packet;
                                 uint32_t timeout_usec = 1000;
-                                if (WaitForPacketWithTimeoutMicroSecondsNoLock (extra_stop_reply_packet, timeout_usec, false) == PacketResult::Success)
+                                if (ReadPacket (extra_stop_reply_packet, timeout_usec, false) == PacketResult::Success)
                                 {
                                     switch (extra_stop_reply_packet.GetChar())
                                     {
@@ -1139,7 +1154,7 @@ GDBRemoteCommunicationClient::SendContinuePacketAndWaitForResponse
         else
         {
             if (log)
-                log->Printf ("GDBRemoteCommunicationClient::%s () WaitForPacket(...) => false", __FUNCTION__);
+                log->Printf ("GDBRemoteCommunicationClient::%s () ReadPacket(...) => false", __FUNCTION__);
             state = eStateInvalid;
         }
     }
diff --git a/lldb/source/Plugins/Process/gdb-remote/ProcessGDBRemote.cpp b/lldb/source/Plugins/Process/gdb-remote/ProcessGDBRemote.cpp
@@ -1151,6 +1151,12 @@ ProcessGDBRemote::ConnectToDebugserver (const char *connect_url)
         return error;
     }
 
+
+    // Start the communications read thread so all incoming data can be
+    // parsed into packets and queued as they arrive.
+    if (GetTarget().GetNonStopModeEnabled())
+        m_gdb_comm.StartReadThread();
+
     // We always seem to be able to open a connection to a local port
     // so we need to make sure we can then send data to it. If we can't
     // then we aren't actually connected to anything, so try and do the
