source: libcfa/src/concurrency/locks.cfa @ 43784ac

ADTast-experimentalenumforall-pointer-decayjacob/cs343-translationnew-ast-unique-exprpthread-emulationqualifiedEnum
Last change on this file since 43784ac was 43784ac, checked in by Thierry Delisle <tdelisle@…>, 3 years ago

Changed libcfathread to consistently define _GNU_SOURCE

  • Property mode set to 100644
File size: 11.5 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//-----------------------------------------------------------------------------
27// info_thread
28forall(L & | is_blocking_lock(L)) {
29        struct info_thread {
30                // used to put info_thread on a dl queue (aka sequence)
31                inline Seqable;
32
33                // waiting thread
34                struct $thread * t;
35
36                // shadow field
37                uintptr_t info;
38
39                // lock that is passed to wait() (if one is passed)
40                L * lock;
41
42                // true when signalled and false when timeout wakes thread
43                bool signalled;
44        };
45
46        void ?{}( info_thread(L) & this, $thread * t, uintptr_t info, L * l ) {
47                ((Seqable &) this){};
48                this.t = t;
49                this.info = info;
50                this.lock = l;
51        }
52
53        void ^?{}( info_thread(L) & this ) {}
54
55        info_thread(L) *& Back( info_thread(L) * this ) {
56                return (info_thread(L) *)Back( (Seqable *)this );
57        }
58
59        info_thread(L) *& Next( info_thread(L) * this ) {
60                return (info_thread(L) *)Next( (Colable *)this );
61        }
62}
63
64//-----------------------------------------------------------------------------
65// Blocking Locks
66void ?{}( blocking_lock & this, bool multi_acquisition, bool strict_owner ) {
67        this.lock{};
68        this.blocked_threads{};
69        this.wait_count = 0;
70        this.multi_acquisition = multi_acquisition;
71        this.strict_owner = strict_owner;
72        this.owner = 0p;
73        this.recursion_count = 0;
74}
75
76void ^?{}( blocking_lock & this ) {}
77
78
79void lock( blocking_lock & this ) with( this ) {
80        lock( lock __cfaabi_dbg_ctx2 );
81        $thread * thrd = active_thread();
82
83        // single acquisition lock is held by current thread
84        /* paranoid */ verifyf( owner != thrd || multi_acquisition, "Single acquisition lock holder (%p) attempted to reacquire the lock %p resulting in a deadlock.", owner, &this );
85
86        // lock is held by some other thread
87        if ( owner != 0p && owner != thrd ) {
88                addTail( blocked_threads, *thrd );
89                wait_count++;
90                unlock( lock );
91                park( );
92        }
93        // multi acquisition lock is held by current thread
94        else if ( owner == thrd && multi_acquisition ) {
95                recursion_count++;
96                unlock( lock );
97        }
98        // lock isn't held
99        else {
100                owner = thrd;
101                recursion_count = 1;
102                unlock( lock );
103        }
104}
105
106bool try_lock( blocking_lock & this ) with( this ) {
107        bool ret = false;
108        lock( lock __cfaabi_dbg_ctx2 );
109
110        // lock isn't held
111        if ( owner == 0p ) {
112                owner = active_thread();
113                recursion_count = 1;
114                ret = true;
115        }
116        // multi acquisition lock is held by current thread
117        else if ( owner == active_thread() && multi_acquisition ) {
118                recursion_count++;
119                ret = true;
120        }
121
122        unlock( lock );
123        return ret;
124}
125
126void pop_and_set_new_owner( blocking_lock & this ) with( this ) {
127        $thread * t = &dropHead( blocked_threads );
128        owner = t;
129        recursion_count = ( t ? 1 : 0 );
130        wait_count--;
131        unpark( t );
132}
133
134void unlock( blocking_lock & this ) with( this ) {
135        lock( lock __cfaabi_dbg_ctx2 );
136        /* paranoid */ verifyf( owner != 0p, "Attempt to release lock %p that isn't held", &this );
137        /* paranoid */ verifyf( owner == active_thread() || !strict_owner , "Thread %p other than the owner %p attempted to release owner lock %p", owner, active_thread(), &this );
138        /* 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 );
139
140        // if recursion count is zero release lock and set new owner if one is waiting
141        recursion_count--;
142        if ( recursion_count == 0 ) {
143                pop_and_set_new_owner( this );
144        }
145        unlock( lock );
146}
147
148size_t wait_count( blocking_lock & this ) with( this ) {
149        return wait_count;
150}
151
152void on_notify( blocking_lock & this, $thread * t ) with( this ) {
153        lock( lock __cfaabi_dbg_ctx2 );
154        // lock held
155        if ( owner != 0p ) {
156                addTail( blocked_threads, *t );
157                wait_count++;
158                unlock( lock );
159        }
160        // lock not held
161        else {
162                owner = t;
163                recursion_count = 1;
164                unpark( t );
165                unlock( lock );
166        }
167}
168
169size_t on_wait( blocking_lock & this ) with( this ) {
170        lock( lock __cfaabi_dbg_ctx2 );
171        /* paranoid */ verifyf( owner != 0p, "Attempt to release lock %p that isn't held", &this );
172        /* paranoid */ verifyf( owner == active_thread() || !strict_owner, "Thread %p other than the owner %p attempted to release owner lock %p", owner, active_thread(), &this );
173
174        size_t ret = recursion_count;
175
176        pop_and_set_new_owner( this );
177        unlock( lock );
178        return ret;
179}
180
181void on_wakeup( blocking_lock & this, size_t recursion ) with( this ) {
182        recursion_count = recursion;
183}
184
185//-----------------------------------------------------------------------------
186// alarm node wrapper
187forall(L & | is_blocking_lock(L)) {
188        struct alarm_node_wrap {
189                alarm_node_t alarm_node;
190                condition_variable(L) * cond;
191                info_thread(L) * info_thd;
192        };
193
194        void ?{}( alarm_node_wrap(L) & this, Duration alarm, Duration period, Alarm_Callback callback, condition_variable(L) * c, info_thread(L) * i ) {
195                this.alarm_node{ callback, alarm, period };
196                this.cond = c;
197                this.info_thd = i;
198        }
199
200        void ^?{}( alarm_node_wrap(L) & this ) { }
201
202        void timeout_handler ( alarm_node_wrap(L) & this ) with( this ) {
203                // This condition_variable member is called from the kernel, and therefore, cannot block, but it can spin.
204                lock( cond->lock __cfaabi_dbg_ctx2 );
205
206                // this check is necessary to avoid a race condition since this timeout handler
207                //      may still be called after a thread has been removed from the queue but
208                //      before the alarm is unregistered
209                if ( listed(info_thd) ) {       // is thread on queue
210                        info_thd->signalled = false;
211                        // remove this thread O(1)
212                        remove( cond->blocked_threads, *info_thd );
213                        cond->count--;
214                        if( info_thd->lock ) {
215                                // call lock's on_notify if a lock was passed
216                                on_notify(*info_thd->lock, info_thd->t);
217                        } else {
218                                // otherwise wake thread
219                                unpark( info_thd->t );
220                        }
221                }
222                unlock( cond->lock );
223        }
224
225        // this casts the alarm node to our wrapped type since we used type erasure
226        void alarm_node_wrap_cast( alarm_node_t & a ) { timeout_handler( (alarm_node_wrap(L) &)a ); }
227}
228
229//-----------------------------------------------------------------------------
230// condition variable
231forall(L & | is_blocking_lock(L)) {
232
233        void ?{}( condition_variable(L) & this ){
234                this.lock{};
235                this.blocked_threads{};
236                this.count = 0;
237        }
238
239        void ^?{}( condition_variable(L) & this ){ }
240
241        void process_popped( condition_variable(L) & this, info_thread(L) & popped ) with( this ) {
242                if(&popped != 0p) {
243                        popped.signalled = true;
244                        count--;
245                        if (popped.lock) {
246                                // if lock passed call on_notify
247                                on_notify(*popped.lock, popped.t);
248                        } else {
249                                // otherwise wake thread
250                                unpark(popped.t);
251                        }
252                }
253        }
254
255        bool notify_one( condition_variable(L) & this ) with( this ) {
256                lock( lock __cfaabi_dbg_ctx2 );
257                bool ret = !empty(blocked_threads);
258                process_popped(this, dropHead( blocked_threads ));
259                unlock( lock );
260                return ret;
261        }
262
263        bool notify_all( condition_variable(L) & this ) with(this) {
264                lock( lock __cfaabi_dbg_ctx2 );
265                bool ret = !empty(blocked_threads);
266                while( !empty(blocked_threads) ) {
267                        process_popped(this, dropHead( blocked_threads ));
268                }
269                unlock( lock );
270                return ret;
271        }
272
273        uintptr_t front( condition_variable(L) & this ) with(this) {
274                return empty(blocked_threads) ? NULL : head(blocked_threads).info;
275        }
276
277        bool empty( condition_variable(L) & this ) with(this) {
278                lock( lock __cfaabi_dbg_ctx2 );
279                bool ret = empty(blocked_threads);
280                unlock( lock );
281                return ret;
282        }
283
284        int counter( condition_variable(L) & this ) with(this) { return count; }
285
286        size_t queue_and_get_recursion( condition_variable(L) & this, info_thread(L) * i ) with(this) {
287                // add info_thread to waiting queue
288                addTail( blocked_threads, *i );
289                count++;
290                size_t recursion_count = 0;
291                if (i->lock) {
292                        // if lock was passed get recursion count to reset to after waking thread
293                        recursion_count = on_wait( *i->lock );
294                }
295                return recursion_count;
296        }
297
298        // helper for wait()'s' with no timeout
299        void queue_info_thread( condition_variable(L) & this, info_thread(L) & i ) with(this) {
300                lock( lock __cfaabi_dbg_ctx2 );
301                size_t recursion_count = queue_and_get_recursion(this, &i);
302                unlock( lock );
303
304                // blocks here
305                park( );
306
307                // resets recursion count here after waking
308                if (i.lock) on_wakeup(*i.lock, recursion_count);
309        }
310
311        #define WAIT( u, l ) \
312                info_thread( L ) i = { active_thread(), u, l }; \
313                queue_info_thread( this, i );
314
315        // helper for wait()'s' with a timeout
316        void queue_info_thread_timeout( condition_variable(L) & this, info_thread(L) & info, Duration t, Alarm_Callback callback ) with(this) {
317                lock( lock __cfaabi_dbg_ctx2 );
318                size_t recursion_count = queue_and_get_recursion(this, &info);
319                alarm_node_wrap(L) node_wrap = { t, 0`s, callback, &this, &info };
320                register_self( &node_wrap.alarm_node );
321                unlock( lock );
322
323                // blocks here
324                park();
325
326                // unregisters alarm so it doesn't go off if this happens first
327                unregister_self( &node_wrap.alarm_node );
328
329                // resets recursion count here after waking
330                if (info.lock) on_wakeup(*info.lock, recursion_count);
331        }
332
333        #define WAIT_TIME( u, l, t ) \
334                info_thread( L ) i = { active_thread(), u, l }; \
335                queue_info_thread_timeout(this, i, t, alarm_node_wrap_cast ); \
336                return i.signalled;
337
338        void wait( condition_variable(L) & this                        ) with(this) { WAIT( 0, 0p    ) }
339        void wait( condition_variable(L) & this, uintptr_t info        ) with(this) { WAIT( info, 0p ) }
340        void wait( condition_variable(L) & this, L & l                 ) with(this) { WAIT( 0, &l    ) }
341        void wait( condition_variable(L) & this, L & l, uintptr_t info ) with(this) { WAIT( info, &l ) }
342
343        bool wait( condition_variable(L) & this, Duration duration                        ) with(this) { WAIT_TIME( 0   , 0p , duration ) }
344        bool wait( condition_variable(L) & this, uintptr_t info, Duration duration        ) with(this) { WAIT_TIME( info, 0p , duration ) }
345        bool wait( condition_variable(L) & this, L & l, Duration duration                 ) with(this) { WAIT_TIME( 0   , &l , duration ) }
346        bool wait( condition_variable(L) & this, L & l, uintptr_t info, Duration duration ) with(this) { WAIT_TIME( info, &l , duration ) }
347}
348
349//-----------------------------------------------------------------------------
350// Semaphore
351void  ?{}( semaphore & this, int count = 1 ) {
352        (this.lock){};
353        this.count = count;
354        (this.waiting){};
355}
356void ^?{}(semaphore & this) {}
357
358bool P(semaphore & this) with( this ){
359        lock( lock __cfaabi_dbg_ctx2 );
360        count -= 1;
361        if ( count < 0 ) {
362                // queue current task
363                append( waiting, active_thread() );
364
365                // atomically release spin lock and block
366                unlock( lock );
367                park();
368                return true;
369        }
370        else {
371            unlock( lock );
372            return false;
373        }
374}
375
376$thread * V (semaphore & this, const bool doUnpark ) with( this ) {
377        $thread * thrd = 0p;
378        lock( lock __cfaabi_dbg_ctx2 );
379        count += 1;
380        if ( count <= 0 ) {
381                // remove task at head of waiting list
382                thrd = pop_head( waiting );
383        }
384
385        unlock( lock );
386
387        // make new owner
388        if( doUnpark ) unpark( thrd );
389
390        return thrd;
391}
392
393bool V(semaphore & this) with( this ) {
394        $thread * thrd = V(this, true);
395        return thrd != 0p;
396}
397
398bool V(semaphore & this, unsigned diff) with( this ) {
399        $thread * thrd = 0p;
400        lock( lock __cfaabi_dbg_ctx2 );
401        int release = max(-count, (int)diff);
402        count += diff;
403        for(release) {
404                unpark( pop_head( waiting ) );
405        }
406
407        unlock( lock );
408
409        return thrd != 0p;
410}
Note: See TracBrowser for help on using the repository browser.