Skip to content

Non-blocking read support on Windows #60

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
eed3si9n merged 1 commit into from
Oct 26, 2025
Merged

Non-blocking read support on Windows #60

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 22 additions & 4 deletions jni/org_scalasbt_ipcsocket_JNIWin32NamedPipeLibraryProvider.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,15 @@
(*env)->ThrowNew(env, exClass, _buf); \
} \
} while (0);
#define THROW_SOCKET_TIMEOUT(prefix, ...) \
do { \
char _buf[1024]; \
snprintf(_buf, 1024, prefix ? prefix : "%s", __VA_ARGS__); \
jclass exClass = (*env)->FindClass(env, "java/net/SocketTimeoutException"); \
if (exClass != NULL) { \
(*env)->ThrowNew(env, exClass, _buf); \
} \
} while (0);

#define FILL_ERROR(prefix, buf) \
do { \
Expand Down Expand Up @@ -123,8 +132,8 @@ jlong JNICALL
Java_org_scalasbt_ipcsocket_JNIWin32NamedPipeLibraryProvider_PeekNamedPipeNative(
UNUSED JNIEnv *env, UNUSED jobject object, jlong handlePointer) {
DWORD n = 0;
BOOL immediate = PeekNamedPipe((HANDLE)handlePointer, NULL, 0, NULL, &n, NULL);
if (!immediate) {
BOOL ok = PeekNamedPipe((HANDLE)handlePointer, NULL, 0, NULL, &n, NULL);
if (!ok) {
if (GetLastError() != ERROR_IO_PENDING) {
char buf[256];
FILL_ERROR("PeekNamedPipe() failed: %s (error code %ld)", buf);
Expand All @@ -137,7 +146,8 @@ Java_org_scalasbt_ipcsocket_JNIWin32NamedPipeLibraryProvider_PeekNamedPipeNative
jint JNICALL
Java_org_scalasbt_ipcsocket_JNIWin32NamedPipeLibraryProvider_readNative(
JNIEnv *env, UNUSED jobject object, jlong waitable, jlong hFile,
jbyteArray buffer, jint offset, jint length, jboolean strict) {
jbyteArray buffer, jint offset, jint length, jboolean strict,
jint timeoutMillis) {
HANDLE handle = (HANDLE)hFile;
OVERLAPPED olap = {0};
olap.hEvent = (HANDLE)waitable;
Expand All @@ -151,8 +161,16 @@ Java_org_scalasbt_ipcsocket_JNIWin32NamedPipeLibraryProvider_readNative(
FILL_ERROR("ReadFile() failed: %s (error code %ld)", buf);
THROW_IO(NULL, buf);
}
if (timeoutMillis > 0) {
DWORD res = WaitForSingleObject(olap.hEvent, timeoutMillis);
if (res == WAIT_TIMEOUT) {
CancelIoEx(handle, NULL);
char buf[256];
FILL_ERROR("ReadFile() timed out: %s (error code %ld)", buf);
THROW_SOCKET_TIMEOUT(NULL, buf);
}
}
}

if (!GetOverlappedResult(handle, &olap, &bytes_read, TRUE)) {
char buf[256];
FILL_ERROR(
Expand Down
4 changes: 2 additions & 2 deletions jni/org_scalasbt_ipcsocket_JNIWin32NamedPipeLibraryProvider.h
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

19 changes: 12 additions & 7 deletions src/main/java/org/scalasbt/ipcsocket/JNAWin32NamedPipeLibraryProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package org.scalasbt.ipcsocket;

import java.io.IOException;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;

import com.sun.jna.*;
Expand Down Expand Up @@ -107,12 +108,10 @@ public boolean DisconnectNamedPipe(Handle handle) {
public long PeekNamedPipe(Handle hFile) throws IOException {
HANDLE handle = getHandle(hFile);
IntByReference n = new IntByReference();
boolean immediate = delegate.PeekNamedPipe(handle, null, 0, null, n, null);
if (!immediate) {
boolean ok = delegate.PeekNamedPipe(handle, null, 0, null, n, null);
if (!ok) {
int lastError = delegate.GetLastError();
if (lastError != WinError.ERROR_IO_PENDING) {
throw new IOException("ReadFile() failed: " + lastError);
}
throw new IOException("PeekNamedPipe() failed: " + lastError);
}
return n.getValue();
}
Expand All @@ -124,7 +123,8 @@ public int read(
byte[] buffer,
int offset,
int len,
boolean requireStrictLength)
boolean requireStrictLength,
int timeoutMillis)
throws IOException {
HANDLE readerWaitable = getHandle(waitable);
HANDLE handle = getHandle(hFile);
Expand All @@ -140,8 +140,13 @@ public int read(
if (lastError != WinError.ERROR_IO_PENDING) {
throw new IOException("ReadFile() failed: " + lastError);
}
if (timeoutMillis > 0) {
int res = delegate.WaitForSingleObject(olap.hEvent, timeoutMillis);
if (res == WinError.WAIT_TIMEOUT) {
throw new SocketTimeoutException("ReadFile() timed out " + res);
}
}
}

IntByReference r = new IntByReference();
if (!delegate.GetOverlappedResult(handle, olap.getPointer(), r, true)) {
int lastError = delegate.GetLastError();
Expand Down
14 changes: 11 additions & 3 deletions src/main/java/org/scalasbt/ipcsocket/JNIWin32NamedPipeLibraryProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,19 +97,27 @@ public int read(
byte[] buffer,
int offset,
int len,
boolean requireStrictLength)
boolean requireStrictLength,
int timeoutMillis)
throws IOException {
return readNative(
getHandlePointer(waitable),
getHandlePointer(hFile),
buffer,
offset,
len,
requireStrictLength);
requireStrictLength,
timeoutMillis);
}

native int readNative(
long waitable, long hFile, byte[] buffer, int offset, int len, boolean requireStrictLength)
long waitable,
long hFile,
byte[] buffer,
int offset,
int len,
boolean requireStrictLength,
int timeoutMillis)
throws IOException;

@Override
Expand Down
45 changes: 26 additions & 19 deletions src/main/java/org/scalasbt/ipcsocket/SocketChannels.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,7 @@ public static String readLine(SocketChannel channel) throws IOException {
}

/** Utility function to read until newline from a non-blocking channel. */
public static String readLine(SocketChannel channel, int readTimeoutMilis) throws IOException {
public static String readLine(SocketChannel channel, int readTimeoutMillis) throws IOException {
int readBytes;
byte b;
final List<Byte> values = new ArrayList<>();
Expand All @@ -96,19 +96,22 @@ public static String readLine(SocketChannel channel, int readTimeoutMilis) throw
if (isJava17Plus() && !ServerSocketChannels.isWin) {
if (!channel.isBlocking()) {
channel.register(sel, SelectionKey.OP_READ);
numOfKeys = sel.select(readTimeoutMilis);
numOfKeys = sel.select(readTimeoutMillis);
}
} else {
if (readTimeoutMilis > 0) {
throw new IOException("timeout requires JDK 17 and non-Windows");
}
if (channel.supportedOptions().contains(SO_TIMEOUT)) {
channel.setOption(SO_TIMEOUT, Integer.valueOf(readTimeoutMilis));
if (readTimeoutMillis > 0) {
if (ServerSocketChannels.isWin) {
if (channel.supportedOptions().contains(SO_TIMEOUT)) {
channel.setOption(SO_TIMEOUT, Integer.valueOf(readTimeoutMillis));
}
} else {
throw new IOException("timeout requires JDK 17 or Windows");
}
}
}
if (numOfKeys == 0) {
throw new SocketTimeoutException(
"readLine timed out after " + Integer.toString(readTimeoutMilis) + " msec");
"readLine timed out after " + Integer.toString(readTimeoutMillis) + " msec");
} else {
readBytes = channel.read(buf);
}
Expand All @@ -133,7 +136,8 @@ public static ByteBuffer readAll(SocketChannel channel) throws IOException {
}

/** Utility function to read all buffer from a non-blocking channel. */
public static ByteBuffer readAll(SocketChannel channel, int readTimeoutMilis) throws IOException {
public static ByteBuffer readAll(SocketChannel channel, int readTimeoutMillis)
throws IOException {
int readBytes;
final List<Byte> values = new ArrayList<>();
final int bufSize = 1024 * 1024;
Expand All @@ -145,22 +149,25 @@ public static ByteBuffer readAll(SocketChannel channel, int readTimeoutMilis) th
if (isJava17Plus() && !ServerSocketChannels.isWin) {
if (!channel.isBlocking()) {
channel.register(sel, SelectionKey.OP_READ);
numOfKeys = sel.select(readTimeoutMilis);
numOfKeys = sel.select(readTimeoutMillis);
}
} else {
if (readTimeoutMilis > 0) {
throw new IOException("timeout requires JDK 17 and non-Windows");
}
// The following operation gets blocked on JDK 8
// channel.register(sel, SelectionKey.OP_READ);
// numOfKeys = sel.select(readTimeoutMilis);
if (channel.supportedOptions().contains(SO_TIMEOUT)) {
channel.setOption(SO_TIMEOUT, Integer.valueOf(readTimeoutMilis));
if (readTimeoutMillis > 0) {
// The following operation gets blocked on JDK 8
// channel.register(sel, SelectionKey.OP_READ);
// numOfKeys = sel.select(readTimeoutMillis);
if (ServerSocketChannels.isWin) {
if (channel.supportedOptions().contains(SO_TIMEOUT)) {
channel.setOption(SO_TIMEOUT, Integer.valueOf(readTimeoutMillis));
}
} else {
throw new IOException("timeout requires JDK 17 or Windows");
}
}
}
if (numOfKeys == 0) {
throw new SocketTimeoutException(
"readAll timed out after " + Integer.toString(readTimeoutMilis) + " msec");
"readAll timed out after " + Integer.toString(readTimeoutMillis) + " msec");
} else {
readBytes = channel.read(buf);
}
Expand Down
2 changes: 2 additions & 0 deletions src/main/java/org/scalasbt/ipcsocket/Win32NamedPipeLibrary.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,8 @@ boolean GetOverlappedResult(

boolean CancelIoEx(HANDLE hObject, Pointer lpOverlapped);

int WaitForSingleObject(HANDLE hObject, int dwMilliseconds);

HANDLE CreateEvent(
SECURITY_ATTRIBUTES lpEventAttributes,
boolean bManualReset,
Expand Down
3 changes: 2 additions & 1 deletion src/main/java/org/scalasbt/ipcsocket/Win32NamedPipeLibraryProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,8 @@ int read(
byte[] buffer,
int offset,
int len,
boolean requireStrictLength)
boolean requireStrictLength,
int timeoutMillis)
throws IOException;

void write(Handle waitable, Handle hFile, byte[] lpBuffer, int offset, int len)
Expand Down
3 changes: 2 additions & 1 deletion src/main/java/org/scalasbt/ipcsocket/Win32NamedPipeSocket.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,7 +140,8 @@ public int read() throws IOException {

@Override
public int read(byte[] b, int off, int len) throws IOException {
return provider.read(readerWaitable, handle, b, off, len, requireStrictLength);
return provider.read(
readerWaitable, handle, b, off, len, requireStrictLength, getSoTimeout());
}
}

Expand Down
Binary file modified src/main/resources/win32/x86_64/sbtipcsocket.dll
View file
Open in desktop
Binary file not shown.
4 changes: 2 additions & 2 deletions src/test/java/org/scalasbt/ipcsocket/SocketChannelTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ public void testNonBlockingEchoServer() throws IOException, InterruptedException
"SocketChannelTest#testNonBlockingEchoServer(" + Boolean.toString(useJNI()) + ")");
withSocket(
sock -> {
if (isJava17Plus() && !ServerSocketChannels.isWin) {
if (isJava17Plus() || ServerSocketChannels.isWin) {
String line = nonBlockingEchoServerTest(sock, 100, 600);
assertEquals("echo did not return the content", "hello", line);
}
Expand All @@ -34,7 +34,7 @@ public void testTimeout() throws IOException, InterruptedException {
System.out.println("SocketChannelTest#testTimeout(" + Boolean.toString(useJNI()) + ")");
withSocket(
sock -> {
if (isJava17Plus() && !ServerSocketChannels.isWin) {
if (isJava17Plus() || ServerSocketChannels.isWin) {
String line = nonBlockingEchoServerTest(sock, 6000, 600);
assertEquals("echo did not timeout", "<unavailable>", line);
}
Expand Down