8 "github.com/anacrolix/chansync"
9 "github.com/anacrolix/log"
10 "github.com/anacrolix/sync"
12 pp "github.com/anacrolix/torrent/peer_protocol"
15 func (pc *PeerConn) startWriter() {
16 w := &pc.messageWriter
17 *w = peerConnMsgWriter{
18 fillWriteBuffer: func() {
20 defer pc.locker().Unlock()
21 if pc.closed.IsSet() {
29 keepAlive: func() bool {
31 defer pc.locker().Unlock()
34 writeBuffer: new(bytes.Buffer),
37 defer pc.locker().Unlock()
39 defer pc.locker().Lock()
40 pc.messageWriter.run(pc.t.cl.config.KeepAliveTimeout)
44 type peerConnMsgWriter struct {
45 // Must not be called with the local mutex held, as it will call back into the write method.
46 fillWriteBuffer func()
47 closed *chansync.SetOnce
53 writeCond chansync.BroadcastCond
54 // Pointer so we can swap with the "front buffer".
55 writeBuffer *bytes.Buffer
58 // Routine that writes to the peer. Some of what to write is buffered by
59 // activity elsewhere in the Client, and some is determined locally when the
60 // connection is writable.
61 func (cn *peerConnMsgWriter) run(keepAliveTimeout time.Duration) {
62 lastWrite := time.Now()
63 keepAliveTimer := time.NewTimer(keepAliveTimeout)
64 frontBuf := new(bytes.Buffer)
66 if cn.closed.IsSet() {
70 keepAlive := cn.keepAlive()
72 if cn.writeBuffer.Len() == 0 && time.Since(lastWrite) >= keepAliveTimeout && keepAlive {
73 cn.writeBuffer.Write(pp.Message{Keepalive: true}.MustMarshalBinary())
74 torrent.Add("written keepalives", 1)
76 if cn.writeBuffer.Len() == 0 {
77 writeCond := cn.writeCond.Signaled()
80 case <-cn.closed.Done():
82 case <-keepAliveTimer.C:
87 frontBuf, cn.writeBuffer = cn.writeBuffer, frontBuf
89 n, err := cn.w.Write(frontBuf.Bytes())
91 lastWrite = time.Now()
92 keepAliveTimer.Reset(keepAliveTimeout)
95 cn.logger.WithDefaultLevel(log.Debug).Printf("error writing: %v", err)
98 if n != frontBuf.Len() {
105 func (cn *peerConnMsgWriter) write(msg pp.Message) bool {
108 cn.writeBuffer.Write(msg.MustMarshalBinary())
109 cn.writeCond.Broadcast()
110 return !cn.writeBufferFull()
113 func (cn *peerConnMsgWriter) writeBufferFull() bool {
114 return cn.writeBuffer.Len() >= writeBufferHighWaterLen