Vihren Atlas

fair_task_writer.go ×17

Code frontier

unlabeled · c_590d23b76764

78 tests · 3807 LOC · 153 files · introduces 0 tests · 158 LOC · 5 files

Introduces — evidence that enters the hierarchy at this concept

Code
36 ranges158 lines · 5 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
648 ranges3807 lines · 153 files · Browse complete extent
All tests (intent)
78 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.

5 files ranked by introduced lines: 158 introduced LOC across 36 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/matching/fair_task_writer.go 66 introduced LOC · 17 ranges

Open complete file

67 case <-w.backlogMgr.tqCtx.Done():
68 return errShutdown
69 > default: fair_task_writer.go
70 // noop
71 }
72
73 > startTime := time.Now().UTC() fair_task_writer.go
74 > ch := make(chan error, 1)
75 > req := &writeTaskRequest{
76 > taskInfo: taskInfo,
77 > responseCh: ch,
78 > subqueue: subqueue,
79 > }
80 >
81 > select {
82 > case w.appendCh <- req:
83 > select {
84 > case err := <-ch:
85 > metrics.TaskWriteLatencyPerTaskQueue.With(w.backlogMgr.metricsHandler).Record(time.Since(startTime))
86 > return err
87 case <-w.backlogMgr.tqCtx.Done():
88 // if we are shutting down, this request will never make
100 }
101
102 > func (w *fairTaskWriter) allocTaskIDs(reqs []*writeTaskRequest) error { fair_task_writer.go
103 > for i := range reqs {
104 > if w.taskIDBlock.start > w.taskIDBlock.end {
105 // we ran out of current allocation block
106 newBlock, err := w.allocTaskIDBlock(w.taskIDBlock.end)
110 w.taskIDBlock = newBlock
111 }
112 > reqs[i].id = w.taskIDBlock.start fair_task_writer.go
113 > w.taskIDBlock.start++
114 }
115 > return nil fair_task_writer.go
116 }
117
118 > func (w *fairTaskWriter) pickPasses(tasks []*writeTaskRequest, bases []fairLevel) { fair_task_writer.go
119 > // Fetch latest fairness weight overrides from the partition's rate limit manager via pqMgr
120 > overrides := w.backlogMgr.pqMgr.GetFairnessWeightOverrides()
121 > dither := w.config.FairnessPassDither()
122 >
123 > for i, task := range tasks {
124 > pri := task.taskInfo.Priority
125 > key := pri.GetFairnessKey()
126 > weight := getEffectiveWeight(overrides, pri)
127 > inc := max(1, int64(strideFactor/weight))
128 > base := bases[task.subqueue].pass
129 > if dither {
130 base = ditherPass(w.ditherSeed, key, base, inc)
131 }
132 > cntr := w.counters[task.subqueue] fair_task_writer.go
133 > if cntr == nil {
134 > cntr = w.counterFactory(task.subqueue)
135 > w.counters[task.subqueue] = cntr
136 > }
137 > pass := cntr.GetPass(key, base, inc)
138 > softassert.That(w.logger, pass >= base, "counter returned pass below base")
139 > tasks[i].pass = pass
140 }
141 }
174 case <-w.backlogMgr.tqCtx.Done():
175 return
176 > case req := <-w.appendCh: fair_task_writer.go
177 > // read a batch of requests from the channel
178 > reqs = append(reqs, req)
179 > reqs = w.getWriteBatch(reqs)
180 }
181
182 > err := w.allocTaskIDs(reqs) fair_task_writer.go
183 > if err == nil {
184 > err = w.writeBatch(reqs)
185 > }
186
187 > for _, req := range reqs { fair_task_writer.go
188 > req.responseCh <- err
189 > }
190
191 // maybe persist fairness key counts if it's time
192 > select { fair_task_writer.go
193 case <-persistFairnessKeys:
194 for subqueue, cntr := range w.counters {
200 }
201
202 > func (w *fairTaskWriter) getWriteBatch(reqs []*writeTaskRequest) []*writeTaskRequest { fair_task_writer.go
203 > for range w.config.MaxTaskBatchSize() - 1 {
204 > select {
205 case req := <-w.appendCh:
206 reqs = append(reqs, req)
207 > default: // channel is empty, don't block fair_task_writer.go
208 > return reqs
209 }
210 }
212 }
213
214 > func (w *fairTaskWriter) writeBatch(reqs []*writeTaskRequest) (retErr error) { fair_task_writer.go
215 > bases, unpin := w.backlogMgr.getAndPinAckLevels()
216 > defer func() { unpin(retErr) }()
217
218 > w.pickPasses(reqs, bases) fair_task_writer.go
219 > resp, err := w.db.CreateFairTasks(w.backlogMgr.tqCtx, reqs)
220 > if err == nil {
221 w.backlogMgr.wroteNewTasks(resp) // must be called before unpin()
222 > } else { fair_task_writer.go
223 w.logger.Error("Persistent store operation failure", tag.StoreOperationCreateTask, tag.Error(err))
224 w.backlogMgr.signalIfFatal(err)
225 }
226 > return err fair_task_writer.go
227 }
228
go.temporal.io/server/service/matching/db.go 52 introduced LOC · 10 ranges

Open complete file

439 }
440
441 > func (db *taskQueueDB) getTopKFairnessKeys(subqueue subqueueIndex) []counter.TopKEntry { db.go
442 > db.Lock()
443 > defer db.Unlock()
444 >
445 > if subqueue >= subqueueIndex(len(db.subqueues)) {
446 return nil
447 }
448 > counts := db.subqueues[subqueue].TopKFairnessCounts db.go
449 > entries := make([]counter.TopKEntry, len(counts))
450 > for i, count := range counts {
451 entries[i] = counter.TopKEntry{Key: count.Key, Count: count.Count}
452 }
453 > return entries db.go
454 }
455
597 ctx context.Context,
598 reqs []*writeTaskRequest,
599 > ) (createFairTasksResponse, error) { db.go
600 > if db.isDraining {
601 return createFairTasksResponse{}, softassert.UnexpectedInternalErr(db.logger, "CreateTasks can't be used in draining mode", nil)
602 }
603
604 > db.Lock() db.go
605 > defer db.Unlock()
606 >
607 > if len(reqs) == 0 {
608 return nil, nil
609 }
610
611 > newTasks := make(createFairTasksResponse) db.go
612 > newMaxLevel := make(map[subqueueIndex]fairLevel)
613 > allTasks := make([]*persistencespb.AllocatedTaskInfo, len(reqs))
614 > allSubqueues := make([]int, len(reqs))
615 > for i, req := range reqs {
616 > task := &persistencespb.AllocatedTaskInfo{
617 > TaskId: req.id,
618 > TaskPass: req.pass,
619 > Data: req.taskInfo,
620 > }
621 > allTasks[i] = task
622 > allSubqueues[i] = int(req.subqueue)
623 > newTasks[req.subqueue] = append(newTasks[req.subqueue], task)
624 > newMaxLevel[req.subqueue] = newMaxLevel[req.subqueue].max(req.fairLevel)
625 > }
626
627 > for sq, tasks := range newTasks { db.go
628 > db.subqueues[sq].ApproximateBacklogCount += int64(len(tasks))
629 > }
630
631 // Unlike in CreateTasks, we can set the persisted FairMaxReadLevel before persisting.
633 // the FairMaxReadLevel will be more up-to-date. The max read level is not used by
634 // fairTaskReader, so there's no correctness issue with doing this.
635 > for sq, level := range newMaxLevel { db.go
636 > db.subqueues[sq].FairMaxReadLevel = fairLevelFromProto(db.subqueues[sq].FairMaxReadLevel).max(level).toProto()
637 > }
638
639 > updateMetadata := db.shouldUpdateMetadataOnAppendLocked() db.go
640 >
641 > resp, err := db.store.CreateTasks(
642 > ctx,
643 > &persistence.CreateTasksRequest{
644 > TaskQueueInfo: &persistence.PersistedTaskQueueInfo{
645 > Data: db.cachedQueueInfo(),
646 > RangeID: db.rangeID,
647 > },
648 > Tasks: allTasks,
649 > Subqueues: allSubqueues,
650 > UpdateMetadata: updateMetadata,
651 > })
652 >
653 > if err == nil {
654 // Only update lastWrite for persistence implementations that update metadata on CreateTasks,
655 // otherwise we have a change to ApproximateBacklogCount we need to write.
667 }
668 }
669 > return newTasks, err db.go
670 }
671
go.temporal.io/server/service/matching/fair_backlog_manager.go 19 introduced LOC · 4 ranges

Open complete file

242 }
243
244 > func (c *fairBacklogManagerImpl) getAndPinAckLevels() ([]fairLevel, func(error)) { fair_backlog_manager.go
245 > c.subqueueLock.Lock()
246 > subqueues := slices.Clone(c.subqueues)
247 > c.subqueueLock.Unlock()
248 >
249 > levels := make([]fairLevel, len(subqueues))
250 > for i, s := range subqueues {
251 > levels[i] = s.getAndPinAckLevel()
252 > }
253 > unpin := func(writeErr error) {
254 > for _, s := range subqueues {
255 > s.unpinAckLevel(writeErr)
256 > }
257 }
258 > return levels, unpin fair_backlog_manager.go
259 }
260
403 }
404
405 > func (c *fairBacklogManagerImpl) newCounterForSubqueue(subqueue subqueueIndex) counter.Counter { fair_backlog_manager.go
406 > cntr := c.counterFactory()
407 > // restore persisted keys
408 > for _, entry := range c.db.getTopKFairnessKeys(subqueue) {
409 _ = cntr.GetPass(entry.Key, entry.Count, 0)
410 }
411 > return cntr fair_backlog_manager.go
412 }
413
go.temporal.io/server/service/matching/fair_task_reader.go 18 introduced LOC · 4 ranges

Open complete file

597 func (tr *fairTaskReader) advanceAckLevelLocked() {
598 if tr.ackLevelPinnedLocked() {
599 > return fair_task_reader.go
600 > }
601
602 // Adjust the ack level as far as we can
623 }
624
625 > func (tr *fairTaskReader) getAndPinAckLevel() fairLevel { fair_task_reader.go
626 > tr.lock.Lock()
627 > defer tr.lock.Unlock()
628 >
629 > softassert.That(tr.logger, !tr.ackLevelPinnedByWriter, "ack level already pinned")
630 > tr.ackLevelPinnedByWriter = true
631 > return tr.ackLevel
632 > }
633
634 > func (tr *fairTaskReader) unpinAckLevel(writeErr error) { fair_task_reader.go
635 > tr.lock.Lock()
636 > defer tr.lock.Unlock()
637 >
638 > if writeErr != nil {
639 // We got an error writing but the write may have succeeded anyway.
640 // We can't assume we know where the end is anymore.
644 }
645
646 > softassert.That(tr.logger, tr.ackLevelPinnedByWriter, "ack level wasn't pinned") fair_task_reader.go
647 > tr.ackLevelPinnedByWriter = false
648 > tr.advanceAckLevelLocked()
649 }
650
go.temporal.io/server/service/matching/config.go 3 introduced LOC · 1 range

Open complete file

580 return config.FairnessCounter(ns.String(), taskQueueName, taskType)
581 },
582 > FairnessPassDither: func() bool { config.go
583 > return config.FairnessPassDither(ns.String(), taskQueueName, taskType)
584 > },
585 PartitionScaleAllowedDrift: func() dynamicconfig.PartitionScaleAllowedDrift {
586 return config.PartitionScaleAllowedDrift(ns.String(), taskQueueName, taskType)

Direct hierarchy

Parents

Children