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);
        }
}

		vinyl-cache/bin/vinyltest/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
43		#include "vtcp.h"
44		#include "vtim.h"
45		#include "vsa.h"
46
47		enum barrier_e {
48		BARRIER_NONE = 0,
49		BARRIER_COND,
50		BARRIER_SOCK,
51		};
52
53		struct barrier {
54		unsigned magic;
55		#define BARRIER_MAGIC 0x7b54c275
56		char *name;
57		VTAILQ_ENTRY(barrier) list;
58		pthread_mutex_t mtx;
59		pthread_cond_t cond;
60
61		int waiters;
62		int expected;
63		int cyclic;
64
65		enum barrier_e type;
66		union {
67		int cond_cycle;
68		pthread_t sock_thread;
69		};
70		};
71
72		static VTAILQ_HEAD(, barrier) barriers = VTAILQ_HEAD_INITIALIZER(barriers);
73
74		static struct barrier *
75	1215	barrier_new(const char name, struct vtclog vl)
76		{
77		struct barrier *b;
78
79	1215	ALLOC_OBJ(b, BARRIER_MAGIC);
80	1215	AN(b);
81	1215	if (!pthread_equal(pthread_self(), vtc_thread))
82	0	vtc_fatal(vl,
83	0	"Barrier %s can only be created on the top thread", name);
84	1215	REPLACE(b->name, name);
85
86	1215	PTOK(pthread_mutex_init(&b->mtx, NULL));
87	1215	PTOK(pthread_cond_init(&b->cond, NULL));
88	1215	b->waiters = 0;
89	1215	b->expected = 0;
90	1215	VTAILQ_INSERT_TAIL(&barriers, b, list);
91	1215	return (b);
92		}
93
94		/**********************************************************************
95		* Init a barrier
96		*/
97
98		static void
99	1215	barrier_expect(struct barrier b, const char av, struct vtclog *vl)
100		{
101		unsigned expected;
102
103	1215	if (b->type != BARRIER_NONE)
104	0	vtc_fatal(vl,
105	0	"Barrier(%s) use error: already initialized", b->name);
106
107	1215	AZ(b->expected);
108	1215	AZ(b->waiters);
109	1215	expected = strtoul(av, NULL, 0);
110	1215	if (expected < 2)
111	0	vtc_fatal(vl,
112		"Barrier(%s) use error: wrong expectation (%u)",
113	0	b->name, expected);
114
115	1215	b->expected = expected;
116	1215	}
117
118		static void
119	910	barrier_cond(struct barrier b, const char av, struct vtclog *vl)
120		{
121
122	910	CHECK_OBJ_NOTNULL(b, BARRIER_MAGIC);
123	910	PTOK(pthread_mutex_lock(&b->mtx));
124	910	barrier_expect(b, av, vl);
125	910	b->type = BARRIER_COND;
126	910	PTOK(pthread_mutex_unlock(&b->mtx));
127	910	}
128
129		static void *
130	305	barrier_sock_thread(void *priv)
131		{
132		struct barrier *b;
133		struct vtclog *vl;
134		const char *err;
135	305	char buf[vsa_suckaddr_len];
136		const struct suckaddr *sua;
137
138		char abuf[VTCP_ADDRBUFSIZE], pbuf[VTCP_PORTBUFSIZE];
139		int i, sock, *conns;
140		struct pollfd pfd[1];
141
142	305	CAST_OBJ_NOTNULL(b, priv, BARRIER_MAGIC);
143	305	assert(b->type == BARRIER_SOCK);
144
145	305	PTOK(pthread_mutex_lock(&b->mtx));
146
147	305	vl = vtc_logopen("%s", b->name);
148	305	pthread_cleanup_push(vtc_logclose, vl);
149
150	305	sock = VTCP_listen_on(default_listen_addr, NULL, b->expected, &err);
151	305	if (sock < 0) {
152	0	PTOK(pthread_cond_signal(&b->cond));
153	0	PTOK(pthread_mutex_unlock(&b->mtx));
154	0	vtc_fatal(vl, "Barrier(%s) %s fails: %s (errno=%d)",
155	0	b->name, err, strerror(errno), errno);
156		}
157	305	assert(sock > 0);
158	305	VTCP_nonblocking(sock);
159	305	sua = VSA_getsockname(sock, buf, sizeof buf);
160	305	AN(sua);
161	305	VTCP_name(sua, abuf, sizeof abuf, pbuf, sizeof pbuf);
162
163	305	macro_def(vl, b->name, "addr", "%s", abuf);
164	305	macro_def(vl, b->name, "port", "%s", pbuf);
165	305	if (VSA_Get_Proto(sua) == AF_INET)
166	305	macro_def(vl, b->name, "sock", "%s:%s", abuf, pbuf);
167		else
168	0	macro_def(vl, b->name, "sock", "[%s]:%s", abuf, pbuf);
169
170	305	PTOK(pthread_cond_signal(&b->cond));
171	305	PTOK(pthread_mutex_unlock(&b->mtx));
172
173	305	conns = calloc(b->expected, sizeof *conns);
174	305	AN(conns);
175
176	6544	while (!vtc_stop && !vtc_error) {
177	6484	pfd[0].fd = sock;
178	6484	pfd[0].events = POLLIN;
179
180	6484	i = poll(pfd, 1, 100);
181	6484	if (i == 0)
182	5469	continue;
183	1015	if (i < 0) {
184	0	if (errno == EINTR)
185	0	continue;
186	0	closefd(&sock);
187	0	vtc_fatal(vl,
188		"Barrier(%s) select fails: %s (errno=%d)",
189	0	b->name, strerror(errno), errno);
190		}
191	1015	assert(i == 1);
192	1015	assert(b->waiters <= b->expected);
193	1015	if (b->waiters == b->expected)
194	0	vtc_fatal(vl,
195		"Barrier(%s) use error: "
196		"more waiters than the %u expected",
197	0	b->name, b->expected);
198
199	1015	i = accept(sock, NULL, NULL);
200	1015	if (i < 0) {
201	0	closefd(&sock);
202	0	vtc_fatal(vl,
203		"Barrier(%s) accept fails: %s (errno=%d)",
204	0	b->name, strerror(errno), errno);
205		}
206
207		/* NB. We don't keep track of the established connections, only
208		* that connections were made to the barrier's socket.
209		*/
210	1015	conns[b->waiters] = i;
211
212	1015	if (++b->waiters < b->expected) {
213	1270	vtc_log(vl, 4, "Barrier(%s) wait %u of %u",
214	635	b->name, b->waiters, b->expected);
215	635	continue;
216		}
217
218	380	vtc_log(vl, 4, "Barrier(%s) wake %u", b->name, b->expected);
219	1395	for (i = 0; i < b->expected; i++)
220	1015	closefd(&conns[i]);
221
222	380	if (b->cyclic)
223	135	b->waiters = 0;
224		else
225	245	break;
226		}
227
228	305	if (b->waiters % b->expected > 0) {
229		/* wake up outstanding waiters */
230	0	for (i = 0; i < b->waiters; i++)
231	0	closefd(&conns[i]);
232	0	if (!vtc_error)
233	0	vtc_fatal(vl, "Barrier(%s) has %u outstanding waiters",
234	0	b->name, b->waiters);
235	0	}
236
237	305	macro_undef(vl, b->name, "addr");
238	305	macro_undef(vl, b->name, "port");
239	305	macro_undef(vl, b->name, "sock");
240	305	closefd(&sock);
241	305	free(conns);
242	305	pthread_cleanup_pop(0);
243	305	vtc_logclose(vl);
244		return (NULL);
245	305	}
246
247		static void
248	305	barrier_sock(struct barrier b, const char av, struct vtclog *vl)
249		{
250
251	305	CHECK_OBJ_NOTNULL(b, BARRIER_MAGIC);
252	305	PTOK(pthread_mutex_lock(&b->mtx));
253	305	barrier_expect(b, av, vl);
254	305	b->type = BARRIER_SOCK;
255
256		/* NB. We can use the BARRIER_COND's pthread_cond_t to wait until the
257		* socket is ready for convenience.
258		*/
259	305	PTOK(pthread_create(&b->sock_thread, NULL, barrier_sock_thread, b));
260	305	PTOK(pthread_cond_wait(&b->cond, &b->mtx));
261	305	PTOK(pthread_mutex_unlock(&b->mtx));
262	305	}
263
264		static void
265	115	barrier_cyclic(struct barrier b, struct vtclog vl)
266		{
267		enum barrier_e t;
268		int w;
269
270	115	CHECK_OBJ_NOTNULL(b, BARRIER_MAGIC);
271
272	115	PTOK(pthread_mutex_lock(&b->mtx));
273	115	t = b->type;
274	115	w = b->waiters;
275	115	PTOK(pthread_mutex_unlock(&b->mtx));
276
277	115	if (t == BARRIER_NONE)
278	0	vtc_fatal(vl,
279	0	"Barrier(%s) use error: not initialized", b->name);
280
281	115	if (w != 0)
282	0	vtc_fatal(vl,
283	0	"Barrier(%s) use error: already in use", b->name);
284
285	115	PTOK(pthread_mutex_lock(&b->mtx));
286	115	b->cyclic = 1;
287	115	PTOK(pthread_mutex_unlock(&b->mtx));
288	115	}
289
290		/**********************************************************************
291		* Sync a barrier
292		*/
293
294		static void
295	2305	barrier_cond_sync(struct barrier b, struct vtclog vl)
296		{
297		struct timespec ts;
298		int r, w, c;
299
300	2305	CHECK_OBJ_NOTNULL(b, BARRIER_MAGIC);
301	2305	assert(b->type == BARRIER_COND);
302
303	2305	PTOK(pthread_mutex_lock(&b->mtx));
304	2305	w = b->waiters;
305	2305	assert(w <= b->expected);
306
307	2305	if (w == b->expected)
308	0	w = -1;
309		else
310	2305	b->waiters = ++w;
311
312	2305	c = b->cond_cycle;
313	2305	PTOK(pthread_mutex_unlock(&b->mtx));
314
315	2305	if (w < 0)
316	0	vtc_fatal(vl,
317		"Barrier(%s) use error: more waiters than the %u expected",
318	0	b->name, b->expected);
319
320	2305	PTOK(pthread_mutex_lock(&b->mtx));
321	2305	if (w == b->expected) {
322	990	vtc_log(vl, 4, "Barrier(%s) wake %u", b->name, b->expected);
323	990	b->cond_cycle++;
324	990	if (b->cyclic)
325	145	b->waiters = 0;
326	990	PTOK(pthread_cond_broadcast(&b->cond));
327	990	} else {
328	2630	vtc_log(vl, 4, "Barrier(%s) wait %u of %u",
329	1315	b->name, b->waiters, b->expected);
330	1315	do {
331	4142	ts = VTIM_timespec(VTIM_real() + .1);
332	4142	r = pthread_cond_timedwait(&b->cond, &b->mtx, &ts);
333	4142	assert(r == 0 \|\| r == ETIMEDOUT);
334	6971	} while (!vtc_stop && !vtc_error && r == ETIMEDOUT &&
335	2829	c == b->cond_cycle);
336		}
337	2305	PTOK(pthread_mutex_unlock(&b->mtx));
338	2305	}
339
340		static void
341	620	barrier_sock_sync(const struct barrier b, struct vtclog vl)
342		{
343		struct vsb *vsb;
344		const char *err;
345		char buf[32];
346		int i, sock;
347		ssize_t sz;
348
349	620	CHECK_OBJ_NOTNULL(b, BARRIER_MAGIC);
350	620	assert(b->type == BARRIER_SOCK);
351
352	620	vsb = macro_expandf(vl, "${%s_sock}", b->name);
353	620	vtc_log(vl, 4, "Barrier(%s) sync with socket", b->name);
354
355	620	sock = VTCP_open(VSB_data(vsb), NULL, 0., &err);
356	620	if (sock < 0)
357	0	vtc_fatal(vl, "Barrier(%s) connection failed: %s",
358	0	b->name, err);
359
360	620	VSB_destroy(&vsb);
361
362	620	sz = read(sock, buf, sizeof buf); /* XXX loop with timeout? */
363	620	i = errno;
364	620	closefd(&sock);
365
366	620	if (sz < 0)
367	0	vtc_fatal(vl, "Barrier(%s) read failed: %s (errno=%d)",
368	0	b->name, strerror(i), i);
369	620	if (sz > 0)
370	0	vtc_fatal(vl, "Barrier(%s) unexpected data (%zdB)",
371	0	b->name, sz);
372	620	}
373
374		static void
375	2925	barrier_sync(struct barrier b, struct vtclog vl)
376		{
377
378	2925	CHECK_OBJ_NOTNULL(b, BARRIER_MAGIC);
379
380	2925	switch (b->type) {
381		case BARRIER_NONE:
382	0	vtc_fatal(vl,
383	0	"Barrier(%s) use error: not initialized", b->name);
384		break;
385		case BARRIER_COND:
386	2305	barrier_cond_sync(b, vl);
387	2305	break;
388		case BARRIER_SOCK:
389	620	barrier_sock_sync(b, vl);
390	620	break;
391		default:
392	0	WRONG("Wrong barrier type");
393	0	}
394	2925	}
395
396		/* SECTION: barrier barrier
397		*
398		* NOTE: This command is available everywhere commands are given.
399		*
400		* Barriers allows you to synchronize different threads to make sure events
401		* occur in the right order. It's even possible to use them in VCL.
402		*
403		* First, it's necessary to declare the barrier::
404		*
405		* barrier bNAME TYPE NUMBER [-cyclic]
406		*
407		* With the arguments being:
408		*
409		* bNAME
410		* this is the name of the barrier, used to identify it when you'll
411		* create sync points. It must start with 'b'.
412		*
413		* TYPE
414		* it can be "cond" (mutex) or "sock" (socket) and sets internal
415		* behavior. If you don't need VCL synchronization, use cond.
416		*
417		* NUMBER
418		* number of sync point needed to go through the barrier.
419		*
420		* \-cyclic
421		* if present, the barrier will reset itself and be ready for another
422		* round once gotten through.
423		*
424		* Then, to add a sync point::
425		*
426		* barrier bNAME sync
427		*
428		* This will block the parent thread until the number of sync points for bNAME
429		* reaches the NUMBER given in the barrier declaration.
430		*
431		* If you wish to synchronize the VCL, you need to declare a "sock" barrier.
432		* This will emit a macro definition named "bNAME_sock" that you can use in
433		* VCL (after importing the vtc vmod)::
434		*
435		* vtc.barrier_sync("${bNAME_sock}");
436		*
437		* This function returns 0 if everything went well and is the equivalent of
438		* ``barrier bNAME sync`` at the VTC top-level.
439		*
440		*
441		*/
442
443		void
444	9594	cmd_barrier(CMD_ARGS)
445		{
446		struct barrier b, b2;
447		int r;
448
449	9594	(void)priv;
450
451	9594	if (av == NULL) {
452		/* Reset and free */
453	6670	VTAILQ_FOREACH_SAFE(b, &barriers, list, b2) {
454	1215	r = pthread_mutex_trylock(&b->mtx);
455	1215	assert(r == 0 \|\| r == EBUSY);
456	1215	switch (b->type) {
457		case BARRIER_COND:
458	910	break;
459		case BARRIER_SOCK:
460	305	PTOK(pthread_join(b->sock_thread, NULL));
461	305	break;
462		default:
463	0	WRONG("Wrong barrier type");
464	0	}
465	1215	if (r == 0)
466	1215	PTOK(pthread_mutex_unlock(&b->mtx));
467	1215	}
468	5455	return;
469		}
470
471	4139	AZ(strcmp(av[0], "barrier"));
472	4139	av++;
473
474	4139	VTC_CHECK_NAME(vl, av[0], "Barrier", 'b');
475	8022	VTAILQ_FOREACH(b, &barriers, list)
476	6807	if (!strcmp(b->name, av[0]))
477	2924	break;
478	4139	if (b == NULL)
479	1215	b = barrier_new(av[0], vl);
480	4139	av++;
481
482	8394	for (; *av != NULL; av++) {
483	4255	if (!strcmp(*av, "cond")) {
484	910	av++;
485	910	AN(*av);
486	910	barrier_cond(b, *av, vl);
487	910	continue;
488		}
489	3345	if (!strcmp(*av, "sock")) {
490	305	av++;
491	305	AN(*av);
492	305	barrier_sock(b, *av, vl);
493	305	continue;
494		}
495	3040	if (!strcmp(*av, "sync")) {
496	2925	barrier_sync(b, vl);
497	2925	continue;
498		}
499	115	if (!strcmp(*av, "-cyclic")) {