Skip to content
This repository was archived by the owner on Sep 9, 2022. It is now read-only.

call Stream.Reset instead of Stream.Close #76

Merged
merged 7 commits into from
May 22, 2019
Merged

call Stream.Reset instead of Stream.Close #76

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
e972c1f
call Stream.Reset instead of Stream.Close
Stebalien May 8, 2019
4a96ae7
fix tests for reset on close
Stebalien May 21, 2019
1804298
fix bug in conn.SetWriteDeadline
vyzo May 22, 2019
2ec9f71
tag hop relays for active stop connections
vyzo May 22, 2019
623ae60
add comments for tagging logic
vyzo May 22, 2019
5495b30
store the host instead of the Relay instance in Conn
vyzo May 22, 2019
776794b
Merge pull request #77 from libp2p/feat/stop-tags
Stebalien May 22, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 30 additions & 5 deletions conn.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package relay
import (
"fmt"
"net"
"time"

inet "github.com/libp2p/go-libp2p-net"
pstore "github.com/libp2p/go-libp2p-peerstore"
Expand All @@ -11,7 +12,7 @@ import (
)

type Conn struct {
inet.Stream
stream inet.Stream
remote pstore.PeerInfo
}

Expand All @@ -28,29 +29,53 @@ func (n *NetAddr) String() string {
return fmt.Sprintf("relay[%s-%s]", n.Remote, n.Relay)
}

func (c *Conn) Close() error {
return c.stream.Reset()
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe this is relevant here. This is how I ended up doing it in gostream :

https://github.com/hsanjuan/go-libp2p-gostream/blob/master/conn.go#L38

// Close closes the connection.
// Any blocked Read or Write operations will be unblocked and return errors.
func (c *conn) Close() error {
	if err := c.s.Close(); err != nil {
		c.s.Reset()
		return err
	}
	go pnet.AwaitEOF(c.s)
	return nil
}

If I remember well, bluntly resetting libp2p streams on Close() caused errors on the other side on situations where the stream closing was supposed to be a clean operation.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, but we need to be careful:

  1. Concurrent reads aren't safe. We need to interrupt any concurrent reads (somehow) before we can do this. We should be able to do this by setting a read deadline but... we'll have to make sure to fix any/all of our multiplexers/transports (deadlines currently don't interrupt).
  2. We probably want to set some timeouts/deadlines.
  3. We should probably call Close() in the background as well. That will mimic a normal close(tcpFileDescriptor) call (which will let the kernel flush in the background).

IMO, the real issue here is that TCPConn.Close is actually a very fancy best-effort function that does a bunch of dirty work in the background. Ideally, even if we make Close close both directions, we wouldn't be that sloppy.

}

func (c *Conn) Read(buf []byte) (int, error) {
return c.stream.Read(buf)
}

func (c *Conn) Write(buf []byte) (int, error) {
return c.stream.Write(buf)
}

func (c *Conn) SetDeadline(t time.Time) error {
return c.stream.SetDeadline(t)
}

func (c *Conn) SetReadDeadline(t time.Time) error {
return c.stream.SetReadDeadline(t)
}

func (c *Conn) SetWriteDeadline(t time.Time) error {
return c.stream.SetReadDeadline(t)
}

func (c *Conn) RemoteAddr() net.Addr {
return &NetAddr{
Relay: c.Conn().RemotePeer().Pretty(),
Relay: c.stream.Conn().RemotePeer().Pretty(),
Remote: c.remote.ID.Pretty(),
}
}

// TODO: is it okay to cast c.Conn().RemotePeer() into a multiaddr? might be "user input"
func (c *Conn) RemoteMultiaddr() ma.Multiaddr {
proto := ma.ProtocolWithCode(ma.P_P2P).Name
peerid := c.Conn().RemotePeer().Pretty()
peerid := c.stream.Conn().RemotePeer().Pretty()
p2paddr := ma.StringCast(fmt.Sprintf("/%s/%s", proto, peerid))

circaddr := ma.Cast(ma.CodeToVarint(P_CIRCUIT))
return p2paddr.Encapsulate(circaddr)
}

func (c *Conn) LocalMultiaddr() ma.Multiaddr {
return c.Conn().LocalMultiaddr()
return c.stream.Conn().LocalMultiaddr()
}

func (c *Conn) LocalAddr() net.Addr {
na, err := manet.ToNetAddr(c.Conn().LocalMultiaddr())
na, err := manet.ToNetAddr(c.stream.Conn().LocalMultiaddr())
if err != nil {
log.Error("failed to convert local multiaddr to net addr:", err)
return nil
Expand Down
4 changes: 4 additions & 0 deletions go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,10 @@ github.com/google/uuid v1.1.1 h1:Gkbcsh/GbpXz7lPftLA3P6TYMwjCLYm83jiFQZF/3gY=
github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gxed/hashland/keccakpg v0.0.1/go.mod h1:kRzw3HkwxFU1mpmPP8v1WyQzwdGfmKFJ6tItnhQ67kU=
github.com/gxed/hashland/murmur3 v0.0.1/go.mod h1:KjXop02n4/ckmZSnY2+HKcLud/tcmvhST0bie/0lS48=
github.com/hashicorp/errwrap v1.0.0 h1:hLrqtEDnRye3+sgx6z4qVLNuviH3MR5aQ0ykNJa/UYA=
github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
github.com/hashicorp/go-multierror v1.0.0 h1:iVjPR7a6H0tWELX5NxNe7bYopibicUzc7uPribsnS6o=
github.com/hashicorp/go-multierror v1.0.0/go.mod h1:dHtQlpGsu+cZNNAkkCN/P3hoUDHhCYQXV3UM06sGGrk=
github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI=
github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
Expand Down
4 changes: 2 additions & 2 deletions listen.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,10 +25,10 @@ func (r *Relay) Listener() *RelayListener {
func (l *RelayListener) Accept() (manet.Conn, error) {
select {
case c := <-l.incoming:
err := l.Relay().writeResponse(c.Stream, pb.CircuitRelay_SUCCESS)
err := l.Relay().writeResponse(c.stream, pb.CircuitRelay_SUCCESS)
if err != nil {
log.Debugf("error writing relay response: %s", err.Error())
c.Stream.Reset()
c.stream.Reset()
return nil, err
}

Expand Down
4 changes: 2 additions & 2 deletions relay.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,7 +180,7 @@ func (r *Relay) DialPeer(ctx context.Context, relay pstore.PeerInfo, dest pstore
return nil, RelayError{msg.GetCode()}
}

return &Conn{Stream: s, remote: dest}, nil
return &Conn{stream: s, remote: dest}, nil
}

func (r *Relay) Matches(addr ma.Multiaddr) bool {
Expand Down Expand Up @@ -438,7 +438,7 @@ func (r *Relay) handleStopStream(s inet.Stream, msg *pb.CircuitRelay) {
}

select {
case r.incoming <- &Conn{Stream: s, remote: src}:
case r.incoming <- &Conn{stream: s, remote: src}:
case <-time.After(RelayAcceptTimeout):
r.handleError(s, pb.CircuitRelay_STOP_RELAY_REFUSED)
}
Expand Down
118 changes: 96 additions & 22 deletions relay_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,9 @@ import (
"bytes"
"context"
"fmt"
"io"
"io/ioutil"
"net"
"testing"
"time"

Expand Down Expand Up @@ -72,22 +74,38 @@ func TestBasicRelay(t *testing.T) {

r3 := newTestRelay(t, ctx, hosts[2])

var (
conn1, conn2 net.Conn
done = make(chan struct{})
)

defer func() {
<-done
if conn1 != nil {
conn1.Close()
}
if conn2 != nil {
conn2.Close()
}
}()

msg := []byte("relay works!")
go func() {
defer close(done)
list := r3.Listener()

con, err := list.Accept()
var err error
conn1, err = list.Accept()
if err != nil {
t.Error(err)
return
}

_, err = con.Write(msg)
_, err = conn1.Write(msg)
if err != nil {
t.Error(err)
return
}
con.Close()
}()

rinfo := hosts[1].Peerstore().PeerInfo(hosts[1].ID())
Expand All @@ -96,18 +114,20 @@ func TestBasicRelay(t *testing.T) {
rctx, rcancel := context.WithTimeout(ctx, time.Second)
defer rcancel()

con, err := r1.DialPeer(rctx, rinfo, dinfo)
var err error
conn2, err = r1.DialPeer(rctx, rinfo, dinfo)
if err != nil {
t.Fatal(err)
}

data, err := ioutil.ReadAll(con)
result := make([]byte, len(msg))
_, err = io.ReadFull(conn2, result)
if err != nil {
t.Fatal(err)
}

if !bytes.Equal(data, msg) {
t.Fatal("message was incorrect:", string(data))
if !bytes.Equal(result, msg) {
t.Fatal("message was incorrect:", string(result))
}
}

Expand Down Expand Up @@ -186,35 +206,53 @@ func TestBasicRelayDial(t *testing.T) {
_ = newTestRelay(t, ctx, hosts[1], OptHop)
r3 := newTestRelay(t, ctx, hosts[2])

var (
conn1, conn2 net.Conn
done = make(chan struct{})
)

defer func() {
<-done
if conn1 != nil {
conn1.Close()
}
if conn2 != nil {
conn2.Close()
}
}()

msg := []byte("relay works!")
go func() {
defer close(done)
list := r3.Listener()

con, err := list.Accept()
var err error
conn1, err = list.Accept()
if err != nil {
t.Error(err)
return
}

_, err = con.Write(msg)
_, err = conn1.Write(msg)
if err != nil {
t.Error(err)
return
}
con.Close()
}()

addr := ma.StringCast(fmt.Sprintf("/ipfs/%s/p2p-circuit", hosts[1].ID().Pretty()))

rctx, rcancel := context.WithTimeout(ctx, time.Second)
defer rcancel()

con, err := r1.Dial(rctx, addr, hosts[2].ID())
var err error
conn2, err = r1.Dial(rctx, addr, hosts[2].ID())
if err != nil {
t.Fatal(err)
}

data, err := ioutil.ReadAll(con)
data := make([]byte, len(msg))
_, err = io.ReadFull(conn2, data)
if err != nil {
t.Fatal(err)
}
Expand All @@ -241,35 +279,53 @@ func TestUnspecificRelayDial(t *testing.T) {

time.Sleep(100 * time.Millisecond)

var (
conn1, conn2 net.Conn
done = make(chan struct{})
)

defer func() {
<-done
if conn1 != nil {
conn1.Close()
}
if conn2 != nil {
conn2.Close()
}
}()

msg := []byte("relay works!")
go func() {
defer close(done)
list := r3.Listener()

con, err := list.Accept()
var err error
conn1, err = list.Accept()
if err != nil {
t.Error(err)
return
}

_, err = con.Write(msg)
_, err = conn1.Write(msg)
if err != nil {
t.Error(err)
return
}
con.Close()
}()

addr := ma.StringCast(fmt.Sprintf("/p2p-circuit"))

rctx, rcancel := context.WithTimeout(ctx, time.Second)
defer rcancel()

con, err := r1.Dial(rctx, addr, hosts[2].ID())
var err error
conn2, err = r1.Dial(rctx, addr, hosts[2].ID())
if err != nil {
t.Fatal(err)
}

data, err := ioutil.ReadAll(con)
data := make([]byte, len(msg))
_, err = io.ReadFull(conn2, data)
if err != nil {
t.Fatal(err)
}
Expand Down Expand Up @@ -368,22 +424,38 @@ func TestActiveRelay(t *testing.T) {

r3 := newTestRelay(t, ctx, hosts[2])

var (
conn1, conn2 net.Conn
done = make(chan struct{})
)

defer func() {
<-done
if conn1 != nil {
conn1.Close()
}
if conn2 != nil {
conn2.Close()
}
}()

msg := []byte("relay works!")
go func() {
defer close(done)
list := r3.Listener()

con, err := list.Accept()
var err error
conn2, err = list.Accept()
if err != nil {
t.Error(err)
return
}

_, err = con.Write(msg)
_, err = conn2.Write(msg)
if err != nil {
t.Error(err)
return
}
con.Close()
}()

rinfo := hosts[1].Peerstore().PeerInfo(hosts[1].ID())
Expand All @@ -392,12 +464,14 @@ func TestActiveRelay(t *testing.T) {
rctx, rcancel := context.WithTimeout(ctx, time.Second)
defer rcancel()

con, err := r1.DialPeer(rctx, rinfo, dinfo)
var err error
conn2, err = r1.DialPeer(rctx, rinfo, dinfo)
if err != nil {
t.Fatal(err)
}

data, err := ioutil.ReadAll(con)
data := make([]byte, len(msg))
_, err = io.ReadFull(conn2, data)
if err != nil {
t.Fatal(err)
}
Expand Down