Integrate uvloop async engine

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

@@ -14,4 +14,5 @@ kubernetes_asyncio==31.1.0
 psutil==6.1.0
 referencing==0.35.1
 rdflib==7.0.0
+uvloop==0.21.0
 yattag==1.16.0

streamflow/main.py

@@ -4,8 +4,10 @@
 import asyncio
 import logging
 import os
+import platform
 import sys
 import uuid
+import uvloop
 from typing import Any, MutableMapping
 
 from streamflow import VERSION, report
@@ -245,16 +247,21 @@ def build_context(config: MutableMapping[str, Any]) -> StreamFlowContext:
 def main(args):
     try:
         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
         if args.context == "ext":
-            asyncio.run(_async_ext(args))
+            engine.run(_async_ext(args))
         elif args.context == "list":
-            asyncio.run(_async_list(args))
+            engine.run(_async_list(args))
         elif args.context == "plugin":
-            asyncio.run(_async_plugin(args))
+            engine.run(_async_plugin(args))
         elif args.context == "prov":
-            asyncio.run(_async_prov(args))
+            engine.run(_async_prov(args))
         elif args.context == "report":
-            asyncio.run(_async_report(args))
+            engine.run(_async_report(args))
         elif args.context == "run":
             if args.quiet:
                 logger.setLevel(logging.WARNING)
@@ -266,7 +273,7 @@ def main(args):
                 logger.handlers = []
                 logger.addHandler(colored_stream_handler)
                 logger.addFilter(HighlitingFilter())
-            asyncio.run(_async_run(args))
+            engine.run(_async_run(args))
         elif args.context == "schema":
             load_extensions()
             print(SfSchema().dump(args.version, args.pretty))