def start()

in pyrit/ui/rpc.py [0:0]

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

    def start(self):
        """
        Attempt to start the RPC server. If the server is already running, this method will throw an exception.
        """

        # Check if the server is already running by checking if the port is already in use.
        # If the port is already in use, throw an exception.
        if self._is_instance_running():
            raise RPCAlreadyRunningException()

        self._score_received_semaphore = Semaphore(0)
        self._client_ready_semaphore = Semaphore(0)

        # Start the RPC server.
        self._rpc_service = self.RPCService(
            score_received_semaphore=self._score_received_semaphore, client_ready_semaphore=self._client_ready_semaphore
        )
        self._server = self.rpyc.ThreadedServer(
            self._rpc_service, port=DEFAULT_PORT, protocol_config={"allow_all_attrs": True}
        )
        self._server_thread = Thread(target=self._server.start)
        self._server_thread.start()

        # Start a thread to check if the client is still alive
        self._is_alive_stop = False
        self._is_alive_thread = Thread(target=self._is_alive)
        self._is_alive_thread.start()

        self._server_is_running = True

        logger.info("RPC server started")

        if not is_app_running():
            logger.info("Launching Gradio UI")
            launch_app(open_browser=self._open_browser)
        else:
            logger.info("Gradio UI is already running. Will not launch another instance.")