Changeset 25a9fec in mainline

Timestamp:

2019-08-07T09:54:10Z (6 years ago)

Author:

Matthieu Riolo <matthieu.riolo@…>

Children:

92a7cfb1

Parents:

18377301

git-author:

Michal Koutný <xm.koutny+hos@…> (2015-11-03 21:48:45)

git-committer:

Matthieu Riolo <matthieu.riolo@…> (2019-08-07 09:54:10)

Message:

sysman: Refactor job.c into job_queue.c and job_closure.c

Location:

uspace/srv/sysman

Files:

• Makefile (modified) (1 diff)
• job.c (modified) (3 diffs)
• job.h (modified) (1 diff)
• job_closure.c (added)
• job_closure.h (added)
• job_queue.c (added)
• job_queue.h (added)
• main.c (modified) (1 diff)
• sysman.c (modified) (1 diff)
• test/job_closure.c (modified) (1 diff)
• test/job_queue.c (modified) (1 diff)

uspace/srv/sysman/Makefile

@@ -42 +42 @@
         edge.c \
         job.c \
+        job_closure.c \
+        job_queue.c \
         log.c \
         repo.c \

uspace/srv/sysman/job.c

@@ -38 +38 @@
 #include "sysman.h"
 
-static list_t job_queue;
 
 /*
@@ -44 +43 @@
  */
 
-static int job_add_blocked_job(job_t *blocking_job, job_t *blocked_job)
-{
-        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) {
-                return ENOMEM;
-        }
-        job_add_ref(blocked_job);
-
-        blocking_job->blocked_jobs_count += 1;
-        blocked_job->blocking_jobs += 1;
-
-        return EOK;
-}
 
 /** Remove blocking_job from blocked job structure
@@ -148 +130 @@
 }
 
-static bool job_is_runnable(job_t *job)
-{
-        assert(job->state == JOB_PENDING);
-        return job->blocking_jobs == 0;
-}
-
-/** Pop next runnable job
- *
- * @return runnable job or NULL when there's none
- */
-static job_t *job_queue_pop_runnable(void)
-{
-        job_t *result = NULL;
-
-        /* Select first runnable job */
-        list_foreach(job_queue, job_queue, job_t, candidate) {
-                if (job_is_runnable(candidate)) {
-                        result = candidate;
-                        break;
-                }
-        }
-        if (result) {
-                /* Remove job from queue and pass reference to caller */
-                list_remove(&result->job_queue);
-        }
-
-        return result;
-}
-
-/** Add multiple references to job
- *
- * Non-atomicity doesn't mind as long as individual increments are atomic.
- *
- * @note Function is not exported as other modules shouldn't need it.
- */
-static inline void job_add_refs(job_t *job, size_t refs)
-{
-        for (size_t i = 0; i < refs; ++i) {
-                job_add_ref(job);
-        }
-}
-
-/** Delete multiple references to job
- *
- * Behavior of concurrent runs with job_add_refs aren't specified.
- */
-static inline void job_del_refs(job_t **job_ptr, size_t refs)
-{
-        for (size_t i = 0; i < refs; ++i) {
-                job_del_ref(job_ptr);
-        }
-}
-
-/** 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;
-
-        /*
-         * Note, the sysman_move_observers cannot fail here sice all necessary
-         * allocation is done in job_pre_merge.
-         */
-        size_t observers_refs = sysman_observers_count(other);
-        int rc = sysman_move_observers(other, trunk);
-        assert(rc == EOK);
-
-        /* When we move observers, don't forget to pass their references too. */
-        job_add_refs(trunk, observers_refs);
-        job_del_refs(&other, observers_refs);
-}
-
-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);
-}
-
 /*
  * Non-static functions
  */
-
-void job_queue_init()
-{
-        list_initialize(&job_queue);
-}
-
-/** Consistenly add jobs to the queue
- *
- * @param[in/out]  closure    jobs closure, on success it's emptied, otherwise
- *                            you should take care of remaining jobs
- *
- * @return EOK on success
- * @return EBUSY when any job in closure is conflicting
- */
-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,
-                                    "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);
-                        }
-                } 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->job == NULL);
-                /* Pass reference from the closure to the unit */
-                u->job = job;
-
-                /* Enqueue job (new reference) */
-                job->state = JOB_PENDING;
-                job_add_ref(job);
-                list_append(&job->job_queue, &job_queue);
-        }
-
-        /* We've stolen references from the closure, so erase it */
-        dyn_array_clear(closure);
-
-        return EOK;
-}
-
-/** Process all jobs that aren't transitively blocked
- *
- * Job can be blocked either by another job or by an incoming event, that will
- * be queued after this job_queue_process call.
- *
- * TODO Write down rules from where this function can be called, to avoid stack
- *      overflow.
- */
-void job_queue_process(void)
-{
-        job_t *job;
-        while ((job = job_queue_pop_runnable())) {
-                job_run(job);
-                job_del_ref(&job);
-        }
-}
-
-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);
-
-        /* Check invariant */
-        list_foreach(units, units, unit_t, u) {
-                assert(u->bfs_job == NULL);
-        }
-                
-        unit_t *unit = main_job->unit;
-        job_add_ref(main_job);
-        unit->bfs_job = main_job;
-        list_append(&unit->bfs_link, &units_fifo);
-        
-        while (!list_empty(&units_fifo)) {
-                unit = list_get_instance(list_first(&units_fifo), unit_t,
-                    bfs_link);
-                list_remove(&unit->bfs_link);
-                job_t *job = unit->bfs_job;
-                assert(job != NULL);
-
-                job_add_ref(job);
-                dyn_array_append(job_closure, job_t *, job);
-
-                /*
-                 * Traverse dependencies edges
-                 * According to dependency type and edge direction create
-                 * appropriate jobs (currently "After" only).
-                 */
-                list_foreach(unit->edges_out, edges_out, unit_edge_t, e) {
-                        unit_t *u = e->output;
-                        job_t *blocking_job;
-
-                        if (u->bfs_job == NULL) {
-                                blocking_job = job_create(u, job->target_state);
-                                if (blocking_job == NULL) {
-                                        rc = ENOMEM;
-                                        goto finish;
-                                }
-                                /* 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_t *, job_it) {
-                sysman_log(LVL_DEBUG2, "%s\t%s, refs: %u", __func__,
-                    unit_name((*job_it)->unit), atomic_get(&(*job_it)->refcnt));
-        }
-        rc = EOK;
-
-finish:
-        /* 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);
-        }
-        
-        /* Clean after ourselves (BFS tag jobs) */
-        dyn_array_foreach(*job_closure, job_t *, job_it) {
-                assert(*job_it == (*job_it)->unit->bfs_job);
-                job_del_ref(&(*job_it)->unit->bfs_job);
-                (*job_it)->unit->bfs_job = NULL;
-        }
-
-        return rc;
-}
 
 /** Create job assigned to the unit

uspace/srv/sysman/job.h

@@ -81 +81 @@
 };
 
-extern void job_queue_init(void);
-extern int job_queue_add_closure(dyn_array_t *);
-extern void job_queue_process(void);
-
-extern int job_create_closure(job_t *, dyn_array_t *);
 extern job_t *job_create(unit_t *, unit_state_t);
 

uspace/srv/sysman/main.c

@@ -41 +41 @@
 #include "connection_ctl.h"
 #include "edge.h"
-#include "job.h"
+#include "job_queue.h"
 #include "log.h"
 #include "sysman.h"

uspace/srv/sysman/sysman.c

@@ -33 +33 @@
 #include <stdlib.h>
 
+#include "job_closure.h"
+#include "job_queue.h"
 #include "log.h"
 #include "sysman.h"

uspace/srv/sysman/test/job_closure.c

@@ -31 +31 @@
 #include <stdio.h>
 
-#include "../job.h"
+#include "../job_closure.h"
 
 #include "mock_unit.h"

uspace/srv/sysman/test/job_queue.c

@@ -33 +33 @@
 
 #include "mock_unit.h"
-#include "../job.h"
+#include "../job_queue.h"
 #include "../sysman.h"