Function main

tests/watchdog/worker.py:21–154

(tests_pid)

def main(tests_pid):
    from debugpy.common import log, messaging

    # log.stderr_levels |= {"info"}
    log.timestamp_format = "06.3f"
    log_file = log.to_file(prefix="tests.watchdog")

    stream = messaging.JsonIOStream.from_stdio(f"tests-{tests_pid}")
    log.info("Spawned WatchDog-{0} for tests-{0}", tests_pid)
    tests_process = psutil.Process(tests_pid)
    stream.write_json(["watchdog", log_file.filename])

    spawned_processes = {}  # pid -> ProcessInfo
    try:
        stop = False
        while not stop:
            try:
                message = stream.read_json()
            except Exception:
                break

            command = message[0]
            args = message[1:]

            if command == "stop":
                assert not args
                stop = True

            elif command == "register_spawn":
                pid, name = args
                pid = int(pid)

                log.info(
                    "WatchDog-{0} registering spawned process {1} (pid={2})",
                    tests_pid,
                    name,
                    pid,
                )
                try:
                    _, old_name = spawned_processes[pid]
                except KeyError:
                    pass
                else:
                    log.warning(
                        "WatchDog-{0} already tracks a process with pid={1}: {2}",
                        tests_pid,
                        pid,
                        old_name,
                    )
                spawned_processes[pid] = ProcessInfo(psutil.Process(pid), name)

            elif command == "unregister_spawn":
                pid, name = args
                pid = int(pid)

                log.info(
                    "WatchDog-{0} unregistering spawned process {1} (pid={2})",
                    tests_pid,