cyberpresidentvanellope/tgstation-server
1
0
Fork 0
mirror of https://github.com/tgstation/tgstation-server.git synced 2026-08-27 15:07:03 +01:00
Code Issues Packages Projects Releases Wiki Activity

Make session topic calls exclusive FIFO

Browse Source
This commit is contained in:
Jordan Dominion
2023-07-10 15:30:51 -04:00
parent a4362336fa
commit b6ced2b894
1 changed files with 62 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Tgstation.Server.Host/Components/Session/SessionController.cs
+62 -38
View File
@@ -107,21 +107,6 @@ namespace Tgstation.Server.Host.Components.Session
/// </summary>
public ReattachInformation ReattachInformation { get; }
/// <summary>
/// The <see cref="TaskCompletionSource"/> that completes when DD makes it's first bridge request.
/// </summary>
readonly TaskCompletionSource initialBridgeRequestTcs;
/// <summary>
/// The <see cref="Instance"/> metadata.
/// </summary>
readonly Api.Models.Instance metadata;
/// <summary>
/// A <see cref="CancellationTokenSource"/> used for the topic send operation made on reattaching.
/// </summary>
readonly CancellationTokenSource reattachTopicCts;
/// <summary>
/// The <see cref="global::Byond.TopicSender.ITopicClient"/> for the <see cref="SessionController"/>.
/// </summary>
@@ -157,6 +142,26 @@ namespace Tgstation.Server.Host.Components.Session
/// </summary>
readonly IAsyncDelayer asyncDelayer;
/// <summary>
/// The <see cref="TaskCompletionSource"/> that completes when DD makes it's first bridge request.
/// </summary>
readonly TaskCompletionSource initialBridgeRequestTcs;
/// <summary>
/// The <see cref="FifoSemaphore"/> used to prevent concurrent calls into /world/Topic().
/// </summary>
readonly FifoSemaphore topicSendSemaphore;
/// <summary>
/// The <see cref="Instance"/> metadata.
/// </summary>
readonly Api.Models.Instance metadata;
/// <summary>
/// A <see cref="CancellationTokenSource"/> used for the topic send operation made on reattaching.
/// </summary>
readonly CancellationTokenSource reattachTopicCts;
/// <summary>
/// <see langword="lock"/> <see cref="object"/> for port updates and <see cref="disposed"/>.
/// </summary>
@@ -282,6 +287,8 @@ namespace Tgstation.Server.Host.Components.Session
// Worth further investigation as to if that sequence of events is a reliable crash vector and opening a BYOND bug if it is
initialBridgeRequestTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
reattachTopicCts = new CancellationTokenSource();
topicSendSemaphore = new FifoSemaphore();
synchronizationLock = new object();
if (apiValidate || DMApiAvailable)
@@ -329,6 +336,11 @@ namespace Tgstation.Server.Host.Components.Session
}
Logger.LogTrace("Disposing...");
// yield then acquire the topic semaphore to prevent new calls from starting
await Task.Yield();
(await topicSendSemaphore.Lock(CancellationToken.None)).Dispose(); // DCT: None available
if (!released)
{
process.Terminate();
@@ -345,6 +357,8 @@ namespace Tgstation.Server.Host.Components.Session
if (!released)
await Lifetime; // finish the async callback
topicSendSemaphore.Dispose();
}
/// <inheritdoc />
@@ -387,7 +401,7 @@ namespace Tgstation.Server.Host.Components.Session
{
ArgumentNullException.ThrowIfNull(parameters);
if (Lifetime.IsCompleted)
if (Lifetime.IsCompleted || disposed)
{
Logger.LogWarning(
"Attempted to send a command to an inactive SessionController: {commandType}",
@@ -960,35 +974,45 @@ namespace Tgstation.Server.Host.Components.Session
/// <returns>A <see cref="Task{TResult}"/> resulting in the <see cref="CombinedTopicResponse"/> of the topic request.</returns>
async Task<CombinedTopicResponse> SendRawTopic(string queryString, bool priority, CancellationToken cancellationToken)
{
if (disposed)
{
Logger.LogWarning(
"Attempted to send a topic on a disposed SessionController");
return null;
}
var targetPort = ReattachInformation.Port;
var killedOrRebootedTask = Task.WhenAny(Lifetime, OnReboot);
global::Byond.TopicSender.TopicResponse byondResponse = null;
var firstSend = true;
const int PrioritySendAttempts = 5;
var endpoint = new IPEndPoint(IPAddress.Loopback, targetPort);
for (var i = PrioritySendAttempts - 1; i >= 0 && (priority || firstSend); --i)
try
{
firstSend = false;
if (!killedOrRebootedTask.IsCompleted)
byondResponse = await byondTopicSender.SendTopic(
endpoint,
queryString,
cancellationToken);
break;
}
catch (Exception ex) when (!cancellationToken.IsCancellationRequested)
{
Logger.LogWarning(ex, "SendTopic exception!{retryDetails}", priority ? $" {i} attempts remaining." : String.Empty);
if (priority && i > 0)
using (await topicSendSemaphore.Lock(cancellationToken))
{
const int PrioritySendAttempts = 5;
var endpoint = new IPEndPoint(IPAddress.Loopback, targetPort);
for (var i = PrioritySendAttempts - 1; i >= 0 && (priority || firstSend); --i)
try
{
var delayTask = asyncDelayer.Delay(TimeSpan.FromSeconds(2), cancellationToken);
await Task.WhenAny(killedOrRebootedTask, delayTask);
firstSend = false;
if (!killedOrRebootedTask.IsCompleted)
byondResponse = await byondTopicSender.SendTopic(
endpoint,
queryString,
cancellationToken);
break;
}
}
catch (Exception ex) when (!cancellationToken.IsCancellationRequested)
{
Logger.LogWarning(ex, "SendTopic exception!{retryDetails}", priority ? $" {i} attempts remaining." : String.Empty);
if (priority && i > 0)
{
var delayTask = asyncDelayer.Delay(TimeSpan.FromSeconds(2), cancellationToken);
await Task.WhenAny(killedOrRebootedTask, delayTask);
}
}
}
if (byondResponse == null)
{