some changes
This commit is contained in:
parent
4220b1c81d
commit
947b152163
7 changed files with 70 additions and 25 deletions
|
|
@ -560,7 +560,7 @@ type
|
||||||
mFloat, mFloat32, mFloat64, mFloat128,
|
mFloat, mFloat32, mFloat64, mFloat128,
|
||||||
mBool, mChar, mString, mCstring,
|
mBool, mChar, mString, mCstring,
|
||||||
mPointer, mEmptySet, mIntSetBaseType, mNil, mExpr, mStmt, mTypeDesc,
|
mPointer, mEmptySet, mIntSetBaseType, mNil, mExpr, mStmt, mTypeDesc,
|
||||||
mVoidType, mPNimrodNode, mShared, mGuarded, mLock, mSpawn,
|
mVoidType, mPNimrodNode, mShared, mGuarded, mLock, mSpawn, mDeepCopy,
|
||||||
mIsMainModule, mCompileDate, mCompileTime, mNimrodVersion, mNimrodMajor,
|
mIsMainModule, mCompileDate, mCompileTime, mNimrodVersion, mNimrodMajor,
|
||||||
mNimrodMinor, mNimrodPatch, mCpuEndian, mHostOS, mHostCPU, mAppType,
|
mNimrodMinor, mNimrodPatch, mCpuEndian, mHostOS, mHostCPU, mAppType,
|
||||||
mNaN, mInf, mNegInf,
|
mNaN, mInf, mNegInf,
|
||||||
|
|
|
||||||
|
|
@ -1087,7 +1087,7 @@ proc semTypeNode(c: PContext, n: PNode, prev: PType): PType =
|
||||||
elif n[0].kind notin nkIdentKinds:
|
elif n[0].kind notin nkIdentKinds:
|
||||||
result = semTypeExpr(c, n)
|
result = semTypeExpr(c, n)
|
||||||
else:
|
else:
|
||||||
let op = considerAcc(n.sons[0])
|
let op = considerQuotedIdent(n.sons[0])
|
||||||
if op.id in {ord(wAnd), ord(wOr)} or op.s == "|":
|
if op.id in {ord(wAnd), ord(wOr)} or op.s == "|":
|
||||||
checkSonsLen(n, 3)
|
checkSonsLen(n, 3)
|
||||||
var
|
var
|
||||||
|
|
|
||||||
|
|
@ -6,6 +6,8 @@ Nimrod has two flavors of parallelism:
|
||||||
1) `Structured`:idx parallelism via the ``parallel`` statement.
|
1) `Structured`:idx parallelism via the ``parallel`` statement.
|
||||||
2) `Unstructured`:idx: parallelism via the standalone ``spawn`` statement.
|
2) `Unstructured`:idx: parallelism via the standalone ``spawn`` statement.
|
||||||
|
|
||||||
|
Both need the `threadpool <threadpool.html>`_ module to work.
|
||||||
|
|
||||||
Somewhat confusingly, ``spawn`` is also used in the ``parallel`` statement
|
Somewhat confusingly, ``spawn`` is also used in the ``parallel`` statement
|
||||||
with slightly different semantics. ``spawn`` always takes a call expression of
|
with slightly different semantics. ``spawn`` always takes a call expression of
|
||||||
the form ``f(a, ...)``. Let ``T`` be ``f``'s return type. If ``T`` is ``void``
|
the form ``f(a, ...)``. Let ``T`` be ``f``'s return type. If ``T`` is ``void``
|
||||||
|
|
@ -20,6 +22,25 @@ the overhead of an indirection via ``FlowVar[T]`` to ensure correctness.
|
||||||
Parallel statement
|
Parallel statement
|
||||||
==================
|
==================
|
||||||
|
|
||||||
|
Example:
|
||||||
|
|
||||||
|
.. code-block:: nimrod
|
||||||
|
# Compute PI in an inefficient way
|
||||||
|
import strutils, math, threadpool
|
||||||
|
|
||||||
|
proc term(k: float): float = 4 * math.pow(-1, k) / (2*k + 1)
|
||||||
|
|
||||||
|
proc pi(n: int): float =
|
||||||
|
var ch = newSeq[float](n+1)
|
||||||
|
parallel:
|
||||||
|
for k in 0..ch.high:
|
||||||
|
ch[k] = spawn term(float(k))
|
||||||
|
for k in 0..ch.high:
|
||||||
|
result += ch[k]
|
||||||
|
|
||||||
|
echo formatFloat(pi(5000))
|
||||||
|
|
||||||
|
|
||||||
The parallel statement is the preferred mechanism to introduce parallelism
|
The parallel statement is the preferred mechanism to introduce parallelism
|
||||||
in a Nimrod program. A subset of the Nimrod language is valid within a
|
in a Nimrod program. A subset of the Nimrod language is valid within a
|
||||||
``parallel`` section. This subset is checked to be free of data races at
|
``parallel`` section. This subset is checked to be free of data races at
|
||||||
|
|
@ -34,13 +55,17 @@ restrictions / changes:
|
||||||
``dest`` is part of the pattern ``dest = spawn f(...)`` has to be
|
``dest`` is part of the pattern ``dest = spawn f(...)`` has to be
|
||||||
provable disjoint. This is called the *disjoint check*.
|
provable disjoint. This is called the *disjoint check*.
|
||||||
* Every other complex location ``loc`` that is used in a spawned
|
* Every other complex location ``loc`` that is used in a spawned
|
||||||
proc (``spawn f(loc)``) has to immutable for the duration of
|
proc (``spawn f(loc)``) has to be immutable for the duration of
|
||||||
the ``parallel``. This is called the *immutability check*. Currently it
|
the ``parallel`` section. This is called the *immutability check*. Currently
|
||||||
is not specified what exactly "complex location" means. We need to make that
|
it is not specified what exactly "complex location" means. We need to make
|
||||||
an optimization!
|
this an optimization!
|
||||||
* Every array access has to be provable within bounds.
|
* Every array access has to be provable within bounds. This is called
|
||||||
|
the *bounds check*.
|
||||||
* Slices are optimized so that no copy is performed. This optimization is not
|
* Slices are optimized so that no copy is performed. This optimization is not
|
||||||
yet performed for ordinary slices outside of a ``parallel`` section.
|
yet performed for ordinary slices outside of a ``parallel`` section. Slices
|
||||||
|
are also special in that they currently do not support negative indexes!
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
Spawn statement
|
Spawn statement
|
||||||
|
|
@ -49,9 +74,25 @@ Spawn statement
|
||||||
A standalone ``spawn`` statement is a simple construct. It executes
|
A standalone ``spawn`` statement is a simple construct. It executes
|
||||||
the passed expression on the thread pool and returns a `data flow variable`:idx:
|
the passed expression on the thread pool and returns a `data flow variable`:idx:
|
||||||
``FlowVar[T]`` that can be read from. The reading with the ``^`` operator is
|
``FlowVar[T]`` that can be read from. The reading with the ``^`` operator is
|
||||||
**blocking**. However, one can use ``awaitAny`` to wait on multiple flow variables
|
**blocking**. However, one can use ``awaitAny`` to wait on multiple flow
|
||||||
at the same time.
|
variables at the same time:
|
||||||
|
|
||||||
|
.. code-block:: nimrod
|
||||||
|
import threadpool, ...
|
||||||
|
|
||||||
|
# wait until 2 out of 3 servers received the update:
|
||||||
|
proc main =
|
||||||
|
var responses = newSeq[RawFlowVar](3)
|
||||||
|
for i in 0..2:
|
||||||
|
responses[i] = spawn tellServer(Update, "key", "value")
|
||||||
|
var index = awaitAny(responses)
|
||||||
|
assert index >= 0
|
||||||
|
responses.del(index)
|
||||||
|
discard awaitAny(responses)
|
||||||
|
|
||||||
Like the ``parallel`` statement data flow variables ensure that no data races
|
Like the ``parallel`` statement data flow variables ensure that no data races
|
||||||
are possible.
|
are possible. Due to technical limitations not every type ``T`` is possible in
|
||||||
|
a data flow variable: ``T`` has to be of the type ``ref``, ``string``, ``seq``
|
||||||
|
or of a type that doesn't contain a type that is garbage collected. This
|
||||||
|
restriction will be removed in the future.
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -90,8 +90,8 @@ type
|
||||||
cv: CondVar
|
cv: CondVar
|
||||||
idx: int
|
idx: int
|
||||||
|
|
||||||
RawFlowVar* = ref RawFlowVarObj ## untyped base class for 'FlowVar[T]'
|
FlowVarBase* = ref FlowVarBaseObj ## untyped base class for 'FlowVar[T]'
|
||||||
RawFlowVarObj = object of TObject
|
FlowVarBaseObj = object of TObject
|
||||||
ready, usesCondVar: bool
|
ready, usesCondVar: bool
|
||||||
cv: CondVar #\
|
cv: CondVar #\
|
||||||
# for 'awaitAny' support
|
# for 'awaitAny' support
|
||||||
|
|
@ -100,7 +100,7 @@ type
|
||||||
data: pointer # we incRef and unref it to keep it alive
|
data: pointer # we incRef and unref it to keep it alive
|
||||||
owner: pointer # ptr Worker
|
owner: pointer # ptr Worker
|
||||||
|
|
||||||
FlowVarObj[T] = object of RawFlowVarObj
|
FlowVarObj[T] = object of FlowVarBaseObj
|
||||||
blob: T
|
blob: T
|
||||||
|
|
||||||
FlowVar*{.compilerProc.}[T] = ref FlowVarObj[T] ## a data flow variable
|
FlowVar*{.compilerProc.}[T] = ref FlowVarObj[T] ## a data flow variable
|
||||||
|
|
@ -123,7 +123,7 @@ type
|
||||||
shutdown: bool # the pool requests to shut down this worker thread
|
shutdown: bool # the pool requests to shut down this worker thread
|
||||||
q: ToFreeQueue
|
q: ToFreeQueue
|
||||||
|
|
||||||
proc await*(fv: RawFlowVar) =
|
proc await*(fv: FlowVarBase) =
|
||||||
## waits until the value for the flowVar arrives. Usually it is not necessary
|
## waits until the value for the flowVar arrives. Usually it is not necessary
|
||||||
## to call this explicitly.
|
## to call this explicitly.
|
||||||
if fv.usesCondVar:
|
if fv.usesCondVar:
|
||||||
|
|
@ -131,7 +131,7 @@ proc await*(fv: RawFlowVar) =
|
||||||
await(fv.cv)
|
await(fv.cv)
|
||||||
destroyCondVar(fv.cv)
|
destroyCondVar(fv.cv)
|
||||||
|
|
||||||
proc finished(fv: RawFlowVar) =
|
proc finished(fv: FlowVarBase) =
|
||||||
doAssert fv.ai.isNil, "flowVar is still attached to an 'awaitAny'"
|
doAssert fv.ai.isNil, "flowVar is still attached to an 'awaitAny'"
|
||||||
# we have to protect against the rare cases where the owner of the flowVar
|
# we have to protect against the rare cases where the owner of the flowVar
|
||||||
# simply disregards the flowVar and yet the "flowVarr" has not yet written
|
# simply disregards the flowVar and yet the "flowVarr" has not yet written
|
||||||
|
|
@ -171,11 +171,11 @@ proc fvFinalizer[T](fv: FlowVar[T]) = finished(fv)
|
||||||
proc nimCreateFlowVar[T](): FlowVar[T] {.compilerProc.} =
|
proc nimCreateFlowVar[T](): FlowVar[T] {.compilerProc.} =
|
||||||
new(result, fvFinalizer)
|
new(result, fvFinalizer)
|
||||||
|
|
||||||
proc nimFlowVarCreateCondVar(fv: RawFlowVar) {.compilerProc.} =
|
proc nimFlowVarCreateCondVar(fv: FlowVarBase) {.compilerProc.} =
|
||||||
fv.cv = createCondVar()
|
fv.cv = createCondVar()
|
||||||
fv.usesCondVar = true
|
fv.usesCondVar = true
|
||||||
|
|
||||||
proc nimFlowVarSignal(fv: RawFlowVar) {.compilerProc.} =
|
proc nimFlowVarSignal(fv: FlowVarBase) {.compilerProc.} =
|
||||||
if fv.ai != nil:
|
if fv.ai != nil:
|
||||||
acquire(fv.ai.cv.L)
|
acquire(fv.ai.cv.L)
|
||||||
fv.ai.idx = fv.idx
|
fv.ai.idx = fv.idx
|
||||||
|
|
@ -211,7 +211,7 @@ proc `^`*[T](fv: FlowVar[T]): T =
|
||||||
else:
|
else:
|
||||||
result = fv.blob
|
result = fv.blob
|
||||||
|
|
||||||
proc awaitAny*(flowVars: openArray[RawFlowVar]): int =
|
proc awaitAny*(flowVars: openArray[FlowVarBase]): int =
|
||||||
## awaits any of the given flowVars. Returns the index of one flowVar for
|
## awaits any of the given flowVars. Returns the index of one flowVar for
|
||||||
## which a value arrived. A flowVar only supports one call to 'awaitAny' at
|
## which a value arrived. A flowVar only supports one call to 'awaitAny' at
|
||||||
## the same time. That means if you await([a,b]) and await([b,c]) the second
|
## the same time. That means if you await([a,b]) and await([b,c]) the second
|
||||||
|
|
|
||||||
|
|
@ -2942,6 +2942,10 @@ proc locals*(): TObject {.magic: "Locals", noSideEffect.} =
|
||||||
## # -> B is 1
|
## # -> B is 1
|
||||||
discard
|
discard
|
||||||
|
|
||||||
|
proc deepCopy*[T](x: T): T {.magic: "DeepCopy", noSideEffect.}
|
||||||
|
## performs a deep copy of `x`. This is also used by the code generator
|
||||||
|
## for the implementation of ``spawn``.
|
||||||
|
|
||||||
when not defined(booting):
|
when not defined(booting):
|
||||||
type
|
type
|
||||||
semistatic*[T] = static[T] | T
|
semistatic*[T] = static[T] | T
|
||||||
|
|
|
||||||
|
|
@ -8,7 +8,7 @@ import strutils, math, threadpool
|
||||||
proc term(k: float): float = 4 * math.pow(-1, k) / (2*k + 1)
|
proc term(k: float): float = 4 * math.pow(-1, k) / (2*k + 1)
|
||||||
|
|
||||||
proc piU(n: int): float =
|
proc piU(n: int): float =
|
||||||
var ch = newSeq[Promise[float]](n+1)
|
var ch = newSeq[FlowVar[float]](n+1)
|
||||||
for k in 0..n:
|
for k in 0..n:
|
||||||
ch[k] = spawn term(float(k))
|
ch[k] = spawn term(float(k))
|
||||||
for k in 0..n:
|
for k in 0..n:
|
||||||
|
|
|
||||||
4
todo.txt
4
todo.txt
|
|
@ -4,15 +4,15 @@ version 0.9.6
|
||||||
Concurrency
|
Concurrency
|
||||||
-----------
|
-----------
|
||||||
|
|
||||||
- document the new 'spawn' and 'parallel' statements
|
- the disjoint checker needs to deal with 'a = spawn f(); g = spawn f()'
|
||||||
- implement 'deepCopy' builtin
|
- implement 'deepCopy' builtin
|
||||||
- implement 'foo[1..4] = spawn(f[4..7])'
|
- implement 'foo[1..4] = spawn(f[4..7])'
|
||||||
- the disjoint checker needs to deal with 'a = spawn f(); g = spawn f()'
|
|
||||||
- support for exception propagation
|
- support for exception propagation
|
||||||
- Minor: The copying of the 'ref Promise' into the thead local storage only
|
- Minor: The copying of the 'ref Promise' into the thead local storage only
|
||||||
happens to work due to the write barrier's implementation
|
happens to work due to the write barrier's implementation
|
||||||
- 'gcsafe' inferrence needs to be fixed
|
- 'gcsafe' inferrence needs to be fixed
|
||||||
- implement lock levels --> first without the more complex race avoidance
|
- implement lock levels --> first without the more complex race avoidance
|
||||||
|
- document the new 'spawn' and 'parallel' statements
|
||||||
|
|
||||||
|
|
||||||
Misc
|
Misc
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue