From fcecde6178e7fe70a2b000b0767da0def07dfc9d Mon Sep 17 00:00:00 2001
From: Giuseppe Steduto <giuseppe@steduto.it>
Date: Mon, 29 Jan 2024 15:24:00 +0100
Subject: [PATCH] feat(executor): upgrade to Snakemake v7.32.4 (#81)

Amend the overridden executor to reflect the changes in the new version
of Snakemake, in particular with regard to the change of the
`_wait_for_jobs` method into a coroutine.

Closes #31
Closes reanahub/reana-client#655
---
 Dockerfile                                  | 12 +++++++++---
 reana_workflow_engine_snakemake/executor.py | 13 +++++++------
 setup.py                                    |  4 ++--
 3 files changed, 18 insertions(+), 11 deletions(-)

diff --git a/Dockerfile b/Dockerfile
index c372034..a495ad8 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,5 +1,5 @@
 # This file is part of REANA.
-# Copyright (C) 2021, 2022, 2023 CERN.
+# Copyright (C) 2021, 2022, 2023, 2024 CERN.
 #
 # REANA is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -82,8 +82,6 @@ RUN chmod +x /usr/local/bin/magick
 
 # Are we debugging?
 ARG DEBUG=0
-# hadolint ignore=DL3013
-RUN if [ "${DEBUG}" -gt 0 ]; then pip install --no-cache-dir -e ".[debug,xrootd]"; else pip install --no-cache-dir ".[xrootd]"; fi;
 
 # Are we building with locally-checked-out shared modules?
 # hadolint ignore=DL3013
@@ -95,6 +93,9 @@ RUN if test -e modules/reana-commons; then \
       fi \
     fi
 
+# hadolint ignore=DL3013
+RUN if [ "${DEBUG}" -gt 0 ]; then pip install --no-cache-dir -e ".[debug,xrootd]"; else pip install --no-cache-dir ".[xrootd]"; fi;
+
 # Check for any broken Python dependencies
 RUN pip check
 
@@ -102,6 +103,11 @@ RUN pip check
 ENV TERM=xterm \
     PYTHONPATH=/workdir
 
+# Create and switch to REANA user to be able to create snakemake-specific
+# directories in the home folder.
+RUN useradd reana --uid 1000 --create-home
+USER reana
+
 # Set image labels
 LABEL org.opencontainers.image.authors="team@reanahub.io"
 LABEL org.opencontainers.image.created="2023-12-12"
diff --git a/reana_workflow_engine_snakemake/executor.py b/reana_workflow_engine_snakemake/executor.py
index 02d5cfa..743f973 100644
--- a/reana_workflow_engine_snakemake/executor.py
+++ b/reana_workflow_engine_snakemake/executor.py
@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 #
 # This file is part of REANA.
-# Copyright (C) 2021, 2022, 2023 CERN.
+# Copyright (C) 2021, 2022, 2023, 2024 CERN.
 #
 # REANA is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -10,7 +10,7 @@
 
 import os
 import logging
-import time
+import asyncio
 from collections import namedtuple
 from typing import Callable
 
@@ -18,6 +18,7 @@
 from reana_commons.config import REANA_DEFAULT_SNAKEMAKE_ENV_IMAGE
 from reana_commons.utils import build_progress_message
 from snakemake import snakemake
+from snakemake.common import async_lock
 from snakemake.executors import ClusterExecutor, GenericClusterExecutor
 from snakemake.jobs import Job
 from snakemake import scheduler  # for monkeypatch
@@ -187,13 +188,13 @@ def _get_job_status_from_controller(self, job_id: str) -> str:
             )
             return JobStatus.failed.name
 
-    def _wait_for_jobs(self):
+    async def _wait_for_jobs(self):
         """Override _wait_for_jobs method to poll job-controller for job statuses.
 
         Original GenericClusterExecutor._wait_for_jobs method checks success/failure via .jobfinished or .jobfailed files.
         """
         while True:
-            with self.lock:
+            async with async_lock(self.lock):
                 if not self.wait:
                     return
                 active_jobs = self.active_jobs
@@ -215,7 +216,7 @@ def _wait_for_jobs(self):
                 else:
                     still_running.append(active_job)
 
-            with self.lock:
+            async with async_lock(self.lock):
                 # Even though we have set active_jobs to a new empty list at the
                 # beginning of _wait_for_jobs, here that list might not be empty anymore
                 # as more jobs might have been added while we were fetching the job
@@ -223,7 +224,7 @@ def _wait_for_jobs(self):
                 # list, instead of simply setting active_jobs to still_running.
                 self.active_jobs.extend(still_running)
 
-            time.sleep(POLL_JOBS_STATUS_SLEEP_IN_SECONDS)
+            await asyncio.sleep(POLL_JOBS_STATUS_SLEEP_IN_SECONDS)
 
 
 def submit_job(rjc_api_client, publisher, job_request_body):
diff --git a/setup.py b/setup.py
index f8975f4..7d247cf 100644
--- a/setup.py
+++ b/setup.py
@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 #
 # This file is part of REANA.
-# Copyright (C) 2021, 2022, 2023 CERN.
+# Copyright (C) 2021, 2022, 2023, 2024 CERN.
 #
 # REANA is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -51,7 +51,7 @@
 ]
 
 install_requires = [
-    "reana-commons[snakemake_reports]>=0.9.4,<0.10.0",
+    "reana-commons[snakemake_reports]>=0.9.6,<0.10.0",
 ]
 
 packages = find_packages()