Skip to content

Instantly share code, notes, and snippets.

@pkhuong

pkhuong/ck_ec.c

Last active October 27, 2018 15:33
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save pkhuong/d6ea6cab3c3e2364afcda4acb58317e2 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save pkhuong/d6ea6cab3c3e2364afcda4acb58317e2 to your computer and use it in GitHub Desktop.
Download ZIP
Futex-backed event count for concurrency kit
Raw
ck_ec.c
#define _GNU_SOURCE
#include "ck_ec.h"
#include <ck_limits.h>
#include <ck_stdbool.h>
#include <linux/futex.h>
#include <sys/syscall.h>
#include <time.h>
#include <unistd.h>
#define BUSY_LOOP_ITER 100U
#define INITIAL_WAIT_TIME_NS 1000000L /* Start at 1 ms */
/* Once we've waited for >= 1 sec, go for the full deadline. */
#define FINAL_WAIT_TIME_SEC 1
#define WAIT_TIME_SCALE_FACTOR 1.5 /* Grow the wait time by 50%/iteration. */
#define TIME_MAX ((1ULL << ((sizeof(time_t) * CHAR_BIT) - 1)) - 1)
struct ck_ec32_wait_state {
struct ck_ec32 *ec;
uint32_t flagged_word;
};
struct ck_ec64_wait_state {
struct ck_ec64 *ec;
uint64_t flagged_word;
};
void ck_ec32_wake(struct ck_ec32 *ec)
{
/* Spurious wake-ups are OK. Clear the flag before futexing. */
ck_pr_and_32(&ec->counter, (1U << 31) - 1);
syscall(SYS_futex, (int *)&ec->counter,
FUTEX_WAKE, INT_MAX,
/* ignored arguments */NULL, NULL, 0);
return;
}
void ck_ec64_wake(struct ck_ec64 *ec)
{
ck_pr_and_64(&ec->counter, ~1UL);
syscall(SYS_futex, (int *)&ec->counter,
FUTEX_WAKE, INT_MAX,
/* ignored arguments */NULL, NULL, 0);
return;
}
int ck_ec32_wait_slow(struct ck_ec32 *ec, uint32_t old_value,
const struct timespec *deadline)
{
return ck_ec32_wait_pred_slow(ec, old_value,
NULL, NULL, deadline);
}
int ck_ec64_wait_slow(struct ck_ec64 *ec, uint64_t old_value,
const struct timespec *deadline)
{
return ck_ec64_wait_pred_slow(ec, old_value,
NULL, NULL, deadline);
}
/*
* Increments acc by scale * y. scale * y->tv_nsec must not exceed
* LONG_MAX.
*/
static void timespec_inc(struct timespec *acc,
double scale,
const struct timespec *y)
{
const long nsec_per_sec = 1000 * 1000 * 1000L;
const double delta_sec = y->tv_sec * scale;
unsigned long nsec = acc->tv_nsec + scale * y->tv_nsec;
if (delta_sec >= TIME_MAX - acc->tv_sec) {
/* This is already infinitely in the future. */
acc->tv_sec = TIME_MAX;
return;
}
acc->tv_sec += delta_sec;
acc->tv_sec += nsec / nsec_per_sec;
acc->tv_nsec = nsec % nsec_per_sec;
return;
}
int ck_ec_deadline(struct timespec *new_deadline,
const struct timespec *timeout)
{
struct timespec ret = { .tv_sec = TIME_MAX };
int r;
if (timeout == NULL) {
goto done;
}
r = clock_gettime(CLOCK_MONOTONIC, &ret);
if (r != 0) {
return -1;
}
timespec_inc(&ret, 1, timeout);
done:
*new_deadline = ret;
return 0;
}
/* The rest of the file implements wait_pred_slow. */
/*
* Returns a timespec value for deadline_ptr. If deadline_ptr is NULL,
* returns a timespec far in the future.
*/
static struct timespec canonical_deadline(const struct timespec *deadline_ptr)
{
if (deadline_ptr == NULL) {
return (struct timespec) { .tv_sec = TIME_MAX };
}
return *deadline_ptr;
}
/* Compares two timespecs. Returns -1 if x < y, 0 if x == y, and 1 if x > y. */
static int timespec_cmp(const struct timespec *x, const struct timespec *y)
{
if (x->tv_sec != y->tv_sec) {
return (x->tv_sec < y->tv_sec) ? -1 : 1;
}
if (x->tv_nsec != y->tv_nsec) {
return (x->tv_nsec < y->tv_nsec) ? -1 : 1;
}
return 0;
}
/*
* Overwrites now with the current CLOCK_MONOTONIC time, and returns
* true if the current time is greater than the deadline, or the clock
* is somehow broken.
*/
static bool check_deadline(struct timespec *now,
const struct timespec *deadline)
{
int r;
r = clock_gettime(CLOCK_MONOTONIC, now);
if (r != 0) {
return true;
}
return timespec_cmp(now, deadline) > 0;
}
/*
* Really slow (sleeping) path for ck_ec_wait. Drives the exponential
* backoff scheme to sleep for longer and longer periods of time,
* until either the sleep function returns true (the eventcount's
* value has changed), or the predicate returns non-0 (something else
* has changed).
*
* If deadline is ever reached, returns -1 (timeout).
*
* TODO: add some form of randomisation to the intermediate timeout
* values.
*/
static int exponential_backoff(
bool (*sleep)(const void *sleep_state,
const struct timespec *partial_deadline),
const void *sleep_state,
int (*pred)(void *data, struct timespec *deadline,
const struct timespec *now),
void *data,
const struct timespec *deadline)
{
struct timespec begin;
struct timespec wait_time = { .tv_nsec = INITIAL_WAIT_TIME_NS };
bool first = true;
for (;;) {
struct timespec now;
struct timespec partial_deadline;
if (check_deadline(&now, deadline) == true) {
/* Timeout. */
return -1;
}
if (first) {
begin = now;
first = false;
}
/*
* If we've definitely waited more than
* FINAL_WAIT_TIME_SEC, just wait for the full
* deadline. Until then, stick to exponential backoff.
*/
if (now.tv_sec - begin.tv_sec > FINAL_WAIT_TIME_SEC) {
partial_deadline = *deadline;
} else {
partial_deadline = now;
timespec_inc(&partial_deadline, 1.0, &wait_time);
/*
* Add (SCALE_FACTOR - 1) * wait_time to
* wait_time , i.e., multiply wait_time by
* SCALE_FACTOR.
*/
timespec_inc(&wait_time, WAIT_TIME_SCALE_FACTOR - 1,
&wait_time);
}
if (pred != NULL) {
int r = pred(data, &partial_deadline, &now);
if (r != 0) {
return r;
}
}
/* Canonicalize deadlines in the far future to NULL. */
if (sleep(sleep_state,
((partial_deadline.tv_sec == TIME_MAX)
? NULL : &partial_deadline)) == true) {
return 0;
}
}
}
/*
* Loops up to BUSY_LOOP_ITER times, or until ec's counter value
* (including the flag) differs from old_value.
*
* Returns the new value in ec.
*/
#define DEF_WAIT_EASY(W) \
static uint##W##_t ck_ec##W##_wait_easy(struct ck_ec##W* ec, \
uint##W##_t expected) \
{ \
uint##W##_t current = expected; \
\
for (size_t i = 0; \
i < BUSY_LOOP_ITER && current == expected; \
i++) { \
ck_pr_stall(); \
current = ck_pr_load_##W(&ec->counter); \
} \
\
return current; \
}
DEF_WAIT_EASY(32)
DEF_WAIT_EASY(64)
#undef DEF_WAIT_EASY
/*
* Attempts to upgrade ec->counter from unflagged to flagged.
*
* Returns true if the event count has changed. Otherwise, ec's
* counter word is equal to flagged on return, or has been at some
* time before the return.
*/
#define DEF_UPGRADE(W) \
static bool ck_ec##W##_upgrade(struct ck_ec##W* ec, \
uint##W##_t current, \
uint##W##_t unflagged, \
uint##W##_t flagged) \
{ \
uint##W##_t old_word; \
\
if (current == flagged) { \
/* Nothing to do, no change. */ \
return false; \
} \
\
if (current != unflagged) { \
/* We have a different counter value! */ \
return true; \
} \
\
/* \
* Flag the counter value. The CAS only fails if the \
* counter is already flagged, or has a new value. \
*/ \
return (ck_pr_cas_##W##_value(&ec->counter, \
unflagged, flagged, \
&old_word) == false && \
old_word != flagged); \
}
DEF_UPGRADE(32)
DEF_UPGRADE(64)
#undef DEF_UPGRADE
/*
* Blocks until partial_deadline on the ck_ec. Returns true if the
* eventcount's value has changed. If partial_deadline is NULL, wait
* forever.
*/
static bool ck_ec32_wait_slow_once(const void *vstate,
const struct timespec *partial_deadline)
{
const struct ck_ec32_wait_state *state = vstate;
struct ck_ec32 *ec = state->ec;
const uint32_t flagged_word = state->flagged_word;
(void)syscall(SYS_futex, (int *)&ec->counter,
FUTEX_WAIT_BITSET, (int)flagged_word,
partial_deadline, NULL, FUTEX_BITSET_MATCH_ANY);
return ck_pr_load_32(&ec->counter) != flagged_word;
}
static bool ck_ec64_wait_slow_once(const void *vstate,
const struct timespec *partial_deadline)
{
const struct ck_ec64_wait_state *state = vstate;
struct ck_ec64 *ec = state->ec;
const uint64_t flagged_word = state->flagged_word;
/* futex_wait_bitset will only compare the low 32
* bits. Perform a full comparison here to maximise the
* changes of catching an ABA in the low 32 bits.
*/
if (ck_pr_load_64(&ec->counter) != flagged_word) {
return true;
}
/*
* We're on x86, a little-endian platform. Use the low-order
* bits as the futex int.
*/
(void)syscall(SYS_futex, (int *)&ec->counter,
FUTEX_WAIT_BITSET, (int)flagged_word,
partial_deadline, NULL, FUTEX_BITSET_MATCH_ANY);
return ck_pr_load_64(&ec->counter) != flagged_word;
}
/*
* The full wait logic is a lot of code (> 1KB). Encourage the
* compiler to lay this all out linearly with LIKELY annotations on
* every early exit.
*/
#define WAIT_SLOW_BODY(W, ec, pred, data, deadline_ptr, unflagged, flagged) \
do { \
const struct ck_ec##W##_wait_state state = { \
.ec = ec, \
.flagged_word = flagged \
}; \
const struct timespec deadline = \
canonical_deadline(deadline_ptr); \
uint##W##_t current; \
\
/* Detect infinite past deadlines. */ \
if (CK_CC_LIKELY(deadline.tv_sec <= 0)) { \
return -1; \
} \
\
current = ck_ec##W##_wait_easy(ec, unflagged); \
/* \
* We're about to wait harder (i.e., potentially with \
* futex). Make sure the counter word is flagged. \
*/ \
if (CK_CC_LIKELY( \
ck_ec##W##_upgrade(ec, current, \
unflagged, flagged) == true)) { \
return 0; \
} \
\
/* \
* By now, ec->counter == flagged_word (at some point \
* in the past). Spin some more to heuristically let \
* any in-flight SP inc/add to retire. This does not \
* affect correctness, but practically eliminates lost \
* wake-ups. \
*/ \
current = ck_ec##W##_wait_easy(ec, flagged); \
if (CK_CC_LIKELY(current != flagged_word)) { \
return 0; \
} \
\
return exponential_backoff(ck_ec##W##_wait_slow_once, \
&state, \
pred, data, &deadline); \
} while (0)
int ck_ec32_wait_pred_slow(struct ck_ec32 *ec, uint32_t old_value,
int (*pred)(void *data,
struct timespec *deadline,
const struct timespec *now),
void *data,
const struct timespec *deadline_ptr)
{
const uint32_t unflagged_word = old_value;
const uint32_t flagged_word = old_value | (1UL << 31);
WAIT_SLOW_BODY(32, ec, pred, data, deadline_ptr,
unflagged_word, flagged_word);
}
int ck_ec64_wait_pred_slow(struct ck_ec64 *ec, uint64_t old_value,
int (*pred)(void *data,
struct timespec *deadline,
const struct timespec *now),
void *data,
const struct timespec *deadline_ptr)
{
const uint64_t unflagged_word = old_value << 1;
const uint64_t flagged_word = unflagged_word | 1;
WAIT_SLOW_BODY(64, ec, pred, data, deadline_ptr,
unflagged_word, flagged_word);
}
#undef WAIT_SLOW_BODY
Raw
ck_ec.h
/*
* Copyright 2018 Paul Khuong.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions
* are met:
* 1. Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* 2. Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
*
* THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
* FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
* DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
* OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
* HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
* LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
* OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
* SUCH DAMAGE.
*/
/*
* Overview
* ========
*
* ck_ec implements 32- and 64- bit event counts. Event counts let us
* easily integrate OS-level blocking (futexes) in lock-free
* protocols. Waking up waiters only locks in the OS kernel, and does
* not happen at all when no waiter is blocked.
*
* Waiters only block conditionally, if the event count's value is
* still equal to some old value.
*
* Event counts come in four variants: 32 and 64 bit (with one bit
* stolen for internal signaling, so 31 and 63 bit counters), and
* single or multiple producers (wakers). Waiters are always multiple
* consumers. The 32 bit variants are smaller, and more efficient,
* especially in single producer mode. The 64 bit variants are larger,
* but practically invulnerable to ABA.
*
* A typical usage pattern is:
*
* 1. On the producer side:
*
* - Make changes to some shared data structure, without involving
* the event count at all.
* - After each change, call ck_ec_inc on the event count. The call
* acts as a write-write barrier, and wakes up any consumer blocked
* on the event count (waiting for new changes).
*
* 2. On the consumer side:
*
* - Snapshot ck_ec_value of the event count. The call acts as a
* read barrier.
* - Read and process the shared data structure.
* - Wait for new changes by calling ck_ec_wait with the snapshot value.
*
* Some data structures may opt for tighter integration with their
* event count. For example, an SPMC ring buffer or disruptor might
* use the event count's value as the write pointer. If the buffer is
* regularly full, it might also make sense to store the read pointer
* in an MP event count.
*
* This event count implementation supports tighter integration in two
* ways.
*
* Producers may opt to increment by an arbitrary value (less than
* INT32_MAX / INT64_MAX), in order to encode, e.g., byte
* offsets. Larger increment values make wraparound more likely, so
* the increments should still be relatively small.
*
* Consumers may pass a predicate to ck_ec_wait_pred. This predicate
* can make `ck_ec_wait_pred` return early, before the event count's
* value changes, and can override the deadline passed to futex_wait.
* This lets consumer block on one eventcount, while optimistically
* looking at other waking conditions.
*
* API Reference
* =============
*
* When compiled as C11 or later, this header defines type-generic
* macros for ck_ec32 and ck_ec64; the reference describes this
* type-generic API.
*
* ec is a struct ck_ec32*, or a struct ck_ec64*.
*
* value is an uint32_t for ck_ec32, and an uint64_t for ck_ec64. It
* never exceeds INT32_MAX and INT64_MAX respectively.
*
* mode is CK_EC_SP for single producer mode, or CK_EC_MP for
* multipler producer mode.
*
* deadline is either NULL, or a `const struct timespec *` that will
* be compared against CLOCK_MONOTONIC.
*
* `void ck_ec_init(ec)`: zero-initializes the event count.
*
* `value ck_ec_value(ec)`: returns the current value of the event
* counter. This read acts as an read (acquire) barrier.
*
* `void ck_ec_inc(ec, mode)`: increments the value of the event
* counter by one. This writes acts as a write (release)
* barrier. Wakes up any waiting thread.
*
* `value ck_ec_add(ec, mode, value)`: increments the event counter by
* `value`, and returns the event counter's previous value. This
* write acts as a write (release) barrier. Wakes up any waiting
* thread.
*
* `int ck_ec_deadline(struct timespec *new_deadline,
* const struct timespec *timeout)`:
* computes a deadline `timeout` away from the current time. If
* timeout is NULL, computes a deadline in the infinite future. The
* resulting deadline is written to `new_deadline`. Returns 0 on
* success, and -1 if clock_gettime failed (without touching errno).
*
* `int ck_ec_wait(ec, value, deadline?)`: waits until the event
* counter's value differs from `value`, or, if `deadline` is
* provided and non-NULL, until the current time is after that
* deadline. Use a deadline with tv_sec = 0 for a non-blocking
* execution. Returns 0 if the event counter has changed, and -1 on
* timeout. This function may spuriously return with a timeout if
* clock_gettime fails.
*
* `int ck_ec_wait_pred(ec, value, pred, data, deadline?)`: waits
* until the event counter's value differs from `value`, or until
* `pred` returns non-zero, or, if `deadline` is provided and
* non-NULL, until the current time is after that deadline. Use a
* deadline with tv_sec = 0 for a non-blocking execution. Returns 0 if
* the event counter has changed, `pred`'s return value if non-zero,
* and -1 on timeout. This function may spuriously return with a
* timeout if clock_gettime fails.
*
* `pred` is always called as `pred(data, iteration_deadline, now)`,
* where `iteration_deadline` is a timespec of the deadline for this
* exponential backoff iteration, and `now` is the current time. If
* `pred` returns a non-zero value, that value is immediately returned
* to the waiter. Otherwise, `pred` is free to modify
* `iteration_deadline` (moving it further in the future is a bad
* idea).
*
* Implementation notes
* ====================
*
* This event count implementation is heavily tied to
* [x86-TSO](https://www.cl.cam.ac.uk/~pes20/weakmemory/cacm.pdf), to
* futexes, and to x86's non-atomic read-modify-write instructions
* (e.g., `add m, r`). Porting it to another architecture or OS will
* likely require a complete redesign, or sacrificing single-producer
* performance. The multiple producer case is trivial: every write to
* the `counter` word is LOCKed. Only the single producer case is
* interesting, due to the way it uses both atomic and non-atomic
* writes to the same memory location.
*
* The reason we can mix atomic and non-atomic writes to the `counter`
* word is that every non-atomic write obviates the need for the
* atomically flipped flag bit: we only use non-atomic writes to
* update the event count, and the atomic flag only informs the
* producer that we would like a futex_wake, because of the update.
* We only require the non-atomic RMW counter update to prevent
* preemption from introducing arbitrarily long worst case delays.
*
* Correctness does not rely on the usual ordering argument: in the
* absence of fences, there is no strict ordering between atomic and
* non-atomic writes.
*
* The key is instead x86-TSO's guarantee that a read is satisfied
* from the most recent buffered write in the local store queue if
* there is one, or from memory if there is no write to that address
* in the store queue.
*
* x86-TSO's constraint on reads suffices to guarantee that the
* producer will never forget about a counter update. If the last
* update is still queued, the new update will be based on the queued
* value. Otherwise, the new update will be based on the value in
* memory, which may or may not have had its flag flipped. In either
* case, the value of the counter (modulo flag) is correct.
*
* When the producer forwards the counter's value from its store
* queue, the new update might not preserve a flag flip. Any waiter
* thus has to check from time to time to determine if it wasn't
* woken up because the flag bit was silently cleared.
*
* In reality, the store queue in x86-TSO stands for in-flight
* instructions in the chip's out-of-order backend. In the vast
* majority of cases, instructions will only remain in flight for a
* few hundred or thousand of cycles. That's why ck_ec_wait spins on
* the `counter` word for 100 iterations after flipping its flag bit:
* if the counter hasn't changed after that many iterations, it is
* very likely that the producer's next counter update will observe
* the flag flip.
*
* That's still not a hard guarantee of correctness. Conservatively,
* we can expect that no instruction will remain in flight for more
* than 1 second... if only because some interrupt will have forced
* the chip to store its architectural state in memory, at which point
* an instruction is either fully retired or rolled back. Interrupts,
* particularly the pre-emption timer, are why single-producer updates
* must happen in a single non-atomic read-modify-write instruction.
* Having a single instruction as the critical section means we only
* have to consider the worst-case execution time for that
* instruction. That's easier than doing the same for a pair of
* instructions, which an unlucky pre-emption could delay for
* arbitrarily long.
*
* Thus, after a few iterations of a spin loop, ck_ec_wait enters an
* exponential backoff loop, where each "sleep" is instead a
* futex_wait. The backoff is only necessary to handle rare cases
* where the flag flip was overwritten after the spin
* loop. Eventually, more than one second will have elapsed since the
* flag flip, and the sleep timeout becomes infinite: since the flag
* bit has been set for much longer than the time for which an
* instruction may remain in flight, the flag will definitely be
* observed at the next counter update.
*
* The 64 bit ck_ec_wait pulls another trick: futexes only handle 32
* bit ints, so we must treat the 64 bit counter's low 32 bits as an
* int in futex_wait. That's a bit dodgy, but fine in practice, given
* that the OS's futex code will always read whatever value is
* currently in memory: even if the producer thread were to wait on
* its own event count, the syscall and ring transition would empty
* the store queue (the out-of-order execution backend).
*
* Finally, what happens when the producer is migrated to another core
* or otherwise pre-empted? Migration must already incur a barrier, so
* that thread always sees its own writes, so that's safe. As for
* pre-emption, that requires storing the architectural state, which
* means every instruction must either be executed fully or not at
* all when pre-emption happens.
*/
#ifndef CK_EC_H
#define CK_EC_H
#include <ck_cc.h>
#include <ck_pr.h>
#include <ck_limits.h>
#include <ck_stdint.h>
#include <ck_stddef.h>
#include <sys/time.h>
enum ck_ec_mode {
CK_EC_SP = 0, /* Single producer. */
CK_EC_MP = 1 /* Multiple producers. */
};
#define CK_EC_INITIALIZER { .counter = 0 }
struct ck_ec32 {
/* Flag is "sign" bit, value in bits 0:30. */
uint32_t counter;
};
struct ck_ec64 {
/*
* Flag is bottom bit, value in bits 1:63. Eventcount only
* works on x86-64 (i.e., little endian), so the futex int
* lies in the first 4 (bottom) bytes.
*/
uint64_t counter;
};
static void ck_ec32_init(struct ck_ec32 *ec);
static void ck_ec64_init(struct ck_ec64 *ec);
#if __STDC_VERSION__ >= 201112L
#define ck_ec_init(EC) \
(_Generic(*(EC), \
struct ck_ec32 : ck_ec32_init, \
struct ck_ec64 : ck_ec64_init)((EC)))
#endif
/*
* Returns the counter value in the event count. The value is at most
* INT32_MAX.
*/
static uint32_t ck_ec32_value(const struct ck_ec32* ec);
/*
* Returns the counter value in the event count. The value is at most
* INT64_MAX.
*/
static uint64_t ck_ec64_value(const struct ck_ec64* ec);
#if __STDC_VERSION__ >= 201112L
#define ck_ec_value(EC) \
(_Generic(*(EC), \
struct ck_ec32 : ck_ec32_value, \
struct ck_ec64 : ck_ec64_value)((EC)))
#endif
/*
* Increments the counter value in the event count by one, and wakes
* up any waiter.
*/
static void ck_ec32_inc(struct ck_ec32 *ec, enum ck_ec_mode mode);
static void ck_ec64_inc(struct ck_ec64 *ec, enum ck_ec_mode mode);
#if __STDC_VERSION__ >= 201112L
#define ck_ec_inc(EC, MODE) \
(_Generic(*(EC), \
struct ck_ec32 : ck_ec32_inc, \
struct ck_ec64 : ck_ec64_inc)((EC), (MODE)))
#endif
/*
* Increments the counter value in the event count by delta, wakes
* up any waiter, and returns the previous counter value.
*/
static uint32_t ck_ec32_add(struct ck_ec32 *ec, enum ck_ec_mode mode,
uint32_t delta);
static uint64_t ck_ec64_add(struct ck_ec64 *ec, enum ck_ec_mode mode,
uint64_t delta);
#if __STDC_VERSION__ >= 201112L
#define ck_ec_add(EC, MODE, DELTA) \
(_Generic(*(EC), \
struct ck_ec32 : ck_ec32_add, \
struct ck_ec64 : ck_ec64_add)((EC), (MODE), (DELTA)))
#endif
/*
* Populates `new_deadline` with a deadline `timeout` in the future.
* Returns 0 on success, and -1 if clock_gettime failed, in which
* case errno is left as is.
*/
int ck_ec_deadline(struct timespec *new_deadline,
const struct timespec *timeout);
/*
* Waits until the counter value in the event count differs from
* old_value, or, if deadline is non-NULL, until CLOCK_MONOTONIC is
* past the deadline.
*
* Returns 0 on success, and -1 on timeout.
*/
static int ck_ec32_wait(struct ck_ec32 *ec, uint32_t old_value,
const struct timespec *deadline);
static int ck_ec64_wait(struct ck_ec64 *ec, uint64_t old_value,
const struct timespec *deadline);
#if __STDC_VERSION__ >= 201112L
#define ck_ec_wait(EC, OLD_VALUE, ...) \
ck_ec_wait_impl((EC), (OLD_VALUE), ##__VA_ARGS__, NULL)
#define ck_ec_wait_impl(EC, OLD_VALUE, DEADLINE, ...) \
(_Generic(*EC, \
struct ck_ec32 : ck_ec32_wait, \
struct ck_ec64 : ck_ec64_wait)(EC, OLD_VALUE, (DEADLINE)))
#endif
/*
* Waits until the counter value in the event count differs from
* old_value, pred returns non-zero, or, if deadline is non-NULL,
* until CLOCK_MONOTONIC is past the deadline.
*
* Returns 0 on success, -1 on timeout, and the return value of pred
* if it returns non-zero.
*
* A NULL pred represents a function that always returns 0.
*/
static int ck_ec32_wait_pred(struct ck_ec32 *ec, uint32_t old_value,
int (*pred)(void *data,
struct timespec *deadline,
const struct timespec *now),
void *data,
const struct timespec *deadline);
static int ck_ec64_wait_pred(struct ck_ec64 *ec, uint64_t old_value,
int (*pred)(void *data,
struct timespec *deadline,
const struct timespec *now),
void *data,
const struct timespec *deadline);
#if __STDC_VERSION__ >= 201112L
#define ck_ec_wait_pred(EC, OLD_VALUE, PRED, DATA, ...) \
ck_ec_wait_pred_impl((EC), (OLD_VALUE), (PRED), (DATA), \
##__VA_ARGS__, NULL)
#define ck_ec_wait_pred_impl(EC, OLD_VALUE, PRED, DATA, DEADLINE, ...) \
(_Generic(*EC, \
struct ck_ec32 : ck_ec32_wait_pred, \
struct ck_ec64 : ck_ec64_wait_pred) \
(EC, OLD_VALUE, PRED, DATA, (DEADLINE)))
#endif
/*
* Inline implementation details.
*/
CK_CC_INLINE void ck_ec32_init(struct ck_ec32 *ec)
{
ec->counter = 0;
return;
}
CK_CC_INLINE void ck_ec64_init(struct ck_ec64 *ec)
{
ec->counter = 0;
return;
}
CK_CC_INLINE uint32_t ck_ec32_value(const struct ck_ec32* ec)
{
return ck_pr_load_32(&ec->counter) & ~(1UL << 31);
}
CK_CC_INLINE uint64_t ck_ec64_value(const struct ck_ec64 *ec)
{
return ck_pr_load_64(&ec->counter) >> 1;
}
/* Slow path for ck_ec{32,64}_{inc,add} */
void ck_ec32_wake(struct ck_ec32 *ec);
void ck_ec64_wake(struct ck_ec64 *ec);
CK_CC_INLINE void ck_ec32_inc(struct ck_ec32 *ec, enum ck_ec_mode mode)
{
char flagged;
#if !defined(__GNUC__)
# error "ck_ec requires GCC inline assembly extensions."
#elif __GNUC__ >= 6
/*
* We don't want to wake if the sign bit is 0. We do want to
* wake if the sign bit just flipped from 1 to 0. We don't
* care what happens when our increment caused the sign bit to
* flip from 0 to 1 (that's once per 2^31 increment).
*
* This leaves us with four cases:
*
* old sign bit | new sign bit | SF | OF | ZF
* -------------------------------------------
* 0 | 0 | 0 | 0 | ?
* 0 | 1 | 1 | 0 | ?
* 1 | 1 | 1 | 0 | ?
* 1 | 0 | 0 | 0 | 1
*
* In the first case, we don't want to hit ck_ec32_wake. In
* the last two cases, we do want to call ck_ec32_wake. In the
* second case, we don't care, so we arbitrarily choose to
* call ck_ec32_wake.
*
* The "le" condition checks if SF != OF, or ZF == 1, which
* meets our requirements.
*/
#define CK_EC32_INC_ASM(PREFIX) \
__asm__ volatile(PREFIX " addl $1, %0" \
: "+m"(ec->counter), "=@ccle"(flagged) \
:: "cc", "memory")
#else
#define CK_EC32_INC_ASM(PREFIX) \
__asm__ volatile(PREFIX " addl $1, %0; setle %1" \
: "+m"(ec->counter), "=r"(flagged) \
:: "cc", "memory")
#endif
switch (mode) {
case CK_EC_SP:
CK_EC32_INC_ASM("");
break;
case CK_EC_MP:
default:
CK_EC32_INC_ASM("lock");
break;
}
#undef CK_EC32_INC_ASM
if (CK_CC_UNLIKELY(flagged)) {
ck_ec32_wake(ec);
}
return;
}
CK_CC_FORCE_INLINE void ck_ec64_inc(struct ck_ec64 *ec, enum ck_ec_mode mode)
{
/* We always xadd, so there's no special optimization here. */
(void)ck_ec64_add(ec, mode, 1);
return;
}
CK_CC_INLINE uint32_t ck_ec32_add(struct ck_ec32 *ec, enum ck_ec_mode mode,
uint32_t delta)
{
const uint32_t flag_mask = 1U << 31;
uint32_t old = delta;
uint32_t ret;
#define CK_EC32_ADD_ASM(PREFIX) \
__asm__ volatile(PREFIX " xaddl %1, %0" \
: "+m"(ec->counter), "+r"(old) \
:: "cc", "memory")
switch (mode) {
case CK_EC_SP:
CK_EC32_ADD_ASM("");
break;
case CK_EC_MP:
CK_EC32_ADD_ASM("lock");
break;
}
#undef CK_EC32_ADD_ASM
ret = old & ~flag_mask;
/* These two only differ if the flag bit is set. */
if (CK_CC_UNLIKELY(old != ret)) {
ck_ec32_wake(ec);
}
return ret;
}
CK_CC_INLINE uint64_t ck_ec64_add(struct ck_ec64 *ec, enum ck_ec_mode mode,
uint64_t delta)
{
uint64_t old = 2 * delta; /* The low bit is the flag bit. */
uint64_t ret;
#define CK_EC64_ADD_ASM(PREFIX) \
__asm__ volatile(PREFIX " xaddq %1, %0" \
: "+m"(ec->counter), "+r"(old) \
:: "cc", "memory")
switch (mode) {
case CK_EC_SP:
CK_EC64_ADD_ASM("");
break;
case CK_EC_MP:
CK_EC64_ADD_ASM("lock");
break;
}
#undef CK_EC64_ADD_ASM
ret = old >> 1;
if (CK_CC_UNLIKELY(old & 1)) {
ck_ec64_wake(ec);
}
return ret;
}
int ck_ec32_wait_slow(struct ck_ec32 *ec, uint32_t old_value,
const struct timespec *deadline);
int ck_ec64_wait_slow(struct ck_ec64 *ec, uint64_t old_value,
const struct timespec *deadline);
CK_CC_INLINE int ck_ec32_wait(struct ck_ec32 *ec, uint32_t old_value,
const struct timespec *deadline)
{
if (ck_ec32_value(ec) != old_value) {
return 0;
}
return ck_ec32_wait_slow(ec, old_value, deadline);
}
CK_CC_INLINE int ck_ec64_wait(struct ck_ec64 *ec, uint64_t old_value,
const struct timespec *deadline)
{
if (ck_ec64_value(ec) != old_value) {
return 0;
}
return ck_ec64_wait_slow(ec, old_value, deadline);
}
int ck_ec32_wait_pred_slow(struct ck_ec32 *ec, uint32_t old_value,
int (*pred)(void *data,
struct timespec *deadline,
const struct timespec *now),
void *data,
const struct timespec *deadline);
int ck_ec64_wait_pred_slow(struct ck_ec64 *ec, uint64_t old_value,
int (*pred)(void *data,
struct timespec *deadline,
const struct timespec *now),
void *data,
const struct timespec *deadline);
CK_CC_INLINE int ck_ec32_wait_pred(struct ck_ec32 *ec, uint32_t old_value,
int (*pred)(void *data,
struct timespec *deadline,
const struct timespec *now),
void *data,
const struct timespec *deadline)
{
if (ck_ec32_value(ec) != old_value) {
return 0;
}
return ck_ec32_wait_pred_slow(ec, old_value, pred, data, deadline);
}
CK_CC_INLINE int ck_ec64_wait_pred(struct ck_ec64 *ec, uint64_t old_value,
int (*pred)(void *data,
struct timespec *deadline,
const struct timespec *now),
void *data,
const struct timespec *deadline)
{
if (ck_ec64_value(ec) != old_value) {
return 0;
}
return ck_ec64_wait_pred_slow(ec, old_value, pred, data, deadline);
}
#endif /* !CK_EC_H */
Raw
ck_ec_smoke_test.c
#include "ck_ec.h"
#include <assert.h>
#include <pthread.h>
#include <stdio.h>
#include <unistd.h>
#define TIME_MAX (((unsigned long long)((time_t)-1)) / 2)
static void test_update_counter_32(enum ck_ec_mode mode)
{
struct ck_ec32 ec = CK_EC_INITIALIZER;
assert(ck_ec_value(&ec) == 0);
ck_ec_inc(&ec, mode);
assert(ck_ec_value(&ec) == 1);
uint32_t old = ck_ec_add(&ec, mode, 42);
assert(old == 1);
assert(ck_ec_value(&ec) == 43);
return;
}
static void test_update_counter_64(enum ck_ec_mode mode)
{
struct ck_ec64 ec = CK_EC_INITIALIZER;
assert(ck_ec_value(&ec) == 0);
ck_ec_inc(&ec, mode);
assert(ck_ec_value(&ec) == 1);
uint64_t old = ck_ec_add(&ec, mode, 42);
assert(old == 1);
assert(ck_ec_value(&ec) == 43);
return;
}
static void test_deadline(void)
{
struct timespec deadline;
assert(ck_ec_deadline(&deadline, NULL) == 0);
assert(deadline.tv_sec == TIME_MAX);
{
const struct timespec timeout = {
.tv_sec = 1,
.tv_nsec = 1000
};
const struct timespec no_timeout = {
.tv_sec = 0
};
struct timespec now;
assert(ck_ec_deadline(&deadline, &timeout) == 0);
assert(ck_ec_deadline(&now, &no_timeout) == 0);
double now_sec = now.tv_sec + 1e-9 * now.tv_nsec;
double deadline_sec = deadline.tv_sec + 1e-9 * deadline.tv_nsec;
assert(now_sec < deadline_sec);
assert(deadline_sec <= now_sec + 1 + 1000e-9);
}
{
const struct timespec timeout = {
.tv_sec = TIME_MAX - 1,
.tv_nsec = 1000
};
assert(ck_ec_deadline(&deadline, &timeout) == 0);
assert(deadline.tv_sec == TIME_MAX);
}
return;
}
static void test_wait_32(void)
{
struct timespec deadline = { .tv_sec = 0 };
struct ck_ec32 ec;
ck_ec_init(&ec);
assert(ck_ec_value(&ec) == 0);
assert(ck_ec_wait(&ec, 1) == 0);
assert(ck_ec_wait(&ec, 1, NULL) == 0);
assert(ck_ec_wait(&ec, 0, &deadline) == -1);
{
const struct timespec timeout = { .tv_sec = 0 };
assert(ck_ec_deadline(&deadline, &timeout) == 0);
assert(ck_ec_wait(&ec, 0, &deadline) == -1);
}
return;
}
static void test_wait_64(void)
{
struct timespec deadline = { .tv_sec = 0 };
struct ck_ec64 ec;
ck_ec_init(&ec);
assert(ck_ec_value(&ec) == 0);
assert(ck_ec_wait(&ec, 1) == 0);
assert(ck_ec_wait(&ec, 1, NULL) == 0);
assert(ck_ec_wait(&ec, 0, &deadline) == -1);
{
const struct timespec timeout = { .tv_sec = 0 };
assert(ck_ec_deadline(&deadline, &timeout) == 0);
assert(ck_ec_wait(&ec, 0, &deadline) == -1);
}
return;
}
static int pred(void *data,
struct timespec *deadline,
const struct timespec *now)
{
int *count = data;
(void)deadline;
(void)now;
if ((*count)++ < 3) {
return 0;
}
return (*count)++;
}
/* Check that pred's return value is correctly bubbled up. */
static void test_wait_pred_32(void)
{
struct ck_ec32 ec = CK_EC_INITIALIZER;
int count = 0;
assert(ck_ec_wait_pred(&ec, 0, pred, &count) == 4);
assert(count == 5);
return;
}
static int pred2(void *data,
struct timespec *deadline,
const struct timespec *now)
{
int *count = data;
*deadline = *now;
deadline->tv_sec++;
(*count)++;
return 0;
}
/*
* wait_pred_64 is nearly identical to _32. Now check that deadline
* overriding works.
*/
static void test_wait_pred_64(void)
{
const struct timespec timeout = { .tv_sec = 5 };
struct timespec deadline;
struct ck_ec64 ec = CK_EC_INITIALIZER;
int count = 0;
assert(ck_ec_deadline(&deadline, &timeout) == 0);
assert(ck_ec_wait_pred(&ec, 0, pred2, &count, &deadline) == -1);
assert(count == 5);
return;
}
static int woken = 0;
static void *test_threaded_32_waiter(void *data)
{
struct ck_ec32 *ec = data;
ck_ec_wait(ec, 0);
ck_pr_store_int(&woken, 1);
return NULL;
}
static void test_threaded_inc_32(enum ck_ec_mode mode)
{
struct ck_ec32 ec = CK_EC_INITIALIZER;
pthread_t waiter;
ck_pr_store_int(&woken, 0);
pthread_create(&waiter, NULL, test_threaded_32_waiter, &ec);
usleep(10000);
assert(ck_pr_load_int(&woken) == 0);
ck_ec_inc(&ec, mode);
pthread_join(waiter, NULL);
assert(ck_pr_load_int(&woken) == 1);
return;
}
static void test_threaded_add_32(enum ck_ec_mode mode)
{
struct ck_ec32 ec = CK_EC_INITIALIZER;
pthread_t waiter;
ck_pr_store_int(&woken, 0);
pthread_create(&waiter, NULL, test_threaded_32_waiter, &ec);
usleep(10000);
assert(ck_pr_load_int(&woken) == 0);
ck_ec_add(&ec, mode, 4);
pthread_join(waiter, NULL);
assert(ck_pr_load_int(&woken) == 1);
return;
}
static void *test_threaded_64_waiter(void *data)
{
struct ck_ec64 *ec = data;
ck_ec_wait(ec, 0);
ck_pr_store_int(&woken, 1);
return NULL;
}
static void test_threaded_inc_64(enum ck_ec_mode mode)
{
struct ck_ec64 ec = CK_EC_INITIALIZER;
pthread_t waiter;
ck_pr_store_int(&woken, 0);
pthread_create(&waiter, NULL, test_threaded_64_waiter, &ec);
usleep(10000);
assert(ck_pr_load_int(&woken) == 0);
ck_ec_inc(&ec, mode);
pthread_join(waiter, NULL);
assert(ck_pr_load_int(&woken) == 1);
return;
}
static void test_threaded_add_64(enum ck_ec_mode mode)
{
struct ck_ec64 ec = CK_EC_INITIALIZER;
pthread_t waiter;
ck_pr_store_int(&woken, 0);
pthread_create(&waiter, NULL, test_threaded_64_waiter, &ec);
usleep(10000);
assert(ck_pr_load_int(&woken) == 0);
ck_ec_add(&ec, mode, 4);
pthread_join(waiter, NULL);
assert(ck_pr_load_int(&woken) == 1);
return;
}
int main ()
{
test_update_counter_32(CK_EC_SP);
test_update_counter_64(CK_EC_SP);
printf("test_update_counter SP passed.\n");
test_update_counter_32(CK_EC_MP);
test_update_counter_64(CK_EC_MP);
printf("test_update_counter MP passed.\n");
test_deadline();
printf("test_deadline passed.\n");
test_wait_32();
test_wait_64();
printf("test_wait passed.\n");
test_wait_pred_32();
test_wait_pred_64();
printf("test_wait_pred passed.\n");
test_threaded_inc_32(CK_EC_SP);
test_threaded_add_32(CK_EC_SP);
test_threaded_inc_64(CK_EC_SP);
test_threaded_add_64(CK_EC_SP);
printf("test_threaded SP passed.\n");
test_threaded_inc_32(CK_EC_MP);
test_threaded_add_32(CK_EC_MP);
test_threaded_inc_64(CK_EC_MP);
test_threaded_add_64(CK_EC_MP);
printf("test_threaded MP passed.\n");
return 0;
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment