Fixed: Refreshing Plex Server series in high volume systems

(cherry picked from commit 903aba5dee0284e571fc025883abbf2a9ccf4386)

src/NzbDrone.Common/Cache/CacheManager.cs

@@ -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();

src/NzbDrone.Common/Cache/Cached.cs

@@ -3,6 +3,7 @@ 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);
+                }
+            }
+
+            if (_rollingExpiry && _defaultLifeTime.HasValue)
+            {
+                _store.TryUpdate(key, new CacheItem(cacheItem.Object, _defaultLifeTime.Value), cacheItem);
             }
 
-            return value.Object;
+            return cacheItem.Object;
         }
 
         public void Remove(string key)
@@ -72,20 +87,31 @@ namespace NzbDrone.Common.Cache
         {
             Ensure.That(key, () => key).IsNotNullOrWhiteSpace();
 
+            lifeTime = lifeTime ?? _defaultLifeTime;
+
             CacheItem cacheItem;
-            T value;
 
-            if (!_store.TryGetValue(key, out cacheItem) || cacheItem.IsExpired())
+            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));
+        }
     }
 }

src/NzbDrone.Core/MediaFiles/Events/RenameCompletedEvent.cs

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

src/NzbDrone.Core/MediaFiles/RenameTrackFileService.cs

@@ -157,6 +157,8 @@ namespace NzbDrone.Core.MediaFiles
             _logger.ProgressInfo("Renaming {0} files for {1}", trackFiles.Count, artist.Name);
             RenameFiles(trackFiles, artist);
             _logger.ProgressInfo("Selected track files renamed for {0}", artist.Name);
+
+            _eventAggregator.PublishEvent(new RenameCompletedEvent());
         }
 
         public void Execute(RenameArtistCommand message)
@@ -171,6 +173,8 @@ namespace NzbDrone.Core.MediaFiles
                 RenameFiles(trackFiles, artist);
                 _logger.ProgressInfo("All track files renamed for {0}", artist.Name);
             }
+
+            _eventAggregator.PublishEvent(new RenameCompletedEvent());
         }
     }
 }

src/NzbDrone.Core/Notifications/INotification.cs

@@ -14,6 +14,7 @@ namespace NzbDrone.Core.Notifications
         void OnDownloadFailure(DownloadFailedMessage message);
         void OnImportFailure(AlbumDownloadMessage message);
         void OnTrackRetag(TrackRetagMessage message);
+        void ProcessQueue();
         bool SupportsOnGrab { get; }
         bool SupportsOnReleaseImport { get; }
         bool SupportsOnUpgrade { get; }

src/NzbDrone.Core/Notifications/NotificationBase.cs

@@ -64,6 +64,10 @@ namespace NzbDrone.Core.Notifications
         {
         }
 
+        public virtual void ProcessQueue()
+        {
+        }
+
         public bool SupportsOnGrab => HasConcreteImplementation("OnGrab");
         public bool SupportsOnRename => HasConcreteImplementation("OnRename");
         public bool SupportsOnReleaseImport => HasConcreteImplementation("OnReleaseImport");

src/NzbDrone.Core/Notifications/NotificationService.cs

@@ -21,7 +21,9 @@ namespace NzbDrone.Core.Notifications
           IHandle<HealthCheckFailedEvent>,
           IHandle<DownloadFailedEvent>,
           IHandle<AlbumImportIncompleteEvent>,
-          IHandle<TrackFileRetaggedEvent>
+          IHandle<TrackFileRetaggedEvent>,
+          IHandleAsync<RenameCompletedEvent>,
+          IHandleAsync<HealthCheckCompleteEvent>
     {
         private readonly INotificationFactory _notificationFactory;
         private readonly Logger _logger;
@@ -272,5 +274,30 @@ namespace NzbDrone.Core.Notifications
                 }
             }
         }
+
+        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);
+                }
+            }
+        }
     }
 }

src/NzbDrone.Core/Notifications/Plex/Server/PlexServer.cs

@@ -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.Music;
@@ -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, Artist> Pending { get; } = new Dictionary<int, Artist>();
+            public bool Refreshing { get; set; }
+        }
+
+        private readonly ICached<PlexUpdateQueue> _pendingArtistCache;
+
+        public PlexServer(IPlexServerService plexServerService, IPlexTvService plexTvService, ICacheManager cacheManager, Logger logger)
         {
             _plexServerService = plexServerService;
             _plexTvService = plexTvService;
+            _logger = logger;
+
+            _pendingArtistCache = cacheManager.GetRollingCache<PlexUpdateQueue>(GetType(), "pendingArtists", TimeSpan.FromDays(1));
         }
 
         public override string Link => "https://www.plex.tv/";
@@ -42,7 +57,65 @@ namespace NzbDrone.Core.Notifications.Plex.Server
         {
             if (Settings.UpdateLibrary)
             {
-                _plexServerService.UpdateLibrary(artist, Settings);
+                _logger.Debug("Scheduling library update for artist {0} {1}", artist.Id, artist.Name);
+                var queue = _pendingArtistCache.Get(Settings.Host, () => new PlexUpdateQueue());
+                lock (queue)
+                {
+                    queue.Pending[artist.Id] = artist;
+                }
+            }
+        }
+
+        public override void ProcessQueue()
+        {
+            PlexUpdateQueue queue = _pendingArtistCache.Find(Settings.Host);
+            if (queue == null)
+            {
+                return;
+            }
+
+            lock (queue)
+            {
+                if (queue.Refreshing)
+                {
+                    return;
+                }
+
+                queue.Refreshing = true;
+            }
+
+            try
+            {
+                while (true)
+                {
+                    List<Artist> refreshingArtist;
+                    lock (queue)
+                    {
+                        if (queue.Pending.Empty())
+                        {
+                            queue.Refreshing = false;
+                            return;
+                        }
+
+                        refreshingArtist = queue.Pending.Values.ToList();
+                        queue.Pending.Clear();
+                    }
+
+                    if (Settings.UpdateLibrary)
+                    {
+                        _logger.Debug("Performing library update for {0} artist", refreshingArtist.Count);
+                        _plexServerService.UpdateLibrary(refreshingArtist, Settings);
+                    }
+                }
+            }
+            catch
+            {
+                lock (queue)
+                {
+                    queue.Refreshing = false;
+                }
+
+                throw;
             }
         }
 

src/NzbDrone.Core/Notifications/Plex/Server/PlexServerService.cs

@@ -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(Artist artist, PlexServerSettings settings);
+        void UpdateLibrary(IEnumerable<Artist> artists, PlexServerSettings settings);
         ValidationFailure Test(PlexServerSettings settings);
     }
 
@@ -32,10 +34,16 @@ namespace NzbDrone.Core.Notifications.Plex.Server
         }
 
         public void UpdateLibrary(Artist artist, PlexServerSettings settings)
+        {
+            UpdateLibrary(new[] { artist }, settings);
+        }
+
+        public void UpdateLibrary(IEnumerable<Artist> multipleArtist, 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(artist, sections, settings);
+                    var partiallyUpdated = true;
+
+                    foreach (var artist in multipleArtist)
+                    {
+                        partiallyUpdated &= UpdatePartialSection(artist, 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 Music 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(Artist artist, List<PlexSection> sections, PlexServerSettings settings)
+        private bool UpdatePartialSection(Artist artist, 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 Music sections");
-                sections.ForEach(s => UpdateSection(s.Id, settings));
-            }
+            return partiallyUpdated;
         }
 
         private int? GetMetadataId(int sectionId, Artist artist, 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())