#278

Committed 08 Jul 2024 04:42PM UTC coverage: 77.565% (+0.1%) from 77.469%

Build # #278

Build Type

push

github

Committed by

web-flow

Commit Message

Revert configurable slot provider (#2134)

* Revert "Resource based tuner (#2110)"

This reverts commit 8a2d5cdcc.

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

This reverts commit d2a06fc6f.

* Fix merge conflict

* Keep Publish Test Report step

* Add tests for worker slots

* Fix white space

* One other whitespace change

Run Details

117 of 133 new or added lines in 17 files covered. (87.97%)

5 existing lines in 5 files now uncovered.

19088 of 24609 relevant lines covered (77.57%)

0.78 hits per line

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

84.62

/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.WorkflowTask;
import java.io.Closeable;
import java.util.concurrent.Semaphore;
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 Semaphore executorSlotsSemaphore;

  /**
   * @param dispatchCallback callback into a {@code WorkflowWorker} to dispatch a workflow task.
   * @param executorSlotsSemaphore worker executor slots semaphore that was used to reserve this
   *     dispatch handle on
   */
  public WorkflowTaskDispatchHandle(
      DispatchCallback dispatchCallback, Semaphore executorSlotsSemaphore) {
    this.dispatchCallback = dispatchCallback;
    this.executorSlotsSemaphore = executorSlotsSemaphore;
  }

  /**
   * @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, executorSlotsSemaphore::release));
    } else {
      return false;
    }
  }

  @Override
  public void close() {
    if (completed.compareAndSet(false, true)) {
      executorSlotsSemaphore.release();
    }
  }

  /** 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.WorkflowTask;
26	import java.io.Closeable;
27	import java.util.concurrent.Semaphore;
28	import java.util.concurrent.atomic.AtomicBoolean;
29	import java.util.function.Function;
30	import javax.annotation.Nonnull;
31
32	public class WorkflowTaskDispatchHandle implements Closeable {
33	private final AtomicBoolean completed = new AtomicBoolean();	1✔
34	private final Function<WorkflowTask, Boolean> dispatchCallback;
35	private final Semaphore executorSlotsSemaphore;
36
37	/**
38	* @param dispatchCallback callback into a {@code WorkflowWorker} to dispatch a workflow task.
39	* @param executorSlotsSemaphore worker executor slots semaphore that was used to reserve this
40	* dispatch handle on
41	*/
42	public WorkflowTaskDispatchHandle(
43	DispatchCallback dispatchCallback, Semaphore executorSlotsSemaphore) {	1✔
44	this.dispatchCallback = dispatchCallback;	1✔
45	this.executorSlotsSemaphore = executorSlotsSemaphore;	1✔
46	}	1✔
47
48	/**
49	* @param workflowTask to be fed directly into the workflow worker
50	* @return true is the workflow task was successfully dispatched
51	* @throws IllegalArgumentException if the workflow task doesn't belong to the task queue of the
52	* worker provided this {@link WorkflowTaskDispatchHandle}
53	*/
54	public boolean dispatch(@Nonnull PollWorkflowTaskQueueResponse workflowTask) {
55	Preconditions.checkNotNull(workflowTask, "workflowTask");	1✔
56	if (completed.compareAndSet(false, true)) {	1✔
57	return dispatchCallback.apply(	1✔
58	new WorkflowTask(workflowTask, executorSlotsSemaphore::release));	1✔
59	} else {
60	return false;	×
61	}
62	}
63
64	@Override
65	public void close() {
66	if (completed.compareAndSet(false, true)) {	1✔
NEW 67	executorSlotsSemaphore.release();	×
68	}
69	}	1✔
70
71	/** A callback into a {@code WorkflowWorker} to dispatch a workflow task */
72	@FunctionalInterface
73	public interface DispatchCallback extends Function<WorkflowTask, Boolean> {
74
75	/**
76	* Should dispatch the Workflow Task to the Workflow Worker. Shouldn't block the thread.
77	*
78	* @param workflowTask WorkflowTask to be dispatched
79	* @return true if the dispatch was successful and false otherwise
80	* @throws IllegalArgumentException if {@code workflowTask} doesn't belong to the task queue of the Worker that provided the {@link WorkflowTaskDispatchHandle
81	*/
82	@Override
83	Boolean apply(WorkflowTask workflowTask) throws IllegalArgumentException;
84	}
85	}

temporalio / sdk-java / #278

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