==> src/chan.c <==1/* A ring of fixed-size messages under a mutex, with two condition variables on CLOCK_MONOTONIC. */
2#include "mesh/chan.h"
3#include "mesh/buf.h"
4
5#include <errno.h>
6#include <pthread.h>
7#include <stdlib.h>
8#include <string.h>
9#include <time.h>
10
11struct mesh_chan {
12 pthread_mutex_t mu;
13 pthread_cond_t can_recv, can_send;
14 size_t item, bound;
15 unsigned char *ring;
16 size_t cap, head, len;
17 int refs, senders, rx_open;
18 void (*drop)(void *item);
19};
20
21int64_t mesh_mono_us(void) {
22 struct timespec t;
23 clock_gettime(CLOCK_MONOTONIC, &t);
24 return (int64_t)t.tv_sec * 1000000 + t.tv_nsec / 1000;
25}
26
27int64_t mesh_mono_ns(void) {
28 struct timespec t;
29 clock_gettime(CLOCK_MONOTONIC, &t);
30 return (int64_t)t.tv_sec * 1000000000 + t.tv_nsec;
31}
32
33int64_t mesh_mono_ms(void) { return mesh_mono_us() / 1000; }
34
35void mesh_sleep_us(int64_t us) {
36 if (us <= 0) return;
37 struct timespec t = {(time_t)(us / 1000000), (long)(us % 1000000) * 1000};
38 while (nanosleep(&t, &t) != 0 && errno == EINTR) {}
39}
40
41mesh_chan *mesh_chan_new(size_t item, size_t bound, void (*drop)(void *item)) {
42 mesh_chan *c = mesh_xcalloc(1, sizeof *c);
43 pthread_mutex_init(&c->mu, NULL);
44 pthread_condattr_t a;
45 pthread_condattr_init(&a);
46 pthread_condattr_setclock(&a, CLOCK_MONOTONIC);
47 pthread_cond_init(&c->can_recv, &a);
48 pthread_cond_init(&c->can_send, &a);
49 pthread_condattr_destroy(&a);
50 c->item = item ? item : 1;
51 c->bound = bound;
52 c->cap = bound ? bound : 16;
53 c->ring = mesh_xmalloc(c->cap * c->item);
54 c->refs = 1;
55 c->senders = 1;
56 c->rx_open = 1;
57 c->drop = drop;
58 return c;
59}
60
61static void drain_locked(mesh_chan *c) {
62 while (c->len) {
63 if (c->drop) c->drop(c->ring + c->head * c->item);
64 c->head = (c->head + 1) % c->cap;
65 c->len--;
66 }
67}
68
69void mesh_chan_retain(mesh_chan *c) {
70 pthread_mutex_lock(&c->mu);
71 c->refs++;
72 pthread_mutex_unlock(&c->mu);
73}
74
75void mesh_chan_release(mesh_chan *c) {
76 if (!c) return;
77 pthread_mutex_lock(&c->mu);
78 int last = --c->refs == 0;
79 if (last) drain_locked(c);
80 pthread_mutex_unlock(&c->mu);
81 if (!last) return;
82 pthread_mutex_destroy(&c->mu);
83 pthread_cond_destroy(&c->can_recv);
84 pthread_cond_destroy(&c->can_send);
85 free(c->ring);
86 free(c);
87}
88
89void mesh_chan_add_sender(mesh_chan *c) {
90 pthread_mutex_lock(&c->mu);
91 c->senders++;
92 pthread_mutex_unlock(&c->mu);
93}
94
95void mesh_chan_drop_sender(mesh_chan *c) {
96 pthread_mutex_lock(&c->mu);
97 if (c->senders > 0 && --c->senders == 0) pthread_cond_broadcast(&c->can_recv);
98 pthread_mutex_unlock(&c->mu);
99}
100
101void mesh_chan_close_rx(mesh_chan *c) {
102 pthread_mutex_lock(&c->mu);
103 c->rx_open = 0;
104 drain_locked(c);
105 pthread_cond_broadcast(&c->can_send);
106 pthread_mutex_unlock(&c->mu);
107}
108
109static void push_locked(mesh_chan *c, const void *item) {
110 if (c->len == c->cap) {
111 /* Unbounded: grow, keeping order. */
112 size_t cap = c->cap * 2;
113 unsigned char *r = mesh_xmalloc(cap * c->item);
114 for (size_t i = 0; i < c->len; i++) memcpy(r + i * c->item, c->ring + ((c->head + i) % c->cap) * c->item, c->item);
115 free(c->ring);
116 c->ring = r;
117 c->cap = cap;
118 c->head = 0;
119 }
120 memcpy(c->ring + ((c->head + c->len) % c->cap) * c->item, item, c->item);
121 c->len++;
122 pthread_cond_signal(&c->can_recv);
123}
124
125static int send_(mesh_chan *c, const void *item, int wait) {
126 pthread_mutex_lock(&c->mu);
127 for (;;) {
128 if (!c->rx_open) { pthread_mutex_unlock(&c->mu); return MESH_CHAN_CLOSED; }
129 if (!c->bound || c->len < c->bound) break;
130 if (!wait) { pthread_mutex_unlock(&c->mu); return MESH_CHAN_FULL; }
131 pthread_cond_wait(&c->can_send, &c->mu);
132 }
133 push_locked(c, item);
134 pthread_mutex_unlock(&c->mu);
135 return MESH_CHAN_OK;
136}
137
138int mesh_chan_send(mesh_chan *c, const void *item) { return send_(c, item, 1); }
139int mesh_chan_try_send(mesh_chan *c, const void *item) { return send_(c, item, 0); }
140
141static void deadline(struct timespec *t, int64_t ms) {
142 clock_gettime(CLOCK_MONOTONIC, t);
143 t->tv_sec += (time_t)(ms / 1000);
144 t->tv_nsec += (long)(ms % 1000) * 1000000;
145 if (t->tv_nsec >= 1000000000) { t->tv_sec++; t->tv_nsec -= 1000000000; }
146}
147
148int mesh_chan_recv(mesh_chan *c, void *out, int64_t timeout_ms) {
149 struct timespec until;
150 if (timeout_ms >= 0) deadline(&until, timeout_ms);
151 pthread_mutex_lock(&c->mu);
152 while (!c->len) {
153 if (!c->senders) { pthread_mutex_unlock(&c->mu); return MESH_CHAN_CLOSED; }
154 if (timeout_ms == 0) { pthread_mutex_unlock(&c->mu); return MESH_CHAN_TIMEOUT; }
155 if (timeout_ms < 0) pthread_cond_wait(&c->can_recv, &c->mu);
156 else if (pthread_cond_timedwait(&c->can_recv, &c->mu, &until) == ETIMEDOUT && !c->len) {
157 int closed = !c->senders;
158 pthread_mutex_unlock(&c->mu);
159 return closed ? MESH_CHAN_CLOSED : MESH_CHAN_TIMEOUT;
160 }
161 }
162 memcpy(out, c->ring + c->head * c->item, c->item);
163 c->head = (c->head + 1) % c->cap;
164 c->len--;
165 pthread_cond_signal(&c->can_send);
166 pthread_mutex_unlock(&c->mu);
167 return MESH_CHAN_OK;
168}
169
170int mesh_chan_try_recv(mesh_chan *c, void *out) {
171 pthread_mutex_lock(&c->mu);
172 if (!c->len) {
173 int closed = !c->senders;
174 pthread_mutex_unlock(&c->mu);
175 return closed ? MESH_CHAN_CLOSED : MESH_CHAN_EMPTY;
176 }
177 memcpy(out, c->ring + c->head * c->item, c->item);
178 c->head = (c->head + 1) % c->cap;
179 c->len--;
180 pthread_cond_signal(&c->can_send);
181 pthread_mutex_unlock(&c->mu);
182 return MESH_CHAN_OK;
183}
184
185size_t mesh_chan_len(mesh_chan *c) {
186 pthread_mutex_lock(&c->mu);
187 size_t n = c->len;
188 pthread_mutex_unlock(&c->mu);
189 return n;
190}