keep until server done call

[oweals/gnunet.git] / src / datastore / gnunet-service-datastore.c

diff --git a/src/datastore/gnunet-service-datastore.c b/src/datastore/gnunet-service-datastore.c
index d58f0f8d9d7fdafa7784dd55d24407c04c48d41a..1454504e80589b63b13e4edbbb60e64e5f478488 100644 (file)
--- a/src/datastore/gnunet-service-datastore.c
+++ b/src/datastore/gnunet-service-datastore.c
@@ -174,12 +174,31 @@ typedef void (*TransmitContinuation)(void *cls,
                                     int status);
 
 
+/**
+ * Context for transmitting replies to clients.
+ */
 struct TransmitCallbackContext 
 {
+  
+  /**
+   * We keep these in a doubly-linked list (for cleanup).
+   */
+  struct TransmitCallbackContext *next;
+  
+  /**
+   * We keep these in a doubly-linked list (for cleanup).
+   */
+  struct TransmitCallbackContext *prev;
+  
   /**
    * The message that we're asked to transmit.
    */
   struct GNUNET_MessageHeader *msg;
+  
+  /**
+   * Handle for the transmission request.
+   */
+  struct GNUNET_CONNECTION_TransmitHandle *th;
 
   /**
    * Client that we are transmitting to.
@@ -204,6 +223,17 @@ struct TransmitCallbackContext
   int end;
 };
 
+  
+/**
+ * Head of the doubly-linked list (for cleanup).
+ */
+static struct TransmitCallbackContext *tcc_head;
+
+/**
+ * Tail of the doubly-linked list (for cleanup).
+ */
+static struct TransmitCallbackContext *tcc_tail;
+
 
 /**
  * Task that is used to remove expired entries from
@@ -410,6 +440,10 @@ transmit_callback (void *cls,
   struct TransmitCallbackContext *tcc = cls;
   size_t msize;
   
+  tcc->th = NULL;
+  GNUNET_CONTAINER_DLL_remove (tcc_head,
+                              tcc_tail,
+                              tcc);
   msize = ntohs(tcc->msg->size);
   if (size == 0)
     {
@@ -420,9 +454,8 @@ transmit_callback (void *cls,
       if (tcc->tc != NULL)
        tcc->tc (tcc->tc_cls, GNUNET_SYSERR);
       if (GNUNET_YES == tcc->end)
-       {
-         GNUNET_SERVER_receive_done (tcc->client, GNUNET_SYSERR);
-       }
+       GNUNET_SERVER_receive_done (tcc->client, GNUNET_SYSERR);       
+      GNUNET_SERVER_client_drop (tcc->client);
       GNUNET_free (tcc->msg);
       GNUNET_free (tcc);
       return 0;
@@ -442,6 +475,7 @@ transmit_callback (void *cls,
                  "Response transmitted, more pending!\n");
 #endif
     }
+  GNUNET_SERVER_client_drop (tcc->client);
   GNUNET_free (tcc->msg);
   GNUNET_free (tcc);
   return msize;
@@ -474,13 +508,12 @@ transmit (struct GNUNET_SERVER_Client *client,
   tcc->tc = tc;
   tcc->tc_cls = tc_cls;
   tcc->end = end;
-
   if (NULL ==
-      GNUNET_SERVER_notify_transmit_ready (client,
-                                          ntohs(msg->size),
-                                          GNUNET_TIME_UNIT_FOREVER_REL,
-                                          &transmit_callback,
-                                          tcc))
+      (tcc->th = GNUNET_SERVER_notify_transmit_ready (client,
+                                                     ntohs(msg->size),
+                                                     GNUNET_TIME_UNIT_FOREVER_REL,
+                                                     &transmit_callback,
+                                                     tcc)))
     {
       GNUNET_break (0);
       if (GNUNET_YES == end)
@@ -495,7 +528,12 @@ transmit (struct GNUNET_SERVER_Client *client,
        tc (tc_cls, GNUNET_SYSERR);
       GNUNET_free (msg);
       GNUNET_free (tcc);
+      return;
     }
+  GNUNET_SERVER_client_keep (client);
+  GNUNET_CONTAINER_DLL_insert (tcc_head,
+                              tcc_tail,
+                              tcc);
 }
 
 
@@ -546,8 +584,9 @@ get_next(void *next_cls,
   if (status != GNUNET_OK)
     {
       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
-                 _("Failed to transmit an item to the client; aborting iteration.\n"));    
-      plugin->api->next_request (next_cls, GNUNET_YES);
+                 _("Failed to transmit an item to the client; aborting iteration.\n"));
+      if (plugin != NULL)
+       plugin->api->next_request (next_cls, GNUNET_YES);
       return;
     }
   plugin->api->next_request (next_cls, GNUNET_NO);
@@ -893,6 +932,7 @@ handle_get (void *cls,
       GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
       return;
     }
+  GNUNET_SERVER_client_keep (client);
   msg = (const struct GetMessage*) message;
   if ( (size == sizeof(struct GetMessage)) &&
        (GNUNET_YES != GNUNET_CONTAINER_bloomfilter_test (filter,
@@ -905,13 +945,11 @@ handle_get (void *cls,
                  "GET",
                  GNUNET_h2s (&msg->key));
 #endif 
-      GNUNET_SERVER_client_keep (client);
       transmit_item (client,
                     NULL, NULL, 0, NULL, 0, 0, 0, 
                     GNUNET_TIME_UNIT_ZERO_ABS, 0);
       return;
     }
-  GNUNET_SERVER_client_keep (client);
   plugin->api->get (plugin->api->cls,
                    ((size == sizeof(struct GetMessage)) ? &msg->key : NULL),
                    NULL,
@@ -1193,11 +1231,11 @@ unload_plugin (struct DatastorePlugin *plug)
 
 
 /**
- * Last task run during shutdown.  Disconnects us from
- * the transport and core.
+ * Final task run after shutdown.  Unloads plugins and disconnects us from
+ * statistics.
  */
 static void
-cleaning_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+unload_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
   unload_plugin (plugin);
   plugin = NULL;
@@ -1210,6 +1248,40 @@ cleaning_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 }
 
 
+/**
+ * Last task run during shutdown.  Disconnects us from
+ * the transport and core.
+ */
+static void
+cleaning_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  struct TransmitCallbackContext *tcc;
+
+  while (NULL != (tcc = tcc_head))
+    {
+      GNUNET_CONTAINER_DLL_remove (tcc_head,
+                                  tcc_tail,
+                                  tcc);
+      if (tcc->th != NULL)
+       GNUNET_CONNECTION_notify_transmit_ready_cancel (tcc->th);
+      if (NULL != tcc->tc)
+       tcc->tc (tcc->tc_cls, GNUNET_SYSERR);
+      GNUNET_free (tcc->msg);
+      GNUNET_free (tcc);
+    }
+  if (expired_kill_task != GNUNET_SCHEDULER_NO_TASK)
+    {
+      GNUNET_SCHEDULER_cancel (sched,
+                              expired_kill_task);
+      expired_kill_task = GNUNET_SCHEDULER_NO_TASK;
+    }
+  GNUNET_SCHEDULER_add_continuation (sched,
+                                    &unload_task,
+                                    NULL,
+                                    GNUNET_SCHEDULER_REASON_PREREQ_DONE);
+}
+
+
 /**
  * Function that removes all active reservations made
  * by the given client and releases the space for other
@@ -1227,6 +1299,8 @@ cleanup_reservations (void *cls,
   struct ReservationList *prev;
   struct ReservationList *next;
 
+  if (client == NULL)
+    return;
   prev = NULL;
   pos = reservations;
   while (NULL != pos)