Queue gets corrupted after power cut

[jensanjo]

Queue gets corrupted after power cut

[jensanjo]

I switched to the SQLite queue. It looks like hat got rid of he issue.

[jensanjo]

Here is the test program and queue files to reproduce the problem.
qtest.tar.gz

[peter-wangxu]

which version are you using？

[jensanjo]

I am using 0.7.0 from Pypi Op do 24 feb. 2022 11:48 schreef Peter Wang ***@***.***>:

…

which version are you using？ — Reply to this email directly, view it on GitHub <#180 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAO4VYWOFSAPCTPN3POGNULU4YEHTANCNFSM5PGTBFBQ> . Triage notifications on the go with GitHub Mobile for iOS <https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675> or Android <https://play.google.com/store/apps/details?id=com.github.android&referrer=utm_campaign%3Dnotification-email%26utm_medium%3Demail%26utm_source%3Dgithub>. You are receiving this because you authored the thread.Message ID: ***@***.***>

[peter-wangxu]

thanks, more questions:
1: which is your os kernel and the file system(ext4? xfs? and the mount info)
2: was the queue running within a container?
3: any other environment would be helpful

@jensanjo

[jensanjo]

Thanks for looking into this.

1. The system is running mingw64 under Windows 10 Enterprise. File system: ntfs
2. Not running within a container
3. I have not tried to reproduce in another environment

I hope this helps!

[Kryniowesegryderiusz]

Is there anything more on that? Having same error

[chintal]

It would be useful to know at what point the exceptions are raised (at queue creation / put / get) and to know what the exception will be. I was using pqueue which had the same issue, and I was able to work around it by catching pickle.UnpicklingError on pqueue.get_nowait(). In that case, I did not mind losing some data. It is certainly not ideal, but the data is lost by that point anyway. The exception otherwise requires manual intervention for the program to even continue operating with future data points. This manual intervention is not possible for my application.

[chintal]

I'm using the following wrapper class to avoid this problem. If anyone wishes to use this, you can. Note that the instantiation signature changes path from a positional arg to a keyword arg.

Note that this is exceedingly dangerous code. Not only does it assume that data loss in the queue is acceptable, it also runs shutil.rmtree without any real sanity check on what is essentially user provided input. Make sure that the path and tempdir you provide are completely disposable. Make a mistake there and you could end up accidentally wiping your entire filesystem.

import os
import shutil

class TolerantQueue(object):
    def __init__(self, name=None, logger=None, **kwargs):
        self._name = name or 'unspecified'
        self._logger = logger
        self._path = kwargs.pop('path')
        self._tempdir = kwargs.pop('tempdir', None)
        self._kwargs = kwargs
        self._create()

    def _create(self):
        if not os.path.exists(self._path):
            os.makedirs(self._path)
        if not os.path.exists(self._tempdir):
            os.makedirs(self._tempdir)
        try:
            self._actual_queue = Queue(path=self._path, tempdir=self._tempdir, **self._kwargs)
        except pickle.UnpicklingError as e:
            # info file is truncated
            if self._logger:
                self._logger.warn(f"Unpickling error ({e}) opening persisted queue "
                                  f"{self._name}. Nuking. There may be data loss.")
            self._reset()

    def _reset(self):
        self._actual_queue = None
        if self._path:
            shutil.rmtree(self._path, ignore_errors=True)
        if self._tempdir:
            shutil.rmtree(self._tempdir, ignore_errors=True)
        self._create()

    def get(self, *args, **kwargs):
        try:
            return self._actual_queue.get(*args, **kwargs)
        except (EOFError, pickle.UnpicklingError) as e:
            # q00000 like file is truncated or similar
            if self._logger:
                self._logger.warn(f"Unpickling error ({e}) reading persisted queue "
                                  f"{self._name}. Nuking. There may be data loss.")
            self._reset()
            return None

    def __getattr__(self, item):
        return getattr(self._actual_queue, item)