source: libcfa/src/concurrency/locks.cfa@ 5416b44

ADT ast-experimental pthread-emulation qualifiedEnum
Last change on this file since 5416b44 was c18bf9e, checked in by Thierry Delisle <tdelisle@…>, 3 years ago

Visibility concurrency

  • Property mode set to 100644
File size: 12.9 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
223//-----------------------------------------------------------------------------
224// Synchronization Locks
225forall(L & | is_blocking_lock(L)) {
226
227 //-----------------------------------------------------------------------------
228 // condition variable
229 void ?{}( condition_variable(L) & this ){
230 this.lock{};
231 this.blocked_threads{};
232 this.count = 0;
233 }
234
235 void ^?{}( condition_variable(L) & this ){ }
236
237 static void process_popped( condition_variable(L) & this, info_thread(L) & popped ) with( this ) {
238 if(&popped != 0p) {
239 popped.signalled = true;
240 count--;
241 if (popped.lock) {
242 // if lock passed call on_notify
243 on_notify(*popped.lock, popped.t);
244 } else {
245 // otherwise wake thread
246 unpark(popped.t);
247 }
248 }
249 }
250
251 bool notify_one( condition_variable(L) & this ) with( this ) {
252 lock( lock __cfaabi_dbg_ctx2 );
253 bool ret = ! blocked_threads`isEmpty;
254 process_popped(this, try_pop_front( blocked_threads ));
255 unlock( lock );
256 return ret;
257 }
258
259 bool notify_all( condition_variable(L) & this ) with(this) {
260 lock( lock __cfaabi_dbg_ctx2 );
261 bool ret = ! blocked_threads`isEmpty;
262 while( ! blocked_threads`isEmpty ) {
263 process_popped(this, try_pop_front( blocked_threads ));
264 }
265 unlock( lock );
266 return ret;
267 }
268
269 uintptr_t front( condition_variable(L) & this ) with(this) {
270 return blocked_threads`isEmpty ? NULL : blocked_threads`first.info;
271 }
272
273 bool empty( condition_variable(L) & this ) with(this) {
274 lock( lock __cfaabi_dbg_ctx2 );
275 bool ret = blocked_threads`isEmpty;
276 unlock( lock );
277 return ret;
278 }
279
280 int counter( condition_variable(L) & this ) with(this) { return count; }
281
282 static size_t queue_and_get_recursion( condition_variable(L) & this, info_thread(L) * i ) with(this) {
283 // add info_thread to waiting queue
284 insert_last( blocked_threads, *i );
285 count++;
286 size_t recursion_count = 0;
287 if (i->lock) {
288 // if lock was passed get recursion count to reset to after waking thread
289 recursion_count = on_wait( *i->lock );
290 }
291 return recursion_count;
292 }
293
294 // helper for wait()'s' with no timeout
295 static void queue_info_thread( condition_variable(L) & this, info_thread(L) & i ) with(this) {
296 lock( lock __cfaabi_dbg_ctx2 );
297 size_t recursion_count = queue_and_get_recursion(this, &i);
298 unlock( lock );
299
300 // blocks here
301 park( );
302
303 // resets recursion count here after waking
304 if (i.lock) on_wakeup(*i.lock, recursion_count);
305 }
306
307 #define WAIT( u, l ) \
308 info_thread( L ) i = { active_thread(), u, l }; \
309 queue_info_thread( this, i );
310
311 // helper for wait()'s' with a timeout
312 static void queue_info_thread_timeout( condition_variable(L) & this, info_thread(L) & info, Duration t, Alarm_Callback callback ) with(this) {
313 lock( lock __cfaabi_dbg_ctx2 );
314 size_t recursion_count = queue_and_get_recursion(this, &info);
315 alarm_node_wrap(L) node_wrap = { t, 0`s, callback, &this, &info };
316 register_self( &node_wrap.alarm_node );
317 unlock( lock );
318
319 // blocks here
320 park();
321
322 // unregisters alarm so it doesn't go off if this happens first
323 unregister_self( &node_wrap.alarm_node );
324
325 // resets recursion count here after waking
326 if (info.lock) on_wakeup(*info.lock, recursion_count);
327 }
328
329 #define WAIT_TIME( u, l, t ) \
330 info_thread( L ) i = { active_thread(), u, l }; \
331 queue_info_thread_timeout(this, i, t, alarm_node_wrap_cast ); \
332 return i.signalled;
333
334 void wait( condition_variable(L) & this ) with(this) { WAIT( 0, 0p ) }
335 void wait( condition_variable(L) & this, uintptr_t info ) with(this) { WAIT( info, 0p ) }
336 void wait( condition_variable(L) & this, L & l ) with(this) { WAIT( 0, &l ) }
337 void wait( condition_variable(L) & this, L & l, uintptr_t info ) with(this) { WAIT( info, &l ) }
338
339 bool wait( condition_variable(L) & this, Duration duration ) with(this) { WAIT_TIME( 0 , 0p , duration ) }
340 bool wait( condition_variable(L) & this, uintptr_t info, Duration duration ) with(this) { WAIT_TIME( info, 0p , duration ) }
341 bool wait( condition_variable(L) & this, L & l, Duration duration ) with(this) { WAIT_TIME( 0 , &l , duration ) }
342 bool wait( condition_variable(L) & this, L & l, uintptr_t info, Duration duration ) with(this) { WAIT_TIME( info, &l , duration ) }
343
344 //-----------------------------------------------------------------------------
345 // fast_cond_var
346 void ?{}( fast_cond_var(L) & this ){
347 this.blocked_threads{};
348 #ifdef __CFA_DEBUG__
349 this.lock_used = 0p;
350 #endif
351 }
352 void ^?{}( fast_cond_var(L) & this ){ }
353
354 bool notify_one( fast_cond_var(L) & this ) with(this) {
355 bool ret = ! blocked_threads`isEmpty;
356 if ( ret ) {
357 info_thread(L) & popped = try_pop_front( blocked_threads );
358 on_notify(*popped.lock, popped.t);
359 }
360 return ret;
361 }
362 bool notify_all( fast_cond_var(L) & this ) with(this) {
363 bool ret = ! blocked_threads`isEmpty;
364 while( ! blocked_threads`isEmpty ) {
365 info_thread(L) & popped = try_pop_front( blocked_threads );
366 on_notify(*popped.lock, popped.t);
367 }
368 return ret;
369 }
370
371 uintptr_t front( fast_cond_var(L) & this ) with(this) { return blocked_threads`isEmpty ? NULL : blocked_threads`first.info; }
372 bool empty ( fast_cond_var(L) & this ) with(this) { return blocked_threads`isEmpty; }
373
374 void wait( fast_cond_var(L) & this, L & l ) {
375 wait( this, l, 0 );
376 }
377
378 void wait( fast_cond_var(L) & this, L & l, uintptr_t info ) with(this) {
379 // brand cond lock with lock
380 #ifdef __CFA_DEBUG__
381 if ( lock_used == 0p ) lock_used = &l;
382 else { assert(lock_used == &l); }
383 #endif
384 info_thread( L ) i = { active_thread(), info, &l };
385 insert_last( blocked_threads, i );
386 size_t recursion_count = on_wait( *i.lock );
387 park( );
388 on_wakeup(*i.lock, recursion_count);
389 }
390}
391
392//-----------------------------------------------------------------------------
393// Semaphore
394void ?{}( semaphore & this, int count = 1 ) {
395 (this.lock){};
396 this.count = count;
397 (this.waiting){};
398}
399void ^?{}(semaphore & this) {}
400
401bool P(semaphore & this) with( this ){
402 lock( lock __cfaabi_dbg_ctx2 );
403 count -= 1;
404 if ( count < 0 ) {
405 // queue current task
406 append( waiting, active_thread() );
407
408 // atomically release spin lock and block
409 unlock( lock );
410 park();
411 return true;
412 }
413 else {
414 unlock( lock );
415 return false;
416 }
417}
418
419thread$ * V (semaphore & this, const bool doUnpark ) with( this ) {
420 thread$ * thrd = 0p;
421 lock( lock __cfaabi_dbg_ctx2 );
422 count += 1;
423 if ( count <= 0 ) {
424 // remove task at head of waiting list
425 thrd = pop_head( waiting );
426 }
427
428 unlock( lock );
429
430 // make new owner
431 if( doUnpark ) unpark( thrd );
432
433 return thrd;
434}
435
436bool V(semaphore & this) with( this ) {
437 thread$ * thrd = V(this, true);
438 return thrd != 0p;
439}
440
441bool V(semaphore & this, unsigned diff) with( this ) {
442 thread$ * thrd = 0p;
443 lock( lock __cfaabi_dbg_ctx2 );
444 int release = max(-count, (int)diff);
445 count += diff;
446 for(release) {
447 unpark( pop_head( waiting ) );
448 }
449
450 unlock( lock );
451
452 return thrd != 0p;
453}
Note: See TracBrowser for help on using the repository browser.