fixes
[oweals/gnunet.git] / src / transport / transport_api.c
1 /*
2      This file is part of GNUnet.
3      (C) 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 /**
22  * @file transport/transport_api.c
23  * @brief library to access the low-level P2P IO service
24  * @author Christian Grothoff
25  */
26 #include "platform.h"
27 #include "gnunet_bandwidth_lib.h"
28 #include "gnunet_client_lib.h"
29 #include "gnunet_constants.h"
30 #include "gnunet_container_lib.h"
31 #include "gnunet_arm_service.h"
32 #include "gnunet_hello_lib.h"
33 #include "gnunet_protocols.h"
34 #include "gnunet_server_lib.h"
35 #include "gnunet_time_lib.h"
36 #include "gnunet_transport_service.h"
37 #include "transport.h"
38
39 /**
40  * After how long do we give up on transmitting a HELLO
41  * to the service?
42  */
43 #define OFFER_HELLO_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 30)
44
45 /**
46  * After how long do we automatically retry an unsuccessful
47  * CONNECT request?
48  */
49 #define CONNECT_RETRY_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS, 750)
50
51 /**
52  * How long should ARM wait when starting up the
53  * transport service before reporting back?
54  */
55 #define START_SERVICE_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5)
56
57 /**
58  * How long should ARM wait when stopping the
59  * transport service before reporting back?
60  */
61 #define STOP_SERVICE_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5)
62
63
64 /**
65  * What stage are we in for transmission processing?
66  */
67 enum TransmitStage
68   {
69     /**
70      * No active message.
71      */
72     TS_NEW = 0,
73
74     /**
75      * Message in local queue, not given to service.
76      */
77     TS_QUEUED = 1,
78
79     /**
80      * Message given to service, not confirmed (no SEND_OK).
81      */
82     TS_TRANSMITTED = 2,
83
84     /**
85      * One message was given to service and before it was confirmed,
86      * another one was already queued (waiting for SEND_OK to pass on
87      * to service).
88      */
89     TS_TRANSMITTED_QUEUED = 3
90   };
91
92
93 /**
94  * Handle for a transmission-ready request.
95  */
96 struct GNUNET_TRANSPORT_TransmitHandle
97 {
98
99   /**
100    * Neighbour for this handle, NULL for control-traffic.
101    */
102   struct NeighbourList *neighbour;
103
104   /**
105    * Function to call when notify_size bytes are available
106    * for transmission.
107    */
108   GNUNET_CONNECTION_TransmitReadyNotify notify;
109
110   /**
111    * Closure for notify.
112    */
113   void *notify_cls;
114
115   /**
116    * transmit_ready task Id.  The task is used to introduce the
117    * artificial delay that may be required to maintain the bandwidth
118    * limits.  Later, this will be the ID of the "transmit_timeout"
119    * task which is used to signal a timeout if the transmission could
120    * not be done in a timely fashion.
121    */
122   GNUNET_SCHEDULER_TaskIdentifier notify_delay_task;
123
124   /**
125    * Timeout for this request.
126    */
127   struct GNUNET_TIME_Absolute timeout;
128
129   /**
130    * How many bytes is our notify callback waiting for?
131    */
132   size_t notify_size;
133
134   /**
135    * How important is this message?
136    */
137   unsigned int priority;
138
139 };
140
141
142 /**
143  * Handle for a control message queue entry.
144  */
145 struct ControlMessage
146 {
147
148   /**
149    * This is a doubly-linked list.
150    */
151   struct ControlMessage *next;
152
153   /**
154    * This is a doubly-linked list.
155    */
156   struct ControlMessage *prev;
157
158   /**
159    * Overall transport handle.
160    */
161   struct GNUNET_TRANSPORT_Handle *h;
162
163   /**
164    * Function to call when notify_size bytes are available
165    * for transmission.
166    */
167   GNUNET_CONNECTION_TransmitReadyNotify notify;
168
169   /**
170    * Closure for notify.
171    */
172   void *notify_cls;
173
174   /**
175    * transmit_ready task Id.  The task is used to introduce the
176    * artificial delay that may be required to maintain the bandwidth
177    * limits.  Later, this will be the ID of the "transmit_timeout"
178    * task which is used to signal a timeout if the transmission could
179    * not be done in a timely fashion.
180    */
181   GNUNET_SCHEDULER_TaskIdentifier notify_delay_task;
182
183   /**
184    * How many bytes is our notify callback waiting for?
185    */
186   size_t notify_size;
187
188 };
189
190
191 /**
192  * Entry in linked list of all of our current neighbours.
193  */
194 struct NeighbourList
195 {
196
197   /**
198    * This is a linked list.
199    */
200   struct NeighbourList *next;
201
202   /**
203    * Overall transport handle.
204    */
205   struct GNUNET_TRANSPORT_Handle *h;
206
207   /**
208    * Active transmit handle; available if 'transmit_forbidden'
209    * is GNUNET_NO.
210    */
211   struct GNUNET_TRANSPORT_TransmitHandle transmit_handle;
212
213   /**
214    * Identity of this neighbour.
215    */
216   struct GNUNET_PeerIdentity id;
217
218   /**
219    * Outbound bandwidh tracker.
220    */
221   struct GNUNET_BANDWIDTH_Tracker out_tracker;
222
223   /**
224    * Set to GNUNET_NO if we are currently allowed to accept a
225    * message to the transport service for this peer, GNUNET_YES
226    * if we have one and are waiting for transmission, GNUNET_SYSERR
227    * if we are waiting for confirmation AND have already accepted
228    * yet another message.
229    */
230   enum TransmitStage transmit_stage;
231
232   /**
233    * Have we received a notification that this peer is connected
234    * to us right now?
235    */
236   int is_connected;
237
238 };
239
240
241 /**
242  * Linked list of requests from clients for our HELLO that were
243  * deferred.
244  */
245 struct HelloWaitList
246 {
247
248   /**
249    * This is a linked list.
250    */
251   struct HelloWaitList *next;
252
253   /**
254    * Reference back to our transport handle.
255    */
256   struct GNUNET_TRANSPORT_Handle *handle;
257
258   /**
259    * Callback to call once we got our HELLO.
260    */
261   GNUNET_TRANSPORT_HelloUpdateCallback rec;
262
263   /**
264    * Closure for rec.
265    */
266   void *rec_cls;
267
268 };
269
270
271 /**
272  * Handle for the transport service (includes all of the
273  * state for the transport service).
274  */
275 struct GNUNET_TRANSPORT_Handle
276 {
277
278   /**
279    * Closure for the callbacks.
280    */
281   void *cls;
282
283   /**
284    * Function to call for received data.
285    */
286   GNUNET_TRANSPORT_ReceiveCallback rec;
287
288   /**
289    * function to call on connect events
290    */
291   GNUNET_TRANSPORT_NotifyConnect nc_cb;
292
293   /**
294    * function to call on disconnect events
295    */
296   GNUNET_TRANSPORT_NotifyDisconnect nd_cb;
297
298   /**
299    * Head of DLL of control messages.
300    */
301   struct ControlMessage *control_head;
302
303   /**
304    * Tail of DLL of control messages.
305    */
306   struct ControlMessage *control_tail;
307
308   /**
309    * The current HELLO message for this peer.  Updated
310    * whenever transports change their addresses.
311    */
312   struct GNUNET_HELLO_Message *my_hello;
313
314   /**
315    * My client connection to the transport service.
316    */
317   struct GNUNET_CLIENT_Connection *client;
318
319   /**
320    * Handle to our registration with the client for notification.
321    */
322   struct GNUNET_CLIENT_TransmitHandle *network_handle;
323
324   /**
325    * Linked list of pending requests for our HELLO.
326    */
327   struct HelloWaitList *hwl_head;
328
329   /**
330    * My scheduler.
331    */
332   struct GNUNET_SCHEDULER_Handle *sched;
333
334   /**
335    * My configuration.
336    */
337   const struct GNUNET_CONFIGURATION_Handle *cfg;
338
339   /**
340    * Linked list of the current neighbours of this peer.
341    */
342   struct NeighbourList *neighbours;
343
344   /**
345    * ID of the task trying to reconnect to the service.
346    */
347   GNUNET_SCHEDULER_TaskIdentifier reconnect_task;
348
349   /**
350    * ID of the task trying to trigger transmission for a peer
351    * while maintaining bandwidth quotas.
352    */
353   GNUNET_SCHEDULER_TaskIdentifier quota_task;
354
355   /**
356    * Delay until we try to reconnect.
357    */
358   struct GNUNET_TIME_Relative reconnect_delay;
359   
360   /**
361    * Set once we are in the process of disconnecting from the
362    * service.
363    */
364   int in_disconnect;
365
366 };
367
368
369 // FIXME: replace with hash map!
370 /**
371  * Get the neighbour list entry for the given peer
372  *
373  * @param h our context
374  * @param peer peer to look up
375  * @return NULL if no such peer entry exists
376  */
377 static struct NeighbourList *
378 neighbour_find (struct GNUNET_TRANSPORT_Handle *h,
379                 const struct GNUNET_PeerIdentity *peer)
380 {
381   struct NeighbourList *pos;
382
383   pos = h->neighbours;
384   while ((pos != NULL) &&
385          (0 != memcmp (peer, &pos->id, sizeof (struct GNUNET_PeerIdentity))))
386     pos = pos->next;
387   return pos;
388 }
389
390
391 /**
392  * Schedule the task to send one message, either from the control
393  * list or the peer message queues  to the service.
394  */
395 static void schedule_transmission (struct GNUNET_TRANSPORT_Handle *h);
396
397
398 /**
399  * Function called by the scheduler when the timeout for bandwidth
400  * availablility for the target neighbour is reached.
401  *
402  * @param cls the 'struct GNUNET_TRANSPORT_Handle*'
403  * @param tc scheduler context
404  */
405 static void
406 quota_transmit_ready (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
407 {
408   struct GNUNET_TRANSPORT_Handle *h = cls;
409
410   h->quota_task = GNUNET_SCHEDULER_NO_TASK;
411   schedule_transmission (h);
412 }
413
414
415 /**
416  * Figure out which transmission to a peer can be done right now.
417  * If none can, schedule a task to call 'schedule_transmission'
418  * whenever a peer transmission can be done in the future and
419  * return NULL.  Otherwise return the next transmission to be
420  * performed.
421  *
422  * @param h handle to transport
423  * @return NULL to wait longer before doing any peer transmissions
424  */
425 static struct GNUNET_TRANSPORT_TransmitHandle *
426 schedule_peer_transmission (struct GNUNET_TRANSPORT_Handle *h)
427 {
428   struct GNUNET_TRANSPORT_TransmitHandle *ret;
429   struct GNUNET_TRANSPORT_TransmitHandle *th;
430   struct NeighbourList *n;
431   struct NeighbourList *next;
432   struct GNUNET_TIME_Relative retry_time;
433   struct GNUNET_TIME_Relative duration;
434
435   if (h->quota_task != GNUNET_SCHEDULER_NO_TASK)
436     {
437       GNUNET_SCHEDULER_cancel (h->sched,
438                                h->quota_task);
439       h->quota_task = GNUNET_SCHEDULER_NO_TASK;
440     }
441   retry_time = GNUNET_TIME_UNIT_FOREVER_REL;
442   ret = NULL;
443   next = h->neighbours;
444   while (NULL != (n = next))
445     {
446       next = n->next;
447       if (n->transmit_stage != TS_QUEUED)
448         continue; /* not eligible */
449       th = &n->transmit_handle;
450       GNUNET_break (n == th->neighbour);
451       /* check outgoing quota */
452       duration = GNUNET_BANDWIDTH_tracker_get_delay (&n->out_tracker,
453                                                      th->notify_size - sizeof (struct OutboundMessage));
454       if (th->timeout.value < duration.value)
455         {
456           /* signal timeout! */
457 #if DEBUG_TRANSPORT
458           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
459                       "Would need %llu ms before bandwidth is available for delivery to `%4s', that is too long.  Signaling timeout.\n",
460                       duration.value, 
461                       GNUNET_i2s (&n->id));
462 #endif
463           if (th->notify_delay_task != GNUNET_SCHEDULER_NO_TASK)
464             {
465               GNUNET_SCHEDULER_cancel (h->sched, th->notify_delay_task);
466               th->notify_delay_task = GNUNET_SCHEDULER_NO_TASK;
467             }         
468           n->transmit_stage = TS_NEW;
469           if (NULL != th->notify)
470             GNUNET_assert (0 == th->notify (th->notify_cls, 0, NULL));
471           continue;
472         }
473       if (duration.value > 0)
474         {
475 #if DEBUG_TRANSPORT
476           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
477                       "Need more bandwidth (%u b/s allowed, %u b needed), delaying delivery to `%4s' by %llu ms\n",
478                       (unsigned int) n->out_tracker.available_bytes_per_s__,
479                       (unsigned int) th->notify_size - sizeof (struct OutboundMessage),
480                       GNUNET_i2s (&n->id), 
481                       duration.value);
482 #endif
483           retry_time = GNUNET_TIME_relative_min (retry_time,
484                                                  duration);
485           continue;
486         }
487 #if DEBUG_TRANSPORT
488       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
489                   "Have %u bytes of bandwidth available for transmission to `%4s' right now\n",
490                   th->notify_size - sizeof (struct OutboundMessage),
491                   GNUNET_i2s (&n->id));
492 #endif  
493       if ( (ret == NULL) ||
494            (ret->priority < th->priority) )
495         ret = th;
496     }
497   if (ret == NULL)
498     h->quota_task = GNUNET_SCHEDULER_add_delayed (h->sched,
499                                                   retry_time,
500                                                   &quota_transmit_ready,
501                                                   h);
502   return ret;
503 }
504
505
506 /**
507  * Transmit message(s) to service.
508  *
509  * @param cls handle to transport 
510  * @param size number of bytes available in buf
511  * @param buf where to copy the message
512  * @return number of bytes copied to buf
513  */
514 static size_t
515 transport_notify_ready (void *cls, size_t size, void *buf)
516 {
517   struct GNUNET_TRANSPORT_Handle *h = cls;
518   struct ControlMessage *cm;
519   struct GNUNET_TRANSPORT_TransmitHandle *th;
520   struct NeighbourList *n;
521   struct OutboundMessage obm;
522   size_t ret;
523   size_t mret;
524   size_t nret;
525   char *cbuf;
526
527   h->network_handle = NULL;
528   if (buf == NULL)
529     {
530       schedule_transmission (h);
531       return 0;
532     }
533 #if DEBUG_TRANSPORT
534   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
535               "Ready to transmit %u bytes to transport service\n", size);
536 #endif
537   cbuf = buf;
538   ret = 0;
539   while ( (NULL != (cm = h->control_head)) &&
540           (cm->notify_size <= size) )
541     {
542       if (cm->notify_delay_task != GNUNET_SCHEDULER_NO_TASK)
543         {
544           GNUNET_SCHEDULER_cancel (h->sched, cm->notify_delay_task);
545           cm->notify_delay_task = GNUNET_SCHEDULER_NO_TASK;
546         }
547       GNUNET_CONTAINER_DLL_remove (h->control_head,
548                                    h->control_tail,
549                                    cm);      
550       nret = cm->notify (cm->notify_cls, size, &cbuf[ret]);
551 #if DEBUG_TRANSPORT
552       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
553                   "Added %u bytes of control message at %u\n",
554                   nret,
555                   ret);
556 #endif
557       GNUNET_free (cm);
558       ret += nret;
559       size -= nret;
560     }
561   while ( (NULL != (th = schedule_peer_transmission (h))) &&
562           (th->notify_size <= size) )
563     {
564       if (th->notify_delay_task != GNUNET_SCHEDULER_NO_TASK)
565         {
566           GNUNET_SCHEDULER_cancel (h->sched, th->notify_delay_task);
567           th->notify_delay_task = GNUNET_SCHEDULER_NO_TASK;
568         }
569       n = th->neighbour;
570       switch (n->transmit_stage)
571         {
572         case TS_NEW:
573           GNUNET_break (0);
574           break;
575         case TS_QUEUED:
576           n->transmit_stage = TS_TRANSMITTED;
577           break;
578         case TS_TRANSMITTED:
579           GNUNET_break (0);
580           break;
581         case TS_TRANSMITTED_QUEUED:
582           GNUNET_break (0);
583           break;
584         default:
585           GNUNET_break (0);
586         }
587       GNUNET_assert (size >= sizeof (struct OutboundMessage));
588       mret = th->notify (th->notify_cls, 
589                          size - sizeof (struct OutboundMessage),
590                          &cbuf[ret + sizeof (struct OutboundMessage)]);
591       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
592                   "Added %u bytes of data message at %u\n",
593                   mret,
594                   ret);
595       GNUNET_assert (mret <= size - sizeof (struct OutboundMessage));
596 #if DEBUG_TRANSPORT
597       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
598                   "Message of %u bytes with timeout %llums constructed for `%4s'\n",
599                   (unsigned int) mret, 
600                   (unsigned long long) GNUNET_TIME_absolute_get_remaining (th->timeout).value,
601                   GNUNET_i2s (&n->id));
602 #endif
603       if (mret != 0)    
604         {
605           obm.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SEND);
606           obm.header.size = htons (mret + sizeof (struct OutboundMessage));
607           obm.priority = htonl (th->priority);
608           obm.timeout = GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining (th->timeout));
609           obm.peer = n->id;
610           memcpy (&cbuf[ret], &obm, sizeof (struct OutboundMessage));
611           ret += (mret + sizeof (struct OutboundMessage));
612           size -= (mret + sizeof (struct OutboundMessage));
613           GNUNET_BANDWIDTH_tracker_consume (&n->out_tracker, mret);
614         }
615       else
616         {
617           switch (n->transmit_stage)
618             {
619             case TS_NEW:
620               GNUNET_break (0);
621               break;
622             case TS_QUEUED:
623               GNUNET_break (0);
624               break;
625             case TS_TRANSMITTED:
626               n->transmit_stage = TS_NEW;
627               break;
628             case TS_TRANSMITTED_QUEUED:
629               n->transmit_stage = TS_QUEUED;
630               continue;
631             default:
632               GNUNET_break (0);
633             }
634         }
635     }
636   schedule_transmission (h);
637 #if DEBUG_TRANSPORT
638   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
639               "Transmitting %u bytes to transport service\n", ret);
640 #endif
641   return ret;
642 }
643
644
645 /**
646  * Schedule the task to send one message, either from the control
647  * list or the peer message queues  to the service.
648  */
649 static void
650 schedule_transmission (struct GNUNET_TRANSPORT_Handle *h)
651 {  
652   size_t size;
653   struct GNUNET_TIME_Relative timeout;
654   struct GNUNET_TRANSPORT_TransmitHandle *th;
655
656   if (NULL != h->network_handle)
657     return;
658   if (h->client == NULL)
659     {
660       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
661                   _("Could not yet schedule transmission: we are not yet connected to the transport service!\n"));
662       return;                   /* not yet connected */
663     }
664   if (NULL != h->control_head) 
665     {
666       size = h->control_head->notify_size;
667       timeout = GNUNET_TIME_UNIT_FOREVER_REL;
668     }
669   else
670     {
671       th = schedule_peer_transmission (h);
672       if (th == NULL)
673         {
674           /* no transmission ready right now */
675 #if DEBUG_TRANSPORT
676           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
677                       "Could not yet schedule transmission: none ready\n");
678 #endif
679           return;
680         }
681       size = th->notify_size;
682       timeout = GNUNET_TIME_absolute_get_remaining (th->timeout);
683     }
684   h->network_handle = 
685     GNUNET_CLIENT_notify_transmit_ready (h->client,
686                                          size,
687                                          timeout,
688                                          GNUNET_NO,
689                                          &transport_notify_ready,
690                                          h);
691   GNUNET_assert (NULL != h->network_handle);
692 }
693
694
695 /**
696  * Called when our transmit request timed out before any transport
697  * reported success connecting to the desired peer or before the
698  * transport was ready to receive.  Signal error and free
699  * TransmitHandle.
700  */
701 static void
702 control_transmit_timeout (void *cls,
703                           const struct GNUNET_SCHEDULER_TaskContext *tc)
704 {
705   struct ControlMessage *th = cls;
706
707   th->notify_delay_task = GNUNET_SCHEDULER_NO_TASK;
708   if (NULL != th->notify)
709     th->notify (th->notify_cls, 0, NULL);
710   GNUNET_CONTAINER_DLL_remove (th->h->control_head,
711                                th->h->control_tail,
712                                th);
713   GNUNET_free (th);
714 }
715
716
717 /**
718  * Queue control request for transmission to the transport
719  * service.
720  *
721  * @param h handle to the transport service
722  * @param size number of bytes to be transmitted
723  * @param at_head request must be added to the head of the queue
724  *        (otherwise request will be appended)
725  * @param timeout how long this transmission can wait (at most)
726  * @param notify function to call to get the content
727  * @param notify_cls closure for notify
728  */
729 static void
730 schedule_control_transmit (struct GNUNET_TRANSPORT_Handle *h,
731                            size_t size,
732                            int at_head,
733                            struct GNUNET_TIME_Relative timeout,
734                            GNUNET_CONNECTION_TransmitReadyNotify notify,
735                            void *notify_cls)
736 {
737   struct ControlMessage *th;
738
739 #if DEBUG_TRANSPORT
740   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
741               "Control transmit of %u bytes within %llums requested\n",
742               size, (unsigned long long) timeout.value);
743 #endif
744   th = GNUNET_malloc (sizeof (struct ControlMessage));
745   th->h = h;
746   th->notify = notify;
747   th->notify_cls = notify_cls;
748   th->notify_size = size;
749   th->notify_delay_task
750     = GNUNET_SCHEDULER_add_delayed (h->sched,
751                                     timeout, &control_transmit_timeout, th);
752   if (at_head)    
753     GNUNET_CONTAINER_DLL_insert (h->control_head,
754                                  h->control_tail,
755                                  th);
756   else
757     GNUNET_CONTAINER_DLL_insert_after (h->control_head,
758                                        h->control_tail,
759                                        h->control_tail,
760                                        th);
761   schedule_transmission (h);
762 }
763
764
765 struct SetQuotaContext
766 {
767   struct GNUNET_TRANSPORT_Handle *handle;
768
769   struct GNUNET_PeerIdentity target;
770
771   GNUNET_SCHEDULER_Task cont;
772
773   void *cont_cls;
774
775   struct GNUNET_TIME_Absolute timeout;
776
777   struct GNUNET_BANDWIDTH_Value32NBO quota_in;
778 };
779
780
781 /**
782  * Send SET_QUOTA message to the service.
783  *
784  * @param cls the 'struct SetQuotaContext'
785  * @param size number of bytes available in buf
786  * @param buf where to copy the message
787  * @return number of bytes copied to buf
788  */
789 static size_t
790 send_set_quota (void *cls, size_t size, void *buf)
791 {
792   struct SetQuotaContext *sqc = cls;
793   struct QuotaSetMessage *msg;
794
795   if (buf == NULL)
796     {
797       GNUNET_SCHEDULER_add_continuation (sqc->handle->sched,
798                                          sqc->cont,
799                                          sqc->cont_cls,
800                                          GNUNET_SCHEDULER_REASON_TIMEOUT);
801       GNUNET_free (sqc);
802       return 0;
803     }
804 #if DEBUG_TRANSPORT
805   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
806               "Transmitting `%s' request with respect to `%4s'.\n",
807               "SET_QUOTA", 
808               GNUNET_i2s (&sqc->target));
809 #endif
810   GNUNET_assert (size >= sizeof (struct QuotaSetMessage));
811   msg = buf;
812   msg->header.size = htons (sizeof (struct QuotaSetMessage));
813   msg->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SET_QUOTA);
814   msg->quota = sqc->quota_in;
815   memcpy (&msg->peer, &sqc->target, sizeof (struct GNUNET_PeerIdentity));
816   if (sqc->cont != NULL)
817     GNUNET_SCHEDULER_add_continuation (sqc->handle->sched,
818                                        sqc->cont,
819                                        sqc->cont_cls,
820                                        GNUNET_SCHEDULER_REASON_PREREQ_DONE);
821   GNUNET_free (sqc);
822   return sizeof (struct QuotaSetMessage);
823 }
824
825
826 /**
827  * Set the share of incoming bandwidth for the given
828  * peer to the specified amount.
829  *
830  * @param handle connection to transport service
831  * @param target who's bandwidth quota is being changed
832  * @param quota_in incoming bandwidth quota in bytes per ms
833  * @param quota_out outgoing bandwidth quota in bytes per ms
834  * @param timeout how long to wait until signaling failure if
835  *        we can not communicate the quota change
836  * @param cont continuation to call when done, will be called
837  *        either with reason "TIMEOUT" or with reason "PREREQ_DONE"
838  * @param cont_cls closure for continuation
839  */
840 void
841 GNUNET_TRANSPORT_set_quota (struct GNUNET_TRANSPORT_Handle *handle,
842                             const struct GNUNET_PeerIdentity *target,
843                             struct GNUNET_BANDWIDTH_Value32NBO quota_in,
844                             struct GNUNET_BANDWIDTH_Value32NBO quota_out,
845                             struct GNUNET_TIME_Relative timeout,
846                             GNUNET_SCHEDULER_Task cont, void *cont_cls)
847 {
848   struct NeighbourList *n;
849   struct SetQuotaContext *sqc;
850
851   n = neighbour_find (handle, target);
852   if (n != NULL)
853     {
854 #if DEBUG_TRANSPORT
855       if (ntohl (quota_out.value__) != n->out_tracker.available_bytes_per_s__)
856         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
857                     "Quota changed from %u to %u for peer `%s'\n",
858                     (unsigned int) n->out_tracker.available_bytes_per_s__,
859                     (unsigned int) ntohl (quota_out.value__),
860                     GNUNET_i2s (target));
861       else
862         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
863                     "Quota remains at %u for peer `%s'\n",
864                     (unsigned int) n->out_tracker.available_bytes_per_s__,
865                     GNUNET_i2s (target));
866 #endif
867       GNUNET_BANDWIDTH_tracker_update_quota (&n->out_tracker,
868                                              quota_out);
869     }
870   else
871     {
872 #if DEBUG_TRANSPORT
873       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
874                   "Quota changed to %u for peer `%s', but I have no such neighbour!\n",
875                   (unsigned int) ntohl (quota_out.value__),
876                   GNUNET_i2s (target));
877 #endif
878     }
879   sqc = GNUNET_malloc (sizeof (struct SetQuotaContext));
880   sqc->handle = handle;
881   sqc->target = *target;
882   sqc->cont = cont;
883   sqc->cont_cls = cont_cls;
884   sqc->timeout = GNUNET_TIME_relative_to_absolute (timeout);
885   sqc->quota_in = quota_in;
886   schedule_control_transmit (handle,
887                              sizeof (struct QuotaSetMessage),
888                              GNUNET_NO, timeout, &send_set_quota, sqc);
889 }
890
891
892 /**
893  * Obtain the HELLO message for this peer.
894  *
895  * @param handle connection to transport service
896  * @param rec function to call with the HELLO, sender will be our peer
897  *            identity; message and sender will be NULL on timeout
898  *            (handshake with transport service pending/failed).
899  *             cost estimate will be 0.
900  * @param rec_cls closure for rec
901  */
902 void
903 GNUNET_TRANSPORT_get_hello (struct GNUNET_TRANSPORT_Handle *handle,
904                             GNUNET_TRANSPORT_HelloUpdateCallback rec,
905                             void *rec_cls)
906 {
907   struct HelloWaitList *hwl;
908
909   hwl = GNUNET_malloc (sizeof (struct HelloWaitList));
910   hwl->next = handle->hwl_head;
911   handle->hwl_head = hwl;
912   hwl->handle = handle;
913   hwl->rec = rec;
914   hwl->rec_cls = rec_cls;
915   if (handle->my_hello == NULL)
916     return;    
917   rec (rec_cls, (const struct GNUNET_MessageHeader *) handle->my_hello);
918 }
919
920
921
922 /**
923  * Stop receiving updates about changes to our HELLO message.
924  *
925  * @param handle connection to transport service
926  * @param rec function previously registered to be called with the HELLOs
927  * @param rec_cls closure for rec
928  */
929 void
930 GNUNET_TRANSPORT_get_hello_cancel (struct GNUNET_TRANSPORT_Handle *handle,
931                                    GNUNET_TRANSPORT_HelloUpdateCallback rec,
932                                    void *rec_cls)
933 {
934   struct HelloWaitList *pos;
935   struct HelloWaitList *prev;
936
937   prev = NULL;
938   pos = handle->hwl_head;
939   while (pos != NULL)
940     {
941       if ( (pos->rec == rec) &&
942            (pos->rec_cls == rec_cls) )
943         break;
944       prev = pos;
945       pos = pos->next;
946     }
947   GNUNET_break (pos != NULL);
948   if (pos == NULL)
949     return;
950   if (prev == NULL)
951     handle->hwl_head = pos->next;
952   else
953     prev->next = pos->next;
954   GNUNET_free (pos);
955 }
956
957
958 /**
959  * Send HELLO message to the service.
960  *
961  * @param cls the HELLO message to send
962  * @param size number of bytes available in buf
963  * @param buf where to copy the message
964  * @return number of bytes copied to buf
965  */
966 static size_t
967 send_hello (void *cls, size_t size, void *buf)
968 {
969   struct GNUNET_MessageHeader *hello = cls;
970   uint16_t msize;
971
972   if (buf == NULL)
973     {
974 #if DEBUG_TRANSPORT_TIMEOUT
975       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
976                   "Timeout while trying to transmit `%s' request.\n",
977                   "HELLO");
978 #endif
979       GNUNET_free (hello);
980       return 0;
981     }
982 #if DEBUG_TRANSPORT
983   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
984               "Transmitting `%s' request.\n", "HELLO");
985 #endif
986   msize = ntohs (hello->size);
987   GNUNET_assert (size >= msize);
988   memcpy (buf, hello, msize);
989   GNUNET_free (hello);
990   return msize;
991 }
992
993
994 /**
995  * Offer the transport service the HELLO of another peer.  Note that
996  * the transport service may just ignore this message if the HELLO is
997  * malformed or useless due to our local configuration.
998  *
999  * @param handle connection to transport service
1000  * @param hello the hello message
1001  */
1002 void
1003 GNUNET_TRANSPORT_offer_hello (struct GNUNET_TRANSPORT_Handle *handle,
1004                               const struct GNUNET_MessageHeader *hello)
1005 {
1006   struct GNUNET_MessageHeader *hc;
1007   uint16_t size;
1008   struct GNUNET_PeerIdentity peer;
1009
1010   GNUNET_break (ntohs (hello->type) == GNUNET_MESSAGE_TYPE_HELLO);
1011   size = ntohs (hello->size);
1012   GNUNET_break (size >= sizeof (struct GNUNET_MessageHeader));
1013   if (GNUNET_OK != GNUNET_HELLO_get_id ((const struct GNUNET_HELLO_Message*) hello,
1014                                         &peer))
1015     {
1016       GNUNET_break (0);
1017       return;
1018     }
1019 #if DEBUG_TRANSPORT 
1020   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1021               "Offering `%s' message of `%4s' to transport for validation.\n",
1022               "HELLO",
1023               GNUNET_i2s (&peer));
1024 #endif
1025   hc = GNUNET_malloc (size);
1026   memcpy (hc, hello, size);
1027   schedule_control_transmit (handle,
1028                              size,
1029                              GNUNET_NO, OFFER_HELLO_TIMEOUT, &send_hello, hc);
1030 }
1031
1032
1033 /**
1034  * Transmit START message to service.
1035  *
1036  * @param cls unused
1037  * @param size number of bytes available in buf
1038  * @param buf where to copy the message
1039  * @return number of bytes copied to buf
1040  */
1041 static size_t
1042 send_start (void *cls, size_t size, void *buf)
1043 {
1044   struct GNUNET_MessageHeader *s = buf;
1045
1046   if (buf == NULL)
1047     {
1048       /* Can only be shutdown, just give up */
1049 #if DEBUG_TRANSPORT
1050       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1051                   "Shutdown while trying to transmit `%s' request.\n",
1052                   "START");
1053 #endif
1054       return 0;
1055     }
1056 #if DEBUG_TRANSPORT
1057   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1058               "Transmitting `%s' request.\n", "START");
1059 #endif
1060   GNUNET_assert (size >= sizeof (struct GNUNET_MessageHeader));
1061   s->size = htons (sizeof (struct GNUNET_MessageHeader));
1062   s->type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_START);
1063   return sizeof (struct GNUNET_MessageHeader);
1064 }
1065
1066
1067 /**
1068  * Free neighbour. 
1069  * 
1070  * @param n the entry to free
1071  */
1072 static void
1073 neighbour_free (struct NeighbourList *n)
1074 {
1075   struct GNUNET_TRANSPORT_Handle *h;
1076   struct NeighbourList *prev;
1077   struct NeighbourList *pos;
1078
1079   h = n->h;
1080 #if DEBUG_TRANSPORT
1081   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1082               "Removing neighbour `%s' from list of connected peers.\n",
1083               GNUNET_i2s (&n->id));
1084 #endif
1085   GNUNET_break (n->is_connected == GNUNET_NO);
1086   GNUNET_break (n->transmit_stage == TS_NEW);
1087
1088   prev = NULL;
1089   pos = h->neighbours;
1090   while (pos != n)
1091     {
1092       prev = pos;
1093       pos = pos->next;
1094     }
1095   if (prev == NULL)
1096     h->neighbours = n->next;
1097   else
1098     prev->next = n->next;
1099   GNUNET_free (n);
1100 }
1101
1102
1103 /**
1104  * Mark neighbour as disconnected. 
1105  * 
1106  * @param n the entry to mark as disconnected
1107  */
1108 static void
1109 neighbour_disconnect (struct NeighbourList *n)
1110 {
1111   struct GNUNET_TRANSPORT_Handle *h = n->h;
1112 #if DEBUG_TRANSPORT
1113   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1114               "Removing neighbour `%s' from list of connected peers.\n",
1115               GNUNET_i2s (&n->id));
1116 #endif
1117   GNUNET_break (n->is_connected == GNUNET_YES);
1118   n->is_connected = GNUNET_NO;
1119   if (h->nc_cb != NULL)
1120     h->nd_cb (h->cls, &n->id);
1121   if (n->transmit_stage == TS_NEW)
1122     neighbour_free (n);
1123 }
1124
1125
1126 /**
1127  * Function we use for handling incoming messages.
1128  *
1129  * @param cls closure (struct GNUNET_TRANSPORT_Handle *)
1130  * @param msg message received, NULL on timeout or fatal error
1131  */
1132 static void demultiplexer (void *cls,
1133                            const struct GNUNET_MessageHeader *msg);
1134
1135
1136 /**
1137  * Try again to connect to transport service.
1138  *
1139  * @param cls the handle to the transport service
1140  * @param tc scheduler context
1141  */
1142 static void
1143 reconnect (void *cls, 
1144            const struct GNUNET_SCHEDULER_TaskContext *tc)
1145 {
1146   struct GNUNET_TRANSPORT_Handle *h = cls;
1147   struct ControlMessage *pos;
1148   struct NeighbourList *n;
1149   struct NeighbourList *next;
1150
1151   h->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
1152   if ( (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) != 0)
1153     {
1154       /* shutdown, just give up */
1155       return;
1156     }
1157   /* Forget about all neighbours that we used to be connected to */
1158   n = h->neighbours;
1159   while (NULL != n)
1160     {
1161 #if DEBUG_TRANSPORT_DISCONNECT
1162   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1163               "Disconnecting due to reconnect being called\n");
1164 #endif
1165       next = n->next;
1166       if (n->is_connected)
1167         neighbour_disconnect (n);
1168       n = next;
1169     }
1170 #if DEBUG_TRANSPORT
1171   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, 
1172               "Connecting to transport service.\n");
1173 #endif
1174   GNUNET_assert (h->client == NULL);
1175   h->client = GNUNET_CLIENT_connect (h->sched, "transport", h->cfg);
1176   GNUNET_assert (h->client != NULL);
1177   /* make sure we don't send "START" twice, remove existing entry from
1178      queue (if present) */
1179   pos = h->control_head;
1180   while (pos != NULL)
1181     {
1182       if (pos->notify == &send_start)
1183         {
1184           GNUNET_CONTAINER_DLL_remove (h->control_head,
1185                                        h->control_tail,
1186                                        pos);
1187           if (GNUNET_SCHEDULER_NO_TASK != pos->notify_delay_task)
1188             {
1189               GNUNET_SCHEDULER_cancel (h->sched, pos->notify_delay_task);
1190               pos->notify_delay_task = GNUNET_SCHEDULER_NO_TASK;
1191             }
1192           GNUNET_free (pos);
1193           break;
1194         }
1195       pos = pos->next;
1196     }
1197   schedule_control_transmit (h,
1198                              sizeof (struct GNUNET_MessageHeader),
1199                              GNUNET_YES,
1200                              GNUNET_TIME_UNIT_FOREVER_REL, &send_start, NULL);
1201   GNUNET_CLIENT_receive (h->client,
1202                          &demultiplexer, h, GNUNET_TIME_UNIT_FOREVER_REL);
1203 }
1204
1205
1206 /**
1207  * Function that will schedule the job that will try
1208  * to connect us again to the client.
1209  *
1210  * @param h transport service to reconnect
1211  */
1212 static void
1213 schedule_reconnect (struct GNUNET_TRANSPORT_Handle *h)
1214 {
1215 #if DEBUG_TRANSPORT
1216   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1217               "Scheduling task to reconnect to transport service in %llu ms.\n",
1218               h->reconnect_delay.value);
1219 #endif
1220   GNUNET_assert (h->client == NULL);
1221   GNUNET_assert (h->reconnect_task == GNUNET_SCHEDULER_NO_TASK);
1222   h->reconnect_task
1223     = GNUNET_SCHEDULER_add_delayed (h->sched,
1224                                     h->reconnect_delay, &reconnect, h);
1225   if (h->reconnect_delay.value == 0)
1226     {
1227       h->reconnect_delay = GNUNET_TIME_UNIT_MILLISECONDS;
1228     }
1229   else 
1230     {
1231       h->reconnect_delay = GNUNET_TIME_relative_multiply (h->reconnect_delay, 2);
1232       h->reconnect_delay = GNUNET_TIME_relative_min (GNUNET_TIME_UNIT_SECONDS,
1233                                                      h->reconnect_delay);
1234     }
1235 }
1236
1237
1238 /**
1239  * Add neighbour to our list
1240  *
1241  * @return NULL if this API is currently disconnecting from the service
1242  */
1243 static struct NeighbourList *
1244 neighbour_add (struct GNUNET_TRANSPORT_Handle *h,
1245                const struct GNUNET_PeerIdentity *pid)
1246 {
1247   struct NeighbourList *n;
1248
1249   if (GNUNET_YES == h->in_disconnect)
1250     return NULL;
1251   /* check for duplicates */
1252   if (NULL != (n = neighbour_find (h, pid)))
1253     {
1254       GNUNET_break (0);
1255       return n;
1256     }
1257 #if DEBUG_TRANSPORT
1258   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1259               "Creating entry for neighbour `%4s'.\n", 
1260               GNUNET_i2s (pid));
1261 #endif
1262   n = GNUNET_malloc (sizeof (struct NeighbourList));
1263   n->id = *pid;
1264   GNUNET_BANDWIDTH_tracker_init (&n->out_tracker,
1265                                  GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT,
1266                                  MAX_BANDWIDTH_CARRY_S);
1267   n->next = h->neighbours;
1268   n->h = h;
1269   h->neighbours = n;  
1270   return n;
1271 }
1272
1273
1274 /**
1275  * Connect to the transport service.  Note that the connection may
1276  * complete (or fail) asynchronously.
1277  *
1278  * @param sched scheduler to use
1279  * @param cfg configuration to use
1280  * @param cls closure for the callbacks
1281  * @param rec receive function to call
1282  * @param nc function to call on connect events
1283  * @param nd function to call on disconnect events
1284  */
1285 struct GNUNET_TRANSPORT_Handle *
1286 GNUNET_TRANSPORT_connect (struct GNUNET_SCHEDULER_Handle *sched,
1287                           const struct GNUNET_CONFIGURATION_Handle *cfg,
1288                           void *cls,
1289                           GNUNET_TRANSPORT_ReceiveCallback rec,
1290                           GNUNET_TRANSPORT_NotifyConnect nc,
1291                           GNUNET_TRANSPORT_NotifyDisconnect nd)
1292 {
1293   struct GNUNET_TRANSPORT_Handle *ret;
1294
1295   ret = GNUNET_malloc (sizeof (struct GNUNET_TRANSPORT_Handle));
1296   ret->sched = sched;
1297   ret->cfg = cfg;
1298   ret->cls = cls;
1299   ret->rec = rec;
1300   ret->nc_cb = nc;
1301   ret->nd_cb = nd;
1302   ret->reconnect_delay = GNUNET_TIME_UNIT_ZERO;
1303   schedule_reconnect (ret);
1304   return ret;
1305 }
1306
1307
1308 /**
1309  * Disconnect from the transport service.
1310  */
1311 void
1312 GNUNET_TRANSPORT_disconnect (struct GNUNET_TRANSPORT_Handle *handle)
1313 {
1314   struct GNUNET_TRANSPORT_TransmitHandle *th;
1315   struct NeighbourList *n;
1316   struct HelloWaitList *hwl;
1317   struct GNUNET_CLIENT_Connection *client;
1318   struct ControlMessage *cm;
1319
1320 #if DEBUG_TRANSPORT
1321   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Transport disconnect called!\n");
1322 #endif
1323   handle->in_disconnect = GNUNET_YES;
1324   while (NULL != (n = handle->neighbours))
1325     {
1326       handle->neighbours = n->next;
1327       switch (n->transmit_stage)
1328         {
1329         case TS_NEW:
1330         case TS_TRANSMITTED:
1331           /* nothing to do */
1332           break;
1333         case TS_QUEUED:
1334         case TS_TRANSMITTED_QUEUED:
1335           th = &n->transmit_handle;
1336           if (th->notify_delay_task != GNUNET_SCHEDULER_NO_TASK)
1337             {
1338               GNUNET_SCHEDULER_cancel (handle->sched,
1339                                        th->notify_delay_task);
1340               th->notify_delay_task = GNUNET_SCHEDULER_NO_TASK;
1341             }
1342           GNUNET_assert (0 == th->notify (th->notify_cls, 0, NULL));        
1343           break;
1344         default:
1345           GNUNET_break (0);
1346         }
1347       GNUNET_free (n);
1348     }
1349   while (NULL != (hwl = handle->hwl_head))
1350     {
1351       handle->hwl_head = hwl->next;
1352       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1353                   _
1354                   ("Disconnect while notification for `%s' still registered.\n"),
1355                   "HELLO");
1356       if (hwl->rec != NULL)
1357         hwl->rec (hwl->rec_cls, NULL);
1358       GNUNET_free (hwl);
1359     }
1360
1361   /* Check for still scheduled control messages, cancel delay tasks if so */
1362   /* Added because somehow a notify_delay_task is remaining scheduled and is ever so annoying */
1363   while ( (NULL != (cm = handle->control_head)))
1364     {
1365       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1366                   _("Disconnect before control message sent!\n"));
1367       if (cm->notify_delay_task != GNUNET_SCHEDULER_NO_TASK)
1368         {
1369           GNUNET_SCHEDULER_cancel (handle->sched, cm->notify_delay_task);
1370           cm->notify_delay_task = GNUNET_SCHEDULER_NO_TASK;
1371         }
1372       GNUNET_CONTAINER_DLL_remove (handle->control_head,
1373                                    handle->control_tail,
1374                                    cm);
1375       GNUNET_free (cm);
1376     }
1377   /* end check */
1378
1379   if (handle->reconnect_task != GNUNET_SCHEDULER_NO_TASK)
1380     {
1381       GNUNET_SCHEDULER_cancel (handle->sched, handle->reconnect_task);
1382       handle->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
1383     }
1384   if (handle->quota_task != GNUNET_SCHEDULER_NO_TASK)
1385     {
1386       GNUNET_SCHEDULER_cancel (handle->sched, handle->quota_task);
1387       handle->quota_task = GNUNET_SCHEDULER_NO_TASK;
1388     }
1389   GNUNET_free_non_null (handle->my_hello);
1390   handle->my_hello = NULL;
1391
1392   if (NULL != handle->network_handle)
1393     {
1394       GNUNET_CLIENT_notify_transmit_ready_cancel (handle->network_handle);
1395       handle->network_handle = NULL;
1396     }
1397   if (NULL != (client = handle->client))
1398     {
1399 #if DEBUG_TRANSPORT
1400       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1401                   "Disconnecting from transport service for good.\n");
1402 #endif
1403       handle->client = NULL;
1404       GNUNET_CLIENT_disconnect (client, GNUNET_NO);
1405     }
1406   GNUNET_free (handle);
1407 }
1408
1409
1410 /**
1411  * Function we use for handling incoming messages.
1412  *
1413  * @param cls closure (struct GNUNET_TRANSPORT_Handle *)
1414  * @param msg message received, NULL on timeout or fatal error
1415  */
1416 static void
1417 demultiplexer (void *cls, const struct GNUNET_MessageHeader *msg)
1418 {
1419   struct GNUNET_TRANSPORT_Handle *h = cls;
1420   const struct DisconnectInfoMessage *dim;
1421   const struct ConnectInfoMessage *cim;
1422   const struct InboundMessage *im;
1423   const struct GNUNET_MessageHeader *imm;
1424   const struct SendOkMessage *okm;
1425   struct HelloWaitList *hwl;
1426   struct HelloWaitList *next_hwl;
1427   struct NeighbourList *n;
1428   struct GNUNET_PeerIdentity me;
1429   uint16_t size;
1430
1431   if (h->client == NULL)
1432     {
1433       /* shutdown initiated from 'GNUNET_TRANSPORT_disconnect',
1434          finish clean up work! */
1435       GNUNET_free (h);
1436       return;
1437     }
1438   if (msg == NULL) 
1439     {
1440 #if DEBUG_TRANSPORT
1441       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1442                   "Error receiving from transport service, disconnecting temporarily.\n");
1443 #endif
1444       if (h->network_handle != NULL)
1445         {
1446           GNUNET_CLIENT_notify_transmit_ready_cancel (h->network_handle);
1447           h->network_handle = NULL;
1448         }
1449       GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
1450       h->client = NULL;
1451       schedule_reconnect (h);
1452       return;
1453     }
1454   GNUNET_CLIENT_receive (h->client,
1455                          &demultiplexer, h, GNUNET_TIME_UNIT_FOREVER_REL);
1456   size = ntohs (msg->size);
1457   switch (ntohs (msg->type))
1458     {
1459     case GNUNET_MESSAGE_TYPE_HELLO:
1460       if (GNUNET_OK !=
1461           GNUNET_HELLO_get_id ((const struct GNUNET_HELLO_Message *) msg,
1462                                &me))
1463         {
1464           GNUNET_break (0);
1465           break;
1466         }
1467 #if DEBUG_TRANSPORT
1468       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1469                   "Receiving (my own) `%s' message, I am `%4s'.\n",
1470                   "HELLO", GNUNET_i2s (&me));
1471 #endif
1472       GNUNET_free_non_null (h->my_hello);
1473       h->my_hello = NULL;
1474       if (size < sizeof (struct GNUNET_MessageHeader))
1475         {
1476           GNUNET_break (0);
1477           break;
1478         }
1479       h->my_hello = GNUNET_malloc (size);
1480       memcpy (h->my_hello, msg, size);
1481       hwl = h->hwl_head;
1482       while (NULL != hwl)
1483         {
1484           next_hwl = hwl->next;
1485           hwl->rec (hwl->rec_cls,
1486                     (const struct GNUNET_MessageHeader *) h->my_hello);
1487           hwl = next_hwl;
1488         }
1489       break;
1490     case GNUNET_MESSAGE_TYPE_TRANSPORT_CONNECT:
1491       if (size != sizeof (struct ConnectInfoMessage))
1492         {
1493           GNUNET_break (0);
1494           break;
1495         }
1496       cim = (const struct ConnectInfoMessage *) msg;
1497 #if DEBUG_TRANSPORT
1498       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1499                   "Receiving `%s' message for `%4s'.\n",
1500                   "CONNECT", GNUNET_i2s (&cim->id));
1501 #endif
1502       n = neighbour_find (h, &cim->id);
1503       if (n == NULL)
1504         n = neighbour_add (h,
1505                            &cim->id);
1506       if (n == NULL)
1507         return;
1508       GNUNET_break (n->is_connected == GNUNET_NO);
1509       n->is_connected = GNUNET_YES;
1510       if (h->nc_cb != NULL)
1511         h->nc_cb (h->cls, &n->id,
1512                   GNUNET_TIME_relative_ntoh (cim->latency), 
1513                   ntohl (cim->distance));
1514       break;
1515     case GNUNET_MESSAGE_TYPE_TRANSPORT_DISCONNECT:
1516       if (size != sizeof (struct DisconnectInfoMessage))
1517         {
1518           GNUNET_break (0);
1519           break;
1520         }
1521       dim = (const struct DisconnectInfoMessage *) msg;
1522 #if DEBUG_TRANSPORT_DISCONNECT
1523       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1524                   "Receiving `%s' message for `%4s'.\n",
1525                   "DISCONNECT",
1526                   GNUNET_i2s (&dim->peer));
1527 #endif
1528       n = neighbour_find (h, &dim->peer);
1529       GNUNET_break (n != NULL);      
1530       if (n != NULL)
1531         neighbour_disconnect (n);       
1532       break;
1533     case GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_OK:
1534       if (size != sizeof (struct SendOkMessage))
1535         {
1536           GNUNET_break (0);
1537           break;
1538         }
1539       okm = (const struct SendOkMessage *) msg;
1540 #if DEBUG_TRANSPORT
1541       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1542                   "Receiving `%s' message, transmission %s.\n", "SEND_OK",
1543                   ntohl (okm->success) == GNUNET_OK ? "succeeded" : "failed");
1544 #endif
1545       n = neighbour_find (h, &okm->peer);
1546       GNUNET_assert (n != NULL);
1547       switch (n->transmit_stage)
1548         {
1549         case TS_NEW:
1550           GNUNET_break (0);
1551           break;
1552         case TS_QUEUED:
1553           GNUNET_break (0);
1554           break;
1555         case TS_TRANSMITTED:
1556           n->transmit_stage = TS_NEW;
1557           break;
1558         case TS_TRANSMITTED_QUEUED:
1559           n->transmit_stage = TS_QUEUED;
1560           schedule_transmission (h);
1561           break;
1562         default:
1563           GNUNET_break (0);
1564         }
1565       break;
1566     case GNUNET_MESSAGE_TYPE_TRANSPORT_RECV:
1567 #if DEBUG_TRANSPORT
1568       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1569                   "Receiving `%s' message.\n", "RECV");
1570 #endif
1571       if (size <
1572           sizeof (struct InboundMessage) +
1573           sizeof (struct GNUNET_MessageHeader))
1574         {
1575           GNUNET_break (0);
1576           break;
1577         }
1578       im = (const struct InboundMessage *) msg;
1579       imm = (const struct GNUNET_MessageHeader *) &im[1];
1580       if (ntohs (imm->size) + sizeof (struct InboundMessage) != size)
1581         {
1582           GNUNET_break (0);
1583           break;
1584         }
1585 #if DEBUG_TRANSPORT
1586       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1587                   "Received message of type %u from `%4s'.\n",
1588                   ntohs (imm->type), GNUNET_i2s (&im->peer));
1589 #endif      
1590       n = neighbour_find (h, &im->peer);
1591       if (n == NULL)
1592         {
1593           GNUNET_break (0);
1594           break;
1595         }
1596       if (n->is_connected != GNUNET_YES)
1597         {
1598           GNUNET_break (0);
1599           break;
1600         }
1601       if (h->rec != NULL)
1602         h->rec (h->cls, &im->peer, imm,
1603                 GNUNET_TIME_relative_ntoh (im->latency), ntohl(im->distance));
1604       break;
1605     default:
1606       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1607                   _
1608                   ("Received unexpected message of type %u in %s:%u\n"),
1609                   ntohs (msg->type), __FILE__, __LINE__);
1610       GNUNET_break (0);
1611       break;
1612     }
1613 }
1614
1615
1616 /**
1617  * Called when our transmit request timed out before any transport
1618  * reported success connecting to the desired peer or before the
1619  * transport was ready to receive.  Signal error and free
1620  * TransmitHandle.
1621  *
1622  * @param cls the 'struct GNUNET_TRANSPORT_TransmitHandle*' that is timing out
1623  * @param tc scheduler context
1624  */
1625 static void
1626 peer_transmit_timeout (void *cls,
1627                        const struct GNUNET_SCHEDULER_TaskContext *tc)
1628 {
1629   struct GNUNET_TRANSPORT_TransmitHandle *th = cls;
1630   struct NeighbourList *n;
1631   GNUNET_CONNECTION_TransmitReadyNotify notify;
1632   void *notify_cls;
1633
1634   th->notify_delay_task = GNUNET_SCHEDULER_NO_TASK;
1635   n = th->neighbour;
1636 #if DEBUG_TRANSPORT
1637   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1638               "Triggering timeout for request to transmit to `%4s' (%d)\n",
1639               GNUNET_i2s (&n->id),
1640               n->transmit_stage);
1641 #endif  
1642   notify = th->notify;
1643   notify_cls = th->notify_cls;
1644   switch (n->transmit_stage)
1645     {
1646     case TS_NEW:
1647       GNUNET_break (0);
1648       break;
1649     case TS_QUEUED:
1650       n->transmit_stage = TS_NEW;
1651       if (n->is_connected == GNUNET_NO)
1652         neighbour_free (n);
1653       break;
1654     case TS_TRANSMITTED:
1655       GNUNET_break (0);
1656       break;
1657     case TS_TRANSMITTED_QUEUED:
1658       n->transmit_stage = TS_TRANSMITTED;
1659       break;
1660     default:
1661       GNUNET_break (0);
1662     }
1663   if (NULL != notify)
1664     notify (notify_cls, 0, NULL);
1665 }
1666
1667
1668 /**
1669  * Check if we could queue a message of the given size for
1670  * transmission.  The transport service will take both its
1671  * internal buffers and bandwidth limits imposed by the
1672  * other peer into consideration when answering this query.
1673  *
1674  * @param handle connection to transport service
1675  * @param target who should receive the message
1676  * @param size how big is the message we want to transmit?
1677  * @param priority how important is the message?
1678  * @param timeout after how long should we give up (and call
1679  *        notify with buf NULL and size 0)?
1680  * @param notify function to call when we are ready to
1681  *        send such a message
1682  * @param notify_cls closure for notify
1683  * @return NULL if someone else is already waiting to be notified
1684  *         non-NULL if the notify callback was queued (can be used to cancel
1685  *         using GNUNET_TRANSPORT_notify_transmit_ready_cancel)
1686  */
1687 struct GNUNET_TRANSPORT_TransmitHandle *
1688 GNUNET_TRANSPORT_notify_transmit_ready (struct GNUNET_TRANSPORT_Handle
1689                                         *handle,
1690                                         const struct GNUNET_PeerIdentity
1691                                         *target, size_t size,
1692                                         unsigned int priority,
1693                                         struct GNUNET_TIME_Relative timeout,
1694                                         GNUNET_CONNECTION_TransmitReadyNotify
1695                                         notify, void *notify_cls)
1696 {
1697   struct GNUNET_TRANSPORT_TransmitHandle *th;
1698   struct NeighbourList *n;
1699
1700   if (size + sizeof (struct OutboundMessage) >=
1701       GNUNET_SERVER_MAX_MESSAGE_SIZE)
1702     {
1703 #if DEBUG_TRANSPORT
1704       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1705                   "Message size is %d, max allowed is %d.\n",
1706                   size + sizeof (struct OutboundMessage), GNUNET_SERVER_MAX_MESSAGE_SIZE);
1707 #endif
1708       GNUNET_break (0);
1709       return NULL;
1710     }
1711 #if DEBUG_TRANSPORT
1712   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1713               "Asking transport service for transmission of %u bytes to peer `%4s' within %llu ms.\n",
1714               size, GNUNET_i2s (target),
1715               (unsigned long long) timeout.value);
1716 #endif
1717   n = neighbour_find (handle, target);
1718   if (n == NULL)
1719     n = neighbour_add (handle, target);
1720   if (n == NULL) 
1721     {
1722       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1723                   "Could not create neighbour entry for peer `%s'\n",
1724                   GNUNET_i2s (target));
1725       return NULL;
1726     }
1727   switch (n->transmit_stage)
1728     {
1729     case TS_NEW:
1730       n->transmit_stage = TS_QUEUED;
1731       break;
1732     case TS_QUEUED:
1733       GNUNET_break (0);
1734       return NULL;
1735     case TS_TRANSMITTED:
1736       n->transmit_stage = TS_TRANSMITTED_QUEUED;
1737       break;
1738     case TS_TRANSMITTED_QUEUED:
1739       GNUNET_break (0);
1740       return NULL;
1741     default:
1742       GNUNET_break (0);
1743       return NULL;
1744     }
1745   th = &n->transmit_handle;
1746   th->neighbour = n;
1747   th->notify = notify;
1748   th->notify_cls = notify_cls;
1749   th->timeout = GNUNET_TIME_relative_to_absolute (timeout);
1750   th->notify_size = size + sizeof (struct OutboundMessage);
1751   th->priority = priority;
1752   th->notify_delay_task
1753     = GNUNET_SCHEDULER_add_delayed (handle->sched, timeout,
1754                                     &peer_transmit_timeout, th);
1755   schedule_transmission (handle);
1756   return th;
1757 }
1758
1759
1760 /**
1761  * Cancel the specified transmission-ready notification.
1762  */
1763 void
1764 GNUNET_TRANSPORT_notify_transmit_ready_cancel (struct
1765                                                GNUNET_TRANSPORT_TransmitHandle
1766                                                *th)
1767 {
1768   struct NeighbourList *n;
1769
1770   n = th->neighbour;
1771 #if DEBUG_TRANSPORT
1772   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1773               "Transmission request of %u bytes to `%4s' was cancelled.\n",
1774               th->notify_size - sizeof (struct OutboundMessage),
1775               GNUNET_i2s (&n->id));
1776 #endif
1777   switch (n->transmit_stage)
1778     {
1779     case TS_NEW:
1780       GNUNET_break (0);
1781       break;
1782     case TS_QUEUED:
1783       n->transmit_stage = TS_NEW;
1784       if (n->is_connected == GNUNET_NO)
1785         neighbour_free (n);
1786       break;
1787     case TS_TRANSMITTED:
1788       GNUNET_break (0);
1789       break;
1790     case TS_TRANSMITTED_QUEUED:
1791       n->transmit_stage = TS_TRANSMITTED;
1792       break;
1793     default:
1794       GNUNET_break (0);
1795     }
1796 }
1797
1798
1799 /* end of transport_api.c */