mirrors/apollo-backend
2
0
Fork 0
mirror of https://github.com/christianselig/apollo-backend synced 2024-11-14 16:07:42 +00:00
Code Issues Projects Releases Packages Wiki Activity

parallelize scheduler better

Browse source
This commit is contained in:
Andre Medeiros 2022-06-04 10:21:29 -04:00
parent b29f120eea
commit b23a158471
1 changed files with 34 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

58
internal/cmd/scheduler.go
View file

@ -7,6 +7,7 @@ import (
"net/http" "net/http"
_ "net/http/pprof" _ "net/http/pprof"
"strconv" "strconv"
"sync"
"time" "time"
"github.com/DataDog/datadog-go/statsd" "github.com/DataDog/datadog-go/statsd"
@ -92,7 +93,7 @@ func SchedulerCmd(ctx context.Context) *cobra.Command {
} }
s := gocron.NewScheduler(time.UTC) s := gocron.NewScheduler(time.UTC)
_, _ = s.Every(500).Milliseconds().Do(func() { enqueueAccounts(ctx, logger, statsd, db, redis, luaSha, notifQueue) }) _, _ = s.Every(500).Milliseconds().SingletonMode().Do(func() { enqueueAccounts(ctx, logger, statsd, db, redis, luaSha, notifQueue) })
_, _ = s.Every(5).Second().Do(func() { enqueueSubreddits(ctx, logger, statsd, db, []rmq.Queue{subredditQueue, trendingQueue}) }) _, _ = s.Every(5).Second().Do(func() { enqueueSubreddits(ctx, logger, statsd, db, []rmq.Queue{subredditQueue, trendingQueue}) })
_, _ = s.Every(5).Second().Do(func() { enqueueUsers(ctx, logger, statsd, db, userQueue) }) _, _ = s.Every(5).Second().Do(func() { enqueueUsers(ctx, logger, statsd, db, userQueue) })
_, _ = s.Every(5).Second().Do(func() { cleanQueues(logger, queue) }) _, _ = s.Every(5).Second().Do(func() { cleanQueues(logger, queue) })
@ -441,38 +442,47 @@ func enqueueAccounts(ctx context.Context, logger *zap.Logger, statsd *statsd.Cli
logger.Debug("enqueueing account batch", zap.Int("count", len(ids)), zap.Time("start", now)) logger.Debug("enqueueing account batch", zap.Int("count", len(ids)), zap.Time("start", now))
batches := (idslen / batchSize) + 1
wg := sync.WaitGroup{}
wg.Add(batches)
// Split ids in batches // Split ids in batches
for i := 0; i < idslen; i += batchSize { for i := 0; i < idslen; i += batchSize {
j := i + batchSize go func(offset int) {
if j > idslen { defer wg.Done()
j = idslen
}
batch := Int64Slice(ids[i:j])
logger.Debug("enqueueing batch", zap.Int("len", len(batch))) j := offset + batchSize
if j > idslen {
j = idslen
}
batch := Int64Slice(ids[offset:j])
res, err := redisConn.EvalSha(ctx, luaSha, []string{"locks:accounts"}, batch).Result() logger.Debug("enqueueing batch", zap.Int("len", len(batch)))
if err != nil {
logger.Error("failed to check for locked accounts", zap.Error(err))
}
vals := res.([]interface{}) res, err := redisConn.EvalSha(ctx, luaSha, []string{"locks:accounts"}, batch).Result()
skipped += len(batch) - len(vals) if err != nil {
enqueued += len(vals) logger.Error("failed to check for locked accounts", zap.Error(err))
}
if len(vals) == 0 { vals := res.([]interface{})
continue skipped += len(batch) - len(vals)
} enqueued += len(vals)
batchIds := make([]string, len(vals)) if len(vals) == 0 {
for k, v := range vals { return
batchIds[k] = strconv.FormatInt(v.(int64), 10) }
}
if err = queue.Publish(batchIds...); err != nil { batchIds := make([]string, len(vals))
logger.Error("failed to enqueue account batch", zap.Error(err)) for k, v := range vals {
} batchIds[k] = strconv.FormatInt(v.(int64), 10)
}
if err = queue.Publish(batchIds...); err != nil {
logger.Error("failed to enqueue account batch", zap.Error(err))
}
}(i * batchSize)
} }
wg.Wait()
logger.Debug("done enqueueing account batch", zap.Int("count", enqueued), zap.Int("skipped", skipped), zap.Time("start", now)) logger.Debug("done enqueueing account batch", zap.Int("count", enqueued), zap.Int("skipped", skipped), zap.Time("start", now))
} }