PyBitmessage-2024-12-05/src/singleinstance.py

"""
This is based upon the singleton class from
`tendo <https://github.com/pycontribs/tendo>`_
which is under the Python Software Foundation License version 2
"""

import atexit
import os
import sys

import state

try:
    import fcntl  # @UnresolvedImport
except ImportError:
    pass


class singleinstance(object):
    """
    Implements a single instance application by creating a lock file
    at appdata.
    """
    def __init__(self, flavor_id="", daemon=False):
        self.initialized = False
        self.counter = 0
        self.daemon = daemon
        self.lockPid = None
        self.lockfile = os.path.normpath(
            os.path.join(state.appdata, 'singleton%s.lock' % flavor_id))

        if state.enableGUI and not self.daemon and not state.curses:
            # Tells the already running (if any) application to get focus.
            import bitmessageqt
            bitmessageqt.init()

        self.lock()

        self.initialized = True
        atexit.register(self.cleanup)

    def lock(self):
        """Obtain single instance lock"""
        if self.lockPid is None:
            self.lockPid = os.getpid()
        if sys.platform == 'win32':
            try:
                # file already exists, we try to remove
                # (in case previous execution was interrupted)
                if os.path.exists(self.lockfile):
                    os.unlink(self.lockfile)
                self.fd = os.open(
                    self.lockfile,
                    os.O_CREAT | os.O_EXCL | os.O_RDWR | os.O_TRUNC
                )
            except OSError as e:
                if e.errno == 13:
                    sys.exit(
                        'Another instance of this application is'
                        ' already running')
                raise
            else:
                pidLine = "%i\n" % self.lockPid
                os.write(self.fd, pidLine)
        else:  # non Windows
            self.fp = open(self.lockfile, 'a+')
            try:
                if self.daemon and self.lockPid != os.getpid():
                    # wait for parent to finish
                    fcntl.lockf(self.fp, fcntl.LOCK_EX)
                else:
                    fcntl.lockf(self.fp, fcntl.LOCK_EX | fcntl.LOCK_NB)
                self.lockPid = os.getpid()
            except IOError:
                sys.exit(
                    'Another instance of this application is'
                    ' already running')
            else:
                pidLine = "%i\n" % self.lockPid
                self.fp.truncate(0)
                self.fp.write(pidLine)
                self.fp.flush()

    def cleanup(self):
        """Release single instance lock"""
        if not self.initialized:
            return
        if self.daemon and self.lockPid == os.getpid():
            # these are the two initial forks while daemonizing
            try:
                if sys.platform == 'win32':
                    if hasattr(self, 'fd'):
                        os.close(self.fd)
                else:
                    fcntl.lockf(self.fp, fcntl.LOCK_UN)
            except (IOError, OSError):
                pass

            return

        try:
            if sys.platform == 'win32':
                if hasattr(self, 'fd'):
                    os.close(self.fd)
                    os.unlink(self.lockfile)
            else:
                fcntl.lockf(self.fp, fcntl.LOCK_UN)
                if os.path.isfile(self.lockfile):
                    os.unlink(self.lockfile)
        except (IOError, OSError):
            pass