diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-03-09 13:19:22 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-03-09 13:19:22 +0000 |
commit | c21c3b0befeb46a51b6bf3758ffa30813bea0ff0 (patch) | |
tree | 9754ff1ca740f6346cf8483ec915d4054bc5da2d /web/server/h2o/libh2o/lib/common/multithread.c | |
parent | Adding upstream version 1.43.2. (diff) | |
download | netdata-upstream/1.44.3.tar.xz netdata-upstream/1.44.3.zip |
Adding upstream version 1.44.3.upstream/1.44.3
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to '')
-rw-r--r-- | web/server/h2o/libh2o/lib/common/multithread.c | 274 |
1 files changed, 274 insertions, 0 deletions
diff --git a/web/server/h2o/libh2o/lib/common/multithread.c b/web/server/h2o/libh2o/lib/common/multithread.c new file mode 100644 index 000000000..b4e8ba836 --- /dev/null +++ b/web/server/h2o/libh2o/lib/common/multithread.c @@ -0,0 +1,274 @@ +/* + * Copyright (c) 2015-2016 DeNA Co., Ltd., Kazuho Oku, Tatsuhiko Kubo, + * Chul-Woong Yang + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + */ +#include <assert.h> +#include <pthread.h> +#include "cloexec.h" +#include "h2o/multithread.h" + +struct st_h2o_multithread_queue_t { +#if H2O_USE_LIBUV + uv_async_t async; +#else + struct { + int write; + h2o_socket_t *read; + } async; +#endif + pthread_mutex_t mutex; + struct { + h2o_linklist_t active; + h2o_linklist_t inactive; + } receivers; +}; + +static void queue_cb(h2o_multithread_queue_t *queue) +{ + pthread_mutex_lock(&queue->mutex); + + while (!h2o_linklist_is_empty(&queue->receivers.active)) { + h2o_multithread_receiver_t *receiver = + H2O_STRUCT_FROM_MEMBER(h2o_multithread_receiver_t, _link, queue->receivers.active.next); + /* detach all the messages from the receiver */ + h2o_linklist_t messages; + h2o_linklist_init_anchor(&messages); + h2o_linklist_insert_list(&messages, &receiver->_messages); + /* relink the receiver to the inactive list */ + h2o_linklist_unlink(&receiver->_link); + h2o_linklist_insert(&queue->receivers.inactive, &receiver->_link); + + /* dispatch the messages */ + pthread_mutex_unlock(&queue->mutex); + receiver->cb(receiver, &messages); + assert(h2o_linklist_is_empty(&messages)); + pthread_mutex_lock(&queue->mutex); + } + + pthread_mutex_unlock(&queue->mutex); +} + +#ifdef H2O_NO_64BIT_ATOMICS +pthread_mutex_t h2o_conn_id_mutex = PTHREAD_MUTEX_INITIALIZER; +#endif + +#if H2O_USE_LIBUV +#else + +#include <errno.h> +#include <fcntl.h> +#include <unistd.h> + +static void on_read(h2o_socket_t *sock, const char *err) +{ + if (err != NULL) { + fprintf(stderr, "pipe error\n"); + abort(); + } + + h2o_buffer_consume(&sock->input, sock->input->size); + queue_cb(sock->data); +} + +static void init_async(h2o_multithread_queue_t *queue, h2o_loop_t *loop) +{ + int fds[2]; + + if (cloexec_pipe(fds) != 0) { + perror("pipe"); + abort(); + } + fcntl(fds[1], F_SETFL, O_NONBLOCK); + queue->async.write = fds[1]; + queue->async.read = h2o_evloop_socket_create(loop, fds[0], 0); + queue->async.read->data = queue; + h2o_socket_read_start(queue->async.read, on_read); +} + +#endif + +h2o_multithread_queue_t *h2o_multithread_create_queue(h2o_loop_t *loop) +{ + h2o_multithread_queue_t *queue = h2o_mem_alloc(sizeof(*queue)); + memset(queue, 0, sizeof(*queue)); + +#if H2O_USE_LIBUV + uv_async_init(loop, &queue->async, (uv_async_cb)queue_cb); +#else + init_async(queue, loop); +#endif + pthread_mutex_init(&queue->mutex, NULL); + h2o_linklist_init_anchor(&queue->receivers.active); + h2o_linklist_init_anchor(&queue->receivers.inactive); + + return queue; +} + +void h2o_multithread_destroy_queue(h2o_multithread_queue_t *queue) +{ + assert(h2o_linklist_is_empty(&queue->receivers.active)); + assert(h2o_linklist_is_empty(&queue->receivers.inactive)); +#if H2O_USE_LIBUV + uv_close((uv_handle_t *)&queue->async, (uv_close_cb)free); +#else + h2o_socket_read_stop(queue->async.read); + h2o_socket_close(queue->async.read); + close(queue->async.write); +#endif + pthread_mutex_destroy(&queue->mutex); +} + +void h2o_multithread_register_receiver(h2o_multithread_queue_t *queue, h2o_multithread_receiver_t *receiver, + h2o_multithread_receiver_cb cb) +{ + receiver->queue = queue; + receiver->_link = (h2o_linklist_t){NULL}; + h2o_linklist_init_anchor(&receiver->_messages); + receiver->cb = cb; + + pthread_mutex_lock(&queue->mutex); + h2o_linklist_insert(&queue->receivers.inactive, &receiver->_link); + pthread_mutex_unlock(&queue->mutex); +} + +void h2o_multithread_unregister_receiver(h2o_multithread_queue_t *queue, h2o_multithread_receiver_t *receiver) +{ + assert(queue == receiver->queue); + assert(h2o_linklist_is_empty(&receiver->_messages)); + pthread_mutex_lock(&queue->mutex); + h2o_linklist_unlink(&receiver->_link); + pthread_mutex_unlock(&queue->mutex); +} + +void h2o_multithread_send_message(h2o_multithread_receiver_t *receiver, h2o_multithread_message_t *message) +{ + int do_send = 0; + + pthread_mutex_lock(&receiver->queue->mutex); + if (message != NULL) { + assert(!h2o_linklist_is_linked(&message->link)); + if (h2o_linklist_is_empty(&receiver->_messages)) { + h2o_linklist_unlink(&receiver->_link); + h2o_linklist_insert(&receiver->queue->receivers.active, &receiver->_link); + do_send = 1; + } + h2o_linklist_insert(&receiver->_messages, &message->link); + } else { + if (h2o_linklist_is_empty(&receiver->_messages)) + do_send = 1; + } + pthread_mutex_unlock(&receiver->queue->mutex); + + if (do_send) { +#if H2O_USE_LIBUV + uv_async_send(&receiver->queue->async); +#else + while (write(receiver->queue->async.write, "", 1) == -1 && errno == EINTR) + ; +#endif + } +} + +void h2o_multithread_create_thread(pthread_t *tid, const pthread_attr_t *attr, void *(*func)(void *), void *arg) +{ + if (pthread_create(tid, attr, func, arg) != 0) { + perror("pthread_create"); + abort(); + } +} + +void h2o_sem_init(h2o_sem_t *sem, ssize_t capacity) +{ + pthread_mutex_init(&sem->_mutex, NULL); + pthread_cond_init(&sem->_cond, NULL); + sem->_cur = capacity; + sem->_capacity = capacity; +} + +void h2o_sem_destroy(h2o_sem_t *sem) +{ + assert(sem->_cur == sem->_capacity); + pthread_cond_destroy(&sem->_cond); + pthread_mutex_destroy(&sem->_mutex); +} + +void h2o_sem_wait(h2o_sem_t *sem) +{ + pthread_mutex_lock(&sem->_mutex); + while (sem->_cur <= 0) + pthread_cond_wait(&sem->_cond, &sem->_mutex); + --sem->_cur; + pthread_mutex_unlock(&sem->_mutex); +} + +void h2o_sem_post(h2o_sem_t *sem) +{ + pthread_mutex_lock(&sem->_mutex); + ++sem->_cur; + pthread_cond_signal(&sem->_cond); + pthread_mutex_unlock(&sem->_mutex); +} + +void h2o_sem_set_capacity(h2o_sem_t *sem, ssize_t new_capacity) +{ + pthread_mutex_lock(&sem->_mutex); + sem->_cur += new_capacity - sem->_capacity; + sem->_capacity = new_capacity; + pthread_cond_broadcast(&sem->_cond); + pthread_mutex_unlock(&sem->_mutex); +} + +/* barrier */ + +void h2o_barrier_init(h2o_barrier_t *barrier, size_t count) +{ + pthread_mutex_init(&barrier->_mutex, NULL); + pthread_cond_init(&barrier->_cond, NULL); + barrier->_count = count; +} + +int h2o_barrier_wait(h2o_barrier_t *barrier) +{ + int ret; + pthread_mutex_lock(&barrier->_mutex); + barrier->_count--; + if (barrier->_count == 0) { + pthread_cond_broadcast(&barrier->_cond); + ret = 1; + } else { + while (barrier->_count) + pthread_cond_wait(&barrier->_cond, &barrier->_mutex); + ret = 0; + } + pthread_mutex_unlock(&barrier->_mutex); + return ret; +} + +int h2o_barrier_done(h2o_barrier_t *barrier) +{ + return __sync_add_and_fetch(&barrier->_count, 0) == 0; +} + +void h2o_barrier_destroy(h2o_barrier_t *barrier) +{ + pthread_mutex_destroy(&barrier->_mutex); + pthread_cond_destroy(&barrier->_cond); +} |