a few cleanup updates, turning back on basic parallel scheduling, update to pool schedule to enable dynamically adding kernels, need thought into adding destination nodes or not

Author: jonathan-beard

CMakeLists.txt

@@ -99,7 +99,8 @@ set( TESTAPPS allocate portTypeException dynallocate stdlibsupport split
      staticJoinRetStruct 
      staticLongSplitChainRetStruct
      staticSplitChainJoinRetStruct
-     staticSplitJoinRetStruct ) 
+     staticSplitJoinRetStruct 
+     chainMultiplePorts ) 
 
 if( BUILDRANDOM )
     list( APPEND TESTAPPS gamma uniform gaussian exponential sequential ) 

raftinc/join.tcc

@@ -42,7 +42,12 @@ public:
 
    virtual raft::kstatus run()
    {
-      /** multiple calls to allocate will return same reference **/
+      /** 
+       * NOTE: multiple calls to allocate will return same reference,
+       * however we need to deallocate if we want the run-time to be 
+       * able to dynamically re-allocate or move the memory backing
+       * the stream, so call deallocate below if unused
+       */
       auto &output_port( output[ "0" ] );
       T &mem( output_port.template allocate< T >() );
       raft::signal temp_signal;

raftinc/map.hpp

@@ -88,7 +88,7 @@ class map : public MapBase
 #endif             
              , 
              class allocator           = dynalloc,
-             class parallelism_monitor = no_parallel > 
+             class parallelism_monitor = basic_parallel > 
       void exe()
    {
       {

raftinc/poolschedule.hpp

@@ -96,6 +96,7 @@ class pool_schedule : public Schedule
        bool          finished  = false;
        core_id_t     loc       = -1;
     };
+    std::mutex                  thread_data_mutex;
     std::vector< thread_data* > thread_data_pool;
     std::mutex                  tail_mutex;
     std::vector< thread_data* > tail;

src/poolschedule.cpp

@@ -62,9 +62,26 @@ pool_schedule::~pool_schedule()
 void
 pool_schedule::handleSchedule( raft::kernel * const kernel )
 {
-   //TODO implement me 
-   UNUSED( kernel );
-   assert( false );
+    auto *td( new thread_data( kernel ) );
+    thread_data_mutex.lock();
+    thread_data_pool.emplace_back( td );
+    thread_data_mutex.unlock();
+    if( ! kernel->output.hasPorts() /** has no outputs, only 0 > inputs **/ )
+    {
+        std::lock_guard< std::mutex > tail_lock( tail_mutex );
+        /** destination kernel **/
+        tail.emplace_back( td );
+    }
+    qthread_spawn( pool_schedule::pool_run,
+                   (void*) td,
+                   0,
+                   0,
+                   0,
+                   nullptr,
+                   NO_SHEPHERD,
+                   0 );
+    /** done **/
+    return;
 }
 
 void
@@ -78,9 +95,12 @@ pool_schedule::start()
     for( auto * const k : container )
     {  
         auto *td( new thread_data( k ) );
+        thread_data_mutex.lock();
         thread_data_pool.emplace_back( td );
+        thread_data_mutex.unlock();
         if( ! k->output.hasPorts() /** has no outputs, only 0 > inputs **/ )
         {
+            std::lock_guard< std::mutex > tail_lock( tail_mutex );
             /** destination kernel **/
             tail.emplace_back( td );
         }
@@ -106,14 +126,16 @@ pool_schedule::start()
 START:        
     std::chrono::milliseconds dura( 3 );
     std::this_thread::sleep_for( dura );
-    std::lock_guard< std::mutex > lock( tail_mutex );
+    tail_mutex.lock();
     for( auto * const td : tail )
     {
         if( ! td->finished  )
         {
+            tail_mutex.unlock();
             goto START;
         }
     }
+    tail_mutex.unlock();
     return;
 }
 

testsuite/CMakeLists.txt

@@ -43,7 +43,8 @@ set( TESTAPPS allocate portTypeException dynallocate stdlibsupport split
      staticJoinRetStruct
      staticLongSplitChainRetStruct
      staticSplitChainJoinRetStruct
-     staticSplitJoinRetStruct )
+     staticSplitJoinRetStruct 
+     chainMultiplePorts )
 
 if( BUILDRANDOM )
 list( APPEND TESTAPP gamma uniform gaussian exponential sequential )