516
ctx context.Context,
517
reqs []*writeTaskRequest,
519
>
if db.isDraining {
520
return createTasksResponse{}, softassert.UnexpectedInternalErr(db.logger, "CreateTasks can't be used in draining mode", nil)
521
}
522
524
>
defer db.Unlock()
525
>
526
>
if len(reqs) == 0 {
527
return createTasksResponse{}, nil
528
}
529
531
>
allTasks := make([]*persistencespb.AllocatedTaskInfo, len(reqs))
532
>
allSubqueues := make([]int, len(reqs))
533
>
for i, req := range reqs {
534
>
task := &persistencespb.AllocatedTaskInfo{
535
>
TaskId: req.id,
536
>
Data: req.taskInfo,
537
>
}
538
>
allTasks[i] = task
539
>
allSubqueues[i] = int(req.subqueue)
540
>
541
>
u := updates[req.subqueue]
542
>
updates[req.subqueue] = subqueueCreateTasksResponse{
543
>
tasks: append(u.tasks, task),
544
>
maxReadLevelBefore: db.getMaxReadLevelLocked(req.subqueue),
545
>
maxReadLevelAfter: task.TaskId, // task ids are in order so this is the max
546
>
}
547
>
}
548
550
>
db.subqueues[sq].ApproximateBacklogCount += int64(len(update.tasks))
551
>
}
552
553
// Decide whether to include metadata in the write. We always need the LWT for the
554
// range ID check, but updating the full metadata blob on every append has extra cost.
555
// We piggyback the metadata update if enough time has passed since the last write.
557
>
558
>
resp, err := db.store.CreateTasks(
559
>
ctx,
560
>
&persistence.CreateTasksRequest{
561
>
TaskQueueInfo: &persistence.PersistedTaskQueueInfo{
562
>
Data: db.cachedQueueInfo(),
563
>
RangeID: db.rangeID,
564
>
},
565
>
Tasks: allTasks,
566
>
Subqueues: allSubqueues,
567
>
UpdateMetadata: updateMetadata,
568
>
})
569
>
570
>
// Update the maxReadLevel after the writes are completed, but before we send the response,
571
>
// so that taskReader is guaranteed to see the new read level when SpoolTask wakes it up.
572
>
// Do this even if the write fails, we won't reuse the task ids.
573
>
for sq, update := range updates {
574
>
db.subqueues[sq].maxReadLevel = update.maxReadLevelAfter
575
>
}
576
578
// Only update lastWrite for persistence implementations that update metadata on CreateTasks,
579
// otherwise we have a change to ApproximateBacklogCount we need to write.