diff --git a/Duplicati/WebserverCore/Abstractions/ITaskCacheService.cs b/Duplicati/Library/RestAPI/Abstractions/IQueueRunnerService.cs similarity index 51% rename from Duplicati/WebserverCore/Abstractions/ITaskCacheService.cs rename to Duplicati/Library/RestAPI/Abstractions/IQueueRunnerService.cs index a471cf8c3..554ae4bae 100644 --- a/Duplicati/WebserverCore/Abstractions/ITaskCacheService.cs +++ b/Duplicati/Library/RestAPI/Abstractions/IQueueRunnerService.cs @@ -1,3 +1,4 @@ + // Copyright (C) 2025, The Duplicati Team // https://duplicati.com, hello@duplicati.com // @@ -18,11 +19,14 @@ // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER // DEALINGS IN THE SOFTWARE. -using Duplicati.WebserverCore.Dto; +#nullable enable + +using System; +using System.Collections.Generic; +using Duplicati.Server.Serialization.Interface; namespace Duplicati.WebserverCore.Abstractions; - /// /// A cached task result /// @@ -33,21 +37,63 @@ namespace Duplicati.WebserverCore.Abstractions; /// The exception that was thrown public sealed record CachedTaskResult(long TaskID, string? BackupId, DateTime? TaskStarted, DateTime? TaskFinished, Exception? Exception); - /// -/// Interface for the task result cache service +/// Class to encapsulate a thread that runs a list of queued operations /// -public interface ITaskCacheService +/// The type to operate on +public interface IQueueRunnerService { + /// + /// Returns a copy of the current tasks in the queue + /// + /// A list of queued tasks + List GetCurrentTasks(); + /// + /// Gets a flag indicating if the queue is currently executing a task + /// + /// True if the queue is executing a task, false otherwise + bool GetIsActive(); + /// + /// Returns the currently executing task in the queue + /// + /// The currently executing task, or null if no task is executing + IQueuedTask? GetCurrentTask(); + /// /// Gets the cached task results for a given task ID /// /// The task ID /// The cached task result CachedTaskResult? GetCachedTaskResults(long taskID); + /// - /// Adds a task result to the cache + /// Adds a task to the queue /// - /// The task result to add - void AddTaskResult(CachedTaskResult taskResult); + /// The task to add + long AddTask(IQueuedTask task); + /// + /// Adds a task to the queue, optionally skipping the queue + /// + /// The task to add + /// Whether to skip the queue + long AddTask(IQueuedTask task, bool skipQueue); + /// + /// Removes a task from the queue + /// + /// Whether to wait for the task to finish + void Terminate(bool wait); + /// + /// Resumes processing items in the queue + /// + void Resume(); + /// + /// Pauses processing items in the queue + /// + void Pause(); + + /// + /// Gets the IDs of the tasks in the worker queue + /// + /// A list of tuples containing the task ID and backup ID + IList> GetQueueWithIds(); } \ No newline at end of file diff --git a/Duplicati/Library/RestAPI/Abstractions/IScheduler.cs b/Duplicati/Library/RestAPI/Abstractions/ISchedulerService.cs similarity index 72% rename from Duplicati/Library/RestAPI/Abstractions/IScheduler.cs rename to Duplicati/Library/RestAPI/Abstractions/ISchedulerService.cs index 1140bf84a..398543cdb 100644 --- a/Duplicati/Library/RestAPI/Abstractions/IScheduler.cs +++ b/Duplicati/Library/RestAPI/Abstractions/ISchedulerService.cs @@ -20,25 +20,12 @@ // DEALINGS IN THE SOFTWARE. using System; using System.Collections.Generic; -using Duplicati.Library.Utility; -using Duplicati.Server; using Duplicati.Server.Serialization.Interface; namespace Duplicati.WebserverCore.Abstractions; -public interface IScheduler +public interface ISchedulerService { - /// - /// Initializes scheduler - /// - /// The worker thread - void Init(WorkerThread worker); - - /// - /// Gets the current ids in the scheduler queue - /// - IList> GetSchedulerQueueIds(); - /// /// Gets the current proposed schedule /// @@ -50,21 +37,11 @@ public interface IScheduler /// True if the call should block until the thread has exited, false otherwise void Terminate(bool wait); - /// - /// Subscribes to the event that is triggered when the schedule changes - /// - void SubScribeToNewSchedule(Action handler); - /// /// A snapshot copy of the current schedule list /// List> Schedule { get; } - /// - /// A snapshot copy of the current worker queue, that is items that are scheduled, but waiting for execution - /// - List WorkerQueue { get; } - /// /// Forces the scheduler to re-evaluate the order. /// Call this method if something changes diff --git a/Duplicati/Library/RestAPI/Abstractions/IWorkerThreadsManager.cs b/Duplicati/Library/RestAPI/Abstractions/IWorkerThreadsManager.cs deleted file mode 100644 index aceea26b9..000000000 --- a/Duplicati/Library/RestAPI/Abstractions/IWorkerThreadsManager.cs +++ /dev/null @@ -1,37 +0,0 @@ -// Copyright (C) 2025, The Duplicati Team -// https://duplicati.com, hello@duplicati.com -// -// Permission is hereby granted, free of charge, to any person obtaining a -// copy of this software and associated documentation files (the "Software"), -// to deal in the Software without restriction, including without limitation -// the rights to use, copy, modify, merge, publish, distribute, sublicense, -// and/or sell copies of the Software, and to permit persons to whom the -// Software is furnished to do so, subject to the following conditions: -// -// The above copyright notice and this permission notice shall be included in -// all copies or substantial portions of the Software. -// -// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS -// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING -// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER -// DEALINGS IN THE SOFTWARE. -#nullable enable -using System; -using Duplicati.Library.Utility; -using Duplicati.Server; - -namespace Duplicati.Library.RestAPI.Abstractions; - -public interface IWorkerThreadsManager -{ - void Spawn(Action item); - - Tuple? CurrentTask { get; } - WorkerThread? WorkerThread { get; } - void UpdateThrottleSpeeds(string? uploadSpeed, string? downloadSpeed); - - long AddTask(Runner.IRunnerData data, bool skipQueue = false); -} \ No newline at end of file diff --git a/Duplicati/Library/RestAPI/Database/Connection.cs b/Duplicati/Library/RestAPI/Database/Connection.cs index 90bb7bbda..2c4a8a91b 100644 --- a/Duplicati/Library/RestAPI/Database/Connection.cs +++ b/Duplicati/Library/RestAPI/Database/Connection.cs @@ -32,6 +32,7 @@ using Duplicati.Library.AutoUpdater; using System.Data; using Duplicati.Library.Main.Database; using System.Globalization; +using Microsoft.Extensions.DependencyInjection; #nullable enable @@ -47,6 +48,10 @@ namespace Duplicati.Server.Database private readonly Dictionary m_temporaryBackups = new Dictionary(); private readonly bool m_encryptSensitiveFields; private readonly EncryptedFieldHelper.KeyInstance? m_key; + private IServiceProvider? m_serviceProvider; + private INotificationUpdateService? m_notificationUpdateService; + private EventPollNotify? m_eventPollNotifyer; + private static readonly HashSet _encryptedFields = BackendLoader.Backends.SelectMany(x => x.SupportedCommands ?? []) .Concat(EncryptionLoader.Modules.SelectMany(x => x.SupportedCommands ?? [])) @@ -76,6 +81,22 @@ namespace Duplicati.Server.Database this.ApplicationSettings = new ServerSettings(this); } + /// + /// The service provider is used to resolve dependencies + /// + internal IServiceProvider? ServiceProvider => m_serviceProvider; + + /// + /// Set the service provider to be used for resolving dependencies + /// + /// The service provider + public void SetServiceProvider(IServiceProvider sp) + { + m_serviceProvider = sp; + m_notificationUpdateService = sp?.GetRequiredService(); + m_eventPollNotifyer = sp?.GetRequiredService(); + } + public bool IsEncryptingFields => m_encryptSensitiveFields; public void ReWriteAllFieldsIfEncryptionChanged() @@ -125,7 +146,7 @@ namespace Duplicati.Server.Database this.ApplicationSettings.PreloadSettingsHash = settingsHash; } - public void LogError(string backupid, string message, Exception ex) + public void LogError(string? backupid, string message, Exception ex) { lock (m_lock) { @@ -608,8 +629,8 @@ namespace Duplicati.Server.Database } } - FIXMEGlobal.NotificationUpdateService.IncrementLastDataUpdateId(); - FIXMEGlobal.StatusEventNotifyer.SignalNewEvent(); + m_notificationUpdateService?.IncrementLastDataUpdateId(); + m_eventPollNotifyer?.SignalNewEvent(); } private void AddOrUpdateBackup(IBackup item, bool updateSchedule, ISchedule? schedule) @@ -722,8 +743,8 @@ namespace Duplicati.Server.Database } tr.Commit(); - FIXMEGlobal.NotificationUpdateService.IncrementLastDataUpdateId(); - FIXMEGlobal.StatusEventNotifyer.SignalNewEvent(); + m_notificationUpdateService?.IncrementLastDataUpdateId(); + m_eventPollNotifyer?.SignalNewEvent(); } } } @@ -735,8 +756,8 @@ namespace Duplicati.Server.Database { AddOrUpdateSchedule(item, tr); tr.Commit(); - FIXMEGlobal.NotificationUpdateService.IncrementLastDataUpdateId(); - FIXMEGlobal.StatusEventNotifyer.SignalNewEvent(); + m_notificationUpdateService?.IncrementLastDataUpdateId(); + m_eventPollNotifyer?.SignalNewEvent(); } } @@ -800,8 +821,8 @@ namespace Duplicati.Server.Database } } - FIXMEGlobal.NotificationUpdateService.IncrementLastDataUpdateId(); - FIXMEGlobal.StatusEventNotifyer.SignalNewEvent(); + m_notificationUpdateService?.IncrementLastDataUpdateId(); + m_eventPollNotifyer?.SignalNewEvent(); } public void DeleteBackup(IBackup backup) @@ -820,8 +841,8 @@ namespace Duplicati.Server.Database lock (m_lock) DeleteFromDb("Schedule", ID); - FIXMEGlobal.NotificationUpdateService.IncrementLastDataUpdateId(); - FIXMEGlobal.StatusEventNotifyer.SignalNewEvent(); + m_notificationUpdateService?.IncrementLastDataUpdateId(); + m_eventPollNotifyer?.SignalNewEvent(); } public void DeleteSchedule(ISchedule schedule) @@ -906,21 +927,27 @@ namespace Duplicati.Server.Database return false; DeleteFromDb(typeof(Notification).Name, id); - FIXMEGlobal.DataConnection.ApplicationSettings.UnackedError = notifications.Any(x => x.ID != id && x.Type == Duplicati.Server.Serialization.NotificationType.Error); - FIXMEGlobal.DataConnection.ApplicationSettings.UnackedWarning = notifications.Any(x => x.ID != id && x.Type == Duplicati.Server.Serialization.NotificationType.Warning); + this.ApplicationSettings.UnackedError = notifications.Any(x => x.ID != id && x.Type == Duplicati.Server.Serialization.NotificationType.Error); + this.ApplicationSettings.UnackedWarning = notifications.Any(x => x.ID != id && x.Type == Duplicati.Server.Serialization.NotificationType.Warning); } - // Guard against dismissing notifications before the provider is initialized - if (FIXMEGlobal.Provider != null) - { - FIXMEGlobal.NotificationUpdateService.IncrementLastNotificationUpdateId(); - FIXMEGlobal.StatusEventNotifyer.SignalNewEvent(); - } + m_notificationUpdateService?.IncrementLastNotificationUpdateId(); + m_eventPollNotifyer?.SignalNewEvent(); return true; } - public void RegisterNotification(Serialization.NotificationType type, string title, string message, Exception ex, string backupid, string action, string logid, string messageid, string logtag, Func conflicthandler) + public void RegisterNotification( + Serialization.NotificationType type, + string title, + string message, + Exception? ex, + string? backupid, + string action, + string? logid, + string? messageid, + string? logtag, + Func conflicthandler) { lock (m_lock) { @@ -949,13 +976,13 @@ namespace Duplicati.Server.Database OverwriteAndUpdateDb(null, null, [notification], false); if (type == Serialization.NotificationType.Error) - FIXMEGlobal.DataConnection.ApplicationSettings.UnackedError = true; + ApplicationSettings.UnackedError = true; else if (type == Serialization.NotificationType.Warning) - FIXMEGlobal.DataConnection.ApplicationSettings.UnackedWarning = true; + ApplicationSettings.UnackedWarning = true; } - FIXMEGlobal.NotificationUpdateService.IncrementLastNotificationUpdateId(); - FIXMEGlobal.StatusEventNotifyer.SignalNewEvent(); + m_notificationUpdateService?.IncrementLastNotificationUpdateId(); + m_eventPollNotifyer?.SignalNewEvent(); } //Workaround to clean up the database after invalid settings update diff --git a/Duplicati/Library/RestAPI/Database/ServerSettings.cs b/Duplicati/Library/RestAPI/Database/ServerSettings.cs index dea19e8ec..c1fc38b30 100644 --- a/Duplicati/Library/RestAPI/Database/ServerSettings.cs +++ b/Duplicati/Library/RestAPI/Database/ServerSettings.cs @@ -1,4 +1,4 @@ -// Copyright (C) 2025, The Duplicati Team +// Copyright (C) 2025, The Duplicati Team // https://duplicati.com, hello@duplicati.com // // Permission is hereby granted, free of charge, to any person obtaining a @@ -29,6 +29,8 @@ using System.Text; using System.Security.Cryptography.X509Certificates; using Duplicati.Library.Utility; using Duplicati.Library.AutoUpdater; +using Microsoft.Extensions.DependencyInjection; +using Duplicati.WebserverCore.Abstractions; #nullable enable @@ -142,12 +144,13 @@ namespace Duplicati.Server.Database Value = n.Value }, Database.Connection.SERVER_SETTINGS_ID); - if (FIXMEGlobal.IsServerStarted) + var provider = databaseConnection.ServiceProvider; + if (provider != null) { - FIXMEGlobal.NotificationUpdateService.IncrementLastDataUpdateId(); - FIXMEGlobal.StatusEventNotifyer.SignalNewEvent(); + provider?.GetRequiredService()?.IncrementLastDataUpdateId(); + provider?.GetRequiredService()?.SignalNewEvent(); // If throttle options were changed, update now - FIXMEGlobal.WorkerThreadsManager.UpdateThrottleSpeeds(UploadSpeedLimit, DownloadSpeedLimit); + provider?.GetRequiredService()?.GetCurrentTask()?.UpdateThrottleSpeeds(UploadSpeedLimit, DownloadSpeedLimit); } // In case the usage reporter is enabled or disabled, refresh now @@ -602,7 +605,7 @@ namespace Duplicati.Server.Database lock (databaseConnection.m_lock) settings[CONST.UPDATE_CHECK_INTERVAL] = value; SaveSettings(); - FIXMEGlobal.UpdatePoller.Reschedule(); + databaseConnection?.ServiceProvider?.GetRequiredService()?.Reschedule(); } } diff --git a/Duplicati/Library/RestAPI/FIXMEGlobal.cs b/Duplicati/Library/RestAPI/FIXMEGlobal.cs index 9d3d66839..5a977f0c2 100644 --- a/Duplicati/Library/RestAPI/FIXMEGlobal.cs +++ b/Duplicati/Library/RestAPI/FIXMEGlobal.cs @@ -21,8 +21,6 @@ using Duplicati.Server; using System; -using Duplicati.Library.RestAPI.Abstractions; -using Duplicati.Library.Utility; using Duplicati.WebserverCore.Abstractions; using Microsoft.Extensions.DependencyInjection; using Duplicati.Library.Interface; @@ -66,15 +64,6 @@ namespace Duplicati.Library.RestAPI /// public static bool IsServerStarted => Provider != null; - /// - /// This is the working thread - /// - public static WorkerThread WorkThread => - Provider.GetRequiredService().WorkerThread; - - public static IWorkerThreadsManager WorkerThreadsManager => - Provider.GetRequiredService(); - public static Action StartOrStopUsageReporter; /// @@ -85,7 +74,7 @@ namespace Duplicati.Library.RestAPI /// /// This is the scheduling thread /// - public static IScheduler Scheduler => Provider.GetRequiredService(); + public static ISchedulerService Scheduler => Provider.GetRequiredService(); /// /// The log redirect handler diff --git a/Duplicati/Library/RestAPI/Runner.cs b/Duplicati/Library/RestAPI/Runner.cs index aa73ceabc..3480d1795 100644 --- a/Duplicati/Library/RestAPI/Runner.cs +++ b/Duplicati/Library/RestAPI/Runner.cs @@ -19,6 +19,8 @@ // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER // DEALINGS IN THE SOFTWARE. +#nullable enable + using System; using System.Linq; using System.Collections.Generic; @@ -26,51 +28,48 @@ using Duplicati.Library.Interface; using Duplicati.Server.Serialization; using Duplicati.Library.RestAPI; using Duplicati.Library.Utility; +using System.Threading.Tasks; namespace Duplicati.Server { public static class Runner { - public interface IRunnerData : Duplicati.Server.Serialization.Interface.IQueuedTask + public interface IRunnerData : Serialization.Interface.IQueuedTask { - Duplicati.Server.Serialization.Interface.IBackup Backup { get; } - IDictionary ExtraOptions { get; } - string[] FilterStrings { get; } - string[] ExtraArguments { get; } + Serialization.Interface.IBackup? Backup { get; } + IDictionary? ExtraOptions { get; } + string[]? FilterStrings { get; } + string[]? ExtraArguments { get; } int PageSize { get; } int PageOffset { get; } - void Stop(); - void Abort(); - void Pause(bool alsoTransfers); - void Resume(); - void UpdateThrottleSpeed(string uploadSpeed, string downloadSpeed); - void SetController(Duplicati.Library.Main.Controller controller); - DateTime? TaskStarted { get; set; } - DateTime? TaskFinished { get; set; } + void SetController(Library.Main.Controller? controller); } private class RunnerData : IRunnerData { private static long RunnerTaskID = 1; - public Duplicati.Server.Serialization.DuplicatiOperation Operation { get; internal set; } - public Duplicati.Server.Serialization.Interface.IBackup Backup { get; internal set; } - public IDictionary ExtraOptions { get; internal set; } - public string[] FilterStrings { get; internal set; } + public Func? OnStarting { get; set; } + public Func? OnFinished { get; set; } - public string BackupID { get { return Backup.ID; } } + public DuplicatiOperation Operation { get; internal set; } + public Serialization.Interface.IBackup? Backup { get; internal set; } + public IDictionary? ExtraOptions { get; internal set; } + public string[]? FilterStrings { get; internal set; } + + public string? BackupID { get { return Backup?.ID; } } public long TaskID { get { return m_taskID; } } - public string[] ExtraArguments { get; internal set; } + public string[]? ExtraArguments { get; internal set; } public int PageSize { get; internal set; } = 0; public int PageOffset { get; internal set; } = 0; public DateTime? TaskStarted { get; set; } public DateTime? TaskFinished { get; set; } - internal Duplicati.Library.Main.Controller Controller { get; set; } + internal Library.Main.Controller? Controller { get; set; } - public void SetController(Duplicati.Library.Main.Controller controller) + public void SetController(Library.Main.Controller? controller) { Controller = controller; } @@ -98,7 +97,7 @@ namespace Duplicati.Server public long OriginalUploadSpeed { get; set; } public long OriginalDownloadSpeed { get; set; } - public void UpdateThrottleSpeed(string uploadSpeed, string downloadSpeed) + public void UpdateThrottleSpeeds(string? uploadSpeed, string? downloadSpeed) { var controller = this.Controller; if (controller == null) @@ -113,14 +112,14 @@ namespace Duplicati.Server try { if (!string.IsNullOrWhiteSpace(uploadSpeed)) - server_upload_throttle = Duplicati.Library.Utility.Sizeparser.ParseSize(uploadSpeed, "kb"); + server_upload_throttle = Sizeparser.ParseSize(uploadSpeed, "kb"); } catch { } try { if (!string.IsNullOrWhiteSpace(downloadSpeed)) - server_download_throttle = Duplicati.Library.Utility.Sizeparser.ParseSize(downloadSpeed, "kb"); + server_download_throttle = Sizeparser.ParseSize(downloadSpeed, "kb"); } catch { } @@ -142,6 +141,9 @@ namespace Duplicati.Server { m_taskID = System.Threading.Interlocked.Increment(ref RunnerTaskID); } + + public Task Execute() + => Task.Run(() => Runner.Run(this, true)); } private class CustomRunnerTask : RunnerData @@ -164,13 +166,13 @@ namespace Duplicati.Server return new CustomRunnerTask(runner); } - public static IRunnerData CreateTask(Duplicati.Server.Serialization.DuplicatiOperation operation, Duplicati.Server.Serialization.Interface.IBackup backup, IDictionary extraOptions = null, string[] filterStrings = null, string[] extraArguments = null, int pageSize = 0, int pageOffset = 0) + public static IRunnerData CreateTask(DuplicatiOperation operation, Serialization.Interface.IBackup backup, IDictionary? extraOptions = null, string[]? filterStrings = null, string[]? extraArguments = null, int pageSize = 0, int pageOffset = 0) { return new RunnerData() { Operation = operation, Backup = backup, - ExtraOptions = extraOptions, + ExtraOptions = extraOptions ?? new Dictionary(), FilterStrings = filterStrings, ExtraArguments = extraArguments, PageSize = pageSize, @@ -178,15 +180,15 @@ namespace Duplicati.Server }; } - public static IRunnerData CreateListTask(Duplicati.Server.Serialization.Interface.IBackup backup, string[] filters, bool onlyPrefix, bool allVersions, bool folderContents, DateTime time) + public static IRunnerData CreateListTask(Serialization.Interface.IBackup backup, string[]? filters, bool onlyPrefix, bool allVersions, bool folderContents, DateTime time) { - var dict = new Dictionary(); + var dict = new Dictionary(); if (onlyPrefix) dict["list-prefix-only"] = "true"; if (allVersions) dict["all-versions"] = "true"; if (time.Ticks > 0) - dict["time"] = Duplicati.Library.Utility.Utility.SerializeDateTime(time.ToUniversalTime()); + dict["time"] = Utility.SerializeDateTime(time.ToUniversalTime()); if (folderContents) dict["list-folder-contents"] = "true"; @@ -197,19 +199,19 @@ namespace Duplicati.Server filters); } - public static IRunnerData CreateListFilesetsTask(Duplicati.Server.Serialization.Interface.IBackup backup, Dictionary extraOptions = null) + public static IRunnerData CreateListFilesetsTask(Serialization.Interface.IBackup backup, Dictionary? extraOptions = null) { return CreateTask( DuplicatiOperation.ListFilesets, backup, - extraOptions ?? new Dictionary()); + extraOptions ?? new Dictionary()); } - public static IRunnerData CreateListFolderContents(Duplicati.Server.Serialization.Interface.IBackup backup, string[] folders, DateTime time, int pageSize, int pageOffset) + public static IRunnerData CreateListFolderContents(Serialization.Interface.IBackup backup, string[] folders, DateTime time, int pageSize, int pageOffset) { - var dict = new Dictionary(); + var dict = new Dictionary(); if (time.Ticks > 0) - dict["time"] = Duplicati.Library.Utility.Utility.SerializeDateTime(time.ToUniversalTime()); + dict["time"] = Utility.SerializeDateTime(time.ToUniversalTime()); return CreateTask( DuplicatiOperation.ListFolderContents, @@ -220,9 +222,9 @@ namespace Duplicati.Server pageOffset: pageOffset); } - public static IRunnerData ListFileVersionsTask(Duplicati.Server.Serialization.Interface.IBackup backup, string[] filepaths, int pageSize, int pageOffset) + public static IRunnerData ListFileVersionsTask(Serialization.Interface.IBackup backup, string[] filepaths, int pageSize, int pageOffset) { - var dict = new Dictionary(); + var dict = new Dictionary(); return CreateTask( DuplicatiOperation.ListFileVersions, backup, @@ -232,11 +234,11 @@ namespace Duplicati.Server pageOffset: pageOffset); } - public static IRunnerData CreateSearchEntriesTask(Duplicati.Server.Serialization.Interface.IBackup backup, string[] filters, string[] folders, DateTime time, int pageSize, int pageOffset) + public static IRunnerData CreateSearchEntriesTask(Serialization.Interface.IBackup backup, string[] filters, string[] folders, DateTime time, int pageSize, int pageOffset) { - var dict = new Dictionary(); + var dict = new Dictionary(); if (time.Ticks > 0) - dict["time"] = Duplicati.Library.Utility.Utility.SerializeDateTime(time.ToUniversalTime()); + dict["time"] = Utility.SerializeDateTime(time.ToUniversalTime()); return CreateTask( DuplicatiOperation.SearchEntries, @@ -249,17 +251,17 @@ namespace Duplicati.Server } - public static IRunnerData CreateRestoreTask(Duplicati.Server.Serialization.Interface.IBackup backup, string[] filters, - DateTime time, string restoreTarget, bool overwrite, bool restore_permissions, - bool skip_metadata, string passphrase) + public static IRunnerData CreateRestoreTask(Serialization.Interface.IBackup backup, string[]? filters, + DateTime time, string? restoreTarget, bool overwrite, bool restore_permissions, + bool skip_metadata, string? passphrase) { - var dict = new Dictionary + var dict = new Dictionary { - ["time"] = Library.Utility.Utility.SerializeDateTime(time.ToUniversalTime()), - ["overwrite"] = overwrite ? Boolean.TrueString : Boolean.FalseString, - ["restore-permissions"] = restore_permissions ? Boolean.TrueString : Boolean.FalseString, - ["skip-metadata"] = skip_metadata ? Boolean.TrueString : Boolean.FalseString, - ["allow-passphrase-change"] = Boolean.TrueString + ["time"] = Utility.SerializeDateTime(time.ToUniversalTime()), + ["overwrite"] = overwrite ? bool.TrueString : bool.FalseString, + ["restore-permissions"] = restore_permissions ? bool.TrueString : bool.FalseString, + ["skip-metadata"] = skip_metadata ? bool.TrueString : bool.FalseString, + ["allow-passphrase-change"] = bool.TrueString }; if (!string.IsNullOrWhiteSpace(restoreTarget)) dict["restore-path"] = SpecialFolders.ExpandEnvironmentVariables(restoreTarget); @@ -272,26 +274,26 @@ namespace Duplicati.Server dict, filters); } - private class MessageSink : Duplicati.Library.Main.IMessageSink + private class MessageSink : Library.Main.IMessageSink { - private class ProgressState : Server.Serialization.Interface.IProgressEventData + private class ProgressState : Serialization.Interface.IProgressEventData { - private readonly string m_backupID; + private readonly string? m_backupID; private readonly long m_taskID; - internal Duplicati.Library.Main.BackendActionType m_backendAction; - internal string m_backendPath; + internal Library.Main.BackendActionType m_backendAction; + internal string? m_backendPath; internal long m_backendFileSize; internal long m_backendFileProgress; internal long m_backendSpeed; internal bool m_backendIsBlocking; - internal string m_currentFilename; + internal string? m_currentFilename; internal long m_currentFilesize; internal long m_currentFileoffset; internal bool m_currentFilecomplete; - internal Duplicati.Library.Main.OperationPhase m_phase; + internal Library.Main.OperationPhase m_phase; internal float m_overallProgress; internal long m_processedFileCount; internal long m_processedFileSize; @@ -299,7 +301,7 @@ namespace Duplicati.Server internal long m_totalFileSize; internal bool m_stillCounting; - public ProgressState(long taskId, string backupId) + public ProgressState(long taskId, string? backupId) { m_backupID = backupId; m_taskID = taskId; @@ -311,15 +313,15 @@ namespace Duplicati.Server } #region IProgressEventData implementation - public string BackupID { get { return m_backupID; } } + public string? BackupID { get { return m_backupID; } } public long TaskID { get { return m_taskID; } } public string BackendAction { get { return m_backendAction.ToString(); } } - public string BackendPath { get { return m_backendPath; } } + public string? BackendPath { get { return m_backendPath; } } public long BackendFileSize { get { return m_backendFileSize; } } public long BackendFileProgress { get { return m_backendFileProgress; } } public long BackendSpeed { get { return m_backendSpeed; } } public bool BackendIsBlocking { get { return m_backendIsBlocking; } } - public string CurrentFilename { get { return m_currentFilename; } } + public string? CurrentFilename { get { return m_currentFilename; } } public long CurrentFilesize { get { return m_currentFilesize; } } public long CurrentFileoffset { get { return m_currentFileoffset; } } public bool CurrentFilecomplete { get { return m_currentFilecomplete; } } @@ -334,16 +336,16 @@ namespace Duplicati.Server } private readonly ProgressState m_state; - private Duplicati.Library.Main.IBackendProgress m_backendProgress; - private Duplicati.Library.Main.IOperationProgress m_operationProgress; + private Library.Main.IBackendProgress? m_backendProgress; + private Library.Main.IOperationProgress? m_operationProgress; private readonly object m_lock = new object(); - public MessageSink(long taskId, string backupId) + public MessageSink(long taskId, string? backupId) { m_state = new ProgressState(taskId, backupId); } - public Server.Serialization.Interface.IProgressEventData Copy() + public Serialization.Interface.IProgressEventData Copy() { lock (m_lock) { @@ -395,6 +397,8 @@ namespace Duplicati.Server public static string GetCommandLine(IRunnerData data) { var backup = data.Backup; + if (backup == null) + throw new ArgumentNullException(nameof(backup)); var options = ApplyOptions(backup, GetCommonOptions()); if (data.ExtraOptions != null) @@ -416,22 +420,22 @@ namespace Duplicati.Server ); var cmd = new System.Text.StringBuilder(); - cmd.Append(Library.Utility.Utility.WrapAsCommandLine(new string[] { exe, "backup", backup.TargetURL }, false)); + cmd.Append(Utility.WrapAsCommandLine([exe, "backup", backup.TargetURL], false)); cmd.Append(" "); - cmd.Append(Library.Utility.Utility.WrapAsCommandLine(sources, true)); + cmd.Append(Utility.WrapAsCommandLine(sources, true)); // TODO: We should check each option to see if it is a path, and allow expansion on that foreach (var opt in options) - cmd.AppendFormat(" --{0}={1}", opt.Key, Library.Utility.Utility.WrapCommandLineElement(opt.Value, false)); + cmd.AppendFormat(" --{0}={1}", opt.Key, Utility.WrapCommandLineElement(opt.Value, false)); if (cf != null) foreach (var f in cf) - cmd.AppendFormat(" --{0}={1}", f.Include ? "include" : "exclude", Library.Utility.Utility.WrapCommandLineElement(f.Expression, true)); + cmd.AppendFormat(" --{0}={1}", f.Include ? "include" : "exclude", Utility.WrapCommandLineElement(f.Expression, true)); if (bf != null) foreach (var f in bf) - cmd.AppendFormat(" --{0}={1}", f.Include ? "include" : "exclude", Library.Utility.Utility.WrapCommandLineElement(f.Expression, true)); + cmd.AppendFormat(" --{0}={1}", f.Include ? "include" : "exclude", Utility.WrapCommandLineElement(f.Expression, true)); return cmd.ToString(); } @@ -439,6 +443,8 @@ namespace Duplicati.Server public static string[] GetCommandLineParts(IRunnerData data) { var backup = data.Backup; + if (backup == null) + throw new ArgumentNullException(nameof(backup)); var options = ApplyOptions(backup, GetCommonOptions()); if (data.ExtraOptions != null) @@ -474,7 +480,7 @@ namespace Duplicati.Server return parts.ToArray(); } - public static Duplicati.Library.Interface.IBasicResults Run(IRunnerData data, bool fromQueue) + public static IBasicResults? Run(IRunnerData data, bool fromQueue) { data.TaskStarted = DateTime.Now; if (data is CustomRunnerTask task) @@ -500,12 +506,11 @@ namespace Duplicati.Server } var backup = data.Backup; - if (backup.Metadata == null) - { - backup.Metadata = new Dictionary(); - } + if (backup == null) + throw new ArgumentNullException(nameof(backup)); - Duplicati.Library.Utility.TempFolder tempfolder = null; + backup.Metadata ??= new Dictionary(); + TempFolder? tempfolder = null; try { @@ -552,16 +557,17 @@ namespace Duplicati.Server catch { } ((RunnerData)data).Controller = controller; - data.UpdateThrottleSpeed(FIXMEGlobal.DataConnection.ApplicationSettings.UploadSpeedLimit, FIXMEGlobal.DataConnection.ApplicationSettings.DownloadSpeedLimit); + var appSettings = FIXMEGlobal.DataConnection.ApplicationSettings; + data.UpdateThrottleSpeeds(appSettings.UploadSpeedLimit, appSettings.DownloadSpeedLimit); // Pass on the provider, will be replaced if configured in the backup controller.SetSecretProvider(FIXMEGlobal.SecretProvider); if (backup.Metadata.ContainsKey("LastCompactFinished")) - controller.LastCompact = Library.Utility.Utility.DeserializeDateTime(backup.Metadata["LastCompactFinished"]); + controller.LastCompact = Utility.DeserializeDateTime(backup.Metadata["LastCompactFinished"]); if (backup.Metadata.ContainsKey("LastVacuumFinished")) - controller.LastVacuum = Library.Utility.Utility.DeserializeDateTime(backup.Metadata["LastVacuumFinished"]); + controller.LastVacuum = Utility.DeserializeDateTime(backup.Metadata["LastVacuumFinished"]); switch (data.Operation) { @@ -622,12 +628,12 @@ namespace Duplicati.Server } case DuplicatiOperation.CreateReport: { - using (var tf = new Duplicati.Library.Utility.TempFile()) + using (var tf = new TempFile()) { var r = controller.CreateLogDatabase(tf); var tempid = FIXMEGlobal.DataConnection.RegisterTempFile("create-bug-report", r.TargetPath, DateTime.Now.AddDays(3)); - if (string.Equals(tf, r.TargetPath, Library.Utility.Utility.ClientFilenameStringComparison)) + if (string.Equals(tf, r.TargetPath, Utility.ClientFilenameStringComparison)) tf.Protected = true; FIXMEGlobal.DataConnection.RegisterNotification( @@ -656,16 +662,18 @@ namespace Duplicati.Server case DuplicatiOperation.Delete: { - if (Library.Utility.Utility.ParseBoolOption(data.ExtraOptions.AsReadOnly(), "delete-remote-files")) - controller.DeleteAllRemoteFiles(); - - if (Library.Utility.Utility.ParseBoolOption(data.ExtraOptions.AsReadOnly(), "delete-local-db")) + if (data.ExtraOptions != null) { - string dbpath; - options.TryGetValue("dbpath", out dbpath); + if (Utility.ParseBoolOption(data.ExtraOptions.AsReadOnly(), "delete-remote-files")) + controller.DeleteAllRemoteFiles(); - if (!string.IsNullOrWhiteSpace(dbpath) && System.IO.File.Exists(dbpath)) - System.IO.File.Delete(dbpath); + if (Utility.ParseBoolOption(data.ExtraOptions.AsReadOnly(), "delete-local-db")) + { + options.TryGetValue("dbpath", out var dbpath); + + if (!string.IsNullOrWhiteSpace(dbpath) && System.IO.File.Exists(dbpath)) + System.IO.File.Delete(dbpath); + } } FIXMEGlobal.DataConnection.DeleteBackup(backup); FIXMEGlobal.Scheduler.Reschedule(); @@ -706,8 +714,9 @@ namespace Duplicati.Server } catch (Exception ex) { - FIXMEGlobal.DataConnection.LogError(data.Backup.ID, string.Format("Failed while executing {0} \"{1}\" (id: {2})", data.Operation, data.Backup.Name, data.Backup.ID), ex); - UpdateMetadataError(data.Backup, ex); + FIXMEGlobal.DataConnection.LogError(data.Backup?.ID, string.Format("Failed while executing {0} \"{1}\" (id: {2})", data.Operation, data.Backup?.Name, data.Backup?.ID), ex); + if (data.Backup != null) + UpdateMetadataError(data.Backup, ex); Library.UsageReporter.Reporter.Report(ex); if (!fromQueue) @@ -722,23 +731,26 @@ namespace Duplicati.Server } } - private static Duplicati.Library.Utility.TempFolder StoreTaskConfigAndGetTempFolder(IRunnerData data, Dictionary options) + private static TempFolder? StoreTaskConfigAndGetTempFolder(IRunnerData data, Dictionary options) { + if (data.Backup == null) + throw new ArgumentNullException(nameof(data.Backup)); + var all_tasks = string.Equals(options["store-task-config"], "all", StringComparison.OrdinalIgnoreCase) || string.Equals(options["store-task-config"], "*", StringComparison.OrdinalIgnoreCase); - var this_task = Duplicati.Library.Utility.Utility.ParseBool(options["store-task-config"], false); + var this_task = Utility.ParseBool(options["store-task-config"], false); options.Remove("store-task-config"); - Duplicati.Library.Utility.TempFolder tempfolder = null; + TempFolder? tempfolder = null; if (all_tasks || this_task) { - tempfolder = new Duplicati.Library.Utility.TempFolder(); + tempfolder = new TempFolder(); var temppath = System.IO.Path.Combine(tempfolder, "task-setup.json"); - using (var tempfile = Duplicati.Library.Utility.TempFile.WrapExistingFile(temppath)) + using (var tempfile = TempFile.WrapExistingFile(temppath)) { - object taskdata = null; + object? taskdata = null; if (all_tasks) - taskdata = FIXMEGlobal.DataConnection.Backups.Where(x => !x.IsTemporary).Select(x => FIXMEGlobal.DataConnection.PrepareBackupForExport(FIXMEGlobal.DataConnection.GetBackup(x.ID))); + taskdata = FIXMEGlobal.DataConnection.Backups.Where(x => !x.IsTemporary).Select(x => FIXMEGlobal.DataConnection.PrepareBackupForExport(FIXMEGlobal.DataConnection.GetBackup(x.ID)!)); else taskdata = new[] { FIXMEGlobal.DataConnection.PrepareBackupForExport(data.Backup) }; @@ -748,7 +760,7 @@ namespace Duplicati.Server tempfile.Protected = true; - options.TryGetValue("control-files", out string controlfiles); + options.TryGetValue("control-files", out var controlfiles); if (string.IsNullOrWhiteSpace(controlfiles)) controlfiles = tempfile; @@ -769,7 +781,7 @@ namespace Duplicati.Server if (!backup.IsTemporary) FIXMEGlobal.DataConnection.SetMetadata(backup.Metadata, long.Parse(backup.ID), null); - string messageid = null; + string? messageid = null; if (ex is UserInformationException exception) messageid = exception.HelpID; @@ -973,23 +985,20 @@ namespace Duplicati.Server return true; } - private static void DisableModule(string module, Dictionary options) + private static void DisableModule(string module, Dictionary options) { - string disabledModules; - string enabledModules; - - if (options.TryGetValue("enable-module", out enabledModules)) + if (options.TryGetValue("enable-module", out var enabledModules)) { var emods = (enabledModules ?? "").Split(new char[] { ',' }, StringSplitOptions.RemoveEmptyEntries); options["enable-module"] = string.Join(",", emods.Where(x => module.Equals(x, StringComparison.OrdinalIgnoreCase))); } - options.TryGetValue("disable-module", out disabledModules); + options.TryGetValue("disable-module", out var disabledModules); var mods = (disabledModules ?? "").Split(new char[] { ',' }, StringSplitOptions.RemoveEmptyEntries); options["disable-module"] = string.Join(",", mods.Union(new string[] { module }).Distinct(StringComparer.OrdinalIgnoreCase)); } - internal static Dictionary ApplyOptions(Duplicati.Server.Serialization.Interface.IBackup backup, Dictionary options) + internal static Dictionary ApplyOptions(Serialization.Interface.IBackup backup, Dictionary options) { options["backup-name"] = backup.Name; options["dbpath"] = backup.DBPath; @@ -1023,7 +1032,7 @@ namespace Duplicati.Server return options; } - private static Library.Utility.IFilter ApplyFilter(Serialization.Interface.IBackup backup, Library.Utility.IFilter filter) + private static IFilter? ApplyFilter(Serialization.Interface.IBackup backup, IFilter? filter) { var f2 = backup.Filters; if (f2 != null && f2.Length > 0) @@ -1035,24 +1044,24 @@ namespace Duplicati.Server ? SpecialFolders.ExpandEnvironmentVariablesRegexp(n.Expression) : SpecialFolders.ExpandEnvironmentVariables(n.Expression) orderby n.Order - select (Library.Utility.IFilter)(new Library.Utility.FilterExpression(exp, n.Include))) - .Aggregate((a, b) => Library.Utility.FilterExpression.Combine(a, b)); + select (IFilter)new FilterExpression(exp, n.Include)) + .Aggregate((a, b) => FilterExpression.Combine(a, b)); - filter = Library.Utility.FilterExpression.Combine(filter, nf); + filter = FilterExpression.Combine(filter, nf); } return filter; } - public static Dictionary GetCommonOptions() + public static Dictionary GetCommonOptions() { return (from n in FIXMEGlobal.DataConnection.Settings where TestIfOptionApplies() - select n).ToDictionary(k => k.Name.StartsWith("--", StringComparison.Ordinal) ? k.Name.Substring(2) : k.Name, k => k.Value); + select n).ToDictionary(k => k.Name.StartsWith("--", StringComparison.Ordinal) ? k.Name.Substring(2) : k.Name, k => (string?)k.Value); } - private static Duplicati.Library.Utility.IFilter GetCommonFilter() + private static IFilter? GetCommonFilter() { var filters = FIXMEGlobal.DataConnection.Filters; if (filters == null || filters.Length == 0) @@ -1062,8 +1071,8 @@ namespace Duplicati.Server (from n in filters orderby n.Order let exp = Environment.ExpandEnvironmentVariables(n.Expression) - select (Duplicati.Library.Utility.IFilter)(new Duplicati.Library.Utility.FilterExpression(exp, n.Include))) - .Aggregate((a, b) => Duplicati.Library.Utility.FilterExpression.Combine(a, b)); + select (IFilter)new FilterExpression(exp, n.Include)) + .Aggregate((a, b) => FilterExpression.Combine(a, b)); } } } diff --git a/Duplicati/Library/RestAPI/Scheduler.cs b/Duplicati/Library/RestAPI/Scheduler.cs index fc744e41c..388a850b7 100644 --- a/Duplicati/Library/RestAPI/Scheduler.cs +++ b/Duplicati/Library/RestAPI/Scheduler.cs @@ -19,6 +19,8 @@ // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER // DEALINGS IN THE SOFTWARE. +#nullable enable + using Duplicati.Server.Serialization.Interface; using System; @@ -28,6 +30,8 @@ using System.Linq; using System.Threading; using Duplicati.Library.Utility; using Duplicati.Library.RestAPI; +using System.Threading.Tasks; +using Duplicati.WebserverCore.Abstractions; // TODO: Rewrite this class. // It should just signal what new backups to run, and not mix with the worker thread. @@ -39,7 +43,7 @@ namespace Duplicati.Server /// public class Scheduler { - private static readonly string LOGTAG = Duplicati.Library.Logging.Log.LogTagFromType(); + private static readonly string LOGTAG = Library.Logging.Log.LogTagFromType(); /// /// The thread that runs the scheduler @@ -51,11 +55,6 @@ namespace Duplicati.Server /// private volatile bool m_terminate; - /// - /// The worker thread that is invoked to do work - /// - private WorkerThread m_worker; - /// /// The wait event /// @@ -67,9 +66,9 @@ namespace Duplicati.Server private readonly object m_lock = new object(); /// - /// An event that is raised when the schedule changes + /// The queue runner service /// - public event EventHandler NewSchedule; + private readonly IQueueRunnerService m_queueRunnerService; /// /// The currently scheduled items @@ -79,41 +78,24 @@ namespace Duplicati.Server /// /// List of update tasks, used to set the timestamp on the schedule once completed /// - private Dictionary> m_updateTasks; + private Dictionary> m_updateTasks; /// /// Constructs a new scheduler /// - public Scheduler() + public Scheduler(IQueueRunnerService queueRunnerService) { - } - - /// - /// Initializes scheduler - /// - /// The worker thread - public void Init(WorkerThread worker) - { - m_worker = worker; + m_queueRunnerService = queueRunnerService; m_thread = new Thread(new ThreadStart(Runner)); - m_worker.CompletedWork += OnCompleted; - m_worker.StartingWork += OnStartingWork; - m_schedule = new KeyValuePair[0]; + m_schedule = []; m_terminate = false; m_event = new AutoResetEvent(false); - m_updateTasks = new Dictionary>(); + m_updateTasks = new Dictionary>(); m_thread.IsBackground = true; m_thread.Name = "TaskScheduler"; m_thread.Start(); } - public IList> GetSchedulerQueueIds() - { - return (from n in WorkerQueue - where n.Backup != null - select new Tuple(n.TaskID, n.Backup.ID)).ToList(); - } - public IList> GetProposedSchedule() { return ( @@ -147,14 +129,6 @@ namespace Duplicati.Server } } - /// - /// A snapshot copy of the current worker queue, that is items that are scheduled, but waiting for execution - /// - public List WorkerQueue - { - get { return m_worker?.CurrentTasks?.Where(t => t != null)?.ToList() ?? []; } - } - /// /// Terminates the thread. Any items still in queue will be removed /// @@ -233,9 +207,9 @@ namespace Duplicati.Server return res; } - private void OnCompleted(WorkerThread worker, Runner.IRunnerData task) + private Task OnCompleted(Runner.IRunnerData task) { - Tuple t = null; + Tuple? t = null; lock (m_lock) { if (task != null && m_updateTasks.TryGetValue(task, out t)) @@ -248,24 +222,26 @@ namespace Duplicati.Server t.Item1.LastRun = t.Item3; FIXMEGlobal.DataConnection.AddOrUpdateSchedule(t.Item1); } + + return Task.CompletedTask; } - private void OnStartingWork(WorkerThread worker, Runner.IRunnerData task) + private Task OnStartingWork(Runner.IRunnerData task) { if (task is null) - { - return; - } + return Task.CompletedTask; lock (m_lock) { - if (m_updateTasks.TryGetValue(task, out Tuple scheduleInfo)) + if (m_updateTasks.TryGetValue(task, out var scheduleInfo)) { // Item2 is the scheduled start time (Time in the Schedule table). // Item3 is the actual start time (LastRun in the Schedule table). m_updateTasks[task] = Tuple.Create(scheduleInfo.Item1, scheduleInfo.Item2, DateTime.UtcNow); } } + + return Task.CompletedTask; } /// @@ -328,33 +304,33 @@ namespace Duplicati.Server .Select(x => x.ToString())) { //See if it is already queued - var tmplst = from n in m_worker.CurrentTasks - where n.Operation == Duplicati.Server.Serialization.DuplicatiOperation.Backup - select n.Backup; - var tastTemp = m_worker.CurrentTask; + var tmplst = from n in m_queueRunnerService.GetCurrentTasks() + where n.Operation == Serialization.DuplicatiOperation.Backup + select n.BackupID; + var tastTemp = m_queueRunnerService.GetCurrentTask(); if (tastTemp != null && tastTemp.Operation == - Duplicati.Server.Serialization.DuplicatiOperation.Backup) - tmplst = tmplst.Union(new[] { tastTemp.Backup }); + Serialization.DuplicatiOperation.Backup) + tmplst = tmplst.Union(new[] { tastTemp.BackupID }); //If it is not already in queue, put it there - if (!tmplst.Any(x => x.ID == id)) + if (!tmplst.Any(x => x == id)) { var entry = FIXMEGlobal.DataConnection.GetBackup(id); if (entry != null) { - Dictionary options = Duplicati.Server.Runner.GetCommonOptions(); - Duplicati.Server.Runner.ApplyOptions(entry, options); - if ((new Duplicati.Library.Main.Options(options)).DisableOnBattery && - (Duplicati.Library.Utility.Power.PowerSupply.GetSource() == - Duplicati.Library.Utility.Power.PowerSupply.Source.Battery)) + var options = Server.Runner.GetCommonOptions(); + Server.Runner.ApplyOptions(entry, options); + if (new Library.Main.Options(options).DisableOnBattery && + (Library.Utility.Power.PowerSupply.GetSource() == + Library.Utility.Power.PowerSupply.Source.Battery)) { - Duplicati.Library.Logging.Log.WriteInformationMessage(LOGTAG, + Library.Logging.Log.WriteInformationMessage(LOGTAG, "BackupDisabledOnBattery", "Scheduled backup disabled while on battery power."); } else { - Dictionary taskOptions = null; + Dictionary? taskOptions = null; try { var nextRun = GetNextValidTime(start, @@ -362,15 +338,17 @@ namespace Duplicati.Server Math.Max(DateTime.UtcNow.AddSeconds(1).Ticks, start.AddSeconds(1).Ticks), DateTimeKind.Utc), sc.Repeat, sc.AllowedDays, timeZoneInfo); - taskOptions = new Dictionary() + taskOptions = new Dictionary() { { "next-scheduled-run", Utility.SerializeDateTime(nextRun.ToUniversalTime()) } }; } catch { } - jobsToRun.Add(Server.Runner.CreateTask( - Serialization.DuplicatiOperation.Backup, entry, taskOptions)); + var job = Server.Runner.CreateTask(Serialization.DuplicatiOperation.Backup, entry, taskOptions); + job.OnStarting = () => OnStartingWork(job); + job.OnFinished = (_) => OnCompleted(job); + jobsToRun.Add(job); } } } @@ -391,7 +369,7 @@ namespace Duplicati.Server continue; } - Server.Runner.IRunnerData lastJob = jobsToRun.LastOrDefault(); + var lastJob = jobsToRun.LastOrDefault(); if (lastJob != null) { lock (m_lock) @@ -403,7 +381,7 @@ namespace Duplicati.Server } foreach (var job in jobsToRun) - m_worker.AddTask(job); + m_queueRunnerService.AddTask(job); if (start < DateTime.UtcNow) { @@ -428,11 +406,6 @@ namespace Duplicati.Server foreach (var c in (from n in scheduled where !existing.ContainsKey(n.Key) select n.Key).ToArray()) scheduled.Remove(c); - //Raise event if needed - // TODO: This triggers a new data event and a reconnect with long-poll - if (NewSchedule != null) - NewSchedule(this, null); - int waittime = 0; //Figure out a sensible amount of time to sleep the thread diff --git a/Duplicati/Library/Utility/Utility.cs b/Duplicati/Library/Utility/Utility.cs index c83243532..7f54fb5a4 100644 --- a/Duplicati/Library/Utility/Utility.cs +++ b/Duplicati/Library/Utility/Utility.cs @@ -1397,7 +1397,7 @@ namespace Duplicati.Library.Utility /// The wrapped commandline element. /// The argument to wrap. /// A flag indicating if environment variables are allowed to be expanded - public static string WrapCommandLineElement(string arg, bool allowEnvExpansion) + public static string WrapCommandLineElement(string? arg, bool allowEnvExpansion) { if (string.IsNullOrWhiteSpace(arg)) return arg; diff --git a/Duplicati/Library/Utility/WorkerThread.cs b/Duplicati/Library/Utility/WorkerThread.cs deleted file mode 100644 index 285aac182..000000000 --- a/Duplicati/Library/Utility/WorkerThread.cs +++ /dev/null @@ -1,401 +0,0 @@ -// Copyright (C) 2025, The Duplicati Team -// https://duplicati.com, hello@duplicati.com -// -// Permission is hereby granted, free of charge, to any person obtaining a -// copy of this software and associated documentation files (the "Software"), -// to deal in the Software without restriction, including without limitation -// the rights to use, copy, modify, merge, publish, distribute, sublicense, -// and/or sell copies of the Software, and to permit persons to whom the -// Software is furnished to do so, subject to the following conditions: -// -// The above copyright notice and this permission notice shall be included in -// all copies or substantial portions of the Software. -// -// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS -// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING -// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER -// DEALINGS IN THE SOFTWARE. - -using System; -using System.Collections.Generic; -using System.Text; -using System.Threading; - - -// TODO: Delete this class. -// It is essentially a queue that is processed by a worker thread, and can be implemented using a BlockingCollection or similar. - -namespace Duplicati.Library.Utility -{ - /// - /// Class to encapsulate a thread that runs a list of queued operations - /// - /// The type to operate on - public class WorkerThread where Tx : class - { - /// - /// Locking object for shared data - /// - private readonly object m_lock = new object(); - /// - /// The wait event - /// - private readonly AutoResetEvent m_event; - /// - /// The internal list of tasks to perform - /// - private Queue m_tasks; - /// - /// A flag used to terminate the thread - /// - private volatile bool m_terminate; - /// - /// The coordinating thread - /// - private Thread m_thread; - - /// - /// A value indicating if the coordinating thread is running - /// - private volatile bool m_active; - - /// - /// The current task being processed - /// - private Tx m_currentTask; - /// - /// A callback that performs the actual work on the item - /// - private readonly Action m_delegate; - - /// - /// An event that is raised when the runner state changes - /// - public event Action, RunState> WorkerStateChanged; - - /// - /// Event that occurs when a new operation is being processed - /// - public event Action, Tx> StartingWork; - /// - /// Event that occurs when an operation has completed - /// - public event Action, Tx> CompletedWork; - /// - /// Event that occurs when an error is detected - /// - public event Action, Tx, Exception> OnError; - /// - /// An event that occurs when a new task is added to the queue or an existing one is removed - /// - public event Action> WorkQueueChanged; - - /// - /// The internal state - /// - private volatile RunState m_state; - - /// - /// The states the scheduler can take - /// - public enum RunState - { - /// - /// The program is running as normal - /// - Run, - /// - /// The program is suspended by the user - /// - Paused - } - - /// - /// Constructs a new WorkerThread - /// - /// The callback that performs the work - public WorkerThread(Action item, bool paused) - { - m_delegate = item; - m_event = new AutoResetEvent(paused); - m_terminate = false; - m_tasks = new Queue(); - m_state = paused ? WorkerThread.RunState.Paused : WorkerThread.RunState.Run; - - m_thread = new Thread(new ThreadStart(Runner)); - m_thread.IsBackground = true; - m_thread.Name = "WorkerThread<" + typeof(Tx).Name + ">"; - m_thread.Start(); - } - - /// - /// Gets a copy of the current queue - /// - public List CurrentTasks - { - get - { - lock (m_lock) - return new List(m_tasks); - } - - } - - /// - /// Gets a value indicating if the worker is running - /// - public bool Active - { - get { return m_active; } - } - - /// - /// Adds a task to the queue - /// - /// The task to add - public void AddTask(Tx task) - { - lock (m_lock) - { - m_tasks.Enqueue(task); - m_event.Set(); - } - - if (WorkQueueChanged != null) - WorkQueueChanged(this); - } - - /// - /// An overloaded AddTask method that allows a task to skip to the front of a queue - /// It does this by creating a new queue, adding the new task first, and then adding - /// all the old tasks to the new queue. It's cleaner to use a linked list, - /// but the performance difference is negligible on such a small queue. - /// - /// Task. - /// If set to true skip queue. - public void AddTask(Tx task, bool skipQueue) - { - if (!skipQueue) - { - // Fall back to default AddTask method - AddTask(task); - return; - } - - lock (m_lock) - { - Queue newQueue = new Queue(); - newQueue.Enqueue(task); - while (m_tasks.Count > 0) - { - Tx n = m_tasks.Dequeue(); - newQueue.Enqueue(n); - } - m_tasks = newQueue; - m_event.Set(); - } - - if (WorkQueueChanged != null) - WorkQueueChanged(this); - } - - - /// - /// Removes a task from the queue, does not remove the task if it is currently running - /// - /// The task to remove - public void RemoveTask(Tx task) - { - lock (m_lock) - { - Queue tmp = new Queue(); - while (m_tasks.Count > 0) - { - Tx n = m_tasks.Dequeue(); - if (n != task) - tmp.Enqueue(n); - } - - m_tasks = tmp; - } - - if (WorkQueueChanged != null) - WorkQueueChanged(this); - } - - /// - /// This will clear the pending queue - /// True if the current running thread should be aborted - /// - public void ClearQueue(bool abortThread) - { - lock (m_lock) - m_tasks.Clear(); - - if (abortThread) - { - try - { - m_thread.Interrupt(); - m_thread.Join(500); - } - catch - { - } - - m_thread = new Thread(new ThreadStart(Runner)); - m_thread.Start(); - } - } - - /// - /// Gets a reference to the currently executing task. - /// BEWARE: This is not protected by a mutex, DO NOT MODIFY IT!!!! - /// - public Tx CurrentTask - { - get - { - return m_currentTask; - } - } - - /// - /// Terminates the thread. Any items still in queue will be removed - /// - /// True if the call should block until the thread has exited, false otherwise - public void Terminate(bool wait) - { - m_terminate = true; - m_event.Set(); - - if (wait) - m_thread.Join(); - } - - /// - /// This is the thread entry point - /// - private void Runner() - { - while (!m_terminate) - { - m_currentTask = null; - - lock (m_lock) - if (m_state == WorkerThread.RunState.Run && m_tasks.Count > 0) - m_currentTask = m_tasks.Dequeue(); - - if (m_currentTask == null && !m_terminate) - { - if (m_state == WorkerThread.RunState.Run) - m_event.WaitOne(); //Sleep until signaled - else - { - if (WorkerStateChanged != null) - WorkerStateChanged(this, m_state); - - //Sleep for brief periods, until signaled - while (!m_terminate && m_state != WorkerThread.RunState.Run) - m_event.WaitOne(1000 * 60 * 5, false); - - //If we were not terminated, we are now ready to run - if (!m_terminate) - { - m_state = WorkerThread.RunState.Run; - if (WorkerStateChanged != null) - WorkerStateChanged(this, m_state); - } - } - } - - if (m_terminate) - return; - - if (m_currentTask == null && m_state == WorkerThread.RunState.Run) - lock (m_lock) - if (m_tasks.Count > 0) - m_currentTask = m_tasks.Dequeue(); - - if (m_currentTask == null) - continue; - - if (StartingWork != null) - StartingWork(this, m_currentTask); - - try - { - m_active = true; - m_delegate(m_currentTask); - } - catch (Exception ex) - { - //TODO: Here where Thread.ResetAbort() was called we shall integrate the CancelationToken pattern. - if (OnError != null) - try { OnError(this, m_currentTask, ex); } - catch - { - // ignored - } - } - finally - { - //TODO: Here where Thread.ResetAbort() was called we shall integrate the CancelationToken pattern. - m_active = false; - } - - var task = m_currentTask; - m_currentTask = null; - - if (CompletedWork != null) - try { CompletedWork(this, task); } - catch (Exception ex) - { - try { OnError(this, task, ex); } - catch - { - // ignored - } - } - } - } - - /// - /// Gets the current run state - /// - public RunState State { get { return m_state; } } - - /// - /// Instructs Duplicati to run scheduled backups - /// - public void Resume() - { - m_state = RunState.Run; - m_event.Set(); - } - - /// - /// Instructs Duplicati to pause scheduled backups - /// - public void Pause() - { - m_state = RunState.Paused; - m_event.Set(); - } - - /// - /// Waits the specified number of milliseconds for the thread to terminate - /// - /// The number of milliseconds to wait - /// True if the thread is terminated, false if a timeout occured - public bool Join(int millisecondTimeout) - { - if (m_thread != null) - return m_thread.Join(millisecondTimeout); - return true; - } - } -} \ No newline at end of file diff --git a/Duplicati/Server/Duplicati.Server.Serialization/Interface/IQueuedTask.cs b/Duplicati/Server/Duplicati.Server.Serialization/Interface/IQueuedTask.cs index 385287292..cb95b681a 100644 --- a/Duplicati/Server/Duplicati.Server.Serialization/Interface/IQueuedTask.cs +++ b/Duplicati/Server/Duplicati.Server.Serialization/Interface/IQueuedTask.cs @@ -17,16 +17,75 @@ // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER -// DEALINGS IN THE SOFTWARE. - -namespace Duplicati.Server.Serialization.Interface -{ - public interface IQueuedTask - { - long TaskID { get; } - string BackupID { get; } - Duplicati.Server.Serialization.DuplicatiOperation Operation { get; } - - } -} - +// DEALINGS IN THE SOFTWARE. + +#nullable enable + +using System; +using System.Threading.Tasks; + +namespace Duplicati.Server.Serialization.Interface; + +/// +/// Represents a queued task. +/// +public interface IQueuedTask +{ + /// + /// The task ID. + /// + long TaskID { get; } + /// + /// The backup ID, if applicable. + /// + string? BackupID { get; } + /// + /// The operation type of the task. + /// + DuplicatiOperation Operation { get; } + /// + /// Callback to be executed when the task is starting. + /// + Func? OnStarting { get; set; } + /// + /// Callback to be executed when the task is finished. + /// If the task completes successfully, the exception parameter will be null. + /// + Func? OnFinished { get; set; } + + /// + /// That action that performs the task. + /// + Task Execute(); + /// + /// Updates the throttle speeds for the task. + /// + /// The upload speed to set. + /// The download speed to set. + void UpdateThrottleSpeeds(string? uploadSpeed, string? downloadSpeed); + /// + /// The time when the task was starting to execute. + /// + DateTime? TaskStarted { get; set; } + /// + /// The time when the task was finished executing. + /// + DateTime? TaskFinished { get; set; } + /// + /// Stops the task. + /// + void Stop(); + /// + /// Aborts the task. + /// + void Abort(); + /// + /// Pauses the task. + /// + /// If true, also pauses transfers. + void Pause(bool alsoTransfers); + /// + /// Resumes the task. + /// + void Resume(); +} diff --git a/Duplicati/Server/Program.cs b/Duplicati/Server/Program.cs index 55c734049..933cca135 100644 --- a/Duplicati/Server/Program.cs +++ b/Duplicati/Server/Program.cs @@ -102,7 +102,7 @@ namespace Duplicati.Server /// /// This is the scheduling thread /// - public static IScheduler Scheduler { get => FIXMEGlobal.Scheduler; } + public static ISchedulerService Scheduler { get => FIXMEGlobal.Scheduler; } /// /// The thread running the ping-pong handler @@ -243,6 +243,7 @@ namespace Duplicati.Server var crashed = false; var terminated = false; + IQueueRunnerService queueRunner = null; try { DataConnection = GetDatabaseConnection(commandlineOptions, silentConsole); @@ -266,13 +267,14 @@ namespace Duplicati.Server DuplicatiWebserver = StartWebServer(commandlineOptions, DataConnection).Await(); + queueRunner = DuplicatiWebserver.Provider.GetRequiredService(); + DataConnection.SetServiceProvider(DuplicatiWebserver.Provider); + UpdatePoller.Init(Library.Utility.Utility.ParseBoolOption(commandlineOptions, DISABLE_UPDATE_CHECK_OPTION)); SetPurgeTempFilesTimer(commandlineOptions); - LiveControl.StateChanged = LiveControl_StateChanged; - - SetWorkerThread(); + LiveControl.StateChanged = (e) => { LiveControl_StateChanged(queueRunner, DataConnection, StatusEventNotifyer, e); }; if (Library.Utility.Utility.ParseBoolOption(commandlineOptions, PING_PONG_KEEPALIVE_OPTION)) { @@ -363,7 +365,7 @@ namespace Duplicati.Server () => { if (ShutdownModernWebserver != null) ShutdownModernWebserver(); }, () => UpdatePoller?.Terminate(), () => Scheduler?.Terminate(true), - () => FIXMEGlobal.WorkThread?.Terminate(true), + () => queueRunner?.Terminate(true), () => ApplicationInstance?.Dispose(), () => PurgeTempFilesTimer?.Dispose(), () => Library.UsageReporter.Reporter.ShutDown(), @@ -432,36 +434,6 @@ namespace Duplicati.Server return server; } - private static void SetWorkerThread() - { - FIXMEGlobal.WorkerThreadsManager.Spawn(x => { Runner.Run(x, true); }); - FIXMEGlobal.WorkThread.StartingWork += (worker, task) => - { - SignalNewEvent(null, null); - task.TaskStarted = DateTime.Now; - }; - FIXMEGlobal.WorkThread.CompletedWork += (worker, task) => - { - SignalNewEvent(null, null); - FIXMEGlobal.Provider.GetRequiredService()?.AddTaskResult(new CachedTaskResult(task.TaskID, task.BackupID, task.TaskStarted, task.TaskFinished ?? DateTime.Now, null)); - }; - FIXMEGlobal.WorkThread.WorkQueueChanged += (worker) => { SignalNewEvent(null, null); }; - FIXMEGlobal.Scheduler.SubScribeToNewSchedule(() => SignalNewEvent(null, null)); - FIXMEGlobal.WorkThread.OnError += (worker, task, exception) => - { - DataConnection.LogError(task?.BackupID, "Error in worker", exception); - FIXMEGlobal.Provider.GetRequiredService()?.AddTaskResult(new CachedTaskResult(task.TaskID, task.BackupID, task.TaskStarted, task.TaskFinished ?? DateTime.Now, exception)); - }; - - var lastScheduleId = FIXMEGlobal.NotificationUpdateService.LastDataUpdateId; - StatusEventNotifyer.NewEvent += (sender, e) => - { - if (lastScheduleId == FIXMEGlobal.NotificationUpdateService.LastDataUpdateId) return; - lastScheduleId = FIXMEGlobal.NotificationUpdateService.LastDataUpdateId; - Scheduler.Reschedule(); - }; - } - private static void SetPurgeTempFilesTimer(Dictionary commandlineOptions) { var lastPurge = new DateTime(0); @@ -881,23 +853,22 @@ namespace Duplicati.Server /// This event handler updates the trayicon menu with the current state of the runner. /// /// - private static void LiveControl_StateChanged(LiveControls.LiveControlEvent e) + private static void LiveControl_StateChanged(IQueueRunnerService queueRunnerService, Connection connection, EventPollNotify eventPollNotify, LiveControls.LiveControlEvent e) { - var worker = FIXMEGlobal.WorkThread; - var appSettings = FIXMEGlobal.DataConnection.ApplicationSettings; + var appSettings = connection.ApplicationSettings; switch (e.State) { case LiveControls.LiveControlState.Paused: { - worker.Pause(); - worker.CurrentTask?.Pause(e.TransfersPaused); + queueRunnerService.Pause(); + queueRunnerService.GetCurrentTask()?.Pause(e.TransfersPaused); appSettings.PausedUntil = e.WaitTimeExpiration; break; } case LiveControls.LiveControlState.Running: { - worker.Resume(); - worker.CurrentTask?.Resume(); + queueRunnerService.Resume(); + queueRunnerService.GetCurrentTask()?.Resume(); appSettings.PausedUntil = null; break; } @@ -906,7 +877,7 @@ namespace Duplicati.Server break; } - StatusEventNotifyer.SignalNewEvent(); + eventPollNotify.SignalNewEvent(); } diff --git a/Duplicati/WebserverCore/Dto/ServerStatusDto.cs b/Duplicati/WebserverCore/Dto/ServerStatusDto.cs index 8426d0675..97b09011a 100644 --- a/Duplicati/WebserverCore/Dto/ServerStatusDto.cs +++ b/Duplicati/WebserverCore/Dto/ServerStatusDto.cs @@ -30,7 +30,7 @@ public sealed record ServerStatusDto /// /// Gets or sets the active task. /// - public required Tuple? ActiveTask { get; init; } + public required Tuple? ActiveTask { get; init; } /// /// Gets or sets the state of the program. @@ -40,7 +40,7 @@ public sealed record ServerStatusDto /// /// Gets the IDs of the tasks in the scheduler queue. /// - public required IList> SchedulerQueueIds { get; init; } = []; + public required IList> SchedulerQueueIds { get; init; } = []; /// /// Gets or sets the proposed schedule. diff --git a/Duplicati/WebserverCore/Endpoints/V1/Backup/BackupGet.cs b/Duplicati/WebserverCore/Endpoints/V1/Backup/BackupGet.cs index 038a1a429..acad0cfac 100644 --- a/Duplicati/WebserverCore/Endpoints/V1/Backup/BackupGet.cs +++ b/Duplicati/WebserverCore/Endpoints/V1/Backup/BackupGet.cs @@ -21,7 +21,6 @@ using System.Text.Json; using System.Text.Json.Serialization; using Duplicati.Library.Interface; -using Duplicati.Library.RestAPI.Abstractions; using Duplicati.Server; using Duplicati.Server.Database; using Duplicati.Server.Serialization; @@ -86,8 +85,8 @@ public class BackupGet : IEndpointV1 => ExecuteGetIsdbUsedElsewhere(GetBackup(connection, id))) .RequireAuthorization(); - group.MapGet("/backup/{id}/isactive", ([FromServices] Connection connection, [FromServices] IWorkerThreadsManager workerThreadsManager, [FromRoute] string id) - => ExecuteGetIsActive(workerThreadsManager, GetBackup(connection, id))) + group.MapGet("/backup/{id}/isactive", ([FromServices] Connection connection, [FromServices] IQueueRunnerService queueRunnerService, [FromRoute] string id) + => ExecuteGetIsActive(queueRunnerService, GetBackup(connection, id))) .RequireAuthorization(); } @@ -152,7 +151,7 @@ public class BackupGet : IEndpointV1 if (!allVersions) time = Library.Utility.Timeparser.ParseTimeInterval(timestring, DateTime.Now); - var r = Runner.Run(Runner.CreateListTask(backup, [filter], prefixOnly, allVersions, folderContents, time), false) as Duplicati.Library.Interface.IListResults; + var r = Runner.Run(Runner.CreateListTask(backup, filter == null ? null : [filter], prefixOnly, allVersions, folderContents, time), false) as Duplicati.Library.Interface.IListResults; if (r == null) throw new ServerErrorException("No result from list operation"); @@ -205,7 +204,7 @@ public class BackupGet : IEndpointV1 private static IEnumerable ExecuteGetFilesets(IBackup bk, bool includeMetadata, bool fromRemoteOnly) { - var extra = new Dictionary + var extra = new Dictionary { ["list-sets-only"] = "true" }; @@ -304,17 +303,14 @@ public class BackupGet : IEndpointV1 private static Dto.IsDbUsedElsewhereDto ExecuteGetIsdbUsedElsewhere(IBackup bk) => new Dto.IsDbUsedElsewhereDto(Library.Main.CLIDatabaseLocator.IsDatabasePathInUse(bk.DBPath)); - private static Dto.IsBackupActiveDto ExecuteGetIsActive(IWorkerThreadsManager workerThreadsManager, IBackup bk) + private static Dto.IsBackupActiveDto ExecuteGetIsActive(IQueueRunnerService queueRunnerService, IBackup bk) { - if (workerThreadsManager.WorkerThread == null) - throw new InvalidOperationException("Worker thread not available"); - - var t = workerThreadsManager.WorkerThread.CurrentTask; - var bt = t?.Backup; - if (bt != null && bk.ID == bt.ID) + var t = queueRunnerService.GetCurrentTask(); + var bt = t?.BackupID; + if (bt != null && bk.ID == bt) return new Dto.IsBackupActiveDto("OK", true); - if (workerThreadsManager.WorkerThread.CurrentTasks.Any(x => x?.Backup == null || x.Backup.ID == bk.ID)) + if (queueRunnerService.GetCurrentTasks().Any(x => x?.BackupID == null || x.BackupID == bk.ID)) return new Dto.IsBackupActiveDto("OK", true); return new Dto.IsBackupActiveDto("OK", false); diff --git a/Duplicati/WebserverCore/Endpoints/V1/Backup/BackupPost.cs b/Duplicati/WebserverCore/Endpoints/V1/Backup/BackupPost.cs index 8b58ff846..3b40ff06d 100644 --- a/Duplicati/WebserverCore/Endpoints/V1/Backup/BackupPost.cs +++ b/Duplicati/WebserverCore/Endpoints/V1/Backup/BackupPost.cs @@ -18,7 +18,6 @@ // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER // DEALINGS IN THE SOFTWARE. -using Duplicati.Library.RestAPI.Abstractions; using Duplicati.Server; using Duplicati.Server.Database; using Duplicati.Server.Serialization; @@ -45,44 +44,44 @@ public class BackupPost : IEndpointV1 => UpdateDatabasePath(connection, GetBackup(connection, id), input.path, false)) .RequireAuthorization(); - group.MapPost("/backup/{id}/restore", ([FromServices] Connection connection, [FromServices] IWorkerThreadsManager workerThreadsManager, [FromRoute] string id, [FromBody] Dto.RestoreInputDto input) - => ExecuteRestore(GetBackup(connection, id), workerThreadsManager, input)) + group.MapPost("/backup/{id}/restore", ([FromServices] Connection connection, [FromServices] IQueueRunnerService queueRunnerService, [FromRoute] string id, [FromBody] Dto.RestoreInputDto input) + => ExecuteRestore(GetBackup(connection, id), queueRunnerService, input)) .RequireAuthorization(); - group.MapPost("/backup/{id}/createreport", ([FromServices] Connection connection, [FromServices] IWorkerThreadsManager workerThreadsManager, [FromRoute] string id) - => ExecuteCreateReport(GetBackup(connection, id), workerThreadsManager)) + group.MapPost("/backup/{id}/createreport", ([FromServices] Connection connection, [FromServices] IQueueRunnerService queueRunnerService, [FromRoute] string id) + => ExecuteCreateReport(GetBackup(connection, id), queueRunnerService)) .RequireAuthorization(); - group.MapPost("/backup/{id}/repair", ([FromServices] Connection connection, [FromServices] IWorkerThreadsManager workerThreadsManager, [FromRoute] string id, Dto.RepairInputDto? input) - => ExecuteRepair(GetBackup(connection, id), workerThreadsManager, input)) + group.MapPost("/backup/{id}/repair", ([FromServices] Connection connection, [FromServices] IQueueRunnerService queueRunnerService, [FromRoute] string id, Dto.RepairInputDto? input) + => ExecuteRepair(GetBackup(connection, id), queueRunnerService, input)) .RequireAuthorization(); - group.MapPost("/backup/{id}/repairupdate", ([FromServices] Connection connection, [FromServices] IWorkerThreadsManager workerThreadsManager, [FromRoute] string id, Dto.RepairInputDto? input) - => ExecuteRepairUpdate(GetBackup(connection, id), workerThreadsManager, input)) + group.MapPost("/backup/{id}/repairupdate", ([FromServices] Connection connection, [FromServices] IQueueRunnerService queueRunnerService, [FromRoute] string id, Dto.RepairInputDto? input) + => ExecuteRepairUpdate(GetBackup(connection, id), queueRunnerService, input)) .RequireAuthorization(); - group.MapPost("/backup/{id}/vacuum", ([FromServices] Connection connection, [FromServices] IWorkerThreadsManager workerThreadsManager, [FromRoute] string id) - => ExecuteVacuum(GetBackup(connection, id), workerThreadsManager)) + group.MapPost("/backup/{id}/vacuum", ([FromServices] Connection connection, [FromServices] IQueueRunnerService queueRunnerService, [FromRoute] string id) + => ExecuteVacuum(GetBackup(connection, id), queueRunnerService)) .RequireAuthorization(); - group.MapPost("/backup/{id}/verify", ([FromServices] Connection connection, [FromServices] IWorkerThreadsManager workerThreadsManager, [FromRoute] string id) - => ExecuteVerify(GetBackup(connection, id), workerThreadsManager)) + group.MapPost("/backup/{id}/verify", ([FromServices] Connection connection, [FromServices] IQueueRunnerService queueRunnerService, [FromRoute] string id) + => ExecuteVerify(GetBackup(connection, id), queueRunnerService)) .RequireAuthorization(); - group.MapPost("/backup/{id}/compact", ([FromServices] Connection connection, [FromServices] IWorkerThreadsManager workerThreadsManager, [FromRoute] string id) - => ExecuteCompact(GetBackup(connection, id), workerThreadsManager)) + group.MapPost("/backup/{id}/compact", ([FromServices] Connection connection, [FromServices] IQueueRunnerService queueRunnerService, [FromRoute] string id) + => ExecuteCompact(GetBackup(connection, id), queueRunnerService)) .RequireAuthorization(); - group.MapPost("/backup/{id}/start", ([FromServices] Connection connection, [FromServices] IWorkerThreadsManager workerThreadsManager, [FromRoute] string id) - => ExecuteRunBackup(GetBackup(connection, id), workerThreadsManager)) + group.MapPost("/backup/{id}/start", ([FromServices] Connection connection, [FromServices] IQueueRunnerService queueRunnerService, [FromRoute] string id) + => ExecuteRunBackup(GetBackup(connection, id), queueRunnerService)) .RequireAuthorization(); - group.MapPost("/backup/{id}/run", ([FromServices] Connection connection, [FromServices] IWorkerThreadsManager workerThreadsManager, [FromRoute] string id) - => ExecuteRunBackup(GetBackup(connection, id), workerThreadsManager)) + group.MapPost("/backup/{id}/run", ([FromServices] Connection connection, [FromServices] IQueueRunnerService queueRunnerService, [FromRoute] string id) + => ExecuteRunBackup(GetBackup(connection, id), queueRunnerService)) .RequireAuthorization(); - group.MapPost("/backup/{id}/report-remote-size", ([FromServices] Connection connection, [FromServices] IWorkerThreadsManager workerThreadsManager, [FromRoute] string id) - => ExecuteReportRemoteSize(GetBackup(connection, id), workerThreadsManager)) + group.MapPost("/backup/{id}/report-remote-size", ([FromServices] Connection connection, [FromServices] IQueueRunnerService queueRunnerService, [FromRoute] string id) + => ExecuteReportRemoteSize(GetBackup(connection, id), queueRunnerService)) .RequireAuthorization(); group.MapPost("/backup/{id}/copytotemp", ([FromServices] Connection connection, [FromRoute] string id) @@ -114,8 +113,8 @@ public class BackupPost : IEndpointV1 - private static Dto.TaskStartedDto ExecuteRestore(IBackup backup, IWorkerThreadsManager workerThreadsManager, Dto.RestoreInputDto input) - => new Dto.TaskStartedDto("OK", workerThreadsManager.AddTask(Runner.CreateRestoreTask( + private static Dto.TaskStartedDto ExecuteRestore(IBackup backup, IQueueRunnerService queueRunnerService, Dto.RestoreInputDto input) + => new Dto.TaskStartedDto("OK", queueRunnerService.AddTask(Runner.CreateRestoreTask( backup, input.paths ?? [], Library.Utility.Timeparser.ParseTimeInterval(input.time, DateTime.Now), @@ -125,31 +124,31 @@ public class BackupPost : IEndpointV1 input.skip_metadata ?? false, string.IsNullOrWhiteSpace(input.passphrase) ? null : input.passphrase))); - private static Dto.TaskStartedDto ExecuteCreateReport(IBackup backup, IWorkerThreadsManager workerThreadsManager) - => new Dto.TaskStartedDto("OK", workerThreadsManager.AddTask(Runner.CreateTask(DuplicatiOperation.CreateReport, backup))); + private static Dto.TaskStartedDto ExecuteCreateReport(IBackup backup, IQueueRunnerService queueRunnerService) + => new Dto.TaskStartedDto("OK", queueRunnerService.AddTask(Runner.CreateTask(DuplicatiOperation.CreateReport, backup))); - private static Dto.TaskStartedDto ExecuteReportRemoteSize(IBackup backup, IWorkerThreadsManager workerThreadsManager) - => new Dto.TaskStartedDto("OK", workerThreadsManager.AddTask(Runner.CreateTask(DuplicatiOperation.ListRemote, backup))); + private static Dto.TaskStartedDto ExecuteReportRemoteSize(IBackup backup, IQueueRunnerService queueRunnerService) + => new Dto.TaskStartedDto("OK", queueRunnerService.AddTask(Runner.CreateTask(DuplicatiOperation.ListRemote, backup))); - private static Dto.TaskStartedDto ExecuteRepair(IBackup backup, IWorkerThreadsManager workerThreadsManager, Dto.RepairInputDto? input) - => DoRepair(backup, false, workerThreadsManager, input); + private static Dto.TaskStartedDto ExecuteRepair(IBackup backup, IQueueRunnerService queueRunnerService, Dto.RepairInputDto? input) + => DoRepair(backup, false, queueRunnerService, input); - private static Dto.TaskStartedDto ExecuteRepairUpdate(IBackup backup, IWorkerThreadsManager workerThreadsManager, Dto.RepairInputDto? input) - => DoRepair(backup, true, workerThreadsManager, input); + private static Dto.TaskStartedDto ExecuteRepairUpdate(IBackup backup, IQueueRunnerService queueRunnerService, Dto.RepairInputDto? input) + => DoRepair(backup, true, queueRunnerService, input); - private static Dto.TaskStartedDto ExecuteVacuum(IBackup backup, IWorkerThreadsManager workerThreadsManager) - => new Dto.TaskStartedDto("OK", workerThreadsManager.AddTask(Runner.CreateTask(DuplicatiOperation.Vacuum, backup))); + private static Dto.TaskStartedDto ExecuteVacuum(IBackup backup, IQueueRunnerService queueRunnerService) + => new Dto.TaskStartedDto("OK", queueRunnerService.AddTask(Runner.CreateTask(DuplicatiOperation.Vacuum, backup))); - private static Dto.TaskStartedDto ExecuteVerify(IBackup backup, IWorkerThreadsManager workerThreadsManager) - => new Dto.TaskStartedDto("OK", workerThreadsManager.AddTask(Runner.CreateTask(DuplicatiOperation.Verify, backup))); + private static Dto.TaskStartedDto ExecuteVerify(IBackup backup, IQueueRunnerService queueRunnerService) + => new Dto.TaskStartedDto("OK", queueRunnerService.AddTask(Runner.CreateTask(DuplicatiOperation.Verify, backup))); - private static Dto.TaskStartedDto ExecuteCompact(IBackup backup, IWorkerThreadsManager workerThreadsManager) - => new Dto.TaskStartedDto("OK", workerThreadsManager.AddTask(Runner.CreateTask(DuplicatiOperation.Compact, backup))); + private static Dto.TaskStartedDto ExecuteCompact(IBackup backup, IQueueRunnerService queueRunnerService) + => new Dto.TaskStartedDto("OK", queueRunnerService.AddTask(Runner.CreateTask(DuplicatiOperation.Compact, backup))); - private static Dto.TaskStartedDto DoRepair(IBackup backup, bool repairUpdate, IWorkerThreadsManager workerThreadsManager, Dto.RepairInputDto? input) + private static Dto.TaskStartedDto DoRepair(IBackup backup, bool repairUpdate, IQueueRunnerService queueRunnerService, Dto.RepairInputDto? input) { // These are all props on the input object - var extra = new Dictionary(); + var extra = new Dictionary(); if (input != null) { if (input.only_paths.HasValue) @@ -168,23 +167,23 @@ public class BackupPost : IEndpointV1 var filters = input?.paths ?? []; - return new Dto.TaskStartedDto("OK", workerThreadsManager.AddTask(Runner.CreateTask(repairUpdate ? DuplicatiOperation.RepairUpdate : DuplicatiOperation.Repair, backup, extra, filters))); + return new Dto.TaskStartedDto("OK", queueRunnerService.AddTask(Runner.CreateTask(repairUpdate ? DuplicatiOperation.RepairUpdate : DuplicatiOperation.Repair, backup, extra, filters))); } - private static Dto.TaskStartedDto ExecuteRunBackup(IBackup backup, IWorkerThreadsManager workerThreadsManager) + private static Dto.TaskStartedDto ExecuteRunBackup(IBackup backup, IQueueRunnerService queueRunnerService) { - var t = workerThreadsManager.WorkerThread?.CurrentTask; - var bt = t?.Backup; + var t = queueRunnerService.GetCurrentTask(); + var bt = t?.BackupID; // Already running - if (bt != null && backup.ID == bt.ID) + if (bt != null && backup.ID == bt) return new Dto.TaskStartedDto("OK", t!.TaskID); - t = workerThreadsManager.WorkerThread?.CurrentTasks.FirstOrDefault(x => x?.Backup != null && x.Backup.ID == backup.ID); + t = queueRunnerService.GetCurrentTasks().FirstOrDefault(x => x.BackupID == backup.ID); if (t != null) return new Dto.TaskStartedDto("OK", t.TaskID); - return new Dto.TaskStartedDto("OK", workerThreadsManager.AddTask(Runner.CreateTask(DuplicatiOperation.Backup, backup), true)); + return new Dto.TaskStartedDto("OK", queueRunnerService.AddTask(Runner.CreateTask(DuplicatiOperation.Backup, backup), true)); } private class WrappedBackup : Server.Database.Backup diff --git a/Duplicati/WebserverCore/Endpoints/V1/Backup/BackupPutDelete.cs b/Duplicati/WebserverCore/Endpoints/V1/Backup/BackupPutDelete.cs index 8bb4ce4b4..754cab235 100644 --- a/Duplicati/WebserverCore/Endpoints/V1/Backup/BackupPutDelete.cs +++ b/Duplicati/WebserverCore/Endpoints/V1/Backup/BackupPutDelete.cs @@ -18,8 +18,6 @@ // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER // DEALINGS IN THE SOFTWARE. -using System.Text.Json; -using Duplicati.Library.RestAPI.Abstractions; using Duplicati.Server; using Duplicati.Server.Database; using Duplicati.Server.Serialization; @@ -38,9 +36,9 @@ public class BackupPutDelete : IEndpointV1 => ExecutePut(GetBackup(connection, id), connection, input)) .RequireAuthorization(); - group.MapDelete("/backup/{id}", ([FromServices] Connection connection, [FromServices] IWorkerThreadsManager workerThreadsManager, [FromServices] LiveControls liveControls, [FromServices] IHttpContextAccessor httpContextAccessor, [FromRoute] string id, [FromQuery(Name = "delete-remote-files")] bool? delete_remote_files, [FromQuery(Name = "delete-local-db")] bool? delete_local_db, [FromQuery] bool? force) => + group.MapDelete("/backup/{id}", ([FromServices] Connection connection, [FromServices] IQueueRunnerService queueRunnerService, [FromServices] LiveControls liveControls, [FromServices] IHttpContextAccessor httpContextAccessor, [FromRoute] string id, [FromQuery(Name = "delete-remote-files")] bool? delete_remote_files, [FromQuery(Name = "delete-local-db")] bool? delete_local_db, [FromQuery] bool? force) => { - var res = ExecuteDelete(GetBackup(connection, id), workerThreadsManager, liveControls, delete_remote_files ?? false, delete_local_db, force ?? false); + var res = ExecuteDelete(GetBackup(connection, id), queueRunnerService, liveControls, delete_remote_files ?? false, delete_local_db, force ?? false); if (res.Status != "OK" && httpContextAccessor.HttpContext != null) httpContextAccessor.HttpContext.Response.StatusCode = 500; return res; @@ -132,67 +130,56 @@ public class BackupPutDelete : IEndpointV1 } } - private static Dto.DeleteBackupOutputDto ExecuteDelete(IBackup backup, IWorkerThreadsManager workerThreadsManager, LiveControls liveControls, bool delete_remote_files, bool? delete_local_db, bool force) + private static Dto.DeleteBackupOutputDto ExecuteDelete(IBackup backup, IQueueRunnerService queueRunnerService, LiveControls liveControls, bool delete_remote_files, bool? delete_local_db, bool force) { - if (workerThreadsManager.WorkerThread!.Active) + try { - try + var nt = queueRunnerService.GetCurrentTask(); + if (backup.ID == nt?.BackupID) { - //TODO: It's not safe to access the values like this, - //because the runner thread might interfere - var nt = workerThreadsManager.WorkerThread.CurrentTask; - if (backup.Equals(nt?.Backup)) + if (!force) + return new Dto.DeleteBackupOutputDto("failed", "backup-in-progress", nt?.TaskID); + + + bool hasPaused = liveControls.State != LiveControls.LiveControlState.Paused; + if (hasPaused) + liveControls.Pause(true); + nt.Abort(); + + for (int i = 0; i < 10; i++) { - if (!force) - return new Dto.DeleteBackupOutputDto("failed", "backup-in-progress", nt?.TaskID); - - - bool hasPaused = liveControls.State != LiveControls.LiveControlState.Paused; - if (hasPaused) - liveControls.Pause(true); - nt.Abort(); - - for (int i = 0; i < 10; i++) - if (workerThreadsManager.WorkerThread.Active) - { - var t = workerThreadsManager.WorkerThread.CurrentTask; - if (backup.Equals(t == null ? null : t.Backup)) - Thread.Sleep(1000); - else - break; - } - else - break; - - if (workerThreadsManager.WorkerThread.Active) - { - var t = workerThreadsManager.WorkerThread.CurrentTask; - if (backup.Equals(t == null ? null : t.Backup)) - { - if (hasPaused) - liveControls.Resume(); - - return new Dto.DeleteBackupOutputDto("failed", "backup-unstoppable", t?.TaskID); - } - } + var tt = queueRunnerService.GetCurrentTask(); + if (backup.ID == tt?.BackupID) + Thread.Sleep(1000); + else + break; + } + var t = queueRunnerService.GetCurrentTask(); + if (backup.ID == t?.BackupID) + { if (hasPaused) liveControls.Resume(); + + return new Dto.DeleteBackupOutputDto("failed", "backup-unstoppable", t?.TaskID); } - } - catch (Exception ex) - { - return new Dto.DeleteBackupOutputDto("error", ex.Message, null); + + if (hasPaused) + liveControls.Resume(); } } + catch (Exception ex) + { + return new Dto.DeleteBackupOutputDto("error", ex.Message, null); + } - var extra = new Dictionary(); + var extra = new Dictionary(); if (delete_local_db.HasValue) extra["delete-local-db"] = delete_local_db.Value.ToString(); if (delete_remote_files) extra["delete-remote-files"] = "true"; - return new Dto.DeleteBackupOutputDto("OK", null, workerThreadsManager.AddTask(Runner.CreateTask(DuplicatiOperation.Delete, backup, extra))); + return new Dto.DeleteBackupOutputDto("OK", null, queueRunnerService.AddTask(Runner.CreateTask(DuplicatiOperation.Delete, backup, extra))); } } diff --git a/Duplicati/WebserverCore/Endpoints/V1/Tasks.cs b/Duplicati/WebserverCore/Endpoints/V1/Tasks.cs index a1a72cca2..1da85989d 100644 --- a/Duplicati/WebserverCore/Endpoints/V1/Tasks.cs +++ b/Duplicati/WebserverCore/Endpoints/V1/Tasks.cs @@ -34,35 +34,35 @@ public class Tasks : IEndpointV1 } public static void Map(RouteGroupBuilder group) { - group.MapGet("/tasks", Execute).RequireAuthorization(); - group.MapGet("/task/{taskid}", ([FromRoute] long taskId, [FromServices] ITaskCacheService taskCacheService) => ExecuteGet(taskCacheService, taskId)).RequireAuthorization(); - group.MapPost("/task/{taskid}/stop", ([FromRoute] long taskId) => ExecutePost(taskId, TaskStopState.Stop)).RequireAuthorization(); - group.MapPost("/task/{taskid}/abort", ([FromRoute] long taskId) => ExecutePost(taskId, TaskStopState.Abort)).RequireAuthorization(); + group.MapGet("/tasks", ([FromServices] IQueueRunnerService queueRunnerService) => Execute(queueRunnerService)).RequireAuthorization(); + group.MapGet("/task/{taskid}", ([FromRoute] long taskId, [FromServices] IQueueRunnerService queueRunnerService) => ExecuteGet(queueRunnerService, taskId)).RequireAuthorization(); + group.MapPost("/task/{taskid}/stop", ([FromRoute] long taskId, [FromServices] IQueueRunnerService queueRunnerService) => ExecutePost(queueRunnerService, taskId, TaskStopState.Stop)).RequireAuthorization(); + group.MapPost("/task/{taskid}/abort", ([FromRoute] long taskId, [FromServices] IQueueRunnerService queueRunnerService) => ExecutePost(queueRunnerService, taskId, TaskStopState.Abort)).RequireAuthorization(); } - private static IEnumerable Execute() + private static IEnumerable Execute(IQueueRunnerService queueRunnerService) { - var cur = FIXMEGlobal.WorkThread.CurrentTask; - var n = FIXMEGlobal.WorkThread.CurrentTasks; + var cur = queueRunnerService.GetCurrentTask(); + var n = queueRunnerService.GetCurrentTasks(); if (cur != null) n.Insert(0, cur); - return n; + return n.OfType(); } - private static Dto.GetTaskStateDto ExecuteGet(ITaskCacheService taskCacheService, long taskid) + private static Dto.GetTaskStateDto ExecuteGet(IQueueRunnerService queueRunnerService, long taskid) { - var task = FIXMEGlobal.WorkThread.CurrentTask; - var tasks = FIXMEGlobal.WorkThread.CurrentTasks; + var task = queueRunnerService.GetCurrentTask(); + var tasks = queueRunnerService.GetCurrentTasks(); if (task != null && task.TaskID == taskid) return new Dto.GetTaskStateDto("Running", taskid, task.TaskStarted, task.TaskFinished); if (tasks.FirstOrDefault(x => x.TaskID == taskid) == null) { - var res = taskCacheService.GetCachedTaskResults(taskid); + var res = queueRunnerService.GetCachedTaskResults(taskid); if (res == null) throw new NotFoundException("No such task found"); @@ -79,10 +79,10 @@ public class Tasks : IEndpointV1 return new Dto.GetTaskStateDto("Waiting", taskid, null, null); } - private static void ExecutePost(long taskid, TaskStopState stopState) + private static void ExecutePost(IQueueRunnerService queueRunnerService, long taskid, TaskStopState stopState) { - var task = FIXMEGlobal.WorkThread.CurrentTask; - var tasks = FIXMEGlobal.WorkThread.CurrentTasks; + var task = queueRunnerService.GetCurrentTask(); + var tasks = queueRunnerService.GetCurrentTasks(); if (task != null) tasks.Insert(0, task); diff --git a/Duplicati/WebserverCore/Extensions/ServiceCollectionsExtensions.cs b/Duplicati/WebserverCore/Extensions/ServiceCollectionsExtensions.cs index 45264c9fe..ae2a52593 100644 --- a/Duplicati/WebserverCore/Extensions/ServiceCollectionsExtensions.cs +++ b/Duplicati/WebserverCore/Extensions/ServiceCollectionsExtensions.cs @@ -20,7 +20,6 @@ // DEALINGS IN THE SOFTWARE. using Duplicati.Library.IO; using Duplicati.Library.RestAPI; -using Duplicati.Library.RestAPI.Abstractions; using Duplicati.Server; using Duplicati.Server.Database; using Duplicati.Server.Serialization; @@ -44,7 +43,6 @@ public static class ServiceCollectionsExtensions .AddSingleton(Serializer.JsonSettings) .AddSingleton() .AddSingleton() - .AddSingleton() .AddSingleton(connection); @@ -57,8 +55,7 @@ public static class ServiceCollectionsExtensions .AddTransient() .AddTransient() .AddSingleton() - .AddSingleton() - .AddSingleton() + .AddSingleton() .AddSingleton() .AddTransient() .AddSingleton() @@ -69,7 +66,7 @@ public static class ServiceCollectionsExtensions .AddSingleton() .AddSingleton() .AddSingleton() - .AddSingleton(); + .AddSingleton(); return services; } diff --git a/Duplicati/WebserverCore/Services/CommandlineRunService.cs b/Duplicati/WebserverCore/Services/CommandlineRunService.cs index 81bff62b4..3d9747ab1 100644 --- a/Duplicati/WebserverCore/Services/CommandlineRunService.cs +++ b/Duplicati/WebserverCore/Services/CommandlineRunService.cs @@ -20,13 +20,12 @@ // DEALINGS IN THE SOFTWARE. using System.Text; using Duplicati.Library.RestAPI; -using Duplicati.Library.RestAPI.Abstractions; using Duplicati.Server; using Duplicati.WebserverCore.Abstractions; namespace Duplicati.WebserverCore.Services; -public class CommandlineRunService(IWorkerThreadsManager workerThreadsManager) : ICommandlineRunService +public class CommandlineRunService(IQueueRunnerService queueRunnerService) : ICommandlineRunService { private static readonly string LOGTAG = Library.Logging.Log.LogTagFromType(); @@ -180,7 +179,7 @@ public class CommandlineRunService(IWorkerThreadsManager workerThreadsManager) : } }); - workerThreadsManager.AddTask(k.Task); + queueRunnerService.AddTask(k.Task); return k.ID; } diff --git a/Duplicati/WebserverCore/Services/QueueRunnerService.cs b/Duplicati/WebserverCore/Services/QueueRunnerService.cs new file mode 100644 index 000000000..9335bcc93 --- /dev/null +++ b/Duplicati/WebserverCore/Services/QueueRunnerService.cs @@ -0,0 +1,208 @@ +// Copyright (C) 2025, The Duplicati Team +// https://duplicati.com, hello@duplicati.com +// +// Permission is hereby granted, free of charge, to any person obtaining a +// copy of this software and associated documentation files (the "Software"), +// to deal in the Software without restriction, including without limitation +// the rights to use, copy, modify, merge, publish, distribute, sublicense, +// and/or sell copies of the Software, and to permit persons to whom the +// Software is furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS +// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +// DEALINGS IN THE SOFTWARE. + +using Duplicati.Library.Utility; +using Duplicati.Server; +using Duplicati.Server.Database; +using Duplicati.Server.Serialization.Interface; +using Duplicati.WebserverCore.Abstractions; + +namespace Duplicati.WebserverCore.Services; + +/// +/// Simple queue that will run the given task +/// +public class QueueRunnerService(Connection connection, EventPollNotify eventPollNotify) : IQueueRunnerService +{ + private readonly object _lock = new(); + /// + /// A thread-safe dictionary to store cached task results. + /// + private readonly Dictionary _taskCache = new(); + + /// + /// The maximum number of completed task results to keep in memory + /// + private static readonly int MAX_TASK_RESULT_CACHE_SIZE = 100; + + private readonly List _tasks = new(); + private Task? _currentTask; + private IQueuedTask? _activeTask; + private bool _isPaused; + private bool _isTerminated; + + public long AddTask(IQueuedTask task) + => AddTask(task, false); + + public long AddTask(IQueuedTask task, bool skipQueue) + { + lock (_lock) + if (skipQueue) + _tasks.Insert(0, task); + else + _tasks.Add(task); + + eventPollNotify.SignalNewEvent(); + StartNextTask(); + return task.TaskID; + } + + public bool GetIsActive() + => _activeTask != null; + + public IQueuedTask? GetCurrentTask() + => _activeTask; + + public List GetCurrentTasks() + { + lock (_lock) + return [.. _tasks]; + } + + public void Pause() + { + lock (_lock) + _isPaused = true; + } + + public void Resume() + { + lock (_lock) + _isPaused = false; + + StartNextTask(); + } + + public void Terminate(bool wait) + { + _isTerminated = true; + if (wait) + { + var task = _currentTask; + if (task != null) + task.Await(); + } + } + + private void StartNextTask() + { + lock (_lock) + { + if (_isTerminated || _isPaused || (_currentTask != null && !_currentTask.IsCompleted)) + return; + + // Clean up completed tasks + if (_currentTask != null && _currentTask.IsCompleted) + { + _currentTask = null; + _activeTask = null; + } + + if (_tasks.Count == 0) + return; + + _activeTask = _tasks[0]; + _tasks.RemoveAt(0); + var task = _activeTask; + _currentTask = Task.Run(() => RunTask(task), CancellationToken.None); + } + } + + private async Task RunTask(IQueuedTask task) + { + var completed = false; + try + { + eventPollNotify.SignalNewEvent(); + task.TaskStarted = DateTime.UtcNow; + if (task.OnStarting != null) + await task.OnStarting().ConfigureAwait(false); + + await task.Execute(); + + // If the task is completed, don't call OnFinished again + completed = true; + AddTaskResult(new CachedTaskResult(task.TaskID, task.BackupID, task.TaskStarted, task.TaskFinished ?? DateTime.Now, null)); + if (task.OnFinished != null) + await task.OnFinished(null).ConfigureAwait(false); + } + catch (Exception ex) + { + connection.LogError(task.BackupID, "Error in worker", ex); + if (!completed) + { + AddTaskResult(new CachedTaskResult(task.TaskID, task.BackupID, task.TaskStarted, task.TaskFinished ?? DateTime.Now, ex)); + if (task.OnFinished != null) + await task.OnFinished(ex).ConfigureAwait(false); + } + } + finally + { + task.TaskFinished = DateTime.UtcNow; + _currentTask = null; + _activeTask = null; + eventPollNotify.SignalNewEvent(); + StartNextTask(); + } + } + + public IList> GetQueueWithIds() + { + return (from n in GetCurrentTasks() + where n.BackupID != null + select new Tuple(n.TaskID, n.BackupID)).ToList(); + } + + + /// + public CachedTaskResult? GetCachedTaskResults(long taskID) + { + lock (_lock) + { + _taskCache.TryGetValue(taskID, out var result); + return result; + } + } + + private void AddTaskResult(CachedTaskResult taskResult) + { + lock (_lock) + { + // If the task result is already in the cache, remove it + if (_taskCache.TryGetValue(taskResult.TaskID, out var existingResult)) + { + // If the stored task result has an exception, do not overwrite it + if (existingResult.Exception != null) + return; + } + + // Add/update the new task result in the cache + _taskCache[taskResult.TaskID] = taskResult; + + // If the cache size exceeds the maximum, remove the oldest entry + while (_taskCache.Count >= MAX_TASK_RESULT_CACHE_SIZE) + { + var oldestTaskID = _taskCache.Keys.Min(); + _taskCache.Remove(oldestTaskID); + } + } + } +} \ No newline at end of file diff --git a/Duplicati/WebserverCore/Services/SchedulerService.cs b/Duplicati/WebserverCore/Services/SchedulerService.cs index 15296edf2..b327c83ba 100644 --- a/Duplicati/WebserverCore/Services/SchedulerService.cs +++ b/Duplicati/WebserverCore/Services/SchedulerService.cs @@ -19,31 +19,32 @@ // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER // DEALINGS IN THE SOFTWARE. -using Duplicati.Library.Utility; +using Duplicati.Library.RestAPI; using Duplicati.Server; using Duplicati.Server.Serialization.Interface; using Duplicati.WebserverCore.Abstractions; namespace WebserverCore.Services; -public class SchedulerService : IScheduler +public class SchedulerService : ISchedulerService { - private readonly Duplicati.Server.Scheduler scheduler; - public SchedulerService(Duplicati.Server.Scheduler scheduler) + private readonly Scheduler scheduler; + public SchedulerService(EventPollNotify eventPollNotify, INotificationUpdateService notificationUpdateService, IQueueRunnerService queueRunnerService) { - this.scheduler = scheduler; + this.scheduler = new Scheduler(queueRunnerService); + var lastScheduleId = notificationUpdateService.LastDataUpdateId; + eventPollNotify.NewEvent += (sender, e) => + { + if (lastScheduleId != notificationUpdateService.LastDataUpdateId) + { + lastScheduleId = notificationUpdateService.LastDataUpdateId; + Reschedule(); + } + }; } public List> Schedule => scheduler.Schedule; - public List WorkerQueue => scheduler.WorkerQueue; - - public void SubScribeToNewSchedule(Action handler) - => scheduler.NewSchedule += (_, _) => handler(); - - public IList> GetSchedulerQueueIds() - => scheduler.GetSchedulerQueueIds(); - public IList> GetProposedSchedule() => scheduler.GetProposedSchedule(); @@ -52,7 +53,4 @@ public class SchedulerService : IScheduler public void Terminate(bool wait) => scheduler.Terminate(wait); - - public void Init(WorkerThread worker) - => scheduler.Init(worker); } diff --git a/Duplicati/WebserverCore/Services/StatusService.cs b/Duplicati/WebserverCore/Services/StatusService.cs index 7c3845b8b..336ea261e 100644 --- a/Duplicati/WebserverCore/Services/StatusService.cs +++ b/Duplicati/WebserverCore/Services/StatusService.cs @@ -19,7 +19,6 @@ // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER // DEALINGS IN THE SOFTWARE. using Duplicati.Library.RestAPI; -using Duplicati.Library.RestAPI.Abstractions; using Duplicati.Server; using Duplicati.Server.Serialization; using Duplicati.WebserverCore.Abstractions; @@ -31,22 +30,24 @@ public class StatusService( LiveControls liveControls, UpdatePollThread updatePollThread, IUpdateService updateService, - IWorkerThreadsManager workerThreadsManager, + IQueueRunnerService queueRunnerService, ISettingsService settingsService, - IScheduler scheduler, + ISchedulerService scheduler, EventPollNotify eventPollNotify, INotificationUpdateService notificationUpdateService) : IStatusService { public ServerStatusDto GetStatus() { + var task = queueRunnerService.GetCurrentTask(); + var status = new ServerStatusDto { UpdatedVersion = GetUpdatedVersion(), UpdaterState = updatePollThread.ThreadState, UpdateDownloadProgress = updatePollThread.DownloadProgess, - ActiveTask = workerThreadsManager.CurrentTask, - SchedulerQueueIds = scheduler.GetSchedulerQueueIds(), + ActiveTask = task == null ? null : new Tuple(task.TaskID, task.BackupID), + SchedulerQueueIds = queueRunnerService.GetQueueWithIds(), ProposedSchedule = scheduler.GetProposedSchedule(), LastEventID = eventPollNotify.EventNo, LastDataUpdateID = notificationUpdateService.LastDataUpdateId, @@ -72,7 +73,8 @@ public class StatusService( private SuggestedStatusIcon MapStateToIcon() { - if (workerThreadsManager.CurrentTask == null) + var task = queueRunnerService.GetCurrentTask(); + if (task == null) { if (liveControls.State == LiveControls.LiveControlState.Paused) return SuggestedStatusIcon.Paused; diff --git a/Duplicati/WebserverCore/Services/TaskCacheResults.cs b/Duplicati/WebserverCore/Services/TaskCacheResults.cs deleted file mode 100644 index 0f8d171a3..000000000 --- a/Duplicati/WebserverCore/Services/TaskCacheResults.cs +++ /dev/null @@ -1,75 +0,0 @@ -// Copyright (C) 2025, The Duplicati Team -// https://duplicati.com, hello@duplicati.com -// -// Permission is hereby granted, free of charge, to any person obtaining a -// copy of this software and associated documentation files (the "Software"), -// to deal in the Software without restriction, including without limitation -// the rights to use, copy, modify, merge, publish, distribute, sublicense, -// and/or sell copies of the Software, and to permit persons to whom the -// Software is furnished to do so, subject to the following conditions: -// -// The above copyright notice and this permission notice shall be included in -// all copies or substantial portions of the Software. -// -// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS -// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING -// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER -// DEALINGS IN THE SOFTWARE. -using Duplicati.WebserverCore.Abstractions; - -namespace Duplicati.WebserverCore.Services; - -/// -/// Service for caching task results. -/// -public class TaskCacheService : ITaskCacheService -{ - /// - /// A thread-safe dictionary to store cached task results. - /// - private readonly Dictionary _taskCache = new(); - - /// - /// The maximum number of completed task results to keep in memory - /// - private static readonly int MAX_TASK_RESULT_CACHE_SIZE = 100; - - /// - public CachedTaskResult? GetCachedTaskResults(long taskID) - { - lock (_taskCache) - { - _taskCache.TryGetValue(taskID, out var result); - return result; - } - } - - /// - public void AddTaskResult(CachedTaskResult taskResult) - { - lock (_taskCache) - { - // If the task result is already in the cache, remove it - if (_taskCache.TryGetValue(taskResult.TaskID, out var existingResult)) - { - // If the stored task result has an exception, do not overwrite it - if (existingResult.Exception != null) - return; - } - - // Add/update the new task result in the cache - _taskCache[taskResult.TaskID] = taskResult; - - // If the cache size exceeds the maximum, remove the oldest entry - while (_taskCache.Count >= MAX_TASK_RESULT_CACHE_SIZE) - { - var oldestTaskID = _taskCache.Keys.Min(); - _taskCache.Remove(oldestTaskID); - } - } - } - -} diff --git a/Duplicati/WebserverCore/Services/WorkerThreadsManager.cs b/Duplicati/WebserverCore/Services/WorkerThreadsManager.cs deleted file mode 100644 index 829837f4f..000000000 --- a/Duplicati/WebserverCore/Services/WorkerThreadsManager.cs +++ /dev/null @@ -1,61 +0,0 @@ -// Copyright (C) 2025, The Duplicati Team -// https://duplicati.com, hello@duplicati.com -// -// Permission is hereby granted, free of charge, to any person obtaining a -// copy of this software and associated documentation files (the "Software"), -// to deal in the Software without restriction, including without limitation -// the rights to use, copy, modify, merge, publish, distribute, sublicense, -// and/or sell copies of the Software, and to permit persons to whom the -// Software is furnished to do so, subject to the following conditions: -// -// The above copyright notice and this permission notice shall be included in -// all copies or substantial portions of the Software. -// -// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS -// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING -// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER -// DEALINGS IN THE SOFTWARE. -#nullable enable -using Duplicati.Library.IO; -using Duplicati.Library.RestAPI; -using Duplicati.Library.RestAPI.Abstractions; -using Duplicati.Library.Utility; -using Duplicati.Server; -using Duplicati.WebserverCore.Abstractions; - -namespace Duplicati.WebserverCore.Services; - -public class WorkerThreadsManager(ILiveControls liveControls, IScheduler scheduler) : IWorkerThreadsManager -{ - public WorkerThread? WorkerThread { get; private set; } - - public void Spawn(Action item) - { - WorkerThread = new WorkerThread(item, liveControls.IsPaused); - scheduler.Init(WorkerThread); - } - - public Tuple? CurrentTask - { - get - { - var t = WorkerThread?.CurrentTask; - return t == null ? null : new Tuple(t.TaskID, t.Backup.ID); - } - } - - public void UpdateThrottleSpeeds(string? uploadSpeed, string? downloadSpeed) - { - WorkerThread?.CurrentTask?.UpdateThrottleSpeed(uploadSpeed, downloadSpeed); - } - - public long AddTask(Runner.IRunnerData data, bool skipQueue = false) - { - WorkerThread!.AddTask(data, skipQueue); - FIXMEGlobal.StatusEventNotifyer.SignalNewEvent(); - return data.TaskID; - } -} \ No newline at end of file