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, whoami = 0;
87
88 for (n = 0; n < (int)LWS_ARRAY_SIZE(vhd->pthread_spam); n++)
89 if (pthread_equal(pthread_self(), vhd->pthread_spam[n]))
90 whoami = n + 1;
91
92 do {
93 /* don't generate output if nobody connected */
94 if (!vhd->pss_list)
95 goto wait;
96
97 pthread_mutex_lock(&vhd->lock_ring); /* --------- ring lock { */
98
99 /* only create if space in ringbuffer */
100 n = (int)lws_ring_get_count_free_elements(vhd->ring);
101 if (!n) {
102 lwsl_user("dropping!\n");
103 goto wait_unlock;
104 }
105
106 amsg.payload = malloc((unsigned int)(LWS_PRE + len));
107 if (!amsg.payload) {
108 lwsl_user("OOM: dropping\n");
109 goto wait_unlock;
110 }
111 n = lws_snprintf((char *)amsg.payload + LWS_PRE, (unsigned int)len,
112 "%s: spam tid: %d, msg: %d", vhd->config,
113 whoami, index++);
114 amsg.len = (unsigned int)n;
115 n = (int)lws_ring_insert(vhd->ring, &amsg, 1);
116 if (n != 1) {
117 __minimal_destroy_message(&amsg);
118 lwsl_user("dropping!\n");
119 } else
120 /*
121 * This will cause a LWS_CALLBACK_EVENT_WAIT_CANCELLED
122 * in the lws service thread context.
123 */
124 lws_cancel_service(vhd->context);
125
126 wait_unlock:
127 pthread_mutex_unlock(&vhd->lock_ring); /* } ring lock ------- */
128
129 wait:
130 usleep(100000);
131
132 } while (!vhd->finished);
133
134 lwsl_notice("thread_spam %d exiting\n", whoami);
135
136 pthread_exit(NULL);
137
138 return NULL;
139 }
140
141 /* this runs under the lws service thread context only */
142
143 static int
callback_minimal(struct lws * wsi,enum lws_callback_reasons reason,void * user,void * in,size_t len)144 callback_minimal(struct lws *wsi, enum lws_callback_reasons reason,
145 void *user, void *in, size_t len)
146 {
147 struct per_session_data__minimal *pss =
148 (struct per_session_data__minimal *)user;
149 struct per_vhost_data__minimal *vhd =
150 (struct per_vhost_data__minimal *)
151 lws_protocol_vh_priv_get(lws_get_vhost(wsi),
152 lws_get_protocol(wsi));
153 const struct lws_protocol_vhost_options *pvo;
154 const struct msg *pmsg;
155 char temp[LWS_PRE + 256];
156 void *retval;
157 int n, m, r = 0;
158
159 switch (reason) {
160 case LWS_CALLBACK_PROTOCOL_INIT:
161 /* create our per-vhost struct */
162 vhd = lws_protocol_vh_priv_zalloc(lws_get_vhost(wsi),
163 lws_get_protocol(wsi),
164 sizeof(struct per_vhost_data__minimal));
165 if (!vhd)
166 return 1;
167
168 pthread_mutex_init(&vhd->lock_ring, NULL);
169
170 /* recover the pointer to the globals struct */
171 pvo = lws_pvo_search(
172 (const struct lws_protocol_vhost_options *)in,
173 "config");
174 if (!pvo || !pvo->value) {
175 lwsl_err("%s: Can't find \"config\" pvo\n", __func__);
176 return 1;
177 }
178 vhd->config = pvo->value;
179
180 vhd->context = lws_get_context(wsi);
181 vhd->protocol = lws_get_protocol(wsi);
182 vhd->vhost = lws_get_vhost(wsi);
183
184 vhd->ring = lws_ring_create(sizeof(struct msg), 8,
185 __minimal_destroy_message);
186 if (!vhd->ring) {
187 lwsl_err("%s: failed to create ring\n", __func__);
188 return 1;
189 }
190
191 /* start the content-creating threads */
192
193 for (n = 0; n < (int)LWS_ARRAY_SIZE(vhd->pthread_spam); n++)
194 if (pthread_create(&vhd->pthread_spam[n], NULL,
195 thread_spam, vhd)) {
196 lwsl_err("thread creation failed\n");
197 r = 1;
198 goto init_fail;
199 }
200 break;
201
202 case LWS_CALLBACK_PROTOCOL_DESTROY:
203 init_fail:
204 vhd->finished = 1;
205 for (n = 0; n < (int)LWS_ARRAY_SIZE(vhd->pthread_spam); n++)
206 pthread_join(vhd->pthread_spam[n], &retval);
207
208 if (vhd->ring)
209 lws_ring_destroy(vhd->ring);
210
211 pthread_mutex_destroy(&vhd->lock_ring);
212 break;
213
214 case LWS_CALLBACK_ESTABLISHED:
215 /* add ourselves to the list of live pss held in the vhd */
216 lws_ll_fwd_insert(pss, pss_list, vhd->pss_list);
217 pss->tail = lws_ring_get_oldest_tail(vhd->ring);
218 pss->wsi = wsi;
219 break;
220
221 case LWS_CALLBACK_CLOSED:
222 /* remove our closing pss from the list of live pss */
223 lws_ll_fwd_remove(struct per_session_data__minimal, pss_list,
224 pss, vhd->pss_list);
225 break;
226
227 case LWS_CALLBACK_SERVER_WRITEABLE:
228 pthread_mutex_lock(&vhd->lock_ring); /* --------- ring lock { */
229
230 pmsg = lws_ring_get_element(vhd->ring, &pss->tail);
231 if (!pmsg) {
232 pthread_mutex_unlock(&vhd->lock_ring); /* } ring lock ------- */
233 break;
234 }
235
236 n = lws_snprintf(temp + LWS_PRE, sizeof(temp) - LWS_PRE,
237 "svc, %s",
238 (char *)pmsg->payload + LWS_PRE);
239
240 /* notice we allowed for LWS_PRE in the payload already */
241 m = lws_write(wsi, (unsigned char *)temp + LWS_PRE, (unsigned int)n,
242 LWS_WRITE_TEXT);
243 if (m < n) {
244 pthread_mutex_unlock(&vhd->lock_ring); /* } ring lock ------- */
245 lwsl_err("ERROR %d writing to ws socket\n", m);
246 return -1;
247 }
248
249 lws_ring_consume_and_update_oldest_tail(
250 vhd->ring, /* lws_ring object */
251 struct per_session_data__minimal, /* type of objects with tails */
252 &pss->tail, /* tail of guy doing the consuming */
253 1, /* number of payload objects being consumed */
254 vhd->pss_list, /* head of list of objects with tails */
255 tail, /* member name of tail in objects with tails */
256 pss_list /* member name of next object in objects with tails */
257 );
258
259 /* more to do? */
260 if (lws_ring_get_element(vhd->ring, &pss->tail))
261 /* come back as soon as we can write more */
262 lws_callback_on_writable(pss->wsi);
263
264 pthread_mutex_unlock(&vhd->lock_ring); /* } ring lock ------- */
265 break;
266
267 case LWS_CALLBACK_RECEIVE:
268 break;
269
270 case LWS_CALLBACK_EVENT_WAIT_CANCELLED:
271 lwsl_notice("LWS_CALLBACK_EVENT_WAIT_CANCELLED in svc\n");
272 if (!vhd)
273 break;
274 /*
275 * When the "spam" threads add a message to the ringbuffer,
276 * they create this event in the lws service thread context
277 * using lws_cancel_service().
278 *
279 * We respond by scheduling a writable callback for all
280 * connected clients.
281 */
282 lws_start_foreach_llp(struct per_session_data__minimal **,
283 ppss, vhd->pss_list) {
284 lws_callback_on_writable((*ppss)->wsi);
285 } lws_end_foreach_llp(ppss, pss_list);
286 break;
287
288 default:
289 break;
290 }
291
292 return r;
293 }
294
295 #define LWS_PLUGIN_PROTOCOL_MINIMAL \
296 { \
297 "lws-minimal", \
298 callback_minimal, \
299 sizeof(struct per_session_data__minimal), \
300 128, \
301 0, NULL, 0 \
302 }
303