-
/*
This file is part of GNUnet.
Copyright (C) 2001-2017 GNUnet e.V.
* @author Christian Grothoff
*/
#include "platform.h"
+#include "gnunet-service-cadet-new.h"
#include "gnunet-service-cadet-new_channel.h"
+#include "gnunet-service-cadet-new_connection.h"
#include "gnunet-service-cadet-new_paths.h"
#include "gnunet-service-cadet-new_peer.h"
-#include "gnunet-service-cadet-new_connection.h"
+#include "gnunet-service-cadet-new_tunnels.h"
#include "gnunet_cadet_service.h"
+#include "gnunet_statistics_service.h"
#include "cadet_protocol.h"
+#define LOG(level, ...) GNUNET_log_from(level,"cadet-con",__VA_ARGS__)
+
+
/**
* All the states a connection can be in.
*/
CADET_CONNECTION_SENT,
/**
- * Connection confirmed, ready to carry traffic.
+ * We are an inbound connection, and received a CREATE. Need to
+ * send an CREATE_ACK back.
*/
- CADET_CONNECTION_READY,
+ CADET_CONNECTION_CREATE_RECEIVED,
/**
- * Connection to be destroyed, just waiting to empty queues.
+ * Connection confirmed, ready to carry traffic.
*/
- CADET_CONNECTION_DESTROYED,
+ CADET_CONNECTION_READY
- /**
- * Connection to be destroyed because of a distant peer, same as DESTROYED.
- */
- CADET_CONNECTION_BROKEN
};
*/
struct GNUNET_MQ_Envelope *env;
- /**
- * Message queue to the first hop, or NULL if we have no connection yet.
- */
- struct GNUNET_MQ_Handle *mq;
-
/**
* Handle for calling #GCP_request_mq_cancel() once we are finished.
*/
*/
struct GNUNET_SCHEDULER_Task *task;
+ /**
+ * Queue entry for keepalive messages.
+ */
+ struct CadetTunnelQueueEntry *keepalive_qe;
+
/**
* Function to call once we are ready to transmit.
*/
- GNUNET_SCHEDULER_TaskCallback ready_cb;
+ GCC_ReadyCallback ready_cb;
/**
* Closure for @e ready_cb.
*/
struct GNUNET_TIME_Relative retry_delay;
+ /**
+ * Performance metrics for this connection.
+ */
+ struct CadetConnectionMetrics metrics;
+
/**
* State of the connection.
*/
enum CadetConnectionState state;
+ /**
+ * Options for the route, control buffering.
+ */
+ enum GNUNET_CADET_ChannelOption options;
+
+ /**
+ * How many latency observations did we make for this connection?
+ */
+ unsigned int latency_datapoints;
+
/**
* Offset of our @e destination in @e path.
*/
unsigned int off;
+ /**
+ * Are we ready to transmit via @e mq_man right now?
+ */
+ int mqm_ready;
+
};
/**
- * Is the given connection currently ready for transmission?
+ * Lookup a connection by its identifier.
+ *
+ * @param cid identifier to resolve
+ * @return NULL if connection was not found
+ */
+struct CadetConnection *
+GCC_lookup (const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
+{
+ return GNUNET_CONTAINER_multishortmap_get (connections,
+ &cid->connection_of_tunnel);
+}
+
+
+/**
+ * Update the connection state. Also triggers the necessary
+ * MQM notifications.
*
- * @param cc connection to transmit on
- * @return #GNUNET_YES if we could transmit
+ * @param cc connection to update the state for
+ * @param new_state new state for @a cc
+ * @param new_mqm_ready new `mqm_ready` state for @a cc
*/
-int
-GCC_is_ready (struct CadetConnection *cc)
+static void
+update_state (struct CadetConnection *cc,
+ enum CadetConnectionState new_state,
+ int new_mqm_ready)
{
- return ( (NULL != cc->mq) &&
- (CADET_CONNECTION_READY == cc->state) &&
- (NULL == cc->env) ) ? GNUNET_YES : GNUNET_NO;
+ int old_ready;
+ int new_ready;
+
+ if ( (new_state == cc->state) &&
+ (new_mqm_ready == cc->mqm_ready) )
+ return; /* no change, nothing to do */
+ old_ready = ( (CADET_CONNECTION_READY == cc->state) &&
+ (GNUNET_YES == cc->mqm_ready) );
+ new_ready = ( (CADET_CONNECTION_READY == new_state) &&
+ (GNUNET_YES == new_mqm_ready) );
+ cc->state = new_state;
+ cc->mqm_ready = new_mqm_ready;
+ if (old_ready != new_ready)
+ cc->ready_cb (cc->ready_cb_cls,
+ new_ready);
}
/**
- * Destroy a connection.
+ * Destroy a connection, part of the internal implementation. Called
+ * only from #GCC_destroy_from_core() or #GCC_destroy_from_tunnel().
*
* @param cc connection to destroy
*/
-void
+static void
GCC_destroy (struct CadetConnection *cc)
{
- if (NULL != cc->env)
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Destroying %s\n",
+ GCC_2s (cc));
+ if (NULL != cc->mq_man)
{
- if (NULL != cc->mq)
- GNUNET_MQ_send_cancel (cc->env);
- else
- GNUNET_MQ_discard (cc->env);
- cc->env = NULL;
+ GCP_request_mq_cancel (cc->mq_man,
+ NULL);
+ cc->mq_man = NULL;
}
- if ( (NULL != cc->mq) &&
- (CADET_CONNECTION_SENDING_CREATE != cc->state) )
+ if (NULL != cc->task)
{
- /* Need to notify next hop that we are down. */
- struct GNUNET_MQ_Envelope *env;
- struct GNUNET_CADET_ConnectionDestroy *destroy_msg;
-
- env = GNUNET_MQ_msg (destroy_msg,
- GNUNET_MESSAGE_TYPE_CADET_CONNECTION_DESTROY);
- destroy_msg->cid = cc->cid;
- GNUNET_MQ_send (cc->mq,
- env);
+ GNUNET_SCHEDULER_cancel (cc->task);
+ cc->task = NULL;
+ }
+ if (NULL != cc->keepalive_qe)
+ {
+ GCT_send_cancel (cc->keepalive_qe);
+ cc->keepalive_qe = NULL;
}
- cc->mq = NULL;
- GCP_request_mq_cancel (cc->mq_man);
- cc->mq_man = NULL;
GCPP_del_connection (cc->path,
cc->off,
cc);
+ for (unsigned int i=0;i<cc->off;i++)
+ GCP_remove_connection (GCPP_get_peer_at_offset (cc->path,
+ i),
+ cc);
+ GNUNET_assert (GNUNET_YES ==
+ GNUNET_CONTAINER_multishortmap_remove (connections,
+ &GCC_get_id (cc)->connection_of_tunnel,
+ cc));
GNUNET_free (cc);
}
+
/**
- * Expand the shorter CADET hash to a full GNUnet hash.
+ * Destroy a connection, called when the CORE layer is already done
+ * (i.e. has received a BROKEN message), but if we still have to
+ * communicate the destruction of the connection to the tunnel (if one
+ * exists).
*
- * @param id hash to expand
- * @return expanded hash
+ * @param cc connection to destroy
*/
-static const struct GNUNET_HashCode *
-h2hc (const struct GNUNET_CADET_Hash *id)
+void
+GCC_destroy_without_core (struct CadetConnection *cc)
{
- static struct GNUNET_HashCode hc;
- char *ptr = (char *) &hc;
-
- GNUNET_assert (sizeof (hc) == 2 * sizeof (*id));
- GNUNET_memcpy (ptr,
- id,
- sizeof (*id));
- GNUNET_memcpy (&ptr[sizeof (*id)],
- id,
- sizeof (*id));
- return &hc;
+ if (NULL != cc->ct)
+ {
+ GCT_connection_lost (cc->ct);
+ cc->ct = NULL;
+ }
+ GCC_destroy (cc);
}
/**
- * Get the connection ID as a full hash.
+ * Destroy a connection, called if the tunnel association with the
+ * connection was already broken, but we still need to notify the CORE
+ * layer about the breakage.
*
- * @param cc Connection to get the ID from.
- * @return full hash ID of the connection.
+ * @param cc connection to destroy
*/
-const struct GNUNET_HashCode *
-GCC_get_h (const struct CadetConnection *cc)
+void
+GCC_destroy_without_tunnel (struct CadetConnection *cc)
{
- return h2hc (&cc->cid.connection_of_tunnel);
+ cc->ct = NULL;
+ if ( (CADET_CONNECTION_SENDING_CREATE != cc->state) &&
+ (NULL != cc->mq_man) )
+ {
+ struct GNUNET_MQ_Envelope *env;
+ struct GNUNET_CADET_ConnectionDestroyMessage *destroy_msg;
+
+ /* Need to notify next hop that we are down. */
+ env = GNUNET_MQ_msg (destroy_msg,
+ GNUNET_MESSAGE_TYPE_CADET_CONNECTION_DESTROY);
+ destroy_msg->cid = cc->cid;
+ GCP_request_mq_cancel (cc->mq_man,
+ env);
+ cc->mq_man = NULL;
+ }
+ GCC_destroy (cc);
}
/**
- * An ACK was received for this connection, process it.
+ * Obtain performance @a metrics from @a cc.
*
- * @param cc the connection that got the ACK.
+ * @param cc connection to query
+ * @return the metrics
*/
-void
-GCC_handle_ack (struct CadetConnection *cc)
+const struct CadetConnectionMetrics *
+GCC_get_metrics (struct CadetConnection *cc)
{
- GNUNET_SCHEDULER_cancel (cc->task);
-#if FIXME
- cc->task = GNUNET_SCHEDULER_add_delayed (cc->keepalive_period,
- &send_keepalive,
- cc);
-#endif
- cc->state = CADET_CONNECTION_READY;
- cc->ready_cb (cc->ready_cb_cls);
+ return &cc->metrics;
}
/**
- * Send a CREATE message to the first hop.
+ * Send a #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_KEEPALIVE through the
+ * tunnel to prevent it from timing out.
*
- * @param cls the `struct CadetConnection` to initiate
+ * @param cls the `struct CadetConnection` to keep alive.
*/
static void
-send_create (void *cls);
+send_keepalive (void *cls);
/**
- * We finished transmission of the create message, now wait for
- * ACK or retransmit.
+ * Keepalive was transmitted. Remember this, and possibly
+ * schedule the next one.
*
- * @param cls the `struct CadetConnection` that sent the create message
+ * @param cls the `struct CadetConnection` to keep alive.
+ * @param cid identifier of the connection within the tunnel, NULL
+ * if transmission failed
*/
static void
-transmit_create_done_cb (void *cls)
+keepalive_done (void *cls,
+ const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
{
struct CadetConnection *cc = cls;
- cc->state = CADET_CONNECTION_SENT;
- cc->env = NULL;
- /* FIXME: at some point, we need to reset the delay back to 0! */
- cc->retry_delay = GNUNET_TIME_STD_BACKOFF (cc->retry_delay);
- cc->task = GNUNET_SCHEDULER_add_delayed (cc->retry_delay,
- &send_create,
- cc);
+ cc->keepalive_qe = NULL;
+ if ( (GNUNET_YES == cc->mqm_ready) &&
+ (NULL == cc->task) )
+ cc->task = GNUNET_SCHEDULER_add_delayed (keepalive_period,
+ &send_keepalive,
+ cc);
+}
+
+
+/**
+ * Send a #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_KEEPALIVE through the
+ * tunnel to prevent it from timing out.
+ *
+ * @param cls the `struct CadetConnection` to keep alive.
+ */
+static void
+send_keepalive (void *cls)
+{
+ struct CadetConnection *cc = cls;
+ struct GNUNET_MessageHeader msg;
+
+ cc->task = NULL;
+ if (CADET_TUNNEL_KEY_OK != GCT_get_estate (cc->ct->t))
+ {
+ /* Tunnel not yet ready, wait with keepalives... */
+ cc->task = GNUNET_SCHEDULER_add_delayed (keepalive_period,
+ &send_keepalive,
+ cc);
+ return;
+ }
+ GNUNET_assert (NULL != cc->ct);
+ GNUNET_assert (GNUNET_YES == cc->mqm_ready);
+ GNUNET_assert (NULL == cc->keepalive_qe);
+ LOG (GNUNET_ERROR_TYPE_INFO,
+ "Sending KEEPALIVE on behalf of %s via %s\n",
+ GCC_2s (cc),
+ GCT_2s (cc->ct->t));
+ GNUNET_STATISTICS_update (stats,
+ "# keepalives sent",
+ 1,
+ GNUNET_NO);
+ msg.size = htons (sizeof (msg));
+ msg.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_KEEPALIVE);
+
+ cc->keepalive_qe
+ = GCT_send (cc->ct->t,
+ &msg,
+ &keepalive_done,
+ cc);
+}
+
+
+/**
+ * We sent a message for which we expect to receive an ACK via
+ * the connection identified by @a cti.
+ *
+ * @param cid connection identifier where we expect an ACK
+ */
+void
+GCC_ack_expected (const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
+{
+ struct CadetConnection *cc;
+
+ cc = GCC_lookup (cid);
+ if (NULL == cc)
+ return; /* whopise, connection alredy down? */
+ cc->metrics.num_acked_transmissions++;
+}
+
+
+/**
+ * We observed an ACK for a message that was originally sent via
+ * the connection identified by @a cti.
+ *
+ * @param cti connection identifier where we got an ACK for a message
+ * that was originally sent via this connection (the ACK
+ * may have gotten back to us via a different connection).
+ */
+void
+GCC_ack_observed (const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
+{
+ struct CadetConnection *cc;
+
+ cc = GCC_lookup (cid);
+ if (NULL == cc)
+ return; /* whopise, connection alredy down? */
+ cc->metrics.num_successes++;
+}
+
+
+/**
+ * We observed some the given @a latency on the connection
+ * identified by @a cti. (The same connection was taken
+ * in both directions.)
+ *
+ * @param cid connection identifier where we measured latency
+ * @param latency the observed latency
+ */
+void
+GCC_latency_observed (const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid,
+ struct GNUNET_TIME_Relative latency)
+{
+ struct CadetConnection *cc;
+ double weight;
+ double result;
+
+ cc = GCC_lookup (cid);
+ if (NULL == cc)
+ return; /* whopise, connection alredy down? */
+ GNUNET_STATISTICS_update (stats,
+ "# latencies observed",
+ 1,
+ GNUNET_NO);
+ cc->latency_datapoints++;
+ if (cc->latency_datapoints >= 7)
+ weight = 7.0;
+ else
+ weight = cc->latency_datapoints;
+ /* Compute weighted average, giving at MOST weight 7 to the
+ existing values, or less if that value is based on fewer than 7
+ measurements. */
+ result = (weight * cc->metrics.aged_latency.rel_value_us) + 1.0 * latency.rel_value_us;
+ result /= (weight + 1.0);
+ cc->metrics.aged_latency.rel_value_us = (uint64_t) result;
+}
+
+
+/**
+ * A #GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE_ACK was received for this connection, implying
+ * that the end-to-end connection is up. Process it.
+ *
+ * @param cc the connection that got the ACK.
+ */
+void
+GCC_handle_connection_create_ack (struct CadetConnection *cc)
+{
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Received CADET_CONNECTION_CREATE_ACK for %s in state %d (%s)\n",
+ GCC_2s (cc),
+ cc->state,
+ (GNUNET_YES == cc->mqm_ready) ? "MQM ready" : "MQM busy");
+ if (CADET_CONNECTION_READY == cc->state)
+ return; /* Duplicate ACK, ignore */
+ if (NULL != cc->task)
+ {
+ GNUNET_SCHEDULER_cancel (cc->task);
+ cc->task = NULL;
+ }
+ cc->metrics.age = GNUNET_TIME_absolute_get ();
+ update_state (cc,
+ CADET_CONNECTION_READY,
+ cc->mqm_ready);
+ if ( (NULL == cc->keepalive_qe) &&
+ (GNUNET_YES == cc->mqm_ready) &&
+ (NULL == cc->task) )
+ cc->task = GNUNET_SCHEDULER_add_delayed (keepalive_period,
+ &send_keepalive,
+ cc);
+}
+
+
+/**
+ * Handle KX message.
+ *
+ * @param cc connection that received encrypted message
+ * @param msg the key exchange message
+ */
+void
+GCC_handle_kx (struct CadetConnection *cc,
+ const struct GNUNET_CADET_TunnelKeyExchangeMessage *msg)
+{
+ if (CADET_CONNECTION_SENT == cc->state)
+ {
+ /* We didn't get the CADET_CONNECTION_CREATE_ACK, but instead got payload. That's fine,
+ clearly something is working, so pretend we got an ACK. */
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Faking connection CADET_CONNECTION_CREATE_ACK for %s due to KX\n",
+ GCC_2s (cc));
+ GCC_handle_connection_create_ack (cc);
+ }
+ GCT_handle_kx (cc->ct,
+ msg);
+}
+
+
+/**
+ * Handle KX_AUTH message.
+ *
+ * @param cc connection that received encrypted message
+ * @param msg the key exchange message
+ */
+void
+GCC_handle_kx_auth (struct CadetConnection *cc,
+ const struct GNUNET_CADET_TunnelKeyExchangeAuthMessage *msg)
+{
+ if (CADET_CONNECTION_SENT == cc->state)
+ {
+ /* We didn't get the CADET_CONNECTION_CREATE_ACK, but instead got payload. That's fine,
+ clearly something is working, so pretend we got an ACK. */
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Faking connection CADET_CONNECTION_CREATE_ACK for %s due to KX\n",
+ GCC_2s (cc));
+ GCC_handle_connection_create_ack (cc);
+ }
+ GCT_handle_kx_auth (cc->ct,
+ msg);
}
/**
- * Send a CREATE message to the first hop.
+ * Handle encrypted message.
+ *
+ * @param cc connection that received encrypted message
+ * @param msg the encrypted message to decrypt
+ */
+void
+GCC_handle_encrypted (struct CadetConnection *cc,
+ const struct GNUNET_CADET_TunnelEncryptedMessage *msg)
+{
+ if (CADET_CONNECTION_SENT == cc->state)
+ {
+ /* We didn't get the CREATE_ACK, but instead got payload. That's fine,
+ clearly something is working, so pretend we got an ACK. */
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Faking connection ACK for %s due to ENCRYPTED payload\n",
+ GCC_2s (cc));
+ GCC_handle_connection_create_ack (cc);
+ }
+ cc->metrics.last_use = GNUNET_TIME_absolute_get ();
+ GCT_handle_encrypted (cc->ct,
+ msg);
+}
+
+
+/**
+ * Send a #GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE message to the
+ * first hop.
*
* @param cls the `struct CadetConnection` to initiate
*/
send_create (void *cls)
{
struct CadetConnection *cc = cls;
- struct GNUNET_CADET_ConnectionCreate *create_msg;
+ struct GNUNET_CADET_ConnectionCreateMessage *create_msg;
struct GNUNET_PeerIdentity *pids;
struct GNUNET_MQ_Envelope *env;
unsigned int path_length;
cc->task = NULL;
- GNUNET_assert (NULL != cc->mq);
+ GNUNET_assert (GNUNET_YES == cc->mqm_ready);
path_length = GCPP_get_length (cc->path);
env = GNUNET_MQ_msg_extra (create_msg,
- path_length * sizeof (struct GNUNET_PeerIdentity),
+ (1 + path_length) * sizeof (struct GNUNET_PeerIdentity),
GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE);
+ create_msg->options = htonl ((uint32_t) cc->options);
create_msg->cid = cc->cid;
pids = (struct GNUNET_PeerIdentity *) &create_msg[1];
+ pids[0] = my_full_id;
for (unsigned int i=0;i<path_length;i++)
- pids[i] = *GCP_get_id (GCPP_get_peer_at_offset (cc->path,
- i));
+ pids[i + 1] = *GCP_get_id (GCPP_get_peer_at_offset (cc->path,
+ i));
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Sending CADET_CONNECTION_CREATE message for %s\n",
+ GCC_2s (cc));
+ cc->env = env;
+ update_state (cc,
+ CADET_CONNECTION_SENT,
+ GNUNET_NO);
+ GCP_send (cc->mq_man,
+ env);
+}
+
+
+/**
+ * Send a CREATE_ACK message towards the origin.
+ *
+ * @param cls the `struct CadetConnection` to initiate
+ */
+static void
+send_create_ack (void *cls)
+{
+ struct CadetConnection *cc = cls;
+ struct GNUNET_CADET_ConnectionCreateAckMessage *ack_msg;
+ struct GNUNET_MQ_Envelope *env;
+
+ cc->task = NULL;
+ GNUNET_assert (CADET_CONNECTION_CREATE_RECEIVED == cc->state);
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Sending CONNECTION_CREATE_ACK message for %s\n",
+ GCC_2s (cc));
+ GNUNET_assert (GNUNET_YES == cc->mqm_ready);
+ env = GNUNET_MQ_msg (ack_msg,
+ GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE_ACK);
+ ack_msg->cid = cc->cid;
cc->env = env;
- GNUNET_MQ_notify_sent (env,
- &transmit_create_done_cb,
- cc);
- GNUNET_MQ_send (cc->mq,
- env);
+ update_state (cc,
+ CADET_CONNECTION_READY,
+ GNUNET_NO);
+ GCP_send (cc->mq_man,
+ env);
+}
+
+
+/**
+ * We got a #GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE for a
+ * connection that we already have. Either our ACK got lost
+ * or something is fishy. Consider retransmitting the ACK.
+ *
+ * @param cc connection that got the duplicate CREATE
+ */
+void
+GCC_handle_duplicate_create (struct CadetConnection *cc)
+{
+ if (GNUNET_YES == cc->mqm_ready)
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Got duplicate CREATE for %s, scheduling another ACK (%s)\n",
+ GCC_2s (cc),
+ (GNUNET_YES == cc->mqm_ready) ? "MQM ready" : "MQM busy");
+ /* Revert back to the state of having only received the 'CREATE',
+ and immediately proceed to send the CREATE_ACK. */
+ update_state (cc,
+ CADET_CONNECTION_CREATE_RECEIVED,
+ cc->mqm_ready);
+ if (NULL != cc->task)
+ GNUNET_SCHEDULER_cancel (cc->task);
+ cc->task = GNUNET_SCHEDULER_add_now (&send_create_ack,
+ cc);
+ }
+ else
+ {
+ /* We are currently sending something else back, which
+ can only be an ACK or payload, either of which would
+ do. So actually no need to do anything. */
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Got duplicate CREATE for %s. MQ is busy, not queueing another ACK\n",
+ GCC_2s (cc));
+ }
}
* peer at the first hop. Adjust accordingly.
*
* @param cls the `struct CadetConnection`
- * @param mq NULL if the CORE connection was lost, non-NULL if
- * it became available
+ * @param available #GNUNET_YES if sending is now possible,
+ * #GNUNET_NO if sending is no longer possible
+ * #GNUNET_SYSERR if sending is no longer possible
+ * and the last envelope was discarded
*/
static void
manage_first_hop_mq (void *cls,
- struct GNUNET_MQ_Handle *mq)
+ int available)
{
struct CadetConnection *cc = cls;
- if (NULL == mq)
+ if (GNUNET_YES != available)
{
/* Connection is down, for now... */
- cc->mq = NULL;
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Core MQ for %s went down\n",
+ GCC_2s (cc));
+ update_state (cc,
+ CADET_CONNECTION_NEW,
+ GNUNET_NO);
+ cc->retry_delay = GNUNET_TIME_UNIT_ZERO;
if (NULL != cc->task)
{
GNUNET_SCHEDULER_cancel (cc->task);
return;
}
- cc->mq = mq;
- cc->state = CADET_CONNECTION_SENDING_CREATE;
-
- /* Now repeat sending connection creation messages
- down the path, until we get an ACK! */
- cc->task = GNUNET_SCHEDULER_add_now (&send_create,
- cc);
+ update_state (cc,
+ cc->state,
+ GNUNET_YES);
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Core MQ for %s became available in state %d\n",
+ GCC_2s (cc),
+ cc->state);
+ switch (cc->state)
+ {
+ case CADET_CONNECTION_NEW:
+ /* Transmit immediately */
+ cc->task = GNUNET_SCHEDULER_add_now (&send_create,
+ cc);
+ break;
+ case CADET_CONNECTION_SENDING_CREATE:
+ /* Should not be possible to be called in this state. */
+ GNUNET_assert (0);
+ break;
+ case CADET_CONNECTION_SENT:
+ /* Retry a bit later... */
+ cc->retry_delay = GNUNET_TIME_STD_BACKOFF (cc->retry_delay);
+ cc->task = GNUNET_SCHEDULER_add_delayed (cc->retry_delay,
+ &send_create,
+ cc);
+ break;
+ case CADET_CONNECTION_CREATE_RECEIVED:
+ /* We got the 'CREATE' (incoming connection), should send the CREATE_ACK */
+ cc->metrics.age = GNUNET_TIME_absolute_get ();
+ cc->task = GNUNET_SCHEDULER_add_now (&send_create_ack,
+ cc);
+ break;
+ case CADET_CONNECTION_READY:
+ if ( (NULL == cc->keepalive_qe) &&
+ (GNUNET_YES == cc->mqm_ready) &&
+ (NULL == cc->task) )
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Scheduling keepalive for %s in %s\n",
+ GCC_2s (cc),
+ GNUNET_STRINGS_relative_time_to_string (keepalive_period,
+ GNUNET_YES));
+ cc->task = GNUNET_SCHEDULER_add_delayed (keepalive_period,
+ &send_keepalive,
+ cc);
+ }
+ break;
+ }
}
/**
- * Create a connection to @a destination via @a path and
- * notify @a cb whenever we are ready for more data.
+ * Create a connection to @a destination via @a path and notify @a cb
+ * whenever we are ready for more data. Shared logic independent of
+ * who is initiating the connection.
*
* @param destination where to go
* @param path which path to take (may not be the full path)
- * @param ct tunnel that uses the connection
+ * @param off offset of @a destination on @a path
+ * @param options options for the connection
+ * @param ct which tunnel uses this connection
+ * @param init_state initial state for the connection
* @param ready_cb function to call when ready to transmit
* @param ready_cb_cls closure for @a cb
* @return handle to the connection
*/
-struct CadetConnection *
-GCC_create (struct CadetPeer *destination,
- struct CadetPeerPath *path,
- struct CadetTConnection *ct,
- GNUNET_SCHEDULER_TaskCallback ready_cb,
- void *ready_cb_cls)
+static struct CadetConnection *
+connection_create (struct CadetPeer *destination,
+ struct CadetPeerPath *path,
+ unsigned int off,
+ enum GNUNET_CADET_ChannelOption options,
+ struct CadetTConnection *ct,
+ const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid,
+ enum CadetConnectionState init_state,
+ GCC_ReadyCallback ready_cb,
+ void *ready_cb_cls)
{
struct CadetConnection *cc;
struct CadetPeer *first_hop;
- unsigned int off;
- off = GCPP_find_peer (path,
- destination);
- GNUNET_assert (UINT_MAX > off);
cc = GNUNET_new (struct CadetConnection);
+ cc->options = options;
+ cc->state = init_state;
cc->ct = ct;
- GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_NONCE,
- &cc->cid,
- sizeof (cc->cid));
+ cc->cid = *cid;
GNUNET_assert (GNUNET_OK ==
- GNUNET_CONTAINER_multihashmap_put (connections,
- GCC_get_h (cc),
- cc,
- GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
+ GNUNET_CONTAINER_multishortmap_put (connections,
+ &GCC_get_id (cc)->connection_of_tunnel,
+ cc,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
cc->ready_cb = ready_cb;
cc->ready_cb_cls = ready_cb_cls;
cc->path = path;
cc->off = off;
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Creating %s using path %s\n",
+ GCC_2s (cc),
+ GCPP_2s (path));
GCPP_add_connection (path,
off,
cc);
for (unsigned int i=0;i<off;i++)
GCP_add_connection (GCPP_get_peer_at_offset (path,
- off),
+ i),
cc);
first_hop = GCPP_get_peer_at_offset (path,
/**
- * We finished transmission of a message, if we are still ready, tell
- * the tunnel!
+ * Create a connection to @a destination via @a path and
+ * notify @a cb whenever we are ready for more data. This
+ * is an inbound tunnel, so we must use the existing @a cid
*
- * @param cls our `struct CadetConnection`
+ * @param destination where to go
+ * @param path which path to take (may not be the full path)
+ * @param options options for the connection
+ * @param ct which tunnel uses this connection
+ * @param ready_cb function to call when ready to transmit
+ * @param ready_cb_cls closure for @a cb
+ * @return handle to the connection, NULL if we already have
+ * a connection that takes precedence on @a path
*/
-static void
-transmit_done_cb (void *cls)
+struct CadetConnection *
+GCC_create_inbound (struct CadetPeer *destination,
+ struct CadetPeerPath *path,
+ enum GNUNET_CADET_ChannelOption options,
+ struct CadetTConnection *ct,
+ const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid,
+ GCC_ReadyCallback ready_cb,
+ void *ready_cb_cls)
{
- struct CadetConnection *cc = cls;
+ struct CadetConnection *cc;
+ unsigned int off;
- cc->env = NULL;
- if ( (NULL != cc->mq) &&
- (CADET_CONNECTION_READY == cc->state) )
- cc->ready_cb (cc->ready_cb_cls);
+ off = GCPP_find_peer (path,
+ destination);
+ GNUNET_assert (UINT_MAX != off);
+ cc = GCPP_get_connection (path,
+ destination,
+ off);
+ if (NULL != cc)
+ {
+ int cmp;
+
+ cmp = memcmp (cid,
+ &cc->cid,
+ sizeof (*cid));
+ if (0 == cmp)
+ {
+ /* Two peers picked the SAME random connection identifier at the
+ same time for the same path? Must be malicious. Drop
+ connection (existing and inbound), even if it is the only
+ one. */
+ GNUNET_break_op (0);
+ GCT_connection_lost (cc->ct);
+ GCC_destroy_without_tunnel (cc);
+ return NULL;
+ }
+ if (0 < cmp)
+ {
+ /* drop existing */
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Got two connections on %s, dropping my existing %s\n",
+ GCPP_2s (path),
+ GCC_2s (cc));
+ GCT_connection_lost (cc->ct);
+ GCC_destroy_without_tunnel (cc);
+ }
+ else
+ {
+ /* keep existing */
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Got two connections on %s, keeping my existing %s\n",
+ GCPP_2s (path),
+ GCC_2s (cc));
+ return NULL;
+ }
+ }
+
+ return connection_create (destination,
+ path,
+ off,
+ options,
+ ct,
+ cid,
+ CADET_CONNECTION_CREATE_RECEIVED,
+ ready_cb,
+ ready_cb_cls);
+}
+
+
+/**
+ * Create a connection to @a destination via @a path and
+ * notify @a cb whenever we are ready for more data.
+ *
+ * @param destination where to go
+ * @param path which path to take (may not be the full path)
+ * @param off offset of @a destination on @a path
+ * @param options options for the connection
+ * @param ct tunnel that uses the connection
+ * @param ready_cb function to call when ready to transmit
+ * @param ready_cb_cls closure for @a cb
+ * @return handle to the connection
+ */
+struct CadetConnection *
+GCC_create (struct CadetPeer *destination,
+ struct CadetPeerPath *path,
+ unsigned int off,
+ enum GNUNET_CADET_ChannelOption options,
+ struct CadetTConnection *ct,
+ GCC_ReadyCallback ready_cb,
+ void *ready_cb_cls)
+{
+ struct GNUNET_CADET_ConnectionTunnelIdentifier cid;
+
+ GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_NONCE,
+ &cid,
+ sizeof (cid));
+ return connection_create (destination,
+ path,
+ off,
+ options,
+ ct,
+ &cid,
+ CADET_CONNECTION_NEW,
+ ready_cb,
+ ready_cb_cls);
}
* connection is right now ready for transmission.
*
* @param cc connection identification
- * @param env envelope with message to transmit
+ * @param env envelope with message to transmit; must NOT
+ * yet have a #GNUNET_MQ_notify_sent() callback attached to it
*/
void
GCC_transmit (struct CadetConnection *cc,
struct GNUNET_MQ_Envelope *env)
{
- GNUNET_assert (NULL == cc->env);
- cc->env = env;
- GNUNET_MQ_notify_sent (env,
- &transmit_done_cb,
- cc);
- if ( (NULL != cc->mq) &&
- (CADET_CONNECTION_READY == cc->state) )
- GNUNET_MQ_send (cc->mq,
- env);
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Scheduling message for transmission on %s\n",
+ GCC_2s (cc));
+ GNUNET_assert (GNUNET_YES == cc->mqm_ready);
+ GNUNET_assert (CADET_CONNECTION_READY == cc->state);
+ cc->metrics.last_use = GNUNET_TIME_absolute_get ();
+ cc->mqm_ready = GNUNET_NO;
+ if (NULL != cc->task)
+ {
+ GNUNET_SCHEDULER_cancel (cc->task);
+ cc->task = NULL;
+ }
+ GCP_send (cc->mq_man,
+ env);
}
}
+/**
+ * Get a (static) string for a connection.
+ *
+ * @param cc Connection.
+ */
+const char *
+GCC_2s (const struct CadetConnection *cc)
+{
+ static char buf[128];
+
+ if (NULL == cc)
+ return "Connection(NULL)";
+
+ if (NULL != cc->ct)
+ {
+ GNUNET_snprintf (buf,
+ sizeof (buf),
+ "Connection %s (%s)",
+ GNUNET_sh2s (&cc->cid.connection_of_tunnel),
+ GCT_2s (cc->ct->t));
+ return buf;
+ }
+ GNUNET_snprintf (buf,
+ sizeof (buf),
+ "Connection %s",
+ GNUNET_sh2s (&cc->cid.connection_of_tunnel));
+ return buf;
+}
+
+
+#define LOG2(level, ...) GNUNET_log_from_nocheck(level,"cadet-con",__VA_ARGS__)
+
+
/**
* Log connection info.
*
GCC_debug (struct CadetConnection *cc,
enum GNUNET_ErrorType level)
{
- GNUNET_break (0); // FIXME: implement...
+ int do_log;
+
+ do_log = GNUNET_get_log_call_status (level & (~GNUNET_ERROR_TYPE_BULK),
+ "cadet-con",
+ __FILE__, __FUNCTION__, __LINE__);
+ if (0 == do_log)
+ return;
+ if (NULL == cc)
+ {
+ LOG2 (level,
+ "Connection (NULL)\n");
+ return;
+ }
+ LOG2 (level,
+ "%s to %s via path %s in state %d is %s\n",
+ GCC_2s (cc),
+ GCP_2s (cc->destination),
+ GCPP_2s (cc->path),
+ cc->state,
+ (GNUNET_YES == cc->mqm_ready) ? "ready" : "busy");
}
/* end of gnunet-service-cadet-new_connection.c */
projects / oweals / gnunet.git / blobdiff