外观
concurrent
原文: https://finproducts.atlassian.net/wiki/spaces/JIDD3/pages/34793488393/FIN+5.3.1+concurrent 中文整理基于 FIN 5.3.1
导读： 并发编程工具
以下为 API 参考，保留英文以便对照源码；目录已列出（英文标题便于检索）；概述已尽量中译。
目录（导读）
- mixins
- classes
- enums
- errs
- Slot Details
- locals
- make
- makeCoalescing
- pool
- receive
- send
- sendLater
- sendWhenComplete
- sleep
- actorLocalsKey
- a
- b
- c
- d
- e
- equals
- hash
- id
- make0
- make1
- make2
- make3
- make4
- make5
- toStr
- isDone
- isStopped
- join
- kill
- maxQueue
- maxThreads
- name
- stop
- compareAndSet
- getAndSet
- val
- add
- addAndGet
- decrement
- decrementAndGet
- getAndAdd
- getAndDecrement
- getAndIncrement
- increment
- incrementAndGet
- clear
- containsKey
- each
- eachWhile
- get
- getOrAdd
- isEmpty
- keys
- remove
- set
- setAll
- size
- vals
- cancel
- complete
- completeErr
- makeCompletable
- status
- waitFor
- waitForAll
- cancelled
- err
- fromStr
- isCancelled
- isComplete
- isErr
- isOk
- isPending
- ok
- pending
- lock
- makeReentrant
- tryLock
- unlock
API 正文（英文）
enums
| FutureStatus | State of a Future's asynchronous computation |
errs
| QueueOverflowErr | QueueOverflowErr is raised by a Future for messages sent to actor that has exceeded the max queue size. |
Actor
@Js const class Actor : Obj
Actor is a worker who processes messages asynchronously.
Constructors
| make | Create an actor whose execution is controlled by the given ActorPool. |
| makeCoalescing | Create an actor with a coalescing message loop. |
Methods
| locals | Return the map of "global" variables visibile only to the current actor (similar to how thread locals work in Java). |
| pool | The pool used to control execution of this actor. |
| receive | The receive behavior for this actor is handled by overriding this method or by passing a function to the constructor. |
| send | Asynchronously send a message to this actor for processing. |
| sendLater | Schedule a message for delivery after the specified period of duration has elapsed. |
| sendWhenComplete | Schedule a message for delivery after the given future has completed. |
| sleep | Put the currently executing actor thread to sleep for the specified period. |
Slot Details
locals
static Str:Obj? locals()
Return the map of "global" variables visibile only to the current actor (similar to how thread locals work in Java). These variables are keyed by a string name - by convention use a dotted notation beginning with your pod name to avoid naming collisions.
make
new make(ActorPool pool, |Obj?->Obj?|? receive := null)
Create an actor whose execution is controlled by the given ActorPool. If receive is non-null, then it is used to process messages sent to this actor. If receive is specified then it must be an immutable function (it cannot capture state from the calling thread), otherwise NotImmutableErr is thrown. If receive is null, then you must subclass Actor and override the receive method.
makeCoalescing
new makeCoalescing(ActorPool pool, |Obj?->Obj?|? toKey, |Obj?,Obj?->Obj?|? coalesce, |Obj?->Obj?|? receive := null)
Create an actor with a coalescing message loop. This constructor follows the same semantics as make, but has the ability to coalesce the messages pending in the thread's message queue. Coalesced messages are merged into a single pending message with a shared Future.
The toKey function is used to derive a key for each message, or if null then the message itself is used as the key. If the toKey function returns null, then the message is not considered for coalescing. Internally messages are indexed by key for efficient coalescing.
If an incoming message has the same key as a pending message in the queue, then the coalesce function is called to coalesce the messages into a new merged message. If coalesce function itself is null, then we use the incoming message (last one wins). The coalesced message occupies the same position in the queue as the original and reuses the original message's Future instance.
Both the toKey and coalesce functions are called while holding an internal lock on the queue. So the functions must be efficient and never attempt to interact with other actors.
pool
ActorPool pool()
The pool used to control execution of this actor.
receive
protected virtual Obj? receive(Obj? msg)
The receive behavior for this actor is handled by overriding this method or by passing a function to the constructor. Return the result made available by the Future. If an exception is raised by this method, then it is raised by Future.get.
send
Future send(Obj? msg)
Asynchronously send a message to this actor for processing. If msg is not immutable, then NotImmutableErr is thrown. Throw Err if this actor's pool has been stopped. Return a future which may be used to obtain the result once it the actor has processed the message. If the message is coalesced then this method returns the original message's future reference. Also see sendLater and sendWhenComplete.
sendLater
Future sendLater(Duration d, Obj? msg)
Schedule a message for delivery after the specified period of duration has elapsed. Once the period has elapsed the message is appended to the end of this actor's queue. Accuracy of scheduling is dependent on thread coordination and pending messages in the queue. Scheduled messages are not guaranteed to be processed if the actor's pool is stopped. Scheduled messages are never coalesced. Also see send and sendWhenComplete.
sendWhenComplete
Future sendWhenComplete(Future f, Obj? msg)
Schedule a message for delivery after the given future has completed. Completion may be due to the future returning a result, throwing an exception, or cancellation. Send-when-complete messages are never coalesced. The given future must be an actor based future. Also see send and sendLater.
sleep
static Void sleep(Duration duration)
Put the currently executing actor thread to sleep for the specified period. If the thread is interrupted for any reason while sleeping, then InterruptedErr is thrown.
ActorContext
@Js mixin ActorContext
ActorContext is standardized thread local state stored in by "cx" key
Fields
| actorLocalsKey | Actor.locals key "cx" for context state |
Slot Details
actorLocalsKey
const static Str actorLocalsKey := "cx"
Actor.locals key "cx" for context state
ActorMsg
@Js const class ActorMsg : Obj
ActorMsg provides simple immutable tuple to use for actor messages.
Constructors
| make0 | Constructor with zero arguments |
| make1 | Constructor with one argument |
| make2 | Constructor with two arguments |
| make3 | Constructor with three arguments |
| make4 | Constructor with four arguments |
| make5 | Constructor with five arguments |
Fields
| a | Argument a |
| b | Argument b |
| c | Argument c |
| d | Argument d |
| e | Argument e |
| id | Message identifier key, typically string or enum |
Methods
| equals | Equality is based on id and arguments |
| hash | Hash is based on id and arguments |
| toStr | Return debug string representation |
Slot Details
a
const Obj? a
Argument a
b
const Obj? b
Argument b
c
const Obj? c
Argument c
d
const Obj? d
Argument d
e
const Obj? e
Argument e
equals
virtual override Bool equals(Obj? that)
Equality is based on id and arguments
hash
virtual override Int hash()
Hash is based on id and arguments
id
const Obj id
Message identifier key, typically string or enum
make0
new make0(Obj id)
Constructor with zero arguments
make1
new make1(Obj id, Obj? a)
Constructor with one argument
make2
new make2(Obj id, Obj? a, Obj? b)
Constructor with two arguments
make3
new make3(Obj id, Obj? a, Obj? b, Obj? c)
Constructor with three arguments
make4
new make4(Obj id, Obj? a, Obj? b, Obj? c, Obj? d)
Constructor with four arguments
make5
new make5(Obj id, Obj? a, Obj? b, Obj? c, Obj? d, Obj? e)
Constructor with five arguments
toStr
virtual override Str toStr()
Return debug string representation
ActorPool
@Js const class ActorPool : Obj
Controller for a group of actors which manages their execution using pooled thread resources.
See Actors
Constructors
| make | It-block constructor |
Fields
| maxQueue | Max number of messages that may be queued by actors in this pool. |
| maxThreads | Max number of threads which are used by this pool for concurrent actor execution. |
| name | Name to use for the pool and associated threads. |
Methods
| isDone | Return true if this pool has been stopped or killed and all its actors have completed processing. |
| isStopped | Return true if this pool has been stopped or killed. |
| join | Wait for this pool's actors to fully terminate or until the given timeout occurs. |
| kill | Perform an unorderly shutdown. |
| stop | Perform an orderly shutdown. |
Slot Details
isDone
Bool isDone()
Return true if this pool has been stopped or killed and all its actors have completed processing. If this pool was stopped then true indicates that all pending messages in the queues before the stop have been fully processed. If this pool was killed, then this method returns true once all actors have exited their thread. See join to block until done.
isStopped
Bool isStopped()
Return true if this pool has been stopped or killed. Once a a pool is stopped, new messages may not be delivered to any of its actors. A stopped pool is not necessarily done until all its actors have finished processing. Also see isDone and join.
join
This join(Duration? timeout := null)
Wait for this pool's actors to fully terminate or until the given timeout occurs. A null timeout blocks forever. If this method times out, then TimeoutErr is thrown. Throw Err if the pool is not stopped. Return this.
kill
This kill()
Perform an unorderly shutdown. Any pending messages which have not started processing are cancelled. Actors which are currently processing a message will be interrupted. See stop to perform an orderly shutdown. If the pool has already been killed, then do nothing.
make
new make(|This|? f := null)
It-block constructor
maxQueue
const Int maxQueue := 100000000
Max number of messages that may be queued by actors in this pool. Once this limit is reached, any messages sent are immediately rejected and their future will raise QueueOverflowErr. This limit only applies to messages sent via send and sendWhenDone. No limit checking is applied to sendLater.
maxThreads
const Int maxThreads := 100
Max number of threads which are used by this pool for concurrent actor execution. This value must be at least one or greater.
name
const Str name := "ActorPool"
Name to use for the pool and associated threads.
stop
This stop()
Perform an orderly shutdown. Once stopped, no new messages may be sent to this pool's actors. However, any pending messages will be processed. Note that scheduled messages are not guaranteed to be processed, only those delivered with Actor.send.
Use join to wait for all actors to complete their message queue. To perform an immediate shutdown use kill. If the pool has already been stopped, then do nothing. Return this.
AtomicBool
@Js const class AtomicBool : Obj
AtomicBool is used to manage a boolean variable shared between actor/threads with atomic updates.
Constructors
| make | Construct with initial value |
Fields
| val | The current boolean value |
Methods
| compareAndSet | Atomically set the value to update if current value is equivalent to the expect value. |
| getAndSet | Atomically set the value and return the previous value. |
| toStr | Return val.toStr |
Slot Details
compareAndSet
Bool compareAndSet(Bool expect, Bool update)
Atomically set the value to update if current value is equivalent to the expect value. Return true if updated, or false if current value was not equal to the expected value.
getAndSet
Bool getAndSet(Bool val)
Atomically set the value and return the previous value.
make
new make(Bool val := false)
Construct with initial value
toStr
virtual override Str toStr()
Return val.toStr
val
Bool val
The current boolean value
AtomicInt
@Js const class AtomicInt : Obj
AtomicInt is used to manage an integer variable shared between actor/threads with atomic updates.
Constructors
| make | Construct with initial value |
Fields
| val | The current integer value |
Methods
| add | Atomically add the given value to the current value |
| addAndGet | Atomically add the given value to the current value and return the updated value. |
| compareAndSet | Atomically set the value to update if current value is equivalent to the expect value. |
| decrement | Atomically decrement the value by one |
| decrementAndGet | Atomically decrement the current value by one and return the updated value. |
| getAndAdd | Atomically add the given value to the current value and return the previous value. |
| getAndDecrement | Atomically decrement the current value by one and return the previous value. |
| getAndIncrement | Atomically increment the current value by one and return the previous value. |
| getAndSet | Atomically set the value and return the previous value. |
| increment | Atomically increment the value by one |
| incrementAndGet | Atomically increment the current value by one and return the updated value. |
| toStr | Return val.toStr |
Slot Details
add
Void add(Int delta)
Atomically add the given value to the current value
addAndGet
Int addAndGet(Int delta)
Atomically add the given value to the current value and return the updated value.
compareAndSet
Bool compareAndSet(Int expect, Int update)
Atomically set the value to update if current value is equivalent to the expect value. Return true if updated, or false if current value was not equal to the expected value.
decrement
Void decrement()
Atomically decrement the value by one
decrementAndGet
Int decrementAndGet()
Atomically decrement the current value by one and return the updated value.
getAndAdd
Int getAndAdd(Int delta)
Atomically add the given value to the current value and return the previous value.
getAndDecrement
Int getAndDecrement()
Atomically decrement the current value by one and return the previous value.
getAndIncrement
Int getAndIncrement()
Atomically increment the current value by one and return the previous value.
getAndSet
Int getAndSet(Int val)
Atomically set the value and return the previous value.
increment
Void increment()
Atomically increment the value by one
incrementAndGet
Int incrementAndGet()
Atomically increment the current value by one and return the updated value.
make
new make(Int val := 0)
Construct with initial value
toStr
virtual override Str toStr()
Return val.toStr
val
Int val
The current integer value
AtomicRef
@Js const class AtomicRef : Obj
AtomicRef is used to manage a object reference shared between actor/threads with atomic updates. Only immutable objects may be shared.
Constructors
| make | Construct with initial value. |
Fields
| val | The current value. |
Methods
| compareAndSet | Atomically set the value to update if current value is equivalent to the expect value compared using === operator. |
| getAndSet | Atomically set the value and return the previous value. |
| toStr | Return val.toStr |
Slot Details
compareAndSet
Bool compareAndSet(Obj? expect, Obj? update)
Atomically set the value to update if current value is equivalent to the expect value compared using === operator. Return true if updated, or false if current value was not equal to the expected value. Throw NotImmutableErr if update is mutable.
getAndSet
Obj? getAndSet(Obj? val)
Atomically set the value and return the previous value. Throw NotImmutableErr if val is mutable.
make
new make(Obj? val := null)
Construct with initial value. Throw NotImmutableErr if initialized to a mutable value.
toStr
virtual override Str toStr()
Return val.toStr
val
Obj? val
The current value. Throw NotImmutableErr if set to a mutable value.
ConcurrentMap
@Js const class ConcurrentMap : Obj
ConcurrentMap is a Fantom wrapper around Java's ConcurrentHashMap. It provides high performance concurrency and allows many operations to be performed without locking. Refer to the ConcurrentHashMap Javadoc for the detailed semanatics on behavior and performance.
Constructors
| make | Make with initial capacity |
Methods
| add | Add a value by key, raise exception if key was already mapped |
| clear | Remove all the key/value pairs |
| containsKey | Return true if the specified key is mapped |
| each | Iterate the map's key value pairs |
| eachWhile | Iterate the map's key value pairs until given function returns non-null and return that as the result of this method. |
| get | Get a value by its key or return null |
| getAndSet | Set a value by key and return old value. |
| getOrAdd | Get the value for the specified key, or if it doesn't exist then automatically add it with the given default value. |
| isEmpty | Return if size is zero (this is expensive and requires full segment traveral) |
| keys | Return list of keys |
| remove | Remove a value by key, ignore if key not mapped |
| set | Set a value by key |
| setAll | Append the specified map to this map be setting every key/value from m in this map. |
| size | Return size (this is expensive and requires full segment traveral) |
| vals | Return list of values |
Slot Details
add
Void add(Obj key, Obj val)
Add a value by key, raise exception if key was already mapped
clear
Void clear()
Remove all the key/value pairs
containsKey
Bool containsKey(Obj key)
Return true if the specified key is mapped
each
Void each(|Obj,Obj| f)
Iterate the map's key value pairs
eachWhile
Obj? eachWhile(|Obj,Obj->Obj?| f)
Iterate the map's key value pairs until given function returns non-null and return that as the result of this method. Otherwise itereate every pair and return null
get
@Operator Obj? get(Obj key)
Get a value by its key or return null
getAndSet
Obj? getAndSet(Obj key, Obj val)
Set a value by key and return old value. Return the old value mapped by the key or null if it is not currently mapped.
getOrAdd
Obj getOrAdd(Obj key, Obj defVal)
Get the value for the specified key, or if it doesn't exist then automatically add it with the given default value.
isEmpty
Bool isEmpty()
Return if size is zero (this is expensive and requires full segment traveral)
keys
Obj[] keys(Type of)
Return list of keys
make
new make(Int initialCapacity := 256)
Make with initial capacity
remove
Obj? remove(Obj key)
Remove a value by key, ignore if key not mapped
set
@Operator Void set(Obj key, Obj val)
Set a value by key
setAll
This setAll(Map m)
Append the specified map to this map be setting every key/value from m in this map. Keys in m not yet mapped are added and keys already mapped are overwritten. Return this.
size
Int size()
Return size (this is expensive and requires full segment traveral)
vals
Obj[] vals(Type of)
Return list of values
Future
@Js abstract const class Future : Obj
Future represents the result of an asynchronous computation.
See Actors
Constructors
| make | Subclass constructor |
Methods
| cancel | Cancel this computation if it has not begun processing. |
| complete | Complete the future successfully with given value. |
| completeErr | Complete the future with a failure condition using given exception. |
| get | Block current thread until result is ready. |
| makeCompletable | Construct a completable future instance in the pending state. |
| status | Current state of asynchronous computation |
| waitFor | Block until this future transitions to a completed state (ok, err, or canceled). |
| waitForAll | Block on a list of futures until they all transition to a completed state. |
Slot Details
cancel
abstract Void cancel()
Cancel this computation if it has not begun processing. No guarantee is made that the computation will be cancelled.
complete
abstract This complete(Obj? val)
Complete the future successfully with given value. Raise an exception if value is not immutable or the future is already complete (ignore this call if cancelled). Raise UnsupportedErr if this future is not completable. Return this. This method is subject to change.
completeErr
abstract This completeErr(Err err)
Complete the future with a failure condition using given exception. Raise an exception if the future is already complete (ignore this call if cancelled). Return this. Raise UnsupportedErr if this future is not completable. This method is subject to change.
get
abstract Obj? get(Duration? timeout := null)
Block current thread until result is ready. If timeout occurs then TimeoutErr is raised. A null timeout blocks forever. If an exception was raised by the asynchronous computation, then it is raised to the caller of this method.
make
new make()
Subclass constructor
makeCompletable
static Future makeCompletable()
Construct a completable future instance in the pending state. This method is subject to change.
status
abstract FutureStatus status()
Current state of asynchronous computation
waitFor
abstract This waitFor(Duration? timeout := null)
Block until this future transitions to a completed state (ok, err, or canceled). If timeout is null then block forever, otherwise raise a TimeoutErr if timeout elapses. Return this.
waitForAll
static Void waitForAll(Future[] futures, Duration? timeout := null)
Block on a list of futures until they all transition to a completed state. If timeout is null block forever, otherwise raise TimeoutErr if any one of the futures does not complete before the timeout elapses.
FutureStatus
@Js @Serializable { simple=true } enum class FutureStatus : Enum
State of a Future's asynchronous computation
Enum
| pending | |
| ok | |
| err | |
| cancelled |
Constructors
| fromStr | Return the FutureStatus instance for the specified name. |
Fields
| vals | List of FutureStatus values indexed by ordinal |
Methods
| isCancelled | Return if the cancelled state |
| isComplete | Return if in any completed state: ok, err, or cancelled |
| isErr | Return if the err state |
| isOk | Return if the ok state |
| isPending | Return if pending state |
Slot Details
cancelled
const static FutureStatus cancelled := ...
err
const static FutureStatus err := ...
fromStr
static new fromStr(Str name, Bool checked := true)
Return the FutureStatus instance for the specified name. If not a valid name and checked is false return null, otherwise throw ParseErr.
isCancelled
Bool isCancelled()
Return if the cancelled state
isComplete
Bool isComplete()
Return if in any completed state: ok, err, or cancelled
isErr
Bool isErr()
Return if the err state
isOk
Bool isOk()
Return if the ok state
isPending
Bool isPending()
Return if pending state
ok
const static FutureStatus ok := ...
pending
const static FutureStatus pending := ...
vals
const static FutureStatus[] vals := ...
List of FutureStatus values indexed by ordinal
Lock
@Js const class Lock : Obj
Lock for synchronization between actors.
Methods
| lock | Acquire the lock; if not available then block forever until its available. |
| makeReentrant | Construct mutual exclusion lock. |
| tryLock | Acquire the lock if its free and immediately return true. |
| unlock | Release the lock. |
Slot Details
lock
Void lock()
Acquire the lock; if not available then block forever until its available.
makeReentrant
static Lock makeReentrant()
Construct mutual exclusion lock.
tryLock
Bool tryLock(Duration? timeout := null)
Acquire the lock if its free and immediately return true. Or if the lock is not available and timeout is null, then immediately return false. If timeout is non-null, then block up to the given timeout waiting for the lock and return true if lock acquired or false on timeout.
unlock
Void unlock()
Release the lock. Raise exception if not holding the lock.
QueueOverflowErr
const class QueueOverflowErr : Err
QueueOverflowErr is raised by a Future for messages sent to actor that has exceeded the max queue size.
Constructors
| make | Construct with specified error message and optional root cause. |
Slot Details
make
new make(Str msg := "", Err? cause := null)
Construct with specified error message and optional root cause.