GCOV - Varnish Cache


/*-
 * Copyright (c) 2005 Varnish Software AS
 * All rights reserved.
 *
 * Author: Dridi Boukelmoune <dridi@varnish-software.com>
 *
 * SPDX-License-Identifier: BSD-2-Clause
 *
 * 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 THE AUTHOR AND CONTRIBUTORS ``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 AUTHOR 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 "config.h"

#include <poll.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include <sys/socket.h>
#include <sys/time.h> /* for MUSL */

#include "vtc.h"
#include "vtcp.h"
#include "vtim.h"
#include "vsa.h"

enum barrier_e {
        BARRIER_NONE = 0,
        BARRIER_COND,
        BARRIER_SOCK,
};

struct barrier {
        unsigned                magic;
#define BARRIER_MAGIC           0x7b54c275
        char                    *name;
        VTAILQ_ENTRY(barrier)   list;
        pthread_mutex_t         mtx;
        pthread_cond_t          cond;

        int                     waiters;
        int                     expected;
        int                     cyclic;

        enum barrier_e          type;
        union {
                int             cond_cycle;
                pthread_t       sock_thread;
        };
};

static VTAILQ_HEAD(, barrier)   barriers = VTAILQ_HEAD_INITIALIZER(barriers);

static struct barrier *
barrier_new(const char *name, struct vtclog *vl)
{
        struct barrier *b;

        ALLOC_OBJ(b, BARRIER_MAGIC);
        AN(b);
        if (!pthread_equal(pthread_self(), vtc_thread))
                vtc_fatal(vl,
                    "Barrier %s can only be created on the top thread", name);
        REPLACE(b->name, name);

        PTOK(pthread_mutex_init(&b->mtx, NULL));
        PTOK(pthread_cond_init(&b->cond, NULL));
        b->waiters = 0;
        b->expected = 0;
        VTAILQ_INSERT_TAIL(&barriers, b, list);
        return (b);
}

/**********************************************************************
 * Init a barrier
 */

static void
barrier_expect(struct barrier *b, const char *av, struct vtclog *vl)
{
        unsigned expected;

        if (b->type != BARRIER_NONE)
                vtc_fatal(vl,
                    "Barrier(%s) use error: already initialized", b->name);

        AZ(b->expected);
        AZ(b->waiters);
        expected = strtoul(av, NULL, 0);
        if (expected < 2)
                vtc_fatal(vl,
                    "Barrier(%s) use error: wrong expectation (%u)",
                    b->name, expected);

        b->expected = expected;
}

static void
barrier_cond(struct barrier *b, const char *av, struct vtclog *vl)
{

        CHECK_OBJ_NOTNULL(b, BARRIER_MAGIC);
        PTOK(pthread_mutex_lock(&b->mtx));
        barrier_expect(b, av, vl);
        b->type = BARRIER_COND;
        PTOK(pthread_mutex_unlock(&b->mtx));
}

static void *
barrier_sock_thread(void *priv)
{
        struct barrier *b;
        struct vtclog *vl;
        const char *err;
        char buf[vsa_suckaddr_len];
        const struct suckaddr *sua;

        char abuf[VTCP_ADDRBUFSIZE], pbuf[VTCP_PORTBUFSIZE];
        int i, sock, *conns;
        struct pollfd pfd[1];

        CAST_OBJ_NOTNULL(b, priv, BARRIER_MAGIC);
        assert(b->type == BARRIER_SOCK);

        PTOK(pthread_mutex_lock(&b->mtx));

        vl = vtc_logopen("%s", b->name);
        pthread_cleanup_push(vtc_logclose, vl);

        sock = VTCP_listen_on(default_listen_addr, NULL, b->expected, &err);
        if (sock < 0) {
                PTOK(pthread_cond_signal(&b->cond));
                PTOK(pthread_mutex_unlock(&b->mtx));
                vtc_fatal(vl, "Barrier(%s) %s fails: %s (errno=%d)",
                    b->name, err, strerror(errno), errno);
        }
        assert(sock > 0);
        VTCP_nonblocking(sock);
        sua = VSA_getsockname(sock, buf, sizeof buf);
        AN(sua);
        VTCP_name(sua, abuf, sizeof abuf, pbuf, sizeof pbuf);

        macro_def(vl, b->name, "addr", "%s", abuf);
        macro_def(vl, b->name, "port", "%s", pbuf);
        if (VSA_Get_Proto(sua) == AF_INET)
                macro_def(vl, b->name, "sock", "%s:%s", abuf, pbuf);
        else
                macro_def(vl, b->name, "sock", "[%s]:%s", abuf, pbuf);

        PTOK(pthread_cond_signal(&b->cond));
        PTOK(pthread_mutex_unlock(&b->mtx));

        conns = calloc(b->expected, sizeof *conns);
        AN(conns);

        while (!vtc_stop && !vtc_error) {
                pfd[0].fd = sock;
                pfd[0].events = POLLIN;

                i = poll(pfd, 1, 100);
                if (i == 0)
                        continue;
                if (i < 0) {
                        if (errno == EINTR)
                                continue;
                        closefd(&sock);
                        vtc_fatal(vl,
                            "Barrier(%s) select fails: %s (errno=%d)",
                            b->name, strerror(errno), errno);
                }
                assert(i == 1);
                assert(b->waiters <= b->expected);
                if (b->waiters == b->expected)
                        vtc_fatal(vl,
                            "Barrier(%s) use error: "
                            "more waiters than the %u expected",
                            b->name, b->expected);

                i = accept(sock, NULL, NULL);
                if (i < 0) {
                        closefd(&sock);
                        vtc_fatal(vl,
                            "Barrier(%s) accept fails: %s (errno=%d)",
                            b->name, strerror(errno), errno);
                }

                /* NB. We don't keep track of the established connections, only
                 *     that connections were made to the barrier's socket.
                 */
                conns[b->waiters] = i;

                if (++b->waiters < b->expected) {
                        vtc_log(vl, 4, "Barrier(%s) wait %u of %u",
                            b->name, b->waiters, b->expected);
                        continue;
                }

                vtc_log(vl, 4, "Barrier(%s) wake %u", b->name, b->expected);
                for (i = 0; i < b->expected; i++)
                        closefd(&conns[i]);

                if (b->cyclic)
                        b->waiters = 0;
                else
                        break;
        }

        if (b->waiters % b->expected > 0) {
                /* wake up outstanding waiters */
                for (i = 0; i < b->waiters; i++)
                        closefd(&conns[i]);
                if (!vtc_error)
                        vtc_fatal(vl, "Barrier(%s) has %u outstanding waiters",
                            b->name, b->waiters);
        }

        macro_undef(vl, b->name, "addr");
        macro_undef(vl, b->name, "port");
        macro_undef(vl, b->name, "sock");
        closefd(&sock);
        free(conns);
        pthread_cleanup_pop(0);
        vtc_logclose(vl);
        return (NULL);
}

static void
barrier_sock(struct barrier *b, const char *av, struct vtclog *vl)
{

        CHECK_OBJ_NOTNULL(b, BARRIER_MAGIC);
        PTOK(pthread_mutex_lock(&b->mtx));
        barrier_expect(b, av, vl);
        b->type = BARRIER_SOCK;

        /* NB. We can use the BARRIER_COND's pthread_cond_t to wait until the
         *     socket is ready for convenience.
         */
        PTOK(pthread_create(&b->sock_thread, NULL, barrier_sock_thread, b));
        PTOK(pthread_cond_wait(&b->cond, &b->mtx));
        PTOK(pthread_mutex_unlock(&b->mtx));
}

static void
barrier_cyclic(struct barrier *b, struct vtclog *vl)
{
        enum barrier_e t;
        int w;

        CHECK_OBJ_NOTNULL(b, BARRIER_MAGIC);

        PTOK(pthread_mutex_lock(&b->mtx));
        t = b->type;
        w = b->waiters;
        PTOK(pthread_mutex_unlock(&b->mtx));

        if (t == BARRIER_NONE)
                vtc_fatal(vl,
                    "Barrier(%s) use error: not initialized", b->name);

        if (w != 0)
                vtc_fatal(vl,
                    "Barrier(%s) use error: already in use", b->name);

        PTOK(pthread_mutex_lock(&b->mtx));
        b->cyclic = 1;
        PTOK(pthread_mutex_unlock(&b->mtx));
}

/**********************************************************************
 * Sync a barrier
 */

static void
barrier_cond_sync(struct barrier *b, struct vtclog *vl)
{
        struct timespec ts;
        int r, w, c;

        CHECK_OBJ_NOTNULL(b, BARRIER_MAGIC);
        assert(b->type == BARRIER_COND);

        PTOK(pthread_mutex_lock(&b->mtx));
        w = b->waiters;
        assert(w <= b->expected);

        if (w == b->expected)
                w = -1;
        else
                b->waiters = ++w;

        c = b->cond_cycle;
        PTOK(pthread_mutex_unlock(&b->mtx));

        if (w < 0)
                vtc_fatal(vl,
                    "Barrier(%s) use error: more waiters than the %u expected",
                    b->name, b->expected);

        PTOK(pthread_mutex_lock(&b->mtx));
        if (w == b->expected) {
                vtc_log(vl, 4, "Barrier(%s) wake %u", b->name, b->expected);
                b->cond_cycle++;
                if (b->cyclic)
                        b->waiters = 0;
                PTOK(pthread_cond_broadcast(&b->cond));
        } else {
                vtc_log(vl, 4, "Barrier(%s) wait %u of %u",
                    b->name, b->waiters, b->expected);
                do {
                        ts = VTIM_timespec(VTIM_real() + .1);
                        r = pthread_cond_timedwait(&b->cond, &b->mtx, &ts);
                        assert(r == 0 || r == ETIMEDOUT);
                } while (!vtc_stop && !vtc_error && r == ETIMEDOUT &&
                    c == b->cond_cycle);
        }
        PTOK(pthread_mutex_unlock(&b->mtx));
}

static void
barrier_sock_sync(const struct barrier *b, struct vtclog *vl)
{
        struct vsb *vsb;
        const char *err;
        char buf[32];
        int i, sock;
        ssize_t sz;

        CHECK_OBJ_NOTNULL(b, BARRIER_MAGIC);
        assert(b->type == BARRIER_SOCK);

        vsb = macro_expandf(vl, "${%s_sock}", b->name);
        vtc_log(vl, 4, "Barrier(%s) sync with socket", b->name);

        sock = VTCP_open(VSB_data(vsb), NULL, 0., &err);
        if (sock < 0)
                vtc_fatal(vl, "Barrier(%s) connection failed: %s",
                    b->name, err);

        VSB_destroy(&vsb);

        sz = read(sock, buf, sizeof buf); /* XXX loop with timeout? */
        i = errno;
        closefd(&sock);

        if (sz < 0)
                vtc_fatal(vl, "Barrier(%s) read failed: %s (errno=%d)",
                    b->name, strerror(i), i);
        if (sz > 0)
                vtc_fatal(vl, "Barrier(%s) unexpected data (%zdB)",
                    b->name, sz);
}

static void
barrier_sync(struct barrier *b, struct vtclog *vl)
{

        CHECK_OBJ_NOTNULL(b, BARRIER_MAGIC);

        switch (b->type) {
        case BARRIER_NONE:
                vtc_fatal(vl,
                    "Barrier(%s) use error: not initialized", b->name);
                break;
        case BARRIER_COND:
                barrier_cond_sync(b, vl);
                break;
        case BARRIER_SOCK:
                barrier_sock_sync(b, vl);
                break;
        default:
                WRONG("Wrong barrier type");
        }
}

/* SECTION: barrier barrier
 *
 * NOTE: This command is available everywhere commands are given.
 *
 * Barriers allows you to synchronize different threads to make sure events
 * occur in the right order. It's even possible to use them in VCL.
 *
 * First, it's necessary to declare the barrier::
 *
 *         barrier bNAME TYPE NUMBER [-cyclic]
 *
 * With the arguments being:
 *
 * bNAME
 *         this is the name of the barrier, used to identify it when you'll
 *         create sync points. It must start with 'b'.
 *
 * TYPE
 *         it can be "cond" (mutex) or "sock" (socket) and sets internal
 *         behavior. If you don't need VCL synchronization, use cond.
 *
 * NUMBER
 *         number of sync point needed to go through the barrier.
 *
 * \-cyclic
 *         if present, the barrier will reset itself and be ready for another
 *         round once gotten through.
 *
 * Then, to add a sync point::
 *
 *         barrier bNAME sync
 *
 * This will block the parent thread until the number of sync points for bNAME
 * reaches the NUMBER given in the barrier declaration.
 *
 * If you wish to synchronize the VCL, you need to declare a "sock" barrier.
 * This will emit a macro definition named "bNAME_sock" that you can use in
 * VCL (after importing the vtc vmod)::
 *
 *         vtc.barrier_sync("${bNAME_sock}");
 *
 * This function returns 0 if everything went well and is the equivalent of
 * ``barrier bNAME sync`` at the VTC top-level.
 *
 *
 */

void
cmd_barrier(CMD_ARGS)
{
        struct barrier *b, *b2;
        int r;

        (void)priv;

        if (av == NULL) {
                /* Reset and free */
                VTAILQ_FOREACH_SAFE(b, &barriers, list, b2) {
                        r = pthread_mutex_trylock(&b->mtx);
                        assert(r == 0 || r == EBUSY);
                        switch (b->type) {
                        case BARRIER_COND:
                                break;
                        case BARRIER_SOCK:
                                PTOK(pthread_join(b->sock_thread, NULL));
                                break;
                        default:
                                WRONG("Wrong barrier type");
                        }
                        if (r == 0)
                                PTOK(pthread_mutex_unlock(&b->mtx));
                }
                return;
        }

        AZ(strcmp(av[0], "barrier"));
        av++;

        VTC_CHECK_NAME(vl, av[0], "Barrier", 'b');
        VTAILQ_FOREACH(b, &barriers, list)
                if (!strcmp(b->name, av[0]))
                        break;
        if (b == NULL)
                b = barrier_new(av[0], vl);
        av++;

        for (; *av != NULL; av++) {
                if (!strcmp(*av, "cond")) {
                        av++;
                        AN(*av);
                        barrier_cond(b, *av, vl);
                        continue;
                }
                if (!strcmp(*av, "sock")) {
                        av++;
                        AN(*av);
                        barrier_sock(b, *av, vl);
                        continue;
                }
                if (!strcmp(*av, "sync")) {
                        barrier_sync(b, vl);
                        continue;
                }
                if (!strcmp(*av, "-cyclic")) {
                        barrier_cyclic(b, vl);
                        continue;
                }
                vtc_fatal(vl, "Unknown barrier argument: %s", *av);
        }
}

		varnish-cache/bin/varnishtest/vtest2/src/vtc_barrier.c
0		/*-
1		* Copyright (c) 2005 Varnish Software AS
2		* All rights reserved.
3		*
4		* Author: Dridi Boukelmoune <dridi@varnish-software.com>
5		*
6		* SPDX-License-Identifier: BSD-2-Clause
7		*
8		* Redistribution and use in source and binary forms, with or without
9		* modification, are permitted provided that the following conditions
10		* are met:
11		* 1. Redistributions of source code must retain the above copyright
12		* notice, this list of conditions and the following disclaimer.
13		* 2. Redistributions in binary form must reproduce the above copyright
14		* notice, this list of conditions and the following disclaimer in the
15		* documentation and/or other materials provided with the distribution.
16		*
17		* THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
18		* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
19		* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
20		* ARE DISCLAIMED. IN NO EVENT SHALL AUTHOR OR CONTRIBUTORS BE LIABLE
21		* FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
22		* DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
23		* OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
24		* HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
25		* LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
26		* OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
27		* SUCH DAMAGE.
28		*/
29
30		#include "config.h"
31
32		#include <poll.h>
33		#include <stdio.h>
34		#include <stdlib.h>
35		#include <string.h>
36		#include <unistd.h>
37
38		#include <sys/socket.h>
39		#include <sys/time.h> /* for MUSL */
40
41		#include "vtc.h"
42		#include "vtcp.h"
43		#include "vtim.h"
44		#include "vsa.h"
45
46		enum barrier_e {
47		BARRIER_NONE = 0,
48		BARRIER_COND,
49		BARRIER_SOCK,
50		};
51
52		struct barrier {
53		unsigned magic;
54		#define BARRIER_MAGIC 0x7b54c275
55		char *name;
56		VTAILQ_ENTRY(barrier) list;
57		pthread_mutex_t mtx;
58		pthread_cond_t cond;
59
60		int waiters;
61		int expected;
62		int cyclic;
63
64		enum barrier_e type;
65		union {
66		int cond_cycle;
67		pthread_t sock_thread;
68		};
69		};
70
71		static VTAILQ_HEAD(, barrier) barriers = VTAILQ_HEAD_INITIALIZER(barriers);
72
73		static struct barrier *
74	9200	barrier_new(const char name, struct vtclog vl)
75		{
76		struct barrier *b;
77
78	9200	ALLOC_OBJ(b, BARRIER_MAGIC);
79	9200	AN(b);
80	9200	if (!pthread_equal(pthread_self(), vtc_thread))
81	0	vtc_fatal(vl,
82	0	"Barrier %s can only be created on the top thread", name);
83	9200	REPLACE(b->name, name);
84
85	9200	PTOK(pthread_mutex_init(&b->mtx, NULL));
86	9200	PTOK(pthread_cond_init(&b->cond, NULL));
87	9200	b->waiters = 0;
88	9200	b->expected = 0;
89	9200	VTAILQ_INSERT_TAIL(&barriers, b, list);
90	9200	return (b);
91		}
92
93		/**********************************************************************
94		* Init a barrier
95		*/
96
97		static void
98	9200	barrier_expect(struct barrier b, const char av, struct vtclog *vl)
99		{
100		unsigned expected;
101
102	9200	if (b->type != BARRIER_NONE)
103	0	vtc_fatal(vl,
104	0	"Barrier(%s) use error: already initialized", b->name);
105
106	9200	AZ(b->expected);
107	9200	AZ(b->waiters);
108	9200	expected = strtoul(av, NULL, 0);
109	9200	if (expected < 2)
110	0	vtc_fatal(vl,
111		"Barrier(%s) use error: wrong expectation (%u)",
112	0	b->name, expected);
113
114	9200	b->expected = expected;
115	9200	}
116
117		static void
118	6920	barrier_cond(struct barrier b, const char av, struct vtclog *vl)
119		{
120
121	6920	CHECK_OBJ_NOTNULL(b, BARRIER_MAGIC);
122	6920	PTOK(pthread_mutex_lock(&b->mtx));
123	6920	barrier_expect(b, av, vl);
124	6920	b->type = BARRIER_COND;
125	6920	PTOK(pthread_mutex_unlock(&b->mtx));
126	6920	}
127
128		static void *
129	2280	barrier_sock_thread(void *priv)
130		{
131		struct barrier *b;
132		struct vtclog *vl;
133		const char *err;
134	2280	char buf[vsa_suckaddr_len];
135		const struct suckaddr *sua;
136
137		char abuf[VTCP_ADDRBUFSIZE], pbuf[VTCP_PORTBUFSIZE];
138		int i, sock, *conns;
139		struct pollfd pfd[1];
140
141	2280	CAST_OBJ_NOTNULL(b, priv, BARRIER_MAGIC);
142	2280	assert(b->type == BARRIER_SOCK);
143
144	2280	PTOK(pthread_mutex_lock(&b->mtx));
145
146	2280	vl = vtc_logopen("%s", b->name);
147	2280	pthread_cleanup_push(vtc_logclose, vl);
148
149	2280	sock = VTCP_listen_on(default_listen_addr, NULL, b->expected, &err);
150	2280	if (sock < 0) {
151	0	PTOK(pthread_cond_signal(&b->cond));
152	0	PTOK(pthread_mutex_unlock(&b->mtx));
153	0	vtc_fatal(vl, "Barrier(%s) %s fails: %s (errno=%d)",
154	0	b->name, err, strerror(errno), errno);
155		}
156	2280	assert(sock > 0);
157	2280	VTCP_nonblocking(sock);
158	2280	sua = VSA_getsockname(sock, buf, sizeof buf);
159	2280	AN(sua);
160	2280	VTCP_name(sua, abuf, sizeof abuf, pbuf, sizeof pbuf);
161
162	2280	macro_def(vl, b->name, "addr", "%s", abuf);
163	2280	macro_def(vl, b->name, "port", "%s", pbuf);
164	2280	if (VSA_Get_Proto(sua) == AF_INET)
165	2280	macro_def(vl, b->name, "sock", "%s:%s", abuf, pbuf);
166		else
167	0	macro_def(vl, b->name, "sock", "[%s]:%s", abuf, pbuf);
168
169	2280	PTOK(pthread_cond_signal(&b->cond));
170	2280	PTOK(pthread_mutex_unlock(&b->mtx));
171
172	2280	conns = calloc(b->expected, sizeof *conns);
173	2280	AN(conns);
174
175	49141	while (!vtc_stop && !vtc_error) {
176	48741	pfd[0].fd = sock;
177	48741	pfd[0].events = POLLIN;
178
179	48741	i = poll(pfd, 1, 100);
180	48741	if (i == 0)
181	41421	continue;
182	7320	if (i < 0) {
183	0	if (errno == EINTR)
184	0	continue;
185	0	closefd(&sock);
186	0	vtc_fatal(vl,
187		"Barrier(%s) select fails: %s (errno=%d)",
188	0	b->name, strerror(errno), errno);
189		}
190	7320	assert(i == 1);
191	7320	assert(b->waiters <= b->expected);
192	7320	if (b->waiters == b->expected)
193	0	vtc_fatal(vl,
194		"Barrier(%s) use error: "
195		"more waiters than the %u expected",
196	0	b->name, b->expected);
197
198	7320	i = accept(sock, NULL, NULL);
199	7320	if (i < 0) {
200	0	closefd(&sock);
201	0	vtc_fatal(vl,
202		"Barrier(%s) accept fails: %s (errno=%d)",
203	0	b->name, strerror(errno), errno);
204		}
205
206		/* NB. We don't keep track of the established connections, only
207		* that connections were made to the barrier's socket.
208		*/
209	7320	conns[b->waiters] = i;
210
211	7320	if (++b->waiters < b->expected) {
212	9040	vtc_log(vl, 4, "Barrier(%s) wait %u of %u",
213	4520	b->name, b->waiters, b->expected);
214	4520	continue;
215		}
216
217	2800	vtc_log(vl, 4, "Barrier(%s) wake %u", b->name, b->expected);
218	10120	for (i = 0; i < b->expected; i++)
219	7320	closefd(&conns[i]);
220
221	2800	if (b->cyclic)
222	920	b->waiters = 0;
223		else
224	1880	break;
225		}
226
227	2280	if (b->waiters % b->expected > 0) {
228		/* wake up outstanding waiters */
229	0	for (i = 0; i < b->waiters; i++)
230	0	closefd(&conns[i]);
231	0	if (!vtc_error)
232	0	vtc_fatal(vl, "Barrier(%s) has %u outstanding waiters",
233	0	b->name, b->waiters);
234	0	}
235
236	2280	macro_undef(vl, b->name, "addr");
237	2280	macro_undef(vl, b->name, "port");
238	2280	macro_undef(vl, b->name, "sock");
239	2280	closefd(&sock);
240	2280	free(conns);
241	2280	pthread_cleanup_pop(0);
242	2280	vtc_logclose(vl);
243		return (NULL);
244	2280	}
245
246		static void
247	2280	barrier_sock(struct barrier b, const char av, struct vtclog *vl)
248		{
249
250	2280	CHECK_OBJ_NOTNULL(b, BARRIER_MAGIC);
251	2280	PTOK(pthread_mutex_lock(&b->mtx));
252	2280	barrier_expect(b, av, vl);
253	2280	b->type = BARRIER_SOCK;
254
255		/* NB. We can use the BARRIER_COND's pthread_cond_t to wait until the
256		* socket is ready for convenience.
257		*/
258	2280	PTOK(pthread_create(&b->sock_thread, NULL, barrier_sock_thread, b));
259	2280	PTOK(pthread_cond_wait(&b->cond, &b->mtx));
260	2280	PTOK(pthread_mutex_unlock(&b->mtx));
261	2280	}
262
263		static void
264	760	barrier_cyclic(struct barrier b, struct vtclog vl)
265		{
266		enum barrier_e t;
267		int w;
268
269	760	CHECK_OBJ_NOTNULL(b, BARRIER_MAGIC);
270
271	760	PTOK(pthread_mutex_lock(&b->mtx));
272	760	t = b->type;
273	760	w = b->waiters;
274	760	PTOK(pthread_mutex_unlock(&b->mtx));
275
276	760	if (t == BARRIER_NONE)
277	0	vtc_fatal(vl,
278	0	"Barrier(%s) use error: not initialized", b->name);
279
280	760	if (w != 0)
281	0	vtc_fatal(vl,
282	0	"Barrier(%s) use error: already in use", b->name);
283
284	760	PTOK(pthread_mutex_lock(&b->mtx));
285	760	b->cyclic = 1;
286	760	PTOK(pthread_mutex_unlock(&b->mtx));
287	760	}
288
289		/**********************************************************************
290		* Sync a barrier
291		*/
292
293		static void
294	17160	barrier_cond_sync(struct barrier b, struct vtclog vl)
295		{
296		struct timespec ts;
297		int r, w, c;
298
299	17160	CHECK_OBJ_NOTNULL(b, BARRIER_MAGIC);
300	17160	assert(b->type == BARRIER_COND);
301
302	17160	PTOK(pthread_mutex_lock(&b->mtx));
303	17160	w = b->waiters;
304	17160	assert(w <= b->expected);
305
306	17160	if (w == b->expected)
307	0	w = -1;
308		else
309	17160	b->waiters = ++w;
310
311	17160	c = b->cond_cycle;
312	17160	PTOK(pthread_mutex_unlock(&b->mtx));
313
314	17160	if (w < 0)
315	0	vtc_fatal(vl,
316		"Barrier(%s) use error: more waiters than the %u expected",
317	0	b->name, b->expected);
318
319	17160	PTOK(pthread_mutex_lock(&b->mtx));
320	17160	if (w == b->expected) {
321	7480	vtc_log(vl, 4, "Barrier(%s) wake %u", b->name, b->expected);
322	7480	b->cond_cycle++;
323	7480	if (b->cyclic)
324	1000	b->waiters = 0;
325	7480	PTOK(pthread_cond_broadcast(&b->cond));
326	7480	} else {
327	19360	vtc_log(vl, 4, "Barrier(%s) wait %u of %u",
328	9680	b->name, b->waiters, b->expected);
329	9680	do {
330	31538	ts = VTIM_timespec(VTIM_real() + .1);
331	31538	r = pthread_cond_timedwait(&b->cond, &b->mtx, &ts);
332	31538	assert(r == 0 \|\| r == ETIMEDOUT);
333	53398	} while (!vtc_stop && !vtc_error && r == ETIMEDOUT &&
334	21860	c == b->cond_cycle);
335		}
336	17160	PTOK(pthread_mutex_unlock(&b->mtx));
337	17160	}
338
339		static void
340	4320	barrier_sock_sync(const struct barrier b, struct vtclog vl)
341		{
342		struct vsb *vsb;
343		const char *err;
344		char buf[32];
345		int i, sock;
346		ssize_t sz;
347
348	4320	CHECK_OBJ_NOTNULL(b, BARRIER_MAGIC);
349	4320	assert(b->type == BARRIER_SOCK);
350
351	4320	vsb = macro_expandf(vl, "${%s_sock}", b->name);
352	4320	vtc_log(vl, 4, "Barrier(%s) sync with socket", b->name);
353
354	4320	sock = VTCP_open(VSB_data(vsb), NULL, 0., &err);
355	4320	if (sock < 0)
356	0	vtc_fatal(vl, "Barrier(%s) connection failed: %s",
357	0	b->name, err);
358
359	4320	VSB_destroy(&vsb);
360
361	4320	sz = read(sock, buf, sizeof buf); /* XXX loop with timeout? */
362	4320	i = errno;
363	4320	closefd(&sock);
364
365	4320	if (sz < 0)
366	0	vtc_fatal(vl, "Barrier(%s) read failed: %s (errno=%d)",
367	0	b->name, strerror(i), i);
368	4320	if (sz > 0)
369	0	vtc_fatal(vl, "Barrier(%s) unexpected data (%zdB)",
370	0	b->name, sz);
371	4320	}
372
373		static void
374	21476	barrier_sync(struct barrier b, struct vtclog vl)
375		{
376
377	21476	CHECK_OBJ_NOTNULL(b, BARRIER_MAGIC);
378
379	21476	switch (b->type) {
380		case BARRIER_NONE:
381	0	vtc_fatal(vl,
382	0	"Barrier(%s) use error: not initialized", b->name);
383		break;
384		case BARRIER_COND:
385	17156	barrier_cond_sync(b, vl);
386	17156	break;
387		case BARRIER_SOCK:
388	4320	barrier_sock_sync(b, vl);
389	4320	break;
390		default:
391	0	WRONG("Wrong barrier type");
392	0	}
393	21476	}
394
395		/* SECTION: barrier barrier
396		*
397		* NOTE: This command is available everywhere commands are given.
398		*
399		* Barriers allows you to synchronize different threads to make sure events
400		* occur in the right order. It's even possible to use them in VCL.
401		*
402		* First, it's necessary to declare the barrier::
403		*
404		* barrier bNAME TYPE NUMBER [-cyclic]
405		*
406		* With the arguments being:
407		*
408		* bNAME
409		* this is the name of the barrier, used to identify it when you'll
410		* create sync points. It must start with 'b'.
411		*
412		* TYPE
413		* it can be "cond" (mutex) or "sock" (socket) and sets internal
414		* behavior. If you don't need VCL synchronization, use cond.
415		*
416		* NUMBER
417		* number of sync point needed to go through the barrier.
418		*
419		* \-cyclic
420		* if present, the barrier will reset itself and be ready for another
421		* round once gotten through.
422		*
423		* Then, to add a sync point::
424		*
425		* barrier bNAME sync
426		*
427		* This will block the parent thread until the number of sync points for bNAME
428		* reaches the NUMBER given in the barrier declaration.
429		*
430		* If you wish to synchronize the VCL, you need to declare a "sock" barrier.
431		* This will emit a macro definition named "bNAME_sock" that you can use in
432		* VCL (after importing the vtc vmod)::
433		*
434		* vtc.barrier_sync("${bNAME_sock}");
435		*
436		* This function returns 0 if everything went well and is the equivalent of
437		* ``barrier bNAME sync`` at the VTC top-level.
438		*
439		*
440		*/
441
442		void
443	71760	cmd_barrier(CMD_ARGS)
444		{
445		struct barrier b, b2;
446		int r;
447
448	71760	(void)priv;
449
450	71760	if (av == NULL) {
451		/* Reset and free */
452	50280	VTAILQ_FOREACH_SAFE(b, &barriers, list, b2) {
453	9200	r = pthread_mutex_trylock(&b->mtx);
454	9200	assert(r == 0 \|\| r == EBUSY);
455	9200	switch (b->type) {
456		case BARRIER_COND:
457	6920	break;
458		case BARRIER_SOCK:
459	2280	PTOK(pthread_join(b->sock_thread, NULL));
460	2280	break;
461		default:
462	0	WRONG("Wrong barrier type");
463	0	}
464	9200	if (r == 0)
465	9200	PTOK(pthread_mutex_unlock(&b->mtx));
466	9200	}
467	41080	return;
468		}
469
470	30680	AZ(strcmp(av[0], "barrier"));
471	30680	av++;
472
473	30680	VTC_CHECK_NAME(vl, av[0], "Barrier", 'b');
474	59514	VTAILQ_FOREACH(b, &barriers, list)
475	50314	if (!strcmp(b->name, av[0]))
476	21480	break;
477	30680	if (b == NULL)
478	9200	b = barrier_new(av[0], vl);
479	30680	av++;
480
481	62119	for (; *av != NULL; av++) {
482	31439	if (!strcmp(*av, "cond")) {
483	6920	av++;
484	6920	AN(*av);
485	6920	barrier_cond(b, *av, vl);
486	6920	continue;
487		}
488	24519	if (!strcmp(*av, "sock")) {
489	2280	av++;
490	2280	AN(*av);
491	2280	barrier_sock(b, *av, vl);
492	2280	continue;
493		}
494	22239	if (!strcmp(*av, "sync")) {
495	21479	barrier_sync(b, vl);
496	21479	continue;
497		}
498	760	if (!strcmp(*av, "-cyclic")) {
499	760	barrier_cyclic(b, vl);