You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by al...@apache.org on 2025/09/06 03:53:15 UTC
(dubbo-go-pixiu) branch develop updated: fix: fix protencial data race problem (#750)
This is an automated email from the ASF dual-hosted git repository.
alexstocks pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/dubbo-go-pixiu.git
The following commit(s) were added to refs/heads/develop by this push:
new e2303654 fix: fix protencial data race problem (#750)
e2303654 is described below
commit e230365431407010ef45b4b9070c897cfe293d3c
Author: Xuetao Li <m1...@163.com>
AuthorDate: Sat Sep 6 11:53:11 2025 +0800
fix: fix protencial data race problem (#750)
* fix prometheus.go data race
---
pkg/common/http/manager_test.go | 28 ++++++-
pkg/filter/llm/tokenizer/tokenizer.go | 142 +++++++++++++++++++---------------
pkg/prometheus/prometheus.go | 9 ++-
3 files changed, 110 insertions(+), 69 deletions(-)
diff --git a/pkg/common/http/manager_test.go b/pkg/common/http/manager_test.go
index 292fd752..33080f38 100644
--- a/pkg/common/http/manager_test.go
+++ b/pkg/common/http/manager_test.go
@@ -25,6 +25,7 @@ import (
"net/http"
"net/http/httptest"
"strings"
+ "sync"
"testing"
"time"
)
@@ -102,7 +103,6 @@ func (f *DemoFilterFactory) PrepareFilterChain(ctx *contexthttp.HttpContext, cha
c := f.conf
str := fmt.Sprintf("%s is drinking in the %s", c.Foo, c.Bar)
demoFilter := &DemoFilter{str: str}
-
chain.AppendDecodeFilters(demoFilter)
chain.AppendEncodeFilters(demoFilter)
return nil
@@ -219,11 +219,11 @@ func TestStreamingResponse(t *testing.T) {
// event waiting test
for {
- receivedEvents := httpCtx.Writer.(*StreamRecorder).receivedBuf
select {
case event := <-eventCh:
logger.Info("Received event: %s", strings.ReplaceAll(event, "\n", "\\n"))
case <-done:
+ receivedEvents := httpCtx.Writer.(*StreamRecorder).GetReceivedBuf()
assert.Equal(t, 3, len(receivedEvents), "Should receive 3 events")
return
case <-time.After(5 * time.Second):
@@ -237,6 +237,7 @@ func TestStreamingResponse(t *testing.T) {
type StreamRecorder struct {
http.ResponseWriter
http.Flusher
+ mu sync.Mutex
receivedBuf []string
headers http.Header
status int
@@ -259,10 +260,20 @@ func (r *StreamRecorder) WriteHeader(statusCode int) {
func (r *StreamRecorder) Write(data []byte) (int, error) {
eventCh <- string(data)
+ r.mu.Lock()
+ defer r.mu.Unlock()
r.receivedBuf = append(r.receivedBuf, string(data))
return len(data), nil
}
+func (r *StreamRecorder) GetReceivedBuf() []string {
+ bufCopy := make([]string, len(r.receivedBuf))
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ copy(bufCopy, r.receivedBuf)
+ return bufCopy
+}
+
func (r *StreamRecorder) Flush() {
}
@@ -286,6 +297,7 @@ func NewTestServerWithURL(URL string, handler http.Handler) (*httptest.Server, e
// StreamHTTPRecorder Used to capture and test streaming HTTP responses over channels
type StreamHTTPRecorder struct {
http.ResponseWriter
+ mu sync.Mutex
receivedBuf []string
headers http.Header
status int
@@ -310,6 +322,8 @@ func (r *StreamHTTPRecorder) WriteHeader(statusCode int) {
func (r *StreamHTTPRecorder) Write(data []byte) (int, error) {
eventCh <- string(data)
+ r.mu.Lock()
+ defer r.mu.Unlock()
r.receivedBuf = append(r.receivedBuf, string(data))
return len(data), nil
}
@@ -318,6 +332,14 @@ func (r *StreamHTTPRecorder) Flush() {
r.flushCount++
}
+func (r *StreamHTTPRecorder) GetReceivedBuf() []string {
+ bufCopy := make([]string, len(r.receivedBuf))
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ copy(bufCopy, r.receivedBuf)
+ return bufCopy
+}
+
// Test a variety of common streaming HTTP response types
func TestStreamableHTTPResponse(t *testing.T) {
// define the type of content you want to test
@@ -418,12 +440,12 @@ func testStreamableResponse(t *testing.T, contentType string) {
// collect and validate responses
receivedChunks := 0
for {
- receivedEvents := httpCtx.Writer.(*StreamHTTPRecorder).receivedBuf
select {
case event := <-eventCh:
logger.Info("Received chunk: %s", event)
receivedChunks++
case <-done:
+ receivedEvents := httpCtx.Writer.(*StreamHTTPRecorder).GetReceivedBuf()
assert.Equal(t, 5, len(receivedEvents), "Should receive 5 chunks")
return
case <-time.After(5 * time.Second):
diff --git a/pkg/filter/llm/tokenizer/tokenizer.go b/pkg/filter/llm/tokenizer/tokenizer.go
index c5ff441f..bc9d6825 100644
--- a/pkg/filter/llm/tokenizer/tokenizer.go
+++ b/pkg/filter/llm/tokenizer/tokenizer.go
@@ -101,6 +101,9 @@ var (
llmTotalDurationSum syncfloat64.Counter
llmTimeToLastTokenSum syncfloat64.Counter
llmStreamingRequestsTotal syncint64.Counter
+
+ registerOnce sync.Once
+ registerErr error
)
func init() {
@@ -381,76 +384,87 @@ func decompress(body io.Reader, encoding string) ([]byte, bool) {
// registerLLMMetrics creates and registers all metrics for this plugin.
func registerLLMMetrics() error {
- meter := global.MeterProvider().Meter(meterName)
- var err error
+ registerOnce.Do(func() {
+ meter := global.MeterProvider().Meter(meterName)
+ var err error
- llmPromptTokens, err = meter.SyncInt64().Counter(metricPromptTokens,
- instrument.WithDescription("Total prompt tokens."),
- instrument.WithUnit("1"))
- if err != nil {
- return err
- }
- llmCompletionTokens, err = meter.SyncInt64().Counter(metricCompletionTokens,
- instrument.WithDescription("Total completion tokens."),
- instrument.WithUnit("1"))
- if err != nil {
- return err
- }
- llmTotalTokens, err = meter.SyncInt64().Counter(metricTotalTokens,
- instrument.WithDescription("Total tokens."),
- instrument.WithUnit("1"))
- if err != nil {
- return err
- }
+ llmPromptTokens, err = meter.SyncInt64().Counter(metricPromptTokens,
+ instrument.WithDescription("Total prompt tokens."),
+ instrument.WithUnit("1"))
+ if err != nil {
+ registerErr = err
+ return
+ }
+ llmCompletionTokens, err = meter.SyncInt64().Counter(metricCompletionTokens,
+ instrument.WithDescription("Total completion tokens."),
+ instrument.WithUnit("1"))
+ if err != nil {
+ registerErr = err
+ return
+ }
+ llmTotalTokens, err = meter.SyncInt64().Counter(metricTotalTokens,
+ instrument.WithDescription("Total tokens."),
+ instrument.WithUnit("1"))
+ if err != nil {
+ registerErr = err
+ return
+ }
- llmUpstreamRequestsTotal, err = meter.SyncInt64().Counter(metricUpstreamRequests,
- instrument.WithDescription("Total requests to upstream endpoints."),
- instrument.WithUnit("1"))
- if err != nil {
- return err
- }
- llmUpstreamRequestsSuccessTotal, err = meter.SyncInt64().Counter(metricUpstreamSuccess,
- instrument.WithDescription("Total successful requests to upstream endpoints."),
- instrument.WithUnit("1"))
- if err != nil {
- return err
- }
- llmUpstreamRequestsFailureTotal, err = meter.SyncInt64().Counter(metricUpstreamFailure,
- instrument.WithDescription("Total failed requests to upstream endpoints."),
- instrument.WithUnit("1"))
- if err != nil {
- return err
- }
+ llmUpstreamRequestsTotal, err = meter.SyncInt64().Counter(metricUpstreamRequests,
+ instrument.WithDescription("Total requests to upstream endpoints."),
+ instrument.WithUnit("1"))
+ if err != nil {
+ registerErr = err
+ return
+ }
+ llmUpstreamRequestsSuccessTotal, err = meter.SyncInt64().Counter(metricUpstreamSuccess,
+ instrument.WithDescription("Total successful requests to upstream endpoints."),
+ instrument.WithUnit("1"))
+ if err != nil {
+ registerErr = err
+ return
+ }
+ llmUpstreamRequestsFailureTotal, err = meter.SyncInt64().Counter(metricUpstreamFailure,
+ instrument.WithDescription("Total failed requests to upstream endpoints."),
+ instrument.WithUnit("1"))
+ if err != nil {
+ registerErr = err
+ return
+ }
- llmTotalDurationSum, err = meter.SyncFloat64().Counter(
- metricTotalDurationSum,
- instrument.WithDescription("Sum of total duration of LLM requests in microseconds."),
- instrument.WithUnit(unit.Unit("µs")),
- )
- if err != nil {
- return err
- }
+ llmTotalDurationSum, err = meter.SyncFloat64().Counter(
+ metricTotalDurationSum,
+ instrument.WithDescription("Sum of total duration of LLM requests in microseconds."),
+ instrument.WithUnit(unit.Unit("µs")),
+ )
+ if err != nil {
+ registerErr = err
+ return
+ }
- llmTimeToLastTokenSum, err = meter.SyncFloat64().Counter(
- metricTTLTSum,
- instrument.WithDescription("Sum of Time to Last Token for streaming responses in milliseconds."),
- instrument.WithUnit("ms"),
- )
- if err != nil {
- return err
- }
+ llmTimeToLastTokenSum, err = meter.SyncFloat64().Counter(
+ metricTTLTSum,
+ instrument.WithDescription("Sum of Time to Last Token for streaming responses in milliseconds."),
+ instrument.WithUnit("ms"),
+ )
+ if err != nil {
+ registerErr = err
+ return
+ }
- llmStreamingRequestsTotal, err = meter.SyncInt64().Counter(
- metricStreamingRequests,
- instrument.WithDescription("Total number of streaming LLM requests."),
- instrument.WithUnit("1"),
- )
- if err != nil {
- return err
- }
+ llmStreamingRequestsTotal, err = meter.SyncInt64().Counter(
+ metricStreamingRequests,
+ instrument.WithDescription("Total number of streaming LLM requests."),
+ instrument.WithUnit("1"),
+ )
+ if err != nil {
+ registerErr = err
+ return
+ }
- logger.Info("LLM metrics registered successfully.")
- return nil
+ logger.Info("LLM metrics registered successfully.")
+ })
+ return registerErr
}
// --- TeeReadCloser Implementation ---
diff --git a/pkg/prometheus/prometheus.go b/pkg/prometheus/prometheus.go
index 415c79ce..273e1798 100644
--- a/pkg/prometheus/prometheus.go
+++ b/pkg/prometheus/prometheus.go
@@ -220,7 +220,7 @@ type Prometheus struct {
// PushGateway contains the configuration for pushing to a Prometheus pushgateway (optional)
type PushGateway struct {
CounterPush bool
- PushIntervalSeconds time.Duration
+ PushInterval time.Duration
PushIntervalThreshold int
PushGatewayURL string
Job string
@@ -268,7 +268,8 @@ func (p *Prometheus) registerMetrics() {
}
func (p *Prometheus) SetPushGatewayUrl(pushGatewayURL, metricspath string) {
-
+ p.Ppg.mutex.Lock()
+ defer p.Ppg.mutex.Unlock()
p.Ppg.PushGatewayURL = pushGatewayURL
p.MetricsPath = metricspath
}
@@ -279,6 +280,8 @@ func (p *Prometheus) SetPushIntervalThreshold(isTurn bool, pushIntervalThreshold
}
func (p *Prometheus) SetPushGatewayJob(j string) {
+ p.Ppg.mutex.Lock()
+ defer p.Ppg.mutex.Unlock()
p.Ppg.Job = j
}
@@ -323,6 +326,8 @@ func (p *Prometheus) sendMetricsToPushGateway(metrics []byte) {
}
func (p *Prometheus) getPushGatewayURL() string {
+ p.Ppg.mutex.Lock()
+ defer p.Ppg.mutex.Unlock()
h, _ := os.Hostname()
if p.Ppg.Job == "" {
p.Ppg.Job = "pixiu"