Merge pull request #271 from plasma-umass/fix-mysql-profiling-issues

Fix MySQL profiling issues: double delays, stale counters, and custom sync API

Author: emeryberger

coz

@@ -372,13 +372,36 @@ def calculate_speedups(data, min_points=1, min_delta=5):
       if len(measurements) >= min_points:
         # Calculate max speedup
         max_speedup = max(m[1] for m in measurements) if measurements else 0
+
+        # Calculate linear regression slope and R-squared
+        slope = None
+        r_squared = None
+        if len(measurements) >= 2:
+          n = len(measurements)
+          sum_x = sum(m[0] for m in measurements)
+          sum_y = sum(m[1] for m in measurements)
+          sum_xy = sum(m[0] * m[1] for m in measurements)
+          sum_x2 = sum(m[0] ** 2 for m in measurements)
+          sum_y2 = sum(m[1] ** 2 for m in measurements)
+          denom = n * sum_x2 - sum_x ** 2
+          if denom != 0:
+            slope = (n * sum_xy - sum_x * sum_y) / denom
+            # R-squared: coefficient of determination
+            ss_tot = sum_y2 - (sum_y ** 2) / n
+            intercept = (sum_y - slope * sum_x) / n
+            ss_res = sum((y - (intercept + slope * x)) ** 2 for x, y in measurements)
+            if ss_tot > 0:
+              r_squared = 1.0 - ss_res / ss_tot
+
         results.append({
           'line': selected,
           'progress_point': pp_name,
           'measurements': measurements,
           'max_speedup': max_speedup,
           'num_points': len(measurements),
-          'baseline_speedup': baseline_speedup
+          'baseline_speedup': baseline_speedup,
+          'slope': slope,
+          'r_squared': r_squared
         })
 
   # Sort by max speedup (highest first)
@@ -402,15 +425,17 @@ def print_text_summary(profile_path, results, experiment_count, runtime, samples
   max_line_len = max(max_line_len, 11)  # "Source Line" header
 
   # Print header
-  header = f"{'Source Line':<{max_line_len}} | Max Speedup | Points"
+  header = f"{'Source Line':<{max_line_len}} | {'Slope':>7} | {'R²':>5} | Max Speedup | Points"
   print(header)
-  print('-' * max_line_len + '-+-------------+-------')
+  print('-' * max_line_len + '-+---------+-------+-------------+-------')
 
   # Print each result
   for r in results:
     speedup_pct = r['max_speedup'] * 100
     sign = '+' if speedup_pct >= 0 else ''
-    print(f"{r['line']:<{max_line_len}} | {sign}{speedup_pct:>9.1f}% | {r['num_points']:>5}")
+    slope_str = f"{r['slope']:>7.3f}" if r.get('slope') is not None else '    N/A'
+    r2_str = f"{r['r_squared']:>5.2f}" if r.get('r_squared') is not None else '  N/A'
+    print(f"{r['line']:<{max_line_len}} | {slope_str} | {r2_str} | {sign}{speedup_pct:>9.1f}% | {r['num_points']:>5}")
 
 def print_scatter_plot(result):
   """Print an ASCII scatter plot for a single result."""

include/coz.h

@@ -43,6 +43,12 @@ typedef coz_counter_t* (*coz_get_counter_t)(int, const char*);
 // The type of the _coz_add_delays function
 typedef void (*coz_add_delays_t)(void);
 
+// The type of the _coz_pre_block function
+typedef void (*coz_pre_block_t)(void);
+
+// The type of the _coz_post_block function
+typedef void (*coz_post_block_t)(int);
+
 // Locate and invoke _coz_get_counter
 static coz_counter_t* _call_coz_get_counter(int type, const char* name) {
   static unsigned char _initialized = 0;
@@ -88,9 +94,50 @@ static void _call_coz_add_delays(void) {
   if(fn) fn();
 }
 
+// Locate and invoke _coz_pre_block
+static void _call_coz_pre_block(void) {
+  static unsigned char _initialized = 0;
+  static coz_pre_block_t fn;
+
+  if(!_initialized) {
+    if(dlsym) {
+      void* p = dlsym(RTLD_DEFAULT, "_coz_pre_block");
+      memcpy(&fn, &p, sizeof(p));
+    }
+    _initialized = 1;
+  }
+
+  if(fn) fn();
+}
+
+// Locate and invoke _coz_post_block
+static void _call_coz_post_block(int skip_delays) {
+  static unsigned char _initialized = 0;
+  static coz_post_block_t fn;
+
+  if(!_initialized) {
+    if(dlsym) {
+      void* p = dlsym(RTLD_DEFAULT, "_coz_post_block");
+      memcpy(&fn, &p, sizeof(p));
+    }
+    _initialized = 1;
+  }
+
+  if(fn) fn(skip_delays);
+}
+
+// On macOS, per-thread timers are not available so worker threads must check
+// their delay debt at progress points.  On Linux, delays are already applied
+// in the SIGPROF handler via process_samples() -> add_delays(), so calling
+// add_delays() again at every progress-point hit causes double application
+// and TPS collapse under high concurrency.
+#ifdef __APPLE__
+#  define _COZ_CHECK_DELAYS _call_coz_add_delays()
+#else
+#  define _COZ_CHECK_DELAYS ((void)0)
+#endif
+
 // Macro to initialize and increment a counter, then check for pending delays.
-// The delay check is critical on macOS where per-thread timers are not available,
-// ensuring worker threads apply delays at progress points.
 #define COZ_INCREMENT_COUNTER(type, name) \
   if(1) { \
     static unsigned char _initialized = 0; \
@@ -102,7 +149,7 @@ static void _call_coz_add_delays(void) {
     } \
     if(_counter) { \
       __atomic_add_fetch(&_counter->count, 1, __ATOMIC_RELAXED); \
-      _call_coz_add_delays(); \
+      _COZ_CHECK_DELAYS; \
     } \
   }
 
@@ -115,6 +162,25 @@ static void _call_coz_add_delays(void) {
 #define COZ_BEGIN(name) COZ_INCREMENT_COUNTER(COZ_COUNTER_TYPE_BEGIN, name)
 #define COZ_END(name) COZ_INCREMENT_COUNTER(COZ_COUNTER_TYPE_END, name)
 
+// Custom synchronization support.
+// Use these macros around blocking operations that Coz does not intercept
+// (e.g., custom mutexes, futex-based locks, RocksDB internal synchronization).
+//
+//   COZ_PRE_BLOCK;                        // before blocking
+//   my_custom_lock_acquire(&lock);
+//   COZ_POST_BLOCK(1);                    // after blocking (1 = skip delays)
+//
+//   // Before potentially unblocking another thread:
+//   COZ_CATCH_UP;
+//   my_custom_lock_release(&lock);
+//
+// COZ_POST_BLOCK(skip_delays):
+//   skip_delays=1 when woken by another thread (e.g., mutex acquired)
+//   skip_delays=0 when the wake may have been spurious or timed out
+#define COZ_PRE_BLOCK _call_coz_pre_block()
+#define COZ_CATCH_UP _call_coz_add_delays()
+#define COZ_POST_BLOCK(skip_delays) _call_coz_post_block(skip_delays)
+
 #if defined(__cplusplus)
 }
 #endif

libcoz/libcoz.cpp

@@ -54,6 +54,24 @@ extern "C" void _coz_add_delays() {
   }
 }
 
+/**
+ * Called by the application before a custom blocking operation.
+ * For use with synchronization primitives not intercepted by Coz
+ * (e.g., MySQL's custom mutexes, RocksDB internal locks).
+ */
+extern "C" void _coz_pre_block() {
+  if(initialized) profiler::get_instance().pre_block();
+}
+
+/**
+ * Called by the application after a custom blocking operation completes.
+ * If skip_delays is non-zero, delays inserted during the blocked period
+ * are skipped (use when woken by another thread).
+ */
+extern "C" void _coz_post_block(int skip_delays) {
+  if(initialized) profiler::get_instance().post_block(skip_delays != 0);
+}
+
 #ifdef __APPLE__
 /**
  * Helper functions called from mac_interpose.cpp

libcoz/profiler.cpp

@@ -299,19 +299,6 @@ void profiler::profiler_thread(spinlock& l) {
     }
     _latency_points_lock.unlock();
 
-    // Sync all non-blocked threads' local_delay to global_delay before starting.
-    // This prevents stale residual from a previous experiment from causing threads
-    // to skip delays (local > global branch in add_delays).
-    // Skip blocked threads — their pre_block/post_block mechanism handles accounting.
-    {
-      size_t current_global = _global_delay.load();
-      _thread_states.for_each([current_global](pid_t tid, thread_state* state) {
-        if(!state->is_blocked.load()) {
-          state->local_delay.store(current_global);
-        }
-      });
-    }
-
 #ifdef __APPLE__
     // Reset overshoot counter for this experiment
     g_experiment_overshoot.store(0, std::memory_order_relaxed);

libcoz/profiler.h

@@ -169,7 +169,16 @@ class profiler {
     // Handle all samples and add delays as required
     if(_experiment_active) {
       state->set_in_use(true);
+#ifndef __APPLE__
+      // On Linux, samples accumulate in the per-thread perf_event buffer between
+      // timer signals (every 10ms).  Process pending samples so the delay counters
+      // are up-to-date before we potentially unblock another thread.
+      // process_samples() calls add_delays() at the end.
+      process_samples(state);
+#else
+      // On macOS, samples are processed centrally by the profiler thread.
       add_delays(state);
+#endif
       state->set_in_use(false);
     }
   }
@@ -197,7 +206,18 @@ class profiler {
       state->local_delay.fetch_add(_global_delay.load() - state->pre_block_time);
     }
 
+    // Must clear is_blocked before process_samples() because add_delays()
+    // (called at the end of process_samples()) returns early if is_blocked is true.
     state->is_blocked.store(false);
+
+#ifndef __APPLE__
+    // On Linux, process any samples that accumulated while this thread was
+    // blocked to bring its delay counters up to date (BCOZ fix).
+    if(_experiment_active) {
+      process_samples(state);
+    }
+#endif
+
     state->set_in_use(false);
   }