trying to fix bandwidth calculations
[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
362
363 // FIXME: replace with hash map!
364 /**
365  * Get the neighbour list entry for the given peer
366  *
367  * @param h our context
368  * @param peer peer to look up
369  * @return NULL if no such peer entry exists
370  */
371 static struct NeighbourList *
372 neighbour_find (struct GNUNET_TRANSPORT_Handle *h,
373                 const struct GNUNET_PeerIdentity *peer)
374 {
375   struct NeighbourList *pos;
376
377   pos = h->neighbours;
378   while ((pos != NULL) &&
379          (0 != memcmp (peer, &pos->id, sizeof (struct GNUNET_PeerIdentity))))
380     pos = pos->next;
381   return pos;
382 }
383
384
385 /**
386  * Schedule the task to send one message, either from the control
387  * list or the peer message queues  to the service.
388  */
389 static void schedule_transmission (struct GNUNET_TRANSPORT_Handle *h);
390
391
392 /**
393  * Function called by the scheduler when the timeout for bandwidth
394  * availablility for the target neighbour is reached.
395  *
396  * @param cls the 'struct GNUNET_TRANSPORT_Handle*'
397  * @param tc scheduler context
398  */
399 static void
400 quota_transmit_ready (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
401 {
402   struct GNUNET_TRANSPORT_Handle *h = cls;
403
404   h->quota_task = GNUNET_SCHEDULER_NO_TASK;
405   schedule_transmission (h);
406 }
407
408
409 /**
410  * Figure out which transmission to a peer can be done right now.
411  * If none can, schedule a task to call 'schedule_transmission'
412  * whenever a peer transmission can be done in the future and
413  * return NULL.  Otherwise return the next transmission to be
414  * performed.
415  *
416  * @param h handle to transport
417  * @return NULL to wait longer before doing any peer transmissions
418  */
419 static struct GNUNET_TRANSPORT_TransmitHandle *
420 schedule_peer_transmission (struct GNUNET_TRANSPORT_Handle *h)
421 {
422   struct GNUNET_TRANSPORT_TransmitHandle *ret;
423   struct GNUNET_TRANSPORT_TransmitHandle *th;
424   struct NeighbourList *n;
425   struct NeighbourList *next;
426   struct GNUNET_TIME_Relative retry_time;
427   struct GNUNET_TIME_Relative duration;
428
429   if (h->quota_task != GNUNET_SCHEDULER_NO_TASK)
430     {
431       GNUNET_SCHEDULER_cancel (h->sched,
432                                h->quota_task);
433       h->quota_task = GNUNET_SCHEDULER_NO_TASK;
434     }
435   retry_time = GNUNET_TIME_UNIT_FOREVER_REL;
436   ret = NULL;
437   next = h->neighbours;
438   while (NULL != (n = next))
439     {
440       next = n->next;
441       if (n->transmit_stage != TS_QUEUED)
442         continue; /* not eligible */
443       th = &n->transmit_handle;
444       GNUNET_break (n == th->neighbour);
445       /* check outgoing quota */
446       duration = GNUNET_BANDWIDTH_tracker_get_delay (&n->out_tracker,
447                                                      th->notify_size - sizeof (struct OutboundMessage));
448       if (th->timeout.value < duration.value)
449         {
450           /* signal timeout! */
451 #if DEBUG_TRANSPORT
452           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
453                       "Would need %llu ms before bandwidth is available for delivery to `%4s', that is too long.  Signaling timeout.\n",
454                       duration.value, 
455                       GNUNET_i2s (&n->id));
456 #endif
457           if (th->notify_delay_task != GNUNET_SCHEDULER_NO_TASK)
458             {
459               GNUNET_SCHEDULER_cancel (h->sched, th->notify_delay_task);
460               th->notify_delay_task = GNUNET_SCHEDULER_NO_TASK;
461             }         
462           n->transmit_stage = TS_NEW;
463           if (NULL != th->notify)
464             GNUNET_assert (0 == th->notify (th->notify_cls, 0, NULL));
465           continue;
466         }
467       if (duration.value > 0)
468         {
469 #if DEBUG_TRANSPORT
470           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
471                       "Need more bandwidth (%u b/s allowed, %u b needed), delaying delivery to `%4s' by %llu ms\n",
472                       (unsigned int) n->out_tracker.available_bytes_per_s__,
473                       (unsigned int) th->notify_size - sizeof (struct OutboundMessage),
474                       GNUNET_i2s (&n->id), 
475                       duration.value);
476 #endif
477           retry_time = GNUNET_TIME_relative_min (retry_time,
478                                                  duration);
479           continue;
480         }
481 #if DEBUG_TRANSPORT
482       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
483                   "Have %u bytes of bandwidth available for transmission to `%4s' right now\n",
484                   th->notify_size - sizeof (struct OutboundMessage),
485                   GNUNET_i2s (&n->id));
486 #endif  
487       if ( (ret == NULL) ||
488            (ret->priority < th->priority) )
489         ret = th;
490     }
491   if (ret == NULL)
492     h->quota_task = GNUNET_SCHEDULER_add_delayed (h->sched,
493                                                   retry_time,
494                                                   &quota_transmit_ready,
495                                                   h);
496   return ret;
497 }
498
499
500 /**
501  * Transmit message(s) to service.
502  *
503  * @param cls handle to transport 
504  * @param size number of bytes available in buf
505  * @param buf where to copy the message
506  * @return number of bytes copied to buf
507  */
508 static size_t
509 transport_notify_ready (void *cls, size_t size, void *buf)
510 {
511   struct GNUNET_TRANSPORT_Handle *h = cls;
512   struct ControlMessage *cm;
513   struct GNUNET_TRANSPORT_TransmitHandle *th;
514   struct NeighbourList *n;
515   struct OutboundMessage obm;
516   size_t ret;
517   size_t mret;
518   char *cbuf;
519
520   h->network_handle = NULL;
521   if (buf == NULL)
522     {
523       schedule_transmission (h);
524       return 0;
525     }
526 #if DEBUG_TRANSPORT
527   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
528               "Ready to transmit %u bytes to transport service\n", size);
529 #endif
530   cbuf = buf;
531   ret = 0;
532   while ( (NULL != (cm = h->control_head)) &&
533           (cm->notify_size <= size) )
534     {
535       if (cm->notify_delay_task != GNUNET_SCHEDULER_NO_TASK)
536         {
537           GNUNET_SCHEDULER_cancel (h->sched, cm->notify_delay_task);
538           cm->notify_delay_task = GNUNET_SCHEDULER_NO_TASK;
539         }
540       GNUNET_CONTAINER_DLL_remove (h->control_head,
541                                    h->control_tail,
542                                    cm);
543       ret += cm->notify (cm->notify_cls, size, &cbuf[ret]);
544       GNUNET_free (cm);
545       size -= ret;
546     }
547   while ( (NULL != (th = schedule_peer_transmission (h))) &&
548           (th->notify_size <= size) )
549     {
550       if (th->notify_delay_task != GNUNET_SCHEDULER_NO_TASK)
551         {
552           GNUNET_SCHEDULER_cancel (h->sched, th->notify_delay_task);
553           th->notify_delay_task = GNUNET_SCHEDULER_NO_TASK;
554         }
555       n = th->neighbour;
556       switch (n->transmit_stage)
557         {
558         case TS_NEW:
559           GNUNET_break (0);
560           break;
561         case TS_QUEUED:
562           n->transmit_stage = TS_TRANSMITTED;
563           break;
564         case TS_TRANSMITTED:
565           GNUNET_break (0);
566           break;
567         case TS_TRANSMITTED_QUEUED:
568           GNUNET_break (0);
569           break;
570         default:
571           GNUNET_break (0);
572         }
573       GNUNET_assert (size >= sizeof (struct OutboundMessage));
574       mret = th->notify (th->notify_cls, 
575                          size - sizeof (struct OutboundMessage),
576                          &cbuf[ret + sizeof (struct OutboundMessage)]);
577       GNUNET_assert (mret <= size - sizeof (struct OutboundMessage));
578 #if DEBUG_TRANSPORT
579       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
580                   "Message of %u bytes with timeout %llums constructed for `%4s'\n",
581                   (unsigned int) mret, 
582                   (unsigned long long) GNUNET_TIME_absolute_get_remaining (th->timeout).value,
583                   GNUNET_i2s (&n->id));
584 #endif
585       if (mret != 0)    
586         {
587           obm.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SEND);
588           obm.header.size = htons (mret + sizeof (struct OutboundMessage));
589           obm.priority = htonl (th->priority);
590           obm.timeout = GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining (th->timeout));
591           obm.peer = n->id;
592           memcpy (&cbuf[ret], &obm, sizeof (struct OutboundMessage));
593           ret += (mret + sizeof (struct OutboundMessage));
594           size -= (mret + sizeof (struct OutboundMessage));
595           GNUNET_BANDWIDTH_tracker_consume (&n->out_tracker, mret);
596         }
597       else
598         {
599           switch (n->transmit_stage)
600             {
601             case TS_NEW:
602               GNUNET_break (0);
603               break;
604             case TS_QUEUED:
605               GNUNET_break (0);
606               break;
607             case TS_TRANSMITTED:
608               n->transmit_stage = TS_NEW;
609               break;
610             case TS_TRANSMITTED_QUEUED:
611               GNUNET_break (0);
612               break;
613             default:
614               GNUNET_break (0);
615             }
616         }
617     }
618   schedule_transmission (h);
619 #if DEBUG_TRANSPORT
620   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
621               "Transmitting %u bytes to transport service\n", ret);
622 #endif
623   return ret;
624 }
625
626
627 /**
628  * Schedule the task to send one message, either from the control
629  * list or the peer message queues  to the service.
630  */
631 static void
632 schedule_transmission (struct GNUNET_TRANSPORT_Handle *h)
633 {  
634   size_t size;
635   struct GNUNET_TIME_Relative timeout;
636   struct GNUNET_TRANSPORT_TransmitHandle *th;
637
638   if (NULL != h->network_handle)
639     return;
640   if (h->client == NULL)
641     {
642       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
643                   "Could not yet schedule transmission: we are not yet connected to the transport service!\n");
644       return;                   /* not yet connected */
645     }
646   if (NULL != h->control_head) 
647     {
648       size = h->control_head->notify_size;
649       timeout = GNUNET_TIME_UNIT_FOREVER_REL;
650     }
651   else
652     {
653       th = schedule_peer_transmission (h);
654       if (th == NULL)
655         {
656           /* no transmission ready right now */
657           return;
658         }
659       size = th->notify_size;
660       timeout = GNUNET_TIME_absolute_get_remaining (th->timeout);
661     }
662   h->network_handle = 
663     GNUNET_CLIENT_notify_transmit_ready (h->client,
664                                          size,
665                                          timeout,
666                                          GNUNET_NO,
667                                          &transport_notify_ready,
668                                          h);
669   GNUNET_assert (NULL != h->network_handle);
670 }
671
672
673 /**
674  * Called when our transmit request timed out before any transport
675  * reported success connecting to the desired peer or before the
676  * transport was ready to receive.  Signal error and free
677  * TransmitHandle.
678  */
679 static void
680 control_transmit_timeout (void *cls,
681                           const struct GNUNET_SCHEDULER_TaskContext *tc)
682 {
683   struct ControlMessage *th = cls;
684
685   th->notify_delay_task = GNUNET_SCHEDULER_NO_TASK;
686   if (NULL != th->notify)
687     th->notify (th->notify_cls, 0, NULL);
688   GNUNET_CONTAINER_DLL_remove (th->h->control_head,
689                                th->h->control_tail,
690                                th);
691   GNUNET_free (th);
692 }
693
694
695 /**
696  * Queue control request for transmission to the transport
697  * service.
698  *
699  * @param h handle to the transport service
700  * @param size number of bytes to be transmitted
701  * @param at_head request must be added to the head of the queue
702  *        (otherwise request will be appended)
703  * @param timeout how long this transmission can wait (at most)
704  * @param notify function to call to get the content
705  * @param notify_cls closure for notify
706  */
707 static void
708 schedule_control_transmit (struct GNUNET_TRANSPORT_Handle *h,
709                            size_t size,
710                            int at_head,
711                            struct GNUNET_TIME_Relative timeout,
712                            GNUNET_CONNECTION_TransmitReadyNotify notify,
713                            void *notify_cls)
714 {
715   struct ControlMessage *th;
716
717 #if DEBUG_TRANSPORT
718   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
719               "Control transmit of %u bytes within %llums requested\n",
720               size, (unsigned long long) timeout.value);
721 #endif
722   th = GNUNET_malloc (sizeof (struct ControlMessage));
723   th->h = h;
724   th->notify = notify;
725   th->notify_cls = notify_cls;
726   th->notify_size = size;
727   th->notify_delay_task
728     = GNUNET_SCHEDULER_add_delayed (h->sched,
729                                     timeout, &control_transmit_timeout, th);
730   if (at_head)    
731     GNUNET_CONTAINER_DLL_insert (h->control_head,
732                                  h->control_tail,
733                                  th);
734   else
735     GNUNET_CONTAINER_DLL_insert_after (h->control_head,
736                                        h->control_tail,
737                                        h->control_tail,
738                                        th);
739   schedule_transmission (h);
740 }
741
742
743 struct SetQuotaContext
744 {
745   struct GNUNET_TRANSPORT_Handle *handle;
746
747   struct GNUNET_PeerIdentity target;
748
749   GNUNET_SCHEDULER_Task cont;
750
751   void *cont_cls;
752
753   struct GNUNET_TIME_Absolute timeout;
754
755   struct GNUNET_BANDWIDTH_Value32NBO quota_in;
756 };
757
758
759 /**
760  * Send SET_QUOTA message to the service.
761  *
762  * @param cls the 'struct SetQuotaContext'
763  * @param size number of bytes available in buf
764  * @param buf where to copy the message
765  * @return number of bytes copied to buf
766  */
767 static size_t
768 send_set_quota (void *cls, size_t size, void *buf)
769 {
770   struct SetQuotaContext *sqc = cls;
771   struct QuotaSetMessage *msg;
772
773   if (buf == NULL)
774     {
775       GNUNET_SCHEDULER_add_continuation (sqc->handle->sched,
776                                          sqc->cont,
777                                          sqc->cont_cls,
778                                          GNUNET_SCHEDULER_REASON_TIMEOUT);
779       GNUNET_free (sqc);
780       return 0;
781     }
782 #if DEBUG_TRANSPORT
783   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
784               "Transmitting `%s' request with respect to `%4s'.\n",
785               "SET_QUOTA", 
786               GNUNET_i2s (&sqc->target));
787 #endif
788   GNUNET_assert (size >= sizeof (struct QuotaSetMessage));
789   msg = buf;
790   msg->header.size = htons (sizeof (struct QuotaSetMessage));
791   msg->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SET_QUOTA);
792   msg->quota = sqc->quota_in;
793   memcpy (&msg->peer, &sqc->target, sizeof (struct GNUNET_PeerIdentity));
794   if (sqc->cont != NULL)
795     GNUNET_SCHEDULER_add_continuation (sqc->handle->sched,
796                                        sqc->cont,
797                                        sqc->cont_cls,
798                                        GNUNET_SCHEDULER_REASON_PREREQ_DONE);
799   GNUNET_free (sqc);
800   return sizeof (struct QuotaSetMessage);
801 }
802
803
804 /**
805  * Set the share of incoming bandwidth for the given
806  * peer to the specified amount.
807  *
808  * @param handle connection to transport service
809  * @param target who's bandwidth quota is being changed
810  * @param quota_in incoming bandwidth quota in bytes per ms
811  * @param quota_out outgoing bandwidth quota in bytes per ms
812  * @param timeout how long to wait until signaling failure if
813  *        we can not communicate the quota change
814  * @param cont continuation to call when done, will be called
815  *        either with reason "TIMEOUT" or with reason "PREREQ_DONE"
816  * @param cont_cls closure for continuation
817  */
818 void
819 GNUNET_TRANSPORT_set_quota (struct GNUNET_TRANSPORT_Handle *handle,
820                             const struct GNUNET_PeerIdentity *target,
821                             struct GNUNET_BANDWIDTH_Value32NBO quota_in,
822                             struct GNUNET_BANDWIDTH_Value32NBO quota_out,
823                             struct GNUNET_TIME_Relative timeout,
824                             GNUNET_SCHEDULER_Task cont, void *cont_cls)
825 {
826   struct NeighbourList *n;
827   struct SetQuotaContext *sqc;
828
829   n = neighbour_find (handle, target);
830   if (n != NULL)
831     {
832       if (ntohl (quota_out.value__) != n->out_tracker.available_bytes_per_s__)
833         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
834                     "Quota changed from %u to %u for peer `%s'\n",
835                     (unsigned int) n->out_tracker.available_bytes_per_s__,
836                     (unsigned int) ntohl (quota_out.value__),
837                     GNUNET_i2s (target));
838       GNUNET_BANDWIDTH_tracker_update_quota (&n->out_tracker,
839                                              quota_out);
840     }
841   sqc = GNUNET_malloc (sizeof (struct SetQuotaContext));
842   sqc->handle = handle;
843   sqc->target = *target;
844   sqc->cont = cont;
845   sqc->cont_cls = cont_cls;
846   sqc->timeout = GNUNET_TIME_relative_to_absolute (timeout);
847   sqc->quota_in = quota_in;
848   schedule_control_transmit (handle,
849                              sizeof (struct QuotaSetMessage),
850                              GNUNET_NO, timeout, &send_set_quota, sqc);
851 }
852
853
854 /**
855  * Obtain the HELLO message for this peer.
856  *
857  * @param handle connection to transport service
858  * @param rec function to call with the HELLO, sender will be our peer
859  *            identity; message and sender will be NULL on timeout
860  *            (handshake with transport service pending/failed).
861  *             cost estimate will be 0.
862  * @param rec_cls closure for rec
863  */
864 void
865 GNUNET_TRANSPORT_get_hello (struct GNUNET_TRANSPORT_Handle *handle,
866                             GNUNET_TRANSPORT_HelloUpdateCallback rec,
867                             void *rec_cls)
868 {
869   struct HelloWaitList *hwl;
870
871   hwl = GNUNET_malloc (sizeof (struct HelloWaitList));
872   hwl->next = handle->hwl_head;
873   handle->hwl_head = hwl;
874   hwl->handle = handle;
875   hwl->rec = rec;
876   hwl->rec_cls = rec_cls;
877   if (handle->my_hello == NULL)
878     return;    
879   rec (rec_cls, (const struct GNUNET_MessageHeader *) handle->my_hello);
880 }
881
882
883
884 /**
885  * Stop receiving updates about changes to our HELLO message.
886  *
887  * @param handle connection to transport service
888  * @param rec function previously registered to be called with the HELLOs
889  * @param rec_cls closure for rec
890  */
891 void
892 GNUNET_TRANSPORT_get_hello_cancel (struct GNUNET_TRANSPORT_Handle *handle,
893                                    GNUNET_TRANSPORT_HelloUpdateCallback rec,
894                                    void *rec_cls)
895 {
896   struct HelloWaitList *pos;
897   struct HelloWaitList *prev;
898
899   prev = NULL;
900   pos = handle->hwl_head;
901   while (pos != NULL)
902     {
903       if ( (pos->rec == rec) &&
904            (pos->rec_cls == rec_cls) )
905         break;
906       prev = pos;
907       pos = pos->next;
908     }
909   GNUNET_break (pos != NULL);
910   if (pos == NULL)
911     return;
912   if (prev == NULL)
913     handle->hwl_head = pos->next;
914   else
915     prev->next = pos->next;
916   GNUNET_free (pos);
917 }
918
919
920 /**
921  * Send HELLO message to the service.
922  *
923  * @param cls the HELLO message to send
924  * @param size number of bytes available in buf
925  * @param buf where to copy the message
926  * @return number of bytes copied to buf
927  */
928 static size_t
929 send_hello (void *cls, size_t size, void *buf)
930 {
931   struct GNUNET_MessageHeader *hello = cls;
932   uint16_t msize;
933
934   if (buf == NULL)
935     {
936 #if DEBUG_TRANSPORT_TIMEOUT
937       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
938                   "Timeout while trying to transmit `%s' request.\n",
939                   "HELLO");
940 #endif
941       GNUNET_free (hello);
942       return 0;
943     }
944 #if DEBUG_TRANSPORT
945   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
946               "Transmitting `%s' request.\n", "HELLO");
947 #endif
948   msize = ntohs (hello->size);
949   GNUNET_assert (size >= msize);
950   memcpy (buf, hello, msize);
951   GNUNET_free (hello);
952   return msize;
953 }
954
955
956 /**
957  * Offer the transport service the HELLO of another peer.  Note that
958  * the transport service may just ignore this message if the HELLO is
959  * malformed or useless due to our local configuration.
960  *
961  * @param handle connection to transport service
962  * @param hello the hello message
963  */
964 void
965 GNUNET_TRANSPORT_offer_hello (struct GNUNET_TRANSPORT_Handle *handle,
966                               const struct GNUNET_MessageHeader *hello)
967 {
968   struct GNUNET_MessageHeader *hc;
969   uint16_t size;
970
971   GNUNET_break (ntohs (hello->type) == GNUNET_MESSAGE_TYPE_HELLO);
972   size = ntohs (hello->size);
973   GNUNET_break (size >= sizeof (struct GNUNET_MessageHeader));
974   hc = GNUNET_malloc (size);
975   memcpy (hc, hello, size);
976   schedule_control_transmit (handle,
977                              size,
978                              GNUNET_NO, OFFER_HELLO_TIMEOUT, &send_hello, hc);
979 }
980
981
982 /**
983  * Transmit START message to service.
984  *
985  * @param cls unused
986  * @param size number of bytes available in buf
987  * @param buf where to copy the message
988  * @return number of bytes copied to buf
989  */
990 static size_t
991 send_start (void *cls, size_t size, void *buf)
992 {
993   struct GNUNET_MessageHeader *s = buf;
994
995   if (buf == NULL)
996     {
997       /* Can only be shutdown, just give up */
998 #if DEBUG_TRANSPORT
999       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1000                   "Shutdown while trying to transmit `%s' request.\n",
1001                   "START");
1002 #endif
1003       return 0;
1004     }
1005 #if DEBUG_TRANSPORT
1006   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1007               "Transmitting `%s' request.\n", "START");
1008 #endif
1009   GNUNET_assert (size >= sizeof (struct GNUNET_MessageHeader));
1010   s->size = htons (sizeof (struct GNUNET_MessageHeader));
1011   s->type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_START);
1012   return sizeof (struct GNUNET_MessageHeader);
1013 }
1014
1015
1016 /**
1017  * Free neighbour. 
1018  * 
1019  * @param n the entry to free
1020  */
1021 static void
1022 neighbour_free (struct NeighbourList *n)
1023 {
1024   struct GNUNET_TRANSPORT_Handle *h;
1025   struct NeighbourList *prev;
1026   struct NeighbourList *pos;
1027
1028   h = n->h;
1029 #if DEBUG_TRANSPORT
1030   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1031               "Removing neighbour `%s' from list of connected peers.\n",
1032               GNUNET_i2s (&n->id));
1033 #endif
1034   GNUNET_break (n->is_connected == GNUNET_NO);
1035   GNUNET_break (n->transmit_stage == TS_NEW);
1036
1037   prev = NULL;
1038   pos = h->neighbours;
1039   while (pos != n)
1040     {
1041       prev = pos;
1042       pos = pos->next;
1043     }
1044   if (prev == NULL)
1045     h->neighbours = n->next;
1046   else
1047     prev->next = n->next;
1048   GNUNET_free (n);
1049 }
1050
1051
1052 /**
1053  * Mark neighbour as disconnected. 
1054  * 
1055  * @param n the entry to mark as disconnected
1056  */
1057 static void
1058 neighbour_disconnect (struct NeighbourList *n)
1059 {
1060   struct GNUNET_TRANSPORT_Handle *h = n->h;
1061 #if DEBUG_TRANSPORT
1062   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1063               "Removing neighbour `%s' from list of connected peers.\n",
1064               GNUNET_i2s (&n->id));
1065 #endif
1066   GNUNET_break (n->is_connected == GNUNET_YES);
1067   n->is_connected = GNUNET_NO;
1068   if (h->nc_cb != NULL)
1069     h->nd_cb (h->cls, &n->id);
1070   if (n->transmit_stage == TS_NEW)
1071     neighbour_free (n);
1072 }
1073
1074
1075 /**
1076  * Function we use for handling incoming messages.
1077  *
1078  * @param cls closure (struct GNUNET_TRANSPORT_Handle *)
1079  * @param msg message received, NULL on timeout or fatal error
1080  */
1081 static void demultiplexer (void *cls,
1082                            const struct GNUNET_MessageHeader *msg);
1083
1084
1085 /**
1086  * Try again to connect to transport service.
1087  *
1088  * @param cls the handle to the transport service
1089  * @param tc scheduler context
1090  */
1091 static void
1092 reconnect (void *cls, 
1093            const struct GNUNET_SCHEDULER_TaskContext *tc)
1094 {
1095   struct GNUNET_TRANSPORT_Handle *h = cls;
1096   struct ControlMessage *pos;
1097   struct NeighbourList *n;
1098   struct NeighbourList *next;
1099
1100   h->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
1101   if ( (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) != 0)
1102     {
1103       /* shutdown, just give up */
1104       return;
1105     }
1106   /* Forget about all neighbours that we used to be connected to */
1107   n = h->neighbours;
1108   while (NULL != n)
1109     {
1110 #if DEBUG_TRANSPORT_DISCONNECT
1111   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Disconnecting due to reconnect being called\n");
1112 #endif
1113       next = n->next;
1114       if (n->is_connected)
1115         neighbour_disconnect (n);
1116       n = next;
1117     }
1118 #if DEBUG_TRANSPORT
1119   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Connecting to transport service.\n");
1120 #endif
1121   GNUNET_assert (h->client == NULL);
1122   h->client = GNUNET_CLIENT_connect (h->sched, "transport", h->cfg);
1123   GNUNET_assert (h->client != NULL);
1124   /* make sure we don't send "START" twice, remove existing entry from
1125      queue (if present) */
1126   pos = h->control_head;
1127   while (pos != NULL)
1128     {
1129       if (pos->notify == &send_start)
1130         {
1131           GNUNET_CONTAINER_DLL_remove (h->control_head,
1132                                        h->control_tail,
1133                                        pos);
1134           if (GNUNET_SCHEDULER_NO_TASK != pos->notify_delay_task)
1135             {
1136               GNUNET_SCHEDULER_cancel (h->sched, pos->notify_delay_task);
1137               pos->notify_delay_task = GNUNET_SCHEDULER_NO_TASK;
1138             }
1139           GNUNET_free (pos);
1140           break;
1141         }
1142       pos = pos->next;
1143     }
1144   schedule_control_transmit (h,
1145                              sizeof (struct GNUNET_MessageHeader),
1146                              GNUNET_YES,
1147                              GNUNET_TIME_UNIT_FOREVER_REL, &send_start, NULL);
1148   GNUNET_CLIENT_receive (h->client,
1149                          &demultiplexer, h, GNUNET_TIME_UNIT_FOREVER_REL);
1150 }
1151
1152
1153 /**
1154  * Function that will schedule the job that will try
1155  * to connect us again to the client.
1156  *
1157  * @param h transport service to reconnect
1158  */
1159 static void
1160 schedule_reconnect (struct GNUNET_TRANSPORT_Handle *h)
1161 {
1162 #if DEBUG_TRANSPORT
1163   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1164               "Scheduling task to reconnect to transport service in %llu ms.\n",
1165               h->reconnect_delay.value);
1166 #endif
1167   GNUNET_assert (h->client == NULL);
1168   GNUNET_assert (h->reconnect_task == GNUNET_SCHEDULER_NO_TASK);
1169   h->reconnect_task
1170     = GNUNET_SCHEDULER_add_delayed (h->sched,
1171                                     h->reconnect_delay, &reconnect, h);
1172   if (h->reconnect_delay.value == 0)
1173     {
1174       h->reconnect_delay = GNUNET_TIME_UNIT_MILLISECONDS;
1175     }
1176   else 
1177     {
1178       h->reconnect_delay = GNUNET_TIME_relative_multiply (h->reconnect_delay, 2);
1179       h->reconnect_delay = GNUNET_TIME_relative_min (GNUNET_TIME_UNIT_SECONDS,
1180                                                      h->reconnect_delay);
1181     }
1182 }
1183
1184
1185 /**
1186  * Add neighbour to our list
1187  */
1188 static struct NeighbourList *
1189 neighbour_add (struct GNUNET_TRANSPORT_Handle *h,
1190                const struct GNUNET_PeerIdentity *pid)
1191 {
1192   struct NeighbourList *n;
1193
1194   /* check for duplicates */
1195   if (NULL != (n = neighbour_find (h, pid)))
1196     {
1197       GNUNET_break (0);
1198       return n;
1199     }
1200 #if DEBUG_TRANSPORT
1201   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1202               "Creating entry for neighbour `%4s'.\n", 
1203               GNUNET_i2s (pid));
1204 #endif
1205   n = GNUNET_malloc (sizeof (struct NeighbourList));
1206   n->id = *pid;
1207   GNUNET_BANDWIDTH_tracker_init (&n->out_tracker,
1208                                  GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT,
1209                                  MAX_BANDWIDTH_CARRY_S);
1210   n->next = h->neighbours;
1211   n->h = h;
1212   h->neighbours = n;  
1213   return n;
1214 }
1215
1216
1217 /**
1218  * Connect to the transport service.  Note that the connection may
1219  * complete (or fail) asynchronously.
1220  *
1221  * @param sched scheduler to use
1222  * @param cfg configuration to use
1223  * @param cls closure for the callbacks
1224  * @param rec receive function to call
1225  * @param nc function to call on connect events
1226  * @param nd function to call on disconnect events
1227  */
1228 struct GNUNET_TRANSPORT_Handle *
1229 GNUNET_TRANSPORT_connect (struct GNUNET_SCHEDULER_Handle *sched,
1230                           const struct GNUNET_CONFIGURATION_Handle *cfg,
1231                           void *cls,
1232                           GNUNET_TRANSPORT_ReceiveCallback rec,
1233                           GNUNET_TRANSPORT_NotifyConnect nc,
1234                           GNUNET_TRANSPORT_NotifyDisconnect nd)
1235 {
1236   struct GNUNET_TRANSPORT_Handle *ret;
1237
1238   GNUNET_ARM_start_services (cfg, sched, "peerinfo", "transport", NULL);
1239   ret = GNUNET_malloc (sizeof (struct GNUNET_TRANSPORT_Handle));
1240   ret->sched = sched;
1241   ret->cfg = cfg;
1242   ret->cls = cls;
1243   ret->rec = rec;
1244   ret->nc_cb = nc;
1245   ret->nd_cb = nd;
1246   ret->reconnect_delay = GNUNET_TIME_UNIT_ZERO;
1247   schedule_reconnect (ret);
1248   return ret;
1249 }
1250
1251
1252 /**
1253  * Disconnect from the transport service.
1254  */
1255 void
1256 GNUNET_TRANSPORT_disconnect (struct GNUNET_TRANSPORT_Handle *handle)
1257 {
1258   struct GNUNET_TRANSPORT_TransmitHandle *th;
1259   struct NeighbourList *n;
1260   struct HelloWaitList *hwl;
1261   struct GNUNET_CLIENT_Connection *client;
1262   struct ControlMessage *cm;
1263
1264 #if DEBUG_TRANSPORT
1265   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Transport disconnect called!\n");
1266 #endif
1267   while (NULL != (n = handle->neighbours))
1268     {
1269       handle->neighbours = n->next;
1270       switch (n->transmit_stage)
1271         {
1272         case TS_NEW:
1273         case TS_TRANSMITTED:
1274           /* nothing to do */
1275           break;
1276         case TS_QUEUED:
1277         case TS_TRANSMITTED_QUEUED:
1278           th = &n->transmit_handle;
1279           if (th->notify_delay_task != GNUNET_SCHEDULER_NO_TASK)
1280             {
1281               GNUNET_SCHEDULER_cancel (handle->sched,
1282                                        th->notify_delay_task);
1283               th->notify_delay_task = GNUNET_SCHEDULER_NO_TASK;
1284             }
1285           GNUNET_assert (0 == th->notify (th->notify_cls, 0, NULL));        
1286           break;
1287         default:
1288           GNUNET_break (0);
1289         }
1290       GNUNET_free (n);
1291     }
1292   while (NULL != (hwl = handle->hwl_head))
1293     {
1294       handle->hwl_head = hwl->next;
1295       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1296                   _
1297                   ("Disconnect while notification for `%s' still registered.\n"),
1298                   "HELLO");
1299       if (hwl->rec != NULL)
1300         hwl->rec (hwl->rec_cls, NULL);
1301       GNUNET_free (hwl);
1302     }
1303
1304   /* Check for still scheduled control messages, cancel delay tasks if so */
1305   /* Added because somehow a notify_delay_task is remaining scheduled and is ever so annoying */
1306   while ( (NULL != (cm = handle->control_head)))
1307     {
1308       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1309                   _("Disconnect before control message sent!\n"));
1310       if (cm->notify_delay_task != GNUNET_SCHEDULER_NO_TASK)
1311         {
1312           GNUNET_SCHEDULER_cancel (handle->sched, cm->notify_delay_task);
1313           cm->notify_delay_task = GNUNET_SCHEDULER_NO_TASK;
1314         }
1315       GNUNET_CONTAINER_DLL_remove (handle->control_head,
1316                                    handle->control_tail,
1317                                    cm);
1318       GNUNET_free (cm);
1319     }
1320   /* end check */
1321
1322   if (handle->reconnect_task != GNUNET_SCHEDULER_NO_TASK)
1323     {
1324       GNUNET_SCHEDULER_cancel (handle->sched, handle->reconnect_task);
1325       handle->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
1326     }
1327   if (handle->quota_task != GNUNET_SCHEDULER_NO_TASK)
1328     {
1329       GNUNET_SCHEDULER_cancel (handle->sched, handle->quota_task);
1330       handle->quota_task = GNUNET_SCHEDULER_NO_TASK;
1331     }
1332   GNUNET_free_non_null (handle->my_hello);
1333   handle->my_hello = NULL;
1334   GNUNET_ARM_stop_services (handle->cfg, handle->sched, "transport",
1335                             "peerinfo", NULL);
1336   if (NULL != handle->network_handle)
1337     {
1338       GNUNET_CLIENT_notify_transmit_ready_cancel (handle->network_handle);
1339       handle->network_handle = NULL;
1340     }
1341   if (NULL != (client = handle->client))
1342     {
1343 #if DEBUG_TRANSPORT
1344       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1345                   "Disconnecting from transport service for good.\n");
1346 #endif
1347       handle->client = NULL;
1348       GNUNET_CLIENT_disconnect (client);
1349     }
1350   GNUNET_free (handle);
1351 }
1352
1353
1354 /**
1355  * Function we use for handling incoming messages.
1356  *
1357  * @param cls closure (struct GNUNET_TRANSPORT_Handle *)
1358  * @param msg message received, NULL on timeout or fatal error
1359  */
1360 static void
1361 demultiplexer (void *cls, const struct GNUNET_MessageHeader *msg)
1362 {
1363   struct GNUNET_TRANSPORT_Handle *h = cls;
1364   const struct DisconnectInfoMessage *dim;
1365   const struct ConnectInfoMessage *cim;
1366   const struct InboundMessage *im;
1367   const struct GNUNET_MessageHeader *imm;
1368   const struct SendOkMessage *okm;
1369   struct HelloWaitList *hwl;
1370   struct HelloWaitList *next_hwl;
1371   struct NeighbourList *n;
1372   struct GNUNET_PeerIdentity me;
1373   uint16_t size;
1374
1375   if (h->client == NULL)
1376     {
1377       /* shutdown initiated from 'GNUNET_TRANSPORT_disconnect',
1378          finish clean up work! */
1379       GNUNET_free (h);
1380       return;
1381     }
1382   if (msg == NULL) 
1383     {
1384 #if DEBUG_TRANSPORT
1385       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1386                   "Error receiving from transport service, disconnecting temporarily.\n");
1387 #endif
1388       if (h->network_handle != NULL)
1389         {
1390           GNUNET_CLIENT_notify_transmit_ready_cancel (h->network_handle);
1391           h->network_handle = NULL;
1392         }
1393       GNUNET_CLIENT_disconnect (h->client);
1394       h->client = NULL;
1395       schedule_reconnect (h);
1396       return;
1397     }
1398   GNUNET_CLIENT_receive (h->client,
1399                          &demultiplexer, h, GNUNET_TIME_UNIT_FOREVER_REL);
1400   size = ntohs (msg->size);
1401   switch (ntohs (msg->type))
1402     {
1403     case GNUNET_MESSAGE_TYPE_HELLO:
1404       if (GNUNET_OK !=
1405           GNUNET_HELLO_get_id ((const struct GNUNET_HELLO_Message *) msg,
1406                                &me))
1407         {
1408           GNUNET_break (0);
1409           break;
1410         }
1411 #if DEBUG_TRANSPORT
1412       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1413                   "Receiving (my own) `%s' message, I am `%4s'.\n",
1414                   "HELLO", GNUNET_i2s (&me));
1415 #endif
1416       GNUNET_free_non_null (h->my_hello);
1417       h->my_hello = NULL;
1418       if (size < sizeof (struct GNUNET_MessageHeader))
1419         {
1420           GNUNET_break (0);
1421           break;
1422         }
1423       h->my_hello = GNUNET_malloc (size);
1424       memcpy (h->my_hello, msg, size);
1425       hwl = h->hwl_head;
1426       while (NULL != hwl)
1427         {
1428           next_hwl = hwl->next;
1429           hwl->rec (hwl->rec_cls,
1430                     (const struct GNUNET_MessageHeader *) h->my_hello);
1431           hwl = next_hwl;
1432         }
1433       break;
1434     case GNUNET_MESSAGE_TYPE_TRANSPORT_CONNECT:
1435       if (size != sizeof (struct ConnectInfoMessage))
1436         {
1437           GNUNET_break (0);
1438           break;
1439         }
1440       cim = (const struct ConnectInfoMessage *) msg;
1441 #if DEBUG_TRANSPORT
1442       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1443                   "Receiving `%s' message for `%4s'.\n",
1444                   "CONNECT", GNUNET_i2s (&cim->id));
1445 #endif
1446       n = neighbour_find (h, &cim->id);
1447       if (n == NULL)
1448         n = neighbour_add (h,
1449                            &cim->id);
1450       GNUNET_break (n->is_connected == GNUNET_NO);
1451       n->is_connected = GNUNET_YES;
1452       if (h->nc_cb != NULL)
1453         h->nc_cb (h->cls, &n->id,
1454                   GNUNET_TIME_relative_ntoh (cim->latency), 
1455                   ntohs (cim->distance));
1456       break;
1457     case GNUNET_MESSAGE_TYPE_TRANSPORT_DISCONNECT:
1458       if (size != sizeof (struct DisconnectInfoMessage))
1459         {
1460           GNUNET_break (0);
1461           break;
1462         }
1463       dim = (const struct DisconnectInfoMessage *) msg;
1464 #if DEBUG_TRANSPORT_DISCONNECT
1465       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1466                   "Receiving `%s' message for `%4s'.\n",
1467                   "DISCONNECT",
1468                   GNUNET_i2s (&dim->peer));
1469 #endif
1470       n = neighbour_find (h, &dim->peer);
1471       GNUNET_break (n != NULL);      
1472       if (n != NULL)
1473         neighbour_disconnect (n);       
1474       break;
1475     case GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_OK:
1476       if (size != sizeof (struct SendOkMessage))
1477         {
1478           GNUNET_break (0);
1479           break;
1480         }
1481       okm = (const struct SendOkMessage *) msg;
1482 #if DEBUG_TRANSPORT
1483       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1484                   "Receiving `%s' message, transmission %s.\n", "SEND_OK",
1485                   ntohl (okm->success) == GNUNET_OK ? "succeeded" : "failed");
1486 #endif
1487       n = neighbour_find (h, &okm->peer);
1488       GNUNET_assert (n != NULL);
1489       switch (n->transmit_stage)
1490         {
1491         case TS_NEW:
1492           GNUNET_break (0);
1493           break;
1494         case TS_QUEUED:
1495           GNUNET_break (0);
1496           break;
1497         case TS_TRANSMITTED:
1498           n->transmit_stage = TS_NEW;
1499           break;
1500         case TS_TRANSMITTED_QUEUED:
1501           n->transmit_stage = TS_QUEUED;
1502           schedule_transmission (h);
1503           break;
1504         default:
1505           GNUNET_break (0);
1506         }
1507       break;
1508     case GNUNET_MESSAGE_TYPE_TRANSPORT_RECV:
1509 #if DEBUG_TRANSPORT
1510       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1511                   "Receiving `%s' message.\n", "RECV");
1512 #endif
1513       if (size <
1514           sizeof (struct InboundMessage) +
1515           sizeof (struct GNUNET_MessageHeader))
1516         {
1517           GNUNET_break (0);
1518           break;
1519         }
1520       im = (const struct InboundMessage *) msg;
1521       imm = (const struct GNUNET_MessageHeader *) &im[1];
1522       if (ntohs (imm->size) + sizeof (struct InboundMessage) != size)
1523         {
1524           GNUNET_break (0);
1525           break;
1526         }
1527 #if DEBUG_TRANSPORT
1528       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1529                   "Received message of type %u from `%4s'.\n",
1530                   ntohs (imm->type), GNUNET_i2s (&im->peer));
1531 #endif      
1532       n = neighbour_find (h, &im->peer);
1533       if (n == NULL)
1534         n = neighbour_add (h, &im->peer);
1535       if (n == NULL) 
1536         break;
1537       if (h->rec != NULL)
1538         h->rec (h->cls, &im->peer, imm,
1539                 GNUNET_TIME_relative_ntoh (im->latency), ntohs(im->distance));
1540       break;
1541     default:
1542       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1543                   _
1544                   ("Received unexpected message of type %u in %s:%u\n"),
1545                   ntohs (msg->type), __FILE__, __LINE__);
1546       GNUNET_break (0);
1547       break;
1548     }
1549 }
1550
1551
1552 /**
1553  * Called when our transmit request timed out before any transport
1554  * reported success connecting to the desired peer or before the
1555  * transport was ready to receive.  Signal error and free
1556  * TransmitHandle.
1557  *
1558  * @param cls the 'struct GNUNET_TRANSPORT_TransmitHandle*' that is timing out
1559  * @param tc scheduler context
1560  */
1561 static void
1562 peer_transmit_timeout (void *cls,
1563                        const struct GNUNET_SCHEDULER_TaskContext *tc)
1564 {
1565   struct GNUNET_TRANSPORT_TransmitHandle *th = cls;
1566   struct NeighbourList *n;
1567   GNUNET_CONNECTION_TransmitReadyNotify notify;
1568   void *notify_cls;
1569
1570   th->notify_delay_task = GNUNET_SCHEDULER_NO_TASK;
1571   n = th->neighbour;
1572 #if DEBUG_TRANSPORT
1573   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1574               "Triggering timeout for request to transmit to `%4s' (%d)\n",
1575               GNUNET_i2s (&n->id),
1576               n->transmit_stage);
1577 #endif  
1578   notify = th->notify;
1579   notify_cls = th->notify_cls;
1580   switch (n->transmit_stage)
1581     {
1582     case TS_NEW:
1583       GNUNET_break (0);
1584       break;
1585     case TS_QUEUED:
1586       n->transmit_stage = TS_NEW;
1587       if (n->is_connected == GNUNET_NO)
1588         neighbour_free (n);
1589       break;
1590     case TS_TRANSMITTED:
1591       GNUNET_break (0);
1592       break;
1593     case TS_TRANSMITTED_QUEUED:
1594       n->transmit_stage = TS_TRANSMITTED;
1595       break;
1596     default:
1597       GNUNET_break (0);
1598     }
1599   if (NULL != notify)
1600     notify (notify_cls, 0, NULL);
1601 }
1602
1603
1604 /**
1605  * Check if we could queue a message of the given size for
1606  * transmission.  The transport service will take both its
1607  * internal buffers and bandwidth limits imposed by the
1608  * other peer into consideration when answering this query.
1609  *
1610  * @param handle connection to transport service
1611  * @param target who should receive the message
1612  * @param size how big is the message we want to transmit?
1613  * @param priority how important is the message?
1614  * @param timeout after how long should we give up (and call
1615  *        notify with buf NULL and size 0)?
1616  * @param notify function to call when we are ready to
1617  *        send such a message
1618  * @param notify_cls closure for notify
1619  * @return NULL if someone else is already waiting to be notified
1620  *         non-NULL if the notify callback was queued (can be used to cancel
1621  *         using GNUNET_TRANSPORT_notify_transmit_ready_cancel)
1622  */
1623 struct GNUNET_TRANSPORT_TransmitHandle *
1624 GNUNET_TRANSPORT_notify_transmit_ready (struct GNUNET_TRANSPORT_Handle
1625                                         *handle,
1626                                         const struct GNUNET_PeerIdentity
1627                                         *target, size_t size,
1628                                         unsigned int priority,
1629                                         struct GNUNET_TIME_Relative timeout,
1630                                         GNUNET_CONNECTION_TransmitReadyNotify
1631                                         notify, void *notify_cls)
1632 {
1633   struct GNUNET_TRANSPORT_TransmitHandle *th;
1634   struct NeighbourList *n;
1635
1636   if (size + sizeof (struct OutboundMessage) >=
1637       GNUNET_SERVER_MAX_MESSAGE_SIZE)
1638     {
1639 #if DEBUG_TRANSPORT
1640       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1641                   "Message size is %d, max allowed is %d.\n",
1642                   size + sizeof (struct OutboundMessage), GNUNET_SERVER_MAX_MESSAGE_SIZE);
1643 #endif
1644       GNUNET_break (0);
1645       return NULL;
1646     }
1647 #if DEBUG_TRANSPORT
1648   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1649               "Asking transport service for transmission of %u bytes to peer `%4s' within %llu ms.\n",
1650               size, GNUNET_i2s (target),
1651               (unsigned long long) timeout.value);
1652 #endif
1653   n = neighbour_find (handle, target);
1654   if (n == NULL)
1655     n = neighbour_add (handle, target);
1656   if (n == NULL) 
1657     return NULL;
1658   switch (n->transmit_stage)
1659     {
1660     case TS_NEW:
1661       n->transmit_stage = TS_QUEUED;
1662       break;
1663     case TS_QUEUED:
1664       GNUNET_break (0);
1665       return NULL;
1666     case TS_TRANSMITTED:
1667       n->transmit_stage = TS_TRANSMITTED_QUEUED;
1668       break;
1669     case TS_TRANSMITTED_QUEUED:
1670       GNUNET_break (0);
1671       return NULL;
1672     default:
1673       GNUNET_break (0);
1674       return NULL;
1675     }
1676   th = &n->transmit_handle;
1677   th->neighbour = n;
1678   th->notify = notify;
1679   th->notify_cls = notify_cls;
1680   th->timeout = GNUNET_TIME_relative_to_absolute (timeout);
1681   th->notify_size = size + sizeof (struct OutboundMessage);
1682   th->priority = priority;
1683   th->notify_delay_task
1684     = GNUNET_SCHEDULER_add_delayed (handle->sched, timeout,
1685                                     &peer_transmit_timeout, th);
1686   schedule_transmission (handle);
1687   return th;
1688 }
1689
1690
1691 /**
1692  * Cancel the specified transmission-ready notification.
1693  */
1694 void
1695 GNUNET_TRANSPORT_notify_transmit_ready_cancel (struct
1696                                                GNUNET_TRANSPORT_TransmitHandle
1697                                                *th)
1698 {
1699   struct NeighbourList *n;
1700
1701   n = th->neighbour;
1702 #if DEBUG_TRANSPORT
1703   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1704               "Transmission request of %u bytes to `%4s' was cancelled.\n",
1705               th->notify_size - sizeof (struct OutboundMessage),
1706               GNUNET_i2s (&n->id));
1707 #endif
1708   switch (n->transmit_stage)
1709     {
1710     case TS_NEW:
1711       GNUNET_break (0);
1712       break;
1713     case TS_QUEUED:
1714       n->transmit_stage = TS_NEW;
1715       if (n->is_connected == GNUNET_NO)
1716         neighbour_free (n);
1717       break;
1718     case TS_TRANSMITTED:
1719       GNUNET_break (0);
1720       break;
1721     case TS_TRANSMITTED_QUEUED:
1722       n->transmit_stage = TS_TRANSMITTED;
1723       break;
1724     default:
1725       GNUNET_break (0);
1726     }
1727 }
1728
1729
1730 /* end of transport_api.c */