Single reader - multiple writer waitable lock-free unreliably ordered stack

Question

The main body of code is located here. In the code linted_error is a platform specific type for error codes. As well, as an optimization the code can use Linux futexes.

• linted/mem.h is just a tiny wrapper around malloc.
• linted/sched.h does a bunch of stuff but is only used in this code to wrap platform specific pause instructions for spinlocks.

Otherwise, there are no other unportabilities.

I think the main problem with this implementation is that it doesn't use memory fences correctly. Am I right that I only need to use memory fences with memory_order_relaxed memory ordering. If so, then it should be possible to convert to usage of memory_order_relaxed easily and possibly obtain a small speedup right?

I also think it may be possible to use memory_order_relaxed without memory fences for the trigger code as triggers are supposed to be unreliable anyway.

Note that for my use case it doesn't matter what order things are inserted onto the stack.

The stack:

 /*
  * Copyright 2015 Steven Stewart-Gallus
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  * implied.  See the License for the specific language governing
  * permissions and limitations under the License.
  */
 #include "config.h"

 #include "linted/stack.h"

 #include "linted/error.h"
 #include "linted/mem.h"
 #include "linted/node.h"
 #include "linted/sched.h"
 #include "linted/trigger.h"

 #include <errno.h>
 #include <stdatomic.h>
 #include <stdint.h>

 typedef _Atomic(struct linted_node *) atomic_node;

 struct linted_stack {
    atomic_node inbox;
    struct linted_node *outbox;
    struct linted_trigger inbox_filled;
 };

 static void refresh_node(struct linted_node *node)
 {
    node->next = 0;
 }

 linted_error linted_stack_create(struct linted_stack **stackp)
 {
    linted_error err = 0;

    struct linted_stack *stack;
    {
        void *xx;
        err = linted_mem_alloc(&xx, sizeof *stack);
        if (err != 0)
            return err;
        stack = xx;
    }

    atomic_node ptr = ATOMIC_VAR_INIT((void *)0);
    stack->inbox = ptr;
    stack->outbox = 0;

    linted_trigger_create(&stack->inbox_filled);

    *stackp = stack;

    return 0;
 }

 void linted_stack_destroy(struct linted_stack *stack)
 {
    linted_trigger_destroy(&stack->inbox_filled);

    linted_mem_free(stack);
 }

 void linted_stack_send(struct linted_stack *stack,
                        struct linted_node *node)
 {
    for (;;) {
        struct linted_node *next = atomic_load_explicit(
            &stack->inbox, memory_order_acquire);

        node->next = next;

        atomic_thread_fence(memory_order_release);

        if (atomic_compare_exchange_weak_explicit(
                &stack->inbox, &next, node,
                memory_order_acq_rel, memory_order_acquire)) {
            break;
        }

        linted_sched_light_yield();
    }

    linted_trigger_set(&stack->inbox_filled);
 }

 void linted_stack_recv(struct linted_stack *stack,
                        struct linted_node **nodep)
 {
    struct linted_node *ret = stack->outbox;
    if (ret != 0) {
        stack->outbox = ret->next;
        goto give_node;
    }

    for (;;) {
        for (uint_fast8_t ii = 0U; ii < 20U; ++ii) {
            ret = atomic_exchange_explicit(
                &stack->inbox, 0, memory_order_acq_rel);
            if (ret != 0)
                goto put_on_outbox;

            linted_sched_light_yield();
        }

        linted_trigger_wait(&stack->inbox_filled);
    }

 put_on_outbox:
    atomic_thread_fence(memory_order_acquire);

    for (struct linted_node *node = ret->next;;) {
        if (0 == node)
            break;

        struct linted_node *next = node->next;

        node->next = stack->outbox;
        stack->outbox = node;

        node = next;
    }

 give_node:
    refresh_node(ret);

    *nodep = ret;
 }

 linted_error linted_stack_try_recv(struct linted_stack *stack,
                                    struct linted_node **nodep)
 {
    struct linted_node *ret = stack->outbox;
    if (ret != 0) {
        stack->outbox = ret->next;
        goto give_node;
    }

    ret = atomic_exchange_explicit(&stack->inbox, 0,
                                   memory_order_acq_rel);
    if (0 == ret)
        return EAGAIN;

    atomic_thread_fence(memory_order_acquire);

    for (struct linted_node *node = ret->next;;) {
        if (0 == node)
            break;

        struct linted_node *next = node->next;

        node->next = stack->outbox;
        stack->outbox = node;

        node = next;
    }

 give_node:
    refresh_node(ret);

    *nodep = ret;
    return 0;
 }

linted/node.h:

 /*
  * Copyright 2015 Steven Stewart-Gallus
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  * implied.  See the License for the specific language governing
  * permissions and limitations under the License.
  */
 #ifndef LINTED_NODE_H
 #define LINTED_NODE_H

 struct linted_node {
    struct linted_node *next;
 };

 #endif /* LINTED_NODE_H */

linted/mem.h:

 /*
  * Copyright 2014, 2015 Steven Stewart-Gallus
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  * implied.  See the License for the specific language governing
  * permissions and limitations under the License.
  */
 #ifndef LINTED_MEM_H
 #define LINTED_MEM_H

 #include "linted/error.h"

 #include <stddef.h>

 /**
  * @file
  *
  * Allocates memory.
  */

 /* For consistent platform behaviour we return the null pointer on
  * zero sized allocations.
  */

 static inline linted_error
 linted_mem_safe_multiply(size_t nmemb, size_t size, size_t *resultp)
 {
    if (size > 0U && ((size_t)-1) / size < nmemb)
        return LINTED_ERROR_OUT_OF_MEMORY;

    *resultp = nmemb *size;
    return 0;
 }

 static inline linted_error linted_mem_alloc(void **memp, size_t size)
 {
    extern void *malloc(size_t size);

    void *memory;
    if (0U == size) {
        memory = 0;
    } else {
        memory = malloc(size);
        if (0 == memory)
            return LINTED_ERROR_OUT_OF_MEMORY;
    }

    *memp = memory;
    return 0;
 }

 static inline linted_error
 linted_mem_alloc_array(void **memp, size_t nmemb, size_t size)
 {
    extern void *malloc(size_t size);

    linted_error err;

    void *memory;
    if (0U == nmemb || 0U == size) {
        memory = 0;
        goto store_mem;
    }

    size_t total;
    err = linted_mem_safe_multiply(nmemb, size, &total);
    if (err != 0)
        return err;

    memory = malloc(total);
    if (0 == memory)
        return LINTED_ERROR_OUT_OF_MEMORY;

 store_mem:
    *memp = memory;
    return 0;
 }

 static inline linted_error linted_mem_alloc_zeroed(void **memp,
                                                    size_t size)
 {
    extern void *calloc(size_t nmemb, size_t size);

    void *memory;
    if (0U == size) {
        memory = 0;
    } else {
        memory = calloc(1U, size);
        if (0 == memory)
            return LINTED_ERROR_OUT_OF_MEMORY;
    }

    *memp = memory;
    return 0;
 }

 static inline linted_error
 linted_mem_alloc_array_zeroed(void **memp, size_t nmemb, size_t size)
 {
    extern void *calloc(size_t nmemb, size_t size);

    void *memory;
    if (0U == nmemb || 0U == size) {
        memory = 0;
    } else {
        memory = calloc(nmemb, size);
        if (0 == memory)
            return LINTED_ERROR_OUT_OF_MEMORY;
    }
    *memp = memory;
    return 0;
 }

 static inline linted_error linted_mem_realloc(void **memp, void *memory,
                                               size_t new_size)
 {
    extern void *realloc(void *ptr, size_t size);
    extern void free(void *ptr);

    void *new_memory;
    if (0U == new_size) {
        free(memory);
        new_memory = 0;
    } else {
        new_memory = realloc(memory, new_size);
        if (0 == new_memory)
            return LINTED_ERROR_OUT_OF_MEMORY;
    }

    *memp = new_memory;
    return 0;
 }

 static inline linted_error linted_mem_realloc_array(void **memp,
                                                     void *memory,
                                                     size_t nmemb,
                                                     size_t size)
 {
    extern void *realloc(void *ptr, size_t size);
    extern void free(void *ptr);

    linted_error err = 0;

    void *new_memory;
    if (0U == nmemb || 0U == size) {
        free(memory);
        new_memory = 0;
        goto store_mem;
    }

    size_t total;
    err = linted_mem_safe_multiply(nmemb, size, &total);
    if (err != 0)
        return err;

    new_memory = realloc(memory, total);
    if (0 == new_memory)
        return LINTED_ERROR_OUT_OF_MEMORY;

 store_mem:
    *memp = new_memory;
    return 0;
 }

 static inline void linted_mem_free(void *memory)
 {
    extern void free(void *ptr);

    /* This is primarily for making debugging easier and not for
     * any sort of optimization */
    if (memory != 0)
        free(memory);
 }

 #endif /* LINTED_MEM_H */

linted/trigger.h:

 /*
  * Copyright 2015 Steven Stewart-Gallus
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  * implied.  See the License for the specific language governing
  * permissions and limitations under the License.
  */
 #ifndef LINTED_TRIGGER_H
 #define LINTED_TRIGGER_H

 #include <stdatomic.h>

 struct linted_trigger {
    atomic_int _triggered;
 };

 void linted_trigger_create(struct linted_trigger *trigger);
 void linted_trigger_destroy(struct linted_trigger *trigger);

 void linted_trigger_set(struct linted_trigger *trigger);
 void linted_trigger_wait(struct linted_trigger *trigger);

 #endif /* LINTED_TRIGGER_H */

The trigger code is below. Triggers are unreliable flag variables for waiting on.

 /*
  * Copyright 2015 Steven Stewart-Gallus
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  * implied.  See the License for the specific language governing
  * permissions and limitations under the License.
  */
 #define _GNU_SOURCE

 #include "config.h"

 #include "linted/trigger.h"

 #include "linted/error.h"
 #include "linted/sched.h"

 #include <errno.h>
 #include <stdatomic.h>
 #include <time.h>
 #include <unistd.h>

 #if defined HAVE_POSIX_API
 #include <syscall.h>

 #include <linux/futex.h>
 #endif

 static linted_error wait_until_different(atomic_int const *uaddr,
                                          int val);
 static linted_error hint_wakeup(atomic_int const *uaddr);

 void linted_trigger_create(struct linted_trigger *trigger)
 {
    atomic_int zero = ATOMIC_VAR_INIT(0);
    trigger->_triggered = zero;
 }

 void linted_trigger_destroy(struct linted_trigger *trigger)
 {
 }

 void linted_trigger_set(struct linted_trigger *trigger)
 {
    atomic_store_explicit(&trigger->_triggered, 1,
                          memory_order_release);
    hint_wakeup(&trigger->_triggered);
 }

 void linted_trigger_wait(struct linted_trigger *trigger)
 {
    wait_until_different(&trigger->_triggered, 0);

    atomic_store_explicit(&trigger->_triggered, 0,
                          memory_order_release);
 }

 #if defined HAVE_POSIX_API
 static linted_error futex_wait(atomic_int const *uaddr, int val,
                                struct timespec const *timeout);
 static linted_error futex_wake(unsigned *restrict wokeupp,
                                atomic_int const *uaddr, int val);

 static linted_error wait_until_different(atomic_int const *uaddr,
                                          int val)
 {
    return futex_wait(uaddr, val, NULL);
 }

 static linted_error hint_wakeup(atomic_int const *uaddr)
 {
    return futex_wake(NULL, uaddr, 1);
 }
 #else
 static linted_error wait_until_different(atomic_int const *uaddr,
                                          int val)
 {
    for (;;) {
        if (atomic_load_explicit(uaddr, memory_order_acquire) !=
            val)
            return 0;
        linted_sched_light_yield();
    }
 }

 static linted_error hint_wakeup(atomic_int const *uaddr)
 {
    return 0;
 }
 #endif

 #if defined HAVE_POSIX_API
 static linted_error futex_wait(atomic_int const *uaddr, int val,
                                struct timespec const *timeout)
 {
    int xx =
        syscall(__NR_futex, (intptr_t)uaddr, (intptr_t)FUTEX_WAIT,
                (intptr_t)val, (intptr_t)timeout);
    if (xx < 0) {
        return errno;
    }

    return 0;
 }

 static linted_error futex_wake(unsigned *restrict wokeupp,
                                atomic_int const *uaddr, int val)
 {
    int xx = syscall(__NR_futex, (intptr_t)uaddr,
                     (intptr_t)FUTEX_WAKE, (intptr_t)val);
    if (xx < 0) {
        return errno;
    }

    if (wokeupp != NULL) {
        *wokeupp = xx;
    }
    return 0;
 }
 #endif

Answer 1

Order of pops is strange

If I am reading your code correctly, your pop ordering will be different from the push ordering. Suppose we push 1 2 3 4 5 in that order. Your inbox will look like:

5 -> 4 -> 3 -> 2 -> 1

So far so good. But then when we pop for the first time, this is what happens:

ret = 5
outbox = 1 -> 2 -> 3 -> 4

Then the next time we pop:

ret = 1
outbox = 2 -> 3 -> 4

So the full sequence of pops becomes: 5 1 2 3 4 when it should be 5 4 3 2 1. Thus, I don't think you can call this a stack because it doesn't behave like one. Note that this is all with only only thread, so it isn't a multithreading issue.

Furthermore, if additional pushes arrive while there are still items in the outbox, the outbox will be emptied before any new items will be popped. I'm not sure if you intended to do this (because you called your stack "unreliably ordered"). I would think that a stack should always pop the newest item off first, though.

Potential fix

For the first problem, you could simply change the put_on_outbox code from this:

 put_on_outbox:
    atomic_thread_fence(memory_order_acquire);

    for (struct linted_node *node = ret->next;;) {
        if (0 == node)
            break;

        struct linted_node *next = node->next;

        node->next = stack->outbox;
        stack->outbox = node;

        node = next;
    }

to this:

 put_on_outbox:
    atomic_thread_fence(memory_order_acquire);
    stack->outbox = ret->next;

Note that when arriving at put_on_outbox:, stack->outbox is guaranteed to be NULL and ret is guaranteed to be non-NULL.

Answer 2

Your futex_wake claims to return a value of type linted_error, but it's actually returning errno, which doesn't necessarily correspond to one of the enumerator values of linted_error, does it? In particular, it could return ENFILE (23), which I don't think your linted_error code handles correctly. I would probably try to deal with this by writing a function linted_error linted_error_from_errno(int) that Does The Right Thing no matter what unsanitized errno value you pass in.

---

The assignment stack->inbox = ptr; is dangerously misleading. Unless I'm mistaken, it corresponds to two atomic operations: first "atomically load ptr into a register t1", and second "atomically store t1 into stack->inbox." If I were you, I'd definitely write this as two separate statements. It doesn't actually matter in this case AFAICT, but since you're apparently trying to come up with best practices for how to write secure and bug-free code... "multiple atomic operations in a single statement" is way high up on my list of things never ever to do.

Alternatively, local variable ptr doesn't really need to be _Atomic. You could just set stack->inbox = 0; without introducing that variable at all.

---

In linted_stack_send, I don't really understand why you're yielding at the bottom of the loop and then reloading the value of inbox at the top of the loop. The only way the CAS operation could have failed is if someone else snuck in and changed the value of inbox very recently, and if that happens, I would think that the last thing you'd want to do would be to yield and allow further threads to sneak in! I'd rather loop around as fast as possible, using the new value that I just got from the CAS, and try modifying inbox again as quickly as possible.

IOW, I'd do this instead:

void linted_stack_send(struct linted_stack *stack,
                       struct linted_node *node)
{
    node->next = stack->inbox;
    while (!atomic_compare_exchange_weak(&stack->inbox, &node->next, node)) { }
    linted_trigger_set(&stack->inbox_filled);
}

But I could totally be missing something here. If there's a good rationale for "yield and load again later", I'd definitely like to know it.

(No comment on your explicit memory orders, or whether my preferred loop above could be safely used with any memory orders besides seq_cst. I'm not qualified to talk about explicit memory orders.)

---

It's a bit scary that you use the same naming convention for linted_stack_destroy and linted_trigger_destroy, even though the former means "destroy and free", whereas the latter means "destroy and don't free". If I were you, I'd seriously consider refactoring my whole library to either

• always "destroy and don't free", and leave memory allocation entirely up to the user; or

• come up with a different naming convention for the "allocate" and "free" versions — e.g., let foo_construct mean "initialize a foo in-place without allocation" and foo_alloc mean "malloc and then call foo_init", and likewise let foo_destruct mean "destroy a foo in-place without deallocation" and foo_free mean "call foo_destruct and then free".

---

Is trigger exactly the same thing as a "condition variable"? If so, consider renaming it.

---

No particular comment on the correctness of the stack implementation. I didn't find any obvious bugs. :)