260 lines
8.5 KiB
Java
260 lines
8.5 KiB
Java
/*
|
|
* Copyright (c) 2014, Oracle and/or its affiliates. All rights reserved.
|
|
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
|
|
*
|
|
* This code is free software; you can redistribute it and/or modify it
|
|
* under the terms of the GNU General Public License version 2 only, as
|
|
* published by the Free Software Foundation.
|
|
*
|
|
* This code is distributed in the hope that it will be useful, but WITHOUT
|
|
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
|
|
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
|
|
* version 2 for more details (a copy is included in the LICENSE file that
|
|
* accompanied this code).
|
|
*
|
|
* You should have received a copy of the GNU General Public License version
|
|
* 2 along with this work; if not, write to the Free Software Foundation,
|
|
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
|
|
*
|
|
* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
|
|
* or visit www.oracle.com if you need additional information or have any
|
|
* questions.
|
|
*/
|
|
package org.openjdk.bench.java.net;
|
|
|
|
import org.openjdk.jmh.annotations.Benchmark;
|
|
import org.openjdk.jmh.annotations.BenchmarkMode;
|
|
import org.openjdk.jmh.annotations.Fork;
|
|
import org.openjdk.jmh.annotations.Measurement;
|
|
import org.openjdk.jmh.annotations.Mode;
|
|
import org.openjdk.jmh.annotations.OutputTimeUnit;
|
|
import org.openjdk.jmh.annotations.Scope;
|
|
import org.openjdk.jmh.annotations.Setup;
|
|
import org.openjdk.jmh.annotations.State;
|
|
import org.openjdk.jmh.annotations.TearDown;
|
|
import org.openjdk.jmh.annotations.Warmup;
|
|
|
|
import java.io.IOException;
|
|
import java.io.InputStream;
|
|
import java.io.OutputStream;
|
|
import java.net.InetAddress;
|
|
import java.net.ServerSocket;
|
|
import java.net.Socket;
|
|
import java.util.concurrent.TimeUnit;
|
|
|
|
/**
|
|
* Micro benchmark for streaming data over a Socket.
|
|
*/
|
|
@BenchmarkMode(Mode.AverageTime)
|
|
@OutputTimeUnit(TimeUnit.MILLISECONDS)
|
|
@State(Scope.Thread)
|
|
@Warmup(iterations = 5, time = 1)
|
|
@Measurement(iterations = 5, time = 1)
|
|
@Fork(value = 3)
|
|
public class SocketStreaming {
|
|
|
|
/** The bytes to write/read. */
|
|
public static final int dataLength = 16383;
|
|
/** setTcpNoDelay(noNagle) */
|
|
public static final boolean noNagle = false;
|
|
|
|
private WriterThread writerThread;
|
|
private Socket readSocket;
|
|
private byte[] bytes;
|
|
|
|
@Setup
|
|
public void prepare() throws Exception {
|
|
bytes = new byte[dataLength];
|
|
|
|
// Setup the writer thread
|
|
writerThread = new WriterThread(dataLength, noNagle);
|
|
writerThread.start();
|
|
|
|
// Wait for a read socket
|
|
readSocket = writerThread.waitForReadSocket();
|
|
}
|
|
|
|
@TearDown
|
|
public void cleanup() throws IOException {
|
|
// Take down the writer thread and the reader socket
|
|
writerThread.finish();
|
|
while (!readSocket.isClosed()) {
|
|
readSocket.close();
|
|
}
|
|
readSocket = null;
|
|
}
|
|
|
|
@Benchmark
|
|
public void testSocketInputStreamRead() throws InterruptedException, IOException {
|
|
InputStream in = readSocket.getInputStream();
|
|
|
|
// Notify the writer thread to add elements to stream
|
|
writerThread.requestSendBytes();
|
|
|
|
// Read these from the stream
|
|
int bytesRead = 0;
|
|
while (bytesRead < dataLength) {
|
|
int lastRead = in.read(bytes);
|
|
if (lastRead < 0) {
|
|
throw new InternalError("Unexpectedly got " + lastRead + " bytes from the socket");
|
|
}
|
|
bytesRead += lastRead;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Thread used to write bytes to a socket.
|
|
*/
|
|
private class WriterThread extends Thread {
|
|
|
|
/** The number of bytes to write. */
|
|
private int dataLength;
|
|
/** setTcpNoDelay(noNagle) */
|
|
private boolean noNagle;
|
|
/** Lock needed to send sendBytes requests. */
|
|
private final Object sendBytesLock = new Object();
|
|
/** Indicates that a sendBytes has been requested. */
|
|
private boolean sendBytesRequested;
|
|
/** Indicates that no more sendBytes will be requested. Time to shutdown. */
|
|
private boolean sendBytesDone;
|
|
/** Lock needed to protect the connectPort variable. */
|
|
private final Object connectLock = new Object();
|
|
/** The port the read socket should connect to. */
|
|
private int connectPort = -1;
|
|
|
|
/**
|
|
* Constructor.
|
|
*
|
|
* @param dataLength The number of bytes to write
|
|
* @param noNagle setTcpNoDelay(noNagle)
|
|
*/
|
|
public WriterThread(int dataLength, boolean noNagle) {
|
|
super("Load producer");
|
|
this.dataLength = dataLength;
|
|
this.noNagle = noNagle;
|
|
}
|
|
|
|
/** Entry point for data sending helper thread. */
|
|
@Override
|
|
public void run() {
|
|
try {
|
|
Socket writeSocket;
|
|
ServerSocket serverSocket = new ServerSocket(0);
|
|
|
|
/* Tell the other thread that we now know the port number.
|
|
* The other thread will now start to connect until the following accept() call succeeds.
|
|
*/
|
|
synchronized (connectLock) {
|
|
connectPort = serverSocket.getLocalPort();
|
|
connectLock.notify();
|
|
}
|
|
|
|
// Wait for the other thread to connect
|
|
writeSocket = serverSocket.accept();
|
|
writeSocket.setTcpNoDelay(noNagle);
|
|
|
|
// No more connects so this can be closed
|
|
serverSocket.close();
|
|
serverSocket = null;
|
|
|
|
OutputStream out = writeSocket.getOutputStream();
|
|
|
|
// Iterate as long as sendBytes are issued
|
|
while (waitForSendBytesRequest()) {
|
|
sendBytes(out);
|
|
}
|
|
|
|
// Time to shutdown
|
|
while (!writeSocket.isClosed()) {
|
|
writeSocket.close();
|
|
}
|
|
writeSocket = null;
|
|
} catch (Exception e) {
|
|
System.exit(1);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Sends bytes to the output stream
|
|
*
|
|
* @param out The output stream
|
|
* @throws IOException
|
|
*/
|
|
private void sendBytes(OutputStream out) throws IOException {
|
|
byte outBytes[] = new byte[dataLength];
|
|
|
|
int bytesToSend = dataLength;
|
|
int bytesSent = 0;
|
|
while (bytesSent < bytesToSend) {
|
|
out.write(outBytes);
|
|
bytesSent += outBytes.length;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Waits for the readSocket and returns it when it is ready.
|
|
*
|
|
* @return The socket to read from
|
|
* @throws InterruptedException
|
|
*/
|
|
@SuppressWarnings("SleepWhileHoldingLock")
|
|
public Socket waitForReadSocket() throws InterruptedException {
|
|
int theConnectPort = waitForConnectPort();
|
|
|
|
while (true) {
|
|
try {
|
|
return new Socket(InetAddress.getByName(null), theConnectPort);
|
|
} catch (IOException e) {
|
|
// Wait some more for the server thread to get going
|
|
Thread.sleep(1000);
|
|
}
|
|
}
|
|
|
|
}
|
|
|
|
/**
|
|
* Waits for next sendBytes request
|
|
*
|
|
* @return <code>true</code> if it is time to sendBytes, <code>false</code> if it is time to shutdown
|
|
* @throws InterruptedException
|
|
*/
|
|
public boolean waitForSendBytesRequest() throws InterruptedException {
|
|
synchronized (sendBytesLock) {
|
|
while (!sendBytesRequested && !sendBytesDone) {
|
|
sendBytesLock.wait();
|
|
}
|
|
|
|
// Clear the flag
|
|
sendBytesRequested = false;
|
|
|
|
return !sendBytesDone;
|
|
}
|
|
}
|
|
|
|
/** Requests a sendBytes. */
|
|
public void requestSendBytes() {
|
|
synchronized (sendBytesLock) {
|
|
sendBytesRequested = true;
|
|
sendBytesLock.notify();
|
|
}
|
|
}
|
|
|
|
/** Tells the writerThread that it is time to shutdown. */
|
|
public void finish() {
|
|
synchronized (sendBytesLock) {
|
|
sendBytesDone = true;
|
|
sendBytesLock.notify();
|
|
}
|
|
}
|
|
|
|
private int waitForConnectPort() throws InterruptedException {
|
|
synchronized (connectLock) {
|
|
while (connectPort == -1) {
|
|
connectLock.wait();
|
|
}
|
|
return connectPort;
|
|
}
|
|
}
|
|
}
|
|
}
|