adding support for outbound message monitoring

3 files changed, 71 insertions, 24 deletions

diff --git a/TODO b/TODO
index 492d5b5ff..f5d5ac56d 100644
--- a/TODO
+++ b/TODO
@@ -17,7 +17,6 @@ Urgent items (before announcing ng.gnunet.org):
   - main service not implemented [Nate]
   - testcases crash & burn (no surprise)
 * CORE:
-  - outbound message monitoring not supported [Christian]
   - test currently fails spectacularly [segv of transport service]
     => need transport to work first!
   - request connect not working [Christian, need transport first]
diff --git a/src/core/gnunet-service-core.c b/src/core/gnunet-service-core.c
index cb74fc8c2..a0e1fb1c7 100644
--- a/src/core/gnunet-service-core.c
+++ b/src/core/gnunet-service-core.c
@@ -23,11 +23,6 @@
  * @brief high-level P2P messaging
  * @author Christian Grothoff
  *
- * TODO:
- * - not all GNUNET_CORE_OPTION_SEND_* flags are fully supported yet
- *   (i.e. no SEND_XXX_OUTBOUND).
- * - 'REQUEST_DISCONNECT' is not implemented (transport API is lacking!)
- *
  * Considerations for later:
  * - check that hostkey used by transport (for HELLOs) is the
  *   same as the hostkey that we are using!
@@ -54,6 +49,13 @@
 #define MAX_WINDOW_TIME (5 * 60 * 1000)
 
 /**
+ * How many messages do we queue up at most for optional
+ * notifications to a client?  (this can cause notifications
+ * about outgoing messages to be dropped).
+ */
+#define MAX_NOTIFY_QUEUE 16
+
+/**
  * Minimum of bytes per minute (out) to assign to any connected peer.
  * Should be rather low; values larger than DEFAULT_BPM_IN_OUT make no
  * sense.
@@ -1350,11 +1352,13 @@ batch_message (struct Neighbour *n,
                struct GNUNET_TIME_Relative *retry_time,
                unsigned int *priority)
 {
+  char ntmb[GNUNET_SERVER_MAX_MESSAGE_SIZE];
+  struct NotifyTrafficMessage *ntm = (struct NotifyTrafficMessage*) ntmb;
   struct MessageEntry *pos;
   struct MessageEntry *prev;
   struct MessageEntry *next;
   size_t ret;
-
+  
   ret = 0;
   *priority = 0;
   *deadline = GNUNET_TIME_UNIT_FOREVER_ABS;
@@ -1366,6 +1370,11 @@ batch_message (struct Neighbour *n,
                   retry_time->value);
       return 0;
     }
+  ntm->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_NOTIFY_OUTBOUND);
+  ntm->distance = htonl (n->last_distance);
+  ntm->latency = GNUNET_TIME_relative_hton (n->last_latency);
+  ntm->peer = n->peer;
+  
   pos = n->messages;
   prev = NULL;
   while ((pos != NULL) && (size >= sizeof (struct GNUNET_MessageHeader)))
@@ -1374,6 +1383,33 @@ batch_message (struct Neighbour *n,
       if (GNUNET_YES == pos->do_transmit)
         {
           GNUNET_assert (pos->size <= size);
+          /* do notifications */
+          /* FIXME: track if we have *any* client that wants
+             full notifications and only do this if that is
+             actually true */
+          if (pos->size < GNUNET_SERVER_MAX_MESSAGE_SIZE - sizeof (struct NotifyTrafficMessage))
+            {
+              memcpy (&ntm[1], &pos[1], pos->size);
+              ntm->header.size = htons (sizeof (struct NotifyTrafficMessage) + 
+                                        sizeof (struct GNUNET_MessageHeader));
+              send_to_all_clients (&ntm->header,
+                                   GNUNET_YES,
+                                   GNUNET_CORE_OPTION_SEND_HDR_OUTBOUND);
+            }
+          else
+            {
+              /* message too large for 'full' notifications, we do at
+                 least the 'hdr' type */
+              memcpy (&ntm[1],
+                      &pos[1],
+                      sizeof (struct GNUNET_MessageHeader));
+            }
+          ntm->header.size = htons (sizeof (struct NotifyTrafficMessage) + 
+                                    pos->size);
+          send_to_all_clients (&ntm->header,
+                               GNUNET_YES,
+                               GNUNET_CORE_OPTION_SEND_FULL_OUTBOUND);   
+          /* copy for encrypted transmission */
           memcpy (&buf[ret], &pos[1], pos->size);
           ret += pos->size;
           size -= pos->size;
@@ -3096,17 +3132,6 @@ run (void *cls,
                                                "CORE",
                                                "TOTAL_QUOTA_OUT",
                                                &bandwidth_target_out)) ||
-#if 0
-       (GNUNET_OK !=
-        GNUNET_CONFIGURATION_get_value_number (c,
-                                               "CORE",
-                                               "YY",
-                                               &qout)) ||
-       (GNUNET_OK !=
-        GNUNET_CONFIGURATION_get_value_number (c,
-                                               "CORE",
-                                               "ZZ_LIMIT", &tneigh)) ||
-#endif
        (GNUNET_OK !=
         GNUNET_CONFIGURATION_get_value_filename (c,
                                                  "GNUNETD",
@@ -3132,7 +3157,8 @@ run (void *cls,
                       sizeof (my_public_key), &my_identity.hashPubKey);
   /* setup notification */
   server = serv;
-  notifier = GNUNET_SERVER_notification_context_create (server, 0);
+  notifier = GNUNET_SERVER_notification_context_create (server, 
+                                                        MAX_NOTIFY_QUEUE);
   GNUNET_SERVER_disconnect_notify (server, &handle_client_disconnect, NULL);
   /* setup transport connection */
   transport = GNUNET_TRANSPORT_connect (sched,
diff --git a/src/include/gnunet_core_service.h b/src/include/gnunet_core_service.h
index d35e6fb85..af431ba56 100644
--- a/src/include/gnunet_core_service.h
+++ b/src/include/gnunet_core_service.h
@@ -148,8 +148,15 @@ typedef void
 
 
 /**
- * Connect to the core service.  Note that the connection may
- * complete (or fail) asynchronously.
+ * Connect to the core service.  Note that the connection may complete
+ * (or fail) asynchronously.  This function primarily causes the given
+ * callback notification functions to be invoked whenever the
+ * specified event happens.  The maximum number of queued
+ * notifications (queue length) is per client but the queue is shared
+ * across all types of notifications.  So a slow client that registers
+ * for 'outbound_notify' also risks missing 'inbound_notify' messages.
+ * Certain events (such as connect/disconnect notifications) are not
+ * subject to queue size limitations.
  *
  * @param sched scheduler to use
  * @param cfg configuration to use
@@ -161,14 +168,29 @@ typedef void
  * @param connects function to call on peer connect, can be NULL
  * @param disconnects function to call on peer disconnect / timeout, can be NULL
  * @param inbound_notify function to call for all inbound messages, can be NULL
+ *                note that the core is allowed to drop notifications about inbound
+ *                messages if the client does not process them fast enough (for this
+ *                notification type, a bounded queue is used)
  * @param inbound_hdr_only set to GNUNET_YES if inbound_notify will only read the
  *                GNUNET_MessageHeader and hence we do not need to give it the full message;
- *                can be used to improve efficiency, ignored if inbound_notify is NULLL
- * @param outbound_notify function to call for all outbound messages, can be NULL
+ *                can be used to improve efficiency, ignored if inbound_notify is NULL
+ *                note that the core is allowed to drop notifications about inbound
+ *                messages if the client does not process them fast enough (for this
+ *                notification type, a bounded queue is used) 
+ * @param outbound_notify function to call for all outbound messages, can be NULL;
+ *                note that the core is allowed to drop notifications about outbound
+ *                messages if the client does not process them fast enough (for this
+ *                notification type, a bounded queue is used)
  * @param outbound_hdr_only set to GNUNET_YES if outbound_notify will only read the
  *                GNUNET_MessageHeader and hence we do not need to give it the full message
- *                can be used to improve efficiency, ignored if outbound_notify is NULLL
+ *                can be used to improve efficiency, ignored if outbound_notify is NULL
+ *                note that the core is allowed to drop notifications about outbound
+ *                messages if the client does not process them fast enough (for this
+ *                notification type, a bounded queue is used)
  * @param handlers callbacks for messages we care about, NULL-terminated
+ *                note that the core is allowed to drop notifications about inbound
+ *                messages if the client does not process them fast enough (for this
+ *                notification type, a bounded queue is used)
  * @return handle to the core service (only useful for disconnect until 'init' is called),
  *           NULL on error (in this case, init is never called)
  */