sync #

Description:

sync provides cross platform handling of concurrency primitives.

fn channel_select(mut channels []&Channel, dir []Direction, mut objrefs []voidptr, timeout time.Duration) int

Wait timeout on any of channels[i] until one of them can push (is_push[i] = true) or pop (is_push[i] = false) object referenced by objrefs[i]. timeout = time.infinite means wait unlimited time. timeout <= 0 means return immediately if no transaction can be performed without waiting. return value: the index of the channel on which a transaction has taken place -1 if waiting for a transaction has exceeded timeout -2 if all channels are closed

fn new_channel[T](n u32) &Channel

fn new_many_times(times u64) &ManyTimes

new_many_times return a new ManyTimes struct.

fn new_mutex() &Mutex

new_mutex creates and initialises a new mutex instance on the heap, then returns a pointer to it.

fn new_once() &Once

new_once return a new Once struct.

fn new_rwmutex() &RwMutex

new_rwmutex creates a new read/write mutex instance on the heap, and returns a pointer to it.

fn new_semaphore() &Semaphore

new_semaphore creates a new initialised Semaphore instance on the heap, and returns a pointer to it. The initial counter value of the semaphore is 0.

fn new_semaphore_init(n u32) &Semaphore

new_semaphore_init creates a new initialised Semaphore instance on the heap, and returns a pointer to it. The n parameter can be used to set the initial counter value of the semaphore.

fn new_waitgroup() &WaitGroup

fn thread_id() u64

thread_id returns a unique identifier for the caller thread. All currently running threads in the same process, will have different thread identifiers.

Note: if a thread finishes, and another starts, the identifier of the old thread may be reused for the newly started thread. In other words, thread IDs are guaranteed to be unique only within a process. A thread ID may be reused after a terminated thread has been joined (with t.wait()), or when the thread has terminated.

enum Direction {
	pop
	push
}

struct C.atomic_uintptr_t {}

struct C.pthread_mutex_t {}

struct C.pthread_rwlock_t {}

struct C.pthread_rwlockattr_t {}

struct C.sem_t {}

struct Channel {
	ringbuf   &u8 = unsafe { nil } // queue for buffered channels
	statusbuf &u8 = unsafe { nil } // flags to synchronize write/read in ringbuf
	objsize   u32
mut: // atomic
	writesem           Semaphore // to wake thread that wanted to write, but buffer was full
	readsem            Semaphore // to wake thread that wanted to read, but buffer was empty
	writesem_im        Semaphore
	readsem_im         Semaphore
	write_adr          C.atomic_uintptr_t // if != NULL the next obj can be written here without wait
	read_adr           C.atomic_uintptr_t // if != NULL an obj can be read from here without wait
	adr_read           C.atomic_uintptr_t // used to identify origin of writesem
	adr_written        C.atomic_uintptr_t // used to identify origin of readsem
	write_free         u32 // for queue state
	read_avail         u32
	buf_elem_write_idx u32
	buf_elem_read_idx  u32
	// for select
	write_subscriber &Subscription = unsafe { nil }
	read_subscriber  &Subscription = unsafe { nil }
	write_sub_mtx    u16
	read_sub_mtx     u16
	closed           u16
pub:
	cap u32 // queue length in #objects
}

fn (ch &Channel) auto_str(typename string) string

fn (mut ch Channel) close()

fn (mut ch Channel) len() int

fn (mut ch Channel) closed() bool

fn (mut ch Channel) push(src voidptr)

fn (mut ch Channel) try_push(src voidptr) ChanState

fn (mut ch Channel) pop(dest voidptr) bool

fn (mut ch Channel) try_pop(dest voidptr) ChanState

struct ManyTimes {
mut:
	m RwMutex
pub:
	times u64 = 1
	count u64
}

fn (mut m ManyTimes) do(f fn ())

do execute the function only setting times.

struct Mutex {
	mutex C.pthread_mutex_t
}

[init_with=new_mutex] // TODO: implement support for this struct attribute, and disallow Mutex{} from outside the sync.new_mutex() function.

fn (mut m Mutex) @lock()

@lock locks the mutex instance (lock is a keyword). If the mutex was already locked, it will block, till it is unlocked.

fn (mut m Mutex) destroy()

destroy frees the resources associated with the mutex instance.

Note: the mutex itself is not freed.

fn (mut m Mutex) init()

init initialises the mutex. It should be called once before the mutex is used, since it creates the associated resources needed for the mutex to work properly.

fn (m &Mutex) str() string

str returns a string representation of the Mutex pointer

fn (mut m Mutex) unlock()

unlock unlocks the mutex instance. The mutex is released, and one of the other threads, that were blocked, because they called @lock can continue.

struct Once {
mut:
	m RwMutex
pub:
	count u64
}

fn (mut o Once) do(f fn ())

do executes the function f() only once

fn (mut o Once) do_with_param(f fn (voidptr), param voidptr)

do_with_param executes f(param) only once` This method can be used as a workaround for passing closures to once.do/1 on Windows (they are not implemented there yet) - just pass your data explicitly. i.e. instead of:

once.do(fn [mut o] () { o.add(5) })

... you can use:

once.do_with_param(fn (mut o One) { o.add(5) }, o)

struct RwMutex {
	mutex C.pthread_rwlock_t
}

fn (mut m RwMutex) @lock()

@lock locks the given RwMutex instance for writing. If the mutex was already locked, it will block, till it is unlocked, then it will try to get the lock, and if it can, it will return, otherwise it will continue waiting for the mutex to become unlocked.

Note: there may be several threads that are waiting for the same lock.

Note: RwMutex has separate read and write locks.

fn (mut m RwMutex) @rlock()

@rlock locks the given RwMutex instance for reading. If the mutex was already locked, it will block, and will try to get the lock, once the lock is released by another thread calling unlock. Once it succeds, it returns.

Note: there may be several threads that are waiting for the same lock.

Note: RwMutex has separate read and write locks.

fn (mut m RwMutex) destroy()

destroy frees the resources associated with the rwmutex instance.

Note: the mutex itself is not freed.

fn (mut m RwMutex) init()

init initialises the RwMutex instance. It should be called once before the rw mutex is used, since it creates the associated resources needed for the mutex to work properly.

fn (mut m RwMutex) runlock()

runlock unlocks the RwMutex instance, locked for reading.

Note: Windows SRWLocks have different function to unlocking. To have a common portable API, there are two methods for unlocking here as well, even though that they do the same on !windows platforms.

fn (m &RwMutex) str() string

str returns a string representation of the RwMutex pointer

fn (mut m RwMutex) unlock()

unlock unlocks the RwMutex instance, locked for writing.

Note: Windows SRWLocks have different function to unlocking. To have a common portable API, there are two methods for unlocking here as well, even though that they do the same on !windows platforms.

struct Semaphore {
	sem C.sem_t
}

fn (mut sem Semaphore) init(n u32)

init initialises the Semaphore instance with n as its initial counter value. It should be called once before the semaphore is used, since it creates the associated resources needed for the semaphore to work properly.

fn (mut sem Semaphore) post()

post increases/unlocks the counter of the semaphore by 1. If the resulting counter value is > 0, and if there is another thread waiting on the semaphore, the waiting thread will decrement the counter by 1 (locking the semaphore), and then will continue running. See also .wait() .

fn (mut sem Semaphore) wait()

wait will just decrement the semaphore count, if it was positive. It it was not positive, it will waits for the semaphore count to reach a positive number. When that happens, it will decrease the semaphore count (lock the semaphore), and will return. In effect, it allows you to block threads, until the semaphore, is posted by another thread. See also .post() .

fn (mut sem Semaphore) try_wait() bool

try_wait tries to decrease the semaphore count by 1, if it was positive. If it succeeds in that, it returns true, otherwise it returns false. try_wait should return as fast as possible so error handling is only done when debugging

fn (mut sem Semaphore) timed_wait(timeout time.Duration) bool

timed_wait is similar to .wait(), but it also accepts a timeout duration, thus it can return false early, if the timeout passed before the semaphore was posted.

fn (mut sem Semaphore) destroy()

destroy frees the resources associated with the Semaphore instance.

Note: the semaphore instance itself is not freed.

struct WaitGroup {
mut:
	task_count u32       // current task count - reading/writing should be atomic
	wait_count u32       // current wait count - reading/writing should be atomic
	sem        Semaphore // This blocks wait() until tast_countreleased by add()
}

WaitGroup Do not copy an instance of WaitGroup, use a ref instead.

usage: in main thread: wg := sync.new_waitgroup() wg.add(nr_jobs)before starting jobs withgo ... wg.wait()` to wait for all jobs to have finished

in each parallel job: wg.done() when finished

[init_with=new_waitgroup] // TODO: implement support for init_with struct attribute, and disallow WaitGroup{} from outside the sync.new_waitgroup() function.

fn (mut wg WaitGroup) init()

fn (mut wg WaitGroup) add(delta int)

add increments (+ve delta) or decrements (-ve delta) task count by delta and unblocks any wait() calls if task count becomes zero. add panics if task count drops below zero.

fn (mut wg WaitGroup) done()

done is a convenience fn for add(-1)

fn (mut wg WaitGroup) wait()

wait blocks until all tasks are done (task count becomes zero)