Home | History | Annotate | Line # | Download | only in isc
      1  1.1  christos /*	$NetBSD: async.c,v 1.2 2025/01/26 16:25:36 christos Exp $	*/
      2  1.1  christos 
      3  1.1  christos /*
      4  1.1  christos  * Copyright (C) Internet Systems Consortium, Inc. ("ISC")
      5  1.1  christos  *
      6  1.1  christos  * SPDX-License-Identifier: MPL-2.0
      7  1.1  christos  *
      8  1.1  christos  * This Source Code Form is subject to the terms of the Mozilla Public
      9  1.1  christos  * License, v. 2.0. If a copy of the MPL was not distributed with this
     10  1.1  christos  * file, you can obtain one at https://mozilla.org/MPL/2.0/.
     11  1.1  christos  *
     12  1.1  christos  * See the COPYRIGHT file distributed with this work for additional
     13  1.1  christos  * information regarding copyright ownership.
     14  1.1  christos  */
     15  1.1  christos 
     16  1.1  christos #include <stdlib.h>
     17  1.1  christos #include <sys/types.h>
     18  1.1  christos #include <unistd.h>
     19  1.1  christos 
     20  1.1  christos #include <isc/async.h>
     21  1.1  christos #include <isc/atomic.h>
     22  1.1  christos #include <isc/barrier.h>
     23  1.1  christos #include <isc/condition.h>
     24  1.1  christos #include <isc/job.h>
     25  1.1  christos #include <isc/loop.h>
     26  1.1  christos #include <isc/magic.h>
     27  1.1  christos #include <isc/mem.h>
     28  1.1  christos #include <isc/mutex.h>
     29  1.1  christos #include <isc/refcount.h>
     30  1.1  christos #include <isc/result.h>
     31  1.1  christos #include <isc/signal.h>
     32  1.1  christos #include <isc/strerr.h>
     33  1.1  christos #include <isc/thread.h>
     34  1.1  christos #include <isc/util.h>
     35  1.1  christos #include <isc/uv.h>
     36  1.1  christos #include <isc/work.h>
     37  1.1  christos 
     38  1.1  christos #include "async_p.h"
     39  1.1  christos #include "job_p.h"
     40  1.1  christos #include "loop_p.h"
     41  1.1  christos 
     42  1.1  christos void
     43  1.1  christos isc_async_run(isc_loop_t *loop, isc_job_cb cb, void *cbarg) {
     44  1.1  christos 	REQUIRE(VALID_LOOP(loop));
     45  1.1  christos 	REQUIRE(cb != NULL);
     46  1.1  christos 
     47  1.1  christos 	isc_job_t *job = isc_mem_get(loop->mctx, sizeof(*job));
     48  1.1  christos 	*job = (isc_job_t){
     49  1.1  christos 		.cb = cb,
     50  1.1  christos 		.cbarg = cbarg,
     51  1.1  christos 	};
     52  1.1  christos 
     53  1.1  christos 	cds_wfcq_node_init(&job->wfcq_node);
     54  1.1  christos 
     55  1.1  christos 	/*
     56  1.1  christos 	 * cds_wfcq_enqueue() is non-blocking and enqueues the job to async
     57  1.1  christos 	 * queue.
     58  1.1  christos 	 *
     59  1.1  christos 	 * The function returns 'false' in case the queue was empty - in such
     60  1.1  christos 	 * case we need to trigger the async callback.
     61  1.1  christos 	 */
     62  1.1  christos 	if (!cds_wfcq_enqueue(&loop->async_jobs.head, &loop->async_jobs.tail,
     63  1.1  christos 			      &job->wfcq_node))
     64  1.1  christos 	{
     65  1.1  christos 		int r = uv_async_send(&loop->async_trigger);
     66  1.1  christos 		UV_RUNTIME_CHECK(uv_async_send, r);
     67  1.1  christos 	}
     68  1.1  christos }
     69  1.1  christos 
     70  1.1  christos void
     71  1.1  christos isc__async_cb(uv_async_t *handle) {
     72  1.1  christos 	isc_loop_t *loop = uv_handle_get_data(handle);
     73  1.1  christos 	isc_jobqueue_t jobs;
     74  1.1  christos 
     75  1.1  christos 	REQUIRE(VALID_LOOP(loop));
     76  1.1  christos 
     77  1.1  christos 	/* Initialize local wfcqueue */
     78  1.1  christos 	__cds_wfcq_init(&jobs.head, &jobs.tail);
     79  1.1  christos 
     80  1.1  christos 	/*
     81  1.1  christos 	 * Move all the elements from loop->async_jobs to a local jobs queue.
     82  1.1  christos 	 *
     83  1.1  christos 	 * __cds_wfcq_splice_blocking() assumes that synchronization is
     84  1.1  christos 	 * done externally - there's no internal locking, unlike
     85  1.1  christos 	 * cds_wfcq_splice_blocking(), and we do not need to check whether
     86  1.1  christos 	 * it needs to block, unlike __cds_wfcq_splice_nonblocking().
     87  1.1  christos 	 *
     88  1.1  christos 	 * The reason we can use __cds_wfcq_splice_blocking() is that the
     89  1.1  christos 	 * only other function we use is cds_wfcq_enqueue() which doesn't
     90  1.1  christos 	 * require any synchronization (see the table in urcu/wfcqueue.h
     91  1.1  christos 	 * for more details).
     92  1.1  christos 	 */
     93  1.1  christos 	enum cds_wfcq_ret ret = __cds_wfcq_splice_blocking(
     94  1.1  christos 		&jobs.head, &jobs.tail, &loop->async_jobs.head,
     95  1.1  christos 		&loop->async_jobs.tail);
     96  1.1  christos 	INSIST(ret != CDS_WFCQ_RET_WOULDBLOCK);
     97  1.1  christos 	if (ret == CDS_WFCQ_RET_SRC_EMPTY) {
     98  1.1  christos 		/*
     99  1.1  christos 		 * Nothing to do, the source queue was empty - most
    100  1.1  christos 		 * probably we were called from isc__async_close() below.
    101  1.1  christos 		 */
    102  1.1  christos 		return;
    103  1.1  christos 	}
    104  1.1  christos 
    105  1.1  christos 	/*
    106  1.1  christos 	 * Walk through the local queue which has now all the members copied
    107  1.1  christos 	 * locally, and call the callbacks and free all the isc_job_t(s).
    108  1.1  christos 	 */
    109  1.1  christos 	struct cds_wfcq_node *node, *next;
    110  1.1  christos 	__cds_wfcq_for_each_blocking_safe(&jobs.head, &jobs.tail, node, next) {
    111  1.1  christos 		isc_job_t *job = caa_container_of(node, isc_job_t, wfcq_node);
    112  1.1  christos 
    113  1.1  christos 		job->cb(job->cbarg);
    114  1.1  christos 
    115  1.1  christos 		isc_mem_put(loop->mctx, job, sizeof(*job));
    116  1.1  christos 	}
    117  1.1  christos }
    118  1.1  christos 
    119  1.1  christos void
    120  1.1  christos isc__async_close(uv_handle_t *handle) {
    121  1.1  christos 	isc_loop_t *loop = uv_handle_get_data(handle);
    122  1.1  christos 
    123  1.1  christos 	isc__async_cb(&loop->async_trigger);
    124  1.1  christos }
    125