diff options
Diffstat (limited to 'tests/async')
69 files changed, 3541 insertions, 0 deletions
diff --git a/tests/async/hello.txt b/tests/async/hello.txt new file mode 100644 index 000000000..854d6c20a --- /dev/null +++ b/tests/async/hello.txt @@ -0,0 +1 @@ +hello humans! \ No newline at end of file diff --git a/tests/async/nim.cfg b/tests/async/nim.cfg new file mode 100644 index 000000000..be50f572c --- /dev/null +++ b/tests/async/nim.cfg @@ -0,0 +1 @@ +--experimental:strictEffects diff --git a/tests/async/t11558.nim b/tests/async/t11558.nim new file mode 100644 index 000000000..99961e7b6 --- /dev/null +++ b/tests/async/t11558.nim @@ -0,0 +1,13 @@ +discard """ +output: "Hello\n9" +""" +import std/asyncdispatch + +proc foo(): Future[string] {.async.} = + "Hello" + +proc bar(): Future[int] {.async.} = + result = 9 + +echo waitFor foo() +echo waitFor bar() diff --git a/tests/async/t12221.nim b/tests/async/t12221.nim new file mode 100644 index 000000000..e8bd9c11a --- /dev/null +++ b/tests/async/t12221.nim @@ -0,0 +1,40 @@ +import asyncdispatch, os, times + +proc doubleSleep(hardSleep: int) {.async.} = + await sleepAsync(50) + sleep(hardSleep) + +template assertTime(target, timeTook: float): untyped {.dirty.} = + doAssert(timeTook*1000 > target - 1000, "Took too short, should've taken " & + $target & "ms, but took " & $(timeTook*1000) & "ms") + doAssert(timeTook*1000 < target + 1000, "Took too long, should've taken " & + $target & "ms, but took " & $(timeTook*1000) & "ms") + +var + start: float + fut: Future[void] + +# NOTE: this uses poll(3000) to limit timing error potential. +start = epochTime() +fut = sleepAsync(40) and sleepAsync(100) and doubleSleep(20) +while not fut.finished: + poll(1000) +assertTime(150, epochTime() - start) + +start = epochTime() +fut = sleepAsync(40) and sleepAsync(100) and doubleSleep(50) +while not fut.finished: + poll(1000) +assertTime(200, epochTime() - start) + +start = epochTime() +fut = sleepAsync(40) and sleepAsync(100) and doubleSleep(20) and sleepAsync(200) +while not fut.finished: + poll(1000) +assertTime(300, epochTime() - start) + +start = epochTime() +fut = (sleepAsync(40) and sleepAsync(100) and doubleSleep(20)) or sleepAsync(300) +while not fut.finished: + poll(1000) +assertTime(150, epochTime() - start) diff --git a/tests/async/t13889.nim b/tests/async/t13889.nim new file mode 100644 index 000000000..fe75fe38a --- /dev/null +++ b/tests/async/t13889.nim @@ -0,0 +1,27 @@ +discard """ + output: ''' +believer Foo is saved:true +believer Bar is saved:true +believer Baz is saved:true +''' +""" + +import asyncdispatch + +var + promise = newFuture[bool]() + +proc believers(name: string) {.async.} = + let v = await promise + echo "believer " & name & " is saved:" & $v + +asyncCheck believers("Foo") +asyncCheck believers("Bar") +asyncCheck believers("Baz") + +proc savior() {.async.} = + await sleepAsync(50) + complete(promise, true) + await sleepAsync(50) # give enough time to see who was saved + +waitFor(savior()) diff --git a/tests/async/t14820.nim b/tests/async/t14820.nim new file mode 100644 index 000000000..359884468 --- /dev/null +++ b/tests/async/t14820.nim @@ -0,0 +1,25 @@ +discard """ +output: ''' +iteration: 1 +iteration: 2 +iteration: 3 +iteration: 4 +async done +iteration: 5 +''' +""" + +import asyncdispatch, times + +var done = false +proc somethingAsync() {.async.} = + yield sleepAsync 5000 + echo "async done" + done = true + +asyncCheck somethingAsync() +var count = 0 +while not done: + count += 1 + drain 1000 + echo "iteration: ", count diff --git a/tests/async/t15148.nim b/tests/async/t15148.nim new file mode 100644 index 000000000..ba14c1157 --- /dev/null +++ b/tests/async/t15148.nim @@ -0,0 +1,12 @@ +import asyncdispatch, asyncfile, os + +const Filename = "t15148.txt" + +proc saveEmpty() {.async.} = + let + text = "" + file = openAsync(Filename, fmWrite) + await file.write(text) + file.close() + +waitFor saveEmpty() diff --git a/tests/async/t15804.nim b/tests/async/t15804.nim new file mode 100644 index 000000000..8de012196 --- /dev/null +++ b/tests/async/t15804.nim @@ -0,0 +1,15 @@ +import asyncdispatch + +type + Foo*[E] = ref object + op: proc(): Future[bool] {.gcsafe.} + +proc newFoo*[E](): Foo[E] = + result = Foo[E]() + result.op = proc(): Future[bool] {.gcsafe,async.} = + await sleepAsync(100) + result = false + +when isMainModule: + let f = newFoo[int]() + echo waitFor f.op() diff --git a/tests/async/t17045.nim b/tests/async/t17045.nim new file mode 100644 index 000000000..2b5acf48a --- /dev/null +++ b/tests/async/t17045.nim @@ -0,0 +1,28 @@ +discard """ + targets: "c cpp" + matrix: "--mm:refc; --mm:arc" +""" + +type Future = ref object + +iterator paths: string = + # without "when nimvm" everything works + when nimvm: + yield "test.md" + else: + yield "test.md" + +template await(f: Future): string = + # need this yield, also the template has to return something + yield f + "hello world" + +proc generatePostContextsAsync() = + iterator generatePostContextsAsyncIter(): Future {.closure.} = + for filePath in paths(): + var temp = await Future() + + # need this line + var nameIterVar = generatePostContextsAsyncIter + +generatePostContextsAsync() \ No newline at end of file diff --git a/tests/async/t20111.nim b/tests/async/t20111.nim new file mode 100644 index 000000000..0aaa7d886 --- /dev/null +++ b/tests/async/t20111.nim @@ -0,0 +1,19 @@ +discard """ + action: "run" +""" +import asyncdispatch +type + Sync = object + Async = object + SyncRes = (Sync, string) + AsyncRes = (Async, string) + +proc foo(val: Sync | Async): Future[(Async, string) | (Sync, string)] {.multisync.} = + return (val, "hello") + +let + myAsync = Async() + mySync = Sync() + +doAssert typeof(waitFor foo(myAsync)) is AsyncRes +doAssert typeof(foo(mySync)) is SyncRes diff --git a/tests/async/t21447.nim b/tests/async/t21447.nim new file mode 100644 index 000000000..e4f7ae31f --- /dev/null +++ b/tests/async/t21447.nim @@ -0,0 +1,6 @@ +discard """ + action: "compile" + cmd: "nim c -d:release -d:futureLogging $file" +""" + +import std/asyncdispatch diff --git a/tests/async/t21893.nim b/tests/async/t21893.nim new file mode 100644 index 000000000..658cb02eb --- /dev/null +++ b/tests/async/t21893.nim @@ -0,0 +1,13 @@ +discard """ +output: "@[97]\ntrue" +""" + +import asyncdispatch + +proc test(): Future[bool] {.async.} = + const S4 = @[byte('a')] + echo S4 + return true + +echo waitFor test() + diff --git a/tests/async/t22210.nim b/tests/async/t22210.nim new file mode 100644 index 000000000..fcf939472 --- /dev/null +++ b/tests/async/t22210.nim @@ -0,0 +1,41 @@ +discard """ +output: ''' +stage 1 +stage 2 +stage 3 +(status: 200, data: "SOMEDATA") +''' +""" + +import std/asyncdispatch + + +# bug #22210 +type + ClientResponse = object + status*: int + data*: string + +proc subFoo1(): Future[int] {.async.} = + await sleepAsync(100) + return 200 + +proc subFoo2(): Future[string] {.async.} = + await sleepAsync(100) + return "SOMEDATA" + +proc testFoo(): Future[ClientResponse] {.async.} = + try: + let status = await subFoo1() + doAssert(status == 200) + let data = await subFoo2() + return ClientResponse(status: status, data: data) + finally: + echo "stage 1" + await sleepAsync(100) + echo "stage 2" + await sleepAsync(200) + echo "stage 3" + +when isMainModule: + echo waitFor testFoo() \ No newline at end of file diff --git a/tests/async/t22210_2.nim b/tests/async/t22210_2.nim new file mode 100644 index 000000000..9db664a32 --- /dev/null +++ b/tests/async/t22210_2.nim @@ -0,0 +1,73 @@ +import std/asyncdispatch + + +# bug #22210 +type + ClientResponse = object + status*: int + data*: string + +proc subFoo1(): Future[int] {.async.} = + await sleepAsync(100) + return 200 + +proc subFoo2(): Future[string] {.async.} = + await sleepAsync(100) + return "SOMEDATA" + + +proc testFoo2(): Future[ClientResponse] {.async.} = + var flag = 0 + try: + let status = await subFoo1() + doAssert(status == 200) + let data = await subFoo2() + result = ClientResponse(status: status, data: data) + finally: + inc flag + await sleepAsync(100) + inc flag + await sleepAsync(200) + inc flag + doAssert flag == 3 + +discard waitFor testFoo2() + +proc testFoo3(): Future[ClientResponse] {.async.} = + var flag = 0 + try: + let status = await subFoo1() + doAssert(status == 200) + let data = await subFoo2() + if false: + return ClientResponse(status: status, data: data) + finally: + inc flag + await sleepAsync(100) + inc flag + await sleepAsync(200) + inc flag + doAssert flag == 3 + +discard waitFor testFoo3() + + +proc testFoo4(): Future[ClientResponse] {.async.} = + var flag = 0 + try: + let status = await subFoo1() + doAssert(status == 200) + let data = await subFoo2() + if status == 200: + return ClientResponse(status: status, data: data) + else: + return ClientResponse() + finally: + inc flag + await sleepAsync(100) + inc flag + await sleepAsync(200) + inc flag + doAssert flag == 3 + +discard waitFor testFoo4() diff --git a/tests/async/t3075.nim b/tests/async/t3075.nim new file mode 100644 index 000000000..d26aa0a36 --- /dev/null +++ b/tests/async/t3075.nim @@ -0,0 +1,29 @@ +import asyncnet, asyncdispatch, strtabs + +type + WebSocketCallback = proc (client: WebSocket, message: WebSocketMessage) {.closure, gcsafe.} + WebSocketRecvClosure = proc (ws: WebSocket): Future[string] {.gcsafe.} + + WebSocketMessage = ref object + msg: string + + WebSocket = ref object + socket: AsyncSocket + header: StringTableRef + onOpen: WebSocketCallback + onMessage: WebSocketCallback + onClose: WebSocketCallback + +proc recv(ws: WebSocket, p: WebSocketRecvClosure): Future[string] {.async.}= + if not ws.socket.isClosed(): + result = await ws.p() + if result == "": + ws.socket.close() + if ws.onClose != nil: + ws.onClose(ws, nil) + return result + +proc reŅvSize(ws: WebSocket, size: int): Future[string] {.async.} = + proc recvSizeClosure(ws: WebSocket): Future[string] {.async.} = + return await ws.socket.recv(size) + return await ws.recv(recvSizeClosure) diff --git a/tests/async/t6846.nim b/tests/async/t6846.nim new file mode 100644 index 000000000..7fe38f3b3 --- /dev/null +++ b/tests/async/t6846.nim @@ -0,0 +1,15 @@ +discard """ + exitcode: 0 + output: "hello world" + disabled: windows +""" + +import asyncdispatch +import asyncfile + +var asyncStdout = 1.AsyncFD.newAsyncFile() +proc doStuff: Future[void] {.async.} = + await asyncStdout.write "hello world\n" + +let fut = doStuff() +doAssert fut.finished, "Poll is needed unnecessarily. See #6846." diff --git a/tests/async/t7192.nim b/tests/async/t7192.nim new file mode 100644 index 000000000..9ac0e07c0 --- /dev/null +++ b/tests/async/t7192.nim @@ -0,0 +1,14 @@ +discard """ +output: ''' +testCallback() +''' +""" + +import asyncdispatch + +proc testCallback() = + echo "testCallback()" + +when true: + callSoon(testCallback) + poll() diff --git a/tests/async/t7758.nim b/tests/async/t7758.nim new file mode 100644 index 000000000..fe6d32ad3 --- /dev/null +++ b/tests/async/t7758.nim @@ -0,0 +1,21 @@ +import asyncdispatch +import std/unittest + +proc task() {.async.} = + const tSleep = 40 + await sleepAsync(tSleep) + +proc main() = + var counter = 0 + var f = task() + while not f.finished: + inc(counter) + poll(10) + + const slack = 1 + # because there is overhead in `async` + `sleepAsync` + # as can be seen by increasing `tSleep` from 40 to 49, which increases the number + # of failures. + check counter <= 4 + slack + +for i in 0 .. 10: main() diff --git a/tests/async/t8982.nim b/tests/async/t8982.nim new file mode 100644 index 000000000..5face7edf --- /dev/null +++ b/tests/async/t8982.nim @@ -0,0 +1,33 @@ +discard """ +output: ''' +timeout +runForever should throw ValueError, this is expected +''' +""" + + +import asyncdispatch + +proc failingAwaitable(p: int) {.async.} = + await sleepAsync(500) + if p > 0: + raise newException(Exception, "my exception") + +proc main() {.async.} = + let fut = failingAwaitable(1) + try: + await fut or sleepAsync(100) + if fut.finished: + echo "finished" + else: + echo "timeout" + except: + echo "failed" + + +# Previously this would raise "An attempt was made to complete a Future more than once." +try: + asyncCheck main() + runForever() +except ValueError: + echo "runForever should throw ValueError, this is expected" diff --git a/tests/async/t9201.nim b/tests/async/t9201.nim new file mode 100644 index 000000000..5aaba7063 --- /dev/null +++ b/tests/async/t9201.nim @@ -0,0 +1,14 @@ +discard """ + exitcode: 0 +""" + +# Derived from issue #9201 +import asyncdispatch, macros + +macro newAsyncProc(name: untyped): untyped = + expectKind name, nnkStrLit + let pName = genSym(nskProc, name.strVal) + result = getAst async quote do: + proc `pName`() = discard + +newAsyncProc("hello") diff --git a/tests/async/tacceptcloserace.nim b/tests/async/tacceptcloserace.nim new file mode 100644 index 000000000..fee6537d2 --- /dev/null +++ b/tests/async/tacceptcloserace.nim @@ -0,0 +1,36 @@ +discard """ + exitcode: 0 + output: "" +""" + +import asyncdispatch, net, os, nativesockets + +# bug: https://github.com/nim-lang/Nim/issues/5279 + +proc setupServerSocket(hostname: string, port: Port): AsyncFD = + let fd = createNativeSocket() + if fd == osInvalidSocket: + raiseOSError(osLastError()) + setSockOptInt(fd, SOL_SOCKET, SO_REUSEADDR, 1) + var aiList = getAddrInfo(hostname, port) + if bindAddr(fd, aiList.ai_addr, aiList.ai_addrlen.Socklen) < 0'i32: + freeAddrInfo(aiList) + raiseOSError(osLastError()) + freeAddrInfo(aiList) + if listen(fd) != 0: + raiseOSError(osLastError()) + setBlocking(fd, false) + result = fd.AsyncFD + register(result) + +const port = Port(5614) +for i in 0..100: + let serverFd = setupServerSocket("localhost", port) + serverFd.accept().callback = proc(fut: Future[AsyncFD]) = + if not fut.failed: + fut.read().closeSocket() + + var fd = createAsyncNativeSocket() + waitFor fd.connect("localhost", port) + serverFd.closeSocket() + fd.closeSocket() diff --git a/tests/async/tasyncRecvLine.nim b/tests/async/tasyncRecvLine.nim new file mode 100644 index 000000000..a13a171c3 --- /dev/null +++ b/tests/async/tasyncRecvLine.nim @@ -0,0 +1,53 @@ +discard """ +output: ''' +Hello World +Hello World +''' +""" + +import asyncdispatch, asyncnet + +const recvLinePort = Port(6047) + +proc setupTestServer(): AsyncSocket = + result = newAsyncSocket() + result.setSockOpt(OptReuseAddr, true) + result.bindAddr(recvLinePort) + result.listen() + +proc testUnbuffered(): Future[void] {.async.} = + let serverSock = setupTestServer() + let serverAcceptClientFut = serverSock.accept() + + let clientSock = newAsyncSocket(buffered = false) + let clientConnectFut = clientSock.connect("localhost", recvLinePort) + + let serverAcceptedClient = await serverAcceptClientFut + await clientConnectFut + + await serverAcceptedClient.send("Hello World\c\L") + + echo await clientSock.recvLine() + + clientSock.close() + serverSock.close() + +proc testBuffered(): Future[void] {.async.} = + let serverSock = setupTestServer() + let serverAcceptClientFut = serverSock.accept() + + let clientSock = newAsyncSocket(buffered = true) + let clientConnectFut = clientSock.connect("localhost", recvLinePort) + + let serverAcceptedClient = await serverAcceptClientFut + await clientConnectFut + + await serverAcceptedClient.send("Hello World\c\L") + + echo await clientSock.recvLine() + + clientSock.close() + serverSock.close() + +waitFor testUnbuffered() +waitFor testBuffered() diff --git a/tests/async/tasync_forward.nim b/tests/async/tasync_forward.nim new file mode 100644 index 000000000..99527032f --- /dev/null +++ b/tests/async/tasync_forward.nim @@ -0,0 +1,18 @@ + +import asyncdispatch + +# bug #1970 + +proc foo {.async.} + +proc foo {.async.} = + discard + +# With additional pragmas: +proc bar {.async, cdecl.} + +proc bar {.async.} = + discard + +proc verifyCdeclPresent(p: proc : Future[void] {.cdecl.}) = discard +verifyCdeclPresent(bar) diff --git a/tests/async/tasync_gcsafe.nim b/tests/async/tasync_gcsafe.nim new file mode 100644 index 000000000..bc0eb4271 --- /dev/null +++ b/tests/async/tasync_gcsafe.nim @@ -0,0 +1,36 @@ +discard """ + cmd: "nim c --threads:on $file" + output: ''' +1 +2 +3 +''' +""" + +doAssert compileOption("threads"), "this test will not do anything useful without --threads:on" + +import asyncdispatch + +var globalDummy: ref int +proc gcUnsafeProc() = + if not globalDummy.isNil: + echo globalDummy[] + echo "1" + +proc gcSafeAsyncProcWithNoAnnotation() {.async.} = + echo "2" + +proc gcSafeAsyncProcWithAnnotation() {.gcsafe, async.} = + echo "3" + +proc gcUnsafeAsyncProc() {.async.} = + # We should be able to call gcUnsafe + gcUnsafeProc() + + # We should be able to call async implicitly gcsafe + await gcSafeAsyncProcWithNoAnnotation() + + # We should be able to call async explicitly gcsafe + await gcSafeAsyncProcWithAnnotation() + +waitFor gcUnsafeAsyncProc() diff --git a/tests/async/tasync_gcunsafe.nim b/tests/async/tasync_gcunsafe.nim new file mode 100644 index 000000000..f3e6bc691 --- /dev/null +++ b/tests/async/tasync_gcunsafe.nim @@ -0,0 +1,30 @@ +discard """ + errormsg: "'anotherGCSafeAsyncProc (Async)' is not GC-safe as it calls 'asyncGCUnsafeProc'" + cmd: "nim c --threads:on $file" + file: "asyncmacro.nim" +""" + +doAssert compileOption("threads"), "this test will not do anything useful without --threads:on" + +import asyncdispatch + +var globalDummy: ref int +proc gcUnsafeProc() = + if not globalDummy.isNil: + echo globalDummy[] + +proc asyncExplicitlyGCSafeProc() {.gcsafe, async.} = + echo "hi" + +proc asyncImplicitlyGCSafeProc() {.async.} = + echo "hi" + +proc asyncGCUnsafeProc() {.async.} = + gcUnsafeProc() + +proc anotherGCSafeAsyncProc() {.async, gcsafe.} = + # We should be able to call other gcsafe procs + await asyncExplicitlyGCSafeProc() + await asyncImplicitlyGCSafeProc() + # But we can't call gcunsafe procs + await asyncGCUnsafeProc() diff --git a/tests/async/tasync_in_seq_constr.nim b/tests/async/tasync_in_seq_constr.nim new file mode 100644 index 000000000..3d6dae245 --- /dev/null +++ b/tests/async/tasync_in_seq_constr.nim @@ -0,0 +1,25 @@ +discard """ + output: ''' +@[1, 2, 3, 4] +123 +''' +""" + +# bug #5314, bug #6626 + +import asyncdispatch + +proc bar(i: int): Future[int] {.async.} = + await sleepAsync(2) + result = i + +proc foo(): Future[seq[int]] {.async.} = + await sleepAsync(2) + result = @[1, 2, await bar(3), 4] # <--- The bug is here + +proc foo2() {.async.} = + await sleepAsync(2) + echo(await bar(1), await bar(2), await bar(3)) + +echo waitFor foo() +waitFor foo2() diff --git a/tests/async/tasync_misc.nim b/tests/async/tasync_misc.nim new file mode 100644 index 000000000..ec1418e8c --- /dev/null +++ b/tests/async/tasync_misc.nim @@ -0,0 +1,83 @@ +import json, asyncdispatch +block: #6100 + let done = newFuture[int]() + done.complete(1) + + proc asyncSum: Future[int] {.async.} = + for _ in 1..1_000_000: + result += await done + + let res = waitFor asyncSum() + doAssert(res == 1_000_000) + +block: #7985 + proc getData(): Future[JsonNode] {.async.} = + result = %*{"value": 1} + + type + MyData = object + value: BiggestInt + + proc main() {.async.} = + let data = to(await(getData()), MyData) + doAssert($data == "(value: 1)") + + waitFor(main()) + +block: #8399 + proc bar(): Future[string] {.async.} = discard + + proc foo(line: string) {.async.} = + var res = + case line[0] + of '+', '-': @[] + of '$': (let x = await bar(); @[""]) + else: @[] + + doAssert(res == @[""]) + + waitFor foo("$asd") + +block: # nkCheckedFieldExpr + proc bar(): Future[JsonNode] {.async.} = + return newJInt(5) + + proc foo() {.async.} = + let n = 10 + (await bar()).num + doAssert(n == 15) + + waitFor foo() + +block: # 12743 + + template templ = await sleepAsync 0 + + proc prc {.async.} = templ + + waitFor prc() + +block: # issue #13899 + proc someConnect() {.async.} = + await sleepAsync(1) + proc someClose() {.async.} = + await sleepAsync(2) + proc testFooFails(): Future[bool] {.async.} = + await someConnect() + defer: + await someClose() + result = true + proc testFooSucceed(): Future[bool] {.async.} = + try: + await someConnect() + finally: + await someClose() + result = true + doAssert waitFor testFooSucceed() + doAssert waitFor testFooFails() + +block: # issue #9313 + doAssert compiles(block: + proc a() {.async.} = + echo "Hi" + quit(0) + ) diff --git a/tests/async/tasync_noasync.nim b/tests/async/tasync_noasync.nim new file mode 100644 index 000000000..0927148bf --- /dev/null +++ b/tests/async/tasync_noasync.nim @@ -0,0 +1,44 @@ +discard """ + cmd: "nim check --hints:off --warnings:off $file" + action: "reject" + nimout: ''' +tasync_noasync.nim(21, 10) Error: Can only 'await' inside a proc marked as 'async'. Use 'waitFor' when calling an 'async' proc in a non-async scope instead +tasync_noasync.nim(25, 12) Error: Can only 'await' inside a proc marked as 'async'. Use 'waitFor' when calling an 'async' proc in a non-async scope instead +tasync_noasync.nim(28, 10) Error: Can only 'await' inside a proc marked as 'async'. Use 'waitFor' when calling an 'async' proc in a non-async scope instead +tasync_noasync.nim(31, 10) Error: Can only 'await' inside a proc marked as 'async'. Use 'waitFor' when calling an 'async' proc in a non-async scope instead +tasync_noasync.nim(35, 10) Error: Can only 'await' inside a proc marked as 'async'. Use 'waitFor' when calling an 'async' proc in a non-async scope instead +tasync_noasync.nim(38, 10) Error: Can only 'await' inside a proc marked as 'async'. Use 'waitFor' when calling an 'async' proc in a non-async scope instead +tasync_noasync.nim(40, 8) Error: Can only 'await' inside a proc marked as 'async'. Use 'waitFor' when calling an 'async' proc in a non-async scope instead +''' +""" +import async + +proc a {.async.} = + discard + +# Bad await usage +proc nonAsyncProc = + await a() + +proc nestedNonAsyncProc {.async.} = + proc nested = + await a() + +iterator customIterator: int = + await a() + +macro awaitInMacro = + await a() + +type DummyRef = ref object of RootObj +method awaitInMethod(_: DummyRef) {.base.} = + await a() + +proc improperMultisync {.multisync.} = + await a() + +await a() + +# if we overload a fallback handler to get +# await only available within {.async.} +# we would need `{.dirty.}` templates for await diff --git a/tests/async/tasync_nofuture.nim b/tests/async/tasync_nofuture.nim new file mode 100644 index 000000000..16155601a --- /dev/null +++ b/tests/async/tasync_nofuture.nim @@ -0,0 +1,11 @@ +discard """ + errormsg: "await expects Future[T], got int" + cmd: "nim c $file" + file: "asyncmacro.nim" +""" +import async + +proc a {.async.} = + await 0 + +waitFor a() diff --git a/tests/async/tasync_traceback.nim b/tests/async/tasync_traceback.nim new file mode 100644 index 000000000..98f71b192 --- /dev/null +++ b/tests/async/tasync_traceback.nim @@ -0,0 +1,122 @@ +discard """ + exitcode: 0 + output: "Matched" +""" +import asyncdispatch, strutils + +# Tests to ensure our exception trace backs are friendly. + +# --- Simple test. --- +# +# What does this look like when it's synchronous? +# +# tasync_traceback.nim(23) tasync_traceback +# tasync_traceback.nim(21) a +# tasync_traceback.nim(18) b +# Error: unhandled exception: b failure [OSError] +# +# Good (not quite ideal, but gotta work within constraints) traceback, +# when exception is unhandled: +# +# <traceback for the unhandled exception> +# <very much a bunch of noise> +# <would be ideal to customise this> +# <(the code responsible is in excpt:raiseExceptionAux)> +# Error: unhandled exception: b failure +# =============== +# Async traceback +# =============== +# +# tasync_traceback.nim(23) tasync_traceback +# +# tasync_traceback.nim(21) a +# tasync_traceback.nim(18) b + +var result = "" + +proc b(): Future[int] {.async.} = + if true: + raise newException(OSError, "b failure") + +proc a(): Future[int] {.async.} = + return await b() + +let aFut = a() +try: + discard waitFor aFut +except Exception as exc: + result.add(exc.msg & "\n") +result.add("\n") + +# From #6803 +proc bar(): Future[string] {.async.} = + await sleepAsync(100) + if true: + raise newException(OSError, "bar failure") + +proc foo(): Future[string] {.async.} = return await bar() + +try: + result.add(waitFor(foo()) & "\n") +except Exception as exc: + result.add(exc.msg & "\n") +result.add("\n") + +# Use re to parse the result +import re +const expected = """ +b failure +Async traceback: + tasync_traceback\.nim\(\d+?\) tasync_traceback + tasync_traceback\.nim\(\d+?\) a \(Async\) + tasync_traceback\.nim\(\d+?\) b \(Async\) +Exception message: b failure + + +bar failure +Async traceback: + tasync_traceback\.nim\(\d+?\) tasync_traceback + asyncdispatch\.nim\(\d+?\) waitFor + asyncdispatch\.nim\(\d+?\) poll + ## Processes asynchronous completion events + asyncdispatch\.nim\(\d+?\) runOnce + asyncdispatch\.nim\(\d+?\) processPendingCallbacks + ## Executes pending callbacks + tasync_traceback\.nim\(\d+?\) bar \(Async\) +Exception message: bar failure + +""" + +# TODO: is asyncmacro good enough location for fooIter traceback/debugging? just put the callsite info for all? + +let resLines = splitLines(result.strip) +let expLines = splitLines(expected.strip) + +when not defined(cpp): # todo fixme + if resLines.len != expLines.len: + echo("Not matched! Wrong number of lines!") + echo expLines.len + echo resLines.len + echo("Expected: -----------") + echo expected + echo("Gotten: -------------") + echo result + echo("---------------------") + quit(QuitFailure) + + var ok = true + for i in 0 ..< resLines.len: + if not resLines[i].match(re(expLines[i])): + echo "Not matched! Line ", i + 1 + echo "Expected:" + echo expLines[i] + echo "Actual:" + echo resLines[i] + ok = false + + if ok: + echo("Matched") + else: + quit(QuitFailure) +else: + echo("Matched") diff --git a/tests/async/tasyncall.nim b/tests/async/tasyncall.nim new file mode 100644 index 000000000..3c318dbf7 --- /dev/null +++ b/tests/async/tasyncall.nim @@ -0,0 +1,93 @@ +discard """ + exitcode: 0 +""" +import times, sequtils +import asyncdispatch + +const + taskCount = 10 + sleepDuration = 50 + +proc futureWithValue(x: int): Future[int] {.async.} = + await sleepAsync(sleepDuration) + return x + +proc futureWithoutValue() {.async.} = + await sleepAsync(sleepDuration) + +proc testFuturesWithValue(x: int): seq[int] = + var tasks = newSeq[Future[int]](taskCount) + + for i in 0..<taskCount: + tasks[i] = futureWithValue(x) + + result = waitFor all(tasks) + +proc testFuturesWithoutValues() = + var tasks = newSeq[Future[void]](taskCount) + + for i in 0..<taskCount: + tasks[i] = futureWithoutValue() + + waitFor all(tasks) + +proc testVarargs(x, y, z: int): seq[int] = + let + a = futureWithValue(x) + b = futureWithValue(y) + c = futureWithValue(z) + + result = waitFor all(a, b, c) + +proc testWithDupes() = + var + tasks = newSeq[Future[void]](taskCount) + fut = futureWithoutValue() + + for i in 0..<taskCount: + tasks[i] = fut + + waitFor all(tasks) + +block: + let + startTime = cpuTime() + results = testFuturesWithValue(42) + expected = repeat(42, taskCount) + execTime = cpuTime() - startTime + + doAssert execTime * 1000 < taskCount * sleepDuration + doAssert results == expected + +block: + let startTime = cpuTime() + testFuturesWithoutValues() + let execTime = cpuTime() - startTime + + doAssert execTime * 1000 < taskCount * sleepDuration + +block: + let startTime = cpuTime() + testWithDupes() + let execTime = cpuTime() - startTime + + doAssert execTime * 1000 < taskCount * sleepDuration + +block: + let + startTime = cpuTime() + results = testVarargs(1, 2, 3) + expected = @[1, 2, 3] + execTime = cpuTime() - startTime + + doAssert execTime * 100 < taskCount * sleepDuration + doAssert results == expected + +block: + let + noIntFuturesFut = all(newSeq[Future[int]]()) + noVoidFuturesFut = all(newSeq[Future[void]]()) + + doAssert noIntFuturesFut.finished and not noIntFuturesFut.failed + doAssert noVoidFuturesFut.finished and not noVoidFuturesFut.failed + doAssert noIntFuturesFut.read() == @[] diff --git a/tests/async/tasyncawait.nim b/tests/async/tasyncawait.nim new file mode 100644 index 000000000..e86542b2d --- /dev/null +++ b/tests/async/tasyncawait.nim @@ -0,0 +1,56 @@ +import asyncdispatch, asyncnet, nativesockets, net, strutils +from stdtest/netutils import bindAvailablePort +var msgCount = 0 + +const + swarmSize = 40 + messagesToSend = 50 + +var clientCount = 0 + +proc sendMessages(client: AsyncFD) {.async.} = + for i in 0 ..< messagesToSend: + await send(client, "Message " & $i & "\c\L") + +proc launchSwarm(port: Port) {.async.} = + for i in 0 ..< swarmSize: + var sock = createAsyncNativeSocket() + + await connect(sock, "localhost", port) + await sendMessages(sock) + closeSocket(sock) + +proc readMessages(client: AsyncFD) {.async.} = + # wrapping the AsyncFd into a AsyncSocket object + var sockObj = newAsyncSocket(client) + var (ipaddr, port) = sockObj.getPeerAddr() + doAssert ipaddr == "127.0.0.1" + (ipaddr, port) = sockObj.getLocalAddr() + doAssert ipaddr == "127.0.0.1" + while true: + var line = await recvLine(sockObj) + if line == "": + closeSocket(client) + clientCount.inc + break + else: + if line.startsWith("Message "): + msgCount.inc + else: + doAssert false + +proc createServer(server: AsyncFD) {.async.} = + discard server.SocketHandle.listen() + while true: + asyncCheck readMessages(await accept(server)) + +let server = createAsyncNativeSocket() +let port = bindAvailablePort(server.SocketHandle) +asyncCheck createServer(server) +asyncCheck launchSwarm(port) +while true: + poll() + if clientCount == swarmSize: break + +doAssert msgCount == swarmSize * messagesToSend +doAssert msgCount == 2000 diff --git a/tests/async/tasyncclosestall.nim b/tests/async/tasyncclosestall.nim new file mode 100644 index 000000000..d1c7a5fba --- /dev/null +++ b/tests/async/tasyncclosestall.nim @@ -0,0 +1,101 @@ +discard """ + disabled: "windows" + outputsub: "send has errored. As expected. All good!" + exitcode: 0 +""" +import asyncdispatch, asyncnet + +when defined(windows): + from winlean import ERROR_NETNAME_DELETED +else: + from posix import EBADF + +# This reproduces a case where a socket remains stuck waiting for writes +# even when the socket is closed. +const + timeout = 8000 +var port = Port(0) + +var sent = 0 + +proc keepSendingTo(c: AsyncSocket) {.async.} = + while true: + # This write will eventually get stuck because the client is not reading + # its messages. + let sendFut = c.send("Foobar" & $sent & "\n", flags = {}) + if not await withTimeout(sendFut, timeout): + # The write is stuck. Let's simulate a scenario where the socket + # does not respond to PING messages, and we close it. The above future + # should complete after the socket is closed, not continue stalling. + echo("Socket has stalled, closing it") + c.close() + + let timeoutFut = withTimeout(sendFut, timeout) + yield timeoutFut + if timeoutFut.failed: + let errCode = ((ref OSError)(timeoutFut.error)).errorCode + # The behaviour differs across platforms. On Windows ERROR_NETNAME_DELETED + # is raised which we classif as a "diconnection error", hence we overwrite + # the flags above in the `send` call so that this error is raised. + # + # On Linux the EBADF error code is raised, this is because the socket + # is closed. + # + # This means that by default the behaviours will differ between Windows + # and Linux. I think this is fine though, it makes sense mainly because + # Windows doesn't use a IO readiness model. We can fix this later if + # necessary to reclassify ERROR_NETNAME_DELETED as not a "disconnection + # error" (TODO) + when defined(windows): + if errCode == ERROR_NETNAME_DELETED: + echo("send has errored. As expected. All good!") + quit(QuitSuccess) + else: + raise newException(ValueError, "Test failed. Send failed with code " & $errCode) + else: + if errCode == EBADF: + echo("send has errored. As expected. All good!") + quit(QuitSuccess) + else: + raise newException(ValueError, "Test failed. Send failed with code " & $errCode) + + # The write shouldn't succeed and also shouldn't be stalled. + if timeoutFut.read(): + raise newException(ValueError, "Test failed. Send was expected to fail.") + else: + raise newException(ValueError, "Test failed. Send future is still stalled.") + sent.inc(1) + +proc startClient() {.async.} = + let client = newAsyncSocket() + await client.connect("localhost", port) + echo("Connected") + + let firstLine = await client.recvLine() + echo("Received first line as a client: ", firstLine) + echo("Now not reading anymore") + while true: await sleepAsync(1000) + +proc debug() {.async.} = + while true: + echo("Sent ", sent) + await sleepAsync(1000) + +proc server() {.async.} = + var s = newAsyncSocket() + s.setSockOpt(OptReuseAddr, true) + s.bindAddr(port) + s.listen() + let (addr2, port2) = s.getLocalAddr + port = port2 + + # We're now ready to accept connections, so start the client + asyncCheck startClient() + asyncCheck debug() + + while true: + let client = await accept(s) + asyncCheck keepSendingTo(client) + +when isMainModule: + waitFor server() diff --git a/tests/async/tasyncconnect.nim b/tests/async/tasyncconnect.nim new file mode 100644 index 000000000..564f6c67c --- /dev/null +++ b/tests/async/tasyncconnect.nim @@ -0,0 +1,33 @@ +discard """ + outputsub: "Error: unhandled exception: Connection refused" + exitcode: 1 +""" + +import + asyncdispatch, + posix + + +const + testHost = "127.0.0.1" + testPort = Port(17357) + + +when defined(windows) or defined(nimdoc): + # TODO: just make it work on Windows for now. + quit("Error: unhandled exception: Connection refused") +else: + proc testAsyncConnect() {.async.} = + var s = createAsyncNativeSocket() + + await s.connect(testHost, testPort) + + var peerAddr: SockAddr + var addrSize = Socklen(sizeof(peerAddr)) + var ret = SocketHandle(s).getpeername(addr(peerAddr), addr(addrSize)) + + if ret < 0: + echo("`connect(...)` failed but no exception was raised.") + quit(2) + + waitFor(testAsyncConnect()) diff --git a/tests/async/tasyncdial.nim b/tests/async/tasyncdial.nim new file mode 100644 index 000000000..f0377dfd5 --- /dev/null +++ b/tests/async/tasyncdial.nim @@ -0,0 +1,52 @@ +discard """ + output: ''' +OK AF_INET +OK AF_INET6 +''' +""" + +import + nativesockets, os, asyncdispatch + +proc setupServerSocket(hostname: string, port: Port, domain: Domain): AsyncFD = + ## Creates a socket, binds it to the specified address, and starts listening for connections. + ## Registers the descriptor with the dispatcher of the current thread + ## Raises OSError in case of an error. + let fd = createNativeSocket(domain) + setSockOptInt(fd, SOL_SOCKET, SO_REUSEADDR, 1) + var aiList = getAddrInfo(hostname, port, domain) + if bindAddr(fd, aiList.ai_addr, aiList.ai_addrlen.Socklen) < 0'i32: + freeAddrInfo(aiList) + raiseOSError(osLastError()) + freeAddrInfo(aiList) + if listen(fd) != 0: + raiseOSError(osLastError()) + setBlocking(fd, false) + result = fd.AsyncFD + register(result) + +proc doTest(domain: static[Domain]) {.async.} = + const + testHost = when domain == Domain.AF_INET6: "::1" else: "127.0.0.1" + testPort = Port(17384) + let serverFd = setupServerSocket(testHost, testPort, domain) + let acceptFut = serverFd.accept() + let clientFdFut = dial(testHost, testPort) + + let serverClientFd = await acceptFut + serverFd.closeSocket() + + let clientFd = await clientFdFut + + let recvFut = serverClientFd.recv(2) + await clientFd.send("Hi") + let msg = await recvFut + + serverClientFd.closeSocket() + clientFd.closeSocket() + + if msg == "Hi": + echo "OK ", domain + +waitFor(doTest(Domain.AF_INET)) +waitFor(doTest(Domain.AF_INET6)) diff --git a/tests/async/tasyncdiscard.nim b/tests/async/tasyncdiscard.nim new file mode 100644 index 000000000..64e6021c3 --- /dev/null +++ b/tests/async/tasyncdiscard.nim @@ -0,0 +1,39 @@ +discard """ + output: ''' +1 +2 +3 +4 +1 +2 +1 +6 +''' +""" +import asyncdispatch, asyncnet + +proc main {.async.} = + proc f: Future[int] {.async.} = + discard + echo 1 + discard + result = 2 + discard + + let x = await f() + echo x + echo 3 + + proc g: Future[int] {.async.} = + discard + echo 4 + discard + result = 6 + discard + echo await f() + discard await f() + + discard await g() + echo 6 + +waitFor(main()) diff --git a/tests/async/tasynceagain.nim b/tests/async/tasynceagain.nim new file mode 100644 index 000000000..94c3645dc --- /dev/null +++ b/tests/async/tasynceagain.nim @@ -0,0 +1,67 @@ +discard """ + disabled: "windows" + exitcode: 0 +""" +# AsyncSocketBug.nim +# Jens Alfke (@snej) -- 16 July 2020 +# Demonstrates data loss by Nim's AsyncSocket. +# Just run it, and it will raise an assertion failure within a minute. + +import asyncdispatch, asyncnet, strformat, strutils, sugar + +const FrameSize = 9999 # Exact size not important, but larger sizes fail quicker + +proc runServer() {.async.} = + # Server side: + var server = newAsyncSocket() + server.bindAddr(Port(9001)) + server.listen() + let client = await server.accept() + echo "Server got client connection" + var lastN = 0 + while true: + let frame = await client.recv(FrameSize) + doAssert frame.len == FrameSize + let n = frame[0..<6].parseInt() + echo "RCVD #", n, ": ", frame[0..80], "..." + if n != lastN + 1: + echo &"******** ERROR: Server received #{n}, but last was #{lastN}!" + doAssert n == lastN + 1 + lastN = n + await sleepAsync 100 + + +proc main() {.async.} = + asyncCheck runServer() + + # Client side: + let socket = newAsyncSocket(buffered = false) + await socket.connect("localhost", Port(9001)) + echo "Client socket connected" + + var sentCount = 0 + var completedCount = 0 + + while sentCount < 2000: + sentCount += 1 + let n = sentCount + + var message = &"{n:06} This is message #{n} of â. Please stay tuned for more. " + #echo ">>> ", message + while message.len < FrameSize: + message = message & message + let frame = message[0..<FrameSize] + + capture n: + socket.send(frame).addCallback proc(f: Future[void]) = + # Callback when the send completes: + assert not f.failed + echo "SENT #", n + if n != completedCount + 1: + echo &"******** ERROR: Client completed #{n}, but last completed was #{completedCount}!" + # If this assert is enabled, it will trigger earlier than the server-side assert above: + assert n == completedCount + 1 + completedCount = n + await sleepAsync 1 + +waitFor main() \ No newline at end of file diff --git a/tests/async/tasyncexceptions.nim b/tests/async/tasyncexceptions.nim new file mode 100644 index 000000000..de61c099d --- /dev/null +++ b/tests/async/tasyncexceptions.nim @@ -0,0 +1,40 @@ +discard """ + outputsub: "Error: unhandled exception: foobar" + exitcode: 1 +""" +import asyncdispatch + +# Note: This is a test case for a bug. + +proc accept(): Future[int] {.async.} = + await sleepAsync(100) + result = 4 + +proc recvLine(fd: int): Future[string] {.async.} = + await sleepAsync(100) + return "get" + +proc processClient(fd: int) {.async.} = + # these finish synchronously, we need some async delay to emulate this bug. + var line = await recvLine(fd) + var foo = line[0] + if foo == 'g': + raise newException(Exception, "foobar") + +proc serve() {.async.} = + + while true: + var fut = await accept() + await processClient(fut) + +when true: + proc main = + var fut = serve() + fut.callback = + proc () = + if fut.failed: + # This test ensures that this exception crashes the application + # as it is not handled. + raise fut.error + runForever() + main() diff --git a/tests/async/tasyncfile.nim b/tests/async/tasyncfile.nim new file mode 100644 index 000000000..d95850c31 --- /dev/null +++ b/tests/async/tasyncfile.nim @@ -0,0 +1,61 @@ +discard """ +output: ''' +13 +hello humans! +13 +''' +""" +import asyncfile, asyncdispatch, os + +proc main() {.async.} = + let fn = getTempDir() / "foobar.txt" + removeFile(fn) + + # Simple write/read test. + block: + var file = openAsync(fn, fmReadWrite) + await file.write("testing") + file.setFilePos(0) + await file.write("foo") + file.setFileSize(4) + file.setFilePos(0) + let data = await file.readAll() + doAssert data == "foot" + file.close() + + # Append test + block: + var file = openAsync(fn, fmAppend) + await file.write("\ntest2") + let errorTest = file.readAll() + yield errorTest + doAssert errorTest.failed + file.close() + file = openAsync(fn, fmRead) + let data = await file.readAll() + + doAssert data == "foot\ntest2" + file.close() + + # Issue #5531 + block: + removeFile(fn) + var file = openAsync(fn, fmWrite) + await file.write("test2") + file.close() + file = openAsync(fn, fmWrite) + await file.write("t3") + file.close() + file = openAsync(fn, fmRead) + let data = await file.readAll() + doAssert data == "t3" + file.close() + + # Issue #7347 + block: + var file = openAsync( parentDir(currentSourcePath) / "hello.txt") + echo file.getFileSize() + echo await file.readAll() + echo file.getFilePos() + +waitFor main() diff --git a/tests/async/tasyncfilewrite.nim b/tests/async/tasyncfilewrite.nim new file mode 100644 index 000000000..72a2df0b0 --- /dev/null +++ b/tests/async/tasyncfilewrite.nim @@ -0,0 +1,20 @@ +discard """ + output: '''string 1 +string 2 +string 3 +''' +""" +# bug #5532 +import os, asyncfile, asyncdispatch + +const F = "test_async.txt" + +removeFile(F) +let f = openAsync(F, fmWrite) +var futs = newSeq[Future[void]]() +for i in 1..3: + futs.add(f.write("string " & $i & "\n")) +waitFor(all(futs)) +f.close() +echo readFile(F) +removeFile(F) diff --git a/tests/async/tasyncintemplate.nim b/tests/async/tasyncintemplate.nim new file mode 100644 index 000000000..4bddb1d18 --- /dev/null +++ b/tests/async/tasyncintemplate.nim @@ -0,0 +1,62 @@ +discard """ + output: ''' +42 +43 +43 +1 +2 +3 +4 +''' +""" + +# xxx move to tests/async/tasyncintemplate.nim +import asyncdispatch + +block: # bug #16159 + template foo() = + proc temp(): Future[int] {.async.} = return 42 + proc tempVoid(): Future[void] {.async.} = echo await temp() + foo() + waitFor tempVoid() + +block: # aliasing `void` + template foo() = + type Foo = void + proc temp(): Future[int] {.async.} = return 43 + proc tempVoid(): Future[Foo] {.async.} = echo await temp() + proc tempVoid2() {.async.} = echo await temp() + foo() + waitFor tempVoid() + waitFor tempVoid2() + +block: # sanity check + template foo() = + proc bad(): int {.async.} = discard + doAssert not compiles(bad()) + +block: # bug #16786 + block: + proc main(a: int|string)= + proc bar(b: int|string) = echo b + bar(a) + main(1) + + block: + proc main(a: int) : Future[void] {.async.} = + proc bar(b: int): Future[void] {.async.} = echo b + await bar(a) + waitFor main(2) + + block: + proc main(a: int) : Future[void] {.async.} = + proc bar(b: int | string): Future[void] {.async.} = echo b + await bar(a) + waitFor main(3) + + block: + # bug + proc main(a: int|string) = + proc bar(b: int): Future[void] {.async.} = echo b + waitFor bar(a) + main(4) diff --git a/tests/async/tasyncnetudp.nim b/tests/async/tasyncnetudp.nim new file mode 100644 index 000000000..dade96fb2 --- /dev/null +++ b/tests/async/tasyncnetudp.nim @@ -0,0 +1,90 @@ +# It is a reproduction of the 'tnewasyncudp' test code, but using a high level +# of asynchronous procedures. Output: "5000" +import asyncdispatch, asyncnet, nativesockets, net, strutils + +var msgCount = 0 +var recvCount = 0 + +const + messagesToSend = 100 + swarmSize = 50 + serverPort = 10333 + +var + sendports = 0 + recvports = 0 + +proc saveSendingPort(port: Port) = + sendports = sendports + int(port) + +proc saveReceivedPort(port: Port) = + recvports = recvports + int(port) + +proc launchSwarm(serverIp: string, serverPort: Port) {.async.} = + var i = 0 + + while i < swarmSize: + var sock = newAsyncSocket(nativesockets.AF_INET, nativesockets.SOCK_DGRAM, + Protocol.IPPROTO_UDP, false) + + bindAddr(sock, address = "127.0.0.1") + + let (null, localPort) = getLocalAddr(sock) + + var k = 0 + + while k < messagesToSend: + let message = "Message " & $(i * messagesToSend + k) + + await asyncnet.sendTo(sock, serverIp, serverPort, message) + + let (data, fromIp, fromPort) = await recvFrom(sock, 16384) + + if data == message: + saveSendingPort(localPort) + + inc(recvCount) + + inc(k) + + close(sock) + + inc(i) + +proc readMessages(server: AsyncSocket) {.async.} = + let maxResponses = (swarmSize * messagesToSend) + + var i = 0 + + while i < maxResponses: + let (data, fromIp, fromPort) = await recvFrom(server, 16384) + + if data.startsWith("Message ") and fromIp == "127.0.0.1": + await sendTo(server, fromIp, fromPort, data) + + inc(msgCount) + + saveReceivedPort(fromPort) + + inc(i) + +proc createServer() {.async.} = + var server = newAsyncSocket(nativesockets.AF_INET, nativesockets.SOCK_DGRAM, Protocol.IPPROTO_UDP, false) + + bindAddr(server, Port(serverPort), "127.0.0.1") + + asyncCheck readMessages(server) + +asyncCheck createServer() +asyncCheck launchSwarm("127.0.0.1", Port(serverPort)) + +while true: + poll() + + if recvCount == swarmSize * messagesToSend: + break + +doAssert msgCount == swarmSize * messagesToSend +doAssert sendports == recvports + +echo msgCount \ No newline at end of file diff --git a/tests/async/tasyncrecursion.nim b/tests/async/tasyncrecursion.nim new file mode 100644 index 000000000..7c12dbb0e --- /dev/null +++ b/tests/async/tasyncrecursion.nim @@ -0,0 +1,21 @@ +discard """ +output: "50005000" +""" +import asyncdispatch + +proc asyncRecursionCycle*(counter: int): Future[int] = + var retFuture = newFuture[int]("asyncRecursionTest") + retFuture.complete(counter + 1) + return retFuture + +proc asyncRecursionTest*(): Future[int] {.async.} = + var i = 0 + result = 0 + while i < 10_000: + inc(result, await asyncRecursionCycle(i)) + inc(i) + +when true: + setGlobalDispatcher(newDispatcher()) + var i = waitFor asyncRecursionTest() + echo i diff --git a/tests/async/tasyncsend4757.nim b/tests/async/tasyncsend4757.nim new file mode 100644 index 000000000..29873a905 --- /dev/null +++ b/tests/async/tasyncsend4757.nim @@ -0,0 +1,24 @@ +import asyncdispatch, asyncnet + +var port: Port +proc createServer() {.async.} = + var server = newAsyncSocket() + server.setSockOpt(OptReuseAddr, true) + bindAddr(server) + port = getLocalAddr(server)[1] + server.listen() + while true: + let client = await server.accept() + discard await client.recvLine() + +asyncCheck createServer() + +var done = false +proc f(): Future[void] {.async.} = + let s = createAsyncNativeSocket() + await s.connect("localhost", port) + await s.send("123") + done = true + +waitFor f() +doAssert done diff --git a/tests/async/tasyncssl.nim b/tests/async/tasyncssl.nim new file mode 100644 index 000000000..57de3271d --- /dev/null +++ b/tests/async/tasyncssl.nim @@ -0,0 +1,74 @@ +discard """ + cmd: "nim $target --hints:on --define:ssl $options $file" + disabled: osx +""" + +import asyncdispatch, asyncnet, net, strutils +import stdtest/testutils + +when defined(ssl): + var port0: Port + var msgCount = 0 + + const + swarmSize = 10 + messagesToSend = 50 + + var clientCount = 0 + + proc sendMessages(client: AsyncSocket) {.async.} = + for i in 0 ..< messagesToSend: + await send(client, "Message " & $i & "\c\L") + + proc launchSwarm(port: Port) {.async.} = + for i in 0 ..< swarmSize: + var sock = newAsyncSocket() + var clientContext = newContext(verifyMode = CVerifyNone) + clientContext.wrapSocket(sock) + await connect(sock, "localhost", port) + await sendMessages(sock) + close(sock) + + proc readMessages(client: AsyncSocket) {.async.} = + while true: + var line = await recvLine(client) + if line == "": + close(client) + inc(clientCount) + break + else: + if line.startsWith("Message "): + inc(msgCount) + else: + doAssert false + + proc createServer() {.async.} = + let serverContext = newContext(verifyMode = CVerifyNone, + certFile = "tests/testdata/mycert.pem", + keyFile = "tests/testdata/mycert.pem") + var server = newAsyncSocket() + serverContext.wrapSocket(server) + server.setSockOpt(OptReuseAddr, true) + bindAddr(server) + port0 = getLocalAddr(server)[1] + server.listen() + while true: + let client = await accept(server) + serverContext.wrapConnectedSocket(client, handshakeAsServer) + asyncCheck readMessages(client) + + asyncCheck createServer() + asyncCheck launchSwarm(port0) + while true: + poll() + if clientCount == swarmSize: break + + template cond(): bool = msgCount == swarmSize * messagesToSend + when defined(windows): + # currently: msgCount == 0 + flakyAssert cond() + elif defined(linux) and int.sizeof == 8: + # currently: msgCount == 10 + flakyAssert cond() + doAssert msgCount > 0 + else: doAssert cond(), $msgCount diff --git a/tests/async/tasynctry.nim b/tests/async/tasynctry.nim new file mode 100644 index 000000000..25eab87fb --- /dev/null +++ b/tests/async/tasynctry.nim @@ -0,0 +1,118 @@ +discard """ +output: ''' +Generic except: Test +Specific except +Multiple idents in except +Multiple except branches +Multiple except branches 2 +success +''' +targets: "c" +""" +import asyncdispatch, strutils + +# Here we are testing the ability to catch exceptions. + +proc foobar() {.async.} = + if 5 == 5: + raise newException(IndexDefect, "Test") + +proc catch() {.async.} = + # TODO: Create a test for when exceptions are not caught. + try: + await foobar() + except: + echo("Generic except: ", getCurrentExceptionMsg().splitLines[0]) + + try: + await foobar() + except IndexDefect: + echo("Specific except") + + try: + await foobar() + except OSError, FieldDefect, IndexDefect: + echo("Multiple idents in except") + + try: + await foobar() + except OSError, FieldDefect: + assert false + except IndexDefect: + echo("Multiple except branches") + + try: + await foobar() + except IndexDefect: + echo("Multiple except branches 2") + except OSError, FieldDefect: + assert false + +waitFor catch() + +proc test(): Future[bool] {.async.} = + result = false + try: + raise newException(OSError, "Foobar") + except: + result = true + return + +proc foo(): Future[bool] {.async.} = discard + +proc test2(): Future[bool] {.async.} = + result = false + try: + discard await foo() + raise newException(OSError, "Foobar") + except: + result = true + return + +proc test3(): Future[int] {.async.} = + result = 0 + try: + try: + discard await foo() + raise newException(OSError, "Hello") + except: + result = 1 + raise + except: + result = 2 + return + +proc test4(): Future[int] {.async.} = + try: + discard await foo() + raise newException(ValueError, "Test4") + except OSError: + result = 1 + except: + result = 2 + +var x = test() +assert x.waitFor() + +x = test2() +assert x.waitFor() + +var y = test3() +assert y.waitFor() == 2 + +y = test4() +assert y.waitFor() == 2 + +# bug #14279 + +proc expandValue: Future[int] {.async.} = + return 0 + +proc a(b: int): Future[void] {.async.} = + return + +proc b: Future[void] {.async.} = + await a(await expandValue()) + echo "success" + +waitFor(b()) diff --git a/tests/async/tawaitsemantics.nim b/tests/async/tawaitsemantics.nim new file mode 100644 index 000000000..67903cc5e --- /dev/null +++ b/tests/async/tawaitsemantics.nim @@ -0,0 +1,95 @@ +discard """ +output: ''' +Error can be caught using yield +Infix `or` raises +Infix `and` raises +All() raises +Awaiting a async procedure call raises +Awaiting a future raises +''' +""" + +import asyncdispatch + +# This tests the behaviour of 'await' under different circumstances. +# Specifically, when an awaited future raises an exception then `await` should +# also raise that exception by `read`'ing that future. In cases where you don't +# want this behaviour, you can use `yield`. +# https://github.com/nim-lang/Nim/issues/4170 + +proc thrower(): Future[void] = + result = newFuture[void]() + result.fail(newException(Exception, "Test")) + +proc dummy: Future[void] = + result = newFuture[void]() + result.complete() + +proc testInfixOr() {.async.} = + # Test the infix `or` operator semantics. + var fut = thrower() + var fut2 = dummy() + await fut or fut2 # Should raise! + +proc testInfixAnd() {.async.} = + # Test the infix `and` operator semantics. + var fut = thrower() + var fut2 = dummy() + await fut and fut2 # Should raise! + +proc testAll() {.async.} = + # Test the `all` semantics. + var fut = thrower() + var fut2 = dummy() + await all(fut, fut2) # Should raise! + +proc testCall() {.async.} = + await thrower() + +proc testAwaitFut() {.async.} = + var fut = thrower() + await fut # This should raise. + +proc tester() {.async.} = + # Test that we can handle exceptions without 'try' + var fut = thrower() + doAssert fut.finished + doAssert fut.failed + doAssert fut.error.msg == "Test" + yield fut # We are yielding a 'Future', so no `read` occurs. + doAssert fut.finished + doAssert fut.failed + doAssert fut.error.msg == "Test" + echo("Error can be caught using yield") + + fut = testInfixOr() + yield fut + doAssert fut.finished + doAssert fut.failed + echo("Infix `or` raises") + + fut = testInfixAnd() + yield fut + doAssert fut.finished + doAssert fut.failed + echo("Infix `and` raises") + + fut = testAll() + yield fut + doAssert fut.finished + doAssert fut.failed + echo("All() raises") + + fut = testCall() + yield fut + doAssert fut.failed + echo("Awaiting a async procedure call raises") + + # Test that await will read the future and raise an exception. + fut = testAwaitFut() + yield fut + doAssert fut.failed + echo("Awaiting a future raises") + + +waitFor(tester()) diff --git a/tests/async/tbreak_must_exec_finally.nim b/tests/async/tbreak_must_exec_finally.nim new file mode 100644 index 000000000..8780e6149 --- /dev/null +++ b/tests/async/tbreak_must_exec_finally.nim @@ -0,0 +1,25 @@ +discard """ + output: ''' +finally handler 8 +do not duplicate this one +''' +""" + +# bug #15243 + +import asyncdispatch + +proc f() {.async.} = + try: + while true: + try: + await sleepAsync(400) + break + finally: + var localHere = 8 + echo "finally handler ", localHere + finally: + echo "do not duplicate this one" + +when isMainModule: + waitFor f() diff --git a/tests/async/tcallbacks.nim b/tests/async/tcallbacks.nim new file mode 100644 index 000000000..bd82d5824 --- /dev/null +++ b/tests/async/tcallbacks.nim @@ -0,0 +1,21 @@ +discard """ + exitcode: 0 + output: ''' +1 +2 +3 +5 +''' +""" +import asyncfutures + +let f1: Future[int] = newFuture[int]() +f1.addCallback(proc() = echo 1) +f1.addCallback(proc() = echo 2) +f1.addCallback(proc() = echo 3) +f1.complete(10) + +let f2: Future[int] = newFuture[int]() +f2.addCallback(proc() = echo 4) +f2.callback = proc() = echo 5 +f2.complete(10) diff --git a/tests/async/tdiscardableproc.nim b/tests/async/tdiscardableproc.nim new file mode 100644 index 000000000..93cd83be9 --- /dev/null +++ b/tests/async/tdiscardableproc.nim @@ -0,0 +1,9 @@ +discard """ + errormsg: "Cannot make async proc discardable. Futures have to be checked with `asyncCheck` instead of discarded" +""" + +import async + +proc foo {.async, discardable.} = discard + +foo() diff --git a/tests/async/testmanyasyncevents.nim b/tests/async/testmanyasyncevents.nim new file mode 100644 index 000000000..9fdd01b4f --- /dev/null +++ b/tests/async/testmanyasyncevents.nim @@ -0,0 +1,24 @@ +discard """ +output: ''' +hasPendingOperations: false +triggerCount: 100 +''' +disabled: "windows" +""" + +import asyncDispatch + +var triggerCount = 0 +var evs = newSeq[AsyncEvent]() + +for i in 0 ..< 100: # has to be lower than the typical physical fd limit + var ev = newAsyncEvent() + evs.add(ev) + addEvent(ev, proc(fd: AsyncFD): bool {.gcsafe,closure.} = triggerCount += 1; true) + +for ev in evs: + ev.trigger() + +drain() +echo "hasPendingOperations: ", hasPendingOperations() +echo "triggerCount: ", triggerCount diff --git a/tests/async/tfuturestream.nim b/tests/async/tfuturestream.nim new file mode 100644 index 000000000..a019df400 --- /dev/null +++ b/tests/async/tfuturestream.nim @@ -0,0 +1,71 @@ +discard """ +output: ''' +0 +1 +2 +3 +4 +5 +Done +Finished +''' +""" +import asyncdispatch + +var fs = newFutureStream[int]() + +proc alpha() {.async.} = + for i in 0 .. 5: + await fs.write(i) + await sleepAsync(100) + + echo("Done") + fs.complete() + +proc beta() {.async.} = + while not fs.finished: + let (hasValue, value) = await fs.read() + if hasValue: + echo(value) + + echo("Finished") + +asyncCheck alpha() +waitFor beta() + +template ensureCallbacksAreScheduled = + # callbacks are called directly if the dispatcher is not running + discard getGlobalDispatcher() + +proc testCompletion() {.async.} = + ensureCallbacksAreScheduled + + var stream = newFutureStream[string]() + + for i in 1..5: + await stream.write($i) + + var readFuture = stream.readAll() + stream.complete() + yield readFuture + let data = readFuture.read() + doAssert(data.len == 5, "actual data len = " & $data.len) + +waitFor testCompletion() + +# TODO: Something like this should work eventually. +# proc delta(): FutureStream[string] {.async.} = +# for i in 0 .. 5: +# await sleepAsync(1000) +# result.put($i) + +# return "" + +# proc omega() {.async.} = +# let fut = delta() +# while not fut.finished(): +# echo(await fs.takeAsync()) + +# echo("Finished") + +# waitFor omega() diff --git a/tests/async/tfuturevar.nim b/tests/async/tfuturevar.nim new file mode 100644 index 000000000..b70f1d166 --- /dev/null +++ b/tests/async/tfuturevar.nim @@ -0,0 +1,46 @@ +import asyncdispatch + +proc completeOnReturn(fut: FutureVar[string], x: bool) {.async.} = + if x: + fut.mget() = "" + fut.mget.add("foobar") + return + +proc completeOnImplicitReturn(fut: FutureVar[string], x: bool) {.async.} = + if x: + fut.mget() = "" + fut.mget.add("foobar") + +proc failureTest(fut: FutureVar[string], x: bool) {.async.} = + if x: + raise newException(Exception, "Test") + +proc manualComplete(fut: FutureVar[string], x: bool) {.async.} = + if x: + fut.mget() = "Hello World" + fut.complete() + return + +proc main() {.async.} = + var fut: FutureVar[string] + + fut = newFutureVar[string]() + await completeOnReturn(fut, true) + doAssert(fut.read() == "foobar") + + fut = newFutureVar[string]() + await completeOnImplicitReturn(fut, true) + doAssert(fut.read() == "foobar") + + fut = newFutureVar[string]() + let retFut = failureTest(fut, true) + yield retFut + doAssert(fut.read().len == 0) + doAssert(fut.finished) + + fut = newFutureVar[string]() + await manualComplete(fut, true) + doAssert(fut.read() == "Hello World") + + +waitFor main() diff --git a/tests/async/tgeneric_async.nim b/tests/async/tgeneric_async.nim new file mode 100644 index 000000000..bab2d1a31 --- /dev/null +++ b/tests/async/tgeneric_async.nim @@ -0,0 +1,40 @@ +discard """ +output: "1\nmessa" +""" + +import async + +# bug #2377 +proc test[T](v: T) {.async.} = + echo $v + +asyncCheck test[int](1) + +# More complex case involving typedesc and static params +type + SomeMsg = object + data: string + +template msgId(M: type SomeMsg): int = 1 + +proc recvMsg(): Future[tuple[msgId: int, msgData: string]] {.async.} = + return (1, "message") + +proc read(data: string, T: type SomeMsg, maxBytes: int): T = + result.data = data[0 ..< min(data.len, maxBytes)] + +proc nextMsg*(MsgType: typedesc, + maxBytes: static[int]): Future[MsgType] {.async.} = + const wantedId = MsgType.msgId + + while true: + var (nextMsgId, nextMsgData) = await recvMsg() + if nextMsgId == wantedId: + return nextMsgData.read(MsgType, maxBytes) + +proc main {.async.} = + let msg = await nextMsg(SomeMsg, 5) + echo msg.data + +asyncCheck main() + diff --git a/tests/async/tgenericasync.nim b/tests/async/tgenericasync.nim new file mode 100644 index 000000000..ab704238a --- /dev/null +++ b/tests/async/tgenericasync.nim @@ -0,0 +1,14 @@ +discard """ + output: '''123 +abc''' +""" + +# bug #4856 + +import asyncdispatch + +proc say[T](t: T): Future[void] {.async.} = + echo $t + +waitFor(say(123)) +waitFor(say("abc")) diff --git a/tests/async/tioselectors.nim b/tests/async/tioselectors.nim new file mode 100644 index 000000000..f53767408 --- /dev/null +++ b/tests/async/tioselectors.nim @@ -0,0 +1,643 @@ +discard """ + output: "All tests passed!" +""" +import selectors + +const hasThreadSupport = compileOption("threads") + +template processTest(t, x: untyped) = + #stdout.write(t) + #stdout.flushFile() + if not x: echo(t & " FAILED\r\n") + +when not defined(windows): + import os, posix, nativesockets + + when ioselSupportedPlatform: + import osproc + + proc socket_notification_test(): bool = + proc create_test_socket(): SocketHandle = + var sock = posix.socket(posix.AF_INET, posix.SOCK_STREAM, + posix.IPPROTO_TCP) + var x: int = fcntl(sock, F_GETFL, 0) + if x == -1: raiseOSError(osLastError()) + else: + var mode = x or O_NONBLOCK + if fcntl(sock, F_SETFL, mode) == -1: + raiseOSError(osLastError()) + result = sock + + var client_message = "SERVER HELLO =>" + var server_message = "CLIENT HELLO" + var buffer : array[128, char] + + var selector = newSelector[int]() + var client_socket = create_test_socket() + var server_socket = create_test_socket() + + var option : int32 = 1 + if setsockopt(server_socket, cint(SOL_SOCKET), cint(SO_REUSEADDR), + addr(option), sizeof(option).SockLen) < 0: + raiseOSError(osLastError()) + + var aiList = getAddrInfo("0.0.0.0", Port(13337)) + if bindAddr(server_socket, aiList.ai_addr, + aiList.ai_addrlen.Socklen) < 0'i32: + freeAddrInfo(aiList) + raiseOSError(osLastError()) + if server_socket.listen() == -1: + raiseOSError(osLastError()) + freeAddrInfo(aiList) + + aiList = getAddrInfo("127.0.0.1", Port(13337)) + discard posix.connect(client_socket, aiList.ai_addr, + aiList.ai_addrlen.Socklen) + + registerHandle(selector, server_socket, {Event.Read}, 0) + registerHandle(selector, client_socket, {Event.Write}, 0) + + freeAddrInfo(aiList) + + # make sure both sockets are selected + var nevs = 0 + while nevs < 2: + nevs += selector.select(100).len + + var sockAddress: SockAddr + var addrLen = sizeof(sockAddress).Socklen + var server2_socket = accept(server_socket, + cast[ptr SockAddr](addr(sockAddress)), + addr(addrLen)) + assert(server2_socket != osInvalidSocket) + selector.registerHandle(server2_socket, {Event.Read}, 0) + + if posix.send(client_socket, addr(client_message[0]), + len(client_message), 0) == -1: + raiseOSError(osLastError()) + + selector.updateHandle(client_socket, {Event.Read}) + + var rc2 = selector.select(100) + assert(len(rc2) == 1) + + var read_count = posix.recv(server2_socket, addr buffer[0], 128, 0) + if read_count == -1: + raiseOSError(osLastError()) + + assert(read_count == len(client_message)) + var test1 = true + for i in 0..<read_count: + if client_message[i] != buffer[i]: + test1 = false + break + assert(test1) + + selector.updateHandle(server2_socket, {Event.Write}) + var rc3 = selector.select(0) + assert(len(rc3) == 1) + if posix.send(server2_socket, addr(server_message[0]), + len(server_message), 0) == -1: + raiseOSError(osLastError()) + selector.updateHandle(server2_socket, {Event.Read}) + + var rc4 = selector.select(100) + assert(len(rc4) == 1) + read_count = posix.recv(client_socket, addr(buffer[0]), 128, 0) + if read_count == -1: + raiseOSError(osLastError()) + + assert(read_count == len(server_message)) + var test2 = true + for i in 0..<read_count: + if server_message[i] != buffer[i]: + test2 = false + break + assert(test2) + + selector.unregister(server_socket) + selector.unregister(server2_socket) + selector.unregister(client_socket) + discard posix.close(server_socket) + discard posix.close(server2_socket) + discard posix.close(client_socket) + assert(selector.isEmpty()) + close(selector) + result = true + + proc event_notification_test(): bool = + var selector = newSelector[int]() + var event = newSelectEvent() + selector.registerEvent(event, 1) + var rc0 = selector.select(0) + event.trigger() + var rc1 = selector.select(0) + event.trigger() + var rc2 = selector.select(0) + var rc3 = selector.select(0) + assert(len(rc0) == 0 and len(rc1) == 1 and len(rc2) == 1 and len(rc3) == 0) + var ev1 = selector.getData(rc1[0].fd) + var ev2 = selector.getData(rc2[0].fd) + assert(ev1 == 1 and ev2 == 1) + selector.unregister(event) + event.close() + assert(selector.isEmpty()) + selector.close() + result = true + + when ioselSupportedPlatform: + proc timer_notification_test(): bool = + var selector = newSelector[int]() + var timer = selector.registerTimer(100, false, 0) + var rc1 = selector.select(10000) + var rc2 = selector.select(10000) + # if this flakes, see tests/m14634.nim + assert len(rc1) == 1 and len(rc2) == 1, $(len(rc1), len(rc2)) + selector.unregister(timer) + discard selector.select(0) + selector.registerTimer(100, true, 0) + var rc4 = selector.select(10000) + var rc5 = selector.select(1000) # this will be an actual wait, keep it small + assert len(rc4) == 1 and len(rc5) == 0, $(len(rc4), len(rc5)) + assert(selector.isEmpty()) + selector.close() + result = true + + proc process_notification_test(): bool = + var selector = newSelector[int]() + var process2 = startProcess("sleep", "", ["2"], nil, + {poStdErrToStdOut, poUsePath}) + discard startProcess("sleep", "", ["1"], nil, + {poStdErrToStdOut, poUsePath}) + + selector.registerProcess(process2.processID, 0) + var rc1 = selector.select(1500) + var rc2 = selector.select(1500) + var r = len(rc1) + len(rc2) + assert(r == 1) + result = true + + proc signal_notification_test(): bool = + var sigset1n, sigset1o, sigset2n, sigset2o: Sigset + var pid = posix.getpid() + + discard sigemptyset(sigset1n) + discard sigemptyset(sigset1o) + discard sigemptyset(sigset2n) + discard sigemptyset(sigset2o) + + when hasThreadSupport: + if pthread_sigmask(SIG_BLOCK, sigset1n, sigset1o) == -1: + raiseOSError(osLastError()) + else: + if sigprocmask(SIG_BLOCK, sigset1n, sigset1o) == -1: + raiseOSError(osLastError()) + + var selector = newSelector[int]() + var s1 = selector.registerSignal(SIGUSR1, 1) + var s2 = selector.registerSignal(SIGUSR2, 2) + var s3 = selector.registerSignal(SIGTERM, 3) + discard selector.select(0) + discard posix.kill(pid, SIGUSR1) + discard posix.kill(pid, SIGUSR2) + discard posix.kill(pid, SIGTERM) + var rc = selector.select(0) + var cd0 = selector.getData(rc[0].fd) + var cd1 = selector.getData(rc[1].fd) + var cd2 = selector.getData(rc[2].fd) + selector.unregister(s1) + selector.unregister(s2) + selector.unregister(s3) + + when hasThreadSupport: + if pthread_sigmask(SIG_BLOCK, sigset2n, sigset2o) == -1: + raiseOSError(osLastError()) + else: + if sigprocmask(SIG_BLOCK, sigset2n, sigset2o) == -1: + raiseOSError(osLastError()) + + assert(len(rc) == 3) + assert(cd0 + cd1 + cd2 == 6, $(cd0 + cd1 + cd2)) # 1 + 2 + 3 + assert(equalMem(addr sigset1o, addr sigset2o, sizeof(Sigset))) + assert(selector.isEmpty()) + result = true + + when defined(macosx) or defined(freebsd) or defined(openbsd) or + defined(netbsd): + + proc rename(frompath: cstring, topath: cstring): cint + {.importc: "rename", header: "<stdio.h>".} + + proc createFile(name: string): cint = + result = posix.open(cstring(name), posix.O_CREAT or posix.O_RDWR) + if result == -1: + raiseOsError(osLastError()) + + proc writeFile(name: string, data: string) = + let fd = posix.open(cstring(name), posix.O_APPEND or posix.O_RDWR) + if fd == -1: + raiseOsError(osLastError()) + let length = len(data).cint + if posix.write(fd, cast[pointer](addr data[0]), + len(data).cint) != length: + raiseOsError(osLastError()) + if posix.close(fd) == -1: + raiseOsError(osLastError()) + + proc closeFile(fd: cint) = + if posix.close(fd) == -1: + raiseOsError(osLastError()) + + proc removeFile(name: string) = + let err = posix.unlink(cstring(name)) + if err == -1: + raiseOsError(osLastError()) + + proc createDir(name: string) = + let err = posix.mkdir(cstring(name), 0x1FF) + if err == -1: + raiseOsError(osLastError()) + + proc removeDir(name: string) = + let err = posix.rmdir(cstring(name)) + if err == -1: + raiseOsError(osLastError()) + + proc chmodPath(name: string, mode: cint) = + let err = posix.chmod(cstring(name), Mode(mode)) + if err == -1: + raiseOsError(osLastError()) + + proc renameFile(names: string, named: string) = + let err = rename(cstring(names), cstring(named)) + if err == -1: + raiseOsError(osLastError()) + + proc symlink(names: string, named: string) = + let err = posix.symlink(cstring(names), cstring(named)) + if err == -1: + raiseOsError(osLastError()) + + proc openWatch(name: string): cint = + result = posix.open(cstring(name), posix.O_RDONLY) + if result == -1: + raiseOsError(osLastError()) + + const + testDirectory = "/tmp/kqtest" + + type + valType = object + fd: cint + events: set[Event] + + proc vnode_test(): bool = + proc validate(test: openArray[ReadyKey], + check: openArray[valType]): bool = + result = false + if len(test) == len(check): + for checkItem in check: + result = false + for testItem in test: + if testItem.fd == checkItem.fd and + checkItem.events <= testItem.events: + result = true + break + if not result: + break + + var res: seq[ReadyKey] + var selector = newSelector[int]() + var events = {Event.VnodeWrite, Event.VnodeDelete, Event.VnodeExtend, + Event.VnodeAttrib, Event.VnodeLink, Event.VnodeRename, + Event.VnodeRevoke} + + result = true + discard posix.unlink(testDirectory) + + createDir(testDirectory) + var dirfd = posix.open(cstring(testDirectory), posix.O_RDONLY) + if dirfd == -1: + raiseOsError(osLastError()) + + selector.registerVnode(dirfd, events, 1) + discard selector.select(0) + + # chmod testDirectory to 0777 + chmodPath(testDirectory, 0x1FF) + res = selector.select(0) + doAssert(len(res) == 1) + doAssert(len(selector.select(0)) == 0) + doAssert(res[0].fd == dirfd and + {Event.Vnode, Event.VnodeAttrib} <= res[0].events) + + # create subdirectory + createDir(testDirectory & "/test") + res = selector.select(0) + doAssert(len(res) == 1) + doAssert(len(selector.select(0)) == 0) + doAssert(res[0].fd == dirfd and + {Event.Vnode, Event.VnodeWrite, + Event.VnodeLink} <= res[0].events) + + # open test directory for watching + var testfd = openWatch(testDirectory & "/test") + selector.registerVnode(testfd, events, 2) + doAssert(len(selector.select(0)) == 0) + + # rename test directory + renameFile(testDirectory & "/test", testDirectory & "/renamed") + res = selector.select(0) + doAssert(len(res) == 2) + doAssert(len(selector.select(0)) == 0) + doAssert(validate(res, + [valType(fd: dirfd, events: {Event.Vnode, Event.VnodeWrite}), + valType(fd: testfd, + events: {Event.Vnode, Event.VnodeRename})]) + ) + + # remove test directory + removeDir(testDirectory & "/renamed") + res = selector.select(0) + doAssert(len(res) == 2) + doAssert(len(selector.select(0)) == 0) + doAssert(validate(res, + [valType(fd: dirfd, events: {Event.Vnode, Event.VnodeWrite, + Event.VnodeLink}), + valType(fd: testfd, + events: {Event.Vnode, Event.VnodeDelete})]) + ) + # create file new test file + testfd = createFile(testDirectory & "/testfile") + res = selector.select(0) + doAssert(len(res) == 1) + doAssert(len(selector.select(0)) == 0) + doAssert(res[0].fd == dirfd and + {Event.Vnode, Event.VnodeWrite} <= res[0].events) + + # close new test file + closeFile(testfd) + doAssert(len(selector.select(0)) == 0) + doAssert(len(selector.select(0)) == 0) + + # chmod test file with 0666 + chmodPath(testDirectory & "/testfile", 0x1B6) + doAssert(len(selector.select(0)) == 0) + + testfd = openWatch(testDirectory & "/testfile") + selector.registerVnode(testfd, events, 1) + discard selector.select(0) + + # write data to test file + writeFile(testDirectory & "/testfile", "TESTDATA") + res = selector.select(0) + doAssert(len(res) == 1) + doAssert(len(selector.select(0)) == 0) + doAssert(res[0].fd == testfd and + {Event.Vnode, Event.VnodeWrite, + Event.VnodeExtend} <= res[0].events) + + # symlink test file + symlink(testDirectory & "/testfile", testDirectory & "/testlink") + res = selector.select(0) + doAssert(len(res) == 1) + doAssert(len(selector.select(0)) == 0) + doAssert(res[0].fd == dirfd and + {Event.Vnode, Event.VnodeWrite} <= res[0].events) + + # remove test file + removeFile(testDirectory & "/testfile") + res = selector.select(0) + doAssert(len(res) == 2) + doAssert(len(selector.select(0)) == 0) + doAssert(validate(res, + [valType(fd: testfd, events: {Event.Vnode, Event.VnodeDelete}), + valType(fd: dirfd, events: {Event.Vnode, Event.VnodeWrite})]) + ) + + # remove symlink + removeFile(testDirectory & "/testlink") + res = selector.select(0) + doAssert(len(res) == 1) + doAssert(len(selector.select(0)) == 0) + doAssert(res[0].fd == dirfd and + {Event.Vnode, Event.VnodeWrite} <= res[0].events) + + # remove testDirectory + removeDir(testDirectory) + res = selector.select(0) + doAssert(len(res) == 1) + doAssert(len(selector.select(0)) == 0) + doAssert(res[0].fd == dirfd and + {Event.Vnode, Event.VnodeDelete} <= res[0].events) + + proc pipe_test(): bool = + # closing the read end of a pipe will result in it automatically + # being removed from the kqueue; make sure no exception is raised + var s = newSelector[int]() + var fds: array[2, cint] + discard pipe(fds) + s.registerHandle(fds[1], {Write}, 0) + discard close(fds[0]) + let res = s.select(-1) + doAssert(res.len == 1) + s.unregister(fds[1]) + discard close(fds[1]) + return true + + when hasThreadSupport: + + var counter = 0 + + proc event_wait_thread(event: SelectEvent) {.thread.} = + var selector = newSelector[int]() + selector.registerEvent(event, 1) + var rc = selector.select(1000) + if len(rc) == 1: + inc(counter) + selector.unregister(event) + assert(selector.isEmpty()) + + proc mt_event_test(): bool = + var + thr: array[0..7, Thread[SelectEvent]] + var selector = newSelector[int]() + var sock = createNativeSocket() + var event = newSelectEvent() + for i in 0..high(thr): + createThread(thr[i], event_wait_thread, event) + selector.registerHandle(sock, {Event.Read}, 1) + discard selector.select(500) + selector.unregister(sock) + event.trigger() + joinThreads(thr) + assert(counter == 1) + result = true + + processTest("Socket notification test...", socket_notification_test()) + processTest("User event notification test...", event_notification_test()) + when hasThreadSupport: + processTest("Multithreaded user event notification test...", + mt_event_test()) + when ioselSupportedPlatform: + processTest("Timer notification test...", timer_notification_test()) + processTest("Process notification test...", process_notification_test()) + processTest("Signal notification test...", signal_notification_test()) + when defined(macosx) or defined(freebsd) or defined(openbsd) or + defined(netbsd): + processTest("File notification test...", vnode_test()) + processTest("Pipe test...", pipe_test()) + echo("All tests passed!") +else: + import nativesockets, winlean, os, osproc + + proc socket_notification_test(): bool = + proc create_test_socket(): SocketHandle = + var sock = createNativeSocket() + setBlocking(sock, false) + result = sock + + var client_message = "SERVER HELLO =>" + var server_message = "CLIENT HELLO" + var buffer : array[128, char] + + var selector = newSelector[int]() + var client_socket = create_test_socket() + var server_socket = create_test_socket() + + selector.registerHandle(server_socket, {Event.Read}, 0) + selector.registerHandle(client_socket, {Event.Write}, 0) + + var option : int32 = 1 + if setsockopt(server_socket, cint(SOL_SOCKET), cint(SO_REUSEADDR), + addr(option), sizeof(option).SockLen) < 0: + raiseOSError(osLastError()) + + var aiList = getAddrInfo("0.0.0.0", Port(13337)) + if bindAddr(server_socket, aiList.ai_addr, + aiList.ai_addrlen.Socklen) < 0'i32: + freeAddrInfo(aiList) + raiseOSError(osLastError()) + discard server_socket.listen() + freeAddrInfo(aiList) + + aiList = getAddrInfo("127.0.0.1", Port(13337)) + discard connect(client_socket, aiList.ai_addr, + aiList.ai_addrlen.Socklen) + freeAddrInfo(aiList) + # for some reason Windows select doesn't return both + # descriptors from first call, so we need to make 2 calls + var n = 0 + var rcm = selector.select(1000) + while n < 10 and len(rcm) < 2: + sleep(1000) + rcm = selector.select(1000) + inc(n) + + assert(len(rcm) == 2) + + var sockAddress = SockAddr() + var addrLen = sizeof(sockAddress).Socklen + var server2_socket = accept(server_socket, + cast[ptr SockAddr](addr(sockAddress)), + addr(addrLen)) + assert(server2_socket != osInvalidSocket) + selector.registerHandle(server2_socket, {Event.Read}, 0) + + if send(client_socket, cast[pointer](addr(client_message[0])), + cint(len(client_message)), 0) == -1: + raiseOSError(osLastError()) + + selector.updateHandle(client_socket, {Event.Read}) + + var rc2 = selector.select(1000) + assert(len(rc2) == 1) + + var read_count = recv(server2_socket, addr buffer[0], 128, 0) + if read_count == -1: + raiseOSError(osLastError()) + + assert(read_count == len(client_message)) + var test1 = true + for i in 0..<read_count: + if client_message[i] != buffer[i]: + test1 = false + break + assert(test1) + + if send(server2_socket, cast[pointer](addr(server_message[0])), + cint(len(server_message)), 0) == -1: + raiseOSError(osLastError()) + + var rc3 = selector.select(0) + assert(len(rc3) == 1) + read_count = recv(client_socket, addr(buffer[0]), 128, 0) + if read_count == -1: + raiseOSError(osLastError()) + + assert(read_count == len(server_message)) + var test2 = true + for i in 0..<read_count: + if server_message[i] != buffer[i]: + test2 = false + break + assert(test2) + + selector.unregister(server_socket) + selector.unregister(server2_socket) + selector.unregister(client_socket) + close(server_socket) + close(server2_socket) + close(client_socket) + assert(selector.isEmpty()) + close(selector) + result = true + + proc event_notification_test(): bool = + var selector = newSelector[int]() + var event = newSelectEvent() + selector.registerEvent(event, 1) + discard selector.select(0) + event.trigger() + var rc1 = selector.select(0) + event.trigger() + var rc2 = selector.select(0) + var rc3 = selector.select(0) + assert(len(rc1) == 1 and len(rc2) == 1 and len(rc3) == 0) + var ev1 = selector.getData(rc1[0].fd) + var ev2 = selector.getData(rc2[0].fd) + assert(ev1 == 1 and ev2 == 1) + selector.unregister(event) + event.close() + assert(selector.isEmpty()) + selector.close() + result = true + + when hasThreadSupport: + var counter = 0 + + proc event_wait_thread(event: SelectEvent) {.thread.} = + var selector = newSelector[int]() + selector.registerEvent(event, 1) + var rc = selector.select(1500) + if len(rc) == 1: + inc(counter) + selector.unregister(event) + assert(selector.isEmpty()) + + proc mt_event_test(): bool = + var thr: array[0..7, Thread[SelectEvent]] + var event = newSelectEvent() + for i in 0..high(thr): + createThread(thr[i], event_wait_thread, event) + event.trigger() + joinThreads(thr) + assert(counter == 1) + result = true + + processTest("Socket notification test...", socket_notification_test()) + processTest("User event notification test...", event_notification_test()) + when hasThreadSupport: + processTest("Multithreaded user event notification test...", + mt_event_test()) + echo("All tests passed!") diff --git a/tests/async/tioselectors.nim.cfg b/tests/async/tioselectors.nim.cfg new file mode 100644 index 000000000..b1b896858 --- /dev/null +++ b/tests/async/tioselectors.nim.cfg @@ -0,0 +1 @@ +threads:on -d:threadsafe diff --git a/tests/async/tjsandnativeasync.nim b/tests/async/tjsandnativeasync.nim new file mode 100644 index 000000000..c4db3bcfb --- /dev/null +++ b/tests/async/tjsandnativeasync.nim @@ -0,0 +1,30 @@ +discard """ + output: '''hi +bye''' +""" + +import async, times +when defined(js): + proc sleepAsync(t: int): Future[void] = + var promise = newPromise() do(resolve: proc()): + {.emit: """ + setTimeout(function(){ + `resolve`(); + }, `t`); + """.} + result = promise +else: + from asyncdispatch import sleepAsync, waitFor + +proc foo() {.async.} = + echo "hi" + var s = epochTime() + await sleepAsync(200) + var e = epochTime() + doAssert(e - s > 0.1) + echo "bye" + +when defined(js): + discard foo() +else: + waitFor foo() diff --git a/tests/async/tlambda.nim b/tests/async/tlambda.nim new file mode 100644 index 000000000..8f570689b --- /dev/null +++ b/tests/async/tlambda.nim @@ -0,0 +1,58 @@ + +# bug 2007 + +import asyncdispatch, asyncnet, logging, json, uri, strutils, sugar + +type + Builder = ref object + client: Client + build: Build + + ProgressCB* = proc (message: string): Future[void] {.closure, gcsafe.} + + Build* = ref object + onProgress*: ProgressCB + + Client = ref ClientObj + ClientObj = object + onMessage: proc (client: Client, msg: JsonNode): Future[void] + +proc newClient*(name: string, + onMessage: (Client, JsonNode) -> Future[void]): Client = + new result + result.onMessage = onMessage + +proc newBuild*(onProgress: ProgressCB): Build = + new result + result.onProgress = onProgress + +proc start(build: Build, repo, hash: string) {.async.} = + let path = repo.parseUri().path.toLowerAscii() + +proc onProgress(builder: Builder, message: string) {.async.} = + debug($message) + +proc onMessage(builder: Builder, message: JsonNode) {.async.} = + debug("onMessage") + +proc newBuilder(): Builder = + var cres: Builder + new cres + + cres.client = newClient("builder", (client, msg) => (onMessage(cres, msg))) + cres.build = newBuild( + proc (msg: string): Future[void] {.closure, gcsafe.} = onProgress(cres, msg)) + return cres + +proc main() = + # Set up logging. + var console = newConsoleLogger(fmtStr = verboseFmtStr) + addHandler(console) + + var builder = newBuilder() + + # Test {.async.} pragma with do notation: #5995 + builder.client = newClient("builder") do(client: Client, msg: JsonNode) {.async.}: + await onMessage(builder, msg) + +main() diff --git a/tests/async/tmultisync.nim b/tests/async/tmultisync.nim new file mode 100644 index 000000000..9ef9b105c --- /dev/null +++ b/tests/async/tmultisync.nim @@ -0,0 +1,8 @@ +import asyncdispatch, net, asyncnet + +proc recvTwice(socket: Socket | AsyncSocket, + size: int): Future[string] {.multisync.} = + var x = await socket.recv(size) + var y = await socket.recv(size+1) + return x & "aboo" & y + diff --git a/tests/async/tnestedpfuturetypeparam.nim b/tests/async/tnestedpfuturetypeparam.nim new file mode 100644 index 000000000..bf346ff8e --- /dev/null +++ b/tests/async/tnestedpfuturetypeparam.nim @@ -0,0 +1,8 @@ +import asyncdispatch, asyncnet + +proc main {.async.} = + proc f: Future[seq[int]] {.async.} = + await newAsyncSocket().connect("www.google.com", Port(80)) + let x = await f() + +asyncCheck main() diff --git a/tests/async/tnewasyncudp.nim b/tests/async/tnewasyncudp.nim new file mode 100644 index 000000000..68de796a0 --- /dev/null +++ b/tests/async/tnewasyncudp.nim @@ -0,0 +1,111 @@ +discard """ + output: "5000" +""" +import asyncdispatch, nativesockets, net, strutils, os + +when defined(windows): + import winlean +else: + import posix + +var msgCount = 0 +var recvCount = 0 + +const + messagesToSend = 100 + swarmSize = 50 + serverPort = 10333 + +var + sendports = 0 + recvports = 0 + +proc saveSendingPort(port: int) = + sendports = sendports + port + +proc saveReceivedPort(port: int) = + recvports = recvports + port + +proc prepareAddress(intaddr: uint32, intport: uint16): ptr Sockaddr_in = + result = cast[ptr Sockaddr_in](alloc0(sizeof(Sockaddr_in))) + result.sin_family = typeof(result.sin_family)(toInt(nativesockets.AF_INET)) + result.sin_port = nativesockets.htons(intport) + result.sin_addr.s_addr = nativesockets.htonl(intaddr) + +proc launchSwarm(name: ptr SockAddr) {.async.} = + var i = 0 + var k = 0 + var buffer: array[16384, char] + var slen = sizeof(Sockaddr_in).SockLen + var saddr = Sockaddr_in() + while i < swarmSize: + var peeraddr = prepareAddress(INADDR_LOOPBACK, 0) + var sock = createAsyncNativeSocket(nativesockets.AF_INET, + nativesockets.SOCK_DGRAM, + Protocol.IPPROTO_UDP) + if bindAddr(sock.SocketHandle, cast[ptr SockAddr](peeraddr), + sizeof(Sockaddr_in).Socklen) < 0'i32: + raiseOSError(osLastError()) + let sockport = getSockName(sock.SocketHandle).int + k = 0 + while k < messagesToSend: + zeroMem(addr(buffer[0]), 16384) + zeroMem(cast[pointer](addr(saddr)), sizeof(Sockaddr_in)) + var message = "Message " & $(i * messagesToSend + k) + await sendTo(sock, addr message[0], len(message), + name, sizeof(Sockaddr_in).SockLen) + var size = await recvFromInto(sock, cast[pointer](addr buffer[0]), + 16384, cast[ptr SockAddr](addr saddr), + addr slen) + size = 0 + var grammString = $cast[cstring](addr buffer) + if grammString == message: + saveSendingPort(sockport) + inc(recvCount) + inc(k) + closeSocket(sock) + inc(i) + +proc readMessages(server: AsyncFD) {.async.} = + var buffer: array[16384, char] + var slen = sizeof(Sockaddr_in).SockLen + var saddr = Sockaddr_in() + var maxResponses = (swarmSize * messagesToSend) + + var i = 0 + while i < maxResponses: + zeroMem(addr(buffer[0]), 16384) + zeroMem(cast[pointer](addr(saddr)), sizeof(Sockaddr_in)) + var size = await recvFromInto(server, cast[cstring](addr buffer[0]), + 16384, cast[ptr SockAddr](addr(saddr)), + addr(slen)) + size = 0 + var grammString = $cast[cstring](addr buffer) + if grammString.startsWith("Message ") and + saddr.sin_addr.s_addr == nativesockets.ntohl(INADDR_LOOPBACK.uint32): + await sendTo(server, addr grammString[0], len(grammString), + cast[ptr SockAddr](addr saddr), slen) + inc(msgCount) + saveReceivedPort(nativesockets.ntohs(saddr.sin_port).int) + inc(i) + +proc createServer() {.async.} = + var name = prepareAddress(INADDR_LOOPBACK, serverPort) + var server = createAsyncNativeSocket(nativesockets.AF_INET, + nativesockets.SOCK_DGRAM, + Protocol.IPPROTO_UDP) + if bindAddr(server.SocketHandle, cast[ptr SockAddr](name), + sizeof(Sockaddr_in).Socklen) < 0'i32: + raiseOSError(osLastError()) + asyncCheck readMessages(server) + +var name = prepareAddress(INADDR_LOOPBACK, serverPort) # 127.0.0.1 +asyncCheck createServer() +asyncCheck launchSwarm(cast[ptr SockAddr](name)) +while true: + poll() + if recvCount == swarmSize * messagesToSend: + break +assert msgCount == swarmSize * messagesToSend +assert sendports == recvports +echo msgCount diff --git a/tests/async/tnimcall_to_closure.nim b/tests/async/tnimcall_to_closure.nim new file mode 100644 index 000000000..748b67cb1 --- /dev/null +++ b/tests/async/tnimcall_to_closure.nim @@ -0,0 +1,17 @@ + +import asyncdispatch + +proc defaultOnProgressChanged() = discard + +proc ask(x: proc()) = x() + +proc retrFile*(onProgressChanged: proc() {.nimcall.}): Future[void] = + var retFuture = newFuture[void]("retrFile") + iterator retrFileIter(): FutureBase {.closure.} = + ask(onProgressChanged) + complete(retFuture) + + var nameIterVar = retrFileIter + return retFuture + +discard retrFile(defaultOnProgressChanged) diff --git a/tests/async/tpendingcheck.nim b/tests/async/tpendingcheck.nim new file mode 100644 index 000000000..4eceb0353 --- /dev/null +++ b/tests/async/tpendingcheck.nim @@ -0,0 +1,18 @@ +discard """ + output: "" +""" + +import asyncdispatch + +doAssert(not hasPendingOperations()) + +proc test() {.async.} = + await sleepAsync(50) + +var f = test() +while not f.finished: + doAssert(hasPendingOperations()) + poll(10) +f.read + +doAssert(not hasPendingOperations()) diff --git a/tests/async/tpolltimeouts.nim b/tests/async/tpolltimeouts.nim new file mode 100644 index 000000000..dac33732d --- /dev/null +++ b/tests/async/tpolltimeouts.nim @@ -0,0 +1,19 @@ +discard """ + output: "true" +""" +# Issue https://github.com/nim-lang/Nim/issues/4262 +import asyncdispatch, times + +proc foo(): Future[int] {.async.} = + return 1 + +proc bar(): Future[int] {.async.} = + return await foo() + +let start = epochTime() +let barFut = bar() + +while not barFut.finished: + poll(2000) + +echo(epochTime() - start < 1.0) diff --git a/tests/async/treturn_await.nim b/tests/async/treturn_await.nim new file mode 100644 index 000000000..8d266d665 --- /dev/null +++ b/tests/async/treturn_await.nim @@ -0,0 +1,23 @@ + +# bug #4371 + +import strutils, asyncdispatch, asynchttpserver + +type + List[A] = ref object + value: A + next: List[A] + StrPair* = tuple[k, v: string] + Context* = object + position*: int + accept*: bool + headers*: List[StrPair] + Handler* = proc(req: ref Request, ctx: Context): Future[Context] + +proc logging*(handler: Handler): auto = + proc h(req: ref Request, ctx: Context): Future[Context] {.async.} = + let ret = handler(req, ctx) + debugEcho "$3 $1 $2".format(req.reqMethod, req.url.path, req.hostname) + return await ret + + return h diff --git a/tests/async/ttemplateinasync.nim b/tests/async/ttemplateinasync.nim new file mode 100644 index 000000000..f4a2da538 --- /dev/null +++ b/tests/async/ttemplateinasync.nim @@ -0,0 +1,11 @@ +discard """ + output: 42 +""" + +import asyncdispatch + +proc foo(): Future[int] {.async.} = + template ret() = return 42 + ret() + +echo (waitFor foo()) diff --git a/tests/async/tupcoming_async.nim b/tests/async/tupcoming_async.nim new file mode 100644 index 000000000..0b6e53454 --- /dev/null +++ b/tests/async/tupcoming_async.nim @@ -0,0 +1,157 @@ +discard """ + output: ''' +OK +''' +""" + +when defined(upcoming): + import asyncdispatch, times, streams, posix + from ioselectors import ioselSupportedPlatform + + proc delayedSet(ev: AsyncEvent, timeout: int): Future[void] {.async.} = + await sleepAsync(timeout) + ev.trigger() + + proc waitEvent(ev: AsyncEvent, closeEvent = false): Future[void] = + var retFuture = newFuture[void]("waitEvent") + proc cb(fd: AsyncFD): bool = + retFuture.complete() + if closeEvent: + return true + else: + return false + addEvent(ev, cb) + return retFuture + + proc eventTest() = + var event = newAsyncEvent() + var fut = waitEvent(event) + asyncCheck(delayedSet(event, 500)) + waitFor(fut or sleepAsync(1000)) + if not fut.finished: + echo "eventTest: Timeout expired before event received!" + + proc eventTest5304() = + # Event should not be signaled if it was uregistered, + # even in case, when poll() was not called yet. + # Issue #5304. + var unregistered = false + let e = newAsyncEvent() + addEvent(e) do (fd: AsyncFD) -> bool: + assert(not unregistered) + e.trigger() + e.unregister() + unregistered = true + poll() + + proc eventTest5298() = + # Event must raise `AssertionDefect` if event was unregistered twice. + # Issue #5298. + let e = newAsyncEvent() + var eventReceived = false + addEvent(e) do (fd: AsyncFD) -> bool: + eventReceived = true + return true + e.trigger() + while not eventReceived: + poll() + try: + e.unregister() + except AssertionDefect: + discard + e.close() + + proc eventTest5331() = + # Event must not raise any exceptions while was unregistered inside of + # own callback. + # Issue #5331. + let e = newAsyncEvent() + addEvent(e) do (fd: AsyncFD) -> bool: + e.unregister() + e.close() + e.trigger() + poll() + + when ioselSupportedPlatform or defined(windows): + + import osproc + + proc waitTimer(timeout: int): Future[void] = + var retFuture = newFuture[void]("waitTimer") + proc cb(fd: AsyncFD): bool = + retFuture.complete() + addTimer(timeout, true, cb) + return retFuture + + proc waitProcess(p: Process): Future[void] = + var retFuture = newFuture[void]("waitProcess") + proc cb(fd: AsyncFD): bool = + retFuture.complete() + addProcess(p.processID(), cb) + return retFuture + + proc timerTest() = + waitFor(waitTimer(200)) + + proc processTest() = + when defined(windows): + var process = startProcess("ping.exe", "", + ["127.0.0.1", "-n", "2", "-w", "100"], nil, + {poStdErrToStdOut, poUsePath, poInteractive, + poDaemon}) + else: + var process = startProcess("sleep", "", ["1"], nil, + {poStdErrToStdOut, poUsePath}) + var fut = waitProcess(process) + waitFor(fut or waitTimer(2000)) + if fut.finished and process.peekExitCode() == 0: + discard + else: + echo "processTest: Timeout expired before process exited!" + + when ioselSupportedPlatform: + + proc waitSignal(signal: int): Future[void] = + var retFuture = newFuture[void]("waitSignal") + proc cb(fd: AsyncFD): bool = + retFuture.complete() + addSignal(signal, cb) + return retFuture + + proc delayedSignal(signal: int, timeout: int): Future[void] {.async.} = + await waitTimer(timeout) + var pid = posix.getpid() + discard posix.kill(pid, signal.cint) + + proc signalTest() = + var fut = waitSignal(posix.SIGINT) + asyncCheck(delayedSignal(posix.SIGINT, 500)) + waitFor(fut or waitTimer(1000)) + if not fut.finished: + echo "signalTest: Timeout expired before signal received!" + + when ioselSupportedPlatform: + timerTest() + eventTest() + eventTest5304() + eventTest5298() + eventTest5331() + processTest() + signalTest() + echo "OK" + elif defined(windows): + timerTest() + eventTest() + eventTest5304() + eventTest5298() + eventTest5331() + processTest() + echo "OK" + else: + eventTest() + eventTest5304() + eventTest5298() + eventTest5331() + echo "OK" +else: + echo "OK" diff --git a/tests/async/twinasyncrw.nim b/tests/async/twinasyncrw.nim new file mode 100644 index 000000000..f0a8f6a62 --- /dev/null +++ b/tests/async/twinasyncrw.nim @@ -0,0 +1,243 @@ +when defined(windows): + import asyncdispatch, nativesockets, net, strutils, os, winlean + from stdtest/netutils import bindAvailablePort + var msgCount = 0 + + const + swarmSize = 50 + messagesToSend = 100 + + var clientCount = 0 + + proc winConnect*(socket: AsyncFD, address: string, port: Port, + domain = Domain.AF_INET): Future[void] = + var retFuture = newFuture[void]("winConnect") + proc cb(fd: AsyncFD): bool = + var ret = SocketHandle(fd).getSockOptInt(cint(SOL_SOCKET), cint(SO_ERROR)) + if ret == 0: + # We have connected. + retFuture.complete() + return true + else: + retFuture.fail(newOSError(OSErrorCode(ret))) + return true + + var aiList = getAddrInfo(address, port, domain) + var success = false + var lastError: OSErrorCode = OSErrorCode(0) + var it = aiList + while it != nil: + var ret = nativesockets.connect(socket.SocketHandle, it.ai_addr, it.ai_addrlen.Socklen) + if ret == 0: + # Request to connect completed immediately. + success = true + retFuture.complete() + break + else: + lastError = osLastError() + if lastError.int32 == WSAEWOULDBLOCK: + success = true + addWrite(socket, cb) + break + else: + success = false + it = it.ai_next + + freeAddrInfo(aiList) + if not success: + retFuture.fail(newOSError(lastError)) + return retFuture + + proc winRecv*(socket: AsyncFD, size: int, + flags = {SocketFlag.SafeDisconn}): Future[string] = + var retFuture = newFuture[string]("recv") + + var readBuffer = newString(size) + + proc cb(sock: AsyncFD): bool = + result = true + let res = recv(sock.SocketHandle, addr readBuffer[0], size.cint, + flags.toOSFlags()) + if res < 0: + let lastError = osLastError() + if flags.isDisconnectionError(lastError): + retFuture.complete("") + else: + retFuture.fail(newOSError(lastError)) + elif res == 0: + # Disconnected + retFuture.complete("") + else: + readBuffer.setLen(res) + retFuture.complete(readBuffer) + # TODO: The following causes a massive slowdown. + #if not cb(socket): + addRead(socket, cb) + return retFuture + + proc winRecvInto*(socket: AsyncFD, buf: cstring, size: int, + flags = {SocketFlag.SafeDisconn}): Future[int] = + var retFuture = newFuture[int]("winRecvInto") + + proc cb(sock: AsyncFD): bool = + result = true + let res = nativesockets.recv(sock.SocketHandle, buf, size.cint, + flags.toOSFlags()) + if res < 0: + let lastError = osLastError() + if flags.isDisconnectionError(lastError): + retFuture.complete(0) + else: + retFuture.fail(newOSError(lastError)) + else: + retFuture.complete(res) + # TODO: The following causes a massive slowdown. + #if not cb(socket): + addRead(socket, cb) + return retFuture + + proc winSend*(socket: AsyncFD, data: string, + flags = {SocketFlag.SafeDisconn}): Future[void] = + var retFuture = newFuture[void]("winSend") + + var written = 0 + + proc cb(sock: AsyncFD): bool = + result = true + let netSize = data.len-written + var d = data.cstring + let res = nativesockets.send(sock.SocketHandle, addr d[written], netSize.cint, 0) + if res < 0: + let lastError = osLastError() + if flags.isDisconnectionError(lastError): + retFuture.complete() + else: + retFuture.fail(newOSError(lastError)) + else: + written.inc(res) + if res != netSize: + result = false # We still have data to send. + else: + retFuture.complete() + # TODO: The following causes crashes. + #if not cb(socket): + addWrite(socket, cb) + return retFuture + + proc winAcceptAddr*(socket: AsyncFD, flags = {SocketFlag.SafeDisconn}): + Future[tuple[address: string, client: AsyncFD]] = + var retFuture = newFuture[tuple[address: string, + client: AsyncFD]]("winAcceptAddr") + proc cb(sock: AsyncFD): bool = + result = true + if not retFuture.finished: + var sockAddress = Sockaddr() + var addrLen = sizeof(sockAddress).Socklen + var client = nativesockets.accept(sock.SocketHandle, + cast[ptr SockAddr](addr(sockAddress)), addr(addrLen)) + if client == osInvalidSocket: + retFuture.fail(newOSError(osLastError())) + else: + retFuture.complete((getAddrString(cast[ptr SockAddr](addr sockAddress)), client.AsyncFD)) + + addRead(socket, cb) + return retFuture + + proc winAccept*(socket: AsyncFD, + flags = {SocketFlag.SafeDisconn}): Future[AsyncFD] = + ## Accepts a new connection. Returns a future containing the client socket + ## corresponding to that connection. + ## The future will complete when the connection is successfully accepted. + var retFut = newFuture[AsyncFD]("winAccept") + var fut = winAcceptAddr(socket, flags) + fut.callback = + proc (future: Future[tuple[address: string, client: AsyncFD]]) = + assert future.finished + if future.failed: + retFut.fail(future.error) + else: + retFut.complete(future.read.client) + return retFut + + + proc winRecvLine*(socket: AsyncFD): Future[string] {.async.} = + ## Reads a line of data from ``socket``. Returned future will complete once + ## a full line is read or an error occurs. + ## + ## If a full line is read ``\r\L`` is not + ## added to ``line``, however if solely ``\r\L`` is read then ``line`` + ## will be set to it. + ## + ## If the socket is disconnected, ``line`` will be set to ``""``. + ## + ## If the socket is disconnected in the middle of a line (before ``\r\L`` + ## is read) then line will be set to ``""``. + ## The partial line **will be lost**. + ## + ## **Warning**: This assumes that lines are delimited by ``\r\L``. + ## + ## **Note**: This procedure is mostly used for testing. You likely want to + ## use ``asyncnet.recvLine`` instead. + + template addNLIfEmpty() = + if result.len == 0: + result.add("\c\L") + + result = "" + var c = "" + while true: + c = await winRecv(socket, 1) + if c.len == 0: + return "" + if c == "\r": + c = await winRecv(socket, 1) + assert c == "\l" + addNLIfEmpty() + return + elif c == "\L": + addNLIfEmpty() + return + add(result, c) + + proc sendMessages(client: AsyncFD) {.async.} = + for i in 0 ..< messagesToSend: + await winSend(client, "Message " & $i & "\c\L") + + proc launchSwarm(port: Port) {.async.} = + for i in 0 ..< swarmSize: + var sock = createNativeSocket() + setBlocking(sock, false) + + await winConnect(AsyncFD(sock), "localhost", port) + await sendMessages(AsyncFD(sock)) + discard closeSocket(sock) + + proc readMessages(client: AsyncFD) {.async.} = + while true: + var line = await winRecvLine(client) + if line == "": + closeSocket(client) + clientCount.inc + break + else: + if line.startsWith("Message "): + msgCount.inc + else: + doAssert false + + proc createServer(server: SocketHandle) {.async.} = + discard server.listen() + while true: + asyncCheck readMessages(await winAccept(AsyncFD(server))) + + var server = createNativeSocket() + setBlocking(server, false) + let port = bindAvailablePort(server) + asyncCheck createServer(server) + asyncCheck launchSwarm(port) + while true: + poll() + if clientCount == swarmSize: break + + assert msgCount == swarmSize * messagesToSend + doAssert msgCount == 5000 |