status-im/torrent
2
0
Fork 0
mirror of synced 2025-02-24 06:38:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #79 from shoenig/normalize

Browse Source 
refactor struct identifiers to follow conventional go names
This commit is contained in:
Matt Joiner 2016-04-19 17:10:42 +10:00
commit eac0d6f8da
37 changed files with 688 additions and 690 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
bencode/api.go
View File

@ -18,8 +18,8 @@ type MarshalTypeError struct {
Type reflect.Type Type reflect.Type
} }
func (this *MarshalTypeError) Error() string { func (e *MarshalTypeError) Error() string {
return "bencode: unsupported type: " + this.Type.String() return "bencode: unsupported type: " + e.Type.String()
} }
// Unmarshal argument must be a non-nil value of some pointer type. // Unmarshal argument must be a non-nil value of some pointer type.

6
bencode/tags.go
View File

@ -13,12 +13,12 @@ func parse_tag(tag string) (string, tag_options) {
return tag, tag_options("") return tag, tag_options("")
} }
func (this tag_options) contains(option_name string) bool { func (opts tag_options) contains(option_name string) bool {
if len(this) == 0 { if len(opts) == 0 {
return false return false
} }
s := string(this) s := string(opts)
for s != "" { for s != "" {
var next string var next string
i := strings.Index(s, ",") i := strings.Index(s, ",")

368
client.go
View File

@ -86,27 +86,27 @@ type Client struct {
torrents map[metainfo.Hash]*Torrent torrents map[metainfo.Hash]*Torrent
} }
func (me *Client) IPBlockList() iplist.Ranger { func (cl *Client) IPBlockList() iplist.Ranger {
me.mu.Lock() cl.mu.Lock()
defer me.mu.Unlock() defer cl.mu.Unlock()
return me.ipBlockList return cl.ipBlockList
} }
func (me *Client) SetIPBlockList(list iplist.Ranger) { func (cl *Client) SetIPBlockList(list iplist.Ranger) {
me.mu.Lock() cl.mu.Lock()
defer me.mu.Unlock() defer cl.mu.Unlock()
me.ipBlockList = list cl.ipBlockList = list
if me.dHT != nil { if cl.dHT != nil {
me.dHT.SetIPBlockList(list) cl.dHT.SetIPBlockList(list)
} }
} }
func (me *Client) PeerID() string { func (cl *Client) PeerID() string {
return string(me.peerID[:]) return string(cl.peerID[:])
} }
func (me *Client) ListenAddr() (addr net.Addr) { func (cl *Client) ListenAddr() (addr net.Addr) {
for _, l := range me.listeners { for _, l := range cl.listeners {
addr = l.Addr() addr = l.Addr()
break break
} }
@ -117,16 +117,16 @@ type hashSorter struct {
Hashes []metainfo.Hash Hashes []metainfo.Hash
} }
func (me hashSorter) Len() int { func (hs hashSorter) Len() int {
return len(me.Hashes) return len(hs.Hashes)
} }
func (me hashSorter) Less(a, b int) bool { func (hs hashSorter) Less(a, b int) bool {
return (&big.Int{}).SetBytes(me.Hashes[a][:]).Cmp((&big.Int{}).SetBytes(me.Hashes[b][:])) < 0 return (&big.Int{}).SetBytes(hs.Hashes[a][:]).Cmp((&big.Int{}).SetBytes(hs.Hashes[b][:])) < 0
} }
func (me hashSorter) Swap(a, b int) { func (hs hashSorter) Swap(a, b int) {
me.Hashes[a], me.Hashes[b] = me.Hashes[b], me.Hashes[a] hs.Hashes[a], hs.Hashes[b] = hs.Hashes[b], hs.Hashes[a]
} }
func (cl *Client) sortedTorrents() (ret []*Torrent) { func (cl *Client) sortedTorrents() (ret []*Torrent) {
@ -293,20 +293,20 @@ func NewClient(cfg *Config) (cl *Client, err error) {
// Stops the client. All connections to peers are closed and all activity will // Stops the client. All connections to peers are closed and all activity will
// come to a halt. // come to a halt.
func (me *Client) Close() { func (cl *Client) Close() {
me.mu.Lock() cl.mu.Lock()
defer me.mu.Unlock() defer cl.mu.Unlock()
me.closed.Set() cl.closed.Set()
if me.dHT != nil { if cl.dHT != nil {
me.dHT.Close() cl.dHT.Close()
} }
for _, l := range me.listeners { for _, l := range cl.listeners {
l.Close() l.Close()
} }
for _, t := range me.torrents { for _, t := range cl.torrents {
t.close() t.close()
} }
me.event.Broadcast() cl.event.Broadcast()
} }
var ipv6BlockRange = iplist.Range{Description: "non-IPv4 address"} var ipv6BlockRange = iplist.Range{Description: "non-IPv4 address"}
@ -405,8 +405,8 @@ func (cl *Client) Torrent(ih metainfo.Hash) (t *Torrent, ok bool) {
return return
} }
func (me *Client) torrent(ih metainfo.Hash) *Torrent { func (cl *Client) torrent(ih metainfo.Hash) *Torrent {
return me.torrents[ih] return cl.torrents[ih]
} }
type dialResult struct { type dialResult struct {
@ -439,66 +439,66 @@ func reducedDialTimeout(max time.Duration, halfOpenLimit int, pendingPeers int)
} }
// Returns whether an address is known to connect to a client with our own ID. // Returns whether an address is known to connect to a client with our own ID.
func (me *Client) dopplegangerAddr(addr string) bool { func (cl *Client) dopplegangerAddr(addr string) bool {
_, ok := me.dopplegangerAddrs[addr] _, ok := cl.dopplegangerAddrs[addr]
return ok return ok
} }
// Start the process of connecting to the given peer for the given torrent if // Start the process of connecting to the given peer for the given torrent if
// appropriate. // appropriate.
func (me *Client) initiateConn(peer Peer, t *Torrent) { func (cl *Client) initiateConn(peer Peer, t *Torrent) {
if peer.Id == me.peerID { if peer.Id == cl.peerID {
return return
} }
addr := net.JoinHostPort(peer.IP.String(), fmt.Sprintf("%d", peer.Port)) addr := net.JoinHostPort(peer.IP.String(), fmt.Sprintf("%d", peer.Port))
if me.dopplegangerAddr(addr) || t.addrActive(addr) { if cl.dopplegangerAddr(addr) || t.addrActive(addr) {
duplicateConnsAvoided.Add(1) duplicateConnsAvoided.Add(1)
return return
} }
if r, ok := me.ipBlockRange(peer.IP); ok { if r, ok := cl.ipBlockRange(peer.IP); ok {
log.Printf("outbound connect to %s blocked by IP blocklist rule %s", peer.IP, r) log.Printf("outbound connect to %s blocked by IP blocklist rule %s", peer.IP, r)
return return
} }
t.halfOpen[addr] = struct{}{} t.halfOpen[addr] = struct{}{}
go me.outgoingConnection(t, addr, peer.Source) go cl.outgoingConnection(t, addr, peer.Source)
} }
func (me *Client) dialTimeout(t *Torrent) time.Duration { func (cl *Client) dialTimeout(t *Torrent) time.Duration {
me.mu.Lock() cl.mu.Lock()
pendingPeers := len(t.peers) pendingPeers := len(t.peers)
me.mu.Unlock() cl.mu.Unlock()
return reducedDialTimeout(nominalDialTimeout, me.halfOpenLimit, pendingPeers) return reducedDialTimeout(nominalDialTimeout, cl.halfOpenLimit, pendingPeers)
} }
func (me *Client) dialTCP(addr string, t *Torrent) (c net.Conn, err error) { func (cl *Client) dialTCP(addr string, t *Torrent) (c net.Conn, err error) {
c, err = net.DialTimeout("tcp", addr, me.dialTimeout(t)) c, err = net.DialTimeout("tcp", addr, cl.dialTimeout(t))
if err == nil { if err == nil {
c.(*net.TCPConn).SetLinger(0) c.(*net.TCPConn).SetLinger(0)
} }
return return
} }
func (me *Client) dialUTP(addr string, t *Torrent) (c net.Conn, err error) { func (cl *Client) dialUTP(addr string, t *Torrent) (c net.Conn, err error) {
return me.utpSock.DialTimeout(addr, me.dialTimeout(t)) return cl.utpSock.DialTimeout(addr, cl.dialTimeout(t))
} }
// Returns a connection over UTP or TCP, whichever is first to connect. // Returns a connection over UTP or TCP, whichever is first to connect.
func (me *Client) dialFirst(addr string, t *Torrent) (conn net.Conn, utp bool) { func (cl *Client) dialFirst(addr string, t *Torrent) (conn net.Conn, utp bool) {
// Initiate connections via TCP and UTP simultaneously. Use the first one // Initiate connections via TCP and UTP simultaneously. Use the first one
// that succeeds. // that succeeds.
left := 0 left := 0
if !me.config.DisableUTP { if !cl.config.DisableUTP {
left++ left++
} }
if !me.config.DisableTCP { if !cl.config.DisableTCP {
left++ left++
} }
resCh := make(chan dialResult, left) resCh := make(chan dialResult, left)
if !me.config.DisableUTP { if !cl.config.DisableUTP {
go doDial(me.dialUTP, resCh, true, addr, t) go doDial(cl.dialUTP, resCh, true, addr, t)
} }
if !me.config.DisableTCP { if !cl.config.DisableTCP {
go doDial(me.dialTCP, resCh, false, addr, t) go doDial(cl.dialTCP, resCh, false, addr, t)
} }
var res dialResult var res dialResult
// Wait for a successful connection. // Wait for a successful connection.
@ -521,17 +521,17 @@ func (me *Client) dialFirst(addr string, t *Torrent) (conn net.Conn, utp bool) {
return return
} }
func (me *Client) noLongerHalfOpen(t *Torrent, addr string) { func (cl *Client) noLongerHalfOpen(t *Torrent, addr string) {
if _, ok := t.halfOpen[addr]; !ok { if _, ok := t.halfOpen[addr]; !ok {
panic("invariant broken") panic("invariant broken")
} }
delete(t.halfOpen, addr) delete(t.halfOpen, addr)
me.openNewConns(t) cl.openNewConns(t)
} }
// Performs initiator handshakes and returns a connection. Returns nil // Performs initiator handshakes and returns a connection. Returns nil
// *connection if no connection for valid reasons. // *connection if no connection for valid reasons.
func (me *Client) handshakesConnection(nc net.Conn, t *Torrent, encrypted, utp bool) (c *connection, err error) { func (cl *Client) handshakesConnection(nc net.Conn, t *Torrent, encrypted, utp bool) (c *connection, err error) {
c = newConnection() c = newConnection()
c.conn = nc c.conn = nc
c.rw = nc c.rw = nc
@ -541,7 +541,7 @@ func (me *Client) handshakesConnection(nc net.Conn, t *Torrent, encrypted, utp b
if err != nil { if err != nil {
return return
} }
ok, err := me.initiateHandshakes(c, t) ok, err := cl.initiateHandshakes(c, t)
if !ok { if !ok {
c = nil c = nil
} }
@ -550,12 +550,12 @@ func (me *Client) handshakesConnection(nc net.Conn, t *Torrent, encrypted, utp b
// Returns nil connection and nil error if no connection could be established // Returns nil connection and nil error if no connection could be established
// for valid reasons. // for valid reasons.
func (me *Client) establishOutgoingConn(t *Torrent, addr string) (c *connection, err error) { func (cl *Client) establishOutgoingConn(t *Torrent, addr string) (c *connection, err error) {
nc, utp := me.dialFirst(addr, t) nc, utp := cl.dialFirst(addr, t)
if nc == nil { if nc == nil {
return return
} }
c, err = me.handshakesConnection(nc, t, !me.config.DisableEncryption, utp) c, err = cl.handshakesConnection(nc, t, !cl.config.DisableEncryption, utp)
if err != nil { if err != nil {
nc.Close() nc.Close()
return return
@ -563,22 +563,22 @@ func (me *Client) establishOutgoingConn(t *Torrent, addr string) (c *connection,
return return
} }
nc.Close() nc.Close()
if me.config.DisableEncryption { if cl.config.DisableEncryption {
// We already tried without encryption. // We already tried without encryption.
return return
} }
// Try again without encryption, using whichever protocol type worked last // Try again without encryption, using whichever protocol type worked last
// time. // time.
if utp { if utp {
nc, err = me.dialUTP(addr, t) nc, err = cl.dialUTP(addr, t)
} else { } else {
nc, err = me.dialTCP(addr, t) nc, err = cl.dialTCP(addr, t)
} }
if err != nil { if err != nil {
err = fmt.Errorf("error dialing for unencrypted connection: %s", err) err = fmt.Errorf("error dialing for unencrypted connection: %s", err)
return return
} }
c, err = me.handshakesConnection(nc, t, false, utp) c, err = cl.handshakesConnection(nc, t, false, utp)
if err != nil || c == nil { if err != nil || c == nil {
nc.Close() nc.Close()
} }
@ -587,15 +587,15 @@ func (me *Client) establishOutgoingConn(t *Torrent, addr string) (c *connection,
// Called to dial out and run a connection. The addr we're given is already // Called to dial out and run a connection. The addr we're given is already
// considered half-open. // considered half-open.
func (me *Client) outgoingConnection(t *Torrent, addr string, ps peerSource) { func (cl *Client) outgoingConnection(t *Torrent, addr string, ps peerSource) {
c, err := me.establishOutgoingConn(t, addr) c, err := cl.establishOutgoingConn(t, addr)
me.mu.Lock() cl.mu.Lock()
defer me.mu.Unlock() defer cl.mu.Unlock()
// Don't release lock between here and addConnection, unless it's for // Don't release lock between here and addConnection, unless it's for
// failure. // failure.
me.noLongerHalfOpen(t, addr) cl.noLongerHalfOpen(t, addr)
if err != nil { if err != nil {
if me.config.Debug { if cl.config.Debug {
log.Printf("error establishing outgoing connection: %s", err) log.Printf("error establishing outgoing connection: %s", err)
} }
return return
@ -605,9 +605,9 @@ func (me *Client) outgoingConnection(t *Torrent, addr string, ps peerSource) {
} }
defer c.Close() defer c.Close()
c.Discovery = ps c.Discovery = ps
err = me.runInitiatedHandshookConn(c, t) err = cl.runInitiatedHandshookConn(c, t)
if err != nil { if err != nil {
if me.config.Debug { if cl.config.Debug {
log.Printf("error in established outgoing connection: %s", err) log.Printf("error in established outgoing connection: %s", err)
} }
} }
@ -656,16 +656,16 @@ type (
peerID [20]byte peerID [20]byte
) )
func (me *peerExtensionBytes) SupportsExtended() bool { func (pex *peerExtensionBytes) SupportsExtended() bool {
return me[5]&0x10 != 0 return pex[5]&0x10 != 0
} }
func (me *peerExtensionBytes) SupportsDHT() bool { func (pex *peerExtensionBytes) SupportsDHT() bool {
return me[7]&0x01 != 0 return pex[7]&0x01 != 0
} }
func (me *peerExtensionBytes) SupportsFast() bool { func (pex *peerExtensionBytes) SupportsFast() bool {
return me[7]&0x04 != 0 return pex[7]&0x04 != 0
} }
type handshakeResult struct { type handshakeResult struct {
@ -748,13 +748,13 @@ type deadlineReader struct {
r io.Reader r io.Reader
} }
func (me deadlineReader) Read(b []byte) (n int, err error) { func (r deadlineReader) Read(b []byte) (n int, err error) {
// Keep-alives should be received every 2 mins. Give a bit of gracetime. // Keep-alives should be received every 2 mins. Give a bit of gracetime.
err = me.nc.SetReadDeadline(time.Now().Add(150 * time.Second)) err = r.nc.SetReadDeadline(time.Now().Add(150 * time.Second))
if err != nil { if err != nil {
err = fmt.Errorf("error setting read deadline: %s", err) err = fmt.Errorf("error setting read deadline: %s", err)
} }
n, err = me.r.Read(b) n, err = r.r.Read(b)
// Convert common errors into io.EOF. // Convert common errors into io.EOF.
// if err != nil { // if err != nil {
// if opError, ok := err.(*net.OpError); ok && opError.Op == "read" && opError.Err == syscall.ECONNRESET { // if opError, ok := err.(*net.OpError); ok && opError.Op == "read" && opError.Err == syscall.ECONNRESET {
@ -799,14 +799,14 @@ func (cl *Client) receiveSkeys() (ret [][]byte) {
return return
} }
func (me *Client) initiateHandshakes(c *connection, t *Torrent) (ok bool, err error) { func (cl *Client) initiateHandshakes(c *connection, t *Torrent) (ok bool, err error) {
if c.encrypted { if c.encrypted {
c.rw, err = mse.InitiateHandshake(c.rw, t.infoHash[:], nil) c.rw, err = mse.InitiateHandshake(c.rw, t.infoHash[:], nil)
if err != nil { if err != nil {
return return
} }
} }
ih, ok, err := me.connBTHandshake(c, &t.infoHash) ih, ok, err := cl.connBTHandshake(c, &t.infoHash)
if ih != t.infoHash { if ih != t.infoHash {
ok = false ok = false
} }
@ -908,8 +908,8 @@ func (cl *Client) runHandshookConn(c *connection, t *Torrent) (err error) {
return return
} }
func (me *Client) sendInitialMessages(conn *connection, torrent *Torrent) { func (cl *Client) sendInitialMessages(conn *connection, torrent *Torrent) {
if conn.PeerExtensionBytes.SupportsExtended() && me.extensionBytes.SupportsExtended() { if conn.PeerExtensionBytes.SupportsExtended() && cl.extensionBytes.SupportsExtended() {
conn.Post(pp.Message{ conn.Post(pp.Message{
Type: pp.Extended, Type: pp.Extended,
ExtendedID: pp.HandshakeExtendedID, ExtendedID: pp.HandshakeExtendedID,
@ -918,7 +918,7 @@ func (me *Client) sendInitialMessages(conn *connection, torrent *Torrent) {
"m": func() (ret map[string]int) { "m": func() (ret map[string]int) {
ret = make(map[string]int, 2) ret = make(map[string]int, 2)
ret["ut_metadata"] = metadataExtendedId ret["ut_metadata"] = metadataExtendedId
if !me.config.DisablePEX { if !cl.config.DisablePEX {
ret["ut_pex"] = pexExtendedId ret["ut_pex"] = pexExtendedId
} }
return return
@ -927,13 +927,13 @@ func (me *Client) sendInitialMessages(conn *connection, torrent *Torrent) {
// No upload queue is implemented yet. // No upload queue is implemented yet.
"reqq": 64, "reqq": 64,
} }
if !me.config.DisableEncryption { if !cl.config.DisableEncryption {
d["e"] = 1 d["e"] = 1
} }
if torrent.metadataSizeKnown() { if torrent.metadataSizeKnown() {
d["metadata_size"] = torrent.metadataSize() d["metadata_size"] = torrent.metadataSize()
} }
if p := me.incomingPeerPort(); p != 0 { if p := cl.incomingPeerPort(); p != 0 {
d["p"] = p d["p"] = p
} }
yourip, err := addrCompactIP(conn.remoteAddr()) yourip, err := addrCompactIP(conn.remoteAddr())
@ -953,20 +953,20 @@ func (me *Client) sendInitialMessages(conn *connection, torrent *Torrent) {
} }
if torrent.haveAnyPieces() { if torrent.haveAnyPieces() {
conn.Bitfield(torrent.bitfield()) conn.Bitfield(torrent.bitfield())
} else if me.extensionBytes.SupportsFast() && conn.PeerExtensionBytes.SupportsFast() { } else if cl.extensionBytes.SupportsFast() && conn.PeerExtensionBytes.SupportsFast() {
conn.Post(pp.Message{ conn.Post(pp.Message{
Type: pp.HaveNone, Type: pp.HaveNone,
}) })
} }
if conn.PeerExtensionBytes.SupportsDHT() && me.extensionBytes.SupportsDHT() && me.dHT != nil { if conn.PeerExtensionBytes.SupportsDHT() && cl.extensionBytes.SupportsDHT() && cl.dHT != nil {
conn.Post(pp.Message{ conn.Post(pp.Message{
Type: pp.Port, Type: pp.Port,
Port: uint16(missinggo.AddrPort(me.dHT.Addr())), Port: uint16(missinggo.AddrPort(cl.dHT.Addr())),
}) })
} }
} }
func (me *Client) peerUnchoked(torrent *Torrent, conn *connection) { func (cl *Client) peerUnchoked(torrent *Torrent, conn *connection) {
conn.updateRequests() conn.updateRequests()
} }
@ -1086,14 +1086,14 @@ func (cl *Client) gotMetadataExtensionMsg(payload []byte, t *Torrent, c *connect
return return
} }
func (me *Client) upload(t *Torrent, c *connection) { func (cl *Client) upload(t *Torrent, c *connection) {
if me.config.NoUpload { if cl.config.NoUpload {
return return
} }
if !c.PeerInterested { if !c.PeerInterested {
return return
} }
seeding := me.seeding(t) seeding := cl.seeding(t)
if !seeding && !t.connHasWantedPieces(c) { if !seeding && !t.connHasWantedPieces(c) {
return return
} }
@ -1101,7 +1101,7 @@ another:
for seeding || c.chunksSent < c.UsefulChunksReceived+6 { for seeding || c.chunksSent < c.UsefulChunksReceived+6 {
c.Unchoke() c.Unchoke()
for r := range c.PeerRequests { for r := range c.PeerRequests {
err := me.sendChunk(t, c, r) err := cl.sendChunk(t, c, r)
if err != nil { if err != nil {
if t.pieceComplete(int(r.Index)) && err == io.ErrUnexpectedEOF { if t.pieceComplete(int(r.Index)) && err == io.ErrUnexpectedEOF {
// We had the piece, but not anymore. // We had the piece, but not anymore.
@ -1123,7 +1123,7 @@ another:
c.Choke() c.Choke()
} }
func (me *Client) sendChunk(t *Torrent, c *connection, r request) error { func (cl *Client) sendChunk(t *Torrent, c *connection, r request) error {
// Count the chunk being sent, even if it isn't. // Count the chunk being sent, even if it isn't.
b := make([]byte, r.Length) b := make([]byte, r.Length)
p := t.info.Piece(int(r.Index)) p := t.info.Piece(int(r.Index))
@ -1148,17 +1148,17 @@ func (me *Client) sendChunk(t *Torrent, c *connection, r request) error {
// Processes incoming bittorrent messages. The client lock is held upon entry // Processes incoming bittorrent messages. The client lock is held upon entry
// and exit. // and exit.
func (me *Client) connectionLoop(t *Torrent, c *connection) error { func (cl *Client) connectionLoop(t *Torrent, c *connection) error {
decoder := pp.Decoder{ decoder := pp.Decoder{
R: bufio.NewReader(c.rw), R: bufio.NewReader(c.rw),
MaxLength: 256 * 1024, MaxLength: 256 * 1024,
} }
for { for {
me.mu.Unlock() cl.mu.Unlock()
var msg pp.Message var msg pp.Message
err := decoder.Decode(&msg) err := decoder.Decode(&msg)
me.mu.Lock() cl.mu.Lock()
if me.closed.IsSet() || c.closed.IsSet() || err == io.EOF { if cl.closed.IsSet() || c.closed.IsSet() || err == io.EOF {
return nil return nil
} }
if err != nil { if err != nil {
@ -1177,14 +1177,14 @@ func (me *Client) connectionLoop(t *Torrent, c *connection) error {
// We can then reset our interest. // We can then reset our interest.
c.updateRequests() c.updateRequests()
case pp.Reject: case pp.Reject:
me.connDeleteRequest(t, c, newRequest(msg.Index, msg.Begin, msg.Length)) cl.connDeleteRequest(t, c, newRequest(msg.Index, msg.Begin, msg.Length))
c.updateRequests() c.updateRequests()
case pp.Unchoke: case pp.Unchoke:
c.PeerChoked = false c.PeerChoked = false
me.peerUnchoked(t, c) cl.peerUnchoked(t, c)
case pp.Interested: case pp.Interested:
c.PeerInterested = true c.PeerInterested = true
me.upload(t, c) cl.upload(t, c)
case pp.NotInterested: case pp.NotInterested:
c.PeerInterested = false c.PeerInterested = false
c.Choke() c.Choke()
@ -1210,7 +1210,7 @@ func (me *Client) connectionLoop(t *Torrent, c *connection) error {
c.PeerRequests = make(map[request]struct{}, maxRequests) c.PeerRequests = make(map[request]struct{}, maxRequests)
} }
c.PeerRequests[newRequest(msg.Index, msg.Begin, msg.Length)] = struct{}{} c.PeerRequests[newRequest(msg.Index, msg.Begin, msg.Length)] = struct{}{}
me.upload(t, c) cl.upload(t, c)
case pp.Cancel: case pp.Cancel:
req := newRequest(msg.Index, msg.Begin, msg.Length) req := newRequest(msg.Index, msg.Begin, msg.Length)
if !c.PeerCancel(req) { if !c.PeerCancel(req) {
@ -1223,7 +1223,7 @@ func (me *Client) connectionLoop(t *Torrent, c *connection) error {
case pp.HaveNone: case pp.HaveNone:
err = c.peerSentHaveNone() err = c.peerSentHaveNone()
case pp.Piece: case pp.Piece:
me.downloadedChunk(t, c, &msg) cl.downloadedChunk(t, c, &msg)
case pp.Extended: case pp.Extended:
switch msg.ExtendedID { switch msg.ExtendedID {
case pp.HandshakeExtendedID: case pp.HandshakeExtendedID:
@ -1277,19 +1277,19 @@ func (me *Client) connectionLoop(t *Torrent, c *connection) error {
if !ok { if !ok {
log.Printf("bad metadata_size type: %T", metadata_sizeUntyped) log.Printf("bad metadata_size type: %T", metadata_sizeUntyped)
} else { } else {
t.setMetadataSize(metadata_size, me) t.setMetadataSize(metadata_size, cl)
} }
} }
if _, ok := c.PeerExtensionIDs["ut_metadata"]; ok { if _, ok := c.PeerExtensionIDs["ut_metadata"]; ok {
me.requestPendingMetadata(t, c) cl.requestPendingMetadata(t, c)
} }
case metadataExtendedId: case metadataExtendedId:
err = me.gotMetadataExtensionMsg(msg.ExtendedPayload, t, c) err = cl.gotMetadataExtensionMsg(msg.ExtendedPayload, t, c)
if err != nil { if err != nil {
err = fmt.Errorf("error handling metadata extension message: %s", err) err = fmt.Errorf("error handling metadata extension message: %s", err)
} }
case pexExtendedId: case pexExtendedId:
if me.config.DisablePEX { if cl.config.DisablePEX {
break break
} }
var pexMsg peerExchangeMessage var pexMsg peerExchangeMessage
@ -1299,8 +1299,8 @@ func (me *Client) connectionLoop(t *Torrent, c *connection) error {
break break
} }
go func() { go func() {
me.mu.Lock() cl.mu.Lock()
me.addPeers(t, func() (ret []Peer) { cl.addPeers(t, func() (ret []Peer) {
for i, cp := range pexMsg.Added { for i, cp := range pexMsg.Added {
p := Peer{ p := Peer{
IP: make([]byte, 4), IP: make([]byte, 4),
@ -1315,7 +1315,7 @@ func (me *Client) connectionLoop(t *Torrent, c *connection) error {
} }
return return
}()) }())
me.mu.Unlock() cl.mu.Unlock()
}() }()
default: default:
err = fmt.Errorf("unexpected extended message ID: %v", msg.ExtendedID) err = fmt.Errorf("unexpected extended message ID: %v", msg.ExtendedID)
@ -1329,7 +1329,7 @@ func (me *Client) connectionLoop(t *Torrent, c *connection) error {
} }
} }
case pp.Port: case pp.Port:
if me.dHT == nil { if cl.dHT == nil {
break break
} }
pingAddr, err := net.ResolveUDPAddr("", c.remoteAddr().String()) pingAddr, err := net.ResolveUDPAddr("", c.remoteAddr().String())
@ -1339,7 +1339,7 @@ func (me *Client) connectionLoop(t *Torrent, c *connection) error {
if msg.Port != 0 { if msg.Port != 0 {
pingAddr.Port = int(msg.Port) pingAddr.Port = int(msg.Port)
} }
me.dHT.Ping(pingAddr) cl.dHT.Ping(pingAddr)
default: default:
err = fmt.Errorf("received unknown message type: %#v", msg.Type) err = fmt.Errorf("received unknown message type: %#v", msg.Type)
} }
@ -1350,7 +1350,7 @@ func (me *Client) connectionLoop(t *Torrent, c *connection) error {
} }
// Returns true if connection is removed from torrent.Conns. // Returns true if connection is removed from torrent.Conns.
func (me *Client) deleteConnection(t *Torrent, c *connection) bool { func (cl *Client) deleteConnection(t *Torrent, c *connection) bool {
for i0, _c := range t.conns { for i0, _c := range t.conns {
if _c != c { if _c != c {
continue continue
@ -1365,17 +1365,17 @@ func (me *Client) deleteConnection(t *Torrent, c *connection) bool {
return false return false
} }
func (me *Client) dropConnection(t *Torrent, c *connection) { func (cl *Client) dropConnection(t *Torrent, c *connection) {
me.event.Broadcast() cl.event.Broadcast()
c.Close() c.Close()
if me.deleteConnection(t, c) { if cl.deleteConnection(t, c) {
me.openNewConns(t) cl.openNewConns(t)
} }
} }
// Returns true if the connection is added. // Returns true if the connection is added.
func (me *Client) addConnection(t *Torrent, c *connection) bool { func (cl *Client) addConnection(t *Torrent, c *connection) bool {
if me.closed.IsSet() { if cl.closed.IsSet() {
return false return false
} }
select { select {
@ -1383,7 +1383,7 @@ func (me *Client) addConnection(t *Torrent, c *connection) bool {
return false return false
default: default:
} }
if !me.wantConns(t) { if !cl.wantConns(t) {
return false return false
} }
for _, c0 := range t.conns { for _, c0 := range t.conns {
@ -1394,15 +1394,15 @@ func (me *Client) addConnection(t *Torrent, c *connection) bool {
} }
} }
if len(t.conns) >= socketsPerTorrent { if len(t.conns) >= socketsPerTorrent {
c := t.worstBadConn(me) c := t.worstBadConn(cl)
if c == nil { if c == nil {
return false return false
} }
if me.config.Debug && missinggo.CryHeard() { if cl.config.Debug && missinggo.CryHeard() {
log.Printf("%s: dropping connection to make room for new one:\n %s", t, c) log.Printf("%s: dropping connection to make room for new one:\n %s", t, c)
} }
c.Close() c.Close()
me.deleteConnection(t, c) cl.deleteConnection(t, c)
} }
if len(t.conns) >= socketsPerTorrent { if len(t.conns) >= socketsPerTorrent {
panic(len(t.conns)) panic(len(t.conns))
@ -1445,27 +1445,27 @@ func (cl *Client) usefulConn(t *Torrent, c *connection) bool {
return t.connHasWantedPieces(c) return t.connHasWantedPieces(c)
} }
func (me *Client) wantConns(t *Torrent) bool { func (cl *Client) wantConns(t *Torrent) bool {
if !me.seeding(t) && !t.needData() { if !cl.seeding(t) && !t.needData() {
return false return false
} }
if len(t.conns) < socketsPerTorrent { if len(t.conns) < socketsPerTorrent {
return true return true
} }
return t.worstBadConn(me) != nil return t.worstBadConn(cl) != nil
} }
func (me *Client) openNewConns(t *Torrent) { func (cl *Client) openNewConns(t *Torrent) {
select { select {
case <-t.ceasingNetworking: case <-t.ceasingNetworking:
return return
default: default:
} }
for len(t.peers) != 0 { for len(t.peers) != 0 {
if !me.wantConns(t) { if !cl.wantConns(t) {
return return
} }
if len(t.halfOpen) >= me.halfOpenLimit { if len(t.halfOpen) >= cl.halfOpenLimit {
return return
} }
var ( var (
@ -1476,27 +1476,27 @@ func (me *Client) openNewConns(t *Torrent) {
break break
} }
delete(t.peers, k) delete(t.peers, k)
me.initiateConn(p, t) cl.initiateConn(p, t)
} }
t.wantPeers.Broadcast() t.wantPeers.Broadcast()
} }
func (me *Client) addPeers(t *Torrent, peers []Peer) { func (cl *Client) addPeers(t *Torrent, peers []Peer) {
for _, p := range peers { for _, p := range peers {
if me.dopplegangerAddr(net.JoinHostPort( if cl.dopplegangerAddr(net.JoinHostPort(
p.IP.String(), p.IP.String(),
strconv.FormatInt(int64(p.Port), 10), strconv.FormatInt(int64(p.Port), 10),
)) { )) {
continue continue
} }
if _, ok := me.ipBlockRange(p.IP); ok { if _, ok := cl.ipBlockRange(p.IP); ok {
continue continue
} }
if p.Port == 0 { if p.Port == 0 {
// The spec says to scrub these yourselves. Fine. // The spec says to scrub these yourselves. Fine.
continue continue
} }
t.addPeer(p, me) t.addPeer(p, cl)
} }
} }
@ -1811,8 +1811,8 @@ func (cl *Client) AddTorrentSpec(spec *TorrentSpec) (t *Torrent, new bool, err e
return return
} }
func (me *Client) dropTorrent(infoHash metainfo.Hash) (err error) { func (cl *Client) dropTorrent(infoHash metainfo.Hash) (err error) {
t, ok := me.torrents[infoHash] t, ok := cl.torrents[infoHash]
if !ok { if !ok {
err = fmt.Errorf("no such torrent") err = fmt.Errorf("no such torrent")
return return
@ -1821,7 +1821,7 @@ func (me *Client) dropTorrent(infoHash metainfo.Hash) (err error) {
if err != nil { if err != nil {
panic(err) panic(err)
} }
delete(me.torrents, infoHash) delete(cl.torrents, infoHash)
return return
} }
@ -2047,26 +2047,26 @@ func (cl *Client) allTorrentsCompleted() bool {
// Returns true when all torrents are completely downloaded and false if the // Returns true when all torrents are completely downloaded and false if the
// client is stopped before that. // client is stopped before that.
func (me *Client) WaitAll() bool { func (cl *Client) WaitAll() bool {
me.mu.Lock() cl.mu.Lock()
defer me.mu.Unlock() defer cl.mu.Unlock()
for !me.allTorrentsCompleted() { for !cl.allTorrentsCompleted() {
if me.closed.IsSet() { if cl.closed.IsSet() {
return false return false
} }
me.event.Wait() cl.event.Wait()
} }
return true return true
} }
// Handle a received chunk from a peer. // Handle a received chunk from a peer.
func (me *Client) downloadedChunk(t *Torrent, c *connection, msg *pp.Message) { func (cl *Client) downloadedChunk(t *Torrent, c *connection, msg *pp.Message) {
chunksReceived.Add(1) chunksReceived.Add(1)
req := newRequest(msg.Index, msg.Begin, pp.Integer(len(msg.Piece))) req := newRequest(msg.Index, msg.Begin, pp.Integer(len(msg.Piece)))
// Request has been satisfied. // Request has been satisfied.
if me.connDeleteRequest(t, c, req) { if cl.connDeleteRequest(t, c, req) {
defer c.updateRequests() defer c.updateRequests()
} else { } else {
unexpectedChunksReceived.Add(1) unexpectedChunksReceived.Add(1)
@ -2076,7 +2076,7 @@ func (me *Client) downloadedChunk(t *Torrent, c *connection, msg *pp.Message) {
piece := &t.pieces[index] piece := &t.pieces[index]
// Do we actually want this chunk? // Do we actually want this chunk?
if !t.wantChunk(req) { if !t.wantPiece(req) {
unwantedChunksReceived.Add(1) unwantedChunksReceived.Add(1)
c.UnwantedChunksReceived++ c.UnwantedChunksReceived++
return return
@ -2085,7 +2085,7 @@ func (me *Client) downloadedChunk(t *Torrent, c *connection, msg *pp.Message) {
c.UsefulChunksReceived++ c.UsefulChunksReceived++
c.lastUsefulChunkReceived = time.Now() c.lastUsefulChunkReceived = time.Now()
me.upload(t, c) cl.upload(t, c)
// Need to record that it hasn't been written yet, before we attempt to do // Need to record that it hasn't been written yet, before we attempt to do
// anything with it. // anything with it.
@ -2095,15 +2095,15 @@ func (me *Client) downloadedChunk(t *Torrent, c *connection, msg *pp.Message) {
// Cancel pending requests for this chunk. // Cancel pending requests for this chunk.
for _, c := range t.conns { for _, c := range t.conns {
if me.connCancel(t, c, req) { if cl.connCancel(t, c, req) {
c.updateRequests() c.updateRequests()
} }
} }
me.mu.Unlock() cl.mu.Unlock()
// Write the chunk out. // Write the chunk out.
err := t.writeChunk(int(msg.Index), int64(msg.Begin), msg.Piece) err := t.writeChunk(int(msg.Index), int64(msg.Begin), msg.Piece)
me.mu.Lock() cl.mu.Lock()
piece.decrementPendingWrites() piece.decrementPendingWrites()
@ -2117,7 +2117,7 @@ func (me *Client) downloadedChunk(t *Torrent, c *connection, msg *pp.Message) {
// It's important that the piece is potentially queued before we check if // It's important that the piece is potentially queued before we check if
// the piece is still wanted, because if it is queued, it won't be wanted. // the piece is still wanted, because if it is queued, it won't be wanted.
if t.pieceAllDirty(index) { if t.pieceAllDirty(index) {
me.queuePieceCheck(t, int(req.Index)) cl.queuePieceCheck(t, int(req.Index))
} }
if c.peerTouchedPieces == nil { if c.peerTouchedPieces == nil {
@ -2125,14 +2125,14 @@ func (me *Client) downloadedChunk(t *Torrent, c *connection, msg *pp.Message) {
} }
c.peerTouchedPieces[index] = struct{}{} c.peerTouchedPieces[index] = struct{}{}
me.event.Broadcast() cl.event.Broadcast()
t.publishPieceChange(int(req.Index)) t.publishPieceChange(int(req.Index))
return return
} }
// Return the connections that touched a piece, and clear the entry while // Return the connections that touched a piece, and clear the entry while
// doing it. // doing it.
func (me *Client) reapPieceTouches(t *Torrent, piece int) (ret []*connection) { func (cl *Client) reapPieceTouches(t *Torrent, piece int) (ret []*connection) {
for _, c := range t.conns { for _, c := range t.conns {
if _, ok := c.peerTouchedPieces[piece]; ok { if _, ok := c.peerTouchedPieces[piece]; ok {
ret = append(ret, c) ret = append(ret, c)
@ -2142,7 +2142,7 @@ func (me *Client) reapPieceTouches(t *Torrent, piece int) (ret []*connection) {
return return
} }
func (me *Client) pieceHashed(t *Torrent, piece int, correct bool) { func (cl *Client) pieceHashed(t *Torrent, piece int, correct bool) {
p := &t.pieces[piece] p := &t.pieces[piece]
if p.EverHashed { if p.EverHashed {
// Don't score the first time a piece is hashed, it could be an // Don't score the first time a piece is hashed, it could be an
@ -2155,7 +2155,7 @@ func (me *Client) pieceHashed(t *Torrent, piece int, correct bool) {
} }
} }
p.EverHashed = true p.EverHashed = true
touchers := me.reapPieceTouches(t, piece) touchers := cl.reapPieceTouches(t, piece)
if correct { if correct {
err := p.Storage().MarkComplete() err := p.Storage().MarkComplete()
if err != nil { if err != nil {
@ -2165,13 +2165,13 @@ func (me *Client) pieceHashed(t *Torrent, piece int, correct bool) {
} else if len(touchers) != 0 { } else if len(touchers) != 0 {
log.Printf("dropping %d conns that touched piece", len(touchers)) log.Printf("dropping %d conns that touched piece", len(touchers))
for _, c := range touchers { for _, c := range touchers {
me.dropConnection(t, c) cl.dropConnection(t, c)
} }
} }
me.pieceChanged(t, piece) cl.pieceChanged(t, piece)
} }
func (me *Client) onCompletedPiece(t *Torrent, piece int) { func (cl *Client) onCompletedPiece(t *Torrent, piece int) {
t.pendingPieces.Remove(piece) t.pendingPieces.Remove(piece)
t.pendAllChunkSpecs(piece) t.pendAllChunkSpecs(piece)
for _, conn := range t.conns { for _, conn := range t.conns {
@ -2183,18 +2183,18 @@ func (me *Client) onCompletedPiece(t *Torrent, piece int) {
} }
// Could check here if peer doesn't have piece, but due to caching // Could check here if peer doesn't have piece, but due to caching
// some peers may have said they have a piece but they don't. // some peers may have said they have a piece but they don't.
me.upload(t, conn) cl.upload(t, conn)
} }
} }
func (me *Client) onFailedPiece(t *Torrent, piece int) { func (cl *Client) onFailedPiece(t *Torrent, piece int) {
if t.pieceAllDirty(piece) { if t.pieceAllDirty(piece) {
t.pendAllChunkSpecs(piece) t.pendAllChunkSpecs(piece)
} }
if !t.wantPiece(piece) { if !t.wantPieceIndex(piece) {
return return
} }
me.openNewConns(t) cl.openNewConns(t)
for _, conn := range t.conns { for _, conn := range t.conns {
if conn.PeerHasPiece(piece) { if conn.PeerHasPiece(piece) {
conn.updateRequests() conn.updateRequests()
@ -2202,13 +2202,13 @@ func (me *Client) onFailedPiece(t *Torrent, piece int) {
} }
} }
func (me *Client) pieceChanged(t *Torrent, piece int) { func (cl *Client) pieceChanged(t *Torrent, piece int) {
correct := t.pieceComplete(piece) correct := t.pieceComplete(piece)
defer me.event.Broadcast() defer cl.event.Broadcast()
if correct { if correct {
me.onCompletedPiece(t, piece) cl.onCompletedPiece(t, piece)
} else { } else {
me.onFailedPiece(t, piece) cl.onFailedPiece(t, piece)
} }
if t.updatePiecePriority(piece) { if t.updatePiecePriority(piece) {
t.piecePriorityChanged(piece) t.piecePriorityChanged(piece)
@ -2244,45 +2244,45 @@ func (cl *Client) verifyPiece(t *Torrent, piece int) {
} }
// Returns handles to all the torrents loaded in the Client. // Returns handles to all the torrents loaded in the Client.
func (me *Client) Torrents() (ret []*Torrent) { func (cl *Client) Torrents() (ret []*Torrent) {
me.mu.Lock() cl.mu.Lock()
for _, t := range me.torrents { for _, t := range cl.torrents {
ret = append(ret, t) ret = append(ret, t)
} }
me.mu.Unlock() cl.mu.Unlock()
return return
} }
func (me *Client) AddMagnet(uri string) (T *Torrent, err error) { func (cl *Client) AddMagnet(uri string) (T *Torrent, err error) {
spec, err := TorrentSpecFromMagnetURI(uri) spec, err := TorrentSpecFromMagnetURI(uri)
if err != nil { if err != nil {
return return
} }
T, _, err = me.AddTorrentSpec(spec) T, _, err = cl.AddTorrentSpec(spec)
return return
} }
func (me *Client) AddTorrent(mi *metainfo.MetaInfo) (T *Torrent, err error) { func (cl *Client) AddTorrent(mi *metainfo.MetaInfo) (T *Torrent, err error) {
T, _, err = me.AddTorrentSpec(TorrentSpecFromMetaInfo(mi)) T, _, err = cl.AddTorrentSpec(TorrentSpecFromMetaInfo(mi))
var ss []string var ss []string
missinggo.CastSlice(&ss, mi.Nodes) missinggo.CastSlice(&ss, mi.Nodes)
me.AddDHTNodes(ss) cl.AddDHTNodes(ss)
return return
} }
func (me *Client) AddTorrentFromFile(filename string) (T *Torrent, err error) { func (cl *Client) AddTorrentFromFile(filename string) (T *Torrent, err error) {
mi, err := metainfo.LoadFromFile(filename) mi, err := metainfo.LoadFromFile(filename)
if err != nil { if err != nil {
return return
} }
return me.AddTorrent(mi) return cl.AddTorrent(mi)
} }
func (me *Client) DHT() *dht.Server { func (cl *Client) DHT() *dht.Server {
return me.dHT return cl.dHT
} }
func (me *Client) AddDHTNodes(nodes []string) { func (cl *Client) AddDHTNodes(nodes []string) {
for _, n := range nodes { for _, n := range nodes {
hmp := missinggo.SplitHostMaybePort(n) hmp := missinggo.SplitHostMaybePort(n)
ip := net.ParseIP(hmp.Host) ip := net.ParseIP(hmp.Host)
@ -2296,6 +2296,6 @@ func (me *Client) AddDHTNodes(nodes []string) {
Port: hmp.Port, Port: hmp.Port,
}), }),
} }
me.DHT().AddNode(ni) cl.DHT().AddNode(ni)
} }
} }

22
client_test.go
View File

@ -443,15 +443,15 @@ func TestMergingTrackersByAddingSpecs(t *testing.T) {
type badStorage struct{} type badStorage struct{}
func (me badStorage) OpenTorrent(*metainfo.InfoEx) (storage.Torrent, error) { func (bs badStorage) OpenTorrent(*metainfo.InfoEx) (storage.Torrent, error) {
return me, nil return bs, nil
} }
func (me badStorage) Close() error { func (bs badStorage) Close() error {
return nil return nil
} }
func (me badStorage) Piece(p metainfo.Piece) storage.Piece { func (bs badStorage) Piece(p metainfo.Piece) storage.Piece {
return badStoragePiece{p} return badStoragePiece{p}
} }
@ -459,25 +459,25 @@ type badStoragePiece struct {
p metainfo.Piece p metainfo.Piece
} }
func (me badStoragePiece) WriteAt(b []byte, off int64) (int, error) { func (p badStoragePiece) WriteAt(b []byte, off int64) (int, error) {
return 0, nil return 0, nil
} }
func (me badStoragePiece) GetIsComplete() bool { func (p badStoragePiece) GetIsComplete() bool {
return true return true
} }
func (me badStoragePiece) MarkComplete() error { func (p badStoragePiece) MarkComplete() error {
return errors.New("psyyyyyyyche") return errors.New("psyyyyyyyche")
} }
func (me badStoragePiece) randomlyTruncatedDataString() string { func (p badStoragePiece) randomlyTruncatedDataString() string {
return "hello, world\n"[:rand.Intn(14)] return "hello, world\n"[:rand.Intn(14)]
} }
func (me badStoragePiece) ReadAt(b []byte, off int64) (n int, err error) { func (p badStoragePiece) ReadAt(b []byte, off int64) (n int, err error) {
r := strings.NewReader(me.randomlyTruncatedDataString()) r := strings.NewReader(p.randomlyTruncatedDataString())
return r.ReadAt(b, off+me.p.Offset()) return r.ReadAt(b, off+p.p.Offset())
} }
// We read from a piece which is marked completed, but is missing data. // We read from a piece which is marked completed, but is missing data.

264
connection.go
View File

@ -208,40 +208,40 @@ func (cn *connection) WriteStatus(w io.Writer, t *Torrent) {
) )
} }
func (c *connection) Close() { func (cn *connection) Close() {
c.closed.Set() cn.closed.Set()
c.discardPieceInclination() cn.discardPieceInclination()
c.pieceRequestOrder.Clear() cn.pieceRequestOrder.Clear()
// TODO: This call blocks sometimes, why? // TODO: This call blocks sometimes, why?
go c.conn.Close() go cn.conn.Close()
} }
func (c *connection) PeerHasPiece(piece int) bool { func (cn *connection) PeerHasPiece(piece int) bool {
return c.peerHasAll || c.peerPieces.Contains(piece) return cn.peerHasAll || cn.peerPieces.Contains(piece)
} }
func (c *connection) Post(msg pp.Message) { func (cn *connection) Post(msg pp.Message) {
select { select {
case c.post <- msg: case cn.post <- msg:
postedMessageTypes.Add(strconv.FormatInt(int64(msg.Type), 10), 1) postedMessageTypes.Add(strconv.FormatInt(int64(msg.Type), 10), 1)
case <-c.closed.C(): case <-cn.closed.C():
} }
} }
func (c *connection) RequestPending(r request) bool { func (cn *connection) RequestPending(r request) bool {
_, ok := c.Requests[r] _, ok := cn.Requests[r]
return ok return ok
} }
func (c *connection) requestMetadataPiece(index int) { func (cn *connection) requestMetadataPiece(index int) {
eID := c.PeerExtensionIDs["ut_metadata"] eID := cn.PeerExtensionIDs["ut_metadata"]
if eID == 0 { if eID == 0 {
return return
} }
if index < len(c.metadataRequests) && c.metadataRequests[index] { if index < len(cn.metadataRequests) && cn.metadataRequests[index] {
return return
} }
c.Post(pp.Message{ cn.Post(pp.Message{
Type: pp.Extended, Type: pp.Extended,
ExtendedID: eID, ExtendedID: eID,
ExtendedPayload: func() []byte { ExtendedPayload: func() []byte {
@ -255,19 +255,19 @@ func (c *connection) requestMetadataPiece(index int) {
return b return b
}(), }(),
}) })
for index >= len(c.metadataRequests) { for index >= len(cn.metadataRequests) {
c.metadataRequests = append(c.metadataRequests, false) cn.metadataRequests = append(cn.metadataRequests, false)
} }
c.metadataRequests[index] = true cn.metadataRequests[index] = true
} }
func (c *connection) requestedMetadataPiece(index int) bool { func (cn *connection) requestedMetadataPiece(index int) bool {
return index < len(c.metadataRequests) && c.metadataRequests[index] return index < len(cn.metadataRequests) && cn.metadataRequests[index]
} }
// The actual value to use as the maximum outbound requests. // The actual value to use as the maximum outbound requests.
func (c *connection) nominalMaxRequests() (ret int) { func (cn *connection) nominalMaxRequests() (ret int) {
ret = c.PeerMaxRequests ret = cn.PeerMaxRequests
if ret > 64 { if ret > 64 {
ret = 64 ret = 64
} }
@ -275,26 +275,26 @@ func (c *connection) nominalMaxRequests() (ret int) {
} }
// Returns true if more requests can be sent. // Returns true if more requests can be sent.
func (c *connection) Request(chunk request) bool { func (cn *connection) Request(chunk request) bool {
if len(c.Requests) >= c.nominalMaxRequests() { if len(cn.Requests) >= cn.nominalMaxRequests() {
return false return false
} }
if !c.PeerHasPiece(int(chunk.Index)) { if !cn.PeerHasPiece(int(chunk.Index)) {
return true return true
} }
if c.RequestPending(chunk) { if cn.RequestPending(chunk) {
return true return true
} }
c.SetInterested(true) cn.SetInterested(true)
if c.PeerChoked { if cn.PeerChoked {
return false return false
} }
if c.Requests == nil { if cn.Requests == nil {
c.Requests = make(map[request]struct{}, c.PeerMaxRequests) cn.Requests = make(map[request]struct{}, cn.PeerMaxRequests)
} }
c.Requests[chunk] = struct{}{} cn.Requests[chunk] = struct{}{}
c.requestsLowWater = len(c.Requests) / 2 cn.requestsLowWater = len(cn.Requests) / 2
c.Post(pp.Message{ cn.Post(pp.Message{
Type: pp.Request, Type: pp.Request,
Index: chunk.Index, Index: chunk.Index,
Begin: chunk.Begin, Begin: chunk.Begin,
@ -304,15 +304,15 @@ func (c *connection) Request(chunk request) bool {
} }
// Returns true if an unsatisfied request was canceled. // Returns true if an unsatisfied request was canceled.
func (c *connection) Cancel(r request) bool { func (cn *connection) Cancel(r request) bool {
if c.Requests == nil { if cn.Requests == nil {
return false return false
} }
if _, ok := c.Requests[r]; !ok { if _, ok := cn.Requests[r]; !ok {
return false return false
} }
delete(c.Requests, r) delete(cn.Requests, r)
c.Post(pp.Message{ cn.Post(pp.Message{
Type: pp.Cancel, Type: pp.Cancel,
Index: r.Index, Index: r.Index,
Begin: r.Begin, Begin: r.Begin,
@ -322,43 +322,43 @@ func (c *connection) Cancel(r request) bool {
} }
// Returns true if an unsatisfied request was canceled. // Returns true if an unsatisfied request was canceled.
func (c *connection) PeerCancel(r request) bool { func (cn *connection) PeerCancel(r request) bool {
if c.PeerRequests == nil { if cn.PeerRequests == nil {
return false return false
} }
if _, ok := c.PeerRequests[r]; !ok { if _, ok := cn.PeerRequests[r]; !ok {
return false return false
} }
delete(c.PeerRequests, r) delete(cn.PeerRequests, r)
return true return true
} }
func (c *connection) Choke() { func (cn *connection) Choke() {
if c.Choked { if cn.Choked {
return return
} }
c.Post(pp.Message{ cn.Post(pp.Message{
Type: pp.Choke, Type: pp.Choke,
}) })
c.PeerRequests = nil cn.PeerRequests = nil
c.Choked = true cn.Choked = true
} }
func (c *connection) Unchoke() { func (cn *connection) Unchoke() {
if !c.Choked { if !cn.Choked {
return return
} }
c.Post(pp.Message{ cn.Post(pp.Message{
Type: pp.Unchoke, Type: pp.Unchoke,
}) })
c.Choked = false cn.Choked = false
} }
func (c *connection) SetInterested(interested bool) { func (cn *connection) SetInterested(interested bool) {
if c.Interested == interested { if cn.Interested == interested {
return return
} }
c.Post(pp.Message{ cn.Post(pp.Message{
Type: func() pp.MessageType { Type: func() pp.MessageType {
if interested { if interested {
return pp.Interested return pp.Interested
@ -367,7 +367,7 @@ func (c *connection) SetInterested(interested bool) {
} }
}(), }(),
}) })
c.Interested = interested cn.Interested = interested
} }
var ( var (
@ -378,19 +378,19 @@ var (
) )
// Writes buffers to the socket from the write channel. // Writes buffers to the socket from the write channel.
func (conn *connection) writer() { func (cn *connection) writer() {
defer func() { defer func() {
conn.t.cl.mu.Lock() cn.t.cl.mu.Lock()
defer conn.t.cl.mu.Unlock() defer cn.t.cl.mu.Unlock()
conn.Close() cn.Close()
}() }()
// Reduce write syscalls. // Reduce write syscalls.
buf := bufio.NewWriter(conn.rw) buf := bufio.NewWriter(cn.rw)
for { for {
if buf.Buffered() == 0 { if buf.Buffered() == 0 {
// There's nothing to write, so block until we get something. // There's nothing to write, so block until we get something.
select { select {
case b, ok := <-conn.writeCh: case b, ok := <-cn.writeCh:
if !ok { if !ok {
return return
} }
@ -399,14 +399,14 @@ func (conn *connection) writer() {
if err != nil { if err != nil {
return return
} }
case <-conn.closed.C(): case <-cn.closed.C():
return return
} }
} else { } else {
// We already have something to write, so flush if there's nothing // We already have something to write, so flush if there's nothing
// more to write. // more to write.
select { select {
case b, ok := <-conn.writeCh: case b, ok := <-cn.writeCh:
if !ok { if !ok {
return return
} }
@ -415,7 +415,7 @@ func (conn *connection) writer() {
if err != nil { if err != nil {
return return
} }
case <-conn.closed.C(): case <-cn.closed.C():
return return
default: default:
connectionWriterFlush.Add(1) connectionWriterFlush.Add(1)
@ -428,15 +428,15 @@ func (conn *connection) writer() {
} }
} }
func (conn *connection) writeOptimizer(keepAliveDelay time.Duration) { func (cn *connection) writeOptimizer(keepAliveDelay time.Duration) {
defer close(conn.writeCh) // Responsible for notifying downstream routines. defer close(cn.writeCh) // Responsible for notifying downstream routines.
pending := list.New() // Message queue. pending := list.New() // Message queue.
var nextWrite []byte // Set to nil if we need to need to marshal the next message. var nextWrite []byte // Set to nil if we need to need to marshal the next message.
timer := time.NewTimer(keepAliveDelay) timer := time.NewTimer(keepAliveDelay)
defer timer.Stop() defer timer.Stop()
lastWrite := time.Now() lastWrite := time.Now()
for { for {
write := conn.writeCh // Set to nil if there's nothing to write. write := cn.writeCh // Set to nil if there's nothing to write.
if pending.Len() == 0 { if pending.Len() == 0 {
write = nil write = nil
} else if nextWrite == nil { } else if nextWrite == nil {
@ -459,7 +459,7 @@ func (conn *connection) writeOptimizer(keepAliveDelay time.Duration) {
} }
pending.PushBack(pp.Message{Keepalive: true}) pending.PushBack(pp.Message{Keepalive: true})
postedKeepalives.Add(1) postedKeepalives.Add(1)
case msg, ok := <-conn.post: case msg, ok := <-cn.post:
if !ok { if !ok {
return return
} }
@ -481,7 +481,7 @@ func (conn *connection) writeOptimizer(keepAliveDelay time.Duration) {
if pending.Len() == 0 { if pending.Len() == 0 {
timer.Reset(keepAliveDelay) timer.Reset(keepAliveDelay)
} }
case <-conn.closed.C(): case <-cn.closed.C():
return return
} }
} }
@ -512,134 +512,134 @@ func (cn *connection) Bitfield(haves []bool) {
cn.sentHaves = haves cn.sentHaves = haves
} }
func (c *connection) updateRequests() { func (cn *connection) updateRequests() {
if !c.t.haveInfo() { if !cn.t.haveInfo() {
return return
} }
if c.Interested { if cn.Interested {
if c.PeerChoked { if cn.PeerChoked {
return return
} }
if len(c.Requests) > c.requestsLowWater { if len(cn.Requests) > cn.requestsLowWater {
return return
} }
} }
c.fillRequests() cn.fillRequests()
if len(c.Requests) == 0 && !c.PeerChoked { if len(cn.Requests) == 0 && !cn.PeerChoked {
// So we're not choked, but we don't want anything right now. We may // So we're not choked, but we don't want anything right now. We may
// have completed readahead, and the readahead window has not rolled // have completed readahead, and the readahead window has not rolled
// over to the next piece. Better to stay interested in case we're // over to the next piece. Better to stay interested in case we're
// going to want data in the near future. // going to want data in the near future.
c.SetInterested(!c.t.haveAllPieces()) cn.SetInterested(!cn.t.haveAllPieces())
} }
} }
func (c *connection) fillRequests() { func (cn *connection) fillRequests() {
c.pieceRequestOrder.IterTyped(func(piece int) (more bool) { cn.pieceRequestOrder.IterTyped(func(piece int) (more bool) {
if c.t.cl.config.Debug && c.t.havePiece(piece) { if cn.t.cl.config.Debug && cn.t.havePiece(piece) {
panic(piece) panic(piece)
} }
return c.requestPiecePendingChunks(piece) return cn.requestPiecePendingChunks(piece)
}) })
} }
func (c *connection) requestPiecePendingChunks(piece int) (again bool) { func (cn *connection) requestPiecePendingChunks(piece int) (again bool) {
return c.t.connRequestPiecePendingChunks(c, piece) return cn.t.connRequestPiecePendingChunks(cn, piece)
} }
func (c *connection) stopRequestingPiece(piece int) { func (cn *connection) stopRequestingPiece(piece int) {
c.pieceRequestOrder.Remove(piece) cn.pieceRequestOrder.Remove(piece)
} }
func (c *connection) updatePiecePriority(piece int) { func (cn *connection) updatePiecePriority(piece int) {
tpp := c.t.piecePriority(piece) tpp := cn.t.piecePriority(piece)
if !c.PeerHasPiece(piece) { if !cn.PeerHasPiece(piece) {
tpp = PiecePriorityNone tpp = PiecePriorityNone
} }
if tpp == PiecePriorityNone { if tpp == PiecePriorityNone {
c.stopRequestingPiece(piece) cn.stopRequestingPiece(piece)
return return
} }
prio := c.getPieceInclination()[piece] prio := cn.getPieceInclination()[piece]
switch tpp { switch tpp {
case PiecePriorityNormal: case PiecePriorityNormal:
case PiecePriorityReadahead: case PiecePriorityReadahead:
prio -= c.t.numPieces() prio -= cn.t.numPieces()
case PiecePriorityNext, PiecePriorityNow: case PiecePriorityNext, PiecePriorityNow:
prio -= 2 * c.t.numPieces() prio -= 2 * cn.t.numPieces()
default: default:
panic(tpp) panic(tpp)
} }
prio += piece prio += piece
c.pieceRequestOrder.Set(piece, prio) cn.pieceRequestOrder.Set(piece, prio)
c.updateRequests() cn.updateRequests()
} }
func (c *connection) getPieceInclination() []int { func (cn *connection) getPieceInclination() []int {
if c.pieceInclination == nil { if cn.pieceInclination == nil {
c.pieceInclination = c.t.getConnPieceInclination() cn.pieceInclination = cn.t.getConnPieceInclination()
} }
return c.pieceInclination return cn.pieceInclination
} }
func (c *connection) discardPieceInclination() { func (cn *connection) discardPieceInclination() {
if c.pieceInclination == nil { if cn.pieceInclination == nil {
return return
} }
c.t.putPieceInclination(c.pieceInclination) cn.t.putPieceInclination(cn.pieceInclination)
c.pieceInclination = nil cn.pieceInclination = nil
} }
func (c *connection) peerHasPieceChanged(piece int) { func (cn *connection) peerHasPieceChanged(piece int) {
c.updatePiecePriority(piece) cn.updatePiecePriority(piece)
} }
func (c *connection) peerPiecesChanged() { func (cn *connection) peerPiecesChanged() {
if c.t.haveInfo() { if cn.t.haveInfo() {
for i := range iter.N(c.t.numPieces()) { for i := range iter.N(cn.t.numPieces()) {
c.peerHasPieceChanged(i) cn.peerHasPieceChanged(i)
} }
} }
} }
func (c *connection) raisePeerMinPieces(newMin int) { func (cn *connection) raisePeerMinPieces(newMin int) {
if newMin > c.peerMinPieces { if newMin > cn.peerMinPieces {
c.peerMinPieces = newMin cn.peerMinPieces = newMin
} }
} }
func (c *connection) peerSentHave(piece int) error { func (cn *connection) peerSentHave(piece int) error {
if c.t.haveInfo() && piece >= c.t.numPieces() { if cn.t.haveInfo() && piece >= cn.t.numPieces() {
return errors.New("invalid piece") return errors.New("invalid piece")
} }
if c.PeerHasPiece(piece) { if cn.PeerHasPiece(piece) {
return nil return nil
} }
c.raisePeerMinPieces(piece + 1) cn.raisePeerMinPieces(piece + 1)
c.peerPieces.Set(piece, true) cn.peerPieces.Set(piece, true)
c.peerHasPieceChanged(piece) cn.peerHasPieceChanged(piece)
return nil return nil
} }
func (c *connection) peerSentBitfield(bf []bool) error { func (cn *connection) peerSentBitfield(bf []bool) error {
c.peerHasAll = false cn.peerHasAll = false
if len(bf)%8 != 0 { if len(bf)%8 != 0 {
panic("expected bitfield length divisible by 8") panic("expected bitfield length divisible by 8")
} }
// We know that the last byte means that at most the last 7 bits are // We know that the last byte means that at most the last 7 bits are
// wasted. // wasted.
c.raisePeerMinPieces(len(bf) - 7) cn.raisePeerMinPieces(len(bf) - 7)
if c.t.haveInfo() { if cn.t.haveInfo() {
// Ignore known excess pieces. // Ignore known excess pieces.
bf = bf[:c.t.numPieces()] bf = bf[:cn.t.numPieces()]
} }
for i, have := range bf { for i, have := range bf {
if have { if have {
c.raisePeerMinPieces(i + 1) cn.raisePeerMinPieces(i + 1)
} }
c.peerPieces.Set(i, have) cn.peerPieces.Set(i, have)
} }
c.peerPiecesChanged() cn.peerPiecesChanged()
return nil return nil
} }
@ -650,9 +650,9 @@ func (cn *connection) peerSentHaveAll() error {
return nil return nil
} }
func (c *connection) peerSentHaveNone() error { func (cn *connection) peerSentHaveNone() error {
c.peerPieces.Clear() cn.peerPieces.Clear()
c.peerHasAll = false cn.peerHasAll = false
c.peerPiecesChanged() cn.peerPiecesChanged()
return nil return nil
} }

100
dht/announce.go
View File

@ -30,10 +30,10 @@ type Announce struct {
} }
// Returns the number of distinct remote addresses the announce has queried. // Returns the number of distinct remote addresses the announce has queried.
func (me *Announce) NumContacted() int { func (a *Announce) NumContacted() int {
me.mu.Lock() a.mu.Lock()
defer me.mu.Unlock() defer a.mu.Unlock()
return me.numContacted return a.numContacted
} }
// This is kind of the main thing you want to do with DHT. It traverses the // This is kind of the main thing you want to do with DHT. It traverses the
@ -95,57 +95,57 @@ func (s *Server) Announce(infoHash string, port int, impliedPort bool) (*Announc
return disc, nil return disc, nil
} }
func (me *Announce) gotNodeAddr(addr Addr) { func (a *Announce) gotNodeAddr(addr Addr) {
if addr.UDPAddr().Port == 0 { if addr.UDPAddr().Port == 0 {
// Not a contactable address. // Not a contactable address.
return return
} }
if me.triedAddrs.Test([]byte(addr.String())) { if a.triedAddrs.Test([]byte(addr.String())) {
return return
} }
if me.server.ipBlocked(addr.UDPAddr().IP) { if a.server.ipBlocked(addr.UDPAddr().IP) {
return return
} }
me.server.mu.Lock() a.server.mu.Lock()
if me.server.badNodes.Test([]byte(addr.String())) { if a.server.badNodes.Test([]byte(addr.String())) {
me.server.mu.Unlock() a.server.mu.Unlock()
return return
} }
me.server.mu.Unlock() a.server.mu.Unlock()
me.contact(addr) a.contact(addr)
} }
func (me *Announce) contact(addr Addr) { func (a *Announce) contact(addr Addr) {
me.numContacted++ a.numContacted++
me.triedAddrs.Add([]byte(addr.String())) a.triedAddrs.Add([]byte(addr.String()))
if err := me.getPeers(addr); err != nil { if err := a.getPeers(addr); err != nil {
log.Printf("error sending get_peers request to %s: %#v", addr, err) log.Printf("error sending get_peers request to %s: %#v", addr, err)
return return
} }
me.pending++ a.pending++
} }
func (me *Announce) transactionClosed() { func (a *Announce) transactionClosed() {
me.pending-- a.pending--
if me.pending == 0 { if a.pending == 0 {
me.close() a.close()
return return
} }
} }
func (me *Announce) responseNode(node NodeInfo) { func (a *Announce) responseNode(node NodeInfo) {
me.gotNodeAddr(node.Addr) a.gotNodeAddr(node.Addr)
} }
func (me *Announce) closingCh() chan struct{} { func (a *Announce) closingCh() chan struct{} {
return me.stop return a.stop
} }
// Announce to a peer, if appropriate. // Announce to a peer, if appropriate.
func (me *Announce) maybeAnnouncePeer(to Addr, token, peerId string) { func (a *Announce) maybeAnnouncePeer(to Addr, token, peerId string) {
me.server.mu.Lock() a.server.mu.Lock()
defer me.server.mu.Unlock() defer a.server.mu.Unlock()
if !me.server.config.NoSecurity { if !a.server.config.NoSecurity {
if len(peerId) != 20 { if len(peerId) != 20 {
return return
} }
@ -153,27 +153,27 @@ func (me *Announce) maybeAnnouncePeer(to Addr, token, peerId string) {
return return
} }
} }
err := me.server.announcePeer(to, me.infoHash, me.announcePort, token, me.announcePortImplied) err := a.server.announcePeer(to, a.infoHash, a.announcePort, token, a.announcePortImplied)
if err != nil { if err != nil {
logonce.Stderr.Printf("error announcing peer: %s", err) logonce.Stderr.Printf("error announcing peer: %s", err)
} }
} }
func (me *Announce) getPeers(addr Addr) error { func (a *Announce) getPeers(addr Addr) error {
me.server.mu.Lock() a.server.mu.Lock()
defer me.server.mu.Unlock() defer a.server.mu.Unlock()
t, err := me.server.getPeers(addr, me.infoHash) t, err := a.server.getPeers(addr, a.infoHash)
if err != nil { if err != nil {
return err return err
} }
t.SetResponseHandler(func(m Msg, ok bool) { t.SetResponseHandler(func(m Msg, ok bool) {
// Register suggested nodes closer to the target info-hash. // Register suggested nodes closer to the target info-hash.
if m.R != nil { if m.R != nil {
me.mu.Lock() a.mu.Lock()
for _, n := range m.R.Nodes { for _, n := range m.R.Nodes {
me.responseNode(n) a.responseNode(n)
} }
me.mu.Unlock() a.mu.Unlock()
if vs := m.R.Values; len(vs) != 0 { if vs := m.R.Values; len(vs) != 0 {
nodeInfo := NodeInfo{ nodeInfo := NodeInfo{
@ -181,7 +181,7 @@ func (me *Announce) getPeers(addr Addr) error {
} }
copy(nodeInfo.ID[:], m.SenderID()) copy(nodeInfo.ID[:], m.SenderID())
select { select {
case me.values <- PeersValues{ case a.values <- PeersValues{
Peers: func() (ret []Peer) { Peers: func() (ret []Peer) {
for _, cp := range vs { for _, cp := range vs {
ret = append(ret, Peer(cp)) ret = append(ret, Peer(cp))
@ -190,16 +190,16 @@ func (me *Announce) getPeers(addr Addr) error {
}(), }(),
NodeInfo: nodeInfo, NodeInfo: nodeInfo,
}: }:
case <-me.stop: case <-a.stop:
} }
} }
me.maybeAnnouncePeer(addr, m.R.Token, m.SenderID()) a.maybeAnnouncePeer(addr, m.R.Token, m.SenderID())
} }
me.mu.Lock() a.mu.Lock()
me.transactionClosed() a.transactionClosed()
me.mu.Unlock() a.mu.Unlock()
}) })
return nil return nil
} }
@ -213,16 +213,16 @@ type PeersValues struct {
} }
// Stop the announce. // Stop the announce.
func (me *Announce) Close() { func (a *Announce) Close() {
me.mu.Lock() a.mu.Lock()
defer me.mu.Unlock() defer a.mu.Unlock()
me.close() a.close()
} }
func (ps *Announce) close() { func (a *Announce) close() {
select { select {
case <-ps.stop: case <-a.stop:
default: default:
close(ps.stop) close(a.stop)
} }
} }

32
dht/closest_nodes.go
View File

@ -9,23 +9,23 @@ type nodeMaxHeap struct {
Target nodeID Target nodeID
} }
func (me nodeMaxHeap) Len() int { return len(me.IDs) } func (mh nodeMaxHeap) Len() int { return len(mh.IDs) }
func (me nodeMaxHeap) Less(i, j int) bool { func (mh nodeMaxHeap) Less(i, j int) bool {
m := me.IDs[i].Distance(&me.Target) m := mh.IDs[i].Distance(&mh.Target)
n := me.IDs[j].Distance(&me.Target) n := mh.IDs[j].Distance(&mh.Target)
return m.Cmp(&n) > 0 return m.Cmp(&n) > 0
} }
func (me *nodeMaxHeap) Pop() (ret interface{}) { func (mh *nodeMaxHeap) Pop() (ret interface{}) {
ret, me.IDs = me.IDs[len(me.IDs)-1], me.IDs[:len(me.IDs)-1] ret, mh.IDs = mh.IDs[len(mh.IDs)-1], mh.IDs[:len(mh.IDs)-1]
return return
} }
func (me *nodeMaxHeap) Push(val interface{}) { func (mh *nodeMaxHeap) Push(val interface{}) {
me.IDs = append(me.IDs, val.(nodeID)) mh.IDs = append(mh.IDs, val.(nodeID))
} }
func (me nodeMaxHeap) Swap(i, j int) { func (mh nodeMaxHeap) Swap(i, j int) {
me.IDs[i], me.IDs[j] = me.IDs[j], me.IDs[i] mh.IDs[i], mh.IDs[j] = mh.IDs[j], mh.IDs[i]
} }
type closestNodesSelector struct { type closestNodesSelector struct {
@ -33,15 +33,15 @@ type closestNodesSelector struct {
k int k int
} }
func (me *closestNodesSelector) Push(id nodeID) { func (cns *closestNodesSelector) Push(id nodeID) {
heap.Push(&me.closest, id) heap.Push(&cns.closest, id)
if me.closest.Len() > me.k { if cns.closest.Len() > cns.k {
heap.Pop(&me.closest) heap.Pop(&cns.closest)
} }
} }
func (me *closestNodesSelector) IDs() []nodeID { func (cns *closestNodesSelector) IDs() []nodeID {
return me.closest.IDs return cns.closest.IDs
} }
func newKClosestNodesSelector(k int, targetID nodeID) (ret closestNodesSelector) { func newKClosestNodesSelector(k int, targetID nodeID) (ret closestNodesSelector) {

12
dht/compactNodeInfo.go
View File

@ -13,7 +13,7 @@ type CompactIPv4NodeInfo []NodeInfo
var _ bencode.Unmarshaler = &CompactIPv4NodeInfo{} var _ bencode.Unmarshaler = &CompactIPv4NodeInfo{}
func (me *CompactIPv4NodeInfo) UnmarshalBencode(_b []byte) (err error) { func (i *CompactIPv4NodeInfo) UnmarshalBencode(_b []byte) (err error) {
var b []byte var b []byte
err = bencode.Unmarshal(_b, &b) err = bencode.Unmarshal(_b, &b)
if err != nil { if err != nil {
@ -23,20 +23,20 @@ func (me *CompactIPv4NodeInfo) UnmarshalBencode(_b []byte) (err error) {
err = fmt.Errorf("bad length: %d", len(b)) err = fmt.Errorf("bad length: %d", len(b))
return return
} }
for i := 0; i < len(b); i += CompactIPv4NodeInfoLen { for k := 0; k < len(b); k += CompactIPv4NodeInfoLen {
var ni NodeInfo var ni NodeInfo
err = ni.UnmarshalCompactIPv4(b[i : i+CompactIPv4NodeInfoLen]) err = ni.UnmarshalCompactIPv4(b[k : k+CompactIPv4NodeInfoLen])
if err != nil { if err != nil {
return return
} }
*me = append(*me, ni) *i = append(*i, ni)
} }
return return
} }
func (me CompactIPv4NodeInfo) MarshalBencode() (ret []byte, err error) { func (i CompactIPv4NodeInfo) MarshalBencode() (ret []byte, err error) {
var buf bytes.Buffer var buf bytes.Buffer
for _, ni := range me { for _, ni := range i {
buf.Write(ni.ID[:]) buf.Write(ni.ID[:])
if ni.Addr == nil { if ni.Addr == nil {
err = errors.New("nil addr in node info") err = errors.New("nil addr in node info")

4
dht/dht.go
View File

@ -194,8 +194,8 @@ type Peer struct {
Port int Port int
} }
func (me *Peer) String() string { func (p *Peer) String() string {
return net.JoinHostPort(me.IP.String(), strconv.FormatInt(int64(me.Port), 10)) return net.JoinHostPort(p.IP.String(), strconv.FormatInt(int64(p.Port), 10))
} }
func bootstrapAddrs(nodeAddrs []string) (addrs []*net.UDPAddr, err error) { func bootstrapAddrs(nodeAddrs []string) (addrs []*net.UDPAddr, err error) {

16
dht/krpcError.go
View File

@ -18,7 +18,7 @@ var (
_ error = KRPCError{} _ error = KRPCError{}
) )
func (me *KRPCError) UnmarshalBencode(_b []byte) (err error) { func (e *KRPCError) UnmarshalBencode(_b []byte) (err error) {
var _v interface{} var _v interface{}
err = bencode.Unmarshal(_b, &_v) err = bencode.Unmarshal(_b, &_v)
if err != nil { if err != nil {
@ -26,20 +26,20 @@ func (me *KRPCError) UnmarshalBencode(_b []byte) (err error) {
} }
switch v := _v.(type) { switch v := _v.(type) {
case []interface{}: case []interface{}:
me.Code = int(v[0].(int64)) e.Code = int(v[0].(int64))
me.Msg = v[1].(string) e.Msg = v[1].(string)
case string: case string:
me.Msg = v e.Msg = v
default: default:
err = fmt.Errorf(`KRPC error bencode value has unexpected type: %T`, _v) err = fmt.Errorf(`KRPC error bencode value has unexpected type: %T`, _v)
} }
return return
} }
func (me KRPCError) MarshalBencode() (ret []byte, err error) { func (e KRPCError) MarshalBencode() (ret []byte, err error) {
return bencode.Marshal([]interface{}{me.Code, me.Msg}) return bencode.Marshal([]interface{}{e.Code, e.Msg})
} }
func (me KRPCError) Error() string { func (e KRPCError) Error() string {
return fmt.Sprintf("KRPC error %d: %s", me.Code, me.Msg) return fmt.Sprintf("KRPC error %d: %s", e.Code, e.Msg)
} }

6
dht/nodeinfo.go
View File

@ -33,12 +33,12 @@ func (ni *NodeInfo) PutCompact(b []byte) error {
return nil return nil
} }
func (cni *NodeInfo) UnmarshalCompactIPv4(b []byte) error { func (ni *NodeInfo) UnmarshalCompactIPv4(b []byte) error {
if len(b) != CompactIPv4NodeInfoLen { if len(b) != CompactIPv4NodeInfoLen {
return errors.New("expected 26 bytes") return errors.New("expected 26 bytes")
} }
missinggo.CopyExact(cni.ID[:], b[:20]) missinggo.CopyExact(ni.ID[:], b[:20])
cni.Addr = NewAddr(&net.UDPAddr{ ni.Addr = NewAddr(&net.UDPAddr{
IP: append(make([]byte, 0, 4), b[20:24]...), IP: append(make([]byte, 0, 4), b[20:24]...),
Port: int(binary.BigEndian.Uint16(b[24:26])), Port: int(binary.BigEndian.Uint16(b[24:26])),
}) })

6
dht/server.go
View File

@ -709,7 +709,7 @@ func (s *Server) closestNodes(k int, target nodeID, filter func(*node) bool) []*
return ret return ret
} }
func (me *Server) badNode(addr Addr) { func (s *Server) badNode(addr Addr) {
me.badNodes.Add([]byte(addr.String())) s.badNodes.Add([]byte(addr.String()))
delete(me.nodes, addr.String()) delete(s.nodes, addr.String())
} }

4
file.go
View File

@ -24,11 +24,11 @@ func (f *File) Offset() int64 {
return f.offset return f.offset
} }
func (f File) FileInfo() metainfo.FileInfo { func (f *File) FileInfo() metainfo.FileInfo {
return f.fi return f.fi
} }
func (f File) Path() string { func (f *File) Path() string {
return f.path return f.path
} }

26
fs/torrentfs.go
View File

@ -227,15 +227,15 @@ func (dn dirNode) Attr(ctx context.Context, attr *fuse.Attr) error {
return nil return nil
} }
func (me rootNode) Lookup(ctx context.Context, name string) (_node fusefs.Node, err error) { func (rn rootNode) Lookup(ctx context.Context, name string) (_node fusefs.Node, err error) {
for _, t := range me.fs.Client.Torrents() { for _, t := range rn.fs.Client.Torrents() {
info := t.Info() info := t.Info()
if t.Name() != name || info == nil { if t.Name() != name || info == nil {
continue continue
} }
__node := node{ __node := node{
metadata: info, metadata: info,
FS: me.fs, FS: rn.fs,
t: t, t: t,
} }
if !info.IsDir() { if !info.IsDir() {
@ -251,8 +251,8 @@ func (me rootNode) Lookup(ctx context.Context, name string) (_node fusefs.Node,
return return
} }
func (me rootNode) ReadDirAll(ctx context.Context) (dirents []fuse.Dirent, err error) { func (rn rootNode) ReadDirAll(ctx context.Context) (dirents []fuse.Dirent, err error) {
for _, t := range me.fs.Client.Torrents() { for _, t := range rn.fs.Client.Torrents() {
info := t.Info() info := t.Info()
if info == nil { if info == nil {
continue continue
@ -271,28 +271,28 @@ func (me rootNode) ReadDirAll(ctx context.Context) (dirents []fuse.Dirent, err e
return return
} }
func (rootNode) Attr(ctx context.Context, attr *fuse.Attr) error { func (rn rootNode) Attr(ctx context.Context, attr *fuse.Attr) error {
attr.Mode = os.ModeDir attr.Mode = os.ModeDir
return nil return nil
} }
// TODO(anacrolix): Why should rootNode implement this? // TODO(anacrolix): Why should rootNode implement this?
func (me rootNode) Forget() { func (rn rootNode) Forget() {
me.fs.Destroy() rn.fs.Destroy()
} }
func (tfs *TorrentFS) Root() (fusefs.Node, error) { func (tfs *TorrentFS) Root() (fusefs.Node, error) {
return rootNode{tfs}, nil return rootNode{tfs}, nil
} }
func (me *TorrentFS) Destroy() { func (tfs *TorrentFS) Destroy() {
me.mu.Lock() tfs.mu.Lock()
select { select {
case <-me.destroyed: case <-tfs.destroyed:
default: default:
close(me.destroyed) close(tfs.destroyed)
} }
me.mu.Unlock() tfs.mu.Unlock()
} }
func New(cl *torrent.Client) *TorrentFS { func New(cl *torrent.Client) *TorrentFS {

4
fs/torrentfs_test.go
View File

@ -60,8 +60,8 @@ type testLayout struct {
Metainfo *metainfo.MetaInfo Metainfo *metainfo.MetaInfo
} }
func (me *testLayout) Destroy() error { func (tl *testLayout) Destroy() error {
return os.RemoveAll(me.BaseDir) return os.RemoveAll(tl.BaseDir)
} }
func newGreetingLayout() (tl testLayout, err error) { func newGreetingLayout() (tl testLayout, err error) {

22
iplist/iplist.go
View File

@ -42,16 +42,16 @@ func New(initSorted []Range) *IPList {
} }
} }
func (me *IPList) NumRanges() int { func (ipl *IPList) NumRanges() int {
if me == nil { if ipl == nil {
return 0 return 0
} }
return len(me.ranges) return len(ipl.ranges)
} }
// Return the range the given IP is in. Returns nil if no range is found. // Return the range the given IP is in. Returns nil if no range is found.
func (me *IPList) Lookup(ip net.IP) (r Range, ok bool) { func (ipl *IPList) Lookup(ip net.IP) (r Range, ok bool) {
if me == nil { if ipl == nil {
return return
} }
// TODO: Perhaps all addresses should be converted to IPv6, if the future // TODO: Perhaps all addresses should be converted to IPv6, if the future
@ -59,14 +59,14 @@ func (me *IPList) Lookup(ip net.IP) (r Range, ok bool) {
// memory for IPv4 addresses? // memory for IPv4 addresses?
v4 := ip.To4() v4 := ip.To4()
if v4 != nil { if v4 != nil {
r, ok = me.lookup(v4) r, ok = ipl.lookup(v4)
if ok { if ok {
return return
} }
} }
v6 := ip.To16() v6 := ip.To16()
if v6 != nil { if v6 != nil {
return me.lookup(v6) return ipl.lookup(v6)
} }
if v4 == nil && v6 == nil { if v4 == nil && v6 == nil {
r = Range{ r = Range{
@ -103,12 +103,12 @@ func lookup(
} }
// Return the range the given IP is in. Returns nil if no range is found. // Return the range the given IP is in. Returns nil if no range is found.
func (me *IPList) lookup(ip net.IP) (Range, bool) { func (ipl *IPList) lookup(ip net.IP) (Range, bool) {
return lookup(func(i int) net.IP { return lookup(func(i int) net.IP {
return me.ranges[i].First return ipl.ranges[i].First
}, func(i int) Range { }, func(i int) Range {
return me.ranges[i] return ipl.ranges[i]
}, len(me.ranges), ip) }, len(ipl.ranges), ip)
} }
func minifyIP(ip *net.IP) { func minifyIP(ip *net.IP) {

40
iplist/packed.go
View File

@ -21,9 +21,9 @@ const (
packedRangeLen = 20 packedRangeLen = 20
) )
func (me *IPList) WritePacked(w io.Writer) (err error) { func (ipl *IPList) WritePacked(w io.Writer) (err error) {
descOffsets := make(map[string]int64, len(me.ranges)) descOffsets := make(map[string]int64, len(ipl.ranges))
descs := make([]string, 0, len(me.ranges)) descs := make([]string, 0, len(ipl.ranges))
var nextOffset int64 var nextOffset int64
// This is a little monadic, no? // This is a little monadic, no?
write := func(b []byte, expectedLen int) { write := func(b []byte, expectedLen int) {
@ -40,9 +40,9 @@ func (me *IPList) WritePacked(w io.Writer) (err error) {
} }
} }
var b [8]byte var b [8]byte
binary.LittleEndian.PutUint64(b[:], uint64(len(me.ranges))) binary.LittleEndian.PutUint64(b[:], uint64(len(ipl.ranges)))
write(b[:], 8) write(b[:], 8)
for _, r := range me.ranges { for _, r := range ipl.ranges {
write(r.First.To4(), 4) write(r.First.To4(), 4)
write(r.Last.To4(), 4) write(r.Last.To4(), 4)
descOff, ok := descOffsets[r.Description] descOff, ok := descOffsets[r.Description]
@ -71,34 +71,34 @@ type PackedIPList []byte
var _ Ranger = PackedIPList{} var _ Ranger = PackedIPList{}
func (me PackedIPList) len() int { func (pil PackedIPList) len() int {
return int(binary.LittleEndian.Uint64(me[:8])) return int(binary.LittleEndian.Uint64(pil[:8]))
} }
func (me PackedIPList) NumRanges() int { func (pil PackedIPList) NumRanges() int {
return me.len() return pil.len()
} }
func (me PackedIPList) getFirst(i int) net.IP { func (pil PackedIPList) getFirst(i int) net.IP {
off := packedRangesOffset + packedRangeLen*i off := packedRangesOffset + packedRangeLen*i
return net.IP(me[off : off+4]) return net.IP(pil[off : off+4])
} }
func (me PackedIPList) getRange(i int) (ret Range) { func (pil PackedIPList) getRange(i int) (ret Range) {
rOff := packedRangesOffset + packedRangeLen*i rOff := packedRangesOffset + packedRangeLen*i
last := me[rOff+4 : rOff+8] last := pil[rOff+4 : rOff+8]
descOff := int(binary.LittleEndian.Uint64(me[rOff+8:])) descOff := int(binary.LittleEndian.Uint64(pil[rOff+8:]))
descLen := int(binary.LittleEndian.Uint32(me[rOff+16:])) descLen := int(binary.LittleEndian.Uint32(pil[rOff+16:]))
descOff += packedRangesOffset + packedRangeLen*me.len() descOff += packedRangesOffset + packedRangeLen*pil.len()
ret = Range{ ret = Range{
me.getFirst(i), pil.getFirst(i),
net.IP(last), net.IP(last),
string(me[descOff : descOff+descLen]), string(pil[descOff : descOff+descLen]),
} }
return return
} }
func (me PackedIPList) Lookup(ip net.IP) (r Range, ok bool) { func (pil PackedIPList) Lookup(ip net.IP) (r Range, ok bool) {
ip4 := ip.To4() ip4 := ip.To4()
if ip4 == nil { if ip4 == nil {
// If the IP list was built successfully, then it only contained IPv4 // If the IP list was built successfully, then it only contained IPv4
@ -111,7 +111,7 @@ func (me PackedIPList) Lookup(ip net.IP) (r Range, ok bool) {
} }
return return
} }
return lookup(me.getFirst, me.getRange, me.len(), ip4) return lookup(pil.getFirst, pil.getRange, pil.len(), ip4)
} }
func MMapPacked(filename string) (ret Ranger, err error) { func MMapPacked(filename string) (ret Ranger, err error) {

12
metainfo/hash.go
View File

@ -5,14 +5,14 @@ import "fmt"
// 20-byte SHA1 hash used for info and pieces. // 20-byte SHA1 hash used for info and pieces.
type Hash [20]byte type Hash [20]byte
func (me Hash) Bytes() []byte { func (h Hash) Bytes() []byte {
return me[:] return h[:]
} }
func (ih *Hash) AsString() string { func (h *Hash) AsString() string {
return string(ih[:]) return string(h[:])
} }
func (ih Hash) HexString() string { func (h Hash) HexString() string {
return fmt.Sprintf("%x", ih[:]) return fmt.Sprintf("%x", h[:])
} }

52
metainfo/metainfo.go
View File

@ -138,45 +138,45 @@ func (info *Info) GeneratePieces(open func(fi FileInfo) (io.ReadCloser, error))
return nil return nil
} }
func (me *Info) TotalLength() (ret int64) { func (info *Info) TotalLength() (ret int64) {
if me.IsDir() { if info.IsDir() {
for _, fi := range me.Files { for _, fi := range info.Files {
ret += fi.Length ret += fi.Length
} }
} else { } else {
ret = me.Length ret = info.Length
} }
return return
} }
func (me *Info) NumPieces() int { func (info *Info) NumPieces() int {
if len(me.Pieces)%20 != 0 { if len(info.Pieces)%20 != 0 {
panic(len(me.Pieces)) panic(len(info.Pieces))
} }
return len(me.Pieces) / 20 return len(info.Pieces) / 20
} }
func (me *InfoEx) Piece(i int) Piece { func (info *InfoEx) Piece(i int) Piece {
return Piece{me, i} return Piece{info, i}
} }
func (i *Info) IsDir() bool { func (info *Info) IsDir() bool {
return len(i.Files) != 0 return len(info.Files) != 0
} }
// The files field, converted up from the old single-file in the parent info // The files field, converted up from the old single-file in the parent info
// dict if necessary. This is a helper to avoid having to conditionally handle // dict if necessary. This is a helper to avoid having to conditionally handle
// single and multi-file torrent infos. // single and multi-file torrent infos.
func (i *Info) UpvertedFiles() []FileInfo { func (info *Info) UpvertedFiles() []FileInfo {
if len(i.Files) == 0 { if len(info.Files) == 0 {
return []FileInfo{{ return []FileInfo{{
Length: i.Length, Length: info.Length,
// Callers should determine that Info.Name is the basename, and // Callers should determine that Info.Name is the basename, and
// thus a regular file. // thus a regular file.
Path: nil, Path: nil,
}} }}
} }
return i.Files return info.Files
} }
// The info dictionary with its hash and raw bytes exposed, as these are // The info dictionary with its hash and raw bytes exposed, as these are
@ -192,23 +192,23 @@ var (
_ bencode.Unmarshaler = &InfoEx{} _ bencode.Unmarshaler = &InfoEx{}
) )
func (this *InfoEx) UnmarshalBencode(data []byte) error { func (ie *InfoEx) UnmarshalBencode(data []byte) error {
this.Bytes = append(make([]byte, 0, len(data)), data...) ie.Bytes = append(make([]byte, 0, len(data)), data...)
h := sha1.New() h := sha1.New()
_, err := h.Write(this.Bytes) _, err := h.Write(ie.Bytes)
if err != nil { if err != nil {
panic(err) panic(err)
} }
this.Hash = new(Hash) ie.Hash = new(Hash)
missinggo.CopyExact(this.Hash, h.Sum(nil)) missinggo.CopyExact(ie.Hash, h.Sum(nil))
return bencode.Unmarshal(data, &this.Info) return bencode.Unmarshal(data, &ie.Info)
} }
func (this InfoEx) MarshalBencode() ([]byte, error) { func (ie InfoEx) MarshalBencode() ([]byte, error) {
if this.Bytes != nil { if ie.Bytes != nil {
return this.Bytes, nil return ie.Bytes, nil
} }
return bencode.Marshal(&this.Info) return bencode.Marshal(&ie.Info)
} }
type MetaInfo struct { type MetaInfo struct {

6
metainfo/nodes.go
View File

@ -14,7 +14,7 @@ var (
_ bencode.Unmarshaler = new(Node) _ bencode.Unmarshaler = new(Node)
) )
func (me *Node) UnmarshalBencode(b []byte) (err error) { func (n *Node) UnmarshalBencode(b []byte) (err error) {
var iface interface{} var iface interface{}
err = bencode.Unmarshal(b, &iface) err = bencode.Unmarshal(b, &iface)
if err != nil { if err != nil {
@ -22,7 +22,7 @@ func (me *Node) UnmarshalBencode(b []byte) (err error) {
} }
switch v := iface.(type) { switch v := iface.(type) {
case string: case string:
*me = Node(v) *n = Node(v)
case []interface{}: case []interface{}:
func() { func() {
defer func() { defer func() {
@ -31,7 +31,7 @@ func (me *Node) UnmarshalBencode(b []byte) (err error) {
err = r.(error) err = r.(error)
} }
}() }()
*me = Node(net.JoinHostPort(v[0].(string), strconv.FormatInt(v[1].(int64), 10))) *n = Node(net.JoinHostPort(v[0].(string), strconv.FormatInt(v[1].(int64), 10)))
}() }()
default: default:
err = fmt.Errorf("unsupported type: %T", iface) err = fmt.Errorf("unsupported type: %T", iface)

16
metainfo/piece.go
View File

@ -7,18 +7,18 @@ type Piece struct {
i int i int
} }
func (me Piece) Length() int64 { func (p Piece) Length() int64 {
if me.i == me.Info.NumPieces()-1 { if p.i == p.Info.NumPieces()-1 {
return me.Info.TotalLength() - int64(me.i)*me.Info.PieceLength return p.Info.TotalLength() - int64(p.i)*p.Info.PieceLength
} }
return me.Info.PieceLength return p.Info.PieceLength
} }
func (me Piece) Offset() int64 { func (p Piece) Offset() int64 {
return int64(me.i) * me.Info.PieceLength return int64(p.i) * p.Info.PieceLength
} }
func (me Piece) Hash() (ret Hash) { func (p Piece) Hash() (ret Hash) {
missinggo.CopyExact(&ret, me.Info.Pieces[me.i*20:(me.i+1)*20]) missinggo.CopyExact(&ret, p.Info.Pieces[p.i*20:(p.i+1)*20])
return return
} }

24
mmap_span/mmap_span.go
View File

@ -11,20 +11,20 @@ type segment struct {
*mmap.MMap *mmap.MMap
} }
func (me segment) Size() int64 { func (s segment) Size() int64 {
return int64(len(*me.MMap)) return int64(len(*s.MMap))
} }
type MMapSpan struct { type MMapSpan struct {
span span
} }
func (me *MMapSpan) Append(mmap mmap.MMap) { func (ms *MMapSpan) Append(mmap mmap.MMap) {
me.span = append(me.span, segment{&mmap}) ms.span = append(ms.span, segment{&mmap})
} }
func (me MMapSpan) Close() error { func (ms MMapSpan) Close() error {
for _, mMap := range me.span { for _, mMap := range ms.span {
err := mMap.(segment).Unmap() err := mMap.(segment).Unmap()
if err != nil { if err != nil {
log.Print(err) log.Print(err)
@ -33,15 +33,15 @@ func (me MMapSpan) Close() error {
return nil return nil
} }
func (me MMapSpan) Size() (ret int64) { func (ms MMapSpan) Size() (ret int64) {
for _, seg := range me.span { for _, seg := range ms.span {
ret += seg.Size() ret += seg.Size()
} }
return return
} }
func (me MMapSpan) ReadAt(p []byte, off int64) (n int, err error) { func (ms MMapSpan) ReadAt(p []byte, off int64) (n int, err error) {
me.ApplyTo(off, func(intervalOffset int64, interval sizer) (stop bool) { ms.ApplyTo(off, func(intervalOffset int64, interval sizer) (stop bool) {
_n := copy(p, (*interval.(segment).MMap)[intervalOffset:]) _n := copy(p, (*interval.(segment).MMap)[intervalOffset:])
p = p[_n:] p = p[_n:]
n += _n n += _n
@ -53,8 +53,8 @@ func (me MMapSpan) ReadAt(p []byte, off int64) (n int, err error) {
return return
} }
func (me MMapSpan) WriteAt(p []byte, off int64) (n int, err error) { func (ms MMapSpan) WriteAt(p []byte, off int64) (n int, err error) {
me.ApplyTo(off, func(iOff int64, i sizer) (stop bool) { ms.ApplyTo(off, func(iOff int64, i sizer) (stop bool) {
mMap := i.(segment) mMap := i.(segment)
_n := copy((*mMap.MMap)[iOff:], p) _n := copy((*mMap.MMap)[iOff:], p)
// err = mMap.Sync(gommap.MS_ASYNC) // err = mMap.Sync(gommap.MS_ASYNC)

4
mmap_span/span.go
View File

@ -6,8 +6,8 @@ type sizer interface {
type span []sizer type span []sizer
func (me span) ApplyTo(off int64, f func(int64, sizer) (stop bool)) { func (s span) ApplyTo(off int64, f func(int64, sizer) (stop bool)) {
for _, interval := range me { for _, interval := range s {
iSize := interval.Size() iSize := interval.Size()
if off >= iSize { if off >= iSize {
off -= iSize off -= iSize

14
mse/mse.go
View File

@ -87,10 +87,10 @@ type cipherReader struct {
r io.Reader r io.Reader
} }
func (me *cipherReader) Read(b []byte) (n int, err error) { func (cr *cipherReader) Read(b []byte) (n int, err error) {
be := make([]byte, len(b)) be := make([]byte, len(b))
n, err = me.r.Read(be) n, err = cr.r.Read(be)
me.c.XORKeyStream(b[:n], be[:n]) cr.c.XORKeyStream(b[:n], be[:n])
return return
} }
@ -103,14 +103,14 @@ type cipherWriter struct {
w io.Writer w io.Writer
} }
func (me *cipherWriter) Write(b []byte) (n int, err error) { func (cr *cipherWriter) Write(b []byte) (n int, err error) {
be := make([]byte, len(b)) be := make([]byte, len(b))
me.c.XORKeyStream(be, b) cr.c.XORKeyStream(be, b)
n, err = me.w.Write(be) n, err = cr.w.Write(be)
if n != len(be) { if n != len(be) {
// The cipher will have advanced beyond the callers stream position. // The cipher will have advanced beyond the callers stream position.
// We can't use the cipher anymore. // We can't use the cipher anymore.
me.c = nil cr.c = nil
} }
return return
} }

6
mse/mse_test.go
View File

@ -109,9 +109,9 @@ type trackReader struct {
n int64 n int64
} }
func (me *trackReader) Read(b []byte) (n int, err error) { func (tr *trackReader) Read(b []byte) (n int, err error) {
n, err = me.r.Read(b) n, err = tr.r.Read(b)
me.n += int64(n) tr.n += int64(n)
return return
} }

6
piece.go
View File

@ -14,9 +14,9 @@ import (
type piecePriority byte type piecePriority byte
func (me *piecePriority) Raise(maybe piecePriority) { func (pp *piecePriority) Raise(maybe piecePriority) {
if maybe > *me { if maybe > *pp {
*me = maybe *pp = maybe
} }
} }

44
storage/file.go
View File

@ -21,28 +21,28 @@ func NewFile(baseDir string) I {
} }
} }
func (me *fileStorage) OpenTorrent(info *metainfo.InfoEx) (Torrent, error) { func (fs *fileStorage) OpenTorrent(info *metainfo.InfoEx) (Torrent, error) {
return fileTorrentStorage{me}, nil return fileTorrentStorage{fs}, nil
} }
type fileTorrentStorage struct { type fileTorrentStorage struct {
*fileStorage *fileStorage
} }
func (me *fileStorage) Piece(p metainfo.Piece) Piece { func (fs *fileStorage) Piece(p metainfo.Piece) Piece {
_io := &fileStorageTorrent{ _io := &fileStorageTorrent{
p.Info, p.Info,
me.baseDir, fs.baseDir,
} }
return &fileStoragePiece{ return &fileStoragePiece{
me, fs,
p, p,
missinggo.NewSectionWriter(_io, p.Offset(), p.Length()), missinggo.NewSectionWriter(_io, p.Offset(), p.Length()),
io.NewSectionReader(_io, p.Offset(), p.Length()), io.NewSectionReader(_io, p.Offset(), p.Length()),
} }
} }
func (me *fileStorage) Close() error { func (fs *fileStorage) Close() error {
return nil return nil
} }
@ -53,15 +53,15 @@ type fileStoragePiece struct {
io.ReaderAt io.ReaderAt
} }
func (me *fileStoragePiece) GetIsComplete() bool { func (fs *fileStoragePiece) GetIsComplete() bool {
return me.completed[me.p.Hash()] return fs.completed[fs.p.Hash()]
} }
func (me *fileStoragePiece) MarkComplete() error { func (fs *fileStoragePiece) MarkComplete() error {
if me.completed == nil { if fs.completed == nil {
me.completed = make(map[[20]byte]bool) fs.completed = make(map[[20]byte]bool)
} }
me.completed[me.p.Hash()] = true fs.completed[fs.p.Hash()] = true
return nil return nil
} }
@ -71,8 +71,8 @@ type fileStorageTorrent struct {
} }
// Returns EOF on short or missing file. // Returns EOF on short or missing file.
func (me *fileStorageTorrent) readFileAt(fi metainfo.FileInfo, b []byte, off int64) (n int, err error) { func (fst *fileStorageTorrent) readFileAt(fi metainfo.FileInfo, b []byte, off int64) (n int, err error) {
f, err := os.Open(me.fileInfoName(fi)) f, err := os.Open(fst.fileInfoName(fi))
if os.IsNotExist(err) { if os.IsNotExist(err) {
// File missing is treated the same as a short file. // File missing is treated the same as a short file.
err = io.EOF err = io.EOF
@ -100,10 +100,10 @@ func (me *fileStorageTorrent) readFileAt(fi metainfo.FileInfo, b []byte, off int
} }
// Only returns EOF at the end of the torrent. Premature EOF is ErrUnexpectedEOF. // Only returns EOF at the end of the torrent. Premature EOF is ErrUnexpectedEOF.
func (me *fileStorageTorrent) ReadAt(b []byte, off int64) (n int, err error) { func (fst *fileStorageTorrent) ReadAt(b []byte, off int64) (n int, err error) {
for _, fi := range me.info.UpvertedFiles() { for _, fi := range fst.info.UpvertedFiles() {
for off < fi.Length { for off < fi.Length {
n1, err1 := me.readFileAt(fi, b, off) n1, err1 := fst.readFileAt(fi, b, off)
n += n1 n += n1
off += int64(n1) off += int64(n1)
b = b[n1:] b = b[n1:]
@ -128,8 +128,8 @@ func (me *fileStorageTorrent) ReadAt(b []byte, off int64) (n int, err error) {
return return
} }
func (me *fileStorageTorrent) WriteAt(p []byte, off int64) (n int, err error) { func (fst *fileStorageTorrent) WriteAt(p []byte, off int64) (n int, err error) {
for _, fi := range me.info.UpvertedFiles() { for _, fi := range fst.info.UpvertedFiles() {
if off >= fi.Length { if off >= fi.Length {
off -= fi.Length off -= fi.Length
continue continue
@ -138,7 +138,7 @@ func (me *fileStorageTorrent) WriteAt(p []byte, off int64) (n int, err error) {
if int64(n1) > fi.Length-off { if int64(n1) > fi.Length-off {
n1 = int(fi.Length - off) n1 = int(fi.Length - off)
} }
name := me.fileInfoName(fi) name := fst.fileInfoName(fi)
os.MkdirAll(filepath.Dir(name), 0770) os.MkdirAll(filepath.Dir(name), 0770)
var f *os.File var f *os.File
f, err = os.OpenFile(name, os.O_WRONLY|os.O_CREATE, 0660) f, err = os.OpenFile(name, os.O_WRONLY|os.O_CREATE, 0660)
@ -160,6 +160,6 @@ func (me *fileStorageTorrent) WriteAt(p []byte, off int64) (n int, err error) {
return return
} }
func (me *fileStorageTorrent) fileInfoName(fi metainfo.FileInfo) string { func (fst *fileStorageTorrent) fileInfoName(fi metainfo.FileInfo) string {
return filepath.Join(append([]string{me.baseDir, me.info.Name}, fi.Path...)...) return filepath.Join(append([]string{fst.baseDir, fst.info.Name}, fi.Path...)...)
} }

28
storage/mmap.go
View File

@ -23,8 +23,8 @@ func NewMMap(baseDir string) I {
} }
} }
func (me *mmapStorage) OpenTorrent(info *metainfo.InfoEx) (t Torrent, err error) { func (s *mmapStorage) OpenTorrent(info *metainfo.InfoEx) (t Torrent, err error) {
span, err := MMapTorrent(&info.Info, me.baseDir) span, err := MMapTorrent(&info.Info, s.baseDir)
t = &mmapTorrentStorage{ t = &mmapTorrentStorage{
span: span, span: span,
} }
@ -36,17 +36,17 @@ type mmapTorrentStorage struct {
completed map[metainfo.Hash]bool completed map[metainfo.Hash]bool
} }
func (me *mmapTorrentStorage) Piece(p metainfo.Piece) Piece { func (ts *mmapTorrentStorage) Piece(p metainfo.Piece) Piece {
return mmapStoragePiece{ return mmapStoragePiece{
storage: me, storage: ts,
p: p, p: p,
ReaderAt: io.NewSectionReader(me.span, p.Offset(), p.Length()), ReaderAt: io.NewSectionReader(ts.span, p.Offset(), p.Length()),
WriterAt: missinggo.NewSectionWriter(me.span, p.Offset(), p.Length()), WriterAt: missinggo.NewSectionWriter(ts.span, p.Offset(), p.Length()),
} }
} }
func (me *mmapTorrentStorage) Close() error { func (ts *mmapTorrentStorage) Close() error {
me.span.Close() ts.span.Close()
return nil return nil
} }
@ -57,15 +57,15 @@ type mmapStoragePiece struct {
io.WriterAt io.WriterAt
} }
func (me mmapStoragePiece) GetIsComplete() bool { func (sp mmapStoragePiece) GetIsComplete() bool {
return me.storage.completed[me.p.Hash()] return sp.storage.completed[sp.p.Hash()]
} }
func (me mmapStoragePiece) MarkComplete() error { func (sp mmapStoragePiece) MarkComplete() error {
if me.storage.completed == nil { if sp.storage.completed == nil {
me.storage.completed = make(map[metainfo.Hash]bool) sp.storage.completed = make(map[metainfo.Hash]bool)
} }
me.storage.completed[me.p.Hash()] = true sp.storage.completed[sp.p.Hash()] = true
return nil return nil
} }

52
storage/piece_file.go
View File

@ -25,16 +25,16 @@ type pieceFileTorrentStorage struct {
s *pieceFileStorage s *pieceFileStorage
} }
func (me *pieceFileStorage) OpenTorrent(info *metainfo.InfoEx) (Torrent, error) { func (s *pieceFileStorage) OpenTorrent(info *metainfo.InfoEx) (Torrent, error) {
return &pieceFileTorrentStorage{me}, nil return &pieceFileTorrentStorage{s}, nil
} }
func (me *pieceFileTorrentStorage) Close() error { func (s *pieceFileTorrentStorage) Close() error {
return nil return nil
} }
func (me *pieceFileTorrentStorage) Piece(p metainfo.Piece) Piece { func (s *pieceFileTorrentStorage) Piece(p metainfo.Piece) Piece {
return pieceFileTorrentStoragePiece{me, p, me.s.fs} return pieceFileTorrentStoragePiece{s, p, s.s.fs}
} }
type pieceFileTorrentStoragePiece struct { type pieceFileTorrentStoragePiece struct {
@ -43,52 +43,52 @@ type pieceFileTorrentStoragePiece struct {
fs missinggo.FileStore fs missinggo.FileStore
} }
func (me pieceFileTorrentStoragePiece) completedPath() string { func (s pieceFileTorrentStoragePiece) completedPath() string {
return path.Join("completed", me.p.Hash().HexString()) return path.Join("completed", s.p.Hash().HexString())
} }
func (me pieceFileTorrentStoragePiece) incompletePath() string { func (s pieceFileTorrentStoragePiece) incompletePath() string {
return path.Join("incomplete", me.p.Hash().HexString()) return path.Join("incomplete", s.p.Hash().HexString())
} }
func (me pieceFileTorrentStoragePiece) GetIsComplete() bool { func (s pieceFileTorrentStoragePiece) GetIsComplete() bool {
fi, err := me.fs.Stat(me.completedPath()) fi, err := s.fs.Stat(s.completedPath())
return err == nil && fi.Size() == me.p.Length() return err == nil && fi.Size() == s.p.Length()
} }
func (me pieceFileTorrentStoragePiece) MarkComplete() error { func (s pieceFileTorrentStoragePiece) MarkComplete() error {
return me.fs.Rename(me.incompletePath(), me.completedPath()) return s.fs.Rename(s.incompletePath(), s.completedPath())
} }
func (me pieceFileTorrentStoragePiece) openFile() (f missinggo.File, err error) { func (s pieceFileTorrentStoragePiece) openFile() (f missinggo.File, err error) {
f, err = me.fs.OpenFile(me.completedPath(), os.O_RDONLY) f, err = s.fs.OpenFile(s.completedPath(), os.O_RDONLY)
if err == nil { if err == nil {
var fi os.FileInfo var fi os.FileInfo
fi, err = f.Stat() fi, err = f.Stat()
if err == nil && fi.Size() == me.p.Length() { if err == nil && fi.Size() == s.p.Length() {
return return
} }
f.Close() f.Close()
} else if !os.IsNotExist(err) { } else if !os.IsNotExist(err) {
return return
} }
f, err = me.fs.OpenFile(me.incompletePath(), os.O_RDONLY) f, err = s.fs.OpenFile(s.incompletePath(), os.O_RDONLY)
if os.IsNotExist(err) { if os.IsNotExist(err) {
err = io.ErrUnexpectedEOF err = io.ErrUnexpectedEOF
} }
return return
} }
func (me pieceFileTorrentStoragePiece) ReadAt(b []byte, off int64) (n int, err error) { func (s pieceFileTorrentStoragePiece) ReadAt(b []byte, off int64) (n int, err error) {
f, err := me.openFile() f, err := s.openFile()
if err != nil { if err != nil {
return return
} }
defer f.Close() defer f.Close()
missinggo.LimitLen(&b, me.p.Length()-off) missinggo.LimitLen(&b, s.p.Length()-off)
n, err = f.ReadAt(b, off) n, err = f.ReadAt(b, off)
off += int64(n) off += int64(n)
if off >= me.p.Length() { if off >= s.p.Length() {
err = io.EOF err = io.EOF
} else if err == io.EOF { } else if err == io.EOF {
err = io.ErrUnexpectedEOF err = io.ErrUnexpectedEOF
@ -96,16 +96,16 @@ func (me pieceFileTorrentStoragePiece) ReadAt(b []byte, off int64) (n int, err e
return return
} }
func (me pieceFileTorrentStoragePiece) WriteAt(b []byte, off int64) (n int, err error) { func (s pieceFileTorrentStoragePiece) WriteAt(b []byte, off int64) (n int, err error) {
if me.GetIsComplete() { if s.GetIsComplete() {
err = errors.New("piece completed") err = errors.New("piece completed")
return return
} }
f, err := me.fs.OpenFile(me.incompletePath(), os.O_WRONLY|os.O_CREATE) f, err := s.fs.OpenFile(s.incompletePath(), os.O_WRONLY|os.O_CREATE)
if err != nil { if err != nil {
return return
} }
defer f.Close() defer f.Close()
missinggo.LimitLen(&b, me.p.Length()-off) missinggo.LimitLen(&b, s.p.Length()-off)
return f.WriteAt(b, off) return f.WriteAt(b, off)
} }

14
torrent.go
View File

@ -670,9 +670,9 @@ func (t *Torrent) haveAllPieces() bool {
return t.completedPieces.Len() == t.numPieces() return t.completedPieces.Len() == t.numPieces()
} }
func (me *Torrent) haveAnyPieces() bool { func (t *Torrent) haveAnyPieces() bool {
for i := range me.pieces { for i := range t.pieces {
if me.pieceComplete(i) { if t.pieceComplete(i) {
return true return true
} }
} }
@ -701,9 +701,8 @@ func chunkIndex(cs chunkSpec, chunkSize pp.Integer) int {
return int(cs.Begin / chunkSize) return int(cs.Begin / chunkSize)
} }
// TODO: This should probably be called wantPiece. func (t *Torrent) wantPiece(r request) bool {
func (t *Torrent) wantChunk(r request) bool { if !t.wantPieceIndex(int(r.Index)) {
if !t.wantPiece(int(r.Index)) {
return false return false
} }
if t.pieces[r.Index].pendingChunk(r.chunkSpec, t.chunkSize) { if t.pieces[r.Index].pendingChunk(r.chunkSpec, t.chunkSize) {
@ -714,8 +713,7 @@ func (t *Torrent) wantChunk(r request) bool {
return false return false
} }
// TODO: This should be called wantPieceIndex. func (t *Torrent) wantPieceIndex(index int) bool {
func (t *Torrent) wantPiece(index int) bool {
if !t.haveInfo() { if !t.haveInfo() {
return false return false
} }

16
tracker/http.go
View File

@ -56,9 +56,9 @@ func (r *httpResponse) UnmarshalPeers() (ret []Peer, err error) {
return return
} }
func (me *httpClient) Announce(ar *AnnounceRequest) (ret AnnounceResponse, err error) { func (c *httpClient) Announce(ar *AnnounceRequest) (ret AnnounceResponse, err error) {
// retain query parameters from announce URL // retain query parameters from announce URL
q := me.url.Query() q := c.url.Query()
q.Set("info_hash", string(ar.InfoHash[:])) q.Set("info_hash", string(ar.InfoHash[:]))
q.Set("peer_id", string(ar.PeerId[:])) q.Set("peer_id", string(ar.PeerId[:]))
@ -73,7 +73,7 @@ func (me *httpClient) Announce(ar *AnnounceRequest) (ret AnnounceResponse, err e
q.Set("compact", "1") q.Set("compact", "1")
// According to https://wiki.vuze.com/w/Message_Stream_Encryption. // According to https://wiki.vuze.com/w/Message_Stream_Encryption.
q.Set("supportcrypto", "1") q.Set("supportcrypto", "1")
var reqURL url.URL = me.url var reqURL url.URL = c.url
reqURL.RawQuery = q.Encode() reqURL.RawQuery = q.Encode()
resp, err := http.Get(reqURL.String()) resp, err := http.Get(reqURL.String())
if err != nil { if err != nil {
@ -103,15 +103,15 @@ func (me *httpClient) Announce(ar *AnnounceRequest) (ret AnnounceResponse, err e
return return
} }
func (me *httpClient) Connect() error { func (c *httpClient) Connect() error {
// HTTP trackers do not require a connecting handshake. // HTTP trackers do not require a connecting handshake.
return nil return nil
} }
func (me *httpClient) String() string { func (c *httpClient) String() string {
return me.URL() return c.URL()
} }
func (me *httpClient) URL() string { func (c *httpClient) URL() string {
return me.url.String() return c.url.String()
} }

30
tracker/server.go
View File

@ -34,27 +34,27 @@ func marshal(parts ...interface{}) (ret []byte, err error) {
return return
} }
func (me *server) respond(addr net.Addr, rh ResponseHeader, parts ...interface{}) (err error) { func (s *server) respond(addr net.Addr, rh ResponseHeader, parts ...interface{}) (err error) {
b, err := marshal(append([]interface{}{rh}, parts...)...) b, err := marshal(append([]interface{}{rh}, parts...)...)
if err != nil { if err != nil {
return return
} }
_, err = me.pc.WriteTo(b, addr) _, err = s.pc.WriteTo(b, addr)
return return
} }
func (me *server) newConn() (ret int64) { func (s *server) newConn() (ret int64) {
ret = rand.Int63() ret = rand.Int63()
if me.conns == nil { if s.conns == nil {
me.conns = make(map[int64]struct{}) s.conns = make(map[int64]struct{})
} }
me.conns[ret] = struct{}{} s.conns[ret] = struct{}{}
return return
} }
func (me *server) serveOne() (err error) { func (s *server) serveOne() (err error) {
b := make([]byte, 0x10000) b := make([]byte, 0x10000)
n, addr, err := me.pc.ReadFrom(b) n, addr, err := s.pc.ReadFrom(b)
if err != nil { if err != nil {
return return
} }
@ -69,8 +69,8 @@ func (me *server) serveOne() (err error) {
if h.ConnectionId != connectRequestConnectionId { if h.ConnectionId != connectRequestConnectionId {
return return
} }
connId := me.newConn() connId := s.newConn()
err = me.respond(addr, ResponseHeader{ err = s.respond(addr, ResponseHeader{
ActionConnect, ActionConnect,
h.TransactionId, h.TransactionId,
}, ConnectionResponse{ }, ConnectionResponse{
@ -78,8 +78,8 @@ func (me *server) serveOne() (err error) {
}) })
return return
case ActionAnnounce: case ActionAnnounce:
if _, ok := me.conns[h.ConnectionId]; !ok { if _, ok := s.conns[h.ConnectionId]; !ok {
me.respond(addr, ResponseHeader{ s.respond(addr, ResponseHeader{
TransactionId: h.TransactionId, TransactionId: h.TransactionId,
Action: ActionError, Action: ActionError,
}, []byte("not connected")) }, []byte("not connected"))
@ -90,12 +90,12 @@ func (me *server) serveOne() (err error) {
if err != nil { if err != nil {
return return
} }
t := me.t[ar.InfoHash] t := s.t[ar.InfoHash]
b, err = t.Peers.MarshalBinary() b, err = t.Peers.MarshalBinary()
if err != nil { if err != nil {
panic(err) panic(err)
} }
err = me.respond(addr, ResponseHeader{ err = s.respond(addr, ResponseHeader{
TransactionId: h.TransactionId, TransactionId: h.TransactionId,
Action: ActionAnnounce, Action: ActionAnnounce,
}, AnnounceResponseHeader{ }, AnnounceResponseHeader{
@ -106,7 +106,7 @@ func (me *server) serveOne() (err error) {
return return
default: default:
err = fmt.Errorf("unhandled action: %d", h.Action) err = fmt.Errorf("unhandled action: %d", h.Action)
me.respond(addr, ResponseHeader{ s.respond(addr, ResponseHeader{
TransactionId: h.TransactionId, TransactionId: h.TransactionId,
Action: ActionError, Action: ActionError,
}, []byte("unhandled action")) }, []byte("unhandled action"))

4
tracker/tracker.go
View File

@ -29,9 +29,9 @@ type AnnounceResponse struct {
type AnnounceEvent int32 type AnnounceEvent int32
func (me AnnounceEvent) String() string { func (e AnnounceEvent) String() string {
// See BEP 3, "event". // See BEP 3, "event".
return []string{"empty", "completed", "started", "stopped"}[me] return []string{"empty", "completed", "started", "stopped"}[e]
} }
type Peer struct { type Peer struct {

6
tracker/udp.go
View File

@ -93,9 +93,9 @@ type udpClient struct {
url url.URL url url.URL
} }
func (me *udpClient) Close() error { func (c *udpClient) Close() error {
if me.socket != nil { if c.socket != nil {
return me.socket.Close() return c.socket.Close()
} }
return nil return nil
} }

38
util/dirwatch/dirwatch.go
View File

@ -42,24 +42,24 @@ type Instance struct {
dirState map[metainfo.Hash]entity dirState map[metainfo.Hash]entity
} }
func (me *Instance) Close() { func (i *Instance) Close() {
me.w.Close() i.w.Close()
} }
func (me *Instance) handleEvents() { func (i *Instance) handleEvents() {
defer close(me.Events) defer close(i.Events)
for e := range me.w.Events { for e := range i.w.Events {
log.Printf("event: %s", e) log.Printf("event: %s", e)
if e.Op == fsnotify.Write { if e.Op == fsnotify.Write {
// TODO: Special treatment as an existing torrent may have changed. // TODO: Special treatment as an existing torrent may have changed.
} else { } else {
me.refresh() i.refresh()
} }
} }
} }
func (me *Instance) handleErrors() { func (i *Instance) handleErrors() {
for err := range me.w.Errors { for err := range i.w.Errors {
log.Printf("error in torrent directory watcher: %s", err) log.Printf("error in torrent directory watcher: %s", err)
} }
} }
@ -150,15 +150,15 @@ func magnetFileURIs(name string) (uris []string, err error) {
return return
} }
func (me *Instance) torrentRemoved(ih metainfo.Hash) { func (i *Instance) torrentRemoved(ih metainfo.Hash) {
me.Events <- Event{ i.Events <- Event{
InfoHash: ih, InfoHash: ih,
Change: Removed, Change: Removed,
} }
} }
func (me *Instance) torrentAdded(e entity) { func (i *Instance) torrentAdded(e entity) {
me.Events <- Event{ i.Events <- Event{
InfoHash: e.Hash, InfoHash: e.Hash,
Change: Added, Change: Added,
MagnetURI: e.MagnetURI, MagnetURI: e.MagnetURI,
@ -166,13 +166,13 @@ func (me *Instance) torrentAdded(e entity) {
} }
} }
func (me *Instance) refresh() { func (i *Instance) refresh() {
_new := scanDir(me.dirName) _new := scanDir(i.dirName)
old := me.dirState old := i.dirState
for ih, _ := range old { for ih, _ := range old {
_, ok := _new[ih] _, ok := _new[ih]
if !ok { if !ok {
me.torrentRemoved(ih) i.torrentRemoved(ih)
} }
} }
for ih, newE := range _new { for ih, newE := range _new {
@ -181,11 +181,11 @@ func (me *Instance) refresh() {
if newE == oldE { if newE == oldE {
continue continue
} }
me.torrentRemoved(ih) i.torrentRemoved(ih)
} }
me.torrentAdded(newE) i.torrentAdded(newE)
} }
me.dirState = _new i.dirState = _new
} }
func New(dirName string) (i *Instance, err error) { func New(dirName string) (i *Instance, err error) {

32
util/types.go
View File

@ -22,19 +22,19 @@ var (
) )
// This allows bencode.Unmarshal to do better than a string or []byte. // This allows bencode.Unmarshal to do better than a string or []byte.
func (me *CompactIPv4Peers) UnmarshalBencode(b []byte) (err error) { func (cps *CompactIPv4Peers) UnmarshalBencode(b []byte) (err error) {
var bb []byte var bb []byte
err = bencode.Unmarshal(b, &bb) err = bencode.Unmarshal(b, &bb)
if err != nil { if err != nil {
return return
} }
*me, err = UnmarshalIPv4CompactPeers(bb) *cps, err = UnmarshalIPv4CompactPeers(bb)
return return
} }
func (me CompactIPv4Peers) MarshalBinary() (ret []byte, err error) { func (cps CompactIPv4Peers) MarshalBinary() (ret []byte, err error) {
ret = make([]byte, len(me)*6) ret = make([]byte, len(cps)*6)
for i, cp := range me { for i, cp := range cps {
copy(ret[6*i:], cp.IP.To4()) copy(ret[6*i:], cp.IP.To4())
binary.BigEndian.PutUint16(ret[6*i+4:], uint16(cp.Port)) binary.BigEndian.PutUint16(ret[6*i+4:], uint16(cp.Port))
} }
@ -52,39 +52,39 @@ var (
_ bencode.Unmarshaler = &CompactPeer{} _ bencode.Unmarshaler = &CompactPeer{}
) )
func (me CompactPeer) MarshalBencode() (ret []byte, err error) { func (cp CompactPeer) MarshalBencode() (ret []byte, err error) {
ip := me.IP ip := cp.IP
if ip4 := ip.To4(); ip4 != nil { if ip4 := ip.To4(); ip4 != nil {
ip = ip4 ip = ip4
} }
ret = make([]byte, len(ip)+2) ret = make([]byte, len(ip)+2)
copy(ret, ip) copy(ret, ip)
binary.BigEndian.PutUint16(ret[len(ip):], uint16(me.Port)) binary.BigEndian.PutUint16(ret[len(ip):], uint16(cp.Port))
return bencode.Marshal(ret) return bencode.Marshal(ret)
} }
func (me *CompactPeer) UnmarshalBinary(b []byte) error { func (cp *CompactPeer) UnmarshalBinary(b []byte) error {
switch len(b) { switch len(b) {
case 18: case 18:
me.IP = make([]byte, 16) cp.IP = make([]byte, 16)
case 6: case 6:
me.IP = make([]byte, 4) cp.IP = make([]byte, 4)
default: default:
return fmt.Errorf("bad compact peer string: %q", b) return fmt.Errorf("bad compact peer string: %q", b)
} }
copy(me.IP, b) copy(cp.IP, b)
b = b[len(me.IP):] b = b[len(cp.IP):]
me.Port = int(binary.BigEndian.Uint16(b)) cp.Port = int(binary.BigEndian.Uint16(b))
return nil return nil
} }
func (me *CompactPeer) UnmarshalBencode(b []byte) (err error) { func (cp *CompactPeer) UnmarshalBencode(b []byte) (err error) {
var _b []byte var _b []byte
err = bencode.Unmarshal(b, &_b) err = bencode.Unmarshal(b, &_b)
if err != nil { if err != nil {
return return
} }
return me.UnmarshalBinary(_b) return cp.UnmarshalBinary(_b)
} }
func UnmarshalIPv4CompactPeers(b []byte) (ret []CompactPeer, err error) { func UnmarshalIPv4CompactPeers(b []byte) (ret []CompactPeer, err error) {

38
worst_conns.go
View File

@ -11,19 +11,19 @@ type worstConns struct {
cl *Client cl *Client
} }
func (me *worstConns) Len() int { return len(me.c) } func (wc *worstConns) Len() int { return len(wc.c) }
func (me *worstConns) Swap(i, j int) { me.c[i], me.c[j] = me.c[j], me.c[i] } func (wc *worstConns) Swap(i, j int) { wc.c[i], wc.c[j] = wc.c[j], wc.c[i] }
func (me *worstConns) Pop() (ret interface{}) { func (wc *worstConns) Pop() (ret interface{}) {
old := me.c old := wc.c
n := len(old) n := len(old)
ret = old[n-1] ret = old[n-1]
me.c = old[:n-1] wc.c = old[:n-1]
return return
} }
func (me *worstConns) Push(x interface{}) { func (wc *worstConns) Push(x interface{}) {
me.c = append(me.c, x.(*connection)) wc.c = append(wc.c, x.(*connection))
} }
type worstConnsSortKey struct { type worstConnsSortKey struct {
@ -32,20 +32,20 @@ type worstConnsSortKey struct {
connected time.Time connected time.Time
} }
func (me worstConnsSortKey) Less(other worstConnsSortKey) bool { func (wc worstConnsSortKey) Less(other worstConnsSortKey) bool {
if me.useful != other.useful { if wc.useful != other.useful {
return !me.useful return !wc.useful
} }
if !me.lastHelpful.Equal(other.lastHelpful) { if !wc.lastHelpful.Equal(other.lastHelpful) {
return me.lastHelpful.Before(other.lastHelpful) return wc.lastHelpful.Before(other.lastHelpful)
} }
return me.connected.Before(other.connected) return wc.connected.Before(other.connected)
} }
func (me *worstConns) key(i int) (key worstConnsSortKey) { func (wc *worstConns) key(i int) (key worstConnsSortKey) {
c := me.c[i] c := wc.c[i]
key.useful = me.cl.usefulConn(me.t, c) key.useful = wc.cl.usefulConn(wc.t, c)
if me.cl.seeding(me.t) { if wc.cl.seeding(wc.t) {
key.lastHelpful = c.lastChunkSent key.lastHelpful = c.lastChunkSent
} }
// Intentionally consider the last time a chunk was received when seeding, // Intentionally consider the last time a chunk was received when seeding,
@ -57,6 +57,6 @@ func (me *worstConns) key(i int) (key worstConnsSortKey) {
return return
} }
func (me worstConns) Less(i, j int) bool { func (wc worstConns) Less(i, j int) bool {
return me.key(i).Less(me.key(j)) return wc.key(i).Less(wc.key(j))
} }