Fix false sharing in IO thread active-time and job-finished counters (#4976)

Follow-up to #4107, from the false-sharing survey in [this
comment](https://github.com/valkey-io/valkey/pull/4107#issuecomment-6311748165).

### Problem

Two counters in `io_threads.c` that IO threads write on every busy loop
iteration are packed with other threads' data:

- `used_active_time_io_thread[]` stores one `_Atomic long long` per IO
thread, so eight slots share a 64-byte line (sixteen on 128-byte-line
targets), and each IO thread adds to its own slot after every iteration
that did work.
- `io_jobs_finished` is updated by every IO thread after each busy
iteration, and the line it's on also holds main-thread state. On a macOS
clang build its neighbour is `io_jobs_submitted`, which the main thread
increments for every job. On Linux aarch64 with GCC 13 and LTO, its
neighbours are `cur_epoll_thread`, `cluster_io_pending_responses` and
the first `shared_querybuf_registry` slots.

### Fix

Following #4107, each `used_active_time_io_thread` slot and
`io_jobs_finished` are wrapped in a struct aligned to `CACHE_LINE_SIZE`,
and a `static_assert` checks each wrapper is exactly one line. Only
`INFO` reads `used_active_time_io_thread`, so padding it costs nothing
on the read side. It's now `static`; nothing outside `io_threads.c`
referenced it.

This doesn't address the true sharing on `io_jobs_finished` itself,
where all IO threads update the same counter. Fixing that would need
per-thread counters summed by the main thread.

Memory cost: `used_active_time_io_thread` grows from 2 KiB to 16 KiB
with 64-byte lines (32 KiB with 128-byte lines), in `.bss`.

### Testing

- Builds with `-Werror`; `clang-format` 18.1.8 is clean.
- Symbol layout (macOS arm64, 128-byte lines): `io_jobs_finished` is
alone on its line, and the array is 256 × 128 B.
- Removing the `_Alignas` makes both `static_assert`s fail.
- New gtests in `src/unit/test_io_threads.cpp` check, through `testOnly`
accessors, that every `used_active_time_io_thread` slot is cache-line
aligned and one line apart, and that `io_jobs_finished` shares no line
with `io_jobs_submitted` or any slot. They pass on Linux aarch64 (GCC
13, `-Werror`) alongside the existing `ClusterIOOffload` tests, and all
three fail against the unpadded layout.
- `unit/io-threads`, `unit/info` and `unit/introspection` pass. `stats:
client input and output buffer limit disconnections` fails under
`--config io-threads 4` on macOS both with and without this change.

### Performance

Measured by Rain Valentine
([review](https://github.com/valkey-io/valkey/pull/4976#pullrequestreview-5631041699)),
cachecannon GET 16B, 400 connections, 5 × 30s:

- clang 15 + LTO, Sapphire Rapids, where
`io_jobs_submitted`/`io_jobs_finished` share a line with
`cur_epoll_thread` and `cluster_io_pending_responses`: +2.6% throughput
(CI95 ±1.4%) at pipeline 1 / io-threads 16, with main-thread L1d misses
per request down 5%. Pipeline 10 / io-threads 7 unchanged.
- GCC 11 + LTO, Sapphire Rapids and Graviton 4, where the two counters
already sit on separate lines: within ±2% in every cell (io-threads
7/16, pipeline 1/10).

---------

Signed-off-by: Omer Rubinstein <omerrubi@amazon.com>

src/io_threads.c

@@ -48,12 +48,26 @@ static mpscQueue io_shared_outbox[JOB_PRIORITY_COUNT] = {0};
 // Main -> IO (Thread-Specific) for tasks that must run on specific IO thread where IO threads check their private inbox before the shared queue
 static spscQueue io_private_inbox[IO_THREADS_MAX_NUM] = {0};
 static size_t io_jobs_submitted;
-static _Atomic(size_t) io_jobs_finished;
+/* Every IO thread adds to io_jobs_finished after each busy loop iteration. The wrapper gives
+ * it a cache line of its own, so those updates don't invalidate the line holding the
+ * main-thread-only state next to it (e.g. io_jobs_submitted). */
+typedef struct {
+    _Alignas(CACHE_LINE_SIZE) _Atomic(size_t) value;
+} io_jobs_finished_counter;
+static_assert(sizeof(io_jobs_finished_counter) == CACHE_LINE_SIZE, "io_jobs_finished must occupy exactly one cache line");
+static io_jobs_finished_counter io_jobs_finished;
 static size_t cluster_io_pending_responses;
 static int io_threads_initialized = 0;
 /* Set during process exit to stop worker response retries. */
 static _Atomic int io_threads_exiting = 0;
-_Atomic long long used_active_time_io_thread[IO_THREADS_MAX_NUM] = {0};
+/* Each IO thread adds to its own slot on every busy loop iteration, so each slot gets a cache
+ * line of its own to keep one thread's update from invalidating its neighbours' slots
+ * (false sharing). Only INFO reads these. */
+typedef struct {
+    _Alignas(CACHE_LINE_SIZE) _Atomic long long value;
+} io_thread_active_time;
+static_assert(sizeof(io_thread_active_time) == CACHE_LINE_SIZE, "each used_active_time_io_thread slot must occupy exactly one cache line");
+static io_thread_active_time used_active_time_io_thread[IO_THREADS_MAX_NUM];
 
 /* Job Types for Tagged Pointers
  * We use the lower 3 bits of the pointer to store the job type.
@@ -97,7 +111,7 @@ void commitIOJobs(void) {
 
 /* Jobs sent but not yet processed by IO threads. */
 static size_t getPendingIOThreadsJobs(void) {
-    return io_jobs_submitted - atomic_load_explicit(&io_jobs_finished, memory_order_acquire);
+    return io_jobs_submitted - atomic_load_explicit(&io_jobs_finished.value, memory_order_acquire);
 }
 
 /* Read/write jobs awaiting response from IO threads. */
@@ -381,7 +395,7 @@ static void *IOThreadMain(void *myid) {
         monotime prev_work_start_time = work_start_time;
         work_start_time = getMonotonicUs();
         if (processed != 0) {
-            atomic_fetch_add_explicit(&used_active_time_io_thread[id],
+            atomic_fetch_add_explicit(&used_active_time_io_thread[id].value,
                                       work_start_time - prev_work_start_time,
                                       memory_order_relaxed);
         }
@@ -421,7 +435,7 @@ static void *IOThreadMain(void *myid) {
         }
 
         if (processed) {
-            atomic_fetch_add_explicit(&io_jobs_finished, processed, memory_order_release);
+            atomic_fetch_add_explicit(&io_jobs_finished.value, processed, memory_order_release);
         }
 
         /* If both queues were empty (no processing done), wait for signal. */
@@ -444,7 +458,7 @@ static void *IOThreadMain(void *myid) {
 }
 
 long long getIOThreadActiveTimeMicroseconds(int id) {
-    return atomic_load_explicit(&used_active_time_io_thread[id], memory_order_relaxed);
+    return atomic_load_explicit(&used_active_time_io_thread[id].value, memory_order_relaxed);
 }
 
 static void createIOThread(int id) {
@@ -515,7 +529,7 @@ static void shutdownIOThread(int id) {
             }
             drained += batch_count;
         }
-        if (drained) atomic_fetch_add_explicit(&io_jobs_finished, drained, memory_order_release);
+        if (drained) atomic_fetch_add_explicit(&io_jobs_finished.value, drained, memory_order_release);
     }
 
     spscFree(&io_private_inbox[id]);
@@ -598,7 +612,7 @@ void initIOThreads(int prev_threads_num) {
             mpscInit(&io_shared_outbox[p], IO_MPSC_QUEUE_SIZE);
         }
         io_jobs_submitted = 0;
-        atomic_init(&io_jobs_finished, 0);
+        atomic_init(&io_jobs_finished.value, 0);
         cluster_io_pending_responses = 0;
         prefetchCommandsBatchInit();
         io_threads_initialized = 1;
@@ -623,7 +637,7 @@ void testOnlyInitIOThreadQueues(void) {
         io_thread_ticket[p] = (mpscTicket){0};
     }
     io_jobs_submitted = 0;
-    atomic_store_explicit(&io_jobs_finished, 0, memory_order_relaxed);
+    atomic_store_explicit(&io_jobs_finished.value, 0, memory_order_relaxed);
     cluster_io_pending_responses = 0;
 }
 
@@ -638,7 +652,7 @@ void testOnlyFreeIOThreadQueues(void) {
         io_thread_ticket[p] = (mpscTicket){0};
     }
     io_jobs_submitted = 0;
-    atomic_store_explicit(&io_jobs_finished, 0, memory_order_relaxed);
+    atomic_store_explicit(&io_jobs_finished.value, 0, memory_order_relaxed);
     cluster_io_pending_responses = 0;
 }
 
@@ -658,6 +672,19 @@ size_t testOnlyGetClusterIOPendingResponses(void) {
     return cluster_io_pending_responses;
 }
 
+/* Addresses of the IO thread counters, so unit tests can check their cache line layout. */
+const void *testOnlyGetIOThreadActiveTimeSlot(int id) {
+    return &used_active_time_io_thread[id].value;
+}
+
+const void *testOnlyGetIOJobsFinishedCounter(void) {
+    return &io_jobs_finished.value;
+}
+
+const void *testOnlyGetIOJobsSubmittedCounter(void) {
+    return &io_jobs_submitted;
+}
+
 /* This function attempts to offload the client's read to an I/O thread.
  * Returns C_OK if the read was offloaded, or if an I/O thread is already reading
  * or writing for this client, so the main thread must not read now.

src/io_threads.h

@@ -55,6 +55,9 @@ void testOnlyInitIOThreadQueues(void);
 void testOnlyFreeIOThreadQueues(void);
 void testOnlyFillIOThreadInbox(void);
 size_t testOnlyGetClusterIOPendingResponses(void);
+const void *testOnlyGetIOThreadActiveTimeSlot(int id);
+const void *testOnlyGetIOJobsFinishedCounter(void);
+const void *testOnlyGetIOJobsSubmittedCounter(void);
 void trySendPollJobToIOThreads(void);
 int trySendAcceptToIOThreads(connection *conn);
 struct clusterLink;

src/unit/test_io_threads.cpp

@@ -0,0 +1,47 @@
+/*
+ * Copyright (c) Valkey Contributors
+ * All rights reserved.
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+#include "generated_wrappers.hpp"
+
+#include <stdint.h>
+
+extern "C" {
+#include "io_threads.h"
+#include "server.h"
+}
+
+/* Each IO thread writes its own counters on every busy loop iteration. These tests
+ * check that no two threads' counters, and no main-thread state, share a cache line. */
+static uintptr_t addr(const void *p) {
+    return (uintptr_t)p;
+}
+
+static uintptr_t cacheLine(const void *p) {
+    return addr(p) / CACHE_LINE_SIZE;
+}
+
+TEST(IOThreadsTest, ActiveTimeSlotsAreCacheLineAligned) {
+    for (int id = 0; id < IO_THREADS_MAX_NUM; id++) {
+        ASSERT_EQ(addr(testOnlyGetIOThreadActiveTimeSlot(id)) % CACHE_LINE_SIZE, 0u) << "slot " << id;
+    }
+}
+
+TEST(IOThreadsTest, ActiveTimeSlotsDoNotShareCacheLines) {
+    for (int id = 1; id < IO_THREADS_MAX_NUM; id++) {
+        ASSERT_EQ(addr(testOnlyGetIOThreadActiveTimeSlot(id)) - addr(testOnlyGetIOThreadActiveTimeSlot(id - 1)),
+                  (uintptr_t)CACHE_LINE_SIZE)
+            << "slots " << id - 1 << " and " << id;
+    }
+}
+
+TEST(IOThreadsTest, JobsFinishedCounterHasItsOwnCacheLine) {
+    const void *finished = testOnlyGetIOJobsFinishedCounter();
+    ASSERT_EQ(addr(finished) % CACHE_LINE_SIZE, 0u);
+    ASSERT_NE(cacheLine(finished), cacheLine(testOnlyGetIOJobsSubmittedCounter()));
+    for (int id = 0; id < IO_THREADS_MAX_NUM; id++) {
+        ASSERT_NE(cacheLine(finished), cacheLine(testOnlyGetIOThreadActiveTimeSlot(id))) << "slot " << id;
+    }
+}