367
}
368
369
>
func (tr *priTaskReader) signalNewTasks(resp subqueueCreateTasksResponse) {
pri_task_reader.go
370
>
tr.lock.Lock()
371
>
372
>
// We have to be very careful not to increment the read level past an ID that will somehow
373
>
// end up in the database, otherwise we might lose a task. We do this by verifying that our
374
>
// read level was equal to the previous max read level (i.e. we were at the end of the
375
>
// queue), and then we set it to the max read level as of CreateTasks.
376
>
// We also check that there's room in memory.
377
>
canAddDirect := tr.readLevel == resp.maxReadLevelBefore &&
378
>
(tr.loadedTasks+len(resp.tasks)) <= tr.backlogMgr.config.GetTasksBatchSize() &&
379
>
!slices.ContainsFunc(resp.tasks, func(t *persistencespb.AllocatedTaskInfo) bool {
380
// Because we checked readLevel, we know that getTasksPump can't have beat us to
381
// adding these tasks to outstandingTasks. So they should definitely not be there.