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

Cleanup lpstream, Connection and BufferStream (#228) 

* count published messages

* don't call `switch.dial` in `subscribeToPeer`

* don't use delegation in connection

* move connection out to own file

* don't breakout on reset

* make sure to call close on secured conn

* add lpstream tracing

* don't breackdown by conn id

* fix import

* remove unused lable

* reset  connection on exception

* add additional metrics for skipped messages

* check for nil in secure.close
This commit is contained in:
Dmitriy Ryajov 2020-06-19 11:29:43 -06:00 committed by GitHub
parent 719744f46a
commit 5b28e8c488
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
39 changed files with 244 additions and 293 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
examples/directchat.nim
View File

@ -8,7 +8,7 @@ import ../libp2p/[switch, # manage transports, a single entry
crypto/crypto, # cryptographic functions
errors, # error handling utilities
protocols/identify, # identify the peer info of a peer
connection, # create and close stream read / write connections
stream/connection, # create and close stream read / write connections
transports/transport, # listen and dial to other peers using p2p protocol
transports/tcptransport, # listen and dial to other peers using client-server protocol
multiaddress, # encode different addressing schemes. For example, /ip4/7.7.7.7/tcp/6543 means it is using IPv4 protocol and TCP

139
libp2p/connection.nim
View File

@ -1,139 +0,0 @@
## Nim-LibP2P
## Copyright (c) 2019 Status Research & Development GmbH
## Licensed under either of
## * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
## * MIT license ([LICENSE-MIT](LICENSE-MIT))
## at your option.
## This file may not be copied, modified, or distributed except according to
## those terms.
import chronos, chronicles, metrics
import peerinfo,
errors,
multiaddress,
stream/lpstream,
peerinfo
when chronicles.enabledLogLevel == LogLevel.TRACE:
import oids
export lpstream
logScope:
topics = "connection"
const
ConnectionTrackerName* = "libp2p.connection"
type
Connection* = ref object of LPStream
peerInfo*: PeerInfo
stream*: LPStream
observedAddr*: Multiaddress
ConnectionTracker* = ref object of TrackerBase
opened*: uint64
closed*: uint64
proc setupConnectionTracker(): ConnectionTracker {.gcsafe.}
proc getConnectionTracker*(): ConnectionTracker {.gcsafe.} =
result = cast[ConnectionTracker](getTracker(ConnectionTrackerName))
if isNil(result):
result = setupConnectionTracker()
proc dumpTracking(): string {.gcsafe.} =
var tracker = getConnectionTracker()
result = "Opened conns: " & $tracker.opened & "\n" &
"Closed conns: " & $tracker.closed
proc leakTransport(): bool {.gcsafe.} =
var tracker = getConnectionTracker()
result = (tracker.opened != tracker.closed)
proc setupConnectionTracker(): ConnectionTracker =
result = new ConnectionTracker
result.opened = 0
result.closed = 0
result.dump = dumpTracking
result.isLeaked = leakTransport
addTracker(ConnectionTrackerName, result)
declareGauge libp2p_open_connection, "open Connection instances"
proc `$`*(conn: Connection): string =
if not isNil(conn.peerInfo):
result = conn.peerInfo.id
proc init[T: Connection](self: var T, stream: LPStream): T =
## create a new Connection for the specified async reader/writer
new self
self.stream = stream
self.initStream()
return self
proc newConnection*(stream: LPStream): Connection =
## create a new Connection for the specified async reader/writer
result.init(stream)
method initStream*(s: Connection) =
procCall LPStream(s).initStream()
trace "created connection", oid = s.oid
inc getConnectionTracker().opened
libp2p_open_connection.inc()
method readExactly*(s: Connection,
pbytes: pointer,
nbytes: int):
Future[void] {.async, gcsafe.} =
await s.stream.readExactly(pbytes, nbytes)
method readOnce*(s: Connection,
pbytes: pointer,
nbytes: int):
Future[int] {.async, gcsafe.} =
result = await s.stream.readOnce(pbytes, nbytes)
method write*(s: Connection,
msg: seq[byte]):
Future[void] {.async, gcsafe.} =
await s.stream.write(msg)
method atEof*(s: Connection): bool {.inline.} =
if isNil(s.stream):
return true
s.stream.atEof
method closed*(s: Connection): bool =
if isNil(s.stream):
return true
result = s.stream.closed
method close*(s: Connection) {.async, gcsafe.} =
try:
if not s.isClosed:
s.isClosed = true
trace "about to close connection", closed = s.closed,
conn = $s,
oid = s.oid
if not isNil(s.stream) and not s.stream.closed:
trace "closing child stream", closed = s.closed,
conn = $s,
oid = s.stream.oid
await s.stream.close()
# s.stream = nil
s.closeEvent.fire()
trace "connection closed", closed = s.closed,
conn = $s,
oid = s.oid
inc getConnectionTracker().closed
libp2p_open_connection.dec()
except CatchableError as exc:
trace "exception closing connections", exc = exc.msg

2
libp2p/multistream.nim
View File

@ -9,7 +9,7 @@
import strutils
import chronos, chronicles, stew/byteutils
import connection,
import stream/connection,
vbuffer,
errors,
protocols/protocol

5
libp2p/muxers/mplex/coder.nim
View File

@ -10,11 +10,10 @@
import chronos
import nimcrypto/utils, chronicles, stew/byteutils
import types,
../../connection,
../../stream/connection,
../../utility,
../../varint,
../../vbuffer,
../../stream/lpstream
../../vbuffer
logScope:
topics = "mplexcoder"

15
libp2p/muxers/mplex/lpchannel.nim
View File

@ -8,17 +8,16 @@
## those terms.
import oids, deques
import chronos, chronicles
import chronos, chronicles, metrics
import types,
coder,
nimcrypto/utils,
../../stream/connection,
../../stream/bufferstream,
../../stream/lpstream,
../../connection,
../../utility,
../../errors
export lpstream
export connection
logScope:
topics = "mplexchannel"
@ -79,6 +78,12 @@ template withEOFExceptions(body: untyped): untyped =
method reset*(s: LPChannel) {.base, async, gcsafe.}
method initStream*(s: LPChannel) =
if s.objName.len == 0:
s.objName = "LPChannel"
procCall BufferStream(s).initStream()
proc newChannel*(id: uint64,
conn: Connection,
initiator: bool,
@ -213,7 +218,6 @@ method close*(s: LPChannel) {.async, gcsafe.} =
name = s.name,
oid = s.oid
await s.closeMessage().wait(2.minutes)
s.closedLocal = true
if s.atEof: # already closed by remote close parent buffer immediately
await procCall BufferStream(s).close()
except AsyncTimeoutError:
@ -227,4 +231,5 @@ method close*(s: LPChannel) {.async, gcsafe.} =
name = s.name,
oid = s.oid
s.closedLocal = true
asyncCheck closeRemote()

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

@ -10,8 +10,7 @@
import tables, sequtils, oids
import chronos, chronicles, stew/byteutils
import ../muxer,
../../connection,
../../stream/lpstream,
../../stream/connection,
../../stream/bufferstream,
../../utility,
../../errors,
@ -26,7 +25,6 @@ type
Mplex* = ref object of Muxer
remote: Table[uint64, LPChannel]
local: Table[uint64, LPChannel]
conns: seq[Connection]
handlerFuts: seq[Future[void]]
currentId*: uint64
maxChannels*: uint64
@ -62,6 +60,10 @@ proc newStreamInternal*(m: Mplex,
initiator,
name,
lazy = lazy)
result.peerInfo = m.connection.peerInfo
result.observedAddr = m.connection.observedAddr
m.getChannelList(initiator)[id] = result
method handle*(m: Mplex) {.async, gcsafe.} =
@ -105,22 +107,16 @@ method handle*(m: Mplex) {.async, gcsafe.} =
chann_iod = channel.oid
if not isNil(m.streamHandler):
let stream = newConnection(channel)
m.conns.add(stream)
stream.peerInfo = m.connection.peerInfo
stream.observedAddr = m.connection.observedAddr
var fut = newFuture[void]()
proc handler() {.async.} =
try:
await m.streamHandler(stream)
await m.streamHandler(channel)
trace "finished handling stream"
# doAssert(stream.closed, "connection not closed by handler!")
# doAssert(channel.closed, "connection not closed by handler!")
except CatchableError as exc:
trace "exception in stream handler", exc = exc.msg
doAssert(stream.closed, "stream not closed by protocol handler")
await channel.reset()
finally:
m.conns.keepItIf(it != stream)
m.handlerFuts.keepItIf(it != fut)
fut = handler()
@ -155,7 +151,6 @@ method handle*(m: Mplex) {.async, gcsafe.} =
await channel.reset()
m.getChannelList(initiator).del(id)
trace "deleted channel"
break
finally:
trace "stopping mplex main loop", oid = m.oid
await m.close()
@ -187,11 +182,9 @@ method newStream*(m: Mplex,
let channel = await m.newStreamInternal(lazy = lazy)
if not lazy:
await channel.open()
result = newConnection(channel)
result.peerInfo = m.connection.peerInfo
result.observedAddr = m.connection.observedAddr
asyncCheck m.cleanupChann(channel)
return Connection(channel)
method close*(m: Mplex) {.async, gcsafe.} =
if m.isClosed:
@ -208,12 +201,6 @@ method close*(m: Mplex) {.async, gcsafe.} =
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))
@ -221,6 +208,5 @@ method close*(m: Mplex) {.async, gcsafe.} =
finally:
m.remote.clear()
m.local.clear()
m.conns = @[]
m.handlerFuts = @[]
m.isClosed = true

6
libp2p/muxers/muxer.nim
View File

@ -9,7 +9,8 @@
import chronos, chronicles
import ../protocols/protocol,
../connection,
../stream/connection,
../peerinfo,
../errors
logScope:
@ -65,3 +66,6 @@ method init(c: MuxerProvider) =
trace "exception in muxer handler", exc = exc.msg
c.handler = handler
proc `$`*(m: Muxer): string =
$m.connection.peerInfo

2
libp2p/protocols/identify.nim
View File

@ -11,7 +11,7 @@ import options
import chronos, chronicles
import ../protobuf/minprotobuf,
../peerinfo,
../connection,
../stream/connection,
../peer,
../crypto/crypto,
../multiaddress,

2
libp2p/protocols/protocol.nim
View File

@ -8,7 +8,7 @@
## those terms.
import chronos
import ../connection
import ../stream/connection
type
LPProtoHandler* = proc (conn: Connection,

2
libp2p/protocols/pubsub/floodsub.nim
View File

@ -14,7 +14,7 @@ import pubsub,
timedcache,
rpc/[messages, message],
../../crypto/crypto,
../../connection,
../../stream/connection,
../../peer,
../../peerinfo,
../../utility,

2
libp2p/protocols/pubsub/gossipsub.nim
View File

@ -18,7 +18,7 @@ import pubsub,
../../crypto/crypto,
../protocol,
../../peerinfo,
../../connection,
../../stream/connection,
../../peer,
../../errors,
../../utility

2
libp2p/protocols/pubsub/pubsub.nim
View File

@ -12,7 +12,7 @@ import chronos, chronicles
import pubsubpeer,
rpc/messages,
../protocol,
../../connection,
../../stream/connection,
../../peerinfo
import metrics

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

@ -13,8 +13,7 @@ import rpc/[messages, message, protobuf],
timedcache,
../../peer,
../../peerinfo,
../../connection,
../../stream/lpstream,
../../stream/connection,
../../crypto/crypto,
../../protobuf/minprotobuf,
../../utility
@ -24,7 +23,8 @@ logScope:
declareCounter(libp2p_pubsub_sent_messages, "number of messages sent", labels = ["id", "topic"])
declareCounter(libp2p_pubsub_received_messages, "number of messages received", labels = ["id", "topic"])
declareCounter(libp2p_pubsub_skipped_messages, "number of skipped messages", labels = ["id"])
declareCounter(libp2p_pubsub_skipped_received_messages, "number of received skipped messages", labels = ["id"])
declareCounter(libp2p_pubsub_skipped_sent_messages, "number of sent skipped messages", labels = ["id"])
type
PubSubObserver* = ref object
@ -78,7 +78,7 @@ proc handle*(p: PubSubPeer, conn: Connection) {.async.} =
let digest = $(sha256.digest(data))
trace "read data from peer", peer = p.id, data = data.shortLog
if digest in p.recvdRpcCache:
libp2p_pubsub_skipped_messages.inc(labelValues = [p.id])
libp2p_pubsub_skipped_received_messages.inc(labelValues = [p.id])
trace "message already received, skipping", peer = p.id
continue
@ -117,6 +117,7 @@ proc send*(p: PubSubPeer, msgs: seq[RPCMsg]) {.async.} =
let digest = $(sha256.digest(encoded.buffer))
if digest in p.sentRpcCache:
trace "message already sent to peer, skipping", peer = p.id
libp2p_pubsub_skipped_sent_messages.inc(labelValues = [p.id])
continue
proc sendToRemote() {.async.} =

9
libp2p/protocols/secure/noise.nim
View File

@ -11,7 +11,7 @@ import chronos
import chronicles
import stew/[endians2, byteutils]
import nimcrypto/[utils, sysrand, sha2, hmac]
import ../../connection
import ../../stream/lpstream
import ../../peer
import ../../peerinfo
import ../../protobuf/minprotobuf
@ -267,12 +267,12 @@ template read_s: untyped =
proc receiveHSMessage(sconn: Connection): Future[seq[byte]] {.async.} =
var besize: array[2, byte]
await sconn.stream.readExactly(addr besize[0], besize.len)
await sconn.readExactly(addr besize[0], besize.len)
let size = uint16.fromBytesBE(besize).int
trace "receiveHSMessage", size
var buffer = newSeq[byte](size)
if buffer.len > 0:
await sconn.stream.readExactly(addr buffer[0], buffer.len)
await sconn.readExactly(addr buffer[0], buffer.len)
return buffer
proc sendHSMessage(sconn: Connection; buf: seq[byte]) {.async.} =
@ -488,8 +488,9 @@ method handshake*(p: Noise, conn: Connection, initiator: bool = false): Future[S
secure.initStream()
secure.stream = conn
secure.closeEvent = newAsyncEvent()
secure.peerInfo = PeerInfo.init(remotePubKey)
secure.observedAddr = conn.observedAddr
if initiator:
secure.readCs = handshakeRes.cs2
secure.writeCs = handshakeRes.cs1

2
libp2p/protocols/secure/plaintext.nim
View File

@ -8,7 +8,7 @@
## those terms.
import chronos
import secure, ../../connection
import secure, ../../stream/connection
const PlainTextCodec* = "/plaintext/1.0.0"

13
libp2p/protocols/secure/secio.nim
View File

@ -9,13 +9,13 @@
import chronos, chronicles, oids, stew/endians2
import nimcrypto/[sysrand, hmac, sha2, sha, hash, rijndael, twofish, bcmode]
import secure,
../../connection,
../../stream/connection,
../../peerinfo,
../../stream/lpstream,
../../crypto/crypto,
../../crypto/ecnist,
../../peer,
../../utility
export hmac, sha2, sha, hash, rijndael, bcmode
logScope:
@ -177,7 +177,7 @@ proc macCheckAndDecode(sconn: SecioConn, data: var seq[byte]): bool =
proc readRawMessage(conn: Connection): Future[seq[byte]] {.async.} =
while true: # Discard 0-length payloads
var lengthBuf: array[4, byte]
await conn.stream.readExactly(addr lengthBuf[0], lengthBuf.len)
await conn.readExactly(addr lengthBuf[0], lengthBuf.len)
let length = uint32.fromBytesBE(lengthBuf)
trace "Recieved message header", header = lengthBuf.shortLog, length = length
@ -188,7 +188,7 @@ proc readRawMessage(conn: Connection): Future[seq[byte]] {.async.} =
if length > 0:
var buf = newSeq[byte](int(length))
await conn.stream.readExactly(addr buf[0], buf.len)
await conn.readExactly(addr buf[0], buf.len)
trace "Received message body",
conn = $conn, length = buf.len, buff = buf.shortLog
return buf
@ -200,7 +200,7 @@ method readMessage*(sconn: SecioConn): Future[seq[byte]] {.async.} =
when chronicles.enabledLogLevel == LogLevel.TRACE:
logScope:
stream_oid = $sconn.stream.oid
var buf = await sconn.readRawMessage()
var buf = await sconn.stream.readRawMessage()
if sconn.macCheckAndDecode(buf):
result = buf
else:
@ -242,7 +242,7 @@ proc newSecioConn(conn: Connection,
secrets: Secret,
order: int,
remotePubKey: PublicKey): SecioConn =
## Create new secure connection, using specified hash algorithm ``hash``,
## Create new secure stream/lpstream, using specified hash algorithm ``hash``,
## cipher algorithm ``cipher``, stretched keys ``secrets`` and order
## ``order``.
new result
@ -266,6 +266,7 @@ proc newSecioConn(conn: Connection,
secrets.ivOpenArray(i1))
result.peerInfo = PeerInfo.init(remotePubKey)
result.observedAddr = conn.observedAddr
proc transactMessage(conn: Connection,
msg: seq[byte]): Future[seq[byte]] {.async.} =

15
libp2p/protocols/secure/secure.nim
View File

@ -11,15 +11,28 @@ import options
import chronos, chronicles
import ../protocol,
../../stream/streamseq,
../../connection,
../../stream/connection,
../../peerinfo
type
Secure* = ref object of LPProtocol # base type for secure managers
SecureConn* = ref object of Connection
stream*: Connection
buf: StreamSeq
method initStream*(s: SecureConn) =
if s.objName.len == 0:
s.objName = "SecureConn"
procCall Connection(s).initStream()
method close*(s: SecureConn) {.async.} =
if not(isNil(s.stream)):
await s.stream.close()
await procCall Connection(s).close()
method readMessage*(c: SecureConn): Future[seq[byte]] {.async, base.} =
doAssert(false, "Not implemented!")

2
libp2p/standard_setup.nim
View File

@ -5,7 +5,7 @@ const
import
options, tables, chronos,
switch, peer, peerinfo, connection, multiaddress,
switch, peer, peerinfo, stream/connection, multiaddress,
crypto/crypto, transports/[transport, tcptransport],
muxers/[muxer, mplex/mplex, mplex/types],
protocols/[identify, secure/secure],

20
libp2p/stream/bufferstream.nim
View File

@ -32,18 +32,16 @@
import deques, math
import chronos, chronicles, metrics
import ../stream/lpstream
import ../stream/connection
when chronicles.enabledLogLevel == LogLevel.TRACE:
import oids
export lpstream
export connection
logScope:
topics = "bufferstream"
declareGauge libp2p_open_bufferstream, "open BufferStream instances"
const
DefaultBufferSize* = 1024
@ -83,7 +81,7 @@ type
# TODO: figure out how to make this generic to avoid casts
WriteHandler* = proc (data: seq[byte]): Future[void] {.gcsafe.}
BufferStream* = ref object of LPStream
BufferStream* = ref object of Connection
maxSize*: int # buffer's max size in bytes
readBuf: Deque[byte] # this is a ring buffer based dequeue
readReqs*: Deque[Future[void]] # use dequeue to fire reads in order
@ -109,11 +107,12 @@ proc requestReadBytes(s: BufferStream): Future[void] =
s.readReqs.addLast(result)
# trace "requestReadBytes(): added a future to readReqs", oid = s.oid
method initStream(s: BufferStream) =
procCall LPStream(s).initStream()
method initStream*(s: BufferStream) =
if s.objName.len == 0:
s.objName = "BufferStream"
procCall Connection(s).initStream()
inc getBufferStreamTracker().opened
libp2p_open_bufferstream.inc()
proc initBufferStream*(s: BufferStream,
handler: WriteHandler = nil,
@ -316,12 +315,9 @@ method close*(s: BufferStream) {.async, gcsafe.} =
r.fail(newLPStreamEOFError())
s.dataReadEvent.fire()
s.readBuf.clear()
s.closeEvent.fire()
s.isClosed = true
await procCall Connection(s).close()
inc getBufferStreamTracker().closed
libp2p_open_bufferstream.dec()
trace "bufferstream closed", oid = s.oid
else:
trace "attempt to close an already closed bufferstream", trace = getStackTrace()

14
libp2p/stream/chronosstream.nim
View File

@ -8,18 +8,24 @@
## those terms.
import chronos, chronicles
import lpstream, ../utility
import connection, ../utility
logScope:
topics = "chronosstream"
type ChronosStream* = ref object of LPStream
type ChronosStream* = ref object of Connection
client: StreamTransport
method initStream*(s: ChronosStream) =
if s.objName.len == 0:
s.objName = "ChronosStream"
procCall Connection(s).initStream()
proc newChronosStream*(client: StreamTransport): ChronosStream =
new result
result.client = client
result.closeEvent = newAsyncEvent()
result.initStream()
template withExceptions(body: untyped) =
try:
@ -82,6 +88,6 @@ method close*(s: ChronosStream) {.async.} =
if not s.client.closed():
await s.client.closeWait()
s.closeEvent.fire()
await procCall Connection(s).close()
except CatchableError as exc:
trace "error closing chronosstream", exc = exc.msg

74
libp2p/stream/connection.nim Normal file
View File

@ -0,0 +1,74 @@
## Nim-LibP2P
## Copyright (c) 2020 Status Research & Development GmbH
## Licensed under either of
## * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
## * MIT license ([LICENSE-MIT](LICENSE-MIT))
## at your option.
## This file may not be copied, modified, or distributed except according to
## those terms.
import chronos, metrics
import lpstream,
../multiaddress,
../peerinfo
export lpstream
const
ConnectionTrackerName* = "libp2p.connection"
type
Connection* = ref object of LPStream
peerInfo*: PeerInfo
observedAddr*: Multiaddress
closeEvent*: AsyncEvent
ConnectionTracker* = ref object of TrackerBase
opened*: uint64
closed*: uint64
proc setupConnectionTracker(): ConnectionTracker {.gcsafe.}
proc getConnectionTracker*(): ConnectionTracker {.gcsafe.} =
result = cast[ConnectionTracker](getTracker(ConnectionTrackerName))
if isNil(result):
result = setupConnectionTracker()
proc dumpTracking(): string {.gcsafe.} =
var tracker = getConnectionTracker()
result = "Opened conns: " & $tracker.opened & "\n" &
"Closed conns: " & $tracker.closed
proc leakTransport(): bool {.gcsafe.} =
var tracker = getConnectionTracker()
result = (tracker.opened != tracker.closed)
proc setupConnectionTracker(): ConnectionTracker =
result = new ConnectionTracker
result.opened = 0
result.closed = 0
result.dump = dumpTracking
result.isLeaked = leakTransport
addTracker(ConnectionTrackerName, result)
proc init*[T: Connection](self: var T, peerInfo: PeerInfo): T =
new self
self.initStream()
method initStream*(s: Connection) =
if s.objName.len == 0:
s.objName = "Connection"
procCall LPStream(s).initStream()
s.closeEvent = newAsyncEvent()
inc getConnectionTracker().opened
method close*(s: Connection) {.async.} =
await procCall LPStream(s).close()
s.closeEvent.fire()
inc getConnectionTracker().closed
proc `$`*(conn: Connection): string =
if not isNil(conn.peerInfo):
result = conn.peerInfo.id

32
libp2p/stream/lpstream.nim
View File

@ -7,20 +7,21 @@
## This file may not be copied, modified, or distributed except according to
## those terms.
import chronicles, chronos
import oids
import chronicles, chronos, metrics
import ../varint,
../vbuffer
../vbuffer,
../peerinfo,
../multiaddress
when chronicles.enabledLogLevel == LogLevel.TRACE:
import oids
declareGauge(libp2p_open_streams, "open stream instances", labels = ["type"])
type
LPStream* = ref object of RootObj
isClosed*: bool
isEof*: bool
closeEvent*: AsyncEvent
when chronicles.enabledLogLevel == LogLevel.TRACE:
oid*: Oid
objName*: string
oid*: Oid
LPStreamError* = object of CatchableError
LPStreamIncompleteError* = object of LPStreamError
@ -67,9 +68,12 @@ proc newLPStreamClosedError*(): ref Exception =
result = newException(LPStreamClosedError, "Stream Closed!")
method initStream*(s: LPStream) {.base.} =
s.closeEvent = newAsyncEvent()
when chronicles.enabledLogLevel == LogLevel.TRACE:
s.oid = genOid()
if s.objName.len == 0:
s.objName = "LPStream"
s.oid = genOid()
libp2p_open_streams.inc(labelValues = [s.objName])
trace "stream created", oid = s.oid
method closed*(s: LPStream): bool {.base, inline.} =
s.isClosed
@ -163,6 +167,8 @@ proc write*(s: LPStream, pbytes: pointer, nbytes: int): Future[void] {.deprecate
proc write*(s: LPStream, msg: string): Future[void] =
s.write(@(toOpenArrayByte(msg, 0, msg.high)))
method close*(s: LPStream)
{.base, async.} =
doAssert(false, "not implemented!")
method close*(s: LPStream) {.base, async.} =
if not s.isClosed:
libp2p_open_streams.dec(labelValues = [s.objName])
s.isClosed = true
trace "stream destroyed", oid = s.oid

7
libp2p/switch.nim
View File

@ -9,7 +9,8 @@
import tables, sequtils, options, strformat, sets
import chronos, chronicles, metrics
import connection,
import stream/connection,
stream/chronosstream,
transports/transport,
multistream,
multiaddress,
@ -97,7 +98,7 @@ proc identify(s: Switch, conn: Connection): Future[PeerInfo] {.async, gcsafe.} =
if info.protos.len > 0:
result.protocols = info.protos
debug "identify", info = shortLog(result)
trace "identify", info = shortLog(result)
except IdentityInvalidMsgError as exc:
error "identify: invalid message", msg = exc.msg
except IdentityNoMatchError as exc:
@ -155,6 +156,7 @@ proc cleanupConn(s: Switch, conn: Connection) {.async, gcsafe.} =
if id in s.connections:
s.connections.del(id)
await conn.close()
s.dialedPubSubPeers.excl(id)
@ -169,6 +171,7 @@ proc cleanupConn(s: Switch, conn: Connection) {.async, gcsafe.} =
proc disconnect*(s: Switch, peer: PeerInfo) {.async, gcsafe.} =
let conn = s.connections.getOrDefault(peer.id)
if not isNil(conn):
trace "disconnecting peer", peer = $peer
await s.cleanupConn(conn)
proc getMuxedStream(s: Switch, peerInfo: PeerInfo): Future[Connection] {.async, gcsafe.} =

5
libp2p/transports/tcptransport.nim
View File

@ -11,10 +11,9 @@ import chronos, chronicles, sequtils, oids
import transport,
../errors,
../wire,
../connection,
../multiaddress,
../multicodec,
../stream/lpstream,
../stream/connection,
../stream/chronosstream
logScope:
@ -63,7 +62,7 @@ proc connHandler*(t: TcpTransport,
client: StreamTransport,
initiator: bool): Connection =
trace "handling connection", address = $client.remoteAddress
let conn: Connection = newConnection(newChronosStream(client))
let conn: Connection = Connection(newChronosStream(client))
conn.observedAddr = MultiAddress.init(client.remoteAddress).tryGet()
if not initiator:
if not isNil(t.handler):

2
libp2p/transports/transport.nim
View File

@ -9,7 +9,7 @@
import sequtils, tables
import chronos, chronicles
import ../connection,
import ../stream/connection,
../multiaddress,
../multicodec,
../errors

4
tests/helpers.nim
View File

@ -2,14 +2,14 @@ import chronos
import ../libp2p/transports/tcptransport
import ../libp2p/stream/bufferstream
import ../libp2p/connection
import ../libp2p/stream/lpstream
const
StreamTransportTrackerName = "stream.transport"
StreamServerTrackerName = "stream.server"
trackerNames = [
ConnectionTrackerName,
# ConnectionTrackerName,
BufferStreamTrackerName,
TcpTransportTrackerName,
StreamTransportTrackerName,

2
tests/pubsub/testfloodsub.nim
View File

@ -14,7 +14,7 @@ import chronos, stew/byteutils
import utils,
../../libp2p/[errors,
switch,
connection,
stream/connection,
stream/bufferstream,
crypto/crypto,
protocols/pubsub/pubsub,

28
tests/pubsub/testgossipinternal.nim
View File

@ -30,7 +30,7 @@ suite "GossipSub internal":
var conns = newSeq[Connection]()
for i in 0..<15:
let conn = newConnection(newBufferStream(noop))
let conn = newBufferStream(noop)
conns &= conn
let peerInfo = PeerInfo.init(PrivateKey.random(ECDSA).get())
conn.peerInfo = peerInfo
@ -59,7 +59,7 @@ suite "GossipSub internal":
var conns = newSeq[Connection]()
for i in 0..<15:
let conn = newConnection(newBufferStream(noop))
let conn = newBufferStream(noop)
conns &= conn
let peerInfo = PeerInfo.init(PrivateKey.random(ECDSA).get())
conn.peerInfo = peerInfo
@ -91,7 +91,7 @@ suite "GossipSub internal":
var conns = newSeq[Connection]()
for i in 0..<15:
let conn = newConnection(newBufferStream(noop))
let conn = newBufferStream(noop)
conns &= conn
var peerInfo = PeerInfo.init(PrivateKey.random(ECDSA).get())
conn.peerInfo = peerInfo
@ -125,7 +125,7 @@ suite "GossipSub internal":
var conns = newSeq[Connection]()
for i in 0..<6:
let conn = newConnection(newBufferStream(noop))
let conn = newBufferStream(noop)
conns &= conn
let peerInfo = PeerInfo.init(PrivateKey.random(ECDSA).get())
conn.peerInfo = peerInfo
@ -163,7 +163,7 @@ suite "GossipSub internal":
var conns = newSeq[Connection]()
for i in 0..<6:
let conn = newConnection(newBufferStream(noop))
let conn = newBufferStream(noop)
conns &= conn
let peerInfo = PeerInfo.init(PrivateKey.random(ECDSA).get())
conn.peerInfo = peerInfo
@ -202,7 +202,7 @@ suite "GossipSub internal":
# generate mesh and fanout peers
for i in 0..<30:
let conn = newConnection(newBufferStream(noop))
let conn = newBufferStream(noop)
conns &= conn
let peerInfo = PeerInfo.init(PrivateKey.random(ECDSA).get())
conn.peerInfo = peerInfo
@ -215,7 +215,7 @@ suite "GossipSub internal":
# generate gossipsub (free standing) peers
for i in 0..<15:
let conn = newConnection(newBufferStream(noop))
let conn = newBufferStream(noop)
conns &= conn
let peerInfo = PeerInfo.init(PrivateKey.random(ECDSA).get())
conn.peerInfo = peerInfo
@ -225,7 +225,7 @@ suite "GossipSub internal":
# generate messages
for i in 0..5:
let conn = newConnection(newBufferStream(noop))
let conn = newBufferStream(noop)
conns &= conn
let peerInfo = PeerInfo.init(PrivateKey.random(ECDSA).get())
conn.peerInfo = peerInfo
@ -262,7 +262,7 @@ suite "GossipSub internal":
gossipSub.gossipsub[topic] = initHashSet[string]()
var conns = newSeq[Connection]()
for i in 0..<30:
let conn = newConnection(newBufferStream(noop))
let conn = newBufferStream(noop)
conns &= conn
let peerInfo = PeerInfo.init(PrivateKey.random(ECDSA).get())
conn.peerInfo = peerInfo
@ -275,7 +275,7 @@ suite "GossipSub internal":
# generate messages
for i in 0..5:
let conn = newConnection(newBufferStream(noop))
let conn = newBufferStream(noop)
conns &= conn
let peerInfo = PeerInfo.init(PrivateKey.random(ECDSA).get())
conn.peerInfo = peerInfo
@ -305,7 +305,7 @@ suite "GossipSub internal":
gossipSub.gossipsub[topic] = initHashSet[string]()
var conns = newSeq[Connection]()
for i in 0..<30:
let conn = newConnection(newBufferStream(noop))
let conn = newBufferStream(noop)
conns &= conn
let peerInfo = PeerInfo.init(PrivateKey.random(ECDSA).get())
conn.peerInfo = peerInfo
@ -318,7 +318,7 @@ suite "GossipSub internal":
# generate messages
for i in 0..5:
let conn = newConnection(newBufferStream(noop))
let conn = newBufferStream(noop)
conns &= conn
let peerInfo = PeerInfo.init(PrivateKey.random(ECDSA).get())
conn.peerInfo = peerInfo
@ -348,7 +348,7 @@ suite "GossipSub internal":
gossipSub.fanout[topic] = initHashSet[string]()
var conns = newSeq[Connection]()
for i in 0..<30:
let conn = newConnection(newBufferStream(noop))
let conn = newBufferStream(noop)
conns &= conn
let peerInfo = PeerInfo.init(PrivateKey.random(ECDSA).get())
conn.peerInfo = peerInfo
@ -361,7 +361,7 @@ suite "GossipSub internal":
# generate messages
for i in 0..5:
let conn = newConnection(newBufferStream(noop))
let conn = newBufferStream(noop)
conns &= conn
let peerInfo = PeerInfo.init(PrivateKey.random(ECDSA).get())
conn.peerInfo = peerInfo

2
tests/pubsub/testgossipsub.nim
View File

@ -15,7 +15,7 @@ import chronicles
import utils, ../../libp2p/[errors,
peer,
peerinfo,
connection,
stream/connection,
crypto/crypto,
stream/bufferstream,
protocols/pubsub/pubsub,

4
tests/testbufferstream.nim
View File

@ -1,7 +1,7 @@
import unittest, strformat
import chronos, stew/byteutils
import ../libp2p/errors
import ../libp2p/stream/bufferstream
import ../libp2p/stream/bufferstream,
../libp2p/errors
when defined(nimHasUsed): {.used.}

9
tests/testconnection.nim
View File

@ -1,13 +1,12 @@
import unittest
import chronos, nimcrypto/utils
import ../libp2p/[connection,
stream/lpstream,
import ../libp2p/[stream/connection,
stream/bufferstream]
suite "Connection":
test "close":
proc test(): Future[bool] {.async.} =
var conn = newConnection(newBufferStream())
var conn = newBufferStream()
await conn.close()
check:
conn.closed == true
@ -20,7 +19,7 @@ suite "Connection":
test "parent close":
proc test(): Future[bool] {.async.} =
var buf = newBufferStream()
var conn = newConnection(buf)
var conn = buf
await conn.close()
check:
@ -36,7 +35,7 @@ suite "Connection":
test "child close":
proc test(): Future[bool] {.async.} =
var buf = newBufferStream()
var conn = newConnection(buf)
var conn = buf
await buf.close()
check:

2
tests/testidentify.nim
View File

@ -4,7 +4,7 @@ import ../libp2p/[protocols/identify,
multiaddress,
peerinfo,
peer,
connection,
stream/connection,
multistream,
transports/transport,
transports/tcptransport,

3
tests/testinterop.nim
View File

@ -14,8 +14,7 @@ import ../libp2p/[daemon/daemonapi,
peer,
peerinfo,
switch,
connection,
stream/lpstream,
stream/connection,
muxers/muxer,
crypto/crypto,
muxers/mplex/mplex,

35
tests/testmplex.nim
View File

@ -1,8 +1,7 @@
import unittest, strformat, strformat, random
import chronos, nimcrypto/utils, chronicles, stew/byteutils
import ../libp2p/[errors,
connection,
stream/lpstream,
stream/connection,
stream/bufferstream,
transports/tcptransport,
transports/transport,
@ -30,7 +29,7 @@ suite "Mplex":
check msg == fromHex("000873747265616d2031")
let stream = newBufferStream(encHandler)
let conn = newConnection(stream)
let conn = stream
await conn.writeMsg(0, MessageType.New, ("stream 1").toBytes)
await conn.close()
@ -42,7 +41,7 @@ suite "Mplex":
check msg == fromHex("88010873747265616d2031")
let stream = newBufferStream(encHandler)
let conn = newConnection(stream)
let conn = stream
await conn.writeMsg(17, MessageType.New, ("stream 1").toBytes)
await conn.close()
@ -54,7 +53,7 @@ suite "Mplex":
check msg == fromHex("020873747265616d2031")
let stream = newBufferStream(encHandler)
let conn = newConnection(stream)
let conn = stream
await conn.writeMsg(0, MessageType.MsgOut, ("stream 1").toBytes)
await conn.close()
@ -66,7 +65,7 @@ suite "Mplex":
check msg == fromHex("8a010873747265616d2031")
let stream = newBufferStream(encHandler)
let conn = newConnection(stream)
let conn = stream
await conn.writeMsg(17, MessageType.MsgOut, ("stream 1").toBytes)
await conn.close()
@ -75,7 +74,7 @@ suite "Mplex":
test "decode header with channel id 0":
proc testDecodeHeader() {.async.} =
let stream = newBufferStream()
let conn = newConnection(stream)
let conn = stream
await stream.pushTo(fromHex("000873747265616d2031"))
let msg = await conn.readMsg()
@ -88,7 +87,7 @@ suite "Mplex":
test "decode header and body with channel id 0":
proc testDecodeHeader() {.async.} =
let stream = newBufferStream()
let conn = newConnection(stream)
let conn = stream
await stream.pushTo(fromHex("021668656C6C6F2066726F6D206368616E6E656C20302121"))
let msg = await conn.readMsg()
@ -102,7 +101,7 @@ suite "Mplex":
test "decode header and body with channel id other than 0":
proc testDecodeHeader() {.async.} =
let stream = newBufferStream()
let conn = newConnection(stream)
let conn = stream
await stream.pushTo(fromHex("8a011668656C6C6F2066726F6D206368616E6E656C20302121"))
let msg = await conn.readMsg()
@ -117,7 +116,7 @@ suite "Mplex":
proc testClosedForWrite(): Future[bool] {.async.} =
proc writeHandler(data: seq[byte]) {.async, gcsafe.} = discard
let
conn = newConnection(newBufferStream(writeHandler))
conn = newBufferStream(writeHandler)
chann = newChannel(1, conn, true)
await chann.close()
try:
@ -134,10 +133,10 @@ suite "Mplex":
test "half closed - channel should close for read by remote":
proc testClosedForRead(): Future[bool] {.async.} =
let
conn = newConnection(newBufferStream(
conn = newBufferStream(
proc (data: seq[byte]) {.gcsafe, async.} =
result = nil
))
)
chann = newChannel(1, conn, true)
await chann.pushTo(("Hello!").toBytes)
@ -161,7 +160,7 @@ suite "Mplex":
proc testResetWrite(): Future[bool] {.async.} =
proc writeHandler(data: seq[byte]) {.async, gcsafe.} = discard
let
conn = newConnection(newBufferStream(writeHandler))
conn = newBufferStream(writeHandler)
chann = newChannel(1, conn, true)
await chann.closeRemote()
try:
@ -179,7 +178,7 @@ suite "Mplex":
proc testResetRead(): Future[bool] {.async.} =
proc writeHandler(data: seq[byte]) {.async, gcsafe.} = discard
let
conn = newConnection(newBufferStream(writeHandler))
conn = newBufferStream(writeHandler)
chann = newChannel(1, conn, true)
await chann.reset()
@ -199,7 +198,7 @@ suite "Mplex":
proc testResetWrite(): Future[bool] {.async.} =
proc writeHandler(data: seq[byte]) {.async, gcsafe.} = discard
let
conn = newConnection(newBufferStream(writeHandler))
conn = newBufferStream(writeHandler)
chann = newChannel(1, conn, true)
await chann.reset()
try:
@ -239,7 +238,7 @@ suite "Mplex":
let mplexDialFut = mplexDial.handle()
let stream = await mplexDial.newStream()
await stream.writeLp("HELLO")
check LPChannel(stream.stream).isOpen # not lazy
check LPChannel(stream).isOpen # not lazy
await stream.close()
await done.wait(1.seconds)
@ -278,9 +277,9 @@ suite "Mplex":
let mplexDial = newMplex(conn)
let stream = await mplexDial.newStream(lazy = true)
let mplexDialFut = mplexDial.handle()
check not LPChannel(stream.stream).isOpen # assert lazy
check not LPChannel(stream).isOpen # assert lazy
await stream.writeLp("HELLO")
check LPChannel(stream.stream).isOpen # assert lazy
check LPChannel(stream).isOpen # assert lazy
await stream.close()
await done.wait(1.seconds)

14
tests/testmultistream.nim
View File

@ -1,11 +1,9 @@
import unittest, strutils, sequtils, strformat, stew/byteutils
import chronos
import ../libp2p/errors,
../libp2p/connection,
../libp2p/multistream,
../libp2p/stream/lpstream,
../libp2p/stream/bufferstream,
../libp2p/connection,
../libp2p/stream/connection,
../libp2p/multiaddress,
../libp2p/transports/transport,
../libp2p/transports/tcptransport,
@ -17,7 +15,7 @@ when defined(nimHasUsed): {.used.}
## Mock stream for select test
type
TestSelectStream = ref object of LPStream
TestSelectStream = ref object of Connection
step*: int
method readExactly*(s: TestSelectStream,
@ -157,7 +155,7 @@ suite "Multistream select":
test "test select custom proto":
proc testSelect(): Future[bool] {.async.} =
let ms = newMultistream()
let conn = newConnection(newTestSelectStream())
let conn = newTestSelectStream()
result = (await ms.select(conn, @["/test/proto/1.0.0"])) == "/test/proto/1.0.0"
await conn.close()
@ -167,7 +165,7 @@ suite "Multistream select":
test "test handle custom proto":
proc testHandle(): Future[bool] {.async.} =
let ms = newMultistream()
let conn = newConnection(newTestSelectStream())
let conn = newTestSelectStream()
var protocol: LPProtocol = new LPProtocol
proc testHandler(conn: Connection,
@ -189,7 +187,7 @@ suite "Multistream select":
let ms = newMultistream()
proc testLsHandler(proto: seq[byte]) {.async, gcsafe.} # forward declaration
let conn = newConnection(newTestLsStream(testLsHandler))
let conn = Connection(newTestLsStream(testLsHandler))
let done = newFuture[void]()
proc testLsHandler(proto: seq[byte]) {.async, gcsafe.} =
var strProto: string = string.fromBytes(proto)
@ -216,7 +214,7 @@ suite "Multistream select":
let ms = newMultistream()
proc testNaHandler(msg: string): Future[void] {.async, gcsafe.}
let conn = newConnection(newTestNaStream(testNaHandler))
let conn = newTestNaStream(testNaHandler)
proc testNaHandler(msg: string): Future[void] {.async, gcsafe.} =
check msg == Na

2
tests/testnative.nim
View File

@ -14,7 +14,7 @@ import testmultibase,
testpeer
import testtransport,
testmultistream,
# testmultistream,
testbufferstream,
testidentify,
testswitch,

2
tests/testnoise.nim
View File

@ -19,7 +19,7 @@ import ../libp2p/[switch,
multistream,
stream/bufferstream,
protocols/identify,
connection,
stream/connection,
transports/transport,
transports/tcptransport,
multiaddress,

13
tests/testswitch.nim
View File

@ -10,7 +10,7 @@ import ../libp2p/[errors,
standard_setup,
stream/bufferstream,
protocols/identify,
connection,
stream/connection,
transports/transport,
transports/tcptransport,
multiaddress,
@ -115,14 +115,14 @@ suite "Switch":
check (BufferStreamTracker(bufferTracker).opened ==
(BufferStreamTracker(bufferTracker).closed + 4.uint64))
var connTracker = getTracker(ConnectionTrackerName)
# var connTracker = getTracker(ConnectionTrackerName)
# echo connTracker.dump()
# plus 8 is for the secured connection and the socket
# and the pubsub streams that won't clean up until
# `disconnect()` or `stop()`
check (ConnectionTracker(connTracker).opened ==
(ConnectionTracker(connTracker).closed + 8.uint64))
# check (ConnectionTracker(connTracker).opened ==
# (ConnectionTracker(connTracker).closed + 8.uint64))
await allFuturesThrowing(
done.wait(5.seconds),
@ -189,6 +189,7 @@ suite "Switch":
let switch2 = newStandardSwitch()
awaiters.add(await switch1.start())
awaiters.add(await switch2.start())
await switch2.connect(switch1.peerInfo)
check switch1.connections.len > 0
@ -202,9 +203,9 @@ suite "Switch":
# echo bufferTracker.dump()
check bufferTracker.isLeaked() == false
var connTracker = getTracker(ConnectionTrackerName)
# var connTracker = getTracker(ConnectionTrackerName)
# echo connTracker.dump()
check connTracker.isLeaked() == false
# check connTracker.isLeaked() == false
check switch1.connections.len == 0
check switch2.connections.len == 0

2
tests/testtransport.nim
View File

@ -2,7 +2,7 @@
import unittest
import chronos, stew/byteutils
import ../libp2p/[connection,
import ../libp2p/[stream/connection,
transports/transport,
transports/tcptransport,
multiaddress,