diff --git a/experimental/demoprobe/main.go b/experimental/demoprobe/main.go index b49370a22..55cb91fb6 100644 --- a/experimental/demoprobe/main.go +++ b/experimental/demoprobe/main.go @@ -27,10 +27,11 @@ func main() { if err != nil { log.Fatal(err) } + rp := xfer.NewReportPublisher(publisher) rand.Seed(time.Now().UnixNano()) for range time.Tick(*publishInterval) { - if err := publisher.Publish(demoReport(*hostCount)); err != nil { + if err := rp.Publish(demoReport(*hostCount)); err != nil { log.Print(err) } } diff --git a/experimental/fixprobe/main.go b/experimental/fixprobe/main.go index f0d0a1812..1bd1294b6 100644 --- a/experimental/fixprobe/main.go +++ b/experimental/fixprobe/main.go @@ -39,7 +39,8 @@ func main() { log.Fatal(err) } + rp := xfer.NewReportPublisher(publisher) for range time.Tick(*publishInterval) { - publisher.Publish(fixedReport) + rp.Publish(fixedReport) } } diff --git a/probe/main.go b/probe/main.go index 161276719..eb4495ff9 100644 --- a/probe/main.go +++ b/probe/main.go @@ -175,6 +175,7 @@ func main() { pubTick = time.Tick(*publishInterval) spyTick = time.Tick(*spyInterval) r = report.MakeReport() + p = xfer.NewReportPublisher(publishers) ) for { @@ -182,7 +183,7 @@ func main() { case <-pubTick: publishTicks.WithLabelValues().Add(1) r.Window = *publishInterval - if err := publishers.Publish(r); err != nil { + if err := p.Publish(r); err != nil { log.Printf("publish: %v", err) } r = report.MakeReport() diff --git a/xfer/publisher.go b/xfer/publisher.go index 1eb1fcdd2..e71621ca3 100644 --- a/xfer/publisher.go +++ b/xfer/publisher.go @@ -1,9 +1,6 @@ package xfer import ( - "bytes" - "compress/gzip" - "encoding/gob" "fmt" "log" "net/http" @@ -11,8 +8,6 @@ import ( "strings" "sync" "time" - - "github.com/weaveworks/scope/report" ) const ( @@ -22,7 +17,7 @@ const ( // Publisher is something which can send a report to a remote collector. type Publisher interface { - Publish(report.Report) error + Publish(*Buffer) error Stop() } @@ -63,16 +58,10 @@ func (p HTTPPublisher) String() string { } // Publish publishes the report to the URL. -func (p HTTPPublisher) Publish(rpt report.Report) error { - gzbuf := bytes.Buffer{} - gzwriter := gzip.NewWriter(&gzbuf) +func (p HTTPPublisher) Publish(buf *Buffer) error { + defer buf.Put() - if err := gob.NewEncoder(gzwriter).Encode(rpt); err != nil { - return err - } - gzwriter.Close() // otherwise the content won't get flushed to the output stream - - req, err := http.NewRequest("POST", p.url, &gzbuf) + req, err := http.NewRequest("POST", p.url, buf) if err != nil { return err } @@ -108,7 +97,7 @@ func AuthorizationHeader(token string) string { // concurrent publishes are dropped. type BackgroundPublisher struct { publisher Publisher - reports chan report.Report + reports chan *Buffer quit chan struct{} } @@ -116,7 +105,7 @@ type BackgroundPublisher struct { func NewBackgroundPublisher(p Publisher) *BackgroundPublisher { result := &BackgroundPublisher{ publisher: p, - reports: make(chan report.Report), + reports: make(chan *Buffer), quit: make(chan struct{}), } go result.loop() @@ -146,9 +135,9 @@ func (b *BackgroundPublisher) loop() { } // Publish implements Publisher -func (b *BackgroundPublisher) Publish(r report.Report) error { +func (b *BackgroundPublisher) Publish(buf *Buffer) error { select { - case b.reports <- r: + case b.reports <- buf: default: } return nil @@ -198,13 +187,18 @@ func (p *MultiPublisher) Add(target string) { } // Publish implements Publisher by emitting the report to all publishers. -func (p *MultiPublisher) Publish(rpt report.Report) error { +func (p *MultiPublisher) Publish(buf *Buffer) error { p.mtx.RLock() defer p.mtx.RUnlock() + // First take a reference for each publisher + for range p.m { + buf.Get() + } + var errs []string for _, publisher := range p.m { - if err := publisher.Publish(rpt); err != nil { + if err := publisher.Publish(buf); err != nil { errs = append(errs, err.Error()) } } diff --git a/xfer/publisher_test.go b/xfer/publisher_test.go index ecf0148b0..195d0b0bf 100644 --- a/xfer/publisher_test.go +++ b/xfer/publisher_test.go @@ -64,7 +64,8 @@ func TestHTTPPublisher(t *testing.T) { if err != nil { t.Fatal(err) } - if err := p.Publish(rpt); err != nil { + rp := xfer.NewReportPublisher(p) + if err := rp.Publish(rpt); err != nil { t.Error(err) } @@ -83,7 +84,7 @@ func TestMultiPublisher(t *testing.T) { ) multiPublisher.Add("first") - if err := multiPublisher.Publish(report.MakeReport()); err != nil { + if err := multiPublisher.Publish(xfer.NewBuffer(nil)); err != nil { t.Error(err) } if want, have := 1, p.count; want != have { @@ -91,7 +92,7 @@ func TestMultiPublisher(t *testing.T) { } multiPublisher.Add("second") // but factory returns same mockPublisher - if err := multiPublisher.Publish(report.MakeReport()); err != nil { + if err := multiPublisher.Publish(xfer.NewBuffer(nil)); err != nil { t.Error(err) } if want, have := 3, p.count; want != have { @@ -101,5 +102,5 @@ func TestMultiPublisher(t *testing.T) { type mockPublisher struct{ count int } -func (p *mockPublisher) Publish(report.Report) error { p.count++; return nil } -func (p *mockPublisher) Stop() {} +func (p *mockPublisher) Publish(*xfer.Buffer) error { p.count++; return nil } +func (p *mockPublisher) Stop() {} diff --git a/xfer/report_publisher.go b/xfer/report_publisher.go new file mode 100644 index 000000000..e2af70af5 --- /dev/null +++ b/xfer/report_publisher.go @@ -0,0 +1,79 @@ +package xfer + +import ( + "bytes" + "compress/gzip" + "encoding/gob" + "sync" + "sync/atomic" + + "github.com/weaveworks/scope/report" +) + +// A Buffer is a reference counted bytes.Buffer, which belongs +// to a sync.Pool +type Buffer struct { + bytes.Buffer + pool *sync.Pool + refs int32 +} + +// NewBuffer creates a new buffer +func NewBuffer(pool *sync.Pool) *Buffer { + return &Buffer{ + pool: pool, + refs: 0, + } +} + +// Get increases the reference count. It is safe for concurrent calls. +func (b *Buffer) Get() { + atomic.AddInt32(&b.refs, 1) +} + +// Put decreases the reference count, and when it hits zero, puts the +// buffer back in the pool. +func (b *Buffer) Put() { + if atomic.AddInt32(&b.refs, -1) == 0 { + b.Reset() + b.pool.Put(b) + } +} + +// NewBufferPool creates a new buffer pool. +func NewBufferPool() *sync.Pool { + result := &sync.Pool{} + result.New = func() interface{} { + return NewBuffer(result) + } + return result +} + +// A ReportPublisher uses a buffer pool to serialise reports, which it +// then passes to a publisher +type ReportPublisher struct { + buffers *sync.Pool + publisher Publisher +} + +// NewReportPublisher creates a new report publisher +func NewReportPublisher(publisher Publisher) *ReportPublisher { + return &ReportPublisher{ + buffers: NewBufferPool(), + publisher: publisher, + } +} + +// Publish serialises and compresses a report, then passes it to a publisher +func (p *ReportPublisher) Publish(r report.Report) error { + buf := p.buffers.Get().(*Buffer) + gzwriter := gzip.NewWriter(buf) + if err := gob.NewEncoder(gzwriter).Encode(r); err != nil { + buf.Reset() + p.buffers.Put(buf) + return err + } + gzwriter.Close() // otherwise the content won't get flushed to the output stream + + return p.publisher.Publish(buf) +}