Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -292,6 +292,9 @@ Integer getVersion(
/** Replay safe random. */
Random newRandom();

/** Replay safe named random stream. */
Random getRandomStream(String name);

/**
* @return scope to be used for metrics reporting.
*/
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,11 @@ public Random newRandom() {
return workflowStateMachines.newRandom();
}

@Override
public Random getRandomStream(String name) {
return workflowStateMachines.getRandomStream(name);
}

@Override
public Scope getMetricsScope() {
return replayAwareWorkflowMetricsScope;
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
package io.temporal.internal.statemachines;

import com.google.common.hash.HashCode;
import com.google.common.hash.Hashing;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import javax.annotation.Nonnull;

final class WorkflowRandomStreams {
private static final String SEED_VERSION = "temporal.sdk.random.v1";

private final Map<String, Random> streams = new HashMap<>();

long deriveSeed(@Nonnull String runId, @Nonnull String name) {
// The separators keep ("ab", "c") from colliding with ("a", "bc")
String seed = String.join("\0", SEED_VERSION, runId, name);
HashCode hash = Hashing.sha256().hashString(seed, StandardCharsets.UTF_8);
return ByteBuffer.wrap(hash.asBytes()).getLong();
}

Random get(@Nonnull String runId, @Nonnull String name) {
return streams.computeIfAbsent(name, key -> new Random(deriveSeed(runId, key)));
}

void reseed(@Nonnull String runId) {
streams.forEach((name, stream) -> stream.setSeed(deriveSeed(runId, name)));
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,8 @@ enum HandleEventStatus {
/** Used Workflow.newRandom and randomUUID together with currentRunId. */
private long idCounter;

private final WorkflowRandomStreams randomStreams = new WorkflowRandomStreams();

/** Current workflow time. */
private long currentTimeMillis = -1;

Expand Down Expand Up @@ -1195,6 +1197,11 @@ public Random newRandom() {
return new Random(randomUUID().getLeastSignificantBits());
}

public Random getRandomStream(String name) {
checkEventLoopExecuting();
return randomStreams.get(currentRunId, name);
}

public void sideEffect(
Functions.Func<Optional<Payloads>> func,
UserMetadata userMetadata,
Expand Down Expand Up @@ -1548,6 +1555,7 @@ public void workflowTaskStarted(
@Override
public void updateRunId(String currentRunId) {
WorkflowStateMachines.this.currentRunId = currentRunId;
randomStreams.reseed(currentRunId);
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -744,6 +744,10 @@ public static Random newRandom() {
return getRootWorkflowContext().newRandom();
}

public static Random getRandomStream(String name) {
return getRootWorkflowContext().getReplayContext().getRandomStream(name);
}

public static Logger getLogger(Class<?> clazz) {
Logger logger = LoggerFactory.getLogger(clazz);
return new ReplayAwareLogger(
Expand Down Expand Up @@ -919,11 +923,11 @@ static SyncWorkflowContext getRootWorkflowContext() {
return DeterministicRunnerImpl.currentThreadInternal().getWorkflowContext();
}

static boolean isReadOnly() {
public static boolean isReadOnly() {
return getRootWorkflowContext().isReadOnly();
}

static void assertNotReadOnly(String action) {
public static void assertNotReadOnly(String action) {
if (isReadOnly()) {
throw new ReadOnlyException(action);
}
Expand Down
18 changes: 18 additions & 0 deletions temporal-sdk/src/main/java/io/temporal/workflow/Workflow.java
Original file line number Diff line number Diff line change
Expand Up @@ -711,6 +711,24 @@ public static Random newRandom() {
return WorkflowInternal.newRandom();
}

/**
* Returns a deterministic pseudorandom stream private to {@code name}.
*
* <p>Calling this method again with the same name returns the same logical stream where earlier
* draws left it. A Workflow Reset replays the same values up to the reset point, then reseeds the
* stream for the new Run. Each Continue-As-New Run gets a new sequence.
*
* <p>Draws are not recorded in Workflow History, so do not draw in read-only code. Use {@link
* WorkflowUnsafe#isReadOnly()} to gate draws.
*
* <p>Use a stable package-style name. Stream names are retained for the life of the Workflow Run.
* The stream is deterministic pseudorandomness and is not cryptographically secure.
*/
@Experimental
public static Random getRandomStream(String name) {
return WorkflowInternal.getRandomStream(name);
}

/**
* True if workflow code is being replayed.
*
Expand Down
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
package io.temporal.workflow.unsafe;

import io.temporal.common.Experimental;
import io.temporal.internal.sync.WorkflowInternal;
import io.temporal.workflow.Functions;

Expand Down Expand Up @@ -46,6 +47,21 @@ public static boolean isReplaying() {
return WorkflowInternal.isReplaying();
}

/**
* Reports whether the current code is running where Workflow state cannot be mutated.
*
* <p>Read-only code includes Query handlers, Update validators, Side Effect functions, Await
* conditions, and other SDK callbacks that must not mutate Workflow state.
*
* <p>Must be called from Workflow code.
*
* @return true in a read-only Workflow context
*/
@Experimental
public static boolean isReadOnly() {
return WorkflowInternal.isReadOnly();
}

/**
* Runs the supplied procedure in the calling thread with disabled deadlock detection if called
* from the workflow thread. Does nothing except the procedure execution if called from a
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,102 @@
package io.temporal.internal.statemachines;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertSame;

import com.google.common.io.BaseEncoding;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import org.junit.Test;

public class WorkflowRandomStreamsTest {
private static final String RUN_ID = "runID";
private static final String NAME = "io.temporal.test";

@Test
public void deriveSeed() {
WorkflowRandomStreams randoms = new WorkflowRandomStreams();
long seed = randoms.deriveSeed(RUN_ID, NAME);
assertNotEquals(seed, randoms.deriveSeed("other", NAME));
assertNotEquals(seed, randoms.deriveSeed(RUN_ID, "other"));
assertNotEquals(seed, randoms.deriveSeed("other", "other"));
}

/** Pins the seed derivation and resulting byte stream. Changing either breaks replay. */
@Test
public void getRandomStreamGolden() {
WorkflowRandomStreams randoms = new WorkflowRandomStreams();
assertEquals(8181915698088084985L, randoms.deriveSeed(RUN_ID, NAME));

byte[] bytes = new byte[32];
randoms.get(RUN_ID, NAME).nextBytes(bytes);
assertEquals(
"1c1d4dd36999ff851d72aa41f660ecd3220d83499109ba3ba24e455a1b776c21",
BaseEncoding.base16().lowerCase().encode(bytes));
}

@Test
public void deriveSeedSeparators() {
WorkflowRandomStreams randoms = new WorkflowRandomStreams();
assertNotEquals(randoms.deriveSeed("ab", "c"), randoms.deriveSeed("a", "bc"));
}

/** A second lookup under the same name continues the sequence rather than restarting it. */
@Test
public void getRandomStreamMemoizes() {
WorkflowRandomStreams randoms = new WorkflowRandomStreams();

Random first = randoms.get(RUN_ID, NAME);
long firstDraw = first.nextLong();

Random second = randoms.get(RUN_ID, NAME);
long secondDraw = second.nextLong();

assertSame(first, second);
assertNotEquals(firstDraw, secondDraw);
}

/**
* Interleaving draws across two names yields the same sequence per name as drawing from each on
* its own, so how often a workflow draws from one name cannot shift another.
*/
@Test
public void getRandomStreamNamesAreIndependent() {
WorkflowRandomStreams randoms = new WorkflowRandomStreams();
Random first = randoms.get(RUN_ID, NAME);
Random second = randoms.get(RUN_ID, "other");

List<Long> interleavedA = new ArrayList<>();
List<Long> interleavedB = new ArrayList<>();
for (int i = 0; i < 3; i++) {
interleavedA.add(first.nextLong());
interleavedB.add(second.nextLong());
}

assertEquals(solo(NAME, 3), interleavedA);
assertEquals(solo("other", 3), interleavedB);
assertNotEquals(interleavedA, interleavedB);
}

@Test
public void reseedRandomsInPlace() {
WorkflowRandomStreams randoms = new WorkflowRandomStreams();

Random first = randoms.get(RUN_ID, NAME);
randoms.reseed("other");
Random second = randoms.get("other", NAME);

assertSame(first, second);
assertEquals(new WorkflowRandomStreams().get("other", NAME).nextLong(), second.nextLong());
}

private static List<Long> solo(String name, int draws) {
Random random = new WorkflowRandomStreams().get(RUN_ID, name);
List<Long> result = new ArrayList<>();
for (int i = 0; i < draws; i++) {
result.add(random.nextLong());
}
return result;
}
}
Loading
Loading