Repository: jina-ai/jina
File: jina/serve/runtimes/asyncio.py (Python)

Code before:
            except (ValueError, RuntimeError) as exc:
                self.logger.warning(
                    f' The runtime {self.__class__.__name__} will not be able to handle termination signals. '
                    f' {repr(exc)}'
                )
        else:
            with ImportExtensions(
                    required=True,
                    logger=self.logger,
                    help_text='''If you see a 'DLL load failed' error, please reinstall `pywin32`.
                If you're using conda, please use the command `conda install -c anaconda pywin32`''',
            ):
                import win32api

            win32api.SetConsoleCtrlHandler(
                lambda *args, **kwargs: self.is_cancel.set(), True
            )

        self._setup_monitoring()
        flow_id = self._flow_id
        telemetry_kwargs = {}

        if flow_id is not None:
            telemetry_kwargs = {'flow_id': flow_id}
        send_telemetry_event(event='start', obj=self, **telemetry_kwargs)
        self._start_time = time.time()
        self._loop.run_until_complete(self.async_setup())

    def run_forever(self):
        """
        Running method to block the main thread.

        Run the event loop until a Future is done.
        """
        self._loop.run_until_complete(self._loop_body())

    def teardown(self):
        """Call async_teardown() and stop and close the event loop."""
        self._loop.run_until_complete(self.async_teardown())
        self._loop.stop()
        self._loop.close()
        super().teardown()
        flow_id = self._flow_id
        self._stop_time = time.time()
        telemetry_kwargs = {'duration': self._stop_time - self._start_time}

        if flow_id is not None:
            telemetry_kwargs = {'flow_id': flow_id}
        send_telemetry_event(event='stop', obj=self, **telemetry_kwargs)

    async def _wait_for_cancel(self):

Review comment (suggestion):
I think it's better to have function `_get_telemetry_extra_kwargs` and override it in gateway runtime to flow id.
I don't like much how we set a `None` flow id for all pods.
I'm also okay with renaming to `pod_id` and then each runtime can implement it on his own

Code after:
            except (ValueError, RuntimeError) as exc:
                self.logger.warning(
                    f' The runtime {self.__class__.__name__} will not be able to handle termination signals. '
                    f' {repr(exc)}'
                )
        else:
            with ImportExtensions(
                    required=True,
                    logger=self.logger,
                    help_text='''If you see a 'DLL load failed' error, please reinstall `pywin32`.
                If you're using conda, please use the command `conda install -c anaconda pywin32`''',
            ):
                import win32api

            win32api.SetConsoleCtrlHandler(
                lambda *args, **kwargs: self.is_cancel.set(), True
            )

        self._setup_monitoring()
        send_telemetry_event(event='start', obj=self, entity_id=self._entity_id)
        self._start_time = time.time()
        self._loop.run_until_complete(self.async_setup())

    def run_forever(self):
        """
        Running method to block the main thread.

        Run the event loop until a Future is done.
        """
        self._loop.run_until_complete(self._loop_body())

    def teardown(self):
        """Call async_teardown() and stop and close the event loop."""
        self._loop.run_until_complete(self.async_teardown())
        self._loop.stop()
        self._loop.close()
        super().teardown()
        self._stop_time = time.time()
        send_telemetry_event(event='stop', obj=self, duration=self._stop_time - self._start_time, entity_id=self._entity_id)

    async def _wait_for_cancel(self):
        """Do NOT override this method when inheriting from :class:`GatewayPod`"""
        # threads are not using asyncio.Event, but threading.Event
        if isinstance(self.is_cancel, asyncio.Event):
            await self.is_cancel.wait()
        else:
            while not self.is_cancel.is_set():
                await asyncio.sleep(0.1)

        await self.async_cancel()