Rework Local Activity scheduling #1507
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,91 @@ | ||
| /* | ||
| * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved. | ||
| * | ||
| * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
| * | ||
| * Modifications copyright (C) 2017 Uber Technologies, Inc. | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this material except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package io.temporal.internal.common; | ||
|
|
||
| import io.grpc.Deadline; | ||
| import io.temporal.api.common.v1.RetryPolicy; | ||
| import io.temporal.common.RetryOptions; | ||
| import io.temporal.failure.ActivityFailure; | ||
| import io.temporal.failure.ApplicationFailure; | ||
| import io.temporal.failure.ChildWorkflowFailure; | ||
| import java.time.Duration; | ||
| import java.util.concurrent.TimeUnit; | ||
| import javax.annotation.Nullable; | ||
|
|
||
| public class RetryOptionsUtils { | ||
| public static boolean isNotRetryable(RetryOptions o, @Nullable Throwable e) { | ||
| if (e == null) { | ||
| return false; | ||
| } | ||
| if (e instanceof ActivityFailure || e instanceof ChildWorkflowFailure) { | ||
| e = e.getCause(); | ||
| } | ||
| String type = | ||
| e instanceof ApplicationFailure | ||
| ? ((ApplicationFailure) e).getType() | ||
| : e.getClass().getName(); | ||
| if (o.getDoNotRetry() != null) { | ||
| for (String doNotRetry : o.getDoNotRetry()) { | ||
| if (doNotRetry.equals(type)) { | ||
| return true; | ||
| } | ||
| } | ||
| } | ||
| return false; | ||
| } | ||
|
|
||
| public static boolean areAttemptsReached(RetryOptions o, long attempt) { | ||
| return (o.getMaximumAttempts() != 0 && attempt >= o.getMaximumAttempts()); | ||
| } | ||
|
|
||
| public static boolean isDeadlineReached(@Nullable Deadline deadline, long sleepTimeMs) { | ||
| return deadline != null && deadline.timeRemaining(TimeUnit.MILLISECONDS) < sleepTimeMs; | ||
| } | ||
|
|
||
| public static RetryOptions toRetryOptions(RetryPolicy retryPolicy) { | ||
| RetryOptions.Builder roBuilder = RetryOptions.newBuilder(); | ||
|
|
||
| Duration maximumInterval = ProtobufTimeUtils.toJavaDuration(retryPolicy.getMaximumInterval()); | ||
| if (!maximumInterval.isZero()) { | ||
| roBuilder.setMaximumInterval(maximumInterval); | ||
| } | ||
|
|
||
| Duration initialInterval = ProtobufTimeUtils.toJavaDuration(retryPolicy.getInitialInterval()); | ||
| if (!initialInterval.isZero()) { | ||
| roBuilder.setInitialInterval(initialInterval); | ||
| } | ||
|
|
||
| if (retryPolicy.getBackoffCoefficient() >= 1) { | ||
| roBuilder.setBackoffCoefficient(retryPolicy.getBackoffCoefficient()); | ||
| } | ||
|
|
||
| if (retryPolicy.getMaximumAttempts() > 0) { | ||
| roBuilder.setMaximumAttempts(retryPolicy.getMaximumAttempts()); | ||
| } | ||
|
|
||
| roBuilder.setDoNotRetry( | ||
| retryPolicy | ||
| .getNonRetryableErrorTypesList() | ||
| .toArray(new String[retryPolicy.getNonRetryableErrorTypesCount()])); | ||
|
|
||
| return roBuilder.validateBuildWithDefaults(); | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -57,7 +57,6 @@ | |
| import java.util.concurrent.TimeUnit; | ||
| import java.util.concurrent.locks.Lock; | ||
| import java.util.concurrent.locks.ReentrantLock; | ||
|
|
||
| /** | ||
| * Implements workflow executor that relies on replay of a workflow code. An instance of this class | ||
|
|
@@ -70,12 +69,12 @@ class ReplayWorkflowRunTaskHandler implements WorkflowRunTaskHandler { | |
|
|
||
| private final Lock lock = new ReentrantLock(); | ||
|
|
||
| private final Functions.Proc1<LocalActivityResult> localActivityCompletionSink; | ||
|
|
||
| private final BlockingQueue<LocalActivityResult> localActivityCompletionQueue = | ||
| new LinkedBlockingDeque<>(); | ||
|
|
||
| private final LocalActivityDispatcher localActivityDispatcher; | ||
|
|
||
| private final ReplayWorkflow workflow; | ||
|
|
||
|
|
@@ -93,15 +92,15 @@ class ReplayWorkflowRunTaskHandler implements WorkflowRunTaskHandler { | |
| PollWorkflowTaskQueueResponseOrBuilder workflowTask, | ||
| SingleWorkerOptions workerOptions, | ||
| Scope metricsScope, | ||
| LocalActivityDispatcher localActivityDispatcher) { | ||
| HistoryEvent startedEvent = workflowTask.getHistory().getEvents(0); | ||
| if (!startedEvent.hasWorkflowExecutionStartedEventAttributes()) { | ||
| throw new IllegalArgumentException( | ||
| "First event in the history is not WorkflowExecutionStarted"); | ||
| } | ||
| this.startedEvent = startedEvent.getWorkflowExecutionStartedEventAttributes(); | ||
| this.metricsScope = metricsScope; | ||
| this.localActivityDispatcher = localActivityDispatcher; | ||
| this.workflow = workflow; | ||
|
|
||
| this.workflowStateMachines = new WorkflowStateMachines(new StatesMachinesCallbackImpl()); | ||
|
|
@@ -286,10 +285,10 @@ private void processLocalActivityRequests(long startTimeNs) throws InterruptedEx | |
| // much sense. I believe we should add ScheduleToStart timeout for the local activities | ||
| // as well. | ||
|
|
||
| long maxWaitTimeNs = Math.max(nextWFTHeartbeatTimeNs - System.nanoTime(), 0); | ||
| laRequest.setScheduleToStartTimeout(Duration.ofNanos(maxWaitTimeNs)); | ||
|
Member
There was a problem hiding this comment. Can user not set sched-to-start timeout? (Honestly, makes sense)
Contributor
Author
There was a problem hiding this comment. It's old logic. It will be gone in #1512. |
||
| boolean accepted = localActivityDispatcher.dispatch(laRequest, localActivityCompletionSink); | ||
| // TODO this needs to be reworked when we implement a proper scheduleToStart to report a | ||
| // Failure in the callback. A proper test is also needed for this scenario. | ||
| Preconditions.checkState( | ||
| accepted, | ||
| "Unable to schedule local activity for execution, " | ||
|
|
@@ -302,7 +301,7 @@ private void processLocalActivityRequests(long startTimeNs) throws InterruptedEx | |
| } | ||
|
|
||
| long maxWaitTimeTillHeartbeatNs = Math.max(nextWFTHeartbeatTimeNs - System.nanoTime(), 0); | ||
| LocalActivityResult laCompletion = | ||
| localActivityCompletionQueue.poll(maxWaitTimeTillHeartbeatNs, TimeUnit.NANOSECONDS); | ||
| if (laCompletion == null) { | ||
| // Need to force a new task as we are out of time | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Is it possible to set an explicit do not retry field in Java or is it only the string matching?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
String matching as it gets converted to protobuf's RetryPolicy, at least at this moment.