libp2p / go-mplex

A port of maxogden's multiplex to go

DEPRECATED: go-mplex

mplex is deprecated. When on TCP, we recommend using Yamux instead. If TCP is not a requirement, consider using QUIC instead.

Ref: Discussion around mplex deprecation: libp2p/specs#553

---

A super simple stream muxing library implementing mplex.

Users should prefer yamux over mplex. We would like to deprecate mplex in the future.

Usage

mplex := multiplex.NewMultiplex(mysocket)

s, _ := mplex.NewStream()
s.Write([]byte("Hello World!"))
s.Close()

os, _ := mplex.Accept()
// echo back everything received
io.Copy(os, os)

---

The last gx published version of this module was: 0.2.35: QmWGQQ6Tz8AdUpxktLf3zgnVN9Vy8fcWVezZJSU3ZmiANj

go-mplex's Issues

Correctly implement write deadlines

Write deadlines should:

1. Apply to in-progress writes.
2. Shouldn't corrupt the state of the entire connection.

mpool.ByteSlicePool leak

		b, err := mp.readNext()
		if err != nil {
			mp.shutdownErr = err
			return
		}
		
		mp.chLock.Lock()
		msch, ok := mp.channels[ch]
		mp.chLock.Unlock()

		switch tag {
		case newStreamTag:
			if ok {
				log.Debugf("received NewStream message for existing stream: %d", ch)
				mp.shutdownErr = ErrInvalidState
				return
			}

			name := string(b)
			msch = mp.newStream(ch, name)
			mp.chLock.Lock()
			mp.channels[ch] = msch
			mp.chLock.Unlock()
			select {
			case mp.nstreams <- msch:
			case <-mp.shutdown:
				return
			}

b, err := mp.readNext() get a byte buffer, but in case newStreamTag: branch, the b doesn't put to mpool.ByteSlicePool.

Re-enable write coalescing

The write coalescing has been temporarily disabled in #61; see libp2p/go-libp2p#644.

Re-enable once the current ipfs release (v0.4.21) ships.

Benchmarks on write performance

To realize #75 , benchmarks are needed to get visibility into write performance and latency.

TestFuzzCloseStream fails with leaked closed streams

=== RUN   TestFuzzCloseStream
--- FAIL: TestFuzzCloseStream (0.05s)
    multiplex_test.go:472: leaked closed streams

Re-enable write coalescing

write coalescing was rolled back in #74. In the long run, it seems advantageous to have, and we should work towards better confidence in being able to enable it

Distinguish between locally reset and remotely closed streams

We currently log a warning and send a reset (again) when handling messages if we locally reset a stream. We should either:

1. Not do this.
2. Remember if the remote side has closed the stream so we correctly log the error.

potential race condition in stream reading loop

See PR description of #114.

Partial Writes

So, I'm a bit worried we can write a partial message if a deadline is reached while writing. This would put the connection into an unusable state.

Stream opening should be lazy

Currently, we end up sending a bunch of tiny open packets.

stalled writer goroutines

Grepping through goroutine traces of our relay infrastructure, I found stalled goroutines:

goroutine 2826624695 [select, 3 minutes]:
github.com/libp2p/go-mplex.(*Multiplex).sendMsg(0xc1f430fc00, 0xf77f00, 0xc0000cc010, 0x18350, 0xc3aefa9990, 0x5, 0x20, 0x0, 0x0)
#011/home/ubuntu/go/pkg/mod/github.com/vyzo/[email protected]/multiplex.go:143 +0xf5
github.com/libp2p/go-mplex.(*Multiplex).NewNamedStream(0xc1f430fc00, 0xc044180008, 0x5, 0xc3aefa9a20, 0xe8ec18, 0xc1720a67a0)
#011/home/ubuntu/go/pkg/mod/github.com/vyzo/[email protected]/multiplex.go:217 +0x1a0
github.com/libp2p/go-mplex.(*Multiplex).NewStream(...)
[...]

When we write control messages, we transfer the deadline from the context to the connection, but the sendMsg method is always called with the background context, which never carries a deadline. So if one write stalls, it'll stall forever and it'll also block any subsequent sending of control messages on that multiplexed connection.

See: https://github.com/libp2p/go-mplex/blob/master/multiplex.go#L152

Closing a stream should cancel pending writes.

PANIC in readNext

A relay has crashed with an index out of range.

Apr 18 20:42:52 ip-172-31-43-83 relay.sh[4778]: panic: runtime error: index out of range
Apr 18 20:42:52 ip-172-31-43-83 relay.sh[4778]: goroutine 25260147699 [running]:
Apr 18 20:42:52 ip-172-31-43-83 relay.sh[4778]: bufio.(*Reader).Read(0xc0f1b2c780, 0xc04347c2eb, 0x3142, 0x3d15, 0x20, 0x0, 0x0)
Apr 18 20:42:52 ip-172-31-43-83 relay.sh[4778]:         /usr/local/go/src/bufio/bufio.go:214 +0x3de
Apr 18 20:42:52 ip-172-31-43-83 relay.sh[4778]: io.ReadAtLeast(0xf26220, 0xc0f1b2c780, 0xc04347c000, 0x342d, 0x4000, 0x342d, 0xc0fc028d78, 0x85961f, 0xf26200)
Apr 18 20:42:52 ip-172-31-43-83 relay.sh[4778]:         /usr/local/go/src/io/io.go:310 +0x88
Apr 18 20:42:52 ip-172-31-43-83 relay.sh[4778]: io.ReadFull(...)
Apr 18 20:42:52 ip-172-31-43-83 relay.sh[4778]:         /usr/local/go/src/io/io.go:329
Apr 18 20:42:52 ip-172-31-43-83 relay.sh[4778]: github.com/libp2p/go-mplex.(*Multiplex).readNext(0xc01eb60230, 0x216, 0x7, 0x0, 0x0, 0xec0295e)
Apr 18 20:42:52 ip-172-31-43-83 relay.sh[4778]:         /home/ubuntu/go/pkg/mod/github.com/libp2p/[email protected]/multiplex.go:448 +0xc0
Apr 18 20:42:52 ip-172-31-43-83 relay.sh[4778]: github.com/libp2p/go-mplex.(*Multiplex).handleIncoming(0xc01eb60230)
Apr 18 20:42:52 ip-172-31-43-83 relay.sh[4778]:         /home/ubuntu/go/pkg/mod/github.com/libp2p/[email protected]/multiplex.go:281 +0x155
Apr 18 20:42:52 ip-172-31-43-83 relay.sh[4778]: created by github.com/libp2p/go-mplex.NewMultiplex
Apr 18 20:42:52 ip-172-31-43-83 relay.sh[4778]:         /home/ubuntu/go/pkg/mod/github.com/libp2p/[email protected]/multiplex.go:78 +0x1de

assumes users always call Serve

users who only send should not need to call Serve.

spec updates

@Stebalien We recently made some changes here, I don't think they are represented in the spec.md. Can we make sure to get that updated?

flaky test: TestSmallPackets

I've noticed TestSmallPackets to be flaky.

It failed on ubuntu with go 1.17 here: https://github.com/libp2p/go-mplex/runs/4485473110?check_suite_focus=true but passed on other OSes and on a rerun with the same setup.

Here's the output from the failed run:

=== RUN   TestSmallPackets
CPU Bound Limit:   158347	     13074 ns/op	3682451.26 MB/s
Latency Bound Limit: 512ms
Bandwidth Bound Limit: 102400Kbps
Network Bound Limit:   122653	     35489 ns/op	1050417.25 MB/s
At MaxProc 1 102400Kbps / 256ms latency:   173366	     23267 ns/op	2265277.76 MB/s
At MaxProc 2 102400Kbps / 256ms latency:   118280	     23510 ns/op	1529489.42 MB/s
Slowdown is 31.775522%
    benchmarks_test.go:58: Slowdown from mplex was >15%: 0.317755
--- FAIL: TestSmallPackets (57.05s)

Max message size

Trying to test out the interop between js-libp2p and go-libp2p. Get stuck when I send things bigger than 1024 bytes, and they become cropped at that size.

I'm using the echo examples from both go-libp2p and js-libp2p-ipfs-nodejs to test the interop.

Progress is in this PR: ipfs-inactive/js-libp2p-ipfs-nodejs#55

Screenshot from it in action:

Add people to this repo

I wanted to assign you a CR request for #5

ERROR: leaked a completely closed stream

I am seeing this error quite a bit in the mplex relay:

ERROR      mplex: leaked a completely closed stream multiplex.go:324

Judging by the comments around the source of the log, this is close to an assertion violation and indicative of a bug.

Move interop tests to libp2p/interop

Would that make sense, @vasco-santos?

libp2p org?

Any hopes to get this renamed to go-libp2p-mplex and move it to the libp2p org? Pretty please? 🙏🏽

Concurrent reads are racy

Do we support them? I expect users will make concurrent calls to read regardless of what we officially support...

possible concurrency bug causing lockup

Maybe related to #115

Had an issue developing starpc where sometimes (~10% of the time) the test would get stuck on recv() on both ends even after one side had sent the packet it was waiting for.

I was not able to figure out where the actual issue is located, but after I switched out mplex for yamux, the issue is gone.

So, I think the lockup issue must be in mplex somewhere.

Update Travis Go version

#37 (comment)

Race condition in Stream deadlines

There's a race condition using Stream.wDeadline.

WARNING: DATA RACE
Read at 0x00c00145e060 by goroutine 1059:
  github.com/libp2p/go-mplex.(*Stream).write()
      /Users/anacrolix/gopath/src/github.com/libp2p/go-mplex/stream.go:145 +0xd6
  github.com/libp2p/go-mplex.(*Stream).Write()
      /Users/anacrolix/gopath/src/github.com/libp2p/go-mplex/stream.go:129 +0xce
  github.com/libp2p/go-libp2p-swarm.(*Stream).Write()
      /Users/anacrolix/gopath/src/github.com/libp2p/go-libp2p-swarm/swarm_stream.go:88 +0x92
  bufio.(*Writer).Flush()
      /Users/anacrolix/src/go1.11/src/bufio/bufio.go:575 +0x136
  github.com/multiformats/go-multistream.delimWriteBuffered()
      /Users/anacrolix/gopath/src/github.com/multiformats/go-multistream/multistream.go:64 +0x20b
  github.com/multiformats/go-multistream.(*MultistreamMuxer).NegotiateLazy.func1()
      /Users/anacrolix/gopath/src/github.com/multiformats/go-multistream/multistream.go:206 +0x24a
  sync.(*Once).Do()
      /Users/anacrolix/src/go1.11/src/sync/once.go:44 +0xde

Previous write at 0x00c00145e060 by goroutine 1168:
  github.com/libp2p/go-mplex.(*Stream).SetDeadline()
      /Users/anacrolix/gopath/src/github.com/libp2p/go-mplex/stream.go:208 +0x90
  github.com/libp2p/go-libp2p-swarm.(*Stream).SetDeadline()
      /Users/anacrolix/gopath/src/github.com/libp2p/go-libp2p-swarm/swarm_stream.go:161 +0x72
  github.com/libp2p/go-libp2p/p2p/host/basic.(*streamWrapper).SetDeadline()
      <autogenerated>:1 +0x84
  github.com/libp2p/go-libp2p/p2p/host/basic.(*BasicHost).newStreamHandler()
      /Users/anacrolix/gopath/src/github.com/libp2p/go-libp2p/p2p/host/basic/basic_host.go:247 +0x97a
  github.com/libp2p/go-libp2p/p2p/host/basic.(*BasicHost).newStreamHandler-fm()
      /Users/anacrolix/gopath/src/github.com/libp2p/go-libp2p/p2p/host/basic/basic_host.go:163 +0x55
  github.com/libp2p/go-libp2p-swarm.(*Conn).start.func1.1()
      /Users/anacrolix/gopath/src/github.com/libp2p/go-libp2p-swarm/swarm_conn.go:115 +0xfc

Goroutine 1059 (running) created at:
  github.com/multiformats/go-multistream.(*MultistreamMuxer).NegotiateLazy()
      /Users/anacrolix/gopath/src/github.com/multiformats/go-multistream/multistream.go:194 +0x2fc
  github.com/libp2p/go-libp2p/p2p/host/basic.(*BasicHost).newStreamHandler()
      /Users/anacrolix/gopath/src/github.com/libp2p/go-libp2p/p2p/host/basic/basic_host.go:225 +0x12a
  github.com/libp2p/go-libp2p/p2p/host/basic.(*BasicHost).newStreamHandler-fm()
      /Users/anacrolix/gopath/src/github.com/libp2p/go-libp2p/p2p/host/basic/basic_host.go:163 +0x55
  github.com/libp2p/go-libp2p-swarm.(*Conn).start.func1.1()
      /Users/anacrolix/gopath/src/github.com/libp2p/go-libp2p-swarm/swarm_conn.go:115 +0xfc

Goroutine 1168 (finished) created at:
  github.com/libp2p/go-libp2p-swarm.(*Conn).start.func1()
      /Users/anacrolix/gopath/src/github.com/libp2p/go-libp2p-swarm/swarm_conn.go:102 +0x10e

Stream.waitForData goroutine leak

> curl https://ipfs.io/ipfs/QmWPNtfibbs7E78ZrNAC6GLkGF5y1U2h5T8WgbJi8KMqpD | stackparse --wait-more-than=30m --summary
[...]
gx/ipfs/QmTKsRYeY4simJyf37K93juSq75Lo8MVCDJ7owjmf46u8W/go-context/io.(*ctxReader).Read                 48
gx/ipfs/QmNhuMfExDSZvjyrUk9FkC1Cb3vqx2ErCicFwxq2f7SHbW/floodsub.(*PubSub).handleSendingMessages        53
gx/ipfs/QmQvbWzZPGpoppaAvBtj6QmyBZPw4ivFD7ryyHesxuYYDa/yamux.(*Session).AcceptStream                   364
gx/ipfs/QmQvbWzZPGpoppaAvBtj6QmyBZPw4ivFD7ryyHesxuYYDa/yamux.(*Stream).Read                            642
gx/ipfs/QmbX2bvPnLs2xPoJcFV7LHyn8bh3hxzwRaoEsZpmgP5hXn/go-multiplex.(*Stream).waitForData              9710

add license