Merge pull request #1175 from trws/qmanager-remove-pending

qmanager: split remove to separate pending

Author: mergify[bot]

qmanager/modules/qmanager_callbacks.cpp

@@ -316,7 +316,7 @@ void qmanager_cb_t::jobmanager_cancel_cb (flux_t *h, const flux_msg_t *msg,
     if ((job = queue->lookup (id)) == nullptr
         || !job->is_pending ())
         return;
-    if (queue->remove (id) < 0) {
+    if (queue->remove_pending (job.get ()) < 0) {
         flux_log_error (h, "%s: remove job (%jd)", __FUNCTION__,
                        static_cast<intmax_t> (id));
         return;

qmanager/policies/base/queue_policy_base.hpp

@@ -101,7 +101,8 @@ class queue_policy_base_impl_t
 {
 public:
     int insert (std::shared_ptr<job_t> job);
-    int remove (flux_jobid_t id);
+    int remove_pending(job_t *job);
+    int remove(flux_jobid_t id);
     const std::shared_ptr<job_t> lookup (flux_jobid_t id);
     bool is_schedulable ();
     void set_schedulability (bool scheduable);
@@ -116,7 +117,7 @@ class queue_policy_base_impl_t
     int process_provisional_cancel ();
     int process_provisional_reprio ();
     int insert_pending_job (std::shared_ptr<job_t> &job, bool into_provisional);
-    int erase_pending_job (std::shared_ptr<job_t> &job, bool &found_in_prov);
+    int erase_pending_job (job_t *job, bool &found_in_prov);
     std::shared_ptr<job_t> pending_pop ();
     std::shared_ptr<job_t> alloced_pop ();
     std::shared_ptr<job_t> rejected_pop ();
@@ -270,6 +271,17 @@ class queue_policy_base_t : public detail::queue_policy_base_impl_t,
      */
     int insert (std::shared_ptr<job_t> pending_job);
 
+    /*! Remove a job whose jobid is id from the pending or maybe_pending queues.
+     * If succeeds, it changes the pending queue or resource state. This queue
+     * becomes "schedulable" if pending job queue is not empty: i.e.,
+     * is_schedulable() returns true;
+     *
+     *  \param id        jobid of flux_jobid_t type.
+     *  \return          0 on success; -1 on error.
+     *                       ENOENT: unknown id.
+     */
+    int remove_pending (job_t *job);
+
     /*! Remove a job whose jobid is id from any internal queues
      *  (e.g., pending queue, running queue, and alloced queue.)
      *  If succeeds, it changes the pending queue or resource

qmanager/policies/base/queue_policy_base_impl.hpp

@@ -186,6 +186,11 @@ int queue_policy_base_t::insert (std::shared_ptr<job_t> job)
     return detail::queue_policy_base_impl_t::insert (job);
 }
 
+int queue_policy_base_t::remove_pending (job_t *job)
+{
+    return detail::queue_policy_base_impl_t::remove_pending (job);
+}
+
 int queue_policy_base_t::remove (flux_jobid_t id)
 {
     return detail::queue_policy_base_impl_t::remove (id);
@@ -336,6 +341,45 @@ int queue_policy_base_impl_t::insert (std::shared_ptr<job_t> job)
     return rc;
 }
 
+int queue_policy_base_impl_t::remove_pending (job_t *job) {
+    int rc = -1;
+
+    if (!job || job->state != job_state_kind_t::PENDING) {
+        errno = EINVAL;
+        return rc;
+    }
+
+    job->t_stamps.canceled_ts = m_cancel_cnt++;
+    if (is_sched_loop_active ()) {
+        // if sched-loop is active, the job's pending state
+        // cannot be determined. There is "MAYBE pending state" where
+        // a request has been sent out to the match service.
+        auto res = m_pending_cancel_provisional.insert (
+                        std::pair<uint64_t, flux_jobid_t> (
+                            job->t_stamps.canceled_ts, job->id));
+        if (!res.second) {
+            errno = EEXIST;
+            goto out;
+        }
+    } else {
+        bool found_in_provisional = false;
+        if (erase_pending_job (job, found_in_provisional) < 0)
+            goto out;
+        job->state = job_state_kind_t::CANCELED;
+        auto res = m_canceled.insert (
+                        std::pair<uint64_t, flux_jobid_t> (
+                            job->t_stamps.canceled_ts, job->id));
+        if (!res.second) {
+            errno = EEXIST;
+            goto out;
+        }
+        m_schedulable = true;
+    }
+    rc = 0;
+out:
+    return rc;
+}
+
 int queue_policy_base_impl_t::remove (flux_jobid_t id)
 {
     int rc = -1;
@@ -349,32 +393,7 @@ int queue_policy_base_impl_t::remove (flux_jobid_t id)
     job = m_jobs[id];
     switch (job->state) {
     case job_state_kind_t::PENDING:
-        job->t_stamps.canceled_ts = m_cancel_cnt++;
-        if (is_sched_loop_active ()) {
-            // if sched-loop is active, the job's pending state
-            // cannot be determined. There is "MAYBE pending state" where
-            // a request has been sent out to the match service.
-            auto res = m_pending_cancel_provisional.insert (
-                           std::pair<uint64_t, flux_jobid_t> (
-                               job->t_stamps.canceled_ts, job->id));
-            if (!res.second) {
-                errno = EEXIST;
-                goto out;
-            }
-        } else {
-            bool found_in_provisional = false;
-            if (erase_pending_job (job, found_in_provisional) < 0)
-                goto out;
-            job->state = job_state_kind_t::CANCELED;
-            auto res = m_canceled.insert (
-                           std::pair<uint64_t, flux_jobid_t> (
-                                job->t_stamps.canceled_ts, job->id));
-            if (!res.second) {
-                errno = EEXIST;
-                goto out;
-            }
-            m_schedulable = true;
-        }
+        this->remove_pending(job.get());
         break;
     case job_state_kind_t::ALLOC_RUNNING:
         m_alloced.erase (job->t_stamps.running_ts);
@@ -593,7 +612,7 @@ int queue_policy_base_impl_t::pending_reprioritize (flux_jobid_t id,
         }
     } else {
         bool found_in_prov = false;
-        if (erase_pending_job (job, found_in_prov) < 0)
+        if (erase_pending_job (job.get (), found_in_prov) < 0)
             return -1;
         job->priority = priority;
         if (insert_pending_job (job, found_in_prov) < 0)
@@ -614,7 +633,7 @@ int queue_policy_base_impl_t::process_provisional_cancel ()
         auto job = m_jobs[id];
         if (job->state == job_state_kind_t::PENDING) {
             bool found_in_provisional = false;
-            if (erase_pending_job (job, found_in_provisional) < 0)
+            if (erase_pending_job (job.get (), found_in_provisional) < 0)
                 return -1;
             job->state = job_state_kind_t::CANCELED;
             auto res = m_canceled.insert (
@@ -644,7 +663,7 @@ int queue_policy_base_impl_t::process_provisional_reprio ()
         auto job = m_jobs[id];
         if (job->state == job_state_kind_t::PENDING) {
             bool found_in_provisional = false;
-            if (erase_pending_job (job, found_in_provisional) < 0)
+            if (erase_pending_job (job.get (), found_in_provisional) < 0)
                 return -1;
             job->priority = priority;
             if (insert_pending_job (job, found_in_provisional) < 0)
@@ -685,7 +704,7 @@ int queue_policy_base_impl_t::insert_pending_job (std::shared_ptr<job_t> &job,
     return 0;
 }
 
-int queue_policy_base_impl_t::erase_pending_job (std::shared_ptr<job_t> &job,
+int queue_policy_base_impl_t::erase_pending_job (job_t *job,
                                                  bool &found_in_prov)
 {
     size_t s;

t/t1024-alloc-check.t

@@ -60,9 +60,33 @@ test_expect_success 'some jobs received timeout exception' '
 test_expect_success 'no jobs received alloc-check exception' '
 	test_must_fail grep "job.exception type=alloc-check" joberr2
 '
+test_expect_success 'clean up' '
+	flux cancel --all &&
+	flux queue idle &&
+	(flux resource undrain 0 || true)
+'
+
+send_sched_cancel() {
+	local JOB_ID=$1
+	shift
+	flux python -c "import flux; from flux.job.JobID import id_parse; flux.Flux().rpc('sched.cancel', {'id': id_parse('$JOB_ID')})"
+}
+
+# ensure sched.cancel doesn't free resources when an epilog is pending
+test_expect_success 'submit a job that cannot run, cancel it during epilog, submit another ' '
+    (flux submit -N 1 --flags=waitable --wait-event epilog-start -c 4 /command/that/does/not/exist > ji1 || true ) &&
+	send_sched_cancel $(cat ji1) &&
+    flux submit -N 1 --exclusive hostname > ji2 &&
+	(flux job wait-event $(cat ji1) epilog-finish || true) &&
+	(flux job info $(cat ji1) eventlog | grep epilog-finish | jq ".timestamp" > time1) &&
+	(flux job info $(cat ji2) eventlog | grep alloc | jq ".timestamp" > time2) &&
+	awk -vt1=$(cat time1) -vt2=$(cat time2) "BEGIN {exit (t1 < t2) ? 0 : 1}"
+'
+
 test_expect_success 'clean up' '
 	cleanup_active_jobs
 '
+
 test_expect_success 'remove fluxion modules' '
 	remove_qmanager &&
 	remove_resource