From d95bb5683b6413b5d2e45dafdb86013a732e70b2 Mon Sep 17 00:00:00 2001
From: ybnd <ybnd@tuta.io>
Date: Tue, 28 Jan 2020 21:19:39 +0100
Subject: [PATCH 01/18] Analyze replaygain in parallel with
 multiprocessing.pool.ThreadPool

* Add `--jobs` or `-j` to `replaygain`-> set the pool size
* Single-threaded execution by default, if `--jobs` is unset

* If multithreaded, calls `Backend.compute_album_gain` or `Backend.compute_track_gain` asynchronously with metadata storing/writing in the callback
---
 beetsplug/replaygain.py | 123 ++++++++++++++++++++++++++++++----------
 1 file changed, 94 insertions(+), 29 deletions(-)

diff --git a/beetsplug/replaygain.py b/beetsplug/replaygain.py
index 1076ac714..e534c6fb8 100644
--- a/beetsplug/replaygain.py
+++ b/beetsplug/replaygain.py
@@ -104,6 +104,8 @@ class Backend(object):
     """An abstract class representing engine for calculating RG values.
     """
 
+    do_parallel = False
+
     def __init__(self, config, log):
         """Initialize the backend with the configuration view for the
         plugin.
@@ -135,6 +137,8 @@ class Bs1770gainBackend(Backend):
         -18: "replaygain",
     }
 
+    do_parallel = True
+
     def __init__(self, config, log):
         super(Bs1770gainBackend, self).__init__(config, log)
         config.add({
@@ -346,6 +350,9 @@ class Bs1770gainBackend(Backend):
 class FfmpegBackend(Backend):
     """A replaygain backend using ffmpeg's ebur128 filter.
     """
+
+    do_parallel = True
+
     def __init__(self, config, log):
         super(FfmpegBackend, self).__init__(config, log)
         self._ffmpeg_path = "ffmpeg"
@@ -588,6 +595,7 @@ class FfmpegBackend(Backend):
 
 # mpgain/aacgain CLI tool backend.
 class CommandBackend(Backend):
+    do_parallel = True
 
     def __init__(self, config, log):
         super(CommandBackend, self).__init__(config, log)
@@ -716,7 +724,6 @@ class CommandBackend(Backend):
 # GStreamer-based backend.
 
 class GStreamerBackend(Backend):
-
     def __init__(self, config, log):
         super(GStreamerBackend, self).__init__(config, log)
         self._import_gst()
@@ -1173,6 +1180,7 @@ class ReplayGainPlugin(BeetsPlugin):
         self.overwrite = self.config['overwrite'].get(bool)
         self.per_disc = self.config['per_disc'].get(bool)
         backend_name = self.config['backend'].as_str()
+
         if backend_name not in self.backends:
             raise ui.UserError(
                 u"Selected ReplayGain backend {0} is not supported. "
@@ -1290,8 +1298,6 @@ class ReplayGainPlugin(BeetsPlugin):
             self._log.info(u'Skipping album {0}', album)
             return
 
-        self._log.info(u'analyzing {0}', album)
-
         if (any([self.should_use_r128(item) for item in album.items()]) and not
                 all(([self.should_use_r128(item) for item in album.items()]))):
             raise ReplayGainError(
@@ -1299,6 +1305,8 @@ class ReplayGainPlugin(BeetsPlugin):
                 u" for some tracks in album {0}".format(album)
             )
 
+        self._log.info(u'analyzing {0}', album)
+
         tag_vals = self.tag_specific_values(album.items())
         store_track_gain, store_album_gain, target_level, peak = tag_vals
 
@@ -1311,27 +1319,56 @@ class ReplayGainPlugin(BeetsPlugin):
         else:
             discs[1] = album.items()
 
-        for discnumber, items in discs.items():
-            try:
-                album_gain = self.backend_instance.compute_album_gain(
-                    items, target_level, peak
+        if hasattr(self, 'pool'):
+            for discnumber, items in discs.items():
+                def _store_album(album_gain):
+                    if len(album_gain.track_gains) != len(items):
+                        raise ReplayGainError(
+                            u"ReplayGain backend failed "
+                            u"for some tracks in album {0}".format(album)
+                        )
+
+                    for item, track_gain in zip(items,
+                                                album_gain.track_gains):
+                        store_track_gain(item, track_gain)
+                        store_album_gain(item, album_gain.album_gain)
+                        if write:
+                            item.try_write()
+                        self._log.debug(u'done analyzing {0}', item)
+
+                self.pool.apply_async(
+                    self.backend_instance.compute_album_gain, args=(),
+                    kwds={
+                        "items": [i for i in items],
+                        "target_level": target_level,
+                        "peak": peak
+                    },
+                    callback=_store_album
                 )
-                if len(album_gain.track_gains) != len(items):
-                    raise ReplayGainError(
-                        u"ReplayGain backend failed "
-                        u"for some tracks in album {0}".format(album)
+        else:
+            for discnumber, items in discs.items():
+                try:
+                    album_gain = self.backend_instance.compute_album_gain(
+                        items, target_level, peak
                     )
 
-                for item, track_gain in zip(items, album_gain.track_gains):
-                    store_track_gain(item, track_gain)
-                    store_album_gain(item, album_gain.album_gain)
-                    if write:
-                        item.try_write()
-            except ReplayGainError as e:
-                self._log.info(u"ReplayGain error: {0}", e)
-            except FatalReplayGainError as e:
-                raise ui.UserError(
-                    u"Fatal replay gain error: {0}".format(e))
+                    if len(album_gain.track_gains) != len(items):
+                        raise ReplayGainError(
+                            u"ReplayGain backend failed "
+                            u"for some tracks in album {0}".format(album)
+                        )
+
+                    for item, track_gain in zip(items,
+                                                album_gain.track_gains):
+                        store_track_gain(item, track_gain)
+                        store_album_gain(item, album_gain.album_gain)
+                        if write:
+                            item.try_write()
+                except ReplayGainError as e:
+                    self._log.info(u"ReplayGain error: {0}", e)
+                except FatalReplayGainError as e:
+                    raise ui.UserError(
+                        u"Fatal replay gain error: {0}".format(e))
 
     def handle_track(self, item, write, force=False):
         """Compute track replay gain and store it in the item.
@@ -1344,23 +1381,38 @@ class ReplayGainPlugin(BeetsPlugin):
             self._log.info(u'Skipping track {0}', item)
             return
 
-        self._log.info(u'analyzing {0}', item)
-
         tag_vals = self.tag_specific_values([item])
         store_track_gain, store_album_gain, target_level, peak = tag_vals
 
-        try:
-            track_gains = self.backend_instance.compute_track_gain(
-                [item], target_level, peak
-            )
+        def _store_track(track_gains):
             if len(track_gains) != 1:
                 raise ReplayGainError(
-                    u"ReplayGain backend failed for track {0}".format(item)
+                    u"ReplayGain backend failed for track {0}".format(
+                        item)
                 )
 
             store_track_gain(item, track_gains[0])
             if write:
                 item.try_write()
+            self._log.debug(u'done analyzing {0}', item)
+
+        try:
+            if hasattr(self, 'pool'):
+                self.pool.apply_async(
+                    self.backend_instance.compute_track_gain, args=(),
+                    kwds={
+                        "items": [item],
+                        "target_level": target_level,
+                        "peak": peak,
+                    },
+                    callback=_store_track
+                )
+            else:
+                _store_track(
+                    self.backend_instance.compute_track_gain(
+                        [item], target_level, peak
+                    )
+                )
         except ReplayGainError as e:
             self._log.info(u"ReplayGain error: {0}", e)
         except FatalReplayGainError as e:
@@ -1381,17 +1433,30 @@ class ReplayGainPlugin(BeetsPlugin):
         def func(lib, opts, args):
             write = ui.should_write(opts.write)
             force = opts.force
+            jobs = opts.jobs
+
+            if self.backend_instance.do_parallel and jobs > 0:
+                from multiprocessing.pool import ThreadPool
+                self.pool = ThreadPool(jobs)
 
             if opts.album:
                 for album in lib.albums(ui.decargs(args)):
                     self.handle_album(album, write, force)
-
             else:
                 for item in lib.items(ui.decargs(args)):
                     self.handle_track(item, write, force)
 
+            if hasattr(self, 'pool'):
+                self.pool.close()
+                self.pool.join()
+                self.pool.terminate()
+
         cmd = ui.Subcommand('replaygain', help=u'analyze for ReplayGain')
         cmd.parser.add_album_option()
+        cmd.parser.add_option(
+            "-j", "--jobs", dest="jobs", type=int, default=0,
+            help=u"worker pool size"
+        )
         cmd.parser.add_option(
             "-f", "--force", dest="force", action="store_true", default=False,
             help=u"analyze all files, including those that "

From 42e895c2396c48dafcd3e853963813b380ced02b Mon Sep 17 00:00:00 2001
From: ybnd <ybnd@tuta.io>
Date: Thu, 30 Jan 2020 08:28:04 +0100
Subject: [PATCH 02/18] Match --jobs default & signature to that of convert
 plugin (--threads)

And change local function `func` to `ReplayGainPlugin` method `replaygain_func` so that `self` is passed explicitly
---
 beetsplug/replaygain.py | 54 ++++++++++++++++++++++-------------------
 1 file changed, 29 insertions(+), 25 deletions(-)

diff --git a/beetsplug/replaygain.py b/beetsplug/replaygain.py
index e534c6fb8..e361016d8 100644
--- a/beetsplug/replaygain.py
+++ b/beetsplug/replaygain.py
@@ -28,7 +28,7 @@ from six.moves import zip
 from beets import ui
 from beets.plugins import BeetsPlugin
 from beets.util import (syspath, command_output, bytestring_path,
-                        displayable_path, py3_path)
+                        displayable_path, py3_path, cpu_count)
 
 
 # Utilities.
@@ -1170,6 +1170,7 @@ class ReplayGainPlugin(BeetsPlugin):
             'overwrite': False,
             'auto': True,
             'backend': u'command',
+            'threads': cpu_count(),
             'per_disc': False,
             'peak': 'true',
             'targetlevel': 89,
@@ -1430,32 +1431,12 @@ class ReplayGainPlugin(BeetsPlugin):
     def commands(self):
         """Return the "replaygain" ui subcommand.
         """
-        def func(lib, opts, args):
-            write = ui.should_write(opts.write)
-            force = opts.force
-            jobs = opts.jobs
-
-            if self.backend_instance.do_parallel and jobs > 0:
-                from multiprocessing.pool import ThreadPool
-                self.pool = ThreadPool(jobs)
-
-            if opts.album:
-                for album in lib.albums(ui.decargs(args)):
-                    self.handle_album(album, write, force)
-            else:
-                for item in lib.items(ui.decargs(args)):
-                    self.handle_track(item, write, force)
-
-            if hasattr(self, 'pool'):
-                self.pool.close()
-                self.pool.join()
-                self.pool.terminate()
-
         cmd = ui.Subcommand('replaygain', help=u'analyze for ReplayGain')
         cmd.parser.add_album_option()
         cmd.parser.add_option(
-            "-j", "--jobs", dest="jobs", type=int, default=0,
-            help=u"worker pool size"
+            "-t", "--threads", dest="threads",
+            help=u'change the number of threads, \
+            defaults to maximum available processors'
         )
         cmd.parser.add_option(
             "-f", "--force", dest="force", action="store_true", default=False,
@@ -1467,5 +1448,28 @@ class ReplayGainPlugin(BeetsPlugin):
         cmd.parser.add_option(
             "-W", "--nowrite", dest="write", action="store_false",
             help=u"don't write metadata (opposite of -w)")
-        cmd.func = func
+        cmd.func = self.replaygain_func
         return [cmd]
+    
+    def replaygain_func(self, lib, opts, args):
+        """Handle `replaygain` ui subcommand
+        """
+        write = ui.should_write(opts.write)
+        force = opts.force
+        threads = opts.threads or self.config['threads'].get(int)
+
+        if self.backend_instance.do_parallel:
+            from multiprocessing.pool import ThreadPool
+            self.pool = ThreadPool(threads)
+
+        if opts.album:
+            for album in lib.albums(ui.decargs(args)):
+                self.handle_album(album, write, force)
+        else:
+            for item in lib.items(ui.decargs(args)):
+                self.handle_track(item, write, force)
+
+        if hasattr(self, 'pool'):
+            self.pool.close()
+            self.pool.join()
+            self.pool.terminate()

From 388d2d2c0de0b7bed66d661645b6adead79c3c09 Mon Sep 17 00:00:00 2001
From: ybnd <ybnd@tuta.io>
Date: Thu, 30 Jan 2020 09:32:03 +0100
Subject: [PATCH 03/18] Consolidate `ThreadPool` checking, opening and closing
 into methods

---
 beetsplug/replaygain.py | 37 +++++++++++++++++++++++++++----------
 1 file changed, 27 insertions(+), 10 deletions(-)

diff --git a/beetsplug/replaygain.py b/beetsplug/replaygain.py
index e361016d8..e540d3d99 100644
--- a/beetsplug/replaygain.py
+++ b/beetsplug/replaygain.py
@@ -22,6 +22,7 @@ import math
 import sys
 import warnings
 import enum
+from multiprocessing.pool import ThreadPool
 import xml.parsers.expat
 from six.moves import zip
 
@@ -1320,7 +1321,7 @@ class ReplayGainPlugin(BeetsPlugin):
         else:
             discs[1] = album.items()
 
-        if hasattr(self, 'pool'):
+        if self.has_pool():
             for discnumber, items in discs.items():
                 def _store_album(album_gain):
                     if len(album_gain.track_gains) != len(items):
@@ -1398,7 +1399,7 @@ class ReplayGainPlugin(BeetsPlugin):
             self._log.debug(u'done analyzing {0}', item)
 
         try:
-            if hasattr(self, 'pool'):
+            if self.has_pool():
                 self.pool.apply_async(
                     self.backend_instance.compute_track_gain, args=(),
                     kwds={
@@ -1420,6 +1421,27 @@ class ReplayGainPlugin(BeetsPlugin):
             raise ui.UserError(
                 u"Fatal replay gain error: {0}".format(e))
 
+    def has_pool(self):
+        """Check whether a `ThreadPool` is running instance in `self.pool`
+        """
+        if hasattr(self, 'pool'):
+            if isinstance(self.pool, ThreadPool) and self.pool._state == 'RUN':
+                return True
+        return False
+
+    def open_pool(self, threads):
+        """Open a `ThreadPool` instance in `self.pool`
+        """
+        if not self.has_pool() and self.backend_instance.do_parallel:
+            self.pool = ThreadPool(threads)
+
+    def close_pool(self):
+        """Close the `ThreadPool` instance in `self.pool` (if there is one)
+        """
+        if self.has_pool():
+            self.pool.close()
+            self.pool.join()
+
     def imported(self, session, task):
         """Add replay gain info to items or albums of ``task``.
         """
@@ -1450,7 +1472,7 @@ class ReplayGainPlugin(BeetsPlugin):
             help=u"don't write metadata (opposite of -w)")
         cmd.func = self.replaygain_func
         return [cmd]
-    
+
     def replaygain_func(self, lib, opts, args):
         """Handle `replaygain` ui subcommand
         """
@@ -1458,9 +1480,7 @@ class ReplayGainPlugin(BeetsPlugin):
         force = opts.force
         threads = opts.threads or self.config['threads'].get(int)
 
-        if self.backend_instance.do_parallel:
-            from multiprocessing.pool import ThreadPool
-            self.pool = ThreadPool(threads)
+        self.open_pool(threads)
 
         if opts.album:
             for album in lib.albums(ui.decargs(args)):
@@ -1469,7 +1489,4 @@ class ReplayGainPlugin(BeetsPlugin):
             for item in lib.items(ui.decargs(args)):
                 self.handle_track(item, write, force)
 
-        if hasattr(self, 'pool'):
-            self.pool.close()
-            self.pool.join()
-            self.pool.terminate()
+        self.close_pool()

From 79c5535cf6ea80993ae1d2f00ea15911f2581c3c Mon Sep 17 00:00:00 2001
From: ybnd <ybnd@tuta.io>
Date: Thu, 30 Jan 2020 09:34:37 +0100
Subject: [PATCH 04/18] Open/close pool at begin/end of import session

---
 beetsplug/replaygain.py | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/beetsplug/replaygain.py b/beetsplug/replaygain.py
index e540d3d99..4d963816d 100644
--- a/beetsplug/replaygain.py
+++ b/beetsplug/replaygain.py
@@ -1204,6 +1204,8 @@ class ReplayGainPlugin(BeetsPlugin):
 
         # On-import analysis.
         if self.config['auto']:
+            self.register_listener('import_begin', self.import_begin)
+            self.register_listener('import', self.import_end)
             self.import_stages = [self.imported]
 
         # Formats to use R128.
@@ -1442,6 +1444,16 @@ class ReplayGainPlugin(BeetsPlugin):
             self.pool.close()
             self.pool.join()
 
+    def import_begin(self, session):
+        """Handle `import_begin` event -> open pool
+        """
+        self.open_pool(self.config['threads'].get(int))
+
+    def import_end(self, paths):
+        """Handle `import` event -> close pool
+        """
+        self.close_pool()
+
     def imported(self, session, task):
         """Add replay gain info to items or albums of ``task``.
         """

From 0fede91bbd1a037dc7a9e277c0219f0adf7fd74b Mon Sep 17 00:00:00 2001
From: ybnd <ybnd@tuta.io>
Date: Thu, 30 Jan 2020 11:43:19 +0100
Subject: [PATCH 05/18] Workaround to pass
 ReplayGainLdnsCliMalformedTest.test_malformed_output ~ Python 3.8

---
 beetsplug/replaygain.py | 17 +++++++++++++----
 1 file changed, 13 insertions(+), 4 deletions(-)

diff --git a/beetsplug/replaygain.py b/beetsplug/replaygain.py
index 4d963816d..12b292af5 100644
--- a/beetsplug/replaygain.py
+++ b/beetsplug/replaygain.py
@@ -323,10 +323,19 @@ class Bs1770gainBackend(Backend):
         try:
             parser.Parse(text, True)
         except xml.parsers.expat.ExpatError:
-            raise ReplayGainError(
-                u'The bs1770gain tool produced malformed XML. '
-                'Using version >=0.4.10 may solve this problem.'
-            )
+            # ReplayGainLdnsCliMalformedTest.test_malformed_output
+            #   fails in Python 3.8 when executed in worker thread
+            #   the test scans log for msg but the exception is not logged,
+            #   so we inject it explicitly in order to pass
+
+            # todo: should log worker ALL thread exceptions without
+            #  having to call `self._log' explicitly for every single one
+
+            msg = u'The bs1770gain tool produced malformed XML. ' \
+                  u'Using version >=0.4.10 may solve this problem.'
+            if sys.version >= '3.8':
+                self._log.info(msg)
+            raise ReplayGainError(msg)
 
         if len(per_file_gain) != len(path_list):
             raise ReplayGainError(

From b126ecafdd6c7f15b4493fe5fc3c43f3e9d06ddb Mon Sep 17 00:00:00 2001
From: ybnd <ybnd@tuta.io>
Date: Thu, 30 Jan 2020 12:38:20 +0100
Subject: [PATCH 06/18] Clean up single/multithreaded execution selection logic

As suggested in https://github.com/beetbox/beets/pull/3478#discussion_r372467445
---
 beetsplug/replaygain.py | 83 ++++++++++++++++-------------------------
 1 file changed, 32 insertions(+), 51 deletions(-)

diff --git a/beetsplug/replaygain.py b/beetsplug/replaygain.py
index 12b292af5..031c8845a 100644
--- a/beetsplug/replaygain.py
+++ b/beetsplug/replaygain.py
@@ -1332,9 +1332,9 @@ class ReplayGainPlugin(BeetsPlugin):
         else:
             discs[1] = album.items()
 
-        if self.has_pool():
-            for discnumber, items in discs.items():
-                def _store_album(album_gain):
+        for discnumber, items in discs.items():
+            def _store_album(album_gain):
+                try:
                     if len(album_gain.track_gains) != len(items):
                         raise ReplayGainError(
                             u"ReplayGain backend failed "
@@ -1348,41 +1348,22 @@ class ReplayGainPlugin(BeetsPlugin):
                         if write:
                             item.try_write()
                         self._log.debug(u'done analyzing {0}', item)
-
-                self.pool.apply_async(
-                    self.backend_instance.compute_album_gain, args=(),
-                    kwds={
-                        "items": [i for i in items],
-                        "target_level": target_level,
-                        "peak": peak
-                    },
-                    callback=_store_album
-                )
-        else:
-            for discnumber, items in discs.items():
-                try:
-                    album_gain = self.backend_instance.compute_album_gain(
-                        items, target_level, peak
-                    )
-
-                    if len(album_gain.track_gains) != len(items):
-                        raise ReplayGainError(
-                            u"ReplayGain backend failed "
-                            u"for some tracks in album {0}".format(album)
-                        )
-
-                    for item, track_gain in zip(items,
-                                                album_gain.track_gains):
-                        store_track_gain(item, track_gain)
-                        store_album_gain(item, album_gain.album_gain)
-                        if write:
-                            item.try_write()
                 except ReplayGainError as e:
                     self._log.info(u"ReplayGain error: {0}", e)
                 except FatalReplayGainError as e:
                     raise ui.UserError(
                         u"Fatal replay gain error: {0}".format(e))
 
+            self._apply(
+                self.backend_instance.compute_album_gain, args=(),
+                kwds={
+                    "items": [i for i in items],
+                    "target_level": target_level,
+                    "peak": peak
+                },
+                callback=_store_album
+            )
+
     def handle_track(self, item, write, force=False):
         """Compute track replay gain and store it in the item.
 
@@ -1410,29 +1391,23 @@ class ReplayGainPlugin(BeetsPlugin):
             self._log.debug(u'done analyzing {0}', item)
 
         try:
-            if self.has_pool():
-                self.pool.apply_async(
-                    self.backend_instance.compute_track_gain, args=(),
-                    kwds={
-                        "items": [item],
-                        "target_level": target_level,
-                        "peak": peak,
-                    },
-                    callback=_store_track
-                )
-            else:
-                _store_track(
-                    self.backend_instance.compute_track_gain(
-                        [item], target_level, peak
-                    )
-                )
+            self._apply(
+                self.backend_instance.compute_track_gain, args=(),
+                kwds={
+                    "items": [item],
+                    "target_level": target_level,
+                    "peak": peak,
+                },
+                callback=_store_track
+            )
+
         except ReplayGainError as e:
             self._log.info(u"ReplayGain error: {0}", e)
         except FatalReplayGainError as e:
             raise ui.UserError(
                 u"Fatal replay gain error: {0}".format(e))
 
-    def has_pool(self):
+    def _has_pool(self):
         """Check whether a `ThreadPool` is running instance in `self.pool`
         """
         if hasattr(self, 'pool'):
@@ -1443,13 +1418,19 @@ class ReplayGainPlugin(BeetsPlugin):
     def open_pool(self, threads):
         """Open a `ThreadPool` instance in `self.pool`
         """
-        if not self.has_pool() and self.backend_instance.do_parallel:
+        if not self._has_pool() and self.backend_instance.do_parallel:
             self.pool = ThreadPool(threads)
 
+    def _apply(self, func, args, kwds, callback):
+        if self._has_pool():
+            self.pool.apply_async(func, args, kwds, callback)
+        else:
+            callback(func(*args, **kwds))
+
     def close_pool(self):
         """Close the `ThreadPool` instance in `self.pool` (if there is one)
         """
-        if self.has_pool():
+        if self._has_pool():
             self.pool.close()
             self.pool.join()
 

From b90358416331f416f36d6857bc3bc10534feadcb Mon Sep 17 00:00:00 2001
From: ybnd <ybnd@tuta.io>
Date: Thu, 30 Jan 2020 13:31:14 +0100
Subject: [PATCH 07/18] Fix `--threads` argument handling

---
 beetsplug/replaygain.py | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/beetsplug/replaygain.py b/beetsplug/replaygain.py
index 031c8845a..b00edcabe 100644
--- a/beetsplug/replaygain.py
+++ b/beetsplug/replaygain.py
@@ -1458,7 +1458,7 @@ class ReplayGainPlugin(BeetsPlugin):
         cmd = ui.Subcommand('replaygain', help=u'analyze for ReplayGain')
         cmd.parser.add_album_option()
         cmd.parser.add_option(
-            "-t", "--threads", dest="threads",
+            "-t", "--threads", dest="threads", type=int,
             help=u'change the number of threads, \
             defaults to maximum available processors'
         )
@@ -1480,9 +1480,10 @@ class ReplayGainPlugin(BeetsPlugin):
         """
         write = ui.should_write(opts.write)
         force = opts.force
-        threads = opts.threads or self.config['threads'].get(int)
 
-        self.open_pool(threads)
+        if opts.threads != 0:
+            threads = opts.threads or self.config['threads'].get(int)
+            self.open_pool(threads)
 
         if opts.album:
             for album in lib.albums(ui.decargs(args)):

From 65ffca215a27b5dd37b9a6ff4abfee97f0e0632a Mon Sep 17 00:00:00 2001
From: ybnd <ybnd@tuta.io>
Date: Thu, 30 Jan 2020 17:58:40 +0100
Subject: [PATCH 08/18] Exception handling in main & worker threads

* With `bs1770gain` installed the `Bs1770gainBackend` tests fail, but this should be fixed by https://github.com/beetbox/beets/pull/3480.
---
 beetsplug/replaygain.py | 64 +++++++++++++++++++++++------------------
 1 file changed, 36 insertions(+), 28 deletions(-)

diff --git a/beetsplug/replaygain.py b/beetsplug/replaygain.py
index b00edcabe..bf913b397 100644
--- a/beetsplug/replaygain.py
+++ b/beetsplug/replaygain.py
@@ -1334,13 +1334,12 @@ class ReplayGainPlugin(BeetsPlugin):
 
         for discnumber, items in discs.items():
             def _store_album(album_gain):
+                if len(album_gain.track_gains) != len(items):
+                    raise ReplayGainError(
+                        u"ReplayGain backend failed "
+                        u"for some tracks in album {0}".format(album)
+                    )
                 try:
-                    if len(album_gain.track_gains) != len(items):
-                        raise ReplayGainError(
-                            u"ReplayGain backend failed "
-                            u"for some tracks in album {0}".format(album)
-                        )
-
                     for item, track_gain in zip(items,
                                                 album_gain.track_gains):
                         store_track_gain(item, track_gain)
@@ -1351,18 +1350,23 @@ class ReplayGainPlugin(BeetsPlugin):
                 except ReplayGainError as e:
                     self._log.info(u"ReplayGain error: {0}", e)
                 except FatalReplayGainError as e:
-                    raise ui.UserError(
-                        u"Fatal replay gain error: {0}".format(e))
+                    raise ui.UserError(u"Fatal ReplayGain error: {0}".format(e))
 
-            self._apply(
-                self.backend_instance.compute_album_gain, args=(),
-                kwds={
-                    "items": [i for i in items],
-                    "target_level": target_level,
-                    "peak": peak
-                },
-                callback=_store_album
-            )
+            try:
+                self._apply(
+                    self.backend_instance.compute_album_gain, args=(),
+                    kwds={
+                        "items": [i for i in items],
+                        "target_level": target_level,
+                        "peak": peak
+                    },
+                    callback=_store_album
+                )
+            except ReplayGainError as e:
+                self._log.info(u"ReplayGain error: {0}", e)
+            except FatalReplayGainError as e:
+                raise ui.UserError(
+                    u"Fatal replay gain error: {0}".format(e))
 
     def handle_track(self, item, write, force=False):
         """Compute track replay gain and store it in the item.
@@ -1379,16 +1383,21 @@ class ReplayGainPlugin(BeetsPlugin):
         store_track_gain, store_album_gain, target_level, peak = tag_vals
 
         def _store_track(track_gains):
-            if len(track_gains) != 1:
-                raise ReplayGainError(
-                    u"ReplayGain backend failed for track {0}".format(
-                        item)
-                )
+            try:
+                if len(track_gains) != 1:
+                    raise ReplayGainError(
+                        u"ReplayGain backend failed for track {0}".format(
+                            item)
+                    )
 
-            store_track_gain(item, track_gains[0])
-            if write:
-                item.try_write()
-            self._log.debug(u'done analyzing {0}', item)
+                store_track_gain(item, track_gains[0])
+                if write:
+                    item.try_write()
+                self._log.debug(u'done analyzing {0}', item)
+            except ReplayGainError as e:
+                self._log.info(u"ReplayGain error: {0}", e)
+            except FatalReplayGainError as e:
+                raise ui.UserError(u"Fatal ReplayGain error: {0}".format(e))
 
         try:
             self._apply(
@@ -1404,8 +1413,7 @@ class ReplayGainPlugin(BeetsPlugin):
         except ReplayGainError as e:
             self._log.info(u"ReplayGain error: {0}", e)
         except FatalReplayGainError as e:
-            raise ui.UserError(
-                u"Fatal replay gain error: {0}".format(e))
+            raise ui.UserError(u"Fatal replay gain error: {0}".format(e))
 
     def _has_pool(self):
         """Check whether a `ThreadPool` is running instance in `self.pool`

From f51a68c7e100aea55b092b976076a04d59f32986 Mon Sep 17 00:00:00 2001
From: ybnd <ybnd@tuta.io>
Date: Thu, 30 Jan 2020 19:48:31 +0100
Subject: [PATCH 09/18] Implement comments

---
 beetsplug/replaygain.py | 82 ++++++++++++++++++-----------------------
 1 file changed, 35 insertions(+), 47 deletions(-)

diff --git a/beetsplug/replaygain.py b/beetsplug/replaygain.py
index bf913b397..2658d2dd3 100644
--- a/beetsplug/replaygain.py
+++ b/beetsplug/replaygain.py
@@ -1339,18 +1339,13 @@ class ReplayGainPlugin(BeetsPlugin):
                         u"ReplayGain backend failed "
                         u"for some tracks in album {0}".format(album)
                     )
-                try:
-                    for item, track_gain in zip(items,
-                                                album_gain.track_gains):
-                        store_track_gain(item, track_gain)
-                        store_album_gain(item, album_gain.album_gain)
-                        if write:
-                            item.try_write()
-                        self._log.debug(u'done analyzing {0}', item)
-                except ReplayGainError as e:
-                    self._log.info(u"ReplayGain error: {0}", e)
-                except FatalReplayGainError as e:
-                    raise ui.UserError(u"Fatal ReplayGain error: {0}".format(e))
+                for item, track_gain in zip(items,
+                                            album_gain.track_gains):
+                    store_track_gain(item, track_gain)
+                    store_album_gain(item, album_gain.album_gain)
+                    if write:
+                        item.try_write()
+                    self._log.debug(u'done analyzing {0}', item)
 
             try:
                 self._apply(
@@ -1383,21 +1378,16 @@ class ReplayGainPlugin(BeetsPlugin):
         store_track_gain, store_album_gain, target_level, peak = tag_vals
 
         def _store_track(track_gains):
-            try:
-                if len(track_gains) != 1:
-                    raise ReplayGainError(
-                        u"ReplayGain backend failed for track {0}".format(
-                            item)
-                    )
+            if len(track_gains) != 1:
+                raise ReplayGainError(
+                    u"ReplayGain backend failed for track {0}".format(
+                        item)
+                )
 
-                store_track_gain(item, track_gains[0])
-                if write:
-                    item.try_write()
-                self._log.debug(u'done analyzing {0}', item)
-            except ReplayGainError as e:
-                self._log.info(u"ReplayGain error: {0}", e)
-            except FatalReplayGainError as e:
-                raise ui.UserError(u"Fatal ReplayGain error: {0}".format(e))
+            store_track_gain(item, track_gains[0])
+            if write:
+                item.try_write()
+            self._log.debug(u'done analyzing {0}', item)
 
         try:
             self._apply(
@@ -1409,7 +1399,6 @@ class ReplayGainPlugin(BeetsPlugin):
                 },
                 callback=_store_track
             )
-
         except ReplayGainError as e:
             self._log.info(u"ReplayGain error: {0}", e)
         except FatalReplayGainError as e:
@@ -1463,6 +1452,24 @@ class ReplayGainPlugin(BeetsPlugin):
     def commands(self):
         """Return the "replaygain" ui subcommand.
         """
+        def func(lib, opts, args):
+            write = ui.should_write(opts.write)
+            force = opts.force
+
+            # Bypass self.open_pool() if called with  `--threads 0`
+            if opts.threads != 0:
+                threads = opts.threads or self.config['threads'].get(int)
+                self.open_pool(threads)
+
+            if opts.album:
+                for album in lib.albums(ui.decargs(args)):
+                    self.handle_album(album, write, force)
+            else:
+                for item in lib.items(ui.decargs(args)):
+                    self.handle_track(item, write, force)
+
+            self.close_pool()
+
         cmd = ui.Subcommand('replaygain', help=u'analyze for ReplayGain')
         cmd.parser.add_album_option()
         cmd.parser.add_option(
@@ -1480,24 +1487,5 @@ class ReplayGainPlugin(BeetsPlugin):
         cmd.parser.add_option(
             "-W", "--nowrite", dest="write", action="store_false",
             help=u"don't write metadata (opposite of -w)")
-        cmd.func = self.replaygain_func
+        cmd.func = func
         return [cmd]
-
-    def replaygain_func(self, lib, opts, args):
-        """Handle `replaygain` ui subcommand
-        """
-        write = ui.should_write(opts.write)
-        force = opts.force
-
-        if opts.threads != 0:
-            threads = opts.threads or self.config['threads'].get(int)
-            self.open_pool(threads)
-
-        if opts.album:
-            for album in lib.albums(ui.decargs(args)):
-                self.handle_album(album, write, force)
-        else:
-            for item in lib.items(ui.decargs(args)):
-                self.handle_track(item, write, force)
-
-        self.close_pool()

From b39df01d3553ee98bbb12e52ba641ac0e12c01e0 Mon Sep 17 00:00:00 2001
From: ybnd <ybnd@tuta.io>
Date: Thu, 30 Jan 2020 20:02:51 +0100
Subject: [PATCH 10/18] Add to documentation

---
 docs/plugins/replaygain.rst | 20 +++++++++++++++++++-
 1 file changed, 19 insertions(+), 1 deletion(-)

diff --git a/docs/plugins/replaygain.rst b/docs/plugins/replaygain.rst
index 9602618da..bda85bb05 100644
--- a/docs/plugins/replaygain.rst
+++ b/docs/plugins/replaygain.rst
@@ -13,12 +13,16 @@ Installation
 This plugin can use one of many backends to compute the ReplayGain values:
 GStreamer, mp3gain (and its cousin, aacgain), Python Audio Tools or ffmpeg.
 ffmpeg and mp3gain can be easier to install. mp3gain supports less audio formats
-then the other backend.
+than the other backend.
 
 Once installed, this plugin analyzes all files during the import process. This
 can be a slow process; to instead analyze after the fact, disable automatic
 analysis and use the ``beet replaygain`` command (see below).
 
+To speed up analysis with some of the avalaible backends, this plugin processes
+tracks or albums (when using the ``-a`` option) in parallel. By default,
+a single thread is used per logical core of your CPU.
+
 GStreamer
 `````````
 
@@ -35,6 +39,8 @@ the GStreamer backend by adding this to your configuration file::
     replaygain:
         backend: gstreamer
 
+The GStreamer backend does not support parallel analysis.
+
 mp3gain and aacgain
 ```````````````````
 
@@ -73,6 +79,8 @@ On OS X, most of the dependencies can be installed with `Homebrew`_::
 
     brew install mpg123 mp3gain vorbisgain faad2 libvorbis
 
+The Python Audio Tools backend does not support parallel analysis.
+
 .. _Python Audio Tools: http://audiotools.sourceforge.net
 
 ffmpeg
@@ -92,6 +100,9 @@ configuration file. The available options are:
 
 - **auto**: Enable ReplayGain analysis during import.
   Default: ``yes``.
+- **threads**: The number of parallel threads to run the analysis in. Overridden
+  by ``--threads`` at the command line.
+  Default: # of logical CPU cores
 - **backend**: The analysis backend; either ``gstreamer``, ``command``, ``audiotools``
   or ``ffmpeg``.
   Default: ``command``.
@@ -143,8 +154,15 @@ whether ReplayGain tags are written into the music files, or stored in the
 beets database only (the default is to use :ref:`the importer's configuration
 <config-import-write>`).
 
+To execute with a different number of threads, call ``beet replaygain --threads N``::
+
+    $ beet replaygain --threads N [-Waf] [QUERY]
+
+with N any integer. To disable parallelism, use ``--threads 0``.
+
 ReplayGain analysis is not fast, so you may want to disable it during import.
 Use the ``auto`` config option to control this::
 
     replaygain:
         auto: no
+

From 4a427182cddb6b4c84340c9b1ba30b1e6017f99c Mon Sep 17 00:00:00 2001
From: ybnd <ybnd@tuta.io>
Date: Fri, 31 Jan 2020 13:50:38 +0100
Subject: [PATCH 11/18] Handle exceptions in pooled threads

* ExceptionWatcher instance running in parallel to the pool, monitoring a queue for exceptions
* Pooled threads push exceptions to this queue, log non-fatal exceptions
* Application exits on fatal exception in pooled thread

* More front end info logs in the CLI
---
 beetsplug/replaygain.py | 112 ++++++++++++++++++++++++++++++++++------
 1 file changed, 97 insertions(+), 15 deletions(-)

diff --git a/beetsplug/replaygain.py b/beetsplug/replaygain.py
index 2658d2dd3..01878a3d4 100644
--- a/beetsplug/replaygain.py
+++ b/beetsplug/replaygain.py
@@ -22,9 +22,11 @@ import math
 import sys
 import warnings
 import enum
-from multiprocessing.pool import ThreadPool
+from multiprocessing.pool import ThreadPool, RUN
+from threading import Thread, Event
 import xml.parsers.expat
-from six.moves import zip
+from six.moves import zip, queue
+import six
 
 from beets import ui
 from beets.plugins import BeetsPlugin
@@ -1153,6 +1155,33 @@ class AudioToolsBackend(Backend):
         )
 
 
+class ExceptionWatcher(Thread):
+    """Monitors a queue for exceptions asynchronously.
+        Once an exception occurs, raise it and execute a callback.
+    """
+
+    def __init__(self, queue, callback):
+        self._queue = queue
+        self._callback = callback
+        self._stopevent = Event()
+        Thread.__init__(self)
+
+    def run(self):
+        while not self._stopevent.is_set():
+            try:
+                exc = self._queue.get_nowait()
+                self._callback()
+                six.reraise(exc[0], exc[1], exc[2])
+            except queue.Empty:
+                # No exceptions yet, loop back to check
+                #  whether `_stopevent` is set
+                pass
+
+    def join(self, timeout=None):
+        self._stopevent.set()
+        Thread.join(self, timeout)
+
+
 # Main plugin logic.
 
 class ReplayGainPlugin(BeetsPlugin):
@@ -1190,13 +1219,15 @@ class ReplayGainPlugin(BeetsPlugin):
 
         self.overwrite = self.config['overwrite'].get(bool)
         self.per_disc = self.config['per_disc'].get(bool)
-        backend_name = self.config['backend'].as_str()
 
-        if backend_name not in self.backends:
+        # Remember which backend is used for CLI feedback
+        self.backend_name = self.config['backend'].as_str()
+
+        if self.backend_name not in self.backends:
             raise ui.UserError(
                 u"Selected ReplayGain backend {0} is not supported. "
                 u"Please select one of: {1}".format(
-                    backend_name,
+                    self.backend_name,
                     u', '.join(self.backends.keys())
                 )
             )
@@ -1221,7 +1252,7 @@ class ReplayGainPlugin(BeetsPlugin):
         self.r128_whitelist = self.config['r128'].as_str_seq()
 
         try:
-            self.backend_instance = self.backends[backend_name](
+            self.backend_instance = self.backends[self.backend_name](
                 self.config, self._log
             )
         except (ReplayGainError, FatalReplayGainError) as e:
@@ -1334,10 +1365,15 @@ class ReplayGainPlugin(BeetsPlugin):
 
         for discnumber, items in discs.items():
             def _store_album(album_gain):
-                if len(album_gain.track_gains) != len(items):
+                if not album_gain or not album_gain.album_gain \
+                        or len(album_gain.track_gains) != len(items):
+                    # In some cases, backends fail to produce a valid
+                    # `album_gain` without throwing FatalReplayGainError
+                    #  => raise non-fatal exception & continue
                     raise ReplayGainError(
-                        u"ReplayGain backend failed "
-                        u"for some tracks in album {0}".format(album)
+                        u"ReplayGain backend `{}` failed "
+                        u"for some tracks in album {}"
+                        .format(self.backend_name, album)
                     )
                 for item, track_gain in zip(items,
                                             album_gain.track_gains):
@@ -1378,10 +1414,13 @@ class ReplayGainPlugin(BeetsPlugin):
         store_track_gain, store_album_gain, target_level, peak = tag_vals
 
         def _store_track(track_gains):
-            if len(track_gains) != 1:
+            if not track_gains or len(track_gains) != 1:
+                # In some cases, backends fail to produce a valid
+                # `track_gains` without throwing FatalReplayGainError
+                #  => raise non-fatal exception & continue
                 raise ReplayGainError(
-                    u"ReplayGain backend failed for track {0}".format(
-                        item)
+                    u"ReplayGain backend `{}` failed for track {}"
+                    .format(self.backend_name, item)
                 )
 
             store_track_gain(item, track_gains[0])
@@ -1408,7 +1447,7 @@ class ReplayGainPlugin(BeetsPlugin):
         """Check whether a `ThreadPool` is running instance in `self.pool`
         """
         if hasattr(self, 'pool'):
-            if isinstance(self.pool, ThreadPool) and self.pool._state == 'RUN':
+            if isinstance(self.pool, ThreadPool) and self.pool._state == RUN:
                 return True
         return False
 
@@ -1417,19 +1456,52 @@ class ReplayGainPlugin(BeetsPlugin):
         """
         if not self._has_pool() and self.backend_instance.do_parallel:
             self.pool = ThreadPool(threads)
+            self.exc_queue = queue.Queue()
+
+            self.exc_watcher = ExceptionWatcher(
+                self.exc_queue,      # threads push exceptions here
+                self.terminate_pool  # abort once an exception occurs
+            )
+            self.exc_watcher.start()
 
     def _apply(self, func, args, kwds, callback):
         if self._has_pool():
+            def catch_exc(func, exc_queue, log):
+                """Wrapper to catch raised exceptions in threads
+                """
+                def wfunc(*args, **kwargs):
+                    try:
+                        return func(*args, **kwargs)
+                    except ReplayGainError as e:
+                        log.info(e.args[0])  # log non-fatal exceptions
+                    except Exception:
+                        exc_queue.put(sys.exc_info())
+                return wfunc
+
+            # Wrap function and callback to catch exceptions
+            func = catch_exc(func, self.exc_queue, self._log)
+            callback = catch_exc(callback, self.exc_queue, self._log)
+
             self.pool.apply_async(func, args, kwds, callback)
         else:
             callback(func(*args, **kwds))
 
+    def terminate_pool(self):
+        """Terminate the `ThreadPool` instance in `self.pool`
+            (e.g. stop execution in case of exception)
+        """
+        if self._has_pool():
+            self.pool.terminate()
+            self.pool.join()
+            self.exc_watcher.join()
+
     def close_pool(self):
         """Close the `ThreadPool` instance in `self.pool` (if there is one)
         """
         if self._has_pool():
             self.pool.close()
             self.pool.join()
+            self.exc_watcher.join()
 
     def import_begin(self, session):
         """Handle `import_begin` event -> open pool
@@ -1462,10 +1534,20 @@ class ReplayGainPlugin(BeetsPlugin):
                 self.open_pool(threads)
 
             if opts.album:
-                for album in lib.albums(ui.decargs(args)):
+                albums = lib.albums(ui.decargs(args))
+                self._log.info(
+                    "Analyzing {} albums ~ {} backend..."
+                    .format(len(albums), self.backend_name)
+                )
+                for album in albums:
                     self.handle_album(album, write, force)
             else:
-                for item in lib.items(ui.decargs(args)):
+                items = lib.items(ui.decargs(args))
+                self._log.info(
+                    "Analyzing {} tracks ~ {} backend..."
+                    .format(len(items), self.backend_name)
+                )
+                for item in items:
                     self.handle_track(item, write, force)
 
             self.close_pool()

From 4970585b0a22a6f131f6b377bfc13733d2a427b7 Mon Sep 17 00:00:00 2001
From: ybnd <ybnd@tuta.io>
Date: Fri, 31 Jan 2020 14:08:28 +0100
Subject: [PATCH 12/18] Remove temporary workaround for silent exceptions

---
 beetsplug/replaygain.py | 16 +++-------------
 1 file changed, 3 insertions(+), 13 deletions(-)

diff --git a/beetsplug/replaygain.py b/beetsplug/replaygain.py
index 01878a3d4..1eb15724f 100644
--- a/beetsplug/replaygain.py
+++ b/beetsplug/replaygain.py
@@ -325,19 +325,9 @@ class Bs1770gainBackend(Backend):
         try:
             parser.Parse(text, True)
         except xml.parsers.expat.ExpatError:
-            # ReplayGainLdnsCliMalformedTest.test_malformed_output
-            #   fails in Python 3.8 when executed in worker thread
-            #   the test scans log for msg but the exception is not logged,
-            #   so we inject it explicitly in order to pass
-
-            # todo: should log worker ALL thread exceptions without
-            #  having to call `self._log' explicitly for every single one
-
-            msg = u'The bs1770gain tool produced malformed XML. ' \
-                  u'Using version >=0.4.10 may solve this problem.'
-            if sys.version >= '3.8':
-                self._log.info(msg)
-            raise ReplayGainError(msg)
+            raise ReplayGainError(
+                u'The bs1770gain tool produced malformed XML. '
+                u'Using version >=0.4.10 may solve this problem.')
 
         if len(per_file_gain) != len(path_list):
             raise ReplayGainError(

From 9bd78424c1a5d5e548370d277063437cb070415e Mon Sep 17 00:00:00 2001
From: ybnd <ybnd@tuta.io>
Date: Tue, 4 Feb 2020 19:06:39 +0100
Subject: [PATCH 13/18] Handle keyboard interrupts more cleanly

---
 beetsplug/replaygain.py | 66 ++++++++++++++++++++++++++---------------
 1 file changed, 42 insertions(+), 24 deletions(-)

diff --git a/beetsplug/replaygain.py b/beetsplug/replaygain.py
index 1eb15724f..cc7ce547e 100644
--- a/beetsplug/replaygain.py
+++ b/beetsplug/replaygain.py
@@ -24,6 +24,7 @@ import warnings
 import enum
 from multiprocessing.pool import ThreadPool, RUN
 from threading import Thread, Event
+import signal
 import xml.parsers.expat
 from six.moves import zip, queue
 import six
@@ -1448,6 +1449,8 @@ class ReplayGainPlugin(BeetsPlugin):
             self.pool = ThreadPool(threads)
             self.exc_queue = queue.Queue()
 
+            signal.signal(signal.SIGINT, self._interrupt)
+
             self.exc_watcher = ExceptionWatcher(
                 self.exc_queue,      # threads push exceptions here
                 self.terminate_pool  # abort once an exception occurs
@@ -1480,11 +1483,22 @@ class ReplayGainPlugin(BeetsPlugin):
         """Terminate the `ThreadPool` instance in `self.pool`
             (e.g. stop execution in case of exception)
         """
-        if self._has_pool():
+        # Don't call self._as_pool() here,
+        # self.pool._state may not be == RUN
+        if hasattr(self, 'pool') and isinstance(self.pool, ThreadPool):
             self.pool.terminate()
             self.pool.join()
             self.exc_watcher.join()
 
+    def _interrupt(self, signal, frame):
+        try:
+            self._log.info('interrupted')
+            self.terminate_pool()
+            exit(0)
+        except SystemExit:
+            # Silence raised SystemExit ~ exit(0)
+            pass
+
     def close_pool(self):
         """Close the `ThreadPool` instance in `self.pool` (if there is one)
         """
@@ -1515,32 +1529,36 @@ class ReplayGainPlugin(BeetsPlugin):
         """Return the "replaygain" ui subcommand.
         """
         def func(lib, opts, args):
-            write = ui.should_write(opts.write)
-            force = opts.force
+            try:
+                write = ui.should_write(opts.write)
+                force = opts.force
 
-            # Bypass self.open_pool() if called with  `--threads 0`
-            if opts.threads != 0:
-                threads = opts.threads or self.config['threads'].get(int)
-                self.open_pool(threads)
+                # Bypass self.open_pool() if called with  `--threads 0`
+                if opts.threads != 0:
+                    threads = opts.threads or self.config['threads'].get(int)
+                    self.open_pool(threads)
 
-            if opts.album:
-                albums = lib.albums(ui.decargs(args))
-                self._log.info(
-                    "Analyzing {} albums ~ {} backend..."
-                    .format(len(albums), self.backend_name)
-                )
-                for album in albums:
-                    self.handle_album(album, write, force)
-            else:
-                items = lib.items(ui.decargs(args))
-                self._log.info(
-                    "Analyzing {} tracks ~ {} backend..."
-                    .format(len(items), self.backend_name)
-                )
-                for item in items:
-                    self.handle_track(item, write, force)
+                if opts.album:
+                    albums = lib.albums(ui.decargs(args))
+                    self._log.info(
+                        "Analyzing {} albums ~ {} backend..."
+                        .format(len(albums), self.backend_name)
+                    )
+                    for album in albums:
+                        self.handle_album(album, write, force)
+                else:
+                    items = lib.items(ui.decargs(args))
+                    self._log.info(
+                        "Analyzing {} tracks ~ {} backend..."
+                        .format(len(items), self.backend_name)
+                    )
+                    for item in items:
+                        self.handle_track(item, write, force)
 
-            self.close_pool()
+                self.close_pool()
+            except (SystemExit, KeyboardInterrupt):
+                # Silence interrupt exceptions
+                pass
 
         cmd = ui.Subcommand('replaygain', help=u'analyze for ReplayGain')
         cmd.parser.add_album_option()

From 9b283be7bf886ff64bd7331bfe7ac28e1210ed94 Mon Sep 17 00:00:00 2001
From: ybnd <ybnd@tuta.io>
Date: Wed, 5 Feb 2020 08:29:17 +0100
Subject: [PATCH 14/18] Add to changelog

---
 docs/changelog.rst | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/docs/changelog.rst b/docs/changelog.rst
index 95fde6888..85bda3cab 100644
--- a/docs/changelog.rst
+++ b/docs/changelog.rst
@@ -107,6 +107,9 @@ New features:
   titles.
   Thanks to :user:`cole-miller`.
   :bug:`3459`
+* :doc:`/plugins/replaygain` now does its analysis in parallel when using
+  the ``command``, ``ffmpeg`` or ``bs1770gain`` backends.
+  :bug:`3478`
 
 Fixes:
 

From 50757d34ad045eb0154a499cffb76fe6d4df0883 Mon Sep 17 00:00:00 2001
From: ybnd <ybnd@tuta.io>
Date: Wed, 12 Aug 2020 11:21:35 +0200
Subject: [PATCH 15/18] Fix conflicts

---
 beetsplug/replaygain.py |   9 ++--
 docs/changelog.rst      | 103 ++++++++++++++++++++++++++++++++++++++--
 2 files changed, 105 insertions(+), 7 deletions(-)

diff --git a/beetsplug/replaygain.py b/beetsplug/replaygain.py
index cc7ce547e..2be09dac5 100644
--- a/beetsplug/replaygain.py
+++ b/beetsplug/replaygain.py
@@ -22,17 +22,18 @@ import math
 import sys
 import warnings
 import enum
+import re
+import xml.parsers.expat
+from six.moves import zip
+
 from multiprocessing.pool import ThreadPool, RUN
 from threading import Thread, Event
 import signal
-import xml.parsers.expat
-from six.moves import zip, queue
-import six
 
 from beets import ui
 from beets.plugins import BeetsPlugin
 from beets.util import (syspath, command_output, bytestring_path,
-                        displayable_path, py3_path, cpu_count)
+                        displayable_path, py3_path)
 
 
 # Utilities.
diff --git a/docs/changelog.rst b/docs/changelog.rst
index 85bda3cab..47159ddc2 100644
--- a/docs/changelog.rst
+++ b/docs/changelog.rst
@@ -6,6 +6,14 @@ Changelog
 
 New features:
 
+* :doc:`/plugins/lastgenre`: Added more heavy metal genres: https://en.wikipedia.org/wiki/Heavy_metal_genres to genres.txt and genres-tree.yaml
+* :doc:`/plugins/subsonicplaylist`: import playlist from a subsonic server.
+* A new :ref:`extra_tags` configuration option allows more tagged metadata
+  to be included in MusicBrainz queries.
+* A new :doc:`/plugins/fish` adds `Fish shell`_ tab autocompletion to beets
+* :doc:`plugins/fetchart` and :doc:`plugins/embedart`: Added a new ``quality``
+  option that controls the quality of the image output when the image is
+  resized.
 * :doc:`plugins/keyfinder`: Added support for `keyfinder-cli`_
   Thanks to :user:`BrainDamage`.
 * :doc:`plugins/fetchart`: Added a new ``high_resolution`` config option to
@@ -107,12 +115,48 @@ New features:
   titles.
   Thanks to :user:`cole-miller`.
   :bug:`3459`
-* :doc:`/plugins/replaygain` now does its analysis in parallel when using
-  the ``command``, ``ffmpeg`` or ``bs1770gain`` backends.
-  :bug:`3478`
+* :doc:`/plugins/fetchart`: Album art can now be fetched from `last.fm`_.
+  :bug:`3530`
+* The classes ``AlbumInfo`` and ``TrackInfo`` now have flexible attributes,
+  allowing to solve :bug:`1547`. 
+  Thanks to :user:`dosoe`.
+* :doc:`/plugins/web`: The query API now interprets backslashes as path
+  separators to support path queries.
+  Thanks to :user:`nmeum`.
+  :bug:`3567`
+* ``beet import`` now handles tar archives with bzip2 or gzip compression.
+  :bug:`3606`
+* :doc:`/plugins/plexupdate`: Add option to use secure connection to Plex
+  server, and to ignore certificate validation errors if necessary.
+  :bug:`2871`
+* :doc:`/plugins/lyrics`: Improved searching Genius backend when artist
+  contained special characters.
+  :bug:`3634`
+* :doc:`/plugins/parentwork`: Also get the composition date of the parent work,
+  instead of just the child work.
+  Thanks to :user:`aereaux`.
+  :bug:`3650`
+* :doc:`/plugins/lyrics`: Fix a bug in the heuristic for detecting valid
+  lyrics in the Google source of the lyrics plugin
+  :bug:`2969`
+* :doc:`/plugins/thumbnails`: Fix a bug where pathlib expected a string instead
+  of bytes for a path.
+  :bug:`3360`
+* :doc:`/plugins/convert`: If ``delete_originals`` is enabled, then the source files will
+  be deleted after importing.
+  Thanks to :user:`logan-arens`.
+  :bug:`2947`
 
 Fixes:
 
+* :doc:`/plugins/the`: Fixed incorrect regex for 'the' that matched any
+  3-letter combination of the letters t, h, e.
+  :bug:`3701`
+* :doc:`/plugins/fetchart`: Fixed a bug that caused fetchart to not take
+  environment variables such as proxy servers into account when making requests
+  :bug:`3450`
+* :doc:`/plugins/fetchart`: Temporary files for fetched album art that fail
+  validation are now removed
 * :doc:`/plugins/inline`: In function-style field definitions that refer to
   flexible attributes, values could stick around from one function invocation
   to the next. This meant that, when displaying a list of objects, later
@@ -154,6 +198,57 @@ Fixes:
 * :doc:`/plugins/bpd`: Fix the transition to next track when in consume mode.
   Thanks to :user:`aereaux`.
   :bug:`3437`
+* :doc:`/plugins/lyrics`: Fix a corner-case with Genius lowercase artist names
+  :bug:`3446`
+* :doc:`/plugins/replaygain`: Support ``bs1770gain`` v0.6.0 and up
+  :bug:`3480`
+* :doc:`/plugins/parentwork`: Don't save tracks when nothing has changed.
+  :bug:`3492`
+* Added a warning when configuration files defined in the `include` directive
+  of the configuration file fail to be imported.
+  :bug:`3498`
+* Added the normalize method to the dbcore.types.INTEGER class which now
+  properly returns integer values, which should avoid problems where fields
+  like ``bpm`` would sometimes store non-integer values.
+  :bug:`762` :bug:`3507` :bug:`3508`
+* Removed ``@classmethod`` decorator from dbcore.query.NoneQuery.match method
+  failing with AttributeError when called. It is now an instance method.
+  :bug:`3516` :bug:`3517`
+* :doc:`/plugins/lyrics`: Tolerate missing lyrics div in Genius scraper.
+  Thanks to :user:`thejli21`.
+  :bug:`3535` :bug:`3554`
+* :doc:`/plugins/lyrics`: Use the artist sort name to search for lyrics, which
+  can help find matches when the artist name has special characters.
+  Thanks to :user:`hashhar`.
+  :bug:`3340` :bug:`3558`
+* :doc:`/plugins/replaygain`: Trying to calculate volume gain for an album
+  consisting of some formats using ``ReplayGain`` and some using ``R128``
+  will no longer crash; instead it is skipped and and a message is logged.
+  The log message has also been rewritten for to improve clarity.
+  Thanks to :user:`autrimpo`.
+  :bug:`3533`
+* :doc:`/plugins/lyrics`: Adapt the Genius backend to changes in markup to
+  reduce the scraping failure rate.
+  :bug:`3535` :bug:`3594`
+* :doc:`/plugins/lyrics`: Fix crash when writing ReST files for a query without
+  results or fetched lyrics
+  :bug:`2805`
+* Adapt to breaking changes in Python's ``ast`` module in 3.8
+* :doc:`/plugins/fetchart`: Attempt to fetch pre-resized thumbnails from Cover
+  Art Archive if the ``maxwidth`` option matches one of the sizes supported by
+  the Cover Art Archive API.
+  Thanks to :user:`trolley`.
+  :bug:`3637`
+* :doc:`/plugins/ipfs`: Fix Python 3 compatibility.
+  Thanks to :user:`musoke`.
+  :bug:`2554`
+* Fix a bug that caused metadata starting with something resembling a drive
+  letter to be incorrectly split into an extra directory after the colon.
+  :bug:`3685`
+* :doc:`/plugins/mpdstats`: Don't record a skip when stopping MPD, as MPD keeps
+  the current track in the queue.
+  Thanks to :user:`aereaux`.
+  :bug:`3722`
 
 For plugin developers:
 
@@ -205,11 +300,13 @@ For packagers:
   the test may no longer be necessary.
 * This version drops support for Python 3.4.
 
+.. _Fish shell: https://fishshell.com/
 .. _MediaFile: https://github.com/beetbox/mediafile
 .. _Confuse: https://github.com/beetbox/confuse
 .. _works: https://musicbrainz.org/doc/Work
 .. _Deezer: https://www.deezer.com
 .. _keyfinder-cli: https://github.com/EvanPurkhiser/keyfinder-cli
+.. _last.fm: https://last.fm
 
 
 1.4.9 (May 30, 2019)

From f4db4ae1bfb3ee8e15862a5adb4bb4e5a28cf7a4 Mon Sep 17 00:00:00 2001
From: ybnd <ybnd@tuta.io>
Date: Wed, 12 Aug 2020 11:59:59 +0200
Subject: [PATCH 16/18] Re-add to changelog

---
 docs/changelog.rst | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/docs/changelog.rst b/docs/changelog.rst
index 47159ddc2..3278c73a8 100644
--- a/docs/changelog.rst
+++ b/docs/changelog.rst
@@ -146,6 +146,9 @@ New features:
   be deleted after importing.
   Thanks to :user:`logan-arens`.
   :bug:`2947`
+* :doc:`/plugins/replaygain` now does its analysis in parallel when using
+  the ``command``, ``ffmpeg`` or ``bs1770gain`` backends.
+  :bug:`3478`
 
 Fixes:
 

From e1876590ba26074d85b38d40b4e36d51b4a14e5a Mon Sep 17 00:00:00 2001
From: ybnd <ybnd@tuta.io>
Date: Wed, 12 Aug 2020 12:49:16 +0200
Subject: [PATCH 17/18] Fix replaygain.py to pass test_replaygain.py

---
 beetsplug/replaygain.py | 25 ++++++++++++++++++-------
 1 file changed, 18 insertions(+), 7 deletions(-)

diff --git a/beetsplug/replaygain.py b/beetsplug/replaygain.py
index 076ca0f5c..09607ffe9 100644
--- a/beetsplug/replaygain.py
+++ b/beetsplug/replaygain.py
@@ -24,16 +24,18 @@ import warnings
 import enum
 import re
 import xml.parsers.expat
-from six.moves import zip
+from six.moves import zip, queue
+import six
 
 from multiprocessing.pool import ThreadPool, RUN
 from threading import Thread, Event
 import signal
+from sqlite3 import OperationalError
 
 from beets import ui
 from beets.plugins import BeetsPlugin
 from beets.util import (syspath, command_output, bytestring_path,
-                        displayable_path, py3_path)
+                        displayable_path, py3_path, cpu_count)
 
 
 # Utilities.
@@ -1307,30 +1309,39 @@ class ReplayGainPlugin(BeetsPlugin):
                 (not item.rg_album_gain or not item.rg_album_peak)
                 for item in album.items()])
 
+    def _store(self, item):
+        try:
+            item.store()
+        except OperationalError:
+            # test_replaygain.py :memory: library can fail with
+            #    `sqlite3.OperationalError: no such table: items`
+            # but the second attempt succeeds
+            item.store()
+
     def store_track_gain(self, item, track_gain):
         item.rg_track_gain = track_gain.gain
         item.rg_track_peak = track_gain.peak
-        item.store()
+        self._store(item)
         self._log.debug(u'applied track gain {0} LU, peak {1} of FS',
                         item.rg_track_gain, item.rg_track_peak)
 
     def store_album_gain(self, item, album_gain):
         item.rg_album_gain = album_gain.gain
         item.rg_album_peak = album_gain.peak
-        item.store()
+        self._store(item)
         self._log.debug(u'applied album gain {0} LU, peak {1} of FS',
                         item.rg_album_gain, item.rg_album_peak)
 
     def store_track_r128_gain(self, item, track_gain):
         item.r128_track_gain = track_gain.gain
-        item.store()
+        self._store(item)
 
         self._log.debug(u'applied r128 track gain {0} LU',
                         item.r128_track_gain)
 
     def store_album_r128_gain(self, item, album_gain):
         item.r128_album_gain = album_gain.gain
-        item.store()
+        self._store(item)
         self._log.debug(u'applied r128 album gain {0} LU',
                         item.r128_album_gain)
 
@@ -1520,7 +1531,7 @@ class ReplayGainPlugin(BeetsPlugin):
         if hasattr(self, 'pool') and isinstance(self.pool, ThreadPool):
             self.pool.terminate()
             self.pool.join()
-            self.exc_watcher.join()
+            # self.exc_watcher.join()
 
     def _interrupt(self, signal, frame):
         try:

From 363f71af2e3544605ac560122a965ab4bf5516d5 Mon Sep 17 00:00:00 2001
From: ybnd <ybnd@tuta.io>
Date: Mon, 14 Dec 2020 22:10:02 +0100
Subject: [PATCH 18/18] Move OperationalError handler to test_replaygain.py

---
 beetsplug/replaygain.py | 16 ++++++++--------
 test/test_replaygain.py | 26 ++++++++++++++++++++++++--
 2 files changed, 32 insertions(+), 10 deletions(-)

diff --git a/beetsplug/replaygain.py b/beetsplug/replaygain.py
index 09607ffe9..31742256a 100644
--- a/beetsplug/replaygain.py
+++ b/beetsplug/replaygain.py
@@ -30,7 +30,6 @@ import six
 from multiprocessing.pool import ThreadPool, RUN
 from threading import Thread, Event
 import signal
-from sqlite3 import OperationalError
 
 from beets import ui
 from beets.plugins import BeetsPlugin
@@ -1310,13 +1309,14 @@ class ReplayGainPlugin(BeetsPlugin):
                 for item in album.items()])
 
     def _store(self, item):
-        try:
-            item.store()
-        except OperationalError:
-            # test_replaygain.py :memory: library can fail with
-            #    `sqlite3.OperationalError: no such table: items`
-            # but the second attempt succeeds
-            item.store()
+        """Store an item to the database.
+             When testing, item.store() sometimes fails non-destructively with
+             sqlite.OperationalError.
+             This method is here to be patched to a retry-once helper function
+             in test_replaygain.py, so that it can still fail appropriately
+             outside of these tests.
+        """
+        item.store()
 
     def store_track_gain(self, item, track_gain):
         item.rg_track_gain = track_gain.gain
diff --git a/test/test_replaygain.py b/test/test_replaygain.py
index 969f5c230..21a5fa5a7 100644
--- a/test/test_replaygain.py
+++ b/test/test_replaygain.py
@@ -22,11 +22,15 @@ import six
 from mock import patch
 from test.helper import TestHelper, capture_log, has_program
 
+from sqlite3 import OperationalError
+
 from beets import config
 from beets.util import CommandOutput
 from mediafile import MediaFile
 from beetsplug.replaygain import (FatalGstreamerPluginReplayGainError,
-                                  GStreamerBackend)
+                                  GStreamerBackend,
+                                  ReplayGainPlugin)
+
 
 try:
     import gi
@@ -55,10 +59,28 @@ def reset_replaygain(item):
     item['rg_album_gain'] = None
     item.write()
     item.store()
+    item.store()
+    item.store()
 
 
+def _store_retry_once(self, item):
+    """Helper method to retry item.store() once in case
+    of a sqlite3.OperationalError exception.
+
+    :param self: `ReplayGainPlugin` instance
+    :param item: a library item to store
+    """
+    try:
+        item.store()
+    except OperationalError:
+        # test_replaygain.py :memory: library can fail with
+        #    `sqlite3.OperationalError: no such table: items`
+        # but the second attempt succeeds
+        item.store()
+
+
+@patch.object(ReplayGainPlugin, '_store', _store_retry_once)
 class ReplayGainCliTestBase(TestHelper):
-
     def setUp(self):
         self.setup_beets()
         self.config['replaygain']['backend'] = self.backend