multiprocessing - How to identify and manage a background daemon process beyond parent process runtime in python - Stack Overflo

admin管理员组

文章数量:1429954

I want to run a python script B in the background every 8 hours on a Windows machine that usually runs 24/7. Also, I have another script A that is manually executed and uses data that is provided by the background script (which queries a large database and extracts relevant data so that it is avl. for the main script w/o long waiting times).

The question is: How can I assure that B is running when A is started?

My idea is when starting A to somehow check whether B already exists, and if not launch it via multiprocessing.Process. But how can I identify this process?

Only thing I came up with would be saving the process id somewhere in a file to disk and then check every time whether a process with this id exists - but afaik this id must not necessarily refer to my process in case that one crashed and Windows gave the same id to another process in the meantime.

I want to run a python script B in the background every 8 hours on a Windows machine that usually runs 24/7. Also, I have another script A that is manually executed and uses data that is provided by the background script (which queries a large database and extracts relevant data so that it is avl. for the main script w/o long waiting times).

The question is: How can I assure that B is running when A is started?

My idea is when starting A to somehow check whether B already exists, and if not launch it via multiprocessing.Process. But how can I identify this process?

Only thing I came up with would be saving the process id somewhere in a file to disk and then check every time whether a process with this id exists - but afaik this id must not necessarily refer to my process in case that one crashed and Windows gave the same id to another process in the meantime.

• python
• multiprocessing
• python-multiprocessing
• background-process

Share Improve this question Follow asked Nov 19, 2024 at 14:16 ascripterascripter 6,2651212 gold badges5454 silver badges7272 bronze badges

Add a comment  | 

2 Answers 2

Sorted by: Reset to default 0

Use a File-Based Lock**. It's simple, robust, and doesn't require external tools.

---

Script B (Background Script):

import os
import time

LOCK_FILE = 'script_b.lock'

def is_running():
    """Check if the lock file exists."""
    return os.path.exists(LOCK_FILE)

def create_lock_file():
    """Create the lock file and write the current process ID."""
    with open(LOCK_FILE, 'w') as f:
        f.write(str(os.getpid()))

def remove_lock_file():
    """Remove the lock file."""
    if os.path.exists(LOCK_FILE):
        os.remove(LOCK_FILE)

try:
    if is_running():
        print("Script B is already running.")
        exit(0)  # Exit if already running
    
    create_lock_file()
    print("Script B is running in the background...")
    
    # Simulate continuous work (replace this with your actual logic)
    while True:
        time.sleep(1)

finally:
    remove_lock_file()

---

Script A (Main Script):

import os
import psutil
import subprocess

LOCK_FILE = 'script_b.lock'

def get_running_pid():
    """Get the PID from the lock file if it exists."""
    if os.path.exists(LOCK_FILE):
        with open(LOCK_FILE, 'r') as f:
            return int(f.read().strip())
    return None

def is_process_running(pid):
    """Check if a process with the given PID is still running."""
    try:
        p = psutil.Process(pid)
        return p.is_running()
    except (psutil.NoSuchProcess, ValueError):
        return False

def start_script_b():
    """Start Script B as a detached process."""
    print("Starting Script B...")
    subprocess.Popen(['python', 'script_b.py'], creationflags=subprocess.DETACHED_PROCESS)

pid = get_running_pid()
if pid and is_process_running(pid):
    print(f"Script B is running with PID {pid}.")
else:
    print("Script B is not running. Starting it now.")
    start_script_b()

---

How It Works:

1. Script B creates a lock file (script_b.lock) with its process ID when it starts. If the lock file exists, it assumes it's already running.
2. Script A checks the lock file for a PID and verifies if the process is active using the psutil library.
3. If Script B is not running, Script A launches it using subprocess.Popen.

Dependencies:

• The psutil library for process checks (pip install psutil).
• Ensure both scripts are in the same directory or adjust file paths accordingly.

This solution keeps things lightweight, Python-only, and easy to manage!

I would create the following module that will start an arbitrary Python script as a detached process if and only if the script is not already running (this requires the psutil and FileLock modules from the PyPi repository). The code recognizes that the required script is running by searching all the currently executing processes looking for the (hopefully) unique script name.

start_script.py

from subprocess import Popen, DETACHED_PROCESS
import sys

from filelock import FileLock
import psutil

def start_script(script_name: str, needs_console=True) -> None:
    """Start script if it is not already running as a new, detached process.
    If needs_console is True (the default), then the Windows start command
    is used to provide a new console.
    """

    lock_file = script_name + '.lock'

    with FileLock(lock_file):
        for process in psutil.process_iter(['cmdline']):
            cmdline = process.info['cmdline']
            if cmdline and len(cmdline) > 1 and cmdline[1] == script_name:
                return # Already running

        # Here if the script is not already running
        if needs_console:
            # Use Windows start command
            Popen(['start', sys.executable, script_name], creationflags=DETACHED_PROCESS, shell=True)
        else:
            Popen([sys.executable, script_name], creationflags=DETACHED_PROCESS)

Then instead of ever executing Script B directly with python scriptB.py, create a new script that will ensure that only a single instance of sciptB.py is ever running: :

start_scriptB

from start_script import start_script

SCRIPT_B_PATH = r'\some-directory-path\scriptB.py'

def startB():
    start_script(SCRIPT_B_PATH)

if __name__ == '__main__:
    startB()

Your scheduler will now periodically run the above script instead.

Finally, Script A would contain:

from start_scriptB import startB

...
startB()  # Start Script B if it is not already running

NB: Testing whether a script is running and then starting the script if it is not constitutes a critical section, i.e. these two operations must appear to be atomic to ensure that you cannot have two instances of the script executing concurrently. That is why both operations are performed under the control of an acquired FileLock instance.

本文标签：