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:47 UTC
(commons-io) branch fix/io-utils-skip created (now fc420bf97)
This is an automated email from the ASF dual-hosted git repository.
pkarwasz pushed a change to branch fix/io-utils-skip
in repository https://gitbox.apache.org/repos/asf/commons-io.git
at fc420bf97 Fix concurrency issue in `IOUtils.skip`
This branch includes the following new commits:
new fc420bf97 Fix concurrency issue in `IOUtils.skip`
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(commons-io) 01/01: Fix concurrency issue in `IOUtils.skip`
Posted by pk...@apache.org.
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();
+ }
+ }
+}