Index: uspace/srv/sysman/job.c =================================================================== --- uspace/srv/sysman/job.c (revision 3f05ef78b4f8b389b4770defbab4631b556220e6) +++ uspace/srv/sysman/job.c (revision dba056b9fb6624592ecae8b4943a89914115602c) @@ -80,5 +80,4 @@ assert(job); assert(u); - assert(u->job == NULL); memset(job, 0, sizeof(*job)); @@ -178,4 +177,28 @@ } +/** 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 * @@ -212,7 +235,15 @@ trunk->blocked_jobs_count = other->blocked_jobs.size; - /* All allocation is done in job_pre_merge, cannot fail here. */ + /* + * 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); } @@ -233,4 +264,12 @@ } +/** 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) { @@ -307,14 +346,16 @@ unit_t *u = job->unit; - assert(u->bfs_job != NULL); assert(u->job == NULL); - u->job = u->bfs_job; - u->bfs_job = NULL; - - + /* Pass reference from the closure to the unit */ + u->job = job; + + /* Enqueue job (new reference) */ job->state = JOB_PENDING; - /* We pass reference from the closure to the queue */ + 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; @@ -347,5 +388,5 @@ /* Check invariant */ list_foreach(units, units, unit_t, u) { - assert(u->bfs_job == false); + assert(u->bfs_job == NULL); } @@ -390,8 +431,9 @@ } } -// 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)); -// } + 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; @@ -402,4 +444,11 @@ 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; } @@ -514,8 +563,9 @@ assert(job->state != JOB_FINISHED); assert(job->retval != JOB_UNDEFINED_); - assert(job->unit->job == job); - - sysman_log(LVL_DEBUG2, "%s(%p) %s ret %i", - __func__, job, unit_name(job->unit), job->retval); + assert(!job->unit->job || job->unit->job == job); + + sysman_log(LVL_DEBUG2, "%s(%p) %s ret %i, ref %i", + __func__, job, unit_name(job->unit), job->retval, + atomic_get(&job->refcnt)); job->state = JOB_FINISHED; @@ -528,9 +578,11 @@ dyn_array_clear(&job->blocked_jobs); - /* - * Remove job from unit and pass the reference from the unit to the - * event. - */ - job->unit->job = NULL; + /* Add reference for event handler */ + if (job->unit->job == NULL) { + job_add_ref(job); + } else { + /* Pass reference from unit */ + job->unit->job = NULL; + } sysman_raise_event(&sysman_event_job_finished, job); } Index: uspace/srv/sysman/sysman.c =================================================================== --- uspace/srv/sysman/sysman.c (revision 3f05ef78b4f8b389b4770defbab4631b556220e6) +++ uspace/srv/sysman/sysman.c (revision dba056b9fb6624592ecae8b4943a89914115602c) @@ -196,5 +196,5 @@ /** Create and queue job for unit * - * If unit already has the same job assigned callback is set to it. + * If unit already has the same job assigned callback is moved to it. * * @param[in] callback (optional) callback must explicitly delete reference @@ -213,4 +213,5 @@ if (callback != NULL) { job_add_ref(job); + //TODO sysman_object_observer is not fibril-safe CANNOT BE CALLED HERE! sysman_object_observer(job, callback, callback_arg); } @@ -328,4 +329,18 @@ } +size_t sysman_observers_count(void *object) +{ + ht_link_t *ht_link = hash_table_find(&observed_objects, &object); + + if (ht_link == NULL) { + return 0; + } + + observed_object_t *observed_object = + hash_table_get_inst(ht_link, observed_object_t, ht_link); + + return list_count(&observed_object->callbacks); +} + /* @@ -355,4 +370,5 @@ goto fail; } + /* We don't need job anymore */ job_del_ref(&job); Index: uspace/srv/sysman/sysman.h =================================================================== --- uspace/srv/sysman/sysman.h (revision 3f05ef78b4f8b389b4770defbab4631b556220e6) +++ uspace/srv/sysman/sysman.h (revision dba056b9fb6624592ecae8b4943a89914115602c) @@ -45,4 +45,5 @@ extern int sysman_object_observer(void *, callback_handler_t, void *); extern int sysman_move_observers(void *, void *); +extern size_t sysman_observers_count(void *); Index: uspace/srv/sysman/test/job_queue.c =================================================================== --- uspace/srv/sysman/test/job_queue.c (revision 3f05ef78b4f8b389b4770defbab4631b556220e6) +++ uspace/srv/sysman/test/job_queue.c (revision dba056b9fb6624592ecae8b4943a89914115602c) @@ -152,4 +152,54 @@ } +PCUT_TEST(merge_jobs_with_callback) { + /* Setup mock behavior */ + unit_type_vmts[UNIT_SERVICE]->start = &mock_unit_vmt_start_async; + unit_type_vmts[UNIT_SERVICE]->exposee_created = + &mock_unit_vmt_exposee_created; + + /* Define mock units */ + unit_t *s0 = mock_units[UNIT_SERVICE][0]; + + /* Create and start first job */ + job_t *j0 = NULL; + int rc = sysman_run_job(s0, STATE_STARTED, &job_finished_cb, &j0); + PCUT_ASSERT_INT_EQUALS(EOK, rc); + + sysman_process_queue(); + /* Job not finished */ + PCUT_ASSERT_NULL(j0); + + + /* + * While s0 is starting in j0, create second job that should be merged + * into the existing one. + */ + job_t *j1 = NULL; + rc = sysman_run_job(s0, STATE_STARTED, &job_finished_cb, &j1); + PCUT_ASSERT_INT_EQUALS(EOK, rc); + + sysman_process_queue(); + /* Job not finished */ + PCUT_ASSERT_NULL(j1); + + sysman_raise_event(&sysman_event_unit_exposee_created, s0); + sysman_process_queue(); + + PCUT_ASSERT_NOT_NULL(j0); + PCUT_ASSERT_NOT_NULL(j1); + + /* + * Jobs were, merged so both callbacks should have been called with the + * same job + */ + PCUT_ASSERT_EQUALS(j0, j1); + + /* And there should be exactly two references (per each callback) */ + job_del_ref(&j0); + job_del_ref(&j0); + + PCUT_ASSERT_NULL(j0); +} + PCUT_EXPORT(job_queue); Index: uspace/srv/sysman/units/unit_svc.c =================================================================== --- uspace/srv/sysman/units/unit_svc.c (revision 3f05ef78b4f8b389b4770defbab4631b556220e6) +++ uspace/srv/sysman/units/unit_svc.c (revision dba056b9fb6624592ecae8b4943a89914115602c) @@ -100,6 +100,5 @@ * in a moment. Applies to naming service only. */ - // TODO this is even hack in the workaround, exposees doesn't work properly - if (true || str_cmp(unit->name, "devman.svc") == 0 || + if (str_cmp(unit->name, "devman.svc") == 0 || str_cmp(unit->name, "logger.svc") == 0 || str_cmp(unit->name, "irc.svc") == 0) {