You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@commons.apache.org by pk...@apache.org on 2025/10/16 20:39:48 UTC
(commons-io) 01/01: Fix concurrency issue in `IOUtils.skip`
This is an automated email from the ASF dual-hosted git repository.
pkarwasz pushed a commit to branch fix/io-utils-skip
in repository https://gitbox.apache.org/repos/asf/commons-io.git
commit fc420bf978257e4e16f7db25c9ba21c9b3d74953
Author: Piotr P. Karwasz <pk...@apache.org>
AuthorDate: Thu Oct 16 22:35:14 2025 +0200
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.
---
src/changes/changes.xml | 1 +
src/main/java/org/apache/commons/io/IOUtils.java | 36 +---
.../java/org/apache/commons/io/IOCaseTest.java | 16 --
.../apache/commons/io/IOUtilsConcurrentTest.java | 212 +++++++++++++++++++++
4 files changed, 216 insertions(+), 49 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/IOUtils.java b/src/main/java/org/apache/commons/io/IOUtils.java
index a1f9b0709..631fa6ac6 100644
--- a/src/main/java/org/apache/commons/io/IOUtils.java
+++ b/src/main/java/org/apache/commons/io/IOUtils.java
@@ -206,21 +206,11 @@ public class IOUtils {
*/
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>
@@ -594,8 +584,6 @@ static void checkFromToIndex(final int fromIndex, final int toIndex, final int l
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);
}
/**
@@ -1868,15 +1856,6 @@ 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.
*
@@ -1886,15 +1865,6 @@ 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 +2497,7 @@ 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);
+ return skip(input, skip, IOUtils::getScratchByteArray);
}
/**
@@ -2636,7 +2606,7 @@ public static long skip(final Reader reader, final long toSkip) throws IOExcepti
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 char[] charArray = getScratchCharArray();
final long n = reader.read(charArray, 0, (int) Math.min(remain, charArray.length));
if (n < 0) { // EOF
break;
@@ -2667,7 +2637,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..ca1ca9bbd 100644
--- a/src/test/java/org/apache/commons/io/IOCaseTest.java
+++ b/src/test/java/org/apache/commons/io/IOCaseTest.java
@@ -302,14 +302,6 @@ void test_getScratchByteArray() {
assert0(IOUtils.getScratchCharArray());
}
- @Test
- void test_getScratchByteArrayWriteOnly() {
- final byte[] array = IOUtils.getScratchByteArrayWriteOnly();
- assert0(array);
- Arrays.fill(array, (byte) 1);
- assert0(IOUtils.getScratchCharArray());
- }
-
@Test
void test_getScratchCharArray() {
final char[] array = IOUtils.getScratchCharArray();
@@ -318,14 +310,6 @@ void test_getScratchCharArray() {
assert0(IOUtils.getScratchCharArray());
}
- @Test
- void test_getScratchCharArrayWriteOnly() {
- final char[] array = IOUtils.getScratchCharArrayWriteOnly();
- assert0(array);
- Arrays.fill(array, (char) 1);
- assert0(IOUtils.getScratchCharArray());
- }
-
@Test
void test_isCaseSensitive() {
assertTrue(IOCase.SENSITIVE.isCaseSensitive());
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();
+ }
+ }
+}