mirror
/
Radarr
mirror of https://github.com/Radarr/Radarr
1
0
Fork 0
Code Issues Releases Wiki Activity

Fixed: Refreshing Plex Server series in high volume systems

This commit is contained in:
Taloth Saldono 2020-06-04 00:11:45 +02:00 committed by Qstick
parent 026c34c3ba
commit 35532b6789
9 changed files with 239 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/NzbDrone.Common/Cache/CacheManager.cs
View File

@ -8,6 +8,7 @@ namespace NzbDrone.Common.Cache
{
ICached<T> GetCache<T>(Type host);
ICached<T> GetCache<T>(Type host, string name);
ICached<T> GetRollingCache<T>(Type host, string name, TimeSpan defaultLifeTime);
ICachedDictionary<T> GetCacheDictionary<T>(Type host, string name, Func<IDictionary<string, T>> fetchFunc = null, TimeSpan? lifeTime = null);
void Clear();
ICollection<ICached> Caches { get; }
@ -43,6 +44,14 @@ namespace NzbDrone.Common.Cache
return (ICached<T>)_cache.Get(host.FullName + "_" + name, () => new Cached<T>());
}
public ICached<T> GetRollingCache<T>(Type host, string name, TimeSpan defaultLifeTime)
{
Ensure.That(host, () => host).IsNotNull();
Ensure.That(name, () => name).IsNotNullOrWhiteSpace();
return (ICached<T>)_cache.Get(host.FullName + "_" + name, () => new Cached<T>(defaultLifeTime, true));
}
public ICachedDictionary<T> GetCacheDictionary<T>(Type host, string name, Func<IDictionary<string, T>> fetchFunc = null, TimeSpan? lifeTime = null)
{
Ensure.That(host, () => host).IsNotNull();

77
src/NzbDrone.Common/Cache/Cached.cs
View File

@ -1,8 +1,9 @@
using System;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using NzbDrone.Common.EnsureThat;
using NzbDrone.Common.Extensions;
namespace NzbDrone.Common.Cache
{
@ -29,35 +30,49 @@ namespace NzbDrone.Common.Cache
}
private readonly ConcurrentDictionary<string, CacheItem> _store;
private readonly TimeSpan? _defaultLifeTime;
private readonly bool _rollingExpiry;
public Cached()
public Cached(TimeSpan? defaultLifeTime = null, bool rollingExpiry = false)
{
_store = new ConcurrentDictionary<string, CacheItem>();
_defaultLifeTime = defaultLifeTime;
_rollingExpiry = rollingExpiry;
}
public void Set(string key, T value, TimeSpan? lifetime = null)
public void Set(string key, T value, TimeSpan? lifeTime = null)
{
Ensure.That(key, () => key).IsNotNullOrWhiteSpace();
_store[key] = new CacheItem(value, lifetime);
_store[key] = new CacheItem(value, lifeTime ?? _defaultLifeTime);
}
public T Find(string key)
{
CacheItem value;
_store.TryGetValue(key, out value);
if (value == null)
CacheItem cacheItem;
if (!_store.TryGetValue(key, out cacheItem))
{
return default(T);
}
if (value.IsExpired())
if (cacheItem.IsExpired())
{
_store.TryRemove(key, out value);
return default(T);
if (TryRemove(key, cacheItem))
{
return default(T);
}
if (!_store.TryGetValue(key, out cacheItem))
{
return default(T);
}
}
return value.Object;
if (_rollingExpiry && _defaultLifeTime.HasValue)
{
_store.TryUpdate(key, new CacheItem(cacheItem.Object, _defaultLifeTime.Value), cacheItem);
}
return cacheItem.Object;
}
public void Remove(string key)
@ -72,20 +87,31 @@ namespace NzbDrone.Common.Cache
{
Ensure.That(key, () => key).IsNotNullOrWhiteSpace();
CacheItem cacheItem;
T value;
lifeTime = lifeTime ?? _defaultLifeTime;
if (!_store.TryGetValue(key, out cacheItem) || cacheItem.IsExpired())
CacheItem cacheItem;
if (_store.TryGetValue(key, out cacheItem) && !cacheItem.IsExpired())
{
value = function();
Set(key, value, lifeTime);
if (_rollingExpiry && lifeTime.HasValue)
{
_store.TryUpdate(key, new CacheItem(cacheItem.Object, lifeTime), cacheItem);
}
}
else
{
value = cacheItem.Object;
var newCacheItem = new CacheItem(function(), lifeTime);
if (cacheItem != null && _store.TryUpdate(key, newCacheItem, cacheItem))
{
cacheItem = newCacheItem;
}
else
{
cacheItem = _store.GetOrAdd(key, newCacheItem);
}
}
return value;
return cacheItem.Object;
}
public void Clear()
@ -95,9 +121,11 @@ namespace NzbDrone.Common.Cache
public void ClearExpired()
{
foreach (var cached in _store.Where(c => c.Value.IsExpired()))
var collection = (ICollection<KeyValuePair<string, CacheItem>>)_store;
foreach (var cached in _store.Where(c => c.Value.IsExpired()).ToList())
{
Remove(cached.Key);
collection.Remove(cached);
}
}
@ -108,5 +136,12 @@ namespace NzbDrone.Common.Cache
return _store.Values.Select(c => c.Object).ToList();
}
}
private bool TryRemove(string key, CacheItem value)
{
var collection = (ICollection<KeyValuePair<string, CacheItem>>)_store;
return collection.Remove(new KeyValuePair<string, CacheItem>(key, value));
}
}
}

8
src/NzbDrone.Core/MediaFiles/Events/RenameCompletedEvent.cs Normal file
View File

@ -0,0 +1,8 @@
using NzbDrone.Common.Messaging;
namespace NzbDrone.Core.MediaFiles.Events
{
public class RenameCompletedEvent : IEvent
{
}
}

4
src/NzbDrone.Core/MediaFiles/RenameMovieFileService.cs
View File

@ -127,6 +127,8 @@ namespace NzbDrone.Core.MediaFiles
_logger.ProgressInfo("Renaming {0} files for {1}", movieFiles.Count, movie.Title);
RenameFiles(movieFiles, movie);
_logger.ProgressInfo("Selected movie files renamed for {0}", movie.Title);
_eventAggregator.PublishEvent(new RenameCompletedEvent());
}
public void Execute(RenameMovieCommand message)
@ -141,6 +143,8 @@ namespace NzbDrone.Core.MediaFiles
RenameFiles(movieFiles, movie);
_logger.ProgressInfo("All movie files renamed for {0}", movie.Title);
}
_eventAggregator.PublishEvent(new RenameCompletedEvent());
}
}
}

3
src/NzbDrone.Core/Notifications/INotification.cs
View File

@ -1,4 +1,4 @@
using NzbDrone.Core.Movies;
using NzbDrone.Core.Movies;
using NzbDrone.Core.ThingiProvider;
namespace NzbDrone.Core.Notifications
@ -12,6 +12,7 @@ namespace NzbDrone.Core.Notifications
void OnMovieRename(Movie movie);
void OnHealthIssue(HealthCheck.HealthCheck healthCheck);
void OnDelete(DeleteMessage deleteMessage);
void ProcessQueue();
bool SupportsOnGrab { get; }
bool SupportsOnDownload { get; }
bool SupportsOnUpgrade { get; }

4
src/NzbDrone.Core/Notifications/NotificationBase.cs
View File

@ -50,6 +50,10 @@ namespace NzbDrone.Core.Notifications
{
}
public virtual void ProcessQueue()
{
}
public bool SupportsOnGrab => HasConcreteImplementation("OnGrab");
public bool SupportsOnRename => HasConcreteImplementation("OnMovieRename");
public bool SupportsOnDownload => HasConcreteImplementation("OnDownload");

65
src/NzbDrone.Core/Notifications/NotificationService.cs
View File

@ -17,7 +17,10 @@ namespace NzbDrone.Core.Notifications
IHandle<MovieGrabbedEvent>,
IHandle<MovieDownloadedEvent>,
IHandle<HealthCheckFailedEvent>,
IHandle<MovieFileDeletedEvent>
IHandle<MovieFileDeletedEvent>,
IHandleAsync<DownloadsProcessedEvent>,
IHandleAsync<RenameCompletedEvent>,
IHandleAsync<HealthCheckCompleteEvent>
{
private readonly INotificationFactory _notificationFactory;
private readonly Logger _logger;
@ -176,26 +179,56 @@ namespace NzbDrone.Core.Notifications
public void Handle(MovieFileDeletedEvent message)
{
var deleteMessage = new DeleteMessage();
deleteMessage.Message = GetMessage(message.MovieFile.Movie, message.MovieFile.Quality);
deleteMessage.MovieFile = message.MovieFile;
deleteMessage.Movie = message.MovieFile.Movie;
deleteMessage.Reason = message.Reason;
var deleteMessage = new DeleteMessage();
deleteMessage.Message = GetMessage(message.MovieFile.Movie, message.MovieFile.Quality);
deleteMessage.MovieFile = message.MovieFile;
deleteMessage.Movie = message.MovieFile.Movie;
deleteMessage.Reason = message.Reason;
foreach (var notification in _notificationFactory.OnDeleteEnabled())
foreach (var notification in _notificationFactory.OnDeleteEnabled())
{
try
{
try
if (ShouldHandleMovie(notification.Definition, message.MovieFile.Movie))
{
if (ShouldHandleMovie(notification.Definition, message.MovieFile.Movie))
{
notification.OnDelete(deleteMessage);
}
}
catch (Exception ex)
{
_logger.Warn(ex, "Unable to send OnDelete notification to: " + notification.Definition.Name);
notification.OnDelete(deleteMessage);
}
}
catch (Exception ex)
{
_logger.Warn(ex, "Unable to send OnDelete notification to: " + notification.Definition.Name);
}
}
}
public void HandleAsync(DownloadsProcessedEvent message)
{
ProcessQueue();
}
public void HandleAsync(RenameCompletedEvent message)
{
ProcessQueue();
}
public void HandleAsync(HealthCheckCompleteEvent message)
{
ProcessQueue();
}
private void ProcessQueue()
{
foreach (var notification in _notificationFactory.GetAvailableProviders())
{
try
{
notification.ProcessQueue();
}
catch (Exception ex)
{
_logger.Warn(ex, "Unable to process notification queue for " + notification.Definition.Name);
}
}
}
}
}

77
src/NzbDrone.Core/Notifications/Plex/Server/PlexServer.cs
View File

@ -1,6 +1,9 @@
using System;
using System.Collections.Generic;
using System.Linq;
using FluentValidation.Results;
using NLog;
using NzbDrone.Common.Cache;
using NzbDrone.Common.Extensions;
using NzbDrone.Core.Exceptions;
using NzbDrone.Core.Movies;
@ -13,11 +16,23 @@ namespace NzbDrone.Core.Notifications.Plex.Server
{
private readonly IPlexServerService _plexServerService;
private readonly IPlexTvService _plexTvService;
private readonly Logger _logger;
public PlexServer(IPlexServerService plexServerService, IPlexTvService plexTvService)
private class PlexUpdateQueue
{
public Dictionary<int, Movie> Pending { get; } = new Dictionary<int, Movie>();
public bool Refreshing { get; set; }
}
private readonly ICached<PlexUpdateQueue> _pendingMoviesCache;
public PlexServer(IPlexServerService plexServerService, IPlexTvService plexTvService, ICacheManager cacheManager, Logger logger)
{
_plexServerService = plexServerService;
_plexTvService = plexTvService;
_logger = logger;
_pendingMoviesCache = cacheManager.GetRollingCache<PlexUpdateQueue>(GetType(), "pendingSeries", TimeSpan.FromDays(1));
}
public override string Link => "https://www.plex.tv/";
@ -37,7 +52,65 @@ namespace NzbDrone.Core.Notifications.Plex.Server
{
if (Settings.UpdateLibrary)
{
_plexServerService.UpdateLibrary(movie, Settings);
_logger.Debug("Scheduling library update for movie {0} {1}", movie.Id, movie.Title);
var queue = _pendingMoviesCache.Get(Settings.Host, () => new PlexUpdateQueue());
lock (queue)
{
queue.Pending[movie.Id] = movie;
}
}
}
public override void ProcessQueue()
{
PlexUpdateQueue queue = _pendingMoviesCache.Find(Settings.Host);
if (queue == null)
{
return;
}
lock (queue)
{
if (queue.Refreshing)
{
return;
}
queue.Refreshing = true;
}
try
{
while (true)
{
List<Movie> refreshingMovies;
lock (queue)
{
if (queue.Pending.Empty())
{
queue.Refreshing = false;
return;
}
refreshingMovies = queue.Pending.Values.ToList();
queue.Pending.Clear();
}
if (Settings.UpdateLibrary)
{
_logger.Debug("Performing library update for {0} movies", refreshingMovies.Count);
_plexServerService.UpdateLibrary(refreshingMovies, Settings);
}
}
}
catch
{
lock (queue)
{
queue.Refreshing = false;
}
throw;
}
}

40
src/NzbDrone.Core/Notifications/Plex/Server/PlexServerService.cs
View File

@ -1,5 +1,6 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text.RegularExpressions;
using FluentValidation.Results;
@ -13,6 +14,7 @@ namespace NzbDrone.Core.Notifications.Plex.Server
public interface IPlexServerService
{
void UpdateLibrary(Movie movie, PlexServerSettings settings);
void UpdateLibrary(IEnumerable<Movie> movie, PlexServerSettings settings);
ValidationFailure Test(PlexServerSettings settings);
}
@ -32,10 +34,16 @@ namespace NzbDrone.Core.Notifications.Plex.Server
}
public void UpdateLibrary(Movie movie, PlexServerSettings settings)
{
UpdateLibrary(new[] { movie }, settings);
}
public void UpdateLibrary(IEnumerable<Movie> multipleMovies, PlexServerSettings settings)
{
try
{
_logger.Debug("Sending Update Request to Plex Server");
var watch = Stopwatch.StartNew();
var version = _versionCache.Get(settings.Host, () => GetVersion(settings), TimeSpan.FromHours(2));
ValidateVersion(version);
@ -45,12 +53,31 @@ namespace NzbDrone.Core.Notifications.Plex.Server
if (partialUpdates)
{
UpdatePartialSection(movie, sections, settings);
var partiallyUpdated = true;
foreach (var movie in multipleMovies)
{
partiallyUpdated &= UpdatePartialSection(movie, sections, settings);
if (!partiallyUpdated)
{
break;
}
}
// Only update complete sections if all partial updates failed
if (!partiallyUpdated)
{
_logger.Debug("Unable to update partial section, updating all Movie sections");
sections.ForEach(s => UpdateSection(s.Id, settings));
}
}
else
{
sections.ForEach(s => UpdateSection(s.Id, settings));
}
_logger.Debug("Finished sending Update Request to Plex Server (took {0} ms)", watch.ElapsedMilliseconds);
}
catch (Exception ex)
{
@ -123,7 +150,7 @@ namespace NzbDrone.Core.Notifications.Plex.Server
_plexServerProxy.Update(sectionId, settings);
}
private void UpdatePartialSection(Movie movie, List<PlexSection> sections, PlexServerSettings settings)
private bool UpdatePartialSection(Movie movie, List<PlexSection> sections, PlexServerSettings settings)
{
var partiallyUpdated = false;
@ -140,12 +167,7 @@ namespace NzbDrone.Core.Notifications.Plex.Server
}
}
// Only update complete sections if all partial updates failed
if (!partiallyUpdated)
{
_logger.Debug("Unable to update partial section, updating all Movie sections");
sections.ForEach(s => UpdateSection(s.Id, settings));
}
return partiallyUpdated;
}
private int? GetMetadataId(int sectionId, Movie movie, string language, PlexServerSettings settings)
@ -159,6 +181,8 @@ namespace NzbDrone.Core.Notifications.Plex.Server
{
try
{
_versionCache.Remove(settings.Host);
_partialUpdateCache.Remove(settings.Host);
var sections = GetSections(settings);
if (sections.Empty())