Changeset dba056b in mainline

Timestamp:

2019-08-06T19:23:56Z (5 years ago)

Author:

Matthieu Riolo <matthieu.riolo@…>

Children:

c6d87c10

Parents:

3f05ef7

git-author:

Michal Koutný <xm.koutny+hos@…> (2015-06-20 01:04:31)

git-committer:

Matthieu Riolo <matthieu.riolo@…> (2019-08-06 19:23:56)

Message:

sysman: Using exposees to detect asynchronous start

• also fixed bug with reference counting of merged jobs
• add test for the faulty behavior

Location:

uspace/srv/sysman

Files:

• job.c (modified) (10 diffs)
• sysman.c (modified) (4 diffs)
• sysman.h (modified) (1 diff)
• test/job_queue.c (modified) (1 diff)
• units/unit_svc.c (modified) (1 diff)

uspace/srv/sysman/job.c

@@ -80 +80 @@
         assert(job);
         assert(u);
-        assert(u->job == NULL);
         memset(job, 0, sizeof(*job));
 
@@ -178 +177 @@
 }
 
+/** 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 +235 @@
         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 +264 @@
 }
 
+/** 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 +346 @@
 
                 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 +388 @@
         /* Check invariant */
         list_foreach(units, units, unit_t, u) {
-                assert(u->bfs_job == false);
+                assert(u->bfs_job == NULL);
         }
                 
@@ -390 +431 @@
                 }
         }
-//      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 +444 @@
                 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 +563 @@
         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 +578 @@
         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);
 }

uspace/srv/sysman/sysman.c

@@ -196 +196 @@
 /** 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 +213 @@
         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 +329 @@
 }
 
+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 +370 @@
                 goto fail;
         }
+
         /* We don't need job anymore */
         job_del_ref(&job);

uspace/srv/sysman/sysman.h

@@ -45 +45 @@
 extern int sysman_object_observer(void *, callback_handler_t, void *);
 extern int sysman_move_observers(void *, void *);
+extern size_t sysman_observers_count(void *);
 
 

uspace/srv/sysman/test/job_queue.c

@@ -152 +152 @@
 }
 
+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);

uspace/srv/sysman/units/unit_svc.c

@@ -100 +100 @@
          * 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) {