/*
This file is part of GNUnet.
- (C)
+ Copyright (C)
GNUnet is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published
*/
struct GNUNET_CADET_TransmitHandle *is_live_task;
+ /**
+ * Identity of the peer
+ */
+ struct GNUNET_PeerIdentity peer_id;
+
/**
* This is pobably followed by 'statistical' data (when we first saw
* him, how did we get his ID, how many pushes (in a timeinterval),
* Globals
***********************************************************************/
+/**
+ * Sampler used for the Brahms protocol itself.
+ */
+static struct RPS_Sampler *prot_sampler;
+
+/**
+ * Sampler used for the clients.
+ */
+static struct RPS_Sampler *client_sampler;
+
/**
* Set of all peers to keep track of them.
*/
struct GNUNET_PeerIdentity *peer;
GNUNET_assert (NULL != peer_list);
+ if (0 == list_size)
+ return NULL;
tmp_size = 0;
tmp_peer_list = NULL;
memcpy (tmp_peer_list, peer_list, list_size * sizeof (struct GNUNET_PeerIdentity));
peer = GNUNET_new (struct GNUNET_PeerIdentity);
- do
+ /**;
+ * Choose the r_index of the peer we want to return
+ * at random from the interval of the gossip list
+ */
+ r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
+ tmp_size);
+ *peer = tmp_peer_list[r_index];
+
+ while (in_arr (ignore_list, ignore_size, peer))
{
+ rem_from_list (tmp_peer_list, &tmp_size, peer);
+
+ if (0 == tmp_size)
+ {
+ GNUNET_free (peer);
+ return NULL;
+ }
+
/**;
* Choose the r_index of the peer we want to return
* at random from the interval of the gossip list
*/
r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
tmp_size);
-
*peer = tmp_peer_list[r_index];
- if (in_arr (tmp_peer_list, list_size, peer))
- {
- rem_from_list (tmp_peer_list, &tmp_size, peer);
- if (0 == tmp_size)
- return NULL;
- continue;
- }
+ }
- } while (NULL == peer);
- GNUNET_free (tmp_peer_list);
+ GNUNET_array_grow (tmp_peer_list, tmp_size, 0);
return peer;
}
* This is given to ntfy_tmt_rdy and called when the channel was
* successfully established.
*/
- size_t
+static size_t
peer_is_live (void *cls, size_t size, void *buf)
{
+ struct PeerContext *ctx = cls;
struct GNUNET_PeerIdentity *peer;
struct PeerContext *peer_ctx;
// 0 == size)
// TODO check
- peer = (struct GNUNET_PeerIdentity *) cls;
+ ctx->is_live_task = NULL;
+ peer = &ctx->peer_id;
peer_ctx = get_peer_ctx (peer_map, peer);
peer_ctx->peer_flags |= LIVING;
GNUNET_array_grow (peer_ctx->outstanding_ops, peer_ctx->num_outstanding_ops, 0);
}
- GNUNET_free (peer);
-
//if (NULL != peer_ctx->is_live_task)
//{
// GNUNET_CADET_notify_transmit_ready_cancel (peer_ctx->is_live_task);
const struct GNUNET_PeerIdentity *peer)
{
struct PeerContext *ctx;
- struct GNUNET_PeerIdentity *tmp_peer;
ctx = get_peer_ctx (peer_map, peer);
if (NULL == ctx->send_channel)
if (NULL == ctx->recv_channel)
{
- tmp_peer = GNUNET_new (struct GNUNET_PeerIdentity);
- *tmp_peer = *peer;
- ctx->is_live_task = GNUNET_CADET_notify_transmit_ready (ctx->send_channel, GNUNET_NO,
- GNUNET_TIME_UNIT_FOREVER_REL,
- 0, peer_is_live, tmp_peer);
+ ctx->peer_id = *peer;
+ ctx->is_live_task =
+ GNUNET_CADET_notify_transmit_ready (ctx->send_channel, GNUNET_NO,
+ GNUNET_TIME_UNIT_FOREVER_REL,
+ sizeof (struct GNUNET_MessageHeader),
+ peer_is_live, ctx);
}
// do I have to explicitly put it in the peer_map?
void
insert_in_sampler (void *cls, const struct GNUNET_PeerIdentity *peer)
{
- RPS_sampler_update_list (peer);
+ RPS_sampler_update (prot_sampler, peer);
+ RPS_sampler_update (client_sampler, peer);
}
/**
* Check whether #insert_in_sampler was already scheduled
*/
- int
+static int
insert_in_sampler_scheduled (const struct PeerContext *peer_ctx)
{
unsigned int i;
/**
* Wrapper around #RPS_sampler_resize()
+ *
+ * If we do not have enough sampler elements, double current sampler size
+ * If we have more than enough sampler elements, halv current sampler size
*/
- void
-resize_wrapper ()
+static void
+resize_wrapper (struct RPS_Sampler *sampler, uint32_t new_size)
+{
+ unsigned int sampler_size;
+
+ // TODO statistics
+ // TODO respect the min, max
+ sampler_size = RPS_sampler_get_size (sampler);
+ if (sampler_size > new_size * 4)
+ { /* Shrinking */
+ RPS_sampler_resize (sampler, sampler_size / 2);
+ }
+ else if (sampler_size < new_size)
+ { /* Growing */
+ RPS_sampler_resize (sampler, sampler_size * 2);
+ }
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "sampler_size is now %u\n", sampler_size);
+}
+
+
+/**
+ * Wrapper around #RPS_sampler_resize() resizing the client sampler
+ */
+static void
+client_resize_wrapper ()
{
uint32_t bigger_size;
unsigned int sampler_size;
// TODO statistics
+ sampler_size = RPS_sampler_get_size (client_sampler);
+
if (sampler_size_est_need > sampler_size_client_need)
- bigger_size = sampler_size_client_need;
- else
bigger_size = sampler_size_est_need;
+ else
+ bigger_size = sampler_size_client_need;
// TODO respect the min, max
- sampler_size = RPS_sampler_get_size ();
- if (sampler_size > bigger_size * 4)
- { /* Shrinking */
- RPS_sampler_resize (sampler_size / 2);
- }
- else if (sampler_size < bigger_size)
- { /* Growing */
- RPS_sampler_resize (sampler_size * 2);
- }
+ resize_wrapper (client_sampler, bigger_size);
LOG (GNUNET_ERROR_TYPE_DEBUG, "sampler_size is now %u\n", sampler_size);
}
max_round_duration.rel_value_us / request_rate.rel_value_us;
/* Resize the sampler */
- resize_wrapper ();
+ client_resize_wrapper ();
}
last_request = GNUNET_TIME_absolute_get ();
}
* /Util functions
***********************************************************************/
+
+
+
+
/**
* Function called by NSE.
*
LOG (GNUNET_ERROR_TYPE_DEBUG,
"Received a ns estimate - logest: %f, std_dev: %f (old_size: %u)\n",
- logestimate, std_dev, RPS_sampler_get_size());
+ logestimate, std_dev, RPS_sampler_get_size (prot_sampler));
//scale = .01;
estimate = GNUNET_NSE_log_estimate_to_n (logestimate);
// GNUNET_NSE_log_estimate_to_n (logestimate);
- estimate = pow (estimate, 1./3);
+ estimate = pow (estimate, 1.0 / 3);
// TODO add if std_dev is a number
// estimate += (std_dev * scale);
- if (0 < ceil (estimate))
+ if (2 < ceil (estimate))
{
LOG (GNUNET_ERROR_TYPE_DEBUG, "Changing estimate to %f\n", estimate);
sampler_size_est_need = estimate;
LOG (GNUNET_ERROR_TYPE_DEBUG, "Not using estimate %f\n", estimate);
/* If the NSE has changed adapt the lists accordingly */
- resize_wrapper ();
+ resize_wrapper (prot_sampler, sampler_size_est_need);
+ client_resize_wrapper ();
}
LOG (GNUNET_ERROR_TYPE_DEBUG, "Client requested %" PRIX32 " random peer(s).\n", num_peers);
- RPS_sampler_get_n_rand_peers (client_respond, client, num_peers, GNUNET_YES);
+ RPS_sampler_get_n_rand_peers (client_sampler, client_respond,
+ client, num_peers, GNUNET_YES);
GNUNET_SERVER_receive_done (client,
GNUNET_OK);
peers = (struct GNUNET_PeerIdentity *) &message[1];
for ( i = 0 ; i < ntohl (in_msg->num_peers) ; i++ )
- RPS_sampler_update_list (&peers[i]);
+ RPS_sampler_update (prot_sampler, &peers[i]);
+ RPS_sampler_update (client_sampler, &peers[i]);
GNUNET_SERVER_receive_done (client,
GNUNET_OK);
// FIXME wait for cadet to change this function
sender_ctx = get_peer_ctx (peer_map, sender);
- if (0 == (peer_ctx->peer_flags || PULL_REPLY_PENDING))
+ if (0 == (sender_ctx->peer_flags || PULL_REPLY_PENDING))
{
GNUNET_break_op (0);
return GNUNET_OK;
else if (GNUNET_NO == insert_in_pull_list_scheduled (peer_ctx))
{
out_op.op = insert_in_pull_list;
+ out_op.op_cls = NULL;
GNUNET_array_append (peer_ctx->outstanding_ops, peer_ctx->num_outstanding_ops, out_op);
}
}
unsigned int *permut;
unsigned int n_peers; /* Number of peers we send pushes/pulls to */
struct GNUNET_MQ_Envelope *ev;
- const struct GNUNET_PeerIdentity *peer;
+ struct GNUNET_PeerIdentity peer;
+ struct GNUNET_PeerIdentity *tmp_peer;
struct GNUNET_MQ_Handle *mq;
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Printing gossip list:\n");
+ for (i = 0 ; i < gossip_list_size ; i++)
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "\t%s\n", GNUNET_i2s (&gossip_list[i]));
// TODO log lists, ...
-
/* Would it make sense to have one shuffeled gossip list and then
* to send PUSHes to first alpha peers, PULL requests to next beta peers and
* use the rest to update sampler?
{
permut = GNUNET_CRYPTO_random_permute (GNUNET_CRYPTO_QUALITY_STRONG,
(unsigned int) gossip_list_size);
- if (0 != gossip_list_size)
+ n_peers = ceil (alpha * gossip_list_size);
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Going to send pushes to %u ceil (%f * %u) peers.\n",
+ n_peers, alpha, gossip_list_size);
+ for (i = 0 ; i < n_peers ; i++)
{
- n_peers = round (alpha * gossip_list_size);
- if (0 == n_peers)
- n_peers = 1;
- LOG (GNUNET_ERROR_TYPE_DEBUG, "Going to send pushes to %u (%f * %u) peers.\n",
- n_peers, alpha, gossip_list_size);
- for ( i = 0 ; i < n_peers ; i++ )
- {
- peer = &gossip_list[permut[i]];
- if (0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity, peer)) // TODO
- { // FIXME if this fails schedule/loop this for later
- LOG (GNUNET_ERROR_TYPE_DEBUG, "Sending PUSH to peer %s of gossiped list.\n", GNUNET_i2s (peer));
-
- ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_RPS_PP_PUSH);
- mq = get_mq (peer_map, peer);
- GNUNET_MQ_send (mq, ev);
- }
+ peer = gossip_list[permut[i]];
+ if (0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity, &peer)) // TODO
+ { // FIXME if this fails schedule/loop this for later
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Sending PUSH to peer %s of gossiped list.\n",
+ GNUNET_i2s (&peer));
+
+ ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_RPS_PP_PUSH);
+ mq = get_mq (peer_map, &peer);
+ GNUNET_MQ_send (mq, ev);
}
}
+ GNUNET_free (permut);
}
/* Send PULL requests */
//permut = GNUNET_CRYPTO_random_permute (GNUNET_CRYPTO_QUALITY_STRONG, (unsigned int) sampler_list->size);
- if (0 != gossip_list_size)
+ n_peers = ceil (beta * gossip_list_size);
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Going to send pulls to %u ceil (%f * %u) peers.\n",
+ n_peers, beta, gossip_list_size);
+ for (i = 0 ; i < n_peers ; i++)
{
- n_peers = round (beta * gossip_list_size);
- if (0 == n_peers)
- n_peers = 1;
- LOG (GNUNET_ERROR_TYPE_DEBUG, "Going to send pulls to %u (%f * %u) peers.\n",
- n_peers, beta, gossip_list_size);
- for ( i = 0 ; i < n_peers ; i++ )
+ tmp_peer = get_rand_peer_ignore_list (gossip_list, gossip_list_size,
+ pending_pull_reply_list, pending_pull_reply_list_size);
+ if (NULL != tmp_peer)
{
- peer = get_rand_peer_ignore_list (gossip_list, gossip_list_size,
- pending_pull_reply_list, pending_pull_reply_list_size);
- if (NULL != peer)
- {
- GNUNET_array_append (pending_pull_reply_list, pending_pull_reply_list_size, *peer);
+ peer = *tmp_peer;
+ GNUNET_free (tmp_peer);
- if (0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity, peer))
- { // FIXME if this fails schedule/loop this for later
- LOG (GNUNET_ERROR_TYPE_DEBUG, "Sending PULL request to peer %s of gossiped list.\n", GNUNET_i2s (peer));
+ GNUNET_array_append (pending_pull_reply_list, pending_pull_reply_list_size, peer);
- ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REQUEST);
- //pull_msg = NULL;
- mq = get_mq (peer_map, peer);
- GNUNET_MQ_send (mq, ev);
- }
+ if (0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity, &peer))
+ { // FIXME if this fails schedule/loop this for later
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Sending PULL request to peer %s of gossiped list.\n",
+ GNUNET_i2s (&peer));
+
+ ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REQUEST);
+ mq = get_mq (peer_map, &peer);
+ GNUNET_MQ_send (mq, ev);
}
}
}
uint32_t first_border;
uint32_t second_border;
- first_border = round (alpha * sampler_size_est_need);
- second_border = first_border + round (beta * sampler_size_est_need);
+ first_border = ceil (alpha * sampler_size_est_need);
+ second_border = first_border + ceil (beta * sampler_size_est_need);
GNUNET_array_grow (gossip_list, gossip_list_size, second_border);
- for ( i = 0 ; i < first_border ; i++ )
+ for (i = 0 ; i < first_border ; i++)
{ // TODO use RPS_sampler_get_n_rand_peers
/* Update gossip list with peers received through PUSHes */
r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
// TODO change the peer_flags accordingly
}
- for ( i = first_border ; i < second_border ; i++ )
+ for (i = first_border ; i < second_border ; i++)
{
/* Update gossip list with peers received through PULLs */
r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
// TODO change the peer_flags accordingly
}
- for ( i = second_border ; i < gossip_list_size ; i++ )
+ for (i = second_border ; i < sampler_size_est_need ; i++)
{
/* Update gossip list with peers from history */
- RPS_sampler_get_n_rand_peers (hist_update, NULL, 1, GNUNET_NO);
+ RPS_sampler_get_n_rand_peers (prot_sampler, hist_update, NULL, 1, GNUNET_NO);
num_hist_update_tasks++;
// TODO change the peer_flags accordingly
}
}
else
{
- LOG (GNUNET_ERROR_TYPE_DEBUG, "No update of the gossip list. ()\n");
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "No update of the gossip list.\n");
}
// TODO independent of that also get some peers from CADET_get_peers()?
for ( i = 0 ; i < push_list_size ; i++ )
{
- RPS_sampler_update_list (&push_list[i]);
+ RPS_sampler_update (prot_sampler, &push_list[i]);
+ RPS_sampler_update (client_sampler, &push_list[i]);
// TODO set in_flag?
}
for ( i = 0 ; i < pull_list_size ; i++ )
{
- RPS_sampler_update_list (&pull_list[i]);
+ RPS_sampler_update (prot_sampler, &push_list[i]);
+ RPS_sampler_update (client_sampler, &push_list[i]);
// TODO set in_flag?
}
* Open a connection to given peer and store channel and mq.
*/
void
-insertCB (void *cls, const struct GNUNET_PeerIdentity *id)
+insertCB (void *cls, struct RPS_Sampler *sampler,
+ const struct GNUNET_PeerIdentity *id)
{
// We open a channel to be notified when this peer goes down.
(void) get_channel (peer_map, id);
* Close the connection to given peer and delete channel and mq.
*/
void
-removeCB (void *cls, const struct GNUNET_PeerIdentity *id)
+removeCB (void *cls, struct RPS_Sampler *sampler,
+ const struct GNUNET_PeerIdentity *id)
{
size_t s;
struct PeerContext *ctx;
- s = RPS_sampler_count_id (id);
+ s = RPS_sampler_count_id (sampler, id);
if ( 1 >= s )
{
if (GNUNET_YES == GNUNET_CONTAINER_multipeermap_contains (peer_map, id))
if (GNUNET_NO == insert_in_gossip_list_scheduled (peer_ctx))
{
out_op.op = insert_in_gossip_list;
+ out_op.op_cls = NULL;
GNUNET_array_append (peer_ctx->outstanding_ops, peer_ctx->num_outstanding_ops, out_op);
}
{
struct PeerContext *peer_ctx;
const struct GNUNET_CADET_Channel *ch = (const struct GNUNET_CADET_Channel *) cls;
+ struct GNUNET_CADET_Channel *recv;
+ struct GNUNET_CADET_Channel *send;
peer_ctx = (struct PeerContext *) value;
+ if (0 != peer_ctx->num_outstanding_ops)
+ GNUNET_array_grow (peer_ctx->outstanding_ops, peer_ctx->num_outstanding_ops, 0);
+
if (NULL != peer_ctx->mq)
GNUNET_MQ_destroy (peer_ctx->mq);
peer_ctx->is_live_task = NULL;
}
- if (NULL != peer_ctx->send_channel
- && ch != peer_ctx->send_channel)
- GNUNET_CADET_channel_destroy (peer_ctx->send_channel);
+ send = peer_ctx->send_channel;
+ peer_ctx->send_channel = NULL;
+ recv = peer_ctx->send_channel;
+ peer_ctx->recv_channel = NULL;
- if (NULL != peer_ctx->recv_channel
- && ch != peer_ctx->recv_channel)
- GNUNET_CADET_channel_destroy (peer_ctx->recv_channel);
+ if (NULL != send
+ && ch != send)
+ {
+ GNUNET_CADET_channel_destroy (send);
+ }
+
+ if (NULL != recv
+ && ch != recv)
+ {
+ GNUNET_CADET_channel_destroy (recv);
+ }
if (GNUNET_NO == GNUNET_CONTAINER_multipeermap_remove_all (peer_map, key))
LOG (GNUNET_ERROR_TYPE_WARNING, "removing peer from peer_map failed\n");
+ else
+ GNUNET_free (peer_ctx);
return GNUNET_YES;
}
}
+ {
if (GNUNET_SYSERR == GNUNET_CONTAINER_multipeermap_iterate (peer_map, peer_remove_cb, NULL))
LOG (GNUNET_ERROR_TYPE_WARNING,
"Iterating over peers to disconnect from them was cancelled\n");
-
- GNUNET_CONTAINER_multipeermap_destroy (peer_map);
+ }
GNUNET_NSE_disconnect (nse);
GNUNET_CADET_disconnect (cadet_handle);
- RPS_sampler_destroy ();
- GNUNET_array_grow (request_deltas, request_deltas_size, 0);
+ RPS_sampler_destroy (prot_sampler);
+ RPS_sampler_destroy (client_sampler);
+ GNUNET_break (0 == GNUNET_CONTAINER_multipeermap_size (peer_map));
+ GNUNET_CONTAINER_multipeermap_destroy (peer_map);
GNUNET_array_grow (gossip_list, gossip_list_size, 0);
GNUNET_array_grow (push_list, push_list_size, 0);
GNUNET_array_grow (pull_list, pull_list_size, 0);
struct GNUNET_PeerIdentity *peer;
struct PeerContext *peer_ctx;
- LOG (GNUNET_ERROR_TYPE_DEBUG, "Channel to remote peer was destroyed.\n");
-
peer = (struct GNUNET_PeerIdentity *) GNUNET_CADET_channel_get_info (
(struct GNUNET_CADET_Channel *) channel, GNUNET_CADET_OPTION_PEER);
// Guess simply casting isn't the nicest way...
// FIXME wait for cadet to change this function
- RPS_sampler_reinitialise_by_value (peer);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Cleaning up channel to peer %s\n",
+ GNUNET_i2s (peer));
+
+ RPS_sampler_reinitialise_by_value (prot_sampler, peer);
+ RPS_sampler_reinitialise_by_value (client_sampler, peer);
- peer_ctx = GNUNET_CONTAINER_multipeermap_get (peer_map, peer);
- /* Somwewhat {ab,re}use the iterator function */
- /* Cast to void is ok, because it's used as void in peer_remove_cb */
- (void) peer_remove_cb ((void *) channel, peer, peer_ctx);
+ if (GNUNET_YES == GNUNET_CONTAINER_multipeermap_contains (peer_map, peer))
+ {
+ peer_ctx = GNUNET_CONTAINER_multipeermap_get (peer_map, peer);
+
+ if (NULL == peer_ctx) /* It could have been removed by shutdown_task */
+ return;
+
+ /* Somwewhat {ab,re}use the iterator function */
+ /* Cast to void is ok, because it's used as void in peer_remove_cb */
+ (void) peer_remove_cb ((void *) channel, peer, peer_ctx);
+ }
}
half_round_interval = GNUNET_TIME_relative_multiply (round_interval, .5);
max_round_interval = GNUNET_TIME_relative_add (round_interval, half_round_interval);
- RPS_sampler_init (sampler_size_est_need, max_round_interval,
+ prot_sampler = RPS_sampler_init (sampler_size_est_need, max_round_interval,
+ insertCB, NULL, removeCB, NULL);
+ client_sampler = RPS_sampler_init (sampler_size_est_need, max_round_interval,
insertCB, NULL, removeCB, NULL);
/* Initialise push and pull maps */
projects / oweals / gnunet.git / blobdiff