Merge pull request #498 from Cyberboss/JobMananger

Job Manager

src/Tgstation.Server.Host/Core/Application.cs

@@ -1,11 +1,11 @@
-using Cyberboss.AspNetCore.AsyncInitializer;
-using Microsoft.AspNetCore.Authentication.JwtBearer;
+using Microsoft.AspNetCore.Authentication.JwtBearer;
 using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.AspNetCore.Identity;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
 using Microsoft.IdentityModel.Tokens;
 using System;
 using System.Globalization;
@@ -34,16 +34,16 @@ namespace Tgstation.Server.Host.Core
 		readonly IConfiguration configuration;
 
 		/// <summary>
-		/// The <see cref="IHostingEnvironment"/> for the <see cref="Application"/>
+		/// The <see cref="Microsoft.AspNetCore.Hosting.IHostingEnvironment"/> for the <see cref="Application"/>
 		/// </summary>
-		readonly IHostingEnvironment hostingEnvironment;
+		readonly Microsoft.AspNetCore.Hosting.IHostingEnvironment hostingEnvironment;
 
 		/// <summary>
 		/// Construct an <see cref="Application"/>
 		/// </summary>
 		/// <param name="configuration">The value of <see cref="configuration"/></param>
 		/// <param name="hostingEnvironment">The value of <see cref="hostingEnvironment"/></param>
-		public Application(IConfiguration configuration, IHostingEnvironment hostingEnvironment)
+		public Application(IConfiguration configuration, Microsoft.AspNetCore.Hosting.IHostingEnvironment hostingEnvironment)
 		{
 			this.configuration = configuration ?? throw new ArgumentNullException(nameof(configuration));
 			this.hostingEnvironment = hostingEnvironment ?? throw new ArgumentNullException(nameof(hostingEnvironment));
@@ -131,6 +131,10 @@ namespace Tgstation.Server.Host.Core
 			services.AddSingleton<IPasswordHasher<User>, PasswordHasher<User>>();
 			services.AddSingleton<ITokenFactory, TokenFactory>();
 			services.AddSingleton<ISystemIdentityFactory, SystemIdentityFactory>();
+
+			services.AddSingleton<JobManager>();
+			services.AddSingleton<IJobManager>(x => x.GetRequiredService<JobManager>());
+			services.AddSingleton<IHostedService>(x => x.GetRequiredService<JobManager>());
 		}
 
 		/// <summary>
@@ -144,13 +148,7 @@ namespace Tgstation.Server.Host.Core
 
 			if (hostingEnvironment.IsDevelopment())
 				applicationBuilder.UseDeveloperExceptionPage();
-
-			applicationBuilder.UseAsyncInitialization(async (cancellationToken) =>
-			{
-				using (var scope = applicationBuilder.ApplicationServices.CreateScope())
-					await scope.ServiceProvider.GetRequiredService<IDatabaseContext>().Initialize(cancellationToken).ConfigureAwait(false);
-			});
-
+			
 			applicationBuilder.UseAuthentication();
 			applicationBuilder.UseMvc();
 		}

src/Tgstation.Server.Host/Core/IJobManager.cs

@@ -0,0 +1,33 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Tgstation.Server.Host.Models;
+
+namespace Tgstation.Server.Host.Core
+{
+    interface IJobManager
+	{
+		/// <summary>
+		/// Registers a given <see cref="Job"/> and begins running it
+		/// </summary>
+		/// <param name="job">The <see cref="Job"/></param>
+		/// <param name="operation">The operation to run</param>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation</param>
+		/// <returns>A <see cref="Task"/> representing a running operation</returns>
+		Task RegisterOperation(Job job, Func<CancellationToken, Task> operation, CancellationToken cancellationToken);
+
+		/// <summary>
+		/// Wait for a given <paramref name="job"/> to complete
+		/// </summary>
+		/// <param name="job">The <see cref="Job"/> to wait for</param>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation</param>
+		/// <returns>A <see cref="Task"/> representing a running operation</returns>
+		Task WaitForJob(Job job, CancellationToken cancellationToken);
+
+		/// <summary>
+		/// Cancels a give <paramref name="job"/>
+		/// </summary>
+		/// <param name="job">The <see cref="Job"/> to cancel</param>
+		void CancelJob(Job job);
+    }
+}

src/Tgstation.Server.Host/Core/JobHandler.cs

@@ -0,0 +1,63 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Tgstation.Server.Host.Core
+{
+	/// <summary>
+	/// Class for pairing <see cref="Task"/>s with <see cref="CancellationTokenSource"/>s
+	/// </summary>
+	sealed class JobHandler : IDisposable
+	{
+		/// <summary>
+		/// The <see cref="Task"/> being run
+		/// </summary>
+		readonly Task task;
+		/// <summary>
+		/// The <see cref="CancellationTokenSource"/> for <see cref="task"/>
+		/// </summary>
+		readonly CancellationTokenSource cancellationTokenSource;
+
+		/// <summary>
+		/// Construct a <see cref="JobHandler"/>
+		/// </summary>
+		/// <param name="task">The value of <see cref="Task"/></param>
+		/// <param name="cancellationTokenSource">The value of <see cref="cancellationTokenSource"/></param>
+		JobHandler(Task task, CancellationTokenSource cancellationTokenSource)
+		{
+			this.task = task;
+			this.cancellationTokenSource = cancellationTokenSource;
+		}
+
+		/// <inehritdoc />
+		public void Dispose() => cancellationTokenSource.Dispose();
+
+		/// <summary>
+		/// Wait for <see cref="task"/> to complete
+		/// </summary>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation</param>
+		/// <returns>A <see cref="Task"/> representing the running operation</returns>
+		public async Task Wait(CancellationToken cancellationToken)
+		{
+			TaskCompletionSource<object> tcs = new TaskCompletionSource<object>();
+			using (cancellationToken.Register(() => tcs.SetCanceled()))
+				await Task.WhenAny(tcs.Task, task).ConfigureAwait(false);
+		}
+
+		/// <summary>
+		/// Cancels <see cref="task"/>
+		/// </summary>
+		public void Cancel() => cancellationTokenSource.Cancel();
+
+		/// <summary>
+		/// Create a <see cref="JobHandler"/>
+		/// </summary>
+		/// <param name="job">A <see cref="Func{T, TResult}"/> taking a <see cref="CancellationToken"/> and returning a <see cref="Task"/> that the <see cref="JobHandler"/> will wrap</param>
+		/// <returns>A new <see cref="JobHandler"/></returns>
+		public static JobHandler Create(Func<CancellationToken, Task> job)
+		{
+			var cts = new CancellationTokenSource();
+			return new JobHandler(job(cts.Token), cts);
+		}
+	}
+}

src/Tgstation.Server.Host/Core/JobManager.cs

@@ -0,0 +1,143 @@
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Tgstation.Server.Host.Models;
+
+namespace Tgstation.Server.Host.Core
+{
+	/// <inheritdoc />
+	sealed class JobManager : IHostedService, IJobManager
+	{
+		/// <summary>
+		/// The <see cref="IServiceProvider"/> for the <see cref="JobManager"/>
+		/// </summary>
+		readonly IServiceProvider serviceProvider;
+		/// <summary>
+		/// <see cref="Dictionary{TKey, TValue}"/> of <see cref="Api.Models.Internal.Job.Id"/> to running <see cref="JobHandler"/>s
+		/// </summary>
+		readonly Dictionary<long, JobHandler> jobs;
+
+		/// <summary>
+		/// Construct a <see cref="JobManager"/>
+		/// </summary>
+		/// <param name="serviceProvider">The value of <see cref="serviceProvider"/></param>
+		public JobManager(IServiceProvider serviceProvider)
+		{
+			this.serviceProvider = serviceProvider ?? throw new ArgumentNullException(nameof(serviceProvider));
+			jobs = new Dictionary<long, JobHandler>();
+		}
+
+		/// <summary>
+		/// Gets the <see cref="JobHandler"/> for a given <paramref name="job"/> if it exists
+		/// </summary>
+		/// <param name="job">The <see cref="Job"/> to get the <see cref="JobHandler"/> for</param>
+		/// <returns>The <see cref="JobHandler"/></returns>
+		JobHandler CheckGetJob(Job job)
+		{
+			lock (this)
+			{
+				if (!jobs.TryGetValue(job.Id, out JobHandler jobHandler))
+					throw new InvalidOperationException("Job not running!");
+				return jobHandler;
+			}
+		}
+
+		/// <summary>
+		/// Runner for <see cref="JobHandler"/>s
+		/// </summary>
+		/// <param name="job">The <see cref="Job"/> being run</param>
+		/// <param name="operation">The operation for the <paramref name="job"/></param>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation</param>
+		/// <returns>A <see cref="Task"/> representing the running operation</returns>
+		async Task RunJob(Job job, Func<CancellationToken, Task> operation, CancellationToken cancellationToken)
+		{
+			bool cancelled;
+			try
+			{
+				await operation(cancellationToken).ConfigureAwait(false);
+				cancelled = false;
+			}
+			catch (OperationCanceledException)
+			{
+				cancelled = true;
+			}
+
+			using (var scope = serviceProvider.CreateScope())
+			{
+				var databaseContext = scope.ServiceProvider.GetRequiredService<IDatabaseContext>();
+				job = new Job { Id = job.Id };
+				databaseContext.Jobs.Attach(job);
+				if (cancelled)
+					job.Cancelled = true;
+				job.StoppedAt = DateTimeOffset.Now;
+				await databaseContext.Save(default).ConfigureAwait(false);
+			}
+		}
+
+		/// <inheritdoc />
+		public async Task RegisterOperation(Job job, Func<CancellationToken, Task> operation, CancellationToken cancellationToken)
+		{
+			using (var scope = serviceProvider.CreateScope())
+			{
+				var databaseContext = scope.ServiceProvider.GetRequiredService<IDatabaseContext>();
+				job.StartedAt = DateTimeOffset.Now;
+				databaseContext.Jobs.Add(job);
+				await databaseContext.Save(cancellationToken).ConfigureAwait(false);
+				var jobHandler = JobHandler.Create(x => RunJob(job, operation, x));
+				lock (this)
+					jobs.Add(job.Id, jobHandler);
+			}
+		}
+
+		/// <inheritdoc />
+		public async Task StartAsync(CancellationToken cancellationToken)
+		{
+			using (var scope = serviceProvider.CreateScope())
+			{
+				var databaseContext = scope.ServiceProvider.GetRequiredService<IDatabaseContext>();
+				await databaseContext.Initialize(cancellationToken).ConfigureAwait(false);
+
+				//mark all jobs as cancelled
+				var enumerator = await databaseContext.Jobs.Where(y => !y.Cancelled && y.StoppedAt == null).Select(y => y.Id).ToAsyncEnumerable().ToList(cancellationToken).ConfigureAwait(false);
+				foreach(var I in enumerator)
+				{
+					var job = new Job { Id = I };
+					databaseContext.Jobs.Attach(job);
+					job.Cancelled = true;
+				}
+				await databaseContext.Save(cancellationToken).ConfigureAwait(false);
+			}
+		}
+
+		/// <inheritdoc />
+		public async Task StopAsync(CancellationToken cancellationToken)
+		{
+			var joinTasks = jobs.Select(x =>
+			{
+				x.Value.Cancel();
+				return x.Value.Wait(cancellationToken);
+			});
+			await Task.WhenAll(joinTasks).ConfigureAwait(false);
+			foreach (var job in jobs)
+				job.Value.Dispose();
+			jobs.Clear();
+		}
+
+		/// <inheritdoc />
+		public async Task WaitForJob(Job job, CancellationToken cancellationToken)
+		{
+			var handler = CheckGetJob(job);
+			await handler.Wait(cancellationToken).ConfigureAwait(false);
+			lock (this)
+				jobs.Remove(job.Id);
+			handler.Dispose();
+		}
+
+		/// <inheritdoc />
+		public void CancelJob(Job job) => CheckGetJob(job).Cancel();
+	}
+}

src/Tgstation.Server.Host/Models/IDatabaseContext.cs

@@ -15,10 +15,15 @@ namespace Tgstation.Server.Host.Models
 		DbSet<User> Users { get; }
 
 		/// <summary>
-		/// The <see cref="Instances"/>s in the <see cref="IDatabaseContext"/>
+		/// The <see cref="Instance"/>s in the <see cref="IDatabaseContext"/>
 		/// </summary>
 		DbSet<Instance> Instances { get; }
 
+		/// <summary>
+		/// The <see cref="Job"/>s in the <see cref="IDatabaseContext"/>
+		/// </summary>
+		DbSet<Job> Jobs { get; }
+
 		/// <summary>
 		/// Get the <see cref="ServerSettings"/> in the <see cref="IDatabaseContext"/>
 		/// </summary>

src/Tgstation.Server.Host/Tgstation.Server.Host.csproj

@@ -34,7 +34,6 @@
 
   <ItemGroup>
     <PackageReference Include="Byond.TopicSender" Version="1.1.0.1" />
-    <PackageReference Include="Cyberboss.AspNetCore.AsyncInitializer" Version="1.1.0" />
     <PackageReference Include="LibGit2Sharp" Version="0.26.0-preview-0017" />
     <PackageReference Include="Microsoft.AspNetCore" Version="2.1.0-preview2-final" />
     <PackageReference Include="Microsoft.AspNetCore.Authentication.JwtBearer" Version="2.1.0-preview2-final" />

src/Tgstation.Server.Host/appsettings.json

@@ -18,7 +18,7 @@
     }
   },
   "Database": {
-    "DatabaseType": "Sqlite",
-    "ConnectionString": "Fake"
+    "DatabaseType": "SqlServer",
+    "ConnectionString": "Data Source=(local);Initial Catalog=TGS;Integrated Security=True"
   }
 }