Repository: peter-wangxu/persist-queue

Issue:
qsize() reports negative value
Using the SQLiteAckQueue multithreaded and I'm trying to get a count on the queue size using qsize(), but sometimes I'll see negative numbers. I think that the issue is the `_unack_cache`  is not included in the total queue size, which I think it probably should be as it hasn't yet received an `ack/task_done`  So in a mutlithreaded situation where you have 10 get calls, you're going to report a queue size 10 less than what actually exists, since those jobs haven't been completed.

This also causes an ever changing qsize as things can be `get`/`nack` regularly, so the size keeps going up and down as the state changes.  I think it should only change the size when a new item is added `put` or an item is removed  `ack/ack_failed/task_done`.  I don't think a `get` or `nack` should change the reported size.
Why is  _MAX_ACKED_LENGTH Hardcoded?
In this code, _MAX_ACKED_LENGTH is hardcoded:
https://github.com/peter-wangxu/persist-queue/blob/5ce10afcfcd81afd46a9eae32ba085d8ae900240/persistqueue/sqlackqueue.py#L35

Why is it hardcoded to 1000? I need to remove about 10 to 100 thousand acked entries each time I run this, is there any reason I shouldn't change _MAX_ACKED_LENGTH to 100_000 and then run that? 

Discussion:

Patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index fdbf99b..629c232 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -3,7 +3,7 @@ jobs:
   py27:
     docker:
       # Primary container image where all steps run.
-      - image: circleci/python:2.7.15
+      - image: circleci/python:2.7.17
     environment:
       - TOXENV: py27
     steps: &common_steps
@@ -34,7 +34,7 @@ jobs:
   py34:
     docker:
       # Primary container image where all steps run.
-      - image: circleci/python:3.4.7
+      - image: circleci/python:3.4.10
     environment:
       - TOXENV: py34
     steps: *common_steps
@@ -42,16 +42,15 @@ jobs:
   py35:
     docker:
       # Primary container image where all steps run.
-      - image: circleci/python:3.5.5
+      - image: circleci/python:3.5.9
     environment:
       - TOXENV: py35
-
     steps: *common_steps
 
   py36:
     docker:
       # Primary container image where all steps run.
-      - image: circleci/python:3.6.5
+      - image: circleci/python:3.6.10
     environment:
       - TOXENV: py36
     steps: *common_steps
@@ -59,16 +58,23 @@ jobs:
   py37:
     docker:
       # Primary container image where all steps run.
-      - image: circleci/python:3.7.0
+      - image: circleci/python:3.7.7
     environment:
       - TOXENV: py37
     steps: *common_steps
 
+  py38:
+    docker:
+      # Primary container image where all steps run.
+      - image: circleci/python:3.8.2
+    environment:
+      - TOXENV: py38
+    steps: *common_steps
 
   pep8:
     docker:
       # Primary container image where all steps run.
-      - image: circleci/python:3.5.4
+      - image: circleci/python:3.5.9
     environment:
       - TOXENV: pep8
     steps: *common_steps
@@ -77,7 +83,7 @@ jobs:
   cover:
     docker:
       # Primary container image where all steps run.
-      - image: circleci/python:3.5.4
+      - image: circleci/python:3.5.9
     environment:
       - TOXENV: cover
     steps: *common_steps
@@ -92,4 +98,5 @@ workflows:
       - py35
       - py36
       - py37
+      - py38
       - cover
diff --git a/.gitignore b/.gitignore
index 4f4b433..2a72270 100644
--- a/.gitignore
+++ b/.gitignore
@@ -90,3 +90,6 @@ ENV/
 
 # IDE specific folders
 .idea/
+
+# MacOS
+.DS_Store
diff --git a/README.rst b/README.rst
index 631b5a2..cbdfc86 100644
--- a/README.rst
+++ b/README.rst
@@ -197,12 +197,16 @@ Example usage of SQLite3 based ``SQLiteAckQueue``/``UniqueAckQ``
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 The core functions:
 
-- ``get``: get from queue and mark item as unack
-- ``ack``: mark item as acked
-- ``nack``: there might be something wrong with current consumer, so mark item as ready and new consumer will get it
-- ``ack_failed``: there might be something wrong during process, so just mark item as failed.
-- ``clear_acked_data``: perform a sql delete agaist sqlite, it remove the latest 1000 items whose status is ``AckStatus.acked`` (note: this does not shrink the file size on disk)
+- ``put``: add item to the queue. Returns ``id``
+- ``get``: get item from queue and mark as unack.  Returns ``item``, Optional paramaters (``block``, ``timeout``, ``id``, ``next_in_order``, ``raw``)
+- ``update``: update an item. Returns ``id``, Paramaters (``item``), Optional parameter if item not in raw format (``id``)
+- ``ack``: mark item as acked. Returns ``id``, Parameters (``item`` or ``id``)
+- ``nack``: there might be something wrong with current consumer, so mark item as ready and new consumer will get it.  Returns ``id``, Parameters (``item`` or ``id``)
+- ``ack_failed``: there might be something wrong during process, so just mark item as failed. Returns ``id``, Parameters (``item`` or ``id``)
+- ``clear_acked_data``: perform a sql delete agaist sqlite. It removes 1000 items, while keeping 1000 of the most recent, whose status is ``AckStatus.acked`` (note: this does not shrink the file size on disk) Optional paramters (``max_delete``, ``keep_latest``, ``clear_ack_failed``)
 - ``shrink_disk_usage`` perform a ``VACUUM`` against the sqlite, and rebuild the database file, this usually takes long time and frees a lot of disk space after ``clear_acked_data``
+- ``queue``: returns the database contents as a Python List[Dict]
+- ``active_size``: The active size changes when an item is added (put) and completed (ack/ack_failed) unlike ``qsize`` which changes when an item is pulled (get) or returned (nack).
 
 .. code-block:: python
 
@@ -215,6 +219,40 @@ The core functions:
    >>> ackq.nack(item) # Else mark item as `nack` so that it can be proceeded again by any worker
    >>> ackq.ack_failed(item) # Or else mark item as `ack_failed` to discard this item
 
+Paramaters:
+
+- ``clear_acked_data``
+    - ``max_delete`` (defaults to 1000): This is the LIMIT.  How many items to delete.
+    - ``keep_latest`` (defaults to 1000): This is the OFFSET.  How many recent items to keep.
+    - ``clear_ack_failed`` (defaults to False): Clears the ``AckStatus.ack_failed`` in addition to the ``AckStatus.ack``.
+    
+- ``get``
+    - ``raw`` (defaults to False): Returns the metadata along with the record, which includes the id (``pqid``) and timestamp.  On the SQLiteAckQueue, the raw results can be ack, nack, ack_failed similar to the normal return.  
+    -  ``id`` (defaults to None): Accepts an `id` or a raw item containing ``pqid``.  Will select the item based on the row id.
+    -  ``next_in_order`` (defaults to False): Requires the ``id`` attribute.  This option tells the SQLiteAckQueue/UniqueAckQ to get the next item based on  ``id``, not the first available.  This allows the user to get, nack, get, nack and progress down the queue, instead of continuing to get the same nack'd item over again.
+    
+``raw`` example:
+
+.. code-block:: python
+
+   >>> q.put('val1')
+   >>> d = q.get(raw=True)
+   >>> print(d)
+   >>> {'pqid': 1, 'data': 'val1', 'timestamp': 1616719225.012912}
+   >>> q.ack(d)
+
+``next_in_order`` example:
+
+.. code-block:: python
+
+   >>> q.put("val1")
+   >>> q.put("val2")
+   >>> q.put("val3")
+   >>> item = q.get()
+   >>> id = q.nack(item)
+   >>> item = q.get(id=id, next_in_order=True)
+   >>> print(item)
+   >>> val2
 
 
 Note:
diff --git a/appveyor.yml b/appveyor.yml
index b3ae4a2..32fc5ef 100644
--- a/appveyor.yml
+++ b/appveyor.yml
@@ -21,6 +21,9 @@ environment:
   - TOXENV: "py37"
     PYTHON: "C:\\Python37-x64"
     DISTUTILS_USE_SDK: "1"
+  - TOXENV: "py38"
+    PYTHON: "C:\\Python38-x64"
+    DISTUTILS_USE_SDK: "1"
   - TOXENV: "cover"
     PYTHON: "C:\\Python36-x64"
     DISTUTILS_USE_SDK: "1"
diff --git a/benchmark/run_benchmark.py b/benchmark/run_benchmark.py
index 1d8a48d..9ee228a 100644
--- a/benchmark/run_benchmark.py
+++ b/benchmark/run_benchmark.py
@@ -28,6 +28,7 @@ class FileQueueBench(object):
 
     def __init__(self, prefix=None):
         self.path = prefix
+
     @time_it
     def benchmark_file_write(self):
         """Writing <BENCHMARK_COUNT> items."""
diff --git a/persistqueue/__init__.py b/persistqueue/__init__.py
index 732ccec..10bb167 100644
--- a/persistqueue/__init__.py
+++ b/persistqueue/__init__.py
@@ -1,7 +1,7 @@
 # coding=utf-8
 __author__ = 'Peter Wang'
 __license__ = 'BSD'
-__version__ = '0.5.1'
+__version__ = '0.6.0'
 
 from .exceptions import Empty, Full  # noqa
 from .queue import Queue  # noqa
diff --git a/persistqueue/sqlackqueue.py b/persistqueue/sqlackqueue.py
index ede6385..ea217e7 100644
--- a/persistqueue/sqlackqueue.py
+++ b/persistqueue/sqlackqueue.py
@@ -32,24 +32,38 @@ class SQLiteAckQueue(sqlbase.SQLiteBase):
 
     _TABLE_NAME = 'ack_queue'
     _KEY_COLUMN = '_id'  # the name of the key column, used in DB CRUD
-    _MAX_ACKED_LENGTH = 1000
+    _MAX_ACKED_LENGTH = 1000  # depreciated
     # SQL to create a table
-    _SQL_CREATE = ('CREATE TABLE IF NOT EXISTS {table_name} ('
-                   '{key_column} INTEGER PRIMARY KEY AUTOINCREMENT, '
-                   'data BLOB, timestamp FLOAT, status INTEGER)')
+    _SQL_CREATE = (
+        'CREATE TABLE IF NOT EXISTS {table_name} ('
+        '{key_column} INTEGER PRIMARY KEY AUTOINCREMENT, '
+        'data BLOB, timestamp FLOAT, status INTEGER)'
+    )
     # SQL to insert a record
-    _SQL_INSERT = 'INSERT INTO {table_name} (data, timestamp, status)' \
-                  ' VALUES (?, ?, %s)' % AckStatus.inited
+    _SQL_INSERT = (
+        'INSERT INTO {table_name} (data, timestamp, status)'
+        ' VALUES (?, ?, %s)' % AckStatus.inited
+    )
     # SQL to select a record
-    _SQL_SELECT = ('SELECT {key_column}, data, status FROM {table_name} '
-                   'WHERE status < %s '
-                   'ORDER BY {key_column} ASC LIMIT 1' % AckStatus.unack)
-    _SQL_MARK_ACK_UPDATE = 'UPDATE {table_name} SET status = ?' \
-                           ' WHERE {key_column} = ?'
-    _SQL_SELECT_WHERE = 'SELECT {key_column}, data FROM {table_name}' \
-                        ' WHERE status < %s AND' \
-                        ' {column} {op} ? ORDER BY {key_column} ASC' \
-                        ' LIMIT 1 ' % AckStatus.unack
+    _SQL_SELECT_ID = (
+        'SELECT {key_column}, data, timestamp, status FROM {table_name} WHERE'
+        ' {key_column} = {rowid}'
+    )
+    _SQL_SELECT = (
+        'SELECT {key_column}, data, timestamp, status FROM {table_name} '
+        'WHERE {key_column} > {rowid} AND status < %s '
+        'ORDER BY {key_column} ASC LIMIT 1' % AckStatus.unack
+    )
+    _SQL_MARK_ACK_UPDATE = (
+        'UPDATE {table_name} SET status = ? WHERE {key_column} = ?'
+    )
+    _SQL_SELECT_WHERE = (
+        'SELECT {key_column}, data, timestamp FROM {table_name}'
+        ' WHERE {key_column} > {rowid} AND status < %s AND'
+        ' {column} {op} ? ORDER BY {key_column} ASC'
+        ' LIMIT 1 ' % AckStatus.unack
+    )
+    _SQL_UPDATE = 'UPDATE {table_name} SET data = ? WHERE {key_column} = ?'
 
     def __init__(self, path, auto_resume=True, **kwargs):
         super(SQLiteAckQueue, self).__init__(path, **kwargs)
@@ -73,9 +87,10 @@ class SQLiteAckQueue(sqlbase.SQLiteBase):
 
     def put(self, item):
         obj = self._serializer.dumps(item)
-        self._insert_into(obj, _time.time())
+        _id = self._insert_into(obj, _time.time())
         self.total += 1
         self.put_event.set()
+        return _id
 
     def _init(self):
         super(SQLiteAckQueue, self)._init()
@@ -84,16 +99,16 @@ class SQLiteAckQueue(sqlbase.SQLiteBase):
         self.total = self._count()
 
     def _count(self):
-        sql = 'SELECT COUNT({}) FROM {}' \
-              ' WHERE status < ?'.format(self._key_column,
-                                         self._table_name)
+        sql = 'SELECT COUNT({}) FROM {} WHERE status < ?'.format(
+            self._key_column, self._table_name
+        )
         row = self._getter.execute(sql, (AckStatus.unack,)).fetchone()
         return row[0] if row else 0
 
     def _ack_count_via_status(self, status):
-        sql = 'SELECT COUNT({}) FROM {}' \
-              ' WHERE status = ?'.format(self._key_column,
-                                         self._table_name)
+        sql = 'SELECT COUNT({}) FROM {} WHERE status = ?'.format(
+            self._key_column, self._table_name
+        )
         row = self._getter.execute(sql, (status,)).fetchone()
         return row[0] if row else 0
 
@@ -111,18 +126,45 @@ class SQLiteAckQueue(sqlbase.SQLiteBase):
 
     @sqlbase.with_conditional_transaction
     def _mark_ack_status(self, key, status):
-        return self._sql_mark_ack_status, (status, key,)
+        return self._sql_mark_ack_status, (
+            status,
+            key,
+        )
 
     @sqlbase.with_conditional_transaction
-    def clear_acked_data(self):
+    def clear_acked_data(
+        self, max_delete=1000, keep_latest=1000, clear_ack_failed=False
+    ):
+        acked_clear_all = ''
+        acked_to_delete = ''
+        acked_to_keep = ''
+        if self._MAX_ACKED_LENGTH != 1000 and not max_delete:
+            # Added for backward compatibility for
+            # those that set the _MAX_ACKED_LENGTH
+            print(
+                "_MAX_ACKED_LENGTH has been depreciated.  \
+                    Use clear_acked_data(keep_latest=1000, max_delete=1000)"
+            )
+            max_delete = self._MAX_ACKED_LENGTH
+        if clear_ack_failed:
+            acked_clear_all = 'OR status = %s' % AckStatus.ack_failed
+        if max_delete and max_delete > 0:
+            acked_to_delete = 'LIMIT %d' % max_delete
+        if keep_latest and keep_latest > 0:
+            acked_to_keep = 'OFFSET %d' % keep_latest
         sql = """DELETE FROM {table_name}
             WHERE {key_column} IN (
-                SELECT _id FROM {table_name} WHERE status = ?
-                ORDER BY {key_column} DESC
-                LIMIT 1000 OFFSET {max_acked_length}
-            )""".format(table_name=self._table_name,
-                        key_column=self._key_column,
-                        max_acked_length=self._MAX_ACKED_LENGTH)
+                SELECT _id FROM {table_name}
+                WHERE status = ? {clear_ack_failed}
+                ORDER BY {key_column}
+                DESC {acked_to_delete} {acked_to_keep}
+            )""".format(
+            table_name=self._table_name,
+            key_column=self._key_column,
+            acked_to_delete=acked_to_delete,
+            acked_to_keep=acked_to_keep,
+            clear_ack_failed=acked_clear_all,
+        )
         return sql, AckStatus.acked
 
     @sqlbase.with_conditional_transaction
@@ -132,12 +174,13 @@ class SQLiteAckQueue(sqlbase.SQLiteBase):
 
     @property
     def _sql_mark_ack_status(self):
-        return self._SQL_MARK_ACK_UPDATE.format(table_name=self._table_name,
-                                                key_column=self._key_column)
+        return self._SQL_MARK_ACK_UPDATE.format(
+            table_name=self._table_name, key_column=self._key_column
+        )
 
-    def _pop(self):
+    def _pop(self, rowid=None, next_in_order=False, raw=False):
         with self.action_lock:
-            row = self._select()
+            row = self._select(next_in_order=next_in_order, rowid=rowid)
             # Perhaps a sqlite3 bug, sometimes (None, None) is returned
             # by select, below can avoid these invalid records.
             if row and row[0] is not None:
@@ -146,81 +189,161 @@ class SQLiteAckQueue(sqlbase.SQLiteBase):
                 item = self._serializer.loads(serialized_data)
                 self._unack_cache[row[0]] = item
                 self.total -= 1
-                return item
+                if raw:
+                    return {'pqid': row[0], 'data': item, 'tiemstamp': row[2]}
+                else:
+                    return item
             return None
 
-    def _find_item_id(self, item):
-        for key, value in self._unack_cache.items():
-            if value is item:
-                return key
-        log.warning("Can't find item in unack cache.")
+    def _find_item_id(self, item, search=True):
+        if item is None:
+            return None
+        elif isinstance(item, dict) and "pqid" in item:
+            return item.get("pqid")
+        elif search:
+            for key, value in self._unack_cache.items():
+                if value is item:
+                    return key
+            log.warning("Can't find item in unack cache.")
+        elif isinstance(item, int) or (
+            isinstance(item, str) and item.isnumeric()
+        ):
+            return int(item)
         return None
 
-    def ack(self, item):
+    def _check_id(self, item, id):
+        if id is not None and item is not None:
+            raise ValueError("Specify an id or an item, not both.")
+        elif id is None and item is None:
+            raise ValueError("Specify an id or an item.")
+        elif id is not None:
+            search = False
+            item = id
+        else:
+            search = True
+        return item, search
+
+    def ack(self, item=None, id=None):
+        item, search = self._check_id(item, id)
         with self.action_lock:
-            _id = self._find_item_id(item)
+            _id = self._find_item_id(item, search)
             if _id is None:
-                return
+                return None
             self._mark_ack_status(_id, AckStatus.acked)
-            self._unack_cache.pop(_id)
+            if _id in self._unack_cache:
+                self._unack_cache.pop(_id)
+        return _id
 
-    def ack_failed(self, item):
+    def ack_failed(self, item=None, id=None):
+        item, search = self._check_id(item, id)
         with self.action_lock:
-            _id = self._find_item_id(item)
+            _id = self._find_item_id(item, search)
             if _id is None:
-                return
+                return None
             self._mark_ack_status(_id, AckStatus.ack_failed)
-            self._unack_cache.pop(_id)
+            if _id in self._unack_cache:
+                self._unack_cache.pop(_id)
+        return _id
 
-    def nack(self, item):
+    def nack(self, item=None, id=None):
+        item, search = self._check_id(item, id)
         with self.action_lock:
-            _id = self._find_item_id(item)
+            _id = self._find_item_id(item, search)
             if _id is None:
-                return
+                return None
             self._mark_ack_status(_id, AckStatus.ready)
-            self._unack_cache.pop(_id)
+            if _id in self._unack_cache:
+                self._unack_cache.pop(_id)
             self.total += 1
-
-    def get(self, block=True, timeout=None):
+        return _id
+
+    def update(self, item, id=None):
+        if isinstance(item, dict) and "pqid" in item:
+            _id = item.get("pqid")
+            item = item.get("data")
+        if id is not None:
+            _id = id
+        if _id is None:
+            raise ValueError("Provide an id or raw item")
+        obj = self._serializer.dumps(item)
+        self._update(_id, obj)
+        return _id
+
+    def get(
+        self, block=True, timeout=None, id=None, next_in_order=False, raw=False
+    ):
+        rowid = self._find_item_id(id, search=False)
+        if rowid is None and next_in_order:
+            raise ValueError(
+                "'next_in_order' requires the preceding 'id' be specified."
+            )
+        if next_in_order and not isinstance(next_in_order, bool):
+            raise ValueError("'next_in_order' must be a boolean (True/False)")
         if not block:
-            serialized = self._pop()
+            serialized = self._pop(
+                next_in_order=next_in_order, raw=raw, rowid=rowid
+            )
             if serialized is None:
                 raise Empty
         elif timeout is None:
             # block until a put event.
-            serialized = self._pop()
+            serialized = self._pop(
+                next_in_order=next_in_order, raw=raw, rowid=rowid
+            )
             while serialized is None:
                 self.put_event.clear()
                 self.put_event.wait(TICK_FOR_WAIT)
-                serialized = self._pop()
+                serialized = self._pop(
+                    next_in_order=next_in_order, raw=raw, rowid=rowid
+                )
         elif timeout < 0:
             raise ValueError("'timeout' must be a non-negative number")
         else:
             # block until the timeout reached
             endtime = _time.time() + timeout
-            serialized = self._pop()
+            serialized = self._pop(
+                next_in_order=next_in_order, raw=raw, rowid=rowid
+            )
             while serialized is None:
                 self.put_event.clear()
                 remaining = endtime - _time.time()
                 if remaining <= 0.0:
                     raise Empty
                 self.put_event.wait(
-                    TICK_FOR_WAIT if TICK_FOR_WAIT < remaining else remaining)
-                serialized = self._pop()
-        item = serialized
-        return item
+                    TICK_FOR_WAIT if TICK_FOR_WAIT < remaining else remaining
+                )
+                serialized = self._pop(
+                    next_in_order=next_in_order, raw=raw, rowid=rowid
+                )
+        return serialized
 
     def task_done(self):
         """Persist the current state if auto_commit=False."""
         if not self.auto_commit:
             self._task_done()
 
+    def queue(self):
+        rows = self._sql_queue()
+        datarows = []
+        for row in rows:
+            item = {
+                'id': row[0],
+                'data': self._serializer.loads(row[1]),
+                'timestamp': row[2],
+                'status': row[3],
+            }
+            datarows.append(item)
+        return datarows
+
     @property
     def size(self):
         return self.total
 
     def qsize(self):
-        return self.size
+        return max(0, self.size)
+
+    def active_size(self):
+        return max(0, self.size + len(self._unack_cache))
 
     def empty(self):
         return self.size == 0
@@ -237,9 +360,11 @@ class FILOSQLiteAckQueue(SQLiteAckQueue):
 
     _TABLE_NAME = 'ack_filo_queue'
     # SQL to select a record
-    _SQL_SELECT = ('SELECT {key_column}, data FROM {table_name} '
-                   'WHERE status < %s '
-                   'ORDER BY {key_column} DESC LIMIT 1' % AckStatus.unack)
+    _SQL_SELECT = (
+        'SELECT {key_column}, data FROM {table_name} '
+        'WHERE {key_column} < {rowid} and status < %s '
+        'ORDER BY {key_column} DESC LIMIT 1' % AckStatus.unack
+    )
 
 
 class UniqueAckQ(SQLiteAckQueue):
@@ -252,10 +377,12 @@ class UniqueAckQ(SQLiteAckQueue):
 
     def put(self, item):
         obj = self._serializer.dumps(item, sort_keys=True)
+        _id = None
         try:
-            self._insert_into(obj, _time.time())
+            _id = self._insert_into(obj, _time.time())
         except sqlite3.IntegrityError:
             pass
         else:
             self.total += 1
             self.put_event.set()
+        return _id
diff --git a/persistqueue/sqlbase.py b/persistqueue/sqlbase.py
index fe42b77..b02767c 100644
--- a/persistqueue/sqlbase.py
+++ b/persistqueue/sqlbase.py
@@ -15,7 +15,9 @@ def with_conditional_transaction(func):
         with obj.tran_lock:
             with obj._putter as tran:
                 stat, param = func(obj, *args, **kwargs)
-                tran.execute(stat, param)
+                cur = tran.cursor()
+                cur.execute(stat, param)
+                return cur.lastrowid
 
     return _execute
 
@@ -32,7 +34,8 @@ def commit_ignore_error(conn):
         if 'no transaction is active' in str(ex):
             log.debug(
                 'Not able to commit the transaction, '
-                'may already be committed.')
+                'may already be committed.'
+            )
         else:
             raise
 
@@ -46,13 +49,20 @@ class SQLiteBase(object):
     _SQL_UPDATE = ''  # SQL to update a record
     _SQL_INSERT = ''  # SQL to insert a record
     _SQL_SELECT = ''  # SQL to select a record
+    _SQL_SELECT_ID = ''  # SQL to select a record with criteria
     _SQL_SELECT_WHERE = ''  # SQL to select a record with criteria
     _MEMORY = ':memory:'  # flag indicating store DB in memory
 
-    def __init__(self, path, name='default', multithreading=False,
-                 timeout=10.0, auto_commit=True,
-                 serializer=persistqueue.serializers.pickle,
-                 db_file_name=None):
+    def __init__(
+        self,
+        path,
+        name='default',
+        multithreading=False,
+        timeout=10.0,
+        auto_commit=True,
+        serializer=persistqueue.serializers.pickle,
+        db_file_name=None,
+    ):
         """Initiate a queue in sqlite3 or memory.
 
         :param path: path for storing DB file.
@@ -90,16 +100,17 @@ class SQLiteBase(object):
 
     def _init(self):
         """Initialize the tables in DB."""
-
         if self.path == self._MEMORY:
             self.memory_sql = True
             log.debug("Initializing Sqlite3 Queue in memory.")
         elif not os.path.exists(self.path):
             os.makedirs(self.path)
             log.debug(
-                'Initializing Sqlite3 Queue with path {}'.format(self.path))
+                'Initializing Sqlite3 Queue with path {}'.format(self.path)
+            )
         self._conn = self._new_db_connection(
-            self.path, self.multithreading, self.timeout)
+            self.path, self.multithreading, self.timeout
+        )
         self._getter = self._conn
         self._putter = self._conn
 
@@ -109,7 +120,8 @@ class SQLiteBase(object):
         if self.multithreading:
             if not self.memory_sql:
                 self._putter = self._new_db_connection(
-                    self.path, self.multithreading, self.timeout)
+                    self.path, self.multithreading, self.timeout
+                )
         self._conn.text_factory = str
         self._putter.text_factory = str
 
@@ -120,12 +132,13 @@ class SQLiteBase(object):
     def _new_db_connection(self, path, multithreading, timeout):
         conn = None
         if path == self._MEMORY:
-            conn = sqlite3.connect(path,
-                                   check_same_thread=not multithreading)
+            conn = sqlite3.connect(path, check_same_thread=not multithreading)
         else:
-            conn = sqlite3.connect('{}/{}'.format(path, self.db_file_name),
-                                   timeout=timeout,
-                                   check_same_thread=not multithreading)
+            conn = sqlite3.connect(
+                '{}/{}'.format(path, self.db_file_name),
+                timeout=timeout,
+                check_same_thread=not multithreading,
+            )
         conn.execute('PRAGMA journal_mode=WAL;')
         return conn
 
@@ -140,29 +153,65 @@ class SQLiteBase(object):
 
     @with_conditional_transaction
     def _delete(self, key, op='='):
-        sql = 'DELETE FROM {} WHERE {} {} ?'.format(self._table_name,
-                                                    self._key_column,
-                                                    op)
+        sql = 'DELETE FROM {} WHERE {} {} ?'.format(
+            self._table_name, self._key_column, op
+        )
         return sql, (key,)
 
     def _select(self, *args, **kwargs):
+        start_key = self._start_key()
         op = kwargs.get('op', None)
         column = kwargs.get('column', None)
-        if op and column:
-            return self._getter.execute(
-                self._sql_select_where(op, column), args).fetchone()
-        return self._getter.execute(self._sql_select, args).fetchone()
+        next_in_order = kwargs.get('next_in_order', False)
+        rowid = kwargs.get('rowid') if kwargs.get('rowid', None) else start_key
+        if not next_in_order and rowid != start_key:
+            # Get the record by the id
+            result = self._getter.execute(
+                self._sql_select_id(rowid), args
+            ).fetchone()
+        elif op and column:
+            # Get the next record with criteria
+            rowid = rowid if next_in_order else start_key
+            result = self._getter.execute(
+                self._sql_select_where(rowid, op, column), args
+            ).fetchone()
+        else:
+            # Get the next record
+            rowid = rowid if next_in_order else start_key
+            result = self._getter.execute(
+                self._sql_select(rowid), args
+            ).fetchone()
+        if (
+            next_in_order
+            and rowid != start_key
+            and (not result or len(result) == 0)
+        ):
+            # sqlackqueue: if we're at the end, start over - loop incremental
+            kwargs['rowid'] = start_key
+            result = self._select(args=args, kwargs=kwargs)
+        return result
 
     def _count(self):
-        sql = 'SELECT COUNT({}) FROM {}'.format(self._key_column,
-                                                self._table_name)
+        sql = 'SELECT COUNT({}) FROM {}'.format(
+            self._key_column, self._table_name
+        )
         row = self._getter.execute(sql).fetchone()
         return row[0] if row else 0
 
+    def _start_key(self):
+        if self._TABLE_NAME == 'ack_filo_queue':
+            return 9223372036854775807  # maxsize
+        else:
+            return 0
+
     def _task_done(self):
         """Only required if auto-commit is set as False."""
         commit_ignore_error(self._putter)
 
+    def _sql_queue(self):
+        sql = 'SELECT * FROM {}'.format(self._table_name)
+        return self._getter.execute(sql)
+
     @property
     def _table_name(self):
         return '`{}_{}`'.format(self._TABLE_NAME, self.name)
@@ -173,29 +222,44 @@ class SQLiteBase(object):
 
     @property
     def _sql_create(self):
-        return self._SQL_CREATE.format(table_name=self._table_name,
-                                       key_column=self._key_column)
+        return self._SQL_CREATE.format(
+            table_name=self._table_name, key_column=self._key_column
+        )
 
     @property
     def _sql_insert(self):
-        return self._SQL_INSERT.format(table_name=self._table_name,
-                                       key_column=self._key_column)
+        return self._SQL_INSERT.format(
+            table_name=self._table_name, key_column=self._key_column
+        )
 
     @property
     def _sql_update(self):
-        return self._SQL_UPDATE.format(table_name=self._table_name,
-                                       key_column=self._key_column)
-
-    @property
-    def _sql_select(self):
-        return self._SQL_SELECT.format(table_name=self._table_name,
-                                       key_column=self._key_column)
-
-    def _sql_select_where(self, op, column):
-        return self._SQL_SELECT_WHERE.format(table_name=self._table_name,
-                                             key_column=self._key_column,
-                                             op=op,
-                                             column=column)
+        return self._SQL_UPDATE.format(
+            table_name=self._table_name, key_column=self._key_column
+        )
+
+    def _sql_select_id(self, rowid):
+        return self._SQL_SELECT_ID.format(
+            table_name=self._table_name,
+            key_column=self._key_column,
+            rowid=rowid,
+        )
+
+    def _sql_select(self, rowid):
+        return self._SQL_SELECT.format(
+            table_name=self._table_name,
+            key_column=self._key_column,
+            rowid=rowid,
+        )
+
+    def _sql_select_where(self, rowid, op, column):
+        return self._SQL_SELECT_WHERE.format(
+            table_name=self._table_name,
+            key_column=self._key_column,
+            rowid=rowid,
+            op=op,
+            column=column,
+        )
 
     def __del__(self):
         """Handles sqlite connection when queue was deleted"""
diff --git a/persistqueue/sqlqueue.py b/persistqueue/sqlqueue.py
index f01eec7..c732572 100644
--- a/persistqueue/sqlqueue.py
+++ b/persistqueue/sqlqueue.py
@@ -24,22 +24,34 @@ class SQLiteQueue(sqlbase.SQLiteBase):
     _TABLE_NAME = 'queue'
     _KEY_COLUMN = '_id'  # the name of the key column, used in DB CRUD
     # SQL to create a table
-    _SQL_CREATE = ('CREATE TABLE IF NOT EXISTS {table_name} ('
-                   '{key_column} INTEGER PRIMARY KEY AUTOINCREMENT, '
-                   'data BLOB, timestamp FLOAT)')
+    _SQL_CREATE = (
+        'CREATE TABLE IF NOT EXISTS {table_name} ('
+        '{key_column} INTEGER PRIMARY KEY AUTOINCREMENT, '
+        'data BLOB, timestamp FLOAT)'
+    )
     # SQL to insert a record
     _SQL_INSERT = 'INSERT INTO {table_name} (data, timestamp) VALUES (?, ?)'
     # SQL to select a record
-    _SQL_SELECT = ('SELECT {key_column}, data FROM {table_name} '
-                   'ORDER BY {key_column} ASC LIMIT 1')
-    _SQL_SELECT_WHERE = 'SELECT {key_column}, data FROM {table_name} WHERE' \
-                        ' {column} {op} ? ORDER BY {key_column} ASC LIMIT 1 '
+    _SQL_SELECT_ID = (
+        'SELECT {key_column}, data, timestamp FROM {table_name} WHERE'
+        ' {key_column} = {rowid}'
+    )
+    _SQL_SELECT = (
+        'SELECT {key_column}, data, timestamp FROM {table_name} '
+        'ORDER BY {key_column} ASC LIMIT 1'
+    )
+    _SQL_SELECT_WHERE = (
+        'SELECT {key_column}, data, timestamp FROM {table_name} WHERE'
+        ' {column} {op} ? ORDER BY {key_column} ASC LIMIT 1 '
+    )
+    _SQL_UPDATE = 'UPDATE {table_name} SET data = ? WHERE {key_column} = ?'
 
     def put(self, item):
         obj = self._serializer.dumps(item)
-        self._insert_into(obj, _time.time())
+        _id = self._insert_into(obj, _time.time())
         self.total += 1
         self.put_event.set()
+        return _id
 
     def _init(self):
         super(SQLiteQueue, self)._init()
@@ -54,53 +66,88 @@ class SQLiteQueue(sqlbase.SQLiteBase):
                 self.cursor = 0
         self.total = self._count()
 
-    def _pop(self):
+    def _pop(self, rowid=None, raw=False):
         with self.action_lock:
             if self.auto_commit:
-                row = self._select()
+                row = self._select(rowid=rowid)
                 # Perhaps a sqlite3 bug, sometimes (None, None) is returned
                 # by select, below can avoid these invalid records.
                 if row and row[0] is not None:
                     self._delete(row[0])
                     self.total -= 1
-                    return row[1]  # serialized data
+                    item = self._serializer.loads(row[1])
+                    if raw:
+                        return {
+                            'pqid': row[0],
+                            'data': item,
+                            'tiemstamp': row[2],
+                        }
+                    else:
+                        return item
             else:
                 row = self._select(
-                    self.cursor, op=">", column=self._KEY_COLUMN)
+                    self.cursor, op=">", column=self._KEY_COLUMN, rowid=rowid
+                )
                 if row and row[0] is not None:
                     self.cursor = row[0]
                     self.total -= 1
-                    return row[1]
+                    item = self._serializer.loads(row[1])
+                    if raw:
+                        return {
+                            'pqid': row[0],
+                            'data': item,
+                            'tiemstamp': row[2],
+                        }
+                    else:
+                        return item
             return None
 
-    def get(self, block=True, timeout=None):
+    def update(self, item, id=None):
+        if isinstance(item, dict) and "pqid" in item:
+            _id = item.get("pqid")
+            item = item.get("data")
+        if id is not None:
+            _id = id
+        if _id is None:
+            raise ValueError("Provide an id or raw item")
+        obj = self._serializer.dumps(item)
+        self._update(_id, obj)
+        return _id
+
+    def get(self, block=True, timeout=None, id=None, raw=False):
+        if isinstance(id, dict) and "pqid" in id:
+            rowid = id.get("pqid")
+        elif isinstance(id, int):
+            rowid = id
+        else:
+            rowid = None
         if not block:
-            serialized = self._pop()
-            if not serialized:
+            serialized = self._pop(raw=raw, rowid=rowid)
+            if serialized is None:
                 raise Empty
         elif timeout is None:
             # block until a put event.
-            serialized = self._pop()
+            serialized = self._pop(raw=raw, rowid=rowid)
             while serialized is None:
                 self.put_event.clear()
                 self.put_event.wait(TICK_FOR_WAIT)
-                serialized = self._pop()
+                serialized = self._pop(raw=raw, rowid=rowid)
         elif timeout < 0:
             raise ValueError("'timeout' must be a non-negative number")
         else:
             # block until the timeout reached
             endtime = _time.time() + timeout
-            serialized = self._pop()
-            while not serialized:
+            serialized = self._pop(raw=raw, rowid=rowid)
+            while serialized is None:
                 self.put_event.clear()
                 remaining = endtime - _time.time()
                 if remaining <= 0.0:
                     raise Empty
                 self.put_event.wait(
-                    TICK_FOR_WAIT if TICK_FOR_WAIT < remaining else remaining)
-                serialized = self._pop()
-        item = self._serializer.loads(serialized)
-        return item
+                    TICK_FOR_WAIT if TICK_FOR_WAIT < remaining else remaining
+                )
+                serialized = self._pop(raw=raw, rowid=rowid)
+        return serialized
 
     def task_done(self):
         """Persist the current state if auto_commit=False."""
@@ -108,12 +155,24 @@ class SQLiteQueue(sqlbase.SQLiteBase):
             self._delete(self.cursor, op='<=')
             self._task_done()
 
+    def queue(self):
+        rows = self._sql_queue()
+        datarows = []
+        for row in rows:
+            item = {
+                'id': row[0],
+                'data': self._serializer.loads(row[1]),
+                'timestamp': row[2],
+            }
+            datarows.append(item)
+        return datarows
+
     @property
     def size(self):
         return self.total
 
     def qsize(self):
-        return self.size
+        return max(0, self.size)
 
     def empty(self):
         return self.size == 0
@@ -130,22 +189,28 @@ class FILOSQLiteQueue(SQLiteQueue):
 
     _TABLE_NAME = 'filo_queue'
     # SQL to select a record
-    _SQL_SELECT = ('SELECT {key_column}, data FROM {table_name} '
-                   'ORDER BY {key_column} DESC LIMIT 1')
+    _SQL_SELECT = (
+        'SELECT {key_column}, data FROM {table_name} '
+        'ORDER BY {key_column} DESC LIMIT 1'
+    )
 
 
 class UniqueQ(SQLiteQueue):
     _TABLE_NAME = 'unique_queue'
-    _SQL_CREATE = ('CREATE TABLE IF NOT EXISTS {table_name} ('
-                   '{key_column} INTEGER PRIMARY KEY AUTOINCREMENT, '
-                   'data BLOB, timestamp FLOAT, UNIQUE (data))')
+    _SQL_CREATE = (
+        'CREATE TABLE IF NOT EXISTS {table_name} ('
+        '{key_column} INTEGER PRIMARY KEY AUTOINCREMENT, '
+        'data BLOB, timestamp FLOAT, UNIQUE (data))'
+    )
 
     def put(self, item):
         obj = self._serializer.dumps(item, sort_keys=True)
+        _id = None
         try:
-            self._insert_into(obj, _time.time())
+            _id = self._insert_into(obj, _time.time())
         except sqlite3.IntegrityError:
             pass
         else:
             self.total += 1
             self.put_event.set()
+        return _id
diff --git a/setup.py b/setup.py
index 605252a..c61f7f5 100644
--- a/setup.py
+++ b/setup.py
@@ -40,6 +40,7 @@ setup(
         'Programming Language :: Python :: 3.5',
         'Programming Language :: Python :: 3.6',
         'Programming Language :: Python :: 3.7',
+        'Programming Language :: Python :: 3.8',
         'Topic :: Software Development :: Libraries'
     ],
 )


Test patch:
diff --git a/persistqueue/tests/test_sqlackqueue.py b/persistqueue/tests/test_sqlackqueue.py
index c1174c6..6f0c162 100644
--- a/persistqueue/tests/test_sqlackqueue.py
+++ b/persistqueue/tests/test_sqlackqueue.py
@@ -10,7 +10,8 @@ from threading import Thread
 from persistqueue.sqlackqueue import (
     SQLiteAckQueue,
     FILOSQLiteAckQueue,
-    UniqueAckQ)
+    UniqueAckQ,
+)
 from persistqueue import Empty
 
 
@@ -96,8 +97,7 @@ class SQLite3AckQueueTest(unittest.TestCase):
         # self.skipTest("Not supported multi-thread.")
 
         m_queue = SQLiteAckQueue(
-            path=self.path, multithreading=True,
-            auto_commit=self.auto_commit
+            path=self.path, multithreading=True, auto_commit=self.auto_commit
         )
 
         def producer():
@@ -122,8 +122,7 @@ class SQLite3AckQueueTest(unittest.TestCase):
     def test_multi_threaded_multi_producer(self):
         """Test sqlqueue can be used by multiple producers."""
         queue = SQLiteAckQueue(
-            path=self.path, multithreading=True,
-            auto_commit=self.auto_commit
+            path=self.path, multithreading=True, auto_commit=self.auto_commit
         )
 
         def producer(seq):
@@ -152,8 +151,7 @@ class SQLite3AckQueueTest(unittest.TestCase):
         """Test sqlqueue can be used by multiple consumers."""
 
         queue = SQLiteAckQueue(
-            path=self.path, multithreading=True,
-            auto_commit=self.auto_commit
+            path=self.path, multithreading=True, auto_commit=self.auto_commit
         )
 
         def producer():
@@ -185,23 +183,25 @@ class SQLite3AckQueueTest(unittest.TestCase):
 
         self.assertEqual(0, queue.qsize())
         for x in range(1000):
-            self.assertNotEqual(0, counter[x],
-                                "not 0 for counter's index %s" % x)
+            self.assertNotEqual(
+                0, counter[x], "not 0 for counter's index %s" % x
+            )
 
     def test_protocol_1(self):
         shutil.rmtree(self.path, ignore_errors=True)
         q = SQLiteAckQueue(path=self.path)
-        self.assertEqual(q._serializer.protocol,
-                         2 if sys.version_info[0] == 2 else 4)
+        self.assertEqual(
+            q._serializer.protocol, 2 if sys.version_info[0] == 2 else 4
+        )
 
     def test_protocol_2(self):
         q = SQLiteAckQueue(path=self.path)
-        self.assertEqual(q._serializer.protocol,
-                         2 if sys.version_info[0] == 2 else 4)
+        self.assertEqual(
+            q._serializer.protocol, 2 if sys.version_info[0] == 2 else 4
+        )
 
     def test_ack_and_clear(self):
         q = SQLiteAckQueue(path=self.path)
-        q._MAX_ACKED_LENGTH = 10
         ret_list = []
         for _ in range(100):
             q.put("val%s" % _)
@@ -210,7 +210,7 @@ class SQLite3AckQueueTest(unittest.TestCase):
         for ret in ret_list:
             q.ack(ret)
         self.assertEqual(q.acked_count(), 100)
-        q.clear_acked_data()
+        q.clear_acked_data(keep_latest=10)
         self.assertEqual(q.acked_count(), 10)
         q.shrink_disk_usage()
 
@@ -267,6 +267,8 @@ class SQLite3AckQueueTest(unittest.TestCase):
         # qsize should be zero when all item is getted from q
         self.assertEqual(q.qsize(), 0)
         self.assertEqual(q.unack_count(), 3)
+        # active size should be equal to qsize + unack_count
+        self.assertEqual(q.active_size(), 3)
         # nack will let the item requeued as ready status
         q.nack(val1)
         self.assertEqual(q.qsize(), 1)
@@ -294,6 +296,72 @@ class SQLite3AckQueueTest(unittest.TestCase):
         d = q.get(block=False)
         self.assertIsNotNone(d)
 
+    def test_get_id(self):
+        q = SQLiteAckQueue(path=self.path)
+        q.put("val1")
+        val2_id = q.put("val2")
+        q.put("val3")
+        item = q.get(id=val2_id)
+        # item id should be 2
+        self.assertEqual(val2_id, 2)
+        # item should get val2
+        self.assertEqual(item, 'val2')
+
+    def test_get_next_in_order(self):
+        q = SQLiteAckQueue(path=self.path)
+        val1_id = q.put("val1")
+        q.put("val2")
+        q.put("val3")
+        item = q.get(id=val1_id, next_in_order=True)
+        # item id should be 1
+        self.assertEqual(val1_id, 1)
+        # item should get val2
+        self.assertEqual(item, 'val2')
+
+    def test_get_raw(self):
+        q = SQLiteAckQueue(path=self.path)
+        q.put("val1")
+        item = q.get(raw=True)
+        # item should get val2
+        self.assertEqual(True, "pqid" in item)
+        self.assertEqual(item.get("data"), 'val1')
+
+    def test_nack_raw(self):
+        q = SQLiteAckQueue(path=self.path)
+        q.put("val1")
+        item = q.get(raw=True)
+        # nack a raw return
+        q.nack(item)
+        # size should be 1 after nack
+        self.assertEqual(q.qsize(), 1)
+
+    def test_ack_active_size(self):
+        q = SQLiteAckQueue(path=self.path)
+        q.put("val1")
+        item = q.get(raw=True)
+        # active_size should be 1 as it hasn't been acked
+        self.assertEqual(q.active_size(), 1)
+        q.ack(item)
+        # active_size should be 0 after ack
+        self.assertEqual(q.active_size(), 0)
+
+    def test_queue(self):
+        q = SQLiteAckQueue(path=self.path)
+        q.put("val1")
+        q.put("val2")
+        q.put("val3")
+        # queue should get the three items
+        d = q.queue()
+        self.assertEqual(len(d), 3)
+        self.assertEqual(d[1].get("data"), "val2")
+
+    def test_update(self):
+        q = SQLiteAckQueue(path=self.path)
+        qid = q.put("val1")
+        q.update(id=qid, item="val2")
+        item = q.get(id=qid)
+        self.assertEqual(item, "val2")
+
 
 class SQLite3QueueInMemory(SQLite3AckQueueTest):
     def setUp(self):
@@ -307,16 +375,19 @@ class SQLite3QueueInMemory(SQLite3AckQueueTest):
         self.skipTest('Memory based sqlite is not persistent.')
 
     def test_multiple_consumers(self):
-        self.skipTest('Skipped due to occasional crash during '
-                      'multithreading mode.')
+        self.skipTest(
+            'Skipped due to occasional crash during multithreading mode.'
+        )
 
     def test_multi_threaded_multi_producer(self):
-        self.skipTest('Skipped due to occasional crash during '
-                      'multithreading mode.')
+        self.skipTest(
+            'Skipped due to occasional crash during multithreading mode.'
+        )
 
     def test_multi_threaded_parallel(self):
-        self.skipTest('Skipped due to occasional crash during '
-                      'multithreading mode.')
+        self.skipTest(
+            'Skipped due to occasional crash during multithreading mode.'
+        )
 
     def test_task_done_with_restart(self):
         self.skipTest('Skipped due to not persistent.')
diff --git a/persistqueue/tests/test_sqlqueue.py b/persistqueue/tests/test_sqlqueue.py
index d71d661..37d5413 100644
--- a/persistqueue/tests/test_sqlqueue.py
+++ b/persistqueue/tests/test_sqlqueue.py
@@ -234,7 +234,7 @@ class SQLite3QueueTest(unittest.TestCase):
                 e=[1]
             ))
         q.put(x)
-        self.assertEquals(q.get(), x)
+        self.assertEqual(q.get(), x)
 
     def test_put_0(self):
         q = SQLiteQueue(path=self.path)
@@ -242,6 +242,42 @@ class SQLite3QueueTest(unittest.TestCase):
         d = q.get(block=False)
         self.assertIsNotNone(d)
 
+    def test_get_id(self):
+        q = SQLiteQueue(path=self.path)
+        q.put("val1")
+        val2_id = q.put("val2")
+        q.put("val3")
+        item = q.get(id=val2_id)
+        # item id should be 2
+        self.assertEqual(val2_id, 2)
+        # item should get val2
+        self.assertEqual(item, 'val2')
+
+    def test_get_raw(self):
+        q = SQLiteQueue(path=self.path)
+        q.put("val1")
+        item = q.get(raw=True)
+        # item should get val2
+        self.assertEqual(True, "pqid" in item)
+        self.assertEqual(item.get("data"), 'val1')
+
+    def test_queue(self):
+        q = SQLiteQueue(path=self.path)
+        q.put("val1")
+        q.put("val2")
+        q.put("val3")
+        # queue should get the three items
+        d = q.queue()
+        self.assertEqual(len(d), 3)
+        self.assertEqual(d[1].get("data"), "val2")
+
+    def test_update(self):
+        q = SQLiteQueue(path=self.path)
+        qid = q.put("val1")
+        q.update(item="val2", id=qid)
+        item = q.get(id=qid)
+        self.assertEqual(item, "val2")
+
 
 class SQLite3QueueNoAutoCommitTest(SQLite3QueueTest):
     def setUp(self):
@@ -265,6 +301,8 @@ class SQLite3QueueNoAutoCommitTest(SQLite3QueueTest):
 
 
 class SQLite3QueueInMemory(SQLite3QueueTest):
+    skipstr = 'Skipped due to occasional crash during multithreading mode.'
+
     def setUp(self):
         self.path = ":memory:"
         self.auto_commit = True
@@ -276,16 +314,13 @@ class SQLite3QueueInMemory(SQLite3QueueTest):
         self.skipTest('Memory based sqlite is not persistent.')
 
     def test_multiple_consumers(self):
-        self.skipTest('Skipped due to occasional crash during '
-                      'multithreading mode.')
+        self.skipTest(self.skipstr)
 
     def test_multi_threaded_multi_producer(self):
-        self.skipTest('Skipped due to occasional crash during '
-                      'multithreading mode.')
+        self.skipTest(self.skipstr)
 
     def test_multi_threaded_parallel(self):
-        self.skipTest('Skipped due to occasional crash during '
-                      'multithreading mode.')
+        self.skipTest(self.skipstr)
 
     def test_task_done_with_restart(self):
         self.skipTest('Skipped due to not persistent.')