def run()

in github-runner-ami/packer/files/runner-supervisor.py [0:0]

• 32 lines of code
• 6 McCabe index (conditional complexity)

    def run(self):
        # Create a signal pipe that we can poll on
        sig_read, sig_write = socket.socketpair()

        sel = selectors.DefaultSelector()

        def sig_handler(signal, frame):
            # no-op
            ...

        sig_read.setblocking(False)
        sig_write.setblocking(False)
        sel.register(sig_read, selectors.EVENT_READ, None)

        proc_socket = self.open_proc_connector_socket()
        proc_socket.setblocking(False)

        signal.signal(signal.SIGINT, sig_handler)
        signal.signal(signal.SIGALRM, sig_handler)
        signal.setitimer(signal.ITIMER_REAL, 30, 30.0)
        signal.set_wakeup_fd(sig_write.fileno(), warn_on_full_buffer=False)

        sel.register(proc_socket, selectors.EVENT_READ, self.handle_proc_event)

        self.pgrep()

        try:
            while True:
                for key, mask in sel.select():

                    if key.fileobj == sig_read:
                        sig = signal.Signals(key.fileobj.recv(1)[0])  # type: ignore
                        if sig == signal.SIGALRM:
                            self.check_still_alive()
                            continue
                        else:
                            log.info(f"Got {sig.name}, exiting")
                            return
                    callback = key.data
                    callback(key.fileobj, mask)
        finally:
            # Disable the timers for any cleanup code to run
            signal.setitimer(signal.ITIMER_REAL, 0)
            signal.set_wakeup_fd(-1)