Skip to content

Commit

Permalink
safety: add defensive copy for input array argument if it used asynch…
Browse files Browse the repository at this point in the history 
…ronously(e.g. use in `thenCompose`), not thread safe if it mutates 🧵
  • Loading branch information
@oldratlee
oldratlee committed Dec 13, 2024
1 parent 8199d0e commit 9d315fe
Showing 1 changed file with 31 additions and 28 deletions.
59 changes: 31 additions & 28 deletions cffu-core/src/main/java/io/foldright/cffu/CompletableFutureUtils.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1513,9 +1513,12 @@ public static <T, U> CompletableFuture<List<U>> thenMApplyFailFastAsync(
CompletableFuture<? extends T> cfThis, Executor executor, Function<? super T, ? extends U>... fns) {
requireNonNull(cfThis, "cfThis is null");
requireNonNull(executor, "executor is null");
requireArrayAndEleNonNull("fn", fns);
// Implementation Note: create defensive copy of input array argument
// since it is used asynchronously in `thenCompose` and could be mutated (not thread-safe).
// this same defensive copying pattern is used in similar methods below.
Function<? super T, ? extends U>[] copy = requireArrayAndEleNonNull("fn", fns).clone();

return cfThis.thenCompose(v -> allResultsOf0(true, wrapFunctions0(executor, v, fns)));
return cfThis.thenCompose(v -> allResultsOf0(true, wrapFunctions0(executor, v, copy)));
}

/**
Expand Down Expand Up @@ -1547,9 +1550,9 @@ public static <T, U> CompletableFuture<List<U>> thenMApplyAllSuccessAsync(
@Nullable U valueIfFailed, Function<? super T, ? extends U>... fns) {
requireNonNull(cfThis, "cfThis is null");
requireNonNull(executor, "executor is null");
requireArrayAndEleNonNull("fn", fns);
Function<? super T, ? extends U>[] copy = requireArrayAndEleNonNull("fn", fns);

return cfThis.thenCompose(v -> allSuccessResultsOf0(valueIfFailed, wrapFunctions0(executor, v, fns)));
return cfThis.thenCompose(v -> allSuccessResultsOf0(valueIfFailed, wrapFunctions0(executor, v, copy)));
}

/**
Expand Down Expand Up @@ -1582,10 +1585,10 @@ public static <T, U> CompletableFuture<List<U>> thenMApplyMostSuccessAsync(
requireNonNull(cfThis, "cfThis is null");
requireNonNull(executor, "executor is null");
requireNonNull(unit, "unit is null");
requireArrayAndEleNonNull("fn", fns);
Function<? super T, ? extends U>[] copy = requireArrayAndEleNonNull("fn", fns);

return cfThis.thenCompose(v -> mostSuccessResultsOf0(
executor, valueIfNotSuccess, timeout, unit, wrapFunctions0(executor, v, fns)));
executor, valueIfNotSuccess, timeout, unit, wrapFunctions0(executor, v, copy)));
}

/**
Expand Down Expand Up @@ -1616,9 +1619,9 @@ public static <T, U> CompletableFuture<List<U>> thenMApplyAsync(
CompletableFuture<? extends T> cfThis, Executor executor, Function<? super T, ? extends U>... fns) {
requireNonNull(cfThis, "cfThis is null");
requireNonNull(executor, "executor is null");
requireArrayAndEleNonNull("fn", fns);
Function<? super T, ? extends U>[] copy = requireArrayAndEleNonNull("fn", fns);

return cfThis.thenCompose(v -> allResultsOf0(false, wrapFunctions0(executor, v, fns)));
return cfThis.thenCompose(v -> allResultsOf0(false, wrapFunctions0(executor, v, copy)));
}

/**
Expand Down Expand Up @@ -1649,9 +1652,9 @@ public static <T, U> CompletableFuture<U> thenMApplyAnySuccessAsync(
CompletableFuture<? extends T> cfThis, Executor executor, Function<? super T, ? extends U>... fns) {
requireNonNull(cfThis, "cfThis is null");
requireNonNull(executor, "executor is null");
requireArrayAndEleNonNull("fn", fns);
Function<? super T, ? extends U>[] copy = requireArrayAndEleNonNull("fn", fns);

return cfThis.thenCompose(v -> anySuccessOf0(wrapFunctions0(executor, v, fns)));
return cfThis.thenCompose(v -> anySuccessOf0(wrapFunctions0(executor, v, copy)));
}

/**
Expand Down Expand Up @@ -1682,9 +1685,9 @@ public static <T, U> CompletableFuture<U> thenMApplyAnyAsync(
CompletableFuture<? extends T> cfThis, Executor executor, Function<? super T, ? extends U>... fns) {
requireNonNull(cfThis, "cfThis is null");
requireNonNull(executor, "executor is null");
requireArrayAndEleNonNull("fn", fns);
Function<? super T, ? extends U>[] copy = requireArrayAndEleNonNull("fn", fns);

return cfThis.thenCompose(v -> f_cast(CompletableFuture.anyOf(wrapFunctions0(executor, v, fns))));
return cfThis.thenCompose(v -> f_cast(CompletableFuture.anyOf(wrapFunctions0(executor, v, copy))));
}

private static <T, U> CompletableFuture<U>[] wrapFunctions0(
Expand Down Expand Up @@ -1725,9 +1728,9 @@ public static <T> CompletableFuture<Void> thenMAcceptFailFastAsync(
CompletableFuture<? extends T> cfThis, Executor executor, Consumer<? super T>... actions) {
requireNonNull(cfThis, "cfThis is null");
requireNonNull(executor, "executor is null");
requireArrayAndEleNonNull("action", actions);
Consumer<? super T>[] copy = requireArrayAndEleNonNull("action", actions);

return cfThis.thenCompose(v -> allFailFastOf0(wrapConsumers0(executor, v, actions)));
return cfThis.thenCompose(v -> allFailFastOf0(wrapConsumers0(executor, v, copy)));
}

/**
Expand Down Expand Up @@ -1763,9 +1766,9 @@ public static <T> CompletableFuture<Void> thenMAcceptAsync(
CompletableFuture<? extends T> cfThis, Executor executor, Consumer<? super T>... actions) {
requireNonNull(cfThis, "cfThis is null");
requireNonNull(executor, "executor is null");
requireArrayAndEleNonNull("action", actions);
Consumer<? super T>[] copy = requireArrayAndEleNonNull("action", actions);

return cfThis.thenCompose(v -> CompletableFuture.allOf(wrapConsumers0(executor, v, actions)));
return cfThis.thenCompose(v -> CompletableFuture.allOf(wrapConsumers0(executor, v, copy)));
}

/**
Expand Down Expand Up @@ -1801,9 +1804,9 @@ public static <T> CompletableFuture<Void> thenMAcceptAnySuccessAsync(
CompletableFuture<? extends T> cfThis, Executor executor, Consumer<? super T>... actions) {
requireNonNull(cfThis, "cfThis is null");
requireNonNull(executor, "executor is null");
requireArrayAndEleNonNull("action", actions);
Consumer<? super T>[] copy = requireArrayAndEleNonNull("action", actions);

return cfThis.thenCompose(v -> anySuccessOf0(wrapConsumers0(executor, v, actions)));
return cfThis.thenCompose(v -> anySuccessOf0(wrapConsumers0(executor, v, copy)));
}

/**
Expand Down Expand Up @@ -1839,9 +1842,9 @@ public static <T> CompletableFuture<Void> thenMAcceptAnyAsync(
CompletableFuture<? extends T> cfThis, Executor executor, Consumer<? super T>... actions) {
requireNonNull(cfThis, "cfThis is null");
requireNonNull(executor, "executor is null");
requireArrayAndEleNonNull("action", actions);
Consumer<? super T>[] copy = requireArrayAndEleNonNull("action", actions);

return cfThis.thenCompose(v -> f_cast(CompletableFuture.anyOf(wrapConsumers0(executor, v, actions))));
return cfThis.thenCompose(v -> f_cast(CompletableFuture.anyOf(wrapConsumers0(executor, v, copy))));
}

private static <T> CompletableFuture<Void>[] wrapConsumers0(Executor executor, T v, Consumer<? super T>[] actions) {
Expand All @@ -1868,9 +1871,9 @@ public static CompletableFuture<Void> thenMRunFailFastAsync(
CompletableFuture<?> cfThis, Executor executor, Runnable... actions) {
requireNonNull(cfThis, "cfThis is null");
requireNonNull(executor, "executor is null");
requireArrayAndEleNonNull("action", actions);
Runnable[] copy = requireArrayAndEleNonNull("action", actions);

return cfThis.thenCompose(unused -> allFailFastOf0(wrapRunnables0(executor, actions)));
return cfThis.thenCompose(unused -> allFailFastOf0(wrapRunnables0(executor, copy)));
}

/**
Expand All @@ -1893,9 +1896,9 @@ public static CompletableFuture<Void> thenMRunAsync(
CompletableFuture<?> cfThis, Executor executor, Runnable... actions) {
requireNonNull(cfThis, "cfThis is null");
requireNonNull(executor, "executor is null");
requireArrayAndEleNonNull("action", actions);
Runnable[] copy = requireArrayAndEleNonNull("action", actions);

return cfThis.thenCompose(unused -> CompletableFuture.allOf(wrapRunnables0(executor, actions)));
return cfThis.thenCompose(unused -> CompletableFuture.allOf(wrapRunnables0(executor, copy)));
}

/**
Expand All @@ -1918,9 +1921,9 @@ public static CompletableFuture<Void> thenMRunAnySuccessAsync(
CompletableFuture<?> cfThis, Executor executor, Runnable... actions) {
requireNonNull(cfThis, "cfThis is null");
requireNonNull(executor, "executor is null");
requireArrayAndEleNonNull("action", actions);
Runnable[] copy = requireArrayAndEleNonNull("action", actions);

return cfThis.thenCompose(unused -> anySuccessOf0(wrapRunnables0(executor, actions)));
return cfThis.thenCompose(unused -> anySuccessOf0(wrapRunnables0(executor, copy)));
}

/**
Expand All @@ -1943,9 +1946,9 @@ public static CompletableFuture<Void> thenMRunAnyAsync(
CompletableFuture<?> cfThis, Executor executor, Runnable... actions) {
requireNonNull(cfThis, "cfThis is null");
requireNonNull(executor, "executor is null");
requireArrayAndEleNonNull("action", actions);
Runnable[] copy = requireArrayAndEleNonNull("action", actions);

return cfThis.thenCompose(unused -> f_cast(CompletableFuture.anyOf(wrapRunnables0(executor, actions))));
return cfThis.thenCompose(unused -> f_cast(CompletableFuture.anyOf(wrapRunnables0(executor, copy))));
}

// endregion
Expand Down

0 comments on commit 9d315fe

Please sign in to comment.