Fixes in acquire phase and threadpool.h

Author: vishalgupta97

src/rt/sched/behaviourcore.h

@@ -454,10 +454,8 @@ namespace verona::rt
 
     /**
      * Remove `n` from the exec_count_down.
-     *
-     * Returns true if this call makes the count_down_zero
      */
-    bool resolve(size_t n = 1, bool fifo = true)
+    void resolve(size_t n = 1, bool fifo = true)
     {
       Logging::cout() << "Behaviour::resolve " << n << " for behaviour "
                       << *this << Logging::endl;
@@ -469,10 +467,7 @@ namespace verona::rt
       {
         Logging::cout() << "Scheduling Behaviour " << *this << Logging::endl;
         Scheduler::schedule(as_work(), fifo);
-        return true;
       }
-
-      return false;
     }
 
     // TODO: When C++ 20 move to span.
@@ -488,6 +483,32 @@ namespace verona::rt
       return pointer_offset<T>(slots, sizeof(Slot) * count);
     }
 
+    static void acquire_with_transfer(Cown* cown, size_t transfer, size_t required)
+    {
+      if (transfer == required)
+        return;
+
+      if (transfer > required)
+      {
+        Logging::cout() << "Releasing references as more transferred than "
+                           "required: transfer: "
+                        << transfer << " required: " << required << " on cown "
+                        << cown << Logging::endl;
+        // Release transfer - required times, we needed one as we woke up
+        // the cown, but the rest were not required.
+        for (int j = 0; j < transfer - required; j++)
+          Cown::release(ThreadAlloc::get(), cown);
+        return;
+      }
+
+      Logging::cout() << "Acquiring addition reference count: transfer: "
+                      << transfer << " required: " << required << " on cown "
+                      << cown << Logging::endl;
+      // We didn't have any RCs passed in, so we need to acquire one.
+      for (int j = 0; j < required - transfer; j++)
+        Cown::acquire(cown);
+    }
+
     /**
      * @brief Constructs a behaviour.  Leaves space for the closure.
      *
@@ -742,15 +763,15 @@ namespace verona::rt
         // Mark the slot as ready for scheduling
         curr_slot->reset_status();
         yield();
-
-        if (curr_slot->is_read_only())
-        {
+        if(curr_slot->is_read_only())
           curr_slot->set_behaviour(body);
-          yield();
-          auto prev_slot =
-            cown->last_slot.exchange(curr_slot, std::memory_order_acq_rel);
-          yield();
-          if (prev_slot == nullptr)
+        yield();
+        auto prev_slot =
+          cown->last_slot.exchange(curr_slot, std::memory_order_acq_rel);
+        
+        if(prev_slot == nullptr) 
+        {
+          if(curr_slot->is_read_only()) 
           {
             yield();
             bool first_reader = cown->read_ref_count.add_read();
@@ -760,136 +781,76 @@ namespace verona::rt
             curr_slot->set_active();
             ec[std::get<0>(indexes[first_chain_index])]++;
             yield();
-            if (transfer_count)
-            {
-              Logging::cout()
-                << "Releasing reader transferred count " << transfer_count
-                << " -1 for first in queue " << *curr_slot << Logging::endl;
-              // Release transfer_count - 1 times, we needed one as we woke up
-              // the cown, but the rest were not required.
-              for (int j = 0; j < transfer_count - 1; j++)
-                Cown::release(ThreadAlloc::get(), cown);
-            }
-            else
-            {
-              Logging::cout() << "Acquiring reader reference count for first "
-                                 "in queue on cown "
-                              << *curr_slot << Logging::endl;
-              // We didn't have any RCs passed in, so we need to acquire one.
-              Cown::acquire(cown);
-            }
-
-            if (first_reader)
-            {
-              Logging::cout()
-                << "Acquiring reference count for the first reader on cown "
-                << *curr_slot << Logging::endl;
-              Cown::acquire(cown);
-            }
-
+            acquire_with_transfer(cown, transfer_count, 1 + first_reader);
             continue;
           }
-          else
+
+          cown->next_writer.store(body, std::memory_order_release);
+          yield();
+          acquire_with_transfer(cown, transfer_count, 1);
+
+          if (
+            !cown->read_ref_count.any_reader() &&
+            cown->next_writer.exchange(nullptr, std::memory_order_acq_rel) ==
+              body)
           {
             yield();
-            Logging::cout() << " Someone in queue cown " << *curr_slot
-                            << " previous " << *prev_slot << Logging::endl;
+            Logging::cout() << " Writer at head of queue and got the cown "
+                            << *curr_slot << Logging::endl;
+            ec[std::get<0>(indexes[first_chain_index])]++;
+            yield();
+            continue;
+          }
+          Logging::cout() << " Writer waiting for previous readers cown "
+                          << *curr_slot << Logging::endl;
+          continue;
+        }
 
-            while (prev_slot->is_wait_2pl())
-            {
-              Systematic::yield_until(
-                [prev_slot]() { return !prev_slot->is_wait_2pl(); });
-              Aal::pause();
-            }
+        yield();
+        Logging::cout() << " Someone in queue cown " << *curr_slot
+                        << " previous " << *prev_slot << Logging::endl;
 
-            if (prev_slot->set_next_slot_reader(curr_slot))
-            {
-              Logging::cout()
-                << " Previous slot is a writer or blocked reader cown "
-                << *curr_slot << " previous " << *prev_slot << Logging::endl;
-              yield();
-              continue;
-            }
-            else
-            {
-              yield();
-              bool first_reader = cown->read_ref_count.add_read();
-              Logging::cout() << " Reader got the cown " << *curr_slot
-                              << " previous " << *prev_slot << Logging::endl;
-              yield();
-              curr_slot->set_active();
-              ec[std::get<0>(indexes[first_chain_index])]++;
-              if (first_reader)
-              {
-                Logging::cout()
-                  << "Acquiring reference count for first reader on cown "
-                  << *curr_slot << Logging::endl;
-                Cown::acquire(cown);
-              }
-              continue;
-            }
-          }
+        while (prev_slot->is_wait_2pl())
+        {
+          Systematic::yield_until(
+            [prev_slot]() { return !prev_slot->is_wait_2pl(); });
+          Aal::pause();
         }
-        else
+
+        if (curr_slot->is_read_only())
         {
-          auto prev_slot =
-            cown->last_slot.exchange(curr_slot, std::memory_order_acq_rel);
-          yield();
-          if (prev_slot == nullptr)
+          if (prev_slot->set_next_slot_reader(curr_slot))
           {
-            cown->next_writer.store(body, std::memory_order_release);
+            Logging::cout()
+              << " Previous slot is a writer or blocked reader cown "
+              << *curr_slot << " previous " << *prev_slot << Logging::endl;
             yield();
-            if (transfer_count)
-            {
-              Logging::cout()
-                << "Releasing writer transferred count " << transfer_count
-                << " -1 for first in queue cown " << *curr_slot
-                << Logging::endl;
-              // Release transfer_count - 1 times, we needed one as we woke up
-              // the cown, but the rest were not required.
-              for (int j = 0; j < transfer_count - 1; j++)
-                Cown::release(ThreadAlloc::get(), cown);
-            }
-            else
-            {
-              Logging::cout() << "Acquiring writer reference count on cown "
-                              << *curr_slot << Logging::endl;
-              // We didn't have any RCs passed in, so we need to acquire one.
-              Cown::acquire(cown);
-            }
-
-            if (
-              !cown->read_ref_count.any_reader() &&
-              cown->next_writer.exchange(nullptr, std::memory_order_acq_rel) ==
-                body)
-            {
-              yield();
-              Logging::cout() << " Writer at head of queue and got the cown "
-                              << *curr_slot << Logging::endl;
-              ec[std::get<0>(indexes[first_chain_index])]++;
-              yield();
-              continue;
-            }
-            Logging::cout() << " Writer waiting for previous readers cown "
-                            << *curr_slot << Logging::endl;
+            continue;
           }
-          else
+          
+          yield();
+          bool first_reader = cown->read_ref_count.add_read();
+          Logging::cout() << " Reader got the cown " << *curr_slot
+                          << " previous " << *prev_slot << Logging::endl;
+          yield();
+          curr_slot->set_active();
+          ec[std::get<0>(indexes[first_chain_index])]++;
+          if (first_reader)
           {
-            while (prev_slot->is_wait_2pl())
-            {
-              Systematic::yield_until(
-                [prev_slot]() { return !prev_slot->is_wait_2pl(); });
-              Aal::pause();
-            }
             Logging::cout()
-              << " Writer waiting for cown " << *curr_slot << Logging::endl;
-            prev_slot->set_next_slot_writer(body);
-            Logging::cout()
-              << " Writer Set next of previous slot cown " << *curr_slot
-              << " previous " << *prev_slot << Logging::endl;
+              << "Acquiring reference count for first reader on cown "
+              << *curr_slot << Logging::endl;
+            Cown::acquire(cown);
           }
+          continue;
         }
-
+        
+        Logging::cout()
+          << " Writer waiting for cown " << *curr_slot << Logging::endl;
+        prev_slot->set_next_slot_writer(body);
+        Logging::cout()
+          << " Writer Set next of previous slot cown " << *curr_slot
+          << " previous " << *prev_slot << Logging::endl;
         yield();
       }
 

src/rt/sched/threadpool.h

@@ -199,38 +199,7 @@ namespace verona::rt
       T::schedule_lifo(core, w);
     }
 
-    void init(size_t count)
-    {
-      Logging::cout() << "Init runtime" << Logging::endl;
-
-      if ((thread_count != 0) || (count == 0))
-        abort();
-
-      thread_count = count;
-      teardown_in_progress = false;
-
-      // Initialize the corepool.
-      core_pool.init(count);
-
-      // For future ids.
-      systematic_ids = count + 1;
-
-      for (; count > 0; count--)
-      {
-        T* t = new T;
-        t->systematic_id = count;
-#ifdef USE_SYSTEMATIC_TESTING
-        t->local_systematic =
-          Systematic::create_systematic_thread(t->systematic_id);
-#endif
-        threads.add_free(t);
-      }
-      Logging::cout() << "Runtime initialised" << Logging::endl;
-      init_barrier();
-    }
-
-    template<typename RunAtTermination>
-    void init(size_t count, RunAtTermination run_at_termination)
+    void init(size_t count, void (*run_at_termination)(void) = nullptr)
     {
       Logging::cout() << "Init runtime" << Logging::endl;
 
@@ -260,6 +229,7 @@ namespace verona::rt
       Logging::cout() << "Runtime initialised" << Logging::endl;
       init_barrier();
     }
+
     void run()
     {
       run_with_startup<>(&nop);