Avoid sending a pending query and fixes

For some reason, dbus requests many times
the same query. That changes try to avoid that
behaviour.

Also fix asynchronous processing and cache handling.

Change-Id: If91631a1ab51ea8530113d015688978b0cf6467a
Signed-off-by: José Bollo <jose.bollo@iot.bzh>

diff --git a/src/cynagora.c b/src/cynagora.c
index aee6a4c..466592d 100644 (file)
--- a/src/cynagora.c
+++ b/src/cynagora.c
@@ -49,20 +49,34 @@
 #define CACHESIZE(x)  ((x) >= MIN_CACHE_SIZE ? (x) : (x) ? MIN_CACHE_SIZE : 0)
 
 typedef struct asreq asreq_t;
+typedef struct ascb  ascb_t;
 typedef struct agent agent_t;
 typedef struct query query_t;
 
-/** recording of asynchronous requests */
-struct asreq
+/** recording of asynchronous request callbacks */
+struct ascb
 {
-       /** link to the next pending request */
-       asreq_t *next;
+       /** link to the next pending callback */
+       ascb_t *next;
 
        /** callback function */
        cynagora_async_check_cb_t *callback;
 
        /** closure of the callback */
        void *closure;
+};
+
+/** recording of asynchronous requests */
+struct asreq
+{
+       /** link to the next pending request */
+       asreq_t *next;
+
+       /** callbacks */
+       ascb_t *callbacks;
+
+       /** key of the request */
+       cynagora_key_t key;
 
        /** id of the request */
        idgen_t id;
@@ -134,7 +148,7 @@ struct cynagora
        /** the declared agents */
        agent_t *agents;
 
-       /** the pending queries */
+       /** the pending agent queries */
        query_t *queries;
 
        /** id generator */
@@ -161,6 +175,7 @@ struct query
 };
 
 static void agent_ask(cynagora_t *cynagora, int count, const char **fields);
+static int async_reply_process(cynagora_t *cynagora, int count);
 
 /**
  * Flush the write buffer of the client
@@ -334,22 +349,30 @@ get_reply(
        cynagora_t *cynagora
 ) {
        int rc;
+       const char *first;
        uint32_t cacheid;
 
        prot_next(cynagora->prot);
        rc = prot_get(cynagora->prot, &cynagora->reply.fields);
        if (rc > 0) {
-               if (0 == strcmp(cynagora->reply.fields[0], _clear_)) {
+               first = cynagora->reply.fields[0];
+               if (0 == strcmp(first, _clear_)) {
+                       /* clearing the cache */
                        cacheid = rc > 1 ? (uint32_t)atol(cynagora->reply.fields[1]) : 0;
                        cache_clear(cynagora->cache, cacheid);
                        rc = 0;
-               } else if (0 == strcmp(cynagora->reply.fields[0], _ask_)) {
+               } else if (0 == strcmp(first, _ask_)) {
                        /* on asking agent */
                        agent_ask(cynagora, rc - 1, &cynagora->reply.fields[1]);
                        rc = 0;
                } else {
-                       if (0 != strcmp(cynagora->reply.fields[0], _item_))
+                       if (0 != strcmp(cynagora->reply.fields[0], _item_)) {
                                cynagora->pending--;
+                               if (strcmp(first, _done_) && strcmp(first, _error_)) {
+                                       if (async_reply_process(cynagora, rc))
+                                               rc = 0;
+                               }
+                       }
                }
        }
        cynagora->reply.count = rc;
@@ -441,6 +464,7 @@ static
 int
 status_check(
        cynagora_t *cynagora,
+       int count,
        time_t *expire
 ) {
        int rc;
@@ -454,7 +478,7 @@ status_check(
        else
                rc = -EPROTO;
 
-       if (cynagora->reply.count < 3)
+       if (count < 3)
                *expire = 0;
        else if (cynagora->reply.fields[2][0] == '-')
                *expire = -1;
@@ -653,11 +677,11 @@ check_or_test(
        if (rc < 0)
                goto end;
 
-       /* ensure there is no clear cache pending */
-       flushr(cynagora);
-
        /* check cache item */
        if (!force) {
+               /* ensure there is no clear cache pending */
+               flushr(cynagora);
+
                rc = cache_search(cynagora->cache, key);
                if (rc >= 0)
                        goto end;
@@ -669,7 +693,7 @@ check_or_test(
                /* get the response */
                rc = wait_pending_reply(cynagora);
                if (rc >= 0) {
-                       rc = status_check(cynagora, &expire);
+                       rc = status_check(cynagora, rc, &expire);
                        if (rc >= 0 && action == _check_)
                                cache_put(cynagora->cache, key, rc, expire, true);
                }
@@ -705,6 +729,37 @@ search_async_request(
        return ar;
 }
 
+static
+int
+async_reply_process(
+       cynagora_t *cynagora,
+       int count
+) {
+       int status;
+       const char *id;
+       asreq_t *ar;
+       ascb_t *ac;
+       time_t expire;
+
+       id = count < 2 ? "" : cynagora->reply.fields[1];
+       ar = search_async_request(cynagora, id, true);
+
+       if (!ar)
+               return 0;
+
+       /* emit the asynchronous answer */
+       status = status_check(cynagora, count, &expire);
+       if (status >= 0)
+               cache_put(cynagora->cache, &ar->key, status, expire, true);
+       while((ac = ar->callbacks) != NULL) {
+               ar->callbacks = ac->next;
+               ac->callback(ac->closure, status);
+               free(ac);
+       }
+       free(ar);
+       return 1;
+}
+
 /******************************************************************************/
 /*** PUBLIC COMMON METHODS                                                  ***/
 /******************************************************************************/
@@ -803,11 +858,16 @@ cynagora_async_setup(
        void *closure
 ) {
        asreq_t *ar;
+       ascb_t *ac;
 
        /* cancel pending requests */
        while((ar = cynagora->async.requests) != NULL) {
                cynagora->async.requests = ar->next;
-               ar->callback(ar->closure, -ECANCELED);
+               while((ac = ar->callbacks) != NULL) {
+                       ar->callbacks = ac->next;
+                       ac->callback(ac->closure, -ECANCELED);
+                       free(ac);
+               }
                free(ar);
        }
 
@@ -828,47 +888,12 @@ cynagora_async_process(
        cynagora_t *cynagora
 ) {
        int rc;
-       const char *first;
-       const char *id;
-       asreq_t *ar;
-       time_t expire;
-       cynagora_key_t key;
 
        for (;;) {
                /* non blocking wait for a reply */
                rc = wait_reply(cynagora, false);
                if (rc < 0)
                        return rc == -EAGAIN ? 0 : rc;
-
-               /* skip empty replies */
-               if (rc == 0)
-                       continue;
-
-               /* skip done/error replies */
-               first = cynagora->reply.fields[0];
-               if (!strcmp(first, _done_)
-                || !strcmp(first, _error_))
-                       continue;
-
-               /* search the request */
-               id = cynagora->reply.count < 2 ? "" : cynagora->reply.fields[1];
-               ar = search_async_request(cynagora, id, true);
-
-               /* ignore unexpected answers */
-               if (ar == NULL)
-                       continue;
-
-               /* emit the asynchronous answer */
-               rc = status_check(cynagora, &expire);
-               if (rc >= 0) {
-                       key.client = (const char*)(ar + 1);
-                       key.session = &key.client[1 + strlen(key.client)];
-                       key.user = &key.session[1 + strlen(key.session)];
-                       key.permission = &key.user[1 + strlen(key.user)];
-                       cache_put(cynagora->cache, &key, rc, expire, true);
-               }
-               ar->callback(ar->closure, rc);
-               free(ar);
        }
 }
 
@@ -895,6 +920,8 @@ cynagora_cache_check(
        cynagora_t *cynagora,
        const cynagora_key_t *key
 ) {
+       /* ensure there is no clear cache pending */
+       flushr(cynagora);
        return cache_search(cynagora->cache, key);
 }
 
@@ -930,17 +957,19 @@ cynagora_async_check(
 ) {
        int rc;
        asreq_t *ar;
+       ascb_t *ac;
+       char *p;
 
        /* ensure connection */
        rc = ensure_opened(cynagora);
        if (rc < 0)
                return rc;
 
-       /* ensure there is no clear cache pending */
-       flushr(cynagora);
-
        /* check cache item */
        if (!force) {
+               /* ensure there is no clear cache pending */
+               flushr(cynagora);
+
                rc = cache_search(cynagora->cache, key);
                if (rc >= 0) {
                        callback(closure, rc);
@@ -948,32 +977,67 @@ cynagora_async_check(
                }
        }
 
-       /* allocate */
+       /* allocates the callback */
+       ac = malloc(sizeof *ac);
+       if (ac == NULL)
+               return -ENOMEM;
+       ac->callback = callback;
+       ac->closure = closure;
+
+       /* search the request */
+       ar = cynagora->async.requests;
+       while (ar && (strcmp(key->client, ar->key.client)
+               || strcmp(key->session, ar->key.session)
+               || strcmp(key->user, ar->key.user)
+               || strcmp(key->permission, ar->key.permission)))
+               ar = ar->next;
+
+       /* a same request is pending, use it */
+       if (ar) {
+               ac->next = ar->callbacks;
+               ar->callbacks = ac;
+               return 0;
+       }
+
+       /* allocate for the request */
        ar = malloc(sizeof *ar + strlen(key->client) + strlen(key->session) + strlen(key->user) + strlen(key->permission) + 4);
-       if (ar == NULL)
+       if (ar == NULL) {
+               free(ac);
                return -ENOMEM;
+       }
 
        /* init */
+       ac->next = NULL;
+       ar->callbacks = ac;
+       p = (char*)(ar + 1);
+       ar->key.client = p;
+       p = stpcpy(p, key->client) + 1;
+       ar->key.session = p;
+       p = stpcpy(p, key->session) + 1;
+       ar->key.user = p;
+       p = stpcpy(p, key->user) + 1;
+       ar->key.permission = p;
+       stpcpy(p, key->permission);
        do {
                idgen_next(cynagora->idgen);
        } while (search_async_request(cynagora, cynagora->idgen, false));
        strcpy(ar->id, cynagora->idgen);
-       ar->callback = callback;
-       ar->closure = closure;
-       stpcpy(1 + stpcpy(1 + stpcpy(1 + stpcpy((char*)(ar + 1), key->client), key->session), key->user), key->permission);
+       ar->next = cynagora->async.requests;
+       cynagora->async.requests = ar;
 
        /* send the request */
        rc = putxkv(cynagora, simple ? _test_ : _check_, ar->id, key, 0);
-       if (rc >= 0)
-               rc = flushw(cynagora);
        if (rc < 0) {
+               ar = search_async_request(cynagora, ar->id, true);
+               while((ac = ar->callbacks)) {
+                       ar->callbacks = ac->next;
+                       free(ac);
+               }
                free(ar);
                return rc;
        }
 
        /* record the request */
-       ar->next = cynagora->async.requests;
-       cynagora->async.requests = ar;
        return 0;
 }
 

diff --git a/src/cynagora.h b/src/cynagora.h
index 363e725..167b9bc 100644 (file)
--- a/src/cynagora.h
+++ b/src/cynagora.h
@@ -257,7 +257,7 @@ cynagora_test(
 );
 
 /**
- * Check the key asynchronousely (async)
+ * Check the key asynchronously (async)
  *
  * @param cynagora  the handler of the client
  * @param key       the key to query