progress with futures
This commit is contained in:
parent
417b9f5a1d
commit
d2dbcf2fa4
8 changed files with 68 additions and 36 deletions
|
|
@ -605,7 +605,7 @@ const
|
||||||
# thus cannot be overloaded (also documented in the spec!):
|
# thus cannot be overloaded (also documented in the spec!):
|
||||||
SpecialSemMagics* = {
|
SpecialSemMagics* = {
|
||||||
mDefined, mDefinedInScope, mCompiles, mLow, mHigh, mSizeOf, mIs, mOf,
|
mDefined, mDefinedInScope, mCompiles, mLow, mHigh, mSizeOf, mIs, mOf,
|
||||||
mEcho, mShallowCopy, mExpandToAst, mParallel}
|
mEcho, mShallowCopy, mExpandToAst, mParallel, mSpawn}
|
||||||
|
|
||||||
type
|
type
|
||||||
PNode* = ref TNode
|
PNode* = ref TNode
|
||||||
|
|
|
||||||
|
|
@ -86,8 +86,14 @@ proc indirectAccess*(a: PNode, b: string, info: TLineInfo): PNode =
|
||||||
# returns a[].b as a node
|
# returns a[].b as a node
|
||||||
var deref = newNodeI(nkHiddenDeref, info)
|
var deref = newNodeI(nkHiddenDeref, info)
|
||||||
deref.typ = a.typ.skipTypes(abstractInst).sons[0]
|
deref.typ = a.typ.skipTypes(abstractInst).sons[0]
|
||||||
assert deref.typ.kind == tyObject
|
var t = deref.typ
|
||||||
let field = getSymFromList(deref.typ.n, getIdent(b))
|
var field: PSym
|
||||||
|
while true:
|
||||||
|
assert t.kind == tyObject
|
||||||
|
field = getSymFromList(t.n, getIdent(b))
|
||||||
|
if field != nil: break
|
||||||
|
t = t.sons[0]
|
||||||
|
if t == nil: break
|
||||||
assert field != nil, b
|
assert field != nil, b
|
||||||
addSon(deref, a)
|
addSon(deref, a)
|
||||||
result = newNodeI(nkDotExpr, info)
|
result = newNodeI(nkDotExpr, info)
|
||||||
|
|
@ -124,6 +130,7 @@ proc callCodegenProc*(name: string, arg1: PNode;
|
||||||
result.add arg1
|
result.add arg1
|
||||||
if arg2 != nil: result.add arg2
|
if arg2 != nil: result.add arg2
|
||||||
if arg3 != nil: result.add arg3
|
if arg3 != nil: result.add arg3
|
||||||
|
result.typ = sym.typ.sons[0]
|
||||||
|
|
||||||
# we have 4 cases to consider:
|
# we have 4 cases to consider:
|
||||||
# - a void proc --> nothing to do
|
# - a void proc --> nothing to do
|
||||||
|
|
@ -152,15 +159,21 @@ discard """
|
||||||
We generate roughly this:
|
We generate roughly this:
|
||||||
|
|
||||||
proc f_wrapper(args) =
|
proc f_wrapper(args) =
|
||||||
|
barrierEnter(args.barrier) # for parallel statement
|
||||||
var a = args.a # copy strings/seqs; thread transfer; not generated for
|
var a = args.a # copy strings/seqs; thread transfer; not generated for
|
||||||
# the 'parallel' statement
|
# the 'parallel' statement
|
||||||
var b = args.b
|
var b = args.b
|
||||||
|
|
||||||
args.fut = createFuture(thread, sizeof(T)) # optional
|
args.fut = nimCreateFuture(thread, sizeof(T)) # optional
|
||||||
|
nimFutureCreateCondVar(args.fut) # optional
|
||||||
nimArgsPassingDone() # signal parent that the work is done
|
nimArgsPassingDone() # signal parent that the work is done
|
||||||
|
#
|
||||||
args.fut.blob = f(a, b, ...)
|
args.fut.blob = f(a, b, ...)
|
||||||
|
nimFutureSignal(args.fut)
|
||||||
|
|
||||||
# - or -
|
# - or -
|
||||||
f(a, b, ...)
|
f(a, b, ...)
|
||||||
|
barrierLeave(args.barrier) # for parallel statement
|
||||||
|
|
||||||
stmtList:
|
stmtList:
|
||||||
var scratchObj
|
var scratchObj
|
||||||
|
|
@ -196,8 +209,12 @@ proc createWrapperProc(f: PNode; threadParam, argsParam: PSym;
|
||||||
|
|
||||||
body.add callCodeGenProc("nimArgsPassingDone", threadParam.newSymNode)
|
body.add callCodeGenProc("nimArgsPassingDone", threadParam.newSymNode)
|
||||||
if fut != nil:
|
if fut != nil:
|
||||||
|
let fk = fut.typ.sons[1].futureKind
|
||||||
|
if fk == futInvalid:
|
||||||
|
localError(f.info, "cannot create a future of type: " &
|
||||||
|
typeToString(fut.typ.sons[1]))
|
||||||
body.add newAsgnStmt(indirectAccess(fut,
|
body.add newAsgnStmt(indirectAccess(fut,
|
||||||
if fut.typ.futureKind==futGC: "data" else: "blob", fut.info), call)
|
if fk == futGC: "data" else: "blob", fut.info), call)
|
||||||
if barrier == nil:
|
if barrier == nil:
|
||||||
body.add callCodeGenProc("nimFutureSignal", fut)
|
body.add callCodeGenProc("nimFutureSignal", fut)
|
||||||
else:
|
else:
|
||||||
|
|
|
||||||
|
|
@ -1579,6 +1579,12 @@ proc semShallowCopy(c: PContext, n: PNode, flags: TExprFlags): PNode =
|
||||||
else:
|
else:
|
||||||
result = semDirectOp(c, n, flags)
|
result = semDirectOp(c, n, flags)
|
||||||
|
|
||||||
|
proc createFuture(c: PContext; t: PType; info: TLineInfo): PType =
|
||||||
|
result = newType(tyGenericInvokation, c.module)
|
||||||
|
addSonSkipIntLit(result, magicsys.getCompilerProc("Future").typ)
|
||||||
|
addSonSkipIntLit(result, t)
|
||||||
|
result = instGenericContainer(c, info, result, allowMetaTypes = false)
|
||||||
|
|
||||||
proc setMs(n: PNode, s: PSym): PNode =
|
proc setMs(n: PNode, s: PSym): PNode =
|
||||||
result = n
|
result = n
|
||||||
n.sons[0] = newSymNode(s)
|
n.sons[0] = newSymNode(s)
|
||||||
|
|
@ -1610,6 +1616,12 @@ proc semMagic(c: PContext, n: PNode, s: PSym, flags: TExprFlags): PNode =
|
||||||
var x = n.lastSon
|
var x = n.lastSon
|
||||||
if x.kind == nkDo: x = x.sons[bodyPos]
|
if x.kind == nkDo: x = x.sons[bodyPos]
|
||||||
result.sons[1] = semStmt(c, x)
|
result.sons[1] = semStmt(c, x)
|
||||||
|
of mSpawn:
|
||||||
|
result = setMs(n, s)
|
||||||
|
result.sons[1] = semExpr(c, n.sons[1])
|
||||||
|
# later passes may transform the type 'Future[T]' back into 'T'
|
||||||
|
if not result[1].typ.isEmptyType:
|
||||||
|
result.typ = createFuture(c, result[1].typ, n.info)
|
||||||
else: result = semDirectOp(c, n, flags)
|
else: result = semDirectOp(c, n, flags)
|
||||||
|
|
||||||
proc semWhen(c: PContext, n: PNode, semCheck = true): PNode =
|
proc semWhen(c: PContext, n: PNode, semCheck = true): PNode =
|
||||||
|
|
|
||||||
|
|
@ -115,12 +115,6 @@ proc semLocals(c: PContext, n: PNode): PNode =
|
||||||
if it.typ.skipTypes({tyGenericInst}).kind == tyVar: a = newDeref(a)
|
if it.typ.skipTypes({tyGenericInst}).kind == tyVar: a = newDeref(a)
|
||||||
result.add(a)
|
result.add(a)
|
||||||
|
|
||||||
proc createFuture(c: PContext; t: PType; info: TLineInfo): PType =
|
|
||||||
result = newType(tyGenericInvokation, c.module)
|
|
||||||
addSonSkipIntLit(result, magicsys.getCompilerProc("Future").typ)
|
|
||||||
addSonSkipIntLit(result, t)
|
|
||||||
result = instGenericContainer(c, info, result, allowMetaTypes = false)
|
|
||||||
|
|
||||||
proc semShallowCopy(c: PContext, n: PNode, flags: TExprFlags): PNode
|
proc semShallowCopy(c: PContext, n: PNode, flags: TExprFlags): PNode
|
||||||
proc magicsAfterOverloadResolution(c: PContext, n: PNode,
|
proc magicsAfterOverloadResolution(c: PContext, n: PNode,
|
||||||
flags: TExprFlags): PNode =
|
flags: TExprFlags): PNode =
|
||||||
|
|
@ -136,9 +130,4 @@ proc magicsAfterOverloadResolution(c: PContext, n: PNode,
|
||||||
of mShallowCopy: result = semShallowCopy(c, n, flags)
|
of mShallowCopy: result = semShallowCopy(c, n, flags)
|
||||||
of mNBindSym: result = semBindSym(c, n)
|
of mNBindSym: result = semBindSym(c, n)
|
||||||
of mLocals: result = semLocals(c, n)
|
of mLocals: result = semLocals(c, n)
|
||||||
of mSpawn:
|
|
||||||
result = n
|
|
||||||
# later passes may transform the type 'Future[T]' back into 'T'
|
|
||||||
if not n[1].typ.isEmptyType:
|
|
||||||
result.typ = createFuture(c, n[1].typ, n.info)
|
|
||||||
else: result = n
|
else: result = n
|
||||||
|
|
|
||||||
|
|
@ -57,7 +57,7 @@ proc openBarrier*(b: ptr Barrier) {.compilerProc.} =
|
||||||
b.cv = createCondVar()
|
b.cv = createCondVar()
|
||||||
|
|
||||||
proc closeBarrier*(b: ptr Barrier) {.compilerProc.} =
|
proc closeBarrier*(b: ptr Barrier) {.compilerProc.} =
|
||||||
await(b.cv)
|
while b.counter > 0: await(b.cv)
|
||||||
destroyCondVar(b.cv)
|
destroyCondVar(b.cv)
|
||||||
|
|
||||||
{.pop.}
|
{.pop.}
|
||||||
|
|
@ -136,8 +136,13 @@ proc nimFutureCreateCondVar(fut: RawFuture) {.compilerProc.} =
|
||||||
fut.usesCondVar = true
|
fut.usesCondVar = true
|
||||||
|
|
||||||
proc nimFutureSignal(fut: RawFuture) {.compilerProc.} =
|
proc nimFutureSignal(fut: RawFuture) {.compilerProc.} =
|
||||||
assert fut.usesCondVar
|
if fut.ai != nil:
|
||||||
signal(fut.cv)
|
acquire(fut.ai.cv.L)
|
||||||
|
fut.ai.idx = fut.idx
|
||||||
|
inc fut.ai.cv.counter
|
||||||
|
release(fut.ai.cv.L)
|
||||||
|
signal(fut.ai.cv.c)
|
||||||
|
if fut.usesCondVar: signal(fut.cv)
|
||||||
|
|
||||||
proc await*[T](fut: Future[T]) =
|
proc await*[T](fut: Future[T]) =
|
||||||
## waits until the value for the future arrives.
|
## waits until the value for the future arrives.
|
||||||
|
|
@ -147,28 +152,21 @@ proc `^`*[T](fut: Future[T]): T =
|
||||||
## blocks until the value is available and then returns this value. Note
|
## blocks until the value is available and then returns this value. Note
|
||||||
## this reading is destructive for reasons of efficiency and convenience.
|
## this reading is destructive for reasons of efficiency and convenience.
|
||||||
## This calls ``finished(fut)``.
|
## This calls ``finished(fut)``.
|
||||||
await(fut)
|
if fut.usesCondVar: await(fut)
|
||||||
when T is string or T is seq or T is ref:
|
when T is string or T is seq or T is ref:
|
||||||
result = cast[T](fut.data)
|
result = cast[T](fut.data)
|
||||||
else:
|
else:
|
||||||
result = fut.payload
|
result = fut.blob
|
||||||
finished(fut)
|
finished(fut)
|
||||||
|
|
||||||
proc notify*(fut: RawFuture) {.compilerproc.} =
|
|
||||||
if fut.ai != nil:
|
|
||||||
acquire(fut.ai.cv.L)
|
|
||||||
fut.ai.idx = fut.idx
|
|
||||||
inc fut.ai.cv.counter
|
|
||||||
release(fut.ai.cv.L)
|
|
||||||
signal(fut.ai.cv.c)
|
|
||||||
if fut.usesCondVar: signal(fut.cv)
|
|
||||||
|
|
||||||
proc awaitAny*(futures: openArray[RawFuture]): int =
|
proc awaitAny*(futures: openArray[RawFuture]): int =
|
||||||
# awaits any of the given futures. Returns the index of one future for which
|
# awaits any of the given futures. Returns the index of one future for which
|
||||||
## a value arrived. A future only supports one call to 'awaitAny' at the
|
## a value arrived. A future only supports one call to 'awaitAny' at the
|
||||||
## same time. That means if you await([a,b]) and await([b,c]) the second
|
## same time. That means if you await([a,b]) and await([b,c]) the second
|
||||||
## call will only await 'c'. If there is no future left to be able to wait
|
## call will only await 'c'. If there is no future left to be able to wait
|
||||||
## on, -1 is returned.
|
## on, -1 is returned.
|
||||||
|
## **Note**: This results in non-deterministic behaviour and so should be
|
||||||
|
## avoided.
|
||||||
var ai: AwaitInfo
|
var ai: AwaitInfo
|
||||||
ai.cv = createCondVar()
|
ai.cv = createCondVar()
|
||||||
var conflicts = 0
|
var conflicts = 0
|
||||||
|
|
@ -245,19 +243,18 @@ proc preferSpawn*(): bool =
|
||||||
## it is not necessary to call this directly; use 'spawnX' instead.
|
## it is not necessary to call this directly; use 'spawnX' instead.
|
||||||
result = gSomeReady.counter > 0
|
result = gSomeReady.counter > 0
|
||||||
|
|
||||||
proc spawn*(call: stmt) {.magic: "Spawn".}
|
proc spawn*(call: expr): expr {.magic: "Spawn".}
|
||||||
## always spawns a new task, so that the 'call' is never executed on
|
## always spawns a new task, so that the 'call' is never executed on
|
||||||
## the calling thread. 'call' has to be proc call 'p(...)' where 'p'
|
## the calling thread. 'call' has to be proc call 'p(...)' where 'p'
|
||||||
## is gcsafe and has 'void' as the return type.
|
## is gcsafe and has 'void' as the return type.
|
||||||
|
|
||||||
template spawnX*(call: stmt) =
|
template spawnX*(call: expr): expr =
|
||||||
## spawns a new task if a CPU core is ready, otherwise executes the
|
## spawns a new task if a CPU core is ready, otherwise executes the
|
||||||
## call in the calling thread. Usually it is advised to
|
## call in the calling thread. Usually it is advised to
|
||||||
## use 'spawn' in order to not block the producer for an unknown
|
## use 'spawn' in order to not block the producer for an unknown
|
||||||
## amount of time. 'call' has to be proc call 'p(...)' where 'p'
|
## amount of time. 'call' has to be proc call 'p(...)' where 'p'
|
||||||
## is gcsafe and has 'void' as the return type.
|
## is gcsafe and has 'void' as the return type.
|
||||||
if preferSpawn(): spawn call
|
(if preferSpawn(): spawn call else: call)
|
||||||
else: call
|
|
||||||
|
|
||||||
proc parallel*(body: stmt) {.magic: "Parallel".}
|
proc parallel*(body: stmt) {.magic: "Parallel".}
|
||||||
## a parallel section can be used to execute a block in parallel. ``body``
|
## a parallel section can be used to execute a block in parallel. ``body``
|
||||||
|
|
|
||||||
17
tests/parallel/tflowvar.nim
Normal file
17
tests/parallel/tflowvar.nim
Normal file
|
|
@ -0,0 +1,17 @@
|
||||||
|
discard """
|
||||||
|
output: '''foobarfoobarbazbearbazbear'''
|
||||||
|
cmd: "nimrod $target --threads:on $options $file"
|
||||||
|
"""
|
||||||
|
|
||||||
|
import threadpool
|
||||||
|
|
||||||
|
proc computeSomething(a, b: string): string = a & b & a & b
|
||||||
|
|
||||||
|
proc main =
|
||||||
|
let fvA = spawn computeSomething("foo", "bar")
|
||||||
|
let fvB = spawn computeSomething("baz", "bear")
|
||||||
|
|
||||||
|
echo(^fvA, ^fvB)
|
||||||
|
|
||||||
|
main()
|
||||||
|
sync()
|
||||||
|
|
@ -1,6 +1,6 @@
|
||||||
discard """
|
discard """
|
||||||
line: 7
|
line: 7
|
||||||
errormsg: "'spawn' takes a call expression of type void"
|
errormsg: "'spawn' takes a call expression"
|
||||||
cmd: "nimrod $target --threads:on $options $file"
|
cmd: "nimrod $target --threads:on $options $file"
|
||||||
"""
|
"""
|
||||||
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue