0 branches 0 tags
40
Change to non-recursive mutex
jonroach 17 Oct 2025 13:20
1 contributor
647 lines15.9 KB
Text Raw Blame
  • Delete
1#include "coroutine.h"
2#include <assert.h>
3#include <setjmp.h>
4#include <stdbool.h>
5#include <stddef.h>
6#include <stdio.h>
7#include <pthread.h>
8#include <stdlib.h>
9#include "cor_thread_local.h"
10
11
12static void *mustmalloc(size_t size){
13 void *p = malloc(size);
14 assert(p);
15 return p;
16}
17
18#define New(type, ...) (type##_ctor((type *)mustmalloc(sizeof(type), ## __VA_ARGS__)))
19#define Delete(ptr, type) ((ptr) ? (type##_dtor(ptr), free(ptr), (ptr) = NULL) : (void)0)
20static void Coroutine_RunNext();
21static void _Coroutine_Continue(Coroutine *cor, void *value, bool early);
22
23///////////////////////////////////////////////////////////////////////////////
24// 2-way linked lists...
25//
26// Brought inline here to avoid namespace polution
27///////////////////////////////////////////////////////////////////////////////
28
29typedef struct List_Link List_Link;
30struct List_Link {
31 List_Link *next;
32 List_Link *prev;
33};
34
35typedef struct List_Head List_Head;
36struct List_Head {
37 union {
38 struct {
39 List_Link link;
40 List_Link *filler;
41 } fwd;
42 struct {
43 List_Link *filler;
44 List_Link link;
45 } back;
46 };
47};
48
49
50static inline bool List_IsEmpty(
51 const List_Head *list
52){
53 return list->fwd.link.next == &list->back.link;
54}
55
56
57static inline List_Link *List_GetHead(
58 const List_Head *list
59){
60 return List_IsEmpty(list) ? NULL : list->fwd.link.next;
61}
62
63
64// static inline List_Link *List_GetTail(
65// const List_Head *list
66// ){
67// return List_IsEmpty(list) ? NULL : list->back.link.prev;
68// }
69
70
71#define OFFSETOF(Container, Field) ((char *)&((Container *)4)->Field - (char *)(Container *)4)
72#define List_Link_Container(Container, Link, link) ((Container *)((char *)(link) - OFFSETOF(Container, Link)))
73
74
75static inline void List_Init(
76 List_Head *list
77){
78 list->fwd.link.next = &list->back.link;
79 list->fwd.link.prev = NULL;
80 list->back.link.prev = &list->fwd.link;
81}
82
83
84static inline void List_AddHead(
85 List_Head *list,
86 List_Link *link
87){
88 List_Link *first = list->fwd.link.next;
89 link->next = first;
90 link->prev = &list->fwd.link;
91 first->prev = link;
92 list->fwd.link.next = link;
93}
94
95
96static inline void List_AddTail(
97 List_Head *list,
98 List_Link *link
99){
100 List_Link *last = list->back.link.prev;
101 link->prev = last;
102 link->next = &list->back.link;
103 last->next = link;
104 list->back.link.prev = link;
105}
106
107
108static inline void List_Remove(
109 List_Link *link
110){
111 link->prev->next = link->next;
112 link->next->prev = link->prev;
113}
114
115///////////////////////////////////////////////////////////////////////////////
116// ...2-way linked lists
117///////////////////////////////////////////////////////////////////////////////
118
119typedef struct Coroutines Coroutines;
120
121enum {
122 Coroutines_Idle,
123 Coroutines_Starting,
124 Coroutines_Started,
125 Coroutines_Active,
126 Coroutines_Stopping
127};
128
129enum {
130 Chunk_Initial,
131 Chunk_Create,
132 Chunk_Enter
133};
134
135typedef enum Coroutine_State {
136 Coroutine_Constructing,
137 Coroutine_Free,
138 Coroutine_Idle,
139 Coroutine_Running,
140 Coroutine_Waiting,
141 Coroutine_Complete
142} Coroutine_State;
143
144enum {
145 Coroutines_Init,
146 Coroutines_AllocatedChunk,
147 Coroutines_CoroutineComplete,
148};
149
150struct Coroutine {
151 Coroutines *coroutines; // so can work with it off-thread
152 List_Link link; // for whichever list it's on
153 jmp_buf buf; // how to get back to it
154 unsigned char *guard; // where the stack overrun guard is
155 Coroutine_Start start; // entry point
156 void *entry_param; // to pass to start
157 void *value; // yielded/returned
158 Coroutine_State state;
159};
160
161struct Coroutines {
162 pthread_mutex_t mutex;
163 jmp_buf controller; // to return from Coroutine_Run
164 jmp_buf chunk_allocated;// for chunk allocation
165 unsigned char *guard; // the stack guard for the startup sequence
166
167 // singletons
168 Coroutine *tip; // top of stack chunk
169 Coroutine *active; // currently running coroutine
170 Coroutine *primary; // Coroutine_Run coroutine
171
172 // lists
173 List_Head free;
174 List_Head inactive; // idle or complete
175 List_Head runable; // running or waiting to run
176 List_Head waiting; // yielded / waiting to run
177 pthread_cond_t waiting_cond;
178
179 // Summary of the system
180 Coroutine_Report report;
181
182 // state
183 char state;
184};
185
186_Cor_thread_local Coroutines *g_c;
187
188static void stack_chunk_chunk(Coroutine *parent);
189static void stack_chunk_base(Coroutine *parent, unsigned char *guard);
190
191
192// Check whether the guard is intact
193static inline bool Check_Guard(
194 unsigned char *guard
195){
196 return guard[0] == 0xde &&
197 guard[1] == 0xad &&
198 guard[2] == 0xbe &&
199 guard[3] == 0xef;
200}
201
202
203static void Coroutine_PrimeStackChunks()
204{
205 unsigned char chunk_of_stack[COROUTINE_STARTUP_STACK_SIZE];
206 for (int i = 0; i < COROUTINE_STARTUP_STACK_SIZE-3; i += 4){
207 chunk_of_stack[i+0] = 0xde;
208 chunk_of_stack[i+1] = 0xad;
209 chunk_of_stack[i+2] = 0xbe;
210 chunk_of_stack[i+3] = 0xef;
211 }
212 assert(Check_Guard(chunk_of_stack));
213
214 // Stacks grow down in memory (almost always), so if the caller of this function changes
215 // the guard before entering the coroutine system, it has overrun the startup stack
216 g_c->guard = chunk_of_stack;
217
218 stack_chunk_base(NULL, NULL);
219}
220
221
222static void stack_chunk_chunk(
223 Coroutine *parent
224){
225 unsigned char chunk_of_stack[COROUTINE_STACK_SIZE];
226 for (int i = 0; i < COROUTINE_STACK_SIZE-3; i += 4){
227 chunk_of_stack[i+0] = 0xde;
228 chunk_of_stack[i+1] = 0xad;
229 chunk_of_stack[i+2] = 0xbe;
230 chunk_of_stack[i+3] = 0xef;
231 }
232 stack_chunk_base(parent, chunk_of_stack);
233}
234
235
236static void stack_chunk_base(
237 Coroutine *parent,
238 unsigned char *guard
239){
240 Coroutine here;
241 here.state = Coroutine_Constructing;
242 switch (setjmp(here.buf)) {
243 case Chunk_Initial:
244 // got here for the first time
245 // parent now has a chunk_of_stack - add it to the free list
246 if (parent) {
247 assert(parent->state == Coroutine_Constructing);
248 assert(Check_Guard(guard));
249 parent->guard = guard;
250 parent->state = Coroutine_Free;
251 List_AddHead(&g_c->free, &parent->link);
252 g_c->report.coroutines_pool_size += 1;
253 }
254 // note that here is the tip of the chunk-claim stack
255 here.coroutines = g_c;
256 g_c->tip = &here;
257
258 // return to the coroutine allocator
259 longjmp(g_c->chunk_allocated, 1);
260 case Chunk_Create:
261 // request to create a new chunk on the stack
262 assert(here.state == Coroutine_Constructing);
263 stack_chunk_chunk(&here);
264 assert(false);
265 case Chunk_Enter:
266 // request to start a coroutine (ie use the chunk for a coroutine)
267 // arrive here with mutex locked
268 assert(here.state == Coroutine_Running);
269 g_c->active = &here;
270 int r = pthread_mutex_unlock(&g_c->mutex);
271 assert(r == 0);
272 here.value = here.start(here.entry_param);
273
274 // check the guard
275 if (!Check_Guard(here.guard)){
276 printf("Coroutine has overrun its stack - checked after returning from coroutine function\n");
277 exit(EXIT_FAILURE);
278 }
279
280 r = pthread_mutex_lock(&g_c->mutex);
281 assert(r == 0);
282 g_c->active = NULL;
283 assert(here.state == Coroutine_Running);
284 List_Remove(&here.link);
285 here.state = Coroutine_Complete;
286 List_AddTail(&g_c->inactive, &here.link);
287 // coroutine has completed
288 if (g_c->primary == &here) {
289 // if primary coroutine - return to Coroutine_Run
290 longjmp(g_c->controller, Coroutines_CoroutineComplete);
291 }
292 r = pthread_mutex_unlock(&g_c->mutex);
293 assert(r == 0);
294 Coroutine_RunNext();
295 assert(false);
296 }
297}
298
299
300static void Coroutine_RunNext()
301{
302 // arrive here with mutex unlocked
303 int r;
304
305 r = pthread_mutex_lock(&g_c->mutex);
306 assert(r == 0);
307 while (List_IsEmpty(&g_c->runable)) {
308 r = pthread_cond_wait(&g_c->waiting_cond, &g_c->mutex);
309 assert(r == 0);
310 }
311 Coroutine *next = List_Link_Container(Coroutine, link, List_GetHead(&g_c->runable));
312 assert(next->state == Coroutine_Running);
313 longjmp(next->buf, Chunk_Enter);
314 assert(false);
315}
316
317
318void Coroutine_StartSystem()
319{
320 assert(!g_c);
321 g_c = mustmalloc(sizeof(Coroutines));
322
323 g_c->state = Coroutines_Starting;
324
325 int r;
326 r = pthread_mutex_init(&g_c->mutex, NULL);
327 assert(r == 0);
328
329 g_c->tip = NULL;
330 g_c->active = NULL;
331
332 List_Init(&g_c->free);
333 List_Init(&g_c->inactive);
334 List_Init(&g_c->runable);
335 List_Init(&g_c->waiting);
336 r = pthread_cond_init(&g_c->waiting_cond, NULL);
337 assert(r == 0);
338
339 g_c->report.coroutines_created = 0;
340 g_c->report.coroutines_pool_size = 0;
341 g_c->report.lowest_headroom = COROUTINE_STACK_SIZE;
342
343 // prime the chunk system
344 if (!setjmp(g_c->chunk_allocated)){
345 Coroutine_PrimeStackChunks();
346 assert(false);
347 }
348
349 assert(g_c->state == Coroutines_Starting);
350 g_c->state = Coroutines_Started;
351}
352
353
354Coroutine_Report Coroutine_StopSystem()
355{
356 int r = pthread_mutex_lock(&g_c->mutex);
357 assert(r == 0);
358 assert(g_c->state == Coroutines_Started);
359 g_c->state = Coroutines_Stopping;
360
361 int stackminheadroom = COROUTINE_STACK_SIZE;
362 for (List_Link *link = g_c->free.fwd.link.next; link->next; link = link->next){
363 Coroutine *cor = List_Link_Container(Coroutine, link, link);
364 for (int i = 4; i < COROUTINE_STACK_SIZE-3; i += 4){
365 if (!Check_Guard(&cor->guard[i])){
366 stackminheadroom = i < stackminheadroom ? i : stackminheadroom;
367 break;
368 }
369 }
370 }
371 g_c->report.lowest_headroom = stackminheadroom;
372
373 assert(List_IsEmpty(&g_c->inactive));
374 r = pthread_cond_destroy(&g_c->waiting_cond);
375 assert(r == 0);
376
377 assert(g_c->state == Coroutines_Stopping);
378 pthread_mutex_unlock(&g_c->mutex);
379 assert(r == 0);
380 g_c->state = Coroutines_Idle;
381 r = pthread_mutex_destroy(&g_c->mutex);
382 assert(r == 0);
383
384 Coroutine_Report res = g_c->report;
385
386 free(g_c);
387 g_c = NULL;
388
389 return res;
390}
391
392
393void Coroutine_Run_Coroutine(
394 Coroutine *cor,
395 void *value
396){
397 Coroutines *cors = cor->coroutines;
398 assert(g_c == cors);
399 int r = pthread_mutex_lock(&cors->mutex);
400 assert(r == 0);
401 assert(cors->state == Coroutines_Started);
402 cors->state = Coroutines_Active;
403 cors->primary = cor;
404
405 _Coroutine_Continue(cor, value, true);
406
407 if (!setjmp(cors->controller)){
408 pthread_mutex_unlock(&cors->mutex);
409 assert(r == 0);
410
411 // check the guard
412 if (!Check_Guard(g_c->guard)){
413 printf("Coroutine startup stack as has overrun - checked on entering the main coroutine\n");
414 exit(EXIT_FAILURE);
415 }
416
417 // start the first coroutine
418 Coroutine_RunNext();
419 }
420 // arrive here with mutex locked
421 assert(List_IsEmpty(&cors->runable));
422 assert(List_IsEmpty(&cors->waiting));
423 assert(cors->state == Coroutines_Active);
424 cors->state = Coroutines_Started;
425 pthread_mutex_unlock(&cors->mutex);
426 assert(r == 0);
427}
428
429
430void *Coroutine_Run(
431 Coroutine_Start start,
432 void *value
433){
434 Coroutine *cor = Coroutine_New(start);
435 Coroutine_Run_Coroutine(cor, value);
436 void *res = Coroutine_GetValue(cor);
437 Coroutine_Delete(cor);
438 return res;
439}
440
441
442Coroutine *Coroutine_New(
443 Coroutine_Start start
444){
445 assert((g_c->state == Coroutines_Started && List_IsEmpty(&g_c->inactive)) || g_c->state == Coroutines_Active);
446
447 // if none free - add one
448 if (List_IsEmpty(&g_c->free)){
449 if (!setjmp(g_c->chunk_allocated)){
450 longjmp(g_c->tip->buf, Chunk_Create);
451 }
452 }
453
454 Coroutine *cor = List_Link_Container(Coroutine, link, List_GetHead(&g_c->free));
455 assert(cor->state == Coroutine_Free);
456 cor->state = Coroutine_Idle;
457 cor->start = start;
458 cor->value = NULL;
459 List_Remove(&cor->link);
460 List_AddHead(&g_c->inactive, &cor->link);
461
462 g_c->report.coroutines_created += 1;
463
464 return cor;
465}
466
467
468void Coroutine_Delete(
469 Coroutine *cor
470){
471 Coroutines *cors = cor->coroutines;
472 int r = pthread_mutex_lock(&cors->mutex);
473 assert(r == 0);
474 assert(cor->state == Coroutine_Idle || cor->state == Coroutine_Complete);
475 cor->state = Coroutine_Free;
476 List_Remove(&cor->link);
477 List_AddTail(&cors->free, &cor->link);
478 r = pthread_mutex_unlock(&cors->mutex);
479 assert(r == 0);
480}
481
482
483// Coroutine_Continue, assuming the mutex is claimed
484static void _Coroutine_Continue(
485 Coroutine *cor,
486 void *value,
487 bool early
488){
489 Coroutines *cors = cor->coroutines;
490 assert(cor->state == Coroutine_Idle || cor->state == Coroutine_Waiting);
491 cor->entry_param = value;
492 cor->state = Coroutine_Running;
493 List_Remove(&cor->link);
494 if ( early ) {
495 List_AddHead(&cors->runable, &cor->link);
496 } else {
497 List_AddTail(&cors->runable, &cor->link);
498 }
499 int r;
500 r = pthread_cond_signal(&cors->waiting_cond);
501 assert(r == 0);
502}
503
504
505void Coroutine_Continue(
506 Coroutine *cor,
507 void *value,
508 bool early
509){
510 Coroutines *cors = cor->coroutines;
511 int r = pthread_mutex_lock(&cors->mutex);
512 assert(r == 0);
513 _Coroutine_Continue(cor, value, early);
514 r = pthread_mutex_unlock(&cors->mutex);
515 assert(r == 0);
516}
517
518
519void *Coroutine_Yield(
520 void *value,
521 Coroutine_YieldCallback on_yield,
522 void *yield_me
523){
524 Coroutine *me = g_c->active;
525 if (!Check_Guard(me->guard)){
526 printf("Coroutine has overrun its stack - checked when yielding coroutine\n");
527 exit(EXIT_FAILURE);
528 }
529
530 int r = pthread_mutex_lock(&g_c->mutex);
531 assert(r == 0);
532 Coroutines *cors = me->coroutines;
533 assert(me && me->state == Coroutine_Running && cors == g_c);
534 me->value = value;
535 me->state = Coroutine_Waiting;
536
537 List_Remove(&me->link);
538 List_AddTail(&cors->waiting, &me->link);
539
540 switch (setjmp(me->buf)){
541 case Chunk_Initial:
542 r = pthread_mutex_unlock(&cors->mutex);
543 assert(r == 0);
544 on_yield(yield_me);
545 Coroutine_RunNext();
546 case Chunk_Create:
547 assert(false);
548 case Chunk_Enter:
549 // arrive here with mutex locked
550 cors->active = me;
551 // when we return here - we are running again
552 assert(me->state == Coroutine_Running);
553 void *res = me->entry_param;
554 r = pthread_mutex_unlock(&cors->mutex);
555 assert(r == 0);
556 return res;
557 }
558 return NULL;
559}
560
561
562void *Coroutine_GetValue(
563 Coroutine *cor
564){
565 return cor->value;
566}
567
568
569Coroutine *Coroutine_GetActive()
570{
571 return g_c->active;
572}
573
574
575int Coroutine_GetStackHeadroom(){
576 unsigned char tbuf[4];
577 return tbuf - g_c->active->guard - 4;
578}
579
580
581bool Coroutine_HasCoroutinesInFreePool(){
582 return !List_IsEmpty(&g_c->free);
583}
584
585
586void *Coroutine_GetCStackTop(){
587 return g_c->tip;
588}
589
590
591struct Coroutine_ChainParam {
592 Coroutine_Start start;
593 void *value;
594 Coroutine *ret;
595};
596
597
598static void *Coroutine_ChainFn(
599 void *param
600){
601 struct Coroutine_ChainParam *params = (struct Coroutine_ChainParam *)param;
602 Coroutine_Continue(params->ret, params->start(params->value), true);
603 return NULL;
604}
605
606
607static void Coroutine_ChainYield(
608 void *unused
609){
610 (void)unused;
611}
612
613
614void *Coroutine_Chain(
615 Coroutine_Start start,
616 void *value
617){
618 if (!Check_Guard(Coroutine_GetActive()->guard)){
619 printf("Coroutine has overrun its stack - checked when chaining\n");
620 exit(EXIT_FAILURE);
621 }
622 Coroutine *cor = Coroutine_New(Coroutine_ChainFn);
623 struct Coroutine_ChainParam params = {
624 start,
625 value,
626 Coroutine_GetActive()
627 };
628 Coroutine_Continue(cor, &params, true);
629 void *res = Coroutine_Yield(NULL, Coroutine_ChainYield, NULL);
630 Coroutine_Delete(cor);
631 return res;
632}
633
634
635bool Coroutine_IsRunning(
636 Coroutine *cor
637)
638{
639 int state = cor->state;
640 return state == Coroutine_Running || state == Coroutine_Waiting;
641}
642
643
644bool Coroutine_IsStarted(){
645 return g_c != NULL;
646}
647