Manage a group of processes, keeping only N running at a time.

procmanage.py

#!/usr/bin/env python3
"""
Manage a group of processes, keeping only N running at a time.

Usage:
  procmanage.py N STRING

Arguments:
  N            An integer representing the number of processes to send the CONT signal to.
  STRING       A string representing the full name to match against.

Options:
  -h --help    Show this help message and exit.

Copyright 2023 Johannes Buchner <johannes.buchner.acad@gmx.com>
"""
# Copyright 2023 Johannes Buchner <johannes.buchner.acad@gmx.com>
# Written with help of ChatGPT
# 
# Redistribution and use in source and binary forms, with or without modification, are permitted #provided that the following conditions are met:
# 
# 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.
# 
# 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.
# 
# 3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission.
# 
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS “AS IS” AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

import os
import sys
import signal
import time
import argparse

class HelpfulParser(argparse.ArgumentParser):
    def error(self, message):
        sys.stderr.write('error: %s\n' % message)
        self.print_help()
        sys.exit(2)

def matches_cmdline_of_process(pid, string):
    """
    Check if the full name of a process matches a provided string.

    Parameters
    ----------
    pid : int
        Process ID to check.
    string : str
        String to match against the full process name.

    Returns
    -------
    bool
        True if the process full name matches the provided string, False otherwise.

    """
    try:
        with open(os.path.join('/proc', pid, 'cmdline'), 'rb') as f:
            return string in f.read().decode('utf-8')
    except IOError:
        return False

def get_matching_processes(string):
    """
    Get a list of processes whose full name matches a provided string.

    Parameters
    ----------
    string : str
        String to match against the full process name.

    Returns
    -------
    list of int
        List of process IDs whose full name matches the provided string.

    """
    # Get the PID of the current Python process
    current_pid = os.getpid()

    # Get list of all running processes
    all_processes = os.listdir('/proc')

    # Filter out processes that do not match the provided string and the current Python process
    matching_processes = [int(pid) for pid in all_processes if pid.isdigit() and pid != str(current_pid) and matches_cmdline_of_process(pid, string)]

    # Sort the matching processes by PID
    matching_processes.sort()

    return matching_processes

def loop(n, string):
    last_list = []

    while True:
        matching_processes = get_matching_processes(string)

        if last_list != matching_processes:
            # Send CONT signal to the first N matching processes
            # Send STOP signal to all but the first N matching processes
            [os.kill(pid, signal.SIGCONT) for pid in matching_processes[:n]]
            [os.kill(pid, signal.SIGSTOP) for pid in matching_processes[n:]]
            del last_list
            last_list = matching_processes
            print("continuing:", matching_processes[:n], "stopping:", matching_processes[n:])

        # Wait for 1 second and repeat
        time.sleep(1)

def main():
    """
    Send a signal to a group of processes whose full name matches a provided string.

    The first N processes matching the provided string will receive a CONT signal, while all other matching processes
    will receive a STOP signal.

    """
    parser = HelpfulParser(
        description=__doc__,
        formatter_class=argparse.ArgumentDefaultsHelpFormatter)

    parser.add_argument("N", type=int, help="Number of processes to send CONT signal to.")
    parser.add_argument("STRING", type=str, help="String to match full process name against.")

    args = parser.parse_args()
    loop(args.N, args.STRING)


if __name__ == '__main__':
    main()