Apply 5 min delay for standby task #920

wxing1292 · 2018-07-03T02:23:53Z

transfer task will be delayed by at most 5 min (meaning the delay can be less than 5 min, i.e. if the standby task can be discarded, or wait until 5 min and re-check)
timer task will be delayed by 5 min.

samarabbas · 2018-07-03T04:42:23Z

service/history/queueProcessor.go

@@ -283,7 +283,13 @@ ProcessRetryLoop:
 			if err != nil {
 				if err == ErrTaskRetry {
 					p.metricsClient.IncCounter(p.options.MetricScope, metrics.HistoryTaskStandbyRetryCounter)
-					<-notificationChan
+				DelayLoop:


Looks like we will delay the task after processing for first time. I thought the idea is to even delay on first processing.

for transfer task, there is no timestamp associated with it.
we can tag the queue processor, however, that require more code.

samarabbas · 2018-07-03T04:45:31Z

service/history/historyEngine.go

@@ -237,7 +237,7 @@ func (e *historyEngineImpl) registerDomainFailoverCallback() {
 				// its length > 0 and has correct timestamp, to trkgger a db scan
 				fakeDecisionTask := []persistence.Task{&persistence.DecisionTask{}}
 				fakeDecisionTimeoutTask := []persistence.Task{&persistence.DecisionTimeoutTask{VisibilityTimestamp: now}}
-				e.txProcessor.NotifyNewTask(e.currentClusterName, now, fakeDecisionTask)
+				e.txProcessor.NotifyNewTask(e.currentClusterName, fakeDecisionTask)


I think we should have the exact same contract on moving the clock on both timer and transfer processors. Timer notification takes in the currentTime as notification while transfer does not. I think we should unify these.

transfer task do not have timestamp associated with it

samarabbas · 2018-07-17T18:37:53Z

common/persistence/cassandraPersistence.go

@@ -136,6 +136,7 @@ const (
 		`create_request_id: ?, ` +
 		`decision_version: ?, ` +
 		`decision_schedule_id: ?, ` +
+		`decision_schedule_time: ?, ` +


Probably we should just have a visibility_ts on transfer task for we have the same mechanism for each kind of task for delaying them.

samarabbas · 2018-07-17T21:36:24Z

common/time_source.go

@@ -47,7 +47,7 @@ func NewRealTimeSource() *RealTimeSource {

 // Now return the real current time
 func (ts *RealTimeSource) Now() time.Time {
-	return time.Now()
+	return time.Now().UTC()


usually the API is UtcNow.

there is no UtcNow() function
ref: https://golang.org/pkg/time/

samarabbas · 2018-07-17T21:38:02Z

service/history/MockMutableState.go

@@ -1347,6 +1347,28 @@ func (_m *mockMutableState) GetHistoryEvent(serializedEvent []byte) (*shared.His
 	return r0, r1
 }

+func (_m *mockMutableState) GetInFlightDecisionTask() (*decisionInfo, bool) {


Make sure you merge my change. This was included as part of that.

…ke start child workflow use UTC for time.Now() in history service when failover, should unblock existing standby task

samarabbas · 2018-07-17T21:45:19Z

service/history/historyEngine.go

@@ -357,6 +357,7 @@ func (e *historyEngineImpl) StartWorkflowExecution(startRequest *h.StartWorkflow
 		}
 	}
 	setTaskVersion(msBuilder.GetCurrentVersion(), transferTasks, timerTasks)
+	setTransferTaskTimestamp(common.NewRealTimeSource().Now(), transferTasks)


Can we just have a single API for setting task information, instead of making 2 separate API calls.

samarabbas · 2018-07-17T21:51:55Z

service/history/historyEventNotifier.go

@@ -177,7 +177,7 @@ func (notifier *historyEventNotifierImpl) dispatchHistoryEventNotification(event

 func (notifier *historyEventNotifierImpl) enqueueHistoryEventNotification(event *historyEventNotification) {
 	// set the timestamp just before enqueuing the event
-	event.timestamp = time.Now()
+	event.timestamp = common.NewRealTimeSource().Now()


Can you have time source as a field on historyEventNotifierImpl so another implementation could be injected.

samarabbas · 2018-07-17T21:55:29Z

service/history/workflowExecutionContext.go

@@ -187,12 +187,13 @@ func (c *workflowExecutionContext) updateWorkflowExecution(transferTasks []persi
 			c.msBuilder.GetExecutionInfo().LastFirstEventID, c.msBuilder.GetExecutionInfo().NextEventID)
 	}

-	return c.updateHelper(nil, transferTasks, timerTasks, c.createReplicationTask, "", currentVersion, transactionID)
+	now := common.NewRealTimeSource().Now()


Lets call the API UtcNow and have a timesource defined on workflowExecutionContext

samarabbas · 2018-07-17T22:26:45Z

service/history/queueProcessor.go

+					timestamp := task.GetVisibilityTimestamp()
+					for {
+						<-notificationChan
+						if p.shard.GetCurrentTime(p.clusterName).After(timestamp) {


We only want to have an upper bound on how long we hold onto certain tasks like activity and decision, not delay processing always for all tasks.

samarabbas · 2018-07-18T00:24:29Z

service/history/historyEngine.go

@@ -356,8 +356,7 @@ func (e *historyEngineImpl) StartWorkflowExecution(startRequest *h.StartWorkflow
 			replicationTasks = append(replicationTasks, replicationTask)
 		}
 	}
-	setTaskVersion(msBuilder.GetCurrentVersion(), transferTasks, timerTasks)
-	setTransferTaskTimestamp(common.NewRealTimeSource().Now(), transferTasks)
+	setTaskInfo(msBuilder.GetCurrentVersion(), time.Now().UTC(), transferTasks, timerTasks)


Can we just get UTC within setTaskInfo?

samarabbas · 2018-07-18T00:37:48Z

common/time_source.go

-func (ts *FakeTimeSource) Update(now time.Time) {
-	ts.now = now
+func (ts *FakeTimeSource) Update(now time.Time) *FakeTimeSource {
+	ts.now = now.UTC()


don't do UTC here.

any reason?

samarabbas · 2018-07-18T00:39:58Z

service/history/historyReplicator.go

@@ -835,8 +844,9 @@ func (r *historyReplicator) terminateWorkflow(ctx context.Context, domainID stri

 func (r *historyReplicator) notify(clusterName string, now time.Time, transferTasks []persistence.Task,
 	timerTasks []persistence.Task) {
+	now = now.Add(-r.shard.GetConfig().StandbyClusterDelay())


there is no subtract?

there is a func (t Time) Sub(u Time) Duration which does something else

samarabbas · 2018-07-18T00:42:26Z

service/history/mutableStateBuilder.go

-	setTaskVersion(newStateBuilder.GetCurrentVersion(), newTransferTasks, nil)
+	setTaskInfo(
+		newStateBuilder.GetCurrentVersion(),
+		common.NewFakeTimeSource().Update(time.Unix(0, startedEvent.GetTimestamp())).Now(),


why are we using fakeTimeSource in production code?

this ReplicateWorkflowExecutionContinuedAsNewEvent can be used by both active and standby, so the only reliable way is to use the event time in the start event

wxing1292 requested a review from samarabbas July 3, 2018 02:23

samarabbas reviewed Jul 3, 2018

View reviewed changes

apply 5 min delay for standby task

509a05c

wxing1292 force-pushed the standby-delay branch from f96b0bd to 509a05c Compare July 17, 2018 06:05

Merge branch 'master' into standby-delay

a04a77a

samarabbas reviewed Jul 17, 2018

View reviewed changes

wxing1292 force-pushed the standby-delay branch from 124292d to 829e62e Compare July 17, 2018 21:40

use transfer task visibility timestamp instead of individual ones, li…

51f6613

…ke start child workflow use UTC for time.Now() in history service when failover, should unblock existing standby task

wxing1292 force-pushed the standby-delay branch from 829e62e to 51f6613 Compare July 17, 2018 21:42

Merge branch 'master' into standby-delay

8fd74cf

samarabbas reviewed Jul 17, 2018

View reviewed changes

wxing1292 force-pushed the standby-delay branch 2 times, most recently from 7f7c270 to 440a3a5 Compare July 17, 2018 22:53

address comments

b2e398b

wxing1292 force-pushed the standby-delay branch from 440a3a5 to b2e398b Compare July 17, 2018 23:03

Merge branch 'master' into standby-delay

6bdaf0e

samarabbas reviewed Jul 18, 2018

View reviewed changes

address comments

f48e345

samarabbas reviewed Jul 18, 2018

View reviewed changes

address comments

e94a537

samarabbas approved these changes Jul 18, 2018

View reviewed changes

fix golang time which are suppose to be equal but not deep equal issue

fe7762f

wxing1292 merged commit 3a79fc0 into master Jul 18, 2018

wxing1292 deleted the standby-delay branch July 18, 2018 02:03

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Apply 5 min delay for standby task #920

Apply 5 min delay for standby task #920

wxing1292 commented Jul 3, 2018 •

edited

Loading

samarabbas Jul 3, 2018

wxing1292 Jul 3, 2018

samarabbas Jul 3, 2018

wxing1292 Jul 3, 2018

samarabbas Jul 17, 2018

samarabbas Jul 17, 2018

wxing1292 Jul 17, 2018

samarabbas Jul 17, 2018

wxing1292 Jul 17, 2018

samarabbas Jul 17, 2018

samarabbas Jul 17, 2018

samarabbas Jul 17, 2018

samarabbas Jul 17, 2018

samarabbas Jul 18, 2018

samarabbas Jul 18, 2018

wxing1292 Jul 18, 2018

samarabbas Jul 18, 2018

wxing1292 Jul 18, 2018

samarabbas Jul 18, 2018

wxing1292 Jul 18, 2018

Apply 5 min delay for standby task #920

Apply 5 min delay for standby task #920

Conversation

wxing1292 commented Jul 3, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wxing1292 commented Jul 3, 2018 •

edited

Loading