Fix 4 critical issues

Signed-off-by: Tim Paine <3105306+timkpaine@users.noreply.github.com>

Author: timkpaine

cpp/csp/core/QueueWaiter.h

@@ -8,6 +8,7 @@
 #pragma comment(lib, "ws2_32.lib")
 #endif
 
+#include <cerrno>
 #include <mutex>
 #include <condition_variable>
 #include <csp/core/Time.h>
@@ -61,6 +62,10 @@ class QueueWaiter
 // FdWaiter provides file descriptor based signaling for integration with
 // external event loops like asyncio. The read fd can be registered with
 // select/poll/epoll and will become readable when notify() is called.
+//
+// Consumers must clear() BEFORE draining the event queue they are guarding, never after.
+// Clearing afterwards discards the signal for any event queued between the drain and the clear,
+// which strands that event until the next unrelated wakeup.
 class FdWaiter
 {
 public:
@@ -78,9 +83,15 @@ class FdWaiter
         {
             m_readFd = fds[0];
             m_writeFd = fds[1];
-            // Set non-blocking
-            fcntl( m_readFd, F_SETFL, O_NONBLOCK );
-            fcntl( m_writeFd, F_SETFL, O_NONBLOCK );
+            // notify() writes on every call and must never block, so non-blocking is mandatory
+            if( fcntl( m_readFd, F_SETFL, O_NONBLOCK ) == -1 ||
+                fcntl( m_writeFd, F_SETFL, O_NONBLOCK ) == -1 )
+            {
+                close( m_readFd );
+                close( m_writeFd );
+                m_readFd = -1;
+                m_writeFd = -1;
+            }
         }
         else
         {
@@ -121,43 +132,61 @@ class FdWaiter
     int readFd() const { return m_readFd; }
 #endif
 
-    // Signal the fd (makes it readable)
+    // Signal the fd ( makes it readable ).  Callable from any producer thread.
+    // Deliberately lock-free: this sits on the push-event hot path.  EAGAIN means the eventfd
+    // counter saturated or the pipe buffer is full, ie the fd is already readable, which is the
+    // state notify() is trying to reach - so it needs no handling.
     void notify()
     {
-        std::lock_guard<std::mutex> guard( m_lock );
-        if( m_notified )
-            return;  // Already notified, avoid filling buffer
-
-        m_notified = true;
-
 #ifdef __linux__
         uint64_t val = 1;
-        [[maybe_unused]] auto rv = write( m_eventfd, &val, sizeof( val ) );
+        ssize_t rv;
+        do { rv = write( m_eventfd, &val, sizeof( val ) ); } while( rv < 0 && errno == EINTR );
 #elif defined(__APPLE__)
         char c = 1;
-        [[maybe_unused]] auto rv = write( m_writeFd, &c, 1 );
+        ssize_t rv;
+        do { rv = write( m_writeFd, &c, 1 ); } while( rv < 0 && errno == EINTR );
 #elif defined(_WIN32)
         char c = 1;
-        send( m_writeFd, &c, 1, 0 );
+        int rv;
+        do { rv = send( m_writeFd, &c, 1, 0 ); } while( rv == SOCKET_ERROR && WSAGetLastError() == WSAEINTR );
 #endif
+        ( void ) rv;
     }
 
-    // Clear the notification (call after processing)
+    // Drain the fd.  See the class comment: call this before draining the guarded event queue.
+    // The drain is bounded - producers are unsynchronized and can refill faster than we read, and
+    // leaving bytes behind is harmless: the fd simply stays readable and costs one extra cycle.
     void clear()
     {
-        std::lock_guard<std::mutex> guard( m_lock );
-        m_notified = false;
-
 #ifdef __linux__
         uint64_t val;
-        [[maybe_unused]] auto rv = read( m_eventfd, &val, sizeof( val ) );
+        ssize_t rv;
+        do { rv = read( m_eventfd, &val, sizeof( val ) ); } while( rv < 0 && errno == EINTR );
 #elif defined(__APPLE__)
-        char buf[64];
-        while( read( m_readFd, buf, sizeof( buf ) ) > 0 ) {}
+        char buf[ DRAIN_BUFFER_SIZE ];
+        ssize_t rv = 0;
+        for( size_t i = 0; i < DRAIN_MAX_READS; ++i )
+        {
+            rv = read( m_readFd, buf, sizeof( buf ) );
+            if( rv < 0 && errno == EINTR )
+                continue;
+            if( rv <= 0 )
+                break;
+        }
 #elif defined(_WIN32)
-        char buf[64];
-        while( recv( m_readFd, buf, sizeof( buf ), 0 ) > 0 ) {}
+        char buf[ DRAIN_BUFFER_SIZE ];
+        int rv = 0;
+        for( size_t i = 0; i < DRAIN_MAX_READS; ++i )
+        {
+            rv = recv( m_readFd, buf, sizeof( buf ), 0 );
+            if( rv == SOCKET_ERROR && WSAGetLastError() == WSAEINTR )
+                continue;
+            if( rv <= 0 )
+                break;
+        }
 #endif
+        ( void ) rv;
     }
 
     bool isValid() const
@@ -170,6 +199,9 @@ class FdWaiter
     }
 
 private:
+    static constexpr size_t DRAIN_BUFFER_SIZE = 64;
+    static constexpr size_t DRAIN_MAX_READS   = 64;
+
 #ifdef _WIN32
     void createSocketPair()
     {
@@ -230,10 +262,16 @@ class FdWaiter
             return;
         }
 
-        // Set non-blocking
+        // Set non-blocking; notify() must never block on a full buffer
         u_long mode = 1;
-        ioctlsocket( m_readFd, FIONBIO, &mode );
-        ioctlsocket( m_writeFd, FIONBIO, &mode );
+        if( ioctlsocket( m_readFd, FIONBIO, &mode ) == SOCKET_ERROR ||
+            ioctlsocket( m_writeFd, FIONBIO, &mode ) == SOCKET_ERROR )
+        {
+            closesocket( m_readFd );
+            closesocket( m_writeFd );
+            m_readFd = INVALID_SOCKET;
+            m_writeFd = INVALID_SOCKET;
+        }
     }
 
     SOCKET m_readFd;
@@ -245,8 +283,6 @@ class FdWaiter
     int m_eventfd;
 #endif
 #endif
-    std::mutex m_lock;
-    bool m_notified = false;
 };
 
 }

cpp/csp/engine/RootEngine.cpp

@@ -93,6 +93,8 @@ RootEngine::RootEngine( const Dictionary & settings ) : Engine( m_cycleStepTable
 
 RootEngine::~RootEngine()
 {
+    // Defense in depth for callers of the decomposed API that abandon an engine without finish()
+    m_fdWaiterEnabled.store( false, std::memory_order_relaxed );
 }
 
 bool RootEngine::interrupted() const
@@ -115,6 +117,8 @@ void RootEngine::preRun( DateTime start, DateTime end )
 void RootEngine::postRun()
 {
     m_state = State::SHUTDOWN;
+    // Disarm before stopping adapters so producer threads stop reaching for the fd on their way out
+    m_fdWaiterEnabled.store( false, std::memory_order_relaxed );
     stop();
 }
 

cpp/csp/engine/RootEngine.h

@@ -14,6 +14,7 @@
 #include <csp/engine/PushEvent.h>
 #include <csp/engine/PushPullEvent.h>
 #include <csp/engine/Scheduler.h>
+#include <atomic>
 #include <memory>
 
 namespace csp
@@ -81,8 +82,8 @@ class RootEngine : public Engine
 
     void     cancelCallback( Scheduler::Handle handle );
 
-    void     schedulePushEvent( PushEvent * event )             { m_pushEventQueue.push( event ); m_fdWaiter.notify(); }
-    void     schedulePushBatch( PushEventQueue::Batch & batch ) { m_pushEventQueue.push( batch ); m_fdWaiter.notify(); }
+    void     schedulePushEvent( PushEvent * event )             { m_pushEventQueue.push( event ); notifyFdWaiter(); }
+    void     schedulePushBatch( PushEventQueue::Batch & batch ) { m_pushEventQueue.push( batch ); notifyFdWaiter(); }
 
     bool     scheduleEndCycleListener( EndCycleListener * l );
 
@@ -106,8 +107,15 @@ class RootEngine : public Engine
     PushPullEventQueue & pushPullEventQueue() { return m_pushPullEventQueue; }
 
     // Native fd-based wakeup for external event loops (asyncio, etc.)
-    // Returns a file descriptor that becomes readable when events are queued
-    int getWakeupFd() const { return m_fdWaiter.readFd(); }
+    // Returns a file descriptor that becomes readable when events are queued.  Asking for the fd
+    // is what arms the signalling: until then the push path does no fd work at all.  Arming is
+    // one-way for the life of the run; it is disarmed only at teardown.
+    int getWakeupFd()
+    {
+        if( m_fdWaiter.isValid() )
+            m_fdWaiterEnabled.store( true, std::memory_order_relaxed );
+        return m_fdWaiter.readFd();
+    }
     void clearWakeupFd() { m_fdWaiter.clear(); }
 
 protected:
@@ -123,6 +131,15 @@ class RootEngine : public Engine
 
     void    processEndCycle();
 
+    // Signals only once a consumer has armed the fd, keeping the push hot path syscall-free.
+    // The flag carries no data, hence relaxed; it is a hint, not a synchronization point, and it
+    // does not make teardown safe - adapters must still be stopped before the engine is destroyed.
+    void    notifyFdWaiter()
+    {
+        if( m_fdWaiterEnabled.load( std::memory_order_relaxed ) ) [[unlikely]]
+            m_fdWaiter.notify();
+    }
+
     struct Settings
     {
         Settings( const Dictionary & );
@@ -161,6 +178,9 @@ class RootEngine : public Engine
     std::mutex                        m_exception_mutex;
     std::unique_ptr<csp::Profiler>    m_profiler;
     mutable FdWaiter                  m_fdWaiter;  // For native fd-based event loop integration
+    // Only armed once a consumer asks for the fd, and disarmed at teardown, so the default
+    // engine.run() path never touches the fd from the push hot path
+    std::atomic<bool>                 m_fdWaiterEnabled{ false };
 
 };
 

cpp/csp/python/CMakeLists.txt

@@ -32,7 +32,6 @@ set(CSPIMPL_PUBLIC_HEADERS
         NumpyConversions.h
         NumpyInputAdapter.h
         PyAdapterManagerWrapper.h
-        PyEventLoop.h
         PyBasketInputProxy.h
         PyBasketOutputProxy.h
         PyCppNode.h
@@ -57,7 +56,6 @@ add_library(cspimpl SHARED
         NumpyConversions.cpp
         PyAdapterManager.cpp
         PyAdapterManagerWrapper.cpp
-        PyEventLoop.cpp
         PyConstAdapter.cpp
         PyCppNode.cpp
         PyEngine.cpp

cpp/csp/python/PyEngine.cpp

@@ -135,6 +135,9 @@ static PyObject * PyEngine_start( PyEngine * self, PyObject * args )
     CSP_TRUE_OR_THROW_RUNTIME( self -> engine() -> isRootEngine(), "engine is not root engine" );
     self -> rootEngine() -> start( start, end );
 
+    if( PyErr_Occurred() )
+        return nullptr;
+
     Py_RETURN_NONE;
     CSP_RETURN_NONE;
 }
@@ -155,6 +158,10 @@ static PyObject * PyEngine_processOneCycle( PyEngine * self, PyObject * args )
     TimeDelta maxWait = TimeDelta::fromNanoseconds( maxWaitNanos );
     bool hasMore = self -> rootEngine() -> processOneCycle( maxWait );
 
+    // dialectLockGIL runs PyErr_CheckSignals, so a KeyboardInterrupt may be pending here
+    if( PyErr_Occurred() )
+        return nullptr;
+
     return PyBool_FromLong( hasMore );
     CSP_RETURN_NONE;
 }
@@ -166,7 +173,7 @@ static PyObject * PyEngine_finish( PyEngine * self, PyObject * args )
     CSP_TRUE_OR_THROW_RUNTIME( self -> engine() -> isRootEngine(), "engine is not root engine" );
     self -> rootEngine() -> finish();
 
-    return self -> collectOutputs();
+    return PyErr_Occurred() ? nullptr : self -> collectOutputs();
     CSP_RETURN_NONE;
 }
 
@@ -233,7 +240,7 @@ static PyMethodDef PyEngine_methods[] = {
     { "now",               ( PyCFunction ) PyEngine_now,               METH_NOARGS,  "get current engine time" },
     { "next_scheduled_time", ( PyCFunction ) PyEngine_nextScheduledTime, METH_NOARGS,  "get next scheduled event time" },
     { "get_wakeup_fd",     ( PyCFunction ) PyEngine_getWakeupFd,       METH_NOARGS,  "get fd that becomes readable when events are queued" },
-    { "clear_wakeup_fd",   ( PyCFunction ) PyEngine_clearWakeupFd,     METH_NOARGS,  "clear the wakeup fd after processing events" },
+    { "clear_wakeup_fd",   ( PyCFunction ) PyEngine_clearWakeupFd,     METH_NOARGS,  "drain the wakeup fd; call before processing events, never after" },
     { NULL }
 };