1
\$\begingroup\$

I replaced the thread by a task so there's no thread locked when there is no job to execute for hours.

Now i'm not sure if the lock without an AutoResetEvent is lock safe when AllowParallelExcuteion is set to false.

 public class BezJobManager : IBezJobManager
{
    private CancellationTokenSource cts = new CancellationTokenSource();
    private readonly IBezServiceResolver _serviceResolver;
    private List<BezExecuteJobDetail> jobs = new List<BezExecuteJobDetail>();
    private Queue<RunningJobData> runningjobs = new Queue<RunningJobData>();
    public event Action<BezExecuteJobDetail> JobStartedEvent;
    public event Action<BezExecuteJobDetail> JobFinishedEvent;
    public event Action<BezExecuteJobDetail> JobErrorEvent;
    private static readonly List<AutoResetEvent> autoResetEventHandlers = new List<AutoResetEvent>();
    public bool IsDisposed { get; private set; }
    public bool AllowParallelExecution { get; set; }

    public BezJobManager(IBezServiceResolver serviceResolver)
    {
        _serviceResolver = serviceResolver;
    }

    public Guid Prepare(string name, Guid userId)
    {
        if (IsDisposed)
            return Guid.Empty;
        var job = Create(name, userId);
        return job.Id;
    }


    public Guid Execute(string name, Guid userId, Action action)
    {
        if (IsDisposed)
            return Guid.Empty;
        var job = Create(name, userId);
        Execute(job, action);
        return job.Id;
    }

    public Guid Execute(Guid jobId, Action action)
    {
        BezExecuteJobDetail job;
        lock (jobs)
            job = jobs.FirstOrDefault(item => item.Id == jobId);
        if (job == null) return Guid.Empty;
        Execute(job,action);
        return job.Id;
    }
    public Guid Execute(BezExecuteJobDetail job, Action action)
    {

        if (IsDisposed)
            return Guid.Empty;
        RunningJobData runningJob;
        lock (runningjobs)
        {
            runningJob = CreateRunningJob(job.Id);
            runningJob.Action = action;
        }
        if (AllowParallelExecution)
        {
            lock (runningjobs)
                runningjobs.Dequeue();
            Task.Run(() => RunJob(runningJob), cts.Token);
        }
        else
            RunSingleThread();
        return job.Id;
    }

    private bool isSingleTaskRunning;
    private object singleTaskLocker = new object();
    private void RunSingleThread()
    {
        lock (singleTaskLocker)
        {
            if (isSingleTaskRunning) return;
            isSingleTaskRunning = true;
        }
        Task.Run(() =>
                 {
                     while (!IsDisposed)
                     {
                         try
                         {

                             RunningJobData jobToRun;
                             lock (runningjobs)
                             {
                                 lock (singleTaskLocker)
                                 {
                                     if (runningjobs.Count == 0 || IsDisposed)
                                     {
                                         Task.Delay(100).Wait(cts.Token);
                                         if (runningjobs.Count == 0 || IsDisposed)
                                         {
                                             isSingleTaskRunning = false;
                                             return;
                                         }
                                     }
                                 }
                                 jobToRun = runningjobs.Dequeue();
                                 // It can be that the finished method is already set but not yet the action.
                                 if (jobToRun.Action == null)
                                 {
                                     runningjobs.Enqueue(jobToRun);
                                     // If there's only one job,; wait a bit, otherwise start immediatly in while loop with new other job.
                                     if (runningjobs.Count == 1)
                                         Task.Delay(50).Wait();
                                     continue;
                                 }

                             }
                             RunJob(jobToRun);
                         }
                         catch (Exception ex)
                         {
                             var logger = _serviceResolver.Resolve<ILogger>();
                             if (logger != null)
                                 logger.LogError("BezJobManager:Error while running jobs", ex);
                         }
                     }
                 }, cts.Token);


    }
    private void RunJob(RunningJobData runningJob)
    {
        var job = runningJob.Job;
        try
        {
            job.Status = BezExecuteJobDetail.BezExecuteJobDetailStatus.Running;
            OnJobStartedEvent(job);
            if (IsDisposed) return;
            runningJob.Action();
            if (cts.Token.IsCancellationRequested) return;
            if (IsDisposed) return;
            job.Status = BezExecuteJobDetail.BezExecuteJobDetailStatus.Finished;
            if (cts.Token.IsCancellationRequested) return;
            if (IsDisposed) return;
            runningJob.Finished(cts.Token);
            runningJob.ClearAll();
            OnJobFinishedEvent(job);
        }
        catch (Exception ex)
        {
            job.Status = BezExecuteJobDetail.BezExecuteJobDetailStatus.Error;
            job.ErrorMessage = ex.Message;
            var logger = _serviceResolver.Resolve<ILogger>();
            if (logger != null)
                logger.LogError("BezJobManager:Error while running job:" + job.Name + ":" + job.Status, ex);
            try
            {
                OnJobErrorEvent(job);
            }
            catch
            {
            }
        }
        finally
        {
            lock (jobs)
                jobs.Remove(job);

        }
    }
    public void ExecuteWhenFinished(Guid jobId, Action action)
    {
        var localJob = CreateRunningJob(jobId);
        localJob.AddFinishedAction(action);
    }

    private RunningJobData CreateRunningJob(Guid jobId)
    {
        RunningJobData localJob;
        lock (runningjobs)
        {
            localJob = runningjobs.FirstOrDefault(item => item.Job.Id == jobId);
            if (localJob == null)
            {
                localJob = new RunningJobData(_serviceResolver)
                           {
                               Job = new BezExecuteJobDetail
                                     {
                                         Id = jobId
                                     }
                           };
                runningjobs.Enqueue(localJob);
            }
        }
        return localJob;
    }


    public Task WaitTillFinished(Guid jobId)
    {
        return Task.Run(() =>
        {

            var waitHandle = new AutoResetEvent(false);
            lock (autoResetEventHandlers)
                autoResetEventHandlers.Add(waitHandle);
            ExecuteWhenFinished(jobId, () => waitHandle.Set());
            waitHandle.WaitOne();
            lock (autoResetEventHandlers)
                autoResetEventHandlers.Remove(waitHandle);
        });
    }

    private BezExecuteJobDetail Create(string name, Guid userId)
    {
        var job = new BezExecuteJobDetail
        {
            Name = name,
            Id = Guid.NewGuid(),
            RequestByUserId = userId,
            Status = BezExecuteJobDetail.BezExecuteJobDetailStatus.Created
        };
        lock (jobs)
            jobs.Add(job);
        return job;
    }

    private void OnJobFinishedEvent(BezExecuteJobDetail obj)
    {
        var handler = JobFinishedEvent;
        if (handler != null) handler(obj);
    }

    private void OnJobErrorEvent(BezExecuteJobDetail obj)
    {
        var handler = JobErrorEvent;
        if (handler != null) handler(obj);
    }

    private void OnJobStartedEvent(BezExecuteJobDetail obj)
    {
        var handler = JobStartedEvent;
        if (handler != null) handler(obj);
    }

    public void Dispose()
    {
        IsDisposed = true;
        cts.Cancel();
    }

    private class RunningJobData
    {
         private readonly IBezServiceResolver _serviceResolver;
        public BezExecuteJobDetail Job { get; set; }
        public Action Action { get; set; }
        private readonly List<Action> _actionsWhenFinished = new List<Action>();

        public RunningJobData(IBezServiceResolver serviceResolver)
        {
            _serviceResolver = serviceResolver;
        }

        public void AddFinishedAction(Action action)
        {
            lock (_actionsWhenFinished)
            _actionsWhenFinished.Add(action);
        }
        public void ClearAll()
        {
            lock (_actionsWhenFinished)
            _actionsWhenFinished.Clear();
        }
        public void Finished(CancellationToken token)
        {
            lock (_actionsWhenFinished)
                foreach (var action in _actionsWhenFinished)
                {
                    var action1 = action;
                    Task.Run(() =>
                    {
                        try
                        {
                            action1();
                        }
                        catch (Exception ex)
                        {
                            _serviceResolver.Resolve<ILogger>().LogError("BezJobClientManager:Error executing " + Job.Name, ex);
                        }
                    }, token);
                }
        }

    }
}
\$\endgroup\$
  • \$\begingroup\$ Welcome to Code Review. Does your code work for you or are you trying to debug it? Only working code is on-topic here. \$\endgroup\$ – chicks Dec 6 '16 at 14:05
  • \$\begingroup\$ The code works, but i'm unsure about the locking part when 2 threads on the same time adds a job, if no deadlock will occur \$\endgroup\$ – Emmanuel Dec 6 '16 at 17:31
  • \$\begingroup\$ @Emmanuel Can you point to where specifically in your code is this "locking part"? \$\endgroup\$ – 404 Dec 6 '16 at 17:59
  • \$\begingroup\$ in the method RunSingleThread() , singleTaskLocker object \$\endgroup\$ – Emmanuel Dec 6 '16 at 18:07

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.