<commit_before>package rain

import (
	"bytes"
	"crypto/sha1"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"sync"
	"time"

	"github.com/cenkalti/rain/internal/bitfield"
	"github.com/cenkalti/rain/internal/logger"
	"github.com/cenkalti/rain/internal/protocol"
)

// All current implementations use 2^14 (16 kiB), and close connections which request an amount greater than that.
const blockSize = 16 * 1024

type peerConn struct {
	conn         net.Conn
	disconnected chan struct{} // will be closed when peer disconnects

	unchokeM       sync.Mutex    // protects unchokeC
	unchokeC       chan struct{} // will be closed when and "unchoke" message is received
	onceInterested sync.Once     // for sending "interested" message only once

	amChoking      bool // this client is choking the peer
	amInterested   bool // this client is interested in the peer
	peerChoking    bool // peer is choking this client
	peerInterested bool // peer is interested in this client
	// peerRequests   map[uint64]bool      // What remote peer requested
	// ourRequests    map[uint64]time.Time // What we requested, when we requested it

	log logger.Logger
}

func newPeerConn(conn net.Conn) *peerConn {
	return &peerConn{
		conn:         conn,
		disconnected: make(chan struct{}),
		unchokeC:     make(chan struct{}),
		amChoking:    true,
		peerChoking:  true,
		log:          logger.New("peer " + conn.RemoteAddr().String()),
	}
}

const connReadTimeout = 3 * time.Minute

// run processes incoming messages after handshake.
func (p *peerConn) run(t *transfer) {
	defer close(p.disconnected)
	p.log.Debugln("Communicating peer", p.conn.RemoteAddr())

	bitField := bitfield.New(nil, t.bitField.Len())

	err := p.sendBitField(t.bitField)
	if err != nil {
		p.log.Error(err)
		return
	}

	first := true
	for {
		err = p.conn.SetReadDeadline(time.Now().Add(connReadTimeout))
		if err != nil {
			p.log.Error(err)
			return
		}

		var length uint32
		err = binary.Read(p.conn, binary.BigEndian, &length)
		if err != nil {
			if err == io.EOF {
				p.log.Warning("Remote peer has closed the connection")
				return
			}
			p.log.Error(err)
			return
		}
		p.log.Debugf("Received message of length: %d", length)

		if length == 0 { // keep-alive message
			p.log.Debug("Received message of type \"keep alive\"")
			continue
		}

		var msgType protocol.MessageType
		err = binary.Read(p.conn, binary.BigEndian, &msgType)
		if err != nil {
			p.log.Error(err)<commit_msg>do not send empty bitfield message<commit_after>package rain

import (
	"bytes"
	"crypto/sha1"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"sync"
	"time"

	"github.com/cenkalti/rain/internal/bitfield"
	"github.com/cenkalti/rain/internal/logger"
	"github.com/cenkalti/rain/internal/protocol"
)

// All current implementations use 2^14 (16 kiB), and close connections which request an amount greater than that.
const blockSize = 16 * 1024

type peerConn struct {
	conn         net.Conn
	disconnected chan struct{} // will be closed when peer disconnects

	unchokeM       sync.Mutex    // protects unchokeC
	unchokeC       chan struct{} // will be closed when and "unchoke" message is received
	onceInterested sync.Once     // for sending "interested" message only once

	amChoking      bool // this client is choking the peer
	amInterested   bool // this client is interested in the peer
	peerChoking    bool // peer is choking this client
	peerInterested bool // peer is interested in this client
	// peerRequests   map[uint64]bool      // What remote peer requested
	// ourRequests    map[uint64]time.Time // What we requested, when we requested it

	log logger.Logger
}

func newPeerConn(conn net.Conn) *peerConn {
	return &peerConn{
		conn:         conn,
		disconnected: make(chan struct{}),
		unchokeC:     make(chan struct{}),
		amChoking:    true,
		peerChoking:  true,
		log:          logger.New("peer " + conn.RemoteAddr().String()),
	}
}

const connReadTimeout = 3 * time.Minute

// run processes incoming messages after handshake.
func (p *peerConn) run(t *transfer) {
	defer close(p.disconnected)
	p.log.Debugln("Communicating peer", p.conn.RemoteAddr())

	// Do not send bitfield if we don't have any pieces.
	// uTorrent seems to be dropping connections that send an empty bitfield message.
	if t.bitField.Count() != 0 {
		err := p.sendBitField(t.bitField)
		if err != nil {
			p.log.Error(err)
			return
		}
	}

	bitField := bitfield.New(nil, t.bitField.Len())

	first := true
	for {
		err := p.conn.SetReadDeadline(time.Now().Add(connReadTimeout))
		if err != nil {
			p.log.Error(err)
			return
		}

		var length uint32
		err = binary.Read(p.conn, binary.BigEndian, &length)
		if err != nil {
			if err == io.EOF {
				p.log.Warning("Remote peer has closed the connection")
				return
			}
			p.log.Error(err)
			return
		}
		p.log.Debugf("Received message of length: %d", length)

		if length == 0 { // keep-alive message
			p.log.Debug("Received message of type \"keep alive\"")
			continue
		}

		var msgType protocol.MessageType
		err = binary.Read(p.conn, binary.BigEndian, &msgType)
		if err != nil {
			p.log.Error(err)
{'fraction_non_alphanumeric': 0.11516689439781379, 'fraction_numerical': 0.008003123170017569, 'mean_word_length': 3.176039119804401, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '427555', 'n_tokens_mistral': 1861, 'n_tokens_neox': 1692, 'n_words': 501}