status-im
/
nim-libp2p
mirror of https://github.com/status-im/nim-libp2p.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

emulate `defered`

This commit is contained in:
Dmitriy Ryajov 2020-05-25 08:33:24 -06:00
parent abf659a01a
commit 1b4876d26d
3 changed files with 117 additions and 125 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

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

37
libp2p/protocols/pubsub/pubsubpeer.nim
View File

@ -66,28 +66,29 @@ proc sendObservers(p: PubSubPeer, msg: var RPCMsg) =
proc handle*(p: PubSubPeer, conn: Connection) {.async.} = proc handle*(p: PubSubPeer, conn: Connection) {.async.} =
trace "handling pubsub rpc", peer = p.id, closed = conn.closed trace "handling pubsub rpc", peer = p.id, closed = conn.closed
try: try:
while not conn.closed: try:
trace "waiting for data", peer = p.id, closed = conn.closed while not conn.closed:
let data = await conn.readLp(64 * 1024) trace "waiting for data", peer = p.id, closed = conn.closed
let digest = $(sha256.digest(data)) let data = await conn.readLp(64 * 1024)
trace "read data from peer", peer = p.id, data = data.shortLog let digest = $(sha256.digest(data))
if digest in p.recvdRpcCache: trace "read data from peer", peer = p.id, data = data.shortLog
trace "message already received, skipping", peer = p.id if digest in p.recvdRpcCache:
continue trace "message already received, skipping", peer = p.id
continue
var msg = decodeRpcMsg(data) var msg = decodeRpcMsg(data)
trace "decoded msg from peer", peer = p.id, msg = msg.shortLog trace "decoded msg from peer", peer = p.id, msg = msg.shortLog
# trigger hooks
for obs in p.observers[]:
obs.onRecv(p, msg)
await p.handler(p, @[msg])
p.recvdRpcCache.put(digest)
finally:
trace "exiting pubsub peer read loop", peer = p.id
await conn.close()
p.recvObservers(msg) # hooks can modify the message
await p.handler(p, @[msg])
p.recvdRpcCache.put(digest)
except CatchableError as exc: except CatchableError as exc:
trace "Exception occurred in PubSubPeer.handle", exc = exc.msg trace "Exception occurred in PubSubPeer.handle", exc = exc.msg
finally:
trace "exiting pubsub peer read loop", peer = p.id
if not conn.closed():
await conn.close()
proc send*(p: PubSubPeer, msgs: seq[RPCMsg]) {.async.} = proc send*(p: PubSubPeer, msgs: seq[RPCMsg]) {.async.} =
for m in msgs.items: for m in msgs.items: