/* * Copyright (c) 2023 Proton AG * * This file is part of ProtonVPN. * * ProtonVPN is free software: you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation, either version 3 of the License, or * (at your option) any later version. * * ProtonVPN is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * * You should have received a copy of the GNU General Public License * along with ProtonVPN. If not, see . */ using System; using System.Threading; using System.Threading.Tasks; namespace ProtonVPN.Common.Legacy.Threading; /// /// Implements consumer in a simple producer/consumer pattern. Multiple producers are producing work, /// and a single consumer processes the work. /// public class CoalescingAction { private readonly Func _action; private readonly CancellationHandle _cancellationHandle; private volatile Task _currentTask = Task.CompletedTask; private volatile int _workRequested; public CoalescingAction(Action action) : this(ct => { action(); return Task.CompletedTask; }) { } public CoalescingAction(Func action) : this(ct => action()) { } public CoalescingAction(Func action) { _action = action; _cancellationHandle = new CancellationHandle(); } public event EventHandler Completed; public bool Running { get; private set; } public void Run() { if (Interlocked.Exchange(ref _workRequested, 1) != 0) { return; } Running = true; TaskCompletionSource taskCompletion = new TaskCompletionSource(); Task newTask = taskCompletion.Task; CancellationToken cancellationToken = _cancellationHandle.Token; while (true) { Task expectedTask = _currentTask; Task previousTask = Interlocked.CompareExchange(ref _currentTask, newTask, expectedTask); if (previousTask != expectedTask) { continue; } // ReSharper disable once PossibleNullReferenceException Task task = previousTask.IsCompleted ? Task.Run(() => Run(cancellationToken), cancellationToken) : previousTask.ContinueWith(_ => Run(cancellationToken), cancellationToken, TaskContinuationOptions.LazyCancellation, TaskScheduler.Current).Unwrap(); task.ContinueWith(t => OnCompleted(t, taskCompletion), TaskContinuationOptions.ExecuteSynchronously); return; } } public void Cancel() { _cancellationHandle.Cancel(); _workRequested = 0; } private Task Run(CancellationToken cancellationToken) { Running = true; if (Interlocked.Exchange(ref _workRequested, 0) == 0) { return Task.CompletedTask; } return _action(cancellationToken); } private void OnCompleted(Task task, TaskCompletionSource taskCompletion) { Running = _workRequested != 0; Completed?.Invoke(this, new TaskCompletedEventArgs(task)); taskCompletion.SetResult(null); } }