sync codepath detector

Author: SeungjinYang

sky/jobs/server/server.py

@@ -206,8 +206,8 @@ async def pool_tail_logs(
         request_cluster_name=common.JOB_CONTROLLER_NAME,
     )
 
-    request_task = api_requests.get_request(request.state.request_id,
-                                            fields=['request_id'])
+    request_task = await api_requests.get_request_async(
+        request.state.request_id, fields=['request_id'])
 
     return stream_utils.stream_response_for_long_request(
         request_id=request_task.request_id,

sky/server/requests/requests.py

@@ -51,6 +51,11 @@
 
 DEFAULT_REQUESTS_RETENTION_HOURS = 24  # 1 day
 
+
+def _is_running_pytest() -> bool:
+    return 'PYTEST_CURRENT_TEST' in os.environ
+
+
 # TODO(zhwu): For scalability, there are several TODOs:
 # [x] Have a way to queue requests.
 # [ ] Move logs to persistent place.
@@ -632,6 +637,8 @@ def update_request(request_id: str) -> Generator[Optional[Request], None, None]:
     """Get and update a SkyPilot API request."""
     # Acquire the lock to avoid race conditions between multiple request
     # operations, e.g. execute and cancel.
+    if not _is_running_pytest() and asyncio_utils.is_running_async():
+        logger.warning('synchronous filelock is being used in an async context')
     with filelock.FileLock(request_lock_path(request_id)):
         request = _get_request_no_lock(request_id)
         yield request
@@ -718,6 +725,8 @@ async def get_latest_request_id_async() -> Optional[str]:
 def get_request(request_id: str,
                 fields: Optional[List[str]] = None) -> Optional[Request]:
     """Get a SkyPilot API request."""
+    if not _is_running_pytest() and asyncio_utils.is_running_async():
+        logger.warning('synchronous filelock is being used in an async context')
     with filelock.FileLock(request_lock_path(request_id)):
         return _get_request_no_lock(request_id, fields)
 

sky/utils/asyncio_utils.py

@@ -76,3 +76,12 @@ async def async_wrapper(*args, **kwargs):
             raise
 
     return async_wrapper
+
+
+def is_running_async() -> bool:
+    """Check if the code is currently running inside an asyncio event loop."""
+    try:
+        asyncio.get_running_loop()
+        return True
+    except RuntimeError:
+        return False