- * How many content messages have we received 'recently' that
- * have not yet been claimed as cover traffic?
- */
-static unsigned int cover_content_count;
-
-/**
- * ID of our task that we use to age the cover counters.
- */
-static GNUNET_SCHEDULER_TaskIdentifier cover_age_task;
-
-
-static void
-age_cover_counters (void *cls,
- const struct GNUNET_SCHEDULER_TaskContext *tc)
-{
- cover_content_count = (cover_content_count * 15) / 16;
- cover_query_count = (cover_query_count * 15) / 16;
- cover_age_task = GNUNET_SCHEDULER_add_delayed (COVER_AGE_FREQUENCY,
- &age_cover_counters,
- NULL);
-}
-
-/**
- * We've just now completed a datastore request. Update our
- * datastore load calculations.
- *
- * @param start time when the datastore request was issued
- */
-static void
-update_datastore_delays (struct GNUNET_TIME_Absolute start)
-{
- struct GNUNET_TIME_Relative delay;
-
- delay = GNUNET_TIME_absolute_get_duration (start);
- GNUNET_LOAD_update (datastore_get_load,
- delay.rel_value);
-}
-
-
-/**
- * Get the filename under which we would store the GNUNET_HELLO_Message
- * for the given host and protocol.
- * @return filename of the form DIRECTORY/HOSTID
- */
-static char *
-get_trust_filename (const struct GNUNET_PeerIdentity *id)
-{
- struct GNUNET_CRYPTO_HashAsciiEncoded fil;
- char *fn;
-
- GNUNET_CRYPTO_hash_to_enc (&id->hashPubKey, &fil);
- GNUNET_asprintf (&fn, "%s%s%s", trustDirectory, DIR_SEPARATOR_STR, &fil);
- return fn;
-}
-
-
-
-/**
- * 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);
-
-
-/* ******************* clean up functions ************************ */
-
-/**
- * Delete the given migration block.
- *
- * @param mb block to delete
- */
-static void
-delete_migration_block (struct MigrationReadyBlock *mb)
-{
- GNUNET_CONTAINER_DLL_remove (mig_head,
- mig_tail,
- mb);
- GNUNET_PEER_decrement_rcs (mb->target_list,
- MIGRATION_LIST_SIZE);
- mig_size--;
- GNUNET_free (mb);
-}
-
-
-/**
- * Compare the distance of two peers to a key.
- *
- * @param key key
- * @param p1 first peer
- * @param p2 second peer
- * @return GNUNET_YES if P1 is closer to key than P2
- */
-static int
-is_closer (const GNUNET_HashCode *key,
- const struct GNUNET_PeerIdentity *p1,
- const struct GNUNET_PeerIdentity *p2)
-{
- return GNUNET_CRYPTO_hash_xorcmp (&p1->hashPubKey,
- &p2->hashPubKey,
- key);
-}
-
-
-/**
- * Consider migrating content to a given peer.
- *
- * @param cls 'struct MigrationReadyBlock*' to select
- * targets for (or NULL for none)
- * @param key ID of the peer
- * @param value 'struct ConnectedPeer' of the peer
- * @return GNUNET_YES (always continue iteration)
- */
-static int
-consider_migration (void *cls,
- const GNUNET_HashCode *key,
- void *value)
-{
- struct MigrationReadyBlock *mb = cls;
- struct ConnectedPeer *cp = value;
- struct MigrationReadyBlock *pos;
- struct GNUNET_PeerIdentity cppid;
- struct GNUNET_PeerIdentity otherpid;
- struct GNUNET_PeerIdentity worstpid;
- size_t msize;
- unsigned int i;
- unsigned int repl;
-
- /* consider 'cp' as a migration target for mb */
- if (GNUNET_TIME_absolute_get_remaining (cp->migration_blocked).rel_value > 0)
- return GNUNET_YES; /* peer has requested no migration! */
- if (mb != NULL)
- {
- GNUNET_PEER_resolve (cp->pid,
- &cppid);
- repl = MIGRATION_LIST_SIZE;
- for (i=0;i<MIGRATION_LIST_SIZE;i++)
- {
- if (mb->target_list[i] == 0)
- {
- mb->target_list[i] = cp->pid;
- GNUNET_PEER_change_rc (mb->target_list[i], 1);
- repl = MIGRATION_LIST_SIZE;
- break;
- }
- GNUNET_PEER_resolve (mb->target_list[i],
- &otherpid);
- if ( (repl == MIGRATION_LIST_SIZE) &&
- is_closer (&mb->query,
- &cppid,
- &otherpid))
- {
- repl = i;
- worstpid = otherpid;
- }
- else if ( (repl != MIGRATION_LIST_SIZE) &&
- (is_closer (&mb->query,
- &worstpid,
- &otherpid) ) )
- {
- repl = i;
- worstpid = otherpid;
- }
- }
- if (repl != MIGRATION_LIST_SIZE)
- {
- GNUNET_PEER_change_rc (mb->target_list[repl], -1);
- mb->target_list[repl] = cp->pid;
- GNUNET_PEER_change_rc (mb->target_list[repl], 1);
- }
- }
-
- /* consider scheduling transmission to cp for content migration */
- if (cp->cth != NULL)
- return GNUNET_YES;
- msize = 0;
- pos = mig_head;
- while (pos != NULL)
- {
- for (i=0;i<MIGRATION_LIST_SIZE;i++)
- {
- if (cp->pid == pos->target_list[i])
- {
- if (msize == 0)
- msize = pos->size;
- else
- msize = GNUNET_MIN (msize,
- pos->size);
- break;
- }
- }
- pos = pos->next;
- }
- if (msize == 0)
- return GNUNET_YES; /* no content available */
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Trying to migrate at least %u bytes to peer `%s'\n",
- msize,
- GNUNET_h2s (key));
-#endif
- if (cp->delayed_transmission_request_task != GNUNET_SCHEDULER_NO_TASK)
- {
- GNUNET_SCHEDULER_cancel (cp->delayed_transmission_request_task);
- cp->delayed_transmission_request_task = GNUNET_SCHEDULER_NO_TASK;
- }
- cp->cth
- = GNUNET_CORE_notify_transmit_ready (core,
- GNUNET_YES,
- 0, GNUNET_TIME_UNIT_FOREVER_REL,
- (const struct GNUNET_PeerIdentity*) key,
- msize + sizeof (struct PutMessage),
- &transmit_to_peer,
- cp);
- return GNUNET_YES;
-}
-
-
-/**
- * Task that is run periodically to obtain blocks for content
- * migration
- *
- * @param cls unused
- * @param tc scheduler context (also unused)
- */
-static void
-gather_migration_blocks (void *cls,
- const struct GNUNET_SCHEDULER_TaskContext *tc);
-
-
-
-
-/**
- * Task that is run periodically to obtain blocks for DHT PUTs.
- *
- * @param cls type of blocks to gather
- * @param tc scheduler context (unused)
- */
-static void
-gather_dht_put_blocks (void *cls,
- const struct GNUNET_SCHEDULER_TaskContext *tc);
-
-
-/**
- * If the migration task is not currently running, consider
- * (re)scheduling it with the appropriate delay.
- */
-static void
-consider_migration_gathering ()
-{
- struct GNUNET_TIME_Relative delay;
-
- if (dsh == NULL)
- return;
- if (mig_qe != NULL)
- return;
- if (mig_task != GNUNET_SCHEDULER_NO_TASK)
- return;
- delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
- mig_size);
- delay = GNUNET_TIME_relative_divide (delay,
- MAX_MIGRATION_QUEUE);
- delay = GNUNET_TIME_relative_max (delay,
- min_migration_delay);
- mig_task = GNUNET_SCHEDULER_add_delayed (delay,
- &gather_migration_blocks,
- NULL);
-}
-
-
-/**
- * If the DHT PUT gathering task is not currently running, consider
- * (re)scheduling it with the appropriate delay.
- */
-static void
-consider_dht_put_gathering (void *cls)
-{
- struct GNUNET_TIME_Relative delay;
-
- if (dsh == NULL)
- return;
- if (dht_qe != NULL)
- return;
- if (dht_task != GNUNET_SCHEDULER_NO_TASK)
- return;
- if (zero_anonymity_count_estimate > 0)
- {
- delay = GNUNET_TIME_relative_divide (GNUNET_DHT_DEFAULT_REPUBLISH_FREQUENCY,
- zero_anonymity_count_estimate);
- delay = GNUNET_TIME_relative_min (delay,
- MAX_DHT_PUT_FREQ);
- }
- else
- {
- /* if we have NO zero-anonymity content yet, wait 5 minutes for some to
- (hopefully) appear */
- delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5);
- }
- dht_task = GNUNET_SCHEDULER_add_delayed (delay,
- &gather_dht_put_blocks,
- cls);
-}
-
-
-/**
- * Process content offered for migration.
- *
- * @param cls closure
- * @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_migration_content (void *cls,
- const GNUNET_HashCode * key,
- size_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 MigrationReadyBlock *mb;
-
- if (key == NULL)
- {
- mig_qe = NULL;
- if (mig_size < MAX_MIGRATION_QUEUE)
- consider_migration_gathering ();
- return;
- }
- if (GNUNET_TIME_absolute_get_remaining (expiration).rel_value <
- MIN_MIGRATION_CONTENT_LIFETIME.rel_value)
- {
- /* content will expire soon, don't bother */
- GNUNET_DATASTORE_iterate_get_next (dsh);
- return;
- }
- if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
- {
- if (GNUNET_OK !=
- GNUNET_FS_handle_on_demand_block (key, size, data,
- type, priority, anonymity,
- expiration, uid,
- &process_migration_content,
- NULL))
- {
- GNUNET_DATASTORE_iterate_get_next (dsh);
- }
- return;
- }
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Retrieved block `%s' of type %u for migration\n",
- GNUNET_h2s (key),
- type);
-#endif
- mb = GNUNET_malloc (sizeof (struct MigrationReadyBlock) + size);
- mb->query = *key;
- mb->expiration = expiration;
- mb->size = size;
- mb->type = type;
- memcpy (&mb[1], data, size);
- GNUNET_CONTAINER_DLL_insert_after (mig_head,
- mig_tail,
- mig_tail,
- mb);
- mig_size++;
- GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
- &consider_migration,
- mb);
- GNUNET_DATASTORE_iterate_get_next (dsh);
-}
-
-
-/**
- * Function called upon completion of the DHT PUT operation.
- */
-static void
-dht_put_continuation (void *cls,
- const struct GNUNET_SCHEDULER_TaskContext *tc)
-{
- GNUNET_DATASTORE_iterate_get_next (dsh);
-}
-
-
-/**
- * Store content in DHT.
- *
- * @param cls closure
- * @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_dht_put_content (void *cls,
- const GNUNET_HashCode * key,
- size_t size,
- const void *data,
- enum GNUNET_BLOCK_Type type,
- uint32_t priority,
- uint32_t anonymity,
- struct GNUNET_TIME_Absolute
- expiration, uint64_t uid)
-{
- static unsigned int counter;
- static GNUNET_HashCode last_vhash;
- static GNUNET_HashCode vhash;
-
- if (key == NULL)
- {
- dht_qe = NULL;
- consider_dht_put_gathering (cls);
- return;
- }
- /* slightly funky code to estimate the total number of values with zero
- anonymity from the maximum observed length of a monotonically increasing
- sequence of hashes over the contents */
- GNUNET_CRYPTO_hash (data, size, &vhash);
- if (GNUNET_CRYPTO_hash_cmp (&vhash, &last_vhash) <= 0)
- {
- if (zero_anonymity_count_estimate > 0)
- zero_anonymity_count_estimate /= 2;
- counter = 0;
- }
- last_vhash = vhash;
- if (counter < 31)
- counter++;
- if (zero_anonymity_count_estimate < (1 << counter))
- zero_anonymity_count_estimate = (1 << counter);
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Retrieved block `%s' of type %u for DHT PUT\n",
- GNUNET_h2s (key),
- type);
-#endif
- GNUNET_DHT_put (dht_handle,
- key,
- DEFAULT_PUT_REPLICATION,
- GNUNET_DHT_RO_NONE,
- type,
- size,
- data,
- expiration,
- GNUNET_TIME_UNIT_FOREVER_REL,
- &dht_put_continuation,
- cls);
-}
-
-
-/**
- * Task that is run periodically to obtain blocks for content
- * migration
- *
- * @param cls unused
- * @param tc scheduler context (also unused)
- */
-static void
-gather_migration_blocks (void *cls,
- const struct GNUNET_SCHEDULER_TaskContext *tc)
-{
- mig_task = GNUNET_SCHEDULER_NO_TASK;
- if (dsh != NULL)
- {
- mig_qe = GNUNET_DATASTORE_get_for_replication (dsh, 0, UINT_MAX,
- GNUNET_TIME_UNIT_FOREVER_REL,
- &process_migration_content, NULL);
- GNUNET_assert (mig_qe != NULL);
- }
-}
-
-
-/**
- * Task that is run periodically to obtain blocks for DHT PUTs.
- *
- * @param cls type of blocks to gather
- * @param tc scheduler context (unused)
- */
-static void
-gather_dht_put_blocks (void *cls,
- const struct GNUNET_SCHEDULER_TaskContext *tc)
-{
- dht_task = GNUNET_SCHEDULER_NO_TASK;
- if (dsh != NULL)
- {
- if (dht_put_type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
- dht_put_type = GNUNET_BLOCK_TYPE_FS_KBLOCK;
- dht_qe = GNUNET_DATASTORE_iterate_zero_anonymity (dsh, 0, UINT_MAX,
- GNUNET_TIME_UNIT_FOREVER_REL,
- dht_put_type++,
- &process_dht_put_content, NULL);
- GNUNET_assert (dht_qe != NULL);
- }
-}
-
-
-/**
- * We're done with a particular message list entry.
- * Free all associated resources.
- *
- * @param pml entry to destroy
- */
-static void
-destroy_pending_message_list_entry (struct PendingMessageList *pml)
-{
- GNUNET_CONTAINER_DLL_remove (pml->req->pending_head,
- pml->req->pending_tail,
- pml);
- GNUNET_CONTAINER_DLL_remove (pml->target->pending_messages_head,
- pml->target->pending_messages_tail,
- pml->pm);
- GNUNET_assert (pml->target->pending_requests > 0);
- pml->target->pending_requests--;
- GNUNET_free (pml->pm);
- GNUNET_free (pml);
-}
-
-
-/**
- * Destroy the given pending message (and call the respective
- * continuation).
- *
- * @param pm message to destroy
- * @param tpid id of peer that the message was delivered to, or 0 for none
- */
-static void
-destroy_pending_message (struct PendingMessage *pm,
- GNUNET_PEER_Id tpid)
-{
- struct PendingMessageList *pml = pm->pml;
- TransmissionContinuation cont;
- void *cont_cls;
-
- cont = pm->cont;
- cont_cls = pm->cont_cls;
- if (pml != NULL)
- {
- GNUNET_assert (pml->pm == pm);
- GNUNET_assert ( (tpid == 0) || (tpid == pml->target->pid) );
- destroy_pending_message_list_entry (pml);
- }
- else
- {
- GNUNET_free (pm);
- }
- if (cont != NULL)
- cont (cont_cls, tpid);
-}
-
-
-/**
- * We're done processing a particular request.
- * Free all associated resources.
- *
- * @param pr request to destroy
- */
-static void
-destroy_pending_request (struct PendingRequest *pr)
-{
- struct GNUNET_PeerIdentity pid;
- unsigned int i;
-
- if (pr->hnode != NULL)
- {
- GNUNET_CONTAINER_heap_remove_node (pr->hnode);
- pr->hnode = NULL;
- }
- if (NULL == pr->client_request_list)
- {
- GNUNET_STATISTICS_update (stats,
- gettext_noop ("# P2P searches active"),
- -1,
- GNUNET_NO);
- }
- else
- {
- GNUNET_STATISTICS_update (stats,
- gettext_noop ("# client searches active"),
- -1,
- GNUNET_NO);
- }
- if (GNUNET_YES ==
- GNUNET_CONTAINER_multihashmap_remove (query_request_map,
- &pr->query,
- pr))
- {
- GNUNET_LOAD_update (rt_entry_lifetime,
- GNUNET_TIME_absolute_get_duration (pr->start_time).rel_value);
- }
- if (pr->qe != NULL)
- {
- GNUNET_DATASTORE_cancel (pr->qe);
- pr->qe = NULL;
- }
- if (pr->dht_get != NULL)
- {
- GNUNET_DHT_get_stop (pr->dht_get);
- pr->dht_get = NULL;
- }
- if (pr->client_request_list != NULL)
- {
- GNUNET_CONTAINER_DLL_remove (pr->client_request_list->client_list->rl_head,
- pr->client_request_list->client_list->rl_tail,
- pr->client_request_list);
- GNUNET_free (pr->client_request_list);
- pr->client_request_list = NULL;
- }
- if (pr->cp != NULL)
- {
- GNUNET_PEER_resolve (pr->cp->pid,
- &pid);
- (void) GNUNET_CONTAINER_multihashmap_remove (peer_request_map,
- &pid.hashPubKey,
- pr);
- pr->cp = NULL;
- }
- if (pr->bf != NULL)
- {
- GNUNET_CONTAINER_bloomfilter_free (pr->bf);
- pr->bf = NULL;
- }
- if (pr->pirc != NULL)
- {
- GNUNET_CORE_peer_change_preference_cancel (pr->pirc->irc);
- pr->pirc->irc = NULL;
- pr->pirc = NULL;
- }
- if (pr->replies_seen != NULL)
- {
- GNUNET_free (pr->replies_seen);
- pr->replies_seen = NULL;
- }
- if (pr->task != GNUNET_SCHEDULER_NO_TASK)
- {
- GNUNET_SCHEDULER_cancel (pr->task);
- pr->task = GNUNET_SCHEDULER_NO_TASK;
- }
- while (NULL != pr->pending_head)
- destroy_pending_message_list_entry (pr->pending_head);
- GNUNET_PEER_change_rc (pr->target_pid, -1);
- if (pr->used_targets != NULL)
- {
- for (i=0;i<pr->used_targets_off;i++)
- GNUNET_PEER_change_rc (pr->used_targets[i].pid, -1);
- GNUNET_free (pr->used_targets);
- pr->used_targets_off = 0;
- pr->used_targets_size = 0;
- pr->used_targets = NULL;
- }
- GNUNET_free (pr);
-}
-
-
-/**
- * Find latency information in 'atsi'.
- *
- * @param atsi performance data
- * @return connection latency
- */
-static struct GNUNET_TIME_Relative
-get_latency (const struct GNUNET_TRANSPORT_ATS_Information *atsi)
-{
- if (atsi == NULL)
- return GNUNET_TIME_UNIT_SECONDS;
- while ( (ntohl (atsi->type) != GNUNET_TRANSPORT_ATS_ARRAY_TERMINATOR) &&
- (ntohl (atsi->type) != GNUNET_TRANSPORT_ATS_QUALITY_NET_DELAY) )
- atsi++;
- if (ntohl (atsi->type) == GNUNET_TRANSPORT_ATS_ARRAY_TERMINATOR)
- {
- GNUNET_break (0);
- /* how can we not have latency data? */
- return GNUNET_TIME_UNIT_SECONDS;
- }
- return GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
- ntohl (atsi->value));
-}
-
-
-/**
- * Method called whenever a given peer connects.
- *
- * @param cls closure, not used
- * @param peer peer identity this notification is about
- * @param atsi performance information
- */
-static void
-peer_connect_handler (void *cls,
- const struct
- GNUNET_PeerIdentity * peer,
- const struct GNUNET_TRANSPORT_ATS_Information *atsi)
-{
- struct ConnectedPeer *cp;
- struct MigrationReadyBlock *pos;
- char *fn;
- uint32_t trust;
- struct GNUNET_TIME_Relative latency;
-
- if (0 == memcmp (&my_id, peer, sizeof (struct GNUNET_PeerIdentity)))
- return;
- latency = get_latency (atsi);
- cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
- &peer->hashPubKey);
- if (NULL != cp)
- {
- GNUNET_break (0);
- return;
- }
- cp = GNUNET_malloc (sizeof (struct ConnectedPeer));
- cp->transmission_delay = GNUNET_LOAD_value_init (latency);
- cp->pid = GNUNET_PEER_intern (peer);
-
- fn = get_trust_filename (peer);
- if ((GNUNET_DISK_file_test (fn) == GNUNET_YES) &&
- (sizeof (trust) == GNUNET_DISK_fn_read (fn, &trust, sizeof (trust))))
- cp->disk_trust = cp->trust = ntohl (trust);
- GNUNET_free (fn);
-
- GNUNET_break (GNUNET_OK ==
- GNUNET_CONTAINER_multihashmap_put (connected_peers,
- &peer->hashPubKey,
- cp,
- GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
-
- pos = mig_head;
- while (NULL != pos)
- {
- (void) consider_migration (pos, &peer->hashPubKey, cp);
- pos = pos->next;
- }
-}
-
-
-/**
- * Method called whenever a given peer has a status change.
- *
- * @param cls closure
- * @param peer peer identity this notification is about
- * @param bandwidth_in available amount of inbound bandwidth
- * @param bandwidth_out available amount of outbound bandwidth
- * @param timeout absolute time when this peer will time out
- * unless we see some further activity from it
- * @param atsi status information
- */
-static void
-peer_status_handler (void *cls,
- const struct
- GNUNET_PeerIdentity * peer,
- struct GNUNET_BANDWIDTH_Value32NBO bandwidth_in,
- struct GNUNET_BANDWIDTH_Value32NBO bandwidth_out,
- struct GNUNET_TIME_Absolute timeout,
- const struct GNUNET_TRANSPORT_ATS_Information *atsi)
-{
- struct ConnectedPeer *cp;
- struct GNUNET_TIME_Relative latency;
-
- latency = get_latency (atsi);
- cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
- &peer->hashPubKey);
- if (cp == NULL)
- {
- GNUNET_break (0);
- return;
- }
- GNUNET_LOAD_value_set_decline (cp->transmission_delay,
- latency);
-}
-
-
-
-/**
- * Increase the host credit by a value.
- *
- * @param host which peer to change the trust value on
- * @param value is the int value by which the
- * host credit is to be increased or decreased
- * @returns the actual change in trust (positive or negative)
- */
-static int
-change_host_trust (struct ConnectedPeer *host, int value)
-{
- if (value == 0)
- return 0;
- GNUNET_assert (host != NULL);
- if (value > 0)
- {
- if (host->trust + value < host->trust)
- {
- value = UINT32_MAX - host->trust;
- host->trust = UINT32_MAX;
- }
- else
- host->trust += value;
- }
- else
- {
- if (host->trust < -value)
- {
- value = -host->trust;
- host->trust = 0;
- }
- else
- host->trust += value;
- }
- return value;
-}
-
-
-/**
- * Write host-trust information to a file - flush the buffer entry!
- */
-static int
-flush_trust (void *cls,
- const GNUNET_HashCode *key,
- void *value)
-{
- struct ConnectedPeer *host = value;
- char *fn;
- uint32_t trust;
- struct GNUNET_PeerIdentity pid;
-
- if (host->trust == host->disk_trust)
- return GNUNET_OK; /* unchanged */
- GNUNET_PEER_resolve (host->pid,
- &pid);
- fn = get_trust_filename (&pid);
- if (host->trust == 0)
- {
- if ((0 != UNLINK (fn)) && (errno != ENOENT))
- GNUNET_log_strerror_file (GNUNET_ERROR_TYPE_WARNING |
- GNUNET_ERROR_TYPE_BULK, "unlink", fn);
- }
- else
- {
- trust = htonl (host->trust);
- if (sizeof(uint32_t) == GNUNET_DISK_fn_write (fn, &trust,
- sizeof(uint32_t),
- GNUNET_DISK_PERM_USER_READ | GNUNET_DISK_PERM_USER_WRITE
- | GNUNET_DISK_PERM_GROUP_READ | GNUNET_DISK_PERM_OTHER_READ))
- host->disk_trust = host->trust;
- }
- GNUNET_free (fn);
- return GNUNET_OK;
-}
-
-/**
- * Call this method periodically to scan data/hosts for new hosts.
- */
-static void
-cron_flush_trust (void *cls,
- const struct GNUNET_SCHEDULER_TaskContext *tc)
-{
-
- if (NULL == connected_peers)
- return;
- GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
- &flush_trust,
- NULL);
- if (NULL == tc)
- return;
- if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
- return;
- GNUNET_SCHEDULER_add_delayed (TRUST_FLUSH_FREQ, &cron_flush_trust, NULL);
-}
-
-
-/**
- * Free (each) request made by the peer.
- *
- * @param cls closure, points to peer that the request belongs to
- * @param key current key code
- * @param value value in the hash map
- * @return GNUNET_YES (we should continue to iterate)
- */
-static int
-destroy_request (void *cls,
- const GNUNET_HashCode * key,
- void *value)
-{
- const struct GNUNET_PeerIdentity * peer = cls;
- struct PendingRequest *pr = value;
-
- GNUNET_break (GNUNET_YES ==
- GNUNET_CONTAINER_multihashmap_remove (peer_request_map,
- &peer->hashPubKey,
- pr));
- destroy_pending_request (pr);
- return GNUNET_YES;
-}
-
-
-/**
- * Method called whenever a peer disconnects.
- *
- * @param cls closure, not used
- * @param peer peer identity this notification is about
- */
-static void
-peer_disconnect_handler (void *cls,
- const struct
- GNUNET_PeerIdentity * peer)
-{
- struct ConnectedPeer *cp;
- struct PendingMessage *pm;
- unsigned int i;
- struct MigrationReadyBlock *pos;
- struct MigrationReadyBlock *next;
-
- if (0 == memcmp (&my_id, peer, sizeof (struct GNUNET_PeerIdentity)))
- return;
- GNUNET_CONTAINER_multihashmap_get_multiple (peer_request_map,
- &peer->hashPubKey,
- &destroy_request,
- (void*) peer);
- cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
- &peer->hashPubKey);
- if (cp == NULL)
- return;
- for (i=0;i<CS2P_SUCCESS_LIST_SIZE;i++)
- {
- if (NULL != cp->last_client_replies[i])
- {
- GNUNET_SERVER_client_drop (cp->last_client_replies[i]);
- cp->last_client_replies[i] = NULL;
- }
- }
- GNUNET_break (GNUNET_YES ==
- GNUNET_CONTAINER_multihashmap_remove (connected_peers,
- &peer->hashPubKey,
- cp));
- if (cp->irc != NULL)
- {
- GNUNET_CORE_peer_change_preference_cancel (cp->irc);
- cp->irc = NULL;
- cp->pr->pirc = NULL;
- cp->pr = NULL;
- }
-
- /* remove this peer from migration considerations; schedule
- alternatives */
- next = mig_head;
- while (NULL != (pos = next))
- {
- next = pos->next;
- for (i=0;i<MIGRATION_LIST_SIZE;i++)
- {
- if (pos->target_list[i] == cp->pid)
- {
- GNUNET_PEER_change_rc (pos->target_list[i], -1);
- pos->target_list[i] = 0;
- }
- }
- if (pos->used_targets >= GNUNET_CONTAINER_multihashmap_size (connected_peers))
- {
- delete_migration_block (pos);
- consider_migration_gathering ();
- continue;
- }
- GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
- &consider_migration,
- pos);
- }
- GNUNET_PEER_change_rc (cp->pid, -1);
- GNUNET_PEER_decrement_rcs (cp->last_p2p_replies, P2P_SUCCESS_LIST_SIZE);
- if (NULL != cp->cth)
- {
- GNUNET_CORE_notify_transmit_ready_cancel (cp->cth);
- cp->cth = NULL;
- }
- if (cp->delayed_transmission_request_task != GNUNET_SCHEDULER_NO_TASK)
- {
- GNUNET_SCHEDULER_cancel (cp->delayed_transmission_request_task);
- cp->delayed_transmission_request_task = GNUNET_SCHEDULER_NO_TASK;
- }
- while (NULL != (pm = cp->pending_messages_head))
- destroy_pending_message (pm, 0 /* delivery failed */);
- GNUNET_LOAD_value_free (cp->transmission_delay);
- GNUNET_break (0 == cp->pending_requests);
- GNUNET_free (cp);
-}
-
-
-/**
- * Iterator over hash map entries that removes all occurences
- * of the given 'client' from the 'last_client_replies' of the
- * given connected peer.
- *
- * @param cls closure, the 'struct GNUNET_SERVER_Client*' to remove
- * @param key current key code (unused)
- * @param value value in the hash map (the 'struct ConnectedPeer*' to change)
- * @return GNUNET_YES (we should continue to iterate)
- */
-static int
-remove_client_from_last_client_replies (void *cls,
- const GNUNET_HashCode * key,
- void *value)
-{
- struct GNUNET_SERVER_Client *client = cls;
- struct ConnectedPeer *cp = value;
- unsigned int i;
-
- for (i=0;i<CS2P_SUCCESS_LIST_SIZE;i++)
- {
- if (cp->last_client_replies[i] == client)
- {
- GNUNET_SERVER_client_drop (cp->last_client_replies[i]);
- cp->last_client_replies[i] = NULL;
- }
- }
- return GNUNET_YES;
-}
-
-
-/**
- * A client disconnected. Remove all of its pending queries.
- *
- * @param cls closure, NULL
- * @param client identification of the client
- */
-static void
-handle_client_disconnect (void *cls,
- struct GNUNET_SERVER_Client
- * client)
-{
- struct ClientList *pos;
- struct ClientList *prev;
- struct ClientRequestList *rcl;
- struct ClientResponseMessage *creply;
-
- if (client == NULL)
- return;
- prev = NULL;
- pos = client_list;
- while ( (NULL != pos) &&
- (pos->client != client) )
- {
- prev = pos;
- pos = pos->next;
- }
- if (pos == NULL)
- return; /* no requests pending for this client */
- while (NULL != (rcl = pos->rl_head))
- {
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Destroying pending request `%s' on disconnect\n",
- GNUNET_h2s (&rcl->req->query));
- destroy_pending_request (rcl->req);
- }
- if (prev == NULL)
- client_list = pos->next;
- else
- prev->next = pos->next;
- if (pos->th != NULL)
- {
- GNUNET_CONNECTION_notify_transmit_ready_cancel (pos->th);
- pos->th = NULL;
- }
- while (NULL != (creply = pos->res_head))
- {
- GNUNET_CONTAINER_DLL_remove (pos->res_head,
- pos->res_tail,
- creply);
- GNUNET_free (creply);
- }
- GNUNET_SERVER_client_drop (pos->client);
- GNUNET_free (pos);
- GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
- &remove_client_from_last_client_replies,
- client);
-}
-
-
-/**
- * Iterator to free peer entries.
- *
- * @param cls closure, unused
- * @param key current key code
- * @param value value in the hash map (peer entry)
- * @return GNUNET_YES (we should continue to iterate)
- */
-static int
-clean_peer (void *cls,
- const GNUNET_HashCode * key,
- void *value)
-{
- peer_disconnect_handler (NULL, (const struct GNUNET_PeerIdentity*) key);
- return GNUNET_YES;
-}
-
-
-/**
- * Task run during shutdown.
- *
- * @param cls unused
- * @param tc unused
- */
-static void
-shutdown_task (void *cls,
- const struct GNUNET_SCHEDULER_TaskContext *tc)
-{
- if (mig_qe != NULL)
- {
- GNUNET_DATASTORE_cancel (mig_qe);
- mig_qe = NULL;
- }
- if (dht_qe != NULL)
- {
- GNUNET_DATASTORE_cancel (dht_qe);
- dht_qe = NULL;
- }
- if (GNUNET_SCHEDULER_NO_TASK != mig_task)
- {
- GNUNET_SCHEDULER_cancel (mig_task);
- mig_task = GNUNET_SCHEDULER_NO_TASK;
- }
- if (GNUNET_SCHEDULER_NO_TASK != dht_task)
- {
- GNUNET_SCHEDULER_cancel (dht_task);
- dht_task = GNUNET_SCHEDULER_NO_TASK;
- }
- while (client_list != NULL)
- handle_client_disconnect (NULL,
- client_list->client);
- cron_flush_trust (NULL, NULL);
- GNUNET_assert (NULL != core);
- GNUNET_CORE_disconnect (core);
- core = NULL;
- 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_LOAD_value_free (rt_entry_lifetime);
- rt_entry_lifetime = NULL;
- GNUNET_break (0 == GNUNET_CONTAINER_multihashmap_size (peer_request_map));
- GNUNET_CONTAINER_multihashmap_destroy (peer_request_map);
- peer_request_map = NULL;
- if (stats != NULL)
- {
- GNUNET_STATISTICS_destroy (stats, GNUNET_NO);
- stats = NULL;
- }
- if (dsh != NULL)
- {
- GNUNET_DATASTORE_disconnect (dsh,
- GNUNET_NO);
- dsh = NULL;
- }
- while (mig_head != NULL)
- delete_migration_block (mig_head);
- GNUNET_assert (0 == mig_size);
- GNUNET_DHT_disconnect (dht_handle);
- dht_handle = NULL;
- GNUNET_LOAD_value_free (datastore_get_load);
- datastore_get_load = NULL;
- GNUNET_LOAD_value_free (datastore_put_load);
- datastore_put_load = NULL;
- GNUNET_BLOCK_context_destroy (block_ctx);
- block_ctx = NULL;
- GNUNET_CONFIGURATION_destroy (block_cfg);
- block_cfg = NULL;
- cfg = NULL;
- GNUNET_free_non_null (trustDirectory);
- trustDirectory = NULL;
- GNUNET_SCHEDULER_cancel (cover_age_task);
- cover_age_task = GNUNET_SCHEDULER_NO_TASK;
-}
-
-
-/* ******************* Utility functions ******************** */
-
-
-/**
- * We've had to delay a request for transmission to core, but now
- * we should be ready. Run it.
- *
- * @param cls the 'struct ConnectedPeer' for which a request was delayed
- * @param tc task context (unused)
- */
-static void
-delayed_transmission_request (void *cls,
- const struct GNUNET_SCHEDULER_TaskContext *tc)
-{
- struct ConnectedPeer *cp = cls;
- struct GNUNET_PeerIdentity pid;
- struct PendingMessage *pm;
-
- pm = cp->pending_messages_head;
- cp->delayed_transmission_request_task = GNUNET_SCHEDULER_NO_TASK;
- GNUNET_assert (cp->cth == NULL);
- if (pm == NULL)
- return;
- GNUNET_PEER_resolve (cp->pid,
- &pid);
- cp->last_transmission_request_start = GNUNET_TIME_absolute_get ();
- cp->cth = GNUNET_CORE_notify_transmit_ready (core,
- GNUNET_YES,
- pm->priority,
- GNUNET_CONSTANTS_SERVICE_TIMEOUT,
- &pid,
- pm->msize,
- &transmit_to_peer,
- cp);
-}
-
-
-/**
- * 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 PendingMessage *pm;
- struct PendingMessage *next_pm;
- struct GNUNET_TIME_Absolute now;
- struct GNUNET_TIME_Relative min_delay;
- struct MigrationReadyBlock *mb;
- struct MigrationReadyBlock *next;
- struct PutMessage migm;
- size_t msize;
- unsigned int i;
- struct GNUNET_PeerIdentity pid;
-
- cp->cth = NULL;
- if (NULL == buf)
- {
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Dropping message, core too busy.\n");
-#endif
- GNUNET_LOAD_update (cp->transmission_delay,
- UINT64_MAX);
-
- if (NULL != (pm = cp->pending_messages_head))
- {
- GNUNET_CONTAINER_DLL_remove (cp->pending_messages_head,
- cp->pending_messages_tail,
- pm);
- GNUNET_assert (cp->pending_requests > 0);
- cp->pending_requests--;
- destroy_pending_message (pm, 0);
- }
- if (NULL != (pm = cp->pending_messages_head))
- {
- GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == cp->delayed_transmission_request_task);
- min_delay = GNUNET_TIME_absolute_get_remaining (pm->delay_until);
- cp->delayed_transmission_request_task
- = GNUNET_SCHEDULER_add_delayed (min_delay,
- &delayed_transmission_request,
- cp);
- }
- return 0;
- }
- GNUNET_LOAD_update (cp->transmission_delay,
- GNUNET_TIME_absolute_get_duration (cp->last_transmission_request_start).rel_value);
- now = GNUNET_TIME_absolute_get ();
- msize = 0;
- min_delay = GNUNET_TIME_UNIT_FOREVER_REL;
- next_pm = cp->pending_messages_head;
- while ( (NULL != (pm = next_pm) ) &&
- (pm->msize <= size) )
- {
- next_pm = pm->next;
- if (pm->delay_until.abs_value > now.abs_value)
- {
- min_delay = GNUNET_TIME_relative_min (min_delay,
- GNUNET_TIME_absolute_get_remaining (pm->delay_until));
- continue;
- }
- memcpy (&cbuf[msize], &pm[1], pm->msize);
- msize += pm->msize;
- size -= pm->msize;
- if (NULL == pm->pml)
- {
- GNUNET_CONTAINER_DLL_remove (cp->pending_messages_head,
- cp->pending_messages_tail,
- pm);
- GNUNET_assert (cp->pending_requests > 0);
- cp->pending_requests--;
- }
- destroy_pending_message (pm, cp->pid);
- }
- if (pm != NULL)
- min_delay = GNUNET_TIME_UNIT_ZERO;
- if (NULL != cp->pending_messages_head)
- {
- GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == cp->delayed_transmission_request_task);
- cp->delayed_transmission_request_task
- = GNUNET_SCHEDULER_add_delayed (min_delay,
- &delayed_transmission_request,
- cp);
- }
- if (pm == NULL)
- {
- GNUNET_PEER_resolve (cp->pid,
- &pid);
- 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++;
- memset (&migm, 0, sizeof (migm));
- 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),
- (unsigned int) 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),
- (unsigned int) 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 with PID %u\n",
- (unsigned int) msize,
- (unsigned int) 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);
- if (pr != 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)
- {
- GNUNET_STATISTICS_update (stats,
- gettext_noop ("# P2P searches discarded (queue length bound)"),
- 1,
- GNUNET_NO);
- 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);
- cp->cth = NULL;
- }
- if (cp->delayed_transmission_request_task != GNUNET_SCHEDULER_NO_TASK)
- {
- GNUNET_SCHEDULER_cancel (cp->delayed_transmission_request_task);
- cp->delayed_transmission_request_task = GNUNET_SCHEDULER_NO_TASK;
- }
- /* need to schedule transmission */
- cp->last_transmission_request_start = GNUNET_TIME_absolute_get ();
- cp->cth = GNUNET_CORE_notify_transmit_ready (core,
- GNUNET_YES,
- 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);
- }
-}
-
-
-/**
- * Test if the DATABASE (GET) load on this peer is too high
- * to even consider processing the query at
- * all.
- *
- * @return GNUNET_YES if the load is too high to do anything (load high)
- * GNUNET_NO to process normally (load normal)
- * GNUNET_SYSERR to process for free (load low)
- */
-static int
-test_get_load_too_high (uint32_t priority)
-{
- double ld;
-
- ld = GNUNET_LOAD_get_load (datastore_get_load);
- if (ld < 1)
- return GNUNET_SYSERR;
- if (ld <= priority)
- return GNUNET_NO;
- return GNUNET_YES;
-}
-
-
-
-
-/**
- * Test if the DATABASE (PUT) load on this peer is too high
- * to even consider processing the query at
- * all.
- *
- * @return GNUNET_YES if the load is too high to do anything (load high)
- * GNUNET_NO to process normally (load normal or low)