Changeset 72c8f77 in mainline

Timestamp:

2019-08-06T18:30:29Z (5 years ago)

Author:

Matthieu Riolo <matthieu.riolo@…>

Children:

63a3276

Parents:

c2d50c8

git-author:

Michal Koutný <xm.koutny+hos@…> (2015-05-26 16:15:58)

git-committer:

Matthieu Riolo <matthieu.riolo@…> (2019-08-06 18:30:29)

Message:

sysman: Separate job closure creation and its enqueuing

• added merge job operation
• created test that failed with previous implementation of job closure

Location:

uspace

Files:

• lib/sysman/include/sysman/broker.h (modified) (1 diff)
• srv/sysman/job.c (modified) (16 diffs)
• srv/sysman/job.h (modified) (4 diffs)
• srv/sysman/sysman.c (modified) (11 diffs)
• srv/sysman/sysman.h (modified) (1 diff)
• srv/sysman/test/job_queue.c (modified) (5 diffs)

uspace/lib/sysman/include/sysman/broker.h

@@ -30 +30 @@
 #define _SYSMAN_BROKER_H
 
+#include <sysman/unit.h>
 #include <task.h>
-#include <sysman/unit.h>
 
 int sysman_broker_register(void);

uspace/srv/sysman/job.c

@@ -46 +46 @@
 static int job_add_blocked_job(job_t *blocking_job, job_t *blocked_job)
 {
-        int rc = dyn_array_append(&blocking_job->blocked_jobs, job_ptr_t,
+        assert(blocking_job->blocked_jobs.size ==
+            blocking_job->blocked_jobs_count);
+
+        int rc = dyn_array_append(&blocking_job->blocked_jobs, job_t *,
             blocked_job);
         if (rc != EOK) {
@@ -53 +56 @@
         job_add_ref(blocked_job);
 
+        blocking_job->blocked_jobs_count += 1;
         blocked_job->blocking_jobs += 1;
 
@@ -86 +90 @@
         job->unit = u;
 
-        u->job = job;
-        job_add_ref(job);
-
-        dyn_array_initialize(&job->blocked_jobs, job_ptr_t);
+        dyn_array_initialize(&job->blocked_jobs, job_t *);
         job->blocking_jobs = 0;
         job->blocking_job_failed = false;
 
-        job->state = JOB_UNQUEUED;
+        job->state = JOB_EMBRYO;
         job->retval = JOB_UNDEFINED_;
 }
@@ -100 +101 @@
 {
         unit_t *u = job->unit;
+
         if (u->state == job->target_state) {
                 job->retval = JOB_OK;
@@ -117 +119 @@
 
         /*
-         * We have one reference from caller for our disposal,   *
+         * We have one reference from caller for our disposal,
          * if needed, pass it to observer.
          */
@@ -138 +140 @@
         assert(!link_used(&job->job_queue));
 
-        dyn_array_foreach(job->blocked_jobs, job_ptr_t, job_it) {
+        dyn_array_foreach(job->blocked_jobs, job_t *, job_it) {
                 job_del_ref(&(*job_it));
         }
@@ -149 +151 @@
 static bool job_is_runnable(job_t *job)
 {
-        return job->state == JOB_QUEUED && job->blocking_jobs == 0;
+        assert(job->state == JOB_PENDING);
+        return job->blocking_jobs == 0;
 }
 
@@ -170 +173 @@
                 /* Remove job from queue and pass reference to caller */
                 list_remove(&result->job_queue);
-                result->state = JOB_DEQUEUED;
         }
 
         return result;
+}
+
+/** Merge two jobs together
+ *
+ * @param[in/out]  trunk  job that
+ * @param[in]      other  job that will be cleared out
+ *
+ * @return EOK on success
+ * @return error code on fail
+ */
+static int job_pre_merge(job_t *trunk, job_t *other)
+{
+        assert(trunk->unit == other->unit);
+        assert(trunk->target_state == other->target_state);
+        assert(trunk->blocked_jobs.size == trunk->blocked_jobs_count);
+        assert(other->merged_into == NULL);
+
+        int rc = dyn_array_concat(&trunk->blocked_jobs, &other->blocked_jobs);
+        if (rc != EOK) {
+                return rc;
+        }
+        dyn_array_clear(&other->blocked_jobs);
+
+        // TODO allocate observed object
+
+        other->merged_into = trunk;
+
+        return EOK;
+}
+
+static void job_finish_merge(job_t *trunk, job_t *other)
+{
+        assert(trunk->blocked_jobs.size >= trunk->blocked_jobs_count);
+        //TODO aggregate merged blocked_jobs
+        trunk->blocked_jobs_count = other->blocked_jobs.size;
+
+        /* All allocation is done in job_pre_merge, cannot fail here. */
+        int rc = sysman_move_observers(other, trunk);
+        assert(rc == EOK);
+}
+
+static void job_undo_merge(job_t *trunk)
+{
+        assert(trunk->blocked_jobs.size >= trunk->blocked_jobs_count);
+        dyn_array_clear_range(&trunk->blocked_jobs,
+            trunk->blocked_jobs_count, trunk->blocked_jobs.size);
 }
 
@@ -185 +233 @@
 }
 
-int job_queue_add_jobs(dyn_array_t *jobs)
-{
-        /* Check consistency with queue. */
-        dyn_array_foreach(*jobs, job_ptr_t, new_job_it) {
-                list_foreach(job_queue, job_queue, job_t, queued_job) {
-                        /*
-                         * Currently we have strict strategy not permitting
-                         * multiple jobs for one unit in the queue at a time.
-                         */
-                        if ((*new_job_it)->unit == queued_job->unit) {
+int job_queue_add_closure(dyn_array_t *closure)
+{
+        bool has_error = false;
+        int rc = EOK;
+
+        /* Check consistency with existing jobs. */
+        dyn_array_foreach(*closure, job_t *, job_it) {
+                job_t *job = *job_it;
+                job_t *other_job = job->unit->job;
+
+                if (other_job == NULL) {
+                        continue;
+                }
+
+                if (other_job->target_state != job->target_state) {
+                        switch (other_job->state) {
+                        case JOB_RUNNING:
                                 sysman_log(LVL_ERROR,
-                                    "Cannot queue multiple jobs for unit '%s'",
-                                    unit_name((*new_job_it)->unit));
-                                return EEXIST;
+                                    "Unit '%s' has already different job running.",
+                                    unit_name(job->unit));
+                                has_error = true;
+                                continue;
+                        case JOB_PENDING:
+                                /*
+                                 * Currently we have strict strategy not
+                                 * permitting multiple jobs for one unit in the
+                                 * queue at a time.
+                                 */
+                                sysman_log(LVL_ERROR,
+                                    "Cannot queue multiple jobs for unit '%s'.",
+                                    unit_name(job->unit));
+                                has_error = true;
+                                continue;
+                        default:
+                                assert(false);
                         }
-                }
-        }
-
-        /* Enqueue jobs */
-        dyn_array_foreach(*jobs, job_ptr_t, job_it) {
-                (*job_it)->state = JOB_QUEUED;
-                list_append(&(*job_it)->job_queue, &job_queue);
+                } else {
+                        // TODO think about other options to merging
+                        //      (replacing, cancelling)
+                        rc = job_pre_merge(other_job, job);
+                        if (rc != EOK) {
+                                break;
+                        }
+                }
+        }
+
+        /* Aggregate merged jobs, or rollback any changes in existing jobs */
+        bool finish_merge = (rc == EOK) && !has_error;
+        dyn_array_foreach(*closure, job_t *, job_it) {
+                if ((*job_it)->merged_into == NULL) {
+                        continue;
+                }
+                if (finish_merge) {
+                        job_finish_merge((*job_it)->merged_into, *job_it);
+                } else {
+                        job_undo_merge((*job_it)->merged_into);
+                }
+        }
+        if (has_error) {
+                return EBUSY;
+        } else if (rc != EOK) {
+                return rc;
+        }
+
+        /* Unmerged jobs are enqueued, merged are disposed */
+        dyn_array_foreach(*closure, job_t *, job_it) {
+                job_t *job = (*job_it);
+                if (job->merged_into != NULL) {
+                        job_del_ref(&job);
+                        continue;
+                }
+
+
+                unit_t *u = job->unit;
+                assert(u->bfs_job != NULL);
+                assert(u->job == NULL);
+                u->job = u->bfs_job;
+                u->bfs_job = NULL;
+
+
+                job->state = JOB_PENDING;
                 /* We pass reference from the closure to the queue */
+                list_append(&job->job_queue, &job_queue);
         }
 
@@ -232 +340 @@
 int job_create_closure(job_t *main_job, dyn_array_t *job_closure)
 {
+        sysman_log(LVL_DEBUG2, "%s(%s)", __func__, unit_name(main_job->unit));
         int rc;
         list_t units_fifo;
         list_initialize(&units_fifo);
 
-        /* Zero BFS tags before use */
+        /* Check invariant */
         list_foreach(units, units, unit_t, u) {
-                u->bfs_tag = false;
+                assert(u->bfs_job == false);
         }
                 
         unit_t *unit = main_job->unit;
+        job_add_ref(main_job);
+        unit->bfs_job = main_job;
         list_append(&unit->bfs_link, &units_fifo);
-        unit->bfs_tag = true;
         
         while (!list_empty(&units_fifo)) {
                 unit = list_get_instance(list_first(&units_fifo), unit_t,
                     bfs_link);
-                assert(unit->job);
                 list_remove(&unit->bfs_link);
-                job_t *job = unit->job;
-
-
-                // TODO more sophisticated check? (unit that is in transitional
-                //      state cannot have currently multiple jobs queued)
-                if (job->target_state == unit->state) {
-                        /*
-                         * Job would do nothing, finish it on spot.
-                         * No need to continue BFS search from it.
-                         */
-                        job->retval = JOB_OK;
-                        job_finish(job);
-                        continue;
-                }
+                job_t *job = unit->bfs_job;
+                assert(job != NULL);
 
                 job_add_ref(job);
-                dyn_array_append(job_closure, job_ptr_t, job);
+                dyn_array_append(job_closure, job_t *, job);
 
                 /*
                  * Traverse dependencies edges
-                 * Depending on dependency type and edge direction create
-                 * appropriate jobs.
+                 * According to dependency type and edge direction create
+                 * appropriate jobs (currently "After" only).
                  */
                 list_foreach(unit->dependencies, dependencies, unit_dependency_t, dep) {
                         unit_t *u = dep->dependency;
                         job_t *blocking_job;
-                        if (u->bfs_tag) {
-                                assert(u->job);
-                                blocking_job = u->job;
-                        } else {
-                                u->bfs_tag = true;
+
+                        if (u->bfs_job == NULL) {
                                 blocking_job = job_create(u, job->target_state);
                                 if (blocking_job == NULL) {
@@ -286 +380 @@
                                         goto finish;
                                 }
-                                /* Reference to job is kept in unit */
-                                job_del_ref(&blocking_job);
+                                /* Pass reference to unit */
+                                u->bfs_job = blocking_job;
                                 list_append(&u->bfs_link, &units_fifo);
+                        } else {
+                                blocking_job = u->bfs_job;
                         }
-                        
+
                         job_add_blocked_job(blocking_job, job);
                 }
         }
-        sysman_log(LVL_DEBUG2, "%s(%s):", __func__, unit_name(main_job->unit));
-        dyn_array_foreach(*job_closure, job_ptr_t, job_it) {
-                sysman_log(LVL_DEBUG2, "%s\t%s", __func__, unit_name((*job_it)->unit));
-        }
+//      sysman_log(LVL_DEBUG2, "%s(%s):", __func__, unit_name(main_job->unit));
+//      dyn_array_foreach(*job_closure, job_t *, job_it) {
+//              sysman_log(LVL_DEBUG2, "%s\t%s", __func__, unit_name((*job_it)->unit));
+//      }
         rc = EOK;
 
 finish:
-        /* Any unprocessed jobs may be referenced by units */
-        list_foreach(units_fifo, bfs_link, unit_t, u) {
-                job_del_ref(&u->job);
-        }
+        /* Unreference any jobs in interrupted BFS queue */
+        list_foreach_safe(units_fifo, cur_link, next_link) {
+                unit_t *u = list_get_instance(cur_link, unit_t, bfs_link);
+                job_del_ref(&u->bfs_job);
+                list_remove(cur_link);
+        }
+
         return rc;
 }
@@ -310 +409 @@
 /** Create job assigned to the unit
  *
- * @param[in]  unit          unit to be modified, its job must be empty
+ * @param[in]  unit
  * @param[in]  target_state
  *
- * @return NULL or newly created job
- *         There are two references to the job, one set in the unit and second
- *         is the return value.
+ * @return NULL or newly created job (there is a single refernce for the creator)
  */
 job_t *job_create(unit_t *u, unit_state_t target_state)
@@ -362 +459 @@
 void job_run(job_t *job)
 {
-        assert(job->state != JOB_RUNNING);
-        assert(job->state != JOB_FINISHED);
-
+        assert(job->state == JOB_PENDING);
+
+        job->state = JOB_RUNNING;
         unit_t *u = job->unit;
         sysman_log(LVL_DEBUG, "%s(%p), %s -> %i",
@@ -377 +474 @@
         switch (job->target_state) {
         case STATE_STARTED:
-                rc = unit_start(u);
+                // TODO put here same evaluation as in job_check
+                //      goal is to have job_run "idempotent"
+                if (u->state == job->target_state) {
+                        rc = EOK;
+                } else {
+                        rc = unit_start(u);
+                }
                 break;
         default:
@@ -412 +515 @@
         assert(job->unit->job == job);
 
-        sysman_log(LVL_DEBUG2, "%s(%p) %s -> %i",
+        sysman_log(LVL_DEBUG2, "%s(%p) %s ret %i",
             __func__, job, unit_name(job->unit), job->retval);
 
@@ -418 +521 @@
 
         /* First remove references, then clear the array */
-        dyn_array_foreach(job->blocked_jobs, job_ptr_t, job_it) {
+        assert(job->blocked_jobs.size == job->blocked_jobs_count);
+        dyn_array_foreach(job->blocked_jobs, job_t *, job_it) {
                 job_unblock(*job_it, job);
         }

uspace/srv/sysman/job.h

@@ -37 +37 @@
 #include "unit.h"
 
-// TODO simplify queue states
 /** Run state of job */
 typedef enum {
-        JOB_UNQUEUED, /**< Job not in queue yet */
-        JOB_QUEUED,
-        JOB_DEQUEUED, /**< Job not in queue already */
+        JOB_EMBRYO, /**< Job after creation */
+        JOB_CLOSURED, /**< Intermmediate when closure is evaluated */
+        JOB_PENDING, /**< Job is queued */
         JOB_RUNNING,
         JOB_FINISHED
@@ -54 +53 @@
 } job_retval_t;
 
+struct job;
+typedef struct job job_t;
+
 struct job {
         link_t job_queue;
@@ -63 +65 @@
         /** Jobs that this job is preventing from running */
         dyn_array_t blocked_jobs;
+        /** No. of jobs that the job is actually blocking (may differ from size
+         * of blocked_jobs for not fully merged job */
+        size_t blocked_jobs_count;
         /** No. of jobs that must finish before this job */
         size_t blocking_jobs;
         /** Any of blocking jobs failed */
         bool blocking_job_failed;
+        /** Job that this job was merged to */
+        job_t *merged_into;
 
         /** See job_state_t */
@@ -74 +81 @@
 };
 
-typedef struct job job_t;
-typedef job_t *job_ptr_t;
-
 extern void job_queue_init(void);
-extern int job_queue_add_jobs(dyn_array_t *);
+extern int job_queue_add_closure(dyn_array_t *);
 extern void job_queue_process(void);
 

uspace/srv/sysman/sysman.c

@@ -69 +69 @@
 static fibril_condvar_t observed_objects_cv;
 
-/* Hash table functions */
+/*
+ * Hash table functions
+ */
 static size_t observed_objects_ht_hash(const ht_link_t *item)
 {
@@ -100 +102 @@
 };
 
+/*
+ * Static functions
+ */
+
+static observed_object_t *observed_object_create(void *object)
+{
+        observed_object_t *result = malloc(sizeof(observed_object_t));
+        if (result) {
+                result->object = object;
+                list_initialize(&result->callbacks);
+                hash_table_insert(&observed_objects, &result->ht_link);
+        }
+        return result;
+}
+
+static void observed_object_destroy(observed_object_t **ptr_observed_object)
+{
+        observed_object_t *observed_object = *ptr_observed_object;
+        if (observed_object == NULL) {
+                return;
+        }
+
+        ht_link_t *item = &observed_object->ht_link;
+        hash_table_remove_item(&observed_objects, item);
+        free(observed_object);
+        *ptr_observed_object = NULL;
+}
+
 static void notify_observers(void *object)
 {
@@ -116 +146 @@
                 free(callback);
         }
+
+        observed_object_destroy(&observed_object);
 }
 
@@ -174 +206 @@
     callback_handler_t callback, void *callback_arg)
 {
-        job_t *job;
-
-        if (unit->job != NULL) {
-                assert(unit->job->state != JOB_UNQUEUED);
-
-                if (unit->job->target_state != target_state) {
-                        return EBUSY;
-                }
-                job = unit->job;
-        } else {
-                job = job_create(unit, target_state);
-                if (job == NULL) {
-                        return ENOMEM;
-                }
-                /* Reference in unit is enough */
-                job_del_ref(&job);
+        job_t *job = job_create(unit, target_state);
+        if (job == NULL) {
+                return ENOMEM;
         }
 
@@ -197 +216 @@
         }
 
-        if (job->state == JOB_UNQUEUED) {
-                job_add_ref(job);
-                sysman_raise_event(&sysman_event_job_process, job);
-        }
+        /* Pass reference to event */
+        sysman_raise_event(&sysman_event_job_process, job);
 
         return EOK;
@@ -242 +259 @@
  *
  * TODO no one handles return value, it's quite fatal to lack memory for
- *      callbacks...  @return EOK on success
+ *      callbacks...
+ *
+ * @return EOK on success
  * @return ENOMEM
  */
@@ -253 +272 @@
 
         if (ht_link == NULL) {
-                observed_object = malloc(sizeof(observed_object_t));
-                if (observed_object == NULL) {
+                new_observed_object = observed_object_create(object);
+                if (new_observed_object == NULL) {
                         rc = ENOMEM;
                         goto fail;
                 }
-                new_observed_object = observed_object;
-
-                observed_object->object = object;
-                list_initialize(&observed_object->callbacks);
-                hash_table_insert(&observed_objects, &observed_object->ht_link);
+                observed_object = new_observed_object;
         } else {
                 observed_object =
@@ -284 +299 @@
 }
 
+int sysman_move_observers(void *src_object, void *dst_object)
+{
+        ht_link_t *src_link = hash_table_find(&observed_objects, &src_object);
+        if (src_link == NULL) {
+                return EOK;
+        }
+
+        ht_link_t *dst_link = hash_table_find(&observed_objects, &dst_object);
+        observed_object_t *dst_observed_object;
+        if (dst_link == NULL) {
+                dst_observed_object = observed_object_create(dst_object);
+                if (dst_observed_object == NULL) {
+                        return ENOMEM;
+                }
+        } else {
+                dst_observed_object =
+                    hash_table_get_inst(dst_link, observed_object_t, ht_link);
+        }
+
+        observed_object_t *src_observed_object =
+            hash_table_get_inst(src_link, observed_object_t, ht_link);
+
+        list_concat(&dst_observed_object->callbacks,
+            &src_observed_object->callbacks);
+        observed_object_destroy(&src_observed_object);
+
+        return EOK;
+}
+
+
 /*
  * Event handlers
@@ -293 +338 @@
         job_t *job = data;
         dyn_array_t job_closure;
-        dyn_array_initialize(&job_closure, job_ptr_t);
+        dyn_array_initialize(&job_closure, job_t *);
 
         int rc = job_create_closure(job, &job_closure);
@@ -306 +351 @@
          * otherwise, we still have the reference.
          */
-        rc = job_queue_add_jobs(&job_closure);
+        rc = job_queue_add_closure(&job_closure);
         if (rc != EOK) {
                 goto fail;
@@ -321 +366 @@
         job_del_ref(&job);
 
-        dyn_array_foreach(job_closure, job_ptr_t, closure_job) {
+        dyn_array_foreach(job_closure, job_t *, closure_job) {
                 job_del_ref(&(*closure_job));
         }

uspace/srv/sysman/sysman.h

@@ -44 +44 @@
 extern void sysman_process_queue(void);
 extern int sysman_object_observer(void *, callback_handler_t, void *);
+extern int sysman_move_observers(void *, void *);
 
 

uspace/srv/sysman/test/job_queue.c

@@ -42 +42 @@
 static fid_t fibril_event_loop;
 
-#if 0
-static bool async_finished;
-static fibril_condvar_t async_finished_cv;
-static fibril_mutex_t async_finished_mtx;
-#endif
-
-static void async_finished_callback(void *object, void *arg)
+static void job_finished_cb(void *object, void *arg)
 {
         job_t *job = object;
@@ -56 +50 @@
         *job_ptr = job;
 }
-
-#if 0
-static void reset_wait(void)
-{
-        fibril_mutex_lock(&async_finished_mtx);
-        async_finished = false;
-        fibril_mutex_unlock(&async_finished_mtx);
-}
-
-static void async_wait()
-{
-        fibril_mutex_lock(&async_finished_mtx);
-        while (!async_finished) {
-                fibril_condvar_wait(&async_finished_cv, &async_finished_mtx);
-        }
-        fibril_mutex_unlock(&async_finished_mtx);
-}
-#endif
 
 PCUT_TEST_SUITE(job_queue);
@@ -107 +83 @@
         job_t *job = NULL;
 
-        int rc = sysman_run_job(u, STATE_STARTED, &async_finished_callback,
+        int rc = sysman_run_job(u, STATE_STARTED, &job_finished_cb,
             &job);
         PCUT_ASSERT_INT_EQUALS(EOK, rc);
@@ -126 +102 @@
         job_t *job = NULL;
 
-        int rc = sysman_run_job(u, STATE_STARTED, &async_finished_callback,
-            &job);
+        int rc = sysman_run_job(u, STATE_STARTED, &job_finished_cb, &job);
         PCUT_ASSERT_INT_EQUALS(EOK, rc);
 
@@ -142 +117 @@
 }
 
+PCUT_TEST(multipath_to_started_unit) {
+        /* Setup mock behavior */
+        unit_type_vmts[UNIT_SERVICE]->start = &mock_unit_vmt_start_sync;
+
+        unit_type_vmts[UNIT_MOUNT]->start = &mock_unit_vmt_start_async;
+        unit_type_vmts[UNIT_MOUNT]->exposee_created =
+            &mock_unit_vmt_exposee_created;
+
+        /* Define mock units */
+        unit_t *s0 = mock_units[UNIT_SERVICE][0];
+        unit_t *s1 = mock_units[UNIT_SERVICE][1];
+        unit_t *m0 = mock_units[UNIT_MOUNT][0];
+
+        /* All services require root fs */
+        mock_add_dependency(s0, m0);
+        mock_add_dependency(s1, m0);
+        
+        /* S1 requires another mount and S0 */
+        mock_add_dependency(s1, s0);
+
+        /* Enforce initial state */
+        m0->state = STATE_STARTED;
+
+        /* Run test */
+        job_t *job = NULL;
+        int rc = sysman_run_job(s1, STATE_STARTED, &job_finished_cb, &job);
+        PCUT_ASSERT_INT_EQUALS(EOK, rc);
+
+        sysman_process_queue();
+
+        PCUT_ASSERT_NOT_NULL(job);
+        PCUT_ASSERT_EQUALS(STATE_STARTED, s0->state);
+        PCUT_ASSERT_EQUALS(STATE_STARTED, s1->state);
+}
+
 
 PCUT_EXPORT(job_queue);