diff --git a/.gitignore b/.gitignore index c97db2f..b032e02 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,7 @@ +.tox/ +.cache/ .python-version dist python_persistent_queue.egg-info *.queue *.coverage -.cache diff --git a/.travis.yml b/.travis.yml index a21176f..5c60502 100644 --- a/.travis.yml +++ b/.travis.yml @@ -4,15 +4,22 @@ cache: pip matrix: fast_finish: true include: - - python: "2.7" - - python: "3.4" - - python: "3.5" + - python: 3.5 + env: TOXENV=lint + - python: 2.7 + env: TOXENV=py27 + - python: 3.4 + env: TOXENV=py34 + - python: 3.5 + env: TOXENV=py35 - python: "3.6-dev" + env: TOXENV=py36 allow_failures: - python: "3.6-dev" + env: TOXENV=py36 install: - - "pip install -r requirements_test.txt" + - "pip install tox" script: - - "py.test -v --cov persistent_queue" + - "tox" after_success: - coveralls diff --git a/persistent_queue/__init__.py b/persistent_queue/__init__.py index 24bff75..3a08262 100644 --- a/persistent_queue/__init__.py +++ b/persistent_queue/__init__.py @@ -1,291 +1,7 @@ -""" -An implementation of a persistent queue. It is optimized for peeking at values -and then deleting them off to top of the queue. -""" +from __future__ import absolute_import -import logging -import os.path -import pickle -import shutil -import struct -import threading -import uuid +from .persistent_queue import PersistentQueue -LENGTH_STRUCT = 'I' -HEADER_STRUCT = 'II' -START_OFFSET = 4 + 4 - -_LOGGER = logging.getLogger(__name__) - - -class PersistentQueue: - def __init__(self, filename, path='.', dumps=pickle.dumps, - loads=pickle.loads, flush_limit=1048576): - self.filename = filename - self.path = path - self.file = self._open_file() - self.loads = loads - self.dumps = dumps - self.flush_limit = flush_limit - self.file_lock = threading.RLock() - self.pop_lock = threading.RLock() - self.pushed_event = threading.Event() - - self.file.seek(0, 0) - self.length = struct.unpack(HEADER_STRUCT[0], self.file.read(4))[0] - - def _open_file(self, mode=None): - filename = os.path.join(self.path, self.filename) - - mode = mode or 'r+b' if os.path.isfile(filename) else 'w+b' - file = open(filename, mode=mode, buffering=0) - - if mode == 'w+b': - # write length and start pointer - file.write(struct.pack(HEADER_STRUCT, 0, START_OFFSET)) - - return file - - def _update_length(self, length): - current_pos = self.file.tell() - - self.file.seek(0, 0) # Go to the beginning of the file - self.file.write(struct.pack(HEADER_STRUCT[0], length)) - self.file.flush() # Probably not necessary since buffering=0 - os.fsync(self.file.fileno()) - - self.length = length - - self.file.seek(current_pos, 0) - - def _get_queue_top(self): - current_pos = self.file.tell() - - self.file.seek(START_OFFSET - 4, 0) # Start at beginning of file - pos = struct.unpack(HEADER_STRUCT[1], self.file.read(4))[0] - - self.file.seek(current_pos, 0) - return pos - - def _set_queue_top(self, top): - current_pos = self.file.tell() - - self.file.seek(START_OFFSET - 4, 0) # Start at beginning of file - self.file.write(struct.pack(HEADER_STRUCT[1], top)) - self.file.flush() # Probably not necessary since buffering=0 - os.fsync(self.file.fileno()) - - self.file.seek(current_pos, 0) - - def clear(self): - """Removes all elements from queue.""" - _LOGGER.debug("Clearing the queue") - with self.file_lock, self.pop_lock: - self.file.close() - self.file = self._open_file(mode='w+b') - self.length = 0 - _LOGGER.debug("The queue has been cleared") - - def copy(self, new_filename, path=None): - """Copies a queue to a new queue.""" - old = os.path.join(self.path, self.filename) - new = os.path.join(path or self.path, new_filename) - shutil.copy2(old, new) - - return PersistentQueue(filename=new_filename, - path=path or self.path, - flush_limit=self.flush_limit) - - def count(self): - """Return the size of the queue.""" - return self.length - - def flush(self): - """ - Removes elements that have been deleted or popped from the queue. This - will be taken care of by pop and delete. - """ - _LOGGER.debug("Flushing the queue") - - with self.file_lock: - pos = self._get_queue_top() - - if pos < self.flush_limit: - # Ignore if the file isn't big enough -- it's not worth it - _LOGGER.debug("Ignoring flush because we haven't met the limit") - return - - # Make a new file - random = str(uuid.uuid4()).replace('-', '') - temp_filename = os.path.join(self.path, self.filename + '-' + random) - new_file = open(temp_filename, mode='w+b', buffering=0) - - # From this point on, the file can not change - with self.file_lock, self.pop_lock: - # Make sure everything is to disk - self.file.flush() - os.fsync(self.file.fileno()) - - start = self._get_queue_top() # Get it again in case it changed - self.file.seek(0, 2) # Go to end of file - end = self.file.tell() - - _LOGGER.debug("Writing data to new file") - # Copy over meta data - new_file.write(struct.pack(HEADER_STRUCT, - self.count(), - START_OFFSET)) - - # Copy over data - # Do it in chunks so we aren't loading tons of data into memory - self.file.seek(start, 0) - bytes_read = 0 - chunk_size = 4096 - while bytes_read < end - start: - bytes_read += chunk_size - new_file.write(self.file.read(bytes_read)) - - new_file.flush() # Probably not necessary since buffering=0 - os.fsync(new_file.fileno()) - new_file.close() - self.file.close() - - # So far everything above this point has been safe. If something - # crashed, the data would still be preserved. Now we are entering - # the danger zone. - - _LOGGER.debug("Replacing old file with new file") - os.remove(os.path.join(self.path, self.filename)) - os.rename(temp_filename, os.path.join(self.path, self.filename)) - self.file = self._open_file() - - _LOGGER.debug("Finished flushing the queue") - - def pop(self, items=1, blocking=False): - """ - Removes and returns a certain amount of items from the queue. If items - is greater than one, a list is returned. - """ - _LOGGER.debug("Popping %s items", items) - - # Ignore requests for zero items - if items == 0: - _LOGGER.debug("Returning empty list") - return [] - - with self.pop_lock: - data, queue_top = self._peek(items, blocking) - - with self.file_lock: - self._set_queue_top(queue_top) - - if isinstance(data, list): - if len(data) > 0: - self._update_length(self.count() - len(data)) - elif data is not None: - self._update_length(self.count() - 1) - - _LOGGER.debug("Returning data from the pop") - return data - - def _peek(self, items=1, blocking=False): - """ - Returns a certain amount of items from the queue. If items is greater - than one, a list is returned. - """ - def read_data(): - length = struct.unpack(LENGTH_STRUCT, self.file.read(4))[0] - data = self.file.read(length) - return self.loads(data) - - _LOGGER.debug("Peeking %s items", items) - - # Ignore requests for zero items - if items == 0: - _LOGGER.debug("Returning empty list") - return [], self.file.tell() - - if blocking: - while self.count() < items: - self.pushed_event.wait() - self.pushed_event.clear() - - with self.file_lock: - self.file.seek(self._get_queue_top(), 0) # Beginning of data - total_items = self.count() if items > self.count() else items - data = [read_data() for i in range(total_items)] - queue_top = self.file.tell() - - if items == 1: - if len(data) == 0: - _LOGGER.debug("No items to peek at so returning None") - return None, queue_top - else: - _LOGGER.debug("Returning data from peek") - return data[0], queue_top - else: - _LOGGER.debug("Returning data from peek") - return data, queue_top - - def peek(self, items=1, blocking=False): - with self.pop_lock: - return self._peek(items, blocking)[0] - - def delete(self, items=1): - """Removes items from queue. Nothing is returned.""" - def read_length(): - length = struct.unpack(LENGTH_STRUCT, self.file.read(4))[0] - self.file.seek(length, 1) - - _LOGGER.debug("Deleting %s items", items) - - # Ignore requests for zero items - if items == 0: - _LOGGER.debug("Ignoring request to delete") - return - - with self.file_lock, self.pop_lock: - self.file.seek(self._get_queue_top(), 0) # Beginning of data - total_items = self.count() if items > self.count() else items - - for _ in range(total_items): - read_length() - - self._set_queue_top(self.file.tell()) - self._update_length(self.count() - total_items) - - _LOGGER.debug("Done deleting data") - - def push(self, items): - """Add items to the queue.""" - def write_data(item): - data = self.dumps(item) - self.file.write(struct.pack(LENGTH_STRUCT, len(data))) - self.file.write(data) - self.file.flush() # Probably not necessary since buffering=0 - os.fsync(self.file.fileno()) - - if not isinstance(items, list): - items = [items] - - _LOGGER.debug("Pushing %s items", len(items)) - - # Ignore requests for adding zero items - if len(items) == 0: - _LOGGER.debug("Pushing zero items, ignoring request") - return - - with self.file_lock: - self.file.seek(0, 2) # Go to end of file - - for i in items: - write_data(i) - - self._update_length(self.count() + len(items)) - - self.pushed_event.set() - _LOGGER.debug("Done pushing data") - - def __len__(self): - """Get size of queue.""" - return self.count() +__all__ = [ + 'PersistentQueue', +] diff --git a/persistent_queue/persistent_queue.py b/persistent_queue/persistent_queue.py new file mode 100644 index 0000000..bdeb5f2 --- /dev/null +++ b/persistent_queue/persistent_queue.py @@ -0,0 +1,470 @@ +""" +An implementation of a persistent queue. It is optimized for peeking at values +and then deleting them off to top of the queue. +""" + +import logging +import os.path +import pickle +import shutil +import struct +import threading +import time +import uuid + +try: + import queue +except ImportError: # pragma: no cover + import Queue as queue + +LENGTH_STRUCT = 'I' +HEADER_STRUCT = 'II' +START_OFFSET = 4 + 4 + +_LOGGER = logging.getLogger(__name__) + + +class PersistentQueue: + def __init__(self, filename, maxsize=0, dumps=pickle.dumps, loads=pickle.loads, flush_limit=1048576): + """ + Creates a new PersistentQueue object and underlying file. + + filename: must be a full path to the new file. + maxsize: upperbound limit of items that can be placed in the queue. + dumps: the function called for persisting queue items to the file. + loads: the function called for loading queue items from the file. + flush_limit: below this filesize, flush() is a no-op. + """ + if maxsize < 0: + maxsize = 0 + + self.maxsize = maxsize + self.filename = os.path.abspath(filename) + self.dumps = dumps + self.loads = loads + self.flush_limit = flush_limit + + self._file = self._open_file() + self._file_lock = threading.RLock() + self._get_lock = threading.RLock() + self._get_event = threading.Event() + self._put_lock = threading.RLock() + self._put_event = threading.Event() + + self._all_tasks_done = threading.Condition() + self._unfinished_tasks = 0 + + self._file.seek(0, 0) + self._length = struct.unpack(HEADER_STRUCT[0], self._file.read(4))[0] + + def _open_file(self, mode=None): + mode = mode or 'r+b' if os.path.isfile(self.filename) else 'w+b' + file = open(self.filename, mode=mode, buffering=0) + + if mode == 'w+b': + # write length and start pointer + file.write(struct.pack(HEADER_STRUCT, 0, START_OFFSET)) + + return file + + def _update_length(self, length): + current_pos = self._file.tell() + + self._file.seek(0, 0) # Go to the beginning of the file + self._file.write(struct.pack(HEADER_STRUCT[0], length)) + self._file.flush() # Probably not necessary since buffering=0 + os.fsync(self._file.fileno()) + + self._length = length + + self._file.seek(current_pos, 0) + + def _get_queue_top(self): + current_pos = self._file.tell() + + self._file.seek(START_OFFSET - 4, 0) # Start at beginning of file + pos = struct.unpack(HEADER_STRUCT[1], self._file.read(4))[0] + + self._file.seek(current_pos, 0) + return pos + + def _set_queue_top(self, top): + current_pos = self._file.tell() + + self._file.seek(START_OFFSET - 4, 0) # Start at beginning of file + self._file.write(struct.pack(HEADER_STRUCT[1], top)) + self._file.flush() # Probably not necessary since buffering=0 + os.fsync(self._file.fileno()) + + self._file.seek(current_pos, 0) + + def _peek(self, block, timeout, items, partial=False): + """ + Returns a certain amount of items from the queue. If items is greater + than one, a list is returned. + """ + def read_data(): + length = struct.unpack(LENGTH_STRUCT, self._file.read(4))[0] + data = self._file.read(length) + return self.loads(data) + + _LOGGER.debug("Peeking %s items", items) + + # Ignore requests for zero items + if items == 0: + _LOGGER.debug("Returning empty list") + return [], self._file.tell() + + if block: + if timeout is not None: + target = time.time() + timeout + while self._length < items: + if self._put_event.wait(timeout) is False: + # Nothing was added to the queue and timeout expired + # This will never happen if timeout is None + raise queue.Empty + self._put_event.clear() + + # Something was added to the queue + # Update timeout, if necessary + if timeout is not None: + timeout = target - time.time() + + elif not partial and self._length < items: + raise queue.Empty + + with self._file_lock: + self._file.seek(self._get_queue_top(), 0) # Beginning of data + total_items = self._length if items > self._length else items + data = [read_data() for i in range(total_items)] + queue_top = self._file.tell() + + if items == 1: + if len(data) == 0: + _LOGGER.debug("No items to peek at so returning None") + return None, queue_top + else: + _LOGGER.debug("Returning data from peek") + return data[0], queue_top + else: + _LOGGER.debug("Returning data from peek") + return data, queue_top + + def qsize(self): + """ + Provides compatibility with stdlib Queue objects. + + Return the approximate size of the queue. Note, qsize() > 0 doesn't + guarantee that a subsequent get() will not block, nor will qsize() < + maxsize guarantee that put() will not block. + """ + return self._length + + def empty(self): + """ + Provides compatibility with stdlib Queue objects. + + Return True if the queue is empty, False otherwise. If empty() returns + True it doesn't guarantee that a subsequent call to put() will not + block. Similarly, if empty() returns False it doesn't guarantee that a + subsequent call to get() will not block. + """ + return self._length == 0 + + def full(self): + """ + Provides compatibility with stdlib Queue objects. + + Return True if the queue is full, False otherwise. If full() returns + True it doesn't guarantee that a subsequent call to get() will not + block. Similarly, if full() returns False it doesn't guarantee that a + subsequent call to put() will not block. + """ + return self.maxsize > 0 and self._length >= self.maxsize + + def put(self, items, block=True, timeout=None): + """ + Provides compatibility with stdlib Queue objects. + When this function returns, all items are guaranteed to be persisted + into the file and the underlying storage. + + items: single object, or a list of objects + + Put item into the queue. If optional args block is true and timeout is + None (the default), block if necessary until a free slot is available. + If timeout is a positive number, it blocks at most timeout seconds and + raises the Full exception if no free slot was available within that + time. Otherwise (block is false), put an item on the queue if a free + slot is immediately available, else raise the Full exception (timeout + is ignored in that case). + """ + def write_data(item): + data = self.dumps(item) + self._file.write(struct.pack(LENGTH_STRUCT, len(data))) + self._file.write(data) + self._file.flush() # Probably not necessary since buffering=0 + os.fsync(self._file.fileno()) + + if not isinstance(items, list): + items = [items] + + _LOGGER.debug("Putting %s items", len(items)) + + # Ignore requests for adding zero items + if len(items) == 0: + _LOGGER.debug("Putting zero items, ignoring request") + return + + with self._put_lock: + if self.maxsize > 0: + if block: + if timeout is not None: + target = time.time() + timeout + while self._length + len(items) > self.maxsize: + if self._get_event.wait(timeout) is False: + # Nothing was removed from the queue and timeout expired + # This will never happen if timeout is None + raise queue.Full + self._get_event.clear() + + # Something was removed from the queue + # Update timeout, if necessary + if timeout is not None: + timeout = target - time.time() + else: + if self._length + len(items) > self.maxsize: + raise queue.Full + + with self._file_lock: + self._file.seek(0, 2) # Go to end of file + + for i in items: + write_data(i) + + self._update_length(self._length + len(items)) + self._unfinished_tasks += len(items) + + self._put_event.set() + _LOGGER.debug("Done putting data") + + def put_nowait(self, item): + """ + Provides compatibility with stdlib Queue objects. + + Equivalent to put(item, False). + """ + self.put(item, block=False) + + def get(self, block=True, timeout=None, items=1): + """ + Provides compatibility with stdlib Queue objects. + items: number of how many items are returned. If items is greater than + one, a list is returned. + + Remove and return an item from the queue. If optional args block is + true and timeout is None (the default), block if necessary until an + item is available. If timeout is a positive number, it blocks at most + timeout seconds and raises the Empty exception if no item was available + within that time. Otherwise (block is false), return an item if one is + immediately available, else raise the Empty exception (timeout is + ignored in that case). + """ + _LOGGER.debug("Getting %s items", items) + + # Ignore requests for zero items + if items == 0: + _LOGGER.debug("Returning empty list") + return [] + + with self._get_lock: + data, queue_top = self._peek(block, timeout, items) + + with self._file_lock: + self._set_queue_top(queue_top) + + if isinstance(data, list): + if len(data) > 0: + self._update_length(self._length - len(data)) + elif data is not None: + self._update_length(self._length - 1) + + self._get_event.set() + _LOGGER.debug("Returning data from get") + return data + + def get_nowait(self): + """ + Provides compatibility with stdlib Queue objects. + + Equivalent to get(False). + + Two methods are offered to support tracking whether enqueued tasks + have been fully processed by daemon consumer threads. + """ + return self.get(block=False) + + def task_done(self, items=1): + """ + Provides compatibility with stdlib Queue objects. + Taken from the Python 3.5 stdlib: lib/python3.5/queue.py + + Indicate that a formerly enqueued task is complete. Used by queue + consumer threads. For each get() used to fetch a task, a subsequent + call to task_done() tells the queue that the processing on the task is + complete. + + If a join() is currently blocking, it will resume when all items have + been processed (meaning that a task_done() call was received for every + item that had been put() into the queue). + + Raises a ValueError if called more times than there were items placed in the queue. + """ + with self._all_tasks_done: + unfinished = self._unfinished_tasks - items + if unfinished < 0: + raise ValueError('task_done() called too many times') + if unfinished == 0: + self._all_tasks_done.notify_all() + self._unfinished_tasks = unfinished + + def join(self): + """ + Provides compatibility with stdlib Queue objects. + Taken from the Python 3.5 stdlib: lib/python3.5/queue.py + + Blocks until all items in the queue have been gotten and processed. + + The count of unfinished tasks goes up whenever an item is added to the + queue. The count goes down whenever a consumer thread calls task_done() + to indicate that the item was retrieved and all work on it is complete. + When the count of unfinished tasks drops to zero, join() unblocks. + """ + with self._all_tasks_done: + while self._unfinished_tasks: + self._all_tasks_done.wait() + + def peek(self, block=False, timeout=None, items=1): + """ + Peeks into the queue and returns items without removing them. + """ + with self._get_lock: + return self._peek(block, timeout, items, partial=True)[0] + + def clear(self): + """ + Removes all elements from queue, by truncating the file and reloading. + """ + _LOGGER.debug("Clearing the queue") + with self._file_lock, self._get_lock: + self._file.close() + self._file = self._open_file(mode='w+b') + self._length = 0 + _LOGGER.debug("The queue has been cleared") + + def copy(self, new_filename): + """ + Copies a queue to a new queue by duplicating the underlying file. + + new_filename: must be a full path to the new file. + """ + shutil.copy2(self.filename, new_filename) + return PersistentQueue(maxsize=self.maxsize, + filename=new_filename, + dumps=self.dumps, + loads=self.loads, + flush_limit=self.flush_limit) + + def flush(self): + """ + Removes elements that have been deleted or gotten from the queue. + """ + _LOGGER.debug("Flushing the queue") + + with self._file_lock: + pos = self._get_queue_top() + + if pos < self.flush_limit: + # Ignore if the file isn't big enough -- it's not worth it + _LOGGER.debug("Ignoring flush because we haven't met the limit") + return + + # Make a new file + random = str(uuid.uuid4()).replace('-', '') + temp_filename = self.filename + '-' + random + new_file = open(temp_filename, mode='w+b', buffering=0) + + # From this point on, the file can not change + with self._file_lock, self._get_lock: + # Make sure everything is to disk + self._file.flush() + os.fsync(self._file.fileno()) + + start = self._get_queue_top() # Get it again in case it changed + self._file.seek(0, 2) # Go to end of file + end = self._file.tell() + + _LOGGER.debug("Writing data to new file") + # Copy over meta data + new_file.write(struct.pack(HEADER_STRUCT, + self._length, + START_OFFSET)) + + # Copy over data + # Do it in chunks so we aren't loading tons of data into memory + self._file.seek(start, 0) + bytes_read = 0 + chunk_size = 4096 + while bytes_read < end - start: + bytes_read += chunk_size + new_file.write(self._file.read(bytes_read)) + + new_file.flush() # Probably not necessary since buffering=0 + os.fsync(new_file.fileno()) + new_file.close() + self._file.close() + + # So far everything above this point has been safe. If something + # crashed, the data would still be preserved. Now we are entering + # the danger zone. + + _LOGGER.debug("Replacing old file with new file") + os.remove(self.filename) + os.rename(temp_filename, self.filename) + self._file = self._open_file() + + _LOGGER.debug("Finished flushing the queue") + + def delete(self, items=1): + """ + Removes items from queue. + + items: number of how many items will be deleted + """ + def read_length(): + length = struct.unpack(LENGTH_STRUCT, self._file.read(4))[0] + self._file.seek(length, 1) + + _LOGGER.debug("Deleting %s items", items) + + # Ignore requests for zero items + if items == 0: + _LOGGER.debug("Ignoring request to delete") + return + + with self._file_lock, self._get_lock: + self._file.seek(self._get_queue_top(), 0) # Beginning of data + total_items = self._length if items > self._length else items + + for _ in range(total_items): + read_length() + + self._set_queue_top(self._file.tell()) + self._update_length(self._length - total_items) + + _LOGGER.debug("Done deleting data") + + def __len__(self): + """ + Get size of queue. + """ + return self._length diff --git a/requirements_test.txt b/requirements_test.txt deleted file mode 100644 index d872458..0000000 --- a/requirements_test.txt +++ /dev/null @@ -1,6 +0,0 @@ -bson==0.4.3 -coveralls>=1.1 -dill==0.2.5 -flake8==3.0.4 -pytest==3.0.1 -pytest-cov==2.3.1 diff --git a/setup.cfg b/setup.cfg new file mode 100644 index 0000000..790df1e --- /dev/null +++ b/setup.cfg @@ -0,0 +1,18 @@ +[flake8] +max-line-length = 140 +max-complexity = 25 +ignore = E251,C901,W503 +addons = file,open,basestring,xrange,unicode,long,cmp + +[tool:pytest] +testpaths = tests +addopts = --capture=no --color=yes + +[coverage:run] +branch = False + +[coverage:report] +show_missing = True +exclude_lines = + pragma: no cover + raise NotImplementedError() diff --git a/tests/test_persistent_queue.py b/tests/test_persistent_queue.py index 1ebfc97..a4f2eec 100644 --- a/tests/test_persistent_queue.py +++ b/tests/test_persistent_queue.py @@ -5,144 +5,180 @@ import unittest import uuid +try: + import queue +except ImportError: + import Queue as queue + from persistent_queue import PersistentQueue class TestPersistentQueue(unittest.TestCase): def setUp(self): random = str(uuid.uuid4()).replace('-', '') - self.filename = '{}_{}'.format(self.id(), random) - self.queue = PersistentQueue(self.filename) + self.filename = '{}_{}.queue'.format(self.id(), random) + self.queue = PersistentQueue(filename=self.filename) def tearDown(self): os.remove(self.filename) - def test_count(self): + def test_simple(self): + random = str(uuid.uuid4()).replace('-', '') + filename = '{}_{}.queue'.format(self.id(), random) + + with self.assertRaises(TypeError): + q = PersistentQueue() + + q = PersistentQueue(filename=filename) + self.assertEqual(q.maxsize, 0) + + q = PersistentQueue(maxsize=-42, filename=filename) + self.assertEqual(q.maxsize, 0) + + os.remove(filename) + + def test_qsize(self): self.assertEqual(len(self.queue), 0) - self.assertEqual(self.queue.count(), 0) + self.assertEqual(self.queue.qsize(), 0) + + self.assertTrue(self.queue.empty()) + self.assertFalse(self.queue.full()) - self.queue.push(1) + self.queue.put(1) self.assertEqual(len(self.queue), 1) - self.assertEqual(self.queue.count(), 1) + self.assertEqual(self.queue.qsize(), 1) - self.queue.push(2) + self.queue.put(2) self.assertEqual(len(self.queue), 2) - self.assertEqual(self.queue.count(), 2) + self.assertEqual(self.queue.qsize(), 2) for i in range(100 + 1): - self.queue.push(i) + self.queue.put(i) self.assertEqual(len(self.queue), 103) - self.assertEqual(self.queue.count(), 103) + self.assertEqual(self.queue.qsize(), 103) - def test_clear(self): - self.queue.push(5) - self.queue.push(50) + self.assertFalse(self.queue.empty()) + self.assertFalse(self.queue.full()) - self.assertEqual(self.queue.peek(2), [5, 50]) - self.assertEqual(len(self.queue), 2) - self.queue.clear() - self.assertEqual(len(self.queue), 0) + def test_put(self): + self.queue.put(5) + self.assertEqual(self.queue.peek(items=1), 5) - def test_push(self): - self.queue.push(5) - self.assertEqual(self.queue.peek(1), 5) + self.queue.put_nowait(5) + self.assertEqual(self.queue.get(), 5) - self.queue.push([10, 15, 20]) - self.assertEqual(self.queue.peek(4), [5, 10, 15, 20]) + self.queue.put([10, 15, 20]) + self.assertEqual(self.queue.peek(items=4), [5, 10, 15, 20]) data = {"a": 1, "b": 2, "c": [1, 2, 3]} - self.queue.push(data) - self.assertEqual(self.queue.peek(5), [5, 10, 15, 20, data]) + self.queue.put(data) + self.assertEqual(self.queue.peek(items=5), [5, 10, 15, 20, data]) + + self.queue.put([]) + self.assertEqual(self.queue.peek(items=5), [5, 10, 15, 20, data]) - self.queue.push([]) - self.assertEqual(self.queue.peek(5), [5, 10, 15, 20, data]) + self.queue.maxsize = 4 + with self.assertRaises(queue.Full): + self.queue.put('full', timeout=1) - def test_pop(self): - self.queue.push('a') - self.queue.push('b') + with self.assertRaises(queue.Full): + self.queue.put('full', block=False) + + def test_get(self): + self.queue.put('a') + self.queue.put('b') self.assertEqual(len(self.queue), 2) - self.assertEqual(self.queue.pop(), 'a') + self.assertEqual(self.queue.get(), 'a') self.assertEqual(len(self.queue), 1) - self.assertEqual(self.queue.pop(1), 'b') + self.assertEqual(self.queue.get(items=1), 'b') self.assertEqual(len(self.queue), 0) - self.queue.push('a') - self.queue.push('b') - self.queue.push('c') - self.queue.push('d') + self.queue.put('a') + self.queue.put('b') + self.queue.put('c') + self.queue.put('d') self.assertEqual(len(self.queue), 4) - self.assertEqual(self.queue.pop(3), ['a', 'b', 'c']) + self.assertEqual(self.queue.get(items=3), ['a', 'b', 'c']) self.assertEqual(len(self.queue), 1) - self.assertEqual(self.queue.pop(100), ['d']) - self.assertEqual(len(self.queue), 0) - - self.assertEqual(self.queue.pop(100), []) - self.assertEqual(self.queue.pop(), None) - - self.queue.push('d') - self.assertEqual(self.queue.pop(0), []) + with self.assertRaises(queue.Empty): + self.assertEqual(self.queue.get(block=False, items=100), ['d']) self.assertEqual(len(self.queue), 1) - def test_pop_blocking(self): + self.queue.put('d') + self.assertEqual(self.queue.get(items=0), []) + self.assertEqual(len(self.queue), 2) + + def test_get_blocking(self): done = [False] + def func(): time.sleep(1) done[0] = True - self.queue.push(5) + self.queue.put(5) t = threading.Thread(target=func) t.start() self.assertFalse(done[0]) - data = self.queue.pop(blocking=True) + data = self.queue.get() self.assertTrue(done[0]) self.assertEqual(data, 5) self.assertEqual(len(self.queue), 0) - def test_pop_no_values(self): - self.assertEqual(self.queue.pop(5), []) - self.assertEqual(self.queue.pop(), None) + with self.assertRaises(queue.Empty): + self.queue.get(timeout=1) + + def test_get_non_blocking_no_values(self): + with self.assertRaises(queue.Empty): + self.assertEqual(self.queue.get(block=False, items=5), []) + + with self.assertRaises(queue.Empty): + self.queue.get(block=False) + + with self.assertRaises(queue.Empty): + self.queue.get_nowait() def test_peek(self): - self.queue.push(1) - self.queue.push(2) - self.queue.push("test") + self.queue.put(1) + self.queue.put(2) + self.queue.put("test") self.assertEqual(self.queue.peek(), 1) - self.assertEqual(self.queue.peek(1), 1) - self.assertEqual(self.queue.peek(2), [1, 2]) - self.assertEqual(self.queue.peek(3), [1, 2, "test"]) + self.assertEqual(self.queue.peek(items=1), 1) + self.assertEqual(self.queue.peek(items=2), [1, 2]) + self.assertEqual(self.queue.peek(items=3), [1, 2, "test"]) - self.assertEqual(self.queue.peek(100), [1, 2, "test"]) + self.assertEqual(self.queue.peek(items=100), [1, 2, "test"]) self.queue.clear() - self.queue.push(1) + self.queue.put(1) self.assertEqual(len(self.queue), 1) self.assertEqual(self.queue.peek(), 1) - self.assertEqual(self.queue.peek(1), 1) - self.assertEqual(self.queue.peek(2), [1]) + self.assertEqual(self.queue.peek(items=1), 1) + self.assertEqual(self.queue.peek(items=2), [1]) - self.assertEqual(self.queue.peek(0), []) + self.assertEqual(self.queue.peek(items=0), []) def test_peek_blocking(self): done = [False] + def func(): time.sleep(1) done[0] = True - self.queue.push(5) + self.queue.put(5) t = threading.Thread(target=func) t.start() self.assertFalse(done[0]) - data = self.queue.peek(blocking=True) + data = self.queue.peek(block=True) self.assertTrue(done[0]) self.assertEqual(data, 5) self.assertEqual(len(self.queue), 1) @@ -154,99 +190,65 @@ def test_peek_blocking_list(self): def func(): for i in range(5): time.sleep(.1) - self.queue.push(i) + self.queue.put(i) self.assertFalse(done_peeking[0]) done_pushing[0] = True t = threading.Thread(target=func) t.start() - data = self.queue.peek(5, blocking=True) + data = self.queue.peek(items=5, block=True) done_peeking[0] = True self.assertTrue(done_pushing[0]) self.assertEqual(data, [0, 1, 2, 3, 4]) self.assertEqual(len(self.queue), 5) def test_peek_no_values(self): - self.assertEqual(self.queue.peek(5), []) + self.assertEqual(self.queue.peek(items=5), []) self.assertEqual(self.queue.peek(), None) - def test_big_file_1(self): - data = {"a": list(range(1000))} - - for i in range(2000): - self.queue.push(data) - - self.assertEqual(len(self.queue), 2000) - - for i in range(1995): - self.assertEqual(self.queue.pop(), data) - self.queue.flush() - - self.assertEqual(len(self.queue), 5) - - def test_big_file_2(self): - data = {"a": list(range(1000))} - - for i in range(2000): - self.queue.push(data) - - self.assertEqual(self.queue.pop(1995), [data for i in range(1995)]) - self.queue.flush() - self.assertEqual(len(self.queue), 5) + def test_clear(self): + self.queue.put(5) + self.queue.put(50) - import time - time.sleep(1) + self.assertEqual(self.queue.peek(items=2), [5, 50]) + self.assertEqual(len(self.queue), 2) + self.queue.clear() + self.assertEqual(len(self.queue), 0) def test_copy(self): new_queue_name = 'another_queue' - self.queue.push([5, 4, 3, 2, 1]) + self.queue.put([5, 4, 3, 2, 1]) self.assertEqual(len(self.queue), 5) - self.assertEqual(self.queue.pop(), 5) + self.assertEqual(self.queue.get(), 5) new_queue = self.queue.copy(new_queue_name) self.assertEqual(len(self.queue), len(new_queue)) - self.assertEqual(self.queue.pop(), new_queue.pop()) - self.assertEqual(self.queue.pop(), new_queue.pop()) - self.assertEqual(self.queue.pop(), new_queue.pop()) - self.assertEqual(self.queue.pop(), new_queue.pop()) + self.assertEqual(self.queue.get(), new_queue.get()) + self.assertEqual(self.queue.get(), new_queue.get()) + self.assertEqual(self.queue.get(), new_queue.get()) + self.assertEqual(self.queue.get(), new_queue.get()) os.remove(new_queue_name) - def test_usage(self): - self.queue.push(1) - self.queue.push(2) - self.queue.push(3) - self.queue.push(['a', 'b', 'c']) - - self.assertEqual(self.queue.peek(), 1) - self.assertEqual(self.queue.peek(4), [1, 2, 3, 'a']) - self.assertEqual(len(self.queue), 6) - - self.queue.push('foobar') - - self.assertEqual(self.queue.pop(), 1) - self.assertEqual(len(self.queue), 6) - self.assertEqual(self.queue.pop(6), [2, 3, 'a', 'b', 'c', 'foobar']) - def test_delete(self): - self.queue.push(2) - self.queue.push(3) - self.queue.push(7) - self.queue.push(11) + self.queue.put(2) + self.queue.put(3) + self.queue.put(7) + self.queue.put(11) self.assertEqual(len(self.queue), 4) self.queue.delete(2) self.assertEqual(len(self.queue), 2) - self.assertEqual(self.queue.peek(2), [7, 11]) - self.assertEqual(self.queue.pop(2), [7, 11]) + self.assertEqual(self.queue.peek(items=2), [7, 11]) + self.assertEqual(self.queue.get(items=2), [7, 11]) - self.queue.push(2) + self.queue.put(2) self.queue.delete(1000) self.assertEqual(len(self.queue), 0) - self.queue.push(2) + self.queue.put(2) self.queue.delete(0) self.assertEqual(len(self.queue), 1) @@ -254,20 +256,69 @@ def test_delete_no_values(self): self.queue.delete() self.queue.delete(100) + def test_big_file_1(self): + data = {"a": list(range(500))} + + for i in range(1000): + self.queue.put(data) + + self.assertEqual(len(self.queue), 1000) + + for i in range(995): + self.assertEqual(self.queue.get(), data) + self.queue.flush() + + self.assertEqual(len(self.queue), 5) + + def test_big_file_2(self): + data = {"a": list(range(500))} + + for i in range(1000): + self.queue.put(data) + + self.assertEqual(self.queue.get(items=995), [data for i in range(995)]) + self.queue.flush() + self.assertEqual(len(self.queue), 5) + + import time + time.sleep(1) + + def test_usage(self): + self.queue.put(1) + self.queue.put(2) + self.queue.put(3) + self.queue.put(['a', 'b', 'c']) + + self.assertEqual(self.queue.peek(), 1) + self.assertEqual(self.queue.peek(items=4), [1, 2, 3, 'a']) + self.assertEqual(len(self.queue), 6) + + self.queue.put('foobar') + + self.assertEqual(self.queue.get(), 1) + self.assertEqual(len(self.queue), 6) + self.assertEqual(self.queue.get(items=6), [2, 3, 'a', 'b', 'c', 'foobar']) + def test_threads(self): def random_stuff(): for i in range(100): random_number = random.randint(0, 1000) if random_number % 3 == 0: - self.queue.peek(random_number % 5) + try: + self.queue.peek(block=False, items=(random_number % 5)) + except queue.Empty: + pass elif random_number % 2 == 0: - self.queue.pop(random_number % 5) + try: + self.queue.get(block=False, items=(random_number % 5)) + except queue.Empty: + pass else: for i in range(random_number % 10): - self.queue.push({"test": [1, 2, 3], "foo": "bar", "1": random_number}) + self.queue.put({"test": [1, 2, 3], "foo": "bar", "1": random_number}) - threads = [threading.Thread(target=random_stuff) for _ in range(50)] + threads = [threading.Thread(target=random_stuff) for _ in range(10)] for t in threads: t.start() @@ -277,7 +328,27 @@ def random_stuff(): # Remove everything that is left so we make sure it is serializable for _ in range(len(self.queue)): - self.queue.pop() + self.queue.get() + + def test_join_on_task_done(self): + def worker(): + while True: + try: + self.queue.get(block=False) + self.queue.task_done() + except queue.Empty: + with self.assertRaises(ValueError): + # called too many times + self.queue.task_done() + return + + self.queue.put(list(range(10))) + + t = threading.Thread(target=worker) + t.start() + + self.queue.join() + self.assertTrue(self.queue.empty()) class TestPersistentQueueWithDill(TestPersistentQueue): @@ -286,7 +357,8 @@ def setUp(self): random = str(uuid.uuid4()).replace('-', '') self.filename = '{}_{}'.format(self.id(), random) - self.queue = PersistentQueue(self.filename, loads=dill.loads, + self.queue = PersistentQueue(filename=self.filename, + loads=dill.loads, dumps=dill.dumps) @@ -296,7 +368,8 @@ def setUp(self): random = str(uuid.uuid4()).replace('-', '') self.filename = '{}_{}'.format(self.id(), random) - self.queue = PersistentQueue(self.filename, loads=bson.loads, + self.queue = PersistentQueue(filename=self.filename, + loads=bson.loads, dumps=bson.dumps) def tearDown(self): @@ -306,12 +379,12 @@ def test_big_file_1(self): data = {"a": list(range(1000))} for i in range(2000): - self.queue.push(data) + self.queue.put(data) self.assertEqual(len(self.queue), 2000) for i in range(1995): - self.assertEqual(self.queue.pop(), data) + self.assertEqual(self.queue.get(), data) self.queue.flush() self.assertEqual(len(self.queue), 5) @@ -320,9 +393,9 @@ def test_big_file_2(self): data = {"a": list(range(1000))} for i in range(2000): - self.queue.push(data) + self.queue.put(data) - self.assertEqual(self.queue.pop(1995), [data for i in range(1995)]) + self.assertEqual(self.queue.get(items=1995), [data for i in range(1995)]) self.queue.flush() self.assertEqual(len(self.queue), 5) diff --git a/tox.ini b/tox.ini new file mode 100644 index 0000000..e210028 --- /dev/null +++ b/tox.ini @@ -0,0 +1,21 @@ +[tox] +envlist = py27, py34, py35, py36, lint +skipsdist = True +toxworkdir = {env:TOX_WORK_DIR:.tox} + +[testenv] +deps = + coveralls>=1.1 + bson==0.4.3 + dill==0.2.5 + pytest==3.0.1 + pytest-cov==2.3.1 +passenv = CODECOV_TOKEN CI CI_* TRAVIS TRAVIS_* +setenv = HOME = {envtmpdir} +commands = py.test -v --cov persistent_queue {posargs} + +[testenv:lint] +deps = + flake8==3.0.4 +commands = + flake8 --count persistent_queue tests