qmanager: split remove to separate pending

problem: The "remove" method was used for both handling the `free` and
`cancel` RPCs. Under some code paths, remove de-allocates resources, so
using it for a pending cancellation, even if it seems to be protected by
checks, is unnecessarily dangerous.  Given the recent double-booking
issue and a lack so far of other theories on how we get there, this is a
possible culprit.

solution: Re-factor queue_policy_base_t remove into remove and
remove_pending, the latter of which only operates on the pending states
and has no code-paths that can result in a resource.cancel RPC being
sent.  The remove_pending is used for both the cancel RPC and the part
of the free RPC where the code was used before, should be a pure
refactor.

Author: trws

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;