* - TTL/priority calculations are absent!
* TODO:
* - have non-zero preference / priority for requests we initiate!
- * - track stats for hot-path routing
* - implement hot-path routing decision procedure
- * - implement: bound_priority, test_load_too_high, validate_nblock
- * - add content migration support (store locally) [or create new service]
+ * - implement: bound_priority, test_load_too_high
* - statistics
*/
#include "platform.h"
#include "gnunet_signatures.h"
#include "gnunet_statistics_service.h"
#include "gnunet_util_lib.h"
-#include "gnunet-service-fs_drq.h"
#include "gnunet-service-fs_indexing.h"
#include "fs.h"
*/
struct PendingMessage;
+/**
+ * Our connection to the datastore.
+ */
+static struct GNUNET_DATASTORE_Handle *dsh;
+
/**
* Function called upon completion of a transmission.
GNUNET_PEER_Id *used_pids;
/**
- * Our entry in the DRQ (non-NULL while we wait for our
+ * Our entry in the queue (non-NULL while we wait for our
* turn to interact with the local database).
*/
- struct DatastoreRequestQueue *drq;
+ struct GNUNET_DATASTORE_QueueEntry *qe;
/**
+
* Size of the 'bf' (in bytes).
*/
size_t bf_size;
/**
* Type of the content that this request is for.
*/
- uint32_t type;
+ enum GNUNET_BLOCK_Type type;
/**
* Remove this request after transmission of the current response.
*/
- int do_remove;
+ int16_t do_remove;
+
+ /**
+ * GNUNET_YES if we should not forward this request to other peers.
+ */
+ int16_t local_only;
};
+/**
+ * Block that is ready for migration to other peers. Actual data is at the end of the block.
+ */
+struct MigrationReadyBlock
+{
+
+ /**
+ * This is a doubly-linked list.
+ */
+ struct MigrationReadyBlock *next;
+
+ /**
+ * This is a doubly-linked list.
+ */
+ struct MigrationReadyBlock *prev;
+
+ /**
+ * Query for the block.
+ */
+ GNUNET_HashCode query;
+
+ /**
+ * When does this block expire?
+ */
+ struct GNUNET_TIME_Absolute expiration;
+
+ /**
+ * Peers we would consider forwarding this
+ * block to. Zero for empty entries.
+ */
+ GNUNET_PEER_Id target_list[MIGRATION_LIST_SIZE];
+
+ /**
+ * Size of the block.
+ */
+ size_t size;
+
+ /**
+ * Number of targets already used.
+ */
+ unsigned int used_targets;
+
+ /**
+ * Type of the block.
+ */
+ enum GNUNET_BLOCK_Type type;
+};
+
+
/**
* Our scheduler.
*/
*/
static struct GNUNET_CORE_Handle *core;
+/**
+ * Head of linked list of blocks that can be migrated.
+ */
+static struct MigrationReadyBlock *mig_head;
+
+/**
+ * Tail of linked list of blocks that can be migrated.
+ */
+static struct MigrationReadyBlock *mig_tail;
+
+/**
+ * Request to datastore for migration (or NULL).
+ */
+static struct GNUNET_DATASTORE_QueueEntry *mig_qe;
+
+/**
+ * ID of task that collects blocks for migration.
+ */
+static GNUNET_SCHEDULER_TaskIdentifier mig_task;
+
+/**
+ * What is the maximum frequency at which we are allowed to
+ * poll the datastore for migration content?
+ */
+static struct GNUNET_TIME_Relative min_migration_delay;
+
+/**
+ * Size of the doubly-linked list of migration blocks.
+ */
+static unsigned int mig_size;
+
+/**
+ * Are we allowed to migrate content to this peer.
+ */
+static int active_migration;
+
+
+/**
+ * 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)2
+ */
+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 (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 */
+ cp->cth
+ = GNUNET_CORE_notify_transmit_ready (core,
+ 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);
+
+
+/**
+ * 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 (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 (GNUNET_TIME_UNIT_SECONDS,
+ MAX_MIGRATION_QUEUE);
+ delay = GNUNET_TIME_relative_max (delay,
+ min_migration_delay);
+ mig_task = GNUNET_SCHEDULER_add_delayed (sched,
+ delay,
+ &gather_migration_blocks,
+ NULL);
+}
+
+
+/**
+ * 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,
+ 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 MigrationReadyBlock *mb;
+
+ if (key == NULL)
+ {
+ mig_qe = NULL;
+ if (mig_size < MAX_MIGRATION_QUEUE)
+ consider_migration_gathering ();
+ return;
+ }
+ if (type == GNUNET_BLOCK_TYPE_ONDEMAND)
+ {
+ if (GNUNET_OK !=
+ GNUNET_FS_handle_on_demand_block (key, size, data,
+ type, priority, anonymity,
+ expiration, uid,
+ &process_migration_content,
+ NULL))
+ GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
+ return;
+ }
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Retrieved block `%s' of type %u for migration\n",
+ GNUNET_h2s (key),
+ type);
+ 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_get_next (dsh, 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)
+{
+ mig_task = GNUNET_SCHEDULER_NO_TASK;
+ mig_qe = GNUNET_DATASTORE_get_random (dsh, 0, -1,
+ GNUNET_TIME_UNIT_FOREVER_REL,
+ &process_migration_content, NULL);
+ GNUNET_assert (mig_qe != NULL);
+}
+
+
/**
* We're done with a particular message list entry.
* Free all associated resources.
(void) GNUNET_CONTAINER_multihashmap_remove (query_request_map,
&pr->query,
pr);
- if (pr->drq != NULL)
- {
- GNUNET_FS_drq_get_cancel (pr->drq);
- pr->drq = NULL;
+ if (pr->qe != NULL)
+ {
+ GNUNET_DATASTORE_cancel (pr->qe);
+ pr->qe = NULL;
}
if (pr->client_request_list != NULL)
{
uint32_t distance)
{
struct ConnectedPeer *cp;
-
+ struct MigrationReadyBlock *pos;
+
cp = GNUNET_malloc (sizeof (struct ConnectedPeer));
cp->pid = GNUNET_PEER_intern (peer);
- GNUNET_CONTAINER_multihashmap_put (connected_peers,
- &peer->hashPubKey,
- cp,
- GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
+ 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;
+ }
}
+
/**
* Free (each) request made by the peer.
*
const struct GNUNET_PeerIdentity * peer = cls;
struct PendingRequest *pr = value;
- GNUNET_CONTAINER_multihashmap_remove (peer_request_map,
- &peer->hashPubKey,
- pr);
+ GNUNET_break (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_remove (peer_request_map,
+ &peer->hashPubKey,
+ pr));
destroy_pending_request (pr);
return GNUNET_YES;
}
struct ConnectedPeer *cp;
struct PendingMessage *pm;
unsigned int i;
+ struct MigrationReadyBlock *pos;
+ struct MigrationReadyBlock *next;
GNUNET_CONTAINER_multihashmap_get_multiple (peer_request_map,
&peer->hashPubKey,
cp->last_client_replies[i] = NULL;
}
}
- GNUNET_CONTAINER_multihashmap_remove (connected_peers,
- &peer->hashPubKey,
- cp);
+ GNUNET_break (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_remove (connected_peers,
+ &peer->hashPubKey,
+ cp));
+ /* 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)
shutdown_task (void *cls,
const struct GNUNET_SCHEDULER_TaskContext *tc)
{
+ 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_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;
}
/**
- * Transmit the given message by copying it to the target buffer
+ * 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).
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)
&transmit_to_peer,
cp);
}
-#if DEBUG_FS > 2
+ 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;
+ break;
+ }
+ }
+ 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 > 3
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Transmitting %u bytes to peer %u.\n",
+ "Transmitting %u bytes to peer %u\n",
msize,
cp->pid);
#endif
cp->pending_requests++;
if (cp->pending_requests > MAX_QUEUE_PER_PEER)
destroy_pending_message (cp->pending_messages_tail, 0);
- if (cp->cth == NULL)
- {
- /* need to schedule transmission */
- GNUNET_PEER_resolve (cp->pid, &pid);
- 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);
- }
+ 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
#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_MIN;
/* (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,
switch (pr->type)
{
- case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
- case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
+ 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_DATASTORE_BLOCKTYPE_ANY:
- case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
- case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
+ case GNUNET_BLOCK_TYPE_ANY:
+ case GNUNET_BLOCK_TYPE_KBLOCK:
+ case GNUNET_BLOCK_TYPE_SBLOCK:
break;
default:
GNUNET_break (0);
}
-/**
- * Check if the given KBlock is well-formed.
- *
- * @param kb the kblock data (or at least "dsize" bytes claiming to be one)
- * @param dsize size of "kb" in bytes; check for < sizeof(struct KBlock)!
- * @param query where to store the query that this block answers
- * @return GNUNET_OK if this is actually a well-formed KBlock
- */
-static int
-check_kblock (const struct KBlock *kb,
- size_t dsize,
- GNUNET_HashCode *query)
-{
- if (dsize < sizeof (struct KBlock))
- {
- GNUNET_break_op (0);
- return GNUNET_SYSERR;
- }
- if (dsize - sizeof (struct KBlock) !=
- ntohl (kb->purpose.size)
- - sizeof (struct GNUNET_CRYPTO_RsaSignaturePurpose)
- - sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded) )
- {
- GNUNET_break_op (0);
- return GNUNET_SYSERR;
- }
- if (GNUNET_OK !=
- GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_FS_KBLOCK,
- &kb->purpose,
- &kb->signature,
- &kb->keyspace))
- {
- GNUNET_break_op (0);
- return GNUNET_SYSERR;
- }
- if (query != NULL)
- GNUNET_CRYPTO_hash (&kb->keyspace,
- sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
- query);
- return GNUNET_OK;
-}
-
-
-/**
- * Check if the given NBlock is well-formed.
- *
- * @param nb the nblock data (or at least "dsize" bytes claiming to be one)
- * @param dsize size of "nb" in bytes; check for < sizeof(struct NBlock)!
- * @param query where to store the query that this block answers
- * @return GNUNET_OK if this is actually a well-formed NBlock
- */
-static int
-check_nblock (const struct NBlock *nb,
- size_t dsize,
- GNUNET_HashCode *query)
-{
- if (dsize < sizeof (struct NBlock))
- {
- GNUNET_break_op (0);
- return GNUNET_SYSERR;
- }
- if (dsize - sizeof (struct NBlock) !=
- ntohl (nb->ns_purpose.size)
- - sizeof (struct GNUNET_CRYPTO_RsaSignaturePurpose)
- - sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded) )
- {
- GNUNET_break_op (0);
- return GNUNET_SYSERR;
- }
- if (dsize !=
- ntohl (nb->ksk_purpose.size) + sizeof (struct GNUNET_CRYPTO_RsaSignature))
- {
- GNUNET_break_op (0);
- return GNUNET_SYSERR;
- }
- if (GNUNET_OK !=
- GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_FS_NBLOCK_KSIG,
- &nb->ksk_purpose,
- &nb->ksk_signature,
- &nb->keyspace))
- {
- GNUNET_break_op (0);
- return GNUNET_SYSERR;
- }
- if (GNUNET_OK !=
- GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_FS_NBLOCK,
- &nb->ns_purpose,
- &nb->ns_signature,
- &nb->subspace))
- {
- GNUNET_break_op (0);
- return GNUNET_SYSERR;
- }
- if (query != NULL)
- GNUNET_CRYPTO_hash (&nb->keyspace,
- sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
- query);
- return GNUNET_OK;
-}
-
-
-/**
- * Check if the given SBlock is well-formed.
- *
- * @param sb the sblock data (or at least "dsize" bytes claiming to be one)
- * @param dsize size of "kb" in bytes; check for < sizeof(struct SBlock)!
- * @param query where to store the query that this block answers
- * @param namespace where to store the namespace that this block belongs to
- * @return GNUNET_OK if this is actually a well-formed SBlock
- */
-static int
-check_sblock (const struct SBlock *sb,
- size_t dsize,
- GNUNET_HashCode *query,
- GNUNET_HashCode *namespace)
-{
- if (dsize < sizeof (struct SBlock))
- {
- GNUNET_break_op (0);
- return GNUNET_SYSERR;
- }
- if (dsize !=
- ntohl (sb->purpose.size) + sizeof (struct GNUNET_CRYPTO_RsaSignature))
- {
- GNUNET_break_op (0);
- return GNUNET_SYSERR;
- }
- if (GNUNET_OK !=
- GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_FS_SBLOCK,
- &sb->purpose,
- &sb->signature,
- &sb->subspace))
- {
- GNUNET_break_op (0);
- return GNUNET_SYSERR;
- }
- if (query != NULL)
- *query = sb->identifier;
- if (namespace != NULL)
- GNUNET_CRYPTO_hash (&sb->subspace,
- sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
- namespace);
- return GNUNET_OK;
-}
-
-
/**
* Transmit the given message by copying it to the target buffer
* "buf". "buf" will be NULL and "size" zero if the socket was closed
*/
const void *data;
- // FIXME: add 'struct ConnectedPeer' to track 'last_xxx_replies' here!
+ /**
+ * Who gave us this reply? NULL for local host.
+ */
+ struct ConnectedPeer *sender;
/**
* When the reply expires.
/**
* Type of the block.
*/
- uint32_t type;
+ 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;
};
struct ClientList *cl;
struct PutMessage *pm;
struct ConnectedPeer *cp;
+ struct GNUNET_TIME_Relative cur_delay;
GNUNET_HashCode chash;
GNUNET_HashCode mhash;
size_t msize;
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_DATASTORE_BLOCKTYPE_DBLOCK:
- case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
+ 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->drq != NULL)
+ if (pr->qe != NULL)
{
if (pr->client_request_list != NULL)
GNUNET_SERVER_receive_done (pr->client_request_list->client_list->client,
GNUNET_YES);
- GNUNET_FS_drq_get_cancel (pr->drq);
- pr->drq = NULL;
+ GNUNET_DATASTORE_cancel (pr->qe);
+ pr->qe = NULL;
}
pr->do_remove = GNUNET_YES;
if (pr->task != GNUNET_SCHEDULER_NO_TASK)
key,
pr));
break;
- case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
+ case GNUNET_BLOCK_TYPE_SBLOCK:
if (pr->namespace == NULL)
{
GNUNET_break (0);
return GNUNET_YES; /* wrong namespace */
}
/* then: fall-through! */
- case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
- case GNUNET_DATASTORE_BLOCKTYPE_NBLOCK:
+ case GNUNET_BLOCK_TYPE_KBLOCK:
+ case GNUNET_BLOCK_TYPE_NBLOCK:
if (pr->bf != NULL)
{
mingle_hash (&chash, pr->mingle, &mhash);
}
prq->priority += pr->remaining_priority;
pr->remaining_priority = 0;
- if (pr->client_request_list != NULL)
+ if (NULL != pr->client_request_list)
{
GNUNET_STATISTICS_update (stats,
gettext_noop ("# replies received for local clients"),
}
GNUNET_break (cl->th != NULL);
if (pr->do_remove)
- destroy_pending_request (pr);
+ {
+ prq->finished = GNUNET_YES;
+ destroy_pending_request (pr);
+ }
}
else
{
memcpy (&pm[1], prq->data, prq->size);
add_to_pending_messages_for_peer (cp, reply, pr);
}
- // FIXME: implement hot-path routing statistics keeping!
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.
*
const struct PutMessage *put;
uint16_t msize;
size_t dsize;
- uint32_t type;
+ enum GNUNET_BLOCK_Type type;
struct GNUNET_TIME_Absolute expiration;
GNUNET_HashCode query;
struct ProcessReplyClosure prq;
+ const struct SBlock *sb;
msize = ntohs (message->size);
if (msize < sizeof (struct PutMessage))
type = ntohl (put->type);
expiration = GNUNET_TIME_absolute_ntoh (put->expiration);
- /* first, validate! */
- switch (type)
+ if (GNUNET_OK !=
+ GNUNET_BLOCK_check_block (type,
+ &put[1],
+ dsize,
+ &query))
{
- case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
- case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
- GNUNET_CRYPTO_hash (&put[1], dsize, &query);
- break;
- case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
- if (GNUNET_OK !=
- check_kblock ((const struct KBlock*) &put[1],
- dsize,
- &query))
- return GNUNET_SYSERR;
- break;
- case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
- if (GNUNET_OK !=
- check_sblock ((const struct SBlock*) &put[1],
- dsize,
- &query,
- &prq.namespace))
- return GNUNET_SYSERR;
- break;
- case GNUNET_DATASTORE_BLOCKTYPE_NBLOCK:
- if (GNUNET_OK !=
- check_nblock ((const struct NBlock*) &put[1],
- dsize,
- &query))
- return GNUNET_SYSERR;
- return GNUNET_OK;
- default:
- /* unknown block type */
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,
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);
- // FIXME: if migration is on and load is low,
- // queue to store data in datastore;
- // use "prq.priority" for that!
+ 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;
}
const GNUNET_HashCode * key,
uint32_t size,
const void *data,
- uint32_t type,
+ enum GNUNET_BLOCK_Type type,
uint32_t priority,
uint32_t anonymity,
struct GNUNET_TIME_Absolute
struct PendingRequest *pr = cls;
struct ProcessReplyClosure prq;
struct CheckDuplicateRequestClosure cdrc;
+ const struct SBlock *sb;
GNUNET_HashCode dhash;
GNUNET_HashCode mhash;
GNUNET_HashCode query;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Done processing local replies, forwarding request to other peers.\n");
#endif
- pr->drq = NULL;
+ pr->qe = NULL;
if (pr->client_request_list != NULL)
{
GNUNET_SERVER_receive_done (pr->client_request_list->client_list->client,
GNUNET_h2s (key),
type);
#endif
- if (type == GNUNET_DATASTORE_BLOCKTYPE_ONDEMAND)
+ if (type == GNUNET_BLOCK_TYPE_ONDEMAND)
{
#if DEBUG_FS
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
anonymity, expiration, uid,
&process_local_reply,
pr))
- GNUNET_FS_drq_get_next (GNUNET_YES);
+ if (pr->qe != NULL)
+ GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
return;
}
/* check for duplicates */
gettext_noop ("# results filtered by query bloomfilter"),
1,
GNUNET_NO);
- GNUNET_FS_drq_get_next (GNUNET_YES);
+ if (pr->qe != NULL)
+ GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
return;
}
#if DEBUG_FS
prq.data = data;
prq.expiration = expiration;
prq.size = size;
- if ( (type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) &&
- (GNUNET_OK != check_sblock ((const struct SBlock*) data,
- size,
- &query,
- &prq.namespace)) )
+ 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);
- /* FIXME: consider removing the block? */
- GNUNET_FS_drq_get_next (GNUNET_YES);
+ 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 ( (type == GNUNET_DATASTORE_BLOCKTYPE_DBLOCK) ||
- (type == GNUNET_DATASTORE_BLOCKTYPE_IBLOCK) )
+ if (prq.finished == GNUNET_YES)
+ return;
+ if (pr->qe == NULL)
+ return; /* done here */
+ if ( (type == GNUNET_BLOCK_TYPE_DBLOCK) ||
+ (type == GNUNET_BLOCK_TYPE_IBLOCK) )
{
- GNUNET_FS_drq_get_next (GNUNET_NO);
+ GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
return;
}
if ( (pr->client_request_list == NULL) &&
gettext_noop ("# processing result set cut short due to load"),
1,
GNUNET_NO);
- GNUNET_FS_drq_get_next (GNUNET_NO);
+ GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
return;
}
- GNUNET_FS_drq_get_next (GNUNET_YES);
+ GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
}
uint32_t bm;
size_t bfsize;
uint32_t ttl_decrement;
- uint32_t type;
+ enum GNUNET_BLOCK_Type type;
double preference;
int have_ns;
type = ntohl (gm->type);
switch (type)
{
- case GNUNET_DATASTORE_BLOCKTYPE_ANY:
- case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
- case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
- case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
- case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
+ case GNUNET_BLOCK_TYPE_ANY:
+ case GNUNET_BLOCK_TYPE_DBLOCK:
+ case GNUNET_BLOCK_TYPE_IBLOCK:
+ case GNUNET_BLOCK_TYPE_KBLOCK:
+ case GNUNET_BLOCK_TYPE_SBLOCK:
break;
default:
GNUNET_break_op (0);
bfsize = msize - sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode);
bm = ntohl (gm->hash_bitmap);
if ( (0 != (bm & GET_MESSAGE_BIT_SKS_NAMESPACE)) &&
- (type != GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) )
+ (type != GNUNET_BLOCK_TYPE_SBLOCK) )
{
GNUNET_break_op (0);
return GNUNET_SYSERR;
gettext_noop ("# requests dropped due TTL underflow"),
1,
GNUNET_NO);
- /* integer underflow => drop (should be very rare)! */
+ /* integer underflow => drop (should be very rare)! */
GNUNET_free (pr);
return GNUNET_OK;
}
}
pr->cp = cp;
- GNUNET_CONTAINER_multihashmap_put (query_request_map,
- &gm->query,
- pr,
- GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
- GNUNET_CONTAINER_multihashmap_put (peer_request_map,
- &other->hashPubKey,
- pr,
- GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
+ GNUNET_break (GNUNET_OK ==
+ GNUNET_CONTAINER_multihashmap_put (query_request_map,
+ &gm->query,
+ pr,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
+ GNUNET_break (GNUNET_OK ==
+ GNUNET_CONTAINER_multihashmap_put (peer_request_map,
+ &other->hashPubKey,
+ pr,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
pr->hnode = GNUNET_CONTAINER_heap_insert (requests_by_expiration_heap,
pr,
cps->inc_preference += preference;
/* process locally */
- if (type == GNUNET_DATASTORE_BLOCKTYPE_DBLOCK)
- type = GNUNET_DATASTORE_BLOCKTYPE_ANY; /* to get on-demand as well */
+ if (type == GNUNET_BLOCK_TYPE_DBLOCK)
+ type = GNUNET_BLOCK_TYPE_ANY; /* to get on-demand as well */
timeout = GNUNET_TIME_relative_multiply (BASIC_DATASTORE_REQUEST_DELAY,
(pr->priority + 1));
- pr->drq = GNUNET_FS_drq_get (&gm->query,
- type,
- &process_local_reply,
- pr,
- timeout,
- GNUNET_NO);
+ pr->qe = GNUNET_DATASTORE_get (dsh,
+ &gm->query,
+ type,
+ pr->priority + 1,
+ MAX_DATASTORE_QUEUE,
+ timeout,
+ &process_local_reply,
+ pr);
/* Are multiple results possible? If so, start processing remotely now! */
switch (pr->type)
{
- case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
- case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
+ case GNUNET_BLOCK_TYPE_DBLOCK:
+ case GNUNET_BLOCK_TYPE_IBLOCK:
/* only one result, wait for datastore */
break;
default:
struct PendingRequest *pr;
uint16_t msize;
unsigned int sc;
- uint32_t type;
+ enum GNUNET_BLOCK_Type type;
msize = ntohs (message->size);
if ( (msize < sizeof (struct SearchMessage)) ||
#endif
switch (type)
{
- case GNUNET_DATASTORE_BLOCKTYPE_ANY:
- case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
- case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
- case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
- case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
- case GNUNET_DATASTORE_BLOCKTYPE_NBLOCK:
+ case GNUNET_BLOCK_TYPE_ANY:
+ case GNUNET_BLOCK_TYPE_DBLOCK:
+ case GNUNET_BLOCK_TYPE_IBLOCK:
+ case GNUNET_BLOCK_TYPE_KBLOCK:
+ case GNUNET_BLOCK_TYPE_SBLOCK:
+ case GNUNET_BLOCK_TYPE_NBLOCK:
break;
default:
GNUNET_break (0);
client_list = cl;
}
/* detect duplicate KBLOCK requests */
- if ( (type == GNUNET_DATASTORE_BLOCKTYPE_KBLOCK) ||
- (type == GNUNET_DATASTORE_BLOCKTYPE_NBLOCK) ||
- (type == GNUNET_DATASTORE_BLOCKTYPE_ANY) )
+ if ( (type == GNUNET_BLOCK_TYPE_KBLOCK) ||
+ (type == GNUNET_BLOCK_TYPE_NBLOCK) ||
+ (type == GNUNET_BLOCK_TYPE_ANY) )
{
crl = cl->rl_head;
while ( (crl != NULL) &&
1,
GNUNET_NO);
pr = GNUNET_malloc (sizeof (struct PendingRequest) +
- ((type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) ? sizeof(GNUNET_HashCode) : 0));
+ ((type == GNUNET_BLOCK_TYPE_SBLOCK) ? sizeof(GNUNET_HashCode) : 0));
crl = GNUNET_malloc (sizeof (struct ClientRequestList));
memset (crl, 0, sizeof (struct ClientRequestList));
crl->client_list = cl;
pr->anonymity_level = ntohl (sm->anonymity_level);
refresh_bloomfilter (pr);
pr->query = sm->query;
+ if (0 == (1 & ntohl (sm->options)))
+ pr->local_only = GNUNET_NO;
+ else
+ pr->local_only = GNUNET_YES;
switch (type)
{
- case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
- case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
+ case GNUNET_BLOCK_TYPE_DBLOCK:
+ case GNUNET_BLOCK_TYPE_IBLOCK:
if (0 != memcmp (&sm->target,
&all_zeros,
sizeof (GNUNET_HashCode)))
pr->target_pid = GNUNET_PEER_intern ((const struct GNUNET_PeerIdentity*) &sm->target);
break;
- case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
+ case GNUNET_BLOCK_TYPE_SBLOCK:
pr->namespace = (GNUNET_HashCode*) &pr[1];
memcpy (&pr[1], &sm->target, sizeof (GNUNET_HashCode));
break;
default:
break;
}
- GNUNET_CONTAINER_multihashmap_put (query_request_map,
- &sm->query,
- pr,
- GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
- if (type == GNUNET_DATASTORE_BLOCKTYPE_DBLOCK)
- type = GNUNET_DATASTORE_BLOCKTYPE_ANY; /* get on-demand blocks too! */
- pr->drq = GNUNET_FS_drq_get (&sm->query,
- type,
- &process_local_reply,
- pr,
- GNUNET_CONSTANTS_SERVICE_TIMEOUT,
- GNUNET_YES);
+ GNUNET_break (GNUNET_OK ==
+ GNUNET_CONTAINER_multihashmap_put (query_request_map,
+ &sm->query,
+ pr,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
+ if (type == GNUNET_BLOCK_TYPE_DBLOCK)
+ type = GNUNET_BLOCK_TYPE_ANY; /* get on-demand blocks too! */
+ pr->qe = GNUNET_DATASTORE_get (dsh,
+ &sm->query,
+ type,
+ -3, -1,
+ GNUNET_CONSTANTS_SERVICE_TIMEOUT,
+ &process_local_reply,
+ pr);
}
sched = s;
cfg = c;
stats = GNUNET_STATISTICS_create (sched, "fs", cfg);
+ min_migration_delay = GNUNET_TIME_UNIT_SECONDS; // FIXME: get from config
connected_peers = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
query_request_map = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
peer_request_map = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
GNUNET_TIME_UNIT_FOREVER_REL,
NULL,
NULL,
- NULL,
&peer_connect_handler,
&peer_disconnect_handler,
NULL, GNUNET_NO,
requests_by_expiration_heap = NULL;
GNUNET_CONTAINER_multihashmap_destroy (peer_request_map);
peer_request_map = NULL;
-
+ if (dsh != NULL)
+ {
+ GNUNET_DATASTORE_disconnect (dsh, GNUNET_NO);
+ dsh = NULL;
+ }
return GNUNET_SYSERR;
- }
+ }
+ /* FIXME: distinguish between sending and storing in options? */
+ if (active_migration)
+ consider_migration_gathering ();
GNUNET_SERVER_disconnect_notify (server,
&handle_client_disconnect,
NULL);
struct GNUNET_SERVER_Handle *server,
const struct GNUNET_CONFIGURATION_Handle *cfg)
{
- if ( (GNUNET_OK != GNUNET_FS_drq_init (sched, cfg)) ||
- (GNUNET_OK != GNUNET_FS_indexing_init (sched, cfg)) ||
+ active_migration = GNUNET_CONFIGURATION_get_value_yesno (cfg,
+ "FS",
+ "ACTIVEMIGRATION");
+ dsh = GNUNET_DATASTORE_connect (cfg,
+ sched);
+ if (dsh == NULL)
+ {
+ GNUNET_SCHEDULER_shutdown (sched);
+ return;
+ }
+ if ( (GNUNET_OK != GNUNET_FS_indexing_init (sched, cfg, dsh)) ||
(GNUNET_OK != main_init (sched, server, cfg)) )
{
GNUNET_SCHEDULER_shutdown (sched);
+ GNUNET_DATASTORE_disconnect (dsh, GNUNET_NO);
+ dsh = NULL;
return;
}
}