Skip to content

Commit 1855574

Browse files
mkargAlan Bateman
mkarg
authored and
Alan Bateman
committedSep 8, 2021
8273038: ChannelInputStream.transferTo() uses FileChannel.transferTo(FileChannel)
Reviewed-by: alanb
·
jdk-25+13jdk-18-ga
1 parent 6750c34 commit 1855574

File tree

4 files changed

+388
-80
lines changed

4 files changed

+388
-80
lines changed
 

‎src/java.base/share/classes/java/nio/channels/Channels.java

+4-79
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@
4141
import java.util.Objects;
4242
import java.util.concurrent.ExecutionException;
4343
import sun.nio.ch.ChannelInputStream;
44+
import sun.nio.ch.ChannelOutputStream;
4445
import sun.nio.cs.StreamDecoder;
4546
import sun.nio.cs.StreamEncoder;
4647

@@ -63,40 +64,6 @@ public final class Channels {
6364

6465
private Channels() { throw new Error("no instances"); }
6566

66-
/**
67-
* Write all remaining bytes in buffer to the given channel.
68-
* If the channel is selectable then it must be configured blocking.
69-
*/
70-
private static void writeFullyImpl(WritableByteChannel ch, ByteBuffer bb)
71-
throws IOException
72-
{
73-
while (bb.remaining() > 0) {
74-
int n = ch.write(bb);
75-
if (n <= 0)
76-
throw new RuntimeException("no bytes written");
77-
}
78-
}
79-
80-
/**
81-
* Write all remaining bytes in buffer to the given channel.
82-
*
83-
* @throws IllegalBlockingModeException
84-
* If the channel is selectable and configured non-blocking.
85-
*/
86-
private static void writeFully(WritableByteChannel ch, ByteBuffer bb)
87-
throws IOException
88-
{
89-
if (ch instanceof SelectableChannel sc) {
90-
synchronized (sc.blockingLock()) {
91-
if (!sc.isBlocking())
92-
throw new IllegalBlockingModeException();
93-
writeFullyImpl(ch, bb);
94-
}
95-
} else {
96-
writeFullyImpl(ch, bb);
97-
}
98-
}
99-
10067
// -- Byte streams from channels --
10168

10269
/**
@@ -136,47 +103,7 @@ public static InputStream newInputStream(ReadableByteChannel ch) {
136103
*/
137104
public static OutputStream newOutputStream(WritableByteChannel ch) {
138105
Objects.requireNonNull(ch, "ch");
139-
140-
return new OutputStream() {
141-
142-
private ByteBuffer bb;
143-
private byte[] bs; // Invoker's previous array
144-
private byte[] b1;
145-
146-
@Override
147-
public synchronized void write(int b) throws IOException {
148-
if (b1 == null)
149-
b1 = new byte[1];
150-
b1[0] = (byte) b;
151-
this.write(b1);
152-
}
153-
154-
@Override
155-
public synchronized void write(byte[] bs, int off, int len)
156-
throws IOException
157-
{
158-
if ((off < 0) || (off > bs.length) || (len < 0) ||
159-
((off + len) > bs.length) || ((off + len) < 0)) {
160-
throw new IndexOutOfBoundsException();
161-
} else if (len == 0) {
162-
return;
163-
}
164-
ByteBuffer bb = ((this.bs == bs)
165-
? this.bb
166-
: ByteBuffer.wrap(bs));
167-
bb.limit(Math.min(off + len, bb.capacity()));
168-
bb.position(off);
169-
this.bb = bb;
170-
this.bs = bs;
171-
Channels.writeFully(ch, bb);
172-
}
173-
174-
@Override
175-
public void close() throws IOException {
176-
ch.close();
177-
}
178-
179-
};
106+
return new ChannelOutputStream(ch);
180107
}
181108

182109
/**
@@ -216,10 +143,8 @@ public synchronized int read() throws IOException {
216143
public synchronized int read(byte[] bs, int off, int len)
217144
throws IOException
218145
{
219-
if ((off < 0) || (off > bs.length) || (len < 0) ||
220-
((off + len) > bs.length) || ((off + len) < 0)) {
221-
throw new IndexOutOfBoundsException();
222-
} else if (len == 0) {
146+
Objects.checkFromIndexSize(off, len, bs.length);
147+
if (len == 0) {
223148
return 0;
224149
}
225150

‎src/java.base/share/classes/sun/nio/ch/ChannelInputStream.java

+26-1
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright (c) 2001, 2019, Oracle and/or its affiliates. All rights reserved.
2+
* Copyright (c) 2001, 2021, Oracle and/or its affiliates. All rights reserved.
33
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
44
*
55
* This code is free software; you can redistribute it and/or modify it
@@ -142,4 +142,29 @@ public void close() throws IOException {
142142
ch.close();
143143
}
144144

145+
@Override
146+
public long transferTo(OutputStream out) throws IOException {
147+
Objects.requireNonNull(out, "out");
148+
149+
if (out instanceof ChannelOutputStream cos
150+
&& ch instanceof FileChannel fc
151+
&& cos.channel() instanceof FileChannel dst) {
152+
return transfer(fc, dst);
153+
}
154+
155+
return super.transferTo(out);
156+
}
157+
158+
private static long transfer(FileChannel src, FileChannel dst) throws IOException {
159+
long initialPos = src.position();
160+
long pos = initialPos;
161+
try {
162+
while (pos < src.size()) {
163+
pos += src.transferTo(pos, Long.MAX_VALUE, dst);
164+
}
165+
} finally {
166+
src.position(pos);
167+
}
168+
return pos - initialPos;
169+
}
145170
}

‎src/java.base/share/classes/sun/nio/ch/ChannelOutputStream.java

+130
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
/*
2+
* Copyright (c) 2021, Oracle and/or its affiliates. All rights reserved.
3+
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
4+
*
5+
* This code is free software; you can redistribute it and/or modify it
6+
* under the terms of the GNU General Public License version 2 only, as
7+
* published by the Free Software Foundation. Oracle designates this
8+
* particular file as subject to the "Classpath" exception as provided
9+
* by Oracle in the LICENSE file that accompanied this code.
10+
*
11+
* This code is distributed in the hope that it will be useful, but WITHOUT
12+
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
13+
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
14+
* version 2 for more details (a copy is included in the LICENSE file that
15+
* accompanied this code).
16+
*
17+
* You should have received a copy of the GNU General Public License version
18+
* 2 along with this work; if not, write to the Free Software Foundation,
19+
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
20+
*
21+
* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
22+
* or visit www.oracle.com if you need additional information or have any
23+
* questions.
24+
*/
25+
26+
package sun.nio.ch;
27+
28+
import java.io.*;
29+
import java.nio.*;
30+
import java.nio.channels.*;
31+
import java.nio.channels.spi.*;
32+
import java.util.Objects;
33+
34+
/**
35+
* This class is defined here rather than in java.nio.channels.Channels
36+
* so that it will be accessible from java.nio.channels.Channels and
37+
* sun.nio.ch.ChannelInputStream.
38+
*
39+
*
40+
* @author Mark Reinhold
41+
* @author Mike McCloskey
42+
* @author JSR-51 Expert Group
43+
* @since 18
44+
*/
45+
public class ChannelOutputStream extends OutputStream {
46+
47+
private final WritableByteChannel ch;
48+
private ByteBuffer bb;
49+
private byte[] bs; // Invoker's previous array
50+
private byte[] b1;
51+
52+
/**
53+
* Write all remaining bytes in buffer to the given channel.
54+
* If the channel is selectable then it must be configured blocking.
55+
*/
56+
private static void writeFullyImpl(WritableByteChannel ch, ByteBuffer bb)
57+
throws IOException
58+
{
59+
while (bb.remaining() > 0) {
60+
int n = ch.write(bb);
61+
if (n <= 0)
62+
throw new RuntimeException("no bytes written");
63+
}
64+
}
65+
66+
/**
67+
* Write all remaining bytes in buffer to the given channel.
68+
*
69+
* @throws IllegalBlockingModeException
70+
* If the channel is selectable and configured non-blocking.
71+
*/
72+
private static void writeFully(WritableByteChannel ch, ByteBuffer bb)
73+
throws IOException
74+
{
75+
if (ch instanceof SelectableChannel sc) {
76+
synchronized (sc.blockingLock()) {
77+
if (!sc.isBlocking())
78+
throw new IllegalBlockingModeException();
79+
writeFullyImpl(ch, bb);
80+
}
81+
} else {
82+
writeFullyImpl(ch, bb);
83+
}
84+
}
85+
86+
/**
87+
* @param ch The channel wrapped by this stream.
88+
*/
89+
public ChannelOutputStream(WritableByteChannel ch) {
90+
this.ch = ch;
91+
}
92+
93+
/**
94+
* @return The channel wrapped by this stream.
95+
*/
96+
WritableByteChannel channel() {
97+
return ch;
98+
}
99+
100+
@Override
101+
public synchronized void write(int b) throws IOException {
102+
if (b1 == null)
103+
b1 = new byte[1];
104+
b1[0] = (byte) b;
105+
this.write(b1);
106+
}
107+
108+
@Override
109+
public synchronized void write(byte[] bs, int off, int len)
110+
throws IOException {
111+
Objects.checkFromIndexSize(off, len, bs.length);
112+
if (len == 0) {
113+
return;
114+
}
115+
ByteBuffer bb = ((this.bs == bs)
116+
? this.bb
117+
: ByteBuffer.wrap(bs));
118+
bb.limit(Math.min(off + len, bb.capacity()));
119+
bb.position(off);
120+
this.bb = bb;
121+
this.bs = bs;
122+
writeFully(ch, bb);
123+
}
124+
125+
@Override
126+
public void close() throws IOException {
127+
ch.close();
128+
}
129+
130+
}

‎test/jdk/java/nio/channels/Channels/TransferTo.java

+228
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,228 @@
1+
/*
2+
* Copyright (c) 2021, Oracle and/or its affiliates. All rights reserved.
3+
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
4+
*
5+
* This code is free software; you can redistribute it and/or modify it
6+
* under the terms of the GNU General Public License version 2 only, as
7+
* published by the Free Software Foundation.
8+
*
9+
* This code is distributed in the hope that it will be useful, but WITHOUT
10+
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
11+
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
12+
* version 2 for more details (a copy is included in the LICENSE file that
13+
* accompanied this code).
14+
*
15+
* You should have received a copy of the GNU General Public License version
16+
* 2 along with this work; if not, write to the Free Software Foundation,
17+
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
18+
*
19+
* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
20+
* or visit www.oracle.com if you need additional information or have any
21+
* questions.
22+
*/
23+
24+
import static java.lang.String.format;
25+
26+
import java.io.ByteArrayInputStream;
27+
import java.io.ByteArrayOutputStream;
28+
import java.io.IOException;
29+
import java.io.InputStream;
30+
import java.io.OutputStream;
31+
import java.nio.channels.Channels;
32+
import java.nio.channels.FileChannel;
33+
import java.nio.file.Files;
34+
import java.nio.file.Path;
35+
import java.nio.file.StandardOpenOption;
36+
import java.util.Arrays;
37+
import java.util.Random;
38+
import java.util.concurrent.atomic.AtomicReference;
39+
import java.util.function.Consumer;
40+
import java.util.function.Supplier;
41+
42+
import jdk.test.lib.RandomFactory;
43+
44+
/*
45+
* @test
46+
* @library /test/lib
47+
* @build jdk.test.lib.RandomFactory
48+
* @run main TransferTo
49+
* @bug 8265891
50+
* @summary tests whether sun.nio.ChannelInputStream.transferTo conforms to the
51+
* InputStream.transferTo contract defined in the javadoc
52+
* @key randomness
53+
*/
54+
public class TransferTo {
55+
private static final int MIN_SIZE = 10_000;
56+
private static final int MAX_SIZE_INCR = 100_000_000 - MIN_SIZE;
57+
58+
private static final int ITERATIONS = 10;
59+
60+
private static final Random RND = RandomFactory.getRandom();
61+
62+
public static void main(String[] args) throws Exception {
63+
test(fileChannelInput(), fileChannelOutput());
64+
test(readableByteChannelInput(), defaultOutput());
65+
}
66+
67+
private static void test(InputStreamProvider inputStreamProvider, OutputStreamProvider outputStreamProvider)
68+
throws Exception {
69+
testNullPointerException(inputStreamProvider);
70+
testStreamContents(inputStreamProvider, outputStreamProvider);
71+
}
72+
73+
private static void testNullPointerException(InputStreamProvider inputStreamProvider) throws Exception {
74+
try (InputStream in = inputStreamProvider.input()) {
75+
assertThrowsNPE(() -> in.transferTo(null), "out");
76+
}
77+
78+
try (InputStream in = inputStreamProvider.input((byte) 1)) {
79+
assertThrowsNPE(() -> in.transferTo(null), "out");
80+
}
81+
82+
try (InputStream in = inputStreamProvider.input((byte) 1, (byte) 2)) {
83+
assertThrowsNPE(() -> in.transferTo(null), "out");
84+
}
85+
}
86+
87+
private static void testStreamContents(InputStreamProvider inputStreamProvider,
88+
OutputStreamProvider outputStreamProvider) throws Exception {
89+
checkTransferredContents(inputStreamProvider, outputStreamProvider, new byte[0]);
90+
checkTransferredContents(inputStreamProvider, outputStreamProvider, createRandomBytes(1024, 4096));
91+
92+
// to span through several batches
93+
checkTransferredContents(inputStreamProvider, outputStreamProvider, createRandomBytes(16384, 16384));
94+
95+
// randomly chosen starting positions within source and target
96+
for (int i = 0; i < ITERATIONS; i++) {
97+
byte[] inBytes = createRandomBytes(MIN_SIZE, MAX_SIZE_INCR);
98+
int posIn = RND.nextInt(inBytes.length);
99+
int posOut = RND.nextInt(MIN_SIZE);
100+
checkTransferredContents(inputStreamProvider, outputStreamProvider, inBytes, posIn, posOut);
101+
}
102+
103+
// beyond source EOF
104+
checkTransferredContents(inputStreamProvider, outputStreamProvider, createRandomBytes(4096, 0), 4096, 0);
105+
106+
// beyond target EOF
107+
checkTransferredContents(inputStreamProvider, outputStreamProvider, createRandomBytes(4096, 0), 0, 4096);
108+
}
109+
110+
private static void checkTransferredContents(InputStreamProvider inputStreamProvider,
111+
OutputStreamProvider outputStreamProvider, byte[] inBytes) throws Exception {
112+
checkTransferredContents(inputStreamProvider, outputStreamProvider, inBytes, 0, 0);
113+
}
114+
115+
private static void checkTransferredContents(InputStreamProvider inputStreamProvider,
116+
OutputStreamProvider outputStreamProvider, byte[] inBytes, int posIn, int posOut) throws Exception {
117+
AtomicReference<Supplier<byte[]>> recorder = new AtomicReference<>();
118+
try (InputStream in = inputStreamProvider.input(inBytes);
119+
OutputStream out = outputStreamProvider.output(recorder::set)) {
120+
// skip bytes till starting position
121+
in.skipNBytes(posIn);
122+
out.write(new byte[posOut]);
123+
124+
long reported = in.transferTo(out);
125+
int count = inBytes.length - posIn;
126+
127+
if (reported != count)
128+
throw new AssertionError(
129+
format("reported %d bytes but should report %d", reported, count));
130+
131+
byte[] outBytes = recorder.get().get();
132+
if (!Arrays.equals(inBytes, posIn, posIn + count, outBytes, posOut, posOut + count))
133+
throw new AssertionError(
134+
format("inBytes.length=%d, outBytes.length=%d", count, outBytes.length));
135+
}
136+
}
137+
138+
private static byte[] createRandomBytes(int min, int maxRandomAdditive) {
139+
byte[] bytes = new byte[min + (maxRandomAdditive == 0 ? 0 : RND.nextInt(maxRandomAdditive))];
140+
RND.nextBytes(bytes);
141+
return bytes;
142+
}
143+
144+
private interface InputStreamProvider {
145+
InputStream input(byte... bytes) throws Exception;
146+
}
147+
148+
private interface OutputStreamProvider {
149+
OutputStream output(Consumer<Supplier<byte[]>> spy) throws Exception;
150+
}
151+
152+
private static OutputStreamProvider defaultOutput() {
153+
return new OutputStreamProvider() {
154+
@Override
155+
public OutputStream output(Consumer<Supplier<byte[]>> spy) {
156+
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
157+
spy.accept(outputStream::toByteArray);
158+
return outputStream;
159+
}
160+
};
161+
}
162+
163+
private static InputStreamProvider fileChannelInput() {
164+
return new InputStreamProvider() {
165+
@Override
166+
public InputStream input(byte... bytes) throws Exception {
167+
Path path = Files.createTempFile(null, null);
168+
Files.write(path, bytes);
169+
FileChannel fileChannel = FileChannel.open(path);
170+
return Channels.newInputStream(fileChannel);
171+
}
172+
};
173+
}
174+
175+
private static InputStreamProvider readableByteChannelInput() {
176+
return new InputStreamProvider() {
177+
@Override
178+
public InputStream input(byte... bytes) throws Exception {
179+
return Channels.newInputStream(Channels.newChannel(new ByteArrayInputStream(bytes)));
180+
}
181+
};
182+
}
183+
184+
private static OutputStreamProvider fileChannelOutput() {
185+
return new OutputStreamProvider() {
186+
public OutputStream output(Consumer<Supplier<byte[]>> spy) throws Exception {
187+
Path path = Files.createTempFile(null, null);
188+
FileChannel fileChannel = FileChannel.open(path, StandardOpenOption.WRITE);
189+
spy.accept(() -> {
190+
try {
191+
return Files.readAllBytes(path);
192+
} catch (IOException e) {
193+
throw new AssertionError("Failed to verify output file", e);
194+
}
195+
});
196+
return Channels.newOutputStream(fileChannel);
197+
}
198+
};
199+
}
200+
201+
public interface Thrower {
202+
public void run() throws Throwable;
203+
}
204+
205+
public static void assertThrowsNPE(Thrower thrower, String message) {
206+
assertThrows(thrower, NullPointerException.class, message);
207+
}
208+
209+
public static <T extends Throwable> void assertThrows(Thrower thrower, Class<T> throwable, String message) {
210+
Throwable thrown;
211+
try {
212+
thrower.run();
213+
thrown = null;
214+
} catch (Throwable caught) {
215+
thrown = caught;
216+
}
217+
218+
if (!throwable.isInstance(thrown)) {
219+
String caught = thrown == null ? "nothing" : thrown.getClass().getCanonicalName();
220+
throw new AssertionError(format("Expected to catch %s, but caught %s", throwable, caught), thrown);
221+
}
222+
223+
if (thrown != null && !message.equals(thrown.getMessage())) {
224+
throw new AssertionError(
225+
format("Expected exception message to be '%s', but it's '%s'", message, thrown.getMessage()));
226+
}
227+
}
228+
}

1 commit comments

Comments
 (1)

openjdk-notifier[bot] commented on Sep 8, 2021

@openjdk-notifier[bot]
openjdk-notifier[bot]
on Sep 8, 2021

Review

Issues

Please sign in to comment.