Skip to content

orphan_process_monitor.py

Orphan process monitor.

__all__ = ['OrphanProcessChecker', 'OrphanProcessMonitor'] module-attribute

OrphanProcessChecker

Orphan process checker.

Only work for Linux platform. On Windows platform, is_orphan is always False and there is no need to do this monitoring on Windows.

Source code in solnlib/orphan_process_monitor.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class OrphanProcessChecker:
    """Orphan process checker.

    Only work for Linux platform. On Windows platform, is_orphan is
    always False and there is no need to do this monitoring on Windows.
    """

    def __init__(self, callback: Callable = None):
        """Initializes OrphanProcessChecker.

        Arguments:
            callback: (optional) Callback for orphan process.
        """
        if os.name == "nt":
            self._ppid = 0
        else:
            self._ppid = os.getppid()
        self._callback = callback

    def is_orphan(self) -> bool:
        """Check process is orphan.

        For windows platform just return False.

        Returns:
            True for orphan process else False.
        """

        if os.name == "nt":
            return False
        return self._ppid != os.getppid()

    def check_orphan(self) -> bool:
        """Check if the process becomes orphan.

        If the process becomes orphan then call callback function
        to handle properly.

        Returns:
            True for orphan process else False.
        """

        res = self.is_orphan()
        if res and self._callback:
            self._callback()
        return res

__init__(callback=None)

Initializes OrphanProcessChecker.

Parameters:

Name Type Description Default
callback Callable

(optional) Callback for orphan process.

 None
Source code in solnlib/orphan_process_monitor.py 
34
35
36
37
38
39
40
41
42
43
44
def __init__(self, callback: Callable = None):
    """Initializes OrphanProcessChecker.

    Arguments:
        callback: (optional) Callback for orphan process.
    """
    if os.name == "nt":
        self._ppid = 0
    else:
        self._ppid = os.getppid()
    self._callback = callback

check_orphan()

Check if the process becomes orphan.

If the process becomes orphan then call callback function to handle properly.

Returns:

Type Description
bool

True for orphan process else False.
Source code in solnlib/orphan_process_monitor.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def check_orphan(self) -> bool:
    """Check if the process becomes orphan.

    If the process becomes orphan then call callback function
    to handle properly.

    Returns:
        True for orphan process else False.
    """

    res = self.is_orphan()
    if res and self._callback:
        self._callback()
    return res

is_orphan()

Check process is orphan.

For windows platform just return False.

Returns:

Type Description
bool

True for orphan process else False.
Source code in solnlib/orphan_process_monitor.py 
46
47
48
49
50
51
52
53
54
55
56
57
def is_orphan(self) -> bool:
    """Check process is orphan.

    For windows platform just return False.

    Returns:
        True for orphan process else False.
    """

    if os.name == "nt":
        return False
    return self._ppid != os.getppid()

OrphanProcessMonitor

Orphan process monitor.

Check if process become orphan in background thread per interval and call callback if process become orphan.

Source code in solnlib/orphan_process_monitor.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
class OrphanProcessMonitor:
    """Orphan process monitor.

    Check if process become orphan in background thread per interval and
    call callback if process become orphan.
    """

    def __init__(self, callback: Callable, interval: int = 1):
        """Initializes OrphanProcessMonitor.

        Arguments:
            callback: Callback for orphan process monitor.
            interval: (optional) Interval to monitor.
        """
        self._checker = OrphanProcessChecker(callback)
        self._thr = threading.Thread(target=self._do_monitor)
        self._thr.daemon = True
        self._started = False
        self._interval = interval

    def start(self):
        """Start orphan process monitor."""

        if self._started:
            return
        self._started = True

        self._thr.start()

    def stop(self):
        """Stop orphan process monitor."""

        joinable = self._started
        self._started = False
        if joinable:
            self._thr.join(timeout=1)

    def _do_monitor(self):
        while self._started:
            if self._checker.check_orphan():
                break

            for _ in range(self._interval):
                if not self._started:
                    break
                time.sleep(1)

__init__(callback, interval=1)

Initializes OrphanProcessMonitor.

Parameters:

Name Type Description Default
callback Callable

Callback for orphan process monitor.

 required
interval int

(optional) Interval to monitor.

 1
Source code in solnlib/orphan_process_monitor.py 
82
83
84
85
86
87
88
89
90
91
92
93
def __init__(self, callback: Callable, interval: int = 1):
    """Initializes OrphanProcessMonitor.

    Arguments:
        callback: Callback for orphan process monitor.
        interval: (optional) Interval to monitor.
    """
    self._checker = OrphanProcessChecker(callback)
    self._thr = threading.Thread(target=self._do_monitor)
    self._thr.daemon = True
    self._started = False
    self._interval = interval

start()

Start orphan process monitor.

Source code in solnlib/orphan_process_monitor.py 
 95
 96
 97
 98
 99
100
101
102
def start(self):
    """Start orphan process monitor."""

    if self._started:
        return
    self._started = True

    self._thr.start()

stop()

Stop orphan process monitor.

Source code in solnlib/orphan_process_monitor.py 
104
105
106
107
108
109
110
def stop(self):
    """Stop orphan process monitor."""

    joinable = self._started
    self._started = False
    if joinable:
        self._thr.join(timeout=1)