70 lines
1.8 KiB
Go
70 lines
1.8 KiB
Go
package model
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
type Execution struct {
|
|
Protocol string `json:"protocol"`
|
|
Operation string `json:"operation"`
|
|
DurationMS float64 `json:"durationMs"`
|
|
At time.Time `json:"at"`
|
|
}
|
|
type executionKey struct{}
|
|
|
|
func WithExecutionObserver(ctx context.Context, observe func(Execution)) context.Context {
|
|
return context.WithValue(ctx, executionKey{}, observe)
|
|
}
|
|
func ObserveExecution(ctx context.Context, protocol, operation string, started time.Time) {
|
|
if observe, ok := ctx.Value(executionKey{}).(func(Execution)); ok {
|
|
observe(Execution{Protocol: protocol, Operation: operation, DurationMS: float64(time.Since(started)) / float64(time.Millisecond), At: time.Now().UTC()})
|
|
}
|
|
}
|
|
|
|
// ExecutionQueue correlates ordered replies without retaining SQL, keys or payloads.
|
|
// shortcut: stop measuring after 4096 outstanding requests; forwarding remains unaffected.
|
|
type ExecutionQueue struct {
|
|
mu sync.Mutex
|
|
pending []executionStart
|
|
disabled bool
|
|
}
|
|
type executionStart struct {
|
|
operation string
|
|
at time.Time
|
|
}
|
|
|
|
func (q *ExecutionQueue) Start(operation string) { q.StartAt(operation, time.Now()) }
|
|
func (q *ExecutionQueue) StartAt(operation string, at time.Time) {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
if q.disabled {
|
|
return
|
|
}
|
|
if len(q.pending) >= 4096 {
|
|
q.disabled = true
|
|
q.pending = nil
|
|
return
|
|
}
|
|
q.pending = append(q.pending, executionStart{operation, at})
|
|
}
|
|
func (q *ExecutionQueue) Disable() {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
q.disabled = true
|
|
q.pending = nil
|
|
}
|
|
func (q *ExecutionQueue) Finish(ctx context.Context, protocol string) {
|
|
q.mu.Lock()
|
|
if len(q.pending) == 0 {
|
|
q.mu.Unlock()
|
|
return
|
|
}
|
|
v := q.pending[0]
|
|
q.pending[0] = executionStart{}
|
|
q.pending = q.pending[1:]
|
|
q.mu.Unlock()
|
|
ObserveExecution(ctx, protocol, v.operation, v.at)
|
|
}
|