Singleton Process with PID Lock File #python

lockpidfile.py

import os
import fcntl
import contextlib
import time


@contextlib.contextmanager
def lockpidfile(filepath):
    with os.fdopen(
        os.open(filepath, os.O_RDWR | os.O_CREAT, mode=0o666),
        mode="r+",
        buffering=1,
        encoding="utf-8",
        newline="",
    ) as f:
        try:
            fcntl.lockf(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except OSError:
            yield False
            return
        f.truncate()
        pid = os.getpid()
        f.write(f"{pid}\n")
        yield True
        fcntl.lockf(f, fcntl.LOCK_UN)
        try:
            os.unlink(filepath)
        except OSError:
            pass


def main():
    with lockpidfile("/tmp/lockpidfile.pid") as lockpidstatus:
        if not lockpidstatus:
            print("NOT GOOD - PROCESS ALREADY STARTED")
        else:
            print("OK!")
            time.sleep(15)


main()

To try the above:

python ./lockpidfile.py &
python ./lockpidfile.py &

Good places to put these sorts of pid files is in:

/run/<systemservice>/<programname>.pid
/run/user/1000/<userservice>/<programname>.pid

The pid file will also be locked. This ensures that the same service won't be started again.

Other information can also be stored in this file...

To extend this concept, we can turn this into a class.

The class can then have state. In that we can have an enum of states of the process.

Then it's possible to use this as IPC to communicate the state of the service.

Either during bootstrapping, started... etc.

This allows us to "await" for a certain state, such as when the service has bound to a port or something.

To avoid polling for this state, you could make use of some eventing system, like how tail +F works or making use of fcntl.

Also to make the above portable: https://gavv.github.io/articles/file-locks/