11477480897

Committed 23 Oct 2024 09:53AM UTC coverage: 85.984% (+0.02%) from 85.968%

Build # 11477480897

Build Type

push

github

Committed by

locker

Commit Message

box: build fix

I got compile error for release build on gcc 14.2.1 20240910 version.

```
In function ‘char* mp_store_double(char*, double)’,
    inlined from ‘char* mp_encode_double(char*, double)’ at /home/shiny/dev/tarantool-ee/tarantool/src/lib/msgpuck/msgpuck.h:2409:24,
    inlined from ‘uint32_t tuple_hash_field(uint32_t*, uint32_t*, const char**, field_type, coll*)’ at /home/shiny/dev/tarantool-ee/tarantool/src/box/tuple_hash.cc:317:46:
/home/shiny/dev/tarantool-ee/tarantool/src/lib/msgpuck/msgpuck.h:340:16: error: ‘value’ may be used uninitialized [-Werror=maybe-uninitialized]
  340 |         cast.d = val;
      |         ~~~~~~~^~~~~
/home/shiny/dev/tarantool-ee/tarantool/src/box/tuple_hash.cc: In function ‘uint32_t tuple_hash_field(uint32_t*, uint32_t*, const char**, field_type, coll*)’:
/home/shiny/dev/tarantool-ee/tarantool/src/box/tuple_hash.cc:311:24: note: ‘value’ was declared here
  311 |                 double value;
      |
```

NO_TEST=build fix
NO_CHANGELOG=build fix
NO_DOC=build fix

(cherry picked from commit 1129c758d)

Run Details

63182 of 114387 branches covered (55.24%)

2 of 2 new or added lines in 1 file covered. (100.0%)

15 existing lines in 9 files now uncovered.

94033 of 109361 relevant lines covered (85.98%)

2767362.1 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.47

/src/lib/core/cbus.c

/*
 * Copyright 2010-2016, Tarantool AUTHORS, please see AUTHORS file.
 *
 * Redistribution and use in source and binary forms, with or
 * without modification, are permitted provided that the following
 * conditions are met:
 *
 * 1. Redistributions of source code must retain the above
 *    copyright notice, this list of conditions and the
 *    following disclaimer.
 *
 * 2. 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.
 *
 * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``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
 * <COPYRIGHT HOLDER> OR CONTRIBUTORS 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.
 */
#include "cbus.h"

#include <limits.h>
#include "fiber.h"
#include "trigger.h"

/**
 * Cord interconnect.
 */
struct cbus {
        /** cbus statistics */
        struct rmean *stats;
        /** A mutex to protect bus join. */
        pthread_mutex_t mutex;
        /** Condition for synchronized start of the bus. */
        pthread_cond_t cond;
        /** Connected endpoints */
        struct rlist endpoints;
};

/** A singleton for all cords. */
static struct cbus cbus;

const char *cbus_stat_strings[CBUS_STAT_LAST] = {
        "EVENTS",
        "LOCKS",
};

/**
 * Find a joined cbus endpoint by name.
 * This is an internal helper method which should be called
 * under cbus::mutex.
 *
 * @return endpoint or NULL if not found
 */
static struct cbus_endpoint *
cbus_find_endpoint_locked(struct cbus *bus, const char *name)
{
        struct cbus_endpoint *endpoint;
        rlist_foreach_entry(endpoint, &bus->endpoints, in_cbus) {
                if (strcmp(endpoint->name, name) == 0)
                        return endpoint;
        }
        return NULL;
}

static struct cbus_endpoint *
cbus_find_endpoint(struct cbus *bus, const char *name)
{
        tt_pthread_mutex_lock(&bus->mutex);
        struct cbus_endpoint *endpoint = cbus_find_endpoint_locked(bus, name);
        tt_pthread_mutex_unlock(&bus->mutex);
        return endpoint;
}

static void
cpipe_flush_cb(ev_loop * /* loop */, struct ev_async *watcher,
               int /* events */);

void
cpipe_create(struct cpipe *pipe, const char *consumer)
{
        stailq_create(&pipe->input);

        pipe->n_input = 0;
        pipe->max_input = INT_MAX;
        pipe->producer = cord()->loop;

        ev_async_init(&pipe->flush_input, cpipe_flush_cb);
        pipe->flush_input.data = pipe;
        rlist_create(&pipe->on_flush);

        tt_pthread_mutex_lock(&cbus.mutex);
        struct cbus_endpoint *endpoint =
                cbus_find_endpoint_locked(&cbus, consumer);
        while (endpoint == NULL) {
                tt_pthread_cond_wait(&cbus.cond, &cbus.mutex);
                endpoint = cbus_find_endpoint_locked(&cbus, consumer);
        }
        pipe->endpoint = endpoint;
        ++pipe->endpoint->n_pipes;
        tt_pthread_mutex_unlock(&cbus.mutex);
}

struct cmsg_poison {
        struct cmsg msg;
        struct cbus_endpoint *endpoint;
};

static void
cbus_endpoint_poison_f(struct cmsg *msg)
{
        struct cbus_endpoint *endpoint = ((struct cmsg_poison *)msg)->endpoint;
        tt_pthread_mutex_lock(&cbus.mutex);
        assert(endpoint->n_pipes > 0);
        --endpoint->n_pipes;
        tt_pthread_mutex_unlock(&cbus.mutex);
        fiber_cond_signal(&endpoint->cond);
        free(msg);
}

void
cpipe_destroy(struct cpipe *pipe)
{
        /*
         * The thread should not be canceled while mutex is locked.
         * And everything else must be protected for consistency.
        */
        int old_cancel_state;
        tt_pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancel_state);

        ev_async_stop(pipe->producer, &pipe->flush_input);

        static const struct cmsg_hop route[1] = {
                {cbus_endpoint_poison_f, NULL}
        };
        trigger_destroy(&pipe->on_flush);

        struct cbus_endpoint *endpoint = pipe->endpoint;
        struct cmsg_poison *poison = malloc(sizeof(struct cmsg_poison));
        cmsg_init(&poison->msg, route);
        poison->endpoint = pipe->endpoint;
        /*
         * Avoid the general purpose cpipe_push_input() since
         * we want to control the way the poison message is
         * delivered.
         */
        tt_pthread_mutex_lock(&endpoint->mutex);
        /* Flush input */
        stailq_concat(&endpoint->output, &pipe->input);
        pipe->n_input = 0;
        /* Add the pipe shutdown message as the last one. */
        stailq_add_tail_entry(&endpoint->output, poison, msg.fifo);
        /* Count statistics */
        rmean_collect(cbus.stats, CBUS_STAT_EVENTS, 1);
        /*
         * Keep the lock for the duration of ev_async_send():
         * this will avoid a race condition between
         * ev_async_send() and execution of the poison
         * message, after which the endpoint may disappear.
         */
        ev_async_send(endpoint->consumer, &endpoint->async);
        tt_pthread_mutex_unlock(&endpoint->mutex);

        tt_pthread_setcancelstate(old_cancel_state, NULL);

        TRASH(pipe);
}

static void
cbus_create(struct cbus *bus)
{
        bus->stats = rmean_new(cbus_stat_strings, CBUS_STAT_LAST);
        if (bus->stats == NULL)
                panic_syserror("cbus_create");

        /* Initialize queue lock mutex. */
        (void) tt_pthread_mutex_init(&bus->mutex, NULL);

        (void) tt_pthread_cond_init(&bus->cond, NULL);

        rlist_create(&bus->endpoints);
}

static void
cbus_destroy(struct cbus *bus)
{
        /*
         * Lock the mutex to ensure we do not destroy a mutex
         * while it is locked, happens in at_exit() handler.
         */
        (void) tt_pthread_mutex_lock(&bus->mutex);
        (void) tt_pthread_mutex_unlock(&bus->mutex);
        (void) tt_pthread_mutex_destroy(&bus->mutex);
        (void) tt_pthread_cond_destroy(&bus->cond);
        rmean_delete(bus->stats);
}

/**
 * Join a new endpoint (message consumer) to the bus. The endpoint
 * must have a unique name. Wakes up all producers (@sa cpipe_create())
 * who are blocked waiting for this endpoint to become available.
 */
int
cbus_endpoint_create(struct cbus_endpoint *endpoint, const char *name,
                     void (*fetch_cb)(ev_loop *, struct ev_watcher *, int),
                     void *fetch_data)
{
        tt_pthread_mutex_lock(&cbus.mutex);
        if (cbus_find_endpoint_locked(&cbus, name) != NULL) {
                tt_pthread_mutex_unlock(&cbus.mutex);
                return 1;
        }

        snprintf(endpoint->name, sizeof(endpoint->name), "%s", name);
        endpoint->consumer = loop();
        endpoint->n_pipes = 0;
        fiber_cond_create(&endpoint->cond);
        tt_pthread_mutex_init(&endpoint->mutex, NULL);
        stailq_create(&endpoint->output);
        ev_async_init(&endpoint->async,
                      (void (*)(ev_loop *, struct ev_async *, int)) fetch_cb);
        endpoint->async.data = fetch_data;
        ev_async_start(endpoint->consumer, &endpoint->async);

        rlist_add_tail(&cbus.endpoints, &endpoint->in_cbus);
        /*
         * Alert all waiting producers.
         *
         * POSIX: pthread_cond_broadcast() function shall
         * have no effect if there are no threads currently
         * blocked on cond.
         */
        tt_pthread_cond_broadcast(&cbus.cond);
        tt_pthread_mutex_unlock(&cbus.mutex);
        return 0;
}

int
cbus_endpoint_destroy(struct cbus_endpoint *endpoint,
                      void (*process_cb)(struct cbus_endpoint *endpoint))
{
        tt_pthread_mutex_lock(&cbus.mutex);
        /*
         * Remove endpoint from cbus registry, so no new pipe can
         * be created for this endpoint.
         */
        rlist_del(&endpoint->in_cbus);
        tt_pthread_mutex_unlock(&cbus.mutex);

        while (true) {
                if (process_cb)
                        process_cb(endpoint);
                if (endpoint->n_pipes == 0 && stailq_empty(&endpoint->output))
                        break;
                 fiber_cond_wait(&endpoint->cond);
        }

        /*
         * Pipe flush func can still lock mutex, so just lock and unlock
         * it.
         */
        tt_pthread_mutex_lock(&endpoint->mutex);
        tt_pthread_mutex_unlock(&endpoint->mutex);
        tt_pthread_mutex_destroy(&endpoint->mutex);
        ev_async_stop(endpoint->consumer, &endpoint->async);
        fiber_cond_destroy(&endpoint->cond);
        TRASH(endpoint);
        return 0;
}

static void
cpipe_flush_cb(ev_loop *loop, struct ev_async *watcher, int events)
{
        (void) loop;
        (void) events;
        struct cpipe *pipe = (struct cpipe *) watcher->data;
        struct cbus_endpoint *endpoint = pipe->endpoint;
        if (pipe->n_input == 0)
                return;

        trigger_run(&pipe->on_flush, pipe);
        /* Trigger task processing when the queue becomes non-empty. */
        bool output_was_empty;

        /*
         * We need to set a thread cancellation guard, because
         * another thread may cancel the current thread
         * (write() is a cancellation point in ev_async_send)
         * and the activation of the ev_async watcher
         * through ev_async_send will fail.
         */

        int old_cancel_state;
        tt_pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancel_state);

        tt_pthread_mutex_lock(&endpoint->mutex);
        output_was_empty = stailq_empty(&endpoint->output);
        /** Flush input */
        stailq_concat(&endpoint->output, &pipe->input);
        tt_pthread_mutex_unlock(&endpoint->mutex);

        pipe->n_input = 0;
        if (output_was_empty) {
                /* Count statistics */
                rmean_collect(cbus.stats, CBUS_STAT_EVENTS, 1);

                ev_async_send(endpoint->consumer, &endpoint->async);
        }

        tt_pthread_setcancelstate(old_cancel_state, NULL);
}

void
cbus_init(void)
{
        cbus_create(&cbus);
}

void
cbus_free(void)
{
        cbus_destroy(&cbus);
}

/* {{{ cmsg */

/**
 * Dispatch the message to the next hop.
 */
static inline void
cmsg_dispatch(struct cpipe *pipe, struct cmsg *msg)
{
        /**
         * 'pipe' pointer saved in class constructor works as
         * a guard that the message is alive. If a message route
         * has the next pipe, then the message mustn't have been
         * destroyed on this hop. Otherwise msg->hop->pipe could
         * be already pointing to garbage.
         */
        if (pipe) {
                /*
                 * Once we pushed the message to the bus,
                 * we relinquished all write access to it,
                 * so we must increase the current hop *before*
                 * push.
                 */
                msg->hop++;
                cpipe_push(pipe, msg);
        }
}

/**
 * Deliver the message and dispatch it to the next hop.
 */
void
cmsg_deliver(struct cmsg *msg)
{
        /*
         * Save the pointer to the last pipe,
         * the memory where it is stored may be destroyed
         * on the last hop.
         */
        struct cpipe *pipe = msg->hop->pipe;
        msg->hop->f(msg);
        cmsg_dispatch(pipe, msg);
}

/* }}} cmsg */

/**
 * Call the target function and store the results (diag, rc) in
 * struct cbus_call_msg.
 */
void
cbus_call_perform(struct cmsg *m)
{
        struct cbus_call_msg *msg = (struct cbus_call_msg *)m;
        msg->rc = msg->func(msg);
        if (msg->rc)
                diag_move(&fiber()->diag, &msg->diag);
}

/**
 * Wake up the caller fiber to reap call results.
 * If the fiber is gone, e.g. in case of call timeout
 * or cancellation, invoke free_cb to free message state.
 */
void
cbus_call_done(struct cmsg *m)
{
        struct cbus_call_msg *msg = (struct cbus_call_msg *)m;
        if (msg->caller == NULL) {
                if (msg->free_cb)
                        msg->free_cb(msg);
                return;
        }
        msg->complete = true;
        fiber_wakeup(msg->caller);
}

int
cbus_call_timeout(struct cpipe *callee, struct cpipe *caller,
                  struct cbus_call_msg *msg, cbus_call_f func,
                  cbus_call_f free_cb, double timeout)
{
        int rc;

        diag_create(&msg->diag);
        msg->caller = fiber();
        msg->complete = false;
        msg->route[0].f = cbus_call_perform;
        msg->route[0].pipe = caller;
        msg->route[1].f = cbus_call_done;
        msg->route[1].pipe = NULL;
        cmsg_init(cmsg(msg), msg->route);

        msg->func = func;
        msg->free_cb = free_cb;
        msg->rc = 0;

        cpipe_push(callee, cmsg(msg));

        ev_tstamp deadline = ev_monotonic_now(loop()) + timeout;
        do {
                bool exceeded = fiber_yield_deadline(deadline);
                if (exceeded) {
                        msg->caller = NULL;
                        diag_set(TimedOut);
                        return -1;
                }
        } while (!msg->complete);

        if ((rc = msg->rc))
                diag_move(&msg->diag, &fiber()->diag);
        return rc;
}

struct cbus_pair_msg {
        struct cmsg cmsg;
        void (*pair_cb)(void *);
        void *pair_arg;
        const char *src_name;
        struct cpipe *src_pipe;
        bool complete;
        struct fiber_cond cond;
};

static void
cbus_pair_complete(struct cmsg *cmsg);

static void
cbus_pair_perform(struct cmsg *cmsg)
{
        struct cbus_pair_msg *msg = container_of(cmsg,
                        struct cbus_pair_msg, cmsg);
        static struct cmsg_hop route[] = {
                {cbus_pair_complete, NULL},
        };
        cmsg_init(cmsg, route);
        cpipe_create(msg->src_pipe, msg->src_name);
        if (msg->pair_cb != NULL)
                msg->pair_cb(msg->pair_arg);
        cpipe_push(msg->src_pipe, cmsg);
}

static void
cbus_pair_complete(struct cmsg *cmsg)
{
        struct cbus_pair_msg *msg = container_of(cmsg,
                        struct cbus_pair_msg, cmsg);
        msg->complete = true;
        fiber_cond_signal(&msg->cond);
}

void
cbus_pair(const char *dest_name, const char *src_name,
          struct cpipe *dest_pipe, struct cpipe *src_pipe,
          void (*pair_cb)(void *), void *pair_arg,
          void (*process_cb)(struct cbus_endpoint *))
{
        static struct cmsg_hop route[] = {
                {cbus_pair_perform, NULL},
        };
        struct cbus_pair_msg msg;

        cmsg_init(&msg.cmsg, route);
        msg.pair_cb = pair_cb;
        msg.pair_arg = pair_arg;
        msg.complete = false;
        msg.src_name = src_name;
        msg.src_pipe = src_pipe;
        fiber_cond_create(&msg.cond);

        struct cbus_endpoint *endpoint = cbus_find_endpoint(&cbus, src_name);
        assert(endpoint != NULL);

        cpipe_create(dest_pipe, dest_name);
        cpipe_push(dest_pipe, &msg.cmsg);

        while (true) {
                if (process_cb != NULL)
                        process_cb(endpoint);
                if (msg.complete)
                        break;
                fiber_cond_wait(&msg.cond);
        }
}

struct cbus_unpair_msg {
        struct cmsg cmsg;
        void (*unpair_cb)(void *);
        void *unpair_arg;
        struct cpipe *src_pipe;
        bool complete;
        struct fiber_cond cond;
};

static void
cbus_unpair_prepare(struct cmsg *cmsg)
{
        struct cbus_unpair_msg *msg = container_of(cmsg,
                        struct cbus_unpair_msg, cmsg);
        if (msg->unpair_cb != NULL)
                msg->unpair_cb(msg->unpair_arg);
}

static void
cbus_unpair_flush(struct cmsg *cmsg)
{
        (void)cmsg;
}

static void
cbus_unpair_complete(struct cmsg *cmsg);

static void
cbus_unpair_perform(struct cmsg *cmsg)
{
        struct cbus_unpair_msg *msg = container_of(cmsg,
                        struct cbus_unpair_msg, cmsg);
        static struct cmsg_hop route[] = {
                {cbus_unpair_complete, NULL},
        };
        cmsg_init(cmsg, route);
        cpipe_push(msg->src_pipe, cmsg);
        cpipe_destroy(msg->src_pipe);
}

static void
cbus_unpair_complete(struct cmsg *cmsg)
{
        struct cbus_unpair_msg *msg = container_of(cmsg,
                        struct cbus_unpair_msg, cmsg);
        msg->complete = true;
        fiber_cond_signal(&msg->cond);
}

void
cbus_unpair(struct cpipe *dest_pipe, struct cpipe *src_pipe,
            void (*unpair_cb)(void *), void *unpair_arg,
            void (*process_cb)(struct cbus_endpoint *))
{
        struct cmsg_hop route[] = {
                {cbus_unpair_prepare, src_pipe},
                {cbus_unpair_flush, dest_pipe},
                {cbus_unpair_perform, NULL},
        };
        struct cbus_unpair_msg msg;

        cmsg_init(&msg.cmsg, route);
        msg.unpair_cb = unpair_cb;
        msg.unpair_arg = unpair_arg;
        msg.src_pipe = src_pipe;
        msg.complete = false;
        fiber_cond_create(&msg.cond);

        cpipe_push(dest_pipe, &msg.cmsg);

        struct cbus_endpoint *endpoint = src_pipe->endpoint;
        while (true) {
                if (process_cb != NULL)
                        process_cb(endpoint);
                if (msg.complete)
                        break;
                fiber_cond_wait(&msg.cond);
        }

        cpipe_destroy(dest_pipe);
}

void
cbus_process(struct cbus_endpoint *endpoint)
{
        struct stailq output;
        stailq_create(&output);
        cbus_endpoint_fetch(endpoint, &output);
        struct cmsg *msg, *msg_next;
        stailq_foreach_entry_safe(msg, msg_next, &output, fifo)
                cmsg_deliver(msg);
}

void
cbus_loop(struct cbus_endpoint *endpoint)
{
        while (true) {
                cbus_process(endpoint);
                fiber_check_gc();
                if (fiber_is_cancelled())
                        break;
                fiber_yield();
        }
}

static void
cbus_stop_loop_f(struct cmsg *msg)
{
        fiber_cancel(fiber());
        free(msg);
}

void
cbus_stop_loop(struct cpipe *pipe)
{
        /*
         * Hack: static message only works because cmsg_deliver()
         * is a no-op on the second hop.
         */
        static const struct cmsg_hop route[1] = {
                {cbus_stop_loop_f, NULL}
        };
        struct cmsg *cancel = malloc(sizeof(struct cmsg));

        cmsg_init(cancel, route);

        cpipe_push(pipe, cancel);
        ev_invoke(pipe->producer, &pipe->flush_input, EV_CUSTOM);
}


1	/*
2	* Copyright 2010-2016, Tarantool AUTHORS, please see AUTHORS file.
3	*
4	* Redistribution and use in source and binary forms, with or
5	* without modification, are permitted provided that the following
6	* conditions are met:
7	*
8	* 1. Redistributions of source code must retain the above
9	* copyright notice, this list of conditions and the
10	* following disclaimer.
11	*
12	* 2. Redistributions in binary form must reproduce the above
13	* copyright notice, this list of conditions and the following
14	* disclaimer in the documentation and/or other materials
15	* provided with the distribution.
16	*
17	* THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
18	* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
19	* TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
20	* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
21	* <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
22	* INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
23	* DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
24	* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
25	* BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
26	* LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
27	* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
28	* THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
29	* SUCH DAMAGE.
30	*/
31	#include "cbus.h"
32
33	#include <limits.h>
34	#include "fiber.h"
35	#include "trigger.h"
36
37	/**
38	* Cord interconnect.
39	*/
40	struct cbus {
41	/** cbus statistics */
42	struct rmean *stats;
43	/** A mutex to protect bus join. */
44	pthread_mutex_t mutex;
45	/** Condition for synchronized start of the bus. */
46	pthread_cond_t cond;
47	/** Connected endpoints */
48	struct rlist endpoints;
49	};
50
51	/** A singleton for all cords. */
52	static struct cbus cbus;
53
54	const char *cbus_stat_strings[CBUS_STAT_LAST] = {
55	"EVENTS",
56	"LOCKS",
57	};
58
59	/**
60	* Find a joined cbus endpoint by name.
61	* This is an internal helper method which should be called
62	* under cbus::mutex.
63	*
64	* @return endpoint or NULL if not found
65	*/
66	static struct cbus_endpoint *
67	cbus_find_endpoint_locked(struct cbus bus, const char name)	70,788✔
68	{
69	struct cbus_endpoint *endpoint;
70	rlist_foreach_entry(endpoint, &bus->endpoints, in_cbus) {	796,870✔
71	if (strcmp(endpoint->name, name) == 0)	366,693✔
72	return endpoint;	39,046✔
73	}
74	return NULL;	31,742✔
75	}
76
77	static struct cbus_endpoint *
78	cbus_find_endpoint(struct cbus bus, const char name)	5,291✔
79	{
80	tt_pthread_mutex_lock(&bus->mutex);	5,291!
81	struct cbus_endpoint *endpoint = cbus_find_endpoint_locked(bus, name);	5,291✔
82	tt_pthread_mutex_unlock(&bus->mutex);	5,291!
83	return endpoint;	5,291✔
84	}
85
86	static void
87	cpipe_flush_cb(ev_loop * /* loop /, struct ev_async watcher,
88	int /* events */);
89
90	void
91	cpipe_create(struct cpipe pipe, const char consumer)	33,755✔
92	{
93	stailq_create(&pipe->input);	33,755✔
94
95	pipe->n_input = 0;	33,755✔
96	pipe->max_input = INT_MAX;	33,755✔
97	pipe->producer = cord()->loop;	33,755!
98
99	ev_async_init(&pipe->flush_input, cpipe_flush_cb);	33,755✔
100	pipe->flush_input.data = pipe;	33,755✔
101	rlist_create(&pipe->on_flush);	33,755✔
102
103	tt_pthread_mutex_lock(&cbus.mutex);	33,755!
104	struct cbus_endpoint *endpoint =
105	cbus_find_endpoint_locked(&cbus, consumer);	33,755✔
106	while (endpoint == NULL) {	44,634✔
107	tt_pthread_cond_wait(&cbus.cond, &cbus.mutex);	10,879!
108	endpoint = cbus_find_endpoint_locked(&cbus, consumer);	10,879✔
109	}
110	pipe->endpoint = endpoint;	33,755✔
111	++pipe->endpoint->n_pipes;	33,755✔
112	tt_pthread_mutex_unlock(&cbus.mutex);	33,755!
113	}	33,755✔
114
115	struct cmsg_poison {
116	struct cmsg msg;
117	struct cbus_endpoint *endpoint;
118	};
119
120	static void
121	cbus_endpoint_poison_f(struct cmsg *msg)	9,260✔
122	{
123	struct cbus_endpoint endpoint = ((struct cmsg_poison )msg)->endpoint;	9,260✔
124	tt_pthread_mutex_lock(&cbus.mutex);	9,260!
125	assert(endpoint->n_pipes > 0);	9,260!
126	--endpoint->n_pipes;	9,260✔
127	tt_pthread_mutex_unlock(&cbus.mutex);	9,260!
128	fiber_cond_signal(&endpoint->cond);	9,260✔
129	free(msg);	9,260✔
130	}	9,260✔
131
132	void
133	cpipe_destroy(struct cpipe *pipe)	12,568✔
134	{
135	/*
136	* The thread should not be canceled while mutex is locked.
137	* And everything else must be protected for consistency.
138	*/
139	int old_cancel_state;
140	tt_pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancel_state);	12,568!
141
142	ev_async_stop(pipe->producer, &pipe->flush_input);	12,568!
143
144	static const struct cmsg_hop route[1] = {
145	{cbus_endpoint_poison_f, NULL}
146	};
147	trigger_destroy(&pipe->on_flush);	12,568!
148
149	struct cbus_endpoint *endpoint = pipe->endpoint;	12,568✔
150	struct cmsg_poison *poison = malloc(sizeof(struct cmsg_poison));	12,568✔
151	cmsg_init(&poison->msg, route);	12,568!
152	poison->endpoint = pipe->endpoint;	12,568✔
153	/*
154	* Avoid the general purpose cpipe_push_input() since
155	* we want to control the way the poison message is
156	* delivered.
157	*/
158	tt_pthread_mutex_lock(&endpoint->mutex);	12,568!
159	/* Flush input */
160	stailq_concat(&endpoint->output, &pipe->input);	12,568!
161	pipe->n_input = 0;	12,568✔
162	/* Add the pipe shutdown message as the last one. */
163	stailq_add_tail_entry(&endpoint->output, poison, msg.fifo);	12,568!
164	/* Count statistics */
165	rmean_collect(cbus.stats, CBUS_STAT_EVENTS, 1);	12,568!
166	/*
167	* Keep the lock for the duration of ev_async_send():
168	* this will avoid a race condition between
169	* ev_async_send() and execution of the poison
170	* message, after which the endpoint may disappear.
171	*/
172	ev_async_send(endpoint->consumer, &endpoint->async);	12,568!
173	tt_pthread_mutex_unlock(&endpoint->mutex);	12,568!
174
175	tt_pthread_setcancelstate(old_cancel_state, NULL);	12,568!
176
177	TRASH(pipe);	12,568✔
178	}	12,568✔
179
180	static void
181	cbus_create(struct cbus *bus)	4,060✔
182	{
183	bus->stats = rmean_new(cbus_stat_strings, CBUS_STAT_LAST);	4,060✔
184	if (bus->stats == NULL)	4,060!
185	panic_syserror("cbus_create");	×
186
187	/* Initialize queue lock mutex. */
188	(void) tt_pthread_mutex_init(&bus->mutex, NULL);	4,060!
189
190	(void) tt_pthread_cond_init(&bus->cond, NULL);	4,060!
191
192	rlist_create(&bus->endpoints);	4,060✔
193	}	4,060✔
194
195	static void
196	cbus_destroy(struct cbus *bus)	3,987✔
197	{
198	/*
199	* Lock the mutex to ensure we do not destroy a mutex
200	* while it is locked, happens in at_exit() handler.
201	*/
202	(void) tt_pthread_mutex_lock(&bus->mutex);	3,987!
203	(void) tt_pthread_mutex_unlock(&bus->mutex);	3,987!
204	(void) tt_pthread_mutex_destroy(&bus->mutex);	3,987!
205	(void) tt_pthread_cond_destroy(&bus->cond);	3,987!
206	rmean_delete(bus->stats);	3,987✔
207	}	3,987✔
208
209	/**
210	* Join a new endpoint (message consumer) to the bus. The endpoint
211	* must have a unique name. Wakes up all producers (@sa cpipe_create())
212	* who are blocked waiting for this endpoint to become available.
213	*/
214	int
215	cbus_endpoint_create(struct cbus_endpoint endpoint, const char name,	20,863✔
216	void (fetch_cb)(ev_loop , struct ev_watcher *, int),
217	void *fetch_data)
218	{
219	tt_pthread_mutex_lock(&cbus.mutex);	20,863!
220	if (cbus_find_endpoint_locked(&cbus, name) != NULL) {	20,863!
221	tt_pthread_mutex_unlock(&cbus.mutex);	×
222	return 1;	×
223	}
224
225	snprintf(endpoint->name, sizeof(endpoint->name), "%s", name);	20,863✔
226	endpoint->consumer = loop();	20,863!
227	endpoint->n_pipes = 0;	20,863✔
228	fiber_cond_create(&endpoint->cond);	20,863✔
229	tt_pthread_mutex_init(&endpoint->mutex, NULL);	20,863!
230	stailq_create(&endpoint->output);	20,863✔
231	ev_async_init(&endpoint->async,	20,863✔
232	(void ()(ev_loop , struct ev_async *, int)) fetch_cb);
233	endpoint->async.data = fetch_data;	20,863✔
234	ev_async_start(endpoint->consumer, &endpoint->async);	20,863✔
235
236	rlist_add_tail(&cbus.endpoints, &endpoint->in_cbus);	20,863✔
237	/*
238	* Alert all waiting producers.
239	*
240	* POSIX: pthread_cond_broadcast() function shall
241	* have no effect if there are no threads currently
242	* blocked on cond.
243	*/
244	tt_pthread_cond_broadcast(&cbus.cond);	20,863!
245	tt_pthread_mutex_unlock(&cbus.mutex);	20,863!
246	return 0;	20,863✔
247	}
248
249	int
250	cbus_endpoint_destroy(struct cbus_endpoint *endpoint,	1,916✔
251	void (process_cb)(struct cbus_endpoint endpoint))
252	{
253	tt_pthread_mutex_lock(&cbus.mutex);	1,916!
254	/*
255	* Remove endpoint from cbus registry, so no new pipe can
256	* be created for this endpoint.
257	*/
258	rlist_del(&endpoint->in_cbus);	1,916✔
259	tt_pthread_mutex_unlock(&cbus.mutex);	1,916!
260
261	while (true) {
262	if (process_cb)	1,916!
263	process_cb(endpoint);	1,916✔
264	if (endpoint->n_pipes == 0 && stailq_empty(&endpoint->output))	1,916!
265	break;	1,916✔
UNCOV 266	fiber_cond_wait(&endpoint->cond);	×
267	}
268
269	/*
270	* Pipe flush func can still lock mutex, so just lock and unlock
271	* it.
272	*/
273	tt_pthread_mutex_lock(&endpoint->mutex);	1,916!
274	tt_pthread_mutex_unlock(&endpoint->mutex);	1,916!
275	tt_pthread_mutex_destroy(&endpoint->mutex);	1,916!
276	ev_async_stop(endpoint->consumer, &endpoint->async);	1,916✔
277	fiber_cond_destroy(&endpoint->cond);	1,916✔
278	TRASH(endpoint);	1,916✔
279	return 0;	1,916✔
280	}
281
282	static void
283	cpipe_flush_cb(ev_loop loop, struct ev_async watcher, int events)	8,355,550✔
284	{
285	(void) loop;
286	(void) events;
287	struct cpipe pipe = (struct cpipe ) watcher->data;	8,355,550✔
288	struct cbus_endpoint *endpoint = pipe->endpoint;	8,355,550✔
289	if (pipe->n_input == 0)	8,355,550✔
290	return;	1,344✔
291
292	trigger_run(&pipe->on_flush, pipe);	8,354,210!
293	/* Trigger task processing when the queue becomes non-empty. */
294	bool output_was_empty;
295
296	/*
297	* We need to set a thread cancellation guard, because
298	* another thread may cancel the current thread
299	* (write() is a cancellation point in ev_async_send)
300	* and the activation of the ev_async watcher
301	* through ev_async_send will fail.
302	*/
303
304	int old_cancel_state;
305	tt_pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancel_state);	8,354,210!
306
307	tt_pthread_mutex_lock(&endpoint->mutex);	8,354,210!
308	output_was_empty = stailq_empty(&endpoint->output);	8,354,210!
309	/** Flush input */
310	stailq_concat(&endpoint->output, &pipe->input);	8,354,210!
311	tt_pthread_mutex_unlock(&endpoint->mutex);	8,354,210!
312
313	pipe->n_input = 0;	8,354,210✔
314	if (output_was_empty) {	8,354,210✔
315	/* Count statistics */
316	rmean_collect(cbus.stats, CBUS_STAT_EVENTS, 1);	8,225,260!
317
318	ev_async_send(endpoint->consumer, &endpoint->async);	8,225,260!
319	}
320
321	tt_pthread_setcancelstate(old_cancel_state, NULL);	8,354,210!
322	}
323
324	void
325	cbus_init(void)	4,060✔
326	{
327	cbus_create(&cbus);	4,060✔
328	}	4,060✔
329
330	void
331	cbus_free(void)	3,987✔
332	{
333	cbus_destroy(&cbus);	3,987✔
334	}	3,987✔
335
336	/* {{{ cmsg */
337
338	/**
339	* Dispatch the message to the next hop.
340	*/
341	static inline void
342	cmsg_dispatch(struct cpipe pipe, struct cmsg msg)	8,597,060✔
343	{
344	/**
345	* 'pipe' pointer saved in class constructor works as
346	* a guard that the message is alive. If a message route
347	* has the next pipe, then the message mustn't have been
348	* destroyed on this hop. Otherwise msg->hop->pipe could
349	* be already pointing to garbage.
350	*/
351	if (pipe) {	8,597,060✔
352	/*
353	* Once we pushed the message to the bus,
354	* we relinquished all write access to it,
355	* so we must increase the current hop before
356	* push.
357	*/
358	msg->hop++;	4,194,100✔
359	cpipe_push(pipe, msg);	4,194,100✔
360	}
361	}	8,597,060✔
362
363	/**
364	* Deliver the message and dispatch it to the next hop.
365	*/
366	void
367	cmsg_deliver(struct cmsg *msg)	8,597,440✔
368	{
369	/*
370	* Save the pointer to the last pipe,
371	* the memory where it is stored may be destroyed
372	* on the last hop.
373	*/
374	struct cpipe *pipe = msg->hop->pipe;	8,597,440✔
375	msg->hop->f(msg);	8,597,440✔
376	cmsg_dispatch(pipe, msg);	8,597,060✔
377	}	8,597,060✔
378
379	/* }}} cmsg */
380
381	/**
382	* Call the target function and store the results (diag, rc) in
383	* struct cbus_call_msg.
384	*/
385	void
386	cbus_call_perform(struct cmsg *m)	158,353✔
387	{
388	struct cbus_call_msg msg = (struct cbus_call_msg )m;	158,353✔
389	msg->rc = msg->func(msg);	158,353✔
390	if (msg->rc)	158,351✔
391	diag_move(&fiber()->diag, &msg->diag);	313!
392	}	158,351✔
393
394	/**
395	* Wake up the caller fiber to reap call results.
396	* If the fiber is gone, e.g. in case of call timeout
397	* or cancellation, invoke free_cb to free message state.
398	*/
399	void
400	cbus_call_done(struct cmsg *m)	158,287✔
401	{
402	struct cbus_call_msg msg = (struct cbus_call_msg )m;	158,287✔
403	if (msg->caller == NULL) {	158,287✔
404	if (msg->free_cb)	2!
405	msg->free_cb(msg);	2✔
406	return;	2✔
407	}
408	msg->complete = true;	158,285✔
409	fiber_wakeup(msg->caller);	158,285✔
410	}
411
412	int
413	cbus_call_timeout(struct cpipe callee, struct cpipe caller,	158,356✔
414	struct cbus_call_msg *msg, cbus_call_f func,
415	cbus_call_f free_cb, double timeout)
416	{
417	int rc;
418
419	diag_create(&msg->diag);	158,356✔
420	msg->caller = fiber();	158,356!
421	msg->complete = false;	158,356✔
422	msg->route[0].f = cbus_call_perform;	158,356✔
423	msg->route[0].pipe = caller;	158,356✔
424	msg->route[1].f = cbus_call_done;	158,356✔
425	msg->route[1].pipe = NULL;	158,356✔
426	cmsg_init(cmsg(msg), msg->route);	158,356✔
427
428	msg->func = func;	158,356✔
429	msg->free_cb = free_cb;	158,356✔
430	msg->rc = 0;	158,356✔
431
432	cpipe_push(callee, cmsg(msg));	158,356✔
433
434	ev_tstamp deadline = ev_monotonic_now(loop()) + timeout;	158,356!
435	do {
436	bool exceeded = fiber_yield_deadline(deadline);	158,389✔
437	if (exceeded) {	158,320✔
438	msg->caller = NULL;	2✔
439	diag_set(TimedOut);	2!
440	return -1;	2✔
441	}
442	} while (!msg->complete);	158,318✔
443
444	if ((rc = msg->rc))	158,285✔
445	diag_move(&msg->diag, &fiber()->diag);	313!
446	return rc;	158,285✔
447	}
448
449	struct cbus_pair_msg {
450	struct cmsg cmsg;
451	void (pair_cb)(void );
452	void *pair_arg;
453	const char *src_name;
454	struct cpipe *src_pipe;
455	bool complete;
456	struct fiber_cond cond;
457	};
458
459	static void
460	cbus_pair_complete(struct cmsg *cmsg);
461
462	static void
463	cbus_pair_perform(struct cmsg *cmsg)	5,290✔
464	{
465	struct cbus_pair_msg *msg = container_of(cmsg,	5,290✔
466	struct cbus_pair_msg, cmsg);
467	static struct cmsg_hop route[] = {
468	{cbus_pair_complete, NULL},
469	};
470	cmsg_init(cmsg, route);	5,290✔
471	cpipe_create(msg->src_pipe, msg->src_name);	5,290✔
472	if (msg->pair_cb != NULL)	5,290✔
473	msg->pair_cb(msg->pair_arg);	2,574✔
474	cpipe_push(msg->src_pipe, cmsg);	5,290✔
475	}	5,290✔
476
477	static void
478	cbus_pair_complete(struct cmsg *cmsg)	5,287✔
479	{
480	struct cbus_pair_msg *msg = container_of(cmsg,	5,287✔
481	struct cbus_pair_msg, cmsg);
482	msg->complete = true;	5,287✔
483	fiber_cond_signal(&msg->cond);	5,287✔
484	}	5,287✔
485
486	void
487	cbus_pair(const char dest_name, const char src_name,	5,291✔
488	struct cpipe dest_pipe, struct cpipe src_pipe,
489	void (pair_cb)(void ), void *pair_arg,
490	void (process_cb)(struct cbus_endpoint ))
491	{
492	static struct cmsg_hop route[] = {
493	{cbus_pair_perform, NULL},
494	};
495	struct cbus_pair_msg msg;
496
497	cmsg_init(&msg.cmsg, route);	5,291!
498	msg.pair_cb = pair_cb;	5,291✔
499	msg.pair_arg = pair_arg;	5,291✔
500	msg.complete = false;	5,291✔
501	msg.src_name = src_name;	5,291✔
502	msg.src_pipe = src_pipe;	5,291✔
503	fiber_cond_create(&msg.cond);	5,291!
504
505	struct cbus_endpoint *endpoint = cbus_find_endpoint(&cbus, src_name);	5,291!
506	assert(endpoint != NULL);	5,291!
507
508	cpipe_create(dest_pipe, dest_name);	5,291!
509	cpipe_push(dest_pipe, &msg.cmsg);	5,291!
510
511	while (true) {
512	if (process_cb != NULL)	10,583✔
513	process_cb(endpoint);	5,151✔
514	if (msg.complete)	10,580✔
515	break;	5,287✔
516	fiber_cond_wait(&msg.cond);	5,293!
517	}
518	}	5,287✔
519
520	struct cbus_unpair_msg {
521	struct cmsg cmsg;
522	void (unpair_cb)(void );
523	void *unpair_arg;
524	struct cpipe *src_pipe;
525	bool complete;
526	struct fiber_cond cond;
527	};
528
529	static void
530	cbus_unpair_prepare(struct cmsg *cmsg)	4,595✔
531	{
532	struct cbus_unpair_msg *msg = container_of(cmsg,	4,595✔
533	struct cbus_unpair_msg, cmsg);
534	if (msg->unpair_cb != NULL)	4,595✔
535	msg->unpair_cb(msg->unpair_arg);	1,879✔
536	}	4,595✔
537
538	static void
539	cbus_unpair_flush(struct cmsg *cmsg)	4,595✔
540	{
541	(void)cmsg;
542	}	4,595✔
543
544	static void
545	cbus_unpair_complete(struct cmsg *cmsg);
546
547	static void
548	cbus_unpair_perform(struct cmsg *cmsg)	4,595✔
549	{
550	struct cbus_unpair_msg *msg = container_of(cmsg,	4,595✔
551	struct cbus_unpair_msg, cmsg);
552	static struct cmsg_hop route[] = {
553	{cbus_unpair_complete, NULL},
554	};
555	cmsg_init(cmsg, route);	4,595✔
556	cpipe_push(msg->src_pipe, cmsg);	4,595✔
557	cpipe_destroy(msg->src_pipe);	4,595✔
558	}	4,595✔
559
560	static void
561	cbus_unpair_complete(struct cmsg *cmsg)	4,595✔
562	{
563	struct cbus_unpair_msg *msg = container_of(cmsg,	4,595✔
564	struct cbus_unpair_msg, cmsg);
565	msg->complete = true;	4,595✔
566	fiber_cond_signal(&msg->cond);	4,595✔
567	}	4,595✔
568
569	void
570	cbus_unpair(struct cpipe dest_pipe, struct cpipe src_pipe,	4,596✔
571	void (unpair_cb)(void ), void *unpair_arg,
572	void (process_cb)(struct cbus_endpoint ))
573	{
574	struct cmsg_hop route[] = {	4,596✔
575	{cbus_unpair_prepare, src_pipe},
576	{cbus_unpair_flush, dest_pipe},
577	{cbus_unpair_perform, NULL},
578	};
579	struct cbus_unpair_msg msg;
580
581	cmsg_init(&msg.cmsg, route);	4,596!
582	msg.unpair_cb = unpair_cb;	4,596✔
583	msg.unpair_arg = unpair_arg;	4,596✔
584	msg.src_pipe = src_pipe;	4,596✔
585	msg.complete = false;	4,596✔
586	fiber_cond_create(&msg.cond);	4,596!
587
588	cpipe_push(dest_pipe, &msg.cmsg);	4,596!
589
590	struct cbus_endpoint *endpoint = src_pipe->endpoint;	4,596✔
591	while (true) {
592	if (process_cb != NULL)	11,256✔
593	process_cb(endpoint);	5,824!
594	if (msg.complete)	11,256✔
595	break;	4,595✔
596	fiber_cond_wait(&msg.cond);	6,661!
597	}
598
599	cpipe_destroy(dest_pipe);	4,595!
600	}	4,595✔
601
602	void
603	cbus_process(struct cbus_endpoint *endpoint)	10,979,600✔
604	{
605	struct stailq output;
606	stailq_create(&output);	10,979,600!
607	cbus_endpoint_fetch(endpoint, &output);	10,979,600!
608	struct cmsg msg, msg_next;
609	stailq_foreach_entry_safe(msg, msg_next, &output, fifo)	19,346,900!
610	cmsg_deliver(msg);	8,367,380✔
611	}	10,979,500✔
612
613	void
614	cbus_loop(struct cbus_endpoint *endpoint)	4,117,150✔
615	{
616	while (true) {
617	cbus_process(endpoint);	4,117,150✔
618	fiber_check_gc();	4,117,150✔
619	if (fiber_is_cancelled())	4,117,150✔
620	break;	3,345✔
621	fiber_yield();	4,113,800✔
622	}
623	}	3,345✔
624
625	static void
626	cbus_stop_loop_f(struct cmsg *msg)	3,343✔
627	{
628	fiber_cancel(fiber());	3,343!
629	free(msg);	3,343✔
630	}	3,343✔
631
632	void
633	cbus_stop_loop(struct cpipe *pipe)	3,343✔
634	{
635	/*
636	* Hack: static message only works because cmsg_deliver()
637	* is a no-op on the second hop.
638	*/
639	static const struct cmsg_hop route[1] = {
640	{cbus_stop_loop_f, NULL}
641	};
642	struct cmsg *cancel = malloc(sizeof(struct cmsg));	3,343✔
643
644	cmsg_init(cancel, route);	3,343✔
645
646	cpipe_push(pipe, cancel);	3,343✔
647	ev_invoke(pipe->producer, &pipe->flush_input, EV_CUSTOM);	3,343✔
648	}	3,343✔
649

tarantool / tarantool / 11477480897

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous