1 /*
2 * ws protocol handler plugin for "lws-minimal" demonstrating multithread
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
10 #if !defined (LWS_PLUGIN_STATIC)
11 #define LWS_DLL
12 #define LWS_INTERNAL
13 #include <libwebsockets.h>
14 #endif
15
16 #include <string.h>
17
18 /* one of these created for each message in the ringbuffer */
19
20 struct msg {
21 void *payload; /* is malloc'd */
22 size_t len;
23 };
24
25 /*
26 * One of these is created for each client connecting to us.
27 *
28 * It is ONLY read or written from the lws service thread context.
29 */
30
31 struct per_session_data__minimal {
32 struct per_session_data__minimal *pss_list;
33 struct lws *wsi;
34 uint32_t tail;
35 };
36
37 /* one of these is created for each vhost our protocol is used with */
38
39 struct per_vhost_data__minimal {
40 struct lws_context *context;
41 struct lws_vhost *vhost;
42 const struct lws_protocols *protocol;
43
44 struct per_session_data__minimal *pss_list; /* linked-list of live pss*/
45 pthread_t pthread_spam[2];
46
47 pthread_mutex_t lock_ring; /* serialize access to the ring buffer */
48 struct lws_ring *ring; /* {lock_ring} ringbuffer holding unsent content */
49
50 const char *config;
51 char finished;
52 };
53
54 #if defined(WIN32)
usleep(unsigned long l)55 static void usleep(unsigned long l) { Sleep(l / 1000); }
56 #endif
57
58 /*
59 * This runs under both lws service and "spam threads" contexts.
60 * Access is serialized by vhd->lock_ring.
61 */
62
63 static void
__minimal_destroy_message(void * _msg)64 __minimal_destroy_message(void *_msg)
65 {
66 struct msg *msg = _msg;
67
68 free(msg->payload);
69 msg->payload = NULL;
70 msg->len = 0;
71 }
72
73 /*
74 * This runs under the "spam thread" thread context only.
75 *
76 * We spawn two threads that generate messages with this.
77 *
78 */
79
80 static void *
thread_spam(void * d)81 thread_spam(void *d)
82 {
83 struct per_vhost_data__minimal *vhd =
84 (struct per_vhost_data__minimal *)d;
85 struct msg amsg;
86 int len = 128, index = 1, n;
87
88 do {
89 /* don't generate output if nobody connected */
90 if (!vhd->pss_list)
91 goto wait;
92
93 pthread_mutex_lock(&vhd->lock_ring); /* --------- ring lock { */
94
95 /* only create if space in ringbuffer */
96 n = (int)lws_ring_get_count_free_elements(vhd->ring);
97 if (!n) {
98 lwsl_user("dropping!\n");
99 goto wait_unlock;
100 }
101
102 amsg.payload = malloc(LWS_PRE + len);
103 if (!amsg.payload) {
104 lwsl_user("OOM: dropping\n");
105 goto wait_unlock;
106 }
107 n = lws_snprintf((char *)amsg.payload + LWS_PRE, len,
108 "%s: tid: %p, msg: %d", vhd->config,
109 (void *)pthread_self(), index++);
110 amsg.len = n;
111 n = lws_ring_insert(vhd->ring, &amsg, 1);
112 if (n != 1) {
113 __minimal_destroy_message(&amsg);
114 lwsl_user("dropping!\n");
115 } else
116 /*
117 * This will cause a LWS_CALLBACK_EVENT_WAIT_CANCELLED
118 * in the lws service thread context.
119 */
120 lws_cancel_service(vhd->context);
121
122 wait_unlock:
123 pthread_mutex_unlock(&vhd->lock_ring); /* } ring lock ------- */
124
125 wait:
126 usleep(100000);
127
128 } while (!vhd->finished);
129
130 lwsl_notice("thread_spam %p exiting\n", (void *)pthread_self());
131
132 pthread_exit(NULL);
133
134 return NULL;
135 }
136
137 /* this runs under the lws service thread context only */
138
139 static int
callback_minimal(struct lws * wsi,enum lws_callback_reasons reason,void * user,void * in,size_t len)140 callback_minimal(struct lws *wsi, enum lws_callback_reasons reason,
141 void *user, void *in, size_t len)
142 {
143 struct per_session_data__minimal *pss =
144 (struct per_session_data__minimal *)user;
145 struct per_vhost_data__minimal *vhd =
146 (struct per_vhost_data__minimal *)
147 lws_protocol_vh_priv_get(lws_get_vhost(wsi),
148 lws_get_protocol(wsi));
149 const struct lws_protocol_vhost_options *pvo;
150 const struct msg *pmsg;
151 void *retval;
152 int n, m, r = 0;
153
154 switch (reason) {
155 case LWS_CALLBACK_PROTOCOL_INIT:
156 /* create our per-vhost struct */
157 vhd = lws_protocol_vh_priv_zalloc(lws_get_vhost(wsi),
158 lws_get_protocol(wsi),
159 sizeof(struct per_vhost_data__minimal));
160 if (!vhd)
161 return 1;
162
163 pthread_mutex_init(&vhd->lock_ring, NULL);
164
165 /* recover the pointer to the globals struct */
166 pvo = lws_pvo_search(
167 (const struct lws_protocol_vhost_options *)in,
168 "config");
169 if (!pvo || !pvo->value) {
170 lwsl_err("%s: Can't find \"config\" pvo\n", __func__);
171 return 1;
172 }
173 vhd->config = pvo->value;
174
175 vhd->context = lws_get_context(wsi);
176 vhd->protocol = lws_get_protocol(wsi);
177 vhd->vhost = lws_get_vhost(wsi);
178
179 vhd->ring = lws_ring_create(sizeof(struct msg), 8,
180 __minimal_destroy_message);
181 if (!vhd->ring) {
182 lwsl_err("%s: failed to create ring\n", __func__);
183 return 1;
184 }
185
186 /* start the content-creating threads */
187
188 for (n = 0; n < (int)LWS_ARRAY_SIZE(vhd->pthread_spam); n++)
189 if (pthread_create(&vhd->pthread_spam[n], NULL,
190 thread_spam, vhd)) {
191 lwsl_err("thread creation failed\n");
192 r = 1;
193 goto init_fail;
194 }
195 break;
196
197 case LWS_CALLBACK_PROTOCOL_DESTROY:
198 init_fail:
199 vhd->finished = 1;
200 for (n = 0; n < (int)LWS_ARRAY_SIZE(vhd->pthread_spam); n++)
201 if (vhd->pthread_spam[n])
202 pthread_join(vhd->pthread_spam[n], &retval);
203
204 if (vhd->ring)
205 lws_ring_destroy(vhd->ring);
206
207 pthread_mutex_destroy(&vhd->lock_ring);
208 break;
209
210 case LWS_CALLBACK_ESTABLISHED:
211 /* add ourselves to the list of live pss held in the vhd */
212 lws_ll_fwd_insert(pss, pss_list, vhd->pss_list);
213 pss->tail = lws_ring_get_oldest_tail(vhd->ring);
214 pss->wsi = wsi;
215 break;
216
217 case LWS_CALLBACK_CLOSED:
218 /* remove our closing pss from the list of live pss */
219 lws_ll_fwd_remove(struct per_session_data__minimal, pss_list,
220 pss, vhd->pss_list);
221 break;
222
223 case LWS_CALLBACK_SERVER_WRITEABLE:
224 pthread_mutex_lock(&vhd->lock_ring); /* --------- ring lock { */
225
226 pmsg = lws_ring_get_element(vhd->ring, &pss->tail);
227 if (!pmsg) {
228 pthread_mutex_unlock(&vhd->lock_ring); /* } ring lock ------- */
229 break;
230 }
231
232 /* notice we allowed for LWS_PRE in the payload already */
233 m = lws_write(wsi, ((unsigned char *)pmsg->payload) + LWS_PRE,
234 pmsg->len, LWS_WRITE_TEXT);
235 if (m < (int)pmsg->len) {
236 pthread_mutex_unlock(&vhd->lock_ring); /* } ring lock ------- */
237 lwsl_err("ERROR %d writing to ws socket\n", m);
238 return -1;
239 }
240
241 lws_ring_consume_and_update_oldest_tail(
242 vhd->ring, /* lws_ring object */
243 struct per_session_data__minimal, /* type of objects with tails */
244 &pss->tail, /* tail of guy doing the consuming */
245 1, /* number of payload objects being consumed */
246 vhd->pss_list, /* head of list of objects with tails */
247 tail, /* member name of tail in objects with tails */
248 pss_list /* member name of next object in objects with tails */
249 );
250
251 /* more to do? */
252 if (lws_ring_get_element(vhd->ring, &pss->tail))
253 /* come back as soon as we can write more */
254 lws_callback_on_writable(pss->wsi);
255
256 pthread_mutex_unlock(&vhd->lock_ring); /* } ring lock ------- */
257 break;
258
259 case LWS_CALLBACK_RECEIVE:
260 break;
261
262 case LWS_CALLBACK_EVENT_WAIT_CANCELLED:
263 if (!vhd)
264 break;
265 /*
266 * When the "spam" threads add a message to the ringbuffer,
267 * they create this event in the lws service thread context
268 * using lws_cancel_service().
269 *
270 * We respond by scheduling a writable callback for all
271 * connected clients.
272 */
273 lws_start_foreach_llp(struct per_session_data__minimal **,
274 ppss, vhd->pss_list) {
275 lws_callback_on_writable((*ppss)->wsi);
276 } lws_end_foreach_llp(ppss, pss_list);
277 break;
278
279 default:
280 break;
281 }
282
283 return r;
284 }
285
286 #define LWS_PLUGIN_PROTOCOL_MINIMAL \
287 { \
288 "lws-minimal", \
289 callback_minimal, \
290 sizeof(struct per_session_data__minimal), \
291 128, \
292 0, NULL, 0 \
293 }
294
295 #if !defined (LWS_PLUGIN_STATIC)
296
297 /* boilerplate needed if we are built as a dynamic plugin */
298
299 static const struct lws_protocols protocols[] = {
300 LWS_PLUGIN_PROTOCOL_MINIMAL
301 };
302
303 int
init_protocol_minimal(struct lws_context * context,struct lws_plugin_capability * c)304 init_protocol_minimal(struct lws_context *context,
305 struct lws_plugin_capability *c)
306 {
307 if (c->api_magic != LWS_PLUGIN_API_MAGIC) {
308 lwsl_err("Plugin API %d, library API %d", LWS_PLUGIN_API_MAGIC,
309 c->api_magic);
310 return 1;
311 }
312
313 c->protocols = protocols;
314 c->count_protocols = LWS_ARRAY_SIZE(protocols);
315 c->extensions = NULL;
316 c->count_extensions = 0;
317
318 return 0;
319 }
320
321 int
destroy_protocol_minimal(struct lws_context * context)322 destroy_protocol_minimal(struct lws_context *context)
323 {
324 return 0;
325 }
326 #endif
327