6 files changed, 251 insertions, 123 deletions

diff --git a/src/transport/gnunet-service-transport.c b/src/transport/gnunet-service-transport.c
index 5bde153e4..c571277e2 100644
--- a/src/transport/gnunet-service-transport.c
+++ b/src/transport/gnunet-service-transport.c
@@ -136,6 +136,71 @@ try_connect_if_allowed (void *cls, const struct GNUNET_PeerIdentity *peer,
 
 
 /**
+ * We received some payload.  Prepare to pass it on to our clients. 
+ *
+ * @param peer (claimed) identity of the other peer
+ * @param message the message, NULL if we only care about
+ *                learning about the delay until we should receive again -- FIXME!
+ * @param ats performance information
+ * @param ats_count number of records in ats
+ * @return how long the plugin should wait until receiving more data
+ */
+static struct GNUNET_TIME_Relative
+process_payload (const struct GNUNET_PeerIdentity *peer,
+                 const struct GNUNET_MessageHeader *message,
+                 const struct GNUNET_ATS_Information *ats,
+                 uint32_t ats_count)
+{
+  struct GNUNET_TIME_Relative ret;
+  int do_forward;
+  struct InboundMessage *im;
+  size_t size = sizeof (struct InboundMessage) + ntohs (message->size);
+  char buf[size];
+  
+  ret = GNUNET_TIME_UNIT_ZERO;
+  do_forward = GNUNET_SYSERR;
+  ret =
+    GST_neighbours_calculate_receive_delay (peer,
+                                            (message ==
+                                             NULL) ? 0 :
+                                            ntohs (message->size),
+                                            &do_forward);
+  im = (struct InboundMessage*) buf;    
+  im->header.size = htons (size);
+  im->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_RECV);
+  im->ats_count = htonl (0);
+  memcpy (&(im->peer), peer, sizeof (struct GNUNET_PeerIdentity));
+  memcpy (&im[1], message, ntohs (message->size));
+
+  switch (do_forward)
+  {
+  case GNUNET_YES:
+    GST_clients_broadcast (&im->header, GNUNET_YES);      
+    break;
+  case GNUNET_NO:
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                _("Discarded %u bytes of type %u from %s: quota violated!\n"),
+                ntohs (message->size),
+                ntohs (message->type),
+                GNUNET_i2s (peer));
+    break;
+  case GNUNET_SYSERR:
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                _("Discarded %u bytes of type %u from %s: connection is down!\n"),
+                ntohs (message->size),
+                ntohs (message->type),
+                GNUNET_i2s (peer));
+    /* FIXME: store until connection is up? This is virtually always a SETKEY and a PING... */
+    break;
+  default:
+    GNUNET_break (0);
+    break;
+  }    
+  return ret;
+}
+
+
+/**
  * Function called by the transport for each received message.
  * This function should also be called with "NULL" for the
  * message to signal that the other peer disconnected.
@@ -167,104 +232,74 @@ plugin_env_receive_callback (void *cls, const struct GNUNET_PeerIdentity *peer,
                              uint16_t sender_address_len)
 {
   const char *plugin_name = cls;
-  int do_forward;
   struct GNUNET_TIME_Relative ret;
   uint16_t type;
   
   ret = GNUNET_TIME_UNIT_ZERO;
-  if (NULL != message)
+  if (NULL == message)
+    goto end;
+  type = ntohs (message->type);
+  switch (type)
   {
-    type = ntohs (message->type);
-    switch (type)
-    {
-    case GNUNET_MESSAGE_TYPE_HELLO:
-      GST_validation_handle_hello (message);
-      return ret;
-    case GNUNET_MESSAGE_TYPE_TRANSPORT_PING:
+  case GNUNET_MESSAGE_TYPE_HELLO:
+    GST_validation_handle_hello (message);
+    return ret;
+  case GNUNET_MESSAGE_TYPE_TRANSPORT_PING:
 #if DEBUG_TRANSPORT
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
-                  "Processing `%s' from `%s'\n", "PING",
-                  (sender_address != NULL) ? GST_plugins_a2s (plugin_name,
-                                                              sender_address,
-                                                              sender_address_len)
-                  : "<inbound>");
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
+                "Processing `%s' from `%s'\n", "PING",
+                (sender_address != NULL) ? GST_plugins_a2s (plugin_name,
+                                                            sender_address,
+                                                            sender_address_len)
+                : "<inbound>");
 #endif
-      GST_validation_handle_ping (peer, message, plugin_name, session,
-                                  sender_address, sender_address_len);
-      break;
-    case GNUNET_MESSAGE_TYPE_TRANSPORT_PONG:
+    GST_validation_handle_ping (peer, message, plugin_name, session,
+                                sender_address, sender_address_len);
+    break;
+  case GNUNET_MESSAGE_TYPE_TRANSPORT_PONG:
 #if DEBUG_TRANSPORT
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
-                  "Processing `%s' from `%s'\n", "PONG",
-                  (sender_address != NULL) ? GST_plugins_a2s (plugin_name,
-                                                              sender_address,
-                                                              sender_address_len)
-                  : "<inbound>");
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
+                "Processing `%s' from `%s'\n", "PONG",
+                (sender_address != NULL) ? GST_plugins_a2s (plugin_name,
+                                                            sender_address,
+                                                            sender_address_len)
+                : "<inbound>");
 #endif
-      GST_validation_handle_pong (peer, message);
-      break;
-    case GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_CONNECT:
-      (void) GST_blacklist_test_allowed (peer, NULL, &try_connect_if_allowed,
-                                         NULL);
-      /* TODO: if 'session != NULL', and timestamp more recent than the
-       * previous one, maybe notify ATS that this is now the preferred
-       * * way to communicate with this peer (other peer switched transport) */
-      break;
-    case GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_DISCONNECT:
-      /* FIXME: do some validation to prevent an attacker from sending
-       * a fake disconnect message... */          
-      GST_neighbours_force_disconnect (peer);
-      break;
-    case GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_KEEPALIVE:
-      GST_neighbours_keepalive (peer);
-      break;
-    default:
-      /* should be payload */
-      do_forward = GNUNET_SYSERR;
-      ret =
-          GST_neighbours_calculate_receive_delay (peer,
-                                                  (message ==
-                                                   NULL) ? 0 :
-                                                  ntohs (message->size),
-                                                  &do_forward);
-      if (do_forward == GNUNET_YES)
-      {
-        struct InboundMessage *im;
-        size_t size = sizeof (struct InboundMessage) + ntohs (message->size);
-
-        im = GNUNET_malloc (size);
-        im->header.size = htons (size);
-        im->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_RECV);
-        im->ats_count = htonl (0);
-        memcpy (&(im->peer), peer, sizeof (struct GNUNET_PeerIdentity));
-        memcpy (&im[1], message, ntohs (message->size));
-        GST_clients_broadcast ((const struct GNUNET_MessageHeader *) im,
-                               GNUNET_YES);
-
-        GNUNET_free (im);
-      }
-      else      
-        GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-                    _("Discarded %u bytes of type %u from %s via plugin %s: connection is down!\n"),
-                    ntohs (message->size),
-                    type,
-                    GNUNET_i2s (peer),
-                    plugin_name);
-      break;
-    }
+    GST_validation_handle_pong (peer, message);
+    break;
+  case GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_CONNECT:
+    GST_neighbours_handle_connect (message,
+                                   peer,
+                                   plugin_name, sender_address, sender_address_len,
+                                   session, ats, ats_count);
+    (void) GST_blacklist_test_allowed (peer, NULL, &try_connect_if_allowed,
+                                       NULL);
+    break;
+  case GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_DISCONNECT:
+    /* FIXME: do some validation to prevent an attacker from sending
+     * a fake disconnect message... */            
+    GST_neighbours_force_disconnect (peer);
+    break;
+  case GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_KEEPALIVE:
+    GST_neighbours_keepalive (peer);
+    break;
+  default:
+    /* should be payload */
+    process_payload (peer,
+                     message,
+                     ats, ats_count);
+    break;
   }
-  /*
-     FIXME: this gives an address that might not have been validated to
-     ATS for 'selection', which is probably not what we want; this 
-     might be particularly wrong (as in, possibly hiding bugs with address
-     validation) as 'GNUNET_ATS_address_update' currently ignores
-     the expiration given.
-  */
+ end:
+#if 1
+  /* FIXME: this should not be needed, and not sure it's good to have it, but without
+     this connections seem to go extra-slow */
   if ((ats_count > 0) && (ats != NULL))
     GNUNET_ATS_address_update (GST_ats, peer,
                                plugin_name, sender_address, sender_address_len,
                                session,
                                ats, ats_count);
+#endif
   return ret;
 }
 
diff --git a/src/transport/gnunet-service-transport_neighbours.c b/src/transport/gnunet-service-transport_neighbours.c
index 9676366a6..db2dce5e7 100644
--- a/src/transport/gnunet-service-transport_neighbours.c
+++ b/src/transport/gnunet-service-transport_neighbours.c
@@ -248,6 +248,11 @@ struct NeighbourMapEntry
   struct GNUNET_BANDWIDTH_Tracker in_tracker;
 
   /**
+   * Timestamp of the 'SESSION_CONNECT' message we got from the other peer
+   */
+  struct GNUNET_TIME_Absolute connect_ts;
+
+  /**
    * How often has the other peer (recently) violated the inbound
    * traffic limit?  Incremented by 10 per violation, decremented by 1
    * per non-violation (for each time interval).
@@ -642,16 +647,18 @@ GST_neighbours_switch_to_address (const struct GNUNET_PeerIdentity *peer,
   n = lookup_neighbour (peer);
   if (NULL == n)
   {
-    /* FIXME: ATS not fully implemented, once ATS only generates
-       these events for 'connected' addresses, things should be better... */
-    // GNUNET_break (0);
+    GNUNET_ATS_address_destroyed (GST_ats,
+                                  peer,
+                                  plugin_name, address,
+                                  address_len, session);
     return;
   }
   was_connected = n->is_connected;
   n->is_connected = GNUNET_YES;
   if (GNUNET_YES != was_connected)
-    n->keepalive_task = GNUNET_SCHEDULER_add_now (&neighbour_keepalive_task,
-                                                  n);
+    n->keepalive_task = GNUNET_SCHEDULER_add_delayed (KEEPALIVE_FREQUENCY,
+                                                      &neighbour_keepalive_task,
+                                                      n);
 
 #if DEBUG_TRANSPORT
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -662,7 +669,6 @@ GST_neighbours_switch_to_address (const struct GNUNET_PeerIdentity *peer,
                                                                   address_len),
               session);
 #endif
-
   GNUNET_free_non_null (n->addr);
   n->addr = GNUNET_malloc (address_len);
   memcpy (n->addr, address, address_len);
@@ -682,7 +688,6 @@ GST_neighbours_switch_to_address (const struct GNUNET_PeerIdentity *peer,
       GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get ());
   GST_neighbours_send (peer, &connect_msg, sizeof (connect_msg),
                        GNUNET_TIME_UNIT_FOREVER_REL, NULL, NULL);
-
   if (GNUNET_YES == was_connected)
     return;
   /* First tell clients about connected neighbours...*/
@@ -692,6 +697,39 @@ GST_neighbours_switch_to_address (const struct GNUNET_PeerIdentity *peer,
   connect_notify_cb (callback_cls, peer, ats, ats_count);
 }
 
+
+/**
+ * Create an entry in the neighbour map for the given peer
+ * 
+ * @param peer peer to create an entry for
+ * @return new neighbour map entry
+ */
+static struct NeighbourMapEntry *
+setup_neighbour (const struct GNUNET_PeerIdentity *peer)
+{
+  struct NeighbourMapEntry *n;
+
+#if DEBUG_TRANSPORT
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Unknown peer `%s', creating new neighbour\n",
+              GNUNET_i2s (peer));
+#endif
+  n = GNUNET_malloc (sizeof (struct NeighbourMapEntry));
+  n->id = *peer;
+  GNUNET_BANDWIDTH_tracker_init (&n->in_tracker,
+                                 GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT,
+                                 MAX_BANDWIDTH_CARRY_S);
+  n->timeout_task =
+    GNUNET_SCHEDULER_add_delayed (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
+                                  &neighbour_timeout_task, n);
+  GNUNET_assert (GNUNET_OK ==
+                 GNUNET_CONTAINER_multihashmap_put (neighbours,
+                                                    &n->id.hashPubKey, n,
+                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
+  return n;
+}
+
+
 /**
  * Try to create a connection to the given target (eventually).
  *
@@ -703,12 +741,10 @@ GST_neighbours_try_connect (const struct GNUNET_PeerIdentity *target)
   struct NeighbourMapEntry *n;
 
   GNUNET_assert (neighbours != NULL);
-
 #if DEBUG_TRANSPORT
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Trying to connect to peer `%s'\n",
               GNUNET_i2s (target));
 #endif
-
   GNUNET_assert (0 !=
                  memcmp (target, &GST_my_identity,
                          sizeof (struct GNUNET_PeerIdentity)));
@@ -716,31 +752,13 @@ GST_neighbours_try_connect (const struct GNUNET_PeerIdentity *target)
   if ((NULL != n) && (GNUNET_YES == n->is_connected))
     return;                     /* already connected */
   if (n == NULL)
-  {
-#if DEBUG_TRANSPORT
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                "Unknown peer `%s', creating new neighbour\n",
-                GNUNET_i2s (target));
-#endif
-    n = GNUNET_malloc (sizeof (struct NeighbourMapEntry));
-    n->id = *target;
-    GNUNET_BANDWIDTH_tracker_init (&n->in_tracker,
-                                   GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT,
-                                   MAX_BANDWIDTH_CARRY_S);
-    n->timeout_task =
-        GNUNET_SCHEDULER_add_delayed (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
-                                      &neighbour_timeout_task, n);
-    GNUNET_assert (GNUNET_OK ==
-                   GNUNET_CONTAINER_multihashmap_put (neighbours,
-                                                      &n->id.hashPubKey, n,
-                                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
-  }
+    n = setup_neighbour (target);
 #if DEBUG_TRANSPORT
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "Asking ATS for suggested address to connect to peer `%s'\n",
-              GNUNET_i2s (target));
+              GNUNET_i2s (&n->id));
 #endif
-   GNUNET_ATS_suggest_address (GST_ats, target);
+   GNUNET_ATS_suggest_address (GST_ats, &n->id);
 }
 
 
@@ -783,7 +801,10 @@ GST_neighbours_session_terminated (const struct GNUNET_PeerIdentity *peer,
               "Session %X to peer `%s' ended \n",
               session, GNUNET_i2s (peer));
 #endif
-
+  GNUNET_ATS_address_destroyed (GST_ats,
+                                peer,
+                                NULL, NULL, 0,
+                                session);
   n = lookup_neighbour (peer);
   if (NULL == n)
     return;
@@ -894,7 +915,8 @@ GST_neighbours_send (const struct GNUNET_PeerIdentity *target, const void *msg,
  * @param sender sender of the message
  * @param size size of the message
  * @param do_forward set to GNUNET_YES if the message should be forwarded to clients
- *                   GNUNET_NO if the neighbour is not connected or violates the quota
+ *                   GNUNET_NO if the neighbour is not connected or violates the quota,
+ *                   GNUNET_SYSERR if the connection is not fully up yet
  * @return how long to wait before reading more from this sender
  */
 struct GNUNET_TIME_Relative
@@ -917,7 +939,7 @@ GST_neighbours_calculate_receive_delay (const struct GNUNET_PeerIdentity
     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                 _("Plugin gave us %d bytes of data but somehow the session is not marked as UP yet!\n"),
                 (int) size);
-    *do_forward = GNUNET_NO;
+    *do_forward = GNUNET_SYSERR;
     return GNUNET_TIME_UNIT_ZERO;
   }
   if (GNUNET_YES == GNUNET_BANDWIDTH_tracker_consume (&n->in_tracker, size))
@@ -1141,4 +1163,53 @@ GST_neighbours_force_disconnect (const struct GNUNET_PeerIdentity *target)
 }
 
 
+/**
+ * We received a 'SESSION_CONNECT' message from the other peer.
+ * Consider switching to it.
+ *
+ * @param message possibly a 'struct SessionConnectMessage' (check format)
+ * @param peer identity of the peer to switch the address for
+ * @param plugin_name name of transport that delivered the PONG
+ * @param address address of the other peer, NULL if other peer
+ *                       connected to us
+ * @param address_len number of bytes in address
+ * @param session session to use (or NULL)
+ * @param ats performance data
+ * @param ats_count number of entries in ats (excluding 0-termination)
+  */
+void
+GST_neighbours_handle_connect (const struct GNUNET_MessageHeader *message,
+                               const struct GNUNET_PeerIdentity *peer,
+                               const char *plugin_name,
+                               const char *sender_address, uint16_t sender_address_len,
+                               struct Session *session,
+                               const struct GNUNET_ATS_Information *ats,
+                               uint32_t ats_count)
+{
+  const struct SessionConnectMessage *scm;
+  struct GNUNET_TIME_Absolute ts;
+  struct NeighbourMapEntry *n;
+
+  if (ntohs (message->size) != sizeof (struct SessionConnectMessage))
+  {
+    GNUNET_break_op (0);
+    return;
+  }
+  scm = (const struct SessionConnectMessage *) message;
+  GNUNET_break_op (ntohl (scm->reserved) == 0);
+  ts = GNUNET_TIME_absolute_ntoh (scm->timestamp);
+  n = lookup_neighbour (peer);
+  if (NULL == n) 
+    n = setup_neighbour (peer);
+  if (ts.abs_value > n->connect_ts.abs_value)
+  {
+    GNUNET_ATS_address_update (GST_ats,
+                               peer,
+                               plugin_name, sender_address, sender_address_len,
+                               session, ats, ats_count);
+    n->connect_ts = ts;
+  }
+}
+
+
 /* end of file gnunet-service-transport_neighbours.c */
diff --git a/src/transport/gnunet-service-transport_neighbours.h b/src/transport/gnunet-service-transport_neighbours.h
index 5b4929c2c..2f6962a67 100644
--- a/src/transport/gnunet-service-transport_neighbours.h
+++ b/src/transport/gnunet-service-transport_neighbours.h
@@ -209,5 +209,29 @@ GST_neighbours_switch_to_address (const struct GNUNET_PeerIdentity *peer,
                                   *ats, uint32_t ats_count);
 
 
+/**
+ * We received a 'SESSION_CONNECT' message from the other peer.
+ * Consider switching to it.
+ *
+ * @param message possibly a 'struct SessionConnectMessage' (check format)
+ * @param peer identity of the peer to switch the address for
+ * @param plugin_name name of transport that delivered the PONG
+ * @param address address of the other peer, NULL if other peer
+ *                       connected to us
+ * @param address_len number of bytes in address
+ * @param session session to use (or NULL)
+ * @param ats performance data
+ * @param ats_count number of entries in ats (excluding 0-termination)
+  */
+void
+GST_neighbours_handle_connect (const struct GNUNET_MessageHeader *message,
+                               const struct GNUNET_PeerIdentity *peer,
+                               const char *plugin_name,
+                               const char *sender_address, uint16_t sender_address_len,
+                               struct Session *session,
+                               const struct GNUNET_ATS_Information *ats,
+                               uint32_t ats_count);
+
+
 #endif
 /* end of file gnunet-service-transport_neighbours.h */
diff --git a/src/transport/plugin_transport_tcp.c b/src/transport/plugin_transport_tcp.c
index d3391e1b2..546c2aa36 100644
--- a/src/transport/plugin_transport_tcp.c
+++ b/src/transport/plugin_transport_tcp.c
@@ -696,6 +696,10 @@ do_transmit (void *cls, size_t size, void *buf)
     GNUNET_CONTAINER_DLL_remove (session->pending_messages_head,
                                  session->pending_messages_tail, pos);
     GNUNET_assert (size >= pos->message_size);
+    GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
+                     "tcp",
+                     "Transmitting message of type %u\n",
+                     ntohs (((struct GNUNET_MessageHeader*)pos->msg)->type));
     /* FIXME: this memcpy can be up to 7% of our total runtime */
     memcpy (cbuf, pos->msg, pos->message_size);
     cbuf += pos->message_size;
@@ -1706,7 +1710,7 @@ handle_tcp_data (void *cls, struct GNUNET_SERVER_Client *client,
     return;
   }
   session->last_activity = GNUNET_TIME_absolute_get ();
-#if DEBUG_TCP > 1
+#if DEBUG_TCP
   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "tcp",
                    "Passing %u bytes of type %u from `%4s' to transport service.\n",
                    (unsigned int) ntohs (message->size),
diff --git a/src/transport/transport.h b/src/transport/transport.h
index 440b0dfed..b4ddd0a7f 100644
--- a/src/transport/transport.h
+++ b/src/transport/transport.h
@@ -205,11 +205,6 @@ struct InboundMessage
   struct GNUNET_MessageHeader header;
 
   /**
-   * Always zero.
-   */
-  uint32_t reserved GNUNET_PACKED;
-
-  /**
    * Number of ATS key-value pairs that follow this struct
    * (excluding the 0-terminator).
    */
diff --git a/src/transport/transport_api.c b/src/transport/transport_api.c
index a258172d5..41ef306b7 100644
--- a/src/transport/transport_api.c
+++ b/src/transport/transport_api.c
@@ -543,7 +543,6 @@ demultiplexer (void *cls, const struct GNUNET_MessageHeader *msg)
       break;
     }
     im = (const struct InboundMessage *) msg;
-    GNUNET_break (0 == ntohl (im->reserved));
     ats_count = ntohl (im->ats_count);
     imm = (const struct GNUNET_MessageHeader *) &((&(im->ats))[ats_count + 1]);