Source code for blackhole.daemon
# -*- coding: utf-8 -*-
# (The MIT License)
#
# Copyright (c) 2013-2021 Kura
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the 'Software'), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED 'AS IS', WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
"""Provides daemonisation functionality."""
import atexit
import logging
import os
from .exceptions import DaemonException
from .utils import Singleton
__all__ = ("Daemon",)
"""Tuple all the things."""
logger = logging.getLogger("blackhole.daemon")
[docs]class Daemon(metaclass=Singleton):
"""An object for handling daemonisation."""
def __init__(self, pidfile):
"""
Create an instance of :class:`Daemon`.
:param str pidfile: Path to store the pid.
.. note::
Registers an :py.func:`atexit.register` signal to delete the pid on
exit.
"""
self.pidfile = pidfile
self.pid = os.getpid()
atexit.register(self._exit)
[docs] def daemonize(self):
"""Daemonize the process."""
self.fork()
os.chdir(os.path.sep)
os.setsid()
os.umask(0)
self.pid = os.getpid()
def _exit(self, *args, **kwargs):
"""Call on exit using :py:func:`atexit.register` or via a signal."""
del self.pid
[docs] def fork(self):
"""
Fork off the process.
:raises SystemExit: With code :py:obj:`os.EX_OK` when fork is
successful.
:raises DaemonException: When fork is unsuccessful.
"""
try:
pid = os.fork()
if pid > 0:
os._exit(os.EX_OK)
except OSError as err:
raise DaemonException(err.strerror)
@property
def pid(self):
"""
Pid of the process, if it's been daemonised.
:raises DaemonException: When pid cannot be read from the filesystem.
:returns: The current pid.
:rtype: :py:obj:`int` or :py:obj:`None`
.. note::
The pid is retrieved from the filestem.
If the pid does not exist in /proc, the pid is deleted.
"""
if os.path.exists(self.pidfile):
try:
with open(self.pidfile, "r") as pidfile:
pid = pidfile.read().strip()
if pid != "":
return int(pid)
except IOError as err:
raise DaemonException(str(err))
return None
@pid.setter
def pid(self, pid):
"""
Write the pid to the filesystem.
:param int pid: Process pid.
:raises DaemonException: When writing to filesystem fails.
"""
pid = str(pid)
try:
with open(self.pidfile, "w+") as pidfile:
pidfile.write(f"{pid}\n")
except IOError as err:
raise DaemonException(str(err))
@pid.deleter
def pid(self):
"""Delete the pid from the filesystem."""
if os.path.exists(self.pidfile):
os.remove(self.pidfile)