this repo has no description
Perf tuning
+18
-13
+4
-4
backfill/next/backfill.go
+4
-4
backfill/next/backfill.go
···
60
60
Store Store
61
61
62
62
perPDSBackfillConcurrency int
63
-
perPDSSyncsPerSecond int
63
+
perPDSSyncsPerSecond float64
64
64
globalRecordCreateConcurrency int
65
65
66
66
globalRecordCreationLimiter *rate.Limiter
···
75
75
76
76
type BackfillerOptions struct {
77
77
PerPDSBackfillConcurrency int
78
-
PerPDSSyncsPerSecond int
78
+
PerPDSSyncsPerSecond float64
79
79
GlobalRecordCreateConcurrency int
80
80
NSIDFilter string
81
81
Client *http.Client
···
196
196
ParallelRecordCreates int
197
197
RecordCreateLimiter *rate.Limiter
198
198
NSIDFilter string
199
-
SyncRequestsPerSecond int
199
+
SyncRequestsPerSecond float64
200
200
Client *http.Client
201
201
}
202
202
···
235
235
recordCreateConcurrency: opts.ParallelRecordCreates,
236
236
recordsProcessed: backfillRecordsProcessed.WithLabelValues(name),
237
237
NSIDFilter: opts.NSIDFilter,
238
-
syncLimiter: rate.NewLimiter(rate.Limit(opts.SyncRequestsPerSecond), opts.SyncRequestsPerSecond),
238
+
syncLimiter: rate.NewLimiter(rate.Limit(opts.SyncRequestsPerSecond), int(opts.SyncRequestsPerSecond)),
239
239
recordCreateLimiter: opts.RecordCreateLimiter,
240
240
stop: make(chan struct{}),
241
241
client: opts.Client,
+13
-8
cmd/linear/main.go
+13
-8
cmd/linear/main.go
···
2
2
3
3
import (
4
4
"context"
5
-
"encoding/json"
6
5
"fmt"
7
6
"log/slog"
8
7
"net/http"
···
16
15
"sync/atomic"
17
16
"syscall"
18
17
"time"
18
+
19
+
"github.com/goccy/go-json"
19
20
20
21
comatproto "github.com/bluesky-social/indigo/api/atproto"
21
22
"github.com/bluesky-social/indigo/atproto/data"
···
156
157
linear.startWriters()
157
158
158
159
opts := backfill.DefaultBackfillerOptions()
159
-
opts.GlobalRecordCreateConcurrency = 100_000
160
-
opts.PerPDSSyncsPerSecond = 10
161
-
opts.PerPDSBackfillConcurrency = 30
160
+
opts.GlobalRecordCreateConcurrency = 300_000
161
+
opts.PerPDSSyncsPerSecond = 9.5
162
+
opts.PerPDSBackfillConcurrency = 20
162
163
163
164
bf := backfill.NewBackfiller("linear-backfiller-v2", store, linear.handleCreate, opts)
164
165
···
172
173
xrpcc := xrpc.Client{}
173
174
xrpcc.Host = cctx.String("relay-host")
174
175
175
-
limiter := rate.NewLimiter(5, 1)
176
+
limiter := rate.NewLimiter(2, 1)
176
177
177
178
newPDSList := []string{}
178
179
···
202
203
logger.Info("discovered PDSs", "count", len(newPDSList), "pdsList", newPDSList)
203
204
}
204
205
206
+
listCtx, cancel := context.WithCancel(ctx)
207
+
defer cancel()
208
+
205
209
go func() {
206
210
for _, pds := range pdsList {
207
211
// Ensure the PDS host is valid
···
227
231
228
232
cursor := ""
229
233
for {
230
-
if err := listLimiter.Wait(ctx); err != nil {
234
+
if err := listLimiter.Wait(listCtx); err != nil {
231
235
logger.Error("failed to wait for rate limiter", "pds", pds, "err", err)
232
236
break
233
237
}
234
238
235
-
page, err := comatproto.SyncListRepos(ctx, &xrpcc, cursor, 1000)
239
+
page, err := comatproto.SyncListRepos(listCtx, &xrpcc, cursor, 1000)
236
240
if err != nil {
237
241
logger.Error("failed to list repos for PDS", "pds", pds, "err", err)
238
242
break
···
246
250
continue
247
251
}
248
252
249
-
if err := bf.EnqueueJob(ctx, pds, repo.Did); err != nil {
253
+
if err := bf.EnqueueJob(listCtx, pds, repo.Did); err != nil {
250
254
logger.Error("failed to enqueue job for PDS", "pds", pds, "repo", repo.Did, "err", err)
251
255
} else {
252
256
logger.Debug("enqueued job for PDS", "pds", pds, "repo", repo.Did)
···
269
273
signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)
270
274
271
275
<-signals
276
+
cancel()
272
277
if err := linear.shutdown(ctx); err != nil {
273
278
logger.Error("shutdown encountered an error", "err", err)
274
279
}
+1
-1
go.mod
+1
-1
go.mod
···
16
16
github.com/dustinkirkland/golang-petname v0.0.0-20231002161417-6a283f1aaaf2
17
17
github.com/flosch/pongo2/v6 v6.0.0
18
18
github.com/go-redis/cache/v9 v9.0.0
19
+
github.com/goccy/go-json v0.10.2
19
20
github.com/gocql/gocql v1.7.0
20
21
github.com/golang-jwt/jwt v3.2.2+incompatible
21
22
github.com/golang-jwt/jwt/v5 v5.2.2
···
91
92
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
92
93
github.com/getsentry/sentry-go v0.27.0 // indirect
93
94
github.com/go-redis/redis v6.15.9+incompatible // indirect
94
-
github.com/goccy/go-json v0.10.2 // indirect
95
95
github.com/golang/snappy v0.0.4 // indirect
96
96
github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed // indirect
97
97
github.com/hashicorp/golang-lru v1.0.2 // indirect