source: libcfa/src/concurrency/locks.cfa@ f835806

ADT ast-experimental pthread-emulation qualifiedEnum
Last change on this file since f835806 was 7f958c4, checked in by caparsons <caparson@…>, 3 years ago

added fast lock/cond var

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