Vihren Atlas

pri_backlog_manager.go ×9

Code frontier

unlabeled · c_6464e494b225

232 tests · 3242 LOC · 149 files · introduces 0 tests · 88 LOC · 3 files

Introduces — evidence that enters the hierarchy at this concept

Code
24 ranges88 lines · 3 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
505 ranges3242 lines · 149 files · Browse complete extent
All tests (intent)
232 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.

3 files ranked by introduced lines: 88 introduced LOC across 24 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/matching/pri_backlog_manager.go 35 introduced LOC · 9 ranges

Open complete file

106 // of a newer lease by another backlogManager. Returns true if the unload signal
107 // is emitted, false otherwise.
108 > func (c *priBacklogManagerImpl) signalIfFatal(err error) bool { pri_backlog_manager.go
109 > if err == nil {
110 > return false
111 > }
112 var condfail *persistence.ConditionFailedError
113 if errors.As(err, &condfail) {
120 }
121
122 > func (c *priBacklogManagerImpl) Start() { pri_backlog_manager.go
123 > c.taskWriter.Start()
124 > }
125
126 func (c *priBacklogManagerImpl) Stop() {
146 }
147
148 > func (c *priBacklogManagerImpl) initState(state taskQueueState, err error) { pri_backlog_manager.go
149 > defer c.initializedError.Set(struct{}{}, err)
150 >
151 > if err != nil {
152 // We can't recover from here without starting over, so unload the whole task queue.
153 // Skip final update since we never initialized.
159 // Pass scale info back to physical tq from unversioned (default) queue.
160 // This must be done before c.initializedError.Set().
161 > if c.queueKey().Partition().IsRoot() && !c.queueKey().IsVersioned() && !c.isDraining { pri_backlog_manager.go
162 c.pqMgr.StartScaleManager(state.scaleState)
163 }
164
165 > if state.otherHasTasks { pri_backlog_manager.go
166 c.pqMgr.SetupDraining()
167 }
168
169 > c.subqueueLock.Lock() pri_backlog_manager.go
170 > defer c.subqueueLock.Unlock()
171 >
172 > c.loadSubqueuesLocked(state.subqueues)
173 > go c.periodicSync()
174 }
175
179 }
180
181 > func (c *priBacklogManagerImpl) loadSubqueuesLocked(subqueues []persistencespb.SubqueueInfo) { pri_backlog_manager.go
182 > // TODO(pri): This assumes that subqueues never shrinks, and priority/fairness index of
183 > // existing subqueues never changes. If we change that, this logic will need to change.
184 > for i := range subqueues {
185 > if i >= len(c.subqueues) {
186 > r := newPriTaskReader(c, subqueueIndex(i), subqueues[i].AckLevel)
187 > r.Start()
188 > c.subqueues = append(c.subqueues, r)
189 > }
190 > c.subqueuesByPriority[priorityKey(subqueues[i].Key.Priority)] = subqueueIndex(i)
191 > c.priorityBySubqueue[subqueueIndex(i)] = priorityKey(subqueues[i].Key.Priority)
192 }
193 }
228 }
229
230 > func (c *priBacklogManagerImpl) periodicSync() { pri_backlog_manager.go
231 > for {
232 > select {
233 case <-c.tqCtx.Done():
234 return
397 // }
398
399 > func (c *priBacklogManagerImpl) queueKey() *PhysicalTaskQueueKey { pri_backlog_manager.go
400 > return c.pqMgr.QueueKey()
401 > }
402
403 func (c *priBacklogManagerImpl) getDB() *taskQueueDB {
go.temporal.io/server/service/matching/pri_task_writer.go 28 introduced LOC · 7 ranges

Open complete file

62
63 // Start priTaskWriter background goroutine.
64 > func (w *priTaskWriter) Start() { pri_task_writer.go
65 > go w.taskWriterLoop()
66 > }
67
68 func (w *priTaskWriter) appendTask(
138 }
139
140 > func (w *priTaskWriter) initState() error { pri_task_writer.go
141 > state, err := w.renewLeaseWithRetry(foreverRetryPolicy, common.IsPersistenceTransientError)
142 > if err != nil {
143 w.backlogMgr.initState(taskQueueState{}, err)
144 return err
145 }
146 > w.taskIDBlock = rangeIDToTaskIDBlock(state.rangeID, w.config.RangeSize) pri_task_writer.go
147 > w.currentTaskIDBlock = w.taskIDBlock
148 > w.backlogMgr.initState(state, nil)
149 > return nil
150 }
151
152 > func (w *priTaskWriter) taskWriterLoop() { pri_task_writer.go
153 > if w.initState() != nil {
154 return
155 }
156
157 > var reqs []*writeTaskRequest pri_task_writer.go
158 > for {
159 > atomic.StoreInt64(&w.currentTaskIDBlock.start, w.taskIDBlock.start)
160 > atomic.StoreInt64(&w.currentTaskIDBlock.end, w.taskIDBlock.end)
161 >
162 > select {
163 case request := <-w.appendCh:
164 // read a batch of requests from the channel
195 retryPolicy backoff.RetryPolicy,
196 retryErrors backoff.IsRetryable,
197 > ) (taskQueueState, error) { pri_task_writer.go
198 > var newState taskQueueState
199 > op := func(ctx context.Context) (err error) {
200 > newState, err = w.db.RenewLease(ctx)
201 > return
202 > }
203 > metrics.LeaseRequestPerTaskQueueCounter.With(w.backlogMgr.metricsHandler).Record(1)
204 > err := backoff.ThrottleRetryContext(w.backlogMgr.tqCtx, op, retryPolicy, retryErrors)
205 > if err != nil {
206 metrics.LeaseFailurePerTaskQueueCounter.With(w.backlogMgr.metricsHandler).Record(1)
207 return newState, err
208 }
209 > return newState, nil pri_task_writer.go
210 }
211
go.temporal.io/server/service/matching/pri_task_reader.go 25 introduced LOC · 8 ranges

Open complete file

92
93 // Start priTaskReader background goroutines.
94 > func (tr *priTaskReader) Start() { pri_task_reader.go
95 > go tr.getTasksPump()
96 > }
97
98 > func (tr *priTaskReader) SignalTaskLoading() { pri_task_reader.go
99 > select {
100 > case tr.notifyC <- struct{}{}:
101 default: // channel already has an event, don't block
102 }
156
157 // nolint:revive // can simplify later
158 > func (tr *priTaskReader) getTasksPump() { pri_task_reader.go
159 > ctx := tr.backlogMgr.tqCtx
160 >
161 > tr.SignalTaskLoading() // prime pump
162 > for {
163 > select {
164 case <-ctx.Done():
165 return
166 > case <-tr.notifyC: pri_task_reader.go
167 }
168
169 > if tr.getLoadedTasks() > tr.backlogMgr.config.GetTasksReloadAt() { pri_task_reader.go
170 // Too many loaded already, ignore this signal. We'll get another signal when
171 // loadedTasks drops low enough.
173 }
174
175 > batch, err := tr.getTaskBatch(ctx) pri_task_reader.go
176 > tr.backlogMgr.signalIfFatal(err)
177 > if err != nil {
178 // TODO: Should we ever stop retrying on db errors?
179 if common.IsResourceExhausted(err) {
184 continue
185 }
186 > tr.retrier.Reset() pri_task_reader.go
187 >
188 > if len(batch.tasks) == 0 {
189 tr.setReadLevelAfterGap(batch.readLevel)
190 if !batch.isReadBatchDone {
417 // ack manager
418
419 > func (tr *priTaskReader) getLoadedTasks() int { pri_task_reader.go
420 > tr.lock.Lock()
421 > defer tr.lock.Unlock()
422 > return tr.loadedTasks
423 > }
424
425 // isDrained returns true if this subqueue has been fully drained:

Direct hierarchy

Parents

Children