Update vendor/github.com/anacrolix (#909)

- This fixes a bug where the torrent client didn't unlock the bolt db
  See: anacrolix/torrent/issues/158

Godeps/Godeps.json

@@ -23,55 +23,55 @@
 		},
 		{
 			"ImportPath": "github.com/anacrolix/missinggo",
-			"Rev": "00e7d699a9063f958f13814802cd39e5ac9c85b4"
+			"Rev": "7d1b77890fe200fb3e8097af511af1890420b1e0"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/missinggo/bitmap",
-			"Rev": "00e7d699a9063f958f13814802cd39e5ac9c85b4"
+			"Rev": "7d1b77890fe200fb3e8097af511af1890420b1e0"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/missinggo/container/xheap",
-			"Rev": "00e7d699a9063f958f13814802cd39e5ac9c85b4"
+			"Rev": "7d1b77890fe200fb3e8097af511af1890420b1e0"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/missinggo/httptoo",
-			"Rev": "00e7d699a9063f958f13814802cd39e5ac9c85b4"
+			"Rev": "7d1b77890fe200fb3e8097af511af1890420b1e0"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/missinggo/inproc",
-			"Rev": "00e7d699a9063f958f13814802cd39e5ac9c85b4"
+			"Rev": "7d1b77890fe200fb3e8097af511af1890420b1e0"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/missinggo/iter",
-			"Rev": "00e7d699a9063f958f13814802cd39e5ac9c85b4"
+			"Rev": "7d1b77890fe200fb3e8097af511af1890420b1e0"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/missinggo/orderedmap",
-			"Rev": "00e7d699a9063f958f13814802cd39e5ac9c85b4"
+			"Rev": "7d1b77890fe200fb3e8097af511af1890420b1e0"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/missinggo/perf",
-			"Rev": "00e7d699a9063f958f13814802cd39e5ac9c85b4"
+			"Rev": "7d1b77890fe200fb3e8097af511af1890420b1e0"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/missinggo/pproffd",
-			"Rev": "00e7d699a9063f958f13814802cd39e5ac9c85b4"
+			"Rev": "7d1b77890fe200fb3e8097af511af1890420b1e0"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/missinggo/prioritybitmap",
-			"Rev": "00e7d699a9063f958f13814802cd39e5ac9c85b4"
+			"Rev": "7d1b77890fe200fb3e8097af511af1890420b1e0"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/missinggo/pubsub",
-			"Rev": "00e7d699a9063f958f13814802cd39e5ac9c85b4"
+			"Rev": "7d1b77890fe200fb3e8097af511af1890420b1e0"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/missinggo/resource",
-			"Rev": "00e7d699a9063f958f13814802cd39e5ac9c85b4"
+			"Rev": "7d1b77890fe200fb3e8097af511af1890420b1e0"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/missinggo/slices",
-			"Rev": "00e7d699a9063f958f13814802cd39e5ac9c85b4"
+			"Rev": "7d1b77890fe200fb3e8097af511af1890420b1e0"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/sync",
@@ -79,51 +79,51 @@
 		},
 		{
 			"ImportPath": "github.com/anacrolix/torrent",
-			"Rev": "58a24196ce836032e37f48bc3e70571a9cdc735b"
+			"Rev": "77e81061f54e73c30c840d03d0dc408cdc0f73ea"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/torrent/bencode",
-			"Rev": "58a24196ce836032e37f48bc3e70571a9cdc735b"
+			"Rev": "77e81061f54e73c30c840d03d0dc408cdc0f73ea"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/torrent/iplist",
-			"Rev": "58a24196ce836032e37f48bc3e70571a9cdc735b"
+			"Rev": "77e81061f54e73c30c840d03d0dc408cdc0f73ea"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/torrent/logonce",
-			"Rev": "58a24196ce836032e37f48bc3e70571a9cdc735b"
+			"Rev": "77e81061f54e73c30c840d03d0dc408cdc0f73ea"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/torrent/metainfo",
-			"Rev": "58a24196ce836032e37f48bc3e70571a9cdc735b"
+			"Rev": "77e81061f54e73c30c840d03d0dc408cdc0f73ea"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/torrent/mmap_span",
-			"Rev": "58a24196ce836032e37f48bc3e70571a9cdc735b"
+			"Rev": "77e81061f54e73c30c840d03d0dc408cdc0f73ea"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/torrent/mse",
-			"Rev": "58a24196ce836032e37f48bc3e70571a9cdc735b"
+			"Rev": "77e81061f54e73c30c840d03d0dc408cdc0f73ea"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/torrent/peer_protocol",
-			"Rev": "58a24196ce836032e37f48bc3e70571a9cdc735b"
+			"Rev": "77e81061f54e73c30c840d03d0dc408cdc0f73ea"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/torrent/storage",
-			"Rev": "58a24196ce836032e37f48bc3e70571a9cdc735b"
+			"Rev": "77e81061f54e73c30c840d03d0dc408cdc0f73ea"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/torrent/tracker",
-			"Rev": "58a24196ce836032e37f48bc3e70571a9cdc735b"
+			"Rev": "77e81061f54e73c30c840d03d0dc408cdc0f73ea"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/torrent/util",
-			"Rev": "58a24196ce836032e37f48bc3e70571a9cdc735b"
+			"Rev": "77e81061f54e73c30c840d03d0dc408cdc0f73ea"
 		},
 		{
 			"ImportPath": "github.com/anacrolix/utp",
-			"Rev": "1a2093d4dfa2a614da54529a1d63109e7ae34f93"
+			"Rev": "adfb02f3172f79816912bcf58f2b58531624ae68"
 		},
 		{
 			"ImportPath": "github.com/azhao12345/gorm",

service/torrent/metainfoFetcher/operation_test.go

@@ -22,6 +22,7 @@ func TestInvalidHash(t *testing.T) {
 	if err != nil {
 		t.Skipf("Failed to create client, with err %v. Skipping.", err)
 	}
+	defer client.Close()
 
 	fetcher := &MetainfoFetcher{
 		timeout:       5,

vendor/github.com/anacrolix/missinggo/httptoo/closenotify.go

@@ -1,29 +0,0 @@
-package httptoo
-
-import (
-	"context"
-	"net/http"
-)
-
-var closeNotifyContextKey = new(struct{})
-
-func RequestWithCloseNotify(r *http.Request, w http.ResponseWriter) *http.Request {
-	if r.Context().Value(closeNotifyContextKey) != nil {
-		return r
-	}
-	v := make(chan struct{})
-	r = r.WithContext(context.WithValue(r.Context(), closeNotifyContextKey, v))
-	cn := w.(http.CloseNotifier).CloseNotify()
-	go func() {
-		select {
-		case <-cn:
-			close(v)
-		case <-r.Context().Done():
-		}
-	}()
-	return r
-}
-
-func RequestCloseNotifyValue(r *http.Request) <-chan struct{} {
-	return r.Context().Value(closeNotifyContextKey).(chan struct{})
-}

vendor/github.com/anacrolix/missinggo/httptoo/inproc_roundtrip.go

@@ -1,6 +1,7 @@
 package httptoo
 
 import (
+	"context"
 	"io"
 	"net/http"
 	"sync"
@@ -13,18 +14,20 @@ type responseWriter struct {
 	r             http.Response
 	headerWritten missinggo.Event
 	bodyWriter    io.WriteCloser
-	closed        missinggo.SynchronizedEvent
+	bodyClosed    missinggo.SynchronizedEvent
 }
 
 var _ interface {
 	http.ResponseWriter
+	// We're able to emulate this easily enough.
 	http.CloseNotifier
 } = &responseWriter{}
 
+// Use Request.Context.Done instead.
 func (me *responseWriter) CloseNotify() <-chan bool {
 	ret := make(chan bool, 1)
 	go func() {
-		<-me.closed.C()
+		<-me.bodyClosed.C()
 		ret <- true
 	}()
 	return ret
@@ -66,10 +69,20 @@ func (me *responseWriter) runHandler(h http.Handler, req *http.Request) {
 	me.r.Body = struct {
 		io.Reader
 		io.Closer
-	}{pr, eventCloser{pr, &me.closed}}
+	}{pr, eventCloser{pr, &me.bodyClosed}}
+	// Shouldn't be writing to the response after the handler returns.
 	defer me.bodyWriter.Close()
+	// Send a 200 if nothing was written yet.
 	defer me.WriteHeader(200)
-	h.ServeHTTP(me, req)
+	// Wrap the context in the given Request with one that closes when either
+	// the handler returns, or the response body is closed.
+	ctx, cancel := context.WithCancel(req.Context())
+	defer cancel()
+	go func() {
+		<-me.bodyClosed.C()
+		cancel()
+	}()
+	h.ServeHTTP(me, req.WithContext(ctx))
 }
 
 type eventCloser struct {

vendor/github.com/anacrolix/missinggo/perf/mutex.go

@@ -14,7 +14,7 @@ type TimedLocker struct {
 func (me *TimedLocker) Lock() {
 	tr := NewTimer()
 	me.L.Lock()
-	tr.Stop(me.Desc)
+	tr.Mark(me.Desc)
 }
 
 func (me *TimedLocker) Unlock() {
@@ -30,7 +30,7 @@ type TimedRWLocker struct {
 func (me *TimedRWLocker) Lock() {
 	tr := NewTimer()
 	me.RWL.Lock()
-	tr.Stop(me.WriteDesc)
+	tr.Mark(me.WriteDesc)
 }
 
 func (me *TimedRWLocker) Unlock() {
@@ -40,7 +40,7 @@ func (me *TimedRWLocker) Unlock() {
 func (me *TimedRWLocker) RLock() {
 	tr := NewTimer()
 	me.RWL.RLock()
-	tr.Stop(me.ReadDesc)
+	tr.Mark(me.ReadDesc)
 }
 
 func (me *TimedRWLocker) RUnlock() {

vendor/github.com/anacrolix/missinggo/perf/perf.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"expvar"
 	"fmt"
+	"log"
 	"math"
 	"sync"
 	"time"
@@ -18,10 +19,31 @@ var (
 
 type Timer struct {
 	started time.Time
+	log     bool
+	name    string
 }
 
-func NewTimer() Timer {
-	return Timer{time.Now()}
+func NewTimer(opts ...timerOpt) (t Timer) {
+	t.started = time.Now()
+	for _, o := range opts {
+		o(&t)
+	}
+	if t.log && t.name != "" {
+		log.Printf("starting timer %q", t.name)
+	}
+	return
+}
+
+type timerOpt func(*Timer)
+
+func Log(t *Timer) {
+	t.log = true
+}
+
+func Name(name string) func(*Timer) {
+	return func(t *Timer) {
+		t.name = name
+	}
 }
 
 // The exponent is the upper bound of the duration in seconds.
@@ -79,9 +101,11 @@ func (me *buckets) String() string {
 
 var _ expvar.Var = &buckets{}
 
-func (t *Timer) Stop(desc string) time.Duration {
+func (t *Timer) Mark(events ...string) time.Duration {
 	d := time.Since(t.started)
-	t.addDuration(desc, d)
+	for _, e := range events {
+		t.addDuration(e, d)
+	}
 	return d
 }
 
@@ -100,4 +124,11 @@ func (t *Timer) addDuration(desc string, d time.Duration) {
 	}
 	m := _m.(*buckets)
 	m.Add(d)
+	if t.log {
+		if t.name != "" {
+			log.Printf("timer %q got event %q after %s", t.name, desc, d)
+		} else {
+			log.Printf("marking event %q after %s", desc, d)
+		}
+	}
 }

vendor/github.com/anacrolix/missinggo/perf/scope.go

@@ -0,0 +1,18 @@
+package perf
+
+import (
+	"runtime"
+)
+
+func ScopeTimer() func() {
+	t := NewTimer(Name(getCallerName()), Log)
+	return func() { t.Mark("returned") }
+}
+
+func getCallerName() string {
+	var pc [1]uintptr
+	runtime.Callers(3, pc[:])
+	fs := runtime.CallersFrames(pc[:])
+	f, _ := fs.Next()
+	return f.Func.Name()
+}

vendor/github.com/anacrolix/torrent/client.go

@@ -46,6 +46,7 @@ type Client struct {
 	halfOpenLimit  int
 	peerID         [20]byte
 	defaultStorage *storage.Client
+	onClose        []func()
 	tcpListener    net.Listener
 	utpSock        *utp.Socket
 	dHT            *dht.Server
@@ -238,6 +239,12 @@ func NewClient(cfg *Config) (cl *Client, err error) {
 		dopplegangerAddrs: make(map[string]struct{}),
 		torrents:          make(map[metainfo.Hash]*Torrent),
 	}
+	defer func() {
+		if err == nil {
+			return
+		}
+		cl.Close()
+	}()
 	if cfg.UploadRateLimiter == nil {
 		cl.uploadLimit = rate.NewLimiter(rate.Inf, 0)
 	} else {
@@ -253,6 +260,11 @@ func NewClient(cfg *Config) (cl *Client, err error) {
 	storageImpl := cfg.DefaultStorage
 	if storageImpl == nil {
 		storageImpl = storage.NewFile(cfg.DataDir)
+		cl.onClose = append(cl.onClose, func() {
+			if err := storageImpl.Close(); err != nil {
+				log.Printf("error closing default storage: %s", err)
+			}
+		})
 	}
 	cl.defaultStorage = storage.NewClient(storageImpl)
 	if cfg.IPBlocklist != nil {
@@ -337,6 +349,9 @@ func (cl *Client) Close() {
 	for _, t := range cl.torrents {
 		t.close()
 	}
+	for _, f := range cl.onClose {
+		f()
+	}
 	cl.event.Broadcast()
 }
 
@@ -899,7 +914,7 @@ func (cl *Client) runInitiatedHandshookConn(c *connection, t *Torrent) {
 		cl.dopplegangerAddrs[addr] = struct{}{}
 		return
 	}
-	cl.runHandshookConn(c, t)
+	cl.runHandshookConn(c, t, true)
 }
 
 func (cl *Client) runReceivedConn(c *connection) {
@@ -926,14 +941,14 @@ func (cl *Client) runReceivedConn(c *connection) {
 		// doppleganger.
 		return
 	}
-	cl.runHandshookConn(c, t)
+	cl.runHandshookConn(c, t, false)
 }
 
-func (cl *Client) runHandshookConn(c *connection, t *Torrent) {
+func (cl *Client) runHandshookConn(c *connection, t *Torrent, outgoing bool) {
 	c.conn.SetWriteDeadline(time.Time{})
 	c.r = deadlineReader{c.conn, c.r}
 	completedHandshakeConnectionFlags.Add(c.connectionFlags(), 1)
-	if !t.addConnection(c) {
+	if !t.addConnection(c, outgoing) {
 		return
 	}
 	defer t.dropConnection(c)

vendor/github.com/anacrolix/torrent/file.go

@@ -54,6 +54,8 @@ type FilePieceState struct {
 
 // Returns the state of pieces in this file.
 func (f *File) State() (ret []FilePieceState) {
+	f.t.cl.mu.RLock()
+	defer f.t.cl.mu.RUnlock()
 	pieceSize := int64(f.t.usualPieceSize())
 	off := f.offset % pieceSize
 	remaining := f.length
@@ -65,9 +67,7 @@ func (f *File) State() (ret []FilePieceState) {
 		if len1 > remaining {
 			len1 = remaining
 		}
-		f.t.cl.mu.RLock()
 		ps := f.t.pieceState(i)
-		f.t.cl.mu.RUnlock()
 		ret = append(ret, FilePieceState{len1, ps})
 		off = 0
 		remaining -= len1

vendor/github.com/anacrolix/torrent/reader.go

@@ -44,6 +44,11 @@ func (r *Reader) SetResponsive() {
 	r.responsive = true
 }
 
+// Disable responsive mode.
+func (r *Reader) SetNonResponsive() {
+	r.responsive = false
+}
+
 // Configure the number of bytes ahead of a read that should also be
 // prioritized in preparation for further reads.
 func (r *Reader) SetReadahead(readahead int64) {
@@ -55,6 +60,11 @@ func (r *Reader) SetReadahead(readahead int64) {
 	r.posChanged()
 }
 
+// Return reader's current position.
+func (r *Reader) CurrentPos() int64 {
+	return r.pos
+}
+
 func (r *Reader) readable(off int64) (ret bool) {
 	if r.t.closed.IsSet() {
 		return true

vendor/github.com/anacrolix/torrent/storage/boltPieceCompletion.go

@@ -2,6 +2,7 @@ package storage
 
 import (
 	"encoding/binary"
+	"os"
 	"path/filepath"
 	"time"
 
@@ -18,7 +19,8 @@ type boltPieceCompletion struct {
 	db *bolt.DB
 }
 
-func newBoltPieceCompletion(dir string) (ret *boltPieceCompletion, err error) {
+func NewBoltPieceCompletion(dir string) (ret PieceCompletion, err error) {
+	os.MkdirAll(dir, 0770)
 	p := filepath.Join(dir, ".torrent.bolt.db")
 	db, err := bolt.Open(p, 0660, &bolt.Options{
 		Timeout: time.Second,

vendor/github.com/anacrolix/torrent/storage/completion.go

@@ -7,17 +7,17 @@ import (
 )
 
 // Implementations track the completion of pieces. It must be concurrent-safe.
-type pieceCompletion interface {
+type PieceCompletion interface {
 	Get(metainfo.PieceKey) (bool, error)
 	Set(metainfo.PieceKey, bool) error
 	Close() error
 }
 
-func pieceCompletionForDir(dir string) (ret pieceCompletion) {
-	ret, err := newBoltPieceCompletion(dir)
+func pieceCompletionForDir(dir string) (ret PieceCompletion) {
+	ret, err := NewBoltPieceCompletion(dir)
 	if err != nil {
 		log.Printf("couldn't open piece completion db in %q: %s", dir, err)
-		ret = new(mapPieceCompletion)
+		ret = NewMapPieceCompletion()
 	}
 	return
 }

vendor/github.com/anacrolix/torrent/storage/completion_piece_map.go

@@ -11,7 +11,11 @@ type mapPieceCompletion struct {
 	m  map[metainfo.PieceKey]struct{}
 }
 
-func (mapPieceCompletion) Close() error { return nil }
+func NewMapPieceCompletion() PieceCompletion {
+	return &mapPieceCompletion{m: make(map[metainfo.PieceKey]struct{})}
+}
+
+func (*mapPieceCompletion) Close() error { return nil }
 
 func (me *mapPieceCompletion) Get(pk metainfo.PieceKey) (bool, error) {
 	me.mu.Lock()

vendor/github.com/anacrolix/torrent/storage/file.go

@@ -15,7 +15,7 @@ import (
 type fileClientImpl struct {
 	baseDir   string
 	pathMaker func(baseDir string, info *metainfo.Info, infoHash metainfo.Hash) string
-	pc        pieceCompletion
+	pc        PieceCompletion
 }
 
 // The Default path maker just returns the current path
@@ -29,7 +29,11 @@ func infoHashPathMaker(baseDir string, info *metainfo.Info, infoHash metainfo.Ha
 
 // All Torrent data stored in this baseDir
 func NewFile(baseDir string) ClientImpl {
-	return NewFileWithCustomPathMaker(baseDir, nil)
+	return NewFileWithCompletion(baseDir, pieceCompletionForDir(baseDir))
+}
+
+func NewFileWithCompletion(baseDir string, completion PieceCompletion) ClientImpl {
+	return newFileWithCustomPathMakerAndCompletion(baseDir, nil, completion)
 }
 
 // All Torrent data stored in subdirectorys by infohash
@@ -39,13 +43,17 @@ func NewFileByInfoHash(baseDir string) ClientImpl {
 
 // Allows passing a function to determine the path for storing torrent data
 func NewFileWithCustomPathMaker(baseDir string, pathMaker func(baseDir string, info *metainfo.Info, infoHash metainfo.Hash) string) ClientImpl {
+	return newFileWithCustomPathMakerAndCompletion(baseDir, pathMaker, pieceCompletionForDir(baseDir))
+}
+
+func newFileWithCustomPathMakerAndCompletion(baseDir string, pathMaker func(baseDir string, info *metainfo.Info, infoHash metainfo.Hash) string, completion PieceCompletion) ClientImpl {
 	if pathMaker == nil {
 		pathMaker = defaultPathMaker
 	}
 	return &fileClientImpl{
 		baseDir:   baseDir,
 		pathMaker: pathMaker,
-		pc:        pieceCompletionForDir(baseDir),
+		pc:        completion,
 	}
 }
 
@@ -72,7 +80,7 @@ type fileTorrentImpl struct {
 	dir        string
 	info       *metainfo.Info
 	infoHash   metainfo.Hash
-	completion pieceCompletion
+	completion PieceCompletion
 }
 
 func (fts *fileTorrentImpl) Piece(p metainfo.Piece) PieceImpl {

vendor/github.com/anacrolix/torrent/storage/mmap.go

@@ -15,13 +15,17 @@ import (
 
 type mmapStorage struct {
 	baseDir string
-	pc      pieceCompletion
+	pc      PieceCompletion
 }
 
 func NewMMap(baseDir string) ClientImpl {
+	return NewMMapWithCompletion(baseDir, pieceCompletionForDir(baseDir))
+}
+
+func NewMMapWithCompletion(baseDir string, completion PieceCompletion) ClientImpl {
 	return &mmapStorage{
 		baseDir: baseDir,
-		pc:      pieceCompletionForDir(baseDir),
+		pc:      completion,
 	}
 }
 
@@ -40,7 +44,7 @@ func (s *mmapStorage) Close() error {
 
 type mmapTorrentStorage struct {
 	span mmap_span.MMapSpan
-	pc   pieceCompletion
+	pc   PieceCompletion
 }
 
 func (ts *mmapTorrentStorage) Piece(p metainfo.Piece) PieceImpl {
@@ -58,7 +62,7 @@ func (ts *mmapTorrentStorage) Close() error {
 }
 
 type mmapStoragePiece struct {
-	pc pieceCompletion
+	pc PieceCompletion
 	p  metainfo.Piece
 	ih metainfo.Hash
 	io.ReaderAt

vendor/github.com/anacrolix/torrent/storage/wrappers.go

@@ -11,7 +11,7 @@ import (
 )
 
 type Client struct {
-	ClientImpl
+	ci ClientImpl
 }
 
 func NewClient(cl ClientImpl) *Client {
@@ -19,7 +19,7 @@ func NewClient(cl ClientImpl) *Client {
 }
 
 func (cl Client) OpenTorrent(info *metainfo.Info, infoHash metainfo.Hash) (*Torrent, error) {
-	t, err := cl.ClientImpl.OpenTorrent(info, infoHash)
+	t, err := cl.ci.OpenTorrent(info, infoHash)
 	return &Torrent{t}, err
 }
 

vendor/github.com/anacrolix/torrent/t.go

@@ -61,6 +61,14 @@ func (t *Torrent) NumPieces() int {
 	return t.numPieces()
 }
 
+// Get missing bytes count for specific piece.
+func (t *Torrent) PieceBytesMissing(piece int) int64 {
+	t.cl.mu.Lock()
+	defer t.cl.mu.Unlock()
+
+	return int64(t.pieces[piece].bytesLeft())
+}
+
 // Drop the torrent from the client, and close it. It's always safe to do
 // this. No data corruption can, or should occur to either the torrent's data,
 // or connected peers.

vendor/github.com/anacrolix/torrent/torrent.go

@@ -576,7 +576,7 @@ func (t *Torrent) writeChunk(piece int, begin int64, data []byte) (err error) {
 		err = io.ErrShortWrite
 	}
 	if err == nil {
-		tr.Stop("write chunk")
+		tr.Mark("write chunk")
 	}
 	return
 }
@@ -1296,7 +1296,7 @@ func (t *Torrent) numTotalPeers() int {
 }
 
 // Returns true if the connection is added.
-func (t *Torrent) addConnection(c *connection) bool {
+func (t *Torrent) addConnection(c *connection, outgoing bool) bool {
 	if t.cl.closed.IsSet() {
 		return false
 	}
@@ -1307,7 +1307,19 @@ func (t *Torrent) addConnection(c *connection) bool {
 		if c.PeerID == c0.PeerID {
 			// Already connected to a client with that ID.
 			duplicateClientConns.Add(1)
-			return false
+			lower := string(t.cl.peerID[:]) < string(c.PeerID[:])
+			// Retain the connection from initiated from lower peer ID to
+			// higher.
+			if outgoing == lower {
+				// Close the other one.
+				c0.Close()
+				// Is it safe to delete from the map while we're iterating
+				// over it?
+				t.deleteConnection(c0)
+			} else {
+				// Abandon this one.
+				return false
+			}
 		}
 	}
 	if len(t.conns) >= t.maxEstablishedConns {

vendor/github.com/anacrolix/utp/conn.go

@@ -98,8 +98,6 @@ func (c *Conn) wndSize() uint32 {
 }
 
 func (c *Conn) makePacket(_type st, connID, seqNr uint16, payload []byte) (p []byte) {
-	// Always selectively ack the first 64 packets. Don't bother with rest for
-	// now.
 	var selAck selectiveAckBitmask
 	for i := 1; i < len(c.inbound); i++ {
 		if c.inbound[i].seen {
@@ -115,11 +113,17 @@ func (c *Conn) makePacket(_type st, connID, seqNr uint16, payload []byte) (p []b
 		WndSize:       c.wndSize(),
 		Timestamp:     c.timestamp(),
 		TimestampDiff: c.lastTimeDiff,
-		// Currently always send an 8 byte selective ack.
-		Extensions: []extensionField{{
+	}
+	if len(selAck.Bytes) != 0 {
+		// The spec requires the number of bytes for a selective ACK to be at
+		// least 4, and a multiple of 4.
+		if len(selAck.Bytes)%4 != 0 {
+			panic(len(selAck.Bytes))
+		}
+		h.Extensions = append(h.Extensions, extensionField{
 			Type:  extensionTypeSelectiveAck,
 			Bytes: selAck.Bytes,
-		}},
+		})
 	}
 	p = sendBufferPool.Get().([]byte)[:0:minMTU]
 	n := h.Marshal(p)

vendor/github.com/anacrolix/utp/socket.go

@@ -359,16 +359,10 @@ func (s *Socket) network() string {
 	return s.pc.LocalAddr().Network()
 }
 
-// A zero timeout is no timeout. This will fallback onto the write ack
-// timeout.
-func (s *Socket) DialTimeout(addr string, timeout time.Duration) (nc net.Conn, err error) {
-	netAddr, err := s.resolveAddr(addr)
-	if err != nil {
-		return
-	}
-
+func (s *Socket) startOutboundConn(addr net.Addr) (c *Conn, err error) {
 	mu.Lock()
-	c := s.newConn(netAddr)
+	defer mu.Unlock()
+	c = s.newConn(addr)
 	c.recv_id = s.newConnID(resolvedAddrStr(c.RemoteAddr().String()))
 	c.send_id = c.recv_id + 1
 	if logLevel >= 1 {
@@ -383,12 +377,25 @@ func (s *Socket) DialTimeout(addr string, timeout time.Duration) (nc net.Conn, e
 		log.Printf("that's %d connections", len(s.conns))
 	}
 	if err != nil {
-		mu.Unlock()
 		return
 	}
 	c.seq_nr = 1
 	c.writeSyn()
-	mu.Unlock()
+	return
+}
+
+// A zero timeout is no timeout. This will fallback onto the write ack
+// timeout.
+func (s *Socket) DialTimeout(addr string, timeout time.Duration) (nc net.Conn, err error) {
+	netAddr, err := s.resolveAddr(addr)
+	if err != nil {
+		return
+	}
+
+	c, err := s.startOutboundConn(netAddr)
+	if err != nil {
+		return
+	}
 
 	connErr := make(chan error, 1)
 	go func() {