runtime: barriers take locks, and can fail to begin

Author: nickwanninger

runtime/core/Runtime.cpp

@@ -26,7 +26,8 @@ namespace alaska {
   static volatile bool runtime_initialized = false;
 
 
-  Runtime::Runtime(alaska::Configuration config) : handle_table(config) {
+  Runtime::Runtime(alaska::Configuration config)
+      : handle_table(config) {
     // Validate that there is not already a runtime (TODO: atomics?)
     ALASKA_ASSERT(g_runtime == nullptr, "Cannot create more than one runtime");
 
@@ -77,6 +78,19 @@ namespace alaska {
   }
 
 
+  void Runtime::lock_all_thread_caches(void) {
+    tcs_lock.lock();
+
+    for (auto *tc : tcs)
+      tc->lock.lock();
+  }
+  void Runtime::unlock_all_thread_caches(void) {
+    for (auto *tc : tcs)
+      tc->lock.unlock();
+    tcs_lock.unlock();
+  }
+
+
   void wait_for_initialization(void) {
     log_debug("waiting for initialization!\n");
     while (not is_initialized()) {
@@ -86,8 +100,6 @@ namespace alaska {
   }
 
 
-  bool is_initialized(void) {
-    return atomic_get(runtime_initialized);
-  }
+  bool is_initialized(void) { return atomic_get(runtime_initialized); }
 
 }  // namespace alaska

runtime/include/alaska/BarrierManager.hpp

@@ -20,7 +20,7 @@ namespace alaska {
   // pin any handles, the barrier does not need to do anything.
   struct BarrierManager : public alaska::InternalHeapAllocated {
     virtual ~BarrierManager() = default;
-    virtual void begin(void){};
+    virtual bool begin(void) { return true; };
     virtual void end(void){};
 
     unsigned long barrier_count = 0;

runtime/include/alaska/Runtime.hpp

@@ -76,16 +76,25 @@ namespace alaska {
 
     template <typename Fn>
     void with_barrier(Fn &&cb) {
-      barrier_manager->begin();
-      in_barrier = true;
-      barrier_manager->barrier_count++;
-      cb();
-      in_barrier = false;
-      barrier_manager->end();
+      lock_all_thread_caches();
+      if (barrier_manager->begin()) {
+        in_barrier = true;
+        barrier_manager->barrier_count++;
+        cb();
+        in_barrier = false;
+        barrier_manager->end();
+      } else {
+        alaska::printf("Barrier failed\n");
+      }
+      unlock_all_thread_caches();
     }
 
    private:
     int next_thread_cache_id = 0;
+
+
+    void lock_all_thread_caches(void);
+    void unlock_all_thread_caches(void);
   };
 
 

runtime/include/alaska/ThreadCache.hpp

@@ -16,6 +16,7 @@
 #include <alaska/HandleTable.hpp>
 #include <alaska/LocalityPage.hpp>
 #include <alaska/alaska.hpp>
+#include "ck/lock.h"
 
 namespace alaska {
 
@@ -41,6 +42,13 @@ namespace alaska {
     bool localize(alaska::Mapping &m, uint64_t epoch);
     bool localize(void *handle, uint64_t epoch);
 
+
+   protected:
+    friend class LockedThreadCache;
+    friend alaska::Runtime;
+
+    ck::mutex lock;
+
    private:
     // Allocate backing data for a handle, but don't assign it yet.
     void *allocate_backing_data(const alaska::Mapping &m, size_t size);
@@ -79,4 +87,36 @@ namespace alaska {
 
 
 
+  class LockedThreadCache final {
+   public:
+    LockedThreadCache(ThreadCache &tc)
+        : tc(tc) {
+      tc.lock.lock();
+    }
+
+
+    ~LockedThreadCache(void) {
+      tc.lock.unlock();
+    }
+
+    // Delete copy constructor and copy assignment operator
+    LockedThreadCache(const LockedThreadCache &) = delete;
+    LockedThreadCache &operator=(const LockedThreadCache &) = delete;
+
+    // Delete move constructor and move assignment operator
+    LockedThreadCache(LockedThreadCache &&) = delete;
+    LockedThreadCache &operator=(LockedThreadCache &&) = delete;
+
+
+
+
+    ThreadCache &operator*(void) { return tc; }
+    ThreadCache *operator->(void) { return &tc; }
+
+   private:
+    ThreadCache &tc;
+  };
+
+
+
 }  // namespace alaska

runtime/include/alaska/rt/barrier.hpp

@@ -44,7 +44,7 @@ namespace alaska {
 
     // Barrier operational lifetime. It is not recommended to use this interface, and
     // instead use `with_barrier` interface below:
-    void begin();
+    bool begin();
     void end();
 
     // struct BarrierInfo {};

runtime/rt/barrier.cpp

@@ -328,7 +328,7 @@ void dump_thread_states(void) {
 
 
 
-void alaska::barrier::begin(void) {
+bool alaska::barrier::begin(void) {
   // Pseudocode:
   //
   // function begin():
@@ -369,9 +369,15 @@ void alaska::barrier::begin(void) {
   int retries = 0;
   int signals_sent = 0;
 
-  // printf("\n");
+  bool success = true;
+
   // Make sure the threads that are in unmanaged (library) code get signalled.
   while (true) {
+    if (retries >= 1000) {
+      success = false;
+      break;
+    }
+    retries++;
     bool sent_signal = false;
     list_for_each_entry(pos, &all_threads, list_head) {
       if (pos->state->join_status == ALASKA_JOIN_REASON_NOT_JOINED) {
@@ -391,6 +397,9 @@ void alaska::barrier::begin(void) {
 
   (void)retries;
   (void)signals_sent;
+
+
+  return success;
   // printf("%10f ", (end - start) / 1000.0 / 1000.0 / 1000.0);
   // dump_thread_states();
   // printf(" retries = %d, signals = %d\n", retries, signals_sent);
@@ -437,14 +446,14 @@ static void alaska_barrier_signal_handler(int sig, siginfo_t* info, void* ptr) {
       // we need to return back to the thread so it can hit a poll.
 
 
-      // // First, though, we need to wait for the patches to be done.
+      // First, though, we need to wait for the patches to be done.
       // while (!patches_done) {
       // }
-      //
-      // for (auto [start, end] : managed_blob_text_regions) {
-      //   __builtin___clear_cache((char*)start, (char*)end);
-      // }
-      // printf("ManagedUntracked!\n");
+
+      for (auto [start, end] : managed_blob_text_regions) {
+        __builtin___clear_cache((char*)start, (char*)end);
+      }
+      printf("ManagedUntracked!\n");
       // printf("EEP %p %d!\n", return_address, sig);
       return;
 

runtime/rt/halloc.cpp

@@ -19,11 +19,11 @@
 // TODO: don't have this be global!
 static __thread alaska::ThreadCache *g_tc = nullptr;
 
-alaska::ThreadCache *get_tc(void) {
+alaska::LockedThreadCache get_tc(void) {
   if (unlikely(g_tc == nullptr)) {
     g_tc = alaska::Runtime::get().new_threadcache();
   }
-  return g_tc;
+  return *g_tc;
 }
 
 
@@ -44,11 +44,10 @@ void *hcalloc(size_t nmemb, size_t size) { return _halloc(nmemb * size, 1); }
 
 // Reallocate a handle
 void *hrealloc(void *handle, size_t new_size) {
-  auto *tc = get_tc();
   // If the handle is null, then this call is equivalent to malloc(size)
-  if (handle == NULL) {
-    return halloc(new_size);
-  }
+  if (handle == NULL) return halloc(new_size);
+
+
   auto *m = alaska::Mapping::from_handle_safe(handle);
   if (m == NULL) {
     if (!alaska::Runtime::get().heap.huge_allocator.owns(handle)) {
@@ -65,7 +64,7 @@ void *hrealloc(void *handle, size_t new_size) {
     return NULL;
   }
 
-  handle = tc->hrealloc(handle, new_size);
+  handle = get_tc()->hrealloc(handle, new_size);
   return handle;
 }
 

runtime/rt/init.cpp

@@ -28,7 +28,7 @@ static alaska::Runtime *the_runtime = nullptr;
 
 struct CompilerRuntimeBarrierManager : public alaska::BarrierManager {
   ~CompilerRuntimeBarrierManager() override = default;
-  void begin(void) override { alaska::barrier::begin(); }
+  bool begin(void) override { return alaska::barrier::begin(); }
   void end(void) override { alaska::barrier::end(); }
 };
 
@@ -44,8 +44,9 @@ static void *barrier_thread_func(void *) {
     usleep(50 * 1000);
 
     alaska::Runtime::get().with_barrier([]() {
-      long swapped = alaska::Runtime::get().heap.jumble();
-      printf("Swapped %ld\n", swapped);
+      alaska::Runtime::get().heap.compact_sizedpages();
+      // long swapped = alaska::Runtime::get().heap.jumble();
+      // printf("Swapped %ld\n", swapped);
     });
     // alaska::barrier::begin();
     // printf("Barrier.\n");

runtime/rt/sim.cpp

@@ -19,7 +19,7 @@
 #define L2_SETS 32
 #define TOTAL_ENTRIES (L1_WAYS * L1_SETS + L2_WAYS * L2_SETS)
 
-#define RATE 10'000
+#define RATE 1'000'000
 
 static long access_count = 0;
 static alaska::sim::HTLB *g_htlb = NULL;
@@ -56,9 +56,10 @@ static void *sim_background_thread_func(void *) {
     auto &rt = alaska::Runtime::get();
 
     rt.with_barrier([&]() {
-      unsigned long c = rt.heap.jumble();
-      printf("jumbled %lu objects\n", c);
-      // rt.heap.compact_sizedpages();
+      // unsigned long c = rt.heap.jumble();
+      // printf("jumbled %lu objects\n", c);
+      unsigned long c = rt.heap.compact_sizedpages();
+      printf("compacted %lu\n", c);
       return;
 
 

runtime/yukon/yukon.cpp

@@ -35,8 +35,8 @@
 static alaska::ThreadCache *tc = NULL;
 
 static void set_ht_addr(void *addr) {
-  // printf("set htaddr to %p\n", addr);
-  // write_csr(0xc2, addr);
+  // alaska::printf("set htaddr to %p\n", addr);
+  write_csr(0xc2, addr);
 }
 
 static alaska::Runtime *the_runtime = NULL;
@@ -135,7 +135,7 @@ void __attribute__((destructor)) alaska_deinit(void) {
 // #define free hfree
 
 static void *_halloc(size_t sz, int zero) {
-  print_hex("_halloc", sz);
+  // print_hex("_halloc", sz);
   if (dead) {
     return alaska_internal_malloc(sz);
   }