- tor-commits - lists.torproject.org

[stem/master] Allowing the DescriptorReader to be run multiple times
by atagar＠torproject.org 26 Mar '12

26 Mar '12

commit ca837ddcb6f7cc6d118d4e1a774d2c4b22dbda79 Author: Damian Johnson <atagar(a)torproject.org> Date: Mon Mar 12 08:52:17 2012 -0700 Allowing the DescriptorReader to be run multiple times Improving the usability of the DescriptorReader class by making it so callers can reuse instance multiple times to get descriptor changes since the last run. --- stem/descriptor/reader.py | 59 ++++++++++++++++++++++++-------------- test/integ/descriptor/reader.py | 26 ++++++++++++++++- 2 files changed, 62 insertions(+), 23 deletions(-) diff --git a/stem/descriptor/reader.py b/stem/descriptor/reader.py index 9ef6781..27557da 100644 --- a/stem/descriptor/reader.py +++ b/stem/descriptor/reader.py @@ -33,15 +33,15 @@ again... reader.set_processed_files(processed_files) except: pass # could not load, mabye this is the first run - with reader: - start_time = time.time() - - while time.time() - start_time < 60: - # prints any descriptors that have changed since last checked + start_time = time.time() + + while time.time() - start_time < 60: + # prints any descriptors that have changed since last checked + with reader: for descriptor in reader: print descriptor - - time.sleep(1) + + time.sleep(1) save_processed_files(reader.get_processed_files(), "/tmp/used_descriptors") @@ -55,7 +55,6 @@ DescriptorReader - Iterator for descriptor data on the local file system. |- register_skip_listener - adds a listener that's notified of skipped files |- start - begins reading descriptor data |- stop - stops reading descriptor data - |- join - joins on the thread used to process descriptor data |- __enter__ / __exit__ - manages the descriptor reader thread in the context +- __iter__ - iterates over descriptor data in unread files @@ -71,14 +70,9 @@ import threading import mimetypes import Queue -# TODO: Unimplemented concurrency features... -# - restarting when __iter__ is called additional times -# - maximum read-ahead - # TODO: Remianing impementation items... -# - impelment skip listening and add a test for it -# - remove start and join methods from header? # - implement gzip and bz2 reading +# - maximum read-ahead # Maximum number of descriptors that we'll read ahead before waiting for our # caller to fetch some of them. This is included to avoid unbounded memory @@ -182,7 +176,7 @@ def save_processed_files(processed_files, path): output_file.write("%s %i\n" % (path, timestamp)) -class DescriptorReader(threading.Thread): +class DescriptorReader: """ Iterator for the descriptor data on the local file system. This can process text files, tarball archives (gzip or bzip2), or recurse directories. @@ -194,17 +188,19 @@ class DescriptorReader(threading.Thread): """ def __init__(self, targets, follow_links = False): - threading.Thread.__init__(self, name="Descriptor Reader") - self.setDaemon(True) - self._targets = targets self._follow_links = follow_links self._skip_listeners = [] self._processed_files = {} + self._reader_thread = None + self._reader_thread_lock = threading.RLock() + self._iter_lock = threading.RLock() self._iter_notice = threading.Event() + self._is_stopped = threading.Event() + self._is_stopped.set() # Descriptors that we have read but not yet provided to the caller. A # FINISHED entry is used by the reading thread to indicate the end. @@ -256,15 +252,35 @@ class DescriptorReader(threading.Thread): self._skip_listeners.append(listener) + def start(self): + """ + Starts reading our descriptor files. + + Raises: + ValueError if we're already reading the descriptor files + """ + + with self._reader_thread_lock: + if self._reader_thread: + raise ValueError("Already running, you need to call stop() first") + else: + self._is_stopped.clear() + self._reader_thread = threading.Thread(target = self._read_descriptor_files, name="Descriptor Reader") + self._reader_thread.setDaemon(True) + self._reader_thread.start() + def stop(self): """ Stops further reading of descriptor files. """ - self._is_stopped.set() - self._iter_notice.set() + with self._reader_thread_lock: + self._is_stopped.set() + self._iter_notice.set() + self._reader_thread.join() + self._reader_thread = None - def run(self): + def _read_descriptor_files(self): remaining_files = list(self._targets) while remaining_files and not self._is_stopped.is_set(): @@ -349,5 +365,4 @@ class DescriptorReader(threading.Thread): def __exit__(self, type, value, traceback): self.stop() - self.join() diff --git a/test/integ/descriptor/reader.py b/test/integ/descriptor/reader.py index a6190ef..ded3d2d 100644 --- a/test/integ/descriptor/reader.py +++ b/test/integ/descriptor/reader.py @@ -148,6 +148,31 @@ class TestDescriptorReader(unittest.TestCase): # check that we've seen all of the descriptor_entries self.assertTrue(len(remaining_entries) == 0) + def test_multiple_runs(self): + """ + Runs a DescriptorReader instance multiple times over the same content, + making sure that it can be used repeatedly. + """ + + descriptor_path = os.path.join(DESCRIPTOR_TEST_DATA, "example_descriptor") + reader = stem.descriptor.reader.DescriptorReader([descriptor_path]) + + with reader: + self.assertEquals(len(list(reader)), 1) + + # run it a second time, this shouldn't provide any descriptors because we + # have already read it + + with reader: + self.assertEquals(len(list(reader)), 0) + + # clear the DescriptorReader's memory of seeing the file and run it again + + reader.set_processed_files([]) + + with reader: + self.assertEquals(len(list(reader)), 1) + def test_stop(self): """ Runs a DescriptorReader over the root directory, then checks that calling @@ -171,7 +196,6 @@ class TestDescriptorReader(unittest.TestCase): reader.start() time.sleep(0.1) reader.stop() - reader.join() is_test_running = False def test_get_processed_files(self):

1 0

[stem/master] Minor tidying up of integ tests
by atagar＠torproject.org 26 Mar '12

26 Mar '12

commit 63461450b7d7651c11d1f911f3443bf8a202f3f6 Author: Damian Johnson <atagar(a)torproject.org> Date: Mon Mar 12 09:34:15 2012 -0700 Minor tidying up of integ tests Reread the DescriptorReader integ tests and tidied up a few minor stylistic issues. --- test/integ/descriptor/reader.py | 34 ++++++++++++++++++---------------- 1 files changed, 18 insertions(+), 16 deletions(-) diff --git a/test/integ/descriptor/reader.py b/test/integ/descriptor/reader.py index fbacc36..a2383ee 100644 --- a/test/integ/descriptor/reader.py +++ b/test/integ/descriptor/reader.py @@ -158,20 +158,20 @@ class TestDescriptorReader(unittest.TestCase): reader = stem.descriptor.reader.DescriptorReader([descriptor_path]) with reader: - self.assertEquals(len(list(reader)), 1) + self.assertEquals(1, len(list(reader))) # run it a second time, this shouldn't provide any descriptors because we # have already read it with reader: - self.assertEquals(len(list(reader)), 0) + self.assertEquals(0, len(list(reader))) # clear the DescriptorReader's memory of seeing the file and run it again reader.set_processed_files([]) with reader: - self.assertEquals(len(list(reader)), 1) + self.assertEquals(1, len(list(reader))) def test_stop(self): """ @@ -236,7 +236,7 @@ class TestDescriptorReader(unittest.TestCase): self.assertEquals(initial_processed_files, reader.get_processed_files()) with reader: list(reader) # iterates over all of the descriptors - self.assertTrue(len(skip_listener.results) == 1) + self.assertTrue(1, len(skip_listener.results)) skipped_path, skip_exception = skip_listener.results[0] self.assertEqual(test_path, skipped_path) @@ -253,16 +253,17 @@ class TestDescriptorReader(unittest.TestCase): test_path = os.path.join(test.runner.get_runner().get_test_dir(), "test.png") - with open(test_path, "w") as test_file: - test_file.write("test data for test_skip_listener_unrecognized_type()") - try: + test_file = open(test_path, "w") + test_file.write("test data for test_skip_listener_unrecognized_type()") + test_file.close() + skip_listener = SkipListener() reader = stem.descriptor.reader.DescriptorReader([test_path]) reader.register_skip_listener(skip_listener.listener) with reader: list(reader) # iterates over all of the descriptors - self.assertTrue(len(skip_listener.results) == 1) + self.assertTrue(1, len(skip_listener.results)) skipped_path, skip_exception = skip_listener.results[0] self.assertEqual(test_path, skipped_path) @@ -279,18 +280,19 @@ class TestDescriptorReader(unittest.TestCase): test_path = os.path.join(test.runner.get_runner().get_test_dir(), "secret_file") - with open(test_path, "w") as test_file: - test_file.write("test data for test_skip_listener_unrecognized_type()") - - os.chmod(test_path, 0077) # remove read permissions - try: + test_file = open(test_path, "w") + test_file.write("test data for test_skip_listener_unrecognized_type()") + test_file.close() + + os.chmod(test_path, 0077) # remove read permissions + skip_listener = SkipListener() reader = stem.descriptor.reader.DescriptorReader([test_path]) reader.register_skip_listener(skip_listener.listener) with reader: list(reader) # iterates over all of the descriptors - self.assertTrue(len(skip_listener.results) == 1) + self.assertTrue(1, len(skip_listener.results)) skipped_path, skip_exception = skip_listener.results[0] self.assertEqual(test_path, skipped_path) @@ -305,14 +307,14 @@ class TestDescriptorReader(unittest.TestCase): Listens for a file that's skipped because the file doesn't exist. """ - test_path = "/foo/bar/doesn't_exist" + test_path = "/non-existant/path" skip_listener = SkipListener() reader = stem.descriptor.reader.DescriptorReader([test_path]) reader.register_skip_listener(skip_listener.listener) with reader: list(reader) # iterates over all of the descriptors - self.assertTrue(len(skip_listener.results) == 1) + self.assertTrue(1, len(skip_listener.results)) skipped_path, skip_exception = skip_listener.results[0] self.assertEqual(test_path, skipped_path)

1 0

[stem/master] Notifying skip listeners of files that were already read
by atagar＠torproject.org 26 Mar '12

26 Mar '12

commit 34d14afca254eac8f0493f9c9b8aabc7c5400899 Author: Damian Johnson <atagar(a)torproject.org> Date: Mon Mar 12 09:27:31 2012 -0700 Notifying skip listeners of files that were already read Skip listeners are supposed to be notified of all files that we skip, but wasn't accounting for files that we skipped because their 'last modified' timestamp indicated that they'd already been read. Modifying the integ test for set_processed_files() to test this listener condition (more graceful that the prior test that it had been doing). --- stem/descriptor/reader.py | 9 +++++++++ test/integ/descriptor/reader.py | 29 ++++++++++++++++------------- 2 files changed, 25 insertions(+), 13 deletions(-) diff --git a/stem/descriptor/reader.py b/stem/descriptor/reader.py index 27557da..6026408 100644 --- a/stem/descriptor/reader.py +++ b/stem/descriptor/reader.py @@ -59,6 +59,7 @@ DescriptorReader - Iterator for descriptor data on the local file system. +- __iter__ - iterates over descriptor data in unread files FileSkipped - Base exception for a file that was skipped. + |- AlreadyRead - We've already read a file with this last modified timestamp. |- ParsingFailure - Contents can't be parsed as descriptor data. |- UnrecognizedType - File extension indicates non-descriptor data. +- ReadFailed - Wraps an error that was raised while reading the file. @@ -86,6 +87,13 @@ FINISHED = "DONE" class FileSkipped(Exception): "Base error when we can't provide descriptor data from a file." +class AlreadyRead(FileSkipped): + "Already read a file with this 'last modified' timestamp or later." + + def __init__(self, last_modified, last_modified_when_read): + self.last_modified = last_modified + self.last_modified_when_read = last_modified_when_read + class ParsingFailure(FileSkipped): "File contents could not be parsed as descriptor data." @@ -306,6 +314,7 @@ class DescriptorReader: last_used = self._processed_files.get(target) if last_used and last_used >= last_modified: + self._notify_skip_listeners(target, AlreadyRead(last_modified, last_used)) continue else: self._processed_files[target] = last_modified diff --git a/test/integ/descriptor/reader.py b/test/integ/descriptor/reader.py index cb3454a..fbacc36 100644 --- a/test/integ/descriptor/reader.py +++ b/test/integ/descriptor/reader.py @@ -217,28 +217,31 @@ class TestDescriptorReader(unittest.TestCase): self.assertEquals(expected_results, reader.get_processed_files()) - def test_set_processed_files(self): + def test_skip_listener_already_read(self): """ Checks that calling set_processed_files() prior to reading makes us skip - those files. + those files. This also doubles for testing that skip listeners are notified + of files that we've already read. """ - # path and file contents that we want the DescriptorReader to skip - skip_file = os.path.join(DESCRIPTOR_TEST_DATA, "example_descriptor") - - with open(skip_file) as descriptor_file: - skip_contents = descriptor_file.read() - - initial_processed_files = {skip_file: sys.maxint} + # path that we want the DescriptorReader to skip + test_path = os.path.join(DESCRIPTOR_TEST_DATA, "example_descriptor") + initial_processed_files = {test_path: sys.maxint} + skip_listener = SkipListener() reader = stem.descriptor.reader.DescriptorReader([DESCRIPTOR_TEST_DATA]) + reader.register_skip_listener(skip_listener.listener) reader.set_processed_files(initial_processed_files) + self.assertEquals(initial_processed_files, reader.get_processed_files()) + with reader: list(reader) # iterates over all of the descriptors - with reader: - for descriptor in reader: - if str(descriptor) == skip_contents: - self.fail() # we read the file that we were trying to skip + self.assertTrue(len(skip_listener.results) == 1) + + skipped_path, skip_exception = skip_listener.results[0] + self.assertEqual(test_path, skipped_path) + self.assertTrue(isinstance(skip_exception, stem.descriptor.reader.AlreadyRead)) + self.assertEqual(sys.maxint, skip_exception.last_modified_when_read) def test_skip_listener_unrecognized_type(self): """

1 0

[stem/master] Minor pylint corrections
by atagar＠torproject.org 26 Mar '12

26 Mar '12

commit 924e1bd3d3a9b7b0578ad9b9bb86a86d0278ff3b Author: Damian Johnson <atagar(a)torproject.org> Date: Mon Mar 12 19:25:06 2012 -0700 Minor pylint corrections --- stem/descriptor/reader.py | 6 +++++- stem/socket.py | 2 +- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/stem/descriptor/reader.py b/stem/descriptor/reader.py index 6af2fc6..3b6e04a 100644 --- a/stem/descriptor/reader.py +++ b/stem/descriptor/reader.py @@ -81,6 +81,7 @@ class AlreadyRead(FileSkipped): "Already read a file with this 'last modified' timestamp or later." def __init__(self, last_modified, last_modified_when_read): + FileSkipped.__init__(self) self.last_modified = last_modified self.last_modified_when_read = last_modified_when_read @@ -88,18 +89,21 @@ class ParsingFailure(FileSkipped): "File contents could not be parsed as descriptor data." def __init__(self, parsing_exception): + FileSkipped.__init__(self) self.exception = parsing_exception class UnrecognizedType(FileSkipped): "File's mime type indicates that it isn't descriptor data." def __init__(self, mime_type): + FileSkipped.__init__(self) self.mime_type = mime_type class ReadFailed(FileSkipped): "An IOError occured while trying to read the file." def __init__(self, read_exception): + FileSkipped.__init__(self) self.exception = read_exception class FileMissing(ReadFailed): @@ -378,6 +382,6 @@ class DescriptorReader: def __enter__(self): self.start() - def __exit__(self, type, value, traceback): + def __exit__(self, exit_type, value, traceback): self.stop() diff --git a/stem/socket.py b/stem/socket.py index 7f15df6..a9003b0 100644 --- a/stem/socket.py +++ b/stem/socket.py @@ -254,7 +254,7 @@ class ControlSocket: def __enter__(self): return self - def __exit__(self, type, value, traceback): + def __exit__(self, exit_type, value, traceback): self.close() def _connect(self):

1 0

[stem/master] Condensing no-op iterations over DescriptorReaders
by atagar＠torproject.org 26 Mar '12

26 Mar '12

commit 5d61c479081b73eb241bf4a6c541e15ddb7ea713 Author: Damian Johnson <atagar(a)torproject.org> Date: Mon Mar 12 08:55:46 2012 -0700 Condensing no-op iterations over DescriptorReaders At multiple points in the tests we were iterating over all of a DescriptorReaders' items without doing anything. There's no need to do the iteration via a loop - coverting it into a list does the same thing. --- test/integ/descriptor/reader.py | 20 ++++---------------- 1 files changed, 4 insertions(+), 16 deletions(-) diff --git a/test/integ/descriptor/reader.py b/test/integ/descriptor/reader.py index ded3d2d..cb3454a 100644 --- a/test/integ/descriptor/reader.py +++ b/test/integ/descriptor/reader.py @@ -213,10 +213,7 @@ class TestDescriptorReader(unittest.TestCase): expected_results[path] = last_modified reader = stem.descriptor.reader.DescriptorReader([DESCRIPTOR_TEST_DATA]) - - with reader: - for descriptor in reader: - pass + with reader: list(reader) # iterates over all of the descriptors self.assertEquals(expected_results, reader.get_processed_files()) @@ -260,10 +257,7 @@ class TestDescriptorReader(unittest.TestCase): skip_listener = SkipListener() reader = stem.descriptor.reader.DescriptorReader([test_path]) reader.register_skip_listener(skip_listener.listener) - - with reader: - for descriptor in reader: - pass + with reader: list(reader) # iterates over all of the descriptors self.assertTrue(len(skip_listener.results) == 1) @@ -291,10 +285,7 @@ class TestDescriptorReader(unittest.TestCase): skip_listener = SkipListener() reader = stem.descriptor.reader.DescriptorReader([test_path]) reader.register_skip_listener(skip_listener.listener) - - with reader: - for descriptor in reader: - pass + with reader: list(reader) # iterates over all of the descriptors self.assertTrue(len(skip_listener.results) == 1) @@ -316,10 +307,7 @@ class TestDescriptorReader(unittest.TestCase): skip_listener = SkipListener() reader = stem.descriptor.reader.DescriptorReader([test_path]) reader.register_skip_listener(skip_listener.listener) - - with reader: - for descriptor in reader: - pass + with reader: list(reader) # iterates over all of the descriptors self.assertTrue(len(skip_listener.results) == 1)

1 0

[stem/master] Adding tarball support to the DescriptorReader
by atagar＠torproject.org 26 Mar '12

26 Mar '12

commit 828d5dac1c0eda9db996438af45515c0dec0cef0 Author: Damian Johnson <atagar(a)torproject.org> Date: Mon Mar 12 20:28:16 2012 -0700 Adding tarball support to the DescriptorReader Adding support for reading directly from tarballs (which is how metrics are commonly fetched). This supports all forms of compression that the tarfile module does (gzip and bz2 among other). Including some tests and archives to read against. --- stem/descriptor/reader.py | 25 +++++--- test/integ/descriptor/data/descriptor_archive.tar | Bin 0 -> 20480 bytes .../descriptor/data/descriptor_archive.tar.bz2 | Bin 0 -> 3322 bytes .../descriptor/data/descriptor_archive.tar.gz | Bin 0 -> 2844 bytes test/integ/descriptor/reader.py | 59 +++++++++++++++++++- 5 files changed, 74 insertions(+), 10 deletions(-) diff --git a/stem/descriptor/reader.py b/stem/descriptor/reader.py index 3b6e04a..198605b 100644 --- a/stem/descriptor/reader.py +++ b/stem/descriptor/reader.py @@ -67,6 +67,7 @@ FileSkipped - Base exception for a file that was skipped. """ import os +import tarfile import threading import mimetypes import Queue @@ -328,10 +329,9 @@ class DescriptorReader: if target_type[0] in (None, 'text/plain'): # either '.txt' or an unknown type self._handle_descriptor_file(target) - elif target_type == ('application/x-tar', 'gzip'): - self._handle_archive_gzip(target) - elif target_type == ('application/x-tar', 'bzip2'): - self._handle_archive_gzip(target) + elif tarfile.is_tarfile(target): + # handles gzip, bz2, and decompressed tarballs among others + self._handle_archive(target) else: self._notify_skip_listeners(target, UnrecognizedType(target_type)) @@ -355,15 +355,22 @@ class DescriptorReader: # TODO: replace with actual descriptor parsing when we have it target_file = open(target) self._enqueue_descriptor(target_file.read()) + target_file.close() + self._iter_notice.set() except IOError, exc: self._notify_skip_listeners(target, ReadFailed(exc)) - def _handle_archive_gzip(self, target): - pass # TODO: implement - - def _handle_archive_bzip(self, target): - pass # TODO: implement + def _handle_archive(self, target): + with tarfile.open(target) as tar_file: + for tar_entry in tar_file: + if tar_entry.isfile(): + # TODO: replace with actual descriptor parsing when we have it + entry = tar_file.extractfile(tar_entry) + self._enqueue_descriptor(entry.read()) + entry.close() + + self._iter_notice.set() def _enqueue_descriptor(self, descriptor): # blocks until their is either room for the descriptor or we're stopped diff --git a/test/integ/descriptor/data/descriptor_archive.tar b/test/integ/descriptor/data/descriptor_archive.tar new file mode 100644 index 0000000..2c40716 Binary files /dev/null and b/test/integ/descriptor/data/descriptor_archive.tar differ diff --git a/test/integ/descriptor/data/descriptor_archive.tar.bz2 b/test/integ/descriptor/data/descriptor_archive.tar.bz2 new file mode 100644 index 0000000..ba1f239 Binary files /dev/null and b/test/integ/descriptor/data/descriptor_archive.tar.bz2 differ diff --git a/test/integ/descriptor/data/descriptor_archive.tar.gz b/test/integ/descriptor/data/descriptor_archive.tar.gz new file mode 100644 index 0000000..63a6a57 Binary files /dev/null and b/test/integ/descriptor/data/descriptor_archive.tar.gz differ diff --git a/test/integ/descriptor/reader.py b/test/integ/descriptor/reader.py index d071db6..aa204e9 100644 --- a/test/integ/descriptor/reader.py +++ b/test/integ/descriptor/reader.py @@ -6,6 +6,7 @@ import os import sys import time import signal +import tarfile import unittest import stem.descriptor.reader @@ -20,6 +21,8 @@ BASIC_LISTING = """ my_dir = os.path.dirname(__file__) DESCRIPTOR_TEST_DATA = os.path.join(my_dir, "data") +TAR_DESCRIPTORS = [] + def _get_processed_files_path(): return os.path.join(test.runner.get_runner().get_test_dir(), "descriptor_processed_files") @@ -37,6 +40,21 @@ def _make_processed_files_listing(contents): return test_listing_path +def _get_raw_tar_descriptors(): + global TAR_DESCRIPTORS + + if not TAR_DESCRIPTORS: + test_path = os.path.join(DESCRIPTOR_TEST_DATA, "descriptor_archive.tar") + + with tarfile.open(test_path) as tar_file: + for tar_entry in tar_file: + if tar_entry.isfile(): + entry = tar_file.extractfile(tar_entry) + TAR_DESCRIPTORS.append(entry.read()) + entry.close() + + return TAR_DESCRIPTORS + class SkipListener: def __init__(self): self.results = [] # (path, exception) tuples that we've received @@ -133,7 +151,7 @@ class TestDescriptorReader(unittest.TestCase): # running this test multiple times to flush out concurrency issues for i in xrange(15): - reader = stem.descriptor.reader.DescriptorReader([DESCRIPTOR_TEST_DATA]) + reader = stem.descriptor.reader.DescriptorReader([descriptor_path]) remaining_entries = list(descriptor_entries) with reader: @@ -174,6 +192,45 @@ class TestDescriptorReader(unittest.TestCase): with reader: self.assertEquals(1, len(list(reader))) + def test_archived_uncompressed(self): + """ + Checks that we can read descriptors from an uncompressed archive. + """ + + expected_results = _get_raw_tar_descriptors() + test_path = os.path.join(DESCRIPTOR_TEST_DATA, "descriptor_archive.tar") + reader = stem.descriptor.reader.DescriptorReader([test_path]) + + with reader: + read_descriptors = [str(desc) for desc in list(reader)] + self.assertEquals(expected_results, read_descriptors) + + def test_archived_gzip(self): + """ + Checks that we can read descriptors from a gzipped archive. + """ + + expected_results = _get_raw_tar_descriptors() + test_path = os.path.join(DESCRIPTOR_TEST_DATA, "descriptor_archive.tar.gz") + reader = stem.descriptor.reader.DescriptorReader([test_path]) + + with reader: + read_descriptors = [str(desc) for desc in list(reader)] + self.assertEquals(expected_results, read_descriptors) + + def test_archived_bz2(self): + """ + Checks that we can read descriptors from an bzipped archive. + """ + + expected_results = _get_raw_tar_descriptors() + test_path = os.path.join(DESCRIPTOR_TEST_DATA, "descriptor_archive.tar.bz2") + reader = stem.descriptor.reader.DescriptorReader([test_path]) + + with reader: + read_descriptors = [str(desc) for desc in list(reader)] + self.assertEquals(expected_results, read_descriptors) + def test_stop(self): """ Runs a DescriptorReader over the root directory, then checks that calling

1 0

[stem/master] Limiting the number of buffered descriptors
by atagar＠torproject.org 26 Mar '12

26 Mar '12

commit adbc1991fdae353825b893f5e90aee6ee882e0a4 Author: Damian Johnson <atagar(a)torproject.org> Date: Mon Mar 12 19:17:33 2012 -0700 Limiting the number of buffered descriptors Preventing the DescriptorReader from having unbounded memory usage by limiting the number of descriptors that we'll store before we wait for our caller to request some. This doesn't technically make our memory usage bounded since a single descriptor doesn't have a limited size, but if one descripter can trigger the OOM killer then we have a problem. :) This isn't yet tested because we only have a single descriptor in our test data (we need at least two before we can test this). Adding a todo note for now. --- stem/descriptor/reader.py | 34 ++++++++++++++++++++-------------- test/integ/descriptor/reader.py | 1 + 2 files changed, 21 insertions(+), 14 deletions(-) diff --git a/stem/descriptor/reader.py b/stem/descriptor/reader.py index 6026408..6af2fc6 100644 --- a/stem/descriptor/reader.py +++ b/stem/descriptor/reader.py @@ -71,16 +71,6 @@ import threading import mimetypes import Queue -# TODO: Remianing impementation items... -# - implement gzip and bz2 reading -# - maximum read-ahead - -# Maximum number of descriptors that we'll read ahead before waiting for our -# caller to fetch some of them. This is included to avoid unbounded memory -# usage. This condition will be removed if set to zero. - -MAX_STORED_DESCRIPTORS = 20 - # flag to indicate when the reader thread is out of descriptor files to read FINISHED = "DONE" @@ -189,13 +179,19 @@ class DescriptorReader: Iterator for the descriptor data on the local file system. This can process text files, tarball archives (gzip or bzip2), or recurse directories. + By default this limits the number of descriptors that we'll read ahead before + waiting for our caller to fetch some of them. This is included to avoid + unbounded memory usage. + Arguments: targets (list) - paths for files or directories to be read from follow_links (bool) - determines if we'll follow symlinks when transversing directories + buffer_size (int) - descriptors we'll buffer before waiting for some to + be read, this is unbounded if zero """ - def __init__(self, targets, follow_links = False): + def __init__(self, targets, follow_links = False, buffer_size = 100): self._targets = targets self._follow_links = follow_links self._skip_listeners = [] @@ -213,7 +209,7 @@ class DescriptorReader: # Descriptors that we have read but not yet provided to the caller. A # FINISHED entry is used by the reading thread to indicate the end. - self._unreturned_descriptors = Queue.Queue() + self._unreturned_descriptors = Queue.Queue(buffer_size) def get_processed_files(self): """ @@ -335,7 +331,7 @@ class DescriptorReader: else: self._notify_skip_listeners(target, UnrecognizedType(target_type)) - self._unreturned_descriptors.put(FINISHED) + self._enqueue_descriptor(FINISHED) self._iter_notice.set() def __iter__(self): @@ -354,7 +350,7 @@ class DescriptorReader: try: # TODO: replace with actual descriptor parsing when we have it target_file = open(target) - self._unreturned_descriptors.put(target_file.read()) + self._enqueue_descriptor(target_file.read()) self._iter_notice.set() except IOError, exc: self._notify_skip_listeners(target, ReadFailed(exc)) @@ -365,6 +361,16 @@ class DescriptorReader: def _handle_archive_bzip(self, target): pass # TODO: implement + def _enqueue_descriptor(self, descriptor): + # blocks until their is either room for the descriptor or we're stopped + + while True: + try: + self._unreturned_descriptors.put(descriptor, timeout = 0.1) + return + except Queue.Full: + if self._is_stopped.is_set(): return + def _notify_skip_listeners(self, path, exception): for listener in self._skip_listeners: listener(path, exception) diff --git a/test/integ/descriptor/reader.py b/test/integ/descriptor/reader.py index a2383ee..d071db6 100644 --- a/test/integ/descriptor/reader.py +++ b/test/integ/descriptor/reader.py @@ -44,6 +44,7 @@ class SkipListener: def listener(self, path, exception): self.results.append((path, exception)) +# TODO: test buffer_size when we have more descriptor examples class TestDescriptorReader(unittest.TestCase): def tearDown(self): # cleans up 'processed file' listings that we made

1 0

[stem/master] Inverting save_processed_files arg order
by atagar＠torproject.org 26 Mar '12

26 Mar '12

commit 7faf52e5d122c9266baefec4a31e4d3a3b40ce9f Author: Damian Johnson <atagar(a)torproject.org> Date: Tue Mar 13 09:18:24 2012 -0700 Inverting save_processed_files arg order Little nicer when the usually-static argument is first imho. --- stem/descriptor/reader.py | 4 ++-- test/integ/descriptor/reader.py | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/stem/descriptor/reader.py b/stem/descriptor/reader.py index 4a0c351..ca987e8 100644 --- a/stem/descriptor/reader.py +++ b/stem/descriptor/reader.py @@ -151,15 +151,15 @@ def load_processed_files(path): return processed_files -def save_processed_files(processed_files, path): +def save_processed_files(path, processed_files): """ Persists a dictionary of 'path => last modified timestamp' mappings (as provided by the DescriptorReader's get_processed_files() method) so that they can be loaded later and applied to another DescriptorReader. Arguments: - processed_files (dict) - 'path => last modified' mappings path (str) - location to save the processed files dictionary to + processed_files (dict) - 'path => last modified' mappings Raises: IOError if unable to write to the file diff --git a/test/integ/descriptor/reader.py b/test/integ/descriptor/reader.py index a34533a..0344b99 100644 --- a/test/integ/descriptor/reader.py +++ b/test/integ/descriptor/reader.py @@ -116,7 +116,7 @@ class TestDescriptorReader(unittest.TestCase): # saves the initial_listing to a file then reloads it test_listing_path = _get_processed_files_path() - stem.descriptor.reader.save_processed_files(initial_listing, test_listing_path) + stem.descriptor.reader.save_processed_files(test_listing_path, initial_listing) loaded_listing = stem.descriptor.reader.load_processed_files(test_listing_path) self.assertEquals(initial_listing, loaded_listing) @@ -131,7 +131,7 @@ class TestDescriptorReader(unittest.TestCase): string_timestamp = {"/tmp": "123a"} for listing in (missing_filename, relative_filename, string_timestamp): - self.assertRaises(TypeError, stem.descriptor.reader.save_processed_files, listing, "/tmp/foo") + self.assertRaises(TypeError, stem.descriptor.reader.save_processed_files, "/tmp/foo", listing) def test_basic_example(self): """

1 0

[stem/master] Adding test for the DescriptorReader's buffer_size
by atagar＠torproject.org 26 Mar '12

26 Mar '12

commit 985d1c473526748013bc7f09658c5d24df76ef5c Author: Damian Johnson <atagar(a)torproject.org> Date: Tue Mar 13 09:13:35 2012 -0700 Adding test for the DescriptorReader's buffer_size Simple test that we don't read ahead more than the DescriptorReader's buffer size. --- stem/descriptor/reader.py | 13 +++++++++++++ test/integ/descriptor/reader.py | 24 +++++++++++++++++------- 2 files changed, 30 insertions(+), 7 deletions(-) diff --git a/stem/descriptor/reader.py b/stem/descriptor/reader.py index 198605b..4a0c351 100644 --- a/stem/descriptor/reader.py +++ b/stem/descriptor/reader.py @@ -261,6 +261,18 @@ class DescriptorReader: self._skip_listeners.append(listener) + def get_buffered_descriptor_count(self): + """ + Provides the number of descriptors that are waiting to be iterated over. + This is limited to the buffer_size that we were constructed with. + + Returns: + int for the estimated number of currently enqueued descriptors, this is + not entirely reliable + """ + + return self._unreturned_descriptors.qsize() + def start(self): """ Starts reading our descriptor files. @@ -388,6 +400,7 @@ class DescriptorReader: def __enter__(self): self.start() + return self def __exit__(self, exit_type, value, traceback): self.stop() diff --git a/test/integ/descriptor/reader.py b/test/integ/descriptor/reader.py index aa204e9..a34533a 100644 --- a/test/integ/descriptor/reader.py +++ b/test/integ/descriptor/reader.py @@ -62,7 +62,6 @@ class SkipListener: def listener(self, path, exception): self.results.append((path, exception)) -# TODO: test buffer_size when we have more descriptor examples class TestDescriptorReader(unittest.TestCase): def tearDown(self): # cleans up 'processed file' listings that we made @@ -192,6 +191,20 @@ class TestDescriptorReader(unittest.TestCase): with reader: self.assertEquals(1, len(list(reader))) + def test_buffer_size(self): + """ + Checks that we can process sets of descriptors larger than our buffer size, + that we don't exceed it, and that we can still stop midway through reading + them. + """ + + reader = stem.descriptor.reader.DescriptorReader([DESCRIPTOR_TEST_DATA], buffer_size = 2) + + with reader: + self.assertTrue(reader.get_buffered_descriptor_count() <= 2) + time.sleep(0.01) + self.assertTrue(reader.get_buffered_descriptor_count() <= 2) + def test_archived_uncompressed(self): """ Checks that we can read descriptors from an uncompressed archive. @@ -199,9 +212,8 @@ class TestDescriptorReader(unittest.TestCase): expected_results = _get_raw_tar_descriptors() test_path = os.path.join(DESCRIPTOR_TEST_DATA, "descriptor_archive.tar") - reader = stem.descriptor.reader.DescriptorReader([test_path]) - with reader: + with stem.descriptor.reader.DescriptorReader([test_path]) as reader: read_descriptors = [str(desc) for desc in list(reader)] self.assertEquals(expected_results, read_descriptors) @@ -212,9 +224,8 @@ class TestDescriptorReader(unittest.TestCase): expected_results = _get_raw_tar_descriptors() test_path = os.path.join(DESCRIPTOR_TEST_DATA, "descriptor_archive.tar.gz") - reader = stem.descriptor.reader.DescriptorReader([test_path]) - with reader: + with stem.descriptor.reader.DescriptorReader([test_path]) as reader: read_descriptors = [str(desc) for desc in list(reader)] self.assertEquals(expected_results, read_descriptors) @@ -225,9 +236,8 @@ class TestDescriptorReader(unittest.TestCase): expected_results = _get_raw_tar_descriptors() test_path = os.path.join(DESCRIPTOR_TEST_DATA, "descriptor_archive.tar.bz2") - reader = stem.descriptor.reader.DescriptorReader([test_path]) - with reader: + with stem.descriptor.reader.DescriptorReader([test_path]) as reader: read_descriptors = [str(desc) for desc in list(reader)] self.assertEquals(expected_results, read_descriptors)

1 0

[stem/master] Making _get_raw_tar_descriptors() thread safe
by atagar＠torproject.org 26 Mar '12

26 Mar '12

commit 5b7eacfce6ac4f370dfb4fb1baa293c4b934689f Author: Damian Johnson <atagar(a)torproject.org> Date: Tue Mar 13 09:26:01 2012 -0700 Making _get_raw_tar_descriptors() thread safe At present integ tests are single threaded, but there's a project proposal to change that and spawn threads for each target. Fixing this helper function so it won't be sad when that day comes. --- test/integ/descriptor/reader.py | 7 +++++-- 1 files changed, 5 insertions(+), 2 deletions(-) diff --git a/test/integ/descriptor/reader.py b/test/integ/descriptor/reader.py index 55644b4..cdc33c2 100644 --- a/test/integ/descriptor/reader.py +++ b/test/integ/descriptor/reader.py @@ -21,7 +21,7 @@ BASIC_LISTING = """ my_dir = os.path.dirname(__file__) DESCRIPTOR_TEST_DATA = os.path.join(my_dir, "data") -TAR_DESCRIPTORS = [] +TAR_DESCRIPTORS = None def _get_processed_files_path(): return os.path.join(test.runner.get_runner().get_test_dir(), "descriptor_processed_files") @@ -45,13 +45,16 @@ def _get_raw_tar_descriptors(): if not TAR_DESCRIPTORS: test_path = os.path.join(DESCRIPTOR_TEST_DATA, "descriptor_archive.tar") + raw_descriptors = [] with tarfile.open(test_path) as tar_file: for tar_entry in tar_file: if tar_entry.isfile(): entry = tar_file.extractfile(tar_entry) - TAR_DESCRIPTORS.append(entry.read()) + raw_descriptors.append(entry.read()) entry.close() + + TAR_DESCRIPTORS = raw_descriptors return TAR_DESCRIPTORS

1 0