source: mainline/uspace/lib/c/generic/thread/mpsc.c@ bd41ac52

/*
 * Copyright (c) 2018 CZ.NIC, z.s.p.o.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 * - Redistributions of source code must retain the above copyright
 *   notice, this list of conditions and the following disclaimer.
 * - 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.
 * - The name of the author may not be used to endorse or promote products
 *   derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``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 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.
 */

/*
 * Authors:
 *      Jiří Zárevúcky (jzr) <zarevucky.jiri@gmail.com>
 */

#include <fibril.h>
#include <fibril_synch.h>
#include <mem.h>
#include <stdlib.h>

#include "../private/fibril.h"

/*
 * A multi-producer, single-consumer concurrent FIFO channel with unlimited
 * buffering.
 *
 * The current implementation is based on the super simple two-lock queue
 * by Michael and Scott
 * (http://www.cs.rochester.edu/~scott/papers/1996_PODC_queues.pdf)
 *
 * The original algorithm uses one lock on each side. Since this queue is
 * single-consumer, we only use the tail lock.
 */

typedef struct mpsc_node mpsc_node_t;

struct mpsc {
        size_t elem_size;
        fibril_rmutex_t t_lock;
        mpsc_node_t *head;
        mpsc_node_t *tail;
        mpsc_node_t *close_node;
        fibril_event_t event;
};

struct mpsc_node {
        mpsc_node_t *next;
        unsigned char data[];
};

mpsc_t *mpsc_create(size_t elem_size)
{
        mpsc_t *q = calloc(1, sizeof(mpsc_t));
        mpsc_node_t *n = calloc(1, sizeof(mpsc_node_t) + elem_size);
        mpsc_node_t *c = calloc(1, sizeof(mpsc_node_t) + elem_size);

        if (!q || !n || !c) {
                free(q);
                free(n);
                free(c);
                return NULL;
        }

        q->elem_size = elem_size;
        fibril_rmutex_initialize(&q->t_lock);
        q->head = q->tail = n;
        q->close_node = c;
        return q;
}

void mpsc_destroy(mpsc_t *q)
{
        mpsc_node_t *n = q->head;
        mpsc_node_t *next = NULL;
        while (n != NULL) {
                next = n->next;
                free(n);
                n = next;
        }

        // TODO: fibril_rmutex_destroy()

        free(q);
}

static errno_t _mpsc_push(mpsc_t *q, mpsc_node_t *n)
{
        fibril_rmutex_lock(&q->t_lock);

        if (q->tail == q->close_node) {
                fibril_rmutex_unlock(&q->t_lock);
                return EINVAL;
        }

        __atomic_store_n(&q->tail->next, n, __ATOMIC_RELEASE);
        q->tail = n;

        fibril_rmutex_unlock(&q->t_lock);

        fibril_notify(&q->event);
        return EOK;
}

/**
 * Send data on the channel.
 * The length of data is equal to the `elem_size` value set in `mpsc_create`.
 *
 * This function is safe for use under restricted mutex lock.
 *
 * @return ENOMEM if allocation failed, EINVAL if the queue is closed.
 */
errno_t mpsc_send(mpsc_t *q, const void *b)
{
        mpsc_node_t *n = malloc(sizeof(mpsc_node_t) + q->elem_size);
        if (!n)
                return ENOMEM;

        n->next = NULL;
        memcpy(n->data, b, q->elem_size);

        return _mpsc_push(q, n);
}

/**
 * Receive data from the channel.
 *
 * @return ETIMEOUT if deadline expires, ENOENT if the queue is closed and
 * there is no message left in the queue.
 */
errno_t mpsc_receive(mpsc_t *q, void *b, const struct timespec *expires)
{
        mpsc_node_t *n;
        mpsc_node_t *new_head;

        while (true) {
                n = q->head;
                new_head = __atomic_load_n(&n->next, __ATOMIC_ACQUIRE);
                if (new_head)
                        break;

                errno_t rc = fibril_wait_timeout(&q->event, expires);
                if (rc != EOK)
                        return rc;
        }

        if (new_head == q->close_node)
                return ENOENT;

        memcpy(b, new_head->data, q->elem_size);
        q->head = new_head;

        free(n);
        return EOK;
}

/**
 * Close the channel.
 *
 * This function is safe for use under restricted mutex lock.
 */
void mpsc_close(mpsc_t *q)
{
        _mpsc_push(q, q->close_node);
}