2416

Committed 03 Jul 2024 08:33PM UTC coverage: 61.478% (-0.04%) from 61.518%

Build # 2416

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%)

8 existing lines in 4 files now uncovered.

11974 of 19477 relevant lines covered (61.48%)

0.61 hits per line

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

68.42

/src/main/java/com/uber/cadence/internal/common/BackoffThrottler.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.common;

import java.time.Duration;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Used to throttle code execution in presence of failures using exponential backoff logic. The
 * formula used to calculate the next sleep interval is:
 *
 * <p>
 *
 * <pre>
 * min(pow(backoffCoefficient, failureCount - 1) * initialSleep, maxSleep);
 * </pre>
 *
 * <p>Example usage:
 *
 * <p>
 *
 * <pre>
 * BackoffThrottler throttler = new BackoffThrottler(1000, 60000, 2);
 * while(!stopped) {
 *     try {
 *         throttler.throttle();
 *         // some code that can fail and should be throttled
 *         ...
 *         throttler.success();
 *     }
 *     catch (Exception e) {
 *         throttler.failure();
 *     }
 * }
 * </pre>
 *
 * @author fateev
 */
public final class BackoffThrottler {

  private final Duration initialSleep;

  private final Duration maxSleep;

  private final double backoffCoefficient;

  private final AtomicLong failureCount = new AtomicLong();

  /**
   * Construct an instance of the throttler.
   *
   * @param initialSleep time to sleep on the first failure
   * @param maxSleep maximum time to sleep independently of number of failures
   * @param backoffCoefficient coefficient used to calculate the next time to sleep.
   */
  public BackoffThrottler(Duration initialSleep, Duration maxSleep, double backoffCoefficient) {
    Objects.requireNonNull(initialSleep, "initialSleep");
    this.initialSleep = initialSleep;
    this.maxSleep = maxSleep;
    this.backoffCoefficient = backoffCoefficient;
  }

  private long calculateSleepTime() {
    double sleepMillis =
        Math.pow(backoffCoefficient, failureCount.get() - 1) * initialSleep.toMillis();
    if (maxSleep != null) {
      return Math.min((long) sleepMillis, maxSleep.toMillis());
    }
    return (long) sleepMillis;
  }

  /**
   * Sleep if there were failures since the last success call.
   *
   * @throws InterruptedException
   */
  public void throttle() throws InterruptedException {
    if (failureCount.get() > 0) {
      Thread.sleep(calculateSleepTime());
    }
  }

  /** Resent failure count to 0. */
  public void success() {
    failureCount.set(0);
  }

  /** Increment failure count. */
  public void failure() {
    failureCount.incrementAndGet();
  }
}

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.common;
19
20	import java.time.Duration;
21	import java.util.Objects;
22	import java.util.concurrent.atomic.AtomicLong;
23
24	/**
25	* Used to throttle code execution in presence of failures using exponential backoff logic. The
26	* formula used to calculate the next sleep interval is:
27	*
28	* <p>
29	*
30	* <pre>
31	* min(pow(backoffCoefficient, failureCount - 1) * initialSleep, maxSleep);
32	* </pre>
33	*
34	* <p>Example usage:
35	*
36	* <p>
37	*
38	* <pre>
39	* BackoffThrottler throttler = new BackoffThrottler(1000, 60000, 2);
40	* while(!stopped) {
41	* try {
42	* throttler.throttle();
43	* // some code that can fail and should be throttled
44	* ...
45	* throttler.success();
46	* }
47	* catch (Exception e) {
48	* throttler.failure();
49	* }
50	* }
51	* </pre>
52	*
53	* @author fateev
54	*/
55	public final class BackoffThrottler {
56
57	private final Duration initialSleep;
58
59	private final Duration maxSleep;
60
61	private final double backoffCoefficient;
62
63	private final AtomicLong failureCount = new AtomicLong();	1✔
64
65	/**
66	* Construct an instance of the throttler.
67	*
68	* @param initialSleep time to sleep on the first failure
69	* @param maxSleep maximum time to sleep independently of number of failures
70	* @param backoffCoefficient coefficient used to calculate the next time to sleep.
71	*/
72	public BackoffThrottler(Duration initialSleep, Duration maxSleep, double backoffCoefficient) {	1✔
73	Objects.requireNonNull(initialSleep, "initialSleep");	1✔
74	this.initialSleep = initialSleep;	1✔
75	this.maxSleep = maxSleep;	1✔
76	this.backoffCoefficient = backoffCoefficient;	1✔
77	}	1✔
78
79	private long calculateSleepTime() {
UNCOV 80	double sleepMillis =	×
UNCOV 81	Math.pow(backoffCoefficient, failureCount.get() - 1) * initialSleep.toMillis();	×
UNCOV 82	if (maxSleep != null) {	×
UNCOV 83	return Math.min((long) sleepMillis, maxSleep.toMillis());	×
84	}
85	return (long) sleepMillis;	×
86	}
87
88	/**
89	* Sleep if there were failures since the last success call.
90	*
91	* @throws InterruptedException
92	*/
93	public void throttle() throws InterruptedException {
94	if (failureCount.get() > 0) {	1✔
95	Thread.sleep(calculateSleepTime());	×
96	}
97	}	1✔
98
99	/** Resent failure count to 0. */
100	public void success() {
101	failureCount.set(0);	1✔
102	}	1✔
103
104	/** Increment failure count. */
105	public void failure() {
106	failureCount.incrementAndGet();	1✔
107	}	1✔
108	}

uber / cadence-java-client / 2416

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