Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
14 changes: 14 additions & 0 deletions src/FreeDSx/Socket/HasSocketOptions.php
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,8 @@ trait HasSocketOptions

private int $timeoutRead = 15;

private int $timeoutHandshake = 5;

private int $timeoutWrite = 0;

/**
Expand Down Expand Up @@ -269,6 +271,18 @@ public function getTimeoutRead(): int
return $this->timeoutRead;
}

public function setTimeoutHandshake(int $seconds): self
{
$this->timeoutHandshake = $seconds;

return $this;
}

public function getTimeoutHandshake(): int
{
return $this->timeoutHandshake;
}

public function setTimeoutWrite(int $seconds): self
{
$this->timeoutWrite = $seconds;
Expand Down
102 changes: 100 additions & 2 deletions src/FreeDSx/Socket/Socket.php
Original file line number Diff line number Diff line change
Expand Up @@ -19,14 +19,19 @@
use FreeDSx\Socket\Tls\Certificate;
use OpenSSLCertificate;

use function error_get_last;
use function fclose;
use function fmod;
use function fread;
use function fwrite;
use function is_array;
use function is_resource;
use function microtime;
use function sprintf;
use function stream_context_create;
use function stream_context_get_options;
use function stream_get_meta_data;
use function stream_select;
use function stream_set_blocking;
use function stream_set_timeout;
use function stream_socket_client;
Expand All @@ -40,6 +45,8 @@
*/
class Socket
{
private const MICROSECONDS_PER_SECOND = 1_000_000;

protected bool $isEncrypted = false;

/**
Expand Down Expand Up @@ -80,6 +87,17 @@ public function __construct(
public function read(bool $block = true): string|false
{
$stream = $this->getStream();

// Bytes decrypted into the TLS buffer are off the wire already.
$buffered = $block && $this->isEncrypted
? $this->readWithoutBlocking($stream)
: '';
if ($buffered !== '') {
return $buffered;
}
if ($block) {
$this->awaitReadable($stream);
}
stream_set_blocking($stream, $block);

$data = fread(
Expand Down Expand Up @@ -108,6 +126,76 @@ public function read(bool $block = true): string|false
: $data;
}

/**
* Take whatever is already buffered without waiting.
*
* @param resource $stream
* @return string Whatever is in the buffer. Empty string when nothing is.
*/
private function readWithoutBlocking($stream): string
{
stream_set_blocking($stream, false);

$data = fread(
$stream,
$this->options->getBufferSize(),
);

stream_set_blocking($stream, true);

return $data === false
? ''
: $data;
}

/**
* Wait for the socket to become readable. This enforces the read timeout that the blocking read would.
*
* Waiting here returns on EINTR and lets the caller run its signal handlers.
*
* @param resource $stream
* @throws IdleTimeoutException
*/
private function awaitReadable($stream): void
{
$timeout = $this->options->getTimeoutRead();
$deadline = $timeout > 0
? microtime(true) + $timeout
: null;

while (true) {
$remaining = $deadline === null
? null
: $deadline - microtime(true);

if ($remaining !== null && $remaining <= 0) {
throw new IdleTimeoutException(sprintf(
'The connection was idle for longer than the read timeout of %d seconds.',
$timeout,
));
}

$read = [$stream];
$write = [];
$except = [];

$ready = @stream_select(
$read,
$write,
$except,
$remaining === null ? null : (int) $remaining,
$remaining === null ? 0 : (int) (fmod($remaining, 1.0) * self::MICROSECONDS_PER_SECOND),
);

if ($ready > 0) {
return;
}
if (!is_resource($stream)) {
return;
}
}
}

/**
* @param resource $stream
*/
Expand Down Expand Up @@ -220,11 +308,21 @@ public function encrypt(bool $encrypt): static
$stream,
true,
);
$result = stream_socket_enable_crypto(
// Bounded on its own, since the read timeout belongs to an idle session and can be far longer.
stream_set_timeout(
$stream,
$this->options->getTimeoutHandshake(),
);

// Suppressed so a peer that stalls the handshake cannot make the process write to stderr.
$result = @stream_socket_enable_crypto(
$stream,
$encrypt,
$this->options->getSslCryptoMethod(),
);
$lastError = error_get_last();

$this->setStreamOpts();
stream_set_blocking(
$stream,
false,
Expand All @@ -234,7 +332,7 @@ public function encrypt(bool $encrypt): static
throw new ConnectionException(sprintf(
'Unable to %s encryption on TCP connection. %s',
$encrypt ? 'enable' : 'disable',
$this->errorMessage,
$lastError['message'] ?? $this->errorMessage,
));
}
$this->isEncrypted = $encrypt;
Expand Down
5 changes: 5 additions & 0 deletions src/FreeDSx/Socket/SocketOptionsInterface.php
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,11 @@ public function getTimeoutConnect(): int;

public function getTimeoutRead(): int;

/**
* Seconds a TLS handshake may take.
*/
public function getTimeoutHandshake(): int;

public function getTimeoutWrite(): int;

public function getWriteTimeoutEnforcer(): WriteTimeoutEnforcerInterface;
Expand Down
40 changes: 40 additions & 0 deletions tests/unit/SocketTest.php
Original file line number Diff line number Diff line change
Expand Up @@ -285,6 +285,46 @@ public function test_it_should_throw_an_idle_timeout_when_a_blocking_read_exceed
$subject->read();
}

public function test_a_blocking_read_dispatches_a_signal_that_arrives_while_it_waits(): void
{
if (!function_exists('pcntl_async_signals')) {
self::markTestSkipped('The pcntl extension is required to deliver a signal mid-read.');
}

[$local, $remote] = $this->createSocketPair();
$subject = new Socket(
$local,
(new SocketOptions())->setTimeoutRead(5),
);

$handled = false;
pcntl_async_signals(true);
pcntl_signal(
SIGALRM,
function () use (&$handled, $remote): void {
$handled = true;
fwrite($remote, 'woke');
},
);
pcntl_alarm(1);

try {
$data = $subject->read();
} finally {
pcntl_alarm(0);
pcntl_signal(SIGALRM, SIG_DFL);
}

self::assertTrue(
$handled,
'The signal handler never ran.',
);
self::assertSame(
'woke',
$data,
);
}

public function test_it_should_not_throw_an_idle_timeout_on_a_non_blocking_read(): void
{
[$local] = $this->createSocketPair();
Expand Down
Loading