You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@commons.apache.org by gg...@apache.org on 2025/10/17 13:24:01 UTC
(commons-io) branch master updated: Fix concurrency issue in `IOUtils.skip` (#801)
This is an automated email from the ASF dual-hosted git repository.
ggregory pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/commons-io.git
The following commit(s) were added to refs/heads/master by this push:
new f76ee129d Fix concurrency issue in `IOUtils.skip` (#801)
f76ee129d is described below
commit f76ee129d348f04c540d205856853e587ae22766
Author: Piotr P. Karwasz <pk...@apache.org>
AuthorDate: Fri Oct 17 15:23:57 2025 +0200
Fix concurrency issue in `IOUtils.skip` (#801)
* Fix concurrency issue in `IOUtils.skip`
This patch addresses a concurrency problem in `IOUtils.skip`, as reported in [COMPRESS-666](https://issues.apache.org/jira/browse/COMPRESS-666) and [COMPRESS-697](https://issues.apache.org/jira/browse/COMPRESS-697).
Previously, `IOUtils.skip` relied on `InputStream#read` to skip bytes, using a buffer shared across **all** threads. Although `IOUtils.skip` itself does not consume the data read, certain `InputStream` implementations (e.g. `ChecksumInputStream`) may process that data internally.
In concurrent scenarios, this shared buffer could be overwritten by another thread between the `read` and the subsequent internal processing (such as checksum calculation), leading to incorrect behavior.
This change reverts commit c12eaff7f747353a7a9a97df735fd3301f42e313 and restores the use of a **per-thread buffer** in `IOUtils.skip`, ensuring thread safety and correct behavior in concurrent environments.
* Adds a reentrancy guard to the thread-local pool
* Apply suggestion from @Copilot (1)
Co-authored-by: Copilot <17...@users.noreply.github.com>
* Apply suggestions from @Copilot (2)
Co-authored-by: Copilot <17...@users.noreply.github.com>
---------
Co-authored-by: Copilot <17...@users.noreply.github.com>
---
src/changes/changes.xml | 1 +
src/main/java/org/apache/commons/io/CopyUtils.java | 18 +-
src/main/java/org/apache/commons/io/IOUtils.java | 273 ++++++++++++---------
.../java/org/apache/commons/io/IOCaseTest.java | 65 +++--
.../apache/commons/io/IOUtilsConcurrentTest.java | 212 ++++++++++++++++
5 files changed, 419 insertions(+), 150 deletions(-)
diff --git a/src/changes/changes.xml b/src/changes/changes.xml
index ecee3aeca..3aa419401 100644
--- a/src/changes/changes.xml
+++ b/src/changes/changes.xml
@@ -56,6 +56,7 @@ The <action> type attribute can be add,update,fix,remove.
<action type="fix" dev="pkarwasz" due-to="Piotr P. Karwasz">IOUtils.toByteArray(InputStream) now throws IOException on byte array overflow.</action>
<action type="fix" dev="ggregory" due-to="Gary Gregory, Piotr P. Karwasz">Javadoc general improvements.</action>
<action type="fix" dev="ggregory" due-to="Piotr P. Karwasz">IOUtils.toByteArray() now throws EOFException when not enough data is available #796.</action>
+ <action type="fix" dev="pkarwasz" due-to="Piotr P. Karwasz">Fix IOUtils.skip() usage in concurrent scenarios.</action>
<!-- ADD -->
<action dev="ggregory" type="add" due-to="strangelookingnerd, Gary Gregory">FileUtils#byteCountToDisplaySize() supports Zettabyte, Yottabyte, Ronnabyte and Quettabyte #763.</action>
<action dev="ggregory" type="add" due-to="strangelookingnerd, Gary Gregory">Add org.apache.commons.io.FileUtils.ONE_RB #763.</action>
diff --git a/src/main/java/org/apache/commons/io/CopyUtils.java b/src/main/java/org/apache/commons/io/CopyUtils.java
index e22adef83..fdb137d20 100644
--- a/src/main/java/org/apache/commons/io/CopyUtils.java
+++ b/src/main/java/org/apache/commons/io/CopyUtils.java
@@ -278,14 +278,18 @@ public static int copy(
final Reader input,
final Writer output)
throws IOException {
- final char[] buffer = IOUtils.getScratchCharArray();
- int count = 0;
- int n;
- while (EOF != (n = input.read(buffer))) {
- output.write(buffer, 0, n);
- count += n;
+ final char[] buffer = IOUtils.ScratchBufferHolder.getScratchCharArray();
+ try {
+ int count = 0;
+ int n;
+ while (EOF != (n = input.read(buffer))) {
+ output.write(buffer, 0, n);
+ count += n;
+ }
+ return count;
+ } finally {
+ IOUtils.ScratchBufferHolder.releaseScratchCharArray(buffer);
}
- return count;
}
/**
diff --git a/src/main/java/org/apache/commons/io/IOUtils.java b/src/main/java/org/apache/commons/io/IOUtils.java
index a1f9b0709..717be4b49 100644
--- a/src/main/java/org/apache/commons/io/IOUtils.java
+++ b/src/main/java/org/apache/commons/io/IOUtils.java
@@ -132,6 +132,94 @@ public class IOUtils {
// Writer. Each method should take at least one of these as a parameter,
// or return one of them.
+ /**
+ * Holder for per-thread internal scratch buffers.
+ *
+ * <p>Buffers are created lazily and reused within the same thread to reduce allocation overhead. In the rare case of reentrant access, a temporary buffer
+ * is allocated to avoid data corruption.</p>
+ *
+ * <p>Typical usage:</p>
+ *
+ * <pre>{@code
+ * final byte[] buffer = ScratchBufferHolder.getScratchByteArray();
+ * try {
+ * // use the buffer
+ * } finally {
+ * ScratchBufferHolder.releaseScratchByteArray(buffer);
+ * }
+ * }</pre>
+ */
+ static final class ScratchBufferHolder {
+
+ /**
+ * Holder for internal byte array buffer.
+ */
+ private static final ThreadLocal<Object[]> SCRATCH_BYTE_BUFFER_HOLDER = ThreadLocal.withInitial(() -> new Object[] { false, byteArray() });
+
+ /**
+ * Holder for internal char array buffer.
+ */
+ private static final ThreadLocal<Object[]> SCRATCH_CHAR_BUFFER_HOLDER = ThreadLocal.withInitial(() -> new Object[] { false, charArray() });
+
+
+ /**
+ * Gets the internal byte array buffer.
+ *
+ * @return the internal byte array buffer.
+ */
+ static byte[] getScratchByteArray() {
+ final Object[] holder = SCRATCH_BYTE_BUFFER_HOLDER.get();
+ // If already in use, return a new array
+ if ((boolean) holder[0]) {
+ return byteArray();
+ }
+ holder[0] = true;
+ return (byte[]) holder[1];
+ }
+
+ /**
+ * Gets the char array buffer.
+ *
+ * @return the char array buffer.
+ */
+ static char[] getScratchCharArray() {
+ final Object[] holder = SCRATCH_CHAR_BUFFER_HOLDER.get();
+ // If already in use, return a new array
+ if ((boolean) holder[0]) {
+ return charArray();
+ }
+ holder[0] = true;
+ return (char[]) holder[1];
+ }
+
+
+ /**
+ * If the argument is the internal byte array, release it for reuse.
+ *
+ * @param array the byte array to release.
+ */
+ static void releaseScratchByteArray(byte[] array) {
+ final Object[] holder = SCRATCH_BYTE_BUFFER_HOLDER.get();
+ if (array == holder[1]) {
+ Arrays.fill(array, (byte) 0);
+ holder[0] = false;
+ }
+ }
+
+ /**
+ * If the argument is the internal char array, release it for reuse.
+ *
+ * @param array the char array to release.
+ */
+ static void releaseScratchCharArray(char[] array) {
+ final Object[] holder = SCRATCH_CHAR_BUFFER_HOLDER.get();
+ if (array == holder[1]) {
+ Arrays.fill(array, (char) 0);
+ holder[0] = false;
+ }
+ }
+ }
+
/**
* CR char '{@value}'.
*
@@ -201,26 +289,6 @@ public class IOUtils {
*/
public static final String LINE_SEPARATOR_WINDOWS = StandardLineSeparator.CRLF.getString();
- /**
- * Internal byte array buffer, intended for both reading and writing.
- */
- private static final ThreadLocal<byte[]> SCRATCH_BYTE_BUFFER_RW = ThreadLocal.withInitial(IOUtils::byteArray);
-
- /**
- * Internal byte array buffer, intended for write only operations.
- */
- private static final byte[] SCRATCH_BYTE_BUFFER_WO = byteArray();
-
- /**
- * Internal char array buffer, intended for both reading and writing.
- */
- private static final ThreadLocal<char[]> SCRATCH_CHAR_BUFFER_RW = ThreadLocal.withInitial(IOUtils::charArray);
-
- /**
- * Internal char array buffer, intended for write only operations.
- */
- private static final char[] SCRATCH_CHAR_BUFFER_WO = charArray();
-
/**
* The maximum size of an array in many Java VMs.
* <p>
@@ -592,10 +660,8 @@ static void checkFromToIndex(final int fromIndex, final int toIndex, final int l
* @see IO#clear()
*/
static void clear() {
- SCRATCH_BYTE_BUFFER_RW.remove();
- SCRATCH_CHAR_BUFFER_RW.remove();
- Arrays.fill(SCRATCH_BYTE_BUFFER_WO, (byte) 0);
- Arrays.fill(SCRATCH_CHAR_BUFFER_WO, (char) 0);
+ ScratchBufferHolder.SCRATCH_BYTE_BUFFER_HOLDER.remove();
+ ScratchBufferHolder.SCRATCH_CHAR_BUFFER_HOLDER.remove();
}
/**
@@ -1139,39 +1205,43 @@ public static boolean contentEquals(final Reader input1, final Reader input2) th
}
// reuse one
- final char[] array1 = getScratchCharArray();
+ final char[] array1 = ScratchBufferHolder.getScratchCharArray();
// but allocate another
final char[] array2 = charArray();
int pos1;
int pos2;
int count1;
int count2;
- while (true) {
- pos1 = 0;
- pos2 = 0;
- for (int index = 0; index < DEFAULT_BUFFER_SIZE; index++) {
- if (pos1 == index) {
- do {
- count1 = input1.read(array1, pos1, DEFAULT_BUFFER_SIZE - pos1);
- } while (count1 == 0);
- if (count1 == EOF) {
- return pos2 == index && input2.read() == EOF;
+ try {
+ while (true) {
+ pos1 = 0;
+ pos2 = 0;
+ for (int index = 0; index < DEFAULT_BUFFER_SIZE; index++) {
+ if (pos1 == index) {
+ do {
+ count1 = input1.read(array1, pos1, DEFAULT_BUFFER_SIZE - pos1);
+ } while (count1 == 0);
+ if (count1 == EOF) {
+ return pos2 == index && input2.read() == EOF;
+ }
+ pos1 += count1;
}
- pos1 += count1;
- }
- if (pos2 == index) {
- do {
- count2 = input2.read(array2, pos2, DEFAULT_BUFFER_SIZE - pos2);
- } while (count2 == 0);
- if (count2 == EOF) {
- return pos1 == index && input1.read() == EOF;
+ if (pos2 == index) {
+ do {
+ count2 = input2.read(array2, pos2, DEFAULT_BUFFER_SIZE - pos2);
+ } while (count2 == 0);
+ if (count2 == EOF) {
+ return pos1 == index && input1.read() == EOF;
+ }
+ pos2 += count2;
+ }
+ if (array1[index] != array2[index]) {
+ return false;
}
- pos2 += count2;
- }
- if (array1[index] != array2[index]) {
- return false;
}
}
+ } finally {
+ ScratchBufferHolder.releaseScratchCharArray(array1);
}
}
@@ -1651,9 +1721,13 @@ public static long copyLarge(final InputStream inputStream, final OutputStream o
* @throws IOException if an I/O error occurs.
* @since 2.2
*/
- public static long copyLarge(final InputStream input, final OutputStream output, final long inputOffset,
- final long length) throws IOException {
- return copyLarge(input, output, inputOffset, length, getScratchByteArray());
+ public static long copyLarge(final InputStream input, final OutputStream output, final long inputOffset, final long length) throws IOException {
+ final byte[] buffer = ScratchBufferHolder.getScratchByteArray();
+ try {
+ return copyLarge(input, output, inputOffset, length, buffer);
+ } finally {
+ ScratchBufferHolder.releaseScratchByteArray(buffer);
+ }
}
/**
@@ -1723,7 +1797,12 @@ public static long copyLarge(final InputStream input, final OutputStream output,
* @since 1.3
*/
public static long copyLarge(final Reader reader, final Writer writer) throws IOException {
- return copyLarge(reader, writer, getScratchCharArray());
+ final char[] buffer = ScratchBufferHolder.getScratchCharArray();
+ try {
+ return copyLarge(reader, writer, buffer);
+ } finally {
+ ScratchBufferHolder.releaseScratchCharArray(buffer);
+ }
}
/**
@@ -1770,7 +1849,12 @@ public static long copyLarge(final Reader reader, final Writer writer, final cha
* @since 2.2
*/
public static long copyLarge(final Reader reader, final Writer writer, final long inputOffset, final long length) throws IOException {
- return copyLarge(reader, writer, inputOffset, length, getScratchCharArray());
+ final char[] buffer = ScratchBufferHolder.getScratchCharArray();
+ try {
+ return copyLarge(reader, writer, inputOffset, length, buffer);
+ } finally {
+ ScratchBufferHolder.releaseScratchCharArray(buffer);
+ }
}
/**
@@ -1837,64 +1921,6 @@ static UnsynchronizedByteArrayOutputStream copyToOutputStream(
}
}
- /**
- * Fills the given array with 0s.
- *
- * @param arr The non-null array to fill.
- * @return The given array.
- */
- private static byte[] fill0(final byte[] arr) {
- Arrays.fill(arr, (byte) 0);
- return arr;
- }
-
- /**
- * Fills the given array with 0s.
- *
- * @param arr The non-null array to fill.
- * @return The given array.
- */
- private static char[] fill0(final char[] arr) {
- Arrays.fill(arr, (char) 0);
- return arr;
- }
-
- /**
- * Gets the internal byte array buffer, intended for both reading and writing.
- *
- * @return the internal byte array buffer, intended for both reading and writing.
- */
- static byte[] getScratchByteArray() {
- return fill0(SCRATCH_BYTE_BUFFER_RW.get());
- }
-
- /**
- * Gets the internal byte array intended for write only operations.
- *
- * @return the internal byte array intended for write only operations.
- */
- static byte[] getScratchByteArrayWriteOnly() {
- return fill0(SCRATCH_BYTE_BUFFER_WO);
- }
-
- /**
- * Gets the char byte array buffer, intended for both reading and writing.
- *
- * @return the char byte array buffer, intended for both reading and writing.
- */
- static char[] getScratchCharArray() {
- return fill0(SCRATCH_CHAR_BUFFER_RW.get());
- }
-
- /**
- * Gets the internal char array intended for write only operations.
- *
- * @return the internal char array intended for write only operations.
- */
- static char[] getScratchCharArrayWriteOnly() {
- return fill0(SCRATCH_CHAR_BUFFER_WO);
- }
-
/**
* Returns the length of the given array in a null-safe manner.
*
@@ -2527,7 +2553,12 @@ public static URL resourceToURL(final String name, final ClassLoader classLoader
* @since 2.0
*/
public static long skip(final InputStream input, final long skip) throws IOException {
- return skip(input, skip, IOUtils::getScratchByteArrayWriteOnly);
+ final byte[] buffer = ScratchBufferHolder.getScratchByteArray();
+ try {
+ return skip(input, skip, () -> buffer);
+ } finally {
+ ScratchBufferHolder.releaseScratchByteArray(buffer);
+ }
}
/**
@@ -2634,14 +2665,18 @@ public static long skip(final Reader reader, final long toSkip) throws IOExcepti
throw new IllegalArgumentException("Skip count must be non-negative, actual: " + toSkip);
}
long remain = toSkip;
- while (remain > 0) {
- // See https://issues.apache.org/jira/browse/IO-203 for why we use read() rather than delegating to skip()
- final char[] charArray = getScratchCharArrayWriteOnly();
- final long n = reader.read(charArray, 0, (int) Math.min(remain, charArray.length));
- if (n < 0) { // EOF
- break;
+ final char[] charArray = ScratchBufferHolder.getScratchCharArray();
+ try {
+ while (remain > 0) {
+ // See https://issues.apache.org/jira/browse/IO-203 for why we use read() rather than delegating to skip()
+ final long n = reader.read(charArray, 0, (int) Math.min(remain, charArray.length));
+ if (n < 0) { // EOF
+ break;
+ }
+ remain -= n;
}
- remain -= n;
+ } finally {
+ ScratchBufferHolder.releaseScratchCharArray(charArray);
}
return toSkip - remain;
}
@@ -2667,7 +2702,7 @@ public static long skip(final Reader reader, final long toSkip) throws IOExcepti
* @since 2.0
*/
public static void skipFully(final InputStream input, final long toSkip) throws IOException {
- final long skipped = skip(input, toSkip, IOUtils::getScratchByteArrayWriteOnly);
+ final long skipped = skip(input, toSkip);
if (skipped != toSkip) {
throw new EOFException("Bytes to skip: " + toSkip + " actual: " + skipped);
}
diff --git a/src/test/java/org/apache/commons/io/IOCaseTest.java b/src/test/java/org/apache/commons/io/IOCaseTest.java
index 97a990f63..1121b38ee 100644
--- a/src/test/java/org/apache/commons/io/IOCaseTest.java
+++ b/src/test/java/org/apache/commons/io/IOCaseTest.java
@@ -18,6 +18,7 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -296,34 +297,50 @@ void test_getName() {
@Test
void test_getScratchByteArray() {
- final byte[] array = IOUtils.getScratchByteArray();
- assert0(array);
- Arrays.fill(array, (byte) 1);
- assert0(IOUtils.getScratchCharArray());
- }
-
- @Test
- void test_getScratchByteArrayWriteOnly() {
- final byte[] array = IOUtils.getScratchByteArrayWriteOnly();
- assert0(array);
- Arrays.fill(array, (byte) 1);
- assert0(IOUtils.getScratchCharArray());
+ final byte[] array = IOUtils.ScratchBufferHolder.getScratchByteArray();
+ try {
+ assert0(array);
+ Arrays.fill(array, (byte) 1);
+ // Get another array, while the first is still in use
+ final byte[] array2 = IOUtils.ScratchBufferHolder.getScratchByteArray();
+ assert0(array2);
+ assertNotSame(array, array2);
+ } finally {
+ // Release first array
+ IOUtils.ScratchBufferHolder.releaseScratchByteArray(array);
+ }
+ // The first array should be reset and reusable
+ final byte[] array3 = IOUtils.ScratchBufferHolder.getScratchByteArray();
+ try {
+ assert0(array3);
+ assertSame(array, array3);
+ } finally {
+ IOUtils.ScratchBufferHolder.releaseScratchByteArray(array3);
+ }
}
@Test
void test_getScratchCharArray() {
- final char[] array = IOUtils.getScratchCharArray();
- assert0(array);
- Arrays.fill(array, (char) 1);
- assert0(IOUtils.getScratchCharArray());
- }
-
- @Test
- void test_getScratchCharArrayWriteOnly() {
- final char[] array = IOUtils.getScratchCharArrayWriteOnly();
- assert0(array);
- Arrays.fill(array, (char) 1);
- assert0(IOUtils.getScratchCharArray());
+ final char[] array = IOUtils.ScratchBufferHolder.getScratchCharArray();
+ try {
+ assert0(array);
+ Arrays.fill(array, (char) 1);
+ // Get another array, while the first is still in use
+ final char[] array2 = IOUtils.ScratchBufferHolder.getScratchCharArray();
+ assert0(array2);
+ assertNotSame(array, array2);
+ } finally {
+ // Release first array
+ IOUtils.ScratchBufferHolder.releaseScratchCharArray(array);
+ }
+ // The first array should be reset and reusable
+ final char[] array3 = IOUtils.ScratchBufferHolder.getScratchCharArray();
+ try {
+ assert0(array3);
+ assertSame(array, array3);
+ } finally {
+ IOUtils.ScratchBufferHolder.releaseScratchCharArray(array3);
+ }
}
@Test
diff --git a/src/test/java/org/apache/commons/io/IOUtilsConcurrentTest.java b/src/test/java/org/apache/commons/io/IOUtilsConcurrentTest.java
new file mode 100644
index 000000000..0e1fb0571
--- /dev/null
+++ b/src/test/java/org/apache/commons/io/IOUtilsConcurrentTest.java
@@ -0,0 +1,212 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 org.apache.commons.io;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.Reader;
+import java.io.StringReader;
+import java.nio.charset.Charset;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import java.util.zip.CRC32;
+import java.util.zip.Checksum;
+
+import org.apache.commons.io.function.IOConsumer;
+import org.apache.commons.io.input.ChecksumInputStream;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+/**
+ * Tests {@link IOUtils} methods in a concurrent environment.
+ */
+class IOUtilsConcurrentTest {
+
+ private static class ChecksumReader extends Reader {
+ private final CRC32 checksum;
+ private final long expectedChecksumValue;
+ private final Reader reader;
+
+ ChecksumReader(Reader reader, long expectedChecksumValue) {
+ this.reader = reader;
+ this.checksum = new CRC32();
+ this.expectedChecksumValue = expectedChecksumValue;
+ }
+
+ @Override
+ public void close() throws IOException {
+ reader.close();
+ }
+
+ public long getValue() {
+ return checksum.getValue();
+ }
+
+ @Override
+ public int read() throws IOException {
+ return super.read();
+ }
+
+ @Override
+ public int read(char[] cbuf, int off, int len) throws IOException {
+ final int n = reader.read(cbuf, off, len);
+ if (n > 0) {
+ final byte[] bytes = new String(cbuf, off, n).getBytes(Charset.defaultCharset());
+ checksum.update(bytes, 0, bytes.length);
+ }
+ if (n == -1) {
+ final long actual = checksum.getValue();
+ if (actual != expectedChecksumValue) {
+ throw new IOException("Checksum mismatch: expected " + expectedChecksumValue + " but got " + actual);
+ }
+ }
+ return n;
+ }
+ }
+
+ /**
+ * Test data for InputStream tests.
+ */
+ private static final byte[][] BYTE_DATA;
+ /**
+ * Checksum values for {@link #BYTE_DATA}.
+ */
+ private static final long[] BYTE_DATA_CHECKSUM;
+ /**
+ * Number of runs per thread (to increase the chance of collisions).
+ */
+ private static final int RUNS_PER_THREAD = 16;
+ /**
+ * Size of test data.
+ */
+ private static final int SIZE = IOUtils.DEFAULT_BUFFER_SIZE;
+ /**
+ * Test data for Reader tests.
+ */
+ private static final String[] STRING_DATA;
+ /**
+ * Checksum values for {@link #STRING_DATA}.
+ */
+ private static final long[] STRING_DATA_CHECKSUM;
+ /**
+ * Number of threads to use.
+ */
+ private static final int THREAD_COUNT = 16;
+ /**
+ * Number of data variants (to increase the chance of collisions).
+ */
+ private static final int VARIANTS = 16;
+
+ static {
+ final Checksum checksum = new CRC32();
+ // Byte data
+ BYTE_DATA = new byte[VARIANTS][];
+ BYTE_DATA_CHECKSUM = new long[VARIANTS];
+ for (int variant = 0; variant < VARIANTS; variant++) {
+ final byte[] data = new byte[SIZE];
+ for (int i = 0; i < SIZE; i++) {
+ data[i] = (byte) ((i + variant) % 256);
+ }
+ BYTE_DATA[variant] = data;
+ checksum.reset();
+ checksum.update(data, 0 , data.length);
+ BYTE_DATA_CHECKSUM[variant] = checksum.getValue();
+ }
+ // Char data
+ final char[] cdata = new char[SIZE];
+ STRING_DATA = new String[VARIANTS];
+ STRING_DATA_CHECKSUM = new long[VARIANTS];
+ for (int variant = 0; variant < VARIANTS; variant++) {
+ for (int i = 0; i < SIZE; i++) {
+ cdata[i] = (char) ((i + variant) % Character.MAX_VALUE);
+ }
+ STRING_DATA[variant] = new String(cdata);
+ checksum.reset();
+ final byte[] bytes = STRING_DATA[variant].getBytes(Charset.defaultCharset());
+ checksum.update(bytes, 0, bytes.length);
+ STRING_DATA_CHECKSUM[variant] = checksum.getValue();
+ }
+ }
+
+ static Stream<IOConsumer<InputStream>> testConcurrentInputStreamTasks() {
+ return Stream.of(
+ IOUtils::consume,
+ in -> IOUtils.skip(in, Long.MAX_VALUE),
+ in -> IOUtils.skipFully(in, SIZE),
+ IOUtils::toByteArray,
+ in -> IOUtils.toByteArray(in, SIZE),
+ in -> IOUtils.toByteArray(in, SIZE, 512)
+ );
+ }
+
+ static Stream<IOConsumer<Reader>> testConcurrentReaderTasks() {
+ return Stream.of(
+ IOUtils::consume,
+ reader -> IOUtils.skip(reader, Long.MAX_VALUE),
+ reader -> IOUtils.skipFully(reader, SIZE),
+ reader -> IOUtils.toByteArray(reader, Charset.defaultCharset())
+ );
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void testConcurrentInputStreamTasks(IOConsumer<InputStream> consumer) throws InterruptedException {
+ final ExecutorService threadPool = Executors.newFixedThreadPool(THREAD_COUNT);
+ try {
+ final List<Future<Void>> futures = IntStream.range(0, THREAD_COUNT * RUNS_PER_THREAD)
+ .<Future<Void>>mapToObj(i -> threadPool.submit(() -> {
+ try (InputStream in = ChecksumInputStream
+ .builder()
+ .setByteArray(BYTE_DATA[i % VARIANTS])
+ .setChecksum(new CRC32())
+ .setExpectedChecksumValue(BYTE_DATA_CHECKSUM[i % VARIANTS])
+ .get()) {
+ consumer.accept(in);
+ }
+ return null;
+ })).collect(Collectors.toList());
+ futures.forEach(f -> assertDoesNotThrow(() -> f.get()));
+ } finally {
+ threadPool.shutdownNow();
+ }
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void testConcurrentReaderTasks(IOConsumer<Reader> consumer) throws InterruptedException {
+ final ExecutorService threadPool = Executors.newFixedThreadPool(THREAD_COUNT);
+ try {
+ final List<Future<Void>> futures = IntStream.range(0, THREAD_COUNT * RUNS_PER_THREAD)
+ .<Future<Void>>mapToObj(i -> threadPool.submit(() -> {
+ try (Reader reader = new ChecksumReader(new StringReader(STRING_DATA[i % VARIANTS]), STRING_DATA_CHECKSUM[i % VARIANTS])) {
+ consumer.accept(reader);
+ }
+ return null;
+ })).collect(Collectors.toList());
+ futures.forEach(f -> assertDoesNotThrow(() -> f.get()));
+ } finally {
+ threadPool.shutdownNow();
+ }
+ }
+}