The CPU suite never runs anything in parallel — turn it on, under TSan, and fix the deadlock that finds

.github/workflows/ci.yml

@@ -411,6 +411,90 @@ jobs:
  if: always()
  run: ccache -s
 
+ # The same suite a third time, under ThreadSanitizer AND with the thread
+ # pool's grain size forced down to one element.
+ #
+ # Why this exists: with the shipping grain size the CPU suite never runs
+ # anything in parallel at all. `parallel_for` hands work to the pool only
+ # when the loop covers strictly more than GRAIN_SIZE (32768) elements, and
+ # every workload in tests/lib-cpu/ is far below that -- measured, all 13 test
+ # binaries make **zero** `clone` syscalls, so `internal::invoke_parallel`,
+ # the thread pool, and every "accumulate into a shared output" path are
+ # executed single-threaded by a gate reporting 59,886 passing checks. The
+ # scatter (`push`/`count`/`restriction`) paths, the ones that exist precisely
+ # because several threads may land on the same output voxel, are exactly the
+ # ones that concurrency bug would live in.
+ #
+ # * -DFF_GRAIN_SIZE=1 makes every parallel_for in the tree take the pooled
+ # path, so the *existing* checks run concurrently. That is much stronger
+ # coverage than adding one big-volume test case, and it does not move the
+ # check count: same 13 suites, same 59,886 checks, same references.
+ # Verified locally -- 59,886 / 13 / 0 failures at FF_GRAIN_SIZE=1, and
+ # 2 clone calls per binary instead of 0.
+ # * TSan, not ASan/UBSan: the existing sanitize leg answers "is the
+ # behaviour defined?", this one answers "is it defined when two threads
+ # do it at once?". A lost update in a float accumulation is a silent
+ # wrong answer -- no crash, no ASan report, and the reference comparison
+ # only catches it if the race happens to fire on that run. TSan flags the
+ # unsynchronised access whether or not it fires.
+ # * FF_NUM_THREADS=4 rather than the default (hardware_concurrency/2 = 2 on
+ # a 4-vCPU runner): more workers, more interleavings, no extra cost.
+ # * halt_on_error=1 makes the first report fatal, for the same reason
+ # -fno-sanitize-recover=all does in the leg above.
+ #
+ # TSan and ASan cannot be combined in one binary, which is why this is a
+ # separate job rather than another flag on the one above.
+ tsan:
+ name: test-cpu (tsan, grain=1)
+ needs: changes
+ if: needs.changes.outputs.cpu == 'true'
+ runs-on: ubuntu-latest
+ timeout-minutes: 45
+ steps:
+ - uses: actions/checkout@v5
+ - name: Install build toolchain
+ run: sudo apt-get update && sudo apt-get install -y clang ccache
+ # Same requirement as the ASan leg: `clang` alone does not ship
+ # compiler-rt's runtimes, and without them the link fails looking for
+ # libclang_rt.tsan-x86_64.a.
+ - name: Install the matching compiler-rt sanitizer runtimes
+ run: sudo apt-get install -y "libclang-rt-$(clang -dumpversion | cut -d. -f1)-dev"
+ - name: Set CCACHE_DIR
+ run: echo "CCACHE_DIR=$HOME/.ccache" >> "$GITHUB_ENV"
+ - name: Cache ccache directory
+ uses: actions/cache@v4
+ with:
+ path: ~/.ccache
+ key: ${{ runner.os }}-ccache-test-cpu-tsan-${{ hashFiles('include/**', 'src/**', 'tests/**') }}
+ restore-keys: |
+ ${{ runner.os }}-ccache-test-cpu-tsan-
+ - name: Configure ccache
+ run: |
+ ccache --max-size=1G
+ ccache -z
+ - name: Build and run the CPU suite under TSan with a one-element grain
+ env:
+ TSAN_OPTIONS: halt_on_error=1 second_deadlock_stack=1
+ FF_NUM_THREADS: 4
+ run: |
+ make -j2 -C . test-lib-cpu CXX="ccache clang++" \
+ CXXFLAGS="-std=c++11 -O1 -g -fPIC -DFF_GRAIN_SIZE=1 -fsanitize=thread -fno-omit-frame-pointer -ferror-limit=1 -ftemplate-backtrace-limit=0" \
+ BOUNDFLAGS="-DFF_STATIC_BOUNDS=0" \
+ SPLINEFLAGS="-DFF_STATIC_SPLINES=0"
+ # Seconds, and it states in one place what the suite above only implies:
+ # which CPU atomicAdd implementation is actually selected, and that
+ # concurrent accumulation into disjoint slots is exact.
+ - name: What ff::anyAtomicAdd is on the CPU
+ env:
+ TSAN_OPTIONS: halt_on_error=1
+ FF_NUM_THREADS: 4
+ run: |
+ make -C . test-atomics CXX="ccache clang++" \
+ CXXFLAGS="-std=c++11 -O1 -g -DFF_GRAIN_SIZE=1 -fsanitize=thread -fno-omit-frame-pointer -ferror-limit=1 -ftemplate-backtrace-limit=0"
+ - name: Show ccache stats
+ if: always()
+ run: ccache -s
+
  # ==========================================================================
  # The hub's own tests.
  # ==========================================================================

CLAUDE.md

@@ -111,9 +111,18 @@ the CPU path is the tested source of truth and CUDA is **compile+link only**.
 ## CI
 
 `.github/workflows/ci.yml`, path-filtered. `codespell` always; `test-cpu` (a
-3-leg `BOUNDFLAGS`/`SPLINEFLAGS` matrix + a g++ leg) and `sanitize` on
-kernels/cpu/hub changes; `test-hub` on hub changes; `build-cuda` and
-`compile-probe-cuda` on kernels/cuda changes.
+3-leg `BOUNDFLAGS`/`SPLINEFLAGS` matrix + a g++ leg), `sanitize` (ASan+UBSan)
+and `tsan` on kernels/cpu/hub changes; `test-hub` on hub changes; `build-cuda`
+and `compile-probe-cuda` on kernels/cuda changes.
+
+**The `tsan` leg is the only one that runs anything in parallel.** With the
+shipping `GRAIN_SIZE` (32768) every workload in `tests/lib-cpu/` is below the
+threshold at which `parallel_for` hands work to the thread pool, so the whole
+59,886-check suite executes single-threaded -- measured, zero `clone` syscalls
+across all 13 binaries. That leg rebuilds the same suite with
+`-DFF_GRAIN_SIZE=1` so the same checks run concurrently, under ThreadSanitizer.
+It is not a performance knob: results must be identical at any grain size, and
+32768 stays the shipping value.
 
 `build-cuda` builds `libfastfields-cuda.so` **and then links the hub against it**
 (`make lib USE_CUDA=1`). That second step is the point: building the CUDA

Makefile

@@ -11,7 +11,7 @@ GROUP := root
 include make/common.mk
 
 .PHONY: all cpu cuda lib test test-lib test-lib-cpu test-impl-cuda test-kernels \
- clean install
+test-atomics clean install
 
 # `all` produces both shared objects -- build/lib/libfastfields-cpu.so and
 # build/libfastfields.so -- exactly as the hub repo's `all` did. CUDA stays
@@ -53,5 +53,17 @@ test-kernels: | $(TESTDIR)
 	$(CXX) $(CXXFLAGS) $(DIAGFLAGS) $(INCLUDES) -std=c++11 \
  -o $(BUILDDIR)/test/kernels_vector tests/kernels/vector/test.cpp
 
+# What ff::anyAtomicAdd is on the CPU, and that concurrent accumulation into
+# disjoint slots stays exact. Deliberately NOT part of `make test`: the 13
+# suites and 59,886 checks that gate is defined by must not move. It compiles
+# and runs in milliseconds and is built by the tsan CI leg, which is where the
+# threaded case is worth anything. Honours CXXFLAGS, so
+# make test-atomics CXXFLAGS="-std=c++11 -O1 -g -DFF_GRAIN_SIZE=1 -fsanitize=thread"
+# is the interesting invocation.
+test-atomics: | $(TESTDIR)
+	$(CXX) $(CXXFLAGS) $(DIAGFLAGS) $(INCLUDES) -std=c++11 \
+ -o $(BUILDDIR)/test/kernels_atomic tests/kernels/atomic/test.cpp
+	$(BUILDDIR)/test/kernels_atomic
+
 clean:
 	$(DEL) -r $(BUILDDIR)

include/fastfields/impl/kernels/parallel.h

@@ -8,9 +8,29 @@
 #include "fastfields/core/defines.h"
 #include "parallel_impl.h"
 
+/* The minimum number of elements a `parallel_for` must cover before it is
+ * worth handing to the thread pool; below it, `parallel_for` calls `f` inline
+ * on the calling thread and no thread is ever created.
+ *
+ * Overridable at build time (`-DFF_GRAIN_SIZE=<n>`) for one specific reason:
+ * *every* workload in `tests/lib-cpu/` sits below the shipping value, so with
+ * the default the whole suite -- 59,886 checks -- runs single-threaded and
+ * `internal::invoke_parallel`, the thread pool, and every accumulate-into-a-
+ * shared-output path are never executed concurrently (measured: zero `clone`
+ * syscalls across all 13 test binaries). Building the suite with a small
+ * FF_GRAIN_SIZE makes the same checks run multi-threaded without inventing new,
+ * slower, large-volume test cases, which is what the `tsan` CI leg does.
+ *
+ * This is a threshold, never a correctness switch: results must be identical at
+ * any value. Do not use it to tune performance -- 32768 is the shipping value.
+ */
+#ifndef FF_GRAIN_SIZE
+# define FF_GRAIN_SIZE 32768
+#endif
+
 FF_NAMESPACE_BEGIN(FF_NS)
 
-constexpr int64_t GRAIN_SIZE = 32768;
+constexpr int64_t GRAIN_SIZE = FF_GRAIN_SIZE;
 
 template <class F>
 inline void parallel_for(int64_t begin, int64_t end, int64_t grain_size, const F& f)

include/fastfields/impl/kernels/threadpool.h

@@ -5,6 +5,7 @@
 #ifndef FF_THREADPOOL_H
 #define FF_THREADPOOL_H
 
+#include <atomic>
 #include <list>
 #include <deque>
 #include <future>
@@ -103,6 +104,11 @@ class ThreadPool
  mQ.pop_back();
  return true;
  }
+ bool empty()
+ {
+ std::unique_lock<std::mutex> lock(mQMut);
+ return mQ.empty();
+ }
  private:
  std::deque<T> mQ;
  std::mutex mQMut;
@@ -113,8 +119,9 @@ class ThreadPool
  public:
  Worker() = delete;
  Worker(ThreadPool* pool, int id) :
+ mId(id),
  mPool(pool),
-mId(id)
+mExit(false)
  {
  }
  void start()
@@ -127,8 +134,17 @@ class ThreadPool
  wake();
  mPool->requestSteal();
  }
+ // Acquiring mCvMut before notifying is not decoration. The waiter below
+ // evaluates its predicate and blocks while holding mCvMut, so taking
+ // the same mutex here makes it impossible for a notify to land in
+ // between -- which is the classic lost wakeup: the worker checks "no
+ // work, not exiting", the pusher enqueues and notifies, and the worker
+ // then sleeps on a queue that is no longer empty. With one task and one
+ // idle worker that is a hang, not a slowdown.
  void wake()
  {
+ std::unique_lock<std::mutex> lock(mCvMut);
+ lock.unlock();
  mCv.notify_one();
  }
  bool steal(std::function<void()>& work)
@@ -157,7 +173,11 @@ class ThreadPool
  Queue<std::function<void()>> mQue;
  int mId;
  ThreadPool* mPool;
- bool mExit = false;
+ // Written by whoever calls exit() (the pool's destructor, on the main
+ // thread) and read by this worker's own thread, so it must be atomic:
+ // as a plain bool the pair is a data race, which is exactly what
+ // ThreadSanitizer reports the first time the pool is ever exercised.
+ std::atomic<bool> mExit;
 
  void threadFunc()
  {
@@ -179,8 +199,14 @@ class ThreadPool
  }
  else
  {
+ // Predicated wait, not a bare one: re-checking the exit
+ // flag and the queue under mCvMut is the other half of the
+ // handshake wake() takes that mutex for, and it also
+ // absorbs spurious wakeups.
  std::unique_lock<std::mutex> lock(mCvMut);
- mCv.wait(lock);
+ mCv.wait(lock, [this] {
+ return mExit.load() || !mQue.empty();
+ });
  }
  }
  }

tests/kernels/atomic/test.cpp

@@ -0,0 +1,89 @@
+/* What `ff::anyAtomicAdd` actually is on the CPU, pinned.
+ *
+ * `impl/kernels/atomic.h` offers two CPU implementations and picks between
+ * them with `has_atomic_add<T>`:
+ *
+ * AtomicAdd<false> -- a plain, non-atomic `*address += val`
+ * AtomicAdd<true> -- a std::atomic compare-exchange loop
+ *
+ * and `impl/cpu/pushpull.h` / `impl/cpu/restrict.h` branch on the *same*
+ * predicate to choose a parallelisation strategy: `if (has_atomic_add<T>)`
+ * parallelise over every element and let the atomics resolve the collisions,
+ * else parallelise over the batch dimension only, so that concurrent threads
+ * write to disjoint output slices and no atomicity is required.
+ *
+ * The predicate answers **false for every type, in every language standard**
+ * -- `has_fetch_add` probes `&C::fetch_add`, which is an overload set on
+ * `std::atomic` (with and without a memory_order argument), so taking its
+ * address is ambiguous and SFINAE rejects it before the C++ version is ever
+ * relevant. The CAS specialisation is therefore unreachable, and the scatter
+ * paths always take the disjoint-slice branch. That is what makes them
+ * correct, so it is worth a test rather than a comment: someone "fixing" the
+ * detector would silently switch every scatter op onto a strategy whose
+ * atomics are a plain `+=`.
+ *
+ * Not part of `make test` (`make test-atomics` runs it). It links nothing and
+ * takes milliseconds; it is built by the tsan CI leg, where the threaded case
+ * below is the interesting one.
+ */
+#include "fastfields/impl/kernels/atomic.h"
+#include "fastfields/impl/kernels/parallel.h"
+#include <cstdio>
+#include <cstdint>
+#include <vector>
+
+static int g_failures = 0;
+
+static void check(bool ok, const char * what)
+{
+ if (!ok) { ++g_failures; std::printf(" FAIL [%s]\n", what); }
+ else std::printf(" ok: %s\n", what);
+}
+
+int main()
+{
+ /* 1. The predicate. If any of these ever flips, read the header comment
+ * above before changing the expectation -- `anyAtomicAdd` does not even
+ * compile for a type the predicate accepts, because
+ * `AtomicAdd<true>::atomicAdd` returns void while `anyAtomicAdd`
+ * returns T. */
+ check(!ff::has_atomic_add<float>::value, "has_atomic_add<float> is false");
+ check(!ff::has_atomic_add<double>::value, "has_atomic_add<double> is false");
+ check(!ff::has_atomic_add<int32_t>::value, "has_atomic_add<int32_t> is false");
+ check(!ff::has_atomic_add<int64_t>::value, "has_atomic_add<int64_t> is false");
+
+ /* 2. Single-threaded semantics of the implementation that is actually
+ * selected: accumulate, and return the *new* value. (Note the CUDA
+ * `gpuAtomicAdd` this shadows returns the *old* value; nothing in the
+ * tree uses either return value, which is why the divergence is
+ * harmless today.) */
+ {
+ double x = 1.0;
+ double r = ff::anyAtomicAdd(&x, 2.0);
+ check(x == 3.0, "anyAtomicAdd accumulates into the target");
+ check(r == 3.0, "anyAtomicAdd returns the new value (CUDA returns the old one)");
+ ff::anyAtomicAddNoReturn(&x, -3.0);
+ check(x == 0.0, "anyAtomicAddNoReturn accumulates into the target");
+ }
+
+ /* 3. The strategy the scatter ops rely on: many threads accumulating into
+ * *disjoint* slots is race-free even though the add is not atomic.
+ * Under TSan this is the case that must stay clean; it fails there the
+ * moment two workers are given overlapping output. */
+ {
+ const long n = 4096;
+ std::vector<double> out((size_t)n, 0.0);
+ double * p = &out[0];
+ ff::parallel_for(0, n, /*grain_size=*/1, [p](long begin, long end) {
+ for (long i = begin; i < end; ++i)
+ ff::anyAtomicAdd(p + i, 1.0);
+ });
+ bool all_one = true;
+ for (long i = 0; i < n; ++i) if (out[(size_t)i] != 1.0) all_one = false;
+ check(all_one, "disjoint concurrent accumulation is exact");
+ std::printf(" (%zu worker threads available)\n", ff::get_parallel_threads());
+ }
+
+ std::printf("%s\n", g_failures ? "FAILURES" : "All checks passed.");
+ return g_failures ? 1 : 0;
+}