Skip to content
Open
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
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,28 @@ public Task StressTest()
return StressTestCore(CreateServer, CreateClient);
}

[Fact]
public Task RequestTimeout()
{
static TcpServer CreateServer(ILocalMember member, EndPoint address, TimeSpan timeout) => new(address, 2, member, NullLoggerFactory.Instance)
{
MemoryAllocator = MemoryAllocator<byte>.Default,
ReceiveTimeout = timeout,
TransmissionBlockSize = 65535,
GracefulShutdownTimeout = 2000
};

static TcpClient CreateClient(EndPoint address, ILocalMember member, TimeSpan timeout) => new(member, address)
{
MemoryAllocator = MemoryAllocator<byte>.Default,
RequestTimeout = timeout,
ConnectTimeout = timeout,
TransmissionBlockSize = 65535,
};

return RequestTimeoutTest(CreateServer, CreateClient);
}

[Theory]
[InlineData(true)]
[InlineData(false)]
Expand Down Expand Up @@ -393,4 +415,4 @@ private static RaftCluster.TcpConfiguration CreateConfiguration(int port, bool c

return result;
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ private sealed class LocalMember : Assert, ILocalMember
internal ReceiveEntriesBehavior Behavior;
internal byte[] ReceivedConfiguration = [];
internal long ReceivedConfigurationVersion = -1L;
internal TimeSpan VoteDelay;
private readonly ClusterMemberId localId = Random.Shared.Next<ClusterMemberId>();

internal LocalMember(bool smallAmountOfMetadata = false)
Expand Down Expand Up @@ -125,13 +126,17 @@ async ValueTask<bool> ILocalMember.InstallConfigurationAsync<TConfiguration>(lon
return true;
}

ValueTask<Result<bool>> ILocalMember.VoteAsync(ClusterMemberId sender, long term, long lastLogIndex, long lastLogTerm, CancellationToken token)
async ValueTask<Result<bool>> ILocalMember.VoteAsync(ClusterMemberId sender, long term, long lastLogIndex, long lastLogTerm, CancellationToken token)
{
True(token.CanBeCanceled);
Equal(42L, term);
Equal(1L, lastLogIndex);
Equal(56L, lastLogTerm);
return ValueTask.FromResult<Result<bool>>(new() { Term = 43L, Value = true });

if (VoteDelay > TimeSpan.Zero)
await Task.Delay(VoteDelay, token);

return new() { Term = 43L, Value = true };
}

ValueTask<Result<PreVoteResult>> ILocalMember.PreVoteAsync(ClusterMemberId sender, long term, long lastLogIndex, long lastLogTerm, CancellationToken token)
Expand Down Expand Up @@ -212,6 +217,27 @@ private protected async Task StressTestCore(ServerFactory serverFactory, ClientF
});
}

private protected async Task RequestTimeoutTest(ServerFactory serverFactory, ClientFactory clientFactory)
{
var serverAddr = new IPEndPoint(IPAddress.Loopback, 3789);
var serverTimeout = TimeSpan.FromMilliseconds(50D);
var member = new LocalMember { VoteDelay = DefaultTimeout };
await using var server = serverFactory(member, serverAddr, serverTimeout);
await server.StartAsync(TestToken);

using (var client = clientFactory(serverAddr, member, DefaultTimeout))
{
await ThrowsAsync<MemberUnavailableException>(
() => client.As<IRaftClusterMember>().VoteAsync(42L, 1L, 56L, TestToken));
}

member.VoteDelay = TimeSpan.Zero;
using var nextClient = clientFactory(serverAddr, member, DefaultTimeout);
var result = await nextClient.As<IRaftClusterMember>().VoteAsync(42L, 1L, 56L, TestToken);
True(result.Value);
Equal(43L, result.Term);
}

private protected async Task MetadataRequestResponseTest(ServerFactory serverFactory, ClientFactory clientFactory, bool smallAmountOfMetadata)
{
var timeout = DefaultTimeout;
Expand Down Expand Up @@ -494,4 +520,4 @@ private protected async Task LeadershipCore(Func<int, bool, IPersistentState, Ra

protected static WriteAheadLog CreateWal()
=> new(new() { Location = GetTempPath() }, IStateMachine.CreateNoOp());
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -131,6 +131,11 @@ private async void HandleConnection(Socket remoteClient)
{
await ProcessRequestAsync(messageType, protocol, timeoutSource.Token).ConfigureAwait(false);
}
catch (OperationCanceledException e) when (e.CausedByTimeout(timeoutSource))

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

try-catch-finally block is redundant at all. You can do the same thing as in GenericServer:

await ProcessRequestAsync(messageType, protocol, timeoutSource.Token).ConfigureAwait(false);
                
// reset cancellation token
await timeoutSource.DisposeAsync().ConfigureAwait(false);
timeoutSource = default;

and leave the existing handler as-is:

catch (OperationCanceledException e)
{
    // if lifecycleToken is canceled then shutdown socket gracefully without logging
    if (e.CausedByTimeout(timeoutSource))
        logger.RequestTimedOut(clientAddress, e);
}

{
logger.RequestTimedOut(clientAddress, e);
break;
}
finally
{
// reset cancellation token
Expand All @@ -142,11 +147,9 @@ private async void HandleConnection(Socket remoteClient)
{
logger.ConnectionWasResetByClient(clientAddress);
}
catch (OperationCanceledException e)
catch (OperationCanceledException)
{
// if lifecycleToken is canceled then shutdown socket gracefully without logging
if (e.CausedByTimeout(timeoutSource))
logger.RequestTimedOut(clientAddress, e);
// shutdown socket gracefully without logging
}
catch (Exception e)
{
Expand Down Expand Up @@ -267,4 +270,4 @@ protected override async ValueTask DisposeAsyncCore()
logger.TcpGracefulShutdownFailed(GracefulShutdownTimeout);
}
}
}
}
Loading