Vihren Atlas

workflow.go ×25

Code frontier

unlabeled · c_008c0ac2cdab

43 tests · 2642 LOC · 119 files · introduces 0 tests · 133 LOC · 1 file

Introduces — evidence that enters the hierarchy at this concept

Code
25 ranges133 lines · 1 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
419 ranges2642 lines · 119 files · Browse complete extent
All tests (intent)
43 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: 133 introduced LOC across 25 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/worker/scheduler/workflow.go 133 introduced LOC · 25 ranges

Open complete file

362
363 // process backfills if we have any too
364 > s.processBackfills() workflow.go
365 > // try starting workflows in the buffer
366 > //nolint:revive
367 > for s.processBuffer() {
368 }
369 > s.updateMemoAndSearchAttributes() workflow.go
370 >
371 > // if schedule is not paused and out of actions or do not have anything scheduled, exit the schedule workflow after retention period has passed
372 > if exp := s.getRetentionExpiration(nextWakeup); !exp.IsZero() && !exp.After(s.now()) {
373 return nil
374 }
375 > if suggestContinueAsNew && s.pendingUpdate == nil && s.pendingPatch == nil && s.hasMinVersion(CANAfterSignals) { workflow.go
376 // If suggestContinueAsNew was true but we had a pending update or patch, we would
377 // not break above, but process the update/patch. Now that we're done, we should
385 // 2. we got a signal (update, request, refresh)
386 // 3. a workflow that we were watching finished
387 > s.sleep(nextWakeup) workflow.go
388 > s.updateTweakables()
389 }
390
774 }
775
776 > func (s *scheduler) sleep(nextWakeup time.Time) { workflow.go
777 > sel := workflow.NewSelector(s.ctx)
778 >
779 > upCh := workflow.GetSignalChannel(s.ctx, SignalNameUpdate)
780 > sel.AddReceive(upCh, func(ch workflow.ReceiveChannel, _ bool) {
781 ch.Receive(s.ctx, &s.pendingUpdate)
782 })
783
784 > reqCh := workflow.GetSignalChannel(s.ctx, SignalNamePatch) workflow.go
785 > sel.AddReceive(reqCh, func(ch workflow.ReceiveChannel, _ bool) {
786 ch.Receive(s.ctx, &s.pendingPatch)
787 })
788
789 > refreshCh := workflow.GetSignalChannel(s.ctx, SignalNameRefresh) workflow.go
790 > sel.AddReceive(refreshCh, s.handleRefreshSignal)
791 >
792 > forceCAN := workflow.GetSignalChannel(s.ctx, SignalNameForceCAN)
793 > sel.AddReceive(forceCAN, s.handleForceCANSignal)
794 >
795 > migrateCh := workflow.GetSignalChannel(s.ctx, SignalNameMigrateToChasm)
796 > sel.AddReceive(migrateCh, s.handleMigrateSignal)
797 >
798 > if s.hasMoreAllowAllBackfills() {
799 // if we have more allow-all backfills to do, do a short sleep and continue
800 nextWakeup = s.now().Add(1 * time.Second)
801 > } else if s.tweakables.SleepWhilePaused && !s.canTakeScheduledAction(false, false) { workflow.go
802 // if we're paused or out of actions, we don't need to wake up until we get an update
803 nextWakeup = time.Time{}
805
806 // if retention is not zero, it means there is no more job to schedule, so sleep for retention time and then exit if no signal is received
807 > if exp := s.getRetentionExpiration(nextWakeup); !exp.IsZero() { workflow.go
808 nextWakeup = exp
809 }
810
811 > if !nextWakeup.IsZero() { workflow.go
812 sleepTime := nextWakeup.Sub(s.now())
813 // A previous version of this workflow passed around sleep duration instead of wakeup time,
834 }
835
836 > if s.watchingFuture != nil { workflow.go
837 sel.AddFuture(s.watchingFuture, s.wfWatcherReturned)
838 }
839
840 > s.logger.Debug("sleeping", "next-wakeup", nextWakeup, "watching", s.watchingFuture != nil) workflow.go
841 > sel.Select(s.ctx)
842 > for sel.HasPending() {
843 sel.Select(s.ctx)
844 }
845 }
846
847 > func (s *scheduler) processBackfills() { workflow.go
848 > limit := s.tweakables.BackfillsPerIteration
849 >
850 > for len(s.State.OngoingBackfills) > 0 &&
851 > limit > 0 &&
852 > // use only half the buffer for backfills
853 > len(s.State.BufferedStarts) < s.tweakables.MaxBufferSize/2 {
854 bfr := s.State.OngoingBackfills[0]
855 startTime := timestamp.TimeValue(bfr.GetStartTime())
872 }
873
874 > func (s *scheduler) hasMoreAllowAllBackfills() bool { workflow.go
875 > return len(s.State.OngoingBackfills) > 0 &&
876 > s.resolveOverlapPolicy(s.State.OngoingBackfills[0].OverlapPolicy) == enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL
877 > }
878
879 func (s *scheduler) wfWatcherReturned(f workflow.Future) {
1105 // schedule's `RemainingActions` will be returned. Future action times that
1106 // precede the schedule's UpdateTime are not included.
1107 > func (s *scheduler) getFutureActionTimes(inWorkflowContext bool, n int) []*timestamppb.Timestamp { workflow.go
1108 > // Note that `s` may be a `scheduler` created outside of a workflow context, used to
1109 > // compute list info at creation time or in a query. In that case inWorkflowContext will
1110 > // be false, and this function and anything it calls should not use s.ctx.
1111 >
1112 > base := timestamp.TimeValue(s.State.LastProcessedTime)
1113 >
1114 > // Pure version not using workflow context
1115 > next := func(t time.Time) time.Time {
1116 res, _ := s.cspec.GetNextTime(s.jitterSeed(), t)
1117 return res.Next
1118 }
1119
1120 > if inWorkflowContext && s.hasMinVersion(NewCacheAndJitter) { workflow.go
1121 > // We can use the cache here
1122 > next = func(t time.Time) time.Time {
1123 return s.getNextTimeV2(base, t).Next
1124 }
1125 }
1126
1127 > if s.hasMinVersion(AccurateFutureActionTimes) && s.Schedule.State.LimitedActions { workflow.go
1128 n = min(int(s.Schedule.State.RemainingActions), n)
1129 }
1130
1131 > if s.cspec == nil { workflow.go
1132 return nil
1133 }
1194 }
1195
1196 > func (s *scheduler) getListInfo(inWorkflowContext bool) *schedulepb.ScheduleListInfo { workflow.go
1197 > // Note that `s` may be a `scheduler` created outside of a workflow context, used to
1198 > // compute list info at creation time. In that case inWorkflowContext will be false,
1199 > // and this function and anything it calls should not use s.ctx.
1200 >
1201 > spec := common.CloneProto(s.Schedule.Spec)
1202 > // clear fields that are too large/not useful for the list view
1203 > spec.TimezoneData = nil
1204 >
1205 > if s.hasMinVersion(LimitMemoSpecSize) {
1206 > // Limit the number of specs and exclusions stored on the memo.
1207 > limit := s.tweakables.SpecFieldLengthLimit
1208 > spec.ExcludeStructuredCalendar = util.SliceHead(spec.ExcludeStructuredCalendar, limit)
1209 > spec.Interval = util.SliceHead(spec.Interval, limit)
1210 > spec.StructuredCalendar = util.SliceHead(spec.StructuredCalendar, limit)
1211 > }
1212
1213 > return &schedulepb.ScheduleListInfo{ workflow.go
1214 > Spec: spec,
1215 > WorkflowType: s.Schedule.Action.GetStartWorkflow().GetWorkflowType(),
1216 > Notes: s.Schedule.State.Notes,
1217 > Paused: s.Schedule.State.Paused,
1218 > RecentActions: util.SliceTail(s.Info.RecentActions, s.tweakables.RecentActionCountForList),
1219 > FutureActionTimes: s.getFutureActionTimes(inWorkflowContext, s.tweakables.FutureActionCountForList),
1220 > }
1221 }
1222
1268 }
1269
1270 > func (s *scheduler) updateMemoAndSearchAttributes() { workflow.go
1271 > newInfo := s.getListInfo(true)
1272 >
1273 > workflowInfo := workflow.GetInfo(s.ctx)
1274 > currentInfoPayload := workflowInfo.Memo.GetFields()[MemoFieldInfo]
1275 >
1276 > var currentInfoBytes []byte
1277 > var currentInfo schedulepb.ScheduleListInfo
1278 >
1279 > //workflowcheck:ignore (proto.Equal is falsely flagged as non-deterministic)
1280 > if currentInfoPayload == nil ||
1281 > payload.Decode(currentInfoPayload, &currentInfoBytes) != nil ||
1282 > currentInfo.Unmarshal(currentInfoBytes) != nil ||
1283 > !proto.Equal(&currentInfo, newInfo) {
1284 > // marshal manually to get proto encoding (default dataconverter will use json)
1285 > newInfoBytes, err := newInfo.Marshal()
1286 > if err == nil {
1287 > err = workflow.UpsertMemo(s.ctx, map[string]any{
1288 > MemoFieldInfo: newInfoBytes,
1289 > })
1290 > }
1291 > if err != nil {
1292 s.logger.Error("error updating memo", "error", err)
1293 }
1295
1296 //nolint:staticcheck // SA1019: workflowInfo.SearchAttributes is not typed.
1297 > currentPausedPayload := workflowInfo.SearchAttributes.GetIndexedFields()[sadefs.TemporalSchedulePaused] workflow.go
1298 > var currentPaused bool
1299 > if currentPausedPayload == nil ||
1300 > payload.Decode(currentPausedPayload, &currentPaused) != nil ||
1301 > currentPaused != s.Schedule.State.Paused {
1302 > err := workflow.UpsertSearchAttributes(s.ctx, map[string]any{ //nolint:staticcheck // SA1019: untyped search attributes required here
1303 > sadefs.TemporalSchedulePaused: s.Schedule.State.Paused,
1304 > })
1305 > if err != nil {
1306 s.logger.Error("error updating search attributes", "error", err)
1307 }
1374 //
1375 //nolint:revive
1376 > func (s *scheduler) processBuffer() bool { workflow.go
1377 > s.logger.Debug("ProcessBuffer", "buffer", len(s.State.BufferedStarts), "running", len(s.Info.RunningWorkflows), "need-refresh", s.State.NeedRefresh)
1378 >
1379 > // TODO: consider doing this always and removing needRefresh? we only end up here without
1380 > // needRefresh in the case of update, or patch without an immediate run, so it's not much
1381 > // wasted work.
1382 > // TODO: on the other hand, we don't have to refresh if we have one workflow running and a
1383 > // long-poll watcher running, because we would have gotten woken up already.
1384 > if s.State.NeedRefresh {
1385 s.refreshWorkflows(slices.Clone(s.Info.RunningWorkflows))
1386 s.State.NeedRefresh = false
1388
1389 // Make sure we have something to start. If not, we can clear the buffer.
1390 > req := s.Schedule.Action.GetStartWorkflow() workflow.go
1391 > if req == nil || len(s.State.BufferedStarts) == 0 {
1392 s.State.BufferedStarts = nil
1393 return false
1677 }
1678
1679 > func (s *scheduler) getRetentionExpiration(nextWakeup time.Time) time.Time { workflow.go
1680 > // if RetentionTime is not set or the schedule is paused or nextWakeup time is not zero
1681 > // or there is more action to take, there is no need for retention
1682 > if s.tweakables.RetentionTime == 0 ||
1683 > s.Schedule.State.Paused ||
1684 > (!nextWakeup.IsZero() && s.canTakeScheduledAction(false, false)) ||
1685 > s.hasMoreAllowAllBackfills() {
1686 return time.Time{}
1687 }

Direct hierarchy

Parents

Children