status-im/nim-libp2p
2
0
Fork 0
mirror of https://github.com/status-im/nim-libp2p.git synced 2025-02-27 20:10:50 +00:00
Code Issues Projects Releases Wiki Activity

wip

Browse Source
This commit is contained in:
Dmitriy Ryajov 2020-06-02 10:08:02 -06:00
parent 3ddc4bbebd
commit ea2092580a
No known key found for this signature in database
GPG Key ID: DA8C680CE7C657A4
2 changed files with 107 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

203
libp2p/muxers/mplex/mplex.nim
View File

@ -67,102 +67,101 @@ proc newStreamInternal*(m: Mplex,
method handle*(m: Mplex) {.async, gcsafe.} = method handle*(m: Mplex) {.async, gcsafe.} =
trace "starting mplex main loop", oid = m.oid trace "starting mplex main loop", oid = m.oid
try: try:
try: while not m.connection.closed:
while not m.connection.closed: trace "waiting for data", oid = m.oid
trace "waiting for data", oid = m.oid let (id, msgType, data) = await m.connection.readMsg()
let (id, msgType, data) = await m.connection.readMsg() trace "read message from connection", id = id,
trace "read message from connection", id = id,
msgType = msgType,
data = data.shortLog,
oid = m.oid
let initiator = bool(ord(msgType) and 1)
var channel: LPChannel
if MessageType(msgType) != MessageType.New:
let channels = m.getChannelList(initiator)
if id notin channels:
trace "Channel not found, skipping", id = id,
initiator = initiator,
msg = msgType,
oid = m.oid
continue
channel = channels[id]
case msgType:
of MessageType.New:
let name = cast[string](data)
channel = await m.newStreamInternal(false, id, name)
trace "created channel", id = id,
name = name,
inititator = channel.initiator,
channoid = channel.oid,
oid = m.oid
if not isNil(m.streamHandler):
let stream = newConnection(channel)
m.conns.add(stream)
stream.peerInfo = m.connection.peerInfo
var fut = newFuture[void]()
proc handler() {.async.} =
try:
await m.streamHandler(stream)
except CatchableError as exc:
trace "exception in stream handler", exc = exc.msg
finally:
m.conns.keepItIf(it != stream)
m.handlerFuts.keepItIf(it != fut)
fut = handler()
of MessageType.MsgIn, MessageType.MsgOut:
trace "pushing data to channel", id = id,
initiator = initiator,
msgType = msgType, msgType = msgType,
size = data.len, data = data.shortLog,
name = channel.name,
channoid = channel.oid,
oid = m.oid oid = m.oid
let initiator = bool(ord(msgType) and 1)
var channel: LPChannel
if MessageType(msgType) != MessageType.New:
let channels = m.getChannelList(initiator)
if id notin channels:
trace "Channel not found, skipping", id = id,
initiator = initiator,
msg = msgType,
oid = m.oid
continue
channel = channels[id]
if data.len > MaxMsgSize: case msgType:
raise newLPStreamLimitError() of MessageType.New:
await channel.pushTo(data) let name = cast[string](data)
of MessageType.CloseIn, MessageType.CloseOut: channel = await m.newStreamInternal(false, id, name)
trace "closing channel", id = id, trace "created channel", id = id,
initiator = initiator, name = name,
msgType = msgType, inititator = channel.initiator,
name = channel.name, channoid = channel.oid,
channoid = channel.oid, oid = m.oid
oid = m.oid if not isNil(m.streamHandler):
let stream = newConnection(channel)
m.conns.add(stream)
stream.peerInfo = m.connection.peerInfo
await channel.closeRemote() var fut = newFuture[void]()
m.getChannelList(initiator).del(id) proc handler() {.async.} =
trace "deleted channel", id = id, try:
initiator = initiator, await m.streamHandler(stream)
msgType = msgType, except CatchableError as exc:
name = channel.name, trace "exception in stream handler", exc = exc.msg
channoid = channel.oid, finally:
oid = m.oid m.conns.keepItIf(it != stream)
of MessageType.ResetIn, MessageType.ResetOut: m.handlerFuts.keepItIf(it != fut)
trace "resetting channel", id = id,
initiator = initiator,
msgType = msgType,
name = channel.name,
channoid = channel.oid,
oid = m.oid
await channel.reset() fut = handler()
m.getChannelList(initiator).del(id)
trace "deleted channel", id = id, of MessageType.MsgIn, MessageType.MsgOut:
initiator = initiator, trace "pushing data to channel", id = id,
msgType = msgType, initiator = initiator,
name = channel.name, msgType = msgType,
channoid = channel.oid, size = data.len,
oid = m.oid name = channel.name,
break channoid = channel.oid,
finally: oid = m.oid
trace "stopping mplex main loop", oid = m.oid
await m.close() if data.len > MaxMsgSize:
raise newLPStreamLimitError()
await channel.pushTo(data)
of MessageType.CloseIn, MessageType.CloseOut:
trace "closing channel", id = id,
initiator = initiator,
msgType = msgType,
name = channel.name,
channoid = channel.oid,
oid = m.oid
await channel.closeRemote()
m.getChannelList(initiator).del(id)
trace "deleted channel", id = id,
initiator = initiator,
msgType = msgType,
name = channel.name,
channoid = channel.oid,
oid = m.oid
of MessageType.ResetIn, MessageType.ResetOut:
trace "resetting channel", id = id,
initiator = initiator,
msgType = msgType,
name = channel.name,
channoid = channel.oid,
oid = m.oid
await channel.reset()
m.getChannelList(initiator).del(id)
trace "deleted channel", id = id,
initiator = initiator,
msgType = msgType,
name = channel.name,
channoid = channel.oid,
oid = m.oid
break
except CatchableError as exc: except CatchableError as exc:
trace "Exception occurred", exception = exc.msg, oid = m.oid trace "Exception occurred", exception = exc.msg, oid = m.oid
finally:
trace "stopping mplex main loop", oid = m.oid
await m.close()
proc newMplex*(conn: Connection, proc newMplex*(conn: Connection,
maxChanns: uint = MaxChannels): Mplex = maxChanns: uint = MaxChannels): Mplex =
@ -200,16 +199,26 @@ method close*(m: Mplex) {.async, gcsafe.} =
try: try:
trace "closing mplex muxer", oid = m.oid trace "closing mplex muxer", oid = m.oid
await all( let channs = toSeq(m.remote.values) &
toSeq(m.remote.values).mapIt(it.reset()) & toSeq(m.local.values)
toSeq(m.local.values).mapIt(it.reset()))
for chann in channs:
try:
await chann.reset()
except CatchableError as exc:
warn "error resetting channel", exc = exc.msg
for conn in m.conns:
try:
await conn.close()
except CatchableError as exc:
warn "error closing channel's connection"
checkFutures(
await allFinished(m.handlerFuts))
await all(m.conns.mapIt(it.close())) # dispose of channel's connections
await all(m.handlerFuts)
except CatchableError as exc:
trace "exception in mplex close", exc = exc.msg
finally:
await m.connection.close() await m.connection.close()
finally:
m.remote.clear() m.remote.clear()
m.local.clear() m.local.clear()
m.conns = @[] m.conns = @[]

2
libp2p/muxers/muxer.nim
View File

@ -60,7 +60,7 @@ method init(c: MuxerProvider) =
if not isNil(c.muxerHandler): if not isNil(c.muxerHandler):
futs &= c.muxerHandler(muxer) futs &= c.muxerHandler(muxer)
await all(futs) checkFutures(await allFinished(futs))
except CatchableError as exc: except CatchableError as exc:
trace "exception in muxer handler", exc = exc.msg trace "exception in muxer handler", exc = exc.msg