#244

Committed 10 Apr 2024 08:19PM UTC coverage: 77.465% (-0.08%) from 77.549%

Build # #244

Build Type

push

github

Committed by

web-flow

Commit Message

Slot supplier interface & fixed-size implementation (#2014)

https://github.com/temporalio/proposals/blob/master/all-sdk/autotuning.md

Run Details

286 of 388 new or added lines in 25 files covered. (73.71%)

3 existing lines in 3 files now uncovered.

19116 of 24677 relevant lines covered (77.46%)

0.77 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.71

/temporal-sdk/src/main/java/io/temporal/worker/WorkflowTaskDispatchHandle.java

/*
 * 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.worker;

import com.google.common.base.Preconditions;
import io.temporal.api.workflowservice.v1.PollWorkflowTaskQueueResponse;
import io.temporal.internal.worker.TrackingSlotSupplier;
import io.temporal.internal.worker.WorkflowTask;
import io.temporal.worker.tuning.SlotPermit;
import io.temporal.worker.tuning.SlotReleaseReason;
import io.temporal.worker.tuning.WorkflowSlotInfo;
import java.io.Closeable;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import javax.annotation.Nonnull;

public class WorkflowTaskDispatchHandle implements Closeable {
  private final AtomicBoolean completed = new AtomicBoolean();
  private final Function<WorkflowTask, Boolean> dispatchCallback;
  private final TrackingSlotSupplier<WorkflowSlotInfo> slotSupplier;
  private final SlotPermit permit;

  /**
   * @param dispatchCallback callback into a {@code WorkflowWorker} to dispatch a workflow task.
   * @param slotSupplier slot supplier that was used to reserve a slot for this workflow task
   */
  public WorkflowTaskDispatchHandle(
      DispatchCallback dispatchCallback,
      TrackingSlotSupplier<WorkflowSlotInfo> slotSupplier,
      SlotPermit permit) {
    this.dispatchCallback = dispatchCallback;
    this.slotSupplier = slotSupplier;
    this.permit = permit;
  }

  /**
   * @param workflowTask to be fed directly into the workflow worker
   * @return true is the workflow task was successfully dispatched
   * @throws IllegalArgumentException if the workflow task doesn't belong to the task queue of the
   *     worker provided this {@link WorkflowTaskDispatchHandle}
   */
  public boolean dispatch(@Nonnull PollWorkflowTaskQueueResponse workflowTask) {
    Preconditions.checkNotNull(workflowTask, "workflowTask");
    if (completed.compareAndSet(false, true)) {
      return dispatchCallback.apply(
          new WorkflowTask(
              workflowTask,
              () -> slotSupplier.releaseSlot(SlotReleaseReason.taskComplete(), permit)));
    } else {
      return false;
    }
  }

  @Override
  public void close() {
    if (completed.compareAndSet(false, true)) {
      slotSupplier.releaseSlot(SlotReleaseReason.neverUsed(), permit);
    }
  }

  /** A callback into a {@code WorkflowWorker} to dispatch a workflow task */
  @FunctionalInterface
  public interface DispatchCallback extends Function<WorkflowTask, Boolean> {

    /**
     * Should dispatch the Workflow Task to the Workflow Worker. Shouldn't block the thread.
     *
     * @param workflowTask WorkflowTask to be dispatched
     * @return true if the dispatch was successful and false otherwise
     * @throws IllegalArgumentException if {@code workflowTask} doesn't belong to the task queue of the Worker that provided the {@link WorkflowTaskDispatchHandle
     */
    @Override
    Boolean apply(WorkflowTask workflowTask) throws IllegalArgumentException;
  }
}

1	/*
2	* Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
3	*
4	* Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
5	*
6	* Modifications copyright (C) 2017 Uber Technologies, Inc.
7	*
8	* Licensed under the Apache License, Version 2.0 (the "License");
9	* you may not use this material except in compliance with the License.
10	* You may obtain a copy of the License at
11	*
12	* http://www.apache.org/licenses/LICENSE-2.0
13	*
14	* Unless required by applicable law or agreed to in writing, software
15	* distributed under the License is distributed on an "AS IS" BASIS,
16	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17	* See the License for the specific language governing permissions and
18	* limitations under the License.
19	*/
20
21	package io.temporal.worker;
22
23	import com.google.common.base.Preconditions;
24	import io.temporal.api.workflowservice.v1.PollWorkflowTaskQueueResponse;
25	import io.temporal.internal.worker.TrackingSlotSupplier;
26	import io.temporal.internal.worker.WorkflowTask;
27	import io.temporal.worker.tuning.SlotPermit;
28	import io.temporal.worker.tuning.SlotReleaseReason;
29	import io.temporal.worker.tuning.WorkflowSlotInfo;
30	import java.io.Closeable;
31	import java.util.concurrent.atomic.AtomicBoolean;
32	import java.util.function.Function;
33	import javax.annotation.Nonnull;
34
35	public class WorkflowTaskDispatchHandle implements Closeable {
36	private final AtomicBoolean completed = new AtomicBoolean();	1✔
37	private final Function<WorkflowTask, Boolean> dispatchCallback;
38	private final TrackingSlotSupplier<WorkflowSlotInfo> slotSupplier;
39	private final SlotPermit permit;
40
41	/**
42	* @param dispatchCallback callback into a {@code WorkflowWorker} to dispatch a workflow task.
43	* @param slotSupplier slot supplier that was used to reserve a slot for this workflow task
44	*/
45	public WorkflowTaskDispatchHandle(
46	DispatchCallback dispatchCallback,
47	TrackingSlotSupplier<WorkflowSlotInfo> slotSupplier,
48	SlotPermit permit) {	1✔
49	this.dispatchCallback = dispatchCallback;	1✔
50	this.slotSupplier = slotSupplier;	1✔
51	this.permit = permit;	1✔
52	}	1✔
53
54	/**
55	* @param workflowTask to be fed directly into the workflow worker
56	* @return true is the workflow task was successfully dispatched
57	* @throws IllegalArgumentException if the workflow task doesn't belong to the task queue of the
58	* worker provided this {@link WorkflowTaskDispatchHandle}
59	*/
60	public boolean dispatch(@Nonnull PollWorkflowTaskQueueResponse workflowTask) {
61	Preconditions.checkNotNull(workflowTask, "workflowTask");	1✔
62	if (completed.compareAndSet(false, true)) {	1✔
63	return dispatchCallback.apply(	1✔
64	new WorkflowTask(
65	workflowTask,
66	() -> slotSupplier.releaseSlot(SlotReleaseReason.taskComplete(), permit)));	1✔
67	} else {
68	return false;	×
69	}
70	}
71
72	@Override
73	public void close() {
74	if (completed.compareAndSet(false, true)) {	1✔
NEW 75	slotSupplier.releaseSlot(SlotReleaseReason.neverUsed(), permit);	×
76	}
77	}	1✔
78
79	/** A callback into a {@code WorkflowWorker} to dispatch a workflow task */
80	@FunctionalInterface
81	public interface DispatchCallback extends Function<WorkflowTask, Boolean> {
82
83	/**
84	* Should dispatch the Workflow Task to the Workflow Worker. Shouldn't block the thread.
85	*
86	* @param workflowTask WorkflowTask to be dispatched
87	* @return true if the dispatch was successful and false otherwise
88	* @throws IllegalArgumentException if {@code workflowTask} doesn't belong to the task queue of the Worker that provided the {@link WorkflowTaskDispatchHandle
89	*/
90	@Override
91	Boolean apply(WorkflowTask workflowTask) throws IllegalArgumentException;
92	}
93	}

temporalio / sdk-java / #244

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous