- if (mig_qe != NULL)
- {
- GNUNET_DATASTORE_cancel (mig_qe);
- mig_qe = NULL;
- }
- if (GNUNET_SCHEDULER_NO_TASK != mig_task)
- {
- GNUNET_SCHEDULER_cancel (sched, mig_task);
- mig_task = GNUNET_SCHEDULER_NO_TASK;
- }
- while (client_list != NULL)
- handle_client_disconnect (NULL,
- client_list->client);
- GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
- &clean_peer,
- NULL);
- GNUNET_break (0 == GNUNET_CONTAINER_heap_get_size (requests_by_expiration_heap));
- GNUNET_CONTAINER_heap_destroy (requests_by_expiration_heap);
- requests_by_expiration_heap = 0;
- GNUNET_CONTAINER_multihashmap_destroy (connected_peers);
- connected_peers = NULL;
- GNUNET_break (0 == GNUNET_CONTAINER_multihashmap_size (query_request_map));
- GNUNET_CONTAINER_multihashmap_destroy (query_request_map);
- query_request_map = NULL;
- GNUNET_break (0 == GNUNET_CONTAINER_multihashmap_size (peer_request_map));
- GNUNET_CONTAINER_multihashmap_destroy (peer_request_map);
- peer_request_map = NULL;
- GNUNET_assert (NULL != core);
- GNUNET_CORE_disconnect (core);
- core = NULL;
- if (stats != NULL)
- {
- GNUNET_STATISTICS_destroy (stats, GNUNET_NO);
- stats = NULL;
- }
- GNUNET_DATASTORE_disconnect (dsh,
- GNUNET_NO);
- while (mig_head != NULL)
- delete_migration_block (mig_head);
- GNUNET_assert (0 == mig_size);
- dsh = NULL;
- sched = NULL;
- cfg = NULL;
-}
-
-
-/* ******************* Utility functions ******************** */
-
-
-/**
- * Transmit messages by copying it to the target buffer
- * "buf". "buf" will be NULL and "size" zero if the socket was closed
- * for writing in the meantime. In that case, do nothing
- * (the disconnect or shutdown handler will take care of the rest).
- * If we were able to transmit messages and there are still more
- * pending, ask core again for further calls to this function.
- *
- * @param cls closure, pointer to the 'struct ConnectedPeer*'
- * @param size number of bytes available in buf
- * @param buf where the callee should write the message
- * @return number of bytes written to buf
- */
-static size_t
-transmit_to_peer (void *cls,
- size_t size, void *buf)
-{
- struct ConnectedPeer *cp = cls;
- char *cbuf = buf;
- struct GNUNET_PeerIdentity pid;
- struct PendingMessage *pm;
- struct MigrationReadyBlock *mb;
- struct MigrationReadyBlock *next;
- struct PutMessage migm;
- size_t msize;
- unsigned int i;
-
- cp->cth = NULL;
- if (NULL == buf)
- {
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Dropping message, core too busy.\n");
-#endif
- return 0;
- }
- msize = 0;
- while ( (NULL != (pm = cp->pending_messages_head) ) &&
- (pm->msize <= size) )
- {
- memcpy (&cbuf[msize], &pm[1], pm->msize);
- msize += pm->msize;
- size -= pm->msize;
- destroy_pending_message (pm, cp->pid);
- }
- if (NULL != pm)
- {
- GNUNET_PEER_resolve (cp->pid,
- &pid);
- cp->cth = GNUNET_CORE_notify_transmit_ready (core,
- pm->priority,
- GNUNET_CONSTANTS_SERVICE_TIMEOUT,
- &pid,
- pm->msize,
- &transmit_to_peer,
- cp);
- }
- else
- {
- next = mig_head;
- while (NULL != (mb = next))
- {
- next = mb->next;
- for (i=0;i<MIGRATION_LIST_SIZE;i++)
- {
- if ( (cp->pid == mb->target_list[i]) &&
- (mb->size + sizeof (migm) <= size) )
- {
- GNUNET_PEER_change_rc (mb->target_list[i], -1);
- mb->target_list[i] = 0;
- mb->used_targets++;
- migm.header.size = htons (sizeof (migm) + mb->size);
- migm.header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
- migm.type = htonl (mb->type);
- migm.expiration = GNUNET_TIME_absolute_hton (mb->expiration);
- memcpy (&cbuf[msize], &migm, sizeof (migm));
- msize += sizeof (migm);
- size -= sizeof (migm);
- memcpy (&cbuf[msize], &mb[1], mb->size);
- msize += mb->size;
- size -= mb->size;
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Pushing migration block `%s' (%u bytes) to `%s'\n",
- GNUNET_h2s (&mb->query),
- mb->size,
- GNUNET_i2s (&pid));
-#endif
- break;
- }
- else
- {
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Migration block `%s' (%u bytes) is not on migration list for peer `%s'\n",
- GNUNET_h2s (&mb->query),
- mb->size,
- GNUNET_i2s (&pid));
-#endif
- }
- }
- if ( (mb->used_targets >= MIGRATION_TARGET_COUNT) ||
- (mb->used_targets >= GNUNET_CONTAINER_multihashmap_size (connected_peers)) )
- {
- delete_migration_block (mb);
- consider_migration_gathering ();
- }
- }
- consider_migration (NULL,
- &pid.hashPubKey,
- cp);
- }
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Transmitting %u bytes to peer %u\n",
- msize,
- cp->pid);
-#endif
- return msize;
-}
-
-
-/**
- * Add a message to the set of pending messages for the given peer.
- *
- * @param cp peer to send message to
- * @param pm message to queue
- * @param pr request on which behalf this message is being queued
- */
-static void
-add_to_pending_messages_for_peer (struct ConnectedPeer *cp,
- struct PendingMessage *pm,
- struct PendingRequest *pr)
-{
- struct PendingMessage *pos;
- struct PendingMessageList *pml;
- struct GNUNET_PeerIdentity pid;
-
- GNUNET_assert (pm->next == NULL);
- GNUNET_assert (pm->pml == NULL);
- pml = GNUNET_malloc (sizeof (struct PendingMessageList));
- pml->req = pr;
- pml->target = cp;
- pml->pm = pm;
- pm->pml = pml;
- GNUNET_CONTAINER_DLL_insert (pr->pending_head,
- pr->pending_tail,
- pml);
- pos = cp->pending_messages_head;
- while ( (pos != NULL) &&
- (pm->priority < pos->priority) )
- pos = pos->next;
- GNUNET_CONTAINER_DLL_insert_after (cp->pending_messages_head,
- cp->pending_messages_tail,
- pos,
- pm);
- cp->pending_requests++;
- if (cp->pending_requests > MAX_QUEUE_PER_PEER)
- destroy_pending_message (cp->pending_messages_tail, 0);
- GNUNET_PEER_resolve (cp->pid, &pid);
- if (NULL != cp->cth)
- GNUNET_CORE_notify_transmit_ready_cancel (cp->cth);
- /* need to schedule transmission */
- cp->cth = GNUNET_CORE_notify_transmit_ready (core,
- cp->pending_messages_head->priority,
- MAX_TRANSMIT_DELAY,
- &pid,
- cp->pending_messages_head->msize,
- &transmit_to_peer,
- cp);
- if (cp->cth == NULL)
- {
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Failed to schedule transmission with core!\n");
-#endif
- GNUNET_STATISTICS_update (stats,
- gettext_noop ("# CORE transmission failures"),
- 1,
- GNUNET_NO);
- }
-}
-
-
-/**
- * Mingle hash with the mingle_number to produce different bits.
- */
-static void
-mingle_hash (const GNUNET_HashCode * in,
- int32_t mingle_number,
- GNUNET_HashCode * hc)
-{
- GNUNET_HashCode m;
-
- GNUNET_CRYPTO_hash (&mingle_number,
- sizeof (int32_t),
- &m);
- GNUNET_CRYPTO_hash_xor (&m, in, hc);
-}
-
-
-/**
- * Test if the load on this peer is too high
- * to even consider processing the query at
- * all.
- *
- * @return GNUNET_YES if the load is too high, GNUNET_NO otherwise
- */
-static int
-test_load_too_high ()
-{
- return GNUNET_NO; // FIXME
-}
-
-
-/* ******************* Pending Request Refresh Task ******************** */
-
-
-
-/**
- * We use a random delay to make the timing of requests less
- * predictable. This function returns such a random delay. We add a base
- * delay of MAX_CORK_DELAY (1s).
- *
- * FIXME: make schedule dependent on the specifics of the request?
- * Or bandwidth and number of connected peers and load?
- *
- * @return random delay to use for some request, between 1s and 1000+TTL_DECREMENT ms
- */
-static struct GNUNET_TIME_Relative
-get_processing_delay ()
-{
- return
- GNUNET_TIME_relative_add (GNUNET_CONSTANTS_MAX_CORK_DELAY,
- GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
- GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
- TTL_DECREMENT)));
-}
-
-
-/**
- * We're processing a GET request from another peer and have decided
- * to forward it to other peers. This function is called periodically
- * and should forward the request to other peers until we have all
- * possible replies. If we have transmitted the *only* reply to
- * the initiator we should destroy the pending request. If we have
- * many replies in the queue to the initiator, we should delay sending
- * out more queries until the reply queue has shrunk some.
- *
- * @param cls our "struct ProcessGetContext *"
- * @param tc unused
- */
-static void
-forward_request_task (void *cls,
- const struct GNUNET_SCHEDULER_TaskContext *tc);
-
-
-/**
- * Function called after we either failed or succeeded
- * at transmitting a query to a peer.
- *
- * @param cls the requests "struct PendingRequest*"
- * @param tpid ID of receiving peer, 0 on transmission error
- */
-static void
-transmit_query_continuation (void *cls,
- GNUNET_PEER_Id tpid)
-{
- struct PendingRequest *pr = cls;
-
- GNUNET_STATISTICS_update (stats,
- gettext_noop ("# queries scheduled for forwarding"),
- -1,
- GNUNET_NO);
- if (tpid == 0)
- {
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Transmission of request failed, will try again later.\n");
-#endif
- if (pr->task == GNUNET_SCHEDULER_NO_TASK)
- pr->task = GNUNET_SCHEDULER_add_delayed (sched,
- get_processing_delay (),
- &forward_request_task,
- pr);
- return;
- }
- GNUNET_STATISTICS_update (stats,
- gettext_noop ("# queries forwarded"),
- 1,
- GNUNET_NO);
- GNUNET_PEER_change_rc (tpid, 1);
- if (pr->used_pids_off == pr->used_pids_size)
- GNUNET_array_grow (pr->used_pids,
- pr->used_pids_size,
- pr->used_pids_size * 2 + 2);
- pr->used_pids[pr->used_pids_off++] = tpid;
- if (pr->task == GNUNET_SCHEDULER_NO_TASK)
- pr->task = GNUNET_SCHEDULER_add_delayed (sched,
- get_processing_delay (),
- &forward_request_task,
- pr);
-}
-
-
-/**
- * How many bytes should a bloomfilter be if we have already seen
- * entry_count responses? Note that BLOOMFILTER_K gives us the number
- * of bits set per entry. Furthermore, we should not re-size the
- * filter too often (to keep it cheap).
- *
- * Since other peers will also add entries but not resize the filter,
- * we should generally pick a slightly larger size than what the
- * strict math would suggest.
- *
- * @return must be a power of two and smaller or equal to 2^15.
- */
-static size_t
-compute_bloomfilter_size (unsigned int entry_count)
-{
- size_t size;
- unsigned int ideal = (entry_count * BLOOMFILTER_K) / 4;
- uint16_t max = 1 << 15;
-
- if (entry_count > max)
- return max;
- size = 8;
- while ((size < max) && (size < ideal))
- size *= 2;
- if (size > max)
- return max;
- return size;
-}
-
-
-/**
- * Recalculate our bloom filter for filtering replies. This function
- * will create a new bloom filter from scratch, so it should only be
- * called if we have no bloomfilter at all (and hence can create a
- * fresh one of minimal size without problems) OR if our peer is the
- * initiator (in which case we may resize to larger than mimimum size).
- *
- * @param pr request for which the BF is to be recomputed
- */
-static void
-refresh_bloomfilter (struct PendingRequest *pr)
-{
- unsigned int i;
- size_t nsize;
- GNUNET_HashCode mhash;
-
- nsize = compute_bloomfilter_size (pr->replies_seen_off);
- if (nsize == pr->bf_size)
- return; /* size not changed */
- if (pr->bf != NULL)
- GNUNET_CONTAINER_bloomfilter_free (pr->bf);
- pr->bf_size = nsize;
- pr->mingle = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, -1);
- pr->bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
- pr->bf_size,
- BLOOMFILTER_K);
- for (i=0;i<pr->replies_seen_off;i++)
- {
- mingle_hash (&pr->replies_seen[i], pr->mingle, &mhash);
- GNUNET_CONTAINER_bloomfilter_add (pr->bf, &mhash);
- }
-}
-
-
-/**
- * Function called after we've tried to reserve a certain amount of
- * bandwidth for a reply. Check if we succeeded and if so send our
- * query.
- *
- * @param cls the requests "struct PendingRequest*"
- * @param peer identifies the peer
- * @param bpm_in set to the current bandwidth limit (receiving) for this peer
- * @param bpm_out set to the current bandwidth limit (sending) for this peer
- * @param amount set to the amount that was actually reserved or unreserved
- * @param preference current traffic preference for the given peer
- */
-static void
-target_reservation_cb (void *cls,
- const struct
- GNUNET_PeerIdentity * peer,
- struct GNUNET_BANDWIDTH_Value32NBO bpm_in,
- struct GNUNET_BANDWIDTH_Value32NBO bpm_out,
- int amount,
- uint64_t preference)
-{
- struct PendingRequest *pr = cls;
- struct ConnectedPeer *cp;
- struct PendingMessage *pm;
- struct GetMessage *gm;
- GNUNET_HashCode *ext;
- char *bfdata;
- size_t msize;
- unsigned int k;
- int no_route;
- uint32_t bm;
-
- pr->irc = NULL;
- if (peer == NULL)
- {
- /* error in communication with core, try again later */
- if (pr->task == GNUNET_SCHEDULER_NO_TASK)
- pr->task = GNUNET_SCHEDULER_add_delayed (sched,
- get_processing_delay (),
- &forward_request_task,
- pr);
- return;
- }
- // (3) transmit, update ttl/priority
- cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
- &peer->hashPubKey);
- if (cp == NULL)
- {
- /* Peer must have just left */
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Selected peer disconnected!\n");
-#endif
- if (pr->task == GNUNET_SCHEDULER_NO_TASK)
- pr->task = GNUNET_SCHEDULER_add_delayed (sched,
- get_processing_delay (),
- &forward_request_task,
- pr);
- return;
- }
- no_route = GNUNET_NO;
- if (amount == 0)
- {
- if (pr->cp == NULL)
- {
-#if DEBUG_FS > 1
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Failed to reserve bandwidth for reply (got %d/%u bytes only)!\n",
- amount,
- DBLOCK_SIZE);
-#endif
- GNUNET_STATISTICS_update (stats,
- gettext_noop ("# reply bandwidth reservation requests failed"),
- 1,
- GNUNET_NO);
- if (pr->task == GNUNET_SCHEDULER_NO_TASK)
- pr->task = GNUNET_SCHEDULER_add_delayed (sched,
- get_processing_delay (),
- &forward_request_task,
- pr);
- return; /* this target round failed */
- }
- /* FIXME: if we are "quite" busy, we may still want to skip
- this round; need more load detection code! */
- no_route = GNUNET_YES;
- }
-
- GNUNET_STATISTICS_update (stats,
- gettext_noop ("# queries scheduled for forwarding"),
- 1,
- GNUNET_NO);
- /* build message and insert message into priority queue */
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Forwarding request `%s' to `%4s'!\n",
- GNUNET_h2s (&pr->query),
- GNUNET_i2s (peer));
-#endif
- k = 0;
- bm = 0;
- if (GNUNET_YES == no_route)
- {
- bm |= GET_MESSAGE_BIT_RETURN_TO;
- k++;
- }
- if (pr->namespace != NULL)
- {
- bm |= GET_MESSAGE_BIT_SKS_NAMESPACE;
- k++;
- }
- if (pr->target_pid != 0)
- {
- bm |= GET_MESSAGE_BIT_TRANSMIT_TO;
- k++;
- }
- msize = sizeof (struct GetMessage) + pr->bf_size + k * sizeof(GNUNET_HashCode);
- GNUNET_assert (msize < GNUNET_SERVER_MAX_MESSAGE_SIZE);
- pm = GNUNET_malloc (sizeof (struct PendingMessage) + msize);
- pm->msize = msize;
- gm = (struct GetMessage*) &pm[1];
- gm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_GET);
- gm->header.size = htons (msize);
- gm->type = htonl (pr->type);
- pr->remaining_priority /= 2;
- gm->priority = htonl (pr->remaining_priority);
- gm->ttl = htonl (pr->ttl);
- gm->filter_mutator = htonl(pr->mingle);
- gm->hash_bitmap = htonl (bm);
- gm->query = pr->query;
- ext = (GNUNET_HashCode*) &gm[1];
- k = 0;
- if (GNUNET_YES == no_route)
- GNUNET_PEER_resolve (pr->cp->pid, (struct GNUNET_PeerIdentity*) &ext[k++]);
- if (pr->namespace != NULL)
- memcpy (&ext[k++], pr->namespace, sizeof (GNUNET_HashCode));
- if (pr->target_pid != 0)
- GNUNET_PEER_resolve (pr->target_pid, (struct GNUNET_PeerIdentity*) &ext[k++]);
- bfdata = (char *) &ext[k];
- if (pr->bf != NULL)
- GNUNET_CONTAINER_bloomfilter_get_raw_data (pr->bf,
- bfdata,
- pr->bf_size);
- pm->cont = &transmit_query_continuation;
- pm->cont_cls = pr;
- add_to_pending_messages_for_peer (cp, pm, pr);
-}
-
-
-/**
- * Closure used for "target_peer_select_cb".
- */
-struct PeerSelectionContext
-{
- /**
- * The request for which we are selecting
- * peers.
- */
- struct PendingRequest *pr;
-
- /**
- * Current "prime" target.
- */
- struct GNUNET_PeerIdentity target;
-
- /**
- * How much do we like this target?
- */
- double target_score;
-
-};
-
-
-/**
- * Function called for each connected peer to determine
- * which one(s) would make good targets for forwarding.
- *
- * @param cls closure (struct PeerSelectionContext)
- * @param key current key code (peer identity)
- * @param value value in the hash map (struct ConnectedPeer)
- * @return GNUNET_YES if we should continue to
- * iterate,
- * GNUNET_NO if not.
- */
-static int
-target_peer_select_cb (void *cls,
- const GNUNET_HashCode * key,
- void *value)
-{
- struct PeerSelectionContext *psc = cls;
- struct ConnectedPeer *cp = value;
- struct PendingRequest *pr = psc->pr;
- double score;
- unsigned int i;
- unsigned int pc;
-
- /* 1) check that this peer is not the initiator */
- if (cp == pr->cp)
- {
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Skipping initiator in forwarding selection\n");
-#endif
- return GNUNET_YES; /* skip */
- }
-
- /* 2) check if we have already (recently) forwarded to this peer */
- pc = 0;
- for (i=0;i<pr->used_pids_off;i++)
- if (pr->used_pids[i] == cp->pid)
- {
- pc++;
- if (0 != GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
- RETRY_PROBABILITY_INV))
- {
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "NOT re-trying query that was previously transmitted %u times\n",
- (unsigned int) pr->used_pids_off);
-#endif
- return GNUNET_YES; /* skip */
- }
- }
-#if DEBUG_FS
- if (0 < pc)
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Re-trying query that was previously transmitted %u times to this peer\n",
- (unsigned int) pc);
-#endif
- /* 3) calculate how much we'd like to forward to this peer,
- starting with a random value that is strong enough
- to at least give any peer a chance sometimes
- (compared to the other factors that come later) */
- /* 3a) count successful (recent) routes from cp for same source */
- if (pr->cp != NULL)
- {
- score = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
- P2P_SUCCESS_LIST_SIZE);
- for (i=0;i<P2P_SUCCESS_LIST_SIZE;i++)
- if (cp->last_p2p_replies[i] == pr->cp->pid)
- score += 1; /* likely successful based on hot path */
- }
- else
- {
- score = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
- CS2P_SUCCESS_LIST_SIZE);
- for (i=0;i<CS2P_SUCCESS_LIST_SIZE;i++)
- if (cp->last_client_replies[i] == pr->client_request_list->client_list->client)
- score += 1; /* likely successful based on hot path */
- }
- /* 3b) include latency */
- if (cp->avg_delay.value < 4 * TTL_DECREMENT)
- score += 1; /* likely fast based on latency */
- /* 3c) include priorities */
- if (cp->avg_priority <= pr->remaining_priority / 2.0)
- score += 1; /* likely successful based on priorities */
- /* 3d) penalize for queue size */
- score -= (2.0 * cp->pending_requests / (double) MAX_QUEUE_PER_PEER);
- /* 3e) include peer proximity */
- score -= (2.0 * (GNUNET_CRYPTO_hash_distance_u32 (key,
- &pr->query)) / (double) UINT32_MAX);
- /* store best-fit in closure */
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Peer `%s' gets score %f for forwarding query, max is %f\n",
- GNUNET_h2s (key),
- score,
- psc->target_score);
-#endif
- score++; /* avoid zero */
- if (score > psc->target_score)
- {
- psc->target_score = score;
- psc->target.hashPubKey = *key;
- }
- return GNUNET_YES;
-}
-
-
-/**
- * The priority level imposes a bound on the maximum
- * value for the ttl that can be requested.
- *
- * @param ttl_in requested ttl
- * @param prio given priority
- * @return ttl_in if ttl_in is below the limit,
- * otherwise the ttl-limit for the given priority
- */
-static int32_t
-bound_ttl (int32_t ttl_in, uint32_t prio)
-{
- unsigned long long allowed;
-
- if (ttl_in <= 0)
- return ttl_in;
- allowed = ((unsigned long long) prio) * TTL_DECREMENT / 1000;
- if (ttl_in > allowed)
- {
- if (allowed >= (1 << 30))
- return 1 << 30;
- return allowed;
- }
- return ttl_in;
-}
-
-
-/**
- * We're processing a GET request and have decided
- * to forward it to other peers. This function is called periodically
- * and should forward the request to other peers until we have all
- * possible replies. If we have transmitted the *only* reply to
- * the initiator we should destroy the pending request. If we have
- * many replies in the queue to the initiator, we should delay sending
- * out more queries until the reply queue has shrunk some.
- *
- * @param cls our "struct ProcessGetContext *"
- * @param tc unused
- */
-static void
-forward_request_task (void *cls,
- const struct GNUNET_SCHEDULER_TaskContext *tc)
-{
- struct PendingRequest *pr = cls;
- struct PeerSelectionContext psc;
- struct ConnectedPeer *cp;
- struct GNUNET_TIME_Relative delay;
-
- pr->task = GNUNET_SCHEDULER_NO_TASK;
- if (pr->irc != NULL)
- {
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Forwarding of query `%s' not attempted due to pending local lookup!\n",
- GNUNET_h2s (&pr->query));
-#endif
- return; /* already pending */
- }
- if (GNUNET_YES == pr->local_only)
- return; /* configured to not do P2P search */
- /* (1) select target */
- psc.pr = pr;
- psc.target_score = -DBL_MAX;
- GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
- &target_peer_select_cb,
- &psc);
- if (psc.target_score == -DBL_MAX)
- {
- delay = get_processing_delay ();
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "No peer selected for forwarding of query `%s', will try again in %llu ms!\n",
- GNUNET_h2s (&pr->query),
- delay.value);
-#endif
- pr->task = GNUNET_SCHEDULER_add_delayed (sched,
- delay,
- &forward_request_task,
- pr);
- return; /* nobody selected */
- }
- /* (3) update TTL/priority */
- if (pr->client_request_list != NULL)
- {
- /* FIXME: use better algorithm!? */
- if (0 == GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
- 4))
- pr->priority++;
- /* FIXME: bound priority by "customary" priority used by other peers
- at this time! */
- pr->ttl = bound_ttl (pr->ttl + TTL_DECREMENT * 2,
- pr->priority);
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Trying query `%s' with priority %u and TTL %d.\n",
- GNUNET_h2s (&pr->query),
- pr->priority,
- pr->ttl);
-#endif
- }
-
- /* (3) reserve reply bandwidth */
- cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
- &psc.target.hashPubKey);
- GNUNET_assert (NULL != cp);
- pr->irc = GNUNET_CORE_peer_change_preference (sched, cfg,
- &psc.target,
- GNUNET_CONSTANTS_SERVICE_TIMEOUT,
- GNUNET_BANDWIDTH_value_init (UINT32_MAX),
- DBLOCK_SIZE * 2,
- cp->inc_preference,
- &target_reservation_cb,
- pr);
- cp->inc_preference = 0;
-}
-
-
-/* **************************** P2P PUT Handling ************************ */
-
-
-/**
- * Function called after we either failed or succeeded
- * at transmitting a reply to a peer.
- *
- * @param cls the requests "struct PendingRequest*"
- * @param tpid ID of receiving peer, 0 on transmission error
- */
-static void
-transmit_reply_continuation (void *cls,
- GNUNET_PEER_Id tpid)
-{
- struct PendingRequest *pr = cls;
-
- switch (pr->type)
- {
- case GNUNET_BLOCK_TYPE_DBLOCK:
- case GNUNET_BLOCK_TYPE_IBLOCK:
- /* only one reply expected, done with the request! */
- destroy_pending_request (pr);
- break;
- case GNUNET_BLOCK_TYPE_ANY:
- case GNUNET_BLOCK_TYPE_KBLOCK:
- case GNUNET_BLOCK_TYPE_SBLOCK:
- break;
- default:
- GNUNET_break (0);
- break;
- }
-}
-
-
-/**
- * Transmit the given message by copying it to the target buffer
- * "buf". "buf" will be NULL and "size" zero if the socket was closed
- * for writing in the meantime. In that case, do nothing
- * (the disconnect or shutdown handler will take care of the rest).
- * If we were able to transmit messages and there are still more
- * pending, ask core again for further calls to this function.
- *
- * @param cls closure, pointer to the 'struct ClientList*'
- * @param size number of bytes available in buf
- * @param buf where the callee should write the message
- * @return number of bytes written to buf
- */
-static size_t
-transmit_to_client (void *cls,
- size_t size, void *buf)
-{
- struct ClientList *cl = cls;
- char *cbuf = buf;
- struct ClientResponseMessage *creply;
- size_t msize;
-
- cl->th = NULL;
- if (NULL == buf)
- {
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Not sending reply, client communication problem.\n");
-#endif
- return 0;
- }
- msize = 0;
- while ( (NULL != (creply = cl->res_head) ) &&
- (creply->msize <= size) )
- {
- memcpy (&cbuf[msize], &creply[1], creply->msize);
- msize += creply->msize;
- size -= creply->msize;
- GNUNET_CONTAINER_DLL_remove (cl->res_head,
- cl->res_tail,
- creply);
- GNUNET_free (creply);
- }
- if (NULL != creply)
- cl->th = GNUNET_SERVER_notify_transmit_ready (cl->client,
- creply->msize,
- GNUNET_TIME_UNIT_FOREVER_REL,
- &transmit_to_client,
- cl);
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Transmitted %u bytes to client\n",
- (unsigned int) msize);
-#endif
- return msize;
-}
-
-
-/**
- * Closure for "process_reply" function.
- */
-struct ProcessReplyClosure
-{
- /**
- * The data for the reply.
- */
- const void *data;
-
- /**
- * Who gave us this reply? NULL for local host.
- */
- struct ConnectedPeer *sender;
-
- /**
- * When the reply expires.
- */
- struct GNUNET_TIME_Absolute expiration;
-
- /**
- * Size of data.
- */
- size_t size;
-
- /**
- * Namespace that this reply belongs to
- * (if it is of type SBLOCK).
- */
- GNUNET_HashCode namespace;
-
- /**
- * Type of the block.
- */
- enum GNUNET_BLOCK_Type type;
-
- /**
- * How much was this reply worth to us?
- */
- uint32_t priority;
-
- /**
- * Did we finish processing the associated request?
- */
- int finished;
-};
-
-
-/**
- * We have received a reply; handle it!
- *
- * @param cls response (struct ProcessReplyClosure)
- * @param key our query
- * @param value value in the hash map (info about the query)
- * @return GNUNET_YES (we should continue to iterate)
- */
-static int
-process_reply (void *cls,
- const GNUNET_HashCode * key,
- void *value)
-{
- struct ProcessReplyClosure *prq = cls;
- struct PendingRequest *pr = value;
- struct PendingMessage *reply;
- struct ClientResponseMessage *creply;
- struct ClientList *cl;
- struct PutMessage *pm;
- struct ConnectedPeer *cp;
- struct GNUNET_TIME_Relative cur_delay;
- GNUNET_HashCode chash;
- GNUNET_HashCode mhash;
- size_t msize;
-
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Matched result (type %u) for query `%s' with pending request\n",
- (unsigned int) prq->type,
- GNUNET_h2s (key));
-#endif
- GNUNET_STATISTICS_update (stats,
- gettext_noop ("# replies received and matched"),
- 1,
- GNUNET_NO);
- if (prq->sender != NULL)
- {
- /* FIXME: should we be more precise here and not use
- "start_time" but a peer-specific time stamp? */
- cur_delay = GNUNET_TIME_absolute_get_duration (pr->start_time);
- prq->sender->avg_delay.value
- = (prq->sender->avg_delay.value *
- (RUNAVG_DELAY_N - 1) + cur_delay.value) / RUNAVG_DELAY_N;
- prq->sender->avg_priority
- = (prq->sender->avg_priority *
- (RUNAVG_DELAY_N - 1) + pr->priority) / (double) RUNAVG_DELAY_N;
- if (pr->cp != NULL)
- {
- GNUNET_PEER_change_rc (prq->sender->last_p2p_replies
- [prq->sender->last_p2p_replies_woff % P2P_SUCCESS_LIST_SIZE],
- -1);
- GNUNET_PEER_change_rc (pr->cp->pid, 1);
- prq->sender->last_p2p_replies
- [(prq->sender->last_p2p_replies_woff++) % P2P_SUCCESS_LIST_SIZE]
- = pr->cp->pid;
- }
- else
- {
- if (NULL != prq->sender->last_client_replies
- [(prq->sender->last_client_replies_woff) % CS2P_SUCCESS_LIST_SIZE])
- GNUNET_SERVER_client_drop (prq->sender->last_client_replies
- [(prq->sender->last_client_replies_woff) % CS2P_SUCCESS_LIST_SIZE]);
- prq->sender->last_client_replies
- [(prq->sender->last_client_replies_woff++) % CS2P_SUCCESS_LIST_SIZE]
- = pr->client_request_list->client_list->client;
- GNUNET_SERVER_client_keep (pr->client_request_list->client_list->client);
- }
- }
- GNUNET_CRYPTO_hash (prq->data,
- prq->size,
- &chash);
- switch (prq->type)
- {
- case GNUNET_BLOCK_TYPE_DBLOCK:
- case GNUNET_BLOCK_TYPE_IBLOCK:
- /* only possible reply, stop requesting! */
- while (NULL != pr->pending_head)
- destroy_pending_message_list_entry (pr->pending_head);
- if (pr->qe != NULL)
- {
- if (pr->client_request_list != NULL)
- GNUNET_SERVER_receive_done (pr->client_request_list->client_list->client,
- GNUNET_YES);
- GNUNET_DATASTORE_cancel (pr->qe);
- pr->qe = NULL;
- }
- pr->do_remove = GNUNET_YES;
- if (pr->task != GNUNET_SCHEDULER_NO_TASK)
- {
- GNUNET_SCHEDULER_cancel (sched,
- pr->task);
- pr->task = GNUNET_SCHEDULER_NO_TASK;
- }
- GNUNET_break (GNUNET_YES ==
- GNUNET_CONTAINER_multihashmap_remove (query_request_map,
- key,
- pr));
- break;
- case GNUNET_BLOCK_TYPE_SBLOCK:
- if (pr->namespace == NULL)
- {
- GNUNET_break (0);
- return GNUNET_YES;
- }
- if (0 != memcmp (pr->namespace,
- &prq->namespace,
- sizeof (GNUNET_HashCode)))
- {
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- _("Reply mismatched in terms of namespace. Discarded.\n"));
- return GNUNET_YES; /* wrong namespace */
- }
- /* then: fall-through! */
- case GNUNET_BLOCK_TYPE_KBLOCK:
- case GNUNET_BLOCK_TYPE_NBLOCK:
- if (pr->bf != NULL)
- {
- mingle_hash (&chash, pr->mingle, &mhash);
- if (GNUNET_YES == GNUNET_CONTAINER_bloomfilter_test (pr->bf,
- &mhash))
- {
- GNUNET_STATISTICS_update (stats,
- gettext_noop ("# duplicate replies discarded (bloomfilter)"),
- 1,
- GNUNET_NO);
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Duplicate response `%s', discarding.\n",
- GNUNET_h2s (&mhash));
-#endif
- return GNUNET_YES; /* duplicate */
- }
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "New response `%s', adding to filter.\n",
- GNUNET_h2s (&mhash));
-#endif
- }
- if (pr->client_request_list != NULL)
- {
- if (pr->replies_seen_size == pr->replies_seen_off)
- GNUNET_array_grow (pr->replies_seen,
- pr->replies_seen_size,
- pr->replies_seen_size * 2 + 4);
- pr->replies_seen[pr->replies_seen_off++] = chash;
- }
- if ( (pr->bf == NULL) ||
- (pr->client_request_list != NULL) )
- refresh_bloomfilter (pr);
- GNUNET_CONTAINER_bloomfilter_add (pr->bf,
- &mhash);
- break;
- default:
- GNUNET_break (0);
- return GNUNET_YES;
- }
- prq->priority += pr->remaining_priority;
- pr->remaining_priority = 0;
- if (NULL != pr->client_request_list)
- {
- GNUNET_STATISTICS_update (stats,
- gettext_noop ("# replies received for local clients"),
- 1,
- GNUNET_NO);
- cl = pr->client_request_list->client_list;
- msize = sizeof (struct PutMessage) + prq->size;
- creply = GNUNET_malloc (msize + sizeof (struct ClientResponseMessage));
- creply->msize = msize;
- creply->client_list = cl;
- GNUNET_CONTAINER_DLL_insert_after (cl->res_head,
- cl->res_tail,
- cl->res_tail,
- creply);
- pm = (struct PutMessage*) &creply[1];
- pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
- pm->header.size = htons (msize);
- pm->type = htonl (prq->type);
- pm->expiration = GNUNET_TIME_absolute_hton (prq->expiration);
- memcpy (&pm[1], prq->data, prq->size);
- if (NULL == cl->th)
- {
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Transmitting result for query `%s' to client\n",
- GNUNET_h2s (key));
-#endif
- cl->th = GNUNET_SERVER_notify_transmit_ready (cl->client,
- msize,
- GNUNET_TIME_UNIT_FOREVER_REL,
- &transmit_to_client,
- cl);
- }
- GNUNET_break (cl->th != NULL);
- if (pr->do_remove)
- {
- prq->finished = GNUNET_YES;
- destroy_pending_request (pr);
- }
- }
- else
- {
- cp = pr->cp;
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Transmitting result for query `%s' to other peer (PID=%u)\n",
- GNUNET_h2s (key),
- (unsigned int) cp->pid);
-#endif
- GNUNET_STATISTICS_update (stats,
- gettext_noop ("# replies received for other peers"),
- 1,
- GNUNET_NO);
- msize = sizeof (struct PutMessage) + prq->size;
- reply = GNUNET_malloc (msize + sizeof (struct PendingMessage));
- reply->cont = &transmit_reply_continuation;
- reply->cont_cls = pr;
- reply->msize = msize;
- reply->priority = UINT32_MAX; /* send replies first! */
- pm = (struct PutMessage*) &reply[1];
- pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
- pm->header.size = htons (msize);
- pm->type = htonl (prq->type);
- pm->expiration = GNUNET_TIME_absolute_hton (prq->expiration);
- memcpy (&pm[1], prq->data, prq->size);
- add_to_pending_messages_for_peer (cp, reply, pr);
- }
- return GNUNET_YES;
-}
-
-
-/**
- * Continuation called to notify client about result of the
- * operation.
- *
- * @param cls closure
- * @param success GNUNET_SYSERR on failure
- * @param msg NULL on success, otherwise an error message
- */
-static void
-put_migration_continuation (void *cls,
- int success,
- const char *msg)
-{
- /* FIXME */
-}
-
-
-/**
- * Handle P2P "PUT" message.
- *
- * @param cls closure, always NULL
- * @param other the other peer involved (sender or receiver, NULL
- * for loopback messages where we are both sender and receiver)
- * @param message the actual message
- * @param latency reported latency of the connection with 'other'
- * @param distance reported distance (DV) to 'other'
- * @return GNUNET_OK to keep the connection open,
- * GNUNET_SYSERR to close it (signal serious error)
- */
-static int
-handle_p2p_put (void *cls,
- const struct GNUNET_PeerIdentity *other,
- const struct GNUNET_MessageHeader *message,
- struct GNUNET_TIME_Relative latency,
- uint32_t distance)
-{
- const struct PutMessage *put;
- uint16_t msize;
- size_t dsize;
- enum GNUNET_BLOCK_Type type;
- struct GNUNET_TIME_Absolute expiration;
- GNUNET_HashCode query;
- struct ProcessReplyClosure prq;
- const struct SBlock *sb;
- struct ConnectedPeer *cps;
-
- msize = ntohs (message->size);
- if (msize < sizeof (struct PutMessage))
- {
- GNUNET_break_op(0);
- return GNUNET_SYSERR;
- }
- put = (const struct PutMessage*) message;
- dsize = msize - sizeof (struct PutMessage);
- type = ntohl (put->type);
- expiration = GNUNET_TIME_absolute_ntoh (put->expiration);
-
- if (GNUNET_OK !=
- GNUNET_BLOCK_check_block (type,
- &put[1],
- dsize,
- &query))
- {
- GNUNET_break_op (0);
- return GNUNET_SYSERR;
- }
- if (type == GNUNET_BLOCK_TYPE_ONDEMAND)
- return GNUNET_SYSERR;
- if (GNUNET_BLOCK_TYPE_SBLOCK == type)
- {
- sb = (const struct SBlock*) &put[1];
- GNUNET_CRYPTO_hash (&sb->subspace,
- sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
- &prq.namespace);
- }
-
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Received result for query `%s' from peer `%4s'\n",
- GNUNET_h2s (&query),
- GNUNET_i2s (other));
-#endif
- GNUNET_STATISTICS_update (stats,
- gettext_noop ("# replies received (overall)"),
- 1,
- GNUNET_NO);
- /* now, lookup 'query' */
- prq.data = (const void*) &put[1];
- if (other != NULL)
- prq.sender = GNUNET_CONTAINER_multihashmap_get (connected_peers,
- &other->hashPubKey);
- prq.size = dsize;
- prq.type = type;
- prq.expiration = expiration;
- prq.priority = 0;
- prq.finished = GNUNET_NO;
- GNUNET_CONTAINER_multihashmap_get_multiple (query_request_map,
- &query,
- &process_reply,
- &prq);
- cps = GNUNET_CONTAINER_multihashmap_get (connected_peers,
- &other->hashPubKey);
- if (cps != NULL)
- {
- cps->inc_preference += CONTENT_BANDWIDTH_VALUE + 1000 * prq.priority;
- cps->trust_delta += prq.priority;
- }
- else
- {
- GNUNET_break (0);
- }
- if (GNUNET_YES == active_migration)
- {
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Replicating result for query `%s' with priority %u\n",
- GNUNET_h2s (&query),
- prq.priority);
-#endif
- GNUNET_DATASTORE_put (dsh,
- 0, &query, dsize, &put[1],
- type, prq.priority, 1 /* anonymity */,
- expiration,
- 1 + prq.priority, MAX_DATASTORE_QUEUE,
- GNUNET_CONSTANTS_SERVICE_TIMEOUT,
- &put_migration_continuation,
- NULL);
- }
- return GNUNET_OK;
-}
-
-
-/* **************************** P2P GET Handling ************************ */
-
-
-/**
- * Closure for 'check_duplicate_request_{peer,client}'.
- */
-struct CheckDuplicateRequestClosure
-{
- /**
- * The new request we should check if it already exists.
- */
- const struct PendingRequest *pr;
-
- /**
- * Existing request found by the checker, NULL if none.
- */
- struct PendingRequest *have;
-};
-
-
-/**
- * Iterator over entries in the 'query_request_map' that
- * tries to see if we have the same request pending from
- * the same client already.
- *
- * @param cls closure (our 'struct CheckDuplicateRequestClosure')
- * @param key current key code (query, ignored, must match)
- * @param value value in the hash map (a 'struct PendingRequest'
- * that already exists)
- * @return GNUNET_YES if we should continue to
- * iterate (no match yet)
- * GNUNET_NO if not (match found).
- */
-static int
-check_duplicate_request_client (void *cls,
- const GNUNET_HashCode * key,
- void *value)
-{
- struct CheckDuplicateRequestClosure *cdc = cls;
- struct PendingRequest *have = value;
-
- if (have->client_request_list == NULL)
- return GNUNET_YES;
- if ( (cdc->pr->client_request_list->client_list->client == have->client_request_list->client_list->client) &&
- (cdc->pr != have) )
- {
- cdc->have = have;
- return GNUNET_NO;
- }
- return GNUNET_YES;
-}
-
-
-/**
- * We're processing (local) results for a search request
- * from another peer. Pass applicable results to the
- * peer and if we are done either clean up (operation
- * complete) or forward to other peers (more results possible).
- *
- * @param cls our closure (struct LocalGetContext)
- * @param key key for the content
- * @param size number of bytes in data
- * @param data content stored
- * @param type type of the content
- * @param priority priority of the content
- * @param anonymity anonymity-level for the content
- * @param expiration expiration time for the content
- * @param uid unique identifier for the datum;
- * maybe 0 if no unique identifier is available
- */
-static void
-process_local_reply (void *cls,
- const GNUNET_HashCode * key,
- uint32_t size,
- const void *data,
- enum GNUNET_BLOCK_Type type,
- uint32_t priority,
- uint32_t anonymity,
- struct GNUNET_TIME_Absolute
- expiration,
- uint64_t uid)
-{
- struct PendingRequest *pr = cls;
- struct ProcessReplyClosure prq;
- struct CheckDuplicateRequestClosure cdrc;
- const struct SBlock *sb;
- GNUNET_HashCode dhash;
- GNUNET_HashCode mhash;
- GNUNET_HashCode query;
-
- if (NULL == key)
- {
-#if DEBUG_FS > 1
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Done processing local replies, forwarding request to other peers.\n");
-#endif
- pr->qe = NULL;
- if (pr->client_request_list != NULL)
- {
- GNUNET_SERVER_receive_done (pr->client_request_list->client_list->client,
- GNUNET_YES);
- /* Figure out if this is a duplicate request and possibly
- merge 'struct PendingRequest' entries */
- cdrc.have = NULL;
- cdrc.pr = pr;
- GNUNET_CONTAINER_multihashmap_get_multiple (query_request_map,
- &pr->query,
- &check_duplicate_request_client,
- &cdrc);
- if (cdrc.have != NULL)
- {
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Received request for block `%s' twice from client, will only request once.\n",
- GNUNET_h2s (&pr->query));
-#endif
-
- destroy_pending_request (pr);
- return;
- }
- }
-
- /* no more results */
- if (pr->task == GNUNET_SCHEDULER_NO_TASK)
- pr->task = GNUNET_SCHEDULER_add_now (sched,
- &forward_request_task,
- pr);
- return;
- }
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "New local response to `%s' of type %u.\n",
- GNUNET_h2s (key),
- type);
-#endif
- if (type == GNUNET_BLOCK_TYPE_ONDEMAND)
- {
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Found ONDEMAND block, performing on-demand encoding\n");
-#endif
- GNUNET_STATISTICS_update (stats,
- gettext_noop ("# on-demand blocks matched requests"),
- 1,
- GNUNET_NO);
- if (GNUNET_OK !=
- GNUNET_FS_handle_on_demand_block (key, size, data, type, priority,
- anonymity, expiration, uid,
- &process_local_reply,
- pr))
- if (pr->qe != NULL)
- GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
- return;
- }
- /* check for duplicates */
- GNUNET_CRYPTO_hash (data, size, &dhash);
- mingle_hash (&dhash,
- pr->mingle,
- &mhash);
- if ( (pr->bf != NULL) &&
- (GNUNET_YES ==
- GNUNET_CONTAINER_bloomfilter_test (pr->bf,
- &mhash)) )
- {
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Result from datastore filtered by bloomfilter (duplicate).\n");
-#endif
- GNUNET_STATISTICS_update (stats,
- gettext_noop ("# results filtered by query bloomfilter"),
- 1,
- GNUNET_NO);
- if (pr->qe != NULL)
- GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
- return;
- }
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Found result for query `%s' in local datastore\n",
- GNUNET_h2s (key));
-#endif
- GNUNET_STATISTICS_update (stats,
- gettext_noop ("# results found locally"),
- 1,
- GNUNET_NO);
- pr->results_found++;
- memset (&prq, 0, sizeof (prq));
- prq.data = data;
- prq.expiration = expiration;
- prq.size = size;
- if (GNUNET_BLOCK_TYPE_SBLOCK == type)
- {
- sb = (const struct SBlock*) data;
- GNUNET_CRYPTO_hash (&sb->subspace,
- sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
- &prq.namespace);
- }
- if (GNUNET_OK != GNUNET_BLOCK_check_block (type,
- data,
- size,
- &query))
- {
- GNUNET_break (0);
- GNUNET_DATASTORE_remove (dsh,
- key,
- size, data,
- -1, -1,
- GNUNET_TIME_UNIT_FOREVER_REL,
- NULL, NULL);
- GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
- return;
- }
- prq.type = type;
- prq.priority = priority;
- prq.finished = GNUNET_NO;
- process_reply (&prq, key, pr);
- if (prq.finished == GNUNET_YES)
projects / oweals / gnunet.git / blobdiff