Index: uspace/Makefile =================================================================== --- uspace/Makefile (revision d7c5fc0ed63066056e62be12b37714ab838e915b) +++ uspace/Makefile (revision 3f7e1f243981a64f3cb89f89ffb7c6189b539f5c) @@ -118,6 +118,6 @@ srv/net/udp \ srv/ns \ + srv/taskmon \ srv/sysman \ - srv/taskmon \ srv/vfs \ srv/bd/sata_bd \ Index: uspace/srv/sysman/configuration.c =================================================================== --- uspace/srv/sysman/configuration.c (revision d7c5fc0ed63066056e62be12b37714ab838e915b) +++ uspace/srv/sysman/configuration.c (revision 3f7e1f243981a64f3cb89f89ffb7c6189b539f5c) @@ -39,5 +39,4 @@ static hash_table_t units; -static fibril_rwlock_t units_rwl; /* Hash table functions */ @@ -81,5 +80,4 @@ { hash_table_create(&units, 0, 0, &units_ht_ops); - fibril_rwlock_initialize(&units_rwl); } @@ -89,5 +87,4 @@ assert(unit->state == STATE_EMBRYO); assert(unit->name != NULL); - assert(fibril_rwlock_is_write_locked(&units_rwl)); sysman_log(LVL_DEBUG2, "%s('%s')", __func__, unit_name(unit)); @@ -100,7 +97,5 @@ void configuration_start_update(void) { - assert(!fibril_rwlock_is_write_locked(&units_rwl)); - sysman_log(LVL_DEBUG2, "%s", __func__); - fibril_rwlock_write_lock(&units_rwl); + sysman_log(LVL_DEBUG2, "%s", __func__); } @@ -124,5 +119,4 @@ void configuration_commit(void) { - assert(fibril_rwlock_is_write_locked(&units_rwl)); sysman_log(LVL_DEBUG2, "%s", __func__); @@ -132,5 +126,4 @@ */ hash_table_apply(&units, &configuration_commit_unit, NULL); - fibril_rwlock_write_unlock(&units_rwl); } @@ -157,9 +150,7 @@ void configuration_rollback(void) { - assert(fibril_rwlock_is_write_locked(&units_rwl)); sysman_log(LVL_DEBUG2, "%s", __func__); hash_table_apply(&units, &configuration_rollback_unit, NULL); - fibril_rwlock_write_unlock(&units_rwl); } @@ -199,5 +190,4 @@ int configuration_resolve_dependecies(void) { - assert(fibril_rwlock_is_write_locked(&units_rwl)); sysman_log(LVL_DEBUG2, "%s", __func__); Index: uspace/srv/sysman/job.c =================================================================== --- uspace/srv/sysman/job.c (revision d7c5fc0ed63066056e62be12b37714ab838e915b) +++ uspace/srv/sysman/job.c (revision 3f7e1f243981a64f3cb89f89ffb7c6189b539f5c) @@ -27,136 +27,132 @@ */ -#include +#include #include #include -#include -#include -#include -#include - + +#include "dep.h" #include "job.h" #include "log.h" -#include "unit.h" +#include "sysman.h" static list_t job_queue; -static fibril_mutex_t job_queue_mtx; -static fibril_condvar_t job_queue_cv; - -static void job_destroy(job_t **); - - -static int job_run_start(job_t *job) -{ - sysman_log(LVL_DEBUG, "%s(%p)", __func__, job); - unit_t *unit = job->unit; - - int rc = unit_start(unit); + +/* + * Static functions + */ + +static int job_add_blocked_job(job_t *job, job_t *blocked_job) +{ + int rc = dyn_array_append(&job->blocked_jobs, job_ptr_t, blocked_job); if (rc != EOK) { - return rc; - } - - fibril_mutex_lock(&unit->state_mtx); - while (unit->state != STATE_STARTED) { - fibril_condvar_wait(&unit->state_cv, &unit->state_mtx); - } - fibril_mutex_unlock(&unit->state_mtx); - - // TODO react to failed state + return ENOMEM; + } + + job_add_ref(blocked_job); + blocked_job->blocking_jobs += 1; + return EOK; } -static int job_runner(void *arg) -{ - job_t *job = (job_t *)arg; - - int retval = EOK; - - /* Wait for previous jobs */ - list_foreach(job->blocking_jobs, link, job_link_t, jl) { - retval = job_wait(jl->job); - if (retval != EOK) { - break; - } - } - - if (retval != EOK) { - goto finish; - } - - /* Run the job itself */ - fibril_mutex_lock(&job->state_mtx); - job->state = JOB_RUNNING; - fibril_condvar_broadcast(&job->state_cv); - fibril_mutex_unlock(&job->state_mtx); - - switch (job->type) { - case JOB_START: - retval = job_run_start(job); - break; - default: - assert(false); - } - - -finish: - fibril_mutex_lock(&job->state_mtx); - job->state = JOB_FINISHED; - job->retval = retval; - fibril_condvar_broadcast(&job->state_cv); - fibril_mutex_unlock(&job->state_mtx); - - job_del_ref(&job); - - return EOK; -} - -static int job_dispatcher(void *arg) -{ - fibril_mutex_lock(&job_queue_mtx); - while (1) { - while (list_empty(&job_queue)) { - fibril_condvar_wait(&job_queue_cv, &job_queue_mtx); - } - - link_t *link = list_first(&job_queue); - assert(link); - list_remove(link); - - /* - * Note that possible use of fibril pool must hold invariant - * that job is started asynchronously. In the case there exists - * circular dependency between jobs, it may result in a deadlock. - */ - job_t *job = list_get_instance(link, job_t, link); - fid_t runner_fibril = fibril_create(job_runner, job); - fibril_add_ready(runner_fibril); - } - - fibril_mutex_unlock(&job_queue_mtx); - return EOK; -} +static void job_init(job_t *job, unit_t *u, unit_state_t target_state) +{ + assert(job); + assert(u); + + link_initialize(&job->job_queue); + + /* Start with one reference for the creator */ + atomic_set(&job->refcnt, 1); + + job->target_state = target_state; + job->unit = u; + + dyn_array_initialize(&job->blocked_jobs, job_ptr_t, 0); + job->blocking_jobs = 0; + job->blocking_job_failed = false; + + job->state = JOB_UNQUEUED; + job->retval = JOB_UNDEFINED_; +} + +static bool job_eval_retval(job_t *job) +{ + unit_t *u = job->unit; + if (u->state == job->target_state) { + job->retval = JOB_OK; + return true; + } else if (u->state == STATE_FAILED) { + job->retval = JOB_FAILED; + return true; + } else { + return false; + } +} + +static bool job_is_runnable(job_t *job) +{ + return job->state == JOB_QUEUED && job->blocking_jobs == 0; +} + +static void job_check(void *object, void *data) +{ + unit_t *u = object; + job_t *job = data; + + if (job_eval_retval(job)) { + job_finish(job); + } else { + // TODO place for timeout + // TODO add reference to job? + sysman_object_observer(u, &job_check, job); + } +} + + +static void job_unblock(job_t *blocked_job, job_t *blocking_job) +{ + if (blocking_job->retval == JOB_FAILED) { + blocked_job->blocking_job_failed = true; + } + blocked_job->blocking_jobs -= 1; +} + +static void job_destroy(job_t **job_ptr) +{ + job_t *job = *job_ptr; + if (job == NULL) { + return; + } + + assert(!link_used(&job->job_queue)); + dyn_array_destroy(&job->blocked_jobs); + // TODO I should decrease referece of blocked jobs + + free(job); + *job_ptr = NULL; +} + +/* + * Non-static functions + */ void job_queue_init() { list_initialize(&job_queue); - fibril_mutex_initialize(&job_queue_mtx); - fibril_condvar_initialize(&job_queue_cv); - - fid_t dispatcher_fibril = fibril_create(job_dispatcher, NULL); - fibril_add_ready(dispatcher_fibril); -} - -int job_queue_jobs(list_t *jobs) -{ - fibril_mutex_lock(&job_queue_mtx); - +} + +int job_queue_add_jobs(dyn_array_t *jobs) +{ /* Check consistency with queue. */ - list_foreach(*jobs, link, job_t, new_job) { - list_foreach(job_queue, link, job_t, queued_job) { + 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. */ - if (new_job->unit == queued_job->unit) { + if ((*new_job_it)->unit == queued_job->unit) { + sysman_log(LVL_ERROR, + "Cannot queue multiple jobs foor unit '%s'", + unit_name((*new_job_it)->unit)); return EEXIST; } @@ -165,33 +161,112 @@ /* Enqueue jobs */ - list_foreach_safe(*jobs, cur_link, next_link) { - list_remove(cur_link); - list_append(cur_link, &job_queue); - } - - /* Only job dispatcher waits, it's correct to notify one only. */ - fibril_condvar_signal(&job_queue_cv); - fibril_mutex_unlock(&job_queue_mtx); + dyn_array_foreach(*jobs, job_ptr_t, job_it) { + (*job_it)->state = JOB_QUEUED; + list_append(&(*job_it)->job_queue, &job_queue); + // TODO explain this reference + job_add_ref(*job_it); + } return EOK; } -/** Blocking wait for job finishing. - * - * Multiple fibrils may wait for the same job. - * - * @return Return code of the job - */ -int job_wait(job_t *job) -{ - fibril_mutex_lock(&job->state_mtx); - while (job->state != JOB_FINISHED) { - fibril_condvar_wait(&job->state_cv, &job->state_mtx); - } - - int rc = job->retval; - fibril_mutex_unlock(&job->state_mtx); - +/** Pop next runnable job + * + * @return runnable job or NULL when there's none + */ +job_t *job_queue_pop_runnable(void) +{ + job_t *result = NULL; + link_t *first_link = list_first(&job_queue); + bool first_iteration = true; + + list_foreach_safe(job_queue, cur_link, next_link) { + result = list_get_instance(cur_link, job_t, job_queue); + if (job_is_runnable(result)) { + break; + } else if (!first_iteration && cur_link == first_link) { + result = NULL; + break; + } else { + /* + * We make no assuptions about ordering of jobs in the + * queue, so just move the job to the end of the queue. + * If there are exist topologic ordering, eventually + * jobs will be reordered. Furthermore when if there + * exists any runnable job, it's always found. + */ + list_remove(cur_link); + list_append(cur_link, &job_queue); + } + first_iteration = false; + } + + if (result) { + // TODO update refcount + list_remove(&result->job_queue); + result->state = JOB_DEQUEUED; + } + + return result; +} + +int job_create_closure(job_t *main_job, dyn_array_t *job_closure) +{ + // TODO replace hard-coded FIFO size with resizable FIFO + FIFO_INITIALIZE_DYNAMIC(jobs_fifo, job_ptr_t, 10); + void *fifo_data = fifo_create(jobs_fifo); + int rc; + if (fifo_data == NULL) { + rc = ENOMEM; + goto finish; + } + + /* + * Traverse dependency graph in BFS fashion and create jobs for every + * necessary unit. + */ + fifo_push(jobs_fifo, main_job); + job_t *job; + while ((job = fifo_pop(jobs_fifo)) != NULL) { + /* + * Do not increase reference count of job, as we're passing it + * to the closure. + */ + dyn_array_append(job_closure, job_ptr_t, job); + + /* Traverse dependencies edges */ + unit_t *u = job->unit; + list_foreach(u->dependencies, dependencies, unit_dependency_t, dep) { + // TODO prepare for reverse edge direction and + // non-identity state mapping + job_t *new_job = + job_create(dep->dependency, job->target_state); + if (new_job == NULL) { + rc = ENOMEM; + goto finish; + } + job_add_blocked_job(new_job, job); + fifo_push(jobs_fifo, new_job); + } + } + rc = EOK; + +finish: + free(fifo_data); + /* + * Newly created jobs are already passed to the closure, thus not + * deleting them here. + */ return rc; +} + +job_t *job_create(unit_t *u, unit_state_t target_state) +{ + job_t *job = malloc(sizeof(job_t)); + if (job != NULL) { + job_init(job, u, target_state); + } + + return job; } @@ -214,63 +289,61 @@ } -static void job_init(job_t *job, job_type_t type) -{ - assert(job); - - link_initialize(&job->link); - list_initialize(&job->blocking_jobs); - - /* Start with one reference for the creator */ - atomic_set(&job->refcnt, 1); - - job->type = type; - job->unit = NULL; - - job->state = JOB_WAITING; - fibril_mutex_initialize(&job->state_mtx); - fibril_condvar_initialize(&job->state_cv); -} - -job_t *job_create(job_type_t type) -{ - job_t *job = malloc(sizeof(job_t)); - if (job != NULL) { - job_init(job, type); - } - - return job; -} - -int job_add_blocking_job(job_t *job, job_t *blocking_job) -{ - job_link_t *job_link = malloc(sizeof(job_link_t)); - if (job_link == NULL) { - return ENOMEM; - } - - link_initialize(&job_link->link); - list_append(&job_link->link, &job->blocking_jobs); - - job_link->job = blocking_job; - job_add_ref(blocking_job); - - return EOK; -} - -static void job_destroy(job_t **job_ptr) -{ - job_t *job = *job_ptr; - if (job == NULL) { - return; - } - - list_foreach_safe(job->blocking_jobs, cur_link, next_link) { - job_link_t *jl = list_get_instance(cur_link, job_link_t, link); - list_remove(cur_link); - job_del_ref(&jl->job); - free(jl); - } - free(job); - - *job_ptr = NULL; -} +void job_run(job_t *job) +{ + assert(job->state != JOB_RUNNING); + assert(job->state != JOB_FINISHED); + + unit_t *u = job->unit; + sysman_log(LVL_DEBUG, "%s, %s -> %i", + __func__, unit_name(u), job->target_state); + + /* Propagate failure */ + if (job->blocking_job_failed) { + goto fail; + } + + int rc; + switch (job->target_state) { + case STATE_STARTED: + rc = unit_start(u); + break; + default: + // TODO implement other states + assert(false); + } + if (rc != EOK) { + goto fail; + } + + job_check(job->unit, job); + return; + +fail: + job->retval = JOB_FAILED; + job_finish(job); +} + +/** Unblocks blocked jobs and notify observers + * + * @param[in] job job with defined return value + */ +void job_finish(job_t *job) +{ + assert(job->state != JOB_FINISHED); + assert(job->retval != JOB_UNDEFINED_); + + sysman_log(LVL_DEBUG2, "%s(%s) -> %i", + __func__, unit_name(job->unit), job->retval); + + job->state = JOB_FINISHED; + + /* Job finished */ + dyn_array_foreach(job->blocked_jobs, job_ptr_t, job_it) { + job_unblock(*job_it, job); + } + // TODO remove reference from blocked jobs + + // TODO should reference be added (for the created event) + sysman_raise_event(&sysman_event_job_changed, job); +} + Index: uspace/srv/sysman/job.h =================================================================== --- uspace/srv/sysman/job.h (revision d7c5fc0ed63066056e62be12b37714ab838e915b) +++ uspace/srv/sysman/job.h (revision 3f7e1f243981a64f3cb89f89ffb7c6189b539f5c) @@ -30,59 +30,61 @@ #define SYSMAN_JOB_H +#include #include #include +#include #include "unit.h" -struct job; -typedef struct job job_t; - +/** Run state of job */ typedef enum { - JOB_START -} job_type_t; - -typedef enum { - JOB_WAITING, + JOB_UNQUEUED, /**< Job not in queue yet */ + JOB_QUEUED, + JOB_DEQUEUED, /**< Job not in queue already */ JOB_RUNNING, JOB_FINISHED } job_state_t; +/** Return value of job */ +typedef enum { + JOB_OK, + JOB_FAILED, + JOB_UNDEFINED_ = -1 +} job_retval_t; + typedef struct { - link_t link; - job_t *job; -} job_link_t; - -/** Job represents pending or running operation on unit */ -struct job { - /** Link to queue job is in */ - link_t link; - - /** List of jobs (job_link_t ) that are blocking the job. */ - list_t blocking_jobs; - - /** Reference counter for the job structure. */ + link_t job_queue; atomic_t refcnt; - job_type_t type; + unit_state_t target_state; unit_t *unit; + /** Jobs that this job is preventing from running */ + dyn_array_t blocked_jobs; + /** No. of jobs that must finish before this job */ + size_t blocking_jobs; + /** Any of blocking jobs failed */ + bool blocking_job_failed; + + /** See job_state_t */ job_state_t state; - fibril_mutex_t state_mtx; - fibril_condvar_t state_cv; + /** See job_retval_t */ + job_retval_t retval; +} job_t; - /** Return value of the job, defined only when state == JOB_FINISHED */ - int retval; -}; +typedef job_t *job_ptr_t; extern void job_queue_init(void); -extern int job_queue_jobs(list_t *); +extern int job_queue_add_jobs(dyn_array_t *); +extern job_t *job_queue_pop_runnable(void); -extern int job_wait(job_t *); +extern int job_create_closure(job_t *, dyn_array_t *); +extern job_t *job_create(unit_t *, unit_state_t); extern void job_add_ref(job_t *); extern void job_del_ref(job_t **); -extern job_t *job_create(job_type_t type); -extern int job_add_blocking_job(job_t *, job_t *); +extern void job_run(job_t *); +extern void job_finish(job_t *); #endif Index: uspace/srv/sysman/main.c =================================================================== --- uspace/srv/sysman/main.c (revision d7c5fc0ed63066056e62be12b37714ab838e915b) +++ uspace/srv/sysman/main.c (revision 3f7e1f243981a64f3cb89f89ffb7c6189b539f5c) @@ -37,4 +37,5 @@ #include "dep.h" #include "job.h" +#include "log.h" #include "sysman.h" #include "unit.h" @@ -47,11 +48,6 @@ } -static int sysman_entry_point(void *arg) { - /* - * Build hard coded configuration. - * - * Strings are allocated on heap, so that they can be free'd by an - * owning unit. - */ +/** Build hard coded configuration */ +static job_t *create_entry_configuration(void) { int result = EOK; unit_t *mnt_initrd = NULL; @@ -107,13 +103,23 @@ configuration_commit(); - result = sysman_unit_start(tgt_default); - - return result; + job_t *first_job = job_create(tgt_default, STATE_STARTED); + if (first_job == NULL) { + goto fail; + } + return first_job; fail: + // TODO cannot destroy units after they're added to configuration unit_destroy(&tgt_default); unit_destroy(&cfg_init); unit_destroy(&mnt_initrd); - return result; + return NULL; +} + +static void first_job_handler(void *object, void *unused) +{ + job_t *job = object; + sysman_log(LVL_DEBUG, "First job retval: %i.", job->retval); + job_del_ref(&job); } @@ -122,16 +128,29 @@ printf(NAME ": HelenOS system daemon\n"); + /* + * Initialize global structures + */ configuration_init(); + sysman_events_init(); job_queue_init(); /* - * Create and start initial configuration asynchronously - * so that we can start server's fibril that may be used - * when executing the start. + * Create initial configuration while we are in a single fibril, keep + * the job and run it when event loop is running. */ - fid_t entry_fibril = fibril_create(sysman_entry_point, NULL); - fibril_add_ready(entry_fibril); + job_t *first_job = create_entry_configuration(); - /* Prepare and start sysman server */ + /* + * Event loop runs in separate fibril, all consequent access to global + * structure is made from this fibril only. + */ + fid_t event_loop_fibril = fibril_create(sysman_events_loop, NULL); + fibril_add_ready(event_loop_fibril); + + /* Queue first job for processing */ + sysman_object_observer(first_job, &first_job_handler, NULL); + sysman_raise_event(&sysman_event_job_process, first_job); + + /* Start sysman server */ async_set_client_connection(sysman_connection); @@ -139,4 +158,5 @@ async_manager(); + /* not reached */ return 0; } Index: uspace/srv/sysman/sysman.c =================================================================== --- uspace/srv/sysman/sysman.c (revision d7c5fc0ed63066056e62be12b37714ab838e915b) +++ uspace/srv/sysman/sysman.c (revision 3f7e1f243981a64f3cb89f89ffb7c6189b539f5c) @@ -27,69 +27,245 @@ */ +#include #include #include - -#include "dep.h" -#include "job.h" +#include +#include + +#include "log.h" #include "sysman.h" -/** Create jobs for cluser of given unit. - * - * @note Using recursion, limits "depth" of dependency graph. - */ -static int sysman_create_closure_jobs(unit_t *unit, job_t **entry_job_ptr, - list_t *accumulator, job_type_t type) -{ - int rc = EOK; - job_t *job = job_create(type); - if (job == NULL) { + +/* Do not expose this generally named type */ +typedef struct { + link_t event_queue; + + event_handler_t handler; + void *data; +} event_t; + +typedef struct { + link_t callbacks; + + callback_handler_t handler; + void *data; +} obj_callback_t; + +typedef struct { + ht_link_t ht_link; + + void *object; + list_t callbacks; +} observed_object_t; + +static LIST_INITIALIZE(event_queue); +static fibril_mutex_t event_queue_mtx; +static fibril_condvar_t event_queue_cv; + +static hash_table_t observed_objects; +static fibril_mutex_t observed_objects_mtx; +static fibril_condvar_t observed_objects_cv; + +/* Hash table functions */ +static size_t observed_objects_ht_hash(const ht_link_t *item) +{ + observed_object_t *callbacks = + hash_table_get_inst(item, observed_object_t, ht_link); + + return (size_t) callbacks->object; +} + +static size_t observed_objects_ht_key_hash(void *key) +{ + void *object = *(void **) key; + return (size_t) object; +} + +static bool observed_objects_ht_key_equal(void *key, const ht_link_t *item) +{ + void *object = *(void **)key; + return ( + hash_table_get_inst(item, observed_object_t, ht_link)->object == + object); +} + +static hash_table_ops_t observed_objects_ht_ops = { + .hash = &observed_objects_ht_hash, + .key_hash = &observed_objects_ht_key_hash, + .equal = NULL, + .key_equal = &observed_objects_ht_key_equal, + .remove_callback = NULL +}; + +static void notify_observers(void *object) +{ + ht_link_t *item = hash_table_find(&observed_objects, &object); + if (item == NULL) { + return; + } + observed_object_t *observed_object = + hash_table_get_inst(item, observed_object_t, ht_link); + + list_foreach_safe(observed_object->callbacks, cur_link, next_link) { + obj_callback_t *callback = + list_get_instance(cur_link, obj_callback_t, callbacks); + callback->handler(object, callback->data); + list_remove(cur_link); + free(callback); + } +} + +/* + * Non-static functions + */ +void sysman_events_init(void) +{ + fibril_mutex_initialize(&event_queue_mtx); + fibril_condvar_initialize(&event_queue_cv); + + bool table = + hash_table_create(&observed_objects, 0, 0, &observed_objects_ht_ops); + if (!table) { + sysman_log(LVL_FATAL, "%s: Failed initialization", __func__); + abort(); + } + fibril_mutex_initialize(&observed_objects_mtx); + fibril_condvar_initialize(&observed_objects_cv); +} + +int sysman_events_loop(void *unused) +{ + while (1) { + /* Pop event */ + fibril_mutex_lock(&event_queue_mtx); + while (list_empty(&event_queue)) { + fibril_condvar_wait(&event_queue_cv, &event_queue_mtx); + } + + link_t *li_event = list_first(&event_queue); + list_remove(li_event); + event_t *event = + list_get_instance(li_event, event_t, event_queue); + fibril_mutex_unlock(&event_queue_mtx); + + /* Process event */ + event->handler(event->data); + free(event); + } +} + +void sysman_raise_event(event_handler_t handler, void *data) +{ + event_t *event = malloc(sizeof(event_t)); + if (event == NULL) { + sysman_log(LVL_FATAL, "%s: cannot allocate event", __func__); + // TODO think about aborting system critical task + abort(); + } + link_initialize(&event->event_queue); + event->handler = handler; + event->data = data; + + fibril_mutex_lock(&event_queue_mtx); + list_append(&event->event_queue, &event_queue); + /* There's only single event loop, broadcast is unnecessary */ + fibril_condvar_signal(&event_queue_cv); + fibril_mutex_unlock(&event_queue_mtx); +} + +/** Register single-use object observer callback + * + * TODO no one handles return value, it's quite fatal to lack memory for + * callbacks... @return EOK on success + * @return ENOMEM + */ +int sysman_object_observer(void *object, callback_handler_t handler, void *data) +{ + int rc; + observed_object_t *observed_object = NULL; + observed_object_t *new_observed_object = NULL; + ht_link_t *ht_link = hash_table_find(&observed_objects, &object); + + if (ht_link == NULL) { + observed_object = malloc(sizeof(observed_object_t)); + if (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); + } else { + observed_object = + hash_table_get_inst(ht_link, observed_object_t, ht_link); + } + + obj_callback_t *obj_callback = malloc(sizeof(obj_callback_t)); + if (obj_callback == NULL) { rc = ENOMEM; goto fail; } - job->unit = unit; - - list_foreach(unit->dependencies, dependencies, unit_dependency_t, dep) { - job_t *blocking_job = NULL; - rc = sysman_create_closure_jobs(dep->dependency, &blocking_job, - accumulator, type); - if (rc != EOK) { - goto fail; - } - - rc = job_add_blocking_job(job, blocking_job); - if (rc != EOK) { - goto fail; - } - } - - /* Job is passed to the accumulator, i.e. no add_ref. */ - list_append(&job->link, accumulator); - - if (entry_job_ptr != NULL) { - *entry_job_ptr = job; - } + obj_callback->handler = handler; + obj_callback->data = data; + list_append(&obj_callback->callbacks, &observed_object->callbacks); return EOK; fail: - job_del_ref(&job); + free(new_observed_object); return rc; } -int sysman_unit_start(unit_t *unit) -{ - list_t new_jobs; - list_initialize(&new_jobs); - - job_t *job = NULL; - // TODO shouldn't be here read-lock on configuration? - int rc = sysman_create_closure_jobs(unit, &job, &new_jobs, JOB_START); +/* + * Event handlers + */ + +// NOTE must run in main event loop fibril +void sysman_event_job_process(void *arg) +{ + job_t *job = arg; + dyn_array_t job_closure; + dyn_array_initialize(&job_closure, job_ptr_t, 0); + + int rc = job_create_closure(job, &job_closure); if (rc != EOK) { - return rc; - } - - // TODO handle errors when adding job accumulator - job_queue_jobs(&new_jobs); - - return job_wait(job); -} + sysman_log(LVL_ERROR, "Cannot create closure for job %p (%i)", + job, rc); + goto fail; + } + + rc = job_queue_add_jobs(&job_closure); + if (rc != EOK) { + // TODO job_queue_add_jobs should log message + goto fail; + } + + // TODO explain why calling asynchronously + sysman_raise_event(&sysman_event_job_queue_run, NULL); + return; + +fail: + job->retval = JOB_FAILED; + job_finish(job); + // TODO clarify refcount to the main job + dyn_array_foreach(job_closure, job_ptr_t, closure_job) { + job_del_ref(&(*closure_job)); + } + dyn_array_destroy(&job_closure); +} + + +void sysman_event_job_queue_run(void *unused) +{ + job_t *job; + while ((job = job_queue_pop_runnable())) { + job_run(job); + } +} + +void sysman_event_job_changed(void *object) +{ + notify_observers(object); +} Index: uspace/srv/sysman/sysman.h =================================================================== --- uspace/srv/sysman/sysman.h (revision d7c5fc0ed63066056e62be12b37714ab838e915b) +++ uspace/srv/sysman/sysman.h (revision 3f7e1f243981a64f3cb89f89ffb7c6189b539f5c) @@ -30,7 +30,22 @@ #define SYSMAN_SYSMAN_H +#include "job.h" #include "unit.h" -extern int sysman_unit_start(unit_t *); +typedef void (*event_handler_t)(void *); +typedef void (*callback_handler_t)(void *object, void *data); + +extern void sysman_events_init(void); + +extern int sysman_events_loop(void *); + +extern void sysman_raise_event(event_handler_t, void *); + +extern int sysman_object_observer(void *, callback_handler_t, void *); + + +extern void sysman_event_job_process(void *); +extern void sysman_event_job_queue_run(void *); +extern void sysman_event_job_changed(void *); #endif Index: uspace/srv/sysman/unit.c =================================================================== --- uspace/srv/sysman/unit.c (revision d7c5fc0ed63066056e62be12b37714ab838e915b) +++ uspace/srv/sysman/unit.c (revision 3f7e1f243981a64f3cb89f89ffb7c6189b539f5c) @@ -68,6 +68,4 @@ unit->state = STATE_EMBRYO; - fibril_mutex_initialize(&unit->state_mtx); - fibril_condvar_initialize(&unit->state_cv); list_initialize(&unit->dependants); @@ -105,16 +103,15 @@ } -void unit_set_state(unit_t *unit, unit_state_t state) -{ - fibril_mutex_lock(&unit->state_mtx); - unit->state = state; - fibril_condvar_broadcast(&unit->state_cv); - fibril_mutex_unlock(&unit->state_mtx); -} /** Issue request to restarter to start a unit * - * Return from this function only means start request was issued. - * If you need to wait for real start of the unit, use waiting on state_cv. + * Ideally this function is non-blocking synchronous, however, some units + * cannot be started synchronously and thus return from this function generally + * means that start was requested. + * + * Check state of the unit for actual result, start method can end in states: + * - STATE_STARTED, (succesful synchronous start) + * - STATE_STARTING, (succesful asynchronous start request) + * - STATE_FAILED. (error occured) */ int unit_start(unit_t *unit) Index: uspace/srv/sysman/unit.h =================================================================== --- uspace/srv/sysman/unit.h (revision d7c5fc0ed63066056e62be12b37714ab838e915b) +++ uspace/srv/sysman/unit.h (revision 3f7e1f243981a64f3cb89f89ffb7c6189b539f5c) @@ -62,6 +62,4 @@ unit_state_t state; - fibril_mutex_t state_mtx; - fibril_condvar_t state_cv; list_t dependencies; Index: uspace/srv/sysman/units/unit_cfg.c =================================================================== --- uspace/srv/sysman/units/unit_cfg.c (revision d7c5fc0ed63066056e62be12b37714ab838e915b) +++ uspace/srv/sysman/units/unit_cfg.c (revision 3f7e1f243981a64f3cb89f89ffb7c6189b539f5c) @@ -176,5 +176,5 @@ } - assert(unit->state = STATE_EMBRYO); + assert(unit->state == STATE_EMBRYO); configuration_add_unit(unit); } @@ -232,9 +232,4 @@ assert(u_cfg); - /* - * Skip starting state and hold state lock during whole configuration - * load. - */ - fibril_mutex_lock(&unit->state_mtx); int rc = cfg_load_configuration(u_cfg->path); @@ -244,6 +239,4 @@ unit->state = STATE_FAILED; } - fibril_condvar_broadcast(&unit->state_cv); - fibril_mutex_unlock(&unit->state_mtx); return rc; Index: uspace/srv/sysman/units/unit_mnt.c =================================================================== --- uspace/srv/sysman/units/unit_mnt.c (revision d7c5fc0ed63066056e62be12b37714ab838e915b) +++ uspace/srv/sysman/units/unit_mnt.c (revision 3f7e1f243981a64f3cb89f89ffb7c6189b539f5c) @@ -87,28 +87,36 @@ static int unit_mnt_start(unit_t *unit) { + // TODO replace with non-blocking + const bool blocking = true; unit_mnt_t *u_mnt = CAST_MNT(unit); assert(u_mnt); - fibril_mutex_lock(&unit->state_mtx); // TODO think about unit's lifecycle (is STOPPED only acceptable?) assert(unit->state == STATE_STOPPED); - unit->state = STATE_STARTING; - - fibril_condvar_broadcast(&unit->state_cv); - fibril_mutex_unlock(&unit->state_mtx); // TODO use other mount parameters int rc = mount(u_mnt->type, u_mnt->mountpoint, u_mnt->device, "", - IPC_FLAG_BLOCKING, 0); + blocking ? IPC_FLAG_BLOCKING : 0, 0); - if (rc == EOK) { - sysman_log(LVL_NOTE, "Mount ('%s') mounted", unit_name(unit)); - unit_set_state(unit, STATE_STARTED); + if (blocking) { + if (rc == EOK) { + sysman_log(LVL_DEBUG, "Mount ('%s') mounted", unit_name(unit)); + unit->state = STATE_STARTED; + } else { + sysman_log(LVL_ERROR, "Mount ('%s') failed (%i)", + unit_name(unit), rc); + unit->state = STATE_FAILED; + } } else { - sysman_log(LVL_ERROR, "Mount ('%s') failed (%i)", - unit_name(unit), rc); - unit_set_state(unit, STATE_FAILED); + if (rc == EOK) { + sysman_log(LVL_DEBUG, "Mount ('%s') requested", unit_name(unit)); + unit->state = STATE_STARTING; + } else { + sysman_log(LVL_ERROR, "Mount ('%s') request failed (%i)", + unit_name(unit), rc); + unit->state = STATE_FAILED; + } } Index: uspace/srv/sysman/units/unit_tgt.c =================================================================== --- uspace/srv/sysman/units/unit_tgt.c (revision d7c5fc0ed63066056e62be12b37714ab838e915b) +++ uspace/srv/sysman/units/unit_tgt.c (revision 3f7e1f243981a64f3cb89f89ffb7c6189b539f5c) @@ -59,4 +59,5 @@ assert(u_tgt); + unit->state = STATE_STARTED; return EOK; }