Merge pull request #291 from plasma-umass/interpose-semaphores

Interpose semaphores, and don't let delay accounting underflow duration

Author: emeryberger

.github/scripts/check_semaphore_speedup.py

@@ -0,0 +1,80 @@
+#!/usr/bin/env python3
+"""Assert that coz measures a real speedup for benchmarks/sem_toy.
+
+sem_toy's main thread joins its workers on a semaphore. A thread blocked on a
+semaphore is not running, so it must not be charged for the virtual delays
+inserted while it slept. When libcoz does not interpose the semaphore, the main
+thread -- the one that visits the progress point -- pays them all on wake-up,
+the measured period grows with the speedup, and the slope collapses to zero or
+goes negative.
+
+Both of sem_toy's loops inline the same xorshift, so the hot line's slope should
+approach 1.0: removing that work removes the program.
+"""
+import json
+import sys
+
+
+def main() -> int:
+    if len(sys.argv) != 2:
+        print(f"usage: {sys.argv[0]} <profile.jsonl>")
+        return 2
+
+    rows, current = [], None
+    with open(sys.argv[1]) as handle:
+        for line in handle:
+            line = line.strip()
+            if not line:
+                continue
+            record = json.loads(line)
+            if record["type"] == "experiment":
+                current = record
+            elif record["type"] == "throughput-point" and current is not None:
+                current["delta"] = record["delta"]
+                rows.append(current)
+                current = None
+
+    if not rows:
+        print("ERROR: no experiments in profile")
+        return 1
+
+    # A duration past 2^63 means the inserted delay exceeded the experiment's
+    # wall time and the unsigned subtraction wrapped.
+    underflowed = [r for r in rows if r["duration"] > 10**12]
+    if underflowed:
+        print(f"ERROR: {len(underflowed)} experiment(s) with underflowed duration")
+        return 1
+
+    periods = {}
+    for row in rows:
+        if row["delta"]:
+            periods.setdefault(round(row["speedup"], 2), []).append(row["duration"] / row["delta"])
+
+    baseline = periods.get(0.0)
+    if not baseline:
+        print("WARNING: no 0% baseline experiments; skipping slope check")
+        return 0
+
+    best = max(periods)
+    if best < 0.5:
+        print(f"WARNING: largest speedup sampled was {best:.0%}; skipping slope check")
+        return 0
+
+    p0 = sum(baseline) / len(baseline)
+    ps = sum(periods[best]) / len(periods[best])
+    improvement = 1 - ps / p0
+
+    print(f"baseline period {p0:.0f}ns; period at {best:.0%} speedup {ps:.0f}ns")
+    print(f"measured improvement: {improvement:.1%}")
+
+    if improvement < 0.25:
+        print("ERROR: virtual speedup produced no throughput gain -- is the "
+              "blocked main thread being charged for delays it never paid?")
+        return 1
+
+    print("Semaphore interposition validated")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

.github/workflows/ci.yml

@@ -42,14 +42,15 @@ jobs:
       run: |
         cd build
         cmake .. -DBUILD_BENCHMARKS=ON -DCMAKE_BUILD_TYPE=RelWithDebInfo
-        make -j$(nproc) toy lock_test kmeans
+        make -j$(nproc) toy lock_test kmeans sem_toy
 
     - name: Verify build artifacts
       run: |
         test -f build/libcoz/libcoz.so
         test -x build/benchmarks/toy/toy
         test -x build/benchmarks/lock_test/lock_test
         test -x build/benchmarks/kmeans/kmeans
+        test -x build/benchmarks/sem_toy/sem_toy
         echo "Build artifacts verified"
 
     - name: Run unit tests
@@ -232,6 +233,20 @@ jobs:
         print("  - lock_test.cpp:12 (critical section) identified as profiling target")
         VALIDATION_SCRIPT
 
+
+    - name: Run sem_toy (semaphore join) with coz
+      run: |
+        cd build
+        TIMEOUT=timeout
+        command -v gtimeout >/dev/null && TIMEOUT=gtimeout
+        $TIMEOUT 120 ../coz run -o sem_toy_profile.jsonl --- ./benchmarks/sem_toy/sem_toy || true
+        test -s sem_toy_profile.jsonl
+
+    - name: Validate semaphore interposition
+      run: |
+        cd build
+        python3 ../.github/scripts/check_semaphore_speedup.py sem_toy_profile.jsonl
+
     - name: Upload profile artifacts
       uses: actions/upload-artifact@v4
       with:
@@ -265,7 +280,7 @@ jobs:
       run: |
         cd build
         cmake .. -DBUILD_BENCHMARKS=ON -DCMAKE_BUILD_TYPE=RelWithDebInfo
-        make -j$(sysctl -n hw.ncpu) toy lock_test kmeans
+        make -j$(sysctl -n hw.ncpu) toy lock_test kmeans sem_toy
 
     - name: Verify build artifacts
       run: |
@@ -275,6 +290,7 @@ jobs:
         test -x build/benchmarks/toy/toy
         test -x build/benchmarks/lock_test/lock_test
         test -x build/benchmarks/kmeans/kmeans
+        test -x build/benchmarks/sem_toy/sem_toy
         echo "Build artifacts verified"
 
     - name: Run unit tests
@@ -296,6 +312,17 @@ jobs:
         echo "Profile created:"
         cat profile.jsonl
 
+    - name: Run sem_toy (semaphore join) with coz
+      run: |
+        cd build
+        gtimeout 120 ../coz run -o sem_toy_profile.jsonl --- ./benchmarks/sem_toy/sem_toy || true
+        test -s sem_toy_profile.jsonl
+
+    - name: Validate semaphore interposition
+      run: |
+        cd build
+        python3 ../.github/scripts/check_semaphore_speedup.py sem_toy_profile.jsonl
+
   lang-rust:
     name: Rust bindings (${{ matrix.os }})
     runs-on: ${{ matrix.os }}

benchmarks/sem_toy/CMakeLists.txt

@@ -0,0 +1,4 @@
+add_executable(sem_toy sem_toy.cpp)
+target_link_libraries(sem_toy PRIVATE pthread coz-instrumentation)
+
+add_coz_run_target(run_sem_toy COMMAND $<TARGET_FILE:sem_toy>)

benchmarks/sem_toy/sem_toy.cpp

@@ -0,0 +1,78 @@
+/*
+ * Copyright (c) 2015, Charlie Curtsinger and Emery Berger,
+ *                     University of Massachusetts Amherst
+ * This file is part of the Coz project. See LICENSE.md file at the top-level
+ * directory of this distribution and at http://github.com/plasma-umass/coz.
+ */
+
+// Like benchmarks/toy, but the main thread joins its workers through a
+// semaphore rather than pthread_join.
+//
+// This is the regression test for semaphore interposition. A thread blocked on
+// a semaphore is not running, so it must not be charged for virtual delays
+// inserted while it slept. Before libcoz wrapped sem_wait/semaphore_wait, the
+// main thread -- the one that visits the progress point -- paid all of them on
+// wake-up, and the profile came out with a slope near zero or negative
+// (measured: +0.13 with R^2 0.01, and -0.57 with R^2 0.18) instead of the ~1.0
+// this program should show.
+//
+// Both loops inline the same xorshift, so the expected result is a single hot
+// line with a slope near 1.0: removing that work removes the program.
+#include <coz.h>
+#include <pthread.h>
+#include <stdio.h>
+#include <stdint.h>
+
+#ifdef __APPLE__
+#include <dispatch/dispatch.h>
+static dispatch_semaphore_t done;
+static void sem_setup() { done = dispatch_semaphore_create(0); }
+static void sem_wait_one() { dispatch_semaphore_wait(done, DISPATCH_TIME_FOREVER); }
+static void sem_signal() { dispatch_semaphore_signal(done); }
+#else
+#include <semaphore.h>
+static sem_t done;
+static void sem_setup() { sem_init(&done, 0, 0); }
+static void sem_wait_one() { sem_wait(&done); }
+static void sem_signal() { sem_post(&done); }
+#endif
+
+static const uint64_t kIterations = 40000000ULL;
+static volatile uint64_t slow_sink, fast_sink;
+
+static uint64_t xorshift(uint64_t v) {
+  v ^= v << 13; v ^= v >> 7; v ^= v << 17; return v;
+}
+
+static void* slow_work(void*) {
+  uint64_t acc = 0x9E3779B97F4A7C15ULL;
+  for (uint64_t i = 0; i < kIterations; i++) acc = xorshift(acc);
+  slow_sink = acc;
+  sem_signal();
+  return nullptr;
+}
+
+static void* fast_work(void*) {
+  uint64_t acc = 0x9E3779B97F4A7C15ULL;
+  for (uint64_t i = 0; i < kIterations / 2; i++) acc = xorshift(acc);
+  fast_sink = acc;
+  sem_signal();
+  return nullptr;
+}
+
+int main() {
+  sem_setup();
+  printf("Starting.\n");
+  for (int round = 0; round < 100; round++) {
+    pthread_t a, b;
+    pthread_create(&a, nullptr, slow_work, nullptr);
+    pthread_create(&b, nullptr, fast_work, nullptr);
+    sem_wait_one();   // main blocks here -- invisible to coz without the wrappers
+    sem_wait_one();
+    pthread_detach(a);
+    pthread_detach(b);
+    COZ_PROGRESS;
+    printf("."); fflush(stdout);
+  }
+  printf("\nDone. %llu %llu\n", (unsigned long long)slow_sink, (unsigned long long)fast_sink);
+}

libcoz/libcoz.cpp

@@ -452,6 +452,46 @@ extern "C" {
     return real::pthread_mutex_unlock(mutex);
   }
 
+  /**
+   * POSIX semaphores.
+   *
+   * A thread blocked on a semaphore is not running, so it must not be charged
+   * for virtual delays inserted while it slept -- otherwise it pays them all at
+   * once on wake-up, and if it is the thread that visits the progress point,
+   * every line in the profile acquires a negative slope.
+   *
+   * glibc implements sem_wait on futex(2) with an inline syscall, so the futex
+   * cannot be interposed. sem_wait can: it is an ordinary exported libc symbol.
+   * That is the seam, and it covers everything layered on POSIX semaphores,
+   * including swift-corelibs-libdispatch's DispatchSemaphore.
+   */
+  int sem_wait(sem_t* sem) {
+    if(initialized) profiler::get_instance().pre_block();
+    int result = real::sem_wait(sem);
+    // Woken by a sem_post from another thread, so skip the delays.
+    if(initialized) profiler::get_instance().post_block(true);
+    return result;
+  }
+
+  int sem_timedwait(sem_t* sem, const struct timespec* abstime) {
+    if(initialized) profiler::get_instance().pre_block();
+    int result = real::sem_timedwait(sem, abstime);
+    // On timeout nobody handed us the semaphore, so we own our delays.
+    if(initialized) profiler::get_instance().post_block(result == 0);
+    return result;
+  }
+
+  /// Never blocks, so there is nothing to skip.
+  int sem_trywait(sem_t* sem) {
+    return real::sem_trywait(sem);
+  }
+
+  /// May unblock another thread, so pay outstanding delays before it runs.
+  int sem_post(sem_t* sem) {
+    if(initialized) profiler::get_instance().catch_up();
+    return real::sem_post(sem);
+  }
+
   /**
    * Enforce a floor on the alternate signal stack.
    *