Vihren Atlas

fair_task_reader.go ×9

Code frontier

unlabeled · c_8b7d1fea9030

80 tests · 3654 LOC · 152 files · introduces 0 tests · 16 LOC · 1 file

Introduces — evidence that enters the hierarchy at this concept

Code
9 ranges16 lines · 1 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
619 ranges3654 lines · 152 files · Browse complete extent
All tests (intent)
80 testsBrowse complete intent

Neighbourhood graph

The orange circle is the focus. Violet and green circles are every ancestor and descendant, broader and narrower, at any distance; blue squares and pink diamonds are the introduced files and exact introduced tests of every visible concept, not only the focus's. Arrows point from broader to narrower concepts and bridge only concepts omitted from this view. Undirected links show source or test introduction. Concept and file size follows LOC; exact test nodes use test-count units.

Introduced files, introduced tests, and structurally relevant concept specialization

In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the native relationship evidence on this page.

Graph controls are ready.

Interactive rendering requires JavaScript and WebGL. Use the native relationship evidence on this page while the interactive map is unavailable.

Native relationship evidence

Every exact file and test below is linked only from the concept that introduces it.

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

No tests are introduced at this concept. Its intent tests are introduced by other concepts.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

1 file ranked by introduced lines: 16 introduced LOC across 9 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/matching/fair_task_reader.go 16 introduced LOC · 9 ranges

Open complete file

418 // (2) Note these values are *AllocatedTaskInfo.
419 for _, t := range tasks {
420 > level := fairLevelFromAllocatedTask(t) fair_task_reader.go
421 > if !tr.ackLevel.less(level) {
422 // Reads may race with completes/acks such that we read some tasks that are already
423 // acked. We should ignore these.
424 continue
425 > } else if mode == mergeWrite && !tr.atEnd && tr.readLevel.less(level) { fair_task_reader.go
426 // If we're writing and we're not at the end, then we have to ignore tasks
427 // above readLevel since we don't know what's in between readLevel and there.
428 continue
429 > } else if _, have := tr.outstandingTasks.Get(level); have { fair_task_reader.go
430 // If write/read race or we have to re-read a range, we may read something we had
431 // already added to the matcher or acked. Ignore tasks we already have.
435 // regular tasks and are turned back into acks in the final loop below, the same way
436 // expired tasks are handled.
437 > merged.Put(level, t) fair_task_reader.go
438 }
439
445 tasks = tasks[:0] // reuse incoming slice to avoid an allocation
446 for b := 0; b < batchSize && it.Next(); b++ {
447 > if t, ok := it.Value().(*persistencespb.AllocatedTaskInfo); ok { fair_task_reader.go
448 > // new task we need to add to the matcher
449 > tasks = append(tasks, t)
450 > }
451 > highestLevel = it.Key().(fairLevel) // nolint:revive
452 }
453
454 if highestLevel.id != 0 {
455 > // If we have any tasks at all in memory, set readLevel to the maximum of that set. fair_task_reader.go
456 > tr.readLevel = highestLevel
457 } else {
458 // Otherwise start reading at ack level next.
498 internalTasks := make([]*internalTask, 0, len(tasks))
499 for _, t := range tasks {
500 > level := fairLevelFromAllocatedTask(t) fair_task_reader.go
501 > if _, have := tr.evictedAcks.Delete(level); have {
502 // This task was already acked, but its ack was evicted from memory before it could
503 // advance the ack level, and now we've re-read it. Add it back as a pre-acked (nil)
510 continue
511 }
512 > if IsTaskExpired(t) { fair_task_reader.go
513 // Expired tasks are added as pre-acked (nil) so they participate in
514 // readLevel calculation above and advance ackLevel + get GC'd below.
606 if minLevel == nil {
607 break
608 > } else if _, ok := v.(*internalTask); ok { fair_task_reader.go
609 break
610 }

Direct hierarchy

Parents

Children