4c110a134bf81f282513388d18fddfafc86054f6
[oweals/gnunet.git] / src / util / server.c
1 /*
2      This file is part of GNUnet.
3      (C) 2009, 2012 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 util/server.c
23  * @brief library for building GNUnet network servers
24  * @author Christian Grothoff
25  */
26
27 #include "platform.h"
28 #include "gnunet_common.h"
29 #include "gnunet_util_lib.h"
30 #include "gnunet_protocols.h"
31
32 #define LOG(kind,...) GNUNET_log_from (kind, "util", __VA_ARGS__)
33
34 #define LOG_STRERROR(kind,syscall) GNUNET_log_from_strerror (kind, "util", syscall)
35
36 #define LOG_STRERROR_FILE(kind,syscall,filename) GNUNET_log_from_strerror_file (kind, "util", syscall, filename)
37
38
39 /**
40  * List of arrays of message handlers.
41  */
42 struct HandlerList
43 {
44   /**
45    * This is a linked list.
46    */
47   struct HandlerList *next;
48
49   /**
50    * NULL-terminated array of handlers.
51    */
52   const struct GNUNET_SERVER_MessageHandler *handlers;
53 };
54
55
56 /**
57  * List of arrays of message handlers.
58  */
59 struct NotifyList
60 {
61   /**
62    * This is a doubly linked list.
63    */
64   struct NotifyList *next;
65
66   /**
67    * This is a doubly linked list.
68    */
69   struct NotifyList *prev;
70
71   /**
72    * Function to call.
73    */
74   GNUNET_SERVER_DisconnectCallback callback;
75
76   /**
77    * Closure for callback.
78    */
79   void *callback_cls;
80 };
81
82
83 /**
84  * @brief handle for a server
85  */
86 struct GNUNET_SERVER_Handle
87 {
88   /**
89    * List of handlers for incoming messages.
90    */
91   struct HandlerList *handlers;
92
93   /**
94    * List of our current clients.
95    */
96   struct GNUNET_SERVER_Client *clients;
97
98   /**
99    * Head of linked list of functions to call on disconnects by clients.
100    */
101   struct NotifyList *disconnect_notify_list_head;
102
103   /**
104    * Tail of linked list of functions to call on disconnects by clients.
105    */
106   struct NotifyList *disconnect_notify_list_tail;
107
108   /**
109    * Function to call for access control.
110    */
111   GNUNET_CONNECTION_AccessCheck access;
112
113   /**
114    * Closure for access.
115    */
116   void *access_cls;
117
118   /**
119    * NULL-terminated array of sockets used to listen for new
120    * connections.
121    */
122   struct GNUNET_NETWORK_Handle **listen_sockets;
123
124   /**
125    * After how long should an idle connection time
126    * out (on write).
127    */
128   struct GNUNET_TIME_Relative idle_timeout;
129
130   /**
131    * Task scheduled to do the listening.
132    */
133   GNUNET_SCHEDULER_TaskIdentifier listen_task;
134
135   /**
136    * Alternative function to create a MST instance.
137    */
138   GNUNET_SERVER_MstCreateCallback mst_create;
139
140   /**
141    * Alternative function to destroy a MST instance.
142    */
143   GNUNET_SERVER_MstDestroyCallback mst_destroy;
144
145   /**
146    * Alternative function to give data to a MST instance.
147    */
148   GNUNET_SERVER_MstReceiveCallback mst_receive;
149
150   /**
151    * Closure for 'mst_'-callbacks.
152    */
153   void *mst_cls;
154
155   /**
156    * Do we ignore messages of types that we do not understand or do we
157    * require that a handler is found (and if not kill the connection)?
158    */
159   int require_found;
160
161   /**
162    * Set to GNUNET_YES once we are in 'soft' shutdown where we wait for
163    * all non-monitor clients to disconnect before we call
164    * GNUNET_SERVER_destroy.  See 'test_monitor_clients'.  Set to
165    * GNUNET_SYSERR once the final destroy task has been scheduled
166    * (we cannot run it in the same task).
167    */
168   int in_soft_shutdown;
169 };
170
171
172 /**
173  * Handle server returns for aborting transmission to a client.
174  */
175 struct GNUNET_SERVER_TransmitHandle
176 {
177   /**
178    * Function to call to get the message.
179    */
180   GNUNET_CONNECTION_TransmitReadyNotify callback;
181
182   /**
183    * Closure for 'callback'
184    */
185   void *callback_cls;
186
187   /**
188    * Active connection transmission handle.
189    */
190   struct GNUNET_CONNECTION_TransmitHandle *cth;
191
192 };
193
194
195 /**
196  * @brief handle for a client of the server
197  */
198 struct GNUNET_SERVER_Client
199 {
200
201   /**
202    * This is a linked list.
203    */
204   struct GNUNET_SERVER_Client *next;
205
206   /**
207    * Processing of incoming data.
208    */
209   void *mst;
210
211   /**
212    * Server that this client belongs to.
213    */
214   struct GNUNET_SERVER_Handle *server;
215
216   /**
217    * Client closure for callbacks.
218    */
219   struct GNUNET_CONNECTION_Handle *connection;
220
221   /**
222    * ID of task used to restart processing.
223    */
224   GNUNET_SCHEDULER_TaskIdentifier restart_task;
225
226   /**
227    * Task that warns about missing calls to 'GNUNET_SERVER_receive_done'.
228    */
229   GNUNET_SCHEDULER_TaskIdentifier warn_task;
230
231   /**
232    * Time when the warn task was started.
233    */
234   struct GNUNET_TIME_Absolute warn_start;
235
236   /**
237    * Last activity on this socket (used to time it out
238    * if reference_count == 0).
239    */
240   struct GNUNET_TIME_Absolute last_activity;
241
242   /**
243    * Transmission handle we return for this client from
244    * GNUNET_SERVER_notify_transmit_ready.
245    */
246   struct GNUNET_SERVER_TransmitHandle th;
247
248   /**
249    * After how long should an idle connection time
250    * out (on write).
251    */
252   struct GNUNET_TIME_Relative idle_timeout;
253
254   /**
255    * Number of external entities with a reference to
256    * this client object.
257    */
258   unsigned int reference_count;
259
260   /**
261    * Was processing if incoming messages suspended while
262    * we were still processing data already received?
263    * This is a counter saying how often processing was
264    * suspended (once per handler invoked).
265    */
266   unsigned int suspended;
267
268   /**
269    * Are we currently in the "process_client_buffer" function (and
270    * will hence restart the receive job on exit if suspended == 0 once
271    * we are done?).  If this is set, then "receive_done" will
272    * essentially only decrement suspended; if this is not set, then
273    * "receive_done" may need to restart the receive process (either
274    * from the side-buffer or via select/recv).
275    */
276   int in_process_client_buffer;
277
278   /**
279    * We're about to close down this client due to some serious
280    * error.
281    */
282   int shutdown_now;
283
284   /**
285    * Are we currently trying to receive? (YES if we are, NO if we are not,
286    * SYSERR if data is already available in MST).
287    */
288   int receive_pending;
289
290   /**
291    * Finish pending write when disconnecting?
292    */
293   int finish_pending_write;
294
295   /**
296    * Persist the file handle for this client no matter what happens,
297    * force the OS to close once the process actually dies.  Should only
298    * be used in special cases!
299    */
300   int persist;
301
302   /**
303    * Is this client a 'monitor' client that should not be counted
304    * when deciding on destroying the server during soft shutdown?
305    * (see also GNUNET_SERVICE_start)
306    */
307   int is_monitor;
308
309   /**
310    * Type of last message processed (for warn_no_receive_done).
311    */
312   uint16_t warn_type;
313 };
314
315
316 /**
317  * Scheduler says our listen socket is ready.  Process it!
318  *
319  * @param cls handle to our server for which we are processing the listen
320  *        socket
321  * @param tc reason why we are running right now
322  */
323 static void
324 process_listen_socket (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
325 {
326   struct GNUNET_SERVER_Handle *server = cls;
327   struct GNUNET_CONNECTION_Handle *sock;
328   struct GNUNET_SERVER_Client *client;
329   struct GNUNET_NETWORK_FDSet *r;
330   unsigned int i;
331
332   server->listen_task = GNUNET_SCHEDULER_NO_TASK;
333   r = GNUNET_NETWORK_fdset_create ();
334   i = 0;
335   while (NULL != server->listen_sockets[i])
336     GNUNET_NETWORK_fdset_set (r, server->listen_sockets[i++]);
337   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
338   {
339     /* ignore shutdown, someone else will take care of it! */
340     server->listen_task =
341         GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_HIGH,
342                                      GNUNET_TIME_UNIT_FOREVER_REL, r, NULL,
343                                      &process_listen_socket, server);
344     GNUNET_NETWORK_fdset_destroy (r);
345     return;
346   }
347   i = 0;
348   while (NULL != server->listen_sockets[i])
349   {
350     if (GNUNET_NETWORK_fdset_isset (tc->read_ready, server->listen_sockets[i]))
351     {
352       sock =
353           GNUNET_CONNECTION_create_from_accept (server->access,
354                                                 server->access_cls,
355                                                 server->listen_sockets[i]);
356       if (NULL != sock)
357       {
358         LOG (GNUNET_ERROR_TYPE_DEBUG, "Server accepted incoming connection.\n");
359         client = GNUNET_SERVER_connect_socket (server, sock);
360         /* decrement reference count, we don't keep "client" alive */
361         GNUNET_SERVER_client_drop (client);
362       }
363     }
364     i++;
365   }
366   /* listen for more! */
367   server->listen_task =
368       GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_HIGH,
369                                    GNUNET_TIME_UNIT_FOREVER_REL, r, NULL,
370                                    &process_listen_socket, server);
371   GNUNET_NETWORK_fdset_destroy (r);
372 }
373
374
375 /**
376  * Create and initialize a listen socket for the server.
377  *
378  * @param serverAddr address to listen on
379  * @param socklen length of address
380  * @return NULL on error, otherwise the listen socket
381  */
382 static struct GNUNET_NETWORK_Handle *
383 open_listen_socket (const struct sockaddr *serverAddr, socklen_t socklen)
384 {
385   const static int on = 1;
386   struct GNUNET_NETWORK_Handle *sock;
387   uint16_t port;
388   int eno;
389
390   switch (serverAddr->sa_family)
391   {
392   case AF_INET:
393     port = ntohs (((const struct sockaddr_in *) serverAddr)->sin_port);
394     break;
395   case AF_INET6:
396     port = ntohs (((const struct sockaddr_in6 *) serverAddr)->sin6_port);
397     break;
398   case AF_UNIX:
399     port = 0;
400     break;
401   default:
402     GNUNET_break (0);
403     port = 0;
404     break;
405   }
406   sock = GNUNET_NETWORK_socket_create (serverAddr->sa_family, SOCK_STREAM, 0);
407   if (NULL == sock)
408   {
409     LOG_STRERROR (GNUNET_ERROR_TYPE_ERROR, "socket");
410     errno = 0;
411     return NULL;
412   }
413   if (0 != port)
414   {
415     if (GNUNET_NETWORK_socket_setsockopt
416         (sock, SOL_SOCKET, SO_REUSEADDR, &on, sizeof (on)) != GNUNET_OK)
417       LOG_STRERROR (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
418                     "setsockopt");
419 #ifdef IPV6_V6ONLY
420     if ((AF_INET6 == serverAddr->sa_family) &&
421         (GNUNET_NETWORK_socket_setsockopt
422          (sock, IPPROTO_IPV6, IPV6_V6ONLY, &on, sizeof (on)) != GNUNET_OK))
423       LOG_STRERROR (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
424                     "setsockopt");
425 #endif
426   }
427   /* bind the socket */
428   if (GNUNET_OK != GNUNET_NETWORK_socket_bind (sock, serverAddr, socklen))
429   {
430     eno = errno;
431     if (EADDRINUSE != errno)
432     {
433       /* we don't log 'EADDRINUSE' here since an IPv4 bind may
434        * fail if we already took the port on IPv6; if both IPv4 and
435        * IPv6 binds fail, then our caller will log using the
436        * errno preserved in 'eno' */
437       LOG_STRERROR (GNUNET_ERROR_TYPE_ERROR, "bind");
438       if (0 != port)
439         LOG (GNUNET_ERROR_TYPE_ERROR, _("`%s' failed for port %d (%s).\n"),
440              "bind", port,
441              (AF_INET == serverAddr->sa_family) ? "IPv4" : "IPv6");
442       eno = 0;
443     }
444     else
445     {
446       if (0 != port)
447         LOG (GNUNET_ERROR_TYPE_WARNING,
448              _("`%s' failed for port %d (%s): address already in use\n"),
449              "bind", port,
450              (AF_INET == serverAddr->sa_family) ? "IPv4" : "IPv6");
451       else if (AF_UNIX == serverAddr->sa_family)
452         LOG (GNUNET_ERROR_TYPE_WARNING,
453              _("`%s' failed for `%s': address already in use\n"), "bind",
454              ((const struct sockaddr_un *) serverAddr)->sun_path);
455
456     }
457     GNUNET_break (GNUNET_OK == GNUNET_NETWORK_socket_close (sock));
458     errno = eno;
459     return NULL;
460   }
461   if (GNUNET_OK != GNUNET_NETWORK_socket_listen (sock, 5))
462   {
463     LOG_STRERROR (GNUNET_ERROR_TYPE_ERROR, "listen");
464     GNUNET_break (GNUNET_OK == GNUNET_NETWORK_socket_close (sock));
465     errno = 0;
466     return NULL;
467   }
468   if (0 != port)
469     LOG (GNUNET_ERROR_TYPE_DEBUG, "Server starts to listen on port %u.\n",
470          port);
471   return sock;
472 }
473
474
475 /**
476  * Create a new server.
477  *
478  * @param access function for access control
479  * @param access_cls closure for access
480  * @param lsocks NULL-terminated array of listen sockets
481  * @param idle_timeout after how long should we timeout idle connections?
482  * @param require_found if YES, connections sending messages of unknown type
483  *        will be closed
484  * @return handle for the new server, NULL on error
485  *         (typically, "port" already in use)
486  */
487 struct GNUNET_SERVER_Handle *
488 GNUNET_SERVER_create_with_sockets (GNUNET_CONNECTION_AccessCheck access,
489                                    void *access_cls,
490                                    struct GNUNET_NETWORK_Handle **lsocks,
491                                    struct GNUNET_TIME_Relative idle_timeout,
492                                    int require_found)
493 {
494   struct GNUNET_SERVER_Handle *server;
495   struct GNUNET_NETWORK_FDSet *r;
496   int i;
497
498   server = GNUNET_malloc (sizeof (struct GNUNET_SERVER_Handle));
499   server->idle_timeout = idle_timeout;
500   server->listen_sockets = lsocks;
501   server->access = access;
502   server->access_cls = access_cls;
503   server->require_found = require_found;
504   if (NULL != lsocks)
505   {
506     r = GNUNET_NETWORK_fdset_create ();
507     i = 0;
508     while (NULL != server->listen_sockets[i])
509       GNUNET_NETWORK_fdset_set (r, server->listen_sockets[i++]);
510     server->listen_task =
511         GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_HIGH,
512                                      GNUNET_TIME_UNIT_FOREVER_REL, r, NULL,
513                                      &process_listen_socket, server);
514     GNUNET_NETWORK_fdset_destroy (r);
515   }
516   return server;
517 }
518
519
520 /**
521  * Create a new server.
522  *
523  * @param access function for access control
524  * @param access_cls closure for access
525  * @param serverAddr address to listen on (including port), NULL terminated array
526  * @param socklen length of serverAddr
527  * @param idle_timeout after how long should we timeout idle connections?
528  * @param require_found if YES, connections sending messages of unknown type
529  *        will be closed
530  * @return handle for the new server, NULL on error
531  *         (typically, "port" already in use)
532  */
533 struct GNUNET_SERVER_Handle *
534 GNUNET_SERVER_create (GNUNET_CONNECTION_AccessCheck access, void *access_cls,
535                       struct sockaddr *const *serverAddr,
536                       const socklen_t * socklen,
537                       struct GNUNET_TIME_Relative idle_timeout,
538                       int require_found)
539 {
540   struct GNUNET_NETWORK_Handle **lsocks;
541   unsigned int i;
542   unsigned int j;
543   unsigned int k;
544   int seen;
545
546   i = 0;
547   while (NULL != serverAddr[i])
548     i++;
549   if (i > 0)
550   {
551     lsocks = GNUNET_malloc (sizeof (struct GNUNET_NETWORK_Handle *) * (i + 1));
552     i = 0;
553     j = 0;
554     while (NULL != serverAddr[i])
555     {
556       seen = 0;
557       for (k=0;k<i;k++)
558         if ( (socklen[k] == socklen[i]) &&
559              (0 == memcmp (serverAddr[k], serverAddr[i], socklen[i])) )
560         {
561           seen = 1;
562           break;
563         }
564       if (0 != seen)
565       {
566         /* duplicate address, skip */
567         i++;
568         continue;
569       }
570       lsocks[j] = open_listen_socket (serverAddr[i], socklen[i]);
571       if (NULL != lsocks[j])
572         j++;
573       i++;
574     }
575     if (0 == j)
576     {
577       if (0 != errno)
578         LOG_STRERROR (GNUNET_ERROR_TYPE_ERROR, "bind");
579       GNUNET_free (lsocks);
580       lsocks = NULL;
581     }
582   }
583   else
584   {
585     lsocks = NULL;
586   }
587   return GNUNET_SERVER_create_with_sockets (access, access_cls, lsocks,
588                                             idle_timeout, require_found);
589 }
590
591
592 /**
593  * Set the 'monitor' flag on this client.  Clients which have been
594  * marked as 'monitors' won't prevent the server from shutting down
595  * once 'GNUNET_SERVER_stop_listening' has been invoked.  The idea is
596  * that for "normal" clients we likely want to allow them to process
597  * their requests; however, monitor-clients are likely to 'never'
598  * disconnect during shutdown and thus will not be considered when
599  * determining if the server should continue to exist after
600  * 'GNUNET_SERVER_destroy' has been called.
601  *
602  * @param client the client to set the 'monitor' flag on
603  */
604 void
605 GNUNET_SERVER_client_mark_monitor (struct GNUNET_SERVER_Client *client)
606 {
607   client->is_monitor = GNUNET_YES;
608 }
609
610
611 /**
612  * Helper function for 'test_monitor_clients' to trigger
613  * 'GNUNET_SERVER_destroy' after the stack has unwound.
614  *
615  * @param cls the 'struct GNUNET_SERVER_Handle' to destroy
616  * @param tc unused
617  */
618 static void
619 do_destroy (void *cls,
620             const struct GNUNET_SCHEDULER_TaskContext *tc)
621 {
622   struct GNUNET_SERVER_Handle *server = cls;
623   GNUNET_SERVER_destroy (server);
624 }
625
626
627 /**
628  * Check if only 'monitor' clients are left.  If so, destroy the
629  * server completely.
630  *
631  * @param server server to test for full shutdown
632  */
633 static void
634 test_monitor_clients (struct GNUNET_SERVER_Handle *server)
635 {
636   struct GNUNET_SERVER_Client *client;
637
638   if (GNUNET_YES != server->in_soft_shutdown)
639     return;
640   for (client = server->clients; NULL != client; client = client->next)
641     if (GNUNET_NO == client->is_monitor)
642       return; /* not done yet */
643   server->in_soft_shutdown = GNUNET_SYSERR;
644   GNUNET_SCHEDULER_add_continuation (&do_destroy, server,
645                                      GNUNET_SCHEDULER_REASON_PREREQ_DONE);
646 }
647
648
649 /**
650  * Stop the listen socket and get ready to shutdown the server
651  * once only 'monitor' clients are left.
652  *
653  * @param server server to stop listening on
654  */
655 void
656 GNUNET_SERVER_stop_listening (struct GNUNET_SERVER_Handle *server)
657 {
658   unsigned int i;
659
660   LOG (GNUNET_ERROR_TYPE_DEBUG, "Server in soft shutdown\n");
661   if (GNUNET_SCHEDULER_NO_TASK != server->listen_task)
662   {
663     GNUNET_SCHEDULER_cancel (server->listen_task);
664     server->listen_task = GNUNET_SCHEDULER_NO_TASK;
665   }
666   if (NULL != server->listen_sockets)
667   {
668     i = 0;
669     while (NULL != server->listen_sockets[i])
670       GNUNET_break (GNUNET_OK ==
671                     GNUNET_NETWORK_socket_close (server->listen_sockets[i++]));
672     GNUNET_free (server->listen_sockets);
673     server->listen_sockets = NULL;
674   }
675   if (GNUNET_NO == server->in_soft_shutdown)
676     server->in_soft_shutdown = GNUNET_YES;
677   test_monitor_clients (server);
678 }
679
680
681 /**
682  * Free resources held by this server.
683  *
684  * @param server server to destroy
685  */
686 void
687 GNUNET_SERVER_destroy (struct GNUNET_SERVER_Handle *server)
688 {
689   struct HandlerList *hpos;
690   struct NotifyList *npos;
691   unsigned int i;
692
693   LOG (GNUNET_ERROR_TYPE_DEBUG, "Server shutting down.\n");
694   if (GNUNET_SCHEDULER_NO_TASK != server->listen_task)
695   {
696     GNUNET_SCHEDULER_cancel (server->listen_task);
697     server->listen_task = GNUNET_SCHEDULER_NO_TASK;
698   }
699   if (NULL != server->listen_sockets)
700   {
701     i = 0;
702     while (NULL != server->listen_sockets[i])
703       GNUNET_break (GNUNET_OK ==
704                     GNUNET_NETWORK_socket_close (server->listen_sockets[i++]));
705     GNUNET_free (server->listen_sockets);
706     server->listen_sockets = NULL;
707   }
708   while (NULL != server->clients)
709     GNUNET_SERVER_client_disconnect (server->clients);
710   while (NULL != (hpos = server->handlers))
711   {
712     server->handlers = hpos->next;
713     GNUNET_free (hpos);
714   }
715   while (NULL != (npos = server->disconnect_notify_list_head))
716   {
717     npos->callback (npos->callback_cls, NULL);
718     GNUNET_CONTAINER_DLL_remove (server->disconnect_notify_list_head,
719                                  server->disconnect_notify_list_tail,
720                                  npos);
721     GNUNET_free (npos);
722   }
723   GNUNET_free (server);
724 }
725
726
727 /**
728  * Add additional handlers to an existing server.
729  *
730  * @param server the server to add handlers to
731  * @param handlers array of message handlers for
732  *        incoming messages; the last entry must
733  *        have "NULL" for the "callback"; multiple
734  *        entries for the same type are allowed,
735  *        they will be called in order of occurence.
736  *        These handlers can be removed later;
737  *        the handlers array must exist until removed
738  *        (or server is destroyed).
739  */
740 void
741 GNUNET_SERVER_add_handlers (struct GNUNET_SERVER_Handle *server,
742                             const struct GNUNET_SERVER_MessageHandler *handlers)
743 {
744   struct HandlerList *p;
745
746   p = GNUNET_malloc (sizeof (struct HandlerList));
747   p->handlers = handlers;
748   p->next = server->handlers;
749   server->handlers = p;
750 }
751
752
753 /**
754  * Change functions used by the server to tokenize the message stream.
755  * (very rarely used).
756  *
757  * @param server server to modify
758  * @param create new tokenizer initialization function
759  * @param destroy new tokenizer destruction function
760  * @param receive new tokenizer receive function
761  * @param cls closure for 'create', 'receive', 'destroy' 
762  */
763 void
764 GNUNET_SERVER_set_callbacks (struct GNUNET_SERVER_Handle *server,
765                              GNUNET_SERVER_MstCreateCallback create,
766                              GNUNET_SERVER_MstDestroyCallback destroy,
767                              GNUNET_SERVER_MstReceiveCallback receive,
768                              void *cls)
769 {
770   server->mst_create = create;
771   server->mst_destroy = destroy;
772   server->mst_receive = receive;
773   server->mst_cls = cls;
774 }
775
776
777 /**
778  * Task run to warn about missing calls to 'GNUNET_SERVER_receive_done'.
779  *
780  * @param cls our 'struct GNUNET_SERVER_Client*' to process more requests from
781  * @param tc scheduler context (unused)
782  */
783 static void
784 warn_no_receive_done (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
785 {
786   struct GNUNET_SERVER_Client *client = cls;
787
788   client->warn_task =
789       GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
790                                     &warn_no_receive_done, client);
791   if (0 == (GNUNET_SCHEDULER_REASON_SHUTDOWN & tc->reason))
792     LOG (GNUNET_ERROR_TYPE_WARNING,
793          _
794          ("Processing code for message of type %u did not call GNUNET_SERVER_receive_done after %llums\n"),
795          (unsigned int) client->warn_type,
796          (unsigned long long)
797          GNUNET_TIME_absolute_get_duration (client->warn_start).rel_value);
798 }
799
800
801 /**
802  * Disable the warning the server issues if a message is not acknowledged
803  * in a timely fashion.  Use this call if a client is intentionally delayed
804  * for a while.  Only applies to the current message.
805  *
806  * @param client client for which to disable the warning
807  */
808 void
809 GNUNET_SERVER_disable_receive_done_warning (struct GNUNET_SERVER_Client *client)
810 {
811   if (GNUNET_SCHEDULER_NO_TASK != client->warn_task)
812   {
813     GNUNET_SCHEDULER_cancel (client->warn_task);
814     client->warn_task = GNUNET_SCHEDULER_NO_TASK;
815   }
816 }
817
818
819 /**
820  * Inject a message into the server, pretend it came
821  * from the specified client.  Delivery of the message
822  * will happen instantly (if a handler is installed;
823  * otherwise the call does nothing).
824  *
825  * @param server the server receiving the message
826  * @param sender the "pretended" sender of the message
827  *        can be NULL!
828  * @param message message to transmit
829  * @return GNUNET_OK if the message was OK and the
830  *                   connection can stay open
831  *         GNUNET_SYSERR if the connection to the
832  *         client should be shut down
833  */
834 int
835 GNUNET_SERVER_inject (struct GNUNET_SERVER_Handle *server,
836                       struct GNUNET_SERVER_Client *sender,
837                       const struct GNUNET_MessageHeader *message)
838 {
839   struct HandlerList *pos;
840   const struct GNUNET_SERVER_MessageHandler *mh;
841   unsigned int i;
842   uint16_t type;
843   uint16_t size;
844   int found;
845
846   type = ntohs (message->type);
847   size = ntohs (message->size);
848   LOG (GNUNET_ERROR_TYPE_DEBUG,
849        "Server schedules transmission of %u-byte message of type %u to client.\n",
850        size, type);
851   found = GNUNET_NO;
852   for (pos = server->handlers; NULL != pos; pos = pos->next)
853   {
854     i = 0;
855     while (pos->handlers[i].callback != NULL)
856     {
857       mh = &pos->handlers[i];
858       if ((mh->type == type) || (mh->type == GNUNET_MESSAGE_TYPE_ALL))
859       {
860         if ((0 != mh->expected_size) && (mh->expected_size != size))
861         {
862 #if GNUNET8_NETWORK_IS_DEAD
863           LOG (GNUNET_ERROR_TYPE_WARNING,
864                "Expected %u bytes for message of type %u, got %u\n",
865                mh->expected_size, mh->type, size);
866           GNUNET_break_op (0);
867 #endif
868           return GNUNET_SYSERR;
869         }
870         if (NULL != sender)
871         {
872           if (0 == sender->suspended)
873           {
874             sender->warn_start = GNUNET_TIME_absolute_get ();
875             sender->warn_task =
876                 GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
877                                               &warn_no_receive_done, sender);
878             sender->warn_type = type;
879           }
880           sender->suspended++;
881         }
882         mh->callback (mh->callback_cls, sender, message);
883         found = GNUNET_YES;
884       }
885       i++;
886     }
887   }
888   if (GNUNET_NO == found)
889   {
890     LOG (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
891          "Received message of unknown type %d\n", type);
892     if (GNUNET_YES == server->require_found)
893       return GNUNET_SYSERR;
894   }
895   return GNUNET_OK;
896 }
897
898
899 /**
900  * We are receiving an incoming message.  Process it.
901  *
902  * @param cls our closure (handle for the client)
903  * @param buf buffer with data received from network
904  * @param available number of bytes available in buf
905  * @param addr address of the sender
906  * @param addrlen length of addr
907  * @param errCode code indicating errors receiving, 0 for success
908  */
909 static void
910 process_incoming (void *cls, const void *buf, size_t available,
911                   const struct sockaddr *addr, socklen_t addrlen, int errCode);
912
913
914 /**
915  * Process messages from the client's message tokenizer until either
916  * the tokenizer is empty (and then schedule receiving more), or
917  * until some handler is not immediately done (then wait for restart_processing)
918  * or shutdown.
919  *
920  * @param client the client to process, RC must have already been increased
921  *        using GNUNET_SERVER_client_keep and will be decreased by one in this
922  *        function
923  * @param ret GNUNET_NO to start processing from the buffer,
924  *            GNUNET_OK if the mst buffer is drained and we should instantly go back to receiving
925  *            GNUNET_SYSERR if we should instantly abort due to error in a previous step
926  */
927 static void
928 process_mst (struct GNUNET_SERVER_Client *client, int ret)
929 {
930   while ((GNUNET_SYSERR != ret) && (NULL != client->server) &&
931          (GNUNET_YES != client->shutdown_now) && (0 == client->suspended))
932   {
933     if (GNUNET_OK == ret)
934     {
935       LOG (GNUNET_ERROR_TYPE_DEBUG,
936            "Server re-enters receive loop, timeout: %llu.\n",
937            client->idle_timeout.rel_value);
938       client->receive_pending = GNUNET_YES;
939       GNUNET_CONNECTION_receive (client->connection,
940                                  GNUNET_SERVER_MAX_MESSAGE_SIZE - 1,
941                                  client->idle_timeout, &process_incoming,
942                                  client);
943       break;
944     }
945     LOG (GNUNET_ERROR_TYPE_DEBUG,
946          "Server processes additional messages instantly.\n");
947     if (NULL != client->server->mst_receive)
948       ret =
949           client->server->mst_receive (client->server->mst_cls, client->mst,
950                                        client, NULL, 0, GNUNET_NO, GNUNET_YES);
951     else
952       ret =
953           GNUNET_SERVER_mst_receive (client->mst, client, NULL, 0, GNUNET_NO,
954                                      GNUNET_YES);
955   }
956   LOG (GNUNET_ERROR_TYPE_DEBUG,
957        "Server leaves instant processing loop: ret = %d, server = %p, shutdown = %d, suspended = %u\n",
958        ret, client->server, client->shutdown_now, client->suspended);
959   if (GNUNET_NO == ret)
960   {
961     LOG (GNUNET_ERROR_TYPE_DEBUG,
962          "Server has more data pending but is suspended.\n");
963     client->receive_pending = GNUNET_SYSERR;    /* data pending */
964   }
965   if ((GNUNET_SYSERR == ret) || (GNUNET_YES == client->shutdown_now))
966     GNUNET_SERVER_client_disconnect (client);
967   GNUNET_SERVER_client_drop (client);
968 }
969
970
971 /**
972  * We are receiving an incoming message.  Process it.
973  *
974  * @param cls our closure (handle for the client)
975  * @param buf buffer with data received from network
976  * @param available number of bytes available in buf
977  * @param addr address of the sender
978  * @param addrlen length of addr
979  * @param errCode code indicating errors receiving, 0 for success
980  */
981 static void
982 process_incoming (void *cls, const void *buf, size_t available,
983                   const struct sockaddr *addr, socklen_t addrlen, int errCode)
984 {
985   struct GNUNET_SERVER_Client *client = cls;
986   struct GNUNET_SERVER_Handle *server = client->server;
987   struct GNUNET_TIME_Absolute end;
988   struct GNUNET_TIME_Absolute now;
989   int ret;
990
991   GNUNET_assert (GNUNET_YES == client->receive_pending);
992   client->receive_pending = GNUNET_NO;
993   now = GNUNET_TIME_absolute_get ();
994   end = GNUNET_TIME_absolute_add (client->last_activity, client->idle_timeout);
995
996   if ((NULL == buf) && (0 == available) && (NULL == addr) && (0 == errCode) &&
997       (GNUNET_YES != client->shutdown_now) && (NULL != server) &&
998       (GNUNET_YES == GNUNET_CONNECTION_check (client->connection)) &&
999       (end.abs_value > now.abs_value))
1000   {
1001     /* wait longer, timeout changed (i.e. due to us sending) */
1002     LOG (GNUNET_ERROR_TYPE_DEBUG,
1003          "Receive time out, but no disconnect due to sending (%p)\n",
1004          GNUNET_a2s (addr, addrlen));
1005     client->receive_pending = GNUNET_YES;
1006     GNUNET_CONNECTION_receive (client->connection,
1007                                GNUNET_SERVER_MAX_MESSAGE_SIZE - 1,
1008                                GNUNET_TIME_absolute_get_remaining (end),
1009                                &process_incoming, client);
1010     return;
1011   }
1012   if ((NULL == buf) || (0 == available) || (0 != errCode) || (NULL == server) ||
1013       (GNUNET_YES == client->shutdown_now) ||
1014       (GNUNET_YES != GNUNET_CONNECTION_check (client->connection)))
1015   {
1016     /* other side closed connection, error connecting, etc. */
1017     GNUNET_SERVER_client_disconnect (client);
1018     return;
1019   }
1020   LOG (GNUNET_ERROR_TYPE_DEBUG, "Server receives %u bytes from `%s'.\n",
1021        (unsigned int) available, GNUNET_a2s (addr, addrlen));
1022   GNUNET_SERVER_client_keep (client);
1023   client->last_activity = now;
1024
1025   if (NULL != server->mst_receive)
1026     ret =
1027         client->server->mst_receive (client->server->mst_cls, client->mst,
1028                                      client, buf, available, GNUNET_NO, GNUNET_YES);
1029   else
1030     ret =
1031         GNUNET_SERVER_mst_receive (client->mst, client, buf, available, GNUNET_NO,
1032                                    GNUNET_YES);
1033   process_mst (client, ret);
1034 }
1035
1036
1037 /**
1038  * Task run to start again receiving from the network
1039  * and process requests.
1040  *
1041  * @param cls our 'struct GNUNET_SERVER_Client*' to process more requests from
1042  * @param tc scheduler context (unused)
1043  */
1044 static void
1045 restart_processing (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1046 {
1047   struct GNUNET_SERVER_Client *client = cls;
1048
1049   client->restart_task = GNUNET_SCHEDULER_NO_TASK;
1050   if (GNUNET_NO == client->receive_pending)
1051   {
1052     LOG (GNUNET_ERROR_TYPE_DEBUG, "Server begins to read again from client.\n");
1053     client->receive_pending = GNUNET_YES;
1054     GNUNET_CONNECTION_receive (client->connection,
1055                                GNUNET_SERVER_MAX_MESSAGE_SIZE - 1,
1056                                client->idle_timeout, &process_incoming, client);
1057     return;
1058   }
1059   LOG (GNUNET_ERROR_TYPE_DEBUG,
1060        "Server continues processing messages still in the buffer.\n");
1061   GNUNET_SERVER_client_keep (client);
1062   client->receive_pending = GNUNET_NO;
1063   process_mst (client, GNUNET_NO);
1064 }
1065
1066
1067 /**
1068  * This function is called whenever our inbound message tokenizer has
1069  * received a complete message.
1070  *
1071  * @param cls closure (struct GNUNET_SERVER_Handle)
1072  * @param client identification of the client (struct GNUNET_SERVER_Client*)
1073  * @param message the actual message
1074  */
1075 static void
1076 client_message_tokenizer_callback (void *cls, void *client,
1077                                    const struct GNUNET_MessageHeader *message)
1078 {
1079   struct GNUNET_SERVER_Handle *server = cls;
1080   struct GNUNET_SERVER_Client *sender = client;
1081   int ret;
1082
1083   LOG (GNUNET_ERROR_TYPE_DEBUG,
1084        "Tokenizer gives server message of type %u from client\n",
1085        ntohs (message->type));
1086   sender->in_process_client_buffer = GNUNET_YES;
1087   ret = GNUNET_SERVER_inject (server, sender, message);
1088   sender->in_process_client_buffer = GNUNET_NO;
1089   if (GNUNET_OK != ret)
1090     GNUNET_SERVER_client_disconnect (sender);
1091 }
1092
1093
1094 /**
1095  * Add a TCP socket-based connection to the set of handles managed by
1096  * this server.  Use this function for outgoing (P2P) connections that
1097  * we initiated (and where this server should process incoming
1098  * messages).
1099  *
1100  * @param server the server to use
1101  * @param connection the connection to manage (client must
1102  *        stop using this connection from now on)
1103  * @return the client handle (client should call
1104  *         "client_drop" on the return value eventually)
1105  */
1106 struct GNUNET_SERVER_Client *
1107 GNUNET_SERVER_connect_socket (struct GNUNET_SERVER_Handle *server,
1108                               struct GNUNET_CONNECTION_Handle *connection)
1109 {
1110   struct GNUNET_SERVER_Client *client;
1111
1112   client = GNUNET_malloc (sizeof (struct GNUNET_SERVER_Client));
1113   client->connection = connection;
1114   client->reference_count = 1;
1115   client->server = server;
1116   client->last_activity = GNUNET_TIME_absolute_get ();
1117   client->next = server->clients;
1118   client->idle_timeout = server->idle_timeout;
1119   server->clients = client;
1120   if (NULL != server->mst_create)
1121     client->mst =
1122         server->mst_create (server->mst_cls, client);
1123   else
1124     client->mst =
1125         GNUNET_SERVER_mst_create (&client_message_tokenizer_callback, server);
1126   client->receive_pending = GNUNET_YES;
1127   GNUNET_CONNECTION_receive (client->connection,
1128                              GNUNET_SERVER_MAX_MESSAGE_SIZE - 1,
1129                              client->idle_timeout, &process_incoming, client);
1130   return client;
1131 }
1132
1133
1134 /**
1135  * Change the timeout for a particular client.  Decreasing the timeout
1136  * may not go into effect immediately (only after the previous timeout
1137  * times out or activity happens on the socket).
1138  *
1139  * @param client the client to update
1140  * @param timeout new timeout for activities on the socket
1141  */
1142 void
1143 GNUNET_SERVER_client_set_timeout (struct GNUNET_SERVER_Client *client,
1144                                   struct GNUNET_TIME_Relative timeout)
1145 {
1146   client->idle_timeout = timeout;
1147 }
1148
1149
1150 /**
1151  * Notify the server that the given client handle should
1152  * be kept (keeps the connection up if possible, increments
1153  * the internal reference counter).
1154  *
1155  * @param client the client to keep
1156  */
1157 void
1158 GNUNET_SERVER_client_keep (struct GNUNET_SERVER_Client *client)
1159 {
1160   client->reference_count++;
1161 }
1162
1163
1164 /**
1165  * Notify the server that the given client handle is no
1166  * longer required.  Decrements the reference counter.  If
1167  * that counter reaches zero an inactive connection maybe
1168  * closed.
1169  *
1170  * @param client the client to drop
1171  */
1172 void
1173 GNUNET_SERVER_client_drop (struct GNUNET_SERVER_Client *client)
1174 {
1175   GNUNET_assert (client->reference_count > 0);
1176   client->reference_count--;
1177   if ((GNUNET_YES == client->shutdown_now) && (0 == client->reference_count))
1178     GNUNET_SERVER_client_disconnect (client);
1179 }
1180
1181
1182 /**
1183  * Obtain the network address of the other party.
1184  *
1185  * @param client the client to get the address for
1186  * @param addr where to store the address
1187  * @param addrlen where to store the length of the address
1188  * @return GNUNET_OK on success
1189  */
1190 int
1191 GNUNET_SERVER_client_get_address (struct GNUNET_SERVER_Client *client,
1192                                   void **addr, size_t * addrlen)
1193 {
1194   return GNUNET_CONNECTION_get_address (client->connection, addr, addrlen);
1195 }
1196
1197
1198 /**
1199  * Ask the server to notify us whenever a client disconnects.
1200  * This function is called whenever the actual network connection
1201  * is closed; the reference count may be zero or larger than zero
1202  * at this point.
1203  *
1204  * @param server the server manageing the clients
1205  * @param callback function to call on disconnect
1206  * @param callback_cls closure for callback
1207  */
1208 void
1209 GNUNET_SERVER_disconnect_notify (struct GNUNET_SERVER_Handle *server,
1210                                  GNUNET_SERVER_DisconnectCallback callback,
1211                                  void *callback_cls)
1212 {
1213   struct NotifyList *n;
1214
1215   n = GNUNET_malloc (sizeof (struct NotifyList));
1216   n->callback = callback;
1217   n->callback_cls = callback_cls;
1218   GNUNET_CONTAINER_DLL_insert (server->disconnect_notify_list_head,
1219                                server->disconnect_notify_list_tail,
1220                                n);
1221 }
1222
1223
1224 /**
1225  * Ask the server to stop notifying us whenever a client disconnects.
1226  *
1227  * @param server the server manageing the clients
1228  * @param callback function to call on disconnect
1229  * @param callback_cls closure for callback
1230  */
1231 void
1232 GNUNET_SERVER_disconnect_notify_cancel (struct GNUNET_SERVER_Handle *server,
1233                                         GNUNET_SERVER_DisconnectCallback
1234                                         callback, void *callback_cls)
1235 {
1236   struct NotifyList *pos;
1237
1238   for (pos = server->disconnect_notify_list_head; NULL != pos; pos = pos->next)
1239     if ((pos->callback == callback) && (pos->callback_cls == callback_cls))
1240       break;
1241   if (NULL == pos)
1242   {
1243     GNUNET_break (0);
1244     return;
1245   }
1246   GNUNET_CONTAINER_DLL_remove (server->disconnect_notify_list_head,
1247                                server->disconnect_notify_list_tail,
1248                                pos);
1249   GNUNET_free (pos);
1250 }
1251
1252
1253 /**
1254  * Ask the server to disconnect from the given client.
1255  * This is the same as returning GNUNET_SYSERR from a message
1256  * handler, except that it allows dropping of a client even
1257  * when not handling a message from that client.
1258  *
1259  * @param client the client to disconnect from
1260  */
1261 void
1262 GNUNET_SERVER_client_disconnect (struct GNUNET_SERVER_Client *client)
1263 {
1264   struct GNUNET_SERVER_Client *prev;
1265   struct GNUNET_SERVER_Client *pos;
1266   struct GNUNET_SERVER_Handle *server;
1267   struct NotifyList *n;
1268   unsigned int rc;
1269
1270   LOG (GNUNET_ERROR_TYPE_DEBUG,
1271        "Client is being disconnected from the server.\n");
1272   if (GNUNET_SCHEDULER_NO_TASK != client->restart_task)
1273   {
1274     GNUNET_SCHEDULER_cancel (client->restart_task);
1275     client->restart_task = GNUNET_SCHEDULER_NO_TASK;
1276   }
1277   if (GNUNET_SCHEDULER_NO_TASK != client->warn_task)
1278   {
1279     GNUNET_SCHEDULER_cancel (client->warn_task);
1280     client->warn_task = GNUNET_SCHEDULER_NO_TASK;
1281   }
1282   if (GNUNET_YES == client->receive_pending)
1283   {
1284     GNUNET_CONNECTION_receive_cancel (client->connection);
1285     client->receive_pending = GNUNET_NO;
1286   }
1287   server = client->server;
1288   rc = client->reference_count;
1289   if (GNUNET_YES != client->shutdown_now)
1290   {
1291     client->shutdown_now = GNUNET_YES;
1292     prev = NULL;
1293     pos = server->clients;
1294     while ((NULL != pos) && (pos != client))
1295     {
1296       prev = pos;
1297       pos = pos->next;
1298     }
1299     GNUNET_assert (NULL != pos);
1300     if (NULL == prev)
1301       server->clients = pos->next;
1302     else
1303       prev->next = pos->next;
1304     if (GNUNET_SCHEDULER_NO_TASK != client->restart_task)
1305     {
1306       GNUNET_SCHEDULER_cancel (client->restart_task);
1307       client->restart_task = GNUNET_SCHEDULER_NO_TASK;
1308     }
1309     if (GNUNET_SCHEDULER_NO_TASK != client->warn_task)
1310     {
1311       GNUNET_SCHEDULER_cancel (client->warn_task);
1312       client->warn_task = GNUNET_SCHEDULER_NO_TASK;
1313     }
1314     for (n = server->disconnect_notify_list_head; NULL != n; n = n->next)
1315       n->callback (n->callback_cls, client);
1316   }
1317   if (rc > 0)
1318   {
1319     LOG (GNUNET_ERROR_TYPE_DEBUG,
1320          "RC still positive, not destroying everything.\n");
1321     return;
1322   }
1323   if (GNUNET_YES == client->in_process_client_buffer)
1324   {
1325     LOG (GNUNET_ERROR_TYPE_DEBUG,
1326          "Still processing inputs, not destroying everything.\n");
1327     return;
1328   }
1329
1330   if (GNUNET_YES == client->persist)
1331     GNUNET_CONNECTION_persist_ (client->connection);
1332   if (NULL != client->th.cth)
1333     GNUNET_SERVER_notify_transmit_ready_cancel (&client->th);
1334   GNUNET_CONNECTION_destroy (client->connection);
1335
1336   if (NULL != server->mst_destroy)
1337     server->mst_destroy (server->mst_cls, client->mst);
1338   else
1339     GNUNET_SERVER_mst_destroy (client->mst);
1340   GNUNET_free (client);
1341   /* we might be in soft-shutdown, test if we're done */
1342   test_monitor_clients (server);
1343 }
1344
1345
1346 /**
1347  * Disable the "CORK" feature for communication with the given client,
1348  * forcing the OS to immediately flush the buffer on transmission
1349  * instead of potentially buffering multiple messages.
1350  *
1351  * @param client handle to the client
1352  * @return GNUNET_OK on success
1353  */
1354 int
1355 GNUNET_SERVER_client_disable_corking (struct GNUNET_SERVER_Client *client)
1356 {
1357   return GNUNET_CONNECTION_disable_corking (client->connection);
1358 }
1359
1360
1361 /**
1362  * Wrapper for transmission notification that calls the original
1363  * callback and update the last activity time for our connection.
1364  *
1365  * @param cls the 'struct GNUNET_SERVER_Client'
1366  * @param size number of bytes we can transmit
1367  * @param buf where to copy the message
1368  * @return number of bytes actually transmitted
1369  */
1370 static size_t
1371 transmit_ready_callback_wrapper (void *cls, size_t size, void *buf)
1372 {
1373   struct GNUNET_SERVER_Client *client = cls;
1374   GNUNET_CONNECTION_TransmitReadyNotify callback;
1375   size_t ret;
1376
1377   client->th.cth = NULL;
1378   callback = client->th.callback;
1379   client->th.callback = NULL;
1380   ret = callback (client->th.callback_cls, size, buf);
1381   if (ret > 0)
1382     client->last_activity = GNUNET_TIME_absolute_get ();
1383   return ret;
1384 }
1385
1386
1387 /**
1388  * Notify us when the server has enough space to transmit
1389  * a message of the given size to the given client.
1390  *
1391  * @param client client to transmit message to
1392  * @param size requested amount of buffer space
1393  * @param timeout after how long should we give up (and call
1394  *        notify with buf NULL and size 0)?
1395  * @param callback function to call when space is available
1396  * @param callback_cls closure for callback
1397  * @return non-NULL if the notify callback was queued; can be used
1398  *           to cancel the request using
1399  *           GNUNET_SERVER_notify_transmit_ready_cancel.
1400  *         NULL if we are already going to notify someone else (busy)
1401  */
1402 struct GNUNET_SERVER_TransmitHandle *
1403 GNUNET_SERVER_notify_transmit_ready (struct GNUNET_SERVER_Client *client,
1404                                      size_t size,
1405                                      struct GNUNET_TIME_Relative timeout,
1406                                      GNUNET_CONNECTION_TransmitReadyNotify
1407                                      callback, void *callback_cls)
1408 {
1409   if (NULL != client->th.callback)
1410     return NULL;
1411   client->th.callback_cls = callback_cls;
1412   client->th.callback = callback;
1413   client->th.cth = GNUNET_CONNECTION_notify_transmit_ready (client->connection, size,
1414                                                             timeout,
1415                                                             &transmit_ready_callback_wrapper,
1416                                                             client);
1417   return &client->th;
1418 }
1419
1420
1421 /**
1422  * Abort transmission request.
1423  *
1424  * @param th request to abort
1425  */
1426 void
1427 GNUNET_SERVER_notify_transmit_ready_cancel (struct GNUNET_SERVER_TransmitHandle *th)
1428 {
1429   GNUNET_CONNECTION_notify_transmit_ready_cancel (th->cth);
1430   th->cth = NULL;
1431   th->callback = NULL;
1432 }
1433
1434
1435 /**
1436  * Set the persistent flag on this client, used to setup client connection
1437  * to only be killed when the service it's connected to is actually dead.
1438  *
1439  * @param client the client to set the persistent flag on
1440  */
1441 void
1442 GNUNET_SERVER_client_persist_ (struct GNUNET_SERVER_Client *client)
1443 {
1444   client->persist = GNUNET_YES;
1445 }
1446
1447
1448 /**
1449  * Resume receiving from this client, we are done processing the
1450  * current request.  This function must be called from within each
1451  * GNUNET_SERVER_MessageCallback (or its respective continuations).
1452  *
1453  * @param client client we were processing a message of
1454  * @param success GNUNET_OK to keep the connection open and
1455  *                          continue to receive
1456  *                GNUNET_NO to close the connection (normal behavior)
1457  *                GNUNET_SYSERR to close the connection (signal
1458  *                          serious error)
1459  */
1460 void
1461 GNUNET_SERVER_receive_done (struct GNUNET_SERVER_Client *client, int success)
1462 {
1463   if (NULL == client)
1464     return;
1465   GNUNET_assert (client->suspended > 0);
1466   client->suspended--;
1467   if (GNUNET_OK != success)
1468   {
1469     LOG (GNUNET_ERROR_TYPE_DEBUG,
1470          "GNUNET_SERVER_receive_done called with failure indication\n");
1471     GNUNET_SERVER_client_disconnect (client);
1472     return;
1473   }
1474   if (client->suspended > 0)
1475   {
1476     LOG (GNUNET_ERROR_TYPE_DEBUG,
1477          "GNUNET_SERVER_receive_done called, but more clients pending\n");
1478     return;
1479   }
1480   if (GNUNET_SCHEDULER_NO_TASK != client->warn_task)
1481   {
1482     GNUNET_SCHEDULER_cancel (client->warn_task);
1483     client->warn_task = GNUNET_SCHEDULER_NO_TASK;
1484   }
1485   if (GNUNET_YES == client->in_process_client_buffer)
1486   {
1487     LOG (GNUNET_ERROR_TYPE_DEBUG,
1488          "GNUNET_SERVER_receive_done called while still in processing loop\n");
1489     return;
1490   }
1491   if ((NULL == client->server) || (GNUNET_YES == client->shutdown_now))
1492   {
1493     GNUNET_SERVER_client_disconnect (client);
1494     return;
1495   }
1496   LOG (GNUNET_ERROR_TYPE_DEBUG,
1497        "GNUNET_SERVER_receive_done causes restart in reading from the socket\n");
1498   GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == client->restart_task);
1499   client->restart_task = GNUNET_SCHEDULER_add_now (&restart_processing, client);
1500 }
1501
1502
1503 /* end of server.c */