Merge branch 'master' of https://github.com/erigontech/torrent

.github/workflows/codeql-analysis.yml

@@ -16,7 +16,7 @@ jobs:
 
     steps:
     - name: Checkout repository
-      uses: actions/checkout@v3
+      uses: actions/checkout@v4
       with:
         # We must fetch at least the immediate parents so that if this is
         # a pull request then we can checkout the head.

.github/workflows/go.yml

@@ -13,7 +13,7 @@ jobs:
         os: [windows-latest, macos-latest, ubuntu-latest]
       fail-fast: false
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v4
     - uses: ./.github/actions/go-common
     - run: go test -race $(go list ./... | grep -v /fs | grep -v /possum)
       continue-on-error: true
@@ -28,7 +28,7 @@ jobs:
         go-version: [ '1.22' ]
       fail-fast: false
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v4
     - uses: ./.github/actions/go-common
     - run: go test -race -run @ -bench . -benchtime 2x $(go list ./... | grep -v /possum)
 
@@ -39,7 +39,7 @@ jobs:
         go-version: [ '1.22' ]
       fail-fast: false
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v4
     - uses: ./.github/actions/go-common
     - run: go test -run @ -bench . $(go list ./... | grep -v /possum)
 
@@ -62,7 +62,7 @@ jobs:
         go-version: [ '1.22' ]
       fail-fast: false
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v4
     - uses: ./.github/actions/go-common
     - name: Some packages compile for WebAssembly
       run: GOOS=js GOARCH=wasm go build . ./storage ./tracker/...
@@ -77,7 +77,7 @@ jobs:
         os: [ubuntu-latest]
       fail-fast: false
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v4
     - uses: ./.github/actions/go-common
 
     - name: Install godo

.github/workflows/linter.yml

@@ -13,7 +13,7 @@ jobs:
     env:
       GO111MODULE: on
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v4
       - uses: golangci/golangci-lint-action@v2
         with:
           version: latest

go.mod

@@ -67,7 +67,7 @@ require (
 	github.com/bits-and-blooms/bitset v1.2.2 // indirect
 	github.com/cenkalti/backoff/v4 v4.1.3 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
-	github.com/go-llsqlite/crawshaw v0.4.0 // indirect
+	github.com/go-llsqlite/crawshaw v0.5.2-0.20240425034140-f30eb7704568 // indirect
 	github.com/go-logr/logr v1.2.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/golang/protobuf v1.5.3 // indirect

go.sum

@@ -212,8 +212,8 @@ github.com/go-kit/log v0.1.0/go.mod h1:zbhenjAZHb184qTLMA9ZjW7ThYL0H2mk7Q6pNt4vb
 github.com/go-kit/log v0.2.0/go.mod h1:NwTd00d/i8cPZ3xOwwiv2PO5MOcx78fFErGNcVmBjv0=
 github.com/go-llsqlite/adapter v0.0.0-20230927005056-7f5ce7f0c916 h1:OyQmpAN302wAopDgwVjgs2HkFawP9ahIEqkUYz7V7CA=
 github.com/go-llsqlite/adapter v0.0.0-20230927005056-7f5ce7f0c916/go.mod h1:DADrR88ONKPPeSGjFp5iEN55Arx3fi2qXZeKCYDpbmU=
-github.com/go-llsqlite/crawshaw v0.4.0 h1:L02s2jZBBJj80xm1VkkdyB/JlQ/Fi0kLbNHfXA8yrec=
-github.com/go-llsqlite/crawshaw v0.4.0/go.mod h1:/YJdV7uBQaYDE0fwe4z3wwJIZBJxdYzd38ICggWqtaE=
+github.com/go-llsqlite/crawshaw v0.5.2-0.20240425034140-f30eb7704568 h1:3EpZo8LxIzF4q3BT+vttQQlRfA6uTtTb/cxVisWa5HM=
+github.com/go-llsqlite/crawshaw v0.5.2-0.20240425034140-f30eb7704568/go.mod h1:/YJdV7uBQaYDE0fwe4z3wwJIZBJxdYzd38ICggWqtaE=
 github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
 github.com/go-logfmt/logfmt v0.5.0/go.mod h1:wCYkCAKZfumFQihp8CzCvQ3paCTfi41vtzG1KdI/P7A=

peer-conn-msg-writer.go

@@ -117,10 +117,21 @@ func (cn *peerConnMsgWriter) run(keepAliveTimeout time.Duration) {
 	}
 }
 
+func (cn *peerConnMsgWriter) writeToBuffer(msg pp.Message) (err error) {
+	originalLen := cn.writeBuffer.Len()
+	defer func() {
+		if err != nil {
+			// Since an error occurred during buffer write, revert buffer to its original state before the write.
+			cn.writeBuffer.Truncate(originalLen)
+		}
+	}()
+	return msg.WriteTo(cn.writeBuffer)
+}
+
 func (cn *peerConnMsgWriter) write(msg pp.Message) bool {
 	cn.mu.Lock()
 	defer cn.mu.Unlock()
-	cn.writeBuffer.Write(msg.MustMarshalBinary())
+	cn.writeToBuffer(msg)
 	cn.writeCond.Broadcast()
 	return !cn.writeBufferFull()
 }

peer-conn-msg-writer_test.go

@@ -0,0 +1,68 @@
+package torrent
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/dustin/go-humanize"
+
+	pp "github.com/anacrolix/torrent/peer_protocol"
+)
+
+func PieceMsg(length int64) pp.Message {
+	return pp.Message{
+		Type:  pp.Piece,
+		Index: pp.Integer(0),
+		Begin: pp.Integer(0),
+		Piece: make([]byte, length),
+	}
+}
+
+var benchmarkPieceLengths = []int{defaultChunkSize, 1 << 20, 4 << 20, 8 << 20}
+
+func runBenchmarkWriteToBuffer(b *testing.B, length int64) {
+	writer := &peerConnMsgWriter{
+		writeBuffer: &bytes.Buffer{},
+	}
+	msg := PieceMsg(length)
+
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		//b.StopTimer()
+		writer.writeBuffer.Reset()
+		//b.StartTimer()
+		writer.writeToBuffer(msg)
+	}
+}
+
+func BenchmarkWritePieceMsg(b *testing.B) {
+	for _, length := range benchmarkPieceLengths {
+		b.Run(humanize.IBytes(uint64(length)), func(b *testing.B) {
+			b.Run("ToBuffer", func(b *testing.B) {
+				b.SetBytes(int64(length))
+				runBenchmarkWriteToBuffer(b, int64(length))
+			})
+			b.Run("MarshalBinary", func(b *testing.B) {
+				b.SetBytes(int64(length))
+				runBenchmarkMarshalBinaryWrite(b, int64(length))
+			})
+		})
+	}
+}
+
+func runBenchmarkMarshalBinaryWrite(b *testing.B, length int64) {
+	writer := &peerConnMsgWriter{
+		writeBuffer: &bytes.Buffer{},
+	}
+	msg := PieceMsg(length)
+
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		//b.StopTimer()
+		writer.writeBuffer.Reset()
+		//b.StartTimer()
+		writer.writeBuffer.Write(msg.MustMarshalBinary())
+	}
+}

peer_protocol/msg.go

@@ -6,6 +6,7 @@ import (
 	"encoding"
 	"encoding/binary"
 	"fmt"
+	"io"
 )
 
 // This is a lazy union representing all the possible fields for messages. Go doesn't have ADTs, and
@@ -61,13 +62,14 @@ func (msg Message) MustMarshalBinary() []byte {
 	return b
 }
 
-func (msg Message) MarshalBinary() (data []byte, err error) {
-	// It might look like you could have a pool of buffers and preallocate the message length
-	// prefix, but because we have to return []byte, it becomes non-trivial to make this fast. You
-	// will need a benchmark.
-	var buf bytes.Buffer
+type MessageWriter interface {
+	io.ByteWriter
+	io.Writer
+}
+
+func (msg *Message) writePayloadTo(buf MessageWriter) (err error) {
 	mustWrite := func(data any) {
-		err := binary.Write(&buf, binary.BigEndian, data)
+		err := binary.Write(buf, binary.BigEndian, data)
 		if err != nil {
 			panic(err)
 		}
@@ -85,10 +87,10 @@ func (msg Message) MarshalBinary() (data []byte, err error) {
 		switch msg.Type {
 		case Choke, Unchoke, Interested, NotInterested, HaveAll, HaveNone:
 		case Have, AllowedFast, Suggest:
-			err = binary.Write(&buf, binary.BigEndian, msg.Index)
+			err = binary.Write(buf, binary.BigEndian, msg.Index)
 		case Request, Cancel, Reject:
 			for _, i := range []Integer{msg.Index, msg.Begin, msg.Length} {
-				err = binary.Write(&buf, binary.BigEndian, i)
+				err = binary.Write(buf, binary.BigEndian, i)
 				if err != nil {
 					break
 				}
@@ -97,7 +99,7 @@ func (msg Message) MarshalBinary() (data []byte, err error) {
 			_, err = buf.Write(marshalBitfield(msg.Bitfield))
 		case Piece:
 			for _, i := range []Integer{msg.Index, msg.Begin} {
-				err = binary.Write(&buf, binary.BigEndian, i)
+				err = binary.Write(buf, binary.BigEndian, i)
 				if err != nil {
 					return
 				}
@@ -116,19 +118,43 @@ func (msg Message) MarshalBinary() (data []byte, err error) {
 			}
 			_, err = buf.Write(msg.ExtendedPayload)
 		case Port:
-			err = binary.Write(&buf, binary.BigEndian, msg.Port)
+			err = binary.Write(buf, binary.BigEndian, msg.Port)
 		case HashRequest:
 			buf.Write(msg.PiecesRoot[:])
 			writeConsecutive(msg.BaseLayer, msg.Index, msg.Length, msg.ProofLayers)
 		default:
 			err = fmt.Errorf("unknown message type: %v", msg.Type)
 		}
 	}
-	data = make([]byte, 4+buf.Len())
-	binary.BigEndian.PutUint32(data, uint32(buf.Len()))
-	if buf.Len() != copy(data[4:], buf.Bytes()) {
-		panic("bad copy")
+	return
+}
+
+func (msg *Message) WriteTo(w MessageWriter) (err error) {
+	length, err := msg.getPayloadLength()
+	if err != nil {
+		return
+	}
+	err = binary.Write(w, binary.BigEndian, length)
+	if err != nil {
+		return
 	}
+	return msg.writePayloadTo(w)
+}
+
+func (msg *Message) getPayloadLength() (length Integer, err error) {
+	var lw lengthWriter
+	err = msg.writePayloadTo(&lw)
+	length = lw.n
+	return
+}
+
+func (msg Message) MarshalBinary() (data []byte, err error) {
+	// It might look like you could have a pool of buffers and preallocate the message length
+	// prefix, but because we have to return []byte, it becomes non-trivial to make this fast. You
+	// will need a benchmark.
+	var buf bytes.Buffer
+	err = msg.WriteTo(&buf)
+	data = buf.Bytes()
 	return
 }
 
@@ -158,3 +184,18 @@ func (me *Message) UnmarshalBinary(b []byte) error {
 	}
 	return nil
 }
+
+type lengthWriter struct {
+	n Integer
+}
+
+func (l *lengthWriter) WriteByte(c byte) error {
+	l.n++
+	return nil
+}
+
+func (l *lengthWriter) Write(p []byte) (n int, err error) {
+	n = len(p)
+	l.n += Integer(n)
+	return
+}

requesting.go

@@ -9,9 +9,8 @@ import (
 	"time"
 	"unsafe"
 
-	g "github.com/anacrolix/generics"
-
 	"github.com/RoaringBitmap/roaring"
+	g "github.com/anacrolix/generics"
 	"github.com/anacrolix/generics/heap"
 	"github.com/anacrolix/log"
 	"github.com/anacrolix/multiless"
@@ -313,20 +312,13 @@ func (p *Peer) applyRequestState(next desiredRequestState) {
 			panic("changed")
 		}
 
-		// don't add requests on reciept of a reject - because this causes request back
-		// to potentially permanently unresponive peers - which just adds network noise.  If
-		// the peer can handle more requests it will send an "unchoked" message - which
-		// will cause it to get added back to the request queue
-		if p.needRequestUpdate == peerUpdateRequestsRemoteRejectReason {
+		if p.needRequestUpdate == "Peer.remoteRejectedRequest" {
 			continue
 		}
 
 		existing := t.requestingPeer(req)
 		if existing != nil && existing != p {
-			// don't steal on cancel - because this is triggered by t.cancelRequest below
-			// which means that the cancelled can immediately try to steal back a request
-			// it has lost which can lead to circular cancel/add processing
-			if p.needRequestUpdate == peerUpdateRequestsPeerCancelReason {
+			if p.needRequestUpdate == "Peer.cancel" {
 				continue
 			}
 

storage/file.go

@@ -90,6 +90,7 @@ func (fs fileClientImpl) OpenTorrent(info *metainfo.Info, infoHash metainfo.Hash
 	return TorrentImpl{
 		Piece: t.Piece,
 		Close: t.Close,
+		Flush: t.Flush,
 	}, nil
 }
 
@@ -122,6 +123,29 @@ func (fs *fileTorrentImpl) Close() error {
 	return nil
 }
 
+func fsync(filePath string) (err error) {
+	_ = os.MkdirAll(filepath.Dir(filePath), 0o777)
+	var f *os.File
+	f, err = os.OpenFile(filePath, os.O_WRONLY|os.O_CREATE, 0o666)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	if err = f.Sync(); err != nil {
+		return err
+	}
+	return f.Close()
+}
+
+func (fts *fileTorrentImpl) Flush() error {
+	for _, f := range fts.files {
+		if err := fsync(f.path); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // A helper to create zero-length files which won't appear for file-orientated storage since no
 // writes will ever occur to them (no torrent data is associated with a zero-length file). The
 // caller should make sure the file name provided is safe/sanitized.

tests/go.work

@@ -1,6 +1,6 @@
-go 1.22.0
+go 1.22.2
 
 use (
-	.
 	..
+	./issue-930
 )

tests/go.work.sum

@@ -418,8 +418,10 @@ golang.org/x/mod v0.11.0 h1:bUO06HqtnRcc/7l71XBe4WcqTZ+3AH1J59zWDDwLKgU=
 golang.org/x/mod v0.11.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
 golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20220909164309-bea034e7d591/go.mod h1:YDH+HFinaLZZlnHAfSS6ZXJJ9M9t4Dl22yv3iI2vPwk=
+golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.8.0/go.mod h1:QVkue5JL9kW//ek3r6jTKnTFis1tRmNAW2P1shuFdJc=
 golang.org/x/net v0.9.0/go.mod h1:d48xBJpPfHeWQsugry2m+kC02ZBRGRgulfHnEXEuWns=
+golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
 golang.org/x/oauth2 v0.0.0-20221014153046-6fdb5e3db783/go.mod h1:h4gKUeWbJ4rQPri7E0u6Gs4e9Ri2zaLxzw5DI5XGrYg=
 golang.org/x/oauth2 v0.6.0/go.mod h1:ycmewcwgD4Rpr3eZJLSB4Kyyljb3qDh40vJ8STE5HKw=
 golang.org/x/oauth2 v0.7.0 h1:qe6s0zUXlPX80/dITx3440hWZ7GwMwgDDyrSGTPJG/g=
@@ -433,6 +435,8 @@ golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
 golang.org/x/term v0.15.0 h1:y/Oo/a/q3IXu26lQgl04j/gjuBDOBlx7X6Om1j2CPW4=
 golang.org/x/term v0.15.0/go.mod h1:BDl952bC7+uMoWR75FIrCDx79TPU9oHkTZ9yRbYOrX0=
+golang.org/x/term v0.18.0 h1:FcHjZXDMxI8mM3nwhX9HlKop4C0YQvCVCdwYl2wOtE8=
+golang.org/x/term v0.18.0/go.mod h1:ILwASektA3OnRv7amZ1xhE/KTR+u50pbXfZ03+6Nx58=
 golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
 golang.org/x/text v0.8.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
 golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=