source: libcfa/src/concurrency/locks.cfa

Last change on this file was f9b68d6, checked in by caparsons <caparson@…>, 3 months ago

made spacing consistent

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