Fix ProcessIO test failures

[labath]

Summary

There was a race condition regarding the output of the inferior process. The reading of the
output is performed on a separate thread, and there was no guarantee that the output will get
eventually consumed. Because of that, it was happening that calling Process::GetSTDOUT was not
returning anything even though the process was terminated and would definitely not produce any
further output. This was usually happening only under very heavy system load, but it can be
reproduced by placing an usleep in the stdio thread (Process::STDIOReadThreadBytesReceived).

This patch addresses this by adding synchronization capabilities to the Communication thread.
After calling Communication::SynchronizeWithReadThread one can be sure that all pending input has
been processed by the read thread. This function is then called after every public event which
stops the process to obtain the entire process output.

[jingham]

Shouldn't there be implementations of InterruptRead in this patch?

It seems like you are saying InterruptRead will consume all available stdout, then return Interrupted, which is a little odd for an interrupt. But I don't know what InterruptRead actually does...

Jim

[labath]

InterruptRead is already implemented, here I have only added pure virtual method to the superclass, so they can be accessed easily. For implementations, refer to source/Host/posix/ConnectionFileDescriptorPosix.cpp and source/Host/windows/ConnectionGenericFileWindows.cpp.

It seems like you are saying InterruptRead will consume all available stdout, then return Interrupted, which is a little odd for an interrupt.

Yes that is basically what I am saying. Maybe "interrupt" is not exactly the best name for this, but it was already present, so I used it. In the general case, this is exactly what InterruptRead will do though - make the thread blocked on Read() return immediately with eConnectionStatusInterrupted. In this comment, I have just tried to explain what happens in the edge case, when there is input readily available *and* you call InterruptRead() - here, the priority is given to reading the input.

[clayborg]

Looks good.

lldb/trunk/include/lldb/Core/Communication.h

 {
 public:
     enum {
         eBroadcastBitDisconnected           = (1 << 0), ///< Sent when the communications connection is lost.
         eBroadcastBitReadThreadGotBytes     = (1 << 1), ///< Sent by the read thread when bytes become available.
         eBroadcastBitReadThreadDidExit      = (1 << 2), ///< Sent by the read thread when it exits to inform clients.
         eBroadcastBitReadThreadShouldExit   = (1 << 3), ///< Sent by clients that need to cancel the read thread.
         eBroadcastBitPacketAvailable        = (1 << 4), ///< Sent when data received makes a complete packet.
+        eBroadcastBitNoMorePendingInput     = (1 << 5), ///< Sent by the read thread to indicate all pending input has been processed.
         kLoUserBroadcastBit                 = (1 << 16),///< Subclasses can used bits 31:16 for any needed events.
         kHiUserBroadcastBit                 = (1 << 31),
         eAllEventBits                       = 0xffffffff
     };
 
     typedef void (*ReadThreadBytesReceived) (void *baton, const void *src, size_t src_len);
 
 
public:
     //------------------------------------------------------------------
     static lldb::thread_result_t
     ReadThread (lldb::thread_arg_t comm_ptr);
 
     void
     SetReadThreadBytesReceivedCallback (ReadThreadBytesReceived callback,
                                         void *callback_baton);
 
+
+    //------------------------------------------------------------------
+    /// Wait for the read thread to process all outstanding data.
+    ///
+    /// After this function returns, the read thread has processed all data that
+    /// has been waiting in the Connection queue.
+    ///
+    //------------------------------------------------------------------
+    void SynchronizeWithReadThread ();
+
     static const char *
     ConnectionStatusAsCString (lldb::ConnectionStatus status);
 
     bool
     GetCloseOnEOF () const
     { 
         return m_close_on_eof;
     }
private:
     //------------------------------------------------------------------
     DISALLOW_COPY_AND_ASSIGN (Communication);
 
 
 protected:
     lldb::ConnectionSP m_connection_sp; ///< The connection that is current in use by this communications class.
     HostThread m_read_thread;           ///< The read thread handle in case we need to cancel the thread.
     std::atomic<bool> m_read_thread_enabled;
+    std::atomic<bool> m_read_thread_did_exit;
     std::string m_bytes;    ///< A buffer to cache bytes read in the ReadThread function.
     Mutex m_bytes_mutex;    ///< A mutex to protect multi-threaded access to the cached bytes.
     Mutex m_write_mutex;    ///< Don't let multiple threads write at the same time...
+    Mutex m_synchronize_mutex;
     ReadThreadBytesReceived m_callback;
     void *m_callback_baton;
     bool m_close_on_eof;
 
     size_t
     ReadFromConnection (void *dst, 
                         size_t dst_len, 
                         uint32_t timeout_usec,

lldb/trunk/include/lldb/Core/Connection.h

public:
     /// @param[in] dst
     ///     A destination buffer that must be at least \a dst_len bytes
     ///     long.
     ///
     /// @param[in] dst_len
     ///     The number of bytes to attempt to read, and also the max
     ///     number of bytes that can be placed into \a dst.
     ///
+    /// @param[in] timeout_usec
+    ///     The number of microseconds to wait for the data.
+    ///
+    /// @param[out] status
+    ///     On return, indicates whether the call was sucessful or terminated
+    ///     due to some error condition.
+    ///
     /// @param[out] error_ptr
     ///     A pointer to an error object that should be given an
     ///     approriate error value if this method returns zero. This
     ///     value can be NULL if the error value should be ignored.
     ///
     /// @return
     ///     The number of bytes actually read.
     ///
public:
     /// Subclasses may override this function.
     ///
     /// @return
     ///     Returns URI or an empty string if disconnecteds
     //------------------------------------------------------------------
     virtual std::string
     GetURI() = 0;
 
+    //------------------------------------------------------------------
+    /// Interrupts an ongoing Read() operation.
+    ///
+    /// If there is an ongoing read operation in another thread, this operation
+    /// return with status == eConnectionStatusInterrupted. Note that if there
+    /// data waiting to be read and an interrupt request is issued, the Read()
+    /// function will return the data immediately without processing the
+    /// interrupt request (which will remain queued for the next Read()
+    /// operation).
+    ///
+    /// @return
+    ///     Returns true is the interrupt request was sucessful.
+    //------------------------------------------------------------------
+    virtual bool
+    InterruptRead() = 0;
+
 private:
     //------------------------------------------------------------------
     // For Connection only
     //------------------------------------------------------------------
     DISALLOW_COPY_AND_ASSIGN (Connection);
 };
 
 } // namespace lldb_private
 
 #endif  // liblldb_Connection_h_

lldb/trunk/include/lldb/Host/posix/ConnectionFileDescriptorPosix.h

public:
     virtual size_t Read(void *dst, size_t dst_len, uint32_t timeout_usec, lldb::ConnectionStatus &status, Error *error_ptr);
 
     virtual size_t Write(const void *src, size_t src_len, lldb::ConnectionStatus &status, Error *error_ptr);
 
     virtual std::string GetURI();
 
     lldb::ConnectionStatus BytesAvailable(uint32_t timeout_usec, Error *error_ptr);
 
-    bool InterruptRead();
+    bool InterruptRead() override;
 
     lldb::IOObjectSP
     GetReadObject()
     {
         return m_read_sp;
     }
     const lldb::IOObjectSP
     GetReadObject() const

lldb/trunk/include/lldb/Host/windows/ConnectionGenericFileWindows.h

public:
     virtual lldb::ConnectionStatus Disconnect(Error *error_ptr);
 
     virtual size_t Read(void *dst, size_t dst_len, uint32_t timeout_usec, lldb::ConnectionStatus &status, Error *error_ptr);
 
     virtual size_t Write(const void *src, size_t src_len, lldb::ConnectionStatus &status, Error *error_ptr);
 
     virtual std::string GetURI();
 
-    bool InterruptRead();
+    bool InterruptRead() override;
 
   protected:
     OVERLAPPED m_overlapped;
     HANDLE m_file;
     HANDLE m_event_handles[2];
     bool m_owns_file;
     LARGE_INTEGER m_file_position;
 

lldb/trunk/source/Core/Communication.cpp

 
 //----------------------------------------------------------------------
 // Constructor
 //----------------------------------------------------------------------
 Communication::Communication(const char *name) :
     Broadcaster (NULL, name),
     m_connection_sp (),
     m_read_thread_enabled (false),
+    m_read_thread_did_exit (false),
     m_bytes(),
     m_bytes_mutex (Mutex::eMutexTypeRecursive),
     m_write_mutex (Mutex::eMutexTypeNormal),
+    m_synchronize_mutex (Mutex::eMutexTypeNormal),
     m_callback (NULL),
     m_callback_baton (NULL),
     m_close_on_eof (true)
 
 {
     lldb_private::LogIfAnyCategoriesSet (LIBLLDB_LOG_OBJECT | LIBLLDB_LOG_COMMUNICATION,
                                  "%p Communication::Communication (name = %s)",
                                  this, name);
 
     SetEventName (eBroadcastBitDisconnected, "disconnected");
     SetEventName (eBroadcastBitReadThreadGotBytes, "got bytes");
     SetEventName (eBroadcastBitReadThreadDidExit, "read thread did exit");
     SetEventName (eBroadcastBitReadThreadShouldExit, "read thread should exit");
     SetEventName (eBroadcastBitPacketAvailable, "packet available");
+    SetEventName (eBroadcastBitNoMorePendingInput, "no more pending input");
     
     CheckInWithManager();
 }
 
 //----------------------------------------------------------------------
 // Destructor
 //----------------------------------------------------------------------
 Communication::~Communication()
Communication::StartReadThread (Error *error_ptr)
     lldb_private::LogIfAnyCategoriesSet (LIBLLDB_LOG_COMMUNICATION,
                                  "%p Communication::StartReadThread ()", this);
 
 
     char thread_name[1024];
     snprintf(thread_name, sizeof(thread_name), "<lldb.comm.%s>", m_broadcaster_name.AsCString());
 
     m_read_thread_enabled = true;
+    m_read_thread_did_exit = false;
     m_read_thread = ThreadLauncher::LaunchThread(thread_name, Communication::ReadThread, this, error_ptr);
     if (!m_read_thread.IsJoinable())
         m_read_thread_enabled = false;
     return m_read_thread_enabled;
 }
 
 bool
 Communication::StopReadThread (Error *error_ptr)
while (!done && comm->m_read_thread_enabled)
                 done = true;
             }
             if (log)
                 error.LogIfError (log,
                                   "%p Communication::ReadFromConnection () => status = %s",
                                   p,
                                   Communication::ConnectionStatusAsCString (status));
             break;
+        case eConnectionStatusInterrupted:      // Synchronization signal from SynchronizeWithReadThread()
+            // The connection returns eConnectionStatusInterrupted only when there is no
+            // input pending to be read, so we can signal that.
+            comm->BroadcastEvent (eBroadcastBitNoMorePendingInput);
+            break;
         case eConnectionStatusNoConnection:     // No connection
         case eConnectionStatusLostConnection:   // Lost connection while connected to a valid connection
-        case eConnectionStatusInterrupted:      // Interrupted
             done = true;
             // Fall through...
         case eConnectionStatusTimedOut:         // Request timed out
             if (log)
                 error.LogIfError (log,
                                   "%p Communication::ReadFromConnection () => status = %s",
                                   p,
                                   Communication::ConnectionStatusAsCString (status));
             break;
         }
     }
     log = lldb_private::GetLogIfAllCategoriesSet (LIBLLDB_LOG_COMMUNICATION);
     if (log)
         log->Printf ("%p Communication::ReadThread () thread exiting...", p);
 
+    comm->m_read_thread_did_exit = true;
     // Let clients know that this thread is exiting
+    comm->BroadcastEvent (eBroadcastBitNoMorePendingInput);
     comm->BroadcastEvent (eBroadcastBitReadThreadDidExit);
     return NULL;
 }
 
 void
 Communication::SetReadThreadBytesReceivedCallback
 (
     ReadThreadBytesReceived callback,
     void *callback_baton
 )
 {
     m_callback = callback;
     m_callback_baton = callback_baton;
 }
 
 void
+Communication::SynchronizeWithReadThread ()
+{
+    // Only one thread can do the synchronization dance at a time.
+    Mutex::Locker locker(m_synchronize_mutex);
+
+    // First start listening for the synchronization event.
+    Listener listener("Communication::SyncronizeWithReadThread");
+    listener.StartListeningForEvents(this, eBroadcastBitNoMorePendingInput);
+
+    // If the thread is not running, there is no point in synchronizing.
+    if (!m_read_thread_enabled || m_read_thread_did_exit)
+        return;
+
+    // Notify the read thread.
+    m_connection_sp->InterruptRead();
+
+    // Wait for the synchronization event.
+    EventSP event_sp;
+    listener.WaitForEvent(NULL, event_sp);
+}
+
+void
 Communication::SetConnection (Connection *connection)
 {
     Disconnect (NULL);
     StopReadThread(NULL);
     m_connection_sp.reset(connection);
 }
 
 const char *

lldb/trunk/source/Target/Process.cpp

 Process::ShouldBroadcastEvent (Event *event_ptr)
 {
     const StateType state = Process::ProcessEventData::GetStateFromEvent (event_ptr);
     bool return_value = true;
     Log *log(lldb_private::GetLogIfAnyCategoriesSet(LIBLLDB_LOG_EVENTS | LIBLLDB_LOG_PROCESS));
     
     switch (state)
     {
-        case eStateConnected:
-        case eStateAttaching:
-        case eStateLaunching:
         case eStateDetached:
         case eStateExited:
         case eStateUnloaded:
+            m_stdio_communication.SynchronizeWithReadThread();
+            // fall-through
+        case eStateConnected:
+        case eStateAttaching:
+        case eStateLaunching:
             // These events indicate changes in the state of the debugging session, always report them.
             return_value = true;
             break;
         case eStateInvalid:
             // We stopped for no apparent reason, don't report it.
             return_value = false;
             break;
         case eStateRunning:
switch (state)
         case eStateCrashed:
         case eStateSuspended:
         {
             // We've stopped.  First see if we're going to restart the target.
             // If we are going to stop, then we always broadcast the event.
             // If we aren't going to stop, let the thread plans decide if we're going to report this event.
             // If no thread has an opinion, we don't report it.
 
+            m_stdio_communication.SynchronizeWithReadThread();
             RefreshStateAfterStop ();
             if (ProcessEventData::GetInterruptedFromEvent (event_ptr))
             {
                 if (log)
                     log->Printf ("Process::ShouldBroadcastEvent (%p) stopped due to an interrupt, state: %s",
                                  static_cast<void*>(event_ptr),
                                  StateAsCString(state));
                 // Even though we know we are going to stop, we should let the threads have a look at the stop,

lldb/trunk/test/python_api/process/io/TestProcessIO.py

class ProcessIOTestCase(TestBase):
     def test_stdin_by_api_with_dsym(self):
         """Exercise SBProcess.PutSTDIN()."""
         self.buildDsym()
         self.do_stdin_by_api()
 
     @unittest2.skipIf(sys.platform.startswith("win32"), "stdio manipulation unsupported on Windows")
     @python_api_test
     @dwarf_test
-    @expectedFailureLinux # this test fails 7/100 dosep runs
     def test_stdin_by_api_with_dwarf(self):
         """Exercise SBProcess.PutSTDIN()."""
         self.buildDwarf()
         self.do_stdin_by_api()
 
     @unittest2.skipUnless(sys.platform.startswith("darwin"), "dsym requires Darwin")
     @python_api_test
     @dsym_test
     def test_stdin_redirection_with_dsym(self):
         """Exercise SBLaunchInfo::AddOpenFileAction() for STDIN without specifying STDOUT or STDERR."""
         self.buildDsym()
         self.do_stdin_redirection()
 
     @unittest2.skipIf(sys.platform.startswith("win32"), "stdio manipulation unsupported on Windows")
     @python_api_test
     @dwarf_test
-    @expectedFailureLinux # this test fails 4/100 dosep runs
     def test_stdin_redirection_with_dwarf(self):
         """Exercise SBLaunchInfo::AddOpenFileAction() for STDIN without specifying STDOUT or STDERR."""
         self.buildDwarf()
         self.do_stdin_redirection()
 
     @unittest2.skipUnless(sys.platform.startswith("darwin"), "dsym requires Darwin")
     @python_api_test
     @dsym_test
     def test_stdout_redirection_with_dsym(self):
         """Exercise SBLaunchInfo::AddOpenFileAction() for STDOUT without specifying STDIN or STDERR."""
         self.buildDsym()
         self.do_stdout_redirection()
 
     @unittest2.skipIf(sys.platform.startswith("win32"), "stdio manipulation unsupported on Windows")
     @python_api_test
     @dwarf_test
-    @expectedFailureLinux # this test fails 2/100 dosep runs
     def test_stdout_redirection_with_dwarf(self):
         """Exercise SBLaunchInfo::AddOpenFileAction() for STDOUT without specifying STDIN or STDERR."""
         self.buildDwarf()
         self.do_stdout_redirection()
 
     @unittest2.skipUnless(sys.platform.startswith("darwin"), "dsym requires Darwin")
     @python_api_test
     @dsym_test
     def test_stderr_redirection_with_dsym(self):
         """Exercise SBLaunchInfo::AddOpenFileAction() for STDERR without specifying STDIN or STDOUT."""
         self.buildDsym()
         self.do_stderr_redirection()
 
     @unittest2.skipIf(sys.platform.startswith("win32"), "stdio manipulation unsupported on Windows")
     @python_api_test
     @dwarf_test
-    @expectedFailureLinux # this test fails 5/100 dosep runs
     def test_stderr_redirection_with_dwarf(self):
         """Exercise SBLaunchInfo::AddOpenFileAction() for STDERR without specifying STDIN or STDOUT."""
         self.buildDwarf()
         self.do_stderr_redirection()
 
     @unittest2.skipUnless(sys.platform.startswith("darwin"), "dsym requires Darwin")
     @python_api_test
     @dsym_test