diff --git a/.vscode/settings.json b/.vscode/settings.json deleted file mode 100644 index 1dfdcd1..0000000 --- a/.vscode/settings.json +++ /dev/null @@ -1,5 +0,0 @@ -{ - "haxe.displayConfigurations": [ - ["tests.hxml", "-lib", "travix", "-lib", "tink_io"] - ] -} diff --git a/dev.hxml b/dev.hxml index dac2319..d789ba2 100644 --- a/dev.hxml +++ b/dev.hxml @@ -1,5 +1,8 @@ tests.hxml -lib travix -lib tink_io --lib hxnodejs --js bin/node/tests.js \ No newline at end of file + +-neko bin/neko/tests.n + +# -lib hxnodejs +# -js bin/node/tests.js \ No newline at end of file diff --git a/extraParams.hxml b/extraParams.hxml new file mode 100644 index 0000000..e53bde4 --- /dev/null +++ b/extraParams.hxml @@ -0,0 +1 @@ +--macro tink.io.Boot.boot() diff --git a/haxe_libraries/tink_io.hxml b/haxe_libraries/tink_io.hxml index 116fd90..a4ad52f 100644 --- a/haxe_libraries/tink_io.hxml +++ b/haxe_libraries/tink_io.hxml @@ -2,4 +2,5 @@ -cp src -lib tink_chunk --lib tink_streams \ No newline at end of file +-lib tink_streams +extraParams.hxml \ No newline at end of file diff --git a/hxformat.json b/hxformat.json new file mode 100644 index 0000000..17000ac --- /dev/null +++ b/hxformat.json @@ -0,0 +1,46 @@ +{ + "indentation": { + "character": " ", + "tabWidth": 2, + "trailingWhitespace": false + }, + "sameLine": { + "functionBody": "keep", + "anonFunctionBody": "keep", + "ifBody": "keep", + "elseBody": "keep", + "forBody": "keep", + "whileBody": "keep", + "doWhileBody": "keep", + "tryBody": "keep", + "catchBody": "keep", + "caseBody": "keep", + "expressionCase": "keep", + "returnBody": "keep", + "returnBodySingleLine": "keep" + }, + "wrapping": { + "maxLineLength": 160, + "callParameter": { "defaultWrap": "keep" }, + "arrayWrap": { "defaultWrap": "keep" }, + "objectLiteral": { "defaultWrap": "keep" }, + "methodChain": { "defaultWrap": "keep" }, + "opBoolChain": { "defaultWrap": "keep" }, + "opAddSubChain": { "defaultWrap": "keep" } + }, + "emptyLines": { + "finalNewline": false, + "maxAnywhereInFile": 2 + }, + "whitespace": { + "ifPolicy": "after", + "forPolicy": "none", + "whilePolicy": "none", + "catchPolicy": "after", + "doPolicy": "after", + "objectFieldColonPolicy": "after", + "colonPolicy": "none", + "arrowFunctionsPolicy": "around", + "binopPolicy": "around" + } +} diff --git a/src/tink/io/Boot.hx b/src/tink/io/Boot.hx new file mode 100644 index 0000000..5e8d2c5 --- /dev/null +++ b/src/tink/io/Boot.hx @@ -0,0 +1,21 @@ +package tink.io; + +#if macro +// this whole file existed because of https://github.com/HaxeFoundation/haxe/issues/12985 +class Boot { + static function boot() { + tink.SyntaxHub.transformMain.whenever(function(e) { + if (haxe.macro.Context.defined("java")) { + return macro { + @:pos(e.pos) tink.io.java.OnMainThread.init(); + $e; + }; + } else { + return e; + } + }); + } +} +#else +#error +#end \ No newline at end of file diff --git a/src/tink/io/Sink.hx b/src/tink/io/Sink.hx index a1796c4..484c177 100644 --- a/src/tink/io/Sink.hx +++ b/src/tink/io/Sink.hx @@ -61,6 +61,13 @@ abstract SinkYielding(SinkObject) case { worker: w }: w; }); + #if (sys && target.threaded) + static public function ofSocket(name:String, socket:sys.net.Socket, pool:tink.io.std.SelectPool, ?options:{ ?worker:Worker }):RealSink + return tink.io.std.SocketSink.wrap(name, socket, pool, switch options { + case null | { worker: null }: Worker.get(); + case { worker: w }: w; + }); + #end } diff --git a/src/tink/io/Source.hx b/src/tink/io/Source.hx index 86ec43a..a7df61d 100644 --- a/src/tink/io/Source.hx +++ b/src/tink/io/Source.hx @@ -97,6 +97,17 @@ abstract Source(SourceObject) from SourceObject to SourceObject to S case v: v; }), 0); } + + #if (sys && target.threaded) + @:noUsing static public inline function ofSocket(name:String, socket:sys.net.Socket, pool:tink.io.std.SelectPool, ?options:{ ?chunkSize: Int, ?worker:Worker }):RealSource { + if (options == null) + options = {}; + return tink.io.std.SocketSource.wrap(name, socket, pool, options.worker.ensure(), switch options.chunkSize { + case null: 0x10000; + case v: v; + }); + } + #end public function chunked():Stream return this; diff --git a/src/tink/io/java/JavaFileSource.hx b/src/tink/io/java/JavaFileSource.hx index 613c52d..f2af69f 100644 --- a/src/tink/io/java/JavaFileSource.hx +++ b/src/tink/io/java/JavaFileSource.hx @@ -66,4 +66,4 @@ private class ReadHandler implements CompletionHandler { cb.invoke(Fail(Error.withData('Read failed for "${parent.name}", reason: ' + exc.getMessage(), exc))); }); } -} \ No newline at end of file +} diff --git a/src/tink/io/java/JavaSocketSink.hx b/src/tink/io/java/JavaSocketSink.hx index f88a5fd..1e7895c 100644 --- a/src/tink/io/java/JavaSocketSink.hx +++ b/src/tink/io/java/JavaSocketSink.hx @@ -46,7 +46,11 @@ class JavaSocketSink extends SinkBase { }); if (options.end) - ret.handle(function (end) channel.shutdownOutput()); + ret.handle(function (end) { + try channel.shutdownOutput() + catch (e:java.nio.channels.ClosedChannelException) {} + catch (e:java.io.IOException) {} + }); return ret.map(function (c) return c.toResult(Noise)); } diff --git a/src/tink/io/java/OnMainThread.hx b/src/tink/io/java/OnMainThread.hx index 8170b54..44b6ada 100644 --- a/src/tink/io/java/OnMainThread.hx +++ b/src/tink/io/java/OnMainThread.hx @@ -1,17 +1,17 @@ package tink.io.java; -/** - Java NIO async channel callbacks run on a JDK thread pool without a Haxe event loop. - Re-dispatch stream completions so downstream code (e.g. haxe.Timer) runs on a safe thread. -**/ class OnMainThread { - public static function run(fn:Void->Void):Void { - #if tink_runloop - tink.RunLoop.current.work(fn); - #elseif java - haxe.MainLoop.add(fn); - #else - fn(); - #end - } -} + public static function init() { + run(noop); + } + + public static function run(fn:Void->Void):Void { + #if haxe5 + sys.thread.Thread.main().events.run(fn); + #else + haxe.EntryPoint.runInMainThread(fn); + #end + } + + static function noop() {} +} \ No newline at end of file diff --git a/src/tink/io/std/SelectPool.hx b/src/tink/io/std/SelectPool.hx new file mode 100644 index 0000000..3146df9 --- /dev/null +++ b/src/tink/io/std/SelectPool.hx @@ -0,0 +1,232 @@ +package tink.io.std; + +#if (sys && target.threaded) + +import sys.net.Socket; +import sys.thread.Thread; +import sys.thread.Mutex; +import sys.thread.Lock; +import haxe.ds.ObjectMap; + +using tink.CoreApi; + +/** + * Multiplexes readiness of `sys.net.Socket`s across a single dedicated thread + * running `Socket.select()`, so `SocketSource`/`SocketSink` can wait for a + * socket to become readable/writable without dedicating a thread per socket. + */ +abstract SelectPool(SelectPoolObject) from SelectPoolObject to SelectPoolObject { + + /** + * Registers a socket with this pool, switching it to non-blocking mode. + * Idempotent: safe (and cheap) to call more than once for the same socket. + */ + public function register(socket:Socket):Void + this.register(socket); + + public function waitReadable(socket:Socket):Future + return this.wait(socket, true); + + public function waitWritable(socket:Socket):Future + return this.wait(socket, false); + + /** + * Stops the select thread. Any waiters still pending are woken so they can + * observe the socket failing on their next I/O attempt. + */ + public function close():Void + this.close(); + + #if test + /** + * Test-only introspection into how often this pool actually fell back to + * `Socket.select()`, as opposed to resolving I/O eagerly. + */ + public function waitReadableCount():Int + return this.waitReadableCount(); + + public function waitWritableCount():Int + return this.waitWritableCount(); + + public function selectCallCount():Int + return this.selectCallCount(); + #end + + static public function create(?name:String, ?worker:Worker):SelectPool + return (new SelectPoolImpl(name, worker) : SelectPoolObject); +} + +interface SelectPoolObject { + function register(socket:Socket):Void; + function wait(socket:Socket, read:Bool):Future; + function close():Void; + #if test + function waitReadableCount():Int; + function waitWritableCount():Int; + function selectCallCount():Int; + #end +} + +private class SelectPoolImpl implements SelectPoolObject { + + var name:String; + var worker:Worker; + var mainThread:Thread; + + var mutex = new Mutex(); + var lock = new Lock(); + + var readWaiters = new ObjectMap>>(); + var writeWaiters = new ObjectMap>>(); + var registered = new ObjectMap(); + + var started = false; + var shutdown = false; + + #if test + var _waitReadableCount = 0; + var _waitWritableCount = 0; + var _selectCallCount = 0; + #end + + public function new(?name, ?worker) { + this.name = name == null ? 'SelectPool' : name; + this.worker = worker == null ? Worker.get() : worker; + this.mainThread = Thread.current(); + } + + public function register(socket:Socket):Void { + var isNew = false; + + mutex.acquire(); + if (!registered.exists(socket)) { + registered.set(socket, true); + isNew = true; + } + mutex.release(); + + if (isNew) { + socket.setBlocking(false); + ensureStarted(); + } + } + + public function wait(socket:Socket, read:Bool):Future { + register(socket); + + return Future.async(function (cb) { + mutex.acquire(); + var waiters = read ? readWaiters : writeWaiters; + var list = waiters.get(socket); + if (list == null) { + list = []; + waiters.set(socket, list); + } + list.push(cb); + #if test + if (read) _waitReadableCount++; else _waitWritableCount++; + #end + mutex.release(); + + lock.release(); + }); + } + + public function close():Void { + shutdown = true; + lock.release(); + } + + #if test + public function waitReadableCount():Int + return _waitReadableCount; + + public function waitWritableCount():Int + return _waitWritableCount; + + public function selectCallCount():Int + return _selectCallCount; + #end + + function ensureStarted():Void { + if (started) return; + + mutex.acquire(); + var shouldStart = !started; + if (shouldStart) started = true; + mutex.release(); + + if (shouldStart) + Thread.create(loop); + } + + function dispatch(cb:Callback):Void { + if (worker == Worker.EAGER) + mainThread.events.run(function () cb.invoke(Noise)); + else + worker.work(function () { + cb.invoke(Noise); + return Noise; + }); + } + + function loop():Void { + while (true) { + mutex.acquire(); + if (shutdown) { + var toWake = collectAll(); + mutex.release(); + for (cb in toWake) dispatch(cb); + return; + } + + var readArr = [for (s in readWaiters.keys()) s]; + var writeArr = [for (s in writeWaiters.keys()) s]; + var otherArr = [for (s in registered.keys()) s]; + mutex.release(); + + if (readArr.length == 0 && writeArr.length == 0 && otherArr.length == 0) { + lock.wait(0.2); + continue; + } + + #if test + _selectCallCount++; + #end + + var result = + try Socket.select(readArr, writeArr, otherArr, 0.1) + catch (e:Dynamic) { read: [], write: [], others: [] }; + + var toNotify = []; + + mutex.acquire(); + collectReady(readWaiters, result.read.concat(result.others), toNotify); + collectReady(writeWaiters, result.write.concat(result.others), toNotify); + mutex.release(); + + for (cb in toNotify) dispatch(cb); + } + } + + function collectReady(waiters:ObjectMap>>, ready:Array, out:Array>):Void { + for (socket in ready) { + var list = waiters.get(socket); + if (list != null) { + waiters.remove(socket); + for (cb in list) out.push(cb); + } + } + } + + function collectAll():Array> { + var out = []; + for (list in readWaiters) for (cb in list) out.push(cb); + for (list in writeWaiters) for (cb in list) out.push(cb); + readWaiters = new ObjectMap(); + writeWaiters = new ObjectMap(); + return out; + } +} + +#end diff --git a/src/tink/io/std/SocketSink.hx b/src/tink/io/std/SocketSink.hx new file mode 100644 index 0000000..5fc91a2 --- /dev/null +++ b/src/tink/io/std/SocketSink.hx @@ -0,0 +1,93 @@ +package tink.io.std; + +#if (sys && target.threaded) + +import haxe.io.*; +import sys.net.Socket; +import tink.streams.Stream; + +using tink.io.PipeResult; +using tink.CoreApi; + +private enum WriteOutcome { + Wrote(bytes:Int); + Eof; + WouldBlock; + Failed(e:Error); +} + +class SocketSink extends tink.io.Sink.SinkBase { + var name:String; + var socket:Socket; + var pool:SelectPool; + var worker:Worker; + + function new(name, socket, pool, worker) { + this.name = name; + this.socket = socket; + this.pool = pool; + this.worker = worker; + } + + override public function consume(source:Stream, options:PipeOptions):Future> { + var rest = Chunk.EMPTY; + + var ret = source.forEach(function (c:Chunk) return Future.async(function (cb) { + + var pos = 0, + bytes = c.toBytes(); + + // Eager-first: attempt the write immediately and only fall back to the + // select pool (and thus `Socket.select()`) once the socket reports + // `Blocked`. This skips select entirely whenever the send buffer + // already has room. + function write() { + if (pos == bytes.length) cb(Resume); + else worker.work(function ():WriteOutcome { + return try { + Wrote(socket.output.writeBytes(bytes, pos, bytes.length - pos)); + } + catch (e:haxe.io.Eof) { + Eof; + } + catch (e:haxe.io.Error) switch e { + case Blocked: WouldBlock; + default: Failed(Error.withData('Error writing to $name', e)); + } + catch (e:TypedError) { + Failed(cast e); + } + catch (e:Dynamic) { + Failed(Error.withData('Error writing to $name', e)); + } + }).handle(function (o) switch o { + case Eof: + rest = (bytes:Chunk).slice(pos, bytes.length); + cb(Finish); + case WouldBlock: + pool.waitWritable(socket).handle(function (_) write()); + case Wrote(v): + pos += v; + if (pos == bytes.length) cb(Resume); + else write(); + case Failed(e): + cb(Clog(e)); + }); + } + + write(); + })); + + if (options.end) + ret.handle(function (end) try socket.shutdown(false, true) catch (e:Dynamic) {}); + + return ret.map(function (c) return c.toResult(Noise, rest)); + } + + static inline public function wrap(name:String, socket:Socket, pool:SelectPool, worker:Worker) { + pool.register(socket); + return new SocketSink(name, socket, pool, worker); + } +} + +#end diff --git a/src/tink/io/std/SocketSource.hx b/src/tink/io/std/SocketSource.hx new file mode 100644 index 0000000..3632e5a --- /dev/null +++ b/src/tink/io/std/SocketSource.hx @@ -0,0 +1,76 @@ +package tink.io.std; + +#if (sys && target.threaded) + +import haxe.io.*; +import sys.net.Socket; +import tink.streams.Stream; + +using tink.CoreApi; + +private enum ReadOutcome { + Done(step:Step); + WouldBlock; +} + +class SocketSource extends Generator { + public function new(name:String, socket:Socket, pool:SelectPool, worker:Worker, buf:Bytes, offset:Int) { + + function next(buf, offset) + return new SocketSource(name, socket, pool, worker, buf, offset); + + var free = buf.length - offset; + + // Eager-first: attempt the read immediately and only fall back to the + // select pool (and thus `Socket.select()`) once the socket reports + // `Blocked`. This skips select entirely whenever data is already there. + super(Future.async(function (cb) { + function attempt() + worker.work(function ():ReadOutcome { + return try { + var read = socket.input.readBytes(buf, offset, free); + + if (read == 0) + Done(Link(tink.Chunk.EMPTY, next(buf, offset))); + else { + + var nextOffset = + if (free - read < 0x400) 0; + else offset + read; + + var nextBuf = + if (nextOffset == 0) Bytes.alloc(buf.length); + else buf; + + Done(Link( + (buf:Chunk).slice(offset, offset + read), + next(nextBuf, nextOffset) + )); + } + } + catch (e:haxe.io.Eof) { + Done(End); + } + catch (e:haxe.io.Error) switch e { + case Blocked: + WouldBlock; + default: + Done(Fail(Error.withData('Failed to read from $name', e))); + } + }).handle(function (outcome) switch outcome { + case WouldBlock: + pool.waitReadable(socket).handle(function (_) attempt()); + case Done(step): + cb(step); + }); + attempt(); + } #if !tink_core_2 , true #end)); + } + + static inline public function wrap(name:String, socket:Socket, pool:SelectPool, worker:Worker, size:Int) { + pool.register(socket); + return new SocketSource(name, socket, pool, worker, Bytes.alloc(size), 0); + } +} + +#end diff --git a/tests.hxml b/tests.hxml index 73cd01d..918308c 100644 --- a/tests.hxml +++ b/tests.hxml @@ -3,5 +3,6 @@ -dce full -D eval-call-stack-depth=10000 +-D test -lib tink_unittest diff --git a/tests/PipeTest.hx b/tests/PipeTest.hx index 543b7d5..69921a3 100644 --- a/tests/PipeTest.hx +++ b/tests/PipeTest.hx @@ -10,107 +10,105 @@ using sys.io.File; using sys.FileSystem; @:asserts -@:timeout(15000) +// @:timeout(15000) class PipeTest { - public function new() {} - + public function copyFile() { var fsrc = 'example.blob'; var fdst = 'copy.blob'; - if(!fsrc.exists()) { + if (!fsrc.exists()) { var s = 'some random content'; - for (i in 0...20) + for(i in 0...20) s += s; - s+='123456789'; + s += '123456789'; fsrc.saveContent(s); } - - var src = readFile(fsrc, 5249710), - dst = writeFile(fdst); + + var src = readFile(fsrc, 5249710), dst = writeFile(fdst); var start = haxe.Timer.stamp(); src.pipeTo(dst, {end: true}).handle(function(o) { - trace(haxe.Timer.stamp() - start); - asserts.assert(o == AllWritten); - asserts.assert(fsrc.getContent().length == fdst.getContent().length); - asserts.done(); + trace(haxe.Timer.stamp() - start); + asserts.assert(o == AllWritten); + asserts.assert(fsrc.getContent().length == fdst.getContent().length); + asserts.done(); }); - + return asserts; } function readFile(name:String, ?chunkSize) { #if nodejs - return - Source.ofNodeStream( - 'Input from file $name', - js.node.Fs.createReadStream(name), - { chunkSize: chunkSize } - ); + return + Source.ofNodeStream( + 'Input from file $name', + js.node.Fs.createReadStream(name), + {chunkSize: chunkSize} + ); #elseif cs - return - Source.ofCsStream( - 'Input from file $name', - new cs.system.io.FileStream( - name, - cs.system.io.FileMode.Open, - cs.system.io.FileAccess.Read, - cs.system.io.FileShare.ReadWrite - ), - { chunkSize: chunkSize } - ); + return + Source.ofCsStream( + 'Input from file $name', + new cs.system.io.FileStream( + name, + cs.system.io.FileMode.Open, + cs.system.io.FileAccess.Read, + cs.system.io.FileShare.ReadWrite + ), + {chunkSize: chunkSize} + ); #elseif java - var path = java.nio.file.Paths.get(name); - var op:java.nio.file.OpenOption = cast java.nio.file.StandardOpenOption.READ; - var channel = java.nio.channels.AsynchronousFileChannel.open(path, op); - return - Source.ofJavaFileChannel( - 'Input from file $name', - channel, - { chunkSize: chunkSize } - ); + var path = java.nio.file.Paths.get(name); + var op:java.nio.file.OpenOption = cast java.nio.file.StandardOpenOption.READ; + var channel = java.nio.channels.AsynchronousFileChannel.open(path, op); + return + Source.ofJavaFileChannel( + 'Input from file $name', + channel, + {chunkSize: chunkSize} + ); #elseif sys - return - Source.ofInput( - 'Input from file $name', - sys.io.File.read(name, true), - { chunkSize: chunkSize } - ); + return + Source.ofInput( + 'Input from file $name', + sys.io.File.read(name, true), + {chunkSize: chunkSize} + ); #else - #error "not implemented" - #end + #error "not implemented" + #end } function writeFile(name:String) { #if nodejs - return - Sink.ofNodeStream('Output to file $name', js.node.Fs.createWriteStream(name)); + return + Sink.ofNodeStream('Output to file $name', js.node.Fs.createWriteStream(name)); #elseif cs - return - Sink.ofCsStream('Output to file $name', new cs.system.io.FileStream( - name, - cs.system.io.FileMode.OpenOrCreate, - cs.system.io.FileAccess.Write, - cs.system.io.FileShare.ReadWrite + return + Sink.ofCsStream('Output to file $name', new cs.system.io.FileStream( + name, + cs.system.io.FileMode.OpenOrCreate, + cs.system.io.FileAccess.Write, + cs.system.io.FileShare.ReadWrite )); #elseif java - var path = java.nio.file.Paths.get(name); - var create:java.nio.file.OpenOption = cast java.nio.file.StandardOpenOption.CREATE; - var write:java.nio.file.OpenOption = cast java.nio.file.StandardOpenOption.WRITE; - var channel = java.nio.channels.AsynchronousFileChannel.open(path, create, write); - return - Sink.ofJavaFileChannel('Output to file $name', channel); + var path = java.nio.file.Paths.get(name); + var create:java.nio.file.OpenOption = cast java.nio.file.StandardOpenOption.CREATE; + var write:java.nio.file.OpenOption = cast java.nio.file.StandardOpenOption.WRITE; + var channel = java.nio.channels.AsynchronousFileChannel.open(path, create, write); + return + Sink.ofJavaFileChannel('Output to file $name', channel); #elseif sys - return - Sink.ofOutput('Output to file $name', sys.io.File.write(name, true)); + return + Sink.ofOutput('Output to file $name', sys.io.File.write(name, true)); #else - #error "not implemented" + #error "not implemented" #end } - + public function empty() { var dst = 'empty.blob'; var sink = writeFile(dst); @@ -120,7 +118,7 @@ class PipeTest { return Noise; }); } - + Future.ofMany([ _pipe(''), _pipe(Source.EMPTY), @@ -128,8 +126,7 @@ class PipeTest { // asserts.assert(dst.getContent().length == 0); asserts.done(); }); - + return asserts; } - } \ No newline at end of file diff --git a/tests/RunTests.hx b/tests/RunTests.hx index 2377c25..ce91da1 100644 --- a/tests/RunTests.hx +++ b/tests/RunTests.hx @@ -4,23 +4,14 @@ import tink.unit.*; import tink.testrunner.*; class RunTests { - static function main() { - #if (java && jvm) - // Touch EntryPoint on the real main thread before any JDK pool thread - // (e.g. an AsynchronousSocketChannel/AsynchronousFileChannel completion) - // can trigger its static init and wrongly capture a pool thread as - // "main" (which has no event loop, causing "Event loop is not available"). - haxe.MainLoop.add(function() {}); - #end - Runner.run(TestBatch.make([ - #if (sys || nodejs) new PipeTest(),#end + #if (sys || nodejs) new PipeTest(), #end + #if (sys && target.threaded && !jvm) new SocketTest(), #end new SourceTest(), new ParserTest(), new CastTest(), #if (js && !nodejs) new JsTest(), #end ])).handle(Runner.exit); } - } \ No newline at end of file diff --git a/tests/SocketTest.hx b/tests/SocketTest.hx new file mode 100644 index 0000000..41d9a49 --- /dev/null +++ b/tests/SocketTest.hx @@ -0,0 +1,142 @@ +package; + +#if (sys && target.threaded) +import sys.net.Socket; +import sys.net.Host; +import tink.io.Source; +import tink.io.Sink; +import tink.io.PipeResult; +import tink.io.std.SelectPool; + +using tink.io.Source; +using tink.CoreApi; + +@:asserts +@:timeout(15000) +class SocketTest { + public function new() {} + + public function echo() { + var pool = SelectPool.create(); + + var server = new Socket(); + server.bind(new Host('127.0.0.1'), 0); + server.listen(1); + var port = server.host().port; + + var client = new Socket(); + client.connect(new Host('127.0.0.1'), port); + + var accepted = server.accept(); + server.close(); + + var payload = 'hello from the select pool'; + for(i in 0...10) payload += payload; + + var src:IdealSource = payload; + var dst = Sink.ofSocket('client sink', client, pool); + + src.pipeTo(dst, {end: true}).handle(function(o) { + asserts.assert(o == AllWritten); + var readBack = Source.ofSocket('accepted source', accepted, pool); + readBack.all().handle(function(o) switch o { + case Success(chunk): + final ok = chunk == payload; + asserts.assert(ok); + accepted.close(); + pool.close(); + asserts.done(); + case Failure(e): + asserts.fail(e); + }); + }); + + return asserts; + } + + // Starting the read before any data has been sent forces the first read + // attempt to hit `Blocked`, which routes it through `pool.waitReadable()` + // and thus `Socket.select()` (unlike `echo()`, where the payload is + // already fully written by the time the read starts). + public function readWaitsForData() { + var pool = SelectPool.create(); + + var server = new Socket(); + server.bind(new Host('127.0.0.1'), 0); + server.listen(1); + var port = server.host().port; + + var client = new Socket(); + client.connect(new Host('127.0.0.1'), port); + + var accepted = server.accept(); + server.close(); + + var payload = 'data sent only after the reader is already blocked on select()'; + + var readBack = Source.ofSocket('accepted source', accepted, pool); + readBack.all().handle(function(o) switch o { + case Success(chunk): + asserts.assert(chunk == payload); + asserts.assert(pool.waitReadableCount() > 0); + asserts.assert(pool.selectCallCount() > 0); + client.close(); + pool.close(); + asserts.done(); + case Failure(e): + asserts.fail(e); + }); + + var src:IdealSource = payload; + var dst = Sink.ofSocket('client sink', client, pool); + src.pipeTo(dst, {end: true}).handle(function(_) {}); + + return asserts; + } + + // Flooding a peer that isn't draining yet forces the write to fill the + // send/receive buffers and hit `Blocked`, which routes it through + // `pool.waitWritable()` and thus `Socket.select()`. + public function writeWaitsForDrain() { + var pool = SelectPool.create(); + + var server = new Socket(); + server.bind(new Host('127.0.0.1'), 0); + server.listen(1); + var port = server.host().port; + + var client = new Socket(); + client.connect(new Host('127.0.0.1'), port); + + var accepted = server.accept(); + server.close(); + + var payload = 'x'; + for(i in 0...23) payload += payload; // ~8MB, comfortably larger than OS socket buffers + + var src:IdealSource = payload; + var dst = Sink.ofSocket('client sink', client, pool); + + src.pipeTo(dst, {end: true}).handle(function(o) { + asserts.assert(o == AllWritten); + }); + + // Draining is only started after the write above, so it has to fill the + // buffers and register with the select pool before any bytes are consumed. + var readBack = Source.ofSocket('accepted source', accepted, pool); + readBack.all().handle(function(o) switch o { + case Success(chunk): + asserts.assert(chunk.length == payload.length); + asserts.assert(pool.waitWritableCount() > 0); + asserts.assert(pool.selectCallCount() > 0); + accepted.close(); + pool.close(); + asserts.done(); + case Failure(e): + asserts.fail(e); + }); + + return asserts; + } +} +#end