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

Nullify JobService

Browse Source
This commit is contained in:
Jordan Dominion
2023-11-24 15:23:34 -05:00
parent da0da4e6eb
commit 280bbc836f
2 changed files with 38 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Tgstation.Server.Host/Jobs/IJobManager.cs
+1 -1
View File
@@ -44,6 +44,6 @@ namespace Tgstation.Server.Host.Jobs
/// <param name="blocking">If the operation should wait until the job exits before completing.</param>
/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
/// <returns>A <see cref="ValueTask{TResult}"/> resulting in the updated <paramref name="job"/> if it was cancelled, <see langword="null"/> if it couldn't be found.</returns>
ValueTask<Job> CancelJob(Job job, User? user, bool blocking, CancellationToken cancellationToken);
ValueTask<Job?> CancelJob(Job job, User? user, bool blocking, CancellationToken cancellationToken);
}
}
src/Tgstation.Server.Host/Jobs/JobService.cs
+37 -44
View File
@@ -21,8 +21,6 @@ using Tgstation.Server.Host.Models;
using Tgstation.Server.Host.Utils;
using Tgstation.Server.Host.Utils.SignalR;
#nullable disable
namespace Tgstation.Server.Host.Jobs
{
/// <inheritdoc cref="IJobService" />
@@ -121,25 +119,19 @@ namespace Tgstation.Server.Host.Jobs
ArgumentNullException.ThrowIfNull(job);
ArgumentNullException.ThrowIfNull(operation);
if (job.StartedBy != null && job.StartedBy.Name == null)
throw new InvalidOperationException("StartedBy User associated with job does not have a Name!");
job.StartedAt = DateTimeOffset.UtcNow;
job.Cancelled = false;
if (job.StartedBy != null)
{
if (!job.StartedBy.Id.HasValue)
throw new InvalidOperationException("StartedBy User associated with job does not have an Id!");
if (job.StartedBy.Name == null)
throw new InvalidOperationException("StartedBy User associated with job does not have a Name!");
}
var originalStartedBy = job.StartedBy;
await databaseContextFactory.UseContext(
async databaseContext =>
{
job.Instance = new Models.Instance
{
Id = job.Instance.Id.Value,
Id = job.Instance.Require(x => x.Id),
};
databaseContext.Instances.Attach(job.Instance);
@@ -149,14 +141,14 @@ namespace Tgstation.Server.Host.Jobs
.GetTgsUser(
dbUser => new User
{
Id = dbUser.Id.Value,
Id = dbUser.Id!.Value,
Name = dbUser.Name,
},
cancellationToken);
job.StartedBy = new User
{
Id = originalStartedBy.Id.Value,
Id = originalStartedBy.Require(x => x.Id),
};
databaseContext.Users.Attach(job.StartedBy);
@@ -177,7 +169,7 @@ namespace Tgstation.Server.Host.Jobs
bool jobShouldStart;
lock (synchronizationLock)
{
jobs.Add(job.Id.Value, jobHandler);
jobs.Add(job.Require(x => x.Id), jobHandler);
jobShouldStart = !noMoreJobsShouldStart;
}
@@ -201,7 +193,7 @@ namespace Tgstation.Server.Host.Jobs
.Jobs
.AsQueryable()
.Where(y => !y.StoppedAt.HasValue)
.Select(y => y.Id.Value)
.Select(y => y.Id!.Value)
.ToListAsync(cancellationToken);
if (badJobIds.Count > 0)
{
@@ -224,7 +216,7 @@ namespace Tgstation.Server.Host.Jobs
/// <inheritdoc />
public Task StopAsync(CancellationToken cancellationToken)
{
List<ValueTask<Job>> joinTasks;
List<ValueTask<Job?>> joinTasks;
lock (addCancelLock)
lock (synchronizationLock)
{
@@ -241,24 +233,25 @@ namespace Tgstation.Server.Host.Jobs
}
/// <inheritdoc />
public async ValueTask<Job> CancelJob(Job job, User user, bool blocking, CancellationToken cancellationToken)
public async ValueTask<Job?> CancelJob(Job job, User? user, bool blocking, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(job);
JobHandler handler;
var jid = job.Require(x => x.Id);
JobHandler? handler;
lock (addCancelLock)
{
lock (synchronizationLock)
if (!jobs.TryGetValue(job.Id.Value, out handler))
if (!jobs.TryGetValue(jid, out handler))
return null;
logger.LogDebug("Cancelling job ID {jobId}...", job.Id.Value);
logger.LogDebug("Cancelling job ID {jobId}...", jid);
handler.Cancel(); // this will ensure the db update is only done once
}
await databaseContextFactory.UseContext(async databaseContext =>
{
var updatedJob = new Job(job.Id.Value);
var updatedJob = new Job(jid);
databaseContext.Jobs.Attach(updatedJob);
var attachedUser = user == null
? await databaseContext
@@ -266,12 +259,12 @@ namespace Tgstation.Server.Host.Jobs
.GetTgsUser(
dbUser => new User
{
Id = dbUser.Id.Value,
Id = dbUser.Id!.Value,
},
cancellationToken)
: new User
{
Id = user.Id.Value,
Id = user.Require(x => x.Id),
};
databaseContext.Users.Attach(attachedUser);
@@ -298,7 +291,7 @@ namespace Tgstation.Server.Host.Jobs
ArgumentNullException.ThrowIfNull(apiResponse);
lock (synchronizationLock)
{
if (!jobs.TryGetValue(apiResponse.Id.Value, out var handler))
if (!jobs.TryGetValue(apiResponse.Require(x => x.Id), out var handler))
return;
apiResponse.Progress = handler.Progress;
apiResponse.Stage = handler.Stage;
@@ -306,18 +299,18 @@ namespace Tgstation.Server.Host.Jobs
}
/// <inheritdoc />
public async ValueTask<bool?> WaitForJobCompletion(Job job, User canceller, CancellationToken jobCancellationToken, CancellationToken cancellationToken)
public async ValueTask<bool?> WaitForJobCompletion(Job job, User? canceller, CancellationToken jobCancellationToken, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(job);
if (!cancellationToken.CanBeCanceled)
throw new ArgumentException("A cancellable CancellationToken should be provided!", nameof(cancellationToken));
JobHandler handler;
JobHandler? handler;
bool noMoreJobsShouldStart;
lock (synchronizationLock)
{
if (!jobs.TryGetValue(job.Id.Value, out handler))
if (!jobs.TryGetValue(job.Require(x => x.Id), out handler))
return null;
noMoreJobsShouldStart = this.noMoreJobsShouldStart;
@@ -326,7 +319,7 @@ namespace Tgstation.Server.Host.Jobs
if (noMoreJobsShouldStart && !handler.Started)
await Extensions.TaskExtensions.InfiniteTask.WaitAsync(cancellationToken);
var cancelTask = ValueTask.FromResult<Job>(null);
var cancelTask = ValueTask.FromResult<Job?>(null);
bool result;
using (jobCancellationToken.Register(() => cancelTask = CancelJob(job, canceller, true, cancellationToken)))
result = await handler.Wait(cancellationToken);
@@ -364,17 +357,18 @@ namespace Tgstation.Server.Host.Jobs
async Task<bool> RunJob(Job job, JobEntrypoint operation, CancellationToken cancellationToken)
#pragma warning restore CA1506
{
using (LogContext.PushProperty(SerilogContextHelper.JobIdContextProperty, job.Id))
var jid = job.Require(x => x.Id);
using (LogContext.PushProperty(SerilogContextHelper.JobIdContextProperty, jid))
try
{
void LogException(Exception ex) => logger.LogDebug(ex, "Job {jobId} exited with error!", job.Id);
void LogException(Exception ex) => logger.LogDebug(ex, "Job {jobId} exited with error!", jid);
var hubUpdatesTask = Task.CompletedTask;
var result = false;
var firstLogHappened = false;
var hubGroupName = JobsHub.HubGroupName(job);
Stopwatch stopwatch = null;
Stopwatch? stopwatch = null;
void QueueHubUpdate(JobResponse update, bool final)
{
void NextUpdate(bool bypassRate)
@@ -386,7 +380,7 @@ namespace Tgstation.Server.Host.Jobs
if (!firstLogHappened)
{
logger.LogTrace("Sending updates for job {id} to hub group {group}", update.Id.Value, hubGroupName);
logger.LogTrace("Sending updates for job {id} to hub group {group}", jid, hubGroupName);
firstLogHappened = true;
}
@@ -397,7 +391,7 @@ namespace Tgstation.Server.Host.Jobs
.ReceiveJobUpdate(update, CancellationToken.None);
}
Stopwatch enteredLock = null;
Stopwatch? enteredLock = null;
try
{
if (!bypassRate && stopwatch != null)
@@ -418,19 +412,18 @@ namespace Tgstation.Server.Host.Jobs
}
}
var jobId = update.Id.Value;
lock (hubUpdateActions)
if (final)
hubUpdateActions.Remove(jobId);
hubUpdateActions.Remove(jid);
else
hubUpdateActions[jobId] = () => NextUpdate(true);
hubUpdateActions[jid] = () => NextUpdate(true);
NextUpdate(false);
}
try
{
void UpdateProgress(string stage, double? progress)
void UpdateProgress(string? stage, double? progress)
{
if (progress.HasValue
&& (progress.Value < 0 || progress.Value > 1))
@@ -442,7 +435,7 @@ namespace Tgstation.Server.Host.Jobs
int? newProgress = progress.HasValue ? (int)Math.Floor(progress.Value * 100) : null;
lock (synchronizationLock)
if (jobs.TryGetValue(job.Id.Value, out var handler))
if (jobs.TryGetValue(jid, out var handler))
{
handler.Stage = stage;
handler.Progress = newProgress;
@@ -492,7 +485,7 @@ namespace Tgstation.Server.Host.Jobs
{
await databaseContextFactory.UseContext(async databaseContext =>
{
var attachedJob = new Job(job.Id.Value);
var attachedJob = new Job(jid);
databaseContext.Jobs.Attach(attachedJob);
attachedJob.StoppedAt = DateTimeOffset.UtcNow;
@@ -517,7 +510,7 @@ namespace Tgstation.Server.Host.Jobs
.Include(x => x.Instance)
.Include(x => x.StartedBy)
.Include(x => x.CancelledBy)
.Where(dbJob => dbJob.Id == job.Id.Value)
.Where(dbJob => dbJob.Id == jid)
.FirstAsync(CancellationToken.None);
QueueHubUpdate(finalJob.ToApi(), true);
});
@@ -525,7 +518,7 @@ namespace Tgstation.Server.Host.Jobs
catch
{
lock (hubUpdateActions)
hubUpdateActions.Remove(job.Id.Value);
hubUpdateActions.Remove(jid);
throw;
}
@@ -545,8 +538,8 @@ namespace Tgstation.Server.Host.Jobs
{
lock (synchronizationLock)
{
var handler = jobs[job.Id.Value];
jobs.Remove(job.Id.Value);
var handler = jobs[jid];
jobs.Remove(jid);
handler.Dispose();
}
}