WIP

Author: galabovaa

CMakeLists.txt

@@ -556,9 +556,9 @@ elseif (DEBUG_MEMORY STREQUAL "Thread")
   set (CMAKE_EXE_LINKER_FLAGS_RELWITHDEBINFO "${CMAKE_EXE_LINKER_FLAGS_RELWITHDEBINFO} -fsanitize=thread,undefined \
       -fno-omit-frame-pointer")
 
-  set (CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -fsanitize=thread,undefined \
+  set (CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -fsanitize=thread \
       -fno-omit-frame-pointer ")
-  set (CMAKE_EXE_LINKER_FLAGS_DEBUG "${CMAKE_EXE_LINKER_FLAGS_DEBUG} -fsanitize=thread,undefined \
+  set (CMAKE_EXE_LINKER_FLAGS_DEBUG "${CMAKE_EXE_LINKER_FLAGS_DEBUG} -fsanitize=thread \
       -fno-omit-frame-pointer ")
 
 elseif (DEBUG_MEMORY STREQUAL "Leak")

check/TestHighsParallel.cpp

@@ -218,26 +218,26 @@ TEST_CASE("MatrixMultHighs", "[parallel]") {
   matrix_multiplication("highs", parallel::num_threads(), 1);
 }
 
-TEST_CASE("FibonacciTasksHighs", "[parallel]") {
-  HighsTaskExecutor::shutdown();
-  auto beg = std::chrono::high_resolution_clock::now();
-  parallel::initialize_scheduler(numThreads);
-  int64_t result = fib(41);
-  auto end = std::chrono::high_resolution_clock::now();
-
-  if (dev_run)
-    std::cout << "time elapsed for fib(41) with HiGHS work stealing: "
-              << (std::chrono::duration_cast<std::chrono::microseconds>(end -
-                                                                        beg)
-                      .count() /
-                  1e3)
-              << "ms" << std::endl;
-
-  // REQUIRE(result == 4807526976);
-  // REQUIRE(result == 20365011074);
-  // fib 41
-  REQUIRE(result == 267914296);
-}
+// TEST_CASE("FibonacciTasksHighs", "[parallel]") {
+//   HighsTaskExecutor::shutdown();
+//   auto beg = std::chrono::high_resolution_clock::now();
+//   parallel::initialize_scheduler(numThreads);
+//   int64_t result = fib(41);
+//   auto end = std::chrono::high_resolution_clock::now();
+
+//   if (dev_run)
+//     std::cout << "time elapsed for fib(41) with HiGHS work stealing: "
+//               << (std::chrono::duration_cast<std::chrono::microseconds>(end -
+//                                                                         beg)
+//                       .count() /
+//                   1e3)
+//               << "ms" << std::endl;
+
+//   // REQUIRE(result == 4807526976);
+//   // REQUIRE(result == 20365011074);
+//   // fib 41
+//   REQUIRE(result == 267914296);
+// }
 
 #if 0
 TEST_CASE("MatrixMultOmp", "[parallel]") {

highs/parallel/HighsSplitDeque.h

@@ -29,6 +29,28 @@
 #include "util/HighsInt.h"
 #include "util/HighsRandom.h"
 
+#ifdef __has_feature
+#if __has_feature(thread_sanitizer)
+#define TSAN_ENABLED
+#endif
+#endif
+
+#ifdef __SANITIZE_THREAD__
+#define TSAN_ENABLED
+#endif
+
+#ifdef TSAN_ENABLED
+#define TSAN_ANNOTATE_HAPPENS_BEFORE(addr) \
+    AnnotateHappensBefore(__FILE__, __LINE__, (void*)(addr))
+#define TSAN_ANNOTATE_HAPPENS_AFTER(addr) \
+    AnnotateHappensAfter(__FILE__, __LINE__, (void*)(addr))
+extern "C" void AnnotateHappensBefore(const char* f, int l, void* addr);
+extern "C" void AnnotateHappensAfter(const char* f, int l, void* addr);
+#else
+#define TSAN_ANNOTATE_HAPPENS_BEFORE(addr)
+#define TSAN_ANNOTATE_HAPPENS_AFTER(addr)
+#endif
+
 class HighsSplitDeque {
   using cache_aligned = highs::cache_aligned;
 
@@ -134,6 +156,9 @@ class HighsSplitDeque {
 
     void publishWork(HighsSplitDeque* localDeque) {
       HighsSplitDeque* sleeper = popSleeper(localDeque);
+
+      // TSAN_ANNOTATE_HAPPENS_BEFORE(sleeper);
+
       while (sleeper) {
         uint32_t t = localDeque->selfStealAndGetTail();
         if (t == localDeque->ownerData.splitCopy) {
@@ -143,6 +168,7 @@ class HighsSplitDeque {
                                                     std::memory_order_relaxed);
             haveJobs.fetch_add(-1, std::memory_order_release);
           }
+          // TSAN_ANNOTATE_HAPPENS_AFTER(sleeper);
           pushSleeper(sleeper);
           return;
         } else {
@@ -159,17 +185,23 @@ class HighsSplitDeque {
           return;
         }
 
+        // TSAN_ANNOTATE_HAPPENS_AFTER(sleeper);
         sleeper = popSleeper(localDeque);
       }
     }
 
     HighsTask* waitForNewTask(HighsSplitDeque* localDeque) {
       pushSleeper(localDeque);
       localDeque->stealerData.semaphore.acquire();
+      // std::cout << " return injected task read " << &localDeque->stealerData.injectedTask << std::endl;
       return localDeque->stealerData.injectedTask;
     }
   };
 
+  // const HighsTask * getInjectedTask() {
+  //   return stealerData.injectedTask;
+  // }
+
  private:
   static_assert(sizeof(OwnerData) <= 64,
                 "sizeof(OwnerData) exceeds cache line size");
@@ -439,6 +471,13 @@ class HighsSplitDeque {
 
   void injectTaskAndNotify(HighsTask* t) {
     stealerData.injectedTask = t;
+
+    // std::cout << "     injected task write " << &stealerData.injectedTask << std::endl;
+
+
+    // TSAN_ANNOTATE_HAPPENS_BEFORE(&stealerData.injectedTask);
+    // TSAN_ANNOTATE_HAPPENS_AFTER(&stealerData);
+
     stealerData.semaphore.release();
   }
 

highs/parallel/HighsTaskExecutor.h

@@ -28,27 +28,6 @@
 #include "util/HighsInt.h"
 #include "util/HighsRandom.h"
 
-#ifdef __has_feature
-#if __has_feature(thread_sanitizer)
-#define TSAN_ENABLED
-#endif
-#endif
-
-#ifdef __SANITIZE_THREAD__
-#define TSAN_ENABLED
-#endif
-
-#ifdef TSAN_ENABLED
-#define TSAN_ANNOTATE_HAPPENS_BEFORE(addr) \
-    AnnotateHappensBefore(__FILE__, __LINE__, (void*)(addr))
-#define TSAN_ANNOTATE_HAPPENS_AFTER(addr) \
-    AnnotateHappensAfter(__FILE__, __LINE__, (void*)(addr))
-extern "C" void AnnotateHappensBefore(const char* f, int l, void* addr);
-extern "C" void AnnotateHappensAfter(const char* f, int l, void* addr);
-#else
-#define TSAN_ANNOTATE_HAPPENS_BEFORE(addr)
-#define TSAN_ANNOTATE_HAPPENS_AFTER(addr)
-#endif
 
 class HighsTaskExecutor {
  public:
@@ -127,8 +106,16 @@ class HighsTaskExecutor {
       threadLocalWorkerDeque() = localDeque;
 
       HighsTask* currentTask = ptr->workerBunk->waitForNewTask(localDeque);
+
+      // auto ptr = localDeque->getInjectedTask();
+      // TSAN_ANNOTATE_HAPPENS_BEFORE(localDeque->getInjectedTask());
+
       while (currentTask != nullptr) {
         localDeque->runStolenTask(currentTask);
+        // TSAN_ANNOTATE_HAPPENS_BEFORE(currentTask);
+
+        // TSAN_ANNOTATE_HAPPENS_AFTER(currentTask);
+
 
         currentTask = ptr->random_steal_loop(localDeque);
         if (currentTask != nullptr) continue;
@@ -185,6 +172,7 @@ std::cout << "TSAN is OFF" << std::endl;
     // now inject the null task as termination signal to every worker
     for (auto& workerDeque : workerDeques) {
       workerDeque->injectTaskAndNotify(nullptr);
+      // TSAN_ANNOTATE_HAPPENS_AFTER(workerDeque->getInjectedTask());
     }
 
     // only block if called on main thread, otherwise deadlock may occur
@@ -193,6 +181,7 @@ std::cout << "TSAN is OFF" << std::endl;
         // TSAN_ANNOTATE_HAPPENS_AFTER(&workerThread);
         // std::cout <<"happens after thread blocking and isMain" << ": " << &workerThread << std::endl;
         workerThread.join();
+        // std::cout <<"happens after join" << std::endl;
       }
     } else {
       for (auto& workerThread : workerThreads) {