mirror of
https://github.com/golang/go
synced 2024-11-18 21:24:44 -07:00
internal/telemtry: changed to a simpler threading model for stats
this is not a final solution, but it makes it easier to debug and reason about, and does not require a go routine or buffered channel Change-Id: I758758ac80fcd525ab5264e34c48941766a8db11 Reviewed-on: https://go-review.googlesource.com/c/tools/+/208664 Run-TryBot: Ian Cottrell <iancottrell@google.com> TryBot-Result: Gobot Gobot <gobot@golang.org> Reviewed-by: Emmanuel Odeke <emm.odeke@gmail.com>
This commit is contained in:
parent
9fe613bd66
commit
7360bd5c0f
@ -9,6 +9,7 @@ package stats
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"golang.org/x/tools/internal/telemetry/unit"
|
||||
@ -19,6 +20,7 @@ type Int64Measure struct {
|
||||
name string
|
||||
description string
|
||||
unit unit.Unit
|
||||
mu sync.Mutex
|
||||
subscribers []Int64Subscriber
|
||||
}
|
||||
|
||||
@ -27,6 +29,7 @@ type Float64Measure struct {
|
||||
name string
|
||||
description string
|
||||
unit unit.Unit
|
||||
mu sync.Mutex
|
||||
subscribers []Float64Subscriber
|
||||
}
|
||||
|
||||
@ -66,16 +69,20 @@ func (m *Int64Measure) Description() string { return m.description }
|
||||
func (m *Int64Measure) Unit() unit.Unit { return m.unit }
|
||||
|
||||
// Subscribe adds a new subscriber to this measure.
|
||||
func (m *Int64Measure) Subscribe(s Int64Subscriber) { m.subscribers = append(m.subscribers, s) }
|
||||
func (m *Int64Measure) Subscribe(s Int64Subscriber) {
|
||||
m.mu.Lock()
|
||||
m.subscribers = append(m.subscribers, s)
|
||||
m.mu.Unlock()
|
||||
}
|
||||
|
||||
// Record delivers a new value to the subscribers of this measure.
|
||||
func (m *Int64Measure) Record(ctx context.Context, value int64) {
|
||||
at := time.Now()
|
||||
do(func() {
|
||||
for _, s := range m.subscribers {
|
||||
s(ctx, m, value, at)
|
||||
}
|
||||
})
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
for _, s := range m.subscribers {
|
||||
s(ctx, m, value, at)
|
||||
}
|
||||
}
|
||||
|
||||
// Name returns the name this measure was given on construction.
|
||||
@ -88,14 +95,18 @@ func (m *Float64Measure) Description() string { return m.description }
|
||||
func (m *Float64Measure) Unit() unit.Unit { return m.unit }
|
||||
|
||||
// Subscribe adds a new subscriber to this measure.
|
||||
func (m *Float64Measure) Subscribe(s Float64Subscriber) { m.subscribers = append(m.subscribers, s) }
|
||||
func (m *Float64Measure) Subscribe(s Float64Subscriber) {
|
||||
m.mu.Lock()
|
||||
m.subscribers = append(m.subscribers, s)
|
||||
m.mu.Unlock()
|
||||
}
|
||||
|
||||
// Record delivers a new value to the subscribers of this measure.
|
||||
func (m *Float64Measure) Record(ctx context.Context, value float64) {
|
||||
at := time.Now()
|
||||
do(func() {
|
||||
for _, s := range m.subscribers {
|
||||
s(ctx, m, value, at)
|
||||
}
|
||||
})
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
for _, s := range m.subscribers {
|
||||
s(ctx, m, value, at)
|
||||
}
|
||||
}
|
||||
|
@ -1,40 +0,0 @@
|
||||
// Copyright 2019 The Go Authors. All rights reserved.
|
||||
// Use of this source code is governed by a BSD-style
|
||||
// license that can be found in the LICENSE file.
|
||||
|
||||
package stats
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
)
|
||||
|
||||
var (
|
||||
// TODO: Think about whether this is the right concurrency model, and what
|
||||
// TODO: the queue length should be
|
||||
workQueue = make(chan func(), 1000)
|
||||
)
|
||||
|
||||
func init() {
|
||||
go func() {
|
||||
for task := range workQueue {
|
||||
task()
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// do adds a task to the list of things to work on in the background.
|
||||
// All tasks will be handled in submission order, and no two tasks will happen
|
||||
// concurrently so they do not need to do any kind of locking.
|
||||
// It is safe however to call Do concurrently.
|
||||
// No promises are made about when the tasks will be performed.
|
||||
// This function may block, but in general it will return very quickly and
|
||||
// before the task has been run.
|
||||
func do(task func()) {
|
||||
select {
|
||||
case workQueue <- task:
|
||||
default:
|
||||
fmt.Fprint(os.Stderr, "work queue is full\n")
|
||||
workQueue <- task
|
||||
}
|
||||
}
|
Loading…
Reference in New Issue
Block a user