Integrate uvloop async engine

This commit integrates the `uvloop` asynchronous engine as a drop-in
replacement for `asyncio` when Python runs on top of the `Cython`
library. The `uvloop` library is faster than the standard async engine,
improving StreamFlow performance epsecially on HTC workflows.

Author: GlassOfWhiskey

examples/mpi/streamflow.yml

@@ -9,11 +9,11 @@ workflows:
     bindings:
       - step: /compile
         target:
-          deployment: k8s-mpi
+          deployment: helm-mpi
           service: openmpi
       - step: /execute
         target:
-          deployment: k8s-mpi
+          deployment: helm-mpi
           locations: 2
           service: openmpi
 deployments:
@@ -28,7 +28,6 @@ deployments:
     type: helm
     config:
       chart: environment/helm/openmpi
-      kubeconfig: ~/.kube/config-streamflow
       releaseName: openmpi-rel
     workdir: /tmp
   k8s-mpi:
@@ -37,5 +36,4 @@ deployments:
       files:
         - environment/k8s/secrets.yaml
         - environment/k8s/deployment.yaml
-      kubeconfig: ~/.kube/config-streamflow
     workdir: /tmp

requirements.txt

@@ -13,4 +13,5 @@ psutil==7.1.3
 referencing==0.37.0
 rdflib==7.4.0
 typing-extensions==4.15.0
-yattag==1.16.1
+uvloop==0.22.1
+yattag==1.16.1

streamflow/main.py

@@ -4,8 +4,10 @@
 import asyncio
 import logging
 import os
+import platform
 import sys
 import uuid
+import uvloop
 from collections.abc import MutableMapping, Sequence
 from typing import TYPE_CHECKING, Any
 
@@ -252,17 +254,22 @@ def build_context(config: MutableMapping[str, Any]) -> StreamFlowContext:
 def main(args: Sequence[str]) -> int:
     try:
         parsed_args = parser.parse_args(args)
+        if sys.platform != "win32" and platform.python_implementation() == "CPython":
+            logger.info("CPython detected: using uvloop EventLoop implementation")
+            engine = uvloop
+        else:
+            engine = asyncio
         match parsed_args.context:
             case "ext":
-                asyncio.run(_async_ext(parsed_args))
+                engine.run(_async_ext(parsed_args))
             case "list":
-                asyncio.run(_async_list(parsed_args))
+                engine.run(_async_list(parsed_args))
             case "plugin":
-                asyncio.run(_async_plugin(parsed_args))
+                engine.run(_async_plugin(parsed_args))
             case "prov":
-                asyncio.run(_async_prov(parsed_args))
+                engine.run(_async_prov(parsed_args))
             case "report":
-                asyncio.run(_async_report(parsed_args))
+                engine.run(_async_report(parsed_args))
             case "run":
                 if parsed_args.quiet:
                     logger.setLevel(logging.WARNING)
@@ -278,7 +285,7 @@ def main(args: Sequence[str]) -> int:
                     logger.handlers = []
                     logger.addHandler(colored_stream_handler)
                     logger.addFilter(HighlitingFilter())
-                asyncio.run(_async_run(parsed_args))
+                engine.run(_async_run(parsed_args))
             case "schema":
                 load_extensions()
                 print(SfSchema().dump(parsed_args.version, parsed_args.pretty))