Skip to content

Commit 3bd3d2f

Browse files
author
Henrik Johansson
authored
Merge pull request #239 from scylladb/issue_226
gemini: attempts to shutdown more precisely
2 parents b5832da + 12f45d1 commit 3bd3d2f

File tree

14 files changed

+302
-270
lines changed

14 files changed

+302
-270
lines changed

‎.github/workflows/go.yml

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
name: Go
2+
3+
on:
4+
push:
5+
branches: [ master ]
6+
pull_request:
7+
branches: [ master ]
8+
9+
jobs:
10+
build:
11+
name: Build
12+
env:
13+
runs-on: ubuntu-latest
14+
steps:
15+
- steps:
16+
- uses: actions/checkout@v2
17+
- name: Setup Go
18+
uses: actions/setup-go@v2
19+
with:
20+
go-version: latest
21+
22+
- name: Unit Tests
23+
run: go test -v -race ./...
24+
25+
- name: Build
26+
run: go build .

‎cmd/gemini/generators.go

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,11 +15,13 @@
1515
package main
1616

1717
import (
18+
"context"
19+
1820
"github.com/scylladb/gemini"
1921
"go.uber.org/zap"
2022
)
2123

22-
func createGenerators(schema *gemini.Schema, schemaConfig gemini.SchemaConfig, distributionFunc gemini.DistributionFunc, actors, distributionSize uint64, logger *zap.Logger) []*gemini.Generator {
24+
func createGenerators(ctx context.Context, schema *gemini.Schema, schemaConfig gemini.SchemaConfig, distributionFunc gemini.DistributionFunc, actors, distributionSize uint64, logger *zap.Logger) []*gemini.Generator {
2325
partitionRangeConfig := gemini.PartitionRangeConfig{
2426
MaxBlobLength: schemaConfig.MaxBlobLength,
2527
MinBlobLength: schemaConfig.MinBlobLength,
@@ -36,7 +38,7 @@ func createGenerators(schema *gemini.Schema, schemaConfig gemini.SchemaConfig, d
3638
Seed: seed,
3739
PkUsedBufferSize: pkBufferReuseSize,
3840
}
39-
g := gemini.NewGenerator(table, gCfg, logger.Named("generator"))
41+
g := gemini.NewGenerator(ctx, table, gCfg, logger.Named("generator"))
4042
gs = append(gs, g)
4143
}
4244
return gs

‎cmd/gemini/jobs.go

Lines changed: 87 additions & 65 deletions
Original file line numberDiff line numberDiff line change
@@ -24,12 +24,12 @@ import (
2424
"github.com/scylladb/gemini/store"
2525
"go.uber.org/zap"
2626
"golang.org/x/exp/rand"
27-
"gopkg.in/tomb.v2"
27+
"golang.org/x/sync/errgroup"
2828
)
2929

3030
// MutationJob continuously applies mutations against the database
3131
// for as long as the pump is active.
32-
func MutationJob(ctx context.Context, pump <-chan heartBeat, schema *gemini.Schema, schemaCfg gemini.SchemaConfig, table *gemini.Table, s store.Store, r *rand.Rand, p gemini.PartitionRangeConfig, g *gemini.Generator, c chan Status, mode string, warmup time.Duration, logger *zap.Logger) {
32+
func MutationJob(ctx context.Context, pump <-chan heartBeat, schema *gemini.Schema, schemaCfg gemini.SchemaConfig, table *gemini.Table, s store.Store, r *rand.Rand, p gemini.PartitionRangeConfig, g *gemini.Generator, c chan Status, mode string, warmup time.Duration, logger *zap.Logger) error {
3333
schemaConfig := &schemaCfg
3434
logger = logger.Named("mutation_job")
3535
testStatus := Status{}
@@ -38,29 +38,39 @@ func MutationJob(ctx context.Context, pump <-chan heartBeat, schema *gemini.Sche
3838
c <- testStatus
3939
}()
4040
var i int
41-
for hb := range pump {
42-
hb.await()
43-
ind := r.Intn(1000000)
44-
if ind%100000 == 0 {
45-
ddl(ctx, schema, schemaConfig, table, s, r, p, &testStatus, logger)
46-
} else {
47-
mutation(ctx, schema, schemaConfig, table, s, r, p, g, &testStatus, true, logger)
48-
}
49-
if i%1000 == 0 {
50-
c <- testStatus
51-
testStatus = Status{}
52-
}
53-
if failFast && (testStatus.ReadErrors > 0 || testStatus.WriteErrors > 0) {
54-
c <- testStatus
55-
return
41+
for {
42+
select {
43+
case <-ctx.Done():
44+
logger.Debug("mutation job terminated")
45+
return ctx.Err()
46+
case hb := <-pump:
47+
hb.await()
48+
ind := r.Intn(1000000)
49+
if ind%100000 == 0 {
50+
if err := ddl(ctx, schema, schemaConfig, table, s, r, p, &testStatus, logger); err != nil {
51+
return err
52+
}
53+
} else {
54+
if err := mutation(ctx, schema, schemaConfig, table, s, r, p, g, &testStatus, true, logger); err != nil {
55+
return err
56+
}
57+
}
58+
if i%1000 == 0 {
59+
c <- testStatus
60+
testStatus = Status{}
61+
}
62+
if failFast && (testStatus.ReadErrors > 0 || testStatus.WriteErrors > 0) {
63+
c <- testStatus
64+
return nil
65+
}
5666
}
5767
i++
5868
}
5969
}
6070

6171
// ValidationJob continuously applies validations against the database
6272
// for as long as the pump is active.
63-
func ValidationJob(ctx context.Context, pump <-chan heartBeat, schema *gemini.Schema, schemaCfg gemini.SchemaConfig, table *gemini.Table, s store.Store, r *rand.Rand, p gemini.PartitionRangeConfig, g *gemini.Generator, c chan Status, mode string, warmup time.Duration, logger *zap.Logger) {
73+
func ValidationJob(ctx context.Context, pump <-chan heartBeat, schema *gemini.Schema, schemaCfg gemini.SchemaConfig, table *gemini.Table, s store.Store, r *rand.Rand, p gemini.PartitionRangeConfig, g *gemini.Generator, c chan Status, mode string, warmup time.Duration, logger *zap.Logger) error {
6474
schemaConfig := &schemaCfg
6575
logger = logger.Named("validation_job")
6676

@@ -69,41 +79,55 @@ func ValidationJob(ctx context.Context, pump <-chan heartBeat, schema *gemini.Sc
6979
c <- testStatus
7080
}()
7181
var i int
72-
for hb := range pump {
73-
hb.await()
74-
validation(ctx, schema, schemaConfig, table, s, r, p, g, &testStatus, logger)
75-
if i%1000 == 0 {
76-
c <- testStatus
77-
testStatus = Status{}
78-
}
79-
if failFast && (testStatus.ReadErrors > 0 || testStatus.WriteErrors > 0) {
80-
return
82+
for {
83+
select {
84+
case <-ctx.Done():
85+
return ctx.Err()
86+
case hb := <-pump:
87+
hb.await()
88+
if cql, err := validation(ctx, schema, schemaConfig, table, s, r, p, g, &testStatus, logger); err != nil {
89+
e := JobError{
90+
Timestamp: time.Now(),
91+
Message: "Validation failed: " + err.Error(),
92+
Query: cql,
93+
}
94+
testStatus.Errors = append(testStatus.Errors, e)
95+
testStatus.ReadErrors++
96+
} else {
97+
testStatus.ReadOps++
98+
}
99+
100+
if i%1000 == 0 {
101+
c <- testStatus
102+
testStatus = Status{}
103+
}
104+
if failFast && (testStatus.ReadErrors > 0 || testStatus.WriteErrors > 0) {
105+
return nil
106+
}
81107
}
82108
i++
83109
}
84110
}
85111

86112
// WarmupJob continuously applies mutations against the database
87113
// for as long as the pump is active or the supplied duration expires.
88-
func WarmupJob(ctx context.Context, pump <-chan heartBeat, schema *gemini.Schema, schemaCfg gemini.SchemaConfig, table *gemini.Table, s store.Store, r *rand.Rand, p gemini.PartitionRangeConfig, g *gemini.Generator, c chan Status, mode string, warmup time.Duration, logger *zap.Logger) {
114+
func WarmupJob(ctx context.Context, pump <-chan heartBeat, schema *gemini.Schema, schemaCfg gemini.SchemaConfig, table *gemini.Table, s store.Store, r *rand.Rand, p gemini.PartitionRangeConfig, g *gemini.Generator, c chan Status, mode string, warmup time.Duration, logger *zap.Logger) error {
89115
schemaConfig := &schemaCfg
90116
testStatus := Status{}
91117
var i int
92118
warmupTimer := time.NewTimer(warmup)
93119
for {
94120
select {
95-
case _, ok := <-pump:
96-
if !ok {
97-
logger.Debug("warmup job terminated")
98-
c <- testStatus
99-
return
100-
}
101-
}
102-
select {
121+
case <-ctx.Done():
122+
logger.Debug("warmup job terminated")
123+
c <- testStatus
124+
return ctx.Err()
103125
case <-warmupTimer.C:
126+
logger.Debug("warmup job finished")
104127
c <- testStatus
105-
return
128+
return nil
106129
default:
130+
// Do we care about errors during warmup?
107131
mutation(ctx, schema, schemaConfig, table, s, r, p, g, &testStatus, false, logger)
108132
if i%1000 == 0 {
109133
c <- testStatus
@@ -113,8 +137,8 @@ func WarmupJob(ctx context.Context, pump <-chan heartBeat, schema *gemini.Schema
113137
}
114138
}
115139

116-
func job(t *tomb.Tomb, f testJob, actors uint64, schema *gemini.Schema, schemaConfig gemini.SchemaConfig, s store.Store, pump *Pump, generators []*gemini.Generator, result chan Status, logger *zap.Logger) {
117-
workerCtx, _ := context.WithCancel(context.Background())
140+
func job(ctx context.Context, f testJob, actors uint64, schema *gemini.Schema, schemaConfig gemini.SchemaConfig, s store.Store, pump *Pump, generators []*gemini.Generator, result chan Status, logger *zap.Logger) error {
141+
g, gCtx := errgroup.WithContext(ctx)
118142
partitionRangeConfig := gemini.PartitionRangeConfig{
119143
MaxBlobLength: schemaConfig.MaxBlobLength,
120144
MinBlobLength: schemaConfig.MinBlobLength,
@@ -123,39 +147,39 @@ func job(t *tomb.Tomb, f testJob, actors uint64, schema *gemini.Schema, schemaCo
123147
}
124148

125149
for j, table := range schema.Tables {
126-
g := generators[j]
150+
gen := generators[j]
127151
for i := 0; i < int(actors); i++ {
128152
r := rand.New(rand.NewSource(seed))
129-
t.Go(func() error {
130-
f(workerCtx, pump.ch, schema, schemaConfig, table, s, r, partitionRangeConfig, g, result, mode, warmup, logger)
131-
return nil
153+
g.Go(func() error {
154+
return f(gCtx, pump.ch, schema, schemaConfig, table, s, r, partitionRangeConfig, gen, result, mode, warmup, logger)
132155
})
133156
}
134157
}
158+
return g.Wait()
135159
}
136160

137-
func ddl(ctx context.Context, schema *gemini.Schema, sc *gemini.SchemaConfig, table *gemini.Table, s store.Store, r *rand.Rand, p gemini.PartitionRangeConfig, testStatus *Status, logger *zap.Logger) {
161+
func ddl(ctx context.Context, schema *gemini.Schema, sc *gemini.SchemaConfig, table *gemini.Table, s store.Store, r *rand.Rand, p gemini.PartitionRangeConfig, testStatus *Status, logger *zap.Logger) error {
138162
if sc.CQLFeature != gemini.CQL_FEATURE_ALL {
139163
logger.Debug("ddl statements disabled")
140-
return
164+
return nil
141165
}
142166
table.Lock()
143167
if len(table.MaterializedViews) > 0 {
144168
// Scylla does not allow changing the DDL of a table with materialized views.
145-
return
169+
return nil
146170
}
147171
defer table.Unlock()
148172
ddlStmts, postStmtHook, err := schema.GenDDLStmt(table, r, p, sc)
149173
if err != nil {
150174
logger.Error("Failed! Mutation statement generation failed", zap.Error(err))
151175
testStatus.WriteErrors++
152-
return
176+
return err
153177
}
154178
if ddlStmts == nil {
155179
if w := logger.Check(zap.DebugLevel, "no statement generated"); w != nil {
156180
w.Write(zap.String("job", "ddl"))
157181
}
158-
return
182+
return nil
159183
}
160184
defer postStmtHook()
161185
defer func() {
@@ -181,20 +205,21 @@ func ddl(ctx context.Context, schema *gemini.Schema, sc *gemini.SchemaConfig, ta
181205
testStatus.WriteOps++
182206
}
183207
}
208+
return nil
184209
}
185210

186-
func mutation(ctx context.Context, schema *gemini.Schema, _ *gemini.SchemaConfig, table *gemini.Table, s store.Store, r *rand.Rand, p gemini.PartitionRangeConfig, g *gemini.Generator, testStatus *Status, deletes bool, logger *zap.Logger) {
211+
func mutation(ctx context.Context, schema *gemini.Schema, _ *gemini.SchemaConfig, table *gemini.Table, s store.Store, r *rand.Rand, p gemini.PartitionRangeConfig, g *gemini.Generator, testStatus *Status, deletes bool, logger *zap.Logger) error {
187212
mutateStmt, err := schema.GenMutateStmt(table, g, r, p, deletes)
188213
if err != nil {
189214
logger.Error("Failed! Mutation statement generation failed", zap.Error(err))
190215
testStatus.WriteErrors++
191-
return
216+
return err
192217
}
193218
if mutateStmt == nil {
194219
if w := logger.Check(zap.DebugLevel, "no statement generated"); w != nil {
195220
w.Write(zap.String("job", "mutation"))
196221
}
197-
return
222+
return err
198223
}
199224
mutateQuery := mutateStmt.Query
200225
token, mutateValues := mutateStmt.Values()
@@ -217,15 +242,16 @@ func mutation(ctx context.Context, schema *gemini.Schema, _ *gemini.SchemaConfig
217242
} else {
218243
testStatus.WriteOps++
219244
}
245+
return nil
220246
}
221247

222-
func validation(ctx context.Context, schema *gemini.Schema, sc *gemini.SchemaConfig, table *gemini.Table, s store.Store, r *rand.Rand, p gemini.PartitionRangeConfig, g *gemini.Generator, testStatus *Status, logger *zap.Logger) {
248+
func validation(ctx context.Context, schema *gemini.Schema, sc *gemini.SchemaConfig, table *gemini.Table, s store.Store, r *rand.Rand, p gemini.PartitionRangeConfig, g *gemini.Generator, testStatus *Status, logger *zap.Logger) (string, error) {
223249
checkStmt := schema.GenCheckStmt(table, g, r, p)
224250
if checkStmt == nil {
225251
if w := logger.Check(zap.DebugLevel, "no statement generated"); w != nil {
226252
w.Write(zap.String("job", "validation"))
227253
}
228-
return
254+
return "", nil
229255
}
230256
checkQuery := checkStmt.Query
231257
token, checkValues := checkStmt.Values()
@@ -243,23 +269,19 @@ func validation(ctx context.Context, schema *gemini.Schema, sc *gemini.SchemaCon
243269
for attempts := 0; attempts < maxAttempts; attempts++ {
244270
logger.Info("validation failed for possible async operation",
245271
zap.Duration("trying_again_in", delay))
246-
time.Sleep(delay)
272+
select {
273+
case <-time.After(delay):
274+
case <-ctx.Done():
275+
return checkStmt.PrettyCQL(), err
276+
}
247277
// Should we sample all the errors?
248278
if err = s.Check(ctx, table, checkQuery, checkValues...); err == nil {
249279
// Result sets stabilized
250-
return
280+
return "", nil
251281
}
252282
}
253283
}
254-
// De-duplication needed?
255-
e := JobError{
256-
Timestamp: time.Now(),
257-
Message: "Validation failed: " + err.Error(),
258-
Query: checkStmt.PrettyCQL(),
259-
}
260-
testStatus.Errors = append(testStatus.Errors, e)
261-
testStatus.ReadErrors++
262-
} else {
263-
testStatus.ReadOps++
284+
return checkStmt.PrettyCQL(), err
264285
}
286+
return "", nil
265287
}

0 commit comments

Comments
 (0)