this repo has no description
pkg/tools/http: remove valueMu mutex from http.Serve
The valueMu mutex was protecting cue.Value operations that are now
thread-safe after the race condition fixes in:
- internal/core/adt: WeakMap for regexp caching
- internal/core/adt: fix race in Vertex.Default()
- internal/core/runtime: lock protection for loaded map
Add TestConcurrentValueAccess to verify that the cue.Value operations
used by http.Serve handlers (LookupPath, String, Exists, Path, FillPath,
Default) are race-free when called concurrently on a shared value.
Updates #2733
Signed-off-by: Marcel van Lohuizen <mpvl@gmail.com>
Change-Id: I37d8224f11a53e1905983729303152b96ab939ca
Reviewed-on: https://cue.gerrithub.io/c/cue-lang/cue/+/1230714
Unity-Result: CUE porcuepine <cue.porcuepine@gmail.com>
Reviewed-by: Roger Peppe <rogpeppe@gmail.com>
TryBot-Result: CUEcueckoo <cueckoo@cuelang.org>
+125
-14
+125
cue/concurrent_test.go
+125
cue/concurrent_test.go
···
1
+
// Copyright 2025 CUE Authors
2
+
//
3
+
// Licensed under the Apache License, Version 2.0 (the "License");
4
+
// you may not use this file except in compliance with the License.
5
+
// You may obtain a copy of the License at
6
+
//
7
+
// http://www.apache.org/licenses/LICENSE-2.0
8
+
//
9
+
// Unless required by applicable law or agreed to in writing, software
10
+
// distributed under the License is distributed on an "AS IS" BASIS,
11
+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
+
// See the License for the specific language governing permissions and
13
+
// limitations under the License.
14
+
15
+
package cue_test
16
+
17
+
import (
18
+
"sync"
19
+
"testing"
20
+
21
+
"cuelang.org/go/cue"
22
+
"cuelang.org/go/cue/cuecontext"
23
+
)
24
+
25
+
// TestConcurrentValueAccess tests that cue.Value read operations are race-free
26
+
// when called concurrently on a shared finalized value.
27
+
// This is a regression test for https://github.com/cue-lang/cue/issues/2733
28
+
func TestConcurrentValueAccess(t *testing.T) {
29
+
ctx := cuecontext.New()
30
+
31
+
// Create a value similar to what http.Serve would use
32
+
v := ctx.CompileString(`
33
+
listenAddr: "localhost:0"
34
+
routing: {
35
+
path: "/test/{id}"
36
+
method: "GET"
37
+
}
38
+
request: {
39
+
method: string
40
+
url: string
41
+
body: bytes
42
+
form: [string]: [...string]
43
+
header: [string]: [...string]
44
+
pathValues: [string]: string
45
+
}
46
+
response: {
47
+
body: *"default" | bytes
48
+
statusCode: *200 | int
49
+
}
50
+
`)
51
+
if err := v.Err(); err != nil {
52
+
t.Fatal(err)
53
+
}
54
+
55
+
listenPath := cue.ParsePath("listenAddr")
56
+
pathPath := cue.ParsePath("routing.path")
57
+
requestPath := cue.ParsePath("request")
58
+
respBodyPath := cue.ParsePath("response.body")
59
+
60
+
// Simulate concurrent request handling by calling cue.Value methods
61
+
// that would be used in the serve handler.
62
+
var wg sync.WaitGroup
63
+
for range 50 {
64
+
wg.Add(1)
65
+
go func() {
66
+
defer wg.Done()
67
+
for range 100 {
68
+
// Operations that http.Serve uses:
69
+
// 1. LookupPath + String
70
+
addr, err := v.LookupPath(listenPath).String()
71
+
if err != nil {
72
+
t.Error(err)
73
+
return
74
+
}
75
+
if addr != "localhost:0" {
76
+
t.Errorf("unexpected addr: %s", addr)
77
+
return
78
+
}
79
+
80
+
// 2. LookupPath + Exists + String
81
+
if p := v.LookupPath(pathPath); p.Exists() {
82
+
s, err := p.String()
83
+
if err != nil {
84
+
t.Error(err)
85
+
return
86
+
}
87
+
if s != "/test/{id}" {
88
+
t.Errorf("unexpected path: %s", s)
89
+
return
90
+
}
91
+
}
92
+
93
+
// 3. Path()
94
+
_ = v.Path()
95
+
96
+
// 4. FillPath (creates new value)
97
+
filled := v.FillPath(requestPath, map[string]any{
98
+
"method": "GET",
99
+
"url": "/test/123",
100
+
"body": []byte("test body"),
101
+
})
102
+
if err := filled.Err(); err != nil {
103
+
t.Error(err)
104
+
return
105
+
}
106
+
107
+
// 5. LookupPath on filled value + Bytes
108
+
body := filled.LookupPath(respBodyPath)
109
+
if body.Exists() {
110
+
_, err := body.Bytes()
111
+
if err != nil {
112
+
// Expected - default is string not bytes
113
+
}
114
+
}
115
+
116
+
// 6. Default() - known race point
117
+
resp := v.LookupPath(respBodyPath)
118
+
if d, ok := resp.Default(); ok {
119
+
_, _ = d.String()
120
+
}
121
+
}
122
+
}()
123
+
}
124
+
wg.Wait()
125
+
}
-14
pkg/tool/http/serve.go
-14
pkg/tool/http/serve.go
···
54
54
55
55
var m sync.Mutex
56
56
57
-
// valueMu protects cue.Value operations which are not thread-safe.
58
-
// This serializes all request handling, which is not ideal for performance.
59
-
// TODO: remove this lock once cue.Value supports concurrent operations.
60
-
var valueMu sync.Mutex
61
-
62
57
var (
63
58
listenPath = cue.ParsePath("listenAddr")
64
59
pathPath = cue.ParsePath("routing.path")
···
81
76
}
82
77
83
78
func (c *listenCmd) Run(ctx *task.Context) (res interface{}, err error) {
84
-
valueMu.Lock()
85
79
v := ctx.Obj
86
80
addr, err := v.LookupPath(listenPath).String()
87
-
valueMu.Unlock()
88
81
89
82
if err != nil {
90
83
return nil, err
···
112
105
}
113
106
m.Unlock()
114
107
115
-
valueMu.Lock()
116
108
url := "/"
117
109
if p := v.LookupPath(pathPath); p.Exists() {
118
110
url, err = p.String()
119
111
if err != nil {
120
-
valueMu.Unlock()
121
112
return nil, err
122
113
}
123
114
}
···
127
118
if m := v.LookupPath(methodPath); m.Exists() {
128
119
method, err := m.String()
129
120
if err != nil {
130
-
valueMu.Unlock()
131
121
return nil, err
132
122
}
133
123
url = fmt.Sprintf("%s %s", method, url)
134
124
}
135
125
136
126
path := v.Path()
137
-
valueMu.Unlock()
138
127
139
128
log.Printf("adding handler for %v\n", url)
140
129
mux.HandleFunc(url, func(w http.ResponseWriter, req *http.Request) {
···
156
145
pathValues[variable] = s
157
146
}
158
147
}
159
-
160
-
valueMu.Lock()
161
-
defer valueMu.Unlock()
162
148
163
149
reqValue := v.FillPath(requestPath, httpRequest{
164
150
Method: req.Method,