[LLGS] Get rid of the stdio forwarding thread

[labath]

Summary

This commit removes the stdio forwarding thread in lldb-server in favor of a MainLoop callback.
As in some situations we need to forcibly flush the stream ( => Read() is called from multiple
places) and we still have multiple threads, I have had to additionally protect the communication
instance with a mutex.

[ovyalov]

Please see my comments.

source/Plugins/Process/gdb-remote/GDBRemoteCommunicationServerLLGS.cpp

940 ↗

(On Diff #30001)

m_stdio_communication.Disconnect() ?

967 ↗

(On Diff #30001)

If SendProcessOutput is called from ProcessStateChanged within Monitor thread context we may have a race here:

• simultaneous access to m_stdio_handle_up from main & monitor threads.
• MainLoop will be accessed from multiple threads because UnregisterReadObject is called from monitor.

[labath]

source/Plugins/Process/gdb-remote/GDBRemoteCommunicationServerLLGS.cpp

940 ↗

(On Diff #30001)

Added. Note that this will probably send a SIGHUP to the inferior. However, at this point, I don't think we care about that.

967 ↗

(On Diff #30001)

Nice catch. For the first problem, I propose to extend the scope of the m_stdio_communication_mutex to cover m_stdio_handle_up as well (most of the uses are covered by it already). For the second issue, I will prepare a separate change to make MainLoop thread-safe.

[labath]

• Address review comments.

[ovyalov]

source/Plugins/Process/gdb-remote/GDBRemoteCommunicationServerLLGS.cpp

970 ↗

(On Diff #30166)

If eventually lldb-server will be fully single-threaded then then it's ok to keep MainLoop as-is - since anyway monitor will be part of main thread.

1065 ↗

(On Diff #30166)

You may wrap it up into a separate function since there are 2 usages of such code.

[labath]

source/Plugins/Process/gdb-remote/GDBRemoteCommunicationServerLLGS.cpp

970 ↗

(On Diff #30166)

I have committed this with the (previously reverted) commit which removes NativeProcessLinux monitor thread. This makes lldb-server single threaded.

lldb/trunk/source/Plugins/Process/gdb-remote/GDBRemoteCommunicationServerLLGS.h

public:
     DidExec (NativeProcessProtocol *process) override;
 
     Error
     InitializeConnection (std::unique_ptr<Connection> &&connection);
 
 protected:
     lldb::PlatformSP m_platform_sp;
     MainLoop &m_mainloop;
-    MainLoop::ReadHandleUP m_read_handle_up;
+    MainLoop::ReadHandleUP m_network_handle_up;
     lldb::tid_t m_current_tid;
     lldb::tid_t m_continue_tid;
     Mutex m_debugged_process_mutex;
     NativeProcessProtocolSP m_debugged_process_sp;
+
+    Mutex m_stdio_communication_mutex; // Protects m_stdio_communication and m_stdio_handle_up
     Communication m_stdio_communication;
+    MainLoop::ReadHandleUP m_stdio_handle_up;
+
     lldb::StateType m_inferior_prev_state;
     lldb::DataBufferSP m_active_auxv_buffer_sp;
     Mutex m_saved_registers_mutex;
     std::unordered_map<uint32_t, lldb::DataBufferSP> m_saved_registers_map;
     uint32_t m_next_saved_registers_id;
     bool m_handshake_completed : 1;
 
     PacketResult
     SendONotification (const char *buffer, uint32_t len);
 
     PacketResult
     SendWResponse (NativeProcessProtocol *process);
 
     PacketResult
     SendStopReplyPacketForThread (lldb::tid_t tid);
 
     PacketResult
-    SendStopReasonForState (lldb::StateType process_state, bool flush_on_exit);
+    SendStopReasonForState (lldb::StateType process_state);
 
     PacketResult
     Handle_k (StringExtractorGDBRemote &packet);
 
     PacketResult
     Handle_qProcessInfo (StringExtractorGDBRemote &packet);
 
     PacketResult
protected:
     SetContinueThreadID (lldb::tid_t tid);
 
     lldb::tid_t
     GetContinueThreadID () const { return m_continue_tid; }
 
     Error
     SetSTDIOFileDescriptor (int fd);
 
-    static void
-    STDIOReadThreadBytesReceived (void *baton, const void *src, size_t src_len);
-
     FileSpec
     FindModuleFile (const std::string& module_path, const ArchSpec& arch) override;
 
 private:
     bool
     DebuggedProcessReaped (lldb::pid_t pid);
 
     static bool
     ReapDebuggedProcess (void *callback_baton,
                          lldb::pid_t pid,
                          bool exited,
                          int signal,
                          int status);
 
     void
     HandleInferiorState_Exited (NativeProcessProtocol *process);
 
     void
     HandleInferiorState_Stopped (NativeProcessProtocol *process);
 
-    void
-    FlushInferiorOutput ();
-
     NativeThreadProtocolSP
     GetThreadFromSuffix (StringExtractorGDBRemote &packet);
 
     uint32_t
     GetNextSavedRegistersID ();
 
     void
     MaybeCloseInferiorTerminalConnection ();
 
     void
     ClearProcessSpecificData ();
 
     void
     RegisterPacketHandlers ();
 
     void
     DataAvailableCallback ();
 
+    void
+    SendProcessOutput ();
+
+    void
+    StopSTDIOForwarding();
+
     //------------------------------------------------------------------
     // For GDBRemoteCommunicationServerLLGS only
     //------------------------------------------------------------------
     DISALLOW_COPY_AND_ASSIGN (GDBRemoteCommunicationServerLLGS);
 };
 
 } // namespace process_gdb_remote
 } // namespace lldb_private
 
 #endif  // liblldb_GDBRemoteCommunicationServerLLGS_h_

lldb/trunk/source/Plugins/Process/gdb-remote/GDBRemoteCommunicationServerLLGS.cpp

 GDBRemoteCommunicationServerLLGS::HandleInferiorState_Exited (NativeProcessProtocol *process)
 {
     assert (process && "process cannot be NULL");
 
     Log *log (GetLogIfAnyCategoriesSet(LIBLLDB_LOG_PROCESS));
     if (log)
         log->Printf ("GDBRemoteCommunicationServerLLGS::%s called", __FUNCTION__);
 
-    // Send the exit result, and don't flush output.
-    // Note: flushing output here would join the inferior stdio reflection thread, which
-    // would gunk up the waitpid monitor thread that is calling this.
-    PacketResult result = SendStopReasonForState (StateType::eStateExited, false);
+    PacketResult result = SendStopReasonForState(StateType::eStateExited);
     if (result != PacketResult::Success)
     {
         if (log)
             log->Printf ("GDBRemoteCommunicationServerLLGS::%s failed to send stop notification for PID %" PRIu64 ", state: eStateExited", __FUNCTION__, process->GetID ());
     }
 
     // Remove the process from the list of spawned pids.
     {
         Mutex::Locker locker (m_spawned_pids_mutex);
         if (m_spawned_pids.erase (process->GetID ()) < 1)
         {
             if (log)
                 log->Printf ("GDBRemoteCommunicationServerLLGS::%s failed to remove PID %" PRIu64 " from the spawned pids list", __FUNCTION__, process->GetID ());
 
         }
     }
 
-    // FIXME can't do this yet - since process state propagation is currently
-    // synchronous, it is running off the NativeProcessProtocol's innards and
-    // will tear down the NPP while it still has code to execute.
-#if 0
-    // Clear the NativeProcessProtocol pointer.
-    {
-        Mutex::Locker locker (m_debugged_process_mutex);
-        m_debugged_process_sp.reset();
-    }
-#endif
-
     // Close the pipe to the inferior terminal i/o if we launched it
-    // and set one up.  Otherwise, 'k' and its flush of stdio could
-    // end up waiting on a thread join that will never end.  Consider
-    // adding a timeout to the connection thread join call so we
-    // can avoid that scenario altogether.
+    // and set one up.
     MaybeCloseInferiorTerminalConnection ();
 
     // We are ready to exit the debug monitor.
     m_exit_now = true;
 }
 
 void
 GDBRemoteCommunicationServerLLGS::HandleInferiorState_Stopped (NativeProcessProtocol *process)
GDBRemoteCommunicationServerLLGS::HandleInferiorState_Stopped (NativeProcessProtocol *process)
     switch (m_inferior_prev_state)
     {
         case eStateLaunching:
         case eStateAttaching:
             // Don't send anything per debugserver behavior.
             break;
         default:
             // In all other cases, send the stop reason.
-            PacketResult result = SendStopReasonForState (StateType::eStateStopped, false);
+            PacketResult result = SendStopReasonForState(StateType::eStateStopped);
             if (result != PacketResult::Success)
             {
                 if (log)
                     log->Printf ("GDBRemoteCommunicationServerLLGS::%s failed to send stop notification for PID %" PRIu64 ", state: eStateExited", __FUNCTION__, process->GetID ());
             }
             break;
     }
 }
if (log)
                 __FUNCTION__,
                 process->GetID (),
                 StateAsCString (state));
     }
 
     // Make sure we get all of the pending stdout/stderr from the inferior
     // and send it to the lldb host before we send the state change
     // notification
-    m_stdio_communication.SynchronizeWithReadThread();
+    SendProcessOutput();
 
     switch (state)
     {
     case StateType::eStateExited:
         HandleInferiorState_Exited (process);
         break;
 
     case StateType::eStateStopped:
GDBRemoteCommunicationServerLLGS::DataAvailableCallback ()
 
     if (! m_handshake_completed)
     {
         if (! HandshakeWithClient())
         {
             if(log)
                 log->Printf("GDBRemoteCommunicationServerLLGS::%s handshake with client failed, exiting",
                         __FUNCTION__);
-            m_read_handle_up.reset();
             m_mainloop.RequestTermination();
             return;
         }
         m_handshake_completed = true;
     }
 
     bool interrupt = false;
     bool done = false;
     Error error;
     while (true)
     {
         const PacketResult result = GetPacketAndSendResponse (0, error, interrupt, done);
         if (result == PacketResult::ErrorReplyTimeout)
             break; // No more packets in the queue
 
         if ((result != PacketResult::Success))
         {
             if(log)
                 log->Printf("GDBRemoteCommunicationServerLLGS::%s processing a packet failed: %s",
                         __FUNCTION__, error.AsCString());
-            m_read_handle_up.reset();
             m_mainloop.RequestTermination();
             break;
         }
     }
 }
 
 Error
 GDBRemoteCommunicationServerLLGS::InitializeConnection (std::unique_ptr<Connection> &&connection)
 {
     IOObjectSP read_object_sp = connection->GetReadObject();
     GDBRemoteCommunicationServer::SetConnection(connection.release());
 
     Error error;
-    m_read_handle_up = m_mainloop.RegisterReadObject(read_object_sp,
+    m_network_handle_up = m_mainloop.RegisterReadObject(read_object_sp,
             [this] (MainLoopBase &) { DataAvailableCallback(); }, error);
     return error;
 }
 
 GDBRemoteCommunication::PacketResult
 GDBRemoteCommunicationServerLLGS::SendONotification (const char *buffer, uint32_t len)
 {
     if ((buffer == nullptr) || (len == 0))
GDBRemoteCommunicationServerLLGS::SendONotification (const char *buffer, uint32_t len)
     return SendPacketNoLock (response.GetData (), response.GetSize ());
 }
 
 Error
 GDBRemoteCommunicationServerLLGS::SetSTDIOFileDescriptor (int fd)
 {
     Error error;
 
-    // Set up the Read Thread for reading/handling process I/O
+    // Set up the reading/handling of process I/O
     std::unique_ptr<ConnectionFileDescriptor> conn_up (new ConnectionFileDescriptor (fd, true));
     if (!conn_up)
     {
         error.SetErrorString ("failed to create ConnectionFileDescriptor");
         return error;
     }
 
+    Mutex::Locker locker(m_stdio_communication_mutex);
     m_stdio_communication.SetCloseOnEOF (false);
     m_stdio_communication.SetConnection (conn_up.release());
     if (!m_stdio_communication.IsConnected ())
     {
         error.SetErrorString ("failed to set connection for inferior I/O communication");
         return error;
     }
 
     // llgs local-process debugging may specify PTY paths, which will make these
     // file actions non-null
     // process launch -e/o will also make these file actions non-null
     // nullptr means that the traffic is expected to flow over gdb-remote protocol
     if (
         m_process_launch_info.GetFileActionForFD(STDOUT_FILENO) == nullptr ||
         m_process_launch_info.GetFileActionForFD(STDERR_FILENO) == nullptr
         )
     {
         // output from the process must be forwarded over gdb-remote
-        // create a thread to read the handle and send the data
-        m_stdio_communication.SetReadThreadBytesReceivedCallback (STDIOReadThreadBytesReceived, this);
-        m_stdio_communication.StartReadThread();
+        m_stdio_handle_up = m_mainloop.RegisterReadObject(
+                m_stdio_communication.GetConnection()->GetReadObject(),
+                [this] (MainLoopBase &) { SendProcessOutput(); }, error);
+        if (! m_stdio_handle_up)
+        {
+            m_stdio_communication.Disconnect();
+            return error;
+        }
     }
 
-    return error;
+    return Error();
 }
 
 void
-GDBRemoteCommunicationServerLLGS::STDIOReadThreadBytesReceived (void *baton, const void *src, size_t src_len)
+GDBRemoteCommunicationServerLLGS::StopSTDIOForwarding()
 {
-    GDBRemoteCommunicationServerLLGS *server = reinterpret_cast<GDBRemoteCommunicationServerLLGS*> (baton);
-    static_cast<void> (server->SendONotification (static_cast<const char *>(src), src_len));
+    Mutex::Locker locker(m_stdio_communication_mutex);
+    m_stdio_handle_up.reset();
+}
+
+void
+GDBRemoteCommunicationServerLLGS::SendProcessOutput()
+{
+    char buffer[1024];
+    ConnectionStatus status;
+    Error error;
+    Mutex::Locker locker(m_stdio_communication_mutex);
+    while (true)
+    {
+        size_t bytes_read = m_stdio_communication.Read(buffer, sizeof buffer, 0, status, &error);
+        switch (status)
+        {
+        case eConnectionStatusSuccess:
+            SendONotification(buffer, bytes_read);
+            break;
+        case eConnectionStatusLostConnection:
+        case eConnectionStatusEndOfFile:
+        case eConnectionStatusError:
+        case eConnectionStatusNoConnection:
+            if (Log *log = GetLogIfAnyCategoriesSet(LIBLLDB_LOG_PROCESS))
+                log->Printf("GDBRemoteCommunicationServerLLGS::%s Stopping stdio forwarding as communication returned status %d (error: %s)", __FUNCTION__, status, error.AsCString());
+            m_stdio_handle_up.reset();
+            return;
+
+        case eConnectionStatusInterrupted:
+        case eConnectionStatusTimedOut:
+            return;
+        }
+    }
 }
 
 GDBRemoteCommunication::PacketResult
 GDBRemoteCommunicationServerLLGS::Handle_qProcessInfo (StringExtractorGDBRemote &packet)
 {
     // Fail if we don't have a current process.
     if (!m_debugged_process_sp || (m_debugged_process_sp->GetID () == LLDB_INVALID_PROCESS_ID))
         return SendErrorResponse (68);
GDBRemoteCommunicationServerLLGS::Handle_k (StringExtractorGDBRemote &packet)
     {
         lldb::pid_t spawned_pid = *it;
         if (!KillSpawnedProcess (spawned_pid))
         {
             fprintf (stderr, "%s: failed to kill spawned pid %" PRIu64 ", ignoring.\n", __FUNCTION__, spawned_pid);
         }
     }
 
-    FlushInferiorOutput ();
+    StopSTDIOForwarding();
 
     // No OK response for kill packet.
     // return SendOKResponse ();
     return PacketResult::Success;
 }
 
 GDBRemoteCommunication::PacketResult
 GDBRemoteCommunicationServerLLGS::Handle_QSetDisableASLR (StringExtractorGDBRemote &packet)

 GDBRemoteCommunicationServerLLGS::Handle_stop_reason (StringExtractorGDBRemote &packet)
 {
     // Handle the $? gdbremote command.
 
     // If no process, indicate error
     if (!m_debugged_process_sp)
         return SendErrorResponse (02);
 
-    return SendStopReasonForState (m_debugged_process_sp->GetState (), true);
+    return SendStopReasonForState (m_debugged_process_sp->GetState());
 }
 
 GDBRemoteCommunication::PacketResult
-GDBRemoteCommunicationServerLLGS::SendStopReasonForState (lldb::StateType process_state, bool flush_on_exit)
+GDBRemoteCommunicationServerLLGS::SendStopReasonForState (lldb::StateType process_state)
 {
     Log *log (GetLogIfAnyCategoriesSet(LIBLLDB_LOG_PROCESS));
 
     switch (process_state)
     {
         case eStateAttaching:
         case eStateLaunching:
         case eStateRunning:
switch (process_state)
             // the data the gdb will expect.
             SetCurrentThreadID (tid);
             return SendStopReplyPacketForThread (tid);
         }
 
         case eStateInvalid:
         case eStateUnloaded:
         case eStateExited:
-            if (flush_on_exit)
-                FlushInferiorOutput ();
             return SendWResponse(m_debugged_process_sp.get());
 
         default:
             if (log)
             {
                 log->Printf ("GDBRemoteCommunicationServerLLGS::%s pid %" PRIu64 ", current state reporting not handled: %s",
                              __FUNCTION__,
                              m_debugged_process_sp->GetID (),
for (;;)
         if (read == 0)
         {
             break;
         }
         // write directly to stdin *this might block if stdin buffer is full*
         // TODO: enqueue this block in circular buffer and send window size to remote host
         ConnectionStatus status;
         Error error;
+        Mutex::Locker locker(m_stdio_communication_mutex);
         m_stdio_communication.Write(tmp, read, status, &error);
         if (error.Fail())
         {
             return SendErrorResponse (0x15);
         }
     }
 
     return SendOKResponse();
GDBRemoteCommunicationServerLLGS::Handle_vAttach (StringExtractorGDBRemote &packet)
     if (error.Fail ())
     {
         if (log)
             log->Printf ("GDBRemoteCommunicationServerLLGS::%s failed to attach to pid %" PRIu64 ": %s\n", __FUNCTION__, pid, error.AsCString());
         return SendErrorResponse (0x01);
     }
 
     // Notify we attached by sending a stop packet.
-    return SendStopReasonForState (m_debugged_process_sp->GetState (), true);
+    return SendStopReasonForState (m_debugged_process_sp->GetState ());
 }
 
 GDBRemoteCommunication::PacketResult
 GDBRemoteCommunicationServerLLGS::Handle_D (StringExtractorGDBRemote &packet)
 {
     Log *log (GetLogIfAnyCategoriesSet (LIBLLDB_LOG_PROCESS));
 
+    StopSTDIOForwarding();
+
     // Scope for mutex locker.
     Mutex::Locker locker (m_spawned_pids_mutex);
 
     // Fail if we don't have a current process.
     if (!m_debugged_process_sp || (m_debugged_process_sp->GetID () == LLDB_INVALID_PROCESS_ID))
     {
         if (log)
             log->Printf ("GDBRemoteCommunicationServerLLGS::%s failed, no process available", __FUNCTION__);
GDBRemoteCommunicationServerLLGS::Handle_D (StringExtractorGDBRemote &packet)
     }
 
     if (pid != LLDB_INVALID_PROCESS_ID &&
         m_debugged_process_sp->GetID () != pid)
     {
         return SendIllFormedResponse (packet, "Invalid pid");
     }
 
-    if (m_stdio_communication.IsConnected ())
-    {
-        m_stdio_communication.StopReadThread ();
-    }
-
     const Error error = m_debugged_process_sp->Detach ();
     if (error.Fail ())
     {
         if (log)
             log->Printf ("GDBRemoteCommunicationServerLLGS::%s failed to detach from pid %" PRIu64 ": %s\n",
                          __FUNCTION__, m_debugged_process_sp->GetID (), error.AsCString ());
         return SendErrorResponse (0x01);
     }
if (file_load_address == LLDB_INVALID_ADDRESS)
         return SendErrorResponse(1); // File not loaded
 
     StreamGDBRemote response;
     response.PutHex64(file_load_address);
     return SendPacketNoLock(response.GetData(), response.GetSize());
 }
 
 void
-GDBRemoteCommunicationServerLLGS::FlushInferiorOutput ()
-{
-    // If we're not monitoring an inferior's terminal, ignore this.
-    if (!m_stdio_communication.IsConnected())
-        return;
-
-    Log *log (GetLogIfAnyCategoriesSet(LIBLLDB_LOG_PROCESS | LIBLLDB_LOG_THREAD));
-    if (log)
-        log->Printf ("GDBRemoteCommunicationServerLLGS::%s() called", __FUNCTION__);
-
-    // FIXME implement a timeout on the join.
-    m_stdio_communication.JoinReadThread();
-}
-
-void
 GDBRemoteCommunicationServerLLGS::MaybeCloseInferiorTerminalConnection ()
 {
     Log *log (GetLogIfAnyCategoriesSet(LIBLLDB_LOG_PROCESS));
 
+    Mutex::Locker locker(m_stdio_communication_mutex);
     // Tell the stdio connection to shut down.
     if (m_stdio_communication.IsConnected())
     {
         auto connection = m_stdio_communication.GetConnection();
         if (connection)
         {
             Error error;
             connection->Disconnect (&error);