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