task_generator.go ×4

Frontier kind: Code frontier

unlabeled · c_d519f826791a

13 tests · 6531 LOC · 226 files · introduces 0 tests · 66 LOC · 6 files

Introduces — evidence that enters the hierarchy at this concept

Code
12 ranges66 lines · 6 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
1248 ranges6531 lines · 226 files · Browse complete extent
All tests (intent)
13 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.

6 files ranked by introduced lines: 66 introduced LOC across 12 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/history/workflow/task_generator.go 35 introduced LOC · 4 ranges

Open complete file

469 func (r *TaskGeneratorImpl) GenerateScheduleSpeculativeWorkflowTaskTasks(
470 workflowTask *historyi.WorkflowTaskInfo,
471 > ) error { task_generator.go
472 >
473 > var scheduleToStartTimeout time.Duration
474 > if r.mutableState.IsStickyTaskQueueSet() {
475 scheduleToStartTimeout = timestamp.DurationValue(r.mutableState.GetExecutionInfo().StickyScheduleToStartTimeout)
476 > } else { task_generator.go
477 > // Speculative WT has ScheduleToStart timeout even on normal task queue.
478 > // Normally WT should be added to matching right after being created
479 > // (i.e. from UpdateWorkflowExecution API handler), but if this "add" operation failed,
480 > // there is no good way to handle the error.
481 > // In this case WT will be timed out (as if it was on sticky task queue),
482 > // and new normal WT will be created.
483 > // Note: this timer will also fire if workflow received an update,
484 > // but there is no workers available. Speculative WT will time out, and normal WT will be created.
485 > scheduleToStartTimeout = tasks.SpeculativeWorkflowTaskScheduleToStartTimeout
486 > }
487
488 > isSpeculative := workflowTask.Type == enumsspb.WORKFLOW_TASK_TYPE_SPECULATIVE task_generator.go
489 > wttt := &tasks.WorkflowTaskTimeoutTask{
490 > // TaskID is set by shard
491 > WorkflowKey: r.mutableState.GetWorkflowKey(),
492 > VisibilityTimestamp: workflowTask.ScheduledTime.Add(scheduleToStartTimeout),
493 > TimeoutType: enumspb.TIMEOUT_TYPE_SCHEDULE_TO_START,
494 > EventID: workflowTask.ScheduledEventID,
495 > ScheduleAttempt: workflowTask.Attempt,
496 > Version: workflowTask.Version,
497 > Stamp: workflowTask.Stamp,
498 > InMemory: isSpeculative,
499 > }
500 >
501 > if isSpeculative {
502 > // If WT is still speculative, create task in in-memory task queue.
503 > return r.mutableState.SetSpeculativeWorkflowTaskTimeoutTask(wttt)
504 > }
505
506 // This function can be called for speculative WT which just was converted to normal
539
540 if isSpeculative {
541 > // If WT is speculative, create task in in-memory task queue. task_generator.go
542 > return r.mutableState.SetSpeculativeWorkflowTaskTimeoutTask(wttt)
543 > }
544 r.mutableState.AddTasks(wttt)
545 r.mutableState.SetWorkflowTaskStartToCloseTimeoutTask(wttt)
go.temporal.io/server/service/history/workflow/mutable_state_impl.go 15 introduced LOC · 3 ranges

Open complete file

7339 func (ms *MutableStateImpl) SetSpeculativeWorkflowTaskTimeoutTask(
7340 task *tasks.WorkflowTaskTimeoutTask,
7341 > ) error { mutable_state_impl.go
7342 > taskID, err := ms.shard.GenerateTaskID()
7343 > if err != nil {
7344 return err
7345 }
7346 > task.TaskID = taskID mutable_state_impl.go
7347 > // Producer stamps VisibilityTimestamp in the workflow's virtual frame (it's derived from
7348 > // workflowTask.ScheduledTime / StartedTime, which come from the time-skipping wrapper).
7349 > // The in-memory scheduled queue dispatches against real wall-clock, so convert here —
7350 > // same boundary contract as AddTasks for persisted scheduled tasks.
7351 > task.SetVisibilityTime(ms.ToRealTime(task.GetVisibilityTime()))
7352 > ms.speculativeWorkflowTaskTimeoutTask = task
7353 > return ms.shard.AddSpeculativeWorkflowTaskTimeoutTask(task)
7354 }
7355
7362 func (ms *MutableStateImpl) RemoveSpeculativeWorkflowTaskTimeoutTask() {
7363 if ms.speculativeWorkflowTaskTimeoutTask != nil {
7364 > // Cancelling task prevents it from being submitted to scheduler in memoryScheduledQueue. mutable_state_impl.go
7365 > ms.speculativeWorkflowTaskTimeoutTask.Cancel()
7366 > ms.speculativeWorkflowTaskTimeoutTask = nil
7367 > }
7368 }
7369
go.temporal.io/server/service/history/shard/context_impl.go 11 introduced LOC · 2 ranges

Open complete file

506 func (s *ContextImpl) AddSpeculativeWorkflowTaskTimeoutTask(
507 task *tasks.WorkflowTaskTimeoutTask,
508 > ) error { context_impl.go
509 > // Use a cancelled context to avoid blocking if engineFuture is not ready.
510 > cancelledCtx, cancel := context.WithCancel(context.Background())
511 > cancel()
512 >
513 > // err should never be returned here. engineFuture must always be ready.
514 > engine, err := s.engineFuture.Get(cancelledCtx)
515 > if err != nil {
516 return err
517 }
518
519 > engine.NotifyNewTasks(map[tasks.Category][]tasks.Task{task.GetCategory(): []tasks.Task{task}}) context_impl.go
520 >
521 > return nil
522 }
523
go.temporal.io/server/api/enums/v1/workflow_task_type.pb.go 2 introduced LOC · 1 range

Open complete file

64 case WORKFLOW_TASK_TYPE_TRANSIENT:
65 return "Transient"
66 > case WORKFLOW_TASK_TYPE_SPECULATIVE: workflow_task_type.pb.go
67 > return "Speculative"
68 default:
69 return strconv.Itoa(int(x))
go.temporal.io/server/service/history/tasks/workflow_task_timer.go 2 introduced LOC · 1 range

Open complete file

70 func (d *WorkflowTaskTimeoutTask) GetCategory() Category {
71 if d.InMemory {
72 > return CategoryMemoryTimer workflow_task_timer.go
73 > }
74 return CategoryTimer
75 }
go.temporal.io/server/service/history/workflow/workflow_task_state_machine.go 1 introduced LOC · 1 range

Open complete file

391 if !bypassTaskGeneration {
392 if workflowTask.Type == enumsspb.WORKFLOW_TASK_TYPE_SPECULATIVE {
393 > err = m.ms.taskGenerator.GenerateScheduleSpeculativeWorkflowTaskTasks(workflowTask) workflow_task_state_machine.go
394 } else {
395 err = m.ms.taskGenerator.GenerateScheduleWorkflowTaskTasks(scheduledEventID)