source: libcfa/src/concurrency/locks.hfa@ 5a46e09

ADT ast-experimental enum forall-pointer-decay jacob/cs343-translation new-ast-unique-expr pthread-emulation qualifiedEnum
Last change on this file since 5a46e09 was 5a46e09, checked in by caparsons <caparson@…>, 4 years ago

Added Martins SpinCondLock as linear_backoff_then_block lock
  • Property mode set to 100644
File size: 12.7 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 -- PUBLIC
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#pragma once
18
19#include <stdbool.h>
20#include <stdio.h>
21
22#include "bits/weakso_locks.hfa"
23#include "containers/queueLockFree.hfa"
24#include "containers/list.hfa"
25
26#include "thread.hfa"
27
28#include "time_t.hfa"
29#include "time.hfa"
30
31//-----------------------------------------------------------------------------
32// Semaphores
33
34// '0-nary' semaphore
35// Similar to a counting semaphore except the value of one is never reached
36// as a consequence, a V() that would bring the value to 1 *spins* until
37// a P consumes it
38struct Semaphore0nary {
39 __spinlock_t lock; // needed to protect
40 mpsc_queue($thread) queue;
41};
42
43static inline bool P(Semaphore0nary & this, $thread * thrd) {
44 /* paranoid */ verify(!thrd`next);
45 /* paranoid */ verify(!(&(*thrd)`next));
46
47 push(this.queue, thrd);
48 return true;
49}
50
51static inline bool P(Semaphore0nary & this) {
52 $thread * thrd = active_thread();
53 P(this, thrd);
54 park();
55 return true;
56}
57
58static inline $thread * V(Semaphore0nary & this, bool doUnpark = true) {
59 $thread * next;
60 lock(this.lock __cfaabi_dbg_ctx2);
61 for (;;) {
62 next = pop(this.queue);
63 if (next) break;
64 Pause();
65 }
66 unlock(this.lock);
67
68 if (doUnpark) unpark(next);
69 return next;
70}
71
72// Wrapper used on top of any sempahore to avoid potential locking
73struct BinaryBenaphore {
74 volatile ssize_t counter;
75};
76
77static inline {
78 void ?{}(BinaryBenaphore & this) { this.counter = 0; }
79 void ?{}(BinaryBenaphore & this, zero_t) { this.counter = 0; }
80 void ?{}(BinaryBenaphore & this, one_t ) { this.counter = 1; }
81
82 // returns true if no blocking needed
83 bool P(BinaryBenaphore & this) {
84 return __atomic_fetch_sub(&this.counter, 1, __ATOMIC_SEQ_CST) > 0;
85 }
86
87 bool tryP(BinaryBenaphore & this) {
88 ssize_t c = this.counter;
89 return (c >= 1) && __atomic_compare_exchange_n(&this.counter, &c, c-1, false, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED);
90 }
91
92 // returns true if notify needed
93 bool V(BinaryBenaphore & this) {
94 ssize_t c = 0;
95 for () {
96 if (__atomic_compare_exchange_n(&this.counter, &c, c+1, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
97 if (c == 0) return true;
98 /* paranoid */ verify(c < 0);
99 return false;
100 } else {
101 if (c == 1) return true;
102 /* paranoid */ verify(c < 1);
103 Pause();
104 }
105 }
106 }
107}
108
109// Binary Semaphore based on the BinaryBenaphore on top of the 0-nary Semaphore
110struct ThreadBenaphore {
111 BinaryBenaphore ben;
112 Semaphore0nary sem;
113};
114
115static inline void ?{}(ThreadBenaphore & this) {}
116static inline void ?{}(ThreadBenaphore & this, zero_t) { (this.ben){ 0 }; }
117static inline void ?{}(ThreadBenaphore & this, one_t ) { (this.ben){ 1 }; }
118
119static inline bool P(ThreadBenaphore & this) { return P(this.ben) ? false : P(this.sem); }
120static inline bool tryP(ThreadBenaphore & this) { return tryP(this.ben); }
121static inline bool P(ThreadBenaphore & this, bool wait) { return wait ? P(this) : tryP(this); }
122
123static inline $thread * V(ThreadBenaphore & this, bool doUnpark = true) {
124 if (V(this.ben)) return 0p;
125 return V(this.sem, doUnpark);
126}
127
128//-----------------------------------------------------------------------------
129// Semaphore
130struct semaphore {
131 __spinlock_t lock;
132 int count;
133 __queue_t($thread) waiting;
134};
135
136void ?{}(semaphore & this, int count = 1);
137void ^?{}(semaphore & this);
138bool P (semaphore & this);
139bool V (semaphore & this);
140bool V (semaphore & this, unsigned count);
141$thread * V (semaphore & this, bool );
142
143//----------
144struct single_acquisition_lock {
145 inline blocking_lock;
146};
147
148static inline void ?{}( single_acquisition_lock & this ) {((blocking_lock &)this){ false, false };}
149static inline void ^?{}( single_acquisition_lock & this ) {}
150static inline void lock ( single_acquisition_lock & this ) { lock ( (blocking_lock &)this ); }
151static inline bool try_lock ( single_acquisition_lock & this ) { return try_lock( (blocking_lock &)this ); }
152static inline void unlock ( single_acquisition_lock & this ) { unlock ( (blocking_lock &)this ); }
153static inline size_t on_wait ( single_acquisition_lock & this ) { return on_wait ( (blocking_lock &)this ); }
154static inline void on_wakeup( single_acquisition_lock & this, size_t v ) { on_wakeup ( (blocking_lock &)this, v ); }
155static inline void on_notify( single_acquisition_lock & this, struct $thread * t ) { on_notify( (blocking_lock &)this, t ); }
156
157//----------
158struct owner_lock {
159 inline blocking_lock;
160};
161
162static inline void ?{}( owner_lock & this ) {((blocking_lock &)this){ true, true };}
163static inline void ^?{}( owner_lock & this ) {}
164static inline void lock ( owner_lock & this ) { lock ( (blocking_lock &)this ); }
165static inline bool try_lock ( owner_lock & this ) { return try_lock( (blocking_lock &)this ); }
166static inline void unlock ( owner_lock & this ) { unlock ( (blocking_lock &)this ); }
167static inline size_t on_wait ( owner_lock & this ) { return on_wait ( (blocking_lock &)this ); }
168static inline void on_wakeup( owner_lock & this, size_t v ) { on_wakeup ( (blocking_lock &)this, v ); }
169static inline void on_notify( owner_lock & this, struct $thread * t ) { on_notify( (blocking_lock &)this, t ); }
170
171struct fast_lock {
172 $thread * volatile owner;
173 ThreadBenaphore sem;
174};
175
176static inline bool $try_lock(fast_lock & this, $thread * thrd) {
177 $thread * exp = 0p;
178 return __atomic_compare_exchange_n(&this.owner, &exp, thrd, false, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED);
179}
180
181static inline void lock( fast_lock & this ) __attribute__((artificial));
182static inline void lock( fast_lock & this ) {
183 $thread * thrd = active_thread();
184 /* paranoid */verify(thrd != this.owner);
185
186 for (;;) {
187 if ($try_lock(this, thrd)) return;
188 P(this.sem);
189 }
190}
191
192static inline bool try_lock( fast_lock & this ) __attribute__((artificial));
193static inline bool try_lock ( fast_lock & this ) {
194 $thread * thrd = active_thread();
195 /* paranoid */ verify(thrd != this.owner);
196 return $try_lock(this, thrd);
197}
198
199static inline $thread * unlock( fast_lock & this ) __attribute__((artificial));
200static inline $thread * unlock( fast_lock & this ) {
201 /* paranoid */ verify(active_thread() == this.owner);
202
203 // open 'owner' before unlocking anyone
204 // so new and unlocked threads don't park incorrectly.
205 // This may require additional fencing on ARM.
206 this.owner = 0p;
207
208 return V(this.sem);
209}
210
211static inline size_t on_wait( fast_lock & this ) { unlock(this); return 0; }
212static inline void on_wakeup( fast_lock & this, size_t ) { lock(this); }
213static inline void on_notify( fast_lock &, struct $thread * t ) { unpark(t); }
214
215struct mcs_node {
216 mcs_node * volatile next;
217 single_sem sem;
218};
219
220static inline void ?{}(mcs_node & this) { this.next = 0p; }
221
222static inline mcs_node * volatile & ?`next ( mcs_node * node ) {
223 return node->next;
224}
225
226struct mcs_lock {
227 mcs_queue(mcs_node) queue;
228};
229
230static inline void lock(mcs_lock & l, mcs_node & n) {
231 if(push(l.queue, &n))
232 wait(n.sem);
233}
234
235static inline void unlock(mcs_lock & l, mcs_node & n) {
236 mcs_node * next = advance(l.queue, &n);
237 if(next) post(next->sem);
238}
239
240struct linear_backoff_then_block_lock {
241 // Spin lock used for mutual exclusion
242 __spinlock_t spinlock;
243
244 // Current thread owning the lock
245 struct $thread * owner;
246
247 // List of blocked threads
248 dlist( $thread ) blocked_threads;
249
250 // Used for comparing and exchanging
251 volatile size_t lock_value;
252
253 // used for linear backoff spinning
254 int spin_start;
255 int spin_end;
256 int spin_count;
257
258 // after unsuccessful linear backoff yield this many times
259 int yield_count;
260};
261
262static inline void ?{}( linear_backoff_then_block_lock & this, int spin_start, int spin_end, int spin_count, int yield_count ) {
263 this.spinlock{};
264 this.blocked_threads{};
265 this.lock_value = 0;
266 this.spin_start = spin_start;
267 this.spin_end = spin_end;
268 this.spin_count = spin_count;
269 this.yield_count = yield_count;
270}
271static inline void ?{}( linear_backoff_then_block_lock & this ) { this{4, 1024, 16, 0}; }
272static inline void ^?{}( linear_backoff_then_block_lock & this ) {}
273
274static inline bool internal_try_lock(linear_backoff_then_block_lock & this, size_t & compare_val) with(this) {
275 if (__atomic_compare_exchange_n(&lock_value, &compare_val, 1, false, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED)) {
276 owner = active_thread();
277 return true;
278 }
279 return false;
280}
281
282static inline bool try_lock(linear_backoff_then_block_lock & this) { size_t compare_val = 0; return internal_try_lock(this, compare_val); }
283
284static inline bool try_lock_contention(linear_backoff_then_block_lock & this) with(this) {
285 if (__atomic_exchange_n(&lock_value, 2, __ATOMIC_ACQUIRE) == 0) {
286 owner = active_thread();
287 return true;
288 }
289 return false;
290}
291
292static inline bool block(linear_backoff_then_block_lock & this) with(this) {
293 lock( spinlock __cfaabi_dbg_ctx2 );
294 if (lock_value != 2) {
295 unlock( spinlock );
296 return true;
297 }
298 insert_last( blocked_threads, *active_thread() );
299 unlock( spinlock );
300 park( );
301 return true;
302}
303
304static inline bool lock(linear_backoff_then_block_lock & this) with(this) {
305 // if owner just return
306 if (active_thread() == owner) return true;
307 size_t compare_val = 0;
308 int spin = spin_start;
309 // linear backoff
310 for( ;; ) {
311 compare_val = 0;
312 if (internal_try_lock(this, compare_val)) return true;
313 if (2 == compare_val) break;
314 for (int i = 0; i < spin; i++) Pause();
315 if (spin >= spin_end) break;
316 spin += spin;
317 }
318
319 // linear backoff bounded by spin_count
320 spin = spin_start;
321 int spin_counter = 0;
322 int yield_counter = 0;
323 for ( ;; ) {
324 if(try_lock_contention(this)) return true;
325 if(spin_counter < spin_count) {
326 for (int i = 0; i < spin; i++) Pause();
327 if (spin < spin_end) spin += spin;
328 else spin_counter++;
329 } else if (yield_counter < yield_count) {
330 // after linear backoff yield yield_count times
331 yield_counter++;
332 yield();
333 } else { break; }
334 }
335
336 // block until signalled
337 while (block(this)) if(try_lock_contention(this)) return true;
338
339 // this should never be reached as block(this) always returns true
340 return false;
341}
342
343static inline void unlock(linear_backoff_then_block_lock & this) with(this) {
344 verify(lock_value > 0);
345 owner = 0p;
346 if (__atomic_exchange_n(&lock_value, 0, __ATOMIC_RELEASE) == 1) return;
347 lock( spinlock __cfaabi_dbg_ctx2 );
348 $thread * t = &try_pop_front( blocked_threads );
349 unlock( spinlock );
350 unpark( t );
351}
352
353
354void on_notify(linear_backoff_then_block_lock & this, struct $thread * t ) { unpark(t); }
355size_t on_wait(linear_backoff_then_block_lock & this) { unlock(this); return 0; }
356void on_wakeup(linear_backoff_then_block_lock & this, size_t recursion ) { lock(this); }
357
358//-----------------------------------------------------------------------------
359// is_blocking_lock
360trait is_blocking_lock(L & | sized(L)) {
361 // For synchronization locks to use when acquiring
362 void on_notify( L &, struct $thread * );
363
364 // For synchronization locks to use when releasing
365 size_t on_wait( L & );
366
367 // to set recursion count after getting signalled;
368 void on_wakeup( L &, size_t recursion );
369};
370
371//-----------------------------------------------------------------------------
372// // info_thread
373// // the info thread is a wrapper around a thread used
374// // to store extra data for use in the condition variable
375forall(L & | is_blocking_lock(L)) {
376 struct info_thread;
377
378 // // for use by sequence
379 // info_thread(L) *& Back( info_thread(L) * this );
380 // info_thread(L) *& Next( info_thread(L) * this );
381}
382
383//-----------------------------------------------------------------------------
384// Synchronization Locks
385forall(L & | is_blocking_lock(L)) {
386 struct condition_variable {
387 // Spin lock used for mutual exclusion
388 __spinlock_t lock;
389
390 // List of blocked threads
391 dlist( info_thread(L) ) blocked_threads;
392
393 // Count of current blocked threads
394 int count;
395 };
396
397
398 void ?{}( condition_variable(L) & this );
399 void ^?{}( condition_variable(L) & this );
400
401 bool notify_one( condition_variable(L) & this );
402 bool notify_all( condition_variable(L) & this );
403
404 uintptr_t front( condition_variable(L) & this );
405
406 bool empty ( condition_variable(L) & this );
407 int counter( condition_variable(L) & this );
408
409 void wait( condition_variable(L) & this );
410 void wait( condition_variable(L) & this, uintptr_t info );
411 bool wait( condition_variable(L) & this, Duration duration );
412 bool wait( condition_variable(L) & this, uintptr_t info, Duration duration );
413
414 void wait( condition_variable(L) & this, L & l );
415 void wait( condition_variable(L) & this, L & l, uintptr_t info );
416 bool wait( condition_variable(L) & this, L & l, Duration duration );
417 bool wait( condition_variable(L) & this, L & l, uintptr_t info, Duration duration );
418}
Note: See TracBrowser for help on using the repository browser.