chronos/asyncsync

Search:
Group by:
Source   Edit  

This module implements some core synchronization primitives.

Types

AsyncEvent = ref object of RootRef

A primitive event object.

An event manages a flag that can be set to true with the fire() procedure and reset to false with the clear() procedure. The wait() coroutine blocks until the flag is false.

If more than one coroutine blocked in wait() waiting for event state to be signaled, when event get fired, then all coroutines continue proceeds in order, they have entered waiting state.

Source   Edit   
AsyncEventQueue[T] = ref object of RootObj
Source   Edit   
AsyncLock = ref object of RootRef

A primitive lock is a synchronization primitive that is not owned by a particular coroutine when locked. A primitive lock is in one of two states, locked or unlocked.

When more than one coroutine is blocked in acquire() waiting for the state to turn to unlocked, only one coroutine proceeds when a release() call resets the state to unlocked; first coroutine which is blocked in acquire() is being processed.

Source   Edit   
AsyncLockError = object of AsyncError
AsyncLock is either locked or unlocked. Source   Edit   
AsyncQueue[T] = ref object of RootRef

A queue, useful for coordinating producer and consumer coroutines.

If maxsize is less than or equal to zero, the queue size is infinite. If it is an integer greater than 0, then "await put()" will block when the queue reaches maxsize, until an item is removed by "await get()".

Source   Edit   
AsyncQueueEmptyError = object of AsyncError
AsyncQueue is empty. Source   Edit   
AsyncQueueFullError = object of AsyncError
AsyncQueue is full. Source   Edit   
AsyncSemaphore = ref object of RootObj

A semaphore manages an internal number of available slots which is decremented by each acquire() call and incremented by each release() call. The available slots can never go below zero; when acquire() finds that it is zero, it blocks, waiting until some other task calls release().

The size argument gives the initial value for the available slots counter; it defaults to 1. If the value given is less than 1, AssertionDefect is raised.

Source   Edit   
EventQueueKey = distinct uint64
Source   Edit   
EventQueueReader = object
Source   Edit  

Procs

proc `$`[T](aq: AsyncQueue[T]): string {....raises: [].}
Turn an async queue aq into its string representation. Source   Edit   
proc `[]`[T](aq: AsyncQueue[T]; i: BackwardsIndex): T {.inline, ...raises: [].}
Access the i-th element of aq by order from first to last. aq[0] is the first element, aq[^1] is the last element. Source   Edit   
proc `[]`[T](aq: AsyncQueue[T]; i: Natural): T {.inline, ...raises: [].}
Access the i-th element of aq by order from first to last. aq[0] is the first element, aq[^1] is the last element. Source   Edit   
proc `[]=`[T](aq: AsyncQueue[T]; i: BackwardsIndex; item: T) {.inline,
    ...raises: [].}
Change the i-th element of aq. Source   Edit   
proc `[]=`[T](aq: AsyncQueue[T]; i: Natural; item: T) {.inline, ...raises: [].}
Change the i-th element of aq. Source   Edit   
proc acquire(lock: AsyncLock): InternalRaisesFuture[void, (CancelledError,)] {.
    ...stackTrace: false, raises: [], gcsafe, raises: [], tags: [RootEffect],
    forbids: [].}

Acquire a lock lock.

This procedure blocks until the lock lock is unlocked, then sets it to locked and returns.

Source   Edit   
proc acquire(s: AsyncSemaphore): InternalRaisesFuture[void, (CancelledError,)] {.
    ...stackTrace: false, raises: [], gcsafe, raises: [], tags: [], forbids: [].}
Source   Edit   
proc addFirst[T](aq: AsyncQueue[T]; item: T): InternalRaisesFuture[void,
    (CancelledError,)] {....stackTrace: false, raises: [], gcsafe, raises: [].}
Put an item to the beginning of the queue aq. If the queue is full, wait until a free slot is available before adding item. Source   Edit   
proc addFirstNoWait[T](aq: AsyncQueue[T]; item: T) {.
    ...raises: [AsyncQueueFullError], raises: [].}

Put an item item to the beginning of the queue aq immediately.

If queue aq is full, then AsyncQueueFullError exception raised.

Source   Edit   
proc addLast[T](aq: AsyncQueue[T]; item: T): InternalRaisesFuture[void,
    (CancelledError,)] {....stackTrace: false, raises: [], gcsafe, raises: [].}
Put an item to the end of the queue aq. If the queue is full, wait until a free slot is available before adding item. Source   Edit   
proc addLastNoWait[T](aq: AsyncQueue[T]; item: T) {.
    ...raises: [AsyncQueueFullError], raises: [].}

Put an item item at the end of the queue aq immediately.

If queue aq is full, then AsyncQueueFullError exception raised.

Source   Edit   
proc availableSlots(s: AsyncSemaphore): int {....raises: [], tags: [], forbids: [].}
Source   Edit   
proc clear(event: AsyncEvent) {....raises: [], tags: [], forbids: [].}
Reset the internal flag of event to false. Subsequently, tasks calling wait() will block until fire() is called to set the internal flag to true again. Source   Edit   
proc clear[T](aq: AsyncQueue[T]) {.inline, ...raises: [].}
Clears all elements of queue aq. Source   Edit   
proc close(ab: AsyncEventQueue) {....raises: [], raises: [].}
Source   Edit   
proc closeWait(ab: AsyncEventQueue): InternalRaisesFuture[void, void] {.
    ...stackTrace: false, raises: [], gcsafe, raises: [].}
Source   Edit   
proc contains[T](aq: AsyncQueue[T]; item: T): bool {.inline, ...raises: [].}
Return true if item is in aq or false if not found. Usually used via the in operator. Source   Edit   
proc emit[T](ab: AsyncEventQueue[T]; data: T) {....raises: [].}
Source   Edit   
proc empty[T](aq: AsyncQueue[T]): bool {.inline, ...raises: [].}
Return true if the queue is empty, false otherwise. Source   Edit   
proc fire(event: AsyncEvent) {....raises: [], tags: [], forbids: [].}
Set the internal flag of event to true. All tasks waiting for it to become true are awakened. Task that call wait() once the flag is true will not block at all. Source   Edit   
proc full[T](aq: AsyncQueue[T]): bool {.inline, ...raises: [].}

Return true if there are maxsize items in the queue.

Note: If the aq was initialized with maxsize = 0 (default), then full() is never true.

Source   Edit   
proc get[T](aq: AsyncQueue[T]): InternalRaisesFuture[T, (CancelledError,)] {.
    ...stackTrace: false, raises: [], gcsafe, raises: [].}
Source   Edit   
proc getNoWait[T](aq: AsyncQueue[T]): T {....raises: [AsyncQueueEmptyError],
    raises: [].}
Alias of popFirstNoWait(). Source   Edit   
proc isSet(event: AsyncEvent): bool {....raises: [], tags: [], forbids: [].}
Return true if and only if the internal flag of event is true. Source   Edit   
proc len(ab: AsyncEventQueue): int {....raises: [], raises: [].}
Source   Edit   
proc len[T](aq: AsyncQueue[T]): int {.inline, ...raises: [].}
Return the number of elements in aq. Source   Edit   
proc locked(lock: AsyncLock): bool {....raises: [], tags: [], forbids: [].}
Return true if the lock lock is acquired, false otherwise. Source   Edit   
proc newAsyncEvent(): AsyncEvent {....raises: [], tags: [], forbids: [].}

Creates new asyncronous event AsyncEvent.

An event manages a flag that can be set to true with the fire() procedure and reset to false with the clear() procedure. The wait() procedure blocks until the flag is true. The flag is initially false.

Source   Edit   
proc newAsyncEventQueue[T](limitSize = 0): AsyncEventQueue[T] {....raises: [],
    raises: [].}

Creates new AsyncEventBus maximum size of limitSize (default is 0 which means that there no limits).

When number of events emitted exceeds limitSize - emit() procedure will discard new events, consumers which has number of pending events more than limitSize will get AsyncEventQueueFullError error.

Source   Edit   
proc newAsyncLock(): AsyncLock {....raises: [], tags: [], forbids: [].}

Creates new asynchronous lock AsyncLock.

Lock is created in the unlocked state. When the state is unlocked, acquire() changes the state to locked and returns immediately. When the state is locked, acquire() blocks until a call to release() in another coroutine changes it to unlocked.

The release() procedure changes the state to unlocked and returns immediately.

Source   Edit   
proc newAsyncQueue[T](maxsize: int = 0): AsyncQueue[T] {....raises: [].}
Creates a new asynchronous queue AsyncQueue. Source   Edit   
proc newAsyncSemaphore(size: int = 1): AsyncSemaphore {....raises: [], tags: [],
    forbids: [].}
Creates a new asynchronous bounded semaphore AsyncSemaphore with internal available slots set to size. Source   Edit   
proc popFirst[T](aq: AsyncQueue[T]): InternalRaisesFuture[T, (CancelledError,)] {.
    ...stackTrace: false, raises: [], gcsafe, raises: [].}
Remove and return an item from the beginning of the queue aq. If the queue is empty, wait until an item is available. Source   Edit   
proc popFirstNoWait[T](aq: AsyncQueue[T]): T {....raises: [AsyncQueueEmptyError],
    raises: [].}

Get an item from the beginning of the queue aq immediately.

If queue aq is empty, then AsyncQueueEmptyError exception raised.

Source   Edit   
proc popLast[T](aq: AsyncQueue[T]): InternalRaisesFuture[T, (CancelledError,)] {.
    ...stackTrace: false, raises: [], gcsafe, raises: [].}
Remove and return an item from the end of the queue aq. If the queue is empty, wait until an item is available. Source   Edit   
proc popLastNoWait[T](aq: AsyncQueue[T]): T {....raises: [AsyncQueueEmptyError],
    raises: [].}

Get an item from the end of the queue aq immediately.

If queue aq is empty, then AsyncQueueEmptyError exception raised.

Source   Edit   
proc put[T](aq: AsyncQueue[T]; item: T): InternalRaisesFuture[void,
    (CancelledError,)] {....stackTrace: false, raises: [], gcsafe, raises: [].}
Source   Edit   
proc putNoWait[T](aq: AsyncQueue[T]; item: T) {....raises: [AsyncQueueFullError],
    raises: [].}
Alias of addLastNoWait(). Source   Edit   
proc register(ab: AsyncEventQueue): EventQueueKey {....raises: [], raises: [].}
Source   Edit   
proc release(lock: AsyncLock) {....raises: [AsyncLockError], raises: [], tags: [],
                                forbids: [].}

Release a lock lock.

When the lock is locked, reset it to unlocked, and return. If any other coroutines are blocked waiting for the lock to become unlocked, allow exactly one of them to proceed.

Source   Edit   
proc release(s: AsyncSemaphore) {....raises: [AsyncSemaphoreError], raises: [],
                                  tags: [], forbids: [].}
Release a resource from the semaphore, by picking the first future from waiters queue and completing it and incrementing the internal resource count. Source   Edit   
proc size[T](aq: AsyncQueue[T]): int {.inline, ...raises: [].}
Return the maximum number of elements in aq. Source   Edit   
proc tryAcquire(s: AsyncSemaphore): bool {....raises: [], tags: [], forbids: [].}
Attempts to acquire a resource, if successful returns true, otherwise false. Source   Edit   
proc unregister(ab: AsyncEventQueue; key: EventQueueKey) {....raises: [],
    raises: [].}
Source   Edit   
proc wait(event: AsyncEvent): InternalRaisesFuture[void, (CancelledError,)] {.
    ...stackTrace: false, raises: [], gcsafe, raises: [], tags: [], forbids: [].}
Source   Edit   
proc waitEvents[T](ab: AsyncEventQueue[T]; key: EventQueueKey; eventsCount = -1): InternalRaisesFuture[
    seq[T], (AsyncEventQueueFullError, CancelledError)] {....stackTrace: false,
    raises: [], gcsafe, raises: [].}
Wait for events Source   Edit  

Iterators

iterator items[T](aq: AsyncQueue[T]): T {.inline, ...raises: [].}
Yield every element of aq. Source   Edit   
iterator mitems[T](aq: AsyncQueue[T]): var T {.inline, ...raises: [].}
Yield every element of aq. Source   Edit   
iterator pairs[T](aq: AsyncQueue[T]): tuple[key: int, val: T] {.inline,
    ...raises: [].}
Yield every (position, value) of aq. Source   Edit  

Templates

template readerOverflow(ab: AsyncEventQueue; reader: EventQueueReader): bool {..}
Source   Edit  

Exports

NestedPoll, closeSocket, ENOSR, EHOSTUNREACH, EHOSTDOWN, ECONNRESET, EDQUOT, fail, ECANCELED, EMLINK, milliseconds, asyncTimer, +=, $, Second, internalRaiseIfError, getSrcLocation, FutureError, fromNow, weeks, ESPIPE, withTimeout, LocationKind, id, <=, FutureBase, FutureCompletedError, ENOEXEC, ESHUTDOWN, fail, EREMOTEIO, Day, waitFor, complete, internalInitFutureBase, EPROTONOSUPPORT, cancelAndWait, addTimer, Moment, ECOMM, EPROTOTYPE, AsyncExceptionError, FuturePendingError, milliseconds, ENOENT, millis, removeTimer, or, FutureFlag, ENOTCONN, EBUSY, init, ENOTUNIQ, <=, days, +=, race, ==, cancelSoon, AsyncTimeoutError, setThreadDispatcher, EEXIST, ENOLINK, newInternalRaisesFuture, idleAsync, internalRaiseIfError, SomeIntegerI64, ENOMEM, ENOKEY, +, removeWriter2, ENOMSG, EKEYEXPIRED, Week, cancelSoon, awaitne, EAFNOSUPPORT, EWOULDBLOCK, [], EREMOTE, finished, AsyncCallback, EHWPOISON, epochSeconds, newDispatcher, EPERM, microseconds, internalFail, weeks, Finished, ENOTSOCK, cancelCallback=, ETOOMANYREFS, $, EISCONN, callSoon, ESOCKTNOSUPPORT, setGlobalDispatcher, EXDEV, EBADF, EKEYREVOKED, allFinished, TrackerBase, hours, InternalAsyncCallback, EBFONT, ENETDOWN, EACCES, ELOOP, InfiniteDuration, ETIMEDOUT, EINVAL, value, EBADFD, Future, completed, PDispatcher, ESRCH, EL2NSYNC, Finished, low, await, ESTRPIPE, -=, EALREADY, getAsyncTimestamp, EXFULL, Hour, EIDRM, removeReader2, secs, getThreadDispatcher, futureContinue, isZero, ENOANO, EADV, CallbackFunc, contains, cancelAndWait, $, ENFILE, waitFor, high, Microsecond, ENOPKG, ELIBBAD, EOWNERDEAD, handle, ERANGE, done, ENONET, completed, callSoon, toString, cancelSoon, EMSGSIZE, EBADMSG, EILSEQ, ENOPROTOOPT, EDESTADDRREQ, allFutures, -, complete, FutureState, EBADR, nanos, location, FutureDefect, EDOTDOT, one, EKEYREJECTED, <, race, error, one, CancelledError, nanoseconds, EISDIR, sleepAsync, Nanosecond, EOVERFLOW, FutureSeq, cancelAndWait, micros, -, ESRMNT, ENOTEMPTY, EPROTO, TimerCallback, microseconds, error, epochNanoSeconds, ENOTRECOVERABLE, wait, EDOM, value, hours, AsyncError, EBADRQC, ENODEV, FutureStr, addTimer, ENOBUFS, ECHRNG, ENOTDIR, EUNATCH, cancelAndSchedule, failed, nanoseconds, ERESTART, *, ETXTBSY, newFutureStr, EMFILE, LocFinishIndex, withTimeout, addTimer, race, EUSERS, ENOTBLK, Minute, DispatcherHandle, EISNAM, ELIBACC, ENAVAIL, init, race, EUCLEAN, ENOTTY, +, cancelSoon, completed, minutes, cancel, FutureFlags, getGlobalDispatcher, ELIBEXEC, stepsAsync, micros, wait, cancelAndWait, EINPROGRESS, EPFNOSUPPORT, E2BIG, -=, ENOMEDIUM, secs, read, waitFor, EL3HLT, unregisterAndCloseFd, EBADSLT, InternalRaisesFuture, -, <, ENETRESET, fastEpochTime, ENXIO, join, EPIPE, ThreadCallbackFunc, allFutures, race, wait, now, read, asyncSpawn, removeCallback, EAGAIN, raiseOsDefect, EADDRNOTAVAIL, read, removeCallback, TrackerCounter, async, addCallback, init, async, ZeroDuration, EREMCHG, EFAULT, addCallback, seconds, noCancel, low, AsyncFD, ECHILD, Raising, init, ECONNABORTED, state, wait, removeTimer, LocCreateIndex, addReader2, ETIME, ENOLCK, EBADE, EDEADLK, asyncCheck, Duration, ELNRNG, ENOCSI, millis, EMEDIUMTYPE, readError, sleepAsync, isInfinite, callback=, ENETUNREACH, InternalFutureBase, ENOSPC, ENOSYS, ELIBSCN, EIO, EROFS, seconds, ELIBMAX, SrcLoc, ==, ==, div, init, EL2HLT, EOPNOTSUPP, tryCancel, poll, flags, clearTimer, register2, LocCompleteIndex, days, ENOSTR, EADDRINUSE, one, ENAMETOOLONG, EMULTIHOP, ESTALE, await, ENODATA, EFBIG, +, minutes, ERFKILL, cancelled, toException, or, EDEADLOCK, allFutures, MaxEventsCount, wait, read, raiseAsDefect, Millisecond, setTimer, unregister2, failed, waitFor, newFutureSeq, callback=, orImpl, ECONNREFUSED, high, ENOTNAM, nanos, addWriter2, removeTimer, EINTR, join, closeHandle, EL3RST, readError, *, newFuture

Made with Nim. Generated: 2026-09-02 18:01:36 UTC