NAMESTORE: Add record set blocking API

New API that allows the caller to reserve
the mofification of a record set under a label.
The record set cannot be modified by other clients
until released.

5 files changed, 285 insertions, 42 deletions

diff --git a/src/include/gnunet_namestore_service.h b/src/include/gnunet_namestore_service.h
index 7db5e9d9e..619b81aed 100644
--- a/src/include/gnunet_namestore_service.h
+++ b/src/include/gnunet_namestore_service.h
@@ -212,6 +212,61 @@ GNUNET_NAMESTORE_records_lookup (struct GNUNET_NAMESTORE_Handle *h,
 
 
 /**
+ * Open a record set for editing.
+ * Retrieves an exclusive lock on this set.
+ * Must be commited using @a GNUNET_NAMESTORE_records_commit
+ *
+ * @param h handle to the namestore
+ * @param pkey private key of the zone
+ * @param label name that is being mapped
+ * @param error_cb function to call on error (i.e. disconnect or unable to get lock)
+ *        the handle is afterwards invalid
+ * @param error_cb_cls closure for @a error_cb
+ * @param rm function to call with the result (with 0 records if we don't have that label)
+ * @param rm_cls closure for @a rm
+ * @return handle to abort the request
+ */
+struct GNUNET_NAMESTORE_QueueEntry *
+GNUNET_NAMESTORE_records_open (struct GNUNET_NAMESTORE_Handle *h,
+                               const struct
+                               GNUNET_IDENTITY_PrivateKey *pkey,
+                               const char *label,
+                               GNUNET_SCHEDULER_TaskCallback error_cb,
+                               void *error_cb_cls,
+                               GNUNET_NAMESTORE_RecordMonitor rm,
+                               void *rm_cls);
+
+/**
+ * Commit the record set to the namestore.
+ * Releases the lock on the record set.
+ * Use an empty array to
+ * remove all records under the given name.
+ *
+ * The continuation is called after the value has been stored in the
+ * database. Monitors may be notified asynchronously (basically with
+ * a buffer). However, if any monitor is consistently too slow to
+ * keep up with the changes, calling @a cont will be delayed until the
+ * monitors do keep up.
+ *
+ * @param h handle to the namestore
+ * @param pkey private key of the zone
+ * @param label name that is being mapped
+ * @param rd_count number of records in the 'rd' array
+ * @param rd array of records with data to store
+ * @param cont continuation to call when done
+ * @param cont_cls closure for @a cont
+ * @return handle to abort the request
+ */
+struct GNUNET_NAMESTORE_QueueEntry *
+GNUNET_NAMESTORE_records_commit (struct GNUNET_NAMESTORE_Handle *h,
+                                 const struct GNUNET_IDENTITY_PrivateKey *pkey,
+                                 const char *label,
+                                 unsigned int rd_count,
+                                 const struct GNUNET_GNSRECORD_Data *rd,
+                                 GNUNET_NAMESTORE_ContinuationWithStatus cont,
+                                 void *cont_cls);
+
+/**
  * Look for an existing PKEY delegation record for a given public key.
  * Returns at most one result to the processor.
  *
diff --git a/src/namestore/Makefile.am b/src/namestore/Makefile.am
index 51708dd67..2441b864a 100644
--- a/src/namestore/Makefile.am
+++ b/src/namestore/Makefile.am
@@ -39,6 +39,7 @@ if HAVE_SQLITE
 SQLITE_PLUGIN = libgnunet_plugin_namestore_sqlite.la
 SQLITE_TESTS = test_plugin_namestore_sqlite \
  test_namestore_api_store_sqlite \
+ test_namestore_api_store_locking_sqlite \
  test_namestore_api_store_update_sqlite \
  test_namestore_api_zone_iteration_sqlite \
  test_namestore_api_remove_sqlite \
@@ -249,6 +250,16 @@ test_namestore_api_store_sqlite_LDADD = \
   $(top_builddir)/src/identity/libgnunetidentity.la \
   libgnunetnamestore.la
 
+test_namestore_api_store_locking_sqlite_SOURCES = \
+ test_namestore_api_store_locking.c
+test_namestore_api_store_locking_sqlite_LDADD = \
+  $(top_builddir)/src/testing/libgnunettesting.la \
+  $(top_builddir)/src/util/libgnunetutil.la \
+  $(top_builddir)/src/gnsrecord/libgnunetgnsrecord.la \
+  $(top_builddir)/src/identity/libgnunetidentity.la \
+  libgnunetnamestore.la
+
+
 test_namestore_api_store_postgres_SOURCES = \
  test_namestore_api_store.c
 test_namestore_api_store_postgres_LDADD = \
diff --git a/src/namestore/gnunet-service-namestore.c b/src/namestore/gnunet-service-namestore.c
index 2a3a006e8..3f679cacd 100644
--- a/src/namestore/gnunet-service-namestore.c
+++ b/src/namestore/gnunet-service-namestore.c
@@ -121,6 +121,23 @@ struct ZoneIteration
   int send_end;
 };
 
+/**
+ * Lock on a record set
+ */
+struct RecordsLock
+{
+  /* DLL */
+  struct RecordsLock *prev;
+
+  /* DLL */
+  struct RecordsLock *next;
+
+  /* Hash of the locked label */
+  struct GNUNET_HashCode label_hash;
+
+  /* Client locking the zone */
+  struct NamestoreClient *client;
+};
 
 /**
  * A namestore client
@@ -394,6 +411,16 @@ static struct StoreActivity *sa_head;
 static struct StoreActivity *sa_tail;
 
 /**
+ * Head of the DLL of record set locks
+ */
+static struct RecordsLock *locks_head;
+
+/**
+ * Tail of the DLL of record set locks
+ */
+static struct RecordsLock *locks_tail;
+
+/**
  * Notification context shared by all monitors.
  */
 static struct GNUNET_NotificationContext *monitor_nc;
@@ -420,6 +447,7 @@ static void
 cleanup_task (void *cls)
 {
   struct CacheOperation *cop;
+  struct RecordsLock *lock;
 
   (void) cls;
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Stopping namestore service\n");
@@ -431,6 +459,14 @@ cleanup_task (void *cls)
     GNUNET_CONTAINER_DLL_remove (cop_head, cop_tail, cop);
     GNUNET_free (cop);
   }
+  while (NULL != (lock = locks_head))
+  {
+    GNUNET_CONTAINER_DLL_remove (locks_head,
+                                 locks_tail,
+                                 lock);
+    GNUNET_free (lock);
+  }
+
   if (NULL != namecache)
   {
     GNUNET_NAMECACHE_disconnect (namecache);
@@ -1118,6 +1154,7 @@ client_disconnect_cb (void *cls,
   struct NamestoreClient *nc = app_ctx;
   struct ZoneIteration *no;
   struct CacheOperation *cop;
+  struct RecordsLock *lock;
 
   (void) cls;
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Client %p disconnected\n", client);
@@ -1168,6 +1205,15 @@ client_disconnect_cb (void *cls,
   for (cop = cop_head; NULL != cop; cop = cop->next)
     if (nc == cop->nc)
       cop->nc = NULL;
+  for (lock = locks_head; NULL != lock; lock = lock->next)
+  {
+    if (nc != lock->client)
+      continue;
+    GNUNET_CONTAINER_DLL_remove (locks_head,
+                                 locks_tail,
+                                 lock);
+    GNUNET_free (lock);
+  }
   GNUNET_free (nc);
 }
 
@@ -1361,6 +1407,7 @@ check_record_lookup (void *cls, const struct LabelLookupMessage *ll_msg)
   return GNUNET_OK;
 }
 
+
 /**
  * Handles a #GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_LOOKUP message
  *
@@ -1374,13 +1421,14 @@ handle_record_lookup (void *cls, const struct LabelLookupMessage *ll_msg)
   struct GNUNET_MQ_Envelope *env;
   struct LabelLookupResponseMessage *llr_msg;
   struct RecordLookupContext rlc;
+  struct RecordsLock *lock;
+  struct GNUNET_HashCode label_hash;
   const char *name_tmp;
   char *res_name;
   char *conv_name;
   uint32_t name_len;
   int res;
 
-  name_len = ntohl (ll_msg->label_len);
   name_tmp = (const char *) &ll_msg[1];
   GNUNET_SERVICE_client_continue (nc->client);
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -1396,6 +1444,52 @@ handle_record_lookup (void *cls, const struct LabelLookupMessage *ll_msg)
     GNUNET_SERVICE_client_drop (nc->client);
     return;
   }
+  name_len = strlen (conv_name) + 1;
+  if (GNUNET_YES == ntohl (ll_msg->locking))
+  {
+    GNUNET_CRYPTO_hash (conv_name, strlen (conv_name), &label_hash);
+    for (lock = locks_head; NULL != lock; lock = lock->next)
+      if (0 == memcmp (&label_hash, &lock->label_hash, sizeof (label_hash)))
+        break;
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "Record locked: %s\n", (NULL == lock) ? "No" : "Yes");
+    if (NULL != lock)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "Client holds lock: %s\n", (lock->client != nc) ? "No" : "Yes");
+
+      if (lock->client != nc)
+      {
+        GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                    "Lock is held by other client on `%s'\n", conv_name);
+        env =
+          GNUNET_MQ_msg_extra (llr_msg,
+                               name_len,
+                               GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_LOOKUP_RESPONSE);
+        llr_msg->gns_header.r_id = ll_msg->gns_header.r_id;
+        llr_msg->private_key = ll_msg->zone;
+        llr_msg->name_len = htons (name_len);
+        llr_msg->rd_count = htons (0);
+        llr_msg->rd_len = htons (0);
+        llr_msg->found = htons (GNUNET_SYSERR);
+        GNUNET_memcpy (&llr_msg[1], conv_name, name_len);
+        GNUNET_MQ_send (nc->mq, env);
+        GNUNET_free (conv_name);
+        return;
+      }
+    }
+    else
+    {
+      lock = GNUNET_new (struct RecordsLock);
+      lock->client = nc;
+      GNUNET_CRYPTO_hash (conv_name,
+                          strlen (conv_name),
+                          &lock->label_hash);
+      GNUNET_CONTAINER_DLL_insert (locks_head,
+                                   locks_tail,
+                                   lock);
+    }
+  }
   rlc.label = conv_name;
   rlc.found = GNUNET_NO;
   rlc.res_rd_count = 0;
@@ -1407,7 +1501,6 @@ handle_record_lookup (void *cls, const struct LabelLookupMessage *ll_msg)
                                       conv_name,
                                       &lookup_it,
                                       &rlc);
-  GNUNET_free (conv_name);
   env =
     GNUNET_MQ_msg_extra (llr_msg,
                          name_len + rlc.rd_ser_len,
@@ -1419,16 +1512,18 @@ handle_record_lookup (void *cls, const struct LabelLookupMessage *ll_msg)
   llr_msg->rd_len = htons (rlc.rd_ser_len);
   res_name = (char *) &llr_msg[1];
   if ((GNUNET_YES == rlc.found) && (GNUNET_OK == res))
-    llr_msg->found = ntohs (GNUNET_YES);
+    llr_msg->found = htons (GNUNET_YES);
   else
-    llr_msg->found = ntohs (GNUNET_NO);
-  GNUNET_memcpy (&llr_msg[1], name_tmp, name_len);
+    llr_msg->found = htons (GNUNET_NO);
+  GNUNET_memcpy (&llr_msg[1], conv_name, name_len);
   GNUNET_memcpy (&res_name[name_len], rlc.res_rd, rlc.rd_ser_len);
   GNUNET_MQ_send (nc->mq, env);
   GNUNET_free (rlc.res_rd);
+  GNUNET_free (conv_name);
 }
 
 
+
 /**
  * Checks a #GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_STORE message
  *
@@ -1528,6 +1623,8 @@ handle_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
   unsigned int rd_count;
   int res;
   struct StoreActivity *sa;
+  struct RecordsLock *lock;
+  struct GNUNET_HashCode label_hash;
   struct GNUNET_TIME_Absolute existing_block_exp;
   struct GNUNET_TIME_Absolute new_block_exp;
 
@@ -1552,7 +1649,8 @@ handle_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                   "Error normalizing name `%s'\n",
                   name_tmp);
-      send_store_response (nc, GNUNET_SYSERR, _("Error normalizing name."), rid);
+      send_store_response (nc, GNUNET_SYSERR, _ ("Error normalizing name."),
+                           rid);
       GNUNET_SERVICE_client_continue (nc->client);
       return;
     }
@@ -1574,11 +1672,28 @@ handle_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
         GNUNET_GNSRECORD_records_deserialize (rd_ser_len, rd_ser, rd_count, rd))
     {
       send_store_response (nc, GNUNET_SYSERR,
-                           _("Error deserializing records."), rid);
+                           _ ("Error deserializing records."), rid);
       GNUNET_free (conv_name);
       GNUNET_SERVICE_client_continue (nc->client);
       return;
     }
+    if (GNUNET_YES == ntohl (rp_msg->locking))
+    {
+      GNUNET_CRYPTO_hash (conv_name, strlen (conv_name), &label_hash);
+      for (lock = locks_head; NULL != lock; lock = lock->next)
+        if (0 == memcmp (&label_hash, &lock->label_hash, sizeof (label_hash)))
+          break;
+      if ((NULL == lock) ||
+          (lock->client != nc))
+      {
+        send_store_response (nc, res, _ ("Record set locked."), rid);
+        GNUNET_SERVICE_client_continue (nc->client);
+        GNUNET_free (conv_name);
+        return;
+      }
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "Client has lock on `%s', continuing.\n", conv_name);
+    }
 
     GNUNET_STATISTICS_update (statistics,
                               "Well-formed store requests received",
@@ -1683,11 +1798,21 @@ handle_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
     if (GNUNET_OK != res)
     {
       /* store not successful, no need to tell monitors */
-      send_store_response (nc, res, _("Store failed"), rid);
+      send_store_response (nc, res, _ ("Store failed"), rid);
       GNUNET_SERVICE_client_continue (nc->client);
       GNUNET_free (conv_name);
       return;
     }
+    if (GNUNET_YES == ntohl (rp_msg->locking))
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "Releasing lock on `%s'\n", conv_name);
+      GNUNET_assert (NULL != lock);
+      GNUNET_CONTAINER_DLL_remove (locks_head,
+                                   locks_tail,
+                                   lock);
+      GNUNET_free (lock);
+    }
 
     sa = GNUNET_malloc (sizeof(struct StoreActivity)
                         + ntohs (rp_msg->gns_header.header.size));
diff --git a/src/namestore/namestore.h b/src/namestore/namestore.h
index 8391d9d74..0f3ffa837 100644
--- a/src/namestore/namestore.h
+++ b/src/namestore/namestore.h
@@ -68,6 +68,11 @@ struct RecordStoreMessage
   struct GNUNET_TIME_AbsoluteNBO expire;
 
   /**
+   * Unock the label with this request.
+   */
+  uint32_t locking GNUNET_PACKED;
+
+  /**
    * Name length
    */
   uint16_t name_len GNUNET_PACKED;
@@ -146,6 +151,11 @@ struct LabelLookupMessage
   uint32_t label_len GNUNET_PACKED;
 
   /**
+   * Lock the label with this lookup
+   */
+  uint32_t locking GNUNET_PACKED;
+
+  /**
    * The private key of the zone to look up in
    */
   struct GNUNET_IDENTITY_PrivateKey zone;
@@ -185,7 +195,7 @@ struct LabelLookupResponseMessage
    * Was the label found in the database??
    * #GNUNET_YES or #GNUNET_NO
    */
-  uint16_t found GNUNET_PACKED;
+  int16_t found GNUNET_PACKED;
 
   /**
    * The private key of the authority.
diff --git a/src/namestore/namestore_api.c b/src/namestore/namestore_api.c
index 3c8caf961..a7380bbde 100644
--- a/src/namestore/namestore_api.c
+++ b/src/namestore/namestore_api.c
@@ -479,8 +479,10 @@ handle_lookup_result (void *cls, const struct LabelLookupResponseMessage *msg)
   size_t name_len;
   size_t rd_len;
   unsigned int rd_count;
+  int16_t found = (int16_t) ntohs (msg->found);
 
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "Received RECORD_LOOKUP_RESULT\n");
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "Received RECORD_LOOKUP_RESULT (found=%i)\n",
+        found);
   qe = find_qe (h, ntohl (msg->gns_header.r_id));
   if (NULL == qe)
     return;
@@ -488,7 +490,7 @@ handle_lookup_result (void *cls, const struct LabelLookupResponseMessage *msg)
   rd_count = ntohs (msg->rd_count);
   name_len = ntohs (msg->name_len);
   name = (const char *) &msg[1];
-  if (GNUNET_NO == ntohs (msg->found))
+  if (GNUNET_NO == found)
   {
     /* label was not in namestore */
     if (NULL != qe->proc)
@@ -496,6 +498,13 @@ handle_lookup_result (void *cls, const struct LabelLookupResponseMessage *msg)
     free_qe (qe);
     return;
   }
+  if (GNUNET_SYSERR == found)
+  {
+    if (NULL != qe->error_cb)
+      qe->error_cb (qe->error_cb_cls);
+    free_qe (qe);
+    return;
+  }
 
   rd_tmp = &name[name_len];
   {
@@ -1005,14 +1014,15 @@ warn_delay (void *cls)
 }
 
 struct GNUNET_NAMESTORE_QueueEntry *
-GNUNET_NAMESTORE_records_store (
+records_store_ (
   struct GNUNET_NAMESTORE_Handle *h,
   const struct GNUNET_IDENTITY_PrivateKey *pkey,
   const char *label,
   unsigned int rd_count,
   const struct GNUNET_GNSRECORD_Data *rd,
   GNUNET_NAMESTORE_ContinuationWithStatus cont,
-  void *cont_cls)
+  void *cont_cls,
+  int locking)
 {
   struct GNUNET_NAMESTORE_QueueEntry *qe;
   struct GNUNET_MQ_Envelope *env;
@@ -1059,6 +1069,7 @@ GNUNET_NAMESTORE_records_store (
   msg->rd_len = htons (rd_ser_len);
   msg->reserved = ntohs(0);
   msg->private_key = *pkey;
+  msg->locking = htonl (locking);
 
   name_tmp = (char *) &msg[1];
   GNUNET_memcpy (name_tmp, label, name_len);
@@ -1090,27 +1101,45 @@ GNUNET_NAMESTORE_records_store (
   return qe;
 }
 
-/**
- * Lookup an item in the namestore.
- *
- * @param h handle to the namestore
- * @param pkey private key of the zone
- * @param label name that is being mapped (at most 255 characters long)
- * @param error_cb function to call on error (i.e. disconnect)
- * @param error_cb_cls closure for @a error_cb
- * @param rm function to call with the result (with 0 records if we don't have that label)
- * @param rm_cls closure for @a rm
- * @return handle to abort the request
- */
 struct GNUNET_NAMESTORE_QueueEntry *
-GNUNET_NAMESTORE_records_lookup (
+GNUNET_NAMESTORE_records_store (
+  struct GNUNET_NAMESTORE_Handle *h,
+  const struct GNUNET_IDENTITY_PrivateKey *pkey,
+  const char *label,
+  unsigned int rd_count,
+  const struct GNUNET_GNSRECORD_Data *rd,
+  GNUNET_NAMESTORE_ContinuationWithStatus cont,
+  void *cont_cls)
+{
+  return records_store_ (h, pkey, label,
+                         rd_count, rd, cont, cont_cls, GNUNET_NO);
+}
+
+struct GNUNET_NAMESTORE_QueueEntry *
+GNUNET_NAMESTORE_records_commit (
+  struct GNUNET_NAMESTORE_Handle *h,
+  const struct GNUNET_IDENTITY_PrivateKey *pkey,
+  const char *label,
+  unsigned int rd_count,
+  const struct GNUNET_GNSRECORD_Data *rd,
+  GNUNET_NAMESTORE_ContinuationWithStatus cont,
+  void *cont_cls)
+{
+  return records_store_ (h, pkey, label,
+                         rd_count, rd, cont, cont_cls, GNUNET_YES);
+}
+
+
+struct GNUNET_NAMESTORE_QueueEntry *
+records_lookup_ (
   struct GNUNET_NAMESTORE_Handle *h,
   const struct GNUNET_IDENTITY_PrivateKey *pkey,
   const char *label,
   GNUNET_SCHEDULER_TaskCallback error_cb,
   void *error_cb_cls,
   GNUNET_NAMESTORE_RecordMonitor rm,
-  void *rm_cls)
+  void *rm_cls,
+  int locking)
 {
   struct GNUNET_NAMESTORE_QueueEntry *qe;
   struct GNUNET_MQ_Envelope *env;
@@ -1138,6 +1167,7 @@ GNUNET_NAMESTORE_records_lookup (
   msg->gns_header.r_id = htonl (qe->op_id);
   msg->zone = *pkey;
   msg->label_len = htonl (label_len);
+  msg->locking = htonl (locking);
   GNUNET_memcpy (&msg[1], label, label_len);
   if (NULL == h->mq)
     qe->env = env;
@@ -1146,22 +1176,34 @@ GNUNET_NAMESTORE_records_lookup (
   return qe;
 }
 
+struct GNUNET_NAMESTORE_QueueEntry *
+GNUNET_NAMESTORE_records_lookup (
+  struct GNUNET_NAMESTORE_Handle *h,
+  const struct GNUNET_IDENTITY_PrivateKey *pkey,
+  const char *label,
+  GNUNET_SCHEDULER_TaskCallback error_cb,
+  void *error_cb_cls,
+  GNUNET_NAMESTORE_RecordMonitor rm,
+  void *rm_cls)
+{
+  return records_lookup_ (h, pkey, label,
+                          error_cb, error_cb_cls, rm, rm_cls, GNUNET_NO);
+}
+
+struct GNUNET_NAMESTORE_QueueEntry *
+GNUNET_NAMESTORE_records_open (
+  struct GNUNET_NAMESTORE_Handle *h,
+  const struct GNUNET_IDENTITY_PrivateKey *pkey,
+  const char *label,
+  GNUNET_SCHEDULER_TaskCallback error_cb,
+  void *error_cb_cls,
+  GNUNET_NAMESTORE_RecordMonitor rm,
+  void *rm_cls)
+{
+  return records_lookup_ (h, pkey, label,
+                          error_cb, error_cb_cls, rm, rm_cls, GNUNET_YES);
+}
 
-/**
- * Look for an existing PKEY delegation record for a given public key.
- * Returns at most one result to the processor.
- *
- * @param h handle to the namestore
- * @param zone public key of the zone to look up in, never NULL
- * @param value_zone public key of the target zone (value), never NULL
- * @param error_cb function to call on error (i.e. disconnect)
- * @param error_cb_cls closure for @a error_cb
- * @param proc function to call on the matching records, or with
- *        NULL (rd_count == 0) if there are no matching records
- * @param proc_cls closure for @a proc
- * @return a handle that can be used to
- *         cancel
- */
 struct GNUNET_NAMESTORE_QueueEntry *
 GNUNET_NAMESTORE_zone_to_name (
   struct GNUNET_NAMESTORE_Handle *h,