src/util/server.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2009 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  * TODO:
  27  * - fix inefficient memmove in message processing
  28  */
  29
  30 #include "platform.h"
  31 #include "gnunet_common.h"
  32 #include "gnunet_connection_lib.h"
  33 #include "gnunet_scheduler_lib.h"
  34 #include "gnunet_server_lib.h"
  35 #include "gnunet_time_lib.h"
  36 #include "gnunet_disk_lib.h"
  37
  38 #define DEBUG_SERVER GNUNET_NO
  39
  40 /**
  41  * List of arrays of message handlers.
  42  */
  43 struct HandlerList
  44 {
  45   /**
  46    * This is a linked list.
  47    */
  48   struct HandlerList *next;
  49
  50   /**
  51    * NULL-terminated array of handlers.
  52    */
  53   const struct GNUNET_SERVER_MessageHandler *handlers;
  54 };
  55
  56
  57 /**
  58  * List of arrays of message handlers.
  59  */
  60 struct NotifyList
  61 {
  62   /**
  63    * This is a linked list.
  64    */
  65   struct NotifyList *next;
  66
  67   /**
  68    * Function to call.
  69    */
  70   GNUNET_SERVER_DisconnectCallback callback;
  71
  72   /**
  73    * Closure for callback.
  74    */
  75   void *callback_cls;
  76 };
  77
  78
  79 /**
  80  * @brief handle for a server
  81  */
  82 struct GNUNET_SERVER_Handle
  83 {
  84   /**
  85    * My scheduler.
  86    */
  87   struct GNUNET_SCHEDULER_Handle *sched;
  88
  89   /**
  90    * List of handlers for incoming messages.
  91    */
  92   struct HandlerList *handlers;
  93
  94   /**
  95    * List of our current clients.
  96    */
  97   struct GNUNET_SERVER_Client *clients;
  98
  99   /**
 100    * Linked list of functions to call on disconnects by clients.
 101    */
 102   struct NotifyList *disconnect_notify_list;
 103
 104   /**
 105    * Function to call for access control.
 106    */
 107   GNUNET_NETWORK_AccessCheck access;
 108
 109   /**
 110    * Closure for access.
 111    */
 112   void *access_cls;
 113
 114   /**
 115    * After how long should an idle connection time
 116    * out (on write).
 117    */
 118   struct GNUNET_TIME_Relative idle_timeout;
 119
 120   /**
 121    * maximum write buffer size for accepted sockets
 122    */
 123   size_t maxbuf;
 124
 125   /**
 126    * Pipe used to signal shutdown of the server.
 127    */
 128   struct GNUNET_DISK_PipeHandle *shutpipe;
 129
 130   /**
 131    * Socket used to listen for new connections.  Set to
 132    * "-1" by GNUNET_SERVER_destroy to initiate shutdown.
 133    */
 134   struct GNUNET_NETWORK_Descriptor *listen_socket;
 135
 136   /**
 137    * Set to GNUNET_YES if we are shutting down.
 138    */
 139   int do_shutdown;
 140
 141   /**
 142    * Do we ignore messages of types that we do not
 143    * understand or do we require that a handler
 144    * is found (and if not kill the connection)?
 145    */
 146   int require_found;
 147
 148 };
 149
 150
 151 /**
 152  * @brief handle for a client of the server
 153  */
 154 struct GNUNET_SERVER_Client
 155 {
 156
 157   /**
 158    * Size of the buffer for incoming data.  Should be
 159    * first so we get nice alignment.
 160    */
 161   char incoming_buffer[GNUNET_SERVER_MAX_MESSAGE_SIZE];
 162
 163   /**
 164    * This is a linked list.
 165    */
 166   struct GNUNET_SERVER_Client *next;
 167
 168   /**
 169    * Server that this client belongs to.
 170    */
 171   struct GNUNET_SERVER_Handle *server;
 172
 173   /**
 174    * Client closure for callbacks.
 175    */
 176   void *client_closure;
 177
 178   /**
 179    * Callback to receive from client.
 180    */
 181   GNUNET_SERVER_ReceiveCallback receive;
 182
 183   /**
 184    * Callback to cancel receive from client.
 185    */
 186   GNUNET_SERVER_ReceiveCancelCallback receive_cancel;
 187
 188   /**
 189    * Callback to ask about transmit-ready notification.
 190    */
 191   GNUNET_SERVER_TransmitReadyCallback notify_transmit_ready;
 192
 193    /**
 194    * Callback to ask about transmit-ready notification.
 195    */
 196   GNUNET_SERVER_TransmitReadyCancelCallback notify_transmit_ready_cancel;
 197
 198   /**
 199    * Callback to check if client is still valid.
 200    */
 201   GNUNET_SERVER_CheckCallback check;
 202
 203   /**
 204    * Callback to destroy client.
 205    */
 206   GNUNET_SERVER_DestroyCallback destroy;
 207
 208   /**
 209    * Side-buffer for incoming data used when processing
 210    * is suspended.
 211    */
 212   char *side_buf;
 213
 214   /**
 215    * Number of bytes in the side buffer.
 216    */
 217   size_t side_buf_size;
 218
 219   /**
 220    * Last activity on this socket (used to time it out
 221    * if reference_count == 0).
 222    */
 223   struct GNUNET_TIME_Absolute last_activity;
 224
 225   /**
 226    * Current task identifier for the receive call
 227    * (or GNUNET_SCHEDULER_NO_TASK for none).
 228    */
 229   GNUNET_SCHEDULER_TaskIdentifier my_receive;
 230
 231   /**
 232    * How many bytes in the "incoming_buffer" are currently
 233    * valid? (starting at offset 0).
 234    */
 235   size_t receive_pos;
 236
 237   /**
 238    * Number of external entities with a reference to
 239    * this client object.
 240    */
 241   unsigned int reference_count;
 242
 243   /**
 244    * Was processing if incoming messages suspended while
 245    * we were still processing data already received?
 246    * This is a counter saying how often processing was
 247    * suspended (once per handler invoked).
 248    */
 249   unsigned int suspended;
 250
 251   /**
 252    * Are we currently in the "process_client_buffer" function (and
 253    * will hence restart the receive job on exit if suspended == 0 once
 254    * we are done?).  If this is set, then "receive_done" will
 255    * essentially only decrement suspended; if this is not set, then
 256    * "receive_done" may need to restart the receive process (either
 257    * from the side-buffer or via select/recv).
 258    */
 259   int in_process_client_buffer;
 260
 261   /**
 262    * We're about to close down this client due to some serious
 263    * error.
 264    */
 265   int shutdown_now;
 266
 267 };
 268
 269
 270 /**
 271  * Server has been asked to shutdown, free resources.
 272  */
 273 static void
 274 destroy_server (struct GNUNET_SERVER_Handle *server)
 275 {
 276   struct GNUNET_SERVER_Client *pos;
 277   struct HandlerList *hpos;
 278   struct NotifyList *npos;
 279
 280 #if DEBUG_SERVER
 281   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 282               "Server shutting down.\n");
 283 #endif
 284   GNUNET_assert (server->listen_socket == NULL);
 285   GNUNET_break (GNUNET_YES == GNUNET_DISK_pipe_close (server->shutpipe));
 286   while (server->clients != NULL)
 287     {
 288       pos = server->clients;
 289       server->clients = pos->next;
 290       pos->server = NULL;
 291     }
 292   while (NULL != (hpos = server->handlers))
 293     {
 294       server->handlers = hpos->next;
 295       GNUNET_free (hpos);
 296     }
 297   while (NULL != (npos = server->disconnect_notify_list))
 298     {
 299       server->disconnect_notify_list = npos->next;
 300       GNUNET_free (npos);
 301     }
 302   GNUNET_free (server);
 303 }
 304
 305
 306 /**
 307  * Scheduler says our listen socket is ready.
 308  * Process it!
 309  */
 310 static void
 311 process_listen_socket (void *cls,
 312                        const struct GNUNET_SCHEDULER_TaskContext *tc)
 313 {
 314   struct GNUNET_SERVER_Handle *server = cls;
 315   struct GNUNET_NETWORK_ConnectionHandle *sock;
 316   struct GNUNET_SERVER_Client *client;
 317   struct GNUNET_NETWORK_FDSet *r;
 318   const struct GNUNET_DISK_FileHandle *shutpipe;
 319
 320   if ((server->do_shutdown) ||
 321       ((tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) != 0))
 322     {
 323       /* shutdown was initiated */
 324       GNUNET_assert (server->listen_socket != NULL);
 325       GNUNET_break (0 == GNUNET_NETWORK_socket_close (server->listen_socket));
 326       server->listen_socket = NULL;
 327       if (server->do_shutdown)
 328         destroy_server (server);
 329       return;
 330     }
 331   shutpipe = GNUNET_DISK_pipe_handle (server->shutpipe, 0);
 332   GNUNET_assert (GNUNET_NETWORK_fdset_isset (tc->read_ready, server->listen_socket));
 333   GNUNET_assert (!GNUNET_NETWORK_fdset_handle_isset (tc->read_ready, shutpipe));
 334   sock = GNUNET_NETWORK_connection_create_from_accept (tc->sched,
 335                                                    server->access,
 336                                                    server->access_cls,
 337                                                    server->listen_socket,
 338                                                    server->maxbuf);
 339   if (sock != NULL)
 340     {
 341 #if DEBUG_SERVER
 342       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 343                   "Server accepted incoming connection.\n");
 344 #endif
 345       client = GNUNET_SERVER_connect_socket (server, sock);
 346       /* decrement reference count, we don't keep "client" alive */
 347       GNUNET_SERVER_client_drop (client);
 348     }
 349   /* listen for more! */
 350   r = GNUNET_NETWORK_fdset_create ();
 351   GNUNET_NETWORK_fdset_set (r, server->listen_socket);
 352   GNUNET_NETWORK_fdset_handle_set (r, shutpipe);
 353   GNUNET_SCHEDULER_add_select (server->sched,
 354                                GNUNET_YES,
 355                                GNUNET_SCHEDULER_PRIORITY_HIGH,
 356                                GNUNET_SCHEDULER_NO_TASK,
 357                                GNUNET_TIME_UNIT_FOREVER_REL,
 358                                r, NULL,
 359                                &process_listen_socket, server);
 360   GNUNET_NETWORK_fdset_destroy (r);
 361 }
 362
 363
 364 /**
 365  * Create and initialize a listen socket for the server.
 366  *
 367  * @return NULL on error, otherwise the listen socket
 368  */
 369 static struct GNUNET_NETWORK_Descriptor *
 370 open_listen_socket (const struct sockaddr *serverAddr, socklen_t socklen)
 371 {
 372   const static int on = 1;
 373   struct GNUNET_NETWORK_Descriptor *sock;
 374   uint16_t port;
 375
 376   switch (serverAddr->sa_family)
 377     {
 378     case AF_INET:
 379       port = ntohs (((const struct sockaddr_in *) serverAddr)->sin_port);
 380       break;
 381     case AF_INET6:
 382       port = ntohs (((const struct sockaddr_in6 *) serverAddr)->sin6_port);
 383       break;
 384     default:
 385       GNUNET_break (0);
 386       return NULL;
 387     }
 388   sock = GNUNET_NETWORK_socket_socket (serverAddr->sa_family, SOCK_STREAM, 0);
 389   if (NULL == sock)
 390     {
 391       GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR, "socket");
 392       return NULL;
 393     }
 394 #ifndef MINGW
 395   if (GNUNET_OK != GNUNET_NETWORK_socket_set_inheritable (sock))
 396     GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 397                          "fcntl");
 398 #endif
 399   if (GNUNET_NETWORK_socket_setsockopt (sock, SOL_SOCKET, SO_REUSEADDR, &on, sizeof (on)) < 0)
 400     GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 401                          "setsockopt");
 402   /* bind the socket */
 403   if (GNUNET_NETWORK_socket_bind (sock, serverAddr, socklen) < 0)
 404     {
 405       GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR, "bind");
 406       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 407                   _
 408                   ("`%s' failed for port %d. Is the service already running?\n"),
 409                   "bind", port);
 410       GNUNET_break (0 == GNUNET_NETWORK_socket_close (sock));
 411       return NULL;
 412     }
 413   if (0 != GNUNET_NETWORK_socket_listen (sock, 5))
 414     {
 415       GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR, "listen");
 416       GNUNET_break (0 == GNUNET_NETWORK_socket_close (sock));
 417       return NULL;
 418     }
 419 #if DEBUG_SERVER
 420       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 421                   "Server starts to listen on port %u.\n",
 422                   port);
 423 #endif
 424   return sock;
 425 }
 426
 427
 428 /**
 429  * Create a new server.
 430  *
 431  * @param sched scheduler to use
 432  * @param access function for access control
 433  * @param access_cls closure for access
 434  * @param serverAddr address to listen on (including port), use NULL
 435  *        for internal server (no listening)
 436  * @param socklen length of serverAddr
 437  * @param maxbuf maximum write buffer size for accepted sockets
 438  * @param idle_timeout after how long should we timeout idle connections?
 439  * @param require_found if YES, connections sending messages of unknown type
 440  *        will be closed
 441  * @return handle for the new server, NULL on error
 442  *         (typically, "port" already in use)
 443  */
 444 struct GNUNET_SERVER_Handle *
 445 GNUNET_SERVER_create (struct GNUNET_SCHEDULER_Handle *sched,
 446                       GNUNET_NETWORK_AccessCheck access,
 447                       void *access_cls,
 448                       const struct sockaddr *serverAddr,
 449                       socklen_t socklen,
 450                       size_t maxbuf,
 451                       struct GNUNET_TIME_Relative
 452                       idle_timeout, int require_found)
 453 {
 454   struct GNUNET_SERVER_Handle *ret;
 455   struct GNUNET_NETWORK_Descriptor *lsock;
 456   struct GNUNET_NETWORK_FDSet *r;
 457
 458   lsock = NULL; // FIXME NILS: this was -2, does that have a special meaning?
 459   if (serverAddr != NULL)
 460     {
 461       lsock = open_listen_socket (serverAddr, socklen);
 462       if (lsock == NULL)
 463         return NULL;
 464     }
 465   ret = GNUNET_malloc (sizeof (struct GNUNET_SERVER_Handle));
 466   ret->shutpipe = GNUNET_malloc (sizeof (struct GNUNET_DISK_FileDescriptor *[2]));
 467   if ((ret->shutpipe = GNUNET_DISK_pipe (GNUNET_NO)) == NULL)
 468     {
 469       GNUNET_break (0 == GNUNET_NETWORK_socket_close (lsock));
 470       GNUNET_free (ret->shutpipe);
 471       GNUNET_free (ret);
 472       return NULL;
 473     }
 474   ret->sched = sched;
 475   ret->maxbuf = maxbuf;
 476   ret->idle_timeout = idle_timeout;
 477   ret->listen_socket = lsock;
 478   ret->access = access;
 479   ret->access_cls = access_cls;
 480   ret->require_found = require_found;
 481   if (lsock != NULL)
 482     {
 483       r = GNUNET_NETWORK_fdset_create ();
 484       GNUNET_NETWORK_fdset_set (r, ret->listen_socket);
 485       GNUNET_NETWORK_fdset_handle_set (r, GNUNET_DISK_pipe_handle (ret->shutpipe, 0));
 486       GNUNET_SCHEDULER_add_select (sched,
 487                                    GNUNET_YES,
 488                                    GNUNET_SCHEDULER_PRIORITY_HIGH,
 489                                    GNUNET_SCHEDULER_NO_TASK,
 490                                    GNUNET_TIME_UNIT_FOREVER_REL,
 491                                    r,
 492                                    NULL, &process_listen_socket, ret);
 493       GNUNET_NETWORK_fdset_destroy (r);
 494     }
 495   return ret;
 496 }
 497
 498
 499 /**
 500  * Free resources held by this server.
 501  */
 502 void
 503 GNUNET_SERVER_destroy (struct GNUNET_SERVER_Handle *s)
 504 {
 505   static char c;
 506
 507   GNUNET_assert (s->do_shutdown == GNUNET_NO);
 508   s->do_shutdown = GNUNET_YES;
 509   if (s->listen_socket == NULL)
 510     destroy_server (s);
 511   else
 512     GNUNET_break (1 == GNUNET_DISK_file_write (GNUNET_DISK_pipe_handle (s->shutpipe, 1), &c, 1));
 513 }
 514
 515
 516 /**
 517  * Add additional handlers to an existing server.
 518  *
 519  * @param server the server to add handlers to
 520  * @param handlers array of message handlers for
 521  *        incoming messages; the last entry must
 522  *        have "NULL" for the "callback"; multiple
 523  *        entries for the same type are allowed,
 524  *        they will be called in order of occurence.
 525  *        These handlers can be removed later;
 526  *        the handlers array must exist until removed
 527  *        (or server is destroyed).
 528  */
 529 void
 530 GNUNET_SERVER_add_handlers (struct GNUNET_SERVER_Handle *server,
 531                             const struct GNUNET_SERVER_MessageHandler
 532                             *handlers)
 533 {
 534   struct HandlerList *p;
 535
 536   p = GNUNET_malloc (sizeof (struct HandlerList));
 537   p->handlers = handlers;
 538   p->next = server->handlers;
 539   server->handlers = p;
 540 }
 541
 542
 543 /**
 544  * Inject a message into the server, pretend it came
 545  * from the specified client.  Delivery of the message
 546  * will happen instantly (if a handler is installed;
 547  * otherwise the call does nothing).
 548  *
 549  * @param server the server receiving the message
 550  * @param sender the "pretended" sender of the message
 551  *        can be NULL!
 552  * @param message message to transmit
 553  * @return GNUNET_OK if the message was OK and the
 554  *                   connection can stay open
 555  *         GNUNET_SYSERR if the connection to the
 556  *         client should be shut down
 557  */
 558 int
 559 GNUNET_SERVER_inject (struct GNUNET_SERVER_Handle *server,
 560                       struct GNUNET_SERVER_Client *sender,
 561                       const struct GNUNET_MessageHeader *message)
 562 {
 563   struct HandlerList *pos;
 564   const struct GNUNET_SERVER_MessageHandler *mh;
 565   unsigned int i;
 566   uint16_t type;
 567   uint16_t size;
 568   int found;
 569
 570   type = ntohs (message->type);
 571   size = ntohs (message->size);
 572 #if DEBUG_SERVER
 573   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 574               "Server schedules transmission of %u-byte message of type %u to client.\n",
 575               size,
 576               type);
 577 #endif
 578   pos = server->handlers;
 579   found = GNUNET_NO;
 580   while (pos != NULL)
 581     {
 582       i = 0;
 583       while (pos->handlers[i].callback != NULL)
 584         {
 585           mh = &pos->handlers[i];
 586           if (mh->type == type)
 587             {
 588               if ((mh->expected_size != 0) && (mh->expected_size != size))
 589                 {
 590                   GNUNET_break_op (0);
 591                   return GNUNET_SYSERR;
 592                 }
 593               if (sender != NULL)
 594                 sender->suspended++;
 595               mh->callback (mh->callback_cls, sender, message);
 596               found = GNUNET_YES;
 597             }
 598           i++;
 599         }
 600       pos = pos->next;
 601     }
 602   if (found == GNUNET_NO)
 603     {
 604       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
 605                   _("Received message of unknown type %d\n"), type);
 606       if (server->require_found == GNUNET_YES)
 607         return GNUNET_SYSERR;
 608     }
 609   return GNUNET_OK;
 610 }
 611
 612
 613 /**
 614  * We're finished with this client and especially its input
 615  * processing.  If the RC is zero, free all resources otherwise wait
 616  * until RC hits zero to do so.
 617  */
 618 static void
 619 shutdown_incoming_processing (struct GNUNET_SERVER_Client *client)
 620 {
 621   struct GNUNET_SERVER_Client *prev;
 622   struct GNUNET_SERVER_Client *pos;
 623   struct GNUNET_SERVER_Handle *server;
 624   struct NotifyList *n;
 625   unsigned int rc;
 626
 627   GNUNET_assert (client->my_receive == GNUNET_SCHEDULER_NO_TASK);
 628   rc = client->reference_count;
 629   if (client->server != NULL)
 630     {
 631       server = client->server;
 632       client->server = NULL;
 633       prev = NULL;
 634       pos = server->clients;
 635       while ((pos != NULL) && (pos != client))
 636         {
 637           prev = pos;
 638           pos = pos->next;
 639         }
 640       GNUNET_assert (pos != NULL);
 641       if (prev == NULL)
 642         server->clients = pos->next;
 643       else
 644         prev->next = pos->next;
 645       n = server->disconnect_notify_list;
 646       while (n != NULL)
 647         {
 648           n->callback (n->callback_cls, client);
 649           n = n->next;
 650         }
 651     }
 652   /* wait for RC to hit zero, then free */
 653   if (rc > 0)
 654     return;
 655   client->destroy (client->client_closure);
 656   GNUNET_free (client);
 657 }
 658
 659
 660 /**
 661  * Go over the contents of the client buffer; as long as full messages
 662  * are available, pass them on for processing.  Update the buffer
 663  * accordingly.  Handles fatal errors by shutting down the connection.
 664  *
 665  * @param client identifies which client receive buffer to process
 666  */
 667 static void
 668 process_client_buffer (struct GNUNET_SERVER_Client *client)
 669 {
 670   struct GNUNET_SERVER_Handle *server;
 671   const struct GNUNET_MessageHeader *hdr;
 672   size_t msize;
 673
 674   client->in_process_client_buffer = GNUNET_YES;
 675   server = client->server;
 676 #if DEBUG_SERVER
 677   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 678               "Private buffer contains %u bytes; client is %s and we are %s\n",
 679               client->receive_pos,
 680               client->suspended ? "suspended" : "up",
 681               client->shutdown_now ? "in shutdown" : "running");
 682 #endif
 683   while ((client->receive_pos >= sizeof (struct GNUNET_MessageHeader)) &&
 684          (0 == client->suspended) && (GNUNET_YES != client->shutdown_now))
 685     {
 686       hdr = (const struct GNUNET_MessageHeader *) &client->incoming_buffer;
 687       msize = ntohs (hdr->size);
 688       if (msize > client->receive_pos)
 689         {
 690 #if DEBUG_SERVER
 691           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 692                       "Total message size is %u, we only have %u bytes; need more data\n",
 693                       msize,
 694                       client->receive_pos);
 695 #endif
 696           break;
 697         }
 698 #if DEBUG_SERVER
 699       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 700                   "Passing %u bytes to callback for processing\n",
 701                   msize);
 702 #endif
 703       if ((msize < sizeof (struct GNUNET_MessageHeader)) ||
 704           (GNUNET_OK != GNUNET_SERVER_inject (server, client, hdr)))
 705         {
 706           client->in_process_client_buffer = GNUNET_NO;
 707           shutdown_incoming_processing (client);
 708           return;
 709         }
 710       /* FIXME: this is highly inefficient; we should
 711          try to avoid this if the new base address is
 712          already nicely aligned.  See old handler code... */
 713       memmove (client->incoming_buffer,
 714                &client->incoming_buffer[msize], client->receive_pos - msize);
 715       client->receive_pos -= msize;
 716     }
 717   client->in_process_client_buffer = GNUNET_NO;
 718   if (GNUNET_YES == client->shutdown_now)
 719     shutdown_incoming_processing (client);
 720 }
 721
 722
 723 /**
 724  * We are receiving an incoming message.  Process it.
 725  *
 726  * @param cls our closure (handle for the client)
 727  * @param buf buffer with data received from network
 728  * @param available number of bytes available in buf
 729  * @param addr address of the sender
 730  * @param addrlen length of addr
 731  * @param errCode code indicating errors receiving, 0 for success
 732  */
 733 static void
 734 process_incoming (void *cls,
 735                   const void *buf,
 736                   size_t available,
 737                   const struct sockaddr *addr,
 738                   socklen_t addrlen,
 739                   int errCode)
 740 {
 741   struct GNUNET_SERVER_Client *client = cls;
 742   struct GNUNET_SERVER_Handle *server = client->server;
 743   const char *cbuf = buf;
 744   size_t maxcpy;
 745
 746   client->my_receive = GNUNET_SCHEDULER_NO_TASK;
 747   if ((buf == NULL) ||
 748       (available == 0) ||
 749       (errCode != 0) ||
 750       (server == NULL) ||
 751       (client->shutdown_now == GNUNET_YES) ||
 752       (GNUNET_YES != client->check (client->client_closure)))
 753     {
 754       /* other side closed connection, error connecting, etc. */
 755       shutdown_incoming_processing (client);
 756       return;
 757     }
 758 #if DEBUG_SERVER
 759   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 760               "Server receives %u bytes from `%s'.\n",
 761               available,
 762               GNUNET_a2s(addr, addrlen));
 763 #endif
 764   GNUNET_SERVER_client_keep (client);
 765   client->last_activity = GNUNET_TIME_absolute_get ();
 766   /* process data (if available) */
 767   while (available > 0)
 768     {
 769       maxcpy = available;
 770       if (maxcpy > sizeof (client->incoming_buffer) - client->receive_pos)
 771         maxcpy = sizeof (client->incoming_buffer) - client->receive_pos;
 772 #if DEBUG_SERVER
 773       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 774                   "Can copy %u bytes to private buffer\n",
 775                   maxcpy);
 776 #endif
 777       memcpy (&client->incoming_buffer[client->receive_pos], cbuf, maxcpy);
 778       client->receive_pos += maxcpy;
 779       cbuf += maxcpy;
 780       available -= maxcpy;
 781       if (0 < client->suspended)
 782         {
 783           if (available > 0)
 784             {
 785 #if DEBUG_SERVER
 786               GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 787                           "Client has suspended processing; copying %u bytes to side buffer to be used later.\n",
 788                           available);
 789 #endif
 790               GNUNET_assert (client->side_buf_size == 0);
 791               GNUNET_assert (client->side_buf == NULL);
 792               client->side_buf_size = available;
 793               client->side_buf = GNUNET_malloc (available);
 794               memcpy (client->side_buf, cbuf, available);
 795               available = 0;
 796             }
 797           break;                /* do not run next client iteration! */
 798         }
 799 #if DEBUG_SERVER
 800       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 801                   "Now processing messages in private buffer\n");
 802 #endif
 803       process_client_buffer (client);
 804     }
 805   GNUNET_assert (available == 0);
 806   if ((client->suspended == 0) &&
 807       (GNUNET_YES != client->shutdown_now) && (client->server != NULL))
 808     {
 809       /* Finally, keep receiving! */
 810       client->my_receive = client->receive (client->client_closure,
 811                                             GNUNET_SERVER_MAX_MESSAGE_SIZE,
 812                                             server->idle_timeout,
 813                                             &process_incoming, client);
 814     }
 815   if (GNUNET_YES == client->shutdown_now)
 816     shutdown_incoming_processing (client);
 817   GNUNET_SERVER_client_drop (client);
 818 }
 819
 820
 821 /**
 822  * FIXME: document.
 823  */
 824 static void
 825 restart_processing (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 826 {
 827   struct GNUNET_SERVER_Client *client = cls;
 828
 829   process_client_buffer (client);
 830   if (0 == client->suspended)
 831     client->my_receive = client->receive (client->client_closure,
 832                                           GNUNET_SERVER_MAX_MESSAGE_SIZE,
 833                                           client->server->idle_timeout,
 834                                           &process_incoming, client);
 835 }
 836
 837
 838 /**
 839  * Add a client to the set of our clients and
 840  * start receiving.
 841  */
 842 static void
 843 add_client (struct GNUNET_SERVER_Handle *server,
 844             struct GNUNET_SERVER_Client *client)
 845 {
 846   client->server = server;
 847   client->last_activity = GNUNET_TIME_absolute_get ();
 848   client->next = server->clients;
 849   server->clients = client;
 850   client->my_receive = client->receive (client->client_closure,
 851                                         GNUNET_SERVER_MAX_MESSAGE_SIZE,
 852                                         server->idle_timeout,
 853                                         &process_incoming, client);
 854 }
 855
 856
 857 /**
 858  * Create a request for receiving data from a socket.
 859  *
 860  * @param cls identifies the socket to receive from
 861  * @param max how much data to read at most
 862  * @param timeout when should this operation time out
 863  * @param receiver function to call for processing
 864  * @param receiver_cls closure for receiver
 865  * @return task identifier that can be used to cancel the operation
 866  */
 867 static GNUNET_SCHEDULER_TaskIdentifier
 868 sock_receive (void *cls,
 869               size_t max,
 870               struct GNUNET_TIME_Relative timeout,
 871               GNUNET_NETWORK_Receiver receiver, void *receiver_cls)
 872 {
 873   return GNUNET_NETWORK_connection_receive (cls, max, timeout, receiver, receiver_cls);
 874 }
 875
 876
 877 /**
 878  * Wrapper to cancel receiving from a socket.
 879  *
 880  * @param cls handle to the GNUNET_NETWORK_ConnectionHandle to cancel
 881  * @param tc task ID that was returned by GNUNET_NETWORK_connection_receive
 882  */
 883 static void
 884 sock_receive_cancel (void *cls, GNUNET_SCHEDULER_TaskIdentifier ti)
 885 {
 886   GNUNET_NETWORK_connection_receive_cancel (cls, ti);
 887 }
 888
 889
 890 /**
 891  * FIXME: document.
 892  */
 893 static void *
 894 sock_notify_transmit_ready (void *cls,
 895                             size_t size,
 896                             struct GNUNET_TIME_Relative timeout,
 897                             GNUNET_NETWORK_TransmitReadyNotify notify,
 898                             void *notify_cls)
 899 {
 900   return GNUNET_NETWORK_connection_notify_transmit_ready (cls, size, timeout, notify,
 901                                                notify_cls);
 902 }
 903
 904
 905 /**
 906  * FIXME: document.
 907  */
 908 static void
 909 sock_notify_transmit_ready_cancel (void *cls, void *h)
 910 {
 911   GNUNET_NETWORK_connection_notify_transmit_ready_cancel (h);
 912 }
 913
 914
 915 /**
 916  * Check if socket is still valid (no fatal errors have happened so far).
 917  *
 918  * @param cls the socket
 919  * @return GNUNET_YES if valid, GNUNET_NO otherwise
 920  */
 921 static int
 922 sock_check (void *cls)
 923 {
 924   return GNUNET_NETWORK_connection_check (cls);
 925 }
 926
 927
 928 /**
 929  * Destroy this socket (free resources).
 930  *
 931  * @param cls the socket
 932  */
 933 static void
 934 sock_destroy (void *cls)
 935 {
 936   GNUNET_NETWORK_connection_destroy (cls);
 937 }
 938
 939
 940 /**
 941  * Add a TCP socket-based connection to the set of handles managed by
 942  * this server.  Use this function for outgoing (P2P) connections that
 943  * we initiated (and where this server should process incoming
 944  * messages).
 945  *
 946  * @param server the server to use
 947  * @param connection the connection to manage (client must
 948  *        stop using this connection from now on)
 949  * @return the client handle (client should call
 950  *         "client_drop" on the return value eventually)
 951  */
 952 struct GNUNET_SERVER_Client *
 953 GNUNET_SERVER_connect_socket (struct
 954                               GNUNET_SERVER_Handle
 955                               *server,
 956                               struct GNUNET_NETWORK_ConnectionHandle *connection)
 957 {
 958   struct GNUNET_SERVER_Client *client;
 959
 960   client = GNUNET_malloc (sizeof (struct GNUNET_SERVER_Client));
 961   client->client_closure = connection;
 962   client->receive = &sock_receive;
 963   client->receive_cancel = &sock_receive_cancel;
 964   client->notify_transmit_ready = &sock_notify_transmit_ready;
 965   client->notify_transmit_ready_cancel = &sock_notify_transmit_ready_cancel;
 966   client->check = &sock_check;
 967   client->destroy = &sock_destroy;
 968   client->reference_count = 1;
 969   add_client (server, client);
 970   return client;
 971 }
 972
 973
 974 /**
 975  * Add an arbitrary connection to the set of handles managed by this
 976  * server.  This can be used if a sending and receiving does not
 977  * really go over the network (internal transmission) or for servers
 978  * using UDP.
 979  *
 980  * @param server the server to use
 981  * @param chandle opaque handle for the connection
 982  * @param creceive receive function for the connection
 983  * @param ccancel cancel receive function for the connection
 984  * @param cnotify transmit notification function for the connection
 985  * @param cnotify_cancel transmit notification cancellation function for the connection
 986  * @param ccheck function to test if the connection is still up
 987  * @param cdestroy function to close and free the connection
 988  * @return the client handle (client should call
 989  *         "client_drop" on the return value eventually)
 990  */
 991 struct GNUNET_SERVER_Client *
 992 GNUNET_SERVER_connect_callback (struct
 993                                 GNUNET_SERVER_Handle
 994                                 *server,
 995                                 void *chandle,
 996                                 GNUNET_SERVER_ReceiveCallback
 997                                 creceive,
 998                                 GNUNET_SERVER_ReceiveCancelCallback
 999                                 ccancel,
1000                                 GNUNET_SERVER_TransmitReadyCallback
1001                                 cnotify,
1002                                 GNUNET_SERVER_TransmitReadyCancelCallback
1003                                 cnotify_cancel,
1004                                 GNUNET_SERVER_CheckCallback
1005                                 ccheck,
1006                                 GNUNET_SERVER_DestroyCallback cdestroy)
1007 {
1008   struct GNUNET_SERVER_Client *client;
1009
1010   client = GNUNET_malloc (sizeof (struct GNUNET_SERVER_Client));
1011   client->client_closure = chandle;
1012   client->receive = creceive;
1013   client->receive_cancel = ccancel;
1014   client->notify_transmit_ready = cnotify;
1015   client->notify_transmit_ready_cancel = cnotify_cancel;
1016   client->check = ccheck;
1017   client->destroy = cdestroy;
1018   client->reference_count = 1;
1019   add_client (server, client);
1020   return client;
1021 }
1022
1023
1024 /**
1025  * Notify the server that the given client handle should
1026  * be kept (keeps the connection up if possible, increments
1027  * the internal reference counter).
1028  *
1029  * @param client the client to keep
1030  */
1031 void
1032 GNUNET_SERVER_client_keep (struct GNUNET_SERVER_Client *client)
1033 {
1034   client->reference_count++;
1035 }
1036
1037
1038 /**
1039  * Notify the server that the given client handle is no
1040  * longer required.  Decrements the reference counter.  If
1041  * that counter reaches zero an inactive connection maybe
1042  * closed.
1043  *
1044  * @param client the client to drop
1045  */
1046 void
1047 GNUNET_SERVER_client_drop (struct GNUNET_SERVER_Client *client)
1048 {
1049   GNUNET_assert (client->reference_count > 0);
1050   client->reference_count--;
1051   if ((client->server == NULL) && (client->reference_count == 0))
1052     shutdown_incoming_processing (client);
1053 }
1054
1055
1056 /**
1057  * Obtain the network address of the other party.
1058  *
1059  * @param client the client to get the address for
1060  * @param addr where to store the address
1061  * @param addrlen where to store the length of the address
1062  * @return GNUNET_OK on success
1063  */
1064 int
1065 GNUNET_SERVER_client_get_address (struct GNUNET_SERVER_Client *client,
1066                                   void **addr, size_t * addrlen)
1067 {
1068   if (client->receive != &sock_receive)
1069     return GNUNET_SYSERR;       /* not a network client */
1070   return GNUNET_NETWORK_connection_get_address (client->client_closure,
1071                                             addr, addrlen);
1072 }
1073
1074
1075 /**
1076  * Ask the server to notify us whenever a client disconnects.
1077  * This function is called whenever the actual network connection
1078  * is closed; the reference count may be zero or larger than zero
1079  * at this point.
1080  *
1081  * @param server the server manageing the clients
1082  * @param callback function to call on disconnect
1083  * @param callback_cls closure for callback
1084  */
1085 void
1086 GNUNET_SERVER_disconnect_notify (struct GNUNET_SERVER_Handle *server,
1087                                  GNUNET_SERVER_DisconnectCallback callback,
1088                                  void *callback_cls)
1089 {
1090   struct NotifyList *n;
1091
1092   n = GNUNET_malloc (sizeof (struct NotifyList));
1093   n->callback = callback;
1094   n->callback_cls = callback_cls;
1095   n->next = server->disconnect_notify_list;
1096   server->disconnect_notify_list = n;
1097 }
1098
1099
1100 /**
1101  * Ask the server to disconnect from the given client.
1102  * This is the same as returning GNUNET_SYSERR from a message
1103  * handler, except that it allows dropping of a client even
1104  * when not handling a message from that client.
1105  *
1106  * @param client the client to disconnect from
1107  */
1108 void
1109 GNUNET_SERVER_client_disconnect (struct GNUNET_SERVER_Client *client)
1110 {
1111   if (client->server == NULL)
1112     return;                     /* already disconnected */
1113   GNUNET_assert (client->my_receive != GNUNET_SCHEDULER_NO_TASK);
1114   client->receive_cancel (client->client_closure, client->my_receive);
1115   client->my_receive = GNUNET_SCHEDULER_NO_TASK;
1116   shutdown_incoming_processing (client);
1117 }
1118
1119
1120 /**
1121  * Notify us when the server has enough space to transmit
1122  * a message of the given size to the given client.
1123  *
1124  * @param server the server to use
1125  * @param client client to transmit message to
1126  * @param size requested amount of buffer space
1127  * @param timeout after how long should we give up (and call
1128  *        notify with buf NULL and size 0)?
1129  * @param callback function to call when space is available
1130  * @param callback_cls closure for callback
1131  * @return non-NULL if the notify callback was queued; can be used
1132  *           to cancel the request using
1133  *           GNUNET_NETWORK_connection_notify_transmit_ready_cancel.
1134  *         NULL if we are already going to notify someone else (busy)
1135  */
1136 struct GNUNET_NETWORK_TransmitHandle *
1137 GNUNET_SERVER_notify_transmit_ready (struct GNUNET_SERVER_Client *client,
1138                                      size_t size,
1139                                      struct GNUNET_TIME_Relative timeout,
1140                                      GNUNET_NETWORK_TransmitReadyNotify
1141                                      callback, void *callback_cls)
1142 {
1143   return client->notify_transmit_ready (client->client_closure,
1144                                         size,
1145                                         timeout, callback, callback_cls);
1146 }
1147
1148
1149 /**
1150  * Resume receiving from this client, we are done processing the
1151  * current request.  This function must be called from within each
1152  * GNUNET_SERVER_MessageCallback (or its respective continuations).
1153  *
1154  * @param client client we were processing a message of
1155  * @param success GNUNET_OK to keep the connection open and
1156  *                          continue to receive
1157  *                GNUNET_SYSERR to close the connection (signal
1158  *                          serious error)
1159  */
1160 void
1161 GNUNET_SERVER_receive_done (struct GNUNET_SERVER_Client *client, int success)
1162 {
1163   char *sb;
1164
1165   if (client == NULL)
1166     return;
1167   GNUNET_assert (client->suspended > 0);
1168   client->suspended--;
1169   if (success != GNUNET_OK)
1170     client->shutdown_now = GNUNET_YES;
1171   if (client->suspended > 0)
1172     return;
1173   if (client->in_process_client_buffer == GNUNET_YES)
1174     return;
1175   if (client->side_buf_size > 0)
1176     {
1177       /* resume processing from side-buf */
1178       sb = client->side_buf;
1179       client->side_buf = NULL;
1180       /* this will also resume the receive job */
1181       if (GNUNET_YES != client->shutdown_now)
1182         process_incoming (client, sb, client->side_buf_size, NULL, 0, 0);
1183       else
1184         shutdown_incoming_processing (client);
1185       /* finally, free the side-buf */
1186       GNUNET_free (sb);
1187       return;
1188     }
1189   /* resume receive job */
1190   if (GNUNET_YES != client->shutdown_now)
1191     {
1192       GNUNET_SCHEDULER_add_continuation (client->server->sched,
1193                                          GNUNET_NO,
1194                                          &restart_processing,
1195                                          client,
1196                                          GNUNET_SCHEDULER_REASON_PREREQ_DONE);
1197       return;
1198     }
1199   shutdown_incoming_processing (client);
1200 }
1201
1202
1203 /* end of server.c */