- start test only after warmup
[oweals/gnunet.git] / src / mesh / gnunet-service-mesh_connection.c
1 /*
2      This file is part of GNUnet.
3      (C) 2001-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 mesh/gnunet-service-mesh_connection.c
23  * @brief GNUnet MESH service connection handling
24  * @author Bartlomiej Polot
25  */
26
27 #include "platform.h"
28 #include "gnunet_util_lib.h"
29
30 #include "gnunet_statistics_service.h"
31
32 #include "mesh_path.h"
33 #include "mesh_protocol.h"
34 #include "mesh.h"
35 #include "gnunet-service-mesh_connection.h"
36 #include "gnunet-service-mesh_peer.h"
37 #include "gnunet-service-mesh_tunnel.h"
38
39
40 #define LOG(level, ...) GNUNET_log_from (level,"mesh-con",__VA_ARGS__)
41
42 #define MESH_MAX_POLL_TIME      GNUNET_TIME_relative_multiply (\
43                                   GNUNET_TIME_UNIT_MINUTES,\
44                                   10)
45 #define AVG_MSGS                32
46
47
48 /******************************************************************************/
49 /********************************   STRUCTS  **********************************/
50 /******************************************************************************/
51
52 /**
53  * Struct to encapsulate all the Flow Control information to a peer to which
54  * we are directly connected (on a core level).
55  */
56 struct MeshFlowControl
57 {
58   /**
59    * Connection this controls.
60    */
61   struct MeshConnection *c;
62
63   /**
64    * How many messages are in the queue on this connection.
65    */
66   unsigned int queue_n;
67
68   /**
69    * How many messages do we accept in the queue.
70    */
71   unsigned int queue_max;
72
73   /**
74    * Next ID to use.
75    */
76   uint32_t next_pid;
77
78   /**
79    * ID of the last packet sent towards the peer.
80    */
81   uint32_t last_pid_sent;
82
83   /**
84    * ID of the last packet received from the peer.
85    */
86   uint32_t last_pid_recv;
87
88   /**
89    * Last ACK sent to the peer (peer can't send more than this PID).
90    */
91   uint32_t last_ack_sent;
92
93   /**
94    * Last ACK sent towards the origin (for traffic towards leaf node).
95    */
96   uint32_t last_ack_recv;
97
98   /**
99    * Task to poll the peer in case of a lost ACK causes stall.
100    */
101   GNUNET_SCHEDULER_TaskIdentifier poll_task;
102
103   /**
104    * How frequently to poll for ACKs.
105    */
106   struct GNUNET_TIME_Relative poll_time;
107
108   /**
109    * Queued poll message, to cancel if not necessary anymore (got ACK).
110    */
111   struct MeshConnectionQueue *poll_msg;
112
113   /**
114    * Queued poll message, to cancel if not necessary anymore (got ACK).
115    */
116   struct MeshConnectionQueue *ack_msg;
117 };
118
119 /**
120  * Keep a record of the last messages sent on this connection.
121  */
122 struct MeshConnectionPerformance
123 {
124   /**
125    * Circular buffer for storing measurements.
126    */
127   double usecsperbyte[AVG_MSGS];
128
129   /**
130    * Running average of @c usecsperbyte.
131    */
132   double avg;
133
134   /**
135    * How many values of @c usecsperbyte are valid.
136    */
137   uint16_t size;
138
139   /**
140    * Index of the next "free" position in @c usecsperbyte.
141    */
142   uint16_t idx;
143 };
144
145
146 /**
147  * Struct containing all information regarding a connection to a peer.
148  */
149 struct MeshConnection
150 {
151   /**
152    * Tunnel this connection is part of.
153    */
154   struct MeshTunnel3 *t;
155
156   /**
157    * Flow control information for traffic fwd.
158    */
159   struct MeshFlowControl fwd_fc;
160
161   /**
162    * Flow control information for traffic bck.
163    */
164   struct MeshFlowControl bck_fc;
165
166   /**
167    * Measure connection performance on the endpoint.
168    */
169   struct MeshConnectionPerformance *perf;
170
171   /**
172    * ID of the connection.
173    */
174   struct GNUNET_MeshHash id;
175
176   /**
177    * State of the connection.
178    */
179   enum MeshConnectionState state;
180
181   /**
182    * Path being used for the tunnel. At the origin of the connection
183    * it's a pointer to the destination's path pool, otherwise just a copy.
184    */
185   struct MeshPeerPath *path;
186
187   /**
188    * Position of the local peer in the path.
189    */
190   unsigned int own_pos;
191
192   /**
193    * Task to keep the used paths alive at the owner,
194    * time tunnel out on all the other peers.
195    */
196   GNUNET_SCHEDULER_TaskIdentifier fwd_maintenance_task;
197
198   /**
199    * Task to keep the used paths alive at the destination,
200    * time tunnel out on all the other peers.
201    */
202   GNUNET_SCHEDULER_TaskIdentifier bck_maintenance_task;
203
204   /**
205    * Queue handle for maintainance traffic. One handle for FWD and BCK since
206    * one peer never needs to maintain both directions (no loopback connections).
207    */
208   struct MeshPeerQueue *maintenance_q;
209
210   /**
211    * Counter to do exponential backoff when creating a connection (max 64).
212    */
213   unsigned short create_retry;
214
215   /**
216    * Pending message count.
217    */
218   int pending_messages;
219
220   /**
221    * Destroy flag: if true, destroy on last message.
222    */
223   int destroy;
224 };
225
226 /**
227  * Handle for messages queued but not yet sent.
228  */
229 struct MeshConnectionQueue
230 {
231   /**
232    * Peer queue handle, to cancel if necessary.
233    */
234   struct MeshPeerQueue *q;
235
236   /**
237    * Was this a forced message? (Do not account for it)
238    */
239   int forced;
240
241   /**
242    * Continuation to call once sent.
243    */
244   GMC_sent cont;
245
246   /**
247    * Closure for @c cont.
248    */
249   void *cont_cls;
250 };
251
252 /******************************************************************************/
253 /*******************************   GLOBALS  ***********************************/
254 /******************************************************************************/
255
256 /**
257  * Global handle to the statistics service.
258  */
259 extern struct GNUNET_STATISTICS_Handle *stats;
260
261 /**
262  * Local peer own ID (memory efficient handle).
263  */
264 extern GNUNET_PEER_Id myid;
265
266 /**
267  * Local peer own ID (full value).
268  */
269 extern struct GNUNET_PeerIdentity my_full_id;
270
271 /**
272  * Connections known, indexed by cid (MeshConnection).
273  */
274 static struct GNUNET_CONTAINER_MultiHashMap *connections;
275
276 /**
277  * How many connections are we willing to maintain.
278  * Local connections are always allowed, even if there are more connections than max.
279  */
280 static unsigned long long max_connections;
281
282 /**
283  * How many messages *in total* are we willing to queue, divide by number of
284  * connections to get connection queue size.
285  */
286 static unsigned long long max_msgs_queue;
287
288 /**
289  * How often to send path keepalives. Paths timeout after 4 missed.
290  */
291 static struct GNUNET_TIME_Relative refresh_connection_time;
292
293 /**
294  * How often to send path create / ACKs.
295  */
296 static struct GNUNET_TIME_Relative create_connection_time;
297
298
299 /******************************************************************************/
300 /********************************   STATIC  ***********************************/
301 /******************************************************************************/
302
303 #if 0 // avoid compiler warning for unused static function
304 static void
305 fc_debug (struct MeshFlowControl *fc)
306 {
307   LOG (GNUNET_ERROR_TYPE_DEBUG, "    IN: %u/%u\n",
308               fc->last_pid_recv, fc->last_ack_sent);
309   LOG (GNUNET_ERROR_TYPE_DEBUG, "    OUT: %u/%u\n",
310               fc->last_pid_sent, fc->last_ack_recv);
311   LOG (GNUNET_ERROR_TYPE_DEBUG, "    QUEUE: %u/%u\n",
312               fc->queue_n, fc->queue_max);
313 }
314
315 static void
316 connection_debug (struct MeshConnection *c)
317 {
318   if (NULL == c)
319   {
320     LOG (GNUNET_ERROR_TYPE_DEBUG, "*** DEBUG NULL CONNECTION ***\n");
321     return;
322   }
323   LOG (GNUNET_ERROR_TYPE_DEBUG, "Connection %s:%X\n",
324               peer2s (c->t->peer), GMC_2s (c));
325   LOG (GNUNET_ERROR_TYPE_DEBUG, "  state: %u, pending msgs: %u\n",
326               c->state, c->pending_messages);
327   LOG (GNUNET_ERROR_TYPE_DEBUG, "  FWD FC\n");
328   fc_debug (&c->fwd_fc);
329   LOG (GNUNET_ERROR_TYPE_DEBUG, "  BCK FC\n");
330   fc_debug (&c->bck_fc);
331 }
332 #endif
333
334
335 /**
336  * Schedule next keepalive task, taking in consideration
337  * the connection state and number of retries.
338  *
339  * @param c Connection for which to schedule the next keepalive.
340  * @param fwd Direction for the next keepalive.
341  */
342 static void
343 schedule_next_keepalive (struct MeshConnection *c, int fwd);
344
345
346 /**
347  * Resets the connection timeout task, some other message has done the
348  * task's job.
349  * - For the first peer on the direction this means to send
350  *   a keepalive or a path confirmation message (either create or ACK).
351  * - For all other peers, this means to destroy the connection,
352  *   due to lack of activity.
353  * Starts the timeout if no timeout was running (connection just created).
354  *
355  * @param c Connection whose timeout to reset.
356  * @param fwd Is this forward?
357  */
358 static void
359 connection_reset_timeout (struct MeshConnection *c, int fwd);
360
361
362 /**
363  * Get string description for tunnel state. Reentrant.
364  *
365  * @param s Tunnel state.
366  *
367  * @return String representation.
368  */
369 static const char *
370 GMC_state2s (enum MeshConnectionState s)
371 {
372   switch (s)
373   {
374     case MESH_CONNECTION_NEW:
375       return "MESH_CONNECTION_NEW";
376     case MESH_CONNECTION_SENT:
377       return "MESH_CONNECTION_SENT";
378     case MESH_CONNECTION_ACK:
379       return "MESH_CONNECTION_ACK";
380     case MESH_CONNECTION_READY:
381       return "MESH_CONNECTION_READY";
382     case MESH_CONNECTION_DESTROYED:
383       return "MESH_CONNECTION_DESTROYED";
384     default:
385       return "MESH_CONNECTION_STATE_ERROR";
386   }
387 }
388
389
390 /**
391  * Initialize a Flow Control structure to the initial state.
392  *
393  * @param fc Flow Control structure to initialize.
394  */
395 static void
396 fc_init (struct MeshFlowControl *fc)
397 {
398   fc->next_pid = 0;
399   fc->last_pid_sent = (uint32_t) -1; /* Next (expected) = 0 */
400   fc->last_pid_recv = (uint32_t) -1;
401   fc->last_ack_sent = (uint32_t) 0;
402   fc->last_ack_recv = (uint32_t) 0;
403   fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
404   fc->poll_time = GNUNET_TIME_UNIT_SECONDS;
405   fc->queue_n = 0;
406   fc->queue_max = (max_msgs_queue / max_connections) + 1;
407 }
408
409
410 /**
411  * Find a connection.
412  *
413  * @param cid Connection ID.
414  */
415 static struct MeshConnection *
416 connection_get (const struct GNUNET_MeshHash *cid)
417 {
418   return GNUNET_CONTAINER_multihashmap_get (connections, GM_h2hc (cid));
419 }
420
421
422 static void
423 connection_change_state (struct MeshConnection* c,
424                          enum MeshConnectionState state)
425 {
426   LOG (GNUNET_ERROR_TYPE_DEBUG,
427        "Connection %s state %s -> %s\n",
428        GMC_2s (c), GMC_state2s (c->state), GMC_state2s (state));
429   if (MESH_CONNECTION_DESTROYED == c->state)
430   {
431     LOG (GNUNET_ERROR_TYPE_DEBUG, "state not changing anymore\n");
432     return;
433   }
434   c->state = state;
435   if (MESH_CONNECTION_READY == state)
436     c->create_retry = 1;
437 }
438
439
440 /**
441  * Callback called when a queued ACK message is sent.
442  *
443  * @param cls Closure (FC).
444  * @param c Connection this message was on.
445  * @param q Queue handler this call invalidates.
446  * @param type Type of message sent.
447  * @param fwd Was this a FWD going message?
448  * @param size Size of the message.
449  */
450 static void
451 ack_sent (void *cls,
452           struct MeshConnection *c,
453           struct MeshConnectionQueue *q,
454           uint16_t type, int fwd, size_t size)
455 {
456   struct MeshFlowControl *fc = cls;
457
458   fc->ack_msg = NULL;
459 }
460
461
462 /**
463  * Send an ACK on the connection, informing the predecessor about
464  * the available buffer space. Should not be called in case the peer
465  * is origin (no predecessor) in the @c fwd direction.
466  *
467  * Note that for fwd ack, the FWD mean forward *traffic* (root->dest),
468  * the ACK itself goes "back" (dest->root).
469  *
470  * @param c Connection on which to send the ACK.
471  * @param buffer How much space free to advertise?
472  * @param fwd Is this FWD ACK? (Going dest -> root)
473  * @param force Don't optimize out.
474  */
475 static void
476 send_ack (struct MeshConnection *c, unsigned int buffer, int fwd, int force)
477 {
478   struct MeshFlowControl *next_fc;
479   struct MeshFlowControl *prev_fc;
480   struct GNUNET_MESH_ACK msg;
481   uint32_t ack;
482   int delta;
483
484   /* If origin, there is no connection to send ACKs. Wrong function! */
485   if (GMC_is_origin (c, fwd))
486   {
487     LOG (GNUNET_ERROR_TYPE_DEBUG, "connection %s is origin in %s\n",
488          GMC_2s (c), GM_f2s (fwd));
489     GNUNET_break (0);
490     return;
491   }
492
493   next_fc = fwd ? &c->fwd_fc : &c->bck_fc;
494   prev_fc = fwd ? &c->bck_fc : &c->fwd_fc;
495
496   LOG (GNUNET_ERROR_TYPE_DEBUG, "connection send %s ack on %s\n",
497        GM_f2s (fwd), GMC_2s (c));
498
499   /* Check if we need to transmit the ACK. */
500   delta = prev_fc->last_ack_sent - prev_fc->last_pid_recv;
501   if (3 < delta && buffer < delta && GNUNET_NO == force)
502   {
503     LOG (GNUNET_ERROR_TYPE_DEBUG, "Not sending ACK, buffer > 3\n");
504     LOG (GNUNET_ERROR_TYPE_DEBUG,
505          "  last pid recv: %u, last ack sent: %u\n",
506          prev_fc->last_pid_recv, prev_fc->last_ack_sent);
507     return;
508   }
509
510   /* Ok, ACK might be necessary, what PID to ACK? */
511   ack = prev_fc->last_pid_recv + buffer;
512   LOG (GNUNET_ERROR_TYPE_DEBUG, " ACK %u\n", ack);
513   LOG (GNUNET_ERROR_TYPE_DEBUG,
514        " last pid %u, last ack %u, qmax %u, q %u\n",
515        prev_fc->last_pid_recv, prev_fc->last_ack_sent,
516        next_fc->queue_max, next_fc->queue_n);
517   if (ack == prev_fc->last_ack_sent && GNUNET_NO == force)
518   {
519     LOG (GNUNET_ERROR_TYPE_DEBUG, "Not sending FWD ACK, not needed\n");
520     return;
521   }
522
523   /* Check if message is already in queue */
524   if (NULL != prev_fc->ack_msg)
525   {
526     if (GM_is_pid_bigger (ack, prev_fc->last_ack_sent))
527     {
528       LOG (GNUNET_ERROR_TYPE_DEBUG, " canceling old ACK\n");
529       GMC_cancel (prev_fc->ack_msg);
530       /* GMC_cancel triggers ack_sent(), which clears fc->ack_msg */
531     }
532     else
533     {
534       LOG (GNUNET_ERROR_TYPE_DEBUG, " same ACK already in queue\n");
535       return;
536     }
537   }
538
539   prev_fc->last_ack_sent = ack;
540
541   /* Build ACK message and send on connection */
542   msg.header.size = htons (sizeof (msg));
543   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_ACK);
544   msg.ack = htonl (ack);
545   msg.cid = c->id;
546
547   prev_fc->ack_msg = GMC_send_prebuilt_message (&msg.header, c,
548                                                 !fwd, GNUNET_YES,
549                                                 &ack_sent, prev_fc);
550 }
551
552
553 /**
554  * Callback called when a queued message is sent.
555  *
556  * Calculates the average time and connection packet tracking.
557  *
558  * @param cls Closure (ConnectionQueue Handle).
559  * @param c Connection this message was on.
560  * @param type Type of message sent.
561  * @param fwd Was this a FWD going message?
562  * @param size Size of the message.
563  * @param wait Time spent waiting for core (only the time for THIS message)
564  */
565 static void
566 message_sent (void *cls,
567               struct MeshConnection *c, uint16_t type,
568               int fwd, size_t size,
569               struct GNUNET_TIME_Relative wait)
570 {
571   struct MeshConnectionPerformance *p;
572   struct MeshFlowControl *fc;
573   struct MeshConnectionQueue *q = cls;
574   double usecsperbyte;
575   int forced;
576
577   fc = fwd ? &c->fwd_fc : &c->bck_fc;
578   LOG (GNUNET_ERROR_TYPE_DEBUG,
579        "!  sent %s %s\n",
580        GM_f2s (fwd),
581        GM_m2s (type));
582   LOG (GNUNET_ERROR_TYPE_DEBUG, "!  C_P- %p %u\n", c, c->pending_messages);
583   if (NULL != q)
584   {
585     forced = q->forced;
586     if (NULL != q->cont)
587     {
588       LOG (GNUNET_ERROR_TYPE_DEBUG, "!  calling cont\n");
589       q->cont (q->cont_cls, c, q, type, fwd, size);
590     }
591     GNUNET_free (q);
592   }
593   else if (type == GNUNET_MESSAGE_TYPE_MESH_ENCRYPTED)
594   {
595     /* If NULL == q and ENCRYPTED == type, message must have been ch_mngmnt */
596     forced = GNUNET_YES;
597   }
598   else
599   {
600     forced = GNUNET_NO;
601   }
602   c->pending_messages--;
603   if (GNUNET_YES == c->destroy && 0 == c->pending_messages)
604   {
605     LOG (GNUNET_ERROR_TYPE_DEBUG, "!  destroying connection!\n");
606     GMC_destroy (c);
607     return;
608   }
609   /* Send ACK if needed, after accounting for sent ID in fc->queue_n */
610   switch (type)
611   {
612     case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE:
613     case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK:
614       c->maintenance_q = NULL;
615       /* Don't trigger a keepalive for sent ACKs, only SYN and SYNACKs */
616       if (GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE == type || !fwd)
617         schedule_next_keepalive (c, fwd);
618       break;
619
620     case GNUNET_MESSAGE_TYPE_MESH_ENCRYPTED:
621       fc->last_pid_sent++;
622       LOG (GNUNET_ERROR_TYPE_DEBUG, "!  Q_N- %p %u\n", fc, fc->queue_n);
623       if (GNUNET_NO == forced)
624       {
625         fc->queue_n--;
626         LOG (GNUNET_ERROR_TYPE_DEBUG,
627             "!   accounting pid %u\n",
628             fc->last_pid_sent);
629       }
630       else
631       {
632         LOG (GNUNET_ERROR_TYPE_DEBUG,
633              "!   forced, Q_N not accounting pid %u\n",
634              fc->last_pid_sent);
635       }
636       GMC_send_ack (c, fwd, GNUNET_NO);
637       connection_reset_timeout (c, fwd);
638       break;
639
640     case GNUNET_MESSAGE_TYPE_MESH_POLL:
641       fc->poll_msg = NULL;
642       break;
643
644     case GNUNET_MESSAGE_TYPE_MESH_ACK:
645       fc->ack_msg = NULL;
646       break;
647
648     default:
649       break;
650   }
651   LOG (GNUNET_ERROR_TYPE_DEBUG, "!  message sent!\n");
652
653   if (NULL == c->perf)
654     return; /* Only endpoints are interested in timing. */
655
656   p = c->perf;
657   usecsperbyte = ((double) wait.rel_value_us) / size;
658   if (p->size == AVG_MSGS)
659   {
660     /* Array is full. Substract oldest value, add new one and store. */
661     p->avg -= (p->usecsperbyte[p->idx] / AVG_MSGS);
662     p->usecsperbyte[p->idx] = usecsperbyte;
663     p->avg += (p->usecsperbyte[p->idx] / AVG_MSGS);
664   }
665   else
666   {
667     /* Array not yet full. Add current value to avg and store. */
668     p->usecsperbyte[p->idx] = usecsperbyte;
669     p->avg *= p->size;
670     p->avg += p->usecsperbyte[p->idx];
671     p->size++;
672     p->avg /= p->size;
673   }
674   p->idx = (p->idx + 1) % AVG_MSGS;
675 }
676
677
678 /**
679  * Get the previous hop in a connection
680  *
681  * @param c Connection.
682  *
683  * @return Previous peer in the connection.
684  */
685 static struct MeshPeer *
686 get_prev_hop (const struct MeshConnection *c)
687 {
688   GNUNET_PEER_Id id;
689
690   LOG (GNUNET_ERROR_TYPE_DEBUG, " get prev hop %s [%u/%u]\n",
691        GMC_2s (c), c->own_pos, c->path->length);
692   if (0 == c->own_pos || c->path->length < 2)
693     id = c->path->peers[0];
694   else
695     id = c->path->peers[c->own_pos - 1];
696
697   LOG (GNUNET_ERROR_TYPE_DEBUG, "  ID: %s (%u)\n",
698        GNUNET_i2s (GNUNET_PEER_resolve2 (id)), id);
699
700   return GMP_get_short (id);
701 }
702
703
704 /**
705  * Get the next hop in a connection
706  *
707  * @param c Connection.
708  *
709  * @return Next peer in the connection.
710  */
711 static struct MeshPeer *
712 get_next_hop (const struct MeshConnection *c)
713 {
714   GNUNET_PEER_Id id;
715
716   LOG (GNUNET_ERROR_TYPE_DEBUG, " get next hop %s [%u/%u]\n",
717        GMC_2s (c), c->own_pos, c->path->length);
718   if ((c->path->length - 1) == c->own_pos || c->path->length < 2)
719     id = c->path->peers[c->path->length - 1];
720   else
721     id = c->path->peers[c->own_pos + 1];
722
723   LOG (GNUNET_ERROR_TYPE_DEBUG, "  ID: %s (%u)\n",
724        GNUNET_i2s (GNUNET_PEER_resolve2 (id)), id);
725
726   return GMP_get_short (id);
727 }
728
729
730 /**
731  * Get the hop in a connection.
732  *
733  * @param c Connection.
734  * @param fwd Next hop?
735  *
736  * @return Next peer in the connection.
737  */
738 static struct MeshPeer *
739 get_hop (struct MeshConnection *c, int fwd)
740 {
741   if (fwd)
742     return get_next_hop (c);
743   return get_prev_hop (c);
744 }
745
746
747 /**
748  * Is traffic coming from this sender 'FWD' traffic?
749  *
750  * @param c Connection to check.
751  * @param sender Peer identity of neighbor.
752  *
753  * @return #GNUNET_YES in case the sender is the 'prev' hop and therefore
754  *         the traffic is 'FWD'.
755  *         #GNUNET_NO for BCK.
756  *         #GNUNET_SYSERR for errors.
757  */
758 static int
759 is_fwd (const struct MeshConnection *c,
760         const struct GNUNET_PeerIdentity *sender)
761 {
762   GNUNET_PEER_Id id;
763
764   id = GNUNET_PEER_search (sender);
765   if (GMP_get_short_id (get_prev_hop (c)) == id)
766     return GNUNET_YES;
767
768   if (GMP_get_short_id (get_next_hop (c)) == id)
769     return GNUNET_NO;
770
771   GNUNET_break (0);
772   return GNUNET_SYSERR;
773 }
774
775
776 /**
777  * Sends a CONNECTION ACK message in reponse to a received CONNECTION_CREATE
778  * or a first CONNECTION_ACK directed to us.
779  *
780  * @param connection Connection to confirm.
781  * @param fwd Should we send it FWD? (root->dest)
782  *            (First (~SYNACK) goes BCK, second (~ACK) goes FWD)
783  */
784 static void
785 send_connection_ack (struct MeshConnection *connection, int fwd)
786 {
787   struct MeshTunnel3 *t;
788
789   t = connection->t;
790   LOG (GNUNET_ERROR_TYPE_INFO, "=> {%18s ACK} on connection %s\n",
791        GM_f2s (!fwd), GMC_2s (connection));
792   GMP_queue_add (get_hop (connection, fwd), NULL,
793                  GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK,
794                  sizeof (struct GNUNET_MESH_ConnectionACK),
795                  connection, fwd, &message_sent, NULL);
796   connection->pending_messages++;
797   if (MESH_TUNNEL3_NEW == GMT_get_cstate (t))
798     GMT_change_cstate (t, MESH_TUNNEL3_WAITING);
799   if (MESH_CONNECTION_READY != connection->state)
800     connection_change_state (connection, MESH_CONNECTION_SENT);
801 }
802
803
804 /**
805  * Send a notification that a connection is broken.
806  *
807  * @param c Connection that is broken.
808  * @param id1 Peer that has disconnected.
809  * @param id2 Peer that has disconnected.
810  * @param fwd Direction towards which to send it.
811  */
812 static void
813 send_broken (struct MeshConnection *c,
814              const struct GNUNET_PeerIdentity *id1,
815              const struct GNUNET_PeerIdentity *id2,
816              int fwd)
817 {
818   struct GNUNET_MESH_ConnectionBroken msg;
819
820   msg.header.size = htons (sizeof (struct GNUNET_MESH_ConnectionBroken));
821   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_CONNECTION_BROKEN);
822   msg.cid = c->id;
823   msg.peer1 = *id1;
824   msg.peer2 = *id2;
825   GMC_send_prebuilt_message (&msg.header, c, fwd, GNUNET_YES, NULL, NULL);
826 }
827
828
829 /**
830  * Send a notification that a connection is broken, when a connection
831  * isn't even created.
832  *
833  * @param connection_id Connection ID.
834  * @param id1 Peer that has disconnected.
835  * @param id2 Peer that has disconnected.
836  * @param peer Peer to notify (neighbor who sent the connection).
837  */
838 static void
839 send_broken2 (struct GNUNET_MeshHash *connection_id,
840              const struct GNUNET_PeerIdentity *id1,
841              const struct GNUNET_PeerIdentity *id2,
842              GNUNET_PEER_Id peer_id)
843 {
844   struct GNUNET_MESH_ConnectionBroken *msg;
845   struct MeshPeer *neighbor;
846
847   LOG (GNUNET_ERROR_TYPE_INFO, "=> BROKEN on unknown connection %s\n",
848        GNUNET_h2s (GM_h2hc (connection_id)));
849
850   msg = GNUNET_new (struct GNUNET_MESH_ConnectionBroken);
851   msg->header.size = htons (sizeof (struct GNUNET_MESH_ConnectionBroken));
852   msg->header.type = htons (GNUNET_MESSAGE_TYPE_MESH_CONNECTION_BROKEN);
853   msg->cid = *connection_id;
854   msg->peer1 = *id1;
855   msg->peer2 = *id2;
856   neighbor = GMP_get_short (peer_id);
857   GMP_queue_add (neighbor, msg,
858                  GNUNET_MESSAGE_TYPE_MESH_ENCRYPTED,
859                  sizeof (struct GNUNET_MESH_ConnectionBroken),
860                  NULL, GNUNET_SYSERR, /* connection, fwd */
861                  NULL, NULL); /* continuation */
862 }
863
864
865 /**
866  * Send keepalive packets for a connection.
867  *
868  * @param c Connection to keep alive..
869  * @param fwd Is this a FWD keepalive? (owner -> dest).
870  */
871 static void
872 send_connection_keepalive (struct MeshConnection *c, int fwd)
873 {
874   struct GNUNET_MessageHeader msg;
875
876   LOG (GNUNET_ERROR_TYPE_INFO,
877        "keepalive %s for connection %s\n",
878        GM_f2s (fwd), GMC_2s (c));
879   GNUNET_STATISTICS_update (stats, "# keepalives sent", 1, GNUNET_NO);
880
881   GNUNET_assert (NULL != c->t);
882   msg.size = htons (sizeof (msg));
883   msg.type = htons (GNUNET_MESSAGE_TYPE_MESH_KEEPALIVE);
884
885   GNUNET_assert (NULL ==
886                  GMT_send_prebuilt_message (&msg, c->t, c,
887                                             GNUNET_NO, NULL, NULL));
888 }
889
890
891 /**
892  * Send CONNECTION_{CREATE/ACK} packets for a connection.
893  *
894  * @param c Connection for which to send the message.
895  * @param fwd If #GNUNET_YES, send CREATE, otherwise send ACK.
896  */
897 static void
898 connection_recreate (struct MeshConnection *c, int fwd)
899 {
900   LOG (GNUNET_ERROR_TYPE_DEBUG, "sending connection recreate\n");
901   if (fwd)
902     GMC_send_create (c);
903   else
904     send_connection_ack (c, GNUNET_NO);
905 }
906
907
908 /**
909  * Generic connection timer management.
910  * Depending on the role of the peer in the connection will send the
911  * appropriate message (build or keepalive)
912  *
913  * @param c Conncetion to maintain.
914  * @param fwd Is FWD?
915  */
916 static void
917 connection_maintain (struct MeshConnection *c, int fwd)
918 {
919   if (GNUNET_NO != c->destroy)
920     return;
921
922   if (MESH_TUNNEL3_SEARCHING == GMT_get_cstate (c->t))
923   {
924     /* TODO DHT GET with RO_BART */
925     return;
926   }
927   switch (c->state)
928   {
929     case MESH_CONNECTION_NEW:
930       GNUNET_break (0);
931       /* fall-through */
932     case MESH_CONNECTION_SENT:
933       connection_recreate (c, fwd);
934       break;
935     case MESH_CONNECTION_READY:
936       send_connection_keepalive (c, fwd);
937       break;
938     default:
939       break;
940   }
941 }
942
943
944
945 /**
946  * Keep the connection alive.
947  *
948  * @param c Connection to keep alive.
949  * @param fwd Direction.
950  * @param shutdown Are we shutting down? (Don't send traffic)
951  *                 Non-zero value for true, not necessarily GNUNET_YES.
952  */
953 static void
954 connection_keepalive (struct MeshConnection *c, int fwd, int shutdown)
955 {
956   LOG (GNUNET_ERROR_TYPE_DEBUG, "%s keepalive for %s\n",
957        GM_f2s (fwd), GMC_2s (c));
958
959   if (fwd)
960     c->fwd_maintenance_task = GNUNET_SCHEDULER_NO_TASK;
961   else
962     c->bck_maintenance_task = GNUNET_SCHEDULER_NO_TASK;
963
964   if (GNUNET_NO != shutdown)
965     return;
966
967   connection_maintain (c, fwd);
968
969   /* Next execution will be scheduled by message_sent */
970 }
971
972
973 /**
974  * Keep the connection alive in the FWD direction.
975  *
976  * @param cls Closure (connection to keepalive).
977  * @param tc TaskContext.
978  */
979 static void
980 connection_fwd_keepalive (void *cls,
981                           const struct GNUNET_SCHEDULER_TaskContext *tc)
982 {
983   connection_keepalive ((struct MeshConnection *) cls,
984                         GNUNET_YES,
985                         tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN);
986 }
987
988
989 /**
990  * Keep the connection alive in the BCK direction.
991  *
992  * @param cls Closure (connection to keepalive).
993  * @param tc TaskContext.
994  */
995 static void
996 connection_bck_keepalive (void *cls,
997                           const struct GNUNET_SCHEDULER_TaskContext *tc)
998 {
999   connection_keepalive ((struct MeshConnection *) cls,
1000                         GNUNET_NO,
1001                         tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN);
1002 }
1003
1004
1005 /**
1006  * Schedule next keepalive task, taking in consideration
1007  * the connection state and number of retries.
1008  *
1009  * If the peer is not the origin, do nothing.
1010  *
1011  * @param c Connection for which to schedule the next keepalive.
1012  * @param fwd Direction for the next keepalive.
1013  */
1014 static void
1015 schedule_next_keepalive (struct MeshConnection *c, int fwd)
1016 {
1017   struct GNUNET_TIME_Relative delay;
1018   GNUNET_SCHEDULER_TaskIdentifier *task_id;
1019   GNUNET_SCHEDULER_Task keepalive_task;
1020
1021   if (GNUNET_NO == GMC_is_origin (c, fwd))
1022     return;
1023
1024   /* Calculate delay to use, depending on the state of the connection */
1025   if (MESH_CONNECTION_READY == c->state)
1026   {
1027     delay = refresh_connection_time;
1028   }
1029   else
1030   {
1031     if (1 > c->create_retry)
1032       c->create_retry = 1;
1033     delay = GNUNET_TIME_relative_multiply (create_connection_time,
1034                                            c->create_retry);
1035     if (c->create_retry < 64)
1036       c->create_retry *= 2;
1037   }
1038
1039   /* Select direction-dependent parameters */
1040   if (GNUNET_YES == fwd)
1041   {
1042     task_id = &c->fwd_maintenance_task;
1043     keepalive_task = &connection_fwd_keepalive;
1044   }
1045   else
1046   {
1047     task_id = &c->bck_maintenance_task;
1048     keepalive_task = &connection_bck_keepalive;
1049   }
1050
1051   /* Check that no one scheduled it before us */
1052   if (GNUNET_SCHEDULER_NO_TASK != *task_id)
1053   {
1054     /* No need for a _break. It can happen for instance when sending a SYNACK
1055      * for a duplicate SYN: the first SYNACK scheduled the task. */
1056     GNUNET_SCHEDULER_cancel (*task_id);
1057   }
1058
1059   /* Schedule the task */
1060   *task_id = GNUNET_SCHEDULER_add_delayed (delay, keepalive_task, c);
1061   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "next keepalive in %s\n",
1062               GNUNET_STRINGS_relative_time_to_string (delay, GNUNET_YES));
1063 }
1064
1065
1066 /**
1067  * @brief Re-initiate traffic on this connection if necessary.
1068  *
1069  * Check if there is traffic queued towards this peer
1070  * and the core transmit handle is NULL (traffic was stalled).
1071  * If so, call core tmt rdy.
1072  *
1073  * @param c Connection on which initiate traffic.
1074  * @param fwd Is this about fwd traffic?
1075  */
1076 static void
1077 connection_unlock_queue (struct MeshConnection *c, int fwd)
1078 {
1079   struct MeshPeer *peer;
1080
1081   LOG (GNUNET_ERROR_TYPE_DEBUG,
1082               "connection_unlock_queue %s on %s\n",
1083               GM_f2s (fwd), GMC_2s (c));
1084
1085   if (GMC_is_terminal (c, fwd))
1086   {
1087     LOG (GNUNET_ERROR_TYPE_DEBUG, " is terminal!\n");
1088     return;
1089   }
1090
1091   peer = get_hop (c, fwd);
1092   GMP_queue_unlock (peer, c);
1093 }
1094
1095
1096 /**
1097  * Cancel all transmissions that belong to a certain connection.
1098  *
1099  * If the connection is scheduled for destruction and no more messages are left,
1100  * the connection will be destroyed by the continuation call.
1101  *
1102  * @param c Connection which to cancel. Might be destroyed during this call.
1103  * @param fwd Cancel fwd traffic?
1104  */
1105 static void
1106 connection_cancel_queues (struct MeshConnection *c, int fwd)
1107 {
1108   struct MeshFlowControl *fc;
1109   struct MeshPeer *peer;
1110
1111   LOG (GNUNET_ERROR_TYPE_DEBUG,
1112        " *** Cancel %s queues for connection %s\n",
1113        GM_f2s (fwd), GMC_2s (c));
1114   if (NULL == c)
1115   {
1116     GNUNET_break (0);
1117     return;
1118   }
1119
1120   fc = fwd ? &c->fwd_fc : &c->bck_fc;
1121   if (GNUNET_SCHEDULER_NO_TASK != fc->poll_task)
1122   {
1123     GNUNET_SCHEDULER_cancel (fc->poll_task);
1124     fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
1125     LOG (GNUNET_ERROR_TYPE_DEBUG, " *** Cancel POLL in ccq for fc %p\n", fc);
1126   }
1127   peer = get_hop (c, fwd);
1128   GMP_queue_cancel (peer, c);
1129 }
1130
1131
1132 /**
1133  * Function called if a connection has been stalled for a while,
1134  * possibly due to a missed ACK. Poll the neighbor about its ACK status.
1135  *
1136  * @param cls Closure (poll ctx).
1137  * @param tc TaskContext.
1138  */
1139 static void
1140 connection_poll (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
1141
1142
1143 /**
1144  * Callback called when a queued POLL message is sent.
1145  *
1146  * @param cls Closure (FC).
1147  * @param c Connection this message was on.
1148  * @param q Queue handler this call invalidates.
1149  * @param type Type of message sent.
1150  * @param fwd Was this a FWD going message?
1151  * @param size Size of the message.
1152  */
1153 static void
1154 poll_sent (void *cls,
1155            struct MeshConnection *c,
1156            struct MeshConnectionQueue *q,
1157            uint16_t type, int fwd, size_t size)
1158 {
1159   struct MeshFlowControl *fc = cls;
1160
1161   if (2 == c->destroy)
1162   {
1163     LOG (GNUNET_ERROR_TYPE_DEBUG, " *** POLL canceled on shutdown\n");
1164     return;
1165   }
1166   LOG (GNUNET_ERROR_TYPE_DEBUG,
1167        " *** POLL sent for , scheduling new one!\n");
1168   fc->poll_msg = NULL;
1169   fc->poll_time = GNUNET_TIME_STD_BACKOFF (fc->poll_time);
1170   fc->poll_task = GNUNET_SCHEDULER_add_delayed (fc->poll_time,
1171                                                 &connection_poll, fc);
1172   LOG (GNUNET_ERROR_TYPE_DEBUG, " task %u\n", fc->poll_task);
1173
1174 }
1175
1176 /**
1177  * Function called if a connection has been stalled for a while,
1178  * possibly due to a missed ACK. Poll the neighbor about its ACK status.
1179  *
1180  * @param cls Closure (poll ctx).
1181  * @param tc TaskContext.
1182  */
1183 static void
1184 connection_poll (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1185 {
1186   struct MeshFlowControl *fc = cls;
1187   struct GNUNET_MESH_Poll msg;
1188   struct MeshConnection *c;
1189
1190   fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
1191   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1192   {
1193     return;
1194   }
1195
1196   c = fc->c;
1197   LOG (GNUNET_ERROR_TYPE_DEBUG, " *** Polling!\n");
1198   LOG (GNUNET_ERROR_TYPE_DEBUG, " *** connection [%s]\n", GMC_2s (c));
1199   LOG (GNUNET_ERROR_TYPE_DEBUG, " ***   %s\n",
1200        fc == &c->fwd_fc ? "FWD" : "BCK");
1201
1202   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_POLL);
1203   msg.header.size = htons (sizeof (msg));
1204   msg.pid = htonl (fc->last_pid_sent);
1205   LOG (GNUNET_ERROR_TYPE_DEBUG, " *** last pid sent: %u!\n", fc->last_pid_sent);
1206   fc->poll_msg = GMC_send_prebuilt_message (&msg.header, c,
1207                                             fc == &c->fwd_fc, GNUNET_YES,
1208                                             &poll_sent, fc);
1209 }
1210
1211
1212 /**
1213  * Timeout function due to lack of keepalive/traffic from the owner.
1214  * Destroys connection if called.
1215  *
1216  * @param cls Closure (connection to destroy).
1217  * @param tc TaskContext.
1218  */
1219 static void
1220 connection_fwd_timeout (void *cls,
1221                         const struct GNUNET_SCHEDULER_TaskContext *tc)
1222 {
1223   struct MeshConnection *c = cls;
1224
1225   c->fwd_maintenance_task = GNUNET_SCHEDULER_NO_TASK;
1226   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1227     return;
1228
1229   LOG (GNUNET_ERROR_TYPE_DEBUG, "Connection %s FWD timed out. Destroying.\n",
1230        GMC_2s (c));
1231   if (GMC_is_origin (c, GNUNET_YES)) /* If local, leave. */
1232   {
1233     GNUNET_break (0);
1234     return;
1235   }
1236
1237   GMC_destroy (c);
1238 }
1239
1240
1241 /**
1242  * Timeout function due to lack of keepalive/traffic from the destination.
1243  * Destroys connection if called.
1244  *
1245  * @param cls Closure (connection to destroy).
1246  * @param tc TaskContext
1247  */
1248 static void
1249 connection_bck_timeout (void *cls,
1250                         const struct GNUNET_SCHEDULER_TaskContext *tc)
1251 {
1252   struct MeshConnection *c = cls;
1253
1254   c->bck_maintenance_task = GNUNET_SCHEDULER_NO_TASK;
1255   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1256     return;
1257
1258   LOG (GNUNET_ERROR_TYPE_DEBUG, "Connection %s BCK timed out. Destroying.\n",
1259        GMC_2s (c));
1260
1261   if (GMC_is_origin (c, GNUNET_NO)) /* If local, leave. */
1262   {
1263     GNUNET_break (0);
1264     return;
1265   }
1266
1267   GMC_destroy (c);
1268 }
1269
1270
1271 /**
1272  * Resets the connection timeout task, some other message has done the
1273  * task's job.
1274  * - For the first peer on the direction this means to send
1275  *   a keepalive or a path confirmation message (either create or ACK).
1276  * - For all other peers, this means to destroy the connection,
1277  *   due to lack of activity.
1278  * Starts the timeout if no timeout was running (connection just created).
1279  *
1280  * @param c Connection whose timeout to reset.
1281  * @param fwd Is this forward?
1282  *
1283  * TODO use heap to improve efficiency of scheduler.
1284  */
1285 static void
1286 connection_reset_timeout (struct MeshConnection *c, int fwd)
1287 {
1288   LOG (GNUNET_ERROR_TYPE_DEBUG, "Connection %s reset timeout\n", GM_f2s (fwd));
1289
1290   if (GMC_is_origin (c, fwd)) /* Startpoint */
1291   {
1292     schedule_next_keepalive (c, fwd);
1293   }
1294   else /* Relay, endpoint. */
1295   {
1296     struct GNUNET_TIME_Relative delay;
1297     GNUNET_SCHEDULER_TaskIdentifier *ti;
1298     GNUNET_SCHEDULER_Task f;
1299
1300     ti = fwd ? &c->fwd_maintenance_task : &c->bck_maintenance_task;
1301
1302     if (GNUNET_SCHEDULER_NO_TASK != *ti)
1303       GNUNET_SCHEDULER_cancel (*ti);
1304     delay = GNUNET_TIME_relative_multiply (refresh_connection_time, 4);
1305     f = fwd ? &connection_fwd_timeout : &connection_bck_timeout;
1306     *ti = GNUNET_SCHEDULER_add_delayed (delay, f, c);
1307   }
1308 }
1309
1310
1311 /**
1312  * Add the connection to the list of both neighbors.
1313  *
1314  * @param c Connection.
1315  *
1316  * @return #GNUNET_OK if everything went fine
1317  *         #GNUNET_SYSERR if the was an error and @c c is malformed.
1318  */
1319 static int
1320 register_neighbors (struct MeshConnection *c)
1321 {
1322   struct MeshPeer *next_peer;
1323   struct MeshPeer *prev_peer;
1324
1325   next_peer = get_next_hop (c);
1326   prev_peer = get_prev_hop (c);
1327
1328   LOG (GNUNET_ERROR_TYPE_DEBUG, "register neighbors for connection %s\n",
1329        GMC_2s (c));
1330   path_debug (c->path);
1331   LOG (GNUNET_ERROR_TYPE_DEBUG, "own pos %u\n", c->own_pos);
1332   LOG (GNUNET_ERROR_TYPE_DEBUG, "putting connection %s to next peer %p\n",
1333        GMC_2s (c), next_peer);
1334   LOG (GNUNET_ERROR_TYPE_DEBUG, "next peer %p %s\n", next_peer, GMP_2s (next_peer));
1335   LOG (GNUNET_ERROR_TYPE_DEBUG, "putting connection %s to prev peer %p\n",
1336        GMC_2s (c), prev_peer);
1337   LOG (GNUNET_ERROR_TYPE_DEBUG, "prev peer %p %s\n", prev_peer, GMP_2s (prev_peer));
1338
1339   if (GNUNET_NO == GMP_is_neighbor (next_peer)
1340       || GNUNET_NO == GMP_is_neighbor (prev_peer))
1341   {
1342     if (GMC_is_origin (c, GNUNET_YES))
1343       GNUNET_STATISTICS_update (stats, "# local bad paths", 1, GNUNET_NO);
1344     GNUNET_STATISTICS_update (stats, "# bad paths", 1, GNUNET_NO);
1345
1346     LOG (GNUNET_ERROR_TYPE_DEBUG, "  register neighbors failed\n");
1347     LOG (GNUNET_ERROR_TYPE_DEBUG, "  prev: %s, neighbor?: %d\n",
1348          GMP_2s (prev_peer), GMP_is_neighbor (prev_peer));
1349     LOG (GNUNET_ERROR_TYPE_DEBUG, "  next: %s, neighbor?: %d\n",
1350          GMP_2s (next_peer), GMP_is_neighbor (next_peer));
1351     return GNUNET_SYSERR;
1352   }
1353
1354   GMP_add_connection (next_peer, c);
1355   GMP_add_connection (prev_peer, c);
1356
1357   return GNUNET_OK;
1358 }
1359
1360
1361 /**
1362  * Remove the connection from the list of both neighbors.
1363  *
1364  * @param c Connection.
1365  */
1366 static void
1367 unregister_neighbors (struct MeshConnection *c)
1368 {
1369   struct MeshPeer *peer;
1370
1371   peer = get_next_hop (c);
1372   if (GNUNET_OK != GMP_remove_connection (peer, c))
1373   {
1374     GNUNET_assert (MESH_CONNECTION_NEW == c->state
1375                   || MESH_CONNECTION_DESTROYED == c->state);
1376     LOG (GNUNET_ERROR_TYPE_DEBUG, "  cstate: %u\n", c->state);
1377     if (NULL != c->t) GMT_debug (c->t);
1378   }
1379
1380   peer = get_prev_hop (c);
1381   if (GNUNET_OK != GMP_remove_connection (peer, c))
1382   {
1383     GNUNET_assert (MESH_CONNECTION_NEW == c->state
1384                   || MESH_CONNECTION_DESTROYED == c->state);
1385     LOG (GNUNET_ERROR_TYPE_DEBUG, "  cstate: %u\n", c->state);
1386     if (NULL != c->t) GMT_debug (c->t);
1387   }
1388 }
1389
1390
1391 /**
1392  * Bind the connection to the peer and the tunnel to that peer.
1393  *
1394  * If the peer has no tunnel, create one. Update tunnel and connection
1395  * data structres to reflect new status.
1396  *
1397  * @param c Connection.
1398  * @param peer Peer.
1399  */
1400 static void
1401 add_to_peer (struct MeshConnection *c, struct MeshPeer *peer)
1402 {
1403   GMP_add_tunnel (peer);
1404   c->t = GMP_get_tunnel (peer);
1405   GMT_add_connection (c->t, c);
1406 }
1407
1408
1409 /**
1410  * Builds a path from a PeerIdentity array.
1411  *
1412  * @param peers PeerIdentity array.
1413  * @param size Size of the @c peers array.
1414  * @param own_pos Output parameter: own position in the path.
1415  *
1416  * @return Fixed and shortened path.
1417  */
1418 static struct MeshPeerPath *
1419 build_path_from_peer_ids (struct GNUNET_PeerIdentity *peers,
1420                           unsigned int size,
1421                           unsigned int *own_pos)
1422 {
1423   struct MeshPeerPath *path;
1424   GNUNET_PEER_Id shortid;
1425   unsigned int i;
1426   unsigned int j;
1427   unsigned int offset;
1428
1429   /* Create path */
1430   LOG (GNUNET_ERROR_TYPE_DEBUG, "  Creating path...\n");
1431   path = path_new (size);
1432   *own_pos = 0;
1433   offset = 0;
1434   for (i = 0; i < size; i++)
1435   {
1436     LOG (GNUNET_ERROR_TYPE_DEBUG, "  - %u: taking %s\n",
1437          i, GNUNET_i2s (&peers[i]));
1438     shortid = GNUNET_PEER_intern (&peers[i]);
1439
1440     /* Check for loops / duplicates */
1441     for (j = 0; j < i - offset; j++)
1442     {
1443       if (path->peers[j] == shortid)
1444       {
1445         LOG (GNUNET_ERROR_TYPE_DEBUG, "    already exists at pos %u\n", j);
1446         offset = i - j;
1447         LOG (GNUNET_ERROR_TYPE_DEBUG, "    offset now %u\n", offset);
1448         GNUNET_PEER_change_rc (shortid, -1);
1449       }
1450     }
1451     LOG (GNUNET_ERROR_TYPE_DEBUG, "    storing at %u\n", i - offset);
1452     path->peers[i - offset] = shortid;
1453     if (path->peers[i] == myid)
1454       *own_pos = i;
1455   }
1456   path->length -= offset;
1457
1458   if (path->peers[*own_pos] != myid)
1459   {
1460     /* create path: self not found in path through self */
1461     GNUNET_break_op (0);
1462     path_destroy (path);
1463     return NULL;
1464   }
1465
1466   return path;
1467 }
1468
1469
1470 /**
1471  * Log receipt of message on stderr (INFO level).
1472  *
1473  * @param message Message received.
1474  * @param peer Peer who sent the message.
1475  * @param hash Connection ID.
1476  */
1477 static void
1478 log_message (const struct GNUNET_MessageHeader *message,
1479              const struct GNUNET_PeerIdentity *peer,
1480              const struct GNUNET_MeshHash *hash)
1481 {
1482   LOG (GNUNET_ERROR_TYPE_INFO, "<- %s on connection %s from %s\n",
1483        GM_m2s (ntohs (message->type)), GNUNET_h2s (GM_h2hc (hash)),
1484        GNUNET_i2s (peer));
1485 }
1486
1487 /******************************************************************************/
1488 /********************************    API    ***********************************/
1489 /******************************************************************************/
1490
1491 /**
1492  * Core handler for connection creation.
1493  *
1494  * @param cls Closure (unused).
1495  * @param peer Sender (neighbor).
1496  * @param message Message.
1497  *
1498  * @return GNUNET_OK to keep the connection open,
1499  *         GNUNET_SYSERR to close it (signal serious error)
1500  */
1501 int
1502 GMC_handle_create (void *cls, const struct GNUNET_PeerIdentity *peer,
1503                    const struct GNUNET_MessageHeader *message)
1504 {
1505   struct GNUNET_MESH_ConnectionCreate *msg;
1506   struct GNUNET_PeerIdentity *id;
1507   struct GNUNET_MeshHash *cid;
1508   struct MeshPeerPath *path;
1509   struct MeshPeer *dest_peer;
1510   struct MeshPeer *orig_peer;
1511   struct MeshConnection *c;
1512   unsigned int own_pos;
1513   uint16_t size;
1514
1515   /* Check size */
1516   size = ntohs (message->size);
1517   if (size < sizeof (struct GNUNET_MESH_ConnectionCreate))
1518   {
1519     GNUNET_break_op (0);
1520     return GNUNET_OK;
1521   }
1522
1523   /* Calculate hops */
1524   size -= sizeof (struct GNUNET_MESH_ConnectionCreate);
1525   if (size % sizeof (struct GNUNET_PeerIdentity))
1526   {
1527     GNUNET_break_op (0);
1528     return GNUNET_OK;
1529   }
1530   size /= sizeof (struct GNUNET_PeerIdentity);
1531   if (1 > size)
1532   {
1533     GNUNET_break_op (0);
1534     return GNUNET_OK;
1535   }
1536   LOG (GNUNET_ERROR_TYPE_DEBUG, "    path has %u hops.\n", size);
1537
1538   /* Get parameters */
1539   msg = (struct GNUNET_MESH_ConnectionCreate *) message;
1540   cid = &msg->cid;
1541   log_message (message, peer, cid);
1542   id = (struct GNUNET_PeerIdentity *) &msg[1];
1543   LOG (GNUNET_ERROR_TYPE_DEBUG, "    origin: %s\n", GNUNET_i2s (id));
1544
1545   /* Create connection */
1546   c = connection_get (cid);
1547   if (NULL == c)
1548   {
1549     path = build_path_from_peer_ids ((struct GNUNET_PeerIdentity *) &msg[1],
1550                                      size, &own_pos);
1551     if (NULL == path)
1552       return GNUNET_OK;
1553     if (0 == own_pos)
1554     {
1555       GNUNET_break_op (0);
1556       path_destroy (path);
1557       return GNUNET_OK;
1558     }
1559     LOG (GNUNET_ERROR_TYPE_DEBUG, "  Own position: %u\n", own_pos);
1560     LOG (GNUNET_ERROR_TYPE_DEBUG, "  Creating connection\n");
1561     c = GMC_new (cid, NULL, path_duplicate (path), own_pos);
1562     if (NULL == c)
1563     {
1564       if (path->length - 1 == own_pos)
1565       {
1566         /* If we are destination, why did the creation fail? */
1567         GNUNET_break (0);
1568         return GNUNET_OK;
1569       }
1570       send_broken2 (cid, &my_full_id,
1571                     GNUNET_PEER_resolve2 (path->peers[own_pos + 1]),
1572                     path->peers[own_pos - 1]);
1573       path_destroy (path);
1574       return GNUNET_OK;
1575     }
1576     GMP_add_path_to_all (path, GNUNET_NO);
1577     connection_reset_timeout (c, GNUNET_YES);
1578   }
1579   else
1580   {
1581     path = path_duplicate (c->path);
1582   }
1583   if (MESH_CONNECTION_NEW == c->state)
1584     connection_change_state (c, MESH_CONNECTION_SENT);
1585
1586   /* Remember peers */
1587   dest_peer = GMP_get (&id[size - 1]);
1588   orig_peer = GMP_get (&id[0]);
1589
1590   /* Is it a connection to us? */
1591   if (c->own_pos == size - 1)
1592   {
1593     LOG (GNUNET_ERROR_TYPE_DEBUG, "  It's for us!\n");
1594     GMP_add_path_to_origin (orig_peer, path_duplicate (path), GNUNET_YES);
1595
1596     add_to_peer (c, orig_peer);
1597     if (MESH_TUNNEL3_NEW == GMT_get_cstate (c->t))
1598       GMT_change_cstate (c->t,  MESH_TUNNEL3_WAITING);
1599
1600     send_connection_ack (c, GNUNET_NO);
1601     if (MESH_CONNECTION_SENT == c->state)
1602       connection_change_state (c, MESH_CONNECTION_ACK);
1603   }
1604   else
1605   {
1606     /* It's for somebody else! Retransmit. */
1607     LOG (GNUNET_ERROR_TYPE_DEBUG, "  Retransmitting.\n");
1608     GMP_add_path (dest_peer, path_duplicate (path), GNUNET_NO);
1609     GMP_add_path_to_origin (orig_peer, path_duplicate (path), GNUNET_NO);
1610     GMC_send_prebuilt_message (message, c, GNUNET_YES, GNUNET_YES,
1611                                NULL, NULL);
1612   }
1613   path_destroy (path);
1614   return GNUNET_OK;
1615 }
1616
1617
1618 /**
1619  * Core handler for path confirmations.
1620  *
1621  * @param cls closure
1622  * @param message message
1623  * @param peer peer identity this notification is about
1624  *
1625  * @return GNUNET_OK to keep the connection open,
1626  *         GNUNET_SYSERR to close it (signal serious error)
1627  */
1628 int
1629 GMC_handle_confirm (void *cls, const struct GNUNET_PeerIdentity *peer,
1630                     const struct GNUNET_MessageHeader *message)
1631 {
1632   struct GNUNET_MESH_ConnectionACK *msg;
1633   struct MeshConnection *c;
1634   struct MeshPeerPath *p;
1635   struct MeshPeer *pi;
1636   enum MeshConnectionState oldstate;
1637   int fwd;
1638
1639   msg = (struct GNUNET_MESH_ConnectionACK *) message;
1640   log_message (message, peer, &msg->cid);
1641   c = connection_get (&msg->cid);
1642   if (NULL == c)
1643   {
1644     GNUNET_STATISTICS_update (stats, "# control on unknown connection",
1645                               1, GNUNET_NO);
1646     LOG (GNUNET_ERROR_TYPE_DEBUG, "  don't know the connection!\n");
1647     return GNUNET_OK;
1648   }
1649
1650   if (GNUNET_NO != c->destroy)
1651   {
1652     LOG (GNUNET_ERROR_TYPE_DEBUG, "  connection being destroyed\n");
1653     return GNUNET_OK;
1654   }
1655
1656   oldstate = c->state;
1657   LOG (GNUNET_ERROR_TYPE_DEBUG, "  via peer %s\n", GNUNET_i2s (peer));
1658   pi = GMP_get (peer);
1659   if (get_next_hop (c) == pi)
1660   {
1661     LOG (GNUNET_ERROR_TYPE_DEBUG, "  SYNACK\n");
1662     fwd = GNUNET_NO;
1663     if (MESH_CONNECTION_SENT == oldstate)
1664       connection_change_state (c, MESH_CONNECTION_ACK);
1665   }
1666   else if (get_prev_hop (c) == pi)
1667   {
1668     LOG (GNUNET_ERROR_TYPE_DEBUG, "  ACK\n");
1669     fwd = GNUNET_YES;
1670     connection_change_state (c, MESH_CONNECTION_READY);
1671   }
1672   else
1673   {
1674     GNUNET_break_op (0);
1675     return GNUNET_OK;
1676   }
1677
1678   connection_reset_timeout (c, fwd);
1679
1680   /* Add path to peers? */
1681   p = c->path;
1682   if (NULL != p)
1683   {
1684     GMP_add_path_to_all (p, GNUNET_YES);
1685   }
1686   else
1687   {
1688     GNUNET_break (0);
1689   }
1690
1691   /* Message for us as creator? */
1692   if (GMC_is_origin (c, GNUNET_YES))
1693   {
1694     if (GNUNET_NO != fwd)
1695     {
1696       GNUNET_break_op (0);
1697       return GNUNET_OK;
1698     }
1699     LOG (GNUNET_ERROR_TYPE_DEBUG, "  Connection (SYN)ACK for us!\n");
1700
1701     /* If just created, cancel the short timeout and start a long one */
1702     if (MESH_CONNECTION_SENT == oldstate)
1703       connection_reset_timeout (c, GNUNET_YES);
1704
1705     /* Change connection state */
1706     connection_change_state (c, MESH_CONNECTION_READY);
1707     send_connection_ack (c, GNUNET_YES);
1708
1709     /* Change tunnel state, trigger KX */
1710     if (MESH_TUNNEL3_WAITING == GMT_get_cstate (c->t))
1711       GMT_change_cstate (c->t, MESH_TUNNEL3_READY);
1712
1713     return GNUNET_OK;
1714   }
1715
1716   /* Message for us as destination? */
1717   if (GMC_is_terminal (c, GNUNET_YES))
1718   {
1719     if (GNUNET_YES != fwd)
1720     {
1721       GNUNET_break_op (0);
1722       return GNUNET_OK;
1723     }
1724     LOG (GNUNET_ERROR_TYPE_DEBUG, "  Connection ACK for us!\n");
1725
1726     /* If just created, cancel the short timeout and start a long one */
1727     if (MESH_CONNECTION_ACK == oldstate)
1728       connection_reset_timeout (c, GNUNET_NO);
1729
1730     /* Change tunnel state */
1731     if (MESH_TUNNEL3_WAITING == GMT_get_cstate (c->t))
1732       GMT_change_cstate (c->t, MESH_TUNNEL3_READY);
1733
1734     return GNUNET_OK;
1735   }
1736
1737   LOG (GNUNET_ERROR_TYPE_DEBUG, "  not for us, retransmitting...\n");
1738   GMC_send_prebuilt_message (message, c, fwd, GNUNET_YES, NULL, NULL);
1739   return GNUNET_OK;
1740 }
1741
1742
1743 /**
1744  * Core handler for notifications of broken connections.
1745  *
1746  * @param cls Closure (unused).
1747  * @param id Peer identity of sending neighbor.
1748  * @param message Message.
1749  *
1750  * @return GNUNET_OK to keep the connection open,
1751  *         GNUNET_SYSERR to close it (signal serious error)
1752  */
1753 int
1754 GMC_handle_broken (void* cls,
1755                    const struct GNUNET_PeerIdentity* id,
1756                    const struct GNUNET_MessageHeader* message)
1757 {
1758   struct GNUNET_MESH_ConnectionBroken *msg;
1759   struct MeshConnection *c;
1760   int fwd;
1761
1762   msg = (struct GNUNET_MESH_ConnectionBroken *) message;
1763   log_message (message, id, &msg->cid);
1764   LOG (GNUNET_ERROR_TYPE_DEBUG, "  regarding %s\n",
1765               GNUNET_i2s (&msg->peer1));
1766   LOG (GNUNET_ERROR_TYPE_DEBUG, "  regarding %s\n",
1767               GNUNET_i2s (&msg->peer2));
1768   c = connection_get (&msg->cid);
1769   if (NULL == c)
1770   {
1771     LOG (GNUNET_ERROR_TYPE_DEBUG, "  duplicate CONNECTION_BROKEN\n");
1772     return GNUNET_OK;
1773   }
1774
1775   fwd = is_fwd (c, id);
1776   if (GMC_is_terminal (c, fwd))
1777   {
1778     struct GNUNET_MessageHeader *msg;
1779     struct MeshPeer *peer;
1780
1781     peer = get_hop (c, !fwd);
1782     path_invalidate (c->path);
1783     c->state = MESH_CONNECTION_DESTROYED;
1784     while (NULL != (msg = GMP_connection_pop (peer, c)))
1785     {
1786       GNUNET_assert (NULL ==
1787                      GMT_send_prebuilt_message (msg, c->t, NULL, GNUNET_YES,
1788                                                 NULL, NULL));
1789     }
1790
1791     GMC_destroy (c);
1792   }
1793   else
1794   {
1795     GMC_send_prebuilt_message (message, c, fwd, GNUNET_YES, NULL, NULL);
1796     c->destroy = GNUNET_YES;
1797     connection_cancel_queues (c, !fwd);
1798   }
1799
1800   return GNUNET_OK;
1801
1802 }
1803
1804
1805 /**
1806  * Core handler for tunnel destruction
1807  *
1808  * @param cls Closure (unused).
1809  * @param peer Peer identity of sending neighbor.
1810  * @param message Message.
1811  *
1812  * @return GNUNET_OK to keep the connection open,
1813  *         GNUNET_SYSERR to close it (signal serious error)
1814  */
1815 int
1816 GMC_handle_destroy (void *cls, const struct GNUNET_PeerIdentity *peer,
1817                     const struct GNUNET_MessageHeader *message)
1818 {
1819   struct GNUNET_MESH_ConnectionDestroy *msg;
1820   struct MeshConnection *c;
1821   int fwd;
1822
1823   msg = (struct GNUNET_MESH_ConnectionDestroy *) message;
1824   log_message (message, peer, &msg->cid);
1825   c = connection_get (&msg->cid);
1826   if (NULL == c)
1827   {
1828     /* Probably already got the message from another path,
1829      * destroyed the tunnel and retransmitted to children.
1830      * Safe to ignore.
1831      */
1832     GNUNET_STATISTICS_update (stats, "# control on unknown connection",
1833                               1, GNUNET_NO);
1834     LOG (GNUNET_ERROR_TYPE_DEBUG, "  connection unknown: already destroyed?\n");
1835     return GNUNET_OK;
1836   }
1837   fwd = is_fwd (c, peer);
1838   if (GNUNET_SYSERR == fwd)
1839   {
1840     GNUNET_break_op (0); /* FIXME */
1841     return GNUNET_OK;
1842   }
1843   if (GNUNET_NO == GMC_is_terminal (c, fwd))
1844     GMC_send_prebuilt_message (message, c, fwd, GNUNET_YES, NULL, NULL);
1845   else if (0 == c->pending_messages)
1846   {
1847     LOG (GNUNET_ERROR_TYPE_DEBUG, "!  directly destroying connection!\n");
1848     GMC_destroy (c);
1849     return GNUNET_OK;
1850   }
1851   c->destroy = GNUNET_YES;
1852   c->state = MESH_CONNECTION_DESTROYED;
1853   if (NULL != c->t)
1854   {
1855     GMT_remove_connection (c->t, c);
1856     c->t = NULL;
1857   }
1858
1859   return GNUNET_OK;
1860 }
1861
1862 /**
1863  * Generic handler for mesh network encrypted traffic.
1864  *
1865  * @param peer Peer identity this notification is about.
1866  * @param msg Encrypted message.
1867  *
1868  * @return GNUNET_OK to keep the connection open,
1869  *         GNUNET_SYSERR to close it (signal serious error)
1870  */
1871 static int
1872 handle_mesh_encrypted (const struct GNUNET_PeerIdentity *peer,
1873                        const struct GNUNET_MESH_Encrypted *msg)
1874 {
1875   struct MeshConnection *c;
1876   struct MeshPeer *neighbor;
1877   struct MeshFlowControl *fc;
1878   GNUNET_PEER_Id peer_id;
1879   uint32_t pid;
1880   uint32_t ttl;
1881   size_t size;
1882   int fwd;
1883
1884   log_message (&msg->header, peer, &msg->cid);
1885
1886   /* Check size */
1887   size = ntohs (msg->header.size);
1888   if (size <
1889       sizeof (struct GNUNET_MESH_Encrypted) +
1890       sizeof (struct GNUNET_MessageHeader))
1891   {
1892     GNUNET_break_op (0);
1893     return GNUNET_OK;
1894   }
1895
1896   /* Check connection */
1897   c = connection_get (&msg->cid);
1898   if (NULL == c)
1899   {
1900     GNUNET_STATISTICS_update (stats, "# unknown connection", 1, GNUNET_NO);
1901     LOG (GNUNET_ERROR_TYPE_DEBUG, "enc on unknown connection %s\n",
1902          GNUNET_h2s (GM_h2hc (&msg->cid)));
1903     return GNUNET_OK;
1904   }
1905
1906   LOG (GNUNET_ERROR_TYPE_DEBUG, "  connection %s\n", GMC_2s (c));
1907
1908   /* Check if origin is as expected */
1909   neighbor = get_prev_hop (c);
1910   peer_id = GNUNET_PEER_search (peer);
1911   if (peer_id == GMP_get_short_id (neighbor))
1912   {
1913     fwd = GNUNET_YES;
1914   }
1915   else
1916   {
1917     neighbor = get_next_hop (c);
1918     if (peer_id == GMP_get_short_id (neighbor))
1919     {
1920       fwd = GNUNET_NO;
1921     }
1922     else
1923     {
1924       /* Unexpected peer sending traffic on a connection. */
1925       GNUNET_break_op (0);
1926       return GNUNET_OK;
1927     }
1928   }
1929
1930   /* Check PID */
1931   fc = fwd ? &c->bck_fc : &c->fwd_fc;
1932   pid = ntohl (msg->pid);
1933   if (GM_is_pid_bigger (pid, fc->last_ack_sent))
1934   {
1935     GNUNET_STATISTICS_update (stats, "# unsolicited message", 1, GNUNET_NO);
1936     LOG (GNUNET_ERROR_TYPE_DEBUG,
1937                 "WARNING Received PID %u, (prev %u), ACK %u\n",
1938                 pid, fc->last_pid_recv, fc->last_ack_sent);
1939     return GNUNET_OK;
1940   }
1941   if (GNUNET_NO == GM_is_pid_bigger (pid, fc->last_pid_recv))
1942   {
1943     GNUNET_STATISTICS_update (stats, "# duplicate PID", 1, GNUNET_NO);
1944     LOG (GNUNET_ERROR_TYPE_DEBUG,
1945                 " Pid %u not expected (%u+), dropping!\n",
1946                 pid, fc->last_pid_recv + 1);
1947     return GNUNET_OK;
1948   }
1949   if (MESH_CONNECTION_SENT == c->state || MESH_CONNECTION_ACK == c->state)
1950     connection_change_state (c, MESH_CONNECTION_READY);
1951   connection_reset_timeout (c, fwd);
1952   fc->last_pid_recv = pid;
1953
1954   /* Is this message for us? */
1955   if (GMC_is_terminal (c, fwd))
1956   {
1957     LOG (GNUNET_ERROR_TYPE_DEBUG, "  message for us!\n");
1958     GNUNET_STATISTICS_update (stats, "# messages received", 1, GNUNET_NO);
1959
1960     if (NULL == c->t)
1961     {
1962       GNUNET_break (GNUNET_NO != c->destroy);
1963       return GNUNET_OK;
1964     }
1965     fc->last_pid_recv = pid;
1966     GMT_handle_encrypted (c->t, msg);
1967     GMC_send_ack (c, fwd, GNUNET_NO);
1968     return GNUNET_OK;
1969   }
1970
1971   /* Message not for us: forward to next hop */
1972   LOG (GNUNET_ERROR_TYPE_DEBUG, "  not for us, retransmitting...\n");
1973   ttl = ntohl (msg->ttl);
1974   LOG (GNUNET_ERROR_TYPE_DEBUG, "   ttl: %u\n", ttl);
1975   if (ttl == 0)
1976   {
1977     GNUNET_STATISTICS_update (stats, "# TTL drops", 1, GNUNET_NO);
1978     LOG (GNUNET_ERROR_TYPE_WARNING, " TTL is 0, DROPPING!\n");
1979     GMC_send_ack (c, fwd, GNUNET_NO);
1980     return GNUNET_OK;
1981   }
1982
1983   GNUNET_STATISTICS_update (stats, "# messages forwarded", 1, GNUNET_NO);
1984   GMC_send_prebuilt_message (&msg->header, c, fwd, GNUNET_NO, NULL, NULL);
1985
1986   return GNUNET_OK;
1987 }
1988
1989 /**
1990  * Generic handler for mesh network encrypted traffic.
1991  *
1992  * @param peer Peer identity this notification is about.
1993  * @param msg Encrypted message.
1994  *
1995  * @return GNUNET_OK to keep the connection open,
1996  *         GNUNET_SYSERR to close it (signal serious error)
1997  */
1998 static int
1999 handle_mesh_kx (const struct GNUNET_PeerIdentity *peer,
2000                 const struct GNUNET_MESH_KX *msg)
2001 {
2002   struct MeshConnection *c;
2003   struct MeshPeer *neighbor;
2004   GNUNET_PEER_Id peer_id;
2005   size_t size;
2006   int fwd;
2007
2008   log_message (&msg->header, peer, &msg->cid);
2009
2010   /* Check size */
2011   size = ntohs (msg->header.size);
2012   if (size <
2013       sizeof (struct GNUNET_MESH_KX) +
2014       sizeof (struct GNUNET_MessageHeader))
2015   {
2016     GNUNET_break_op (0);
2017     return GNUNET_OK;
2018   }
2019
2020   /* Check connection */
2021   c = connection_get (&msg->cid);
2022   if (NULL == c)
2023   {
2024     GNUNET_STATISTICS_update (stats, "# unknown connection", 1, GNUNET_NO);
2025     LOG (GNUNET_ERROR_TYPE_DEBUG, "kx on unknown connection %s\n",
2026          GNUNET_h2s (GM_h2hc (&msg->cid)));
2027     return GNUNET_OK;
2028   }
2029   LOG (GNUNET_ERROR_TYPE_DEBUG, " on connection %s\n", GMC_2s (c));
2030
2031   /* Check if origin is as expected */
2032   neighbor = get_prev_hop (c);
2033   peer_id = GNUNET_PEER_search (peer);
2034   if (peer_id == GMP_get_short_id (neighbor))
2035   {
2036     fwd = GNUNET_YES;
2037   }
2038   else
2039   {
2040     neighbor = get_next_hop (c);
2041     if (peer_id == GMP_get_short_id (neighbor))
2042     {
2043       fwd = GNUNET_NO;
2044     }
2045     else
2046     {
2047       /* Unexpected peer sending traffic on a connection. */
2048       GNUNET_break_op (0);
2049       return GNUNET_OK;
2050     }
2051   }
2052
2053   /* Count as connection confirmation. */
2054   if (MESH_CONNECTION_SENT == c->state || MESH_CONNECTION_ACK == c->state)
2055   {
2056     connection_change_state (c, MESH_CONNECTION_READY);
2057     if (NULL != c->t)
2058     {
2059       if (MESH_TUNNEL3_WAITING == GMT_get_cstate (c->t))
2060         GMT_change_cstate (c->t, MESH_TUNNEL3_READY);
2061     }
2062   }
2063   connection_reset_timeout (c, fwd);
2064
2065   /* Is this message for us? */
2066   if (GMC_is_terminal (c, fwd))
2067   {
2068     LOG (GNUNET_ERROR_TYPE_DEBUG, "  message for us!\n");
2069     GNUNET_STATISTICS_update (stats, "# messages received", 1, GNUNET_NO);
2070     if (NULL == c->t)
2071     {
2072       GNUNET_break (0);
2073       return GNUNET_OK;
2074     }
2075     GMT_handle_kx (c->t, &msg[1].header);
2076     return GNUNET_OK;
2077   }
2078
2079   /* Message not for us: forward to next hop */
2080   LOG (GNUNET_ERROR_TYPE_DEBUG, "  not for us, retransmitting...\n");
2081   GNUNET_STATISTICS_update (stats, "# messages forwarded", 1, GNUNET_NO);
2082   GMC_send_prebuilt_message (&msg->header, c, fwd, GNUNET_NO, NULL, NULL);
2083
2084   return GNUNET_OK;
2085 }
2086
2087
2088 /**
2089  * Core handler for encrypted mesh network traffic (channel mgmt, data).
2090  *
2091  * @param cls Closure (unused).
2092  * @param message Message received.
2093  * @param peer Peer who sent the message.
2094  *
2095  * @return GNUNET_OK to keep the connection open,
2096  *         GNUNET_SYSERR to close it (signal serious error)
2097  */
2098 int
2099 GMC_handle_encrypted (void *cls, const struct GNUNET_PeerIdentity *peer,
2100                       const struct GNUNET_MessageHeader *message)
2101 {
2102   return handle_mesh_encrypted (peer,
2103                                 (struct GNUNET_MESH_Encrypted *)message);
2104 }
2105
2106
2107 /**
2108  * Core handler for key exchange traffic (ephemeral key, ping, pong).
2109  *
2110  * @param cls Closure (unused).
2111  * @param message Message received.
2112  * @param peer Peer who sent the message.
2113  *
2114  * @return GNUNET_OK to keep the connection open,
2115  *         GNUNET_SYSERR to close it (signal serious error)
2116  */
2117 int
2118 GMC_handle_kx (void *cls, const struct GNUNET_PeerIdentity *peer,
2119                const struct GNUNET_MessageHeader *message)
2120 {
2121   return handle_mesh_kx (peer,
2122                          (struct GNUNET_MESH_KX *) message);
2123 }
2124
2125
2126 /**
2127  * Core handler for mesh network traffic point-to-point acks.
2128  *
2129  * @param cls closure
2130  * @param message message
2131  * @param peer peer identity this notification is about
2132  *
2133  * @return GNUNET_OK to keep the connection open,
2134  *         GNUNET_SYSERR to close it (signal serious error)
2135  */
2136 int
2137 GMC_handle_ack (void *cls, const struct GNUNET_PeerIdentity *peer,
2138                 const struct GNUNET_MessageHeader *message)
2139 {
2140   struct GNUNET_MESH_ACK *msg;
2141   struct MeshConnection *c;
2142   struct MeshFlowControl *fc;
2143   GNUNET_PEER_Id id;
2144   uint32_t ack;
2145   int fwd;
2146
2147   msg = (struct GNUNET_MESH_ACK *) message;
2148   log_message (message, peer, &msg->cid);
2149   c = connection_get (&msg->cid);
2150   if (NULL == c)
2151   {
2152     GNUNET_STATISTICS_update (stats, "# ack on unknown connection", 1,
2153                               GNUNET_NO);
2154     return GNUNET_OK;
2155   }
2156
2157   /* Is this a forward or backward ACK? */
2158   id = GNUNET_PEER_search (peer);
2159   if (GMP_get_short_id (get_next_hop (c)) == id)
2160   {
2161     LOG (GNUNET_ERROR_TYPE_DEBUG, "  FWD ACK\n");
2162     fc = &c->fwd_fc;
2163     fwd = GNUNET_YES;
2164   }
2165   else if (GMP_get_short_id (get_prev_hop (c)) == id)
2166   {
2167     LOG (GNUNET_ERROR_TYPE_DEBUG, "  BCK ACK\n");
2168     fc = &c->bck_fc;
2169     fwd = GNUNET_NO;
2170   }
2171   else
2172   {
2173     GNUNET_break_op (0);
2174     return GNUNET_OK;
2175   }
2176
2177   ack = ntohl (msg->ack);
2178   LOG (GNUNET_ERROR_TYPE_DEBUG, "  ACK %u (was %u)\n",
2179               ack, fc->last_ack_recv);
2180   if (GM_is_pid_bigger (ack, fc->last_ack_recv))
2181     fc->last_ack_recv = ack;
2182
2183   /* Cancel polling if the ACK is big enough. */
2184   if (GNUNET_SCHEDULER_NO_TASK != fc->poll_task &&
2185       GM_is_pid_bigger (fc->last_ack_recv, fc->last_pid_sent))
2186   {
2187     LOG (GNUNET_ERROR_TYPE_DEBUG, "  Cancel poll\n");
2188     GNUNET_SCHEDULER_cancel (fc->poll_task);
2189     fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
2190     fc->poll_time = GNUNET_TIME_UNIT_SECONDS;
2191   }
2192
2193   connection_unlock_queue (c, fwd);
2194
2195   return GNUNET_OK;
2196 }
2197
2198
2199 /**
2200  * Core handler for mesh network traffic point-to-point ack polls.
2201  *
2202  * @param cls closure
2203  * @param message message
2204  * @param peer peer identity this notification is about
2205  *
2206  * @return GNUNET_OK to keep the connection open,
2207  *         GNUNET_SYSERR to close it (signal serious error)
2208  */
2209 int
2210 GMC_handle_poll (void *cls, const struct GNUNET_PeerIdentity *peer,
2211                  const struct GNUNET_MessageHeader *message)
2212 {
2213   struct GNUNET_MESH_Poll *msg;
2214   struct MeshConnection *c;
2215   struct MeshFlowControl *fc;
2216   GNUNET_PEER_Id id;
2217   uint32_t pid;
2218   int fwd;
2219
2220   msg = (struct GNUNET_MESH_Poll *) message;
2221   log_message (message, peer, &msg->cid);
2222   c = connection_get (&msg->cid);
2223   if (NULL == c)
2224   {
2225     GNUNET_STATISTICS_update (stats, "# poll on unknown connection", 1,
2226                               GNUNET_NO);
2227     LOG (GNUNET_ERROR_TYPE_DEBUG,
2228          "WARNING POLL message on unknown connection %s!\n",
2229          GNUNET_h2s (GM_h2hc (&msg->cid)));
2230     return GNUNET_OK;
2231   }
2232
2233   /* Is this a forward or backward ACK?
2234    * Note: a poll should never be needed in a loopback case,
2235    * since there is no possiblility of packet loss there, so
2236    * this way of discerining FWD/BCK should not be a problem.
2237    */
2238   id = GNUNET_PEER_search (peer);
2239   if (GMP_get_short_id (get_next_hop (c)) == id)
2240   {
2241     LOG (GNUNET_ERROR_TYPE_DEBUG, "  FWD FC\n");
2242     fc = &c->fwd_fc;
2243   }
2244   else if (GMP_get_short_id (get_prev_hop (c)) == id)
2245   {
2246     LOG (GNUNET_ERROR_TYPE_DEBUG, "  BCK FC\n");
2247     fc = &c->bck_fc;
2248   }
2249   else
2250   {
2251     GNUNET_break_op (0);
2252     return GNUNET_OK;
2253   }
2254
2255   pid = ntohl (msg->pid);
2256   LOG (GNUNET_ERROR_TYPE_DEBUG, "  PID %u, OLD %u\n", pid, fc->last_pid_recv);
2257   fc->last_pid_recv = pid;
2258   fwd = fc == &c->bck_fc;
2259   GMC_send_ack (c, fwd, GNUNET_YES);
2260
2261   return GNUNET_OK;
2262 }
2263
2264
2265 /**
2266  * Send an ACK on the appropriate connection/channel, depending on
2267  * the direction and the position of the peer.
2268  *
2269  * @param c Which connection to send the hop-by-hop ACK.
2270  * @param fwd Is this a fwd ACK? (will go dest->root).
2271  * @param force Send the ACK even if suboptimal (e.g. requested by POLL).
2272  */
2273 void
2274 GMC_send_ack (struct MeshConnection *c, int fwd, int force)
2275 {
2276   unsigned int buffer;
2277
2278   LOG (GNUNET_ERROR_TYPE_DEBUG,
2279        "GMC send %s ACK on %s\n",
2280        GM_f2s (fwd), GMC_2s (c));
2281
2282   if (NULL == c)
2283   {
2284     GNUNET_break (0);
2285     return;
2286   }
2287
2288   if (GNUNET_NO != c->destroy)
2289   {
2290     LOG (GNUNET_ERROR_TYPE_DEBUG, "  being destroyed, why bother...\n");
2291     return;
2292   }
2293
2294   /* Get available buffer space */
2295   if (GMC_is_terminal (c, fwd))
2296   {
2297     LOG (GNUNET_ERROR_TYPE_DEBUG, "  getting from all channels\n");
2298     buffer = GMT_get_channels_buffer (c->t);
2299   }
2300   else
2301   {
2302     LOG (GNUNET_ERROR_TYPE_DEBUG, "  getting from one connection\n");
2303     buffer = GMC_get_buffer (c, fwd);
2304   }
2305   LOG (GNUNET_ERROR_TYPE_DEBUG, "  buffer available: %u\n", buffer);
2306   if (0 == buffer && GNUNET_NO == force)
2307     return;
2308
2309   /* Send available buffer space */
2310   if (GMC_is_origin (c, fwd))
2311   {
2312     GNUNET_assert (NULL != c->t);
2313     LOG (GNUNET_ERROR_TYPE_DEBUG, "  sending on channels...\n");
2314     GMT_unchoke_channels (c->t);
2315   }
2316   else
2317   {
2318     LOG (GNUNET_ERROR_TYPE_DEBUG, "  sending on connection\n");
2319     send_ack (c, buffer, fwd, force);
2320   }
2321 }
2322
2323
2324 /**
2325  * Initialize the connections subsystem
2326  *
2327  * @param c Configuration handle.
2328  */
2329 void
2330 GMC_init (const struct GNUNET_CONFIGURATION_Handle *c)
2331 {
2332   LOG (GNUNET_ERROR_TYPE_DEBUG, "init\n");
2333   if (GNUNET_OK !=
2334       GNUNET_CONFIGURATION_get_value_number (c, "MESH", "MAX_MSGS_QUEUE",
2335                                              &max_msgs_queue))
2336   {
2337     GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
2338                                "MESH", "MAX_MSGS_QUEUE", "MISSING");
2339     GNUNET_SCHEDULER_shutdown ();
2340     return;
2341   }
2342
2343   if (GNUNET_OK !=
2344       GNUNET_CONFIGURATION_get_value_number (c, "MESH", "MAX_CONNECTIONS",
2345                                              &max_connections))
2346   {
2347     GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
2348                                "MESH", "MAX_CONNECTIONS", "MISSING");
2349     GNUNET_SCHEDULER_shutdown ();
2350     return;
2351   }
2352
2353   if (GNUNET_OK !=
2354       GNUNET_CONFIGURATION_get_value_time (c, "MESH", "REFRESH_CONNECTION_TIME",
2355                                            &refresh_connection_time))
2356   {
2357     GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
2358                                "MESH", "REFRESH_CONNECTION_TIME", "MISSING");
2359     GNUNET_SCHEDULER_shutdown ();
2360     return;
2361   }
2362   create_connection_time = GNUNET_TIME_UNIT_SECONDS;
2363   connections = GNUNET_CONTAINER_multihashmap_create (1024, GNUNET_NO);
2364 }
2365
2366
2367 /**
2368  * Destroy each connection on shutdown.
2369  *
2370  * @param cls Closure (unused).
2371  * @param key Current key code (CID, unused).
2372  * @param value Value in the hash map (connection)
2373  *
2374  * @return #GNUNET_YES, because we should continue to iterate,
2375  */
2376 static int
2377 shutdown_iterator (void *cls,
2378                    const struct GNUNET_HashCode *key,
2379                    void *value)
2380 {
2381   struct MeshConnection *c = value;
2382
2383   GMC_destroy (c);
2384   return GNUNET_YES;
2385 }
2386
2387
2388 /**
2389  * Shut down the connections subsystem.
2390  */
2391 void
2392 GMC_shutdown (void)
2393 {
2394   GNUNET_CONTAINER_multihashmap_iterate (connections, &shutdown_iterator, NULL);
2395   GNUNET_CONTAINER_multihashmap_destroy (connections);
2396   connections = NULL;
2397 }
2398
2399
2400 struct MeshConnection *
2401 GMC_new (const struct GNUNET_MeshHash *cid,
2402          struct MeshTunnel3 *t,
2403          struct MeshPeerPath *p,
2404          unsigned int own_pos)
2405 {
2406   struct MeshConnection *c;
2407
2408   c = GNUNET_new (struct MeshConnection);
2409   c->id = *cid;
2410   GNUNET_assert (GNUNET_OK ==
2411                  GNUNET_CONTAINER_multihashmap_put (connections,
2412                                                     GMC_get_h (c), c,
2413                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
2414   fc_init (&c->fwd_fc);
2415   fc_init (&c->bck_fc);
2416   c->fwd_fc.c = c;
2417   c->bck_fc.c = c;
2418
2419   c->t = t;
2420   GNUNET_assert (own_pos <= p->length - 1);
2421   c->own_pos = own_pos;
2422   c->path = p;
2423
2424   if (GNUNET_OK != register_neighbors (c))
2425   {
2426     if (0 == own_pos)
2427     {
2428       path_invalidate (c->path);
2429       c->t = NULL;
2430       c->path = NULL;
2431     }
2432     GMC_destroy (c);
2433     return NULL;
2434   }
2435
2436   return c;
2437 }
2438
2439
2440 void
2441 GMC_destroy (struct MeshConnection *c)
2442 {
2443   if (NULL == c)
2444   {
2445     GNUNET_break (0);
2446     return;
2447   }
2448
2449   if (2 == c->destroy) /* cancel queues -> GMP_queue_cancel -> q_destroy -> */
2450     return;            /* -> message_sent -> GMC_destroy. Don't loop. */
2451   c->destroy = 2;
2452
2453   LOG (GNUNET_ERROR_TYPE_DEBUG, "destroying connection %s\n", GMC_2s (c));
2454   LOG (GNUNET_ERROR_TYPE_DEBUG, " fc's f: %p, b: %p\n",
2455        &c->fwd_fc, &c->bck_fc);
2456   LOG (GNUNET_ERROR_TYPE_DEBUG, " fc tasks f: %u, b: %u\n",
2457        c->fwd_fc.poll_task, c->bck_fc.poll_task);
2458
2459   /* Cancel all traffic */
2460   if (NULL != c->path)
2461   {
2462     connection_cancel_queues (c, GNUNET_YES);
2463     connection_cancel_queues (c, GNUNET_NO);
2464     unregister_neighbors (c);
2465   }
2466
2467   LOG (GNUNET_ERROR_TYPE_DEBUG, " fc tasks f: %u, b: %u\n",
2468        c->fwd_fc.poll_task, c->bck_fc.poll_task);
2469
2470   /* Cancel maintainance task (keepalive/timeout) */
2471   if (NULL != c->fwd_fc.poll_msg)
2472   {
2473     GMC_cancel (c->fwd_fc.poll_msg);
2474     LOG (GNUNET_ERROR_TYPE_DEBUG, " *** POLL msg FWD canceled\n");
2475   }
2476   if (NULL != c->bck_fc.poll_msg)
2477   {
2478     GMC_cancel (c->bck_fc.poll_msg);
2479     LOG (GNUNET_ERROR_TYPE_DEBUG, " *** POLL msg BCK canceled\n");
2480   }
2481
2482   /* Delete from tunnel */
2483   if (NULL != c->t)
2484     GMT_remove_connection (c->t, c);
2485
2486   if (GNUNET_NO == GMC_is_origin (c, GNUNET_YES) && NULL != c->path)
2487     path_destroy (c->path);
2488   if (GNUNET_SCHEDULER_NO_TASK != c->fwd_maintenance_task)
2489     GNUNET_SCHEDULER_cancel (c->fwd_maintenance_task);
2490   if (GNUNET_SCHEDULER_NO_TASK != c->bck_maintenance_task)
2491     GNUNET_SCHEDULER_cancel (c->bck_maintenance_task);
2492   if (GNUNET_SCHEDULER_NO_TASK != c->fwd_fc.poll_task)
2493   {
2494     GNUNET_SCHEDULER_cancel (c->fwd_fc.poll_task);
2495     LOG (GNUNET_ERROR_TYPE_DEBUG, " *** POLL FWD canceled\n");
2496   }
2497   if (GNUNET_SCHEDULER_NO_TASK != c->bck_fc.poll_task)
2498   {
2499     GNUNET_SCHEDULER_cancel (c->bck_fc.poll_task);
2500     LOG (GNUNET_ERROR_TYPE_DEBUG, " *** POLL BCK canceled\n");
2501   }
2502
2503   GNUNET_break (GNUNET_YES ==
2504                 GNUNET_CONTAINER_multihashmap_remove (connections,
2505                                                       GMC_get_h (c), c));
2506
2507   GNUNET_STATISTICS_update (stats, "# connections", -1, GNUNET_NO);
2508   GNUNET_free (c);
2509 }
2510
2511 /**
2512  * Get the connection ID.
2513  *
2514  * @param c Connection to get the ID from.
2515  *
2516  * @return ID of the connection.
2517  */
2518 const struct GNUNET_MeshHash *
2519 GMC_get_id (const struct MeshConnection *c)
2520 {
2521   return &c->id;
2522 }
2523
2524
2525 /**
2526  * Get the connection ID.
2527  *
2528  * @param c Connection to get the ID from.
2529  *
2530  * @return ID of the connection.
2531  */
2532 const struct GNUNET_HashCode *
2533 GMC_get_h (const struct MeshConnection *c)
2534 {
2535   return GM_h2hc (&c->id);
2536 }
2537
2538
2539 /**
2540  * Get the connection path.
2541  *
2542  * @param c Connection to get the path from.
2543  *
2544  * @return path used by the connection.
2545  */
2546 const struct MeshPeerPath *
2547 GMC_get_path (const struct MeshConnection *c)
2548 {
2549   if (GNUNET_NO == c->destroy)
2550     return c->path;
2551   return NULL;
2552 }
2553
2554
2555 /**
2556  * Get the connection state.
2557  *
2558  * @param c Connection to get the state from.
2559  *
2560  * @return state of the connection.
2561  */
2562 enum MeshConnectionState
2563 GMC_get_state (const struct MeshConnection *c)
2564 {
2565   return c->state;
2566 }
2567
2568 /**
2569  * Get the connection tunnel.
2570  *
2571  * @param c Connection to get the tunnel from.
2572  *
2573  * @return tunnel of the connection.
2574  */
2575 struct MeshTunnel3 *
2576 GMC_get_tunnel (const struct MeshConnection *c)
2577 {
2578   return c->t;
2579 }
2580
2581
2582 /**
2583  * Get free buffer space in a connection.
2584  *
2585  * @param c Connection.
2586  * @param fwd Is query about FWD traffic?
2587  *
2588  * @return Free buffer space [0 - max_msgs_queue/max_connections]
2589  */
2590 unsigned int
2591 GMC_get_buffer (struct MeshConnection *c, int fwd)
2592 {
2593   struct MeshFlowControl *fc;
2594
2595   fc = fwd ? &c->fwd_fc : &c->bck_fc;
2596
2597   return (fc->queue_max - fc->queue_n);
2598 }
2599
2600 /**
2601  * Get how many messages have we allowed to send to us from a direction.
2602  *
2603  * @param c Connection.
2604  * @param fwd Are we asking about traffic from FWD (BCK messages)?
2605  *
2606  * @return last_ack_sent - last_pid_recv
2607  */
2608 unsigned int
2609 GMC_get_allowed (struct MeshConnection *c, int fwd)
2610 {
2611   struct MeshFlowControl *fc;
2612
2613   fc = fwd ? &c->fwd_fc : &c->bck_fc;
2614   if (GM_is_pid_bigger(fc->last_pid_recv, fc->last_ack_sent))
2615   {
2616     return 0;
2617   }
2618   return (fc->last_ack_sent - fc->last_pid_recv);
2619 }
2620
2621 /**
2622  * Get messages queued in a connection.
2623  *
2624  * @param c Connection.
2625  * @param fwd Is query about FWD traffic?
2626  *
2627  * @return Number of messages queued.
2628  */
2629 unsigned int
2630 GMC_get_qn (struct MeshConnection *c, int fwd)
2631 {
2632   struct MeshFlowControl *fc;
2633
2634   fc = fwd ? &c->fwd_fc : &c->bck_fc;
2635
2636   return fc->queue_n;
2637 }
2638
2639
2640 /**
2641  * Allow the connection to advertise a buffer of the given size.
2642  *
2643  * The connection will send an @c fwd ACK message (so: in direction !fwd)
2644  * allowing up to last_pid_recv + buffer.
2645  *
2646  * @param c Connection.
2647  * @param buffer How many more messages the connection can accept.
2648  * @param fwd Is this about FWD traffic? (The ack will go dest->root).
2649  */
2650 void
2651 GMC_allow (struct MeshConnection *c, unsigned int buffer, int fwd)
2652 {
2653   LOG (GNUNET_ERROR_TYPE_DEBUG, "  allowing %s %u messages %s\n",
2654        GMC_2s (c), buffer, GM_f2s (fwd));
2655   send_ack (c, buffer, fwd, GNUNET_NO);
2656 }
2657
2658
2659 /**
2660  * Notify other peers on a connection of a broken link. Mark connections
2661  * to destroy after all traffic has been sent.
2662  *
2663  * @param c Connection on which there has been a disconnection.
2664  * @param peer Peer that disconnected.
2665  */
2666 void
2667 GMC_notify_broken (struct MeshConnection *c,
2668                    struct MeshPeer *peer)
2669 {
2670   int fwd;
2671
2672   LOG (GNUNET_ERROR_TYPE_DEBUG,
2673        " notify broken on %s due to %s disconnect\n",
2674        GMC_2s (c), GMP_2s (peer));
2675
2676   fwd = peer == get_prev_hop (c);
2677
2678   if (GNUNET_YES == GMC_is_terminal (c, fwd))
2679   {
2680     /* Local shutdown, no one to notify about this. */
2681     GMC_destroy (c);
2682     return;
2683   }
2684   if (GNUNET_NO == c->destroy)
2685     send_broken (c, &my_full_id, GMP_get_id (peer), fwd);
2686
2687   /* Connection will have at least one pending message
2688    * (the one we just scheduled), so no point in checking whether to
2689    * destroy immediately. */
2690   c->destroy = GNUNET_YES;
2691   c->state = MESH_CONNECTION_DESTROYED;
2692
2693   /**
2694    * Cancel all queues, if no message is left, connection will be destroyed.
2695    */
2696   connection_cancel_queues (c, !fwd);
2697
2698   return;
2699 }
2700
2701
2702 /**
2703  * Is this peer the first one on the connection?
2704  *
2705  * @param c Connection.
2706  * @param fwd Is this about fwd traffic?
2707  *
2708  * @return #GNUNET_YES if origin, #GNUNET_NO if relay/terminal.
2709  */
2710 int
2711 GMC_is_origin (struct MeshConnection *c, int fwd)
2712 {
2713   if (!fwd && c->path->length - 1 == c->own_pos )
2714     return GNUNET_YES;
2715   if (fwd && 0 == c->own_pos)
2716     return GNUNET_YES;
2717   return GNUNET_NO;
2718 }
2719
2720
2721 /**
2722  * Is this peer the last one on the connection?
2723  *
2724  * @param c Connection.
2725  * @param fwd Is this about fwd traffic?
2726  *            Note that the ROOT is the terminal for BCK traffic!
2727  *
2728  * @return #GNUNET_YES if terminal, #GNUNET_NO if relay/origin.
2729  */
2730 int
2731 GMC_is_terminal (struct MeshConnection *c, int fwd)
2732 {
2733   return GMC_is_origin (c, !fwd);
2734 }
2735
2736
2737 /**
2738  * See if we are allowed to send by the next hop in the given direction.
2739  *
2740  * @param c Connection.
2741  * @param fwd Is this about fwd traffic?
2742  *
2743  * @return #GNUNET_YES in case it's OK to send.
2744  */
2745 int
2746 GMC_is_sendable (struct MeshConnection *c, int fwd)
2747 {
2748   struct MeshFlowControl *fc;
2749
2750   if (NULL == c)
2751   {
2752     GNUNET_break (0);
2753     return GNUNET_YES;
2754   }
2755   fc = fwd ? &c->fwd_fc : &c->bck_fc;
2756   if (GM_is_pid_bigger (fc->last_ack_recv, fc->last_pid_sent))
2757     return GNUNET_YES;
2758   return GNUNET_NO;
2759 }
2760
2761 /**
2762  * Sends an already built message on a connection, properly registering
2763  * all used resources.
2764  *
2765  * @param message Message to send. Function makes a copy of it.
2766  *                If message is not hop-by-hop, decrements TTL of copy.
2767  * @param c Connection on which this message is transmitted.
2768  * @param fwd Is this a fwd message?
2769  * @param force Force the connection to accept the message (buffer overfill).
2770  * @param cont Continuation called once message is sent. Can be NULL.
2771  * @param cont_cls Closure for @c cont.
2772  *
2773  * @return Handle to cancel the message before it's sent.
2774  *         NULL on error or if @c cont is NULL.
2775  *         Invalid on @c cont call.
2776  */
2777 struct MeshConnectionQueue *
2778 GMC_send_prebuilt_message (const struct GNUNET_MessageHeader *message,
2779                            struct MeshConnection *c, int fwd, int force,
2780                            GMC_sent cont, void *cont_cls)
2781 {
2782   struct MeshFlowControl *fc;
2783   struct MeshConnectionQueue *q;
2784   void *data;
2785   size_t size;
2786   uint16_t type;
2787   int droppable;
2788
2789   size = ntohs (message->size);
2790   data = GNUNET_malloc (size);
2791   memcpy (data, message, size);
2792   type = ntohs (message->type);
2793   LOG (GNUNET_ERROR_TYPE_INFO, "-> %s on connection %s (%u bytes)\n",
2794        GM_m2s (type), GMC_2s (c), size);
2795
2796   fc = fwd ? &c->fwd_fc : &c->bck_fc;
2797   droppable = GNUNET_NO == force;
2798   switch (type)
2799   {
2800     struct GNUNET_MESH_Encrypted *emsg;
2801     struct GNUNET_MESH_KX        *kmsg;
2802     struct GNUNET_MESH_ACK       *amsg;
2803     struct GNUNET_MESH_Poll      *pmsg;
2804     struct GNUNET_MESH_ConnectionDestroy *dmsg;
2805     struct GNUNET_MESH_ConnectionBroken  *bmsg;
2806     uint32_t ttl;
2807
2808     case GNUNET_MESSAGE_TYPE_MESH_ENCRYPTED:
2809       emsg = (struct GNUNET_MESH_Encrypted *) data;
2810       ttl = ntohl (emsg->ttl);
2811       if (0 == ttl)
2812       {
2813         GNUNET_break_op (0);
2814         GNUNET_free (data);
2815         return NULL;
2816       }
2817       emsg->cid = c->id;
2818       emsg->ttl = htonl (ttl - 1);
2819       emsg->pid = htonl (fc->next_pid++);
2820       LOG (GNUNET_ERROR_TYPE_DEBUG, "  Q_N+ %p %u\n", fc, fc->queue_n);
2821       if (GNUNET_YES == droppable)
2822       {
2823         fc->queue_n++;
2824         LOG (GNUNET_ERROR_TYPE_DEBUG, "pid %u\n", ntohl (emsg->pid));
2825         LOG (GNUNET_ERROR_TYPE_DEBUG, "last pid sent %u\n", fc->last_pid_sent);
2826         LOG (GNUNET_ERROR_TYPE_DEBUG, "     ack recv %u\n", fc->last_ack_recv);
2827       }
2828       else
2829       {
2830         LOG (GNUNET_ERROR_TYPE_DEBUG, "  not droppable, Q_N stays the same\n");
2831       }
2832       if (GM_is_pid_bigger (fc->last_pid_sent + 1, fc->last_ack_recv))
2833       {
2834         GMC_start_poll (c, fwd);
2835       }
2836       break;
2837
2838     case GNUNET_MESSAGE_TYPE_MESH_KX:
2839       kmsg = (struct GNUNET_MESH_KX *) data;
2840       kmsg->cid = c->id;
2841       break;
2842
2843     case GNUNET_MESSAGE_TYPE_MESH_ACK:
2844       amsg = (struct GNUNET_MESH_ACK *) data;
2845       amsg->cid = c->id;
2846       LOG (GNUNET_ERROR_TYPE_DEBUG, " ack %u\n", ntohl (amsg->ack));
2847       droppable = GNUNET_NO;
2848       break;
2849
2850     case GNUNET_MESSAGE_TYPE_MESH_POLL:
2851       pmsg = (struct GNUNET_MESH_Poll *) data;
2852       pmsg->cid = c->id;
2853       LOG (GNUNET_ERROR_TYPE_DEBUG, " poll %u\n", ntohl (pmsg->pid));
2854       droppable = GNUNET_NO;
2855       break;
2856
2857     case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_DESTROY:
2858       dmsg = (struct GNUNET_MESH_ConnectionDestroy *) data;
2859       dmsg->cid = c->id;
2860       break;
2861
2862     case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_BROKEN:
2863       bmsg = (struct GNUNET_MESH_ConnectionBroken *) data;
2864       bmsg->cid = c->id;
2865       break;
2866
2867     case GNUNET_MESSAGE_TYPE_MESH_KEEPALIVE:
2868       GNUNET_break (0);
2869       /* falltrough */
2870     case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE:
2871     case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK:
2872       break;
2873
2874     default:
2875       GNUNET_break (0);
2876       GNUNET_free (data);
2877       return NULL;
2878   }
2879
2880   if (fc->queue_n > fc->queue_max && droppable)
2881   {
2882     GNUNET_STATISTICS_update (stats, "# messages dropped (buffer full)",
2883                               1, GNUNET_NO);
2884     GNUNET_break (0);
2885     LOG (GNUNET_ERROR_TYPE_DEBUG,
2886                 "queue full: %u/%u\n",
2887                 fc->queue_n, fc->queue_max);
2888     if (GNUNET_MESSAGE_TYPE_MESH_ENCRYPTED == type)
2889     {
2890       fc->queue_n--;
2891       fc->next_pid--;
2892     }
2893     GNUNET_free (data);
2894     return NULL; /* Drop this message */
2895   }
2896
2897   LOG (GNUNET_ERROR_TYPE_DEBUG, "  C_P+ %p %u\n", c, c->pending_messages);
2898   c->pending_messages++;
2899
2900   q = GNUNET_new (struct MeshConnectionQueue);
2901   q->forced = !droppable;
2902   q->q = GMP_queue_add (get_hop (c, fwd), data, type, size, c, fwd,
2903                         &message_sent, q);
2904   if (NULL == q->q)
2905   {
2906     LOG (GNUNET_ERROR_TYPE_DEBUG, "WARNING dropping msg on %s\n", GMC_2s (c));
2907     GNUNET_free (data);
2908     GNUNET_free (q);
2909     return NULL;
2910   }
2911   q->cont = cont;
2912   q->cont_cls = cont_cls;
2913   return q;
2914 }
2915
2916
2917 /**
2918  * Cancel a previously sent message while it's in the queue.
2919  *
2920  * ONLY can be called before the continuation given to the send function
2921  * is called. Once the continuation is called, the message is no longer in the
2922  * queue.
2923  *
2924  * @param q Handle to the queue.
2925  */
2926 void
2927 GMC_cancel (struct MeshConnectionQueue *q)
2928 {
2929   LOG (GNUNET_ERROR_TYPE_DEBUG, "!  GMC cancel message\n");
2930
2931   /* queue destroy calls message_sent, which calls q->cont and frees q */
2932   GMP_queue_destroy (q->q, GNUNET_YES);
2933 }
2934
2935
2936 /**
2937  * Sends a CREATE CONNECTION message for a path to a peer.
2938  * Changes the connection and tunnel states if necessary.
2939  *
2940  * @param connection Connection to create.
2941  */
2942 void
2943 GMC_send_create (struct MeshConnection *connection)
2944 {
2945   enum MeshTunnel3CState state;
2946   size_t size;
2947
2948   size = sizeof (struct GNUNET_MESH_ConnectionCreate);
2949   size += connection->path->length * sizeof (struct GNUNET_PeerIdentity);
2950
2951   LOG (GNUNET_ERROR_TYPE_INFO, "=> %s on connection %s  (%u bytes)\n",
2952        GM_m2s (GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE),
2953        GMC_2s (connection), size);
2954   LOG (GNUNET_ERROR_TYPE_DEBUG, "  C_P+ %p %u (create)\n",
2955        connection, connection->pending_messages);
2956   connection->pending_messages++;
2957
2958   connection->maintenance_q =
2959     GMP_queue_add (get_next_hop (connection), NULL,
2960                    GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE,
2961                    size, connection, GNUNET_YES, &message_sent, NULL);
2962
2963   state = GMT_get_cstate (connection->t);
2964   if (MESH_TUNNEL3_SEARCHING == state || MESH_TUNNEL3_NEW == state)
2965     GMT_change_cstate (connection->t, MESH_TUNNEL3_WAITING);
2966   if (MESH_CONNECTION_NEW == connection->state)
2967     connection_change_state (connection, MESH_CONNECTION_SENT);
2968 }
2969
2970
2971 /**
2972  * Send a message to all peers in this connection that the connection
2973  * is no longer valid.
2974  *
2975  * If some peer should not receive the message, it should be zero'ed out
2976  * before calling this function.
2977  *
2978  * @param c The connection whose peers to notify.
2979  */
2980 void
2981 GMC_send_destroy (struct MeshConnection *c)
2982 {
2983   struct GNUNET_MESH_ConnectionDestroy msg;
2984
2985   if (GNUNET_YES == c->destroy)
2986     return;
2987
2988   msg.header.size = htons (sizeof (msg));
2989   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_CONNECTION_DESTROY);;
2990   msg.cid = c->id;
2991   LOG (GNUNET_ERROR_TYPE_DEBUG,
2992               "  sending connection destroy for connection %s\n",
2993               GMC_2s (c));
2994
2995   if (GNUNET_NO == GMC_is_terminal (c, GNUNET_YES))
2996     GMC_send_prebuilt_message (&msg.header, c,
2997                                GNUNET_YES, GNUNET_YES, NULL, NULL);
2998   if (GNUNET_NO == GMC_is_terminal (c, GNUNET_NO))
2999     GMC_send_prebuilt_message (&msg.header, c,
3000                                GNUNET_NO, GNUNET_YES, NULL, NULL);
3001   c->destroy = GNUNET_YES;
3002   c->state = MESH_CONNECTION_DESTROYED;
3003 }
3004
3005
3006 /**
3007  * @brief Start a polling timer for the connection.
3008  *
3009  * When a neighbor does not accept more traffic on the connection it could be
3010  * caused by a simple congestion or by a lost ACK. Polling enables to check
3011  * for the lastest ACK status for a connection.
3012  *
3013  * @param c Connection.
3014  * @param fwd Should we poll in the FWD direction?
3015  */
3016 void
3017 GMC_start_poll (struct MeshConnection *c, int fwd)
3018 {
3019   struct MeshFlowControl *fc;
3020
3021   fc = fwd ? &c->fwd_fc : &c->bck_fc;
3022   LOG (GNUNET_ERROR_TYPE_DEBUG, " *** POLL %s requested\n",
3023        GM_f2s (fwd));
3024   if (GNUNET_SCHEDULER_NO_TASK != fc->poll_task || NULL != fc->poll_msg)
3025   {
3026     LOG (GNUNET_ERROR_TYPE_DEBUG, " ***   not needed (%u, %p)\n",
3027          fc->poll_task, fc->poll_msg);
3028     return;
3029   }
3030   LOG (GNUNET_ERROR_TYPE_DEBUG, " *** POLL started on request\n");
3031   fc->poll_task = GNUNET_SCHEDULER_add_delayed (fc->poll_time,
3032                                                 &connection_poll,
3033                                                 fc);
3034 }
3035
3036
3037 /**
3038  * @brief Stop polling a connection for ACKs.
3039  *
3040  * Once we have enough ACKs for future traffic, polls are no longer necessary.
3041  *
3042  * @param c Connection.
3043  * @param fwd Should we stop the poll in the FWD direction?
3044  */
3045 void
3046 GMC_stop_poll (struct MeshConnection *c, int fwd)
3047 {
3048   struct MeshFlowControl *fc;
3049
3050   fc = fwd ? &c->fwd_fc : &c->bck_fc;
3051   if (GNUNET_SCHEDULER_NO_TASK != fc->poll_task)
3052   {
3053     GNUNET_SCHEDULER_cancel (fc->poll_task);
3054     fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
3055   }
3056 }
3057
3058 /**
3059  * Get a (static) string for a connection.
3060  *
3061  * @param c Connection.
3062  */
3063 const char *
3064 GMC_2s (const struct MeshConnection *c)
3065 {
3066   if (NULL == c)
3067     return "NULL";
3068
3069   if (NULL != c->t)
3070   {
3071     static char buf[128];
3072
3073     sprintf (buf, "%s (->%s)", GNUNET_h2s (GM_h2hc (GMC_get_id(c))), GMT_2s (c->t));
3074     return buf;
3075   }
3076   return GNUNET_h2s (GM_h2hc (&c->id));
3077 }