MODULE: Protect the `updating` field with mutexes in repo module

src/module/module_repo.c

@@ -47,6 +47,7 @@ struct module_repo_internal {
 
     /* Worker threads */
     struct worker_pool workers;
+    pthread_mutex_t mtx; /* Protect the updating field */
     volatile int updating : REPO_UPDATE_TYPE_COUNT;
 
     /* The database and the queue */
@@ -427,7 +428,10 @@ __worker_update(void *__repo)
 {
     struct module_repo_internal *repo = __repo;
     lkt_queue_send(repo->queue, lkt_event_db_updating, LKT_DB_UPDATING_PROGRESS);
+
+    GOTO_IF(pthread_mutex_lock(&(repo->mtx)), "Failed to lock", end_no_lock);
     repo->updating &= REPO_UPDATE_KARA;
+    GOTO_IF(pthread_mutex_unlock(&(repo->mtx)), "Failed to unlock", end_no_lock);
 
     struct json_object *json;
     LOG_INFO("REPO", "Download kara list from %s (%s), directory is %s",
@@ -443,6 +447,11 @@ __worker_update(void *__repo)
     LOG_INFO("REPO", "Finished to deal with deleted kara");
     database_updated(repo->db);
 
+    GOTO_IF(pthread_mutex_lock(&(repo->mtx)), "Failed to lock", end_no_lock);
+    repo->updating &= (~ REPO_UPDATE_KARA);
+    GOTO_IF(pthread_mutex_unlock(&(repo->mtx)), "Failed to unlock", end_no_lock);
+
+end_no_lock:
     lkt_queue_send(repo->queue, lkt_event_db_updating, LKT_DB_UPDATING_FINISHED);
     pthread_exit(NULL);
 }
@@ -453,14 +462,24 @@ __worker_rescan(void *__repo)
     struct module_repo_internal *repo = __repo;
     char kara_prefix[LKT_LINE_MAX];
     lkt_queue_send(repo->queue, lkt_event_db_updating, LKT_DB_UPDATING_PROGRESS);
+
     if (!database_config_get_text(repo->db, "database", "kara_dir", kara_prefix, LKT_LINE_MAX)) {
         LOG_ERROR("REPO", "Failed to get kara prefix from config");
         pthread_exit(NULL);
     }
+
+    GOTO_IF(pthread_mutex_lock(&(repo->mtx)), "Failed to lock", end_no_lock);
     repo->updating &= REPO_UPDATE_KARA;
+    GOTO_IF(pthread_mutex_unlock(&(repo->mtx)), "Failed to unlock", end_no_lock);
+
     database_update(repo->db, kara_prefix, 0); /* Don't check timestamp.
                                                 * TODO: Sometimes we want to check them */
+
+    GOTO_IF(pthread_mutex_lock(&(repo->mtx)), "Failed to lock", end_no_lock);
     repo->updating &= (~ REPO_UPDATE_KARA);
+    GOTO_IF(pthread_mutex_unlock(&(repo->mtx)), "Failed to unlock", end_no_lock);
+
+end_no_lock:
     lkt_queue_send(repo->queue, lkt_event_db_updating, LKT_DB_UPDATING_FINISHED);
     pthread_exit(NULL);
 }
@@ -646,22 +665,26 @@ mod_update(va_list *va)
     struct module_repo_internal **repo;
     va_copy(copy, *va);
     repo    = (struct module_repo_internal **) va_arg(copy, void **);
+    int ret = 0;
 
+    GOTO_IF(pthread_mutex_lock(&(*repo)->mtx), "Failed to lock", end_no_lock);
     if ((*repo)->updating & REPO_UPDATE_KARA) {
         LOG_WARN("REPO", "Already updating");
-        va_end(copy);
-        return 0;
+        goto end;
     }
     (*repo)->updating &= REPO_UPDATE_KARA;
     if (worker_pool_push(&(*repo)->workers, __worker_update, (void *) *repo)) {
         LOG_ERROR("REPO", "Out of memory");
-        va_end(copy);
-        return 1;
+        ret = 1;
+        goto end;
     }
 
-    va_end(copy);
     LOG_INFO("REPO", "Update started (update)");
-    return 0;
+end:
+    GOTO_IF(pthread_mutex_unlock(&(*repo)->mtx), "Failed to unlock", end_no_lock);
+end_no_lock:
+    va_end(copy);
+    return ret;
 }
 
 static int
@@ -671,22 +694,26 @@ mod_rescan(va_list *va)
     struct module_repo_internal **repo;
     va_copy(copy, *va);
     repo    = (struct module_repo_internal **) va_arg(copy, void **);
+    int ret = 0;
 
+    GOTO_IF(pthread_mutex_lock(&(*repo)->mtx), "Failed to lock", end_no_lock);
     if ((*repo)->updating & REPO_UPDATE_KARA) {
         LOG_WARN("REPO", "Already updating");
-        va_end(copy);
-        return 0;
+        goto end;
     }
     (*repo)->updating &= REPO_UPDATE_KARA;
     if (worker_pool_push(&(*repo)->workers, __worker_rescan, (void *) *repo)) {
         LOG_ERROR("REPO", "Out of memory");
-        va_end(copy);
-        return 1;
+        ret = 1;
+        goto end;
     }
 
-    va_end(copy);
     LOG_INFO("REPO", "Update started (rescan)");
-    return 0;
+end:
+    GOTO_IF(pthread_mutex_unlock(&(*repo)->mtx), "Failed to unlock", end_no_lock);
+end_no_lock:
+    va_end(copy);
+    return ret;
 }
 
 static int
@@ -706,13 +733,38 @@ mod_is_updating(va_list *va)
     return 0;
 }
 
+static int
+mod_import(va_list *va)
+{
+    va_list copy;
+    struct module_repo_internal **repo;
+    va_copy(copy, *va);
+    repo = (struct module_repo_internal **) va_arg(copy, void **);
+
+    GOTO_IF(pthread_mutex_lock(&(*repo)->mtx), "Failed to lock", end_no_lock);
+    if ((*repo)->updating & REPO_UPDATE_FAV) {
+        LOG_WARN("REPO", "Already importing favorites");
+        goto end;
+    }
+    (*repo)->updating &= REPO_UPDATE_FAV;
+
+end:
+    GOTO_IF(pthread_mutex_unlock(&(*repo)->mtx), "Failed to unlock", end_no_lock);
+end_no_lock:
+    va_end(copy);
+    return 0;
+}
+
 REG_BEGIN(repo_reg)
-REG_ADD_NAMED("new",         mod_new)
-REG_ADD_NAMED("free",        mod_free)
-REG_ADD_NAMED("close",       mod_close)
-REG_ADD_NAMED("update",      mod_update)
-REG_ADD_NAMED("rescan",      mod_rescan)
-REG_ADD_NAMED("is_updating", mod_is_updating)
+REG_ADD_NAMED("new",         mod_new)           /* Create the module            */
+REG_ADD_NAMED("free",        mod_free)          /* Close and free ressources    */
+REG_ADD_NAMED("close",       mod_close)         /* Stop the module              */
+REG_ADD_NAMED("update",      mod_update)        /* Update from kurisu           */
+REG_ADD_NAMED("rescan",      mod_rescan)        /* Rescan from hard drive       */
+REG_ADD_NAMED("import",      mod_import)        /* Import favorites from kurisu */
+REG_ADD_NAMED("is_updating", mod_is_updating)   /* Is the module updating the
+                                                 * base, either the favorites
+                                                 * or the bakabase              */
 REG_END()
 #if ! defined (LKT_STATIC_MODULE)
     REG_EXPORT(repo_reg)