FedP2P/peerconn_test.go

package torrent

import (
	"errors"
	"io"
	"net"
	"sync"
	"testing"

	"github.com/frankban/quicktest"
	qt "github.com/frankban/quicktest"
	"github.com/stretchr/testify/require"

	"github.com/anacrolix/torrent/metainfo"
	pp "github.com/anacrolix/torrent/peer_protocol"
	"github.com/anacrolix/torrent/storage"
)

// Ensure that no race exists between sending a bitfield, and a subsequent
// Have that would potentially alter it.
func TestSendBitfieldThenHave(t *testing.T) {
	var cl Client
	cl.init(TestingConfig(t))
	cl.initLogger()
	c := cl.newConnection(nil, false, nil, "io.Pipe", "")
	c.setTorrent(cl.newTorrent(metainfo.Hash{}, nil))
	if err := c.t.setInfo(&metainfo.Info{Pieces: make([]byte, metainfo.HashSize*3)}); err != nil {
		t.Log(err)
	}
	r, w := io.Pipe()
	// c.r = r
	c.w = w
	c.startWriter()
	c.locker().Lock()
	c.t._completedPieces.Add(1)
	c.postBitfield( /*[]bool{false, true, false}*/ )
	c.locker().Unlock()
	c.locker().Lock()
	c.have(2)
	c.locker().Unlock()
	b := make([]byte, 15)
	n, err := io.ReadFull(r, b)
	c.locker().Lock()
	// This will cause connection.writer to terminate.
	c.closed.Set()
	c.locker().Unlock()
	require.NoError(t, err)
	require.EqualValues(t, 15, n)
	// Here we see that the bitfield doesn't have piece 2 set, as that should
	// arrive in the following Have message.
	require.EqualValues(t, "\x00\x00\x00\x02\x05@\x00\x00\x00\x05\x04\x00\x00\x00\x02", string(b))
}

type torrentStorage struct {
	writeSem sync.Mutex
}

func (me *torrentStorage) Close() error { return nil }

func (me *torrentStorage) Piece(mp metainfo.Piece) storage.PieceImpl {
	return me
}

func (me *torrentStorage) Completion() storage.Completion {
	return storage.Completion{}
}

func (me *torrentStorage) MarkComplete() error {
	return nil
}

func (me *torrentStorage) MarkNotComplete() error {
	return nil
}

func (me *torrentStorage) ReadAt([]byte, int64) (int, error) {
	panic("shouldn't be called")
}

func (me *torrentStorage) WriteAt(b []byte, _ int64) (int, error) {
	if len(b) != defaultChunkSize {
		panic(len(b))
	}
	me.writeSem.Unlock()
	return len(b), nil
}

func BenchmarkConnectionMainReadLoop(b *testing.B) {
	c := quicktest.New(b)
	var cl Client
	cl.init(&ClientConfig{
		DownloadRateLimiter: unlimited,
	})
	cl.initLogger()
	ts := &torrentStorage{}
	t := cl.newTorrent(metainfo.Hash{}, nil)
	t.initialPieceCheckDisabled = true
	require.NoError(b, t.setInfo(&metainfo.Info{
		Pieces:      make([]byte, 20),
		Length:      1 << 20,
		PieceLength: 1 << 20,
	}))
	t.storage = &storage.Torrent{TorrentImpl: storage.TorrentImpl{Piece: ts.Piece, Close: ts.Close}}
	t.onSetInfo()
	t._pendingPieces.Add(0)
	r, w := net.Pipe()
	cn := cl.newConnection(r, true, r.RemoteAddr(), r.RemoteAddr().Network(), regularNetConnPeerConnConnString(r))
	cn.setTorrent(t)
	mrlErrChan := make(chan error)
	msg := pp.Message{
		Type:  pp.Piece,
		Piece: make([]byte, defaultChunkSize),
	}
	go func() {
		cl.lock()
		err := cn.mainReadLoop()
		if err != nil {
			mrlErrChan <- err
		}
		close(mrlErrChan)
	}()
	wb := msg.MustMarshalBinary()
	b.SetBytes(int64(len(msg.Piece)))
	go func() {
		ts.writeSem.Lock()
		for i := 0; i < b.N; i += 1 {
			cl.lock()
			// The chunk must be written to storage everytime, to ensure the
			// writeSem is unlocked.
			t.pendAllChunkSpecs(0)
			cn.validReceiveChunks = map[RequestIndex]int{
				t.requestIndexFromRequest(newRequestFromMessage(&msg)): 1,
			}
			cl.unlock()
			n, err := w.Write(wb)
			require.NoError(b, err)
			require.EqualValues(b, len(wb), n)
			ts.writeSem.Lock()
		}
		if err := w.Close(); err != nil {
			panic(err)
		}
	}()
	mrlErr := <-mrlErrChan
	if mrlErr != nil && !errors.Is(mrlErr, io.EOF) {
		c.Fatal(mrlErr)
	}
	c.Assert(cn._stats.ChunksReadUseful.Int64(), quicktest.Equals, int64(b.N))
}

func TestConnPexPeerFlags(t *testing.T) {
	var (
		tcpAddr = &net.TCPAddr{IP: net.IPv6loopback, Port: 4848}
		udpAddr = &net.UDPAddr{IP: net.IPv6loopback, Port: 4848}
	)
	testcases := []struct {
		conn *PeerConn
		f    pp.PexPeerFlags
	}{
		{&PeerConn{Peer: Peer{outgoing: false, PeerPrefersEncryption: false}}, 0},
		{&PeerConn{Peer: Peer{outgoing: false, PeerPrefersEncryption: true}}, pp.PexPrefersEncryption},
		{&PeerConn{Peer: Peer{outgoing: true, PeerPrefersEncryption: false}}, pp.PexOutgoingConn},
		{&PeerConn{Peer: Peer{outgoing: true, PeerPrefersEncryption: true}}, pp.PexOutgoingConn | pp.PexPrefersEncryption},
		{&PeerConn{Peer: Peer{RemoteAddr: udpAddr, Network: udpAddr.Network()}}, pp.PexSupportsUtp},
		{&PeerConn{Peer: Peer{RemoteAddr: udpAddr, Network: udpAddr.Network(), outgoing: true}}, pp.PexOutgoingConn | pp.PexSupportsUtp},
		{&PeerConn{Peer: Peer{RemoteAddr: tcpAddr, Network: tcpAddr.Network(), outgoing: true}}, pp.PexOutgoingConn},
		{&PeerConn{Peer: Peer{RemoteAddr: tcpAddr, Network: tcpAddr.Network()}}, 0},
	}
	for i, tc := range testcases {
		f := tc.conn.pexPeerFlags()
		require.EqualValues(t, tc.f, f, i)
	}
}

func TestConnPexEvent(t *testing.T) {
	var (
		udpAddr     = &net.UDPAddr{IP: net.IPv6loopback, Port: 4848}
		tcpAddr     = &net.TCPAddr{IP: net.IPv6loopback, Port: 4848}
		dialTcpAddr = &net.TCPAddr{IP: net.IPv6loopback, Port: 4747}
		dialUdpAddr = &net.UDPAddr{IP: net.IPv6loopback, Port: 4747}
	)
	testcases := []struct {
		t pexEventType
		c *PeerConn
		e pexEvent
	}{
		{
			pexAdd,
			&PeerConn{Peer: Peer{RemoteAddr: udpAddr, Network: udpAddr.Network()}},
			pexEvent{pexAdd, udpAddr, pp.PexSupportsUtp, nil},
		},
		{
			pexDrop,
			&PeerConn{Peer: Peer{RemoteAddr: tcpAddr, Network: tcpAddr.Network(), outgoing: true, PeerListenPort: dialTcpAddr.Port}},
			pexEvent{pexDrop, tcpAddr, pp.PexOutgoingConn, nil},
		},
		{
			pexAdd,
			&PeerConn{Peer: Peer{RemoteAddr: tcpAddr, Network: tcpAddr.Network(), PeerListenPort: dialTcpAddr.Port}},
			pexEvent{pexAdd, dialTcpAddr, 0, nil},
		},
		{
			pexDrop,
			&PeerConn{Peer: Peer{RemoteAddr: udpAddr, Network: udpAddr.Network(), PeerListenPort: dialUdpAddr.Port}},
			pexEvent{pexDrop, dialUdpAddr, pp.PexSupportsUtp, nil},
		},
	}
	for i, tc := range testcases {
		e := tc.c.pexEvent(tc.t)
		require.EqualValues(t, tc.e, e, i)
	}
}

func TestHaveAllThenBitfield(t *testing.T) {
	c := qt.New(t)
	cl := newTestingClient(t)
	tt := cl.newTorrentForTesting()
	// cl.newConnection()
	pc := PeerConn{
		Peer: Peer{t: tt},
	}
	pc.peerImpl = &pc
	tt.conns[&pc] = struct{}{}
	c.Assert(pc.onPeerSentHaveAll(), qt.IsNil)
	c.Check(pc.t.connsWithAllPieces, qt.DeepEquals, map[*Peer]struct{}{&pc.Peer: {}})
	pc.peerSentBitfield([]bool{false, false, true, false, true, true, false, false})
	c.Check(pc.peerMinPieces, qt.Equals, 6)
	c.Check(pc.t.connsWithAllPieces, qt.HasLen, 0)
	c.Assert(pc.t.setInfo(&metainfo.Info{
		PieceLength: 0,
		Pieces:      make([]byte, pieceHash.Size()*7),
	}), qt.IsNil)
	pc.t.onSetInfo()
	c.Check(tt.numPieces(), qt.Equals, 7)
	c.Check(tt.pieceAvailabilityRuns(), qt.DeepEquals, []pieceAvailabilityRun{
		// The last element of the bitfield is irrelevant, as the Torrent actually only has 7
		// pieces.
		{2, 0}, {1, 1}, {1, 0}, {2, 1}, {1, 0},
	})
}

func TestApplyRequestStateWriteBufferConstraints(t *testing.T) {
	c := qt.New(t)
	c.Check(interestedMsgLen, qt.Equals, 5)
	c.Check(requestMsgLen, qt.Equals, 17)
	c.Check(maxLocalToRemoteRequests >= 8, qt.IsTrue)
	c.Logf("max local to remote requests: %v", maxLocalToRemoteRequests)
}