#387

Committed 15 Sep 2026 10:00PM UTC coverage: 68.408% (+0.04%) from 68.372%

Build # #387

Build Type

push

github

Committed by

web-flow

Commit Message

Account for dynamic workflows when setting WorkflowImplementationOptions (#3042)

Coverage Stats

7844 of 13600 branches covered (57.68%)

Branch coverage included in aggregate %.

6 of 6 new or added lines in 1 file covered. (100.0%)

3 existing lines in 3 files now uncovered.

31842 of 44414 relevant lines covered (71.69%)

0.72 hits per line

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

73.93

/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowThreadImpl.java

package io.temporal.internal.sync;

import com.google.common.base.Preconditions;
import io.temporal.common.context.ContextPropagator;
import io.temporal.failure.CanceledFailure;
import io.temporal.internal.common.NonIdempotentHandle;
import io.temporal.internal.common.SdkFlag;
import io.temporal.internal.context.ContextThreadLocal;
import io.temporal.internal.logging.LoggerTag;
import io.temporal.internal.replay.ReplayWorkflowContext;
import io.temporal.internal.worker.WorkflowExecutorCache;
import io.temporal.workflow.Functions;
import io.temporal.workflow.Promise;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.function.Supplier;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;

class WorkflowThreadImpl implements WorkflowThread {
  /**
   * Runnable passed to the thread that wraps a runnable passed to the WorkflowThreadImpl
   * constructor.
   */
  class RunnableWrapper implements Runnable {

    private final WorkflowThreadContext threadContext;
    // TODO: Move MDC injection logic into an interceptor as this context shouldn't be leaked
    // to the WorkflowThreadImpl
    private final ReplayWorkflowContext replayWorkflowContext;
    private String originalName;
    private String name;
    private final CancellationScopeImpl cancellationScope;
    private final List<ContextPropagator> contextPropagators;
    private final Map<String, Object> propagatedContexts;

    RunnableWrapper(
        WorkflowThreadContext threadContext,
        ReplayWorkflowContext replayWorkflowContext,
        String name,
        boolean detached,
        CancellationScopeImpl parent,
        Runnable runnable,
        List<ContextPropagator> contextPropagators,
        Map<String, Object> propagatedContexts) {
      this.threadContext = threadContext;
      this.replayWorkflowContext = replayWorkflowContext;
      this.name = name;
      boolean deterministicCancellationScopeOrder =
          replayWorkflowContext.checkSdkFlag(SdkFlag.DETERMINISTIC_CANCELLATION_SCOPE_ORDER);
      this.cancellationScope =
          new CancellationScopeImpl(
              detached, deterministicCancellationScopeOrder, runnable, parent);
      Preconditions.checkState(
          context.getStatus() == Status.CREATED, "threadContext not in CREATED state");
      this.contextPropagators = contextPropagators;
      this.propagatedContexts = propagatedContexts;
    }

    @Override
    public void run() {
      Thread thread = Thread.currentThread();
      originalName = thread.getName();
      thread.setName(name);

      threadContext.initializeCurrentThread(thread);
      DeterministicRunnerImpl.setCurrentThreadInternal(WorkflowThreadImpl.this);

      MDC.put(LoggerTag.WORKFLOW_ID, replayWorkflowContext.getWorkflowId());
      MDC.put(LoggerTag.WORKFLOW_TYPE, replayWorkflowContext.getWorkflowType().getName());
      MDC.put(LoggerTag.RUN_ID, replayWorkflowContext.getRunId());
      MDC.put(LoggerTag.TASK_QUEUE, replayWorkflowContext.getTaskQueue());
      MDC.put(LoggerTag.NAMESPACE, replayWorkflowContext.getNamespace());

      // Repopulate the context(s)
      ContextThreadLocal.setContextPropagators(this.contextPropagators);
      ContextThreadLocal.propagateContextToCurrentThread(this.propagatedContexts);
      try {
        // initialYield blocks thread until the first runUntilBlocked is called.
        // Otherwise, r starts executing without control of the sync.
        threadContext.initialYield();
        cancellationScope.run();
      } catch (DestroyWorkflowThreadError e) {
        if (!threadContext.isDestroyRequested()) {
          threadContext.setUnhandledException(e);
        }
      } catch (Error e) {
        threadContext.setUnhandledException(e);
      } catch (CanceledFailure e) {
        if (!isCancelRequested()) {
          threadContext.setUnhandledException(e);
        }
        if (log.isDebugEnabled()) {
          log.debug(String.format("Workflow thread \"%s\" run canceled", name));
        }
      } catch (Throwable e) {
        threadContext.setUnhandledException(e);
      } finally {
        DeterministicRunnerImpl.setCurrentThreadInternal(null);
        threadContext.makeDone();
        thread.setName(originalName);
        MDC.clear();
      }
    }

    public String getName() {
      return name;
    }

    StackTraceElement[] getStackTrace() {
      @Nullable Thread thread = threadContext.getCurrentThread();
      if (thread != null) {
        return thread.getStackTrace();
      }
      return new StackTraceElement[0];
    }

    public void setName(String name) {
      this.name = name;
      @Nullable Thread thread = threadContext.getCurrentThread();
      if (thread != null) {
        thread.setName(name);
      }
    }
  }

  private static final Logger log = LoggerFactory.getLogger(WorkflowThreadImpl.class);

  private final WorkflowThreadExecutor workflowThreadExecutor;
  private final WorkflowThreadContext context;
  private final WorkflowExecutorCache cache;
  private final SyncWorkflowContext syncWorkflowContext;

  private final DeterministicRunnerImpl runner;
  private final RunnableWrapper task;
  private final int priority;
  private Future<?> taskFuture;
  private final Map<WorkflowThreadLocalInternal<?>, Object> threadLocalMap = new HashMap<>();

  WorkflowThreadImpl(
      WorkflowThreadExecutor workflowThreadExecutor,
      SyncWorkflowContext syncWorkflowContext,
      DeterministicRunnerImpl runner,
      @Nonnull String name,
      int priority,
      boolean detached,
      CancellationScopeImpl parentCancellationScope,
      Runnable runnable,
      WorkflowExecutorCache cache,
      List<ContextPropagator> contextPropagators,
      Map<String, Object> propagatedContexts) {
    this.workflowThreadExecutor = workflowThreadExecutor;
    this.syncWorkflowContext = Preconditions.checkNotNull(syncWorkflowContext);
    this.runner = runner;
    this.context = new WorkflowThreadContext(runner.getLock());
    this.cache = cache;
    this.priority = priority;
    this.task =
        new RunnableWrapper(
            context,
            syncWorkflowContext.getReplayContext(),
            Preconditions.checkNotNull(name, "Thread name shouldn't be null"),
            detached,
            parentCancellationScope,
            runnable,
            contextPropagators,
            propagatedContexts);
  }

  @Override
  public void run() {
    throw new UnsupportedOperationException("not used");
  }

  @Override
  public boolean isDetached() {
    return task.cancellationScope.isDetached();
  }

  @Override
  public void cancel() {
    task.cancellationScope.cancel();
  }

  @Override
  public void cancel(String reason) {
    task.cancellationScope.cancel(reason);
  }

  @Override
  public String getCancellationReason() {
    return task.cancellationScope.getCancellationReason();
  }

  @Override
  public boolean isCancelRequested() {
    return task.cancellationScope.isCancelRequested();
  }

  @Override
  public Promise<String> getCancellationRequest() {
    return task.cancellationScope.getCancellationRequest();
  }

  @Override
  public void start() {
    context.verifyAndStart();
    while (true) {
      try {
        taskFuture = workflowThreadExecutor.submit(task);
        return;
      } catch (RejectedExecutionException e) {
        if (cache != null) {
          SyncWorkflowContext workflowContext = getWorkflowContext();
          ReplayWorkflowContext context = workflowContext.getReplayContext();
          boolean evicted =
              cache.evictAnyNotInProcessing(
                  context.getWorkflowExecution(), workflowContext.getMetricsScope());
          if (!evicted) {
            // Note here we need to throw error, not exception. Otherwise it will be
            // translated to workflow execution exception and instead of failing the
            // workflow task we will be failing the workflow.
            throw new WorkflowRejectedExecutionError(e);
          }
        } else {
          throw new WorkflowRejectedExecutionError(e);
        }
      }
    }
  }

  @Override
  public boolean isStarted() {
    return context.getStatus() != Status.CREATED;
  }

  @Override
  public WorkflowThreadContext getWorkflowThreadContext() {
    return context;
  }

  @Override
  public DeterministicRunnerImpl getRunner() {
    return runner;
  }

  @Override
  public SyncWorkflowContext getWorkflowContext() {
    return syncWorkflowContext;
  }

  @Override
  public void setName(String name) {
    task.setName(name);
  }

  @Override
  public String getName() {
    return task.getName();
  }

  @Override
  public long getId() {
    return hashCode();
  }

  @Override
  public int getPriority() {
    return priority;
  }

  @Override
  public boolean runUntilBlocked(long deadlockDetectionTimeoutMs) {
    if (taskFuture == null) {
      start();
    }
    return context.runUntilBlocked(deadlockDetectionTimeoutMs);
  }

  @Override
  public NonIdempotentHandle lockDeadlockDetector() {
    return context.lockDeadlockDetector();
  }

  @Override
  public boolean isDone() {
    return context.isDone();
  }

  @Override
  public Throwable getUnhandledException() {
    return context.getUnhandledException();
  }

  /**
   * Evaluates function in the threadContext of the coroutine without unblocking it. Used to get
   * current coroutine status, like stack trace.
   *
   * @param function Parameter is reason for current goroutine blockage.
   */
  public void evaluateInCoroutineContext(Functions.Proc1<String> function) {
    context.evaluateInCoroutineContext(function);
  }

  /**
   * Interrupt coroutine by throwing DestroyWorkflowThreadError from an await method it is blocked
   * on and return underlying Future to be waited on.
   */
  @Override
  public Future<?> stopNow() {
    // Cannot call destroy() on itself
    @Nullable Thread thread = context.getCurrentThread();
    if (Thread.currentThread().equals(thread)) {
      throw new Error("Cannot call destroy on itself: " + thread.getName());
    }
    context.initiateDestroy();
    if (taskFuture == null) {
      return getCompletedFuture();
    }
    return taskFuture;
  }

  private Future<?> getCompletedFuture() {
    CompletableFuture<String> f = new CompletableFuture<>();
    f.complete("done");
    return f;
  }

  @Override
  public void addStackTrace(StringBuilder result) {
    result.append(getName());
    @Nullable Thread thread = context.getCurrentThread();
    if (thread == null) {
      result.append("(NEW)");
      return;
    }
    result
        .append(": (BLOCKED on ")
        .append(getWorkflowThreadContext().getYieldReason())
        .append(")\n");
    // These numbers might change if implementation changes.
    int omitTop = 5;
    int omitBottom = 7;
    // TODO it's not a good idea to rely on the name to understand the thread type. Instead of that
    // we would better
    // assign an explicit thread type enum to the threads. This will be especially important when we
    // refactor
    // root and workflow-method
    // thread names into names that will include workflowId
    if (DeterministicRunnerImpl.WORKFLOW_ROOT_THREAD_NAME.equals(getName())) {
      // TODO revisit this number
      omitBottom = 11;
    } else if (getName().startsWith(WorkflowMethodThreadNameStrategy.WORKFLOW_MAIN_THREAD_PREFIX)) {
      // TODO revisit this number
      omitBottom = 11;
    }
    StackTraceElement[] stackTrace = thread.getStackTrace();
    for (int i = omitTop; i < stackTrace.length - omitBottom; i++) {
      StackTraceElement e = stackTrace[i];
      if (i == omitTop && "await".equals(e.getMethodName())) continue;
      result.append(e);
      result.append("\n");
    }
  }

  @Override
  public void yield(String reason, Supplier<Boolean> unblockCondition) {
    context.yield(reason, unblockCondition);
  }

  @Override
  public void exitThread() {
    runner.exit();
    throw new DestroyWorkflowThreadError("exit");
  }

  @Override
  public <T> void setThreadLocal(WorkflowThreadLocalInternal<T> key, T value) {
    threadLocalMap.put(key, value);
  }

  /**
   * Retrieve data from thread locals. Returns 1. not found (an empty Optional) 2. found but null
   * (an Optional of an empty Optional) 3. found and non-null (an Optional of an Optional of a
   * value). The type nesting is because Java Optionals cannot understand "Some null" vs "None",
   * which is exactly what we need here.
   *
   * @param key
   * @return one of three cases
   * @param <T>
   */
  @SuppressWarnings("unchecked")
  public <T> Optional<Optional<T>> getThreadLocal(WorkflowThreadLocalInternal<T> key) {
    if (!threadLocalMap.containsKey(key)) {
      return Optional.empty();
    }
    return Optional.of(Optional.ofNullable((T) threadLocalMap.get(key)));
  }

  /**
   * @return stack trace of the coroutine thread
   */
  @Override
  public String getStackTrace() {
    StackTraceElement[] st = task.getStackTrace();
    StringWriter sw = new StringWriter();
    PrintWriter pw = new PrintWriter(sw);
    pw.append(task.getName());
    pw.append("\n");
    for (StackTraceElement se : st) {
      pw.println("\tat " + se);
    }
    return sw.toString();
  }

  static class YieldWithTimeoutCondition implements Supplier<Boolean> {

    private final Supplier<Boolean> unblockCondition;
    private final long blockedUntil;
    private boolean timedOut;

    YieldWithTimeoutCondition(Supplier<Boolean> unblockCondition, long blockedUntil) {
      this.unblockCondition = unblockCondition;
      this.blockedUntil = blockedUntil;
    }

    boolean isTimedOut() {
      return timedOut;
    }

    /**
     * @return true if condition matched or timed out
     */
    @Override
    public Boolean get() {
      boolean result = unblockCondition.get();
      if (result) {
        return true;
      }
      long currentTimeMillis = WorkflowInternal.currentTimeMillis();
      timedOut = currentTimeMillis >= blockedUntil;
      return timedOut;
    }
  }
}

1	package io.temporal.internal.sync;
2
3	import com.google.common.base.Preconditions;
4	import io.temporal.common.context.ContextPropagator;
5	import io.temporal.failure.CanceledFailure;
6	import io.temporal.internal.common.NonIdempotentHandle;
7	import io.temporal.internal.common.SdkFlag;
8	import io.temporal.internal.context.ContextThreadLocal;
9	import io.temporal.internal.logging.LoggerTag;
10	import io.temporal.internal.replay.ReplayWorkflowContext;
11	import io.temporal.internal.worker.WorkflowExecutorCache;
12	import io.temporal.workflow.Functions;
13	import io.temporal.workflow.Promise;
14	import java.io.PrintWriter;
15	import java.io.StringWriter;
16	import java.util.HashMap;
17	import java.util.List;
18	import java.util.Map;
19	import java.util.Optional;
20	import java.util.concurrent.CompletableFuture;
21	import java.util.concurrent.Future;
22	import java.util.concurrent.RejectedExecutionException;
23	import java.util.function.Supplier;
24	import javax.annotation.Nonnull;
25	import javax.annotation.Nullable;
26	import org.slf4j.Logger;
27	import org.slf4j.LoggerFactory;
28	import org.slf4j.MDC;
29
30	class WorkflowThreadImpl implements WorkflowThread {
31	/**
32	* Runnable passed to the thread that wraps a runnable passed to the WorkflowThreadImpl
33	* constructor.
34	*/
35	class RunnableWrapper implements Runnable {
36
37	private final WorkflowThreadContext threadContext;
38	// TODO: Move MDC injection logic into an interceptor as this context shouldn't be leaked
39	// to the WorkflowThreadImpl
40	private final ReplayWorkflowContext replayWorkflowContext;
41	private String originalName;
42	private String name;
43	private final CancellationScopeImpl cancellationScope;
44	private final List<ContextPropagator> contextPropagators;
45	private final Map<String, Object> propagatedContexts;
46
47	RunnableWrapper(
48	WorkflowThreadContext threadContext,
49	ReplayWorkflowContext replayWorkflowContext,
50	String name,
51	boolean detached,
52	CancellationScopeImpl parent,
53	Runnable runnable,
54	List<ContextPropagator> contextPropagators,
55	Map<String, Object> propagatedContexts) {	1✔
56	this.threadContext = threadContext;	1✔
57	this.replayWorkflowContext = replayWorkflowContext;	1✔
58	this.name = name;	1✔
59	boolean deterministicCancellationScopeOrder =	1✔
60	replayWorkflowContext.checkSdkFlag(SdkFlag.DETERMINISTIC_CANCELLATION_SCOPE_ORDER);	1✔
61	this.cancellationScope =	1✔
62	new CancellationScopeImpl(
63	detached, deterministicCancellationScopeOrder, runnable, parent);
64	Preconditions.checkState(	1✔
65	context.getStatus() == Status.CREATED, "threadContext not in CREATED state");	1!
66	this.contextPropagators = contextPropagators;	1✔
67	this.propagatedContexts = propagatedContexts;	1✔
68	}	1✔
69
70	@Override
71	public void run() {
72	Thread thread = Thread.currentThread();	1✔
73	originalName = thread.getName();	1✔
74	thread.setName(name);	1✔
75
76	threadContext.initializeCurrentThread(thread);	1✔
77	DeterministicRunnerImpl.setCurrentThreadInternal(WorkflowThreadImpl.this);	1✔
78
79	MDC.put(LoggerTag.WORKFLOW_ID, replayWorkflowContext.getWorkflowId());	1✔
80	MDC.put(LoggerTag.WORKFLOW_TYPE, replayWorkflowContext.getWorkflowType().getName());	1✔
81	MDC.put(LoggerTag.RUN_ID, replayWorkflowContext.getRunId());	1✔
82	MDC.put(LoggerTag.TASK_QUEUE, replayWorkflowContext.getTaskQueue());	1✔
83	MDC.put(LoggerTag.NAMESPACE, replayWorkflowContext.getNamespace());	1✔
84
85	// Repopulate the context(s)
86	ContextThreadLocal.setContextPropagators(this.contextPropagators);	1✔
87	ContextThreadLocal.propagateContextToCurrentThread(this.propagatedContexts);	1✔
88	try {
89	// initialYield blocks thread until the first runUntilBlocked is called.
90	// Otherwise, r starts executing without control of the sync.
91	threadContext.initialYield();	1✔
92	cancellationScope.run();	1✔
93	} catch (DestroyWorkflowThreadError e) {	1✔
94	if (!threadContext.isDestroyRequested()) {	1✔
95	threadContext.setUnhandledException(e);	1✔
96	}
97	} catch (Error e) {	1✔
98	threadContext.setUnhandledException(e);	1✔
99	} catch (CanceledFailure e) {	×
100	if (!isCancelRequested()) {	×
101	threadContext.setUnhandledException(e);	×
102	}
103	if (log.isDebugEnabled()) {	×
104	log.debug(String.format("Workflow thread \"%s\" run canceled", name));	×
105	}
106	} catch (Throwable e) {	1✔
107	threadContext.setUnhandledException(e);	1✔
108	} finally {
109	DeterministicRunnerImpl.setCurrentThreadInternal(null);	1✔
110	threadContext.makeDone();	1✔
111	thread.setName(originalName);	1✔
112	MDC.clear();	1✔
113	}
114	}	1✔
115
116	public String getName() {
117	return name;	1✔
118	}
119
120	StackTraceElement[] getStackTrace() {
121	@Nullable Thread thread = threadContext.getCurrentThread();	1✔
122	if (thread != null) {	1✔
123	return thread.getStackTrace();	1✔
124	}
125	return new StackTraceElement[0];	1✔
126	}
127
128	public void setName(String name) {
129	this.name = name;	×
130	@Nullable Thread thread = threadContext.getCurrentThread();	×
131	if (thread != null) {	×
132	thread.setName(name);	×
133	}
134	}	×
135	}
136
137	private static final Logger log = LoggerFactory.getLogger(WorkflowThreadImpl.class);	1✔
138
139	private final WorkflowThreadExecutor workflowThreadExecutor;
140	private final WorkflowThreadContext context;
141	private final WorkflowExecutorCache cache;
142	private final SyncWorkflowContext syncWorkflowContext;
143
144	private final DeterministicRunnerImpl runner;
145	private final RunnableWrapper task;
146	private final int priority;
147	private Future<?> taskFuture;
148	private final Map<WorkflowThreadLocalInternal<?>, Object> threadLocalMap = new HashMap<>();	1✔
149
150	WorkflowThreadImpl(
151	WorkflowThreadExecutor workflowThreadExecutor,
152	SyncWorkflowContext syncWorkflowContext,
153	DeterministicRunnerImpl runner,
154	@Nonnull String name,
155	int priority,
156	boolean detached,
157	CancellationScopeImpl parentCancellationScope,
158	Runnable runnable,
159	WorkflowExecutorCache cache,
160	List<ContextPropagator> contextPropagators,
161	Map<String, Object> propagatedContexts) {	1✔
162	this.workflowThreadExecutor = workflowThreadExecutor;	1✔
163	this.syncWorkflowContext = Preconditions.checkNotNull(syncWorkflowContext);	1✔
164	this.runner = runner;	1✔
165	this.context = new WorkflowThreadContext(runner.getLock());	1✔
166	this.cache = cache;	1✔
167	this.priority = priority;	1✔
168	this.task =	1✔
169	new RunnableWrapper(
170	context,
171	syncWorkflowContext.getReplayContext(),	1✔
172	Preconditions.checkNotNull(name, "Thread name shouldn't be null"),	1✔
173	detached,
174	parentCancellationScope,
175	runnable,
176	contextPropagators,
177	propagatedContexts);
178	}	1✔
179
180	@Override
181	public void run() {
182	throw new UnsupportedOperationException("not used");	×
183	}
184
185	@Override
186	public boolean isDetached() {
187	return task.cancellationScope.isDetached();	×
188	}
189
190	@Override
191	public void cancel() {
192	task.cancellationScope.cancel();	1✔
193	}	1✔
194
195	@Override
196	public void cancel(String reason) {
197	task.cancellationScope.cancel(reason);	1✔
198	}	1✔
199
200	@Override
201	public String getCancellationReason() {
202	return task.cancellationScope.getCancellationReason();	×
203	}
204
205	@Override
206	public boolean isCancelRequested() {
207	return task.cancellationScope.isCancelRequested();	×
208	}
209
210	@Override
211	public Promise<String> getCancellationRequest() {
212	return task.cancellationScope.getCancellationRequest();	×
213	}
214
215	@Override
216	public void start() {
217	context.verifyAndStart();	1✔
218	while (true) {
219	try {
220	taskFuture = workflowThreadExecutor.submit(task);	1✔
221	return;	1✔
222	} catch (RejectedExecutionException e) {	1✔
223	if (cache != null) {	1✔
224	SyncWorkflowContext workflowContext = getWorkflowContext();	1✔
225	ReplayWorkflowContext context = workflowContext.getReplayContext();	1✔
226	boolean evicted =	1✔
227	cache.evictAnyNotInProcessing(	1✔
228	context.getWorkflowExecution(), workflowContext.getMetricsScope());	1✔
229	if (!evicted) {	1!
230	// Note here we need to throw error, not exception. Otherwise it will be
231	// translated to workflow execution exception and instead of failing the
232	// workflow task we will be failing the workflow.
UNCOV 233	throw new WorkflowRejectedExecutionError(e);	×
234	}
235	} else {	1✔
236	throw new WorkflowRejectedExecutionError(e);	1✔
237	}
238	}	1✔
239	}
240	}
241
242	@Override
243	public boolean isStarted() {
244	return context.getStatus() != Status.CREATED;	×
245	}
246
247	@Override
248	public WorkflowThreadContext getWorkflowThreadContext() {
249	return context;	1✔
250	}
251
252	@Override
253	public DeterministicRunnerImpl getRunner() {
254	return runner;	1✔
255	}
256
257	@Override
258	public SyncWorkflowContext getWorkflowContext() {
259	return syncWorkflowContext;	1✔
260	}
261
262	@Override
263	public void setName(String name) {
264	task.setName(name);	×
265	}	×
266
267	@Override
268	public String getName() {
269	return task.getName();	1✔
270	}
271
272	@Override
273	public long getId() {
274	return hashCode();	×
275	}
276
277	@Override
278	public int getPriority() {
279	return priority;	1✔
280	}
281
282	@Override
283	public boolean runUntilBlocked(long deadlockDetectionTimeoutMs) {
284	if (taskFuture == null) {	1✔
285	start();	1✔
286	}
287	return context.runUntilBlocked(deadlockDetectionTimeoutMs);	1✔
288	}
289
290	@Override
291	public NonIdempotentHandle lockDeadlockDetector() {
292	return context.lockDeadlockDetector();	1✔
293	}
294
295	@Override
296	public boolean isDone() {
297	return context.isDone();	1✔
298	}
299
300	@Override
301	public Throwable getUnhandledException() {
302	return context.getUnhandledException();	1✔
303	}
304
305	/**
306	* Evaluates function in the threadContext of the coroutine without unblocking it. Used to get
307	* current coroutine status, like stack trace.
308	*
309	* @param function Parameter is reason for current goroutine blockage.
310	*/
311	public void evaluateInCoroutineContext(Functions.Proc1<String> function) {
312	context.evaluateInCoroutineContext(function);	×
313	}	×
314
315	/**
316	* Interrupt coroutine by throwing DestroyWorkflowThreadError from an await method it is blocked
317	* on and return underlying Future to be waited on.
318	*/
319	@Override
320	public Future<?> stopNow() {
321	// Cannot call destroy() on itself
322	@Nullable Thread thread = context.getCurrentThread();	1✔
323	if (Thread.currentThread().equals(thread)) {	1!
324	throw new Error("Cannot call destroy on itself: " + thread.getName());	×
325	}
326	context.initiateDestroy();	1✔
327	if (taskFuture == null) {	1✔
328	return getCompletedFuture();	1✔
329	}
330	return taskFuture;	1✔
331	}
332
333	private Future<?> getCompletedFuture() {
334	CompletableFuture<String> f = new CompletableFuture<>();	1✔
335	f.complete("done");	1✔
336	return f;	1✔
337	}
338
339	@Override
340	public void addStackTrace(StringBuilder result) {
341	result.append(getName());	1✔
342	@Nullable Thread thread = context.getCurrentThread();	1✔
343	if (thread == null) {	1!
344	result.append("(NEW)");	×
345	return;	×
346	}
347	result	1✔
348	.append(": (BLOCKED on ")	1✔
349	.append(getWorkflowThreadContext().getYieldReason())	1✔
350	.append(")\n");	1✔
351	// These numbers might change if implementation changes.
352	int omitTop = 5;	1✔
353	int omitBottom = 7;	1✔
354	// TODO it's not a good idea to rely on the name to understand the thread type. Instead of that
355	// we would better
356	// assign an explicit thread type enum to the threads. This will be especially important when we
357	// refactor
358	// root and workflow-method
359	// thread names into names that will include workflowId
360	if (DeterministicRunnerImpl.WORKFLOW_ROOT_THREAD_NAME.equals(getName())) {	1!
361	// TODO revisit this number
362	omitBottom = 11;	×
363	} else if (getName().startsWith(WorkflowMethodThreadNameStrategy.WORKFLOW_MAIN_THREAD_PREFIX)) {	1✔
364	// TODO revisit this number
365	omitBottom = 11;	1✔
366	}
367	StackTraceElement[] stackTrace = thread.getStackTrace();	1✔
368	for (int i = omitTop; i < stackTrace.length - omitBottom; i++) {	1✔
369	StackTraceElement e = stackTrace[i];	1✔
370	if (i == omitTop && "await".equals(e.getMethodName())) continue;	1!
371	result.append(e);	1✔
372	result.append("\n");	1✔
373	}
374	}	1✔
375
376	@Override
377	public void yield(String reason, Supplier<Boolean> unblockCondition) {
378	context.yield(reason, unblockCondition);	1✔
379	}	1✔
380
381	@Override
382	public void exitThread() {
383	runner.exit();	1✔
384	throw new DestroyWorkflowThreadError("exit");	1✔
385	}
386
387	@Override
388	public <T> void setThreadLocal(WorkflowThreadLocalInternal<T> key, T value) {
389	threadLocalMap.put(key, value);	1✔
390	}	1✔
391
392	/**
393	* Retrieve data from thread locals. Returns 1. not found (an empty Optional) 2. found but null
394	* (an Optional of an empty Optional) 3. found and non-null (an Optional of an Optional of a
395	* value). The type nesting is because Java Optionals cannot understand "Some null" vs "None",
396	* which is exactly what we need here.
397	*
398	* @param key
399	* @return one of three cases
400	* @param <T>
401	*/
402	@SuppressWarnings("unchecked")
403	public <T> Optional<Optional<T>> getThreadLocal(WorkflowThreadLocalInternal<T> key) {
404	if (!threadLocalMap.containsKey(key)) {	1✔
405	return Optional.empty();	1✔
406	}
407	return Optional.of(Optional.ofNullable((T) threadLocalMap.get(key)));	1✔
408	}
409
410	/**
411	* @return stack trace of the coroutine thread
412	*/
413	@Override
414	public String getStackTrace() {
415	StackTraceElement[] st = task.getStackTrace();	1✔
416	StringWriter sw = new StringWriter();	1✔
417	PrintWriter pw = new PrintWriter(sw);	1✔
418	pw.append(task.getName());	1✔
419	pw.append("\n");	1✔
420	for (StackTraceElement se : st) {	1✔
421	pw.println("\tat " + se);	1✔
422	}
423	return sw.toString();	1✔
424	}
425
426	static class YieldWithTimeoutCondition implements Supplier<Boolean> {
427
428	private final Supplier<Boolean> unblockCondition;
429	private final long blockedUntil;
430	private boolean timedOut;
431
432	YieldWithTimeoutCondition(Supplier<Boolean> unblockCondition, long blockedUntil) {	×
433	this.unblockCondition = unblockCondition;	×
434	this.blockedUntil = blockedUntil;	×
435	}	×
436
437	boolean isTimedOut() {
438	return timedOut;	×
439	}
440
441	/**
442	* @return true if condition matched or timed out
443	*/
444	@Override
445	public Boolean get() {
446	boolean result = unblockCondition.get();	×
447	if (result) {	×
448	return true;	×
449	}
450	long currentTimeMillis = WorkflowInternal.currentTimeMillis();	×
451	timedOut = currentTimeMillis >= blockedUntil;	×
452	return timedOut;	×
453	}
454	}
455	}

temporalio / sdk-java / #387

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