Progress: began adding parallel merging and locking of levels

1 files changed, 19 insertions, 1 deletions

diff --git a/include/framework/scheduling/FIFOScheduler.h b/include/framework/scheduling/FIFOScheduler.h
index 4c1db8d..8a4cd8d 100644
--- a/include/framework/scheduling/FIFOScheduler.h
+++ b/include/framework/scheduling/FIFOScheduler.h
@@ -51,13 +51,20 @@ public:
 
     m_sched_thrd.join();
     m_sched_wakeup_thrd.join();
+    m_flush_thread.join();
   }
 
   void schedule_job(std::function<void(void *)> job, size_t size, void *args,
                     size_t type = 0) {
-    std::unique_lock<std::mutex> lk(m_cv_lock);
+
     size_t ts = m_counter.fetch_add(1);
 
+    if (type == 3) {
+      do_flush(Task(size, ts, job, args, type, &m_stats, &m_flush_lock));
+      return;
+    }
+
+    std::unique_lock<std::mutex> lk(m_cv_lock);
     m_stats.job_queued(ts, type, size);
     m_task_queue.push(Task(size, ts, job, args, type, &m_stats));
 
@@ -84,6 +91,9 @@ private:
   std::condition_variable m_cv;
   std::mutex m_queue_lock;
 
+  std::mutex m_flush_lock;
+  std::thread m_flush_thread;
+
   std::thread m_sched_thrd;
   std::thread m_sched_wakeup_thrd;
   ctpl::thread_pool m_thrd_pool;
@@ -121,6 +131,14 @@ private:
       }
     } while (!m_shutdown.load());
   }
+
+  void do_flush(Task task) {
+    m_flush_lock.lock();
+    if (m_flush_thread.joinable()) {
+      m_flush_thread.join();
+    }
+    m_flush_thread = std::thread(task, 0);
+  }
 };
 
 } // namespace de