csp event loop, async bridges

Signed-off-by: Tim Paine <[email protected]>

Author: timkpaine

cpp/csp/engine/RootEngine.cpp

@@ -75,6 +75,7 @@ RootEngine::RootEngine( const Dictionary & settings ) : Engine( m_cycleStepTable
                                                         m_cycleCount( 0 ),
                                                         m_settings( settings ),
                                                         m_inRealtime( false ),
+                                                        m_stepping( false ),
                                                         m_initSignalCount( g_SIGNAL_COUNT ),
                                                         m_pushEventQueue( m_settings.queueWaitTime > TimeDelta::ZERO() )
 {
@@ -222,7 +223,7 @@ void RootEngine::runRealtime( DateTime end )
 
             for( auto * group : dirtyGroups )
                 group -> state = PushGroup::NONE;
-            
+
             dirtyGroups.clear();
             haveEvents = false;
         }
@@ -289,6 +290,144 @@ void RootEngine::shutdown( std::exception_ptr except_ptr )
         m_exception_ptr = except_ptr;
 }
 
+void RootEngine::startStepping( DateTime start, DateTime end )
+{
+    if( m_stepping )
+        CSP_THROW( RuntimeException, "Engine is already in stepping mode" );
+
+    preRun( start, end );
+
+    m_exception_mutex.lock();
+    if( m_state != State::SHUTDOWN )
+        m_state = State::RUNNING;
+    m_exception_mutex.unlock();
+
+    m_stepping = true;
+    m_inRealtime = m_settings.realtime;
+    m_stepDirtyGroups.clear();
+}
+
+bool RootEngine::step( TimeDelta maxWait )
+{
+    if( !m_stepping )
+        CSP_THROW( RuntimeException, "Engine is not in stepping mode. Call startStepping() first." );
+
+    if( m_state != State::RUNNING || interrupted() )
+        return false;
+
+    // Check if we've passed the end time
+    if( m_now > m_endTime )
+        return false;
+
+    bool hasWork = false;
+
+    if( m_inRealtime )
+    {
+        // Realtime mode: check for push events and timers
+        TimeDelta waitTime = maxWait;
+        if( waitTime == TimeDelta::NONE() )
+            waitTime = TimeDelta::ZERO();
+
+        if( !m_pendingPushEvents.hasEvents() )
+        {
+            DateTime now = DateTime::now();
+            if( m_scheduler.hasEvents() )
+                waitTime = std::min( m_scheduler.nextTime() - now, waitTime );
+        }
+
+        // Don't block in step mode - just check for events
+        bool haveEvents = m_pushEventQueue.wait( waitTime );
+
+        m_now = DateTime::now();
+        if( m_now > m_endTime )
+        {
+            m_now = m_endTime;
+            return false;
+        }
+
+        ++m_cycleCount;
+
+        // Execute timers that are ready
+        if( m_scheduler.hasEvents() && m_scheduler.nextTime() <= m_now )
+        {
+            DateTime timerTime = m_scheduler.nextTime();
+            m_now = timerTime;
+            m_scheduler.executeNextEvents( m_now );
+            hasWork = true;
+        }
+        else if( haveEvents )
+        {
+            // Process push events
+            PushEvent * events = m_pushEventQueue.popAll();
+            processPendingPushEvents( m_stepDirtyGroups );
+            processPushEventQueue( events, m_stepDirtyGroups );
+
+            for( auto * group : m_stepDirtyGroups )
+                group -> state = PushGroup::NONE;
+            m_stepDirtyGroups.clear();
+            hasWork = true;
+        }
+    }
+    else
+    {
+        // Sim mode: process next scheduled event
+        if( m_scheduler.hasEvents() )
+        {
+            m_now = m_scheduler.nextTime();
+            if( m_now <= m_endTime )
+            {
+                ++m_cycleCount;
+                m_scheduler.executeNextEvents( m_now );
+                hasWork = true;
+            }
+            else
+            {
+                m_now = m_endTime;
+            }
+        }
+    }
+
+    if( hasWork )
+    {
+        m_cycleStepTable.executeCycle( m_profiler.get() );
+        processEndCycle();
+    }
+
+    // Return true if there's more work to do
+    return m_state == State::RUNNING && !interrupted() &&
+           ( m_scheduler.hasEvents() || m_pendingPushEvents.hasEvents() );
+}
+
+void RootEngine::stopStepping()
+{
+    if( !m_stepping )
+        return;
+
+    try
+    {
+        postRun();
+    }
+    catch( ... )
+    {
+        if( !m_exception_ptr )
+            m_exception_ptr = std::current_exception();
+    }
+
+    m_state = State::DONE;
+    m_stepping = false;
+    m_stepDirtyGroups.clear();
+
+    if( m_exception_ptr )
+        std::rethrow_exception( m_exception_ptr );
+}
+
+DateTime RootEngine::nextScheduledTime()
+{
+    if( m_scheduler.hasEvents() )
+        return m_scheduler.nextTime();
+    return DateTime::NONE();
+}
+
 DictionaryPtr RootEngine::engine_stats() const
 {
     if( !m_profiler )

cpp/csp/engine/RootEngine.h

@@ -25,7 +25,7 @@ class EndCycleListener
 public:
     virtual ~EndCycleListener() {};
     virtual void onEndCycle() = 0;
-    
+
     bool isDirty() const  { return m_dirty; }
     void setDirtyFlag()   { m_dirty = true; }
     void clearDirtyFlag() { m_dirty = false; }
@@ -57,6 +57,14 @@ class RootEngine : public Engine
     void     shutdown();
     void     shutdown( std::exception_ptr except );
 
+    // Step-based execution API for asyncio integration
+    // Allows external event loops to drive CSP execution one step at a time
+    void     startStepping( DateTime start, DateTime end );
+    bool     step( TimeDelta maxWait = TimeDelta::NONE() );  // Returns true if more work pending
+    void     stopStepping();
+    bool     isStepping() const { return m_stepping; }
+    DateTime nextScheduledTime();  // Returns next scheduled event time, or NONE if none
+
     Scheduler::Handle reserveSchedulerHandle();
     Scheduler::Handle scheduleCallback( TimeDelta delta, Scheduler::Callback cb );
     Scheduler::Handle scheduleCallback( DateTime time, Scheduler::Callback cb );
@@ -90,7 +98,7 @@ class RootEngine : public Engine
     bool interrupted() const;
 
     PushPullEventQueue & pushPullEventQueue() { return m_pushPullEventQueue; }
-    
+
 protected:
     enum State { NONE, STARTING, RUNNING, SHUTDOWN, DONE };
     using EndCycleListeners = std::vector<EndCycleListener*>;
@@ -131,8 +139,12 @@ class RootEngine : public Engine
     PendingPushEvents m_pendingPushEvents;
     Settings          m_settings;
     bool              m_inRealtime;
+    bool              m_stepping;  // True when in step-based execution mode
     int               m_initSignalCount;
 
+    // For step-based execution
+    std::vector<PushGroup *> m_stepDirtyGroups;
+
     PushEventQueue     m_pushEventQueue;
     //This queue is managed entirely from the PushPullInputAdapter
     PushPullEventQueue m_pushPullEventQueue;
@@ -168,7 +180,7 @@ inline Scheduler::Handle RootEngine::scheduleCallback( Scheduler::Handle reserve
     if( unlikely( time < m_now ) )
         CSP_THROW( ValueError, "Cannot schedule event in the past.  new time: " << time << " now: " << m_now );
 
-    return m_scheduler.scheduleCallback( reservedHandle, time, std::move( cb ) ); 
+    return m_scheduler.scheduleCallback( reservedHandle, time, std::move( cb ) );
 }
 
 inline Scheduler::Handle RootEngine::rescheduleCallback( Scheduler::Handle id, csp::DateTime time )

cpp/csp/python/CMakeLists.txt

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

cpp/csp/python/PyEngine.cpp

@@ -95,7 +95,7 @@ static void PyEngine_dealloc( PyEngine * self )
 {
     CSP_BEGIN_METHOD;
     self -> ~PyEngine();
-    Py_TYPE( self ) -> tp_free( self ); 
+    Py_TYPE( self ) -> tp_free( self );
     CSP_RETURN;
 }
 
@@ -105,7 +105,7 @@ static PyObject * PyEngine_run( PyEngine * self, PyObject * args )
 
     PyObject * pyStart;
     PyObject * pyEnd;
-    if( !PyArg_ParseTuple( args, "OO", &pyStart, &pyEnd ) ) 
+    if( !PyArg_ParseTuple( args, "OO", &pyStart, &pyEnd ) )
         return nullptr;
 
     auto start = fromPython<DateTime>( pyStart );
@@ -118,8 +118,94 @@ static PyObject * PyEngine_run( PyEngine * self, PyObject * args )
     CSP_RETURN_NONE;
 }
 
+static PyObject * PyEngine_startStepping( PyEngine * self, PyObject * args )
+{
+    CSP_BEGIN_METHOD;
+
+    PyObject * pyStart;
+    PyObject * pyEnd;
+    if( !PyArg_ParseTuple( args, "OO", &pyStart, &pyEnd ) )
+        return nullptr;
+
+    auto start = fromPython<DateTime>( pyStart );
+    auto end   = fromPython<DateTime>( pyEnd );
+
+    CSP_TRUE_OR_THROW_RUNTIME( self -> engine() -> isRootEngine(), "engine is not root engine" );
+    self -> rootEngine() -> startStepping( start, end );
+
+    Py_RETURN_NONE;
+    CSP_RETURN_NONE;
+}
+
+static PyObject * PyEngine_step( PyEngine * self, PyObject * args )
+{
+    CSP_BEGIN_METHOD;
+
+    double maxWaitSeconds = 0.0;
+    if( !PyArg_ParseTuple( args, "|d", &maxWaitSeconds ) )
+        return nullptr;
+
+    CSP_TRUE_OR_THROW_RUNTIME( self -> engine() -> isRootEngine(), "engine is not root engine" );
+
+    TimeDelta maxWait = maxWaitSeconds > 0 ? TimeDelta::fromSeconds( maxWaitSeconds ) : TimeDelta::ZERO();
+    bool hasMore = self -> rootEngine() -> step( maxWait );
+
+    return PyBool_FromLong( hasMore );
+    CSP_RETURN_NONE;
+}
+
+static PyObject * PyEngine_stopStepping( PyEngine * self, PyObject * args )
+{
+    CSP_BEGIN_METHOD;
+
+    CSP_TRUE_OR_THROW_RUNTIME( self -> engine() -> isRootEngine(), "engine is not root engine" );
+    self -> rootEngine() -> stopStepping();
+
+    return self -> collectOutputs();
+    CSP_RETURN_NONE;
+}
+
+static PyObject * PyEngine_isStepping( PyEngine * self, PyObject * args )
+{
+    CSP_BEGIN_METHOD;
+
+    CSP_TRUE_OR_THROW_RUNTIME( self -> engine() -> isRootEngine(), "engine is not root engine" );
+    bool stepping = self -> rootEngine() -> isStepping();
+
+    return PyBool_FromLong( stepping );
+    CSP_RETURN_NONE;
+}
+
+static PyObject * PyEngine_now( PyEngine * self, PyObject * args )
+{
+    CSP_BEGIN_METHOD;
+
+    CSP_TRUE_OR_THROW_RUNTIME( self -> engine() -> isRootEngine(), "engine is not root engine" );
+    DateTime now = self -> rootEngine() -> now();
+
+    return toPython( now );
+    CSP_RETURN_NONE;
+}
+
+static PyObject * PyEngine_nextScheduledTime( PyEngine * self, PyObject * args )
+{
+    CSP_BEGIN_METHOD;
+
+    CSP_TRUE_OR_THROW_RUNTIME( self -> engine() -> isRootEngine(), "engine is not root engine" );
+    DateTime nextTime = self -> rootEngine() -> nextScheduledTime();
+
+    return toPython( nextTime );
+    CSP_RETURN_NONE;
+}
+
 static PyMethodDef PyEngine_methods[] = {
-    { "run",             (PyCFunction) PyEngine_run,            METH_VARARGS, "start and run engine" },
+    { "run",               ( PyCFunction ) PyEngine_run,               METH_VARARGS, "start and run engine" },
+    { "start_stepping",    ( PyCFunction ) PyEngine_startStepping,     METH_VARARGS, "start engine in step mode" },
+    { "step",              ( PyCFunction ) PyEngine_step,              METH_VARARGS, "execute one step, returns True if more work pending" },
+    { "stop_stepping",     ( PyCFunction ) PyEngine_stopStepping,      METH_NOARGS,  "stop stepping and cleanup" },
+    { "is_stepping",       ( PyCFunction ) PyEngine_isStepping,        METH_NOARGS,  "check if in stepping mode" },
+    { "now",               ( PyCFunction ) PyEngine_now,               METH_NOARGS,  "get current engine time" },
+    { "next_scheduled_time", ( PyCFunction ) PyEngine_nextScheduledTime, METH_NOARGS,  "get next scheduled event time" },
     { NULL }
 };