PPI: Fix senf::ppi::init() event hook scheduling

PPI/Module.test.cc

@@ -27,6 +27,8 @@
 //#include "Module.test.ih"
 
 // Custom includes
+#include <boost/scoped_ptr.hpp>
+#include "../Utils/membind.hh"
 #include "DebugEvent.hh"
 #include "DebugModules.hh"
 #include "Setup.hh"
@@ -80,6 +82,57 @@ BOOST_AUTO_UNIT_TEST(module)
     BOOST_CHECK( senf::ClockService::now() - tester.time() < senf::ClockService::seconds(1) );
 }
 
+namespace {
+
+    void timeout() {
+        senf::scheduler::terminate();
+    }
+    
+    class InitTest : public ppi::module::Module
+    {
+        SENF_PPI_MODULE(InitTest);
+    public:
+        InitTest() : init (false) {}
+        void v_init() { init = true; }
+
+        bool init;
+    };
+
+    struct MakeInit {
+        boost::scoped_ptr<InitTest> tester;
+        void make() {
+            tester.reset(new InitTest());
+        }
+        void test() {
+            BOOST_REQUIRE( tester );
+            BOOST_CHECK( tester->init );
+        }
+    };
+
+}
+
+BOOST_AUTO_UNIT_TEST(delayedInit)
+{
+    MakeInit maker;
+    senf::scheduler::TimerEvent timer ( 
+        "delayedInit timer",
+        senf::membind(&MakeInit::make, &maker),
+        senf::ClockService::now() + senf::ClockService::milliseconds(250) );
+    senf::scheduler::TimerEvent testTimer (
+        "delayedInit test",
+        senf::membind(&MakeInit::test, &maker),
+        senf::ClockService::now() + senf::ClockService::milliseconds(500) );
+    senf::scheduler::TimerEvent timeoutTimer (
+        "delayedInit timeout",
+        &timeout,
+        senf::ClockService::now() + senf::ClockService::milliseconds(750) );
+
+    senf::ppi::run();
+
+    BOOST_REQUIRE( maker.tester );
+    BOOST_CHECK( maker.tester->init );
+}
+
 ///////////////////////////////cc.e////////////////////////////////////////
 #undef prefix_
 

PPI/ModuleManager.cc

@@ -71,7 +71,7 @@ prefix_ void senf::ppi::ModuleManager::run()
 prefix_ senf::ppi::ModuleManager::ModuleManager()
     : running_(false), terminate_(false), 
       initRunner_ ("senf::ppi::init", membind(&ModuleManager::init, this), false, 
-                   scheduler::EventHook::POST)
+                   scheduler::EventHook::PRE)
 {}
 
 ///////////////////////////////cc.e////////////////////////////////////////

PPI/ModuleManager.cci

@@ -54,6 +54,10 @@ prefix_ void senf::ppi::ModuleManager::registerInitializable(Initializable & i)
 {
     initQueue_.push_back(&i);
     initRunner_.enable();
+    // This call ensures, that the senf::ppi::init() handler is called as next handler
+    // after this handler returns (this works since the senf::ppi::init() handler is registered as 
+    // PRE hook and thus has very high priority)
+    senf::scheduler::yield();
 }
 
 prefix_ void senf::ppi::ModuleManager::unregisterInitializable(Initializable & i)

PPI/Route.test.cc

@@ -27,6 +27,7 @@
 //#include "Route.test.ih"
 
 // Custom includes
+#include <boost/scoped_ptr.hpp>
 #include "Route.hh"
 #include "DebugEvent.hh"
 #include "DebugModules.hh"
@@ -36,6 +37,7 @@
 #include "Joins.hh"
 #include "PassiveQueue.hh"
 #include "../Utils/membind.hh"
+#include "../Utils/senfassert.hh"
 
 #include "../Utils/auto_unit_test.hh"
 #include <boost/test/test_tools.hpp>
@@ -190,9 +192,10 @@ namespace {
         ModuleConnector( module::PriorityJoin & join)
             : join_( join) {};
         void connect() {
-            module::PassiveQueue * queue = new module::PassiveQueue;
+            queue.reset(new module::PassiveQueue);
             ppi::connect( *queue, join_, 0);
         }
+        boost::scoped_ptr<module::PassiveQueue> queue;
     };
     
     class TestSink : public module::Module
@@ -206,7 +209,7 @@ namespace {
         }
     private:
         void request() {
-            (void) input.read().data();
+            SENF_ASSERT(input());
         }
     };
 }
@@ -217,26 +220,24 @@ BOOST_AUTO_UNIT_TEST(connect_runtime)
     module::ActiveFeeder feeder;
     module::PriorityJoin join;
     module::CloneSource source1 (senf::DataPacket::create());
-    module::PassiveQueue queue;
     
-    ppi::connect( feeder, sink);
-    ppi::connect( join, feeder);
     ppi::connect( source1, join);
+    ppi::connect( join, feeder);
+    ppi::connect( feeder, sink);
     
     ModuleConnector moduleConnector ( join);
-    senf::scheduler::TimerEvent timer ( "connect_runtime timer",
-        senf::membind(&ModuleConnector::connect, moduleConnector),
-        senf::ClockService::now() + senf::ClockService::milliseconds(25));
+    senf::scheduler::TimerEvent timer ( 
+        "connect_runtime timer",
+        senf::membind(&ModuleConnector::connect, &moduleConnector),
+        senf::ClockService::now() + senf::ClockService::milliseconds(250));
     
     senf::scheduler::TimerEvent timeoutTimer (
         "connect_runtime test timeoutTimer", &timeout,
-        senf::ClockService::now() + senf::ClockService::milliseconds(50));
+        senf::ClockService::now() + senf::ClockService::milliseconds(500));
     
-//    senf::ppi::run();
+    senf::ppi::run();
 }
 
-
-
 ///////////////////////////////cc.e////////////////////////////////////////
 #undef prefix_
 

Scheduler/EventHook.cci

@@ -56,8 +56,10 @@ prefix_ void senf::scheduler::EventHook::disable()
 
 prefix_ void senf::scheduler::EventHook::enable()
 {
-    if (! enabled())
+    if (! enabled()) {
         detail::EventHookDispatcher::instance().add(*this);
+        setRunnable();
+    }
 }
 
 prefix_ void senf::scheduler::EventHook::action(Callback const & cb)

Scheduler/FIFORunner.cc

@@ -40,7 +40,7 @@
 
 prefix_ senf::scheduler::detail::FIFORunner::FIFORunner()
     : tasks_ (), next_ (tasks_.end()), watchdogRunning_ (false), watchdogMs_ (1000), 
-      watchdogAbort_ (false), watchdogCount_(0), hangCount_ (0)
+      watchdogAbort_ (false), watchdogCount_(0), hangCount_ (0), yield_ (false)
 {
     struct sigevent ev;
     ::memset(&ev, 0, sizeof(ev));
@@ -134,17 +134,32 @@ prefix_ void senf::scheduler::detail::FIFORunner::dequeue(TaskInfo * task)
 
 prefix_ void senf::scheduler::detail::FIFORunner::run()
 {
-    TaskList::iterator f (tasks_.begin());
-    TaskList::iterator l (TaskList::current(highPriorityEnd_));
-    run(f, l);
+    for(;;) {
+        TaskList::iterator f (tasks_.begin());
+        TaskList::iterator l (TaskList::current(highPriorityEnd_));
+        run(f, l);
+        if (yield_) {
+            yield_ = false;
+            continue;
+        }
 
-    f = l; ++f;
-    l = TaskList::current(normalPriorityEnd_);
-    run(f, l);
-        
-    f = l; ++f;
-    l = tasks_.end();
-    run(f, l);
+        f = l; ++f;
+        l = TaskList::current(normalPriorityEnd_);
+        run(f, l); 
+        if (yield_) {
+            yield_ = false;
+            continue;
+        }
+       
+        f = l; ++f;
+        l = tasks_.end();
+        run(f, l);
+        if (yield_) {
+            yield_ = false;
+            continue;
+        }
+        break;
+    }
 }
 
 prefix_ void senf::scheduler::detail::FIFORunner::run(TaskList::iterator f, TaskList::iterator l)
@@ -172,7 +187,7 @@ prefix_ void senf::scheduler::detail::FIFORunner::run(TaskList::iterator f, Task
     ScopeExit atExit ((
                           var(watchdogCount_) = 0, 
                           var(next_) = l
-                          ));
+                     ));
     
     while (next_ != end) {
         TaskInfo & task (*next_);
@@ -186,7 +201,10 @@ prefix_ void senf::scheduler::detail::FIFORunner::run(TaskList::iterator f, Task
             ++ next_;
             tasks_.splice(l, tasks_, i);
             watchdogCount_ = 1;
+            yield_ = false;
             task.run();
+            if (yield_)
+                return;
         }
         else
             ++ next_;

Scheduler/FIFORunner.cci

@@ -158,6 +158,11 @@ prefix_ senf::scheduler::detail::FIFORunner::iterator senf::scheduler::detail::F
         EventManager::IteratorFilter(), tasks_.end(), tasks_.end());
 }
 
+prefix_ void senf::scheduler::detail::FIFORunner::yield()
+{
+    yield_ = true;
+}
+
 ///////////////////////////////cci.e///////////////////////////////////////
 #undef prefix_
 

Scheduler/FIFORunner.hh

@@ -110,6 +110,8 @@ namespace detail {
         iterator begin() const;
         iterator end() const;
 
+        void yield();
+
     protected:
 
     private:
@@ -146,6 +148,7 @@ namespace detail {
 #   endif
         unsigned watchdogCount_;
         unsigned hangCount_;
+        bool yield_;
 
         friend void senf::scheduler::restart();
         friend class singleton<FIFORunner>;

Scheduler/Scheduler.cc

@@ -49,6 +49,11 @@ prefix_ void senf::scheduler::terminate()
     terminate_ = true;
 }
 
+prefix_ void senf::scheduler::yield()
+{
+    senf::scheduler::detail::FIFORunner::instance().yield();
+}
+
 prefix_ bool senf::scheduler::running()
 {
     return running_;

Scheduler/Scheduler.hh

@@ -245,6 +245,13 @@ namespace scheduler {
      */
     void terminate(); 
 
+    /** \brief Immediately rescheduler
+
+        Calling yield() will cause the scheduler to terminate the current queue run and immediately
+        rescheduler all pending tasks.
+     */
+    void yield();
+
     /** \brief Return timestamp of last event
 
         This is the timestamp, the last event has been signaled. This is the real time at which the