From a0aaa29739853d4503f7fd0d2ddfa444c35da664 Mon Sep 17 00:00:00 2001 From: Dmitriy Ryajov Date: Tue, 3 Sep 2019 19:42:00 -0600 Subject: [PATCH] wip: mplex and mplex tests --- libp2p/muxers/mplex.nim | 97 +++++++++++++++++++++++------------------ tests/testmplex.nim | 51 ++++++++++++++++++---- 2 files changed, 98 insertions(+), 50 deletions(-) diff --git a/libp2p/muxers/mplex.nim b/libp2p/muxers/mplex.nim index 8b2ae5d..31e2067 100644 --- a/libp2p/muxers/mplex.nim +++ b/libp2p/muxers/mplex.nim @@ -7,11 +7,11 @@ ## This file may not be copied, modified, or distributed except according to ## those terms. -import sequtils +import tables, sequtils import chronos import ../varint, ../connection, ../vbuffer, ../protocol, - ../stream/bufferstream, + ../stream/bufferstream, ../stream/lpstream, muxer const MaxMsgSize* = 1 shl 20 # 1mb @@ -31,8 +31,8 @@ type StreamHandler = proc(conn: Connection): Future[void] {.gcsafe.} Mplex* = ref object of Muxer - remote*: seq[Channel] - local*: seq[Channel] + remote*: Table[int, Channel] + local*: Table[int, Channel] currentId*: int maxChannels*: uint streamHandler*: StreamHandler @@ -67,7 +67,7 @@ proc readHeader*(conn: Connection): Future[(uint, MessageType)] {.async, gcsafe. if res != VarintStatus.Success: buffer.setLen(0) return - except TransportIncompleteError: + except LPStreamIncompleteError: buffer.setLen(0) proc writeHeader*(conn: Connection, @@ -77,7 +77,8 @@ proc writeHeader*(conn: Connection, ## write lenght prefixed var buf = initVBuffer() buf.writeVarint(LPSomeUVarint(id.uint shl 3 or msgType.uint)) - buf.writeVarint(LPSomeUVarint(size.uint)) + if size > 0: + buf.writeVarint(LPSomeUVarint(size.uint)) buf.finish() result = conn.write(buf.buffer) @@ -94,55 +95,64 @@ proc newChannel*(mplex: Mplex, result.id = id result.mplex = mplex result.initiator = initiator - result.writeHandler = handler - result.maxSize = size + result.initBufferStream(handler, size) proc closed*(s: Channel): bool = s.closedLocal and s.closedRemote proc close*(s: Channel) {.async.} = discard proc reset*(s: Channel) {.async.} = discard ########################################## -## ## Mplex -## ########################################## -proc getChannelList(m: Mplex, initiator: bool): var seq[Channel] = - if initiator: +proc getChannelList(m: Mplex, initiator: bool): var Table[int, Channel] = + if initiator: result = m.remote else: result = m.local -proc newStream*(m: Mplex, - chanId: int = -1, - initiator: bool = true): - Future[Connection] {.async, gcsafe.} = +proc newStreamInternal*(m: Mplex, + initiator: bool = true, + chanId: int): + Future[Channel] {.async, gcsafe.} = ## create new channel/stream - defer: inc(m.currentId) - let id = if chanId > -1: chanId else: m.currentId + let id = if initiator: m.currentId.inc(); m.currentId else: chanId proc writeHandler(data: seq[byte]): Future[void] {.async, gcsafe.} = - let msgType = if initiator: MessageType.MsgIn else: MessageType.MsgOut + let msgType = if initiator: MessageType.MsgOut else: MessageType.MsgIn await m.connection.writeHeader(id, msgType, data.len) # write header await m.connection.write(data) # write data - let channel = newChannel(m, id, initiator, writeHandler) - m.getChannelList(initiator)[id] = channel - result = newConnection(channel) + result = newChannel(m, id, initiator, writeHandler) + m.getChannelList(initiator)[id] = result -proc handle*(m: Mplex) {.async, gcsafe.} = - while not m.connection.closed: - let (id, msgType) = await m.connection.readHeader() - let initiator = bool(ord(msgType) and 1) - case msgType: - of MessageType.New: - await m.streamHandler(await m.newStream(id.int, false)) - of MessageType.MsgIn, MessageType.MsgOut: - await m.getChannelList(initiator)[id.int].pushTo(await m.connection.readLp()) - of MessageType.CloseIn, MessageType.CloseOut: - await m.getChannelList(initiator)[id.int].close() - of MessageType.ResetIn, MessageType.ResetOut: - await m.getChannelList(initiator)[id.int].reset() - else: raise newMplexUnknownMsgError() +proc newStreamInternal*(m: Mplex): Future[Channel] {.gcsafe.} = + result = m.newStreamInternal(true, 0) + +proc handle*(m: Mplex): Future[void] {.async, gcsafe.} = + try: + while not m.connection.closed: + let (id, msgType) = await m.connection.readHeader() + let initiator = bool(ord(msgType) and 1) + case msgType: + of MessageType.New: + let channel = await m.newStreamInternal(false, id.int) + await m.streamHandler(newConnection(channel)) + of MessageType.MsgIn, MessageType.MsgOut: + let channel = m.getChannelList(initiator)[id.int] + let msg = await m.connection.readLp() + await channel.pushTo(msg) + of MessageType.CloseIn, MessageType.CloseOut: + let channel = m.getChannelList(initiator)[id.int] + await channel.close() + of MessageType.ResetIn, MessageType.ResetOut: + let channel = m.getChannelList(initiator)[id.int] + await channel.reset() + else: raise newMplexUnknownMsgError() + except Exception as exc: + #TODO: add proper loging + discard + finally: + await m.connection.close() proc newMplex*(conn: Connection, streamHandler: StreamHandler, @@ -151,12 +161,15 @@ proc newMplex*(conn: Connection, result.connection = conn result.maxChannels = maxChanns result.streamHandler = streamHandler + result.remote = initTable[int, Channel]() + result.local = initTable[int, Channel]() -method newStream*(m: Mplex): Future[Connection] {.gcsafe.} = - result = m.newStream(true) +method newStream*(m: Mplex): Future[Connection] {.async, gcsafe.} = + let channel = await m.newStreamInternal() + await m.connection.writeHeader(channel.id, MessageType.New, 0) + result = newConnection(channel) -method close(m: Mplex) {.async, gcsafe.} = - let futs = @[allFutures(m.remote.mapIt(it.close())), - allFutures(m.local.mapIt(it.close()))] - await allFutures(futs) +method close*(m: Mplex) {.async, gcsafe.} = + await allFutures(@[allFutures(toSeq(m.remote.values).mapIt(it.close())), + allFutures(toSeq(m.local.values).mapIt(it.close()))]) await m.connection.close() diff --git a/tests/testmplex.nim b/tests/testmplex.nim index ee83867..a4d4f53 100644 --- a/tests/testmplex.nim +++ b/tests/testmplex.nim @@ -43,7 +43,7 @@ suite "Mplex": # check msg == fromHex("880102") # let conn = newConnection(newTestEncodeStream(encHandler)) - # await conn.writeHeader(uint(17), MessageType.New, 2) + # await conn.writeHeader(17, MessageType.New, 2) # result = true # check: @@ -63,15 +63,16 @@ suite "Mplex": # check: # waitFor(testDecodeHeader()) == true - test "e2e - new stream": + test "e2e - read/write initiator": proc testNewStream(): Future[bool] {.async.} = - let ma: MultiAddress = Multiaddress.init("/ip4/127.0.0.1/tcp/53351") + let ma: MultiAddress = Multiaddress.init("/ip4/127.0.0.1/tcp/53380") proc connHandler(conn: Connection) {.async, gcsafe.} = - proc handleListen(stream: Connection) {.async, gcsafe.} = + proc handleMplexListen(stream: Connection) {.async, gcsafe.} = await stream.writeLp("Hello from stream!") + await stream.close() - let mplexListen = newMplex(conn, handleListen) + let mplexListen = newMplex(conn, handleMplexListen) await mplexListen.handle() let transport1: TcpTransport = newTransport(TcpTransport) @@ -79,11 +80,45 @@ suite "Mplex": let transport2: TcpTransport = newTransport(TcpTransport) let conn = await transport2.dial(ma) - proc handleDial(stream: Connection) {.async, gcsafe.} = - let msg = await stream.readLp() + proc handleDial(stream: Connection) {.async, gcsafe.} = discard let mplexDial = newMplex(conn, handleDial) - let handleFut = mplexDial.handle() + let dialFut = mplexDial.handle() + let stream = await mplexDial.newStream() + check cast[string](await stream.readLp()) == "Hello from stream!" + + await conn.close() + await dialFut + result = true + + check: + waitFor(testNewStream()) == true + + test "e2e - read/write receiver": + proc testNewStream(): Future[bool] {.async.} = + let ma: MultiAddress = Multiaddress.init("/ip4/127.0.0.1/tcp/53381") + + proc connHandler(conn: Connection) {.async, gcsafe.} = + proc handleMplexListen(stream: Connection) {.async, gcsafe.} = + let msg = await stream.readLp() + check cast[string](msg) == "Hello from stream!" + await stream.close() + + let mplexListen = newMplex(conn, handleMplexListen) + await mplexListen.handle() + + let transport1: TcpTransport = newTransport(TcpTransport) + await transport1.listen(ma, connHandler) + + let transport2: TcpTransport = newTransport(TcpTransport) + let conn = await transport2.dial(ma) + + proc handleDial(stream: Connection) {.async, gcsafe.} = discard + let mplexDial = newMplex(conn, handleDial) + let dialFut = mplexDial.handle() + let stream = await mplexDial.newStream() + await stream.writeLp("Hello from stream!") + await dialFut result = true check: