Improve worker.hpp

rt/include/typon/core/worker.hpp

@@ -82,6 +82,12 @@ namespace typon
       }
     }
 
+    void add(WorkDeque * deque) noexcept
+    {
+      std::lock_guard lock(_mutex);
+      _pool.push_back(deque);
+    }
+
     bool try_add(WorkDeque * deque) noexcept
     {
       if (!_mutex.try_lock())
@@ -89,19 +95,8 @@ namespace typon
         return false;
       }
       std::lock_guard lock(_mutex, std::adopt_lock);
-      unsafe_add(deque);
-      return true;
-    }
-
-    void add(WorkDeque * deque) noexcept
-    {
-      std::lock_guard lock(_mutex);
-      unsafe_add(deque);
-    }
-
-    void unsafe_add(WorkDeque * deque) noexcept
-    {
       _pool.push_back(deque);
+      return true;
     }
 
     Work try_steal() noexcept
@@ -111,28 +106,35 @@ namespace typon
         return {};
       }
       std::lock_guard lock(_mutex, std::adopt_lock);
-      return unsafe_steal();
-    }
-
-    Work unsafe_steal() noexcept
+      auto active = _active.load();
+      auto total = _pool.size() + bool(active);
+      if (total == 0)
       {
-      auto index = fdt::random::random64() % (_pool.size() + 1);
+        return {};
+      }
+      auto index = fdt::random::random64() % (_pool.size() + bool(active));
       if (index == _pool.size())
-      {
-        if (auto active = _active.load())
       {
         if (auto task = active->steal())
         {
           task->thefts()++;
           return *task;
         }
-        }
         return {};
       }
       auto deque = _pool[index];
       if (!deque->_resumable.load())
       {
-        if (auto task = deque->steal())
+        auto task = deque->steal();
+        if (task.get_flags() == fdt::lock_free::deque<Continuation>::Resize)
+        {
+          if (auto array = deque->shrink())
+          {
+            delete array;
+          }
+          delete deque;
+        }
+        if (task)
         {
           task->thefts()++;
           return *task;