source: libcfa/src/concurrency/locks.cfa @ 4e83bb7

pthread-emulationqualifiedEnum
Last change on this file since 4e83bb7 was 4e83bb7, checked in by caparsons <caparson@…>, 6 months ago

fixed timeout cond var bug and updated pthread_locks test with timeout lock test

  • Property mode set to 100644
File size: 17.3 KB
Line 
1//
2// Cforall Version 1.0.0 Copyright (C) 2021 University of Waterloo
3//
4// The contents of this file are covered under the licence agreement in the
5// file "LICENCE" distributed with Cforall.
6//
7// locks.hfa -- LIBCFATHREAD
8// Runtime locks that used with the runtime thread system.
9//
10// Author           : Colby Alexander Parsons
11// Created On       : Thu Jan 21 19:46:50 2021
12// Last Modified By :
13// Last Modified On :
14// Update Count     :
15//
16
17#define __cforall_thread__
18#define _GNU_SOURCE
19
20#include "locks.hfa"
21#include "kernel/private.hfa"
22
23#include <kernel.hfa>
24#include <stdlib.hfa>
25
26#pragma GCC visibility push(default)
27
28//-----------------------------------------------------------------------------
29// info_thread
30forall(L & | is_blocking_lock(L)) {
31        struct info_thread {
32                // used to put info_thread on a dl queue
33                inline dlink(info_thread(L));
34
35                // waiting thread
36                struct thread$ * t;
37
38                // shadow field
39                uintptr_t info;
40
41                // lock that is passed to wait() (if one is passed)
42                L * lock;
43
44                // true when signalled and false when timeout wakes thread
45                bool signalled;
46        };
47        P9_EMBEDDED( info_thread(L), dlink(info_thread(L)) )
48
49        void ?{}( info_thread(L) & this, thread$ * t, uintptr_t info, L * l ) {
50                this.t = t;
51                this.info = info;
52                this.lock = l;
53        }
54
55        void ^?{}( info_thread(L) & this ) {}
56}
57
58//-----------------------------------------------------------------------------
59// Blocking Locks
60void ?{}( blocking_lock & this, bool multi_acquisition, bool strict_owner ) {
61        this.lock{};
62        this.blocked_threads{};
63        this.wait_count = 0;
64        this.multi_acquisition = multi_acquisition;
65        this.strict_owner = strict_owner;
66        this.owner = 0p;
67        this.recursion_count = 0;
68}
69
70void ^?{}( blocking_lock & this ) {}
71
72
73void lock( blocking_lock & this ) with( this ) {
74        lock( lock __cfaabi_dbg_ctx2 );
75        thread$ * thrd = active_thread();
76
77        // single acquisition lock is held by current thread
78        /* paranoid */ verifyf( owner != thrd || multi_acquisition, "Single acquisition lock holder (%p) attempted to reacquire the lock %p resulting in a deadlock.", owner, &this );
79
80        // lock is held by some other thread
81        if ( owner != 0p && owner != thrd ) {
82                insert_last( blocked_threads, *thrd );
83                wait_count++;
84                unlock( lock );
85                park( );
86        }
87        // multi acquisition lock is held by current thread
88        else if ( owner == thrd && multi_acquisition ) {
89                recursion_count++;
90                unlock( lock );
91        }
92        // lock isn't held
93        else {
94                owner = thrd;
95                recursion_count = 1;
96                unlock( lock );
97        }
98}
99
100bool try_lock( blocking_lock & this ) with( this ) {
101        bool ret = false;
102        lock( lock __cfaabi_dbg_ctx2 );
103
104        // lock isn't held
105        if ( owner == 0p ) {
106                owner = active_thread();
107                recursion_count = 1;
108                ret = true;
109        }
110        // multi acquisition lock is held by current thread
111        else if ( owner == active_thread() && multi_acquisition ) {
112                recursion_count++;
113                ret = true;
114        }
115
116        unlock( lock );
117        return ret;
118}
119
120static void pop_and_set_new_owner( blocking_lock & this ) with( this ) {
121        thread$ * t = &try_pop_front( blocked_threads );
122        owner = t;
123        recursion_count = ( t ? 1 : 0 );
124        if ( t ) wait_count--;
125        unpark( t );
126}
127
128void unlock( blocking_lock & this ) with( this ) {
129        lock( lock __cfaabi_dbg_ctx2 );
130        /* paranoid */ verifyf( owner != 0p, "Attempt to release lock %p that isn't held", &this );
131        /* paranoid */ verifyf( owner == active_thread() || !strict_owner , "Thread %p other than the owner %p attempted to release owner lock %p", owner, active_thread(), &this );
132        /* paranoid */ verifyf( recursion_count == 1 || multi_acquisition, "Thread %p attempted to release owner lock %p which is not recursive but has a recursive count of %zu", active_thread(), &this, recursion_count );
133
134        // if recursion count is zero release lock and set new owner if one is waiting
135        recursion_count--;
136        if ( recursion_count == 0 ) {
137                pop_and_set_new_owner( this );
138        }
139        unlock( lock );
140}
141
142size_t wait_count( blocking_lock & this ) with( this ) {
143        return wait_count;
144}
145
146void on_notify( blocking_lock & this, thread$ * t ) with( this ) {
147        lock( lock __cfaabi_dbg_ctx2 );
148        // lock held
149        if ( owner != 0p ) {
150                insert_last( blocked_threads, *t );
151                wait_count++;
152                unlock( lock );
153        }
154        // lock not held
155        else {
156                owner = t;
157                recursion_count = 1;
158                unpark( t );
159                unlock( lock );
160        }
161}
162
163size_t on_wait( blocking_lock & this ) with( this ) {
164        lock( lock __cfaabi_dbg_ctx2 );
165        /* paranoid */ verifyf( owner != 0p, "Attempt to release lock %p that isn't held", &this );
166        /* paranoid */ verifyf( owner == active_thread() || !strict_owner, "Thread %p other than the owner %p attempted to release owner lock %p", owner, active_thread(), &this );
167
168        size_t ret = recursion_count;
169
170        pop_and_set_new_owner( this );
171        unlock( lock );
172        return ret;
173}
174
175void on_wakeup( blocking_lock & this, size_t recursion ) with( this ) {
176        recursion_count = recursion;
177}
178
179//-----------------------------------------------------------------------------
180// alarm node wrapper
181forall(L & | is_blocking_lock(L)) {
182        struct alarm_node_wrap {
183                alarm_node_t alarm_node;
184                condition_variable(L) * cond;
185                info_thread(L) * info_thd;
186        };
187
188        void ?{}( alarm_node_wrap(L) & this, Duration alarm, Duration period, Alarm_Callback callback, condition_variable(L) * c, info_thread(L) * i ) {
189                this.alarm_node{ callback, alarm, period };
190                this.cond = c;
191                this.info_thd = i;
192        }
193
194        void ^?{}( alarm_node_wrap(L) & this ) { }
195
196        static void timeout_handler ( alarm_node_wrap(L) & this ) with( this ) {
197                // This condition_variable member is called from the kernel, and therefore, cannot block, but it can spin.
198                lock( cond->lock __cfaabi_dbg_ctx2 );
199
200                // this check is necessary to avoid a race condition since this timeout handler
201                //      may still be called after a thread has been removed from the queue but
202                //      before the alarm is unregistered
203                if ( (*info_thd)`isListed ) {   // is thread on queue
204                        info_thd->signalled = false;
205                        // remove this thread O(1)
206                        remove( *info_thd );
207                        cond->count--;
208                        if( info_thd->lock ) {
209                                // call lock's on_notify if a lock was passed
210                                on_notify(*info_thd->lock, info_thd->t);
211                        } else {
212                                // otherwise wake thread
213                                unpark( info_thd->t );
214                        }
215                }
216                unlock( cond->lock );
217        }
218
219        // this casts the alarm node to our wrapped type since we used type erasure
220        static void alarm_node_wrap_cast( alarm_node_t & a ) { timeout_handler( (alarm_node_wrap(L) &)a ); }
221
222        struct pthread_alarm_node_wrap {
223                alarm_node_t alarm_node;
224                pthread_cond_var(L) * cond;
225                info_thread(L) * info_thd;
226        };
227
228        void ?{}( pthread_alarm_node_wrap(L) & this, Duration alarm, Duration period, Alarm_Callback callback, pthread_cond_var(L) * c, info_thread(L) * i ) {
229                this.alarm_node{ callback, alarm, period };
230                this.cond = c;
231                this.info_thd = i;
232        }
233
234        void ^?{}( pthread_alarm_node_wrap(L) & this ) { }
235
236        static void timeout_handler ( pthread_alarm_node_wrap(L) & this ) with( this ) {
237                // This pthread_cond_var member is called from the kernel, and therefore, cannot block, but it can spin.
238                lock( cond->lock __cfaabi_dbg_ctx2 );
239                // this check is necessary to avoid a race condition since this timeout handler
240                //      may still be called after a thread has been removed from the queue but
241                //      before the alarm is unregistered
242                if ( (*info_thd)`isListed ) {   // is thread on queue
243                        info_thd->signalled = false;
244                        // remove this thread O(1)
245                        remove( *info_thd );
246                        on_notify(*info_thd->lock, info_thd->t);
247                }
248                unlock( cond->lock );
249        }
250
251        // this casts the alarm node to our wrapped type since we used type erasure
252        static void pthread_alarm_node_wrap_cast( alarm_node_t & a ) { timeout_handler( (pthread_alarm_node_wrap(L) &)a ); }
253}
254
255//-----------------------------------------------------------------------------
256// Synchronization Locks
257forall(L & | is_blocking_lock(L)) {
258
259        //-----------------------------------------------------------------------------
260        // condition variable
261        void ?{}( condition_variable(L) & this ){
262                this.lock{};
263                this.blocked_threads{};
264                this.count = 0;
265        }
266
267        void ^?{}( condition_variable(L) & this ){ }
268
269        static void process_popped( condition_variable(L) & this, info_thread(L) & popped ) with( this ) {
270                if(&popped != 0p) {
271                        popped.signalled = true;
272                        count--;
273                        if (popped.lock) {
274                                // if lock passed call on_notify
275                                on_notify(*popped.lock, popped.t);
276                        } else {
277                                // otherwise wake thread
278                                unpark(popped.t);
279                        }
280                }
281        }
282
283        bool notify_one( condition_variable(L) & this ) with( this ) {
284                lock( lock __cfaabi_dbg_ctx2 );
285                bool ret = ! blocked_threads`isEmpty;
286                process_popped(this, try_pop_front( blocked_threads ));
287                unlock( lock );
288                return ret;
289        }
290
291        bool notify_all( condition_variable(L) & this ) with(this) {
292                lock( lock __cfaabi_dbg_ctx2 );
293                bool ret = ! blocked_threads`isEmpty;
294                while( ! blocked_threads`isEmpty ) {
295                        process_popped(this, try_pop_front( blocked_threads ));
296                }
297                unlock( lock );
298                return ret;
299        }
300
301        uintptr_t front( condition_variable(L) & this ) with(this) {
302                return blocked_threads`isEmpty ? NULL : blocked_threads`first.info;
303        }
304
305        bool empty( condition_variable(L) & this ) with(this) {
306                lock( lock __cfaabi_dbg_ctx2 );
307                bool ret = blocked_threads`isEmpty;
308                unlock( lock );
309                return ret;
310        }
311
312        int counter( condition_variable(L) & this ) with(this) { return count; }
313
314        static size_t queue_and_get_recursion( condition_variable(L) & this, info_thread(L) * i ) with(this) {
315                // add info_thread to waiting queue
316                insert_last( blocked_threads, *i );
317                count++;
318                size_t recursion_count = 0;
319                if (i->lock) {
320                        // if lock was passed get recursion count to reset to after waking thread
321                        recursion_count = on_wait( *i->lock );
322                }
323                return recursion_count;
324        }
325
326        // helper for wait()'s' with no timeout
327        static void queue_info_thread( condition_variable(L) & this, info_thread(L) & i ) with(this) {
328                lock( lock __cfaabi_dbg_ctx2 );
329                size_t recursion_count = queue_and_get_recursion(this, &i);
330                unlock( lock );
331
332                // blocks here
333                park( );
334
335                // resets recursion count here after waking
336                if (i.lock) on_wakeup(*i.lock, recursion_count);
337        }
338
339        #define WAIT( u, l ) \
340                info_thread( L ) i = { active_thread(), u, l }; \
341                queue_info_thread( this, i );
342
343        // helper for wait()'s' with a timeout
344        static void queue_info_thread_timeout( condition_variable(L) & this, info_thread(L) & info, Duration t, Alarm_Callback callback ) with(this) {
345                lock( lock __cfaabi_dbg_ctx2 );
346                size_t recursion_count = queue_and_get_recursion(this, &info);
347                alarm_node_wrap(L) node_wrap = { t, 0`s, callback, &this, &info };
348                unlock( lock );
349
350                // registers alarm outside cond lock to avoid deadlock
351                register_self( &node_wrap.alarm_node );
352
353                // blocks here
354                park();
355
356                // unregisters alarm so it doesn't go off if this happens first
357                unregister_self( &node_wrap.alarm_node );
358
359                // resets recursion count here after waking
360                if (info.lock) on_wakeup(*info.lock, recursion_count);
361        }
362
363        #define WAIT_TIME( u, l, t ) \
364                info_thread( L ) i = { active_thread(), u, l }; \
365                queue_info_thread_timeout(this, i, t, alarm_node_wrap_cast ); \
366                return i.signalled;
367
368        void wait( condition_variable(L) & this                        ) with(this) { WAIT( 0, 0p    ) }
369        void wait( condition_variable(L) & this, uintptr_t info        ) with(this) { WAIT( info, 0p ) }
370        void wait( condition_variable(L) & this, L & l                 ) with(this) { WAIT( 0, &l    ) }
371        void wait( condition_variable(L) & this, L & l, uintptr_t info ) with(this) { WAIT( info, &l ) }
372
373        bool wait( condition_variable(L) & this, Duration duration                        ) with(this) { WAIT_TIME( 0   , 0p , duration ) }
374        bool wait( condition_variable(L) & this, uintptr_t info, Duration duration        ) with(this) { WAIT_TIME( info, 0p , duration ) }
375        bool wait( condition_variable(L) & this, L & l, Duration duration                 ) with(this) { WAIT_TIME( 0   , &l , duration ) }
376        bool wait( condition_variable(L) & this, L & l, uintptr_t info, Duration duration ) with(this) { WAIT_TIME( info, &l , duration ) }
377
378        //-----------------------------------------------------------------------------
379        // fast_cond_var
380        void  ?{}( fast_cond_var(L) & this ){
381                this.blocked_threads{};
382                #ifdef __CFA_DEBUG__
383                this.lock_used = 0p;
384                #endif
385        }
386        void ^?{}( fast_cond_var(L) & this ){ }
387
388        bool notify_one( fast_cond_var(L) & this ) with(this) {
389                bool ret = ! blocked_threads`isEmpty;
390                if ( ret ) {
391                        info_thread(L) & popped = try_pop_front( blocked_threads );
392                        on_notify(*popped.lock, popped.t);
393                }
394                return ret;
395        }
396        bool notify_all( fast_cond_var(L) & this ) with(this) {
397                bool ret = ! blocked_threads`isEmpty;
398                while( ! blocked_threads`isEmpty ) {
399                        info_thread(L) & popped = try_pop_front( blocked_threads );
400                        on_notify(*popped.lock, popped.t);
401                }
402                return ret;
403        }
404
405        uintptr_t front( fast_cond_var(L) & this ) with(this) { return blocked_threads`isEmpty ? NULL : blocked_threads`first.info; }
406        bool empty ( fast_cond_var(L) & this ) with(this) { return blocked_threads`isEmpty; }
407
408        void wait( fast_cond_var(L) & this, L & l ) {
409                wait( this, l, 0 );
410        }
411
412        void wait( fast_cond_var(L) & this, L & l, uintptr_t info ) with(this) {
413                // brand cond lock with lock
414                #ifdef __CFA_DEBUG__
415                        if ( lock_used == 0p ) lock_used = &l;
416                        else { assert(lock_used == &l); }
417                #endif
418                info_thread( L ) i = { active_thread(), info, &l };
419                insert_last( blocked_threads, i );
420                size_t recursion_count = on_wait( *i.lock );
421                park( );
422                on_wakeup(*i.lock, recursion_count);
423        }
424
425        //-----------------------------------------------------------------------------
426        // pthread_cond_var
427
428        void  ?{}( pthread_cond_var(L) & this ) with(this) {
429                blocked_threads{};
430                lock{};
431        }
432
433        void ^?{}( pthread_cond_var(L) & this ) { }
434
435        bool notify_one( pthread_cond_var(L) & this ) with(this) {
436                lock( lock __cfaabi_dbg_ctx2 );
437                bool ret = ! blocked_threads`isEmpty;
438                if ( ret ) {
439                        info_thread(L) & popped = try_pop_front( blocked_threads );
440                        popped.signalled = true;
441                        on_notify(*popped.lock, popped.t);
442                }
443                unlock( lock );
444                return ret;
445        }
446
447        bool notify_all( pthread_cond_var(L) & this ) with(this) {
448                lock( lock __cfaabi_dbg_ctx2 );
449                bool ret = ! blocked_threads`isEmpty;
450                while( ! blocked_threads`isEmpty ) {
451                        info_thread(L) & popped = try_pop_front( blocked_threads );
452                        popped.signalled = true;
453                        on_notify(*popped.lock, popped.t);
454                }
455                unlock( lock );
456                return ret;
457        }
458
459        uintptr_t front( pthread_cond_var(L) & this ) with(this) { return blocked_threads`isEmpty ? NULL : blocked_threads`first.info; }
460        bool empty ( pthread_cond_var(L) & this ) with(this) { return blocked_threads`isEmpty; }
461
462        static size_t queue_and_get_recursion( pthread_cond_var(L) & this, info_thread(L) * i ) with(this) {
463                // add info_thread to waiting queue
464                insert_last( blocked_threads, *i );
465                size_t recursion_count = 0;
466                recursion_count = on_wait( *i->lock );
467                return recursion_count;
468        }
469       
470        static void queue_info_thread_timeout( pthread_cond_var(L) & this, info_thread(L) & info, Duration t, Alarm_Callback callback ) with(this) {
471                lock( lock __cfaabi_dbg_ctx2 );
472                size_t recursion_count = queue_and_get_recursion(this, &info);
473                pthread_alarm_node_wrap(L) node_wrap = { t, 0`s, callback, &this, &info };
474                unlock( lock );
475
476                // registers alarm outside cond lock to avoid deadlock
477                register_self( &node_wrap.alarm_node );
478
479                // blocks here
480                park();
481
482                // unregisters alarm so it doesn't go off if this happens first
483                unregister_self( &node_wrap.alarm_node );
484
485                // resets recursion count here after waking
486                if (info.lock) on_wakeup(*info.lock, recursion_count);
487        }
488
489        void wait( pthread_cond_var(L) & this, L & l ) with(this) {
490                wait( this, l, 0 );
491        }
492
493        void wait( pthread_cond_var(L) & this, L & l, uintptr_t info ) with(this) {
494                lock( lock __cfaabi_dbg_ctx2 );
495                info_thread( L ) i = { active_thread(), info, &l };
496                size_t recursion_count = queue_and_get_recursion(this, &i);
497                unlock( lock );
498                park( );
499                on_wakeup(*i.lock, recursion_count);
500        }
501
502        #define PTHREAD_WAIT_TIME( u, l, t ) \
503                info_thread( L ) i = { active_thread(), u, l }; \
504                queue_info_thread_timeout(this, i, t, pthread_alarm_node_wrap_cast ); \
505                return i.signalled;
506
507        Duration getDuration(timespec t) {
508                timespec currTime;
509                clock_gettime(CLOCK_REALTIME, &currTime);
510                Duration waitUntil = { t };
511                Duration currDur = { currTime };
512                if ( currDur >= waitUntil ) return currDur - waitUntil;
513                Duration zero = { 0 };
514                return zero;
515        }
516
517        bool wait( pthread_cond_var(L) & this, L & l, timespec t ) {
518                PTHREAD_WAIT_TIME( 0, &l , getDuration( t ) )
519        }
520       
521        bool wait( pthread_cond_var(L) & this, L & l, uintptr_t info, timespec t  ) {
522                PTHREAD_WAIT_TIME( info, &l , getDuration( t ) )
523        }
524}
525//-----------------------------------------------------------------------------
526// Semaphore
527void  ?{}( semaphore & this, int count = 1 ) {
528        (this.lock){};
529        this.count = count;
530        (this.waiting){};
531}
532void ^?{}(semaphore & this) {}
533
534bool P(semaphore & this) with( this ){
535        lock( lock __cfaabi_dbg_ctx2 );
536        count -= 1;
537        if ( count < 0 ) {
538                // queue current task
539                append( waiting, active_thread() );
540
541                // atomically release spin lock and block
542                unlock( lock );
543                park();
544                return true;
545        }
546        else {
547            unlock( lock );
548            return false;
549        }
550}
551
552thread$ * V (semaphore & this, const bool doUnpark ) with( this ) {
553        thread$ * thrd = 0p;
554        lock( lock __cfaabi_dbg_ctx2 );
555        count += 1;
556        if ( count <= 0 ) {
557                // remove task at head of waiting list
558                thrd = pop_head( waiting );
559        }
560
561        unlock( lock );
562
563        // make new owner
564        if( doUnpark ) unpark( thrd );
565
566        return thrd;
567}
568
569bool V(semaphore & this) with( this ) {
570        thread$ * thrd = V(this, true);
571        return thrd != 0p;
572}
573
574bool V(semaphore & this, unsigned diff) with( this ) {
575        thread$ * thrd = 0p;
576        lock( lock __cfaabi_dbg_ctx2 );
577        int release = max(-count, (int)diff);
578        count += diff;
579        for(release) {
580                unpark( pop_head( waiting ) );
581        }
582
583        unlock( lock );
584
585        return thrd != 0p;
586}
Note: See TracBrowser for help on using the repository browser.