CPU status display refactored to Rx/DynamicData

Increased the frequency of status updates to 250 milliseconds.  I believe it's still low CPU usage /w the EnsureUniqueChanges call, but we can dial it back or adjust if someone else finds otherwise

Wabbajack/AppState.cs

@@ -8,6 +8,7 @@ using System.IO;
 using System.IO.Compression;
 using System.Linq;
 using System.Net.Http;
+using System.Reactive.Subjects;
 using System.Reactive.Disposables;
 using System.Reactive.Linq;
 using System.Reflection;
@@ -19,6 +20,8 @@ using System.Windows.Threading;
 using Wabbajack.Common;
 using Wabbajack.NexusApi;
 using Wabbajack.UI;
+using DynamicData;
+using DynamicData.Binding;
 
 namespace Wabbajack
 {
@@ -36,7 +39,8 @@ namespace Wabbajack
         private readonly BitmapImage _wabbajackLogo = UIUtils.BitmapImageFromResource("Wabbajack.UI.banner.png");
         public readonly BitmapImage _noneImage = UIUtils.BitmapImageFromResource("Wabbajack.UI.none.jpg");
 
-        public volatile bool Dirty;
+        private readonly Subject<CPUStatus> _statusSubject = new Subject<CPUStatus>();
+        public ObservableCollectionExtended<CPUStatus> Status { get; } = new ObservableCollectionExtended<CPUStatus>();
 
         private ModList _ModList;
         public ModList ModList { get => _ModList; private set => this.RaiseAndSetIfChanged(ref _ModList, value); }
@@ -83,7 +87,6 @@ namespace Wabbajack
             }
 
             Mode = mode;
-            Dirty = false;
 
             this.OpenReadmeCommand = ReactiveCommand.Create(
                 execute: this.OpenReadmeWindow,
@@ -159,6 +162,22 @@ namespace Wabbajack
                 .Subscribe(_ => _slideShow.UpdateSlideShowItem())
                 .DisposeWith(this.CompositeDisposable);
 
+            // Initialize work queue
+            WorkQueue.Init(
+                report_function: (id, msg, progress) => this._statusSubject.OnNext(new CPUStatus() { ID = id, Msg = msg, Progress = progress }),
+                report_queue_size: (max, current) => this.SetQueueSize(max, current));
+            // Compile progress updates and populate ObservableCollection
+            this._statusSubject
+                .ObserveOn(RxApp.TaskpoolScheduler)
+                .ToObservableChangeSet(x => x.ID)
+                .Batch(TimeSpan.FromMilliseconds(250))
+                .EnsureUniqueChanges()
+                .ObserveOn(RxApp.MainThreadScheduler)
+                .Sort(SortExpressionComparer<CPUStatus>.Ascending(s => s.ID), SortOptimisations.ComparesImmutableValuesOnly)
+                .Bind(this.Status)
+                .Subscribe()
+                .DisposeWith(this.CompositeDisposable);
+
             slideshowThread = new Thread(UpdateLoop)
             {
                 Priority = ThreadPriority.BelowNormal,
@@ -170,7 +189,6 @@ namespace Wabbajack
         public DateTime lastSlideShowUpdate = new DateTime();
 
         public ObservableCollection<string> Log { get; } = new ObservableCollection<string>();
-        public ObservableCollection<CPUStatus> Status { get; } = new ObservableCollection<CPUStatus>();
 
         private string _Location;
         public string Location { get => _Location; set => this.RaiseAndSetIfChanged(ref _Location, value); }
@@ -198,7 +216,6 @@ namespace Wabbajack
         private int _queueProgress;
         public int QueueProgress { get => _queueProgress; set => this.RaiseAndSetIfChanged(ref _queueProgress, value); }
 
-        private List<CPUStatus> InternalStatus { get; } = new List<CPUStatus>();
         public string LogFile { get; }
 
         private void ExecuteChangePath()
@@ -348,21 +365,6 @@ namespace Wabbajack
         {
             while (Running)
             {
-                if (Dirty)
-                    lock (InternalStatus)
-                    {
-                        CPUStatus[] data = InternalStatus.ToArray();
-                        Application.Current.Dispatcher.Invoke(() =>
-                        {
-                            for (var idx = 0; idx < data.Length; idx += 1)
-                                if (idx >= Status.Count)
-                                    Status.Add(data[idx]);
-                                else if (Status[idx] != data[idx])
-                                    Status[idx] = data[idx];
-                        });
-                        Dirty = false;
-                    }
-
                 if (_slideShow.SlidesQueue.Any())
                 {
                     if (DateTime.Now - lastSlideShowUpdate > TimeSpan.FromSeconds(10))
@@ -381,17 +383,6 @@ namespace Wabbajack
             Application.Current.Dispatcher.Invoke(() => Log.Add(msg));
         }
 
-        public void SetProgress(int id, string msg, int progress)
-        {
-            lock (InternalStatus)
-            {
-                Dirty = true;
-                while (id >= InternalStatus.Count) InternalStatus.Add(new CPUStatus());
-
-                InternalStatus[id] = new CPUStatus { ID = id, Msg = msg, Progress = progress };
-            }
-        }
-
         public void SetQueueSize(int max, int current)
         {
             if (max == 0)

Wabbajack/Extensions/ReactiveUIExt.cs

@@ -1,6 +1,9 @@
 using System;
+using System.Linq;
 using System.Linq.Expressions;
 using System.Reactive.Linq;
+using DynamicData;
+using DynamicData.Kernel;
 using ReactiveUI;
 
 namespace Wabbajack
@@ -20,5 +23,54 @@ namespace Wabbajack
         {
             return source.Where(u => u != null);
         }
+
+        /// These snippets were provided by RolandPheasant (author of DynamicData)
+        /// They'll be going into the official library at some point, but are here for now.
+        #region Dynamic Data EnsureUniqueChanges        
+        public static IObservable<IChangeSet<TObject, TKey>> EnsureUniqueChanges<TObject, TKey>(this IObservable<IChangeSet<TObject, TKey>> source)
+        {
+            return source.Select(EnsureUniqueChanges);
+        }
+
+        public static IChangeSet<TObject, TKey> EnsureUniqueChanges<TObject, TKey>(this IChangeSet<TObject, TKey> input)
+        {
+            var changes = input
+                .GroupBy(kvp => kvp.Key)
+                .Select(g => g.Aggregate(Optional<Change<TObject, TKey>>.None, Reduce))
+                .Where(x => x.HasValue)
+                .Select(x => x.Value);
+
+            return new ChangeSet<TObject, TKey>(changes);
+        }
+
+
+        internal static Optional<Change<TObject, TKey>> Reduce<TObject, TKey>(Optional<Change<TObject, TKey>> previous, Change<TObject, TKey> next)
+        {
+            if (!previous.HasValue)
+            {
+                return next;
+            }
+
+            var previousValue = previous.Value;
+
+            switch (previousValue.Reason)
+            {
+                case ChangeReason.Add when next.Reason == ChangeReason.Remove:
+                    return Optional<Change<TObject, TKey>>.None;
+
+                case ChangeReason.Remove when next.Reason == ChangeReason.Add:
+                    return new Change<TObject, TKey>(ChangeReason.Update, next.Key, next.Current, previousValue.Current, next.CurrentIndex, previousValue.CurrentIndex);
+
+                case ChangeReason.Add when next.Reason == ChangeReason.Update:
+                    return new Change<TObject, TKey>(ChangeReason.Add, next.Key, next.Current, next.CurrentIndex);
+
+                case ChangeReason.Update when next.Reason == ChangeReason.Update:
+                    return new Change<TObject, TKey>(ChangeReason.Update, previousValue.Key, next.Current, previousValue.Previous, next.CurrentIndex, previousValue.PreviousIndex);
+
+                default:
+                    return next;
+            }
+        }
+        #endregion
     }
 }

Wabbajack/UI/MainWindow.xaml

@@ -1,4 +1,4 @@
-<Window
+<Window
     x:Class="Wabbajack.MainWindow"
     xmlns="http://schemas.microsoft.com/winfx/2006/xaml/presentation"
     xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"
@@ -17,7 +17,6 @@
     WindowStyle="ToolWindow"
     mc:Ignorable="d">
     <Viewbox Stretch="Uniform">
-        <!--<Grid Width="1280" Height="960">-->
         <Grid Margin="4,0,4,0">
             <Grid.RowDefinitions>
                 <RowDefinition Height="Auto" />

Wabbajack/UI/MainWindow.xaml.cs

@@ -25,8 +25,6 @@ namespace Wabbajack
         public MainWindow(RunMode mode, string source)
         {
             var args = Environment.GetCommandLineArgs();
-            var DebugMode = false;
-            string MO2Folder = null, InstallFolder = null, MO2Profile = null;
 
             InitializeComponent();
 
@@ -34,8 +32,6 @@ namespace Wabbajack
             context.LogMsg($"Wabbajack Build - {ThisAssembly.Git.Sha}");
             SetupHandlers(context);
             DataContext = context;
-            WorkQueue.Init((id, msg, progress) => context.SetProgress(id, msg, progress),
-                (max, current) => context.SetQueueSize(max, current));
 
             Utils.SetLoggerFn(s => context.LogMsg(s));
             Utils.SetStatusFn((msg, progress) => WorkQueue.Report(msg, progress));