Sonarr/NzbDrone.Core/Jobs/JobController.cs

262 lines
8.6 KiB
C#
Raw Normal View History

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using NLog;
2013-02-19 01:13:42 +00:00
using NzbDrone.Core.Lifecycle;
using NzbDrone.Core.Model;
using NzbDrone.Core.Model.Notification;
2011-12-02 01:33:17 +00:00
using NzbDrone.Core.Providers;
2011-12-02 01:33:17 +00:00
namespace NzbDrone.Core.Jobs
{
public interface IJobController
{
Stopwatch StopWatch { get; }
List<JobQueueItem> Queue { get; }
void QueueScheduled();
void QueueJob(Type jobType, dynamic options = null, JobQueueItem.JobSourceType source = JobQueueItem.JobSourceType.User);
bool QueueJob(string jobTypeString);
}
2013-02-19 01:13:42 +00:00
public class JobController : IJobController, IInitializable
{
private readonly NotificationProvider _notificationProvider;
2013-01-03 01:09:13 +00:00
private readonly IEnumerable<IJob> _jobs;
private readonly IJobRepository _jobRepository;
2013-02-19 01:13:42 +00:00
private readonly Logger _logger;
private Timer _timer;
private Thread _jobThread;
public Stopwatch StopWatch { get; private set; }
2011-07-11 00:03:01 +00:00
private readonly object _executionLock = new object();
private readonly List<JobQueueItem> _queue = new List<JobQueueItem>();
private ProgressNotification _notification;
public JobController(NotificationProvider notificationProvider, IEnumerable<IJob> jobs, IJobRepository jobRepository, Logger logger)
{
StopWatch = new Stopwatch();
_notificationProvider = notificationProvider;
_jobs = jobs;
_jobRepository = jobRepository;
2013-02-19 01:13:42 +00:00
_logger = logger;
ResetThread();
}
2013-02-19 01:13:42 +00:00
public void Init()
{
_timer = new Timer(c => QueueScheduled());
_timer.Change(0, 60 * 1000);
}
public List<JobQueueItem> Queue
2011-08-03 16:29:03 +00:00
{
get
{
return _queue;
}
}
public virtual void QueueScheduled()
{
lock (_executionLock)
{
VerifyThreadTime();
if (_jobThread.IsAlive)
{
2013-02-19 01:13:42 +00:00
_logger.Trace("Queue is already running. Ignoring scheduler's request.");
return;
}
}
var pendingJobs = _jobRepository.GetPendingJobs()
.Select(c => _jobs.Single(t => t.GetType().ToString() == c.TypeName)
.GetType()).ToList();
pendingJobs.ForEach(jobType => QueueJob(jobType, source: JobQueueItem.JobSourceType.Scheduler));
2013-02-19 01:13:42 +00:00
_logger.Trace("{0} Scheduled tasks have been added to the queue", pendingJobs.Count);
}
2012-09-10 19:04:17 +00:00
public virtual void QueueJob(Type jobType, dynamic options = null, JobQueueItem.JobSourceType source = JobQueueItem.JobSourceType.User)
{
var queueItem = new JobQueueItem
{
JobType = jobType,
2012-09-10 19:04:17 +00:00
Options = options,
Source = source
};
2013-02-19 01:13:42 +00:00
_logger.Debug("Attempting to queue {0}", queueItem);
lock (_executionLock)
{
VerifyThreadTime();
lock (Queue)
{
if (!Queue.Contains(queueItem))
{
Queue.Add(queueItem);
2013-02-19 01:13:42 +00:00
_logger.Trace("Job {0} added to the queue. current items in queue: {1}", queueItem, Queue.Count);
}
else
{
2013-02-19 01:13:42 +00:00
_logger.Info("{0} already exists in the queue. Skipping. current items in queue: {1}", queueItem, Queue.Count);
}
}
if (_jobThread.IsAlive)
{
2013-02-19 01:13:42 +00:00
_logger.Trace("Queue is already running. No need to start it up.");
return;
}
ResetThread();
StopWatch = Stopwatch.StartNew();
_jobThread.Start();
}
}
public virtual bool QueueJob(string jobTypeString)
{
var type = Type.GetType(jobTypeString);
if (type == null)
return false;
QueueJob(type);
return true;
}
private void ProcessQueue()
{
try
{
do
{
using (NestedDiagnosticsContext.Push(Guid.NewGuid().ToString()))
{
try
2011-07-11 00:03:01 +00:00
{
JobQueueItem job = null;
lock (Queue)
2011-07-27 22:59:48 +00:00
{
if (Queue.Count != 0)
{
job = Queue.OrderBy(c => c.Source).First();
2013-02-19 01:13:42 +00:00
_logger.Trace("Popping {0} from the queue.", job);
Queue.Remove(job);
}
2011-07-27 22:59:48 +00:00
}
if (job != null)
{
Execute(job);
}
}
catch (ThreadAbortException)
2011-07-27 22:59:48 +00:00
{
throw;
}
catch (Exception e)
{
2013-02-19 01:13:42 +00:00
_logger.FatalException("An error has occurred while executing job.", e);
2011-07-11 00:03:01 +00:00
}
}
} while (Queue.Count != 0);
}
catch (ThreadAbortException e)
{
2013-02-19 01:13:42 +00:00
_logger.Warn(e.Message);
}
catch (Exception e)
{
2013-02-19 01:13:42 +00:00
_logger.ErrorException("Error has occurred in queue processor thread", e);
}
finally
{
StopWatch.Stop();
2013-02-19 01:13:42 +00:00
_logger.Trace("Finished processing jobs in the queue.");
}
}
private void Execute(JobQueueItem queueItem)
{
2013-01-14 16:36:16 +00:00
var jobImplementation = _jobs.SingleOrDefault(t => t.GetType() == queueItem.JobType);
2011-05-17 07:24:29 +00:00
if (jobImplementation == null)
{
2013-02-19 01:13:42 +00:00
_logger.Error("Unable to locate implementation for '{0}'. Make sure it is properly registered.", queueItem.JobType);
return;
}
var jobDefinition = _jobRepository.GetDefinition(queueItem.JobType);
2011-05-17 07:24:29 +00:00
using (_notification = new ProgressNotification(jobImplementation.Name))
{
2011-04-22 06:23:29 +00:00
try
{
2013-02-19 01:13:42 +00:00
_logger.Debug("Starting {0}. Last execution {1}", queueItem, jobDefinition.LastExecution);
2011-04-22 06:23:29 +00:00
var sw = Stopwatch.StartNew();
_notificationProvider.Register(_notification);
2012-09-10 19:04:17 +00:00
jobImplementation.Start(_notification, queueItem.Options);
_notification.Status = ProgressNotificationStatus.Completed;
jobDefinition.LastExecution = DateTime.Now;
jobDefinition.Success = true;
2011-04-22 06:23:29 +00:00
sw.Stop();
2013-02-19 01:13:42 +00:00
_logger.Debug("Job {0} successfully completed in {1:0}.{2} seconds.", queueItem, sw.Elapsed.TotalSeconds, sw.Elapsed.Milliseconds / 100,
sw.Elapsed.Seconds);
}
catch (ThreadAbortException)
{
throw;
2011-04-22 06:23:29 +00:00
}
catch (Exception e)
{
2013-02-19 01:13:42 +00:00
_logger.ErrorException("An error has occurred while executing job [" + jobImplementation.Name + "].", e);
2011-04-22 06:23:29 +00:00
_notification.Status = ProgressNotificationStatus.Failed;
_notification.CurrentMessage = jobImplementation.Name + " Failed.";
jobDefinition.LastExecution = DateTime.Now;
jobDefinition.Success = false;
}
}
2011-08-03 16:29:03 +00:00
//Only update last execution status if was triggered by the scheduler
2012-09-10 19:04:17 +00:00
if (queueItem.Options == null)
{
_jobRepository.Update(jobDefinition);
}
}
private void VerifyThreadTime()
{
if (StopWatch.Elapsed.TotalHours > 1)
{
2013-02-19 01:13:42 +00:00
_logger.Warn("Thread job has been running for more than an hour. fuck it!");
ResetThread();
}
}
private void ResetThread()
{
if (_jobThread != null)
{
_jobThread.Abort();
}
2013-02-19 01:13:42 +00:00
_logger.Trace("resetting queue processor thread");
_jobThread = new Thread(ProcessQueue) { Name = "JobQueueThread" };
}
}
}