1 /*
2 * libwebsockets-test-server - libwebsockets test implementation
3 *
4 * Written in 2010-2019 by Andy Green <andy@warmcat.com>
5 *
6 * This file is made available under the Creative Commons CC0 1.0
7 * Universal Public Domain Dedication.
8 *
9 * The person who associated a work with this deed has dedicated
10 * the work to the public domain by waiving all of his or her rights
11 * to the work worldwide under copyright law, including all related
12 * and neighboring rights, to the extent allowed by law. You can copy,
13 * modify, distribute and perform the work, even for commercial purposes,
14 * all without asking permission.
15 *
16 * The test apps are intended to be adapted for use in your code, which
17 * may be proprietary. So unlike the library itself, they are licensed
18 * Public Domain.
19 *
20 * Notice that the lws_pthread... locking apis are all zero-footprint
21 * NOPs in the case LWS_MAX_SMP == 1, which is the default. When lws
22 * is built for multiple service threads though, they resolve to their
23 * pthreads equivalents.
24 */
25
26 #if !defined (LWS_PLUGIN_STATIC)
27 #define LWS_DLL
28 #define LWS_INTERNAL
29 #include <libwebsockets.h>
30 #endif
31
32 #include <string.h>
33 #include <stdlib.h>
34
35 #define QUEUELEN 32
36 /* queue free space below this, rx flow is disabled */
37 #define RXFLOW_MIN (4)
38 /* queue free space above this, rx flow is enabled */
39 #define RXFLOW_MAX ((2 * QUEUELEN) / 3)
40
41 #define MAX_MIRROR_INSTANCES 3
42
43 struct mirror_instance;
44
45 struct per_session_data__lws_mirror {
46 struct lws *wsi;
47 struct mirror_instance *mi;
48 struct per_session_data__lws_mirror *same_mi_pss_list;
49 uint32_t tail;
50 };
51
52 /* this is the element in the ring */
53 struct a_message {
54 void *payload;
55 size_t len;
56 };
57
58 struct mirror_instance {
59 struct mirror_instance *next;
60 lws_pthread_mutex(lock) /* protects all mirror instance data */
61 struct per_session_data__lws_mirror *same_mi_pss_list;
62 /**< must hold the the per_vhost_data__lws_mirror.lock as well
63 * to change mi list membership */
64 struct lws_ring *ring;
65 int messages_allocated;
66 char name[30];
67 char rx_enabled;
68 };
69
70 struct per_vhost_data__lws_mirror {
71 lws_pthread_mutex(lock) /* protects mi_list membership changes */
72 struct mirror_instance *mi_list;
73 };
74
75
76 /* enable or disable rx from all connections to this mirror instance */
77 static void
__mirror_rxflow_instance(struct mirror_instance * mi,int enable)78 __mirror_rxflow_instance(struct mirror_instance *mi, int enable)
79 {
80 lws_start_foreach_ll(struct per_session_data__lws_mirror *,
81 pss, mi->same_mi_pss_list) {
82 lws_rx_flow_control(pss->wsi, enable);
83 } lws_end_foreach_ll(pss, same_mi_pss_list);
84
85 mi->rx_enabled = enable;
86 }
87
88 /*
89 * Find out which connection to this mirror instance has the longest number
90 * of still unread elements in the ringbuffer and update the lws_ring "oldest
91 * tail" with it. Elements behind the "oldest tail" are freed and recycled for
92 * new head content. Elements after the "oldest tail" are still waiting to be
93 * read by somebody.
94 *
95 * If the oldest tail moved on from before, check if it created enough space
96 * in the queue to re-enable RX flow control for the mirror instance.
97 *
98 * Mark connections that are at the oldest tail as being on a 3s timeout to
99 * transmit something, otherwise the connection will be closed. Without this,
100 * a choked or nonresponsive connection can block the FIFO from freeing up any
101 * new space for new data.
102 *
103 * You can skip calling this if on your connection, before processing, the tail
104 * was not equal to the current worst, ie, if the tail you will work on is !=
105 * lws_ring_get_oldest_tail(ring) then no need to call this when the tail
106 * has changed; it wasn't the oldest so it won't change the oldest.
107 *
108 * Returns 0 if oldest unchanged or 1 if oldest changed from this call.
109 */
110 static int
__mirror_update_worst_tail(struct mirror_instance * mi)111 __mirror_update_worst_tail(struct mirror_instance *mi)
112 {
113 uint32_t wai, worst = 0, worst_tail = 0, oldest;
114 struct per_session_data__lws_mirror *worst_pss = NULL;
115
116 oldest = lws_ring_get_oldest_tail(mi->ring);
117
118 lws_start_foreach_ll(struct per_session_data__lws_mirror *,
119 pss, mi->same_mi_pss_list) {
120 wai = (uint32_t)lws_ring_get_count_waiting_elements(mi->ring,
121 &pss->tail);
122 if (wai >= worst) {
123 worst = wai;
124 worst_tail = pss->tail;
125 worst_pss = pss;
126 }
127 } lws_end_foreach_ll(pss, same_mi_pss_list);
128
129 if (!worst_pss)
130 return 0;
131
132 lws_ring_update_oldest_tail(mi->ring, worst_tail);
133 if (oldest == lws_ring_get_oldest_tail(mi->ring))
134 return 0;
135 /*
136 * The oldest tail did move on. Check if we should re-enable rx flow
137 * for the mirror instance since we made some space now.
138 */
139 if (!mi->rx_enabled && /* rx is disabled */
140 lws_ring_get_count_free_elements(mi->ring) >= RXFLOW_MAX)
141 /* there is enough space, let's re-enable rx for our instance */
142 __mirror_rxflow_instance(mi, 1);
143
144 /* if nothing in queue, no timeout needed */
145 if (!worst)
146 return 1;
147
148 /*
149 * The guy(s) with the oldest tail block the ringbuffer from recycling
150 * the FIFO entries he has not read yet. Don't allow those guys to
151 * block the FIFO operation for very long.
152 */
153 lws_start_foreach_ll(struct per_session_data__lws_mirror *,
154 pss, mi->same_mi_pss_list) {
155 if (pss->tail == worst_tail)
156 /*
157 * Our policy is if you are the slowest connection,
158 * you had better transmit something to help with that
159 * within 3s, or we will hang up on you to stop you
160 * blocking the FIFO for everyone else.
161 */
162 lws_set_timeout(pss->wsi,
163 PENDING_TIMEOUT_USER_REASON_BASE, 3);
164 } lws_end_foreach_ll(pss, same_mi_pss_list);
165
166 return 1;
167 }
168
169 static void
__mirror_callback_all_in_mi_on_writable(struct mirror_instance * mi)170 __mirror_callback_all_in_mi_on_writable(struct mirror_instance *mi)
171 {
172 /* ask for WRITABLE callback for every wsi on this mi */
173 lws_start_foreach_ll(struct per_session_data__lws_mirror *,
174 pss, mi->same_mi_pss_list) {
175 lws_callback_on_writable(pss->wsi);
176 } lws_end_foreach_ll(pss, same_mi_pss_list);
177 }
178
179 static void
__mirror_destroy_message(void * _msg)180 __mirror_destroy_message(void *_msg)
181 {
182 struct a_message *msg = _msg;
183
184 free(msg->payload);
185 msg->payload = NULL;
186 msg->len = 0;
187 }
188
189 static int
callback_lws_mirror(struct lws * wsi,enum lws_callback_reasons reason,void * user,void * in,size_t len)190 callback_lws_mirror(struct lws *wsi, enum lws_callback_reasons reason,
191 void *user, void *in, size_t len)
192 {
193 struct per_session_data__lws_mirror *pss =
194 (struct per_session_data__lws_mirror *)user;
195 struct per_vhost_data__lws_mirror *v =
196 (struct per_vhost_data__lws_mirror *)
197 lws_protocol_vh_priv_get(lws_get_vhost(wsi),
198 lws_get_protocol(wsi));
199 char name[300], update_worst, sent_something, *pn = name;
200 struct mirror_instance *mi = NULL;
201 const struct a_message *msg;
202 struct a_message amsg;
203 uint32_t oldest_tail;
204 int n, count_mi = 0;
205
206 switch (reason) {
207 case LWS_CALLBACK_ESTABLISHED:
208 lwsl_info("%s: LWS_CALLBACK_ESTABLISHED\n", __func__);
209
210 /*
211 * mirror instance name... defaults to "", but if URL includes
212 * "?mirror=xxx", will be "xxx"
213 */
214 name[0] = '\0';
215 if (!lws_get_urlarg_by_name(wsi, "mirror", name,
216 sizeof(name) - 1))
217 lwsl_debug("get urlarg failed\n");
218 if (strchr(name, '='))
219 pn = strchr(name, '=') + 1;
220
221 //lwsl_notice("%s: mirror name '%s'\n", __func__, pn);
222
223 /* is there already a mirror instance of this name? */
224
225 lws_pthread_mutex_lock(&v->lock); /* vhost lock { */
226
227 lws_start_foreach_ll(struct mirror_instance *, mi1,
228 v->mi_list) {
229 count_mi++;
230 if (!strcmp(pn, mi1->name)) {
231 /* yes... we will join it */
232 mi = mi1;
233 break;
234 }
235 } lws_end_foreach_ll(mi1, next);
236
237 if (!mi) {
238
239 /* no existing mirror instance for name */
240 if (count_mi == MAX_MIRROR_INSTANCES) {
241 lws_pthread_mutex_unlock(&v->lock); /* } vh lock */
242 return -1;
243 }
244
245 /* create one with this name, and join it */
246 mi = malloc(sizeof(*mi));
247 if (!mi)
248 goto bail1;
249 memset(mi, 0, sizeof(*mi));
250 mi->ring = lws_ring_create(sizeof(struct a_message),
251 QUEUELEN,
252 __mirror_destroy_message);
253 if (!mi->ring) {
254 free(mi);
255 goto bail1;
256 }
257
258 mi->next = v->mi_list;
259 v->mi_list = mi;
260 lws_snprintf(mi->name, sizeof(mi->name) - 1, "%s", pn);
261 mi->rx_enabled = 1;
262
263 lws_pthread_mutex_init(&mi->lock);
264
265 lwsl_notice("Created new mi %p '%s'\n", mi, pn);
266 }
267
268 /* add our pss to list of guys bound to this mi */
269
270 lws_ll_fwd_insert(pss, same_mi_pss_list, mi->same_mi_pss_list);
271
272 /* init the pss */
273
274 pss->mi = mi;
275 pss->tail = lws_ring_get_oldest_tail(mi->ring);
276 pss->wsi = wsi;
277
278 lws_pthread_mutex_unlock(&v->lock); /* } vhost lock */
279 break;
280
281 bail1:
282 lws_pthread_mutex_unlock(&v->lock); /* } vhost lock */
283 return 1;
284
285 case LWS_CALLBACK_CLOSED:
286 /* detach our pss from the mirror instance */
287 mi = pss->mi;
288 if (!mi)
289 break;
290
291 lws_pthread_mutex_lock(&v->lock); /* vhost lock { */
292
293 /* remove our closing pss from its mirror instance list */
294 lws_ll_fwd_remove(struct per_session_data__lws_mirror,
295 same_mi_pss_list, pss, mi->same_mi_pss_list);
296 pss->mi = NULL;
297
298 if (mi->same_mi_pss_list) {
299 /*
300 * Still other pss using the mirror instance. The pss
301 * going away may have had the oldest tail, reconfirm
302 * using the remaining pss what is the current oldest
303 * tail. If the oldest tail moves on, this call also
304 * will re-enable rx flow control when appropriate.
305 */
306 lws_pthread_mutex_lock(&mi->lock); /* mi lock { */
307 __mirror_update_worst_tail(mi);
308 lws_pthread_mutex_unlock(&mi->lock); /* } mi lock */
309 lws_pthread_mutex_unlock(&v->lock); /* } vhost lock */
310 break;
311 }
312
313 /* No more pss using the mirror instance... delete mi */
314
315 lws_start_foreach_llp(struct mirror_instance **,
316 pmi, v->mi_list) {
317 if (*pmi == mi) {
318 *pmi = (*pmi)->next;
319
320 lws_ring_destroy(mi->ring);
321 lws_pthread_mutex_destroy(&mi->lock);
322
323 free(mi);
324 break;
325 }
326 } lws_end_foreach_llp(pmi, next);
327
328 lws_pthread_mutex_unlock(&v->lock); /* } vhost lock */
329 break;
330
331 case LWS_CALLBACK_CONFIRM_EXTENSION_OKAY:
332 return 1; /* disallow compression */
333
334 case LWS_CALLBACK_PROTOCOL_INIT: /* per vhost */
335 lws_protocol_vh_priv_zalloc(lws_get_vhost(wsi),
336 lws_get_protocol(wsi),
337 sizeof(struct per_vhost_data__lws_mirror));
338 v = (struct per_vhost_data__lws_mirror *)
339 lws_protocol_vh_priv_get(lws_get_vhost(wsi),
340 lws_get_protocol(wsi));
341 lws_pthread_mutex_init(&v->lock);
342 break;
343
344 case LWS_CALLBACK_PROTOCOL_DESTROY:
345 lws_pthread_mutex_destroy(&v->lock);
346 break;
347
348 case LWS_CALLBACK_SERVER_WRITEABLE:
349 lws_pthread_mutex_lock(&pss->mi->lock); /* instance lock { */
350 oldest_tail = lws_ring_get_oldest_tail(pss->mi->ring);
351 update_worst = oldest_tail == pss->tail;
352 sent_something = 0;
353
354 do {
355 msg = lws_ring_get_element(pss->mi->ring, &pss->tail);
356 if (!msg)
357 break;
358
359 if (!msg->payload) {
360 lwsl_err("%s: NULL payload: worst = %d,"
361 " pss->tail = %d\n", __func__,
362 oldest_tail, pss->tail);
363 if (lws_ring_consume(pss->mi->ring, &pss->tail,
364 NULL, 1))
365 continue;
366 break;
367 }
368
369 n = lws_write(wsi, (unsigned char *)msg->payload +
370 LWS_PRE, msg->len, LWS_WRITE_TEXT);
371 if (n < 0) {
372 lwsl_info("%s: WRITEABLE: %d\n", __func__, n);
373
374 goto bail2;
375 }
376 sent_something = 1;
377 lws_ring_consume(pss->mi->ring, &pss->tail, NULL, 1);
378
379 } while (!lws_send_pipe_choked(wsi));
380
381 /* if any left for us to send, ask for writeable again */
382 if (lws_ring_get_count_waiting_elements(pss->mi->ring,
383 &pss->tail))
384 lws_callback_on_writable(wsi);
385
386 if (!sent_something || !update_worst)
387 goto done1;
388
389 /*
390 * We are no longer holding the oldest tail (since we sent
391 * something. So free us of the timeout related to hogging the
392 * oldest tail.
393 */
394 lws_set_timeout(pss->wsi, NO_PENDING_TIMEOUT, 0);
395 /*
396 * If we were originally at the oldest fifo position of
397 * all the tails, now we used some up we may have
398 * changed the oldest fifo position and made some space.
399 */
400 __mirror_update_worst_tail(pss->mi);
401
402 done1:
403 lws_pthread_mutex_unlock(&pss->mi->lock); /* } instance lock */
404 break;
405
406 bail2:
407 lws_pthread_mutex_unlock(&pss->mi->lock); /* } instance lock */
408
409 return -1;
410
411 case LWS_CALLBACK_RECEIVE:
412 lws_pthread_mutex_lock(&pss->mi->lock); /* mi lock { */
413 n = (int)lws_ring_get_count_free_elements(pss->mi->ring);
414 if (!n) {
415 lwsl_notice("dropping!\n");
416 if (pss->mi->rx_enabled)
417 __mirror_rxflow_instance(pss->mi, 0);
418 goto req_writable;
419 }
420
421 amsg.payload = malloc(LWS_PRE + len);
422 amsg.len = len;
423 if (!amsg.payload) {
424 lwsl_notice("OOM: dropping\n");
425 goto done2;
426 }
427
428 memcpy((char *)amsg.payload + LWS_PRE, in, len);
429 if (!lws_ring_insert(pss->mi->ring, &amsg, 1)) {
430 __mirror_destroy_message(&amsg);
431 lwsl_notice("dropping!\n");
432 if (pss->mi->rx_enabled)
433 __mirror_rxflow_instance(pss->mi, 0);
434 goto req_writable;
435 }
436
437 if (pss->mi->rx_enabled &&
438 lws_ring_get_count_free_elements(pss->mi->ring) <
439 RXFLOW_MIN)
440 __mirror_rxflow_instance(pss->mi, 0);
441
442 req_writable:
443 __mirror_callback_all_in_mi_on_writable(pss->mi);
444
445 done2:
446 lws_pthread_mutex_unlock(&pss->mi->lock); /* } mi lock */
447 break;
448
449 case LWS_CALLBACK_EVENT_WAIT_CANCELLED:
450 lwsl_info("LWS_CALLBACK_EVENT_WAIT_CANCELLED\n");
451 break;
452
453 default:
454 break;
455 }
456
457 return 0;
458 }
459
460 #define LWS_PLUGIN_PROTOCOL_MIRROR { \
461 "lws-mirror-protocol", \
462 callback_lws_mirror, \
463 sizeof(struct per_session_data__lws_mirror), \
464 4096, /* rx buf size must be >= permessage-deflate rx size */ \
465 0, NULL, 0 \
466 }
467
468 #if !defined (LWS_PLUGIN_STATIC)
469
470 static const struct lws_protocols protocols[] = {
471 LWS_PLUGIN_PROTOCOL_MIRROR
472 };
473
474 LWS_VISIBLE int
init_protocol_lws_mirror(struct lws_context * context,struct lws_plugin_capability * c)475 init_protocol_lws_mirror(struct lws_context *context,
476 struct lws_plugin_capability *c)
477 {
478 if (c->api_magic != LWS_PLUGIN_API_MAGIC) {
479 lwsl_err("Plugin API %d, library API %d", LWS_PLUGIN_API_MAGIC,
480 c->api_magic);
481 return 1;
482 }
483
484 c->protocols = protocols;
485 c->count_protocols = LWS_ARRAY_SIZE(protocols);
486 c->extensions = NULL;
487 c->count_extensions = 0;
488
489 return 0;
490 }
491
492 LWS_VISIBLE int
destroy_protocol_lws_mirror(struct lws_context * context)493 destroy_protocol_lws_mirror(struct lws_context *context)
494 {
495 return 0;
496 }
497 #endif
498