<commit_before>﻿using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;

namespace Mastersign.Tasks
{
    public class WorkerThread : IDisposable
    {
        private TaskQueue _queue;
        private IWorker _worker;
        private string _label;
        private Thread _thread;
        private ManualResetEvent _aliveEvent = new ManualResetEvent(true);
        private ManualResetEvent _workedEvent = new ManualResetEvent(false);
        private ManualResetEvent _busyEvent = new ManualResetEvent(true);
        private CancelationToken _cancelationToken;

        private ThreadPriority ThreadPriority { get; set; }

        public WorkerThread(TaskQueue queue, IWorker worker, string label)
        {
            _queue = queue ?? throw new ArgumentNullException(nameof(queue));
            _worker = worker ?? throw new ArgumentNullException(nameof(worker));
            _label = label;
            ThreadPriority = ThreadPriority.Normal;
        }

        public TaskQueue Queue => _queue;

        public bool IsAlive => _thread != null;

        public event EventHandler IsAliveChanged;

        private void OnIsAliveChanged()
        {
            IsAliveChanged?.Invoke(this, EventArgs.Empty);
        }

        public event EventHandler<TaskEventArgs> TaskBegin;

        private void OnTaskBegin(ITask task)
        {
            TaskBegin?.Invoke(this, new TaskEventArgs(task, TaskState.InProgress));
        }

        public event EventHandler<TaskEventArgs> TaskEnd;

        private void OnTaskEnd(ITask task)
        {
            TaskEnd?.Invoke(this, new TaskEventArgs(task));
        }

        private bool _busy;
        public bool Busy
        {
            get => _busy;
            private set
            {
                if (_busy == value) return;
                _busy = value;
                OnBusyChanged();
                if (_busy)
                    _busyEvent.Reset();
                else
                    _busyEvent.Set();
            }
        }

        public event EventHandler BusyChanged;

        private void OnBusyChanged()
        {
            BusyChanged?.Invoke(this, EventArgs.Empty);
        }

        private ITask _currentTask;
        public ITask CurrentTask
        {
            get => _currentTask;
            set
            {
                if (_currentTask == value) return;
                _currentTask = value;
                OnCurrentTaskChanged();
            }
        }

        public event EventHandler CurrentTaskChanged;

        private void OnCurrentTaskChanged()
        {
            CurrentTaskChanged?.Invoke(this, EventArgs.Empty);
        }

        public string Name => nameof(WorkerThread) + (string.IsNullOrEmpty(_label) ? string.Empty : "_" + _label);

        public void Start()
        {
            if (IsDisposed) throw new ObjectDisposedException(Name);
            if (IsAlive) throw new InvalidOperationException("The thread is already started.");
            _cancelationToken = new CancelationToken();
            _thread = new Thread(ThreadLoop);
            _thread.Name = Name;
            _thread.Priority = ThreadPriority;
            _aliveEvent.Reset();
            OnIsAliveChanged();
            _thread.Start();
        }

        private void ThreadLoop()
        {
            ITask task = null;
            while (!_cancelationToken.IsCanceled && !_queue.IsDisposed && !IsDisposed)
            {
                while (_queue.TryDequeue(ref task))
                {
                    var taskState = task.State;
                    if (taskState != TaskState.Waiting)
                    {
                        OnTaskRejected(task, taskState);
                        continue;
                    }
                    Busy = true;
                    _workedEvent.Set();
                    CurrentTask = task;
                    OnTaskBegin(task);
                    task.UpdateState(TaskState.InProgress);
                    Exception workerError = null;
                    try
                    {
                        _worker.Process(task, _cancelationToken);
                    }
                    catch (Exception e)
                    {
                        workerError = e;
                    }
                    if (workerError != null)
                    {
                        if (task.State == TaskState.InProgress || task.State == TaskState.CleaningUp)
                        {<commit_msg>prevent the WorkerThread from consuming tasks after cancellation<commit_after>﻿using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;

namespace Mastersign.Tasks
{
    public class WorkerThread : IDisposable
    {
        private TaskQueue _queue;
        private IWorker _worker;
        private string _label;
        private Thread _thread;
        private ManualResetEvent _aliveEvent = new ManualResetEvent(true);
        private ManualResetEvent _workedEvent = new ManualResetEvent(false);
        private ManualResetEvent _busyEvent = new ManualResetEvent(true);
        private CancelationToken _cancelationToken;

        private ThreadPriority ThreadPriority { get; set; }

        public WorkerThread(TaskQueue queue, IWorker worker, string label)
        {
            _queue = queue ?? throw new ArgumentNullException(nameof(queue));
            _worker = worker ?? throw new ArgumentNullException(nameof(worker));
            _label = label;
            ThreadPriority = ThreadPriority.Normal;
        }

        public TaskQueue Queue => _queue;

        public bool IsAlive => _thread != null;

        public event EventHandler IsAliveChanged;

        private void OnIsAliveChanged()
        {
            IsAliveChanged?.Invoke(this, EventArgs.Empty);
        }

        public event EventHandler<TaskEventArgs> TaskBegin;

        private void OnTaskBegin(ITask task)
        {
            TaskBegin?.Invoke(this, new TaskEventArgs(task, TaskState.InProgress));
        }

        public event EventHandler<TaskEventArgs> TaskEnd;

        private void OnTaskEnd(ITask task)
        {
            TaskEnd?.Invoke(this, new TaskEventArgs(task));
        }

        private bool _busy;
        public bool Busy
        {
            get => _busy;
            private set
            {
                if (_busy == value) return;
                _busy = value;
                OnBusyChanged();
                if (_busy)
                    _busyEvent.Reset();
                else
                    _busyEvent.Set();
            }
        }

        public event EventHandler BusyChanged;

        private void OnBusyChanged()
        {
            BusyChanged?.Invoke(this, EventArgs.Empty);
        }

        private ITask _currentTask;
        public ITask CurrentTask
        {
            get => _currentTask;
            set
            {
                if (_currentTask == value) return;
                _currentTask = value;
                OnCurrentTaskChanged();
            }
        }

        public event EventHandler CurrentTaskChanged;

        private void OnCurrentTaskChanged()
        {
            CurrentTaskChanged?.Invoke(this, EventArgs.Empty);
        }

        public string Name => nameof(WorkerThread) + (string.IsNullOrEmpty(_label) ? string.Empty : "_" + _label);

        public void Start()
        {
            if (IsDisposed) throw new ObjectDisposedException(Name);
            if (IsAlive) throw new InvalidOperationException("The thread is already started.");
            _cancelationToken = new CancelationToken();
            _thread = new Thread(ThreadLoop);
            _thread.Name = Name;
            _thread.Priority = ThreadPriority;
            _aliveEvent.Reset();
            OnIsAliveChanged();
            _thread.Start();
        }

        private void ThreadLoop()
        {
            ITask task = null;
            while (!_cancelationToken.IsCanceled && !_queue.IsDisposed && !IsDisposed)
            {
                while (!_cancelationToken.IsCanceled && _queue.TryDequeue(ref task))
                {
                    var taskState = task.State;
                    if (taskState != TaskState.Waiting)
                    {
                        OnTaskRejected(task, taskState);
                        continue;
                    }
                    Busy = true;
                    _workedEvent.Set();
                    CurrentTask = task;
                    OnTaskBegin(task);
                    task.UpdateState(TaskState.InProgress);
                    Exception workerError = null;
                    try
                    {
                        _worker.Process(task, _cancelationToken);
                    }
                    catch (Exception e)
                    {
                        workerError = e;
                    }
                    if (workerError != null)
                    {
                        if (task.State == TaskState.InProgress || task.State == TaskState.CleaningUp)
                        {
{'fraction_non_alphanumeric': 0.0736331569664903, 'fraction_numerical': 0.0, 'mean_word_length': 1.4701878573373264, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4303900', 'n_tokens_mistral': 2208, 'n_tokens_neox': 2092, 'n_words': 512}