From 93ce6c03a3707cc66aa086ba60f02cdc59c7fe1d Mon Sep 17 00:00:00 2001 From: "Flavio S. Glock" Date: Tue, 8 Sep 2026 13:41:34 +0200 Subject: [PATCH] fix: preserve IO::Async accepted sockets and binary payloads Keep accepted anonymous sockets alive when callback arguments are assigned to lexicals, and preserve raw octets through internal pipes used by IO::Async channels. Add focused regressions for both behaviors. Generated with Codex (https://openai.com/codex) Co-Authored-By: Codex --- docs/about/changelog.md | 6 +-- .../runtime/io/InternalPipeHandle.java | 23 +++++++++++- .../runtime/operators/IOOperator.java | 2 + .../runtime/runtimetypes/RuntimeGlob.java | 5 +++ .../runtime/runtimetypes/RuntimeScalar.java | 15 +++++++- .../SkipUnsupportedSocketTests.patch | 10 ----- .../unit/accepted_socket_callback_lifetime.t | 37 +++++++++++++++++++ .../unit/pipe_storable_binary_roundtrip.t | 24 ++++++++++++ 8 files changed, 106 insertions(+), 16 deletions(-) create mode 100644 src/test/resources/unit/accepted_socket_callback_lifetime.t create mode 100644 src/test/resources/unit/pipe_storable_binary_roundtrip.t diff --git a/docs/about/changelog.md b/docs/about/changelog.md index 1927c94997..aec390c121 100644 --- a/docs/about/changelog.md +++ b/docs/about/changelog.md @@ -6,9 +6,9 @@ priorities and future plans. ## Work in progress -- Preserve IO::Async thread callback results in scalar and list context on - both execution backends, and align its notifier-loop refcount expectation - with native Perl. +- Preserve IO::Async thread callback results and accepted listener sockets on + both execution backends, retain binary channel payload octets, and align its + notifier-loop refcount expectation with native Perl. - Fix parsing of dense Mo::Inline expressions that use `::` as a bareword. diff --git a/src/main/java/org/perlonjava/runtime/io/InternalPipeHandle.java b/src/main/java/org/perlonjava/runtime/io/InternalPipeHandle.java index d37a03b9f6..a11de7fbec 100644 --- a/src/main/java/org/perlonjava/runtime/io/InternalPipeHandle.java +++ b/src/main/java/org/perlonjava/runtime/io/InternalPipeHandle.java @@ -190,7 +190,28 @@ public RuntimeScalar write(String string) { } try { - byte[] bytes = string.getBytes(StandardCharsets.UTF_8); + // An unlayered Perl handle writes octets, not UTF-8-encoded text. + // RuntimeScalar represents byte strings as chars in the 0..255 + // range, so UTF-8 encoding here expands octets such as 0x8a to + // two bytes. That corrupts framed binary protocols (notably + // Storable over IO::Async::Channel) because their length prefix + // still describes the original octet count. + boolean hasWideChars = false; + for (int i = 0; i < string.length(); i++) { + if (string.charAt(i) > 0xFF) { + hasWideChars = true; + break; + } + } + byte[] bytes; + if (hasWideChars) { + bytes = string.getBytes(StandardCharsets.UTF_8); + } else { + bytes = new byte[string.length()]; + for (int i = 0; i < string.length(); i++) { + bytes[i] = (byte) string.charAt(i); + } + } if (!blocking && inputStream != null) { int free = pipeSize - inputStream.available(); if (free <= 0) { diff --git a/src/main/java/org/perlonjava/runtime/operators/IOOperator.java b/src/main/java/org/perlonjava/runtime/operators/IOOperator.java index 9ebe834928..cedf73690c 100644 --- a/src/main/java/org/perlonjava/runtime/operators/IOOperator.java +++ b/src/main/java/org/perlonjava/runtime/operators/IOOperator.java @@ -2302,6 +2302,7 @@ public static RuntimeScalar accept(int ctx, RuntimeBase... args) { if (targetGlob != null) { targetGlob.setIO(clientRuntimeIO); + targetGlob.acceptedSocket = true; MyVarCleanupStack.retainLiveIoGlobOwners(targetGlob); RuntimeScalar.retainUnstashedIoForDurableSlot(newSocketHandle); } else { @@ -2309,6 +2310,7 @@ public static RuntimeScalar accept(int ctx, RuntimeBase... args) { RuntimeScalar newGlob = new RuntimeScalar(); newGlob.type = RuntimeScalarType.GLOBREFERENCE; RuntimeGlob anonGlob = new RuntimeGlob(null).setIO(clientRuntimeIO); + anonGlob.acceptedSocket = true; newGlob.value = anonGlob; RuntimeIO.registerGlobForFdRecycling(anonGlob, clientRuntimeIO); RuntimeScalar assignedHandle = newSocketHandle.set(newGlob); diff --git a/src/main/java/org/perlonjava/runtime/runtimetypes/RuntimeGlob.java b/src/main/java/org/perlonjava/runtime/runtimetypes/RuntimeGlob.java index 125039747e..df3ee3f669 100644 --- a/src/main/java/org/perlonjava/runtime/runtimetypes/RuntimeGlob.java +++ b/src/main/java/org/perlonjava/runtime/runtimetypes/RuntimeGlob.java @@ -100,6 +100,9 @@ public static RuntimeArray localizedUnderscoreArrayForCurrentCall() { /** Number of scalar wrappers currently pointing at this anonymous IO glob. */ public int ioHolderCount = 0; + /** True when this glob's current IO was created by accept(). */ + public boolean acceptedSocket; + /** * Constructor for RuntimeGlob. * Initializes a new instance of the RuntimeGlob class with the specified glob name. @@ -1205,6 +1208,7 @@ public RuntimeArray getGlobArray() { public RuntimeGlob setIO(RuntimeScalar io) { GlobalVariable.markStashEntryVisible(this.globName); + acceptedSocket = false; // Check if the current IO is the selected handle - if so, update it RuntimeIO oldIO = null; if (this.IO.value instanceof RuntimeIO) { @@ -1232,6 +1236,7 @@ public RuntimeGlob setIO(RuntimeScalar io) { public RuntimeGlob setIO(RuntimeIO io) { GlobalVariable.markStashEntryVisible(this.globName); + acceptedSocket = false; // Set the glob name in the RuntimeIO for proper stringification io.globName = this.globName; // Check if the current IO is the selected handle - if so, update it diff --git a/src/main/java/org/perlonjava/runtime/runtimetypes/RuntimeScalar.java b/src/main/java/org/perlonjava/runtime/runtimetypes/RuntimeScalar.java index c6c8ff0098..7e2c0af651 100644 --- a/src/main/java/org/perlonjava/runtime/runtimetypes/RuntimeScalar.java +++ b/src/main/java/org/perlonjava/runtime/runtimetypes/RuntimeScalar.java @@ -1975,6 +1975,8 @@ private RuntimeScalar setLargeRefCounted(RuntimeScalar value) { boolean assignedFromArgumentAlias = RuntimeCode.isCurrentArgumentAlias(value) || (!value.ioOwner && RuntimeCode.isArgumentFrameActive(value.copiedFromArgumentFrame)); + boolean assignedFromAcceptedSocketArgument = assignedFromArgumentAlias + && isAcceptedSocket(value); boolean transferDetachedIoOwner = durableIoDestination && this != value && !assignedFromArgumentAlias @@ -1994,7 +1996,7 @@ && isUnstashedIoGlob(transferGlob) && isUnstashedIoGlob(newGlob) && hasLiveIo(newGlob) && durableIoDestination - && !assignedFromArgumentAlias + && (!assignedFromArgumentAlias || assignedFromAcceptedSocketArgument) && !transferDetachedIoOwner) { newGlob.ioHolderCount++; } @@ -2099,7 +2101,7 @@ && isSocketIOHandle(oldIo.ioHandle)) { value.ioOwner = false; this.ioOwner = true; } else if (durableIoDestination - && !assignedFromArgumentAlias + && (!assignedFromArgumentAlias || assignedFromAcceptedSocketArgument) && this != value && value.type == GLOBREFERENCE && value.value instanceof RuntimeGlob assignedGlob @@ -4068,6 +4070,15 @@ private static boolean isStreamSocketIOHandle(IOHandle handle) { return handle instanceof SocketIO socket && !socket.isDatagramSocket(); } + private static boolean isAcceptedSocket(RuntimeScalar scalar) { + if (scalar == null || scalar.type != GLOBREFERENCE + || !(scalar.value instanceof RuntimeGlob glob) + || !glob.acceptedSocket) { + return false; + } + return hasLiveIo(glob) && isSocketIOHandle(((RuntimeIO) glob.getIO().value).ioHandle); + } + private static boolean isUnstashedIoGlob(RuntimeGlob glob) { return glob.globName == null || !GlobalVariable.existsGlobalIO(glob.globName); } diff --git a/src/main/perl/lib/PerlOnJava/CpanPatches/IO-Async-0.805/SkipUnsupportedSocketTests.patch b/src/main/perl/lib/PerlOnJava/CpanPatches/IO-Async-0.805/SkipUnsupportedSocketTests.patch index 74a05c8333..50009a0ddb 100644 --- a/src/main/perl/lib/PerlOnJava/CpanPatches/IO-Async-0.805/SkipUnsupportedSocketTests.patch +++ b/src/main/perl/lib/PerlOnJava/CpanPatches/IO-Async-0.805/SkipUnsupportedSocketTests.patch @@ -9,16 +9,6 @@ use IO::Async::OS; use Socket qw( ---- t/05notifier-loop.t.orig -+++ t/05notifier-loop.t -@@ -140,6 +140,11 @@ - '$loop->remove decrements notifiers count' ); - } - --is_refcount( $loop, 2, '$loop has refcount 2 finally' ); -+is_refcount( $loop, 2, '$loop has refcount 2 finally' ); - - done_testing; --- t/10loop-poll-io.t.orig +++ t/10loop-poll-io.t @@ -4,5 +4,6 @@ diff --git a/src/test/resources/unit/accepted_socket_callback_lifetime.t b/src/test/resources/unit/accepted_socket_callback_lifetime.t new file mode 100644 index 0000000000..4540cb5fea --- /dev/null +++ b/src/test/resources/unit/accepted_socket_callback_lifetime.t @@ -0,0 +1,37 @@ +use strict; +use warnings; +use IO::Socket::INET; +use Test::More; + +my $listener = IO::Socket::INET->new( + LocalAddr => '127.0.0.1', + LocalPort => 0, + Listen => 1, + ReuseAddr => 1, +); + +plan skip_all => "loopback listener unavailable: $!" unless $listener; +plan tests => 3; + +my $client = IO::Socket::INET->new( + PeerAddr => '127.0.0.1', + PeerPort => $listener->sockport, +) or die "client connect: $!"; + +accept(my $accepted, $listener) or die "accept: $!"; + +my $callback_socket; +sub invoke_callback { + my ($callback, $socket) = @_; + $callback->($socket); +} + +invoke_callback(sub { $callback_socket = $_[0] }, $accepted); +undef $accepted; + +ok defined fileno($callback_socket), + 'accepted socket remains open after callback argument assignment'; +is length(getpeername($callback_socket)), 16, + 'accepted socket retains its IPv4 peer address'; +is length($client->sockname), 16, + 'connected client remains valid'; diff --git a/src/test/resources/unit/pipe_storable_binary_roundtrip.t b/src/test/resources/unit/pipe_storable_binary_roundtrip.t new file mode 100644 index 0000000000..a53e7c9e3a --- /dev/null +++ b/src/test/resources/unit/pipe_storable_binary_roundtrip.t @@ -0,0 +1,24 @@ +use strict; +use warnings; +use Test::More; +use Storable qw(freeze thaw); + +# IO::Async::Channel frames Storable data over pipes. Its binary stream can +# contain high-bit octets, which must remain single octets on an unlayered +# pipe; otherwise the frame length and payload diverge. +pipe(my $reader, my $writer) or die "pipe: $!"; +binmode $reader; +binmode $writer; + +my $frozen = freeze([10, 20]); +print {$writer} pack('I', length $frozen), $frozen or die "write: $!"; +close $writer or die "close writer: $!"; + +read($reader, my $header, 4) == 4 or die "read header: $!"; +my $length = unpack('I', $header); +read($reader, my $payload, $length) == $length or die "read payload: $!"; + +is_deeply(thaw($payload), [10, 20], + 'binary Storable payload round-trips through a pipe'); + +done_testing;