{
struct GNUNET_DATASTORE_QueueEntry *qe;
+ if (NULL != h->th)
+ {
+ GNUNET_CLIENT_notify_transmit_ready_cancel (h->th);
+ h->th = NULL;
+ }
if (h->client != NULL)
{
GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
h))
return;
GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
+ h->client = NULL;
}
GNUNET_break (0);
}
GNUNET_STATISTICS_destroy (h->stats,
GNUNET_NO);
+ h->stats = NULL;
GNUNET_free (h);
}
1,
GNUNET_NO);
qe->task = GNUNET_SCHEDULER_NO_TASK;
- GNUNET_assert (qe->was_transmitted == GNUNET_NO);
+ GNUNET_assert (qe->was_transmitted == GNUNET_NO);
qe->response_proc (qe->h, NULL);
}
pos = ret->next;
while (pos != NULL)
{
- if (pos->max_queue < h->queue_size)
+ if ( (pos->max_queue < h->queue_size) &&
+ (pos->was_transmitted == GNUNET_NO) )
{
GNUNET_assert (pos->response_proc != NULL);
/* move 'pos' element to head so that it will be
GNUNET_CONTAINER_DLL_insert (h->queue_head,
h->queue_tail,
pos);
+ GNUNET_STATISTICS_update (h->stats,
+ gettext_noop ("# Requests dropped from datastore queue"),
+ 1,
+ GNUNET_NO);
pos->response_proc (h, NULL);
break;
}
}
+/**
+ * Function called whenever we receive a message from
+ * the service. Calls the appropriate handler.
+ *
+ * @param cls the 'struct GNUNET_DATASTORE_Handle'
+ * @param msg the received message
+ */
+static void
+receive_cb (void *cls,
+ const struct GNUNET_MessageHeader *msg)
+{
+ struct GNUNET_DATASTORE_Handle *h = cls;
+ struct GNUNET_DATASTORE_QueueEntry *qe;
+
+ h->in_receive = GNUNET_NO;
+ if (h->skip_next_messages > 0)
+ {
+ h->skip_next_messages--;
+ process_queue (h);
+ return;
+ }
+ if (NULL == (qe = h->queue_head))
+ {
+ GNUNET_break (0);
+ process_queue (h);
+ return;
+ }
+ qe->response_proc (h, msg);
+}
+
+
/**
* Transmit request from queue to datastore service.
*
qe->was_transmitted = GNUNET_YES;
GNUNET_SCHEDULER_cancel (qe->task);
qe->task = GNUNET_SCHEDULER_NO_TASK;
+ GNUNET_assert (GNUNET_NO == h->in_receive);
h->in_receive = GNUNET_YES;
GNUNET_CLIENT_receive (h->client,
- qe->response_proc,
+ &receive_cb,
h,
GNUNET_TIME_absolute_get_remaining (qe->timeout));
GNUNET_STATISTICS_update (h->stats,
#endif
return; /* waiting for reconnect */
}
+ if (GNUNET_YES == h->in_receive)
+ {
+ /* wait for response to previous query */
+ return;
+ }
#if DEBUG_DATASTORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Queueing %u byte request to DATASTORE\n",
GNUNET_YES,
&transmit_request,
h);
+ GNUNET_assert (GNUNET_NO == h->in_receive);
+ GNUNET_break (NULL != h->th);
}
}
+/**
+ * Free a queue entry. Removes the given entry from the
+ * queue and releases associated resources. Does NOT
+ * call the callback.
+ *
+ * @param qe entry to free.
+ */
static void
free_queue_entry (struct GNUNET_DATASTORE_QueueEntry *qe)
{
GNUNET_free (qe);
}
+
/**
* Type of a function to call when we receive a message
* from the service.
int32_t status;
int was_transmitted;
- h->in_receive = GNUNET_NO;
- if (h->skip_next_messages > 0)
- {
- h->skip_next_messages--;
- process_queue (h);
- return;
- }
if (NULL == (qe = h->queue_head))
{
GNUNET_break (0);
_("Failed to receive status response from database."));
if (was_transmitted == GNUNET_YES)
do_disconnect (h);
+ else
+ process_queue (h);
return;
}
GNUNET_assert (GNUNET_YES == qe->was_transmitted);
struct ResultContext rc;
const struct DataMessage *dm;
- h->in_receive = GNUNET_NO;
- if (h->skip_next_messages > 0)
- {
- h->skip_next_messages--;
- process_queue (h);
- return;
- }
if (msg == NULL)
{
qe = h->queue_head;
NULL, 0, NULL, 0, 0, 0,
GNUNET_TIME_UNIT_ZERO_ABS, 0);
}
+ else
+ process_queue (h);
return;
}
if (ntohs(msg->type) == GNUNET_MESSAGE_TYPE_DATASTORE_DATA_END)
#endif
free_queue_entry (qe);
h->retry_time.rel_value = 0;
- rc.proc (rc.proc_cls,
- &dm->key,
- ntohl(dm->size),
- &dm[1],
- ntohl(dm->type),
- ntohl(dm->priority),
- ntohl(dm->anonymity),
- GNUNET_TIME_absolute_ntoh(dm->expiration),
- GNUNET_ntohll(dm->uid));
+ process_queue (h);
+ if (rc.proc != NULL)
+ rc.proc (rc.proc_cls,
+ &dm->key,
+ ntohl(dm->size),
+ &dm[1],
+ ntohl(dm->type),
+ ntohl(dm->priority),
+ ntohl(dm->anonymity),
+ GNUNET_TIME_absolute_ntoh(dm->expiration),
+ GNUNET_ntohll(dm->uid));
}