prevent task starvation

SConstruct

@@ -1284,6 +1284,12 @@ if compiler == 'clang':
                     '-Wno-implicit-int-float-conversion',
                     '-Wno-enum-float-conversion',
                 ]})
+        if compiler_ver[:2] >= (11, 0):
+            for var in ['CXXFLAGS', 'CFLAGS']:
+                env.Append(**{var: [
+                    '-Wno-suggest-override',
+                    '-Wno-suggest-destructor-override',
+                ]})
 
     if platform == 'darwin':
         if compiler_ver[:2] >= (10, 0):

scripts/scons/config.py

@@ -18,7 +18,13 @@ def _run_prog(context, src, suffix):
     # RunProg may incorrectly use cached results from a previous run saved for
     # different file contents but the same invocation number. To prevent this, we
     # monkey patch its global counter with a hashsum of the file contents.
-    SCons.SConf._ac_build_counter = int(hashlib.md5(src.encode()).hexdigest(), 16)
+    # The workaround is needed only for older versions of SCons, where
+    # _ac_build_counter was an integer.
+    try:
+        if type(SCons.SConf._ac_build_counter) is int:
+            SCons.SConf._ac_build_counter = int(hashlib.md5(src.encode()).hexdigest(), 16)
+    except:
+        pass
     return context.RunProg(src, suffix)
 
 def CheckLibWithHeaderExt(context, libs, headers, language, expr='1', run=True):

src/modules/roc_ctl/task_queue.cpp

@@ -49,6 +49,7 @@ TaskQueue::ICompletionHandler::~ICompletionHandler() {
 TaskQueue::TaskQueue()
     : started_(false)
     , stop_(false)
+    , fetch_ready_(true)
     , ready_queue_size_(0) {
     started_ = Thread::start();
 }
@@ -159,10 +160,22 @@ bool TaskQueue::process_tasks_() {
     core::Mutex::Lock lock(task_mutex_);
 
     for (;;) {
-        Task* task = fetch_ready_task_();
-
-        if (!task) {
+        Task* task = NULL;
+
+        if (fetch_ready_) {
+            task = fetch_ready_task_();
+            if (!task) {
+                task = fetch_sleeping_task_();
+            } else {
+                fetch_ready_ = !fetch_ready_;
+            }
+        } else {
             task = fetch_sleeping_task_();
+            if (!task) {
+                task = fetch_ready_task_();
+            } else {
+                fetch_ready_ = !fetch_ready_;
+            }
         }
 
         if (!task) {

src/modules/roc_ctl/task_queue.h

@@ -282,6 +282,7 @@ class TaskQueue : private core::Thread {
 
     bool started_;
     core::Atomic<int> stop_;
+    bool fetch_ready_;
 
     core::Atomic<int> ready_queue_size_;
     core::MpscQueue<Task, core::NoOwnership> ready_queue_;

src/tests/roc_ctl/test_task_queue.cpp

@@ -747,9 +747,9 @@ TEST(task_queue, schedule_at_and_schedule) {
     const core::nanoseconds_t now = core::timestamp();
 
     tq.schedule(tasks[0], &handler);
-    tq.schedule_at(tasks[1], now + core::Millisecond * 3, &handler);
+    tq.schedule_at(tasks[1], now + core::Millisecond * 7, &handler);
     tq.schedule(tasks[2], &handler);
-    tq.schedule_at(tasks[3], now + core::Millisecond * 1, &handler);
+    tq.schedule_at(tasks[3], now + core::Millisecond * 5, &handler);
 
     tq.unblock_one();
     CHECK(handler.wait_called() == &tasks[0]);
@@ -1290,5 +1290,78 @@ TEST(task_queue, reschedule_cancelled) {
     UNSIGNED_LONGS_EQUAL(1, tq.num_tasks());
 }
 
+TEST(task_queue, no_starvation) {
+    TestTaskQueue tq;
+    CHECK(tq.valid());
+
+    enum { NumTasks = 6 };
+
+    UNSIGNED_LONGS_EQUAL(0, tq.num_tasks());
+
+    TestHandler handler;
+    handler.expect_success(true);
+    handler.expect_n_calls(NumTasks);
+
+    TestTaskQueue::Task tasks[NumTasks];
+
+    tq.block();
+
+    const core::nanoseconds_t now = core::timestamp();
+    const core::nanoseconds_t WaitTime = core::Millisecond;
+
+    tq.schedule_at(tasks[0], now + WaitTime, &handler);
+    tq.schedule_at(tasks[1], now + WaitTime * 2, &handler);
+    tq.schedule_at(tasks[2], now + WaitTime * 3, &handler);
+    tq.schedule(tasks[3], &handler);
+    tq.schedule(tasks[4], &handler);
+    tq.schedule(tasks[5], &handler);
+
+    for (size_t i = 0; i < NumTasks; i++) {
+        tq.set_nth_result(i, true);
+    }
+
+    // wait for sleeping task to sync
+    core::sleep_for(WaitTime * (NumTasks / 2));
+
+    // check that the tasks are fetched from alternating queues
+    tq.unblock_one();
+    TaskQueue::Task* temp = handler.wait_called();
+    CHECK(temp == &tasks[0] || temp == &tasks[3]);
+    UNSIGNED_LONGS_EQUAL(1, tq.num_tasks());
+    CHECK(tasks[0].success() || tasks[3].success());
+
+    tq.unblock_one();
+    temp = handler.wait_called();
+    CHECK(temp == &tasks[0] || temp == &tasks[3]);
+    UNSIGNED_LONGS_EQUAL(2, tq.num_tasks());
+    CHECK(tasks[0].success() && tasks[3].success());
+
+    tq.unblock_one();
+    temp = handler.wait_called();
+    CHECK(temp == &tasks[1] || temp == &tasks[4]);
+    UNSIGNED_LONGS_EQUAL(3, tq.num_tasks());
+    CHECK(tasks[1].success() || tasks[4].success());
+
+    tq.unblock_one();
+    temp = handler.wait_called();
+    CHECK(temp == &tasks[1] || temp == &tasks[4]);
+    UNSIGNED_LONGS_EQUAL(4, tq.num_tasks());
+    CHECK(tasks[1].success() && tasks[4].success());
+
+    tq.unblock_one();
+    temp = handler.wait_called();
+    CHECK(temp == &tasks[2] || temp == &tasks[5]);
+    UNSIGNED_LONGS_EQUAL(5, tq.num_tasks());
+    CHECK(tasks[2].success() || tasks[5].success());
+
+    tq.unblock_one();
+    temp = handler.wait_called();
+    CHECK(temp == &tasks[2] || temp == &tasks[5]);
+    UNSIGNED_LONGS_EQUAL(6, tq.num_tasks());
+    CHECK(tasks[2].success() && tasks[5].success());
+
+    tq.check_all_unblocked();
+}
+
 } // namespace ctl
 } // namespace roc