Mirrored_Repos/wabbajack
1
0
Fork 0
mirror of https://github.com/wabbajack-tools/wabbajack.git synced 2024-08-30 18:42:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

WorkQueue taking of queue swapped to TryTake

Browse Source 
Helps shut down threads during quiet periods
This commit is contained in:
Justin Swanson
2020-01-11 17:17:45 -06:00
parent e2b2ea1727
commit 0106d020f9
1 changed files with 13 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
Wabbajack.Common/WorkQueue.cs
View File

@ -5,11 +5,13 @@ using System.Linq;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;
using DynamicData;
using Wabbajack.Common.StatusFeed;
[assembly: InternalsVisibleTo("Wabbajack.Test")]
namespace Wabbajack.Common
{
public class WorkQueue : IDisposable
@ -29,7 +31,7 @@ namespace Wabbajack.Common
public IObservable<CPUStatus> Status => _Status;
private int _nextCpuID = 1; // Start at 1, as 0 is "Unassigned"
private List<(int CpuID, Task Task)> _tasks = new List<(int CpuID, Task Task)>();
internal List<(int CpuID, Task Task)> _tasks = new List<(int CpuID, Task Task)>();
public int DesiredNumWorkers { get; private set; } = 0;
private CancellationTokenSource _shutdown = new CancellationTokenSource();
@ -48,6 +50,8 @@ namespace Wabbajack.Common
private readonly Subject<IObservable<int>> _activeNumThreadsObservable = new Subject<IObservable<int>>();
internal const int PollMS = 200;
/// <summary>
/// Creates a WorkQueue with the given number of threads
/// </summary>
@ -68,7 +72,6 @@ namespace Wabbajack.Common
// Select the latest driving observable
.Select(x => x ?? Observable.Return(Environment.ProcessorCount))
.Switch()
.StartWith(Environment.ProcessorCount)
.DistinctUntilChanged()
// Add new threads if it increases
.SelectTask(AddNewThreadsIfNeeded)
@ -116,17 +119,23 @@ namespace Wabbajack.Common
{
Report("Waiting", 0, false);
if (_shutdown.IsCancellationRequested) return;
Func<Task> f;
bool got;
try
{
f = Queue.Take(_shutdown.Token);
got = Queue.TryTake(out f, PollMS, _shutdown.Token);
}
catch (Exception)
{
throw new OperationCanceledException();
}
await f();
if (got)
{
await f();
}
// Check if we're currently trimming threads
if (DesiredNumWorkers >= _tasks.Count) continue;