Revert "Revert Convert some job_node functions to pybind"

This reverts commit 6b27f63.

Author: andreas-el

src/clib/lib/include/ert/job_queue/job_node.hpp

@@ -62,8 +62,6 @@ struct job_queue_node_struct {
 typedef bool(job_callback_ftype)(void *);
 typedef struct job_queue_node_struct job_queue_node_type;
 
-extern "C" PY_USED submit_status_type job_queue_node_submit_simple(
-    job_queue_node_type *node, queue_driver_type *driver);
 void job_queue_node_fscanf_EXIT(job_queue_node_type *node);
 void job_queue_node_free_data(job_queue_node_type *node);
 
@@ -73,13 +71,9 @@ job_queue_node_alloc(const char *job_name, const char *run_path,
                      const stringlist_type *arguments, int num_cpu,
                      const char *status_file, const char *exit_file);
 
-extern "C" PY_USED bool job_queue_node_kill_simple(job_queue_node_type *node,
-                                                   queue_driver_type *driver);
 extern "C" void job_queue_node_free(job_queue_node_type *node);
 extern "C" job_status_type
 job_queue_node_get_status(const job_queue_node_type *node);
-extern "C" int
-job_queue_node_get_submit_attempt(const job_queue_node_type *node);
 
 int job_queue_node_get_queue_index(const job_queue_node_type *node);
 void job_queue_node_set_queue_index(job_queue_node_type *node, int queue_index);

src/clib/lib/job_queue/job_node.cpp

@@ -142,10 +142,6 @@ job_status_type job_queue_node_get_status(const job_queue_node_type *node) {
     return node->job_status;
 }
 
-int job_queue_node_get_submit_attempt(const job_queue_node_type *node) {
-    return node->submit_attempt;
-}
-
 job_queue_node_type *job_queue_node_alloc(const char *job_name,
                                           const char *run_path,
                                           const char *run_cmd, int argc,
@@ -201,66 +197,6 @@ void job_queue_node_set_status(job_queue_node_type *node,
         return;
 }
 
-submit_status_type job_queue_node_submit_simple(job_queue_node_type *node,
-                                                queue_driver_type *driver) {
-    submit_status_type submit_status;
-    pthread_mutex_lock(&node->data_mutex);
-    job_queue_node_set_status(node, JOB_QUEUE_SUBMITTED);
-    void *job_data = queue_driver_submit_job(
-        driver, node->run_cmd, node->num_cpu, node->run_path, node->job_name,
-        node->argc, (const char **)node->argv);
-
-    if (job_data == NULL) {
-        // In this case the status of the job itself will be
-        // unmodified; i.e. it will still be WAITING, and a new attempt
-        // to submit it will be performed in the next round.
-        submit_status = SUBMIT_DRIVER_FAIL;
-        logger->warning("Failed to submit job {} (attempt {})", node->job_name,
-                        node->submit_attempt);
-        pthread_mutex_unlock(&node->data_mutex);
-        return submit_status;
-    }
-
-    logger->info("Submitted job {} (attempt {})", node->job_name,
-                 node->submit_attempt);
-
-    node->job_data = job_data;
-    node->submit_attempt++;
-    // The status JOB_QUEUE_SUBMITTED is internal, and not exported anywhere.
-    // The job_queue_update_status() will update this to PENDING or RUNNING at
-    // the next call. The important difference between SUBMITTED and WAITING is
-    // that SUBMITTED have job_data != NULL and the job_queue_node free
-    // function must be called on it.
-    submit_status = SUBMIT_OK;
-    job_queue_node_set_status(node, JOB_QUEUE_SUBMITTED);
-    pthread_mutex_unlock(&node->data_mutex);
-    return submit_status;
-}
-
-bool job_queue_node_kill_simple(job_queue_node_type *node,
-                                queue_driver_type *driver) {
-    bool result = false;
-    pthread_mutex_lock(&node->data_mutex);
-    job_status_type current_status = job_queue_node_get_status(node);
-    if (current_status & JOB_QUEUE_CAN_KILL) {
-        // If the job is killed before it is even started no driver specific
-        // job data has been assigned; we therefore must check the
-        // node->job_data pointer before entering.
-        if (node->job_data) {
-            queue_driver_kill_job(driver, node->job_data);
-            queue_driver_free_job(driver, node->job_data);
-            node->job_data = NULL;
-        }
-        job_queue_node_set_status(node, JOB_QUEUE_IS_KILLED);
-        logger->info("job {} set to killed", node->job_name);
-        result = true;
-    } else {
-        logger->warning("node_kill called but cannot kill {}", node->job_name);
-    }
-    pthread_mutex_unlock(&node->data_mutex);
-    return result;
-}
-
 ERT_CLIB_SUBMODULE("queue", m) {
     using namespace py::literals;
     m.def("_refresh_status", [](Cwrap<job_queue_node_type> node,
@@ -316,4 +252,64 @@ ERT_CLIB_SUBMODULE("queue", m) {
         return std::make_pair<int, std::optional<std::string>>(
             int(current_status), std::move(error_msg));
     });
+
+    m.def("_submit", [](Cwrap<job_queue_node_type> node,
+                        Cwrap<queue_driver_type> driver) {
+        pthread_mutex_lock(&node->data_mutex);
+        job_queue_node_set_status(node, JOB_QUEUE_SUBMITTED);
+        void *job_data = queue_driver_submit_job(
+            driver, node->run_cmd, node->num_cpu, node->run_path,
+            node->job_name, node->argc, (const char **)node->argv);
+
+        if (job_data == nullptr) {
+            // In this case the status of the job itself will be
+            // unmodified; i.e. it will still be WAITING, and a new attempt
+            // to submit it will be performed in the next round.
+            logger->warning("Failed to submit job {} (attempt {})",
+                            node->job_name, node->submit_attempt);
+            pthread_mutex_unlock(&node->data_mutex);
+            return static_cast<int>(SUBMIT_DRIVER_FAIL);
+        }
+
+        logger->info("Submitted job {} (attempt {})", node->job_name,
+                     node->submit_attempt);
+
+        node->job_data = job_data;
+        node->submit_attempt++;
+        // The status JOB_QUEUE_SUBMITTED is internal, and not exported anywhere.
+        // The job_queue_update_status() will update this to PENDING or RUNNING at
+        // the next call. The important difference between SUBMITTED and WAITING is
+        // that SUBMITTED have job_data != NULL and the job_queue_node free
+        // function must be called on it.
+        job_queue_node_set_status(node, JOB_QUEUE_SUBMITTED);
+        pthread_mutex_unlock(&node->data_mutex);
+        return static_cast<int>(SUBMIT_OK);
+    });
+    m.def("_kill",
+          [](Cwrap<job_queue_node_type> node, Cwrap<queue_driver_type> driver) {
+              bool result = false;
+              pthread_mutex_lock(&node->data_mutex);
+              job_status_type current_status = job_queue_node_get_status(node);
+              if (current_status & JOB_QUEUE_CAN_KILL) {
+                  // If the job is killed before it is even started no driver specific
+                  // job data has been assigned; we therefore must check the
+                  // node->job_data pointer before entering.
+                  if (node->job_data) {
+                      queue_driver_kill_job(driver, node->job_data);
+                      queue_driver_free_job(driver, node->job_data);
+                      node->job_data = NULL;
+                  }
+                  job_queue_node_set_status(node, JOB_QUEUE_IS_KILLED);
+                  logger->info("job {} set to killed", node->job_name);
+                  result = true;
+              } else {
+                  logger->warning("node_kill called but cannot kill {}",
+                                  node->job_name);
+              }
+              pthread_mutex_unlock(&node->data_mutex);
+              return result;
+          });
+
+    m.def("_get_submit_attempt",
+          [](Cwrap<job_queue_node_type> node) { return node->submit_attempt; });
 }

src/ert/job_queue/job_queue_node.py

@@ -9,8 +9,13 @@
 from cwrap import BaseCClass
 from ecl.util.util import StringList
 
-from ert._clib.queue import _refresh_status  # pylint: disable=import-error
 from ert.callbacks import forward_model_ok
+from ert._clib.queue import (  # pylint: disable=import-error
+    _get_submit_attempt,
+    _kill,
+    _refresh_status,
+    _submit,
+)
 from ert.load_status import LoadStatus
 
 from ..realization_state import RealizationState
@@ -71,20 +76,12 @@ class JobQueueNode(BaseCClass):  # type: ignore
         bind=False,
     )
     _free = ResPrototype("void job_queue_node_free(job_queue_node)")
-    _submit = ResPrototype(
-        "job_submit_status_type_enum job_queue_node_submit_simple(job_queue_node, driver)"  # noqa
-    )
-    _run_kill = ResPrototype("bool job_queue_node_kill_simple(job_queue_node, driver)")
-
     _get_status = ResPrototype(
         "job_status_type_enum job_queue_node_get_status(job_queue_node)"
     )
     _set_queue_status = ResPrototype(
         "void job_queue_node_set_status(job_queue_node, job_status_type_enum)"
     )
-    _get_submit_attempt = ResPrototype(
-        "int job_queue_node_get_submit_attempt(job_queue_node)"
-    )
 
     # pylint: disable=too-many-arguments
     def __init__(
@@ -150,7 +147,7 @@ def timed_out(self) -> bool:
 
     @property
     def submit_attempt(self) -> int:
-        return self._get_submit_attempt()
+        return _get_submit_attempt(self)
 
     def _poll_queue_status(self, driver: "Driver") -> JobStatus:
         result, msg = _refresh_status(self, driver)
@@ -167,7 +164,7 @@ def queue_status(self, value: JobStatus) -> None:
         return self._set_queue_status(value)
 
     def submit(self, driver: "Driver") -> SubmitStatus:
-        return self._submit(driver)
+        return SubmitStatus(_submit(self, driver))
 
     def run_done_callback(self) -> Optional[LoadStatus]:
         callback_status, status_msg = forward_model_ok(self.run_arg)
@@ -342,7 +339,7 @@ def _transition_status(
         self.thread_status = thread_status
 
     def _kill(self, driver: "Driver") -> None:
-        self._run_kill(driver)
+        _kill(self, driver)
         self._tried_killing += 1
 
     def run(self, driver: "Driver", pool_sema: Semaphore, max_submit: int = 2) -> None: