[Reproducible-commits] [debbindiff] 01/01: Work-in-progress: implement parallel processing
Jérémy Bobbio
lunar at moszumanska.debian.org
Thu Jul 30 18:06:18 UTC 2015
This is an automated email from the git hooks/post-receive script.
lunar pushed a commit to branch pu/parellel
in repository debbindiff.
commit 0914ec204eae5f2ab69844513e71d536f0dac464
Author: Jérémy Bobbio <lunar at debian.org>
Date: Thu Jul 30 18:04:40 2015 +0000
Work-in-progress: implement parallel processing
---
debbindiff.py | 7 ++-
debbindiff/__init__.py | 16 ++++-
debbindiff/comparators/__init__.py | 122 +++++++++++++++++++++++++++++++++++++
debbindiff/comparators/binary.py | 15 ++++-
debbindiff/comparators/deb.py | 4 +-
debbindiff/comparators/tar.py | 10 ++-
debbindiff/comparators/utils.py | 27 +++++---
tests/comparators/conftest.py | 5 ++
tests/comparators/test_deb.py | 3 +-
9 files changed, 192 insertions(+), 17 deletions(-)
diff --git a/debbindiff.py b/debbindiff.py
index ff13f1f..c1c7db3 100755
--- a/debbindiff.py
+++ b/debbindiff.py
@@ -27,7 +27,7 @@ import codecs
import os
import sys
import traceback
-from debbindiff import logger, VERSION, set_locale
+from debbindiff import logger, VERSION, set_locale, set_jobs
import debbindiff.comparators
from debbindiff.presenters.html import output_html
from debbindiff.presenters.text import output_text
@@ -43,6 +43,8 @@ def create_parser():
help='show external tools required and exit')
parser.add_argument('--debug', dest='debug', action='store_true',
default=False, help='display debug messages')
+ parser.add_argument('--jobs', metavar='JOBS', dest='jobs', type=int,
+ help='comparisons to run simultaneously (default to CPU count)')
parser.add_argument('--html', metavar='output', dest='html_output',
help='write HTML report to given file (use - for stdout)')
parser.add_argument('--text', metavar='output', dest='text_output',
@@ -84,12 +86,15 @@ class ListToolsAction(argparse.Action):
sys.exit(0)
+
+
def main():
parser = create_parser()
parsed_args = parser.parse_args(sys.argv[1:])
if parsed_args.debug:
logger.setLevel(logging.DEBUG)
set_locale()
+ set_jobs(parsed_args.jobs)
difference = debbindiff.comparators.compare_root_paths(
parsed_args.file1, parsed_args.file2)
if difference:
diff --git a/debbindiff/__init__.py b/debbindiff/__init__.py
index 2f911de..c92d4f5 100644
--- a/debbindiff/__init__.py
+++ b/debbindiff/__init__.py
@@ -19,6 +19,7 @@
from functools import wraps
import logging
+import multiprocessing
from distutils.spawn import find_executable
import os
@@ -32,7 +33,6 @@ logger.addHandler(ch)
formatter = logging.Formatter('%(levelname)8s %(message)s')
ch.setFormatter(formatter)
-
class RequiredToolNotFound(Exception):
PROVIDERS = { 'ar': { 'debian': 'binutils-multiarch' }
, 'bzip2': { 'debian': 'bzip2' }
@@ -109,4 +109,16 @@ def set_locale():
os.environ['TZ'] = 'UTC'
-
+def set_jobs(count):
+ global jobs
+ if not count:
+ try:
+ count = multiprocessing.cpu_count()
+ logger.debug('Detected CPU count: %d', count)
+ except NotImplementedError:
+ logger.warn('Unable to determine CPU count. Please specify --jobs.')
+ count = 1
+ jobs = count
+ if jobs == 1:
+ import debbindiff.comparators
+ debbindiff.comparators.compare_many_files = lambda comparisons: map(lambda args: debbindiff.comparators.compare_files(*args), comparisons)
diff --git a/debbindiff/comparators/__init__.py b/debbindiff/comparators/__init__.py
index cd489e8..f54b6d6 100644
--- a/debbindiff/comparators/__init__.py
+++ b/debbindiff/comparators/__init__.py
@@ -18,11 +18,16 @@
# You should have received a copy of the GNU General Public License
# along with debbindiff. If not, see <http://www.gnu.org/licenses/>.
+from itertools import dropwhile
import magic
+#import multiprocessing
+import multiprocessing.dummy as multiprocessing
+import operator
import os.path
import re
import sys
from debbindiff import logger, tool_required
+import debbindiff
from debbindiff.difference import Difference
from debbindiff.comparators.binary import \
File, FilesystemFile, compare_binary_files
@@ -76,6 +81,123 @@ def compare_files(file1, file2, source=None):
return file1.compare(file2, source)
+class ComparisonPool(object):
+ def __init__(self):
+ self._manager = multiprocessing.Manager()
+ self._condition = self._manager.Condition()
+ self._running = self._manager.Value(int, 0)
+ if not hasattr(debbindiff, 'jobs'):
+ debbindiff.set_jobs(None)
+ self._pool_size = self._manager.Value(int, debbindiff.jobs - 1)
+
+ @property
+ def manager(self):
+ return self._manager
+
+ def grow(self):
+ self._condition.acquire()
+ self._pool_size.value = self._pool_size.value + 1
+ logger.debug('grow pool, new size %s', self._pool_size.value)
+ self._condition.notify()
+ self._condition.release()
+
+ def shrink(self):
+ self._condition.acquire()
+ self._pool_size.value = self._pool_size.value - 1
+ logger.debug('shrink pool, new size %s', self._pool_size.value)
+ assert self._pool_size.value >= 0
+ self._condition.release()
+
+ def task_done(self):
+ self._condition.acquire()
+ self._running.value = self._running.value - 1
+ logger.debug('task_done. running %d/%d', self._running.value, self._pool_size.value)
+ assert self._running.value >= 0
+ self._condition.notify()
+ self._condition.release()
+
+ def run_comparison(self, *args):
+ self._condition.acquire()
+ logger.debug('run comparison. running %d/%d', self._running.value, self._pool_size.value)
+ while self._running.value >= self._pool_size.value:
+ logger.debug('comparison waiting')
+ self._condition.wait()
+ self._running.value = self._running.value + 1
+ logger.debug('done waiting. running %d/%d', self._running.value, self._pool_size.value)
+ self._condition.release()
+ p = multiprocessing.Process(target=compare_for_batch, args=args)
+ p.start()
+
+
+class ComparisonBatch(object):
+ def __init__(self, pool):
+ self._pool = pool
+ self._condition = multiprocessing.Condition()
+ self._remaining = multiprocessing.Value('i', 0, lock=False)
+
+ @property
+ def pool(self):
+ return self._pool
+
+ def process(self, comparisons):
+ self._condition.acquire()
+ assert self._remaining.value == 0
+ self._remaining.value = len(comparisons)
+ self._condition.release()
+ self._outputs = self._pool.manager.list(map(lambda _: None, comparisons))
+ for position, comparison in enumerate(comparisons):
+ self._pool.run_comparison(self, position, comparison)
+
+ def set_result(self, position, result):
+ self._outputs[position] = result
+ self._condition.acquire()
+ if type(result) is Exception:
+ self._remaining.value = 0
+ else:
+ self._remaining.value = self._remaining.value - 1
+ logger.debug('%d now remaining %s', id(self), self._remaining.value)
+ self._condition.notify()
+ self._condition.release()
+
+ def join(self):
+ self._condition.acquire()
+ while self._remaining.value > 0:
+ logger.debug('%d waiting for remaining %s', id(self), self._remaining.value)
+ self._condition.wait()
+ self._condition.release()
+ logger.debug('%d batch is over', id(self))
+ for output in self._outputs:
+ if type(output) is Exception:
+ raise output
+ return self._outputs
+
+
+def compare_for_batch(batch, position, comparison):
+ try:
+ result = compare_files(*comparison)
+ except:
+ logger.error('Exception in compare_files!', exc_info=True)
+ _, result, _ = sys.exc_info()
+ logger.debug('got result %s', result)
+ batch.set_result(position, result)
+ batch.pool.task_done()
+
+
+def compare_many_files(comparisons):
+ if not hasattr(compare_many_files, 'pool'):
+ compare_many_files.pool = ComparisonPool()
+ batch = ComparisonBatch(compare_many_files.pool)
+ # We are going to keep one process waiting, so let's give more space
+ batch.pool.grow()
+ batch.process(comparisons)
+ try:
+ return batch.join()
+ finally:
+ # We're done. Take the space back
+ logger.debug('compare_many_files done')
+ batch.pool.shrink()
+
+
# The order matters! They will be tried in turns.
FILE_CLASSES = (
Directory,
diff --git a/debbindiff/comparators/binary.py b/debbindiff/comparators/binary.py
index afa19a0..9fcd6c0 100644
--- a/debbindiff/comparators/binary.py
+++ b/debbindiff/comparators/binary.py
@@ -26,6 +26,7 @@ import os.path
import re
from stat import S_ISCHR, S_ISBLK
import subprocess
+import threading
import magic
from debbindiff.difference import Difference
from debbindiff import tool_required, RequiredToolNotFound, logger
@@ -82,17 +83,25 @@ class File(object):
if not hasattr(self, '_mimedb'):
self._mimedb = magic.open(magic.NONE)
self._mimedb.load()
- return self._mimedb.file(path)
+ self._mimedb_lock = threading.Lock()
+ self._mimedb_lock.acquire()
+ ret = self._mimedb.file(path)
+ self._mimedb_lock.release()
+ return ret
@classmethod
def guess_encoding(self, path):
if not hasattr(self, '_mimedb_encoding'):
self._mimedb_encoding = magic.open(magic.MAGIC_MIME_ENCODING)
self._mimedb_encoding.load()
- return self._mimedb_encoding.file(path)
+ self._mimedb_encoding_lock = threading.Lock()
+ self._mimedb_encoding_lock.acquire()
+ ret = self._mimedb_encoding.file(path)
+ self._mimedb_encoding_lock.release()
+ return ret
def __repr__(self):
- return '<%s %s %s>' % (self.__class__, self.name, self.path)
+ return '<%s %s %s %s>' % (self.__class__.__name__, id(self), self.name, self.path)
# Path should only be used when accessing the file content (through get_content())
@property
diff --git a/debbindiff/comparators/deb.py b/debbindiff/comparators/deb.py
index 4d975af..8bd6332 100644
--- a/debbindiff/comparators/deb.py
+++ b/debbindiff/comparators/deb.py
@@ -27,7 +27,7 @@ from debbindiff.difference import Difference
import debbindiff.comparators
from debbindiff.comparators.binary import File, needs_content
from debbindiff.comparators.utils import \
- Archive, ArchiveMember, get_ar_content
+ Archive, ArchiveMember, get_ar_content, synchronized
AR_EXTRACTION_BUFFER_SIZE = 32768
@@ -39,9 +39,11 @@ class ArContainer(Archive):
# ArFile don't have to be closed
pass
+ @synchronized
def get_member_names(self):
return self.archive.getnames()
+ @synchronized
def extract(self, member_name, dest_dir):
logger.debug('ar extracting %s to %s', member_name, dest_dir)
member = self.archive.getmember(member_name)
diff --git a/debbindiff/comparators/tar.py b/debbindiff/comparators/tar.py
index d7e0440..e1303b2 100644
--- a/debbindiff/comparators/tar.py
+++ b/debbindiff/comparators/tar.py
@@ -17,6 +17,7 @@
# You should have received a copy of the GNU General Public License
# along with debbindiff. If not, see <http://www.gnu.org/licenses/>.
+from contextlib import contextmanager
import os.path
import re
import stat
@@ -30,7 +31,7 @@ from debbindiff.comparators.binary import File, needs_content
from debbindiff.comparators.device import Device
from debbindiff.comparators.directory import Directory
from debbindiff.comparators.symlink import Symlink
-from debbindiff.comparators.utils import Archive, ArchiveMember
+from debbindiff.comparators.utils import Archive, ArchiveMember, synchronized
class TarMember(ArchiveMember):
def is_directory(self):
@@ -47,6 +48,10 @@ class TarDirectory(Directory, TarMember):
def __init__(self, archive, member_name):
ArchiveMember.__init__(self, archive, member_name)
+ @contextmanager
+ def get_content(self):
+ yield
+
def compare(self, other, source=None):
return None
@@ -96,14 +101,17 @@ class TarContainer(Archive):
def close_archive(self):
self.archive.close()
+ @synchronized
def get_member_names(self):
return self.archive.getnames()
+ @synchronized
def extract(self, member_name, dest_dir):
logger.debug('tar extracting %s to %s', member_name, dest_dir)
self.archive.extract(member_name, dest_dir)
return os.path.join(dest_dir, member_name).decode('utf-8')
+ @synchronized
def get_member(self, member_name):
tarinfo = self.archive.getmember(member_name)
if tarinfo.isdir():
diff --git a/debbindiff/comparators/utils.py b/debbindiff/comparators/utils.py
index 7734a40..a1ae959 100644
--- a/debbindiff/comparators/utils.py
+++ b/debbindiff/comparators/utils.py
@@ -19,6 +19,7 @@
from abc import ABCMeta, abstractmethod
from contextlib import contextmanager
+from functools import wraps
# The following would be shutil.which in Python 3.3
import hashlib
import re
@@ -28,7 +29,7 @@ from stat import S_ISCHR, S_ISBLK
from StringIO import StringIO
import subprocess
import tempfile
-from threading import Thread
+from threading import Thread, RLock
import debbindiff.comparators
from debbindiff.comparators.binary import \
File, compare_binary_files
@@ -173,13 +174,9 @@ class Container(object):
differences = []
my_names = self.get_member_names()
other_names = other.get_member_names()
- for name in sorted(set(my_names).intersection(other_names)):
- logger.debug('compare member %s', name)
- my_file = self.get_member(name)
- other_file = other.get_member(name)
- differences.append(
- debbindiff.comparators.compare_files(
- my_file, other_file, source=name))
+ differences.extend(debbindiff.comparators.compare_many_files(
+ [(self.get_member(name), other.get_member(name), name)
+ for name in sorted(set(my_names).intersection(other_names))]))
return differences
@@ -261,3 +258,17 @@ class Archive(Container):
def get_member(self, member_name):
return ArchiveMember(self, member_name)
+
+
+# decorator
+def synchronized(original_method):
+ @wraps(original_method)
+ def wrapper(self, *args, **kwargs):
+ if not hasattr(self, '__rlock'):
+ self.__rlock = RLock()
+ try:
+ self.__rlock.acquire()
+ return original_method(self, *args, **kwargs)
+ finally:
+ self.__rlock.release()
+ return wrapper
diff --git a/tests/comparators/conftest.py b/tests/comparators/conftest.py
index 51aa2cb..d8557ea 100644
--- a/tests/comparators/conftest.py
+++ b/tests/comparators/conftest.py
@@ -27,5 +27,10 @@ def set_locale():
debbindiff.set_locale()
+ at pytest.fixture
+def no_parallel():
+ debbindiff.set_jobs(1)
+
+
def tool_missing(cmd):
return find_executable(cmd) is None
diff --git a/tests/comparators/test_deb.py b/tests/comparators/test_deb.py
index 265d7fe..4ea80d8 100644
--- a/tests/comparators/test_deb.py
+++ b/tests/comparators/test_deb.py
@@ -25,6 +25,7 @@ import pytest
from debbindiff.comparators import specialize
from debbindiff.comparators.binary import FilesystemFile
from debbindiff.comparators.deb import DebFile, Md5sumsFile
+from conftest import no_parallel
TEST_FILE1_PATH = os.path.join(os.path.dirname(__file__), '../data/test1.deb')
TEST_FILE2_PATH = os.path.join(os.path.dirname(__file__), '../data/test2.deb')
@@ -62,7 +63,7 @@ def test_identification_of_md5sums_outside_deb(tmpdir):
f = specialize(FilesystemFile(path))
assert type(f) is FilesystemFile
-def test_identification_of_md5sums_in_deb(deb1, deb2, monkeypatch):
+def test_identification_of_md5sums_in_deb(deb1, deb2, no_parallel, monkeypatch):
orig_func = Md5sumsFile.recognizes
@staticmethod
def probe(file):
--
Alioth's /usr/local/bin/git-commit-notice on /srv/git.debian.org/git/reproducible/debbindiff.git
More information about the Reproducible-commits
mailing list