Index: uspace/lib/sysman/include/sysman/broker.h =================================================================== --- uspace/lib/sysman/include/sysman/broker.h (revision c2d50c8933899c5ca8fedfaee5dd4803e701c43d) +++ uspace/lib/sysman/include/sysman/broker.h (revision 72c8f77fe49645071c65e51c20256117bc69eb5c) @@ -30,6 +30,6 @@ #define _SYSMAN_BROKER_H +#include #include -#include int sysman_broker_register(void); Index: uspace/srv/sysman/job.c =================================================================== --- uspace/srv/sysman/job.c (revision c2d50c8933899c5ca8fedfaee5dd4803e701c43d) +++ uspace/srv/sysman/job.c (revision 72c8f77fe49645071c65e51c20256117bc69eb5c) @@ -46,5 +46,8 @@ 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,4 +56,5 @@ job_add_ref(blocked_job); + blocking_job->blocked_jobs_count += 1; blocked_job->blocking_jobs += 1; @@ -86,12 +90,9 @@ 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,4 +101,5 @@ { unit_t *u = job->unit; + if (u->state == job->target_state) { job->retval = JOB_OK; @@ -117,5 +119,5 @@ /* - * 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,5 +140,5 @@ 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,5 +151,6 @@ 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,8 +173,53 @@ /* 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,27 +233,87 @@ } -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,53 +340,39 @@ 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,23 +380,28 @@ 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,10 +409,8 @@ /** 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,7 +459,7 @@ 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,5 +474,11 @@ 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,5 +515,5 @@ 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,5 +521,6 @@ /* 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); } Index: uspace/srv/sysman/job.h =================================================================== --- uspace/srv/sysman/job.h (revision c2d50c8933899c5ca8fedfaee5dd4803e701c43d) +++ uspace/srv/sysman/job.h (revision 72c8f77fe49645071c65e51c20256117bc69eb5c) @@ -37,10 +37,9 @@ #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,4 +53,7 @@ } job_retval_t; +struct job; +typedef struct job job_t; + struct job { link_t job_queue; @@ -63,8 +65,13 @@ /** 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,9 +81,6 @@ }; -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); Index: uspace/srv/sysman/sysman.c =================================================================== --- uspace/srv/sysman/sysman.c (revision c2d50c8933899c5ca8fedfaee5dd4803e701c43d) +++ uspace/srv/sysman/sysman.c (revision 72c8f77fe49645071c65e51c20256117bc69eb5c) @@ -69,5 +69,7 @@ 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,4 +102,32 @@ }; +/* + * 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,4 +146,6 @@ free(callback); } + + observed_object_destroy(&observed_object); } @@ -174,20 +206,7 @@ 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,8 +216,6 @@ } - 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,5 +259,7 @@ * * 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,14 +272,10 @@ 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,4 +299,34 @@ } +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,5 +338,5 @@ 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,5 +351,5 @@ * 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,5 +366,5 @@ 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)); } Index: uspace/srv/sysman/sysman.h =================================================================== --- uspace/srv/sysman/sysman.h (revision c2d50c8933899c5ca8fedfaee5dd4803e701c43d) +++ uspace/srv/sysman/sysman.h (revision 72c8f77fe49645071c65e51c20256117bc69eb5c) @@ -44,4 +44,5 @@ extern void sysman_process_queue(void); extern int sysman_object_observer(void *, callback_handler_t, void *); +extern int sysman_move_observers(void *, void *); Index: uspace/srv/sysman/test/job_queue.c =================================================================== --- uspace/srv/sysman/test/job_queue.c (revision c2d50c8933899c5ca8fedfaee5dd4803e701c43d) +++ uspace/srv/sysman/test/job_queue.c (revision 72c8f77fe49645071c65e51c20256117bc69eb5c) @@ -42,11 +42,5 @@ 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,22 +50,4 @@ *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,5 +83,5 @@ 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,6 +102,5 @@ 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,4 +117,39 @@ } +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);