Index: uspace/lib/c/generic/async/server.c =================================================================== --- uspace/lib/c/generic/async/server.c (revision 6340b4d2b8d527b31a31396c1e0dedb44bc5f58b) +++ uspace/lib/c/generic/async/server.c (revision 1de92fb061a30c6722110b2a588542d6cc5cdb29) @@ -124,10 +124,4 @@ #define DPRINTF(...) ((void) 0) -/** Call data */ -typedef struct { - link_t link; - ipc_call_t call; -} msg_t; - /* Client connection data */ typedef struct { @@ -156,15 +150,9 @@ client_t *client; - /** Message event. */ - fibril_event_t msg_arrived; - - /** Messages that should be delivered to this fibril. */ - list_t msg_queue; + /** Channel for messages that should be delivered to this fibril. */ + mpsc_t *msg_channel; /** Call data of the opening call. */ ipc_call_t call; - - /** Identification of the closing call. */ - cap_call_handle_t close_chandle; /** Fibril function that will be used to handle the connection. */ @@ -423,12 +411,20 @@ async_client_put(client); + fibril_rmutex_lock(&conn_mutex); + /* * Remove myself from the connection hash table. */ - fibril_rmutex_lock(&conn_mutex); hash_table_remove(&conn_hash_table, &(conn_key_t){ .task_id = fibril_connection->in_task_id, .phone_hash = fibril_connection->in_phone_hash }); + + /* + * Close the channel, if it isn't closed already. + */ + mpsc_t *c = fibril_connection->msg_channel; + mpsc_close(c); + fibril_rmutex_unlock(&conn_mutex); @@ -436,21 +432,12 @@ * Answer all remaining messages with EHANGUP. */ - while (!list_empty(&fibril_connection->msg_queue)) { - msg_t *msg = - list_get_instance(list_first(&fibril_connection->msg_queue), - msg_t, link); - - list_remove(&msg->link); - ipc_answer_0(msg->call.cap_handle, EHANGUP); - free(msg); - } + ipc_call_t call; + while (mpsc_receive(c, &call, NULL) == EOK) + ipc_answer_0(call.cap_handle, EHANGUP); /* - * If the connection was hung-up, answer the last call, - * i.e. IPC_M_PHONE_HUNGUP. + * Clean up memory. */ - if (fibril_connection->close_chandle) - ipc_answer_0(fibril_connection->close_chandle, EOK); - + mpsc_destroy(c); free(fibril_connection); return EOK; @@ -488,7 +475,5 @@ conn->in_task_id = in_task_id; conn->in_phone_hash = in_phone_hash; - conn->msg_arrived = FIBRIL_EVENT_INIT; - list_initialize(&conn->msg_queue); - conn->close_chandle = CAP_NIL; + conn->msg_channel = mpsc_create(sizeof(ipc_call_t)); conn->handler = handler; conn->data = data; @@ -503,4 +488,5 @@ if (conn->fid == 0) { + mpsc_destroy(conn->msg_channel); free(conn); @@ -606,9 +592,10 @@ * @param call Data of the incoming call. * - * @return False if the call doesn't match any connection. - * @return True if the call was passed to the respective connection fibril. - * - */ -static bool route_call(ipc_call_t *call) + * @return EOK if the call was successfully passed to the respective fibril. + * @return ENOENT if the call doesn't match any connection. + * @return Other error code if routing failed for other reasons. + * + */ +static errno_t route_call(ipc_call_t *call) { assert(call); @@ -622,27 +609,22 @@ if (!link) { fibril_rmutex_unlock(&conn_mutex); - return false; + return ENOENT; } connection_t *conn = hash_table_get_inst(link, connection_t, link); - // FIXME: malloc in critical section - msg_t *msg = malloc(sizeof(*msg)); - if (!msg) { - fibril_rmutex_unlock(&conn_mutex); - return false; - } - - msg->call = *call; - list_append(&msg->link, &conn->msg_queue); - - if (IPC_GET_IMETHOD(*call) == IPC_M_PHONE_HUNGUP) - conn->close_chandle = call->cap_handle; + errno_t rc = mpsc_send(conn->msg_channel, call); + + if (IPC_GET_IMETHOD(*call) == IPC_M_PHONE_HUNGUP) { + /* Close the channel, but let the connection fibril answer. */ + mpsc_close(conn->msg_channel); + // FIXME: Ideally, we should be able to discard/answer the + // hungup message here and just close the channel without + // passing it out. Unfortunatelly, somehow that breaks + // handling of CPU exceptions. + } fibril_rmutex_unlock(&conn_mutex); - - /* If the connection fibril is waiting for an event, activate it */ - fibril_notify(&conn->msg_arrived); - return true; + return rc; } @@ -939,13 +921,4 @@ assert(fibril_connection); - /* - * Why doing this? - * GCC 4.1.0 coughs on fibril_connection-> dereference. - * GCC 4.1.1 happilly puts the rdhwr instruction in delay slot. - * I would never expect to find so many errors in - * a compiler. - */ - connection_t *conn = fibril_connection; - struct timeval tv; struct timeval *expires = NULL; @@ -956,40 +929,19 @@ } - fibril_rmutex_lock(&conn_mutex); - - /* If nothing in queue, wait until something arrives */ - while (list_empty(&conn->msg_queue)) { - if (conn->close_chandle) { - /* - * Handle the case when the connection was already - * closed by the client but the server did not notice - * the first IPC_M_PHONE_HUNGUP call and continues to - * call async_get_call_timeout(). Repeat - * IPC_M_PHONE_HUNGUP until the caller notices. - */ - memset(call, 0, sizeof(ipc_call_t)); - IPC_SET_IMETHOD(*call, IPC_M_PHONE_HUNGUP); - fibril_rmutex_unlock(&conn_mutex); - return true; - } - - // TODO: replace with cvar - fibril_rmutex_unlock(&conn_mutex); - - errno_t rc = fibril_wait_timeout(&conn->msg_arrived, expires); - if (rc == ETIMEOUT) - return false; - - fibril_rmutex_lock(&conn_mutex); - } - - msg_t *msg = list_get_instance(list_first(&conn->msg_queue), - msg_t, link); - list_remove(&msg->link); - - *call = msg->call; - free(msg); - - fibril_rmutex_unlock(&conn_mutex); + errno_t rc = mpsc_receive(fibril_connection->msg_channel, + call, expires); + + if (rc == ETIMEOUT) + return false; + + if (rc != EOK) { + /* + * The async_get_call_timeout() interface doesn't support + * propagating errors. Return a null call instead. + */ + + memset(call, 0, sizeof(ipc_call_t)); + } + return true; } @@ -1071,9 +1023,13 @@ /* Try to route the call through the connection hash table */ - if (route_call(call)) + errno_t rc = route_call(call); + if (rc == EOK) return; - /* Unknown call from unknown phone - hang it up */ - ipc_answer_0(call->cap_handle, EHANGUP); + // TODO: Log the error. + + if (call->cap_handle != CAP_NIL) + /* Unknown call from unknown phone - hang it up */ + ipc_answer_0(call->cap_handle, EHANGUP); } Index: uspace/lib/c/generic/thread/atomic.c =================================================================== --- uspace/lib/c/generic/thread/atomic.c (revision 6340b4d2b8d527b31a31396c1e0dedb44bc5f58b) +++ uspace/lib/c/generic/thread/atomic.c (revision 1de92fb061a30c6722110b2a588542d6cc5cdb29) @@ -36,4 +36,10 @@ */ +void __sync_synchronize(void) +{ + // FIXME: Full memory barrier. We need a syscall for this. + // Should we implement this or is empty definition ok here? +} + unsigned __sync_add_and_fetch_4(volatile void *vptr, unsigned val) { Index: uspace/lib/c/generic/thread/fibril.c =================================================================== --- uspace/lib/c/generic/thread/fibril.c (revision 6340b4d2b8d527b31a31396c1e0dedb44bc5f58b) +++ uspace/lib/c/generic/thread/fibril.c (revision 1de92fb061a30c6722110b2a588542d6cc5cdb29) @@ -723,4 +723,10 @@ } +/** + * Wake up the fibril waiting for the given event. + * Up to one wakeup is remembered if the fibril is not currently waiting. + * + * This function is safe for use under restricted mutex lock. + */ void fibril_notify(fibril_event_t *event) { Index: uspace/lib/c/generic/thread/mpsc.c =================================================================== --- uspace/lib/c/generic/thread/mpsc.c (revision 1de92fb061a30c6722110b2a588542d6cc5cdb29) +++ uspace/lib/c/generic/thread/mpsc.c (revision 1de92fb061a30c6722110b2a588542d6cc5cdb29) @@ -0,0 +1,181 @@ +/* + * 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) + */ + +#include +#include +#include +#include + +/* + * 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 timeval *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); +} +