transport API changes in preparation for the storm
[oweals/gnunet.git] / src / transport / plugin_transport_tcp.c
1 /*
2      This file is part of GNUnet
3      (C) 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010 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 2, 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  * @file transport/plugin_transport_tcp.c
22  * @brief Implementation of the TCP transport service
23  * @author Christian Grothoff
24  */
25 #include "platform.h"
26 #include "gnunet_hello_lib.h"
27 #include "gnunet_connection_lib.h"
28 #include "gnunet_container_lib.h"
29 #include "gnunet_os_lib.h"
30 #include "gnunet_protocols.h"
31 #include "gnunet_resolver_service.h"
32 #include "gnunet_server_lib.h"
33 #include "gnunet_service_lib.h"
34 #include "gnunet_signatures.h"
35 #include "gnunet_statistics_service.h"
36 #include "gnunet_transport_service.h"
37 #include "plugin_transport.h"
38 #include "transport.h"
39
40 #define DEBUG_TCP GNUNET_YES
41
42 /**
43  * How long until we give up on transmitting the welcome message?
44  */
45 #define HOSTNAME_RESOLVE_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5)
46
47
48 /**
49  * Initial handshake message for a session.
50  */
51 struct WelcomeMessage
52 {
53   /**
54    * Type is GNUNET_MESSAGE_TYPE_TRANSPORT_TCP_WELCOME.
55    */
56   struct GNUNET_MessageHeader header;
57
58   /**
59    * Identity of the node connecting (TCP client)
60    */
61   struct GNUNET_PeerIdentity clientIdentity;
62
63 };
64
65
66 /**
67  * Encapsulation of all of the state of the plugin.
68  */
69 struct Plugin;
70
71
72 /**
73  * Information kept for each message that is yet to
74  * be transmitted.
75  */
76 struct PendingMessage
77 {
78
79   /**
80    * This is a doubly-linked list.
81    */
82   struct PendingMessage *next;
83
84   /**
85    * This is a doubly-linked list.
86    */
87   struct PendingMessage *prev;
88
89   /**
90    * The pending message
91    */
92   const char *msg;
93
94   /**
95    * Continuation function to call once the message
96    * has been sent.  Can be NULL if there is no
97    * continuation to call.
98    */
99   GNUNET_TRANSPORT_TransmitContinuation transmit_cont;
100
101   /**
102    * Closure for transmit_cont.
103    */
104   void *transmit_cont_cls;
105
106   /**
107    * Timeout value for the pending message.
108    */
109   struct GNUNET_TIME_Absolute timeout;
110
111   /**
112    * So that the gnunet-service-transport can group messages together,
113    * these pending messages need to accept a message buffer and size
114    * instead of just a GNUNET_MessageHeader.
115    */
116   size_t message_size;
117
118 };
119
120
121 /**
122  * Session handle for TCP connections.
123  */
124 struct Session
125 {
126
127   /**
128    * Stored in a linked list.
129    */
130   struct Session *next;
131
132   /**
133    * Pointer to the global plugin struct.
134    */
135   struct Plugin *plugin;
136
137   /**
138    * The client (used to identify this connection)
139    */
140   struct GNUNET_SERVER_Client *client;
141
142   /**
143    * Messages currently pending for transmission
144    * to this peer, if any.
145    */
146   struct PendingMessage *pending_messages_head;
147
148   /**
149    * Messages currently pending for transmission
150    * to this peer, if any.
151    */
152   struct PendingMessage *pending_messages_tail;
153
154   /**
155    * Handle for pending transmission request.
156    */
157   struct GNUNET_CONNECTION_TransmitHandle *transmit_handle;
158
159   /**
160    * To whom are we talking to (set to our identity
161    * if we are still waiting for the welcome message)
162    */
163   struct GNUNET_PeerIdentity target;
164
165   /**
166    * ID of task used to delay receiving more to throttle sender.
167    */
168   GNUNET_SCHEDULER_TaskIdentifier receive_delay_task;
169
170   /**
171    * Address of the other peer (either based on our 'connect'
172    * call or on our 'accept' call).
173    */
174   void *connect_addr;
175
176   /**
177    * Last activity on this connection.  Used to select preferred
178    * connection.
179    */
180   struct GNUNET_TIME_Absolute last_activity;
181
182   /**
183    * Length of connect_addr.
184    */
185   size_t connect_alen;
186
187   /**
188    * Are we still expecting the welcome message? (GNUNET_YES/GNUNET_NO)
189    */
190   int expecting_welcome;
191
192   /**
193    * Was this a connection that was inbound (we accepted)? (GNUNET_YES/GNUNET_NO)
194    */
195   int inbound;
196
197 };
198
199
200 /**
201  * Encapsulation of all of the state of the plugin.
202  */
203 struct Plugin
204 {
205   /**
206    * Our environment.
207    */
208   struct GNUNET_TRANSPORT_PluginEnvironment *env;
209
210   /**
211    * The listen socket.
212    */
213   struct GNUNET_CONNECTION_Handle *lsock;
214
215   /**
216    * List of open TCP sessions.
217    */
218   struct Session *sessions;
219
220   /**
221    * Handle for the statistics service.
222    */
223   struct GNUNET_STATISTICS_Handle *statistics;
224
225   /**
226    * Handle to the network service.
227    */
228   struct GNUNET_SERVICE_Context *service;
229
230   /**
231    * Handle to the server for this service.
232    */
233   struct GNUNET_SERVER_Handle *server;
234
235   /**
236    * Copy of the handler array where the closures are
237    * set to this struct's instance.
238    */
239   struct GNUNET_SERVER_MessageHandler *handlers;
240
241   /**
242    * Handle for request of hostname resolution, non-NULL if pending.
243    */
244   struct GNUNET_RESOLVER_RequestHandle *hostname_dns;
245
246   /**
247    * ID of task used to update our addresses when one expires.
248    */
249   GNUNET_SCHEDULER_TaskIdentifier address_update_task;
250
251   /**
252    * Port that we are actually listening on.
253    */
254   uint16_t open_port;
255
256   /**
257    * Port that the user said we would have visible to the
258    * rest of the world.
259    */
260   uint16_t adv_port;
261
262 };
263
264
265 /**
266  * Find the session handle for the given client.
267  *
268  * @return NULL if no matching session exists
269  */
270 static struct Session *
271 find_session_by_client (struct Plugin *plugin,
272                         const struct GNUNET_SERVER_Client *client)
273 {
274   struct Session *ret;
275
276   ret = plugin->sessions;
277   while ((ret != NULL) && (client != ret->client))
278     ret = ret->next;
279   return ret;
280 }
281
282
283 /**
284  * Create a new session.  Also queues a welcome message.
285  *
286  * @param plugin us
287  * @param target peer to connect to
288  * @param client client to use
289  * @return new session object
290  */
291 static struct Session *
292 create_session (struct Plugin *plugin,
293                 const struct GNUNET_PeerIdentity *target,
294                 struct GNUNET_SERVER_Client *client)
295 {
296   struct Session *ret;
297   struct PendingMessage *pm;
298   struct WelcomeMessage welcome;
299
300   GNUNET_assert (client != NULL);
301   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
302                    "tcp",
303                    "Creating new session for peer `%4s'\n",
304                    GNUNET_i2s (target));
305   ret = GNUNET_malloc (sizeof (struct Session));
306   ret->last_activity = GNUNET_TIME_absolute_get ();
307   ret->plugin = plugin;
308   ret->next = plugin->sessions;
309   plugin->sessions = ret;
310   ret->client = client;
311   ret->target = *target;
312   ret->expecting_welcome = GNUNET_YES;
313   pm = GNUNET_malloc (sizeof (struct PendingMessage) + sizeof (struct WelcomeMessage));
314   pm->msg = (const char*) &pm[1];
315   pm->message_size = sizeof (struct WelcomeMessage);
316   welcome.header.size = htons (sizeof (struct WelcomeMessage));
317   welcome.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_TCP_WELCOME);
318   welcome.clientIdentity = *plugin->env->my_identity;
319   memcpy (&pm[1], &welcome, sizeof (welcome));
320   pm->timeout = GNUNET_TIME_UNIT_FOREVER_ABS;
321   GNUNET_STATISTICS_update (plugin->env->stats,
322                             gettext_noop ("# bytes currently in TCP buffers"),
323                             pm->message_size,
324                             GNUNET_NO);      
325   GNUNET_CONTAINER_DLL_insert (ret->pending_messages_head,
326                                ret->pending_messages_tail,
327                                pm);
328   GNUNET_STATISTICS_update (plugin->env->stats,
329                             gettext_noop ("# TCP sessions active"),
330                             1,
331                             GNUNET_NO);      
332   return ret;
333 }
334
335
336 /**
337  * If we have pending messages, ask the server to
338  * transmit them (schedule the respective tasks, etc.)
339  *
340  * @param session for which session should we do this
341  */
342 static void process_pending_messages (struct Session *session);
343
344
345 /**
346  * Function called to notify a client about the socket
347  * being ready to queue more data.  "buf" will be
348  * NULL and "size" zero if the socket was closed for
349  * writing in the meantime.
350  *
351  * @param cls closure
352  * @param size number of bytes available in buf
353  * @param buf where the callee should write the message
354  * @return number of bytes written to buf
355  */
356 static size_t
357 do_transmit (void *cls, size_t size, void *buf)
358 {
359   struct Session *session = cls;
360   struct GNUNET_PeerIdentity pid;
361   struct Plugin *plugin;
362   struct PendingMessage *pos;
363   struct PendingMessage *hd;
364   struct PendingMessage *tl;
365   struct GNUNET_TIME_Absolute now;
366   char *cbuf;
367   size_t ret;
368
369   session->transmit_handle = NULL;
370   plugin = session->plugin;
371   if (buf == NULL)
372     {
373 #if DEBUG_TCP
374       GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
375                        "tcp",
376                        "Timeout trying to transmit to peer `%4s', discarding message queue.\n",
377                        GNUNET_i2s (&session->target));
378 #endif
379       /* timeout; cancel all messages that have already expired */
380       hd = NULL;
381       tl = NULL;
382       ret = 0;
383       now = GNUNET_TIME_absolute_get ();
384       while ( (NULL != (pos = session->pending_messages_head)) &&
385               (pos->timeout.value <= now.value) )
386         {
387           GNUNET_CONTAINER_DLL_remove (session->pending_messages_head,
388                                        session->pending_messages_tail,
389                                        pos);
390 #if DEBUG_TCP
391           GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
392                            "tcp",
393                            "Failed to transmit %u byte message to `%4s'.\n",
394                            pos->message_size,
395                            GNUNET_i2s (&session->target));
396 #endif
397           ret += pos->message_size;
398           GNUNET_CONTAINER_DLL_insert_after (hd, tl, tl, pos);
399         }
400       /* do this call before callbacks (so that if callbacks destroy
401          session, they have a chance to cancel actions done by this
402          call) */
403       process_pending_messages (session);  
404       pid = session->target;
405       /* no do callbacks and do not use session again since
406          the callbacks may abort the session */
407       while (NULL != (pos = hd))
408         {
409           GNUNET_CONTAINER_DLL_remove (hd, tl, pos);
410           if (pos->transmit_cont != NULL)
411             pos->transmit_cont (pos->transmit_cont_cls,
412                                 &pid, GNUNET_SYSERR);
413           GNUNET_free (pos);
414         }
415       GNUNET_STATISTICS_update (plugin->env->stats,
416                                 gettext_noop ("# bytes currently in TCP buffers"),
417                                 - (int64_t) ret,
418                                 GNUNET_NO); 
419       GNUNET_STATISTICS_update (plugin->env->stats,
420                                 gettext_noop ("# bytes discarded by TCP (timeout)"),
421                                 ret,
422                                 GNUNET_NO);      
423       return 0;
424     }
425   /* copy all pending messages that would fit */
426   ret = 0;
427   cbuf = buf;
428   hd = NULL;
429   tl = NULL;
430   while (NULL != (pos = session->pending_messages_head)) 
431     {
432       if (ret + pos->message_size > size) 
433         break;
434       GNUNET_CONTAINER_DLL_remove (session->pending_messages_head,
435                                    session->pending_messages_tail,
436                                    pos);
437       GNUNET_assert (size >= pos->message_size);
438       memcpy (cbuf, pos->msg, pos->message_size);
439       cbuf += pos->message_size;
440       ret += pos->message_size;
441       size -= pos->message_size;
442       GNUNET_CONTAINER_DLL_insert_after (hd, tl, tl, pos);
443     }
444   /* schedule 'continuation' before callbacks so that callbacks that
445      cancel everything don't cause us to use a session that no longer
446      exists... */
447   process_pending_messages (session);  
448   session->last_activity = GNUNET_TIME_absolute_get ();
449   pid = session->target;
450   /* we'll now call callbacks that may cancel the session; hence
451      we should not use 'session' after this point */
452   while (NULL != (pos = hd))
453     {
454       GNUNET_CONTAINER_DLL_remove (hd, tl, pos);
455       if (pos->transmit_cont != NULL)
456         pos->transmit_cont (pos->transmit_cont_cls,
457                             &pid, GNUNET_OK);
458       GNUNET_free (pos);
459     }
460   GNUNET_assert (hd == NULL);
461   GNUNET_assert (tl == NULL);
462 #if DEBUG_TCP > 1
463   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
464                    "tcp", "Transmitting %u bytes\n", ret);
465 #endif
466   GNUNET_STATISTICS_update (plugin->env->stats,
467                             gettext_noop ("# bytes currently in TCP buffers"),
468                             - (int64_t) ret,
469                             GNUNET_NO);       
470   GNUNET_STATISTICS_update (plugin->env->stats,
471                             gettext_noop ("# bytes transmitted via TCP"),
472                             ret,
473                             GNUNET_NO);      
474   return ret;
475 }
476
477
478 /**
479  * If we have pending messages, ask the server to
480  * transmit them (schedule the respective tasks, etc.)
481  *
482  * @param session for which session should we do this
483  */
484 static void
485 process_pending_messages (struct Session *session)
486 {
487   struct PendingMessage *pm;
488
489   GNUNET_assert (session->client != NULL);
490   if (session->transmit_handle != NULL)
491     return;
492   if (NULL == (pm = session->pending_messages_head))
493     return;
494   session->transmit_handle
495     = GNUNET_SERVER_notify_transmit_ready (session->client,
496                                            pm->message_size,
497                                            GNUNET_TIME_absolute_get_remaining
498                                            (pm->timeout),
499                                            &do_transmit, session);
500 }
501
502
503 /**
504  * Functions with this signature are called whenever we need
505  * to close a session due to a disconnect or failure to
506  * establish a connection.
507  *
508  * @param session session to close down
509  */
510 static void
511 disconnect_session (struct Session *session)
512 {
513   struct Session *prev;
514   struct Session *pos;
515   struct PendingMessage *pm;
516
517 #if DEBUG_TCP
518   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
519                    "tcp",
520                    "Disconnecting from `%4s' at %s (session %p).\n",
521                    GNUNET_i2s (&session->target),
522                    (session->connect_addr != NULL) ?
523                    GNUNET_a2s (session->connect_addr,
524                                session->connect_alen) : "*", session);
525 #endif
526   /* remove from session list */
527   prev = NULL;
528   pos = session->plugin->sessions;
529   while (pos != session)
530     {
531       prev = pos;
532       pos = pos->next;
533     }
534   if (prev == NULL)
535     session->plugin->sessions = session->next;
536   else
537     prev->next = session->next;
538   /* clean up state */
539   if (session->transmit_handle != NULL)
540     {
541       GNUNET_CONNECTION_notify_transmit_ready_cancel
542         (session->transmit_handle);
543       session->transmit_handle = NULL;
544     }
545   while (NULL != (pm = session->pending_messages_head))
546     {
547 #if DEBUG_TCP
548       GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
549                        "tcp",
550                        pm->transmit_cont != NULL
551                        ? "Could not deliver message to `%4s'.\n"
552                        :
553                        "Could not deliver message to `%4s', notifying.\n",
554                        GNUNET_i2s (&session->target));
555 #endif
556       GNUNET_STATISTICS_update (session->plugin->env->stats,
557                                 gettext_noop ("# bytes currently in TCP buffers"),
558                                 - (int64_t) pm->message_size,
559                                 GNUNET_NO);      
560       GNUNET_STATISTICS_update (session->plugin->env->stats,
561                                 gettext_noop ("# bytes discarded by TCP (disconnect)"),
562                                 pm->message_size,
563                                 GNUNET_NO);      
564       GNUNET_CONTAINER_DLL_remove (session->pending_messages_head,
565                                    session->pending_messages_tail,
566                                    pm);
567       if (NULL != pm->transmit_cont)
568         pm->transmit_cont (pm->transmit_cont_cls,
569                            &session->target, GNUNET_SYSERR);
570       GNUNET_free (pm);
571     }
572   GNUNET_break (session->client != NULL);
573   if (session->receive_delay_task != GNUNET_SCHEDULER_NO_TASK)
574     {
575       GNUNET_SCHEDULER_cancel (session->plugin->env->sched,
576                                session->receive_delay_task);
577       GNUNET_SERVER_receive_done (session->client, 
578                                   GNUNET_SYSERR);       
579     }
580   GNUNET_SERVER_client_drop (session->client);
581   GNUNET_STATISTICS_update (session->plugin->env->stats,
582                             gettext_noop ("# TCP sessions active"),
583                             -1,
584                             GNUNET_NO);      
585   GNUNET_free_non_null (session->connect_addr);
586   GNUNET_free (session);
587 }
588
589
590 /**
591  * Given two otherwise equivalent sessions, pick the better one.
592  * 
593  * @param s1 one session (also default)
594  * @param s2 other session
595  * @return "better" session (more active)
596  */
597 static struct Session *
598 select_better_session (struct Session *s1,
599                        struct Session *s2)
600 {
601   if (s1 == NULL)
602     return s2;
603   if (s2 == NULL)
604     return s1;
605   if ( (s1->expecting_welcome == GNUNET_NO) &&
606        (s2->expecting_welcome == GNUNET_YES) )
607     return s1;
608   if ( (s1->expecting_welcome == GNUNET_YES) &&
609        (s2->expecting_welcome == GNUNET_NO) )
610     return s2;
611   if (s1->last_activity.value < s2->last_activity.value)
612     return s2;
613   if (s1->last_activity.value > s2->last_activity.value)
614     return s1;
615   if ( (GNUNET_YES == s1->inbound) &&
616        (GNUNET_NO  == s2->inbound) )
617     return s1;
618   if ( (GNUNET_NO  == s1->inbound) &&
619        (GNUNET_YES == s2->inbound) )
620     return s2;
621   return s1;
622 }
623
624
625 /**
626  * Function that can be used by the transport service to transmit
627  * a message using the plugin.   Note that in the case of a
628  * peer disconnecting, the continuation MUST be called
629  * prior to the disconnect notification itself.  This function
630  * will be called with this peer's HELLO message to initiate
631  * a fresh connection to another peer.
632  *
633  * @param cls closure
634  * @param target who should receive this message
635  * @param msg the message to transmit
636  * @param msgbuf_size number of bytes in 'msg'
637  * @param priority how important is the message (most plugins will
638  *                 ignore message priority and just FIFO)
639  * @param timeout how long to wait at most for the transmission (does not
640  *                require plugins to discard the message after the timeout,
641  *                just advisory for the desired delay; most plugins will ignore
642  *                this as well)
643  * @param session which session must be used (or NULL for "any")
644  * @param addr the address to use (can be NULL if the plugin
645  *                is "on its own" (i.e. re-use existing TCP connection))
646  * @param addrlen length of the address in bytes
647  * @param force_address GNUNET_YES if the plugin MUST use the given address,
648  *                GNUNET_NO means the plugin may use any other address and
649  *                GNUNET_SYSERR means that only reliable existing
650  *                bi-directional connections should be used (regardless
651  *                of address)
652  * @param cont continuation to call once the message has
653  *        been transmitted (or if the transport is ready
654  *        for the next transmission call; or if the
655  *        peer disconnected...); can be NULL
656  * @param cont_cls closure for cont
657  * @return number of bytes used (on the physical network, with overheads);
658  *         -1 on hard errors (i.e. address invalid); 0 is a legal value
659  *         and does NOT mean that the message was not transmitted (DV)
660  */
661 static ssize_t
662 tcp_plugin_send (void *cls,
663                  const struct GNUNET_PeerIdentity *target,
664                  const char *msg,
665                  size_t msgbuf_size,
666                  uint32_t priority,
667                  struct GNUNET_TIME_Relative timeout,
668                  struct Session *session,
669                  const void *addr,
670                  size_t addrlen,
671                  int force_address,
672                  GNUNET_TRANSPORT_TransmitContinuation cont, void *cont_cls)
673 {
674   struct Plugin *plugin = cls;
675   struct Session *cand_session;
676   struct Session *next;
677   struct PendingMessage *pm;
678   struct GNUNET_CONNECTION_Handle *sa;
679   int af;
680
681   GNUNET_STATISTICS_update (plugin->env->stats,
682                             gettext_noop ("# bytes TCP was asked to transmit"),
683                             msgbuf_size,
684                             GNUNET_NO);      
685   /* FIXME: we could do this a cheaper with a hash table
686      where we could restrict the iteration to entries that match
687      the target peer... */
688   if (session == NULL)
689     {
690       cand_session = NULL;
691       next = plugin->sessions;
692       while (NULL != (session = next)) 
693         {
694           next = session->next;
695           GNUNET_assert (session->client != NULL);
696           if (0 != memcmp (target,
697                            &session->target, 
698                            sizeof (struct GNUNET_PeerIdentity)))
699             continue;
700           if ( ( (GNUNET_SYSERR == force_address) &&
701                  (session->expecting_welcome == GNUNET_NO) ) ||
702                (GNUNET_NO == force_address) )   
703             {
704               cand_session = select_better_session (cand_session,
705                                                     session);
706               continue;
707             }
708           if (GNUNET_SYSERR == force_address)
709             continue;
710           GNUNET_break (GNUNET_YES == force_address);
711           if (addr == NULL)
712             {
713               GNUNET_break (0);
714               break;
715             }
716           if (session->inbound == GNUNET_YES) 
717             continue;
718           if (addrlen != session->connect_alen)
719             continue;
720           if (0 != memcmp (session->connect_addr,
721                            addr,
722                            addrlen))
723             continue;
724           cand_session = select_better_session (cand_session,
725                                                 session);             
726         }
727       session = cand_session;
728     }
729   if ( (session == NULL) &&
730        (addr == NULL) )
731     {
732 #if DEBUG_TCP
733       GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
734                        "tcp",
735                        "Asked to transmit to `%4s' without address and I have no existing connection (failing).\n",
736                        GNUNET_i2s (target));
737 #endif
738       GNUNET_STATISTICS_update (plugin->env->stats,
739                                 gettext_noop ("# bytes discarded by TCP (no address and no connection)"),
740                                 msgbuf_size,
741                                 GNUNET_NO);      
742       return -1;
743     }
744   if (session == NULL)
745     {
746       if (sizeof (struct sockaddr_in) == addrlen)
747         {
748           af = AF_INET;
749         }
750       else if (sizeof (struct sockaddr_in6) == addrlen)
751         {
752           af = AF_INET6;
753         }
754       else
755         {
756           GNUNET_break_op (0);
757           return -1;
758         }
759       sa = GNUNET_CONNECTION_create_from_sockaddr (plugin->env->sched,
760                                                    af, addr, addrlen,
761                                                    GNUNET_SERVER_MAX_MESSAGE_SIZE);
762       if (sa == NULL)
763         {
764 #if DEBUG_TCP
765           GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
766                            "tcp",
767                            "Failed to create connection to `%4s' at `%s'\n",
768                            GNUNET_i2s (target),
769                            GNUNET_a2s (addr, addrlen));
770 #endif
771           GNUNET_STATISTICS_update (plugin->env->stats,
772                                     gettext_noop ("# bytes discarded by TCP (failed to connect)"),
773                                     msgbuf_size,
774                                     GNUNET_NO);      
775           return -1;
776         }
777 #if DEBUG_TCP
778       GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
779                        "tcp",
780                        "Asked to transmit to `%4s', creating fresh session using address `%s'.\n",
781                        GNUNET_i2s (target),
782                        GNUNET_a2s (addr, addrlen));
783 #endif
784       session = create_session (plugin,
785                                 target,
786                                 GNUNET_SERVER_connect_socket (plugin->server,
787                                                               sa));
788       session->connect_addr = GNUNET_malloc (addrlen);
789       memcpy (session->connect_addr,
790               addr,
791               addrlen);
792       session->connect_alen = addrlen;
793     }
794   GNUNET_assert (session != NULL);
795   GNUNET_assert (session->client != NULL);
796   GNUNET_STATISTICS_update (plugin->env->stats,
797                             gettext_noop ("# bytes currently in TCP buffers"),
798                             msgbuf_size,
799                             GNUNET_NO);      
800   /* create new message entry */
801   pm = GNUNET_malloc (sizeof (struct PendingMessage) + msgbuf_size);
802   pm->msg = (const char*) &pm[1];
803   memcpy (&pm[1], msg, msgbuf_size);
804   pm->message_size = msgbuf_size;
805   pm->timeout = GNUNET_TIME_relative_to_absolute (timeout);
806   pm->transmit_cont = cont;
807   pm->transmit_cont_cls = cont_cls;
808
809   /* append pm to pending_messages list */
810   GNUNET_CONTAINER_DLL_insert_after (session->pending_messages_head,
811                                      session->pending_messages_tail,
812                                      session->pending_messages_tail,
813                                      pm);
814 #if DEBUG_TCP
815   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
816                    "tcp",
817                    "Asked to transmit %u bytes to `%s', added message to list.\n",
818                    msgbuf_size,
819                    GNUNET_i2s (target));
820 #endif
821   process_pending_messages (session);
822   return msgbuf_size;
823 }
824
825
826 /**
827  * Function that can be called to force a disconnect from the
828  * specified neighbour.  This should also cancel all previously
829  * scheduled transmissions.  Obviously the transmission may have been
830  * partially completed already, which is OK.  The plugin is supposed
831  * to close the connection (if applicable) and no longer call the
832  * transmit continuation(s).
833  *
834  * Finally, plugin MUST NOT call the services's receive function to
835  * notify the service that the connection to the specified target was
836  * closed after a getting this call.
837  *
838  * @param cls closure
839  * @param target peer for which the last transmission is
840  *        to be cancelled
841  */
842 static void
843 tcp_plugin_disconnect (void *cls,
844                        const struct GNUNET_PeerIdentity *target)
845 {
846   struct Plugin *plugin = cls;
847   struct Session *session;
848   struct Session *next;
849   struct PendingMessage *pm;
850
851 #if DEBUG_TCP
852   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
853                    "tcp",
854                    "Asked to cancel session with `%4s'\n",
855                    GNUNET_i2s (target));
856 #endif
857   next = plugin->sessions;
858   while (NULL != (session = next))
859     {
860       next = session->next;
861       if (0 != memcmp (target,
862                        &session->target,
863                        sizeof (struct GNUNET_PeerIdentity)))
864         continue;
865       pm = session->pending_messages_head;
866       while (pm != NULL)
867         {
868           pm->transmit_cont = NULL;
869           pm->transmit_cont_cls = NULL;
870           pm = pm->next;
871         }
872       disconnect_session (session);
873     }
874 }
875
876
877 /**
878  * Context for address to string conversion.
879  */
880 struct PrettyPrinterContext
881 {
882   /**
883    * Function to call with the result.
884    */
885   GNUNET_TRANSPORT_AddressStringCallback asc;
886
887   /**
888    * Clsoure for 'asc'.
889    */
890   void *asc_cls;
891
892   /**
893    * Port to add after the IP address.
894    */
895   uint16_t port;
896 };
897
898
899 /**
900  * Append our port and forward the result.
901  *
902  * @param cls the 'struct PrettyPrinterContext*'
903  * @param hostname hostname part of the address
904  */
905 static void
906 append_port (void *cls, const char *hostname)
907 {
908   struct PrettyPrinterContext *ppc = cls;
909   char *ret;
910
911   if (hostname == NULL)
912     {
913       ppc->asc (ppc->asc_cls, NULL);
914       GNUNET_free (ppc);
915       return;
916     }
917   GNUNET_asprintf (&ret, "%s:%d", hostname, ppc->port);
918   ppc->asc (ppc->asc_cls, ret);
919   GNUNET_free (ret);
920 }
921
922
923 /**
924  * Convert the transports address to a nice, human-readable
925  * format.
926  *
927  * @param cls closure
928  * @param type name of the transport that generated the address
929  * @param addr one of the addresses of the host, NULL for the last address
930  *        the specific address format depends on the transport
931  * @param addrlen length of the address
932  * @param numeric should (IP) addresses be displayed in numeric form?
933  * @param timeout after how long should we give up?
934  * @param asc function to call on each string
935  * @param asc_cls closure for asc
936  */
937 static void
938 tcp_plugin_address_pretty_printer (void *cls,
939                                    const char *type,
940                                    const void *addr,
941                                    size_t addrlen,
942                                    int numeric,
943                                    struct GNUNET_TIME_Relative timeout,
944                                    GNUNET_TRANSPORT_AddressStringCallback asc,
945                                    void *asc_cls)
946 {
947   struct Plugin *plugin = cls;
948   const struct sockaddr_in *v4;
949   const struct sockaddr_in6 *v6;
950   struct PrettyPrinterContext *ppc;
951
952   if ((addrlen != sizeof (struct sockaddr_in)) &&
953       (addrlen != sizeof (struct sockaddr_in6)))
954     {
955       /* invalid address */
956       GNUNET_break_op (0);
957       asc (asc_cls, NULL);
958       return;
959     }
960   ppc = GNUNET_malloc (sizeof (struct PrettyPrinterContext));
961   ppc->asc = asc;
962   ppc->asc_cls = asc_cls;
963   if (addrlen == sizeof (struct sockaddr_in))
964     {
965       v4 = (const struct sockaddr_in *) addr;
966       ppc->port = ntohs (v4->sin_port);
967     }
968   else
969     {
970       v6 = (const struct sockaddr_in6 *) addr;
971       ppc->port = ntohs (v6->sin6_port);
972
973     }
974   GNUNET_RESOLVER_hostname_get (plugin->env->sched,
975                                 plugin->env->cfg,
976                                 addr,
977                                 addrlen,
978                                 !numeric, timeout, &append_port, ppc);
979 }
980
981
982 /**
983  * Check if the given port is plausible (must be either
984  * our listen port or our advertised port).  If it is
985  * neither, we return one of these two ports at random.
986  *
987  * @param plugin global variables
988  * @param in_port port number to check
989  * @return either in_port or a more plausible port
990  */
991 static uint16_t
992 check_port (struct Plugin *plugin, uint16_t in_port)
993 {
994   if ((in_port == plugin->adv_port) || (in_port == plugin->open_port))
995     return in_port;
996   return (GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
997                                     2) == 0)
998     ? plugin->open_port : plugin->adv_port;
999 }
1000
1001
1002 /**
1003  * Another peer has suggested an address for this peer and transport
1004  * plugin.  Check that this could be a valid address.
1005  *
1006  * @param cls closure, our 'struct Plugin*'
1007  * @param addr pointer to the address
1008  * @param addrlen length of addr
1009  * @return GNUNET_OK if this is a plausible address for this peer
1010  *         and transport
1011  */
1012 static int
1013 tcp_plugin_check_address (void *cls, void *addr, size_t addrlen)
1014 {
1015   struct Plugin *plugin = cls;
1016   char buf[sizeof (struct sockaddr_in6)];
1017   struct sockaddr_in *v4;
1018   struct sockaddr_in6 *v6;
1019
1020   if ((addrlen != sizeof (struct sockaddr_in)) &&
1021       (addrlen != sizeof (struct sockaddr_in6)))
1022     {
1023       GNUNET_break_op (0);
1024       return GNUNET_SYSERR;
1025     }
1026   memcpy (buf, addr, sizeof (struct sockaddr_in6));
1027   if (addrlen == sizeof (struct sockaddr_in))
1028     {
1029       v4 = (struct sockaddr_in *) buf;
1030       v4->sin_port = htons (check_port (plugin, ntohs (v4->sin_port)));
1031     }
1032   else
1033     {
1034       v6 = (struct sockaddr_in6 *) buf;
1035       v6->sin6_port = htons (check_port (plugin, ntohs (v6->sin6_port)));
1036     }
1037 #if DEBUG_TCP
1038   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
1039                    "tcp",
1040                    "Informing transport service about my address `%s'.\n",
1041                    GNUNET_a2s (addr, addrlen));
1042 #endif
1043   return GNUNET_OK;
1044 }
1045
1046
1047 /**
1048  * We've received a welcome from this peer via TCP.  Possibly create a
1049  * fresh client record and send back our welcome.
1050  *
1051  * @param cls closure
1052  * @param client identification of the client
1053  * @param message the actual message
1054  */
1055 static void
1056 handle_tcp_welcome (void *cls,
1057                     struct GNUNET_SERVER_Client *client,
1058                     const struct GNUNET_MessageHeader *message)
1059 {
1060   struct Plugin *plugin = cls;
1061   const struct WelcomeMessage *wm = (const struct WelcomeMessage *) message;
1062   struct Session *session;
1063   size_t alen;
1064   void *vaddr;
1065
1066 #if DEBUG_TCP
1067   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
1068                    "tcp",
1069                    "Received %s message from a `%4s/%p'.\n", 
1070                    "WELCOME",
1071                    GNUNET_i2s (&wm->clientIdentity), client);
1072 #endif
1073   GNUNET_STATISTICS_update (plugin->env->stats,
1074                             gettext_noop ("# TCP WELCOME messages received"),
1075                             1,
1076                             GNUNET_NO);      
1077   session = find_session_by_client (plugin, client);
1078   if (session == NULL)
1079     {
1080       GNUNET_SERVER_client_keep (client);
1081       session = create_session (plugin,
1082                                 &wm->clientIdentity, client);
1083       session->inbound = GNUNET_YES;
1084       if (GNUNET_OK ==
1085           GNUNET_SERVER_client_get_address (client, &vaddr, &alen))
1086         {
1087 #if DEBUG_TCP
1088           GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
1089                            "tcp",
1090                            "Found address `%s' for incoming connection %p\n",
1091                            GNUNET_a2s (vaddr, alen),
1092                            client);
1093 #endif
1094           session->connect_addr = vaddr;
1095           session->connect_alen = alen;
1096         }
1097       else
1098         {
1099 #if DEBUG_TCP
1100           GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
1101                            "tcp",
1102                            "Did not obtain TCP socket address for incoming connection\n");
1103 #endif
1104         }
1105 #if DEBUG_TCP
1106       GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
1107                        "tcp",
1108                        "Creating new session %p for connection %p\n",
1109                        session, client);
1110 #endif
1111       process_pending_messages (session);
1112     }
1113   if (session->expecting_welcome != GNUNET_YES)
1114     {
1115       GNUNET_break_op (0);
1116       GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1117       return;
1118     }
1119   session->last_activity = GNUNET_TIME_absolute_get ();
1120   session->expecting_welcome = GNUNET_NO;
1121   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1122 }
1123
1124
1125 /**
1126  * Task to signal the server that we can continue
1127  * receiving from the TCP client now.
1128  *
1129  * @param cls the 'struct Session*'
1130  * @param tc task context (unused)
1131  */
1132 static void
1133 delayed_done (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1134 {
1135   struct Session *session = cls;
1136   struct GNUNET_TIME_Relative delay;
1137
1138   session->receive_delay_task = GNUNET_SCHEDULER_NO_TASK;
1139   delay = session->plugin->env->receive (session->plugin->env->cls,
1140                                          &session->target,
1141                                          NULL, 0, 
1142                                          session,
1143                                          NULL, 0);
1144   if (delay.value == 0)
1145     GNUNET_SERVER_receive_done (session->client, GNUNET_OK);
1146   else
1147     session->receive_delay_task = 
1148       GNUNET_SCHEDULER_add_delayed (session->plugin->env->sched,
1149                                     delay, &delayed_done, session);
1150 }
1151
1152
1153 /**
1154  * We've received data for this peer via TCP.  Unbox,
1155  * compute latency and forward.
1156  *
1157  * @param cls closure
1158  * @param client identification of the client
1159  * @param message the actual message
1160  */
1161 static void
1162 handle_tcp_data (void *cls,
1163                  struct GNUNET_SERVER_Client *client,
1164                  const struct GNUNET_MessageHeader *message)
1165 {
1166   struct Plugin *plugin = cls;
1167   struct Session *session;
1168   struct GNUNET_TIME_Relative delay;
1169
1170   if (GNUNET_MESSAGE_TYPE_TRANSPORT_TCP_WELCOME == ntohs(message->type))
1171     {
1172       /* We don't want to propagate WELCOME messages up! */
1173       GNUNET_SERVER_receive_done (client, GNUNET_OK);
1174       return; 
1175     }    
1176   session = find_session_by_client (plugin, client);
1177   if ( (NULL == session) || (GNUNET_NO != session->expecting_welcome))
1178     {
1179       GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1180       return;
1181     }
1182   session->last_activity = GNUNET_TIME_absolute_get ();
1183 #if DEBUG_TCP
1184   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
1185                    "tcp", 
1186                    "Passing %u bytes of type %u from `%4s' to transport service.\n",
1187                    (unsigned int) ntohs (message->size), 
1188                    (unsigned int) ntohs (message->type),
1189                    GNUNET_i2s (&session->target));
1190 #endif
1191   GNUNET_STATISTICS_update (plugin->env->stats,
1192                             gettext_noop ("# bytes received via TCP"),
1193                             ntohs (message->size),
1194                             GNUNET_NO); 
1195   delay = plugin->env->receive (plugin->env->cls, &session->target, message, 1,
1196                                 session, 
1197                                 (GNUNET_YES == session->inbound) ? NULL : session->connect_addr,
1198                                 (GNUNET_YES == session->inbound) ? 0 : session->connect_alen);
1199   if (delay.value == 0)
1200     GNUNET_SERVER_receive_done (client, GNUNET_OK);
1201   else
1202     session->receive_delay_task = 
1203       GNUNET_SCHEDULER_add_delayed (session->plugin->env->sched,
1204                                     delay, &delayed_done, session);
1205 }
1206
1207
1208 /**
1209  * Handlers for the various TCP messages.
1210  */
1211 static struct GNUNET_SERVER_MessageHandler my_handlers[] = {
1212   {&handle_tcp_welcome, NULL, GNUNET_MESSAGE_TYPE_TRANSPORT_TCP_WELCOME,
1213    sizeof (struct WelcomeMessage)},
1214   {&handle_tcp_data, NULL, GNUNET_MESSAGE_TYPE_ALL, 0},
1215   {NULL, NULL, 0, 0}
1216 };
1217
1218
1219 /**
1220  * Functions with this signature are called whenever a peer
1221  * is disconnected on the network level.
1222  *
1223  * @param cls closure
1224  * @param client identification of the client
1225  */
1226 static void
1227 disconnect_notify (void *cls, struct GNUNET_SERVER_Client *client)
1228 {
1229   struct Plugin *plugin = cls;
1230   struct Session *session;
1231
1232   if (client == NULL)
1233     return;
1234   session = find_session_by_client (plugin, client);
1235   if (session == NULL)
1236     return;                     /* unknown, nothing to do */
1237 #if DEBUG_TCP
1238   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
1239                    "tcp",
1240                    "Destroying session of `%4s' with %s (%p) due to network-level disconnect.\n",
1241                    GNUNET_i2s (&session->target),
1242                    (session->connect_addr != NULL) ?
1243                    GNUNET_a2s (session->connect_addr,
1244                                session->connect_alen) : "*", client);
1245 #endif
1246   disconnect_session (session);
1247 }
1248
1249
1250 /**
1251  * Add the IP of our network interface to the list of
1252  * our external IP addresses.
1253  */
1254 static int
1255 process_interfaces (void *cls,
1256                     const char *name,
1257                     int isDefault,
1258                     const struct sockaddr *addr, socklen_t addrlen)
1259 {
1260   struct Plugin *plugin = cls;
1261   int af;
1262   struct sockaddr_in *v4;
1263   struct sockaddr_in6 *v6;
1264
1265   af = addr->sa_family;
1266   if (af == AF_INET)
1267     {
1268       v4 = (struct sockaddr_in *) addr;
1269       v4->sin_port = htons (plugin->adv_port);
1270     }
1271   else
1272     {
1273       GNUNET_assert (af == AF_INET6);
1274       v6 = (struct sockaddr_in6 *) addr;
1275       v6->sin6_port = htons (plugin->adv_port);
1276     }
1277   GNUNET_log_from (GNUNET_ERROR_TYPE_INFO |
1278                    GNUNET_ERROR_TYPE_BULK,
1279                    "tcp", _("Found address `%s' (%s)\n"),
1280                    GNUNET_a2s (addr, addrlen), name);
1281   plugin->env->notify_address (plugin->env->cls,
1282                                "tcp",
1283                                addr, addrlen, GNUNET_TIME_UNIT_FOREVER_REL);
1284   return GNUNET_OK;
1285 }
1286
1287
1288 /**
1289  * Function called by the resolver for each address obtained from DNS
1290  * for our own hostname.  Add the addresses to the list of our
1291  * external IP addresses.
1292  *
1293  * @param cls closure
1294  * @param addr one of the addresses of the host, NULL for the last address
1295  * @param addrlen length of the address
1296  */
1297 static void
1298 process_hostname_ips (void *cls,
1299                       const struct sockaddr *addr, socklen_t addrlen)
1300 {
1301   struct Plugin *plugin = cls;
1302
1303   if (addr == NULL)
1304     {
1305       plugin->hostname_dns = NULL;
1306       return;
1307     }
1308   process_interfaces (plugin, "<hostname>", GNUNET_YES, addr, addrlen);
1309 }
1310
1311
1312 /**
1313  * Entry point for the plugin.
1314  */
1315 void *
1316 libgnunet_plugin_transport_tcp_init (void *cls)
1317 {
1318   struct GNUNET_TRANSPORT_PluginEnvironment *env = cls;
1319   struct GNUNET_TRANSPORT_PluginFunctions *api;
1320   struct Plugin *plugin;
1321   struct GNUNET_SERVICE_Context *service;
1322   unsigned long long aport;
1323   unsigned long long bport;
1324   unsigned int i;
1325
1326   service = GNUNET_SERVICE_start ("transport-tcp", env->sched, env->cfg);
1327   if (service == NULL)
1328     {
1329       GNUNET_log_from (GNUNET_ERROR_TYPE_WARNING,
1330                        "tcp",
1331                        _
1332                        ("Failed to start service for `%s' transport plugin.\n"),
1333                        "tcp");
1334       return NULL;
1335     }
1336   aport = 0;
1337   if ((GNUNET_OK !=
1338        GNUNET_CONFIGURATION_get_value_number (env->cfg,
1339                                               "transport-tcp",
1340                                               "PORT",
1341                                               &bport)) ||
1342       (bport > 65535) ||
1343       ((GNUNET_OK ==
1344         GNUNET_CONFIGURATION_get_value_number (env->cfg,
1345                                                "transport-tcp",
1346                                                "ADVERTISED-PORT",
1347                                                &aport)) && (aport > 65535)))
1348     {
1349       GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR,
1350                        "tcp",
1351                        _
1352                        ("Require valid port number for service `%s' in configuration!\n"),
1353                        "transport-tcp");
1354       GNUNET_SERVICE_stop (service);
1355       return NULL;
1356     }
1357   if (aport == 0)
1358     aport = bport;
1359   plugin = GNUNET_malloc (sizeof (struct Plugin));
1360   plugin->open_port = bport;
1361   plugin->adv_port = aport;
1362   plugin->env = env;
1363   plugin->lsock = NULL;
1364   plugin->statistics = NULL;
1365   api = GNUNET_malloc (sizeof (struct GNUNET_TRANSPORT_PluginFunctions));
1366   api->cls = plugin;
1367   api->send = &tcp_plugin_send;
1368   api->disconnect = &tcp_plugin_disconnect;
1369   api->address_pretty_printer = &tcp_plugin_address_pretty_printer;
1370   api->check_address = &tcp_plugin_check_address;
1371   plugin->service = service;
1372   plugin->server = GNUNET_SERVICE_get_server (service);
1373   plugin->handlers = GNUNET_malloc (sizeof (my_handlers));
1374   memcpy (plugin->handlers, my_handlers, sizeof (my_handlers));
1375   for (i = 0;
1376        i <
1377        sizeof (my_handlers) / sizeof (struct GNUNET_SERVER_MessageHandler);
1378        i++)
1379     plugin->handlers[i].callback_cls = plugin;
1380   GNUNET_SERVER_add_handlers (plugin->server, plugin->handlers);
1381
1382   GNUNET_log_from (GNUNET_ERROR_TYPE_INFO,
1383                    "tcp", _("TCP transport listening on port %llu\n"), bport);
1384   if (aport != bport)
1385     GNUNET_log_from (GNUNET_ERROR_TYPE_INFO,
1386                      "tcp",
1387                      _("TCP transport advertises itself as being on port %llu\n"),
1388                      aport);
1389   GNUNET_SERVER_disconnect_notify (plugin->server, 
1390                                    &disconnect_notify,
1391                                    plugin);
1392   /* FIXME: do the two calls below periodically again and
1393      not just once (since the info we get might change...) */
1394   GNUNET_OS_network_interfaces_list (&process_interfaces, plugin);
1395   plugin->hostname_dns = GNUNET_RESOLVER_hostname_resolve (env->sched,
1396                                                            env->cfg,
1397                                                            AF_UNSPEC,
1398                                                            HOSTNAME_RESOLVE_TIMEOUT,
1399                                                            &process_hostname_ips,
1400                                                            plugin);
1401   return api;
1402 }
1403
1404
1405 /**
1406  * Exit point from the plugin.
1407  */
1408 void *
1409 libgnunet_plugin_transport_tcp_done (void *cls)
1410 {
1411   struct GNUNET_TRANSPORT_PluginFunctions *api = cls;
1412   struct Plugin *plugin = api->cls;
1413   struct Session *session;
1414
1415   while (NULL != (session = plugin->sessions))
1416     disconnect_session (session);
1417   if (NULL != plugin->hostname_dns)
1418     {
1419       GNUNET_RESOLVER_request_cancel (plugin->hostname_dns);
1420       plugin->hostname_dns = NULL;
1421     }
1422   GNUNET_SERVICE_stop (plugin->service);
1423   GNUNET_free (plugin->handlers);
1424   GNUNET_free (plugin);
1425   GNUNET_free (api);
1426   return NULL;
1427 }
1428
1429 /* end of plugin_transport_tcp.c */