Why does this have to be so hard? You don't need to worry about EAGAIN or the pipe buffer or whatever. Here's a simple waitable semaphore that never fills the p

Why does this have to be so hard? You don't need to worry about EAGAIN or the pipe buffer or whatever. Here's a simple waitable semaphore that never fills the pipe. struct waitsem { pthread_mutex_t down_lock; int count; int fd_r; int fd_w; }; int waitsem_init(struct waitsem* ws) { memset(ws, 0, sizeof (*ws)); if (pthread_mutex_init(&ws->down_lock)) { return -1; } int pipefd[2]; // Instead use pipe2 and O_CLOEXEC if your OS supports it if (pipe(pipefd)) { posix_mutex_destroy(&ws->down_lock); return -1; } ws->fd_r = pipefd[0]; ws->fd_w = pipefd[1]; return 0; } // This function is async signal safe. void waitsem_up(struct waitsem* ws) { if (!atomic_fetch_add_explict(&ws->count, 1, memory_order_relaxed)) { // We transitioned 0->1, so it's our job to write a byte to the // pipe and wake up concurrent waiters. if (TEMP_FAILURE_RETRY(write(ws->fd_w, ws, 1)) != 1) { abort(); // POSIX prohibits failure. } } } // When ws->fd_r is readable, ws can probably be downed. Does not // block. (To perform a blocking down, wait on ws->fd_r.) Not async // signal safe. bool waitsem_try_down(struct waitsem* ws) { // N.B. We need ws->down_lock to bound the number of bytes in the // pipe. If we didn't take it, an arbitrary number of concurrent // waitsem_try_down callers could stall between decrementing // ws->counter and reading the pipe byte, each allowing an // interleaving waitsem_up to write a byte and, in theory, // eventually overflow the pipe buffer. You might think "well, // waitsem_up() would just block until one of the concurrent // waitsem_try_down functions got around to finishing its read", but // you'd be wrong, since waitsem_up() can run on a thread also // running waitsem_try_down and deadlock! With the mutex in place, // we can have at most two bytes in the pipe at one time, well below // the POSIX minimum PIPE_BUF of 512. // N.B. nothing in this function sleeps, so posix_mutex_lock doesn't // actually block for any meaningful time. Thus, this function // behaves as a non-blocking function even though read(2) can, in // general, sleep. if (posix_mutex_lock(&ws->down_lock)) { abort(); // POSIX prohibits failure } int old_counter = atomic_load_explicit(&ws->count, memory_order_relaxed); bool success = false; do { assert(old_counter >= 0); if (old_counter == 0) { goto out; } } while (!atomic_compare_exchange_weak_explicit(&ws->count, &old_counter, old_counter - 1, memory_order_relaxed, memory_order_relaxed)); if (old_counter == 1) { // We transitioned 1->0, so it's our job to read the byte out of // the pipe. We know the byte is in the pipe, so we don't actually // block here. char dummy; if (TEMP_FAILURE_RETRY(read(&ws->fd_r, &dummy, 1)) != 1) abort(); // POSIX prohibits failure. } success = true; out: if (posix_mutex_unlock(&ws->down_lock)) { abort(); // POSIX prohibits failure. } return success; } // Blocks until we can successfully down WS. void waitsem_down(struct waitsem* ws) { while (!waitsem_try_down(ws)) { struct pollfd pfd = {ws->fd_r, POLL_IN, 0}; poll(&pfd, 1, -1); } }