3b0f0783ad61fc8bfa3a1e9fa00e7fec0660ed0c
[oweals/gnunet.git] / src / dv / gnunet-service-dv.c
1 /*
2      This file is part of GNUnet.
3      (C) 2013 Christian Grothoff (and other contributing authors)
4
5      GNUnet is free software; you can redistribute it and/or modify
6      it under the terms of the GNU General Public License as published
7      by the Free Software Foundation; either version 3, or (at your
8      option) any later version.
9
10      GNUnet is distributed in the hope that it will be useful, but
11      WITHOUT ANY WARRANTY; without even the implied warranty of
12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13      General Public License for more details.
14
15      You should have received a copy of the GNU General Public License
16      along with GNUnet; see the file COPYING.  If not, write to the
17      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
18      Boston, MA 02111-1307, USA.
19 */
20
21 /**
22  * @file dv/gnunet-service-dv.c
23  * @brief the distance vector service, primarily handles gossip of nearby
24  * peers and sending/receiving DV messages from core and decapsulating
25  * them
26  *
27  * @author Christian Grothoff
28  * @author Nathan Evans
29  */
30 #include "platform.h"
31 #include "gnunet_util_lib.h"
32 #include "gnunet_protocols.h"
33 #include "gnunet_core_service.h"
34 #include "gnunet_hello_lib.h"
35 #include "gnunet_peerinfo_service.h"
36 #include "gnunet_statistics_service.h"
37 #include "gnunet_set_service.h"
38 #include "gnunet_ats_service.h"
39 #include "dv.h"
40 #include <gcrypt.h>
41
42
43 /**
44  * How often do we establish the consensu?
45  */
46 #define GNUNET_DV_CONSENSUS_FREQUENCY GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MINUTES, 5)
47
48 /**
49  * Maximum number of messages we queue per peer.
50  */
51 #define MAX_QUEUE_SIZE 16
52
53 /**
54  * Maximum number of messages we queue towards the clients/plugin.
55  */
56 #define MAX_QUEUE_SIZE_PLUGIN 1024
57
58 /**
59  * The default fisheye depth, from how many hops away will
60  * we keep peers?
61  */
62 #define DEFAULT_FISHEYE_DEPTH 3
63
64 /**
65  * How many hops is a direct neighbor away?
66  */
67 #define DIRECT_NEIGHBOR_COST 1
68
69
70 GNUNET_NETWORK_STRUCT_BEGIN
71
72 /**
73  * Information about a peer DV can route to.  These entries are what
74  * we use as the binary format to establish consensus to create our
75  * routing table and as the address format in the HELLOs.
76  */
77 struct Target
78 {
79
80   /**
81    * Identity of the peer we can reach.
82    */
83   struct GNUNET_PeerIdentity peer;
84
85   /**
86    * How many hops (1-3) is this peer away? in network byte order
87    */
88   uint32_t distance GNUNET_PACKED;
89
90 };
91
92
93 /**
94  * Message exchanged between DV services (via core), requesting a
95  * message to be routed.  
96  */
97 struct RouteMessage
98 {
99   /**
100    * Type: GNUNET_MESSAGE_TYPE_DV_ROUTE
101    */
102   struct GNUNET_MessageHeader header;
103
104   /**
105    * Expected (remaining) distance.  Must be always smaller than
106    * DEFAULT_FISHEYE_DEPTH, should be zero at the target.  Must
107    * be decremented by one at each hop.  Peers must not forward
108    * these messages further once the counter has reached zero.
109    */
110   uint32_t distance GNUNET_PACKED;
111
112   /**
113    * The (actual) target of the message (this peer, if distance is zero).
114    */
115   struct GNUNET_PeerIdentity target;
116
117   /**
118    * The (actual) sender of the message.
119    */
120   struct GNUNET_PeerIdentity sender;
121
122 };
123
124 GNUNET_NETWORK_STRUCT_END
125
126
127 /**
128  * Linked list of messages to send to clients.
129  */
130 struct PendingMessage
131 {
132   /**
133    * Pointer to next item in the list
134    */
135   struct PendingMessage *next;
136
137   /**
138    * Pointer to previous item in the list
139    */
140   struct PendingMessage *prev;
141
142   /**
143    * Actual message to be sent, allocated after this struct.
144    */
145   const struct GNUNET_MessageHeader *msg;
146
147   /**
148    * Ultimate target for the message.
149    */
150   struct GNUNET_PeerIdentity ultimate_target;
151
152   /**
153    * Unique ID of the message.
154    */
155   uint32_t uid;
156
157 };
158
159
160 /**
161  * Information about a direct neighbor (core-level, excluding
162  * DV-links, only DV-enabled peers).
163  */
164 struct DirectNeighbor
165 {
166
167   /**
168    * Identity of the peer.
169    */
170   struct GNUNET_PeerIdentity peer;
171
172   /**
173    * Session ID we use whenever we create a set union with
174    * this neighbor; constructed from the XOR of our peer
175    * IDs and then salted with "DV-SALT" to avoid conflicts
176    * with other applications.
177    */
178   struct GNUNET_HashCode real_session_id;
179   
180   /**
181    * Head of linked list of messages to send to this peer.
182    */
183   struct PendingMessage *pm_head;
184
185   /**
186    * Tail of linked list of messages to send to this peer.
187    */
188   struct PendingMessage *pm_tail;
189
190   /**
191    * Transmit handle to core service.
192    */
193   struct GNUNET_CORE_TransmitHandle *cth;
194
195   /**
196    * Routing table of the neighbor, NULL if not yet established.
197    * Keys are peer identities, values are 'struct Target' entries.
198    * Note that the distances in the targets are from the point-of-view
199    * of the peer, not from us!
200    */ 
201   struct GNUNET_CONTAINER_MultiHashMap *neighbor_table;
202
203   /**
204    * Updated routing table of the neighbor, under construction,
205    * NULL if we are not currently building it.
206    * Keys are peer identities, values are 'struct Target' entries.
207    * Note that the distances in the targets are from the point-of-view
208    * of the peer, not from us!
209    */ 
210   struct GNUNET_CONTAINER_MultiHashMap *neighbor_table_consensus;
211
212   /**
213    * Our current (exposed) routing table as a set.
214    */
215   struct GNUNET_SET_Handle *my_set;
216
217   /**
218    * Handle for our current active set union operation.
219    */
220   struct GNUNET_SET_OperationHandle *set_op;
221
222   /**
223    * Handle used if we are listening for this peer, waiting for the
224    * other peer to initiate construction of the set union.  NULL if
225    * we ar the initiating peer.
226    */
227   struct GNUNET_SET_ListenHandle *listen_handle;
228
229   /**
230    * ID of the task we use to (periodically) update our consensus
231    * with this peer.  Used if we are the initiating peer.
232    */
233   GNUNET_SCHEDULER_TaskIdentifier initiate_task;
234
235   /**
236    * At what offset are we, with respect to inserting our own routes
237    * into the consensus?
238    */
239   unsigned int consensus_insertion_offset;
240
241   /**
242    * At what distance are we, with respect to inserting our own routes
243    * into the consensus?
244    */
245   unsigned int consensus_insertion_distance;
246
247   /**
248    * Number of messages currently in the 'pm_XXXX'-DLL.
249    */
250   unsigned int pm_queue_size;
251
252   /**
253    * Flag set within 'check_target_removed' to trigger full global route refresh.
254    */
255   int target_removed;
256
257   /**
258    * Our distance to this peer, 0 for unknown.
259    */
260   uint32_t distance;
261
262   /**
263    * Is this neighbor connected at the core level?
264    */
265   int connected;
266
267 };
268
269
270 /**
271  * A route includes information about the next hop,
272  * the target, and the ultimate distance to the
273  * target.
274  */
275 struct Route
276 {
277
278   /**
279    * Which peer do we need to forward the message to?
280    */
281   struct DirectNeighbor *next_hop;
282
283   /**
284    * What would be the target, and how far is it away?
285    */
286   struct Target target;
287
288   /**
289    * Offset of this target in the respective consensus set.
290    */
291   unsigned int set_offset;
292
293 };
294
295
296 /**
297  * Set of targets we bring to a consensus; all targets in a set have a
298  * distance equal to the sets distance (which is implied by the array
299  * index of the set).
300  */
301 struct ConsensusSet
302 {
303
304   /**
305    * Array of targets in the set, may include NULL entries if a
306    * neighbor has disconnected; the targets are allocated with the
307    * respective container (all_routes), not here.
308    */
309   struct Route **targets;
310
311   /**
312    * Size of the 'targets' array.
313    */
314   unsigned int array_length;
315
316 };
317
318
319 /**
320  * Hashmap of all of our neighbors; processing these usually requires
321  * first checking to see if the peer is core-connected and if the 
322  * distance is 1, in which case they are direct neighbors.
323  */
324 static struct GNUNET_CONTAINER_MultiHashMap *direct_neighbors;
325
326 /**
327  * Hashmap with all routes that we currently support; contains 
328  * routing information for all peers from distance 2
329  * up to distance DEFAULT_FISHEYE_DEPTH.
330  */
331 static struct GNUNET_CONTAINER_MultiHashMap *all_routes;
332
333 /**
334  * Array of consensus sets we expose to the outside world.  Sets
335  * are structured by the distance to the target.
336  */
337 static struct ConsensusSet consensi[DEFAULT_FISHEYE_DEPTH - 1];
338
339 /**
340  * Handle to the core service api.
341  */
342 static struct GNUNET_CORE_Handle *core_api;
343
344 /**
345  * The identity of our peer.
346  */
347 static struct GNUNET_PeerIdentity my_identity;
348
349 /**
350  * The configuration for this service.
351  */
352 static const struct GNUNET_CONFIGURATION_Handle *cfg;
353
354 /**
355  * The client, the DV plugin connected to us (or an event monitor).
356  * Hopefully this client will never change, although if the plugin
357  * dies and returns for some reason it may happen.
358  */
359 static struct GNUNET_SERVER_NotificationContext *nc;
360
361 /**
362  * Handle for the statistics service.
363  */
364 static struct GNUNET_STATISTICS_Handle *stats;
365
366 /**
367  * Handle to ATS service.
368  */
369 static struct GNUNET_ATS_PerformanceHandle *ats;
370  
371
372 /**
373  * Start creating a new DV set union by initiating the connection.
374  *
375  * @param cls the 'struct DirectNeighbor' of the peer we're building
376  *        a routing consensus with
377  * @param tc scheduler context
378  */    
379 static void
380 initiate_set_union (void *cls,
381                     const struct GNUNET_SCHEDULER_TaskContext *tc);
382
383
384 /**
385  * Start creating a new DV set union construction, our neighbour has
386  * asked for it (callback for listening peer).
387  *
388  * @param cls the 'struct DirectNeighbor' of the peer we're building
389  *        a routing consensus with
390  * @param other_peer the other peer
391  * @param context_msg message with application specific information from
392  *        the other peer
393  * @param request request from the other peer, use GNUNET_SET_accept
394  *        to accept it, otherwise the request will be refused
395  *        Note that we don't use a return value here, as it is also
396  *        necessary to specify the set we want to do the operation with,
397  *        whith sometimes can be derived from the context message.
398  *        Also necessary to specify the timeout.
399  */    
400 static void
401 listen_set_union (void *cls,
402                   const struct GNUNET_PeerIdentity *other_peer,
403                   const struct GNUNET_MessageHeader *context_msg,
404                   struct GNUNET_SET_Request *request);
405
406
407 /**
408  * Forward a message from another peer to the plugin.
409  *
410  * @param message the message to send to the plugin
411  * @param origin the original sender of the message
412  * @param distance distance to the original sender of the message
413  */
414 static void
415 send_data_to_plugin (const struct GNUNET_MessageHeader *message, 
416                      const struct GNUNET_PeerIdentity *origin,
417                      uint32_t distance)
418 {
419   struct GNUNET_DV_ReceivedMessage *received_msg;
420   size_t size;
421
422   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
423               "Delivering message from peer `%s'\n",
424               GNUNET_i2s (origin));
425   size = sizeof (struct GNUNET_DV_ReceivedMessage) + 
426     ntohs (message->size);
427   if (size >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
428   {    
429     GNUNET_break (0); /* too big */
430     return;
431   }
432   received_msg = GNUNET_malloc (size);
433   received_msg->header.size = htons (size);
434   received_msg->header.type = htons (GNUNET_MESSAGE_TYPE_DV_RECV);
435   received_msg->distance = htonl (distance);
436   received_msg->sender = *origin;
437   memcpy (&received_msg[1], message, ntohs (message->size));
438   GNUNET_SERVER_notification_context_broadcast (nc, 
439                                                 &received_msg->header,
440                                                 GNUNET_YES);
441   GNUNET_free (received_msg);
442 }
443
444
445 /**
446  * Forward a control message to the plugin.
447  *
448  * @param message the message to send to the plugin
449  */
450 static void
451 send_control_to_plugin (const struct GNUNET_MessageHeader *message)
452 {
453   GNUNET_SERVER_notification_context_broadcast (nc, 
454                                                 message,
455                                                 GNUNET_NO);
456 }
457
458
459 /**
460  * Give an (N)ACK message to the plugin, we transmitted a message for it.
461  *
462  * @param target peer that received the message
463  * @param uid plugin-chosen UID for the message
464  * @param nack GNUNET_NO to send ACK, GNUNET_YES to send NACK
465  */
466 static void
467 send_ack_to_plugin (const struct GNUNET_PeerIdentity *target, 
468                     uint32_t uid,
469                     int nack)
470 {
471   struct GNUNET_DV_AckMessage ack_msg;
472
473   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
474               "Delivering ACK for message to peer `%s'\n",
475               GNUNET_i2s (target));
476   ack_msg.header.size = htons (sizeof (ack_msg));
477   ack_msg.header.type = htons ((GNUNET_YES == nack) 
478                                ? GNUNET_MESSAGE_TYPE_DV_SEND_NACK
479                                : GNUNET_MESSAGE_TYPE_DV_SEND_ACK);
480   ack_msg.uid = htonl (uid);
481   ack_msg.target = *target;
482   send_control_to_plugin (&ack_msg.header);
483 }
484
485
486 /**
487  * Send a DISTANCE_CHANGED message to the plugin.
488  *
489  * @param peer peer with a changed distance
490  * @param distance new distance to the peer
491  */
492 static void
493 send_distance_change_to_plugin (const struct GNUNET_PeerIdentity *peer, 
494                                 uint32_t distance)
495 {
496   struct GNUNET_DV_DistanceUpdateMessage du_msg;
497
498   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
499               "Delivering DISTANCE_CHANGED for message about peer `%s'\n",
500               GNUNET_i2s (peer));
501   du_msg.header.size = htons (sizeof (du_msg));
502   du_msg.header.type = htons (GNUNET_MESSAGE_TYPE_DV_DISTANCE_CHANGED);
503   du_msg.distance = htonl (distance);
504   du_msg.peer = *peer;
505   send_control_to_plugin (&du_msg.header);
506 }
507
508
509 /**
510  * Give a CONNECT message to the plugin.
511  *
512  * @param target peer that connected
513  * @param distance distance to the target
514  */
515 static void
516 send_connect_to_plugin (const struct GNUNET_PeerIdentity *target, 
517                         uint32_t distance)
518 {
519   struct GNUNET_DV_ConnectMessage cm;
520
521   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
522               "Delivering CONNECT about peer `%s'\n",
523               GNUNET_i2s (target));
524   cm.header.size = htons (sizeof (cm));
525   cm.header.type = htons (GNUNET_MESSAGE_TYPE_DV_CONNECT);
526   cm.distance = htonl (distance);
527   cm.peer = *target;
528   send_control_to_plugin (&cm.header);
529 }
530
531
532 /**
533  * Give a DISCONNECT message to the plugin.
534  *
535  * @param target peer that disconnected
536  */
537 static void
538 send_disconnect_to_plugin (const struct GNUNET_PeerIdentity *target)
539 {
540   struct GNUNET_DV_DisconnectMessage dm;
541
542   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
543               "Delivering DISCONNECT about peer `%s'\n",
544               GNUNET_i2s (target));
545   dm.header.size = htons (sizeof (dm));
546   dm.header.type = htons (GNUNET_MESSAGE_TYPE_DV_DISCONNECT);
547   dm.reserved = htonl (0);
548   dm.peer = *target;
549   send_control_to_plugin (&dm.header);
550 }
551
552
553 /**
554  * Function called to transfer a message to another peer
555  * via core.
556  *
557  * @param cls closure with the direct neighbor
558  * @param size number of bytes available in buf
559  * @param buf where the callee should write the message
560  * @return number of bytes written to buf
561  */
562 static size_t
563 core_transmit_notify (void *cls, size_t size, void *buf)
564 {
565   struct DirectNeighbor *dn = cls;
566   char *cbuf = buf;
567   struct PendingMessage *pending;
568   size_t off;
569   size_t msize;
570
571   dn->cth = NULL;
572   if (NULL == buf)
573   {
574     /* peer disconnected */
575     return 0;
576   }
577   off = 0;
578   pending = dn->pm_head;
579   off = 0;
580   while ( (NULL != (pending = dn->pm_head)) &&
581           (size >= off + (msize = ntohs (pending->msg->size))))
582   {
583     dn->pm_queue_size--;
584     GNUNET_CONTAINER_DLL_remove (dn->pm_head,
585                                  dn->pm_tail,
586                                  pending);
587     memcpy (&cbuf[off], pending->msg, msize);
588     if (0 != pending->uid) 
589       send_ack_to_plugin (&pending->ultimate_target,
590                           pending->uid,
591                           GNUNET_NO);
592     GNUNET_free (pending);
593     off += msize;
594   }
595   if (NULL != dn->pm_head)
596     dn->cth =
597       GNUNET_CORE_notify_transmit_ready (core_api,
598                                          GNUNET_YES /* cork */,
599                                          0 /* priority */,
600                                          GNUNET_TIME_UNIT_FOREVER_REL,
601                                          &dn->peer,
602                                          msize,                                  
603                                          &core_transmit_notify, dn);
604   return off;
605 }
606
607
608 /**
609  * Forward the given payload to the given target.
610  *
611  * @param target where to send the message
612  * @param uid unique ID for the message
613  * @param ultimate_target ultimate recipient for the message
614  * @param distance expected (remaining) distance to the target
615  * @param sender original sender of the message
616  * @param payload payload of the message
617  */
618 static void
619 forward_payload (struct DirectNeighbor *target,
620                  uint32_t distance,
621                  uint32_t uid,
622                  const struct GNUNET_PeerIdentity *sender,
623                  const struct GNUNET_PeerIdentity *ultimate_target,
624                  const struct GNUNET_MessageHeader *payload)
625 {
626   struct PendingMessage *pm;
627   struct RouteMessage *rm;
628   size_t msize;
629
630   if ( (target->pm_queue_size >= MAX_QUEUE_SIZE) &&
631        (0 != memcmp (sender,
632                      &my_identity,
633                      sizeof (struct GNUNET_PeerIdentity))) )
634   {
635     GNUNET_break (0 == uid);
636     return;
637   }
638   msize = sizeof (struct RouteMessage) + ntohs (payload->size);
639   if (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
640   {
641     GNUNET_break (0);
642     return;
643   }
644   pm = GNUNET_malloc (sizeof (struct PendingMessage) + msize);
645   pm->ultimate_target = *ultimate_target;
646   pm->uid = uid;
647   pm->msg = (const struct GNUNET_MessageHeader *) &pm[1];
648   rm = (struct RouteMessage *) &pm[1];
649   rm->header.size = htons ((uint16_t) msize);
650   rm->header.type = htons (GNUNET_MESSAGE_TYPE_DV_ROUTE);
651   rm->distance = htonl (distance);
652   rm->target = target->peer;
653   rm->sender = *sender;
654   memcpy (&rm[1], payload, ntohs (payload->size));
655   GNUNET_CONTAINER_DLL_insert_tail (target->pm_head,
656                                     target->pm_tail,
657                                     pm);
658   target->pm_queue_size++;
659   if (NULL == target->cth)
660     target->cth = GNUNET_CORE_notify_transmit_ready (core_api,
661                                                      GNUNET_YES /* cork */,
662                                                      0 /* priority */,
663                                                      GNUNET_TIME_UNIT_FOREVER_REL,
664                                                      &target->peer,
665                                                      msize,                                      
666                                                      &core_transmit_notify, target);
667 }
668
669
670 /**
671  * Find a free slot for storing a 'route' in the 'consensi'
672  * set at the given distance.
673  *
674  * @param distance distance to use for the set slot
675  */
676 static unsigned int
677 get_consensus_slot (uint32_t distance)
678 {
679   struct ConsensusSet *cs;
680   unsigned int i;
681
682   cs = &consensi[distance];
683   i = 0;
684   while ( (i < cs->array_length) &&
685           (NULL != cs->targets[i]) ) i++;
686   if (i == cs->array_length)
687     GNUNET_array_grow (cs->targets,
688                        cs->array_length,
689                        cs->array_length * 2 + 2);
690   return i;
691 }
692
693
694 /**
695  * Allocate a slot in the consensus set for a route.
696  *
697  * @param route route to initialize
698  * @param distance which consensus set to use
699  */
700 static void
701 allocate_route (struct Route *route,
702                 uint32_t distance)
703 {
704   unsigned int i;
705
706   i = get_consensus_slot (distance);
707   route->set_offset = i;
708   consensi[distance].targets[i] = route;
709   route->target.distance = htonl (distance);
710 }
711
712
713 /**
714  * Release a slot in the consensus set for a route.
715  *
716  * @param route route to release the slot from
717  */
718 static void
719 release_route (struct Route *route)
720 {
721   consensi[ntohl (route->target.distance)].targets[route->set_offset] = NULL;
722   route->set_offset = UINT_MAX; /* indicate invalid slot */
723 }
724
725
726 /**
727  * Move a route from one consensus set to another.
728  *
729  * @param route route to move
730  * @param new_distance new distance for the route (destination set)
731  */
732 static void
733 move_route (struct Route *route,
734             uint32_t new_distance)
735 {
736   unsigned int i;
737
738   release_route (route);
739   i = get_consensus_slot (new_distance);
740   route->set_offset = i;
741   consensi[new_distance].targets[i] = route;     
742   route->target.distance = htonl (new_distance);
743 }
744
745
746 /**
747  * Initialize this neighbors 'my_set' and when done give
748  * it to the pending set operation for execution.
749  *
750  * @param cls the neighbor for which we are building the set
751  */
752 static void
753 build_set (void *cls)
754 {
755   struct DirectNeighbor *neighbor = cls;
756   struct GNUNET_SET_Element element;
757
758   while ( (DEFAULT_FISHEYE_DEPTH - 1 > neighbor->consensus_insertion_distance) &&
759           (consensi[neighbor->consensus_insertion_distance].array_length == neighbor->consensus_insertion_offset) )
760   {
761     neighbor->consensus_insertion_offset = 0;
762     neighbor->consensus_insertion_distance++;
763     /* skip over NULL entries */
764     while ( (DEFAULT_FISHEYE_DEPTH - 1 > neighbor->consensus_insertion_distance) &&
765             (consensi[neighbor->consensus_insertion_distance].array_length < neighbor->consensus_insertion_offset) &&
766             (NULL == consensi[neighbor->consensus_insertion_distance].targets[neighbor->consensus_insertion_offset]) )
767       neighbor->consensus_insertion_offset++;
768   }
769   if (DEFAULT_FISHEYE_DEPTH - 1 == neighbor->consensus_insertion_distance)
770   {
771     /* we have added all elements to the set, run the operation */
772 #if 0
773     GNUNET_SET_operation_conclude (neighbor->my_op,
774                                    neighbor->my_set);
775 #endif
776     GNUNET_SET_destroy (neighbor->my_set);
777     neighbor->my_set = NULL;
778     return;
779   }
780   element.size = sizeof (struct Target);
781   element.data = &consensi[neighbor->consensus_insertion_distance].targets[neighbor->consensus_insertion_offset++]->target;
782
783   /* skip over NULL entries */
784   while ( (DEFAULT_FISHEYE_DEPTH - 1 > neighbor->consensus_insertion_distance) &&
785           (consensi[neighbor->consensus_insertion_distance].array_length < neighbor->consensus_insertion_offset) &&
786           (NULL == consensi[neighbor->consensus_insertion_distance].targets[neighbor->consensus_insertion_offset]) )
787     neighbor->consensus_insertion_offset++;  
788   GNUNET_SET_add_element (neighbor->my_set,
789                           &element,
790                           &build_set, neighbor);
791   
792 }
793
794
795 /**
796  * A peer is now connected to us at distance 1.  Initiate DV exchange.
797  *
798  * @param neighbor entry for the neighbor at distance 1
799  */
800 static void
801 handle_direct_connect (struct DirectNeighbor *neighbor)
802 {
803   struct Route *route;
804   struct GNUNET_HashCode session_id;
805
806   GNUNET_STATISTICS_update (stats,
807                             "# peers connected (1-hop)",
808                             1, GNUNET_NO);
809   route = GNUNET_CONTAINER_multihashmap_get (all_routes, 
810                                              &neighbor->peer.hashPubKey);
811   if (NULL != route)  
812   {
813     send_disconnect_to_plugin (&neighbor->peer);
814     release_route (route);
815     GNUNET_free (route);
816   }
817   /* construct session ID seed as XOR of both peer's identities */
818   GNUNET_CRYPTO_hash_xor (&my_identity.hashPubKey, 
819                           &neighbor->peer.hashPubKey, 
820                           &session_id);
821   /* make sure session ID is unique across applications by salting it with 'DV' */
822   GNUNET_CRYPTO_hkdf (&neighbor->real_session_id, sizeof (struct GNUNET_HashCode),
823                       GCRY_MD_SHA512, GCRY_MD_SHA256,
824                       "DV-SALT", 2,
825                       &session_id, sizeof (session_id),
826                       NULL, 0);
827   if (1 == GNUNET_CRYPTO_hash_cmp (&neighbor->peer.hashPubKey,
828                                    &my_identity.hashPubKey))
829     neighbor->initiate_task = GNUNET_SCHEDULER_add_now (&initiate_set_union,
830                                                         neighbor);
831   else
832     neighbor->listen_handle = GNUNET_SET_listen (cfg,
833                                                  GNUNET_SET_OPERATION_UNION,
834                                                  &neighbor->real_session_id,
835                                                  &listen_set_union,
836                                                  neighbor);
837 }
838
839
840 /**
841  * Method called whenever a peer connects.
842  *
843  * @param cls closure
844  * @param peer peer identity this notification is about
845  */
846 static void
847 handle_core_connect (void *cls, const struct GNUNET_PeerIdentity *peer)
848 {
849   struct DirectNeighbor *neighbor;
850  
851   /* Check for connect to self message */
852   if (0 == memcmp (&my_identity, peer, sizeof (struct GNUNET_PeerIdentity)))
853     return;
854   /* check if entry exists */
855   neighbor = GNUNET_CONTAINER_multihashmap_get (direct_neighbors, 
856                                                 &peer->hashPubKey);
857   if (NULL != neighbor)
858   {
859     GNUNET_break (GNUNET_YES != neighbor->connected);
860     neighbor->connected = GNUNET_YES;
861     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
862                 "Core connected to %s (distance %u)\n",
863                 GNUNET_i2s (peer),
864                 (unsigned int) neighbor->distance);
865     if (DIRECT_NEIGHBOR_COST != neighbor->distance)
866       return;
867     handle_direct_connect (neighbor);
868     return;
869   }
870   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
871               "Core connected to %s (distance unknown)\n",
872               GNUNET_i2s (peer));
873   neighbor = GNUNET_malloc (sizeof (struct DirectNeighbor));
874   neighbor->peer = *peer;
875   GNUNET_assert (GNUNET_YES ==
876                  GNUNET_CONTAINER_multihashmap_put (direct_neighbors,
877                                                     &peer->hashPubKey,
878                                                     neighbor,
879                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
880   neighbor->connected = GNUNET_YES;
881   neighbor->distance = 0; /* unknown */
882 }
883
884
885 /**
886  * Called for each 'target' in a neighbor table to free the associated memory.
887  *
888  * @param cls NULL
889  * @param key key of the value
890  * @param value value to free
891  * @return GNUNET_OK to continue to iterate
892  */
893 static int
894 free_targets (void *cls,
895               const struct GNUNET_HashCode *key,
896               void *value)
897 {
898   GNUNET_free (value);
899   return GNUNET_OK;
900 }
901
902
903 /**
904  * Multihashmap iterator for checking if a given route is
905  * (now) useful to this peer.
906  *
907  * @param cls the direct neighbor for the given route
908  * @param key key value stored under
909  * @param value a 'struct Target' that may or may not be useful; not that
910  *        the distance in 'target' does not include the first hop yet
911  * @return GNUNET_YES to continue iteration, GNUNET_NO to stop
912  */
913 static int
914 check_possible_route (void *cls, const struct GNUNET_HashCode * key, void *value)
915 {
916   struct DirectNeighbor *neighbor = cls;
917   struct Target *target = value;
918   struct Route *route;
919   
920   route = GNUNET_CONTAINER_multihashmap_get (all_routes,
921                                            key);
922   if (NULL != route)
923   {
924     if (ntohl (route->target.distance) > ntohl (target->distance) + 1)
925     {
926       /* this 'target' is cheaper than the existing route; switch to alternative route! */
927       move_route (route, ntohl (target->distance) + 1);
928       route->next_hop = neighbor;
929       send_distance_change_to_plugin (&target->peer, ntohl (target->distance) + 1);
930     }
931     return GNUNET_YES; /* got a route to this target already */
932   }
933   route = GNUNET_malloc (sizeof (struct Route));
934   route->next_hop = neighbor;
935   route->target.distance = htonl (ntohl (target->distance) + 1);
936   route->target.peer = target->peer;
937   allocate_route (route, ntohl (route->target.distance));
938   GNUNET_assert (GNUNET_YES ==
939                  GNUNET_CONTAINER_multihashmap_put (all_routes,
940                                                     &route->target.peer.hashPubKey,
941                                                     route,
942                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
943   send_connect_to_plugin (&route->target.peer, ntohl (target->distance));
944   return GNUNET_YES;
945 }
946
947
948 /**
949  * Multihashmap iterator for finding routes that were previously
950  * "hidden" due to a better route (called after a disconnect event).
951  *
952  * @param cls NULL
953  * @param key peer identity of the given direct neighbor
954  * @param value a 'struct DirectNeighbor' to check for additional routes
955  * @return GNUNET_YES to continue iteration
956  */
957 static int
958 refresh_routes (void *cls, const struct GNUNET_HashCode * key, void *value)
959 {
960   struct DirectNeighbor *neighbor = value;
961
962   if ( (GNUNET_YES != neighbor->connected) ||
963        (DIRECT_NEIGHBOR_COST != neighbor->distance) )
964     return GNUNET_YES;    
965   if (NULL != neighbor->neighbor_table)
966     GNUNET_CONTAINER_multihashmap_iterate (neighbor->neighbor_table,
967                                            &check_possible_route,
968                                            neighbor);
969   return GNUNET_YES;
970 }
971
972
973 /**
974  * Get distance information from 'atsi'.
975  *
976  * @param atsi performance data
977  * @param atsi_count number of entries in atsi
978  * @return connected transport distance
979  */
980 static uint32_t
981 get_atsi_distance (const struct GNUNET_ATS_Information *atsi,
982                    uint32_t atsi_count)
983 {
984   uint32_t i;
985
986   for (i = 0; i < atsi_count; i++)
987     if (ntohl (atsi[i].type) == GNUNET_ATS_QUALITY_NET_DISTANCE)
988       return (0 == ntohl (atsi->value)) ? DIRECT_NEIGHBOR_COST : ntohl (atsi->value); // FIXME: 0 check should not be required once ATS is fixed!
989   /* If we do not have explicit distance data, assume direct neighbor. */
990   return DIRECT_NEIGHBOR_COST;
991 }
992
993
994 /**
995  * Multihashmap iterator for freeing routes that go via a particular
996  * neighbor that disconnected and is thus no longer available.
997  *
998  * @param cls the direct neighbor that is now unavailable
999  * @param key key value stored under
1000  * @param value a 'struct Route' that may or may not go via neighbor
1001  *
1002  * @return GNUNET_YES to continue iteration, GNUNET_NO to stop
1003  */
1004 static int
1005 cull_routes (void *cls, const struct GNUNET_HashCode * key, void *value)
1006 {
1007   struct DirectNeighbor *neighbor = cls;
1008   struct Route *route = value;
1009
1010   if (route->next_hop != neighbor)
1011     return GNUNET_YES; /* not affected */
1012   GNUNET_assert (GNUNET_YES ==
1013                  GNUNET_CONTAINER_multihashmap_remove (all_routes, key, value));
1014   release_route (route);
1015   send_disconnect_to_plugin (&route->target.peer);
1016   GNUNET_free (route);
1017   return GNUNET_YES;
1018 }
1019
1020
1021 /**
1022  * Handle the case that a direct connection to a peer is
1023  * disrupted.  Remove all routes via that peer and
1024  * stop the consensus with it.
1025  *
1026  * @param neighbor peer that was disconnected (or at least is no 
1027  *    longer at distance 1)
1028  */
1029 static void
1030 handle_direct_disconnect (struct DirectNeighbor *neighbor)
1031 {
1032   GNUNET_CONTAINER_multihashmap_iterate (all_routes,
1033                                          &cull_routes,
1034                                          neighbor);
1035   if (NULL != neighbor->cth)
1036   {
1037     GNUNET_CORE_notify_transmit_ready_cancel (neighbor->cth);
1038     neighbor->cth = NULL;
1039   }
1040   if (NULL != neighbor->neighbor_table_consensus)
1041   {
1042     GNUNET_CONTAINER_multihashmap_iterate (neighbor->neighbor_table_consensus,
1043                                            &free_targets,
1044                                            NULL);
1045     GNUNET_CONTAINER_multihashmap_destroy (neighbor->neighbor_table_consensus);
1046     neighbor->neighbor_table_consensus = NULL;
1047   }
1048   if (NULL != neighbor->neighbor_table)
1049   {
1050     GNUNET_CONTAINER_multihashmap_iterate (neighbor->neighbor_table,
1051                                            &free_targets,
1052                                            NULL);
1053     GNUNET_CONTAINER_multihashmap_destroy (neighbor->neighbor_table);
1054     neighbor->neighbor_table = NULL;
1055   }
1056   if (NULL != neighbor->set_op)
1057   {
1058     GNUNET_SET_operation_cancel (neighbor->set_op);
1059     neighbor->set_op = NULL;
1060   }
1061   if (NULL != neighbor->my_set)
1062   {
1063     GNUNET_SET_destroy (neighbor->my_set);
1064     neighbor->my_set = NULL;
1065   }
1066   if (NULL != neighbor->listen_handle)
1067   {
1068     GNUNET_SET_listen_cancel (neighbor->listen_handle);
1069     neighbor->listen_handle = NULL;
1070   }
1071   if (GNUNET_SCHEDULER_NO_TASK != neighbor->initiate_task)
1072   {
1073     GNUNET_SCHEDULER_cancel (neighbor->initiate_task);
1074     neighbor->initiate_task = GNUNET_SCHEDULER_NO_TASK;
1075   }
1076 }
1077
1078
1079 /**
1080  * Function that is called with QoS information about an address; used
1081  * to update our current distance to another peer.
1082  *
1083  * @param cls closure
1084  * @param address the address
1085  * @param active is this address in active use
1086  * @param bandwidth_out assigned outbound bandwidth for the connection
1087  * @param bandwidth_in assigned inbound bandwidth for the connection
1088  * @param ats performance data for the address (as far as known)
1089  * @param ats_count number of performance records in 'ats'
1090  */
1091 static void
1092 handle_ats_update (void *cls,
1093                    const struct GNUNET_HELLO_Address *address,
1094                    int active,
1095                    struct GNUNET_BANDWIDTH_Value32NBO bandwidth_out,
1096                    struct GNUNET_BANDWIDTH_Value32NBO bandwidth_in,
1097                    const struct GNUNET_ATS_Information *ats, 
1098                    uint32_t ats_count)
1099 {
1100   struct DirectNeighbor *neighbor;
1101   uint32_t distance;
1102
1103   if (GNUNET_NO == active)
1104         return;
1105   distance = get_atsi_distance (ats, ats_count); 
1106   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1107               "ATS says distance to %s is %u\n",
1108               GNUNET_i2s (&address->peer),
1109               (unsigned int) distance);
1110   /* check if entry exists */
1111   neighbor = GNUNET_CONTAINER_multihashmap_get (direct_neighbors, 
1112                                                 &address->peer.hashPubKey);
1113   if (NULL != neighbor)
1114   {    
1115     if ( (DIRECT_NEIGHBOR_COST == neighbor->distance) &&
1116          (DIRECT_NEIGHBOR_COST == distance) )
1117       return; /* no change */
1118     if (DIRECT_NEIGHBOR_COST == neighbor->distance) 
1119     {
1120       neighbor->distance = distance;
1121       GNUNET_STATISTICS_update (stats,
1122                                 "# peers connected (1-hop)",
1123                                 -1, GNUNET_NO);  
1124       handle_direct_disconnect (neighbor);
1125       GNUNET_CONTAINER_multihashmap_iterate (direct_neighbors,
1126                                              &refresh_routes,
1127                                              NULL);
1128       return;
1129     }
1130     neighbor->distance = distance;
1131     if (DIRECT_NEIGHBOR_COST != neighbor->distance)
1132       return;    
1133     if (GNUNET_YES != neighbor->connected)
1134       return;
1135     handle_direct_connect (neighbor);
1136     return;
1137   }
1138   neighbor = GNUNET_malloc (sizeof (struct DirectNeighbor));
1139   neighbor->peer = address->peer;
1140   GNUNET_assert (GNUNET_YES ==
1141                  GNUNET_CONTAINER_multihashmap_put (direct_neighbors,
1142                                                     &address->peer.hashPubKey,
1143                                                     neighbor,
1144                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
1145   neighbor->connected = GNUNET_NO; /* not yet */
1146   neighbor->distance = distance; 
1147 }
1148
1149
1150 /**
1151  * Check if a target was removed from the set of the other peer; if so,
1152  * if we also used it for our route, we need to remove it from our
1153  * 'all_routes' set (and later check if an alternative path now exists).
1154  *
1155  * @param cls the 'struct DirectNeighbor'
1156  * @param key peer identity for the target
1157  * @param value a 'struct Target' previously reachable via the given neighbor
1158  */
1159 static int
1160 check_target_removed (void *cls,
1161                       const struct GNUNET_HashCode *key,
1162                       void *value)
1163 {
1164   struct DirectNeighbor *neighbor = cls;
1165   struct Target *new_target;
1166   struct Route *current_route;
1167
1168   new_target = GNUNET_CONTAINER_multihashmap_get (neighbor->neighbor_table_consensus,
1169                                                   key);
1170   if (NULL == new_target)
1171   {
1172     /* target was revoked, check if it was used */
1173     current_route = GNUNET_CONTAINER_multihashmap_get (all_routes,
1174                                                        key);
1175     if ( (NULL == current_route) ||
1176          (current_route->next_hop != neighbor) )
1177     {
1178       /* didn't matter, wasn't used */
1179       return GNUNET_OK;
1180     }
1181     /* remove existing route */
1182     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1183                 "Lost route to %s\n",
1184                 GNUNET_i2s (&current_route->target.peer));
1185     GNUNET_assert (GNUNET_YES ==
1186                    GNUNET_CONTAINER_multihashmap_remove (all_routes, key, current_route));
1187     send_disconnect_to_plugin (&current_route->target.peer);
1188     GNUNET_free (current_route);
1189     neighbor->target_removed = GNUNET_YES;
1190     return GNUNET_OK;
1191   }
1192   return GNUNET_OK;
1193 }
1194
1195
1196 /**
1197  * Check if a target was added to the set of the other peer; if it
1198  * was added or impoves the existing route, do the needed updates.
1199  *
1200  * @param cls the 'struct DirectNeighbor'
1201  * @param key peer identity for the target
1202  * @param value a 'struct Target' now reachable via the given neighbor
1203  */
1204 static int
1205 check_target_added (void *cls,
1206                       const struct GNUNET_HashCode *key,
1207                       void *value)
1208 {
1209   struct DirectNeighbor *neighbor = cls;
1210   struct Target *target = value;
1211   struct Route *current_route;
1212
1213   /* target was revoked, check if it was used */
1214   current_route = GNUNET_CONTAINER_multihashmap_get (all_routes,
1215                                                      key);
1216   if (NULL != current_route)
1217   {
1218     /* route exists */
1219     if (current_route->next_hop == neighbor)
1220     {
1221       /* we had the same route before, no change */
1222       if (ntohl (target->distance) + 1 != ntohl (current_route->target.distance))
1223       {
1224         current_route->target.distance = htonl (ntohl (target->distance) + 1);
1225         send_distance_change_to_plugin (&target->peer, ntohl (target->distance) + 1);
1226       }
1227       return GNUNET_OK;
1228     }
1229     if (ntohl (current_route->target.distance) >= ntohl (target->distance) + 1)
1230     {
1231       /* alternative, shorter route exists, ignore */
1232       return GNUNET_OK;
1233     }
1234     /* new route is better than the existing one, take over! */
1235     /* NOTE: minor security issue: malicious peers may advertise
1236        very short routes to take over longer paths; as we don't
1237        check that the shorter routes actually work, a malicious
1238        direct neighbor can use this to DoS our long routes */
1239     current_route->next_hop = neighbor;
1240     current_route->target.distance = htonl (ntohl (target->distance) + 1);
1241     send_distance_change_to_plugin (&target->peer, ntohl (target->distance) + 1);
1242     return GNUNET_OK;
1243   }
1244   /* new route */
1245   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1246               "Discovered new route to %s using %u hops\n",
1247               GNUNET_i2s (&target->peer),
1248               (unsigned int) (ntohl (target->distance) + 1));
1249   current_route = GNUNET_malloc (sizeof (struct Route));
1250   current_route->next_hop = neighbor;
1251   current_route->target.peer = target->peer;
1252   current_route->target.distance = htonl (ntohl (target->distance) + 1);
1253   GNUNET_assert (GNUNET_YES ==
1254                  GNUNET_CONTAINER_multihashmap_put (all_routes,
1255                                                     &current_route->target.peer.hashPubKey,
1256                                                     current_route,
1257                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
1258   send_connect_to_plugin (&current_route->target.peer,
1259                           ntohl (current_route->target.distance));
1260   return GNUNET_OK;
1261 }
1262
1263
1264 /**
1265  * Callback for set operation results. Called for each element
1266  * in the result set.
1267  * We have learned a new route from the other peer.  Add it to the
1268  * route set we're building.
1269  *
1270  * @param cls the 'struct DirectNeighbor' we're building the consensus with
1271  * @param element a result element, only valid if status is GNUNET_SET_STATUS_OK
1272  * @param status see enum GNUNET_SET_Status
1273  */
1274 static void
1275 handle_set_union_result (void *cls,
1276                          const struct GNUNET_SET_Element *element,
1277                          enum GNUNET_SET_Status status)
1278 {
1279   struct DirectNeighbor *neighbor = cls;
1280   struct Target *target;
1281
1282   switch (status)
1283   {
1284   case GNUNET_SET_STATUS_OK:
1285     if (sizeof (struct Target) != element->size)
1286     {
1287       GNUNET_break_op (0);
1288       return;
1289     }
1290     target = GNUNET_malloc (sizeof (struct Target));
1291     memcpy (target, element->data, sizeof (struct Target));
1292     if (GNUNET_YES !=
1293         GNUNET_CONTAINER_multihashmap_put (neighbor->neighbor_table_consensus,
1294                                            &target->peer.hashPubKey,
1295                                            target,
1296                                            GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
1297     {
1298       GNUNET_break_op (0);
1299       GNUNET_free (target);
1300     }
1301     break;
1302   case GNUNET_SET_STATUS_TIMEOUT:
1303   case GNUNET_SET_STATUS_FAILURE:
1304     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1305                 "Failed to establish DV union, will try again later\n");
1306     GNUNET_SET_operation_cancel (neighbor->set_op); /* FIXME: needed? */
1307     neighbor->set_op = NULL;
1308     if (NULL != neighbor->neighbor_table_consensus)
1309     {
1310       GNUNET_CONTAINER_multihashmap_iterate (neighbor->neighbor_table_consensus,
1311                                              &free_targets,
1312                                              NULL);
1313       GNUNET_CONTAINER_multihashmap_destroy (neighbor->neighbor_table_consensus);
1314       neighbor->neighbor_table_consensus = NULL;
1315     }
1316     if (1 == GNUNET_CRYPTO_hash_cmp (&neighbor->peer.hashPubKey,
1317                                      &my_identity.hashPubKey))
1318       neighbor->initiate_task = GNUNET_SCHEDULER_add_delayed (GNUNET_DV_CONSENSUS_FREQUENCY,
1319                                                               &initiate_set_union,
1320                                                               neighbor);
1321     break;
1322   case GNUNET_SET_STATUS_HALF_DONE:
1323     /* we got all of our updates; integrate routing table! */
1324     neighbor->target_removed = GNUNET_NO;
1325     GNUNET_CONTAINER_multihashmap_iterate (neighbor->neighbor_table,
1326                                            &check_target_removed,
1327                                            neighbor);
1328     if (GNUNET_YES == neighbor->target_removed)
1329     {
1330       /* check if we got an alternative for the removed routes */
1331       GNUNET_CONTAINER_multihashmap_iterate (direct_neighbors,
1332                                              &refresh_routes,
1333                                              NULL);    
1334     }
1335     /* add targets that appeared (and check for improved routes) */
1336     GNUNET_CONTAINER_multihashmap_iterate (neighbor->neighbor_table_consensus,
1337                                            &check_target_added,
1338                                            neighbor);
1339     if (NULL != neighbor->neighbor_table)
1340     {
1341       GNUNET_CONTAINER_multihashmap_iterate (neighbor->neighbor_table,
1342                                              &free_targets,
1343                                              NULL);
1344       GNUNET_CONTAINER_multihashmap_destroy (neighbor->neighbor_table);
1345       neighbor->neighbor_table = NULL;
1346     }
1347     neighbor->neighbor_table = neighbor->neighbor_table_consensus;
1348     neighbor->neighbor_table_consensus = NULL;
1349     break;
1350   case GNUNET_SET_STATUS_DONE:
1351     /* operation done, schedule next run! */
1352     GNUNET_SET_operation_cancel (neighbor->set_op); /* FIXME: needed? */
1353     neighbor->set_op = NULL;
1354     if (1 == GNUNET_CRYPTO_hash_cmp (&neighbor->peer.hashPubKey,
1355                                      &my_identity.hashPubKey))
1356       neighbor->initiate_task = GNUNET_SCHEDULER_add_delayed (GNUNET_DV_CONSENSUS_FREQUENCY,
1357                                                               &initiate_set_union,
1358                                                               neighbor);
1359     break;
1360   default:
1361     GNUNET_break (0);
1362     return;
1363   }
1364 }
1365
1366
1367 /**
1368  * Start creating a new DV set union construction, our neighbour has
1369  * asked for it (callback for listening peer).
1370  *
1371  * @param cls the 'struct DirectNeighbor' of the peer we're building
1372  *        a routing consensus with
1373  * @param other_peer the other peer
1374  * @param context_msg message with application specific information from
1375  *        the other peer
1376  * @param request request from the other peer, use GNUNET_SET_accept
1377  *        to accept it, otherwise the request will be refused
1378  *        Note that we don't use a return value here, as it is also
1379  *        necessary to specify the set we want to do the operation with,
1380  *        whith sometimes can be derived from the context message.
1381  *        Also necessary to specify the timeout.
1382  */    
1383 static void
1384 listen_set_union (void *cls,
1385                   const struct GNUNET_PeerIdentity *other_peer,
1386                   const struct GNUNET_MessageHeader *context_msg,
1387                   struct GNUNET_SET_Request *request)
1388 {
1389   struct DirectNeighbor *neighbor = cls;
1390
1391   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1392               "Starting to create consensus with %s!\n",
1393               GNUNET_i2s (&neighbor->peer));
1394   if (NULL != neighbor->set_op)
1395   {
1396     GNUNET_SET_operation_cancel (neighbor->set_op);
1397     neighbor->set_op = NULL;
1398   }
1399   if (NULL != neighbor->my_set)
1400   {
1401     GNUNET_SET_destroy (neighbor->my_set);
1402     neighbor->my_set = NULL;
1403   }
1404   neighbor->my_set = GNUNET_SET_create (cfg,
1405                                         GNUNET_SET_OPERATION_UNION);
1406   neighbor->set_op = GNUNET_SET_accept (request,
1407                                         neighbor->my_set /* FIXME: pass later! */,
1408                                         GNUNET_SET_RESULT_ADDED,
1409                                         &handle_set_union_result,
1410                                         neighbor);
1411   build_set (neighbor);
1412 }
1413
1414
1415 /**
1416  * Start creating a new DV set union by initiating the connection.
1417  *
1418  * @param cls the 'struct DirectNeighbor' of the peer we're building
1419  *        a routing consensus with
1420  * @param tc scheduler context
1421  */    
1422 static void
1423 initiate_set_union (void *cls,
1424                     const struct GNUNET_SCHEDULER_TaskContext *tc)
1425 {
1426   struct DirectNeighbor *neighbor = cls;
1427
1428   neighbor->initiate_task = GNUNET_SCHEDULER_NO_TASK;
1429   neighbor->my_set = GNUNET_SET_create (cfg,
1430                                         GNUNET_SET_OPERATION_UNION);
1431   neighbor->set_op = GNUNET_SET_evaluate (neighbor->my_set /* FIXME: pass later! */,
1432                                           &neighbor->peer,
1433                                           &neighbor->real_session_id,
1434                                           NULL,
1435                                           0 /* FIXME: salt */,
1436                                           GNUNET_SET_RESULT_ADDED,
1437                                           &handle_set_union_result,
1438                                           neighbor);
1439   build_set (neighbor);
1440 }
1441
1442
1443 /**
1444  * Core handler for DV data messages.  Whatever this message
1445  * contains all we really have to do is rip it out of its
1446  * DV layering and give it to our pal the DV plugin to report
1447  * in with.
1448  *
1449  * @param cls closure
1450  * @param peer peer which sent the message (immediate sender)
1451  * @param message the message
1452  * @return GNUNET_OK on success, GNUNET_SYSERR if the other peer violated the protocol
1453  */
1454 static int
1455 handle_dv_route_message (void *cls, const struct GNUNET_PeerIdentity *peer,
1456                          const struct GNUNET_MessageHeader *message)
1457 {
1458   const struct RouteMessage *rm;
1459   const struct GNUNET_MessageHeader *payload;
1460   struct Route *route;
1461
1462   if (ntohs (message->size) < sizeof (struct RouteMessage) + sizeof (struct GNUNET_MessageHeader))
1463   {
1464     GNUNET_break_op (0);
1465     return GNUNET_SYSERR;
1466   }
1467   rm = (const struct RouteMessage *) message;
1468   payload = (const struct GNUNET_MessageHeader *) &rm[1];
1469   if (ntohs (message->size) != sizeof (struct RouteMessage) + ntohs (payload->size))
1470   {
1471     GNUNET_break_op (0);
1472     return GNUNET_SYSERR;
1473   }
1474   if (0 == memcmp (&rm->target,
1475                    &my_identity,
1476                    sizeof (struct GNUNET_PeerIdentity)))
1477   {
1478     /* message is for me, check reverse route! */
1479     route = GNUNET_CONTAINER_multihashmap_get (all_routes,
1480                                                &rm->sender.hashPubKey);
1481     if (NULL == route)
1482     {
1483       /* don't have reverse route, drop */
1484       GNUNET_STATISTICS_update (stats,
1485                                 "# message discarded (no reverse route)",
1486                                 1, GNUNET_NO);
1487       return GNUNET_OK;
1488     }
1489     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1490                 "Delivering %u bytes to myself!\n",
1491                 ntohs (payload->size));
1492     send_data_to_plugin (payload,
1493                          &rm->sender,
1494                          ntohl (route->target.distance));
1495     return GNUNET_OK;
1496   }
1497   route = GNUNET_CONTAINER_multihashmap_get (all_routes,
1498                                              &rm->target.hashPubKey);
1499   if (NULL == route)
1500   {
1501     GNUNET_STATISTICS_update (stats,
1502                               "# messages discarded (no route)",
1503                               1, GNUNET_NO);
1504     return GNUNET_OK;
1505   }
1506   if (ntohl (route->target.distance) > ntohl (rm->distance) + 1)
1507   {
1508     GNUNET_STATISTICS_update (stats,
1509                               "# messages discarded (target too far)",
1510                               1, GNUNET_NO);
1511     return GNUNET_OK;
1512   }
1513   forward_payload (route->next_hop,
1514                    ntohl (route->target.distance),
1515                    0,
1516                    &rm->target,
1517                    &rm->sender,
1518                    payload);
1519   return GNUNET_OK;  
1520 }
1521
1522
1523 /**
1524  * Service server's handler for message send requests (which come
1525  * bubbling up to us through the DV plugin).
1526  *
1527  * @param cls closure
1528  * @param client identification of the client
1529  * @param message the actual message
1530  */
1531 static void
1532 handle_dv_send_message (void *cls, struct GNUNET_SERVER_Client *client,
1533                         const struct GNUNET_MessageHeader *message)
1534 {
1535   struct Route *route;
1536   const struct GNUNET_DV_SendMessage *msg;
1537   const struct GNUNET_MessageHeader *payload;
1538
1539   if (ntohs (message->size) < sizeof (struct GNUNET_DV_SendMessage) + sizeof (struct GNUNET_MessageHeader))
1540   {
1541     GNUNET_break (0);
1542     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1543     return;
1544   }
1545   msg = (const struct GNUNET_DV_SendMessage *) message;
1546   GNUNET_break (0 != ntohl (msg->uid));
1547   payload = (const struct GNUNET_MessageHeader *) &msg[1];
1548   if (ntohs (message->size) != sizeof (struct GNUNET_DV_SendMessage) + ntohs (payload->size))
1549   {
1550     GNUNET_break (0);
1551     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1552     return;
1553   }
1554   route = GNUNET_CONTAINER_multihashmap_get (all_routes,
1555                                              &msg->target.hashPubKey);
1556   if (NULL == route)
1557   {
1558     /* got disconnected */
1559     GNUNET_STATISTICS_update (stats,
1560                               "# local messages discarded (no route)",
1561                               1, GNUNET_NO);
1562     send_ack_to_plugin (&msg->target, ntohl (msg->uid), GNUNET_YES);
1563     GNUNET_SERVER_receive_done (client, GNUNET_OK);
1564     return;
1565   }
1566   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1567               "Forwarding %u bytes to %s\n",
1568               ntohs (payload->size),
1569               GNUNET_i2s (&msg->target));
1570
1571   forward_payload (route->next_hop,
1572                    ntohl (route->target.distance),
1573                    htonl (msg->uid),
1574                    &msg->target,
1575                    &my_identity,
1576                    payload);
1577   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1578 }
1579
1580
1581 /**
1582  * Cleanup all of the data structures associated with a given neighbor.
1583  *
1584  * @param neighbor neighbor to clean up
1585  */
1586 static void
1587 cleanup_neighbor (struct DirectNeighbor *neighbor)
1588 {
1589   struct PendingMessage *pending;
1590
1591   while (NULL != (pending = neighbor->pm_head))
1592   {
1593     neighbor->pm_queue_size--;
1594     GNUNET_CONTAINER_DLL_remove (neighbor->pm_head,
1595                                  neighbor->pm_tail,
1596                                  pending);    
1597     GNUNET_free (pending);
1598   }
1599   handle_direct_disconnect (neighbor);
1600   GNUNET_assert (GNUNET_YES ==
1601                  GNUNET_CONTAINER_multihashmap_remove (direct_neighbors, 
1602                                                        &neighbor->peer.hashPubKey,
1603                                                        neighbor));
1604   GNUNET_free (neighbor);
1605 }
1606
1607
1608 /**
1609  * Method called whenever a given peer disconnects.
1610  *
1611  * @param cls closure
1612  * @param peer peer identity this notification is about
1613  */
1614 static void
1615 handle_core_disconnect (void *cls, const struct GNUNET_PeerIdentity *peer)
1616 {
1617   struct DirectNeighbor *neighbor;
1618
1619   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1620               "Received core peer disconnect message for peer `%s'!\n",
1621               GNUNET_i2s (peer));
1622   /* Check for disconnect from self message */
1623   if (0 == memcmp (&my_identity, peer, sizeof (struct GNUNET_PeerIdentity)))
1624     return;
1625   neighbor =
1626       GNUNET_CONTAINER_multihashmap_get (direct_neighbors, &peer->hashPubKey);
1627   if (NULL == neighbor)
1628   {
1629     GNUNET_break (0);
1630     return;
1631   }
1632   GNUNET_break (GNUNET_YES == neighbor->connected);
1633   neighbor->connected = GNUNET_NO;
1634   if (DIRECT_NEIGHBOR_COST == neighbor->distance)
1635   {
1636     GNUNET_STATISTICS_update (stats,
1637                               "# peers connected (1-hop)",
1638                               -1, GNUNET_NO);  
1639   }
1640   cleanup_neighbor (neighbor);
1641   GNUNET_CONTAINER_multihashmap_iterate (direct_neighbors,
1642                                          &refresh_routes,
1643                                          NULL);
1644 }
1645
1646
1647 /**
1648  * Multihashmap iterator for freeing routes.  Should never be called.
1649  *
1650  * @param cls NULL
1651  * @param key key value stored under
1652  * @param value the route to be freed
1653  *
1654  * @return GNUNET_YES to continue iteration, GNUNET_NO to stop
1655  */
1656 static int
1657 free_route (void *cls, const struct GNUNET_HashCode * key, void *value)
1658 {
1659   struct Route *route = value;
1660
1661   GNUNET_break (0);
1662   GNUNET_assert (GNUNET_YES ==
1663                  GNUNET_CONTAINER_multihashmap_remove (all_routes, key, value));
1664   release_route (route);
1665   send_disconnect_to_plugin (&route->target.peer);
1666   GNUNET_free (route);
1667   return GNUNET_YES;
1668 }
1669
1670
1671 /**
1672  * Multihashmap iterator for freeing direct neighbors. Should never be called.
1673  *
1674  * @param cls NULL
1675  * @param key key value stored under
1676  * @param value the direct neighbor to be freed
1677  *
1678  * @return GNUNET_YES to continue iteration, GNUNET_NO to stop
1679  */
1680 static int
1681 free_direct_neighbors (void *cls, const struct GNUNET_HashCode * key, void *value)
1682 {
1683   struct DirectNeighbor *neighbor = value;
1684
1685   GNUNET_break (0);
1686   cleanup_neighbor (neighbor);
1687   return GNUNET_YES;
1688 }
1689
1690
1691 /**
1692  * Task run during shutdown.
1693  *
1694  * @param cls unused
1695  * @param tc unused
1696  */
1697 static void
1698 shutdown_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1699 {
1700   unsigned int i;
1701
1702   GNUNET_CORE_disconnect (core_api);
1703   core_api = NULL;
1704   GNUNET_ATS_performance_done (ats);
1705   ats = NULL;
1706   GNUNET_CONTAINER_multihashmap_iterate (direct_neighbors,
1707                                          &free_direct_neighbors, NULL);
1708   GNUNET_CONTAINER_multihashmap_iterate (all_routes,
1709                                          &free_route, NULL);
1710   GNUNET_CONTAINER_multihashmap_destroy (direct_neighbors);
1711   GNUNET_CONTAINER_multihashmap_destroy (all_routes);
1712   GNUNET_STATISTICS_destroy (stats, GNUNET_NO);
1713   stats = NULL;
1714   GNUNET_SERVER_notification_context_destroy (nc);
1715   nc = NULL;
1716   for (i=0;i<DEFAULT_FISHEYE_DEPTH - 1;i++)
1717     GNUNET_array_grow (consensi[i].targets,
1718                        consensi[i].array_length,
1719                        0);
1720 }
1721
1722
1723 /**
1724  * Notify newly connected client about an existing route.
1725  *
1726  * @param cls the 'struct GNUNET_SERVER_Client'
1727  * @param key peer identity
1728  * @param value the XXX.
1729  * @return GNUNET_OK (continue to iterate)
1730  */
1731 static int
1732 add_route (void *cls,
1733            const struct GNUNET_HashCode *key,
1734            void *value)
1735 {
1736   struct GNUNET_SERVER_Client *client = cls;
1737   struct Route *route = value;
1738   struct GNUNET_DV_ConnectMessage cm;
1739   
1740   cm.header.size = htons (sizeof (cm));
1741   cm.header.type = htons (GNUNET_MESSAGE_TYPE_DV_CONNECT);
1742   cm.distance = htonl (route->target.distance);
1743   cm.peer = route->target.peer;
1744
1745   GNUNET_SERVER_notification_context_unicast (nc, 
1746                                               client,
1747                                               &cm.header,
1748                                               GNUNET_NO);
1749   return GNUNET_OK;
1750 }
1751
1752
1753 /**
1754  * Handle START-message.  This is the first message sent to us
1755  * by the client (can only be one!).
1756  *
1757  * @param cls closure (always NULL)
1758  * @param client identification of the client
1759  * @param message the actual message
1760  */
1761 static void
1762 handle_start (void *cls, struct GNUNET_SERVER_Client *client,
1763               const struct GNUNET_MessageHeader *message)
1764 {
1765   GNUNET_SERVER_notification_context_add (nc, client);  
1766   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1767   GNUNET_CONTAINER_multihashmap_iterate (all_routes,
1768                                          &add_route,
1769                                          client);
1770 }
1771
1772
1773 /**
1774  * Called on core init.
1775  *
1776  * @param cls unused
1777  * @param server legacy
1778  * @param identity this peer's identity
1779  */
1780 static void
1781 core_init (void *cls, struct GNUNET_CORE_Handle *server,
1782            const struct GNUNET_PeerIdentity *identity)
1783 {
1784   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1785               "I am peer: %s\n",
1786               GNUNET_i2s (identity));
1787   my_identity = *identity;
1788 }
1789
1790
1791 /**
1792  * Process dv requests.
1793  *
1794  * @param cls closure
1795  * @param server the initialized server
1796  * @param c configuration to use
1797  */
1798 static void
1799 run (void *cls, struct GNUNET_SERVER_Handle *server,
1800      const struct GNUNET_CONFIGURATION_Handle *c)
1801 {
1802   static struct GNUNET_CORE_MessageHandler core_handlers[] = {
1803     {&handle_dv_route_message, GNUNET_MESSAGE_TYPE_DV_ROUTE, 0},
1804     {NULL, 0, 0}
1805   };
1806   static struct GNUNET_SERVER_MessageHandler plugin_handlers[] = {
1807     {&handle_start, NULL, 
1808      GNUNET_MESSAGE_TYPE_DV_START, 
1809      sizeof (struct GNUNET_MessageHeader) },
1810     { &handle_dv_send_message, NULL, 
1811       GNUNET_MESSAGE_TYPE_DV_SEND, 
1812       0},
1813     {NULL, NULL, 0, 0}
1814   };
1815
1816   cfg = c;
1817   direct_neighbors = GNUNET_CONTAINER_multihashmap_create (128, GNUNET_NO);
1818   all_routes = GNUNET_CONTAINER_multihashmap_create (65536, GNUNET_NO);
1819   core_api = GNUNET_CORE_connect (cfg, NULL,
1820                                   &core_init, 
1821                                   &handle_core_connect,
1822                                   &handle_core_disconnect,
1823                                   NULL, GNUNET_NO, 
1824                                   NULL, GNUNET_NO, 
1825                                   core_handlers);
1826
1827   if (NULL == core_api)
1828     return;
1829   ats = GNUNET_ATS_performance_init (cfg, &handle_ats_update, NULL);
1830   if (NULL == ats)
1831   {
1832     GNUNET_CORE_disconnect (core_api);
1833     return;
1834   }
1835   nc = GNUNET_SERVER_notification_context_create (server,
1836                                                   MAX_QUEUE_SIZE_PLUGIN);
1837   stats = GNUNET_STATISTICS_create ("dv", cfg);
1838   GNUNET_SERVER_add_handlers (server, plugin_handlers);
1839   GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL,
1840                                 &shutdown_task, NULL);
1841 }
1842
1843
1844 /**
1845  * The main function for the dv service.
1846  *
1847  * @param argc number of arguments from the command line
1848  * @param argv command line arguments
1849  * @return 0 ok, 1 on error
1850  */
1851 int
1852 main (int argc, char *const *argv)
1853 {
1854   return (GNUNET_OK ==
1855           GNUNET_SERVICE_run (argc, argv, "dv", GNUNET_SERVICE_OPTION_NONE,
1856                               &run, NULL)) ? 0 : 1;
1857 }
1858
1859 /* end of gnunet-service-dv.c */