def main()

in compiler_gym/service/runtime/create_and_run_compiler_gym_service.py [0:0]
51 lines of code
8 McCabe index (conditional complexity)

    def main(argv):
        # Register a signal handler for SIGTERM that will set the shutdownSignal
        # future value.
        signal(SIGTERM, _shutdown_handler)

        argv = [x for x in argv if x.strip()]
        if len(argv) > 1:
            print(
                f"ERROR: Unrecognized command line argument '{argv[1]}'",
                file=sys.stderr,
            )
            sys.exit(1)

        working_dir = Path(FLAGS.working_dir or mkdtemp(prefix="compiler_gym-service-"))
        (working_dir / "logs").mkdir(exist_ok=True, parents=True)

        FLAGS.log_dir = str(working_dir / "logs")
        logging.get_absl_handler().use_absl_log_file()
        logging.set_verbosity(dbg.get_logging_level())

        # Create the service.
        server = grpc.server(
            futures.ThreadPoolExecutor(max_workers=FLAGS.rpc_service_threads),
            options=[
                ("grpc.max_send_message_length", MAX_MESSAGE_SIZE_IN_BYTES),
                ("grpc.max_receive_message_length", MAX_MESSAGE_SIZE_IN_BYTES),
            ],
        )
        service = CompilerGymService(
            working_directory=working_dir,
            compilation_session_type=compilation_session_type,
        )
        compiler_gym_service_pb2_grpc.add_CompilerGymServiceServicer_to_server(
            service, server
        )

        address = f"0.0.0.0:{FLAGS.port}" if FLAGS.port else "0.0.0.0:0"
        port = server.add_insecure_port(address)

        with atomic_file_write(working_dir / "port.txt", fileobj=True, mode="w") as f:
            f.write(str(port))

        with atomic_file_write(working_dir / "pid.txt", fileobj=True, mode="w") as f:
            f.write(str(os.getpid()))

        logging.info(
            "Service %s listening on %d, PID = %d", working_dir, port, os.getpid()
        )

        server.start()

        # Block on the RPC service in a separate thread. This enables the
        # current thread to handle the shutdown routine.
        server_thread = Thread(target=server.wait_for_termination)
        server_thread.start()

        # Block until the shutdown signal is received.
        shutdown_signal.wait()
        logging.info("Shutting down the RPC service")
        server.stop(60).wait()
        server_thread.join()

        if len(service.sessions):
            print(
                "ERROR: Killing a service with",
                plural(len(service.session), "active session", "active sessions"),
                file=sys.stderr,
            )
            sys.exit(6)