2409

Committed 03 Jul 2024 08:33PM CUT coverage: 61.467% (-0.05%) from 61.518%

Build # 2409

Build Type

push

buildkite

Committed by

web-flow

Commit Message

Avoid consuming ByteBuffers (#913)

A ByteBuffer is a pointer to a byte[] with a starting position, a current position, and a limit. Any function that reads from its contents updates the current position. Both TracingPropagator and WorkflowUtils copy the entirety of its contents, and in doing so they mutate the current position. WorkflowUtils resets it afterwards but this still isn't thread-safe as another thread may be trying to read it.

By duplicating the ByteBuffer (copying only the metadata, not the actual contents) we avoid modifying it. It doesn't seem likely that there's real impact in either of these cases beyond unit tests, where these ByteBuffers stick around in the workflow history and are repeatedly serialized/deserialized. Modifying them during serialization can create test flakiness as that can trigger exceptions.

Run Details

2 of 2 new or added lines in 2 files covered. (100.0%)

10 existing lines in 4 files now uncovered.

11972 of 19477 relevant lines covered (61.47%)

0.61 hits per line

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

92.65

/src/main/java/com/uber/cadence/internal/sync/CancellationScopeImpl.java

/*
 *  Copyright 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 file except in compliance with the License. A copy of the License is
 *  located at
 *
 *  http://aws.amazon.com/apache2.0
 *
 *  or in the "license" file accompanying this file. This file 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 com.uber.cadence.internal.sync;

import com.uber.cadence.workflow.CancellationScope;
import com.uber.cadence.workflow.CompletablePromise;
import com.uber.cadence.workflow.Functions;
import com.uber.cadence.workflow.Workflow;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.Set;

class CancellationScopeImpl implements CancellationScope {

  private static ThreadLocal<Deque<CancellationScopeImpl>> scopeStack =
      ThreadLocal.withInitial(ArrayDeque::new);
  private boolean detached;
  private CompletablePromise<String> cancellationPromise;

  static CancellationScopeImpl current() {
    if (scopeStack.get().isEmpty()) {
      throw new IllegalStateException("Cannot be called by non workflow thread");
    }
    return scopeStack.get().peekFirst();
  }

  private static void pushCurrent(CancellationScopeImpl scope) {
    scopeStack.get().addFirst(scope);
  }

  private static void popCurrent(CancellationScopeImpl expected) {
    CancellationScopeImpl current = scopeStack.get().pollFirst();
    if (current != expected) {
      throw new Error("Unexpected scope");
    }
    if (!current.detached) {
      current.parent.removeChild(current);
    }
  }

  private final Runnable runnable;
  private CancellationScopeImpl parent;
  private final Set<CancellationScopeImpl> children = new HashSet<>();
  /**
   * When disconnected scope has no parent and thus doesn't receive cancellation requests from it.
   */
  private boolean cancelRequested;

  private String reason;

  CancellationScopeImpl(boolean ignoreParentCancellation, Runnable runnable) {
    this(ignoreParentCancellation, runnable, current());
  }

  CancellationScopeImpl(boolean detached, Runnable runnable, CancellationScopeImpl parent) {
    this.detached = detached;
    this.runnable = runnable;
    setParent(parent);
  }

  public CancellationScopeImpl(
      boolean ignoreParentCancellation, Functions.Proc1<CancellationScope> proc) {
    this.detached = ignoreParentCancellation;
    this.runnable = () -> proc.apply(this);
    setParent(current());
  }

  private void setParent(CancellationScopeImpl parent) {
    if (parent == null) {
      detached = true;
      return;
    }
    if (!detached) {
      this.parent = parent;
      parent.addChild(this);
      if (parent.isCancelRequested()) {
        cancel(parent.getCancellationReason());
      }
    }
  }

  @Override
  public void run() {
    try {
      pushCurrent(this);
      runnable.run();
    } finally {
      popCurrent(this);
    }
  }

  @Override
  public boolean isDetached() {
    return detached;
  }

  @Override
  public void cancel() {
    cancelRequested = true;
    reason = null;
    for (CancellationScopeImpl child : children) {
      child.cancel();
    }
    if (cancellationPromise != null) {
      cancellationPromise.complete(null);
    }
  }

  @Override
  public void cancel(String reason) {
    cancelRequested = true;
    this.reason = reason;
    for (CancellationScopeImpl child : children) {
      child.cancel(reason);
    }
    if (cancellationPromise != null) {
      cancellationPromise.complete(reason);
    }
  }

  @Override
  public String getCancellationReason() {
    return reason;
  }

  @Override
  public boolean isCancelRequested() {
    return cancelRequested;
  }

  @Override
  public CompletablePromise<String> getCancellationRequest() {
    if (cancellationPromise == null) {
      cancellationPromise = Workflow.newPromise();
      if (isCancelRequested()) {
        cancellationPromise.complete(getCancellationReason());
      }
    }
    return cancellationPromise;
  }

  private void addChild(CancellationScopeImpl scope) {
    children.add(scope);
  }

  private void removeChild(CancellationScopeImpl scope) {
    if (!children.remove(scope)) {
      throw new Error("Not a child");
    }
  }
}

1	/*
2	* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
3	*
4	* Modifications copyright (C) 2017 Uber Technologies, Inc.
5	*
6	* Licensed under the Apache License, Version 2.0 (the "License"). You may not
7	* use this file except in compliance with the License. A copy of the License is
8	* located at
9	*
10	* http://aws.amazon.com/apache2.0
11	*
12	* or in the "license" file accompanying this file. This file is distributed on
13	* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
14	* express or implied. See the License for the specific language governing
15	* permissions and limitations under the License.
16	*/
17
18	package com.uber.cadence.internal.sync;
19
20	import com.uber.cadence.workflow.CancellationScope;
21	import com.uber.cadence.workflow.CompletablePromise;
22	import com.uber.cadence.workflow.Functions;
23	import com.uber.cadence.workflow.Workflow;
24	import java.util.ArrayDeque;
25	import java.util.Deque;
26	import java.util.HashSet;
27	import java.util.Set;
28
29	class CancellationScopeImpl implements CancellationScope {
30
31	private static ThreadLocal<Deque<CancellationScopeImpl>> scopeStack =	1✔
32	ThreadLocal.withInitial(ArrayDeque::new);	1✔
33	private boolean detached;
34	private CompletablePromise<String> cancellationPromise;
35
36	static CancellationScopeImpl current() {
37	if (scopeStack.get().isEmpty()) {	1✔
38	throw new IllegalStateException("Cannot be called by non workflow thread");	×
39	}
40	return scopeStack.get().peekFirst();	1✔
41	}
42
43	private static void pushCurrent(CancellationScopeImpl scope) {
44	scopeStack.get().addFirst(scope);	1✔
45	}	1✔
46
47	private static void popCurrent(CancellationScopeImpl expected) {
48	CancellationScopeImpl current = scopeStack.get().pollFirst();	1✔
49	if (current != expected) {	1✔
50	throw new Error("Unexpected scope");	×
51	}
52	if (!current.detached) {	1✔
53	current.parent.removeChild(current);	1✔
54	}
55	}	1✔
56
57	private final Runnable runnable;
58	private CancellationScopeImpl parent;
59	private final Set<CancellationScopeImpl> children = new HashSet<>();	1✔
60	/**
61	* When disconnected scope has no parent and thus doesn't receive cancellation requests from it.
62	*/
63	private boolean cancelRequested;
64
65	private String reason;
66
67	CancellationScopeImpl(boolean ignoreParentCancellation, Runnable runnable) {
68	this(ignoreParentCancellation, runnable, current());	1✔
69	}	1✔
70
71	CancellationScopeImpl(boolean detached, Runnable runnable, CancellationScopeImpl parent) {	1✔
72	this.detached = detached;	1✔
73	this.runnable = runnable;	1✔
74	setParent(parent);	1✔
75	}	1✔
76
77	public CancellationScopeImpl(
78	boolean ignoreParentCancellation, Functions.Proc1<CancellationScope> proc) {	1✔
79	this.detached = ignoreParentCancellation;	1✔
80	this.runnable = () -> proc.apply(this);	1✔
81	setParent(current());	1✔
82	}	1✔
83
84	private void setParent(CancellationScopeImpl parent) {
85	if (parent == null) {	1✔
86	detached = true;	1✔
87	return;	1✔
88	}
89	if (!detached) {	1✔
90	this.parent = parent;	1✔
91	parent.addChild(this);	1✔
92	if (parent.isCancelRequested()) {	1✔
UNCOV 93	cancel(parent.getCancellationReason());	×
94	}
95	}
96	}	1✔
97
98	@Override
99	public void run() {
100	try {
101	pushCurrent(this);	1✔
102	runnable.run();	1✔
103	} finally {
104	popCurrent(this);	1✔
105	}
106	}	1✔
107
108	@Override
109	public boolean isDetached() {
110	return detached;	×
111	}
112
113	@Override
114	public void cancel() {
115	cancelRequested = true;	1✔
116	reason = null;	1✔
117	for (CancellationScopeImpl child : children) {	1✔
118	child.cancel();	1✔
119	}	1✔
120	if (cancellationPromise != null) {	1✔
121	cancellationPromise.complete(null);	1✔
122	}
123	}	1✔
124
125	@Override
126	public void cancel(String reason) {
127	cancelRequested = true;	1✔
128	this.reason = reason;	1✔
129	for (CancellationScopeImpl child : children) {	1✔
130	child.cancel(reason);	1✔
131	}	1✔
132	if (cancellationPromise != null) {	1✔
133	cancellationPromise.complete(reason);	1✔
134	}
135	}	1✔
136
137	@Override
138	public String getCancellationReason() {
139	return reason;	1✔
140	}
141
142	@Override
143	public boolean isCancelRequested() {
144	return cancelRequested;	1✔
145	}
146
147	@Override
148	public CompletablePromise<String> getCancellationRequest() {
149	if (cancellationPromise == null) {	1✔
150	cancellationPromise = Workflow.newPromise();	1✔
151	if (isCancelRequested()) {	1✔
152	cancellationPromise.complete(getCancellationReason());	1✔
153	}
154	}
155	return cancellationPromise;	1✔
156	}
157
158	private void addChild(CancellationScopeImpl scope) {
159	children.add(scope);	1✔
160	}	1✔
161
162	private void removeChild(CancellationScopeImpl scope) {
163	if (!children.remove(scope)) {	1✔
164	throw new Error("Not a child");	×
165	}
166	}	1✔
167	}

uber / cadence-java-client / 2409

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