Introduce socket/torrent limits, work in progress

client.go

@@ -17,6 +17,7 @@ package torrent
 
 import (
 	"bufio"
+	"container/heap"
 	"crypto/rand"
 	"crypto/sha1"
 	"errors"
@@ -55,11 +56,15 @@ var (
 	duplicateConnsAvoided       = expvar.NewInt("duplicateConnsAvoided")
 )
 
-// Justification for set bits follows.
-//
-// Extension protocol: http://www.bittorrent.org/beps/bep_0010.html
-// DHT: http://www.bittorrent.org/beps/bep_0005.html
-const extensionBytes = "\x00\x00\x00\x00\x00\x10\x00\x01"
+const (
+	// Justification for set bits follows.
+	//
+	// Extension protocol: http://www.bittorrent.org/beps/bep_0010.html
+	// DHT: http://www.bittorrent.org/beps/bep_0005.html
+	extensionBytes = "\x00\x00\x00\x00\x00\x10\x00\x01"
+
+	socketsPerTorrent = 40
+)
 
 // Currently doesn't really queue, but should in the future.
 func (cl *Client) queuePieceCheck(t *torrent, pieceIndex pp.Integer) {
@@ -969,6 +974,10 @@ func (me *Client) addConnection(t *torrent, c *connection) bool {
 		}
 	}
 	t.Conns = append(t.Conns, c)
+	if len(t.Conns) > socketsPerTorrent {
+		wcs := t.worstConnsHeap()
+		heap.Pop(wcs).(*connection).Close()
+	}
 	return true
 }
 
@@ -983,6 +992,9 @@ func (me *Client) openNewConns() {
 			if me.halfOpen >= me.halfOpenLimit {
 				return
 			}
+			if me.halfOpen+me.handshaking+len(t.Conns) >= socketsPerTorrent {
+				break
+			}
 			var (
 				k peersKey
 				p Peer

torrent.go

@@ -1,11 +1,13 @@
 package torrent
 
 import (
+	"container/heap"
 	"container/list"
 	"fmt"
 	"io"
 	"log"
 	"net"
+	"sort"
 	"sync"
 
 	"bitbucket.org/anacrolix/go.torrent/util"
@@ -67,6 +69,14 @@ type torrent struct {
 	metadataHave []bool
 }
 
+func (t *torrent) worstConnsHeap() (wcs *worstConnsHeap) {
+	wcs = new(worstConnsHeap)
+	*wcs = make([]*connection, 0, len(t.Conns))
+	*wcs = append(*wcs, t.Conns...)
+	heap.Init(wcs)
+	return
+}
+
 func (t *torrent) CeaseNetworking() {
 	t.stateMu.Lock()
 	defer t.stateMu.Unlock()
@@ -283,6 +293,9 @@ func (t *torrent) WriteStatus(w io.Writer) {
 	fmt.Fprintln(w)
 	fmt.Fprintf(w, "Pending peers: %d\n", len(t.Peers))
 	fmt.Fprintf(w, "Active peers: %d\n", len(t.Conns))
+	wcs := new(worstConnsHeap)
+	*wcs = t.Conns
+	sort.Sort(wcs)
 	for _, c := range t.Conns {
 		c.WriteStatus(w)
 	}