tracing support

tracer.go

@@ -15,22 +15,30 @@ import (
 	"github.com/libp2p/go-libp2p-core/host"
 	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
+	"github.com/libp2p/go-libp2p-core/peerstore"
 	"github.com/libp2p/go-libp2p-core/protocol"
 
 	ggio "github.com/gogo/protobuf/io"
 )
 
 var TraceBufferSize = 1 << 16 // 64K ought to be enough for everyone; famous last words.
+var MinTraceBatchSize = 16
 
 type basicTracer struct {
-	ch    chan struct{}
-	mx    sync.Mutex
-	buf   []*pb.TraceEvent
-	lossy bool
+	ch     chan struct{}
+	mx     sync.Mutex
+	buf    []*pb.TraceEvent
+	lossy  bool
+	closed bool
 }
 
 func (t *basicTracer) Trace(evt *pb.TraceEvent) {
 	t.mx.Lock()
+	if t.closed {
+		t.mx.Unlock()
+		return
+	}
+
 	if t.lossy && len(t.buf) > TraceBufferSize {
 		log.Warningf("trace buffer overflow; dropping trace event")
 	} else {
@@ -45,7 +53,12 @@ func (t *basicTracer) Trace(evt *pb.TraceEvent) {
 }
 
 func (t *basicTracer) Close() {
-	close(t.ch)
+	t.mx.Lock()
+	defer t.mx.Unlock()
+	if !t.closed {
+		t.closed = true
+		close(t.ch)
+	}
 }
 
 // JSONTracer is a tracer that writes events to a file, encoded in ndjson.
@@ -160,12 +173,15 @@ type RemoteTracer struct {
 	basicTracer
 	ctx  context.Context
 	host host.Host
-	pi   peer.AddrInfo
+	peer peer.ID
 }
 
 // NewRemoteTracer constructs a RemoteTracer, tracing to the peer identified by pi
 func NewRemoteTracer(ctx context.Context, host host.Host, pi peer.AddrInfo) (*RemoteTracer, error) {
-	tr := &RemoteTracer{ctx: ctx, host: host, pi: pi, basicTracer: basicTracer{ch: make(chan struct{}, 1), lossy: true}}
+	tr := &RemoteTracer{ctx: ctx, host: host, peer: pi.ID, basicTracer: basicTracer{ch: make(chan struct{}, 1), lossy: true}}
+	for _, addr := range pi.Addrs {
+		host.Peerstore().AddAddr(pi.ID, addr, peerstore.PermanentAddrTTL)
+	}
 	go tr.doWrite()
 	return tr, nil
 }
@@ -187,15 +203,17 @@ func (t *RemoteTracer) doWrite() {
 	for {
 		_, ok := <-t.ch
 
-		// nil out the buffer to gc events when swapping buffers
-		for i := range buf {
-			buf[i] = nil
-		}
-
-		// wait a bit to accumulate a batch
-		time.Sleep(time.Second)
+		// deadline for batch accumulation
+		deadline := time.Now().Add(time.Second)
 
+	again:
 		t.mx.Lock()
+		if len(t.buf) < MinTraceBatchSize && time.Now().Before(deadline) {
+			t.mx.Unlock()
+			time.Sleep(100 * time.Millisecond)
+			goto again
+		}
+
 		tmp := t.buf
 		t.buf = buf[:0]
 		buf = tmp
@@ -220,6 +238,11 @@ func (t *RemoteTracer) doWrite() {
 		}
 
 	end:
+		// nil out the buffer to gc consumed events
+		for i := range buf {
+			buf[i] = nil
+		}
+
 		if !ok {
 			if err != nil {
 				s.Reset()
@@ -243,38 +266,10 @@ func (t *RemoteTracer) doWrite() {
 	}
 }
 
-func (t *RemoteTracer) connect() error {
-	for {
-		ctx, cancel := context.WithTimeout(t.ctx, time.Minute)
-		err := t.host.Connect(ctx, t.pi)
-		cancel()
-		if err != nil {
-			if t.ctx.Err() != nil {
-				return err
-			}
-
-			// wait a minute and try again, to account for transient server downtime
-			select {
-			case <-time.After(time.Minute):
-				continue
-			case <-t.ctx.Done():
-				return t.ctx.Err()
-			}
-		}
-
-		return nil
-	}
-}
-
 func (t *RemoteTracer) openStream() (network.Stream, error) {
 	for {
-		err := t.connect()
-		if err != nil {
-			return nil, err
-		}
-
 		ctx, cancel := context.WithTimeout(t.ctx, time.Minute)
-		s, err := t.host.NewStream(ctx, t.pi.ID, RemoteTracerProtoID)
+		s, err := t.host.NewStream(ctx, t.peer, RemoteTracerProtoID)
 		cancel()
 		if err != nil {
 			if t.ctx.Err() != nil {