src/util/connection.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/connection.c
  23  * @brief  TCP connection management
  24  * @author Christian Grothoff
  25  *
  26  * This code is rather complex.  Only modify it if you
  27  * 1) Have a NEW testcase showing that the new code
  28  *    is needed and correct
  29  * 2) All EXISTING testcases pass with the new code
  30  * These rules should apply in general, but for this
  31  * module they are VERY, VERY important.
  32  */
  33
  34 #include "platform.h"
  35 #include "gnunet_common.h"
  36 #include "gnunet_connection_lib.h"
  37 #include "gnunet_container_lib.h"
  38 #include "gnunet_resolver_service.h"
  39 #include "gnunet_scheduler_lib.h"
  40 #include "gnunet_server_lib.h"
  41
  42 #define DEBUG_CONNECTION GNUNET_NO
  43
  44
  45 /**
  46  * Possible functions to call after connect failed or succeeded.
  47  */
  48 enum ConnectContinuations
  49 {
  50     /**
  51      * Call nothing.
  52      */
  53   COCO_NONE = 0,
  54
  55     /**
  56      * Call "receive_again".
  57      */
  58   COCO_RECEIVE_AGAIN = 1,
  59
  60     /**
  61      * Call "transmit_ready".
  62      */
  63   COCO_TRANSMIT_READY = 2,
  64
  65     /**
  66      * Call "destroy_continuation".
  67      */
  68   COCO_DESTROY_CONTINUATION = 4
  69 };
  70
  71
  72 /**
  73  * Transmission handle.  There can only be one for each connection.
  74  */
  75 struct GNUNET_CONNECTION_TransmitHandle
  76 {
  77
  78   /**
  79    * Function to call if the send buffer has notify_size
  80    * bytes available.
  81    */
  82   GNUNET_CONNECTION_TransmitReadyNotify notify_ready;
  83
  84   /**
  85    * Closure for notify_ready.
  86    */
  87   void *notify_ready_cls;
  88
  89   /**
  90    * Our socket handle.
  91    */
  92   struct GNUNET_CONNECTION_Handle *sh;
  93
  94   /**
  95    * Timeout for receiving (in absolute time).
  96    */
  97   struct GNUNET_TIME_Absolute transmit_timeout;
  98
  99   /**
 100    * Task called on timeout.
 101    */
 102   GNUNET_SCHEDULER_TaskIdentifier timeout_task;
 103
 104   /**
 105    * At what number of bytes available in the
 106    * write buffer should the notify method be called?
 107    */
 108   size_t notify_size;
 109
 110 };
 111
 112
 113 /**
 114  * During connect, we try multiple possible IP addresses
 115  * to find out which one might work.
 116  */
 117 struct AddressProbe
 118 {
 119
 120   /**
 121    * This is a linked list.
 122    */
 123   struct AddressProbe *next;
 124
 125   /**
 126    * This is a doubly-linked list.
 127    */
 128   struct AddressProbe *prev;
 129
 130   /**
 131    * The address; do not free (allocated at the end of this struct).
 132    */
 133   const struct sockaddr *addr;
 134
 135   /**
 136    * Underlying OS's socket.
 137    */
 138   struct GNUNET_NETWORK_Handle *sock;
 139
 140   /**
 141    * Connection for which we are probing.
 142    */
 143   struct GNUNET_CONNECTION_Handle *h;
 144
 145   /**
 146    * Lenth of addr.
 147    */
 148   socklen_t addrlen;
 149
 150   /**
 151    * Task waiting for the socket to finish connecting.
 152    */
 153   GNUNET_SCHEDULER_TaskIdentifier task;
 154 };
 155
 156
 157 /**
 158  * @brief handle for a network socket
 159  */
 160 struct GNUNET_CONNECTION_Handle
 161 {
 162
 163   /**
 164    * Configuration to use.
 165    */
 166   const struct GNUNET_CONFIGURATION_Handle *cfg;
 167
 168   /**
 169    * Linked list of sockets we are currently trying out
 170    * (during connect).
 171    */
 172   struct AddressProbe *ap_head;
 173
 174   /**
 175    * Linked list of sockets we are currently trying out
 176    * (during connect).
 177    */
 178   struct AddressProbe *ap_tail;
 179
 180   /**
 181    * Network address of the other end-point, may be NULL.
 182    */
 183   struct sockaddr *addr;
 184
 185   /**
 186    * Pointer to the hostname if socket was
 187    * created using DNS lookup, otherwise NULL.
 188    */
 189   char *hostname;
 190
 191   /**
 192    * Underlying OS's socket, set to NULL after fatal errors.
 193    */
 194   struct GNUNET_NETWORK_Handle *sock;
 195
 196   /**
 197    * Function to call on data received, NULL if no receive is pending.
 198    */
 199   GNUNET_CONNECTION_Receiver receiver;
 200
 201   /**
 202    * Closure for receiver.
 203    */
 204   void *receiver_cls;
 205
 206   /**
 207    * Pointer to our write buffer.
 208    */
 209   char *write_buffer;
 210
 211   /**
 212    * Current size of our write buffer.
 213    */
 214   size_t write_buffer_size;
 215
 216   /**
 217    * Current write-offset in write buffer (where
 218    * would we write next).
 219    */
 220   size_t write_buffer_off;
 221
 222   /**
 223    * Current read-offset in write buffer (how many
 224    * bytes have already been sent).
 225    */
 226   size_t write_buffer_pos;
 227
 228   /**
 229    * Length of addr.
 230    */
 231   socklen_t addrlen;
 232
 233   /**
 234    * Read task that we may need to wait for.
 235    */
 236   GNUNET_SCHEDULER_TaskIdentifier read_task;
 237
 238   /**
 239    * Write task that we may need to wait for.
 240    */
 241   GNUNET_SCHEDULER_TaskIdentifier write_task;
 242
 243   /**
 244    * Destroy task (if already scheduled).
 245    */
 246   GNUNET_SCHEDULER_TaskIdentifier destroy_task;
 247
 248   /**
 249    * Handle to a pending DNS lookup request.
 250    */
 251   struct GNUNET_RESOLVER_RequestHandle *dns_active;
 252
 253   /**
 254    * The handle we return for GNUNET_CONNECTION_notify_transmit_ready.
 255    */
 256   struct GNUNET_CONNECTION_TransmitHandle nth;
 257
 258   /**
 259    * Timeout for receiving (in absolute time).
 260    */
 261   struct GNUNET_TIME_Absolute receive_timeout;
 262
 263   /**
 264    * Functions to call after connect failed or succeeded.
 265    */
 266   enum ConnectContinuations ccs;
 267
 268   /**
 269    * Maximum number of bytes to read (for receiving).
 270    */
 271   size_t max;
 272
 273   /**
 274    * Ignore GNUNET_SCHEDULER_REASON_SHUTDOWN for this socket.
 275    */
 276   int ignore_shutdown;
 277
 278   /**
 279    * Port to connect to.
 280    */
 281   uint16_t port;
 282
 283   /**
 284    * When shutdown, do not ever actually close the socket, but
 285    * free resources.  Only should ever be set if using program
 286    * termination as a signal (because only then will the leaked
 287    * socket be freed!)
 288    */
 289   int16_t persist;
 290
 291 };
 292
 293 /**
 294  * Set the persist option on this connection handle.  Indicates
 295  * that the underlying socket or fd should never really be closed.
 296  * Used for indicating process death.
 297  *
 298  * @param sock the connection to set persistent
 299  */
 300 void GNUNET_CONNECTION_persist_(struct GNUNET_CONNECTION_Handle *sock)
 301 {
 302   sock->persist = GNUNET_YES;
 303 }
 304
 305 /**
 306  * Create a socket handle by boxing an existing OS socket.  The OS
 307  * socket should henceforth be no longer used directly.
 308  * GNUNET_socket_destroy will close it.
 309  *
 310  * @param osSocket existing socket to box
 311  * @return the boxed socket handle
 312  */
 313 struct GNUNET_CONNECTION_Handle *
 314 GNUNET_CONNECTION_create_from_existing (struct GNUNET_NETWORK_Handle
 315                                         *osSocket)
 316 {
 317   struct GNUNET_CONNECTION_Handle *ret;
 318   ret = GNUNET_malloc (sizeof (struct GNUNET_CONNECTION_Handle));
 319   ret->write_buffer_size = GNUNET_SERVER_MIN_BUFFER_SIZE;
 320   ret->write_buffer = GNUNET_malloc(ret->write_buffer_size);
 321   ret->sock = osSocket;
 322   return ret;
 323 }
 324
 325
 326 /**
 327  * Create a socket handle by accepting on a listen socket.  This
 328  * function may block if the listen socket has no connection ready.
 329  *
 330  * @param access function to use to check if access is allowed
 331  * @param access_cls closure for access
 332  * @param lsock listen socket
 333  * @return the socket handle, NULL on error
 334  */
 335 struct GNUNET_CONNECTION_Handle *
 336 GNUNET_CONNECTION_create_from_accept (GNUNET_CONNECTION_AccessCheck access,
 337                                       void *access_cls,
 338                                       struct GNUNET_NETWORK_Handle *lsock)
 339 {
 340   struct GNUNET_CONNECTION_Handle *ret;
 341   char addr[128];
 342   socklen_t addrlen;
 343   struct GNUNET_NETWORK_Handle *sock;
 344   int aret;
 345   struct sockaddr_in *v4;
 346   struct sockaddr_in6 *v6;
 347   struct sockaddr *sa;
 348   void *uaddr;
 349
 350   addrlen = sizeof (addr);
 351   sock =
 352     GNUNET_NETWORK_socket_accept (lsock, (struct sockaddr *) &addr, &addrlen);
 353   if (NULL == sock)
 354     {
 355       GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING, "accept");
 356       return NULL;
 357     }
 358   if ( (addrlen > sizeof (addr)) ||
 359        (addrlen < sizeof (sa_family_t)) )
 360     {
 361       GNUNET_break (0);
 362       GNUNET_break (GNUNET_OK == GNUNET_NETWORK_socket_close (sock));
 363       return NULL;
 364     }
 365
 366   sa = (struct sockaddr *) addr;
 367   v6 = (struct sockaddr_in6 *) addr;
 368   if ((sa->sa_family == AF_INET6) && (IN6_IS_ADDR_V4MAPPED (&v6->sin6_addr)))
 369     {
 370       /* convert to V4 address */
 371       v4 = GNUNET_malloc (sizeof (struct sockaddr_in));
 372       memset (v4, 0, sizeof (struct sockaddr_in));
 373       v4->sin_family = AF_INET;
 374       memcpy (&v4->sin_addr,
 375               &((char *) &v6->sin6_addr)[sizeof (struct in6_addr) -
 376                                          sizeof (struct in_addr)],
 377               sizeof (struct in_addr));
 378       v4->sin_port = v6->sin6_port;
 379       uaddr = v4;
 380       addrlen = sizeof (struct sockaddr_in);
 381     }
 382   else
 383     {
 384       uaddr = GNUNET_malloc (addrlen);
 385       memcpy (uaddr, addr, addrlen);
 386     }
 387
 388   if ((access != NULL) &&
 389       (GNUNET_YES != (aret = access (access_cls, uaddr, addrlen))))
 390     {
 391       if (aret == GNUNET_NO)
 392         GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 393                     _("Access denied to `%s'\n"),
 394                     GNUNET_a2s (uaddr, addrlen));
 395       GNUNET_break (GNUNET_OK ==
 396                     GNUNET_NETWORK_socket_shutdown (sock, SHUT_RDWR));
 397       GNUNET_break (GNUNET_OK == GNUNET_NETWORK_socket_close (sock));
 398       GNUNET_free (uaddr);
 399       return NULL;
 400     }
 401   ret = GNUNET_malloc (sizeof (struct GNUNET_CONNECTION_Handle));
 402   ret->write_buffer_size = GNUNET_SERVER_MIN_BUFFER_SIZE;
 403   ret->write_buffer = GNUNET_malloc(ret->write_buffer_size);
 404   ret->addr = uaddr;
 405   ret->addrlen = addrlen;
 406   ret->sock = sock;
 407 #if DEBUG_CONNECTION
 408   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 409               _("Accepting connection from `%s': %p\n"),
 410               GNUNET_a2s (uaddr, addrlen), ret);
 411 #endif
 412   return ret;
 413 }
 414
 415 /**
 416  * Obtain the network address of the other party.
 417  *
 418  * @param sock the client to get the address for
 419  * @param addr where to store the address
 420  * @param addrlen where to store the length of the address
 421  * @return GNUNET_OK on success
 422  */
 423 int
 424 GNUNET_CONNECTION_get_address (struct GNUNET_CONNECTION_Handle *sock,
 425                                void **addr, size_t * addrlen)
 426 {
 427   if ((sock->addr == NULL) || (sock->addrlen == 0))
 428     return GNUNET_NO;
 429   *addr = GNUNET_malloc (sock->addrlen);
 430   memcpy (*addr, sock->addr, sock->addrlen);
 431   *addrlen = sock->addrlen;
 432   return GNUNET_OK;
 433 }
 434
 435
 436 /**
 437  * This function is called after establishing a connection either has
 438  * succeeded or timed out.  Note that it is possible that the attempt
 439  * timed out and that we're immediately retrying.  If we are retrying,
 440  * we need to wait again (or timeout); if we succeeded, we need to
 441  * wait for data (or timeout).
 442  *
 443  * @param cls our connection handle
 444  * @param tc task context describing why we are here
 445  */
 446 static void
 447 receive_again (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
 448
 449
 450 /**
 451  * Scheduler let us know that the connect task is finished (or was
 452  * cancelled due to shutdown).  Now really clean up.
 453  *
 454  * @param cls our "struct GNUNET_CONNECTION_Handle *"
 455  * @param tc unused
 456  */
 457 static void
 458 destroy_continuation (void *cls,
 459                       const struct GNUNET_SCHEDULER_TaskContext *tc)
 460 {
 461   struct GNUNET_CONNECTION_Handle *sock = cls;
 462   GNUNET_CONNECTION_TransmitReadyNotify notify;
 463   struct AddressProbe *pos;
 464
 465   sock->destroy_task = GNUNET_SCHEDULER_NO_TASK;
 466   GNUNET_assert (sock->dns_active == NULL);
 467   if (0 != (sock->ccs & COCO_TRANSMIT_READY))
 468     {
 469 #if DEBUG_CONNECTION
 470       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 471                   "Destroy waits for CCS-TR to be done (%p)\n", sock);
 472 #endif
 473       sock->ccs |= COCO_DESTROY_CONTINUATION;
 474       return;
 475     }
 476   if (sock->write_task != GNUNET_SCHEDULER_NO_TASK)
 477     {
 478 #if DEBUG_CONNECTION
 479       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 480                   "Destroy waits for write_task to be done (%p)\n", sock);
 481 #endif
 482       GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == sock->destroy_task);
 483       sock->destroy_task
 484         = GNUNET_SCHEDULER_add_after (sock->write_task,
 485                                       &destroy_continuation, sock);
 486       return;
 487     }
 488   if (0 != (sock->ccs & COCO_RECEIVE_AGAIN))
 489     {
 490       sock->ccs |= COCO_DESTROY_CONTINUATION;
 491       return;
 492     }
 493   if (sock->sock != NULL)
 494     {
 495 #if DEBUG_CONNECTION
 496       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 497                   "Shutting down socket (%p)\n", sock);
 498 #endif
 499       if (sock->persist != GNUNET_YES)
 500         GNUNET_NETWORK_socket_shutdown (sock->sock, SHUT_RDWR);
 501     }
 502   if (sock->read_task != GNUNET_SCHEDULER_NO_TASK)
 503     {
 504       GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == sock->destroy_task);
 505       sock->destroy_task
 506         = GNUNET_SCHEDULER_add_after (sock->read_task,
 507                                       &destroy_continuation, sock);
 508       return;
 509     }
 510 #if DEBUG_CONNECTION
 511   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 512               "Destroy actually runs (%p)!\n", sock);
 513 #endif
 514   while (NULL != (pos = sock->ap_head))
 515     {
 516       GNUNET_break (GNUNET_OK == GNUNET_NETWORK_socket_close (pos->sock));
 517       GNUNET_SCHEDULER_cancel (pos->task);
 518       GNUNET_CONTAINER_DLL_remove (sock->ap_head, sock->ap_tail, pos);
 519       GNUNET_free (pos);
 520     }
 521   GNUNET_assert (sock->nth.timeout_task == GNUNET_SCHEDULER_NO_TASK);
 522   GNUNET_assert (sock->ccs == COCO_NONE);
 523   if (NULL != (notify = sock->nth.notify_ready))
 524     {
 525       sock->nth.notify_ready = NULL;
 526       notify (sock->nth.notify_ready_cls, 0, NULL);
 527     }
 528
 529   if (sock->sock != NULL)
 530     {
 531       if (sock->persist != GNUNET_YES)
 532         GNUNET_break (GNUNET_OK == GNUNET_NETWORK_socket_close (sock->sock));
 533       else
 534         GNUNET_free (sock->sock); /* at least no memory leak (we deliberately
 535                                      leak the socket in this special case) ... */
 536     }
 537   GNUNET_free_non_null (sock->addr);
 538   GNUNET_free_non_null (sock->hostname);
 539   GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == sock->destroy_task);
 540 #if DEBUG_CONNECTION
 541   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 542               "Freeing memory of connection %p.\n", sock);
 543 #endif
 544   GNUNET_free (sock->write_buffer);
 545   GNUNET_free (sock);
 546 }
 547
 548
 549
 550 /**
 551  * See if we are now connected.  If not, wait longer for
 552  * connect to succeed.  If connected, we should be able
 553  * to write now as well, unless we timed out.
 554  *
 555  * @param cls our connection handle
 556  * @param tc task context describing why we are here
 557  */
 558 static void
 559 transmit_ready (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
 560
 561
 562 /**
 563  * We've failed for good to establish a connection.
 564  *
 565  * @param h the connection we tried to establish
 566  */
 567 static void
 568 connect_fail_continuation (struct GNUNET_CONNECTION_Handle *h)
 569 {
 570 #if DEBUG_CONNECTION
 571   GNUNET_log ((0 != strncmp (h->hostname,
 572                              "localhost:",
 573                              10))
 574               ? GNUNET_ERROR_TYPE_INFO
 575               : GNUNET_ERROR_TYPE_WARNING,
 576               _("Failed to establish TCP connection to `%s:%u', no further addresses to try.\n"),
 577               h->hostname, h->port);
 578 #endif
 579   /* connect failed / timed out */
 580   GNUNET_break (h->ap_head == NULL);
 581   GNUNET_break (h->ap_tail == NULL);
 582   GNUNET_break (h->dns_active == GNUNET_NO);
 583   GNUNET_break (h->sock == NULL);
 584
 585   /* trigger jobs that used to wait on "connect_task" */
 586   if (0 != (h->ccs & COCO_RECEIVE_AGAIN))
 587     {
 588 #if DEBUG_CONNECTION
 589       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 590                   "connect_fail_continuation triggers receive_again (%p)\n",
 591                   h);
 592 #endif
 593       h->ccs -= COCO_RECEIVE_AGAIN;
 594       h->read_task = GNUNET_SCHEDULER_add_after (GNUNET_SCHEDULER_NO_TASK,
 595                                                  &receive_again, h);
 596     }
 597   if (0 != (h->ccs & COCO_TRANSMIT_READY))
 598     {
 599 #if DEBUG_CONNECTION
 600       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 601                   "connect_fail_continuation cancels timeout_task, triggers transmit_ready (%p)\n",
 602                   h);
 603 #endif
 604       GNUNET_assert (h->nth.timeout_task != GNUNET_SCHEDULER_NO_TASK);
 605       GNUNET_SCHEDULER_cancel (h->nth.timeout_task);
 606       h->nth.timeout_task = GNUNET_SCHEDULER_NO_TASK;
 607       h->ccs -= COCO_TRANSMIT_READY;
 608       GNUNET_assert (h->write_task == GNUNET_SCHEDULER_NO_TASK);
 609       h->write_task = GNUNET_SCHEDULER_add_after (GNUNET_SCHEDULER_NO_TASK,
 610                                                   &transmit_ready, h);
 611     }
 612   if (0 != (h->ccs & COCO_DESTROY_CONTINUATION))
 613     {
 614 #if DEBUG_CONNECTION
 615       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 616                   "connect_fail_continuation runs destroy_continuation (%p)\n",
 617                   h);
 618 #endif
 619       h->ccs -= COCO_DESTROY_CONTINUATION;
 620       GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == h->destroy_task);
 621       h->destroy_task
 622         = GNUNET_SCHEDULER_add_now (&destroy_continuation,
 623                                     h);
 624     }
 625 }
 626
 627
 628 /**
 629  * We've succeeded in establishing a connection.
 630  *
 631  * @param h the connection we tried to establish
 632  */
 633 static void
 634 connect_success_continuation (struct GNUNET_CONNECTION_Handle *h)
 635 {
 636 #if DEBUG_CONNECTION
 637   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 638               "Connection to `%s' succeeded! (%p)\n",
 639               GNUNET_a2s (h->addr, h->addrlen), h);
 640 #endif
 641   /* trigger jobs that waited for the connection */
 642   if (0 != (h->ccs & COCO_RECEIVE_AGAIN))
 643     {
 644 #if DEBUG_CONNECTION
 645       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 646                   "connect_success_continuation runs receive_again (%p)\n",
 647                   h);
 648 #endif
 649       h->ccs -= COCO_RECEIVE_AGAIN;
 650       h->read_task = GNUNET_SCHEDULER_add_after (GNUNET_SCHEDULER_NO_TASK,
 651                                                  &receive_again, h);
 652     }
 653   if (0 != (h->ccs & COCO_TRANSMIT_READY))
 654     {
 655 #if DEBUG_CONNECTION
 656       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 657                   "connect_success_continuation runs transmit_ready, cancels timeout_task (%p)\n",
 658                   h);
 659 #endif
 660       GNUNET_assert (h->nth.timeout_task != GNUNET_SCHEDULER_NO_TASK);
 661       GNUNET_SCHEDULER_cancel (h->nth.timeout_task);
 662       h->nth.timeout_task = GNUNET_SCHEDULER_NO_TASK;
 663       h->ccs -= COCO_TRANSMIT_READY;
 664       GNUNET_assert (h->write_task == GNUNET_SCHEDULER_NO_TASK);
 665       h->write_task =
 666         GNUNET_SCHEDULER_add_write_net (GNUNET_TIME_absolute_get_remaining
 667                                         (h->nth.transmit_timeout), h->sock,
 668                                         &transmit_ready, h);
 669     }
 670   if (0 != (h->ccs & COCO_DESTROY_CONTINUATION))
 671     {
 672 #if DEBUG_CONNECTION
 673       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 674                   "connect_success_continuation runs destroy_continuation (%p)\n",
 675                   h);
 676 #endif
 677       h->ccs -= COCO_DESTROY_CONTINUATION;
 678       GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == h->destroy_task);
 679       h->destroy_task
 680         = GNUNET_SCHEDULER_add_now (&destroy_continuation,
 681                                                 h);
 682     }
 683 }
 684
 685
 686 /**
 687  * Scheduler let us know that we're either ready to write on the
 688  * socket OR connect timed out.  Do the right thing.
 689  *
 690  * @param cls the "struct AddressProbe*" with the address that we are probing
 691  * @param tc success or failure info about the connect attempt.
 692  */
 693 static void
 694 connect_probe_continuation (void *cls,
 695                             const struct GNUNET_SCHEDULER_TaskContext *tc)
 696 {
 697   struct AddressProbe *ap = cls;
 698   struct GNUNET_CONNECTION_Handle *h = ap->h;
 699   struct AddressProbe *pos;
 700   int error;
 701   socklen_t len;
 702
 703   GNUNET_assert (ap->sock != NULL);
 704   GNUNET_CONTAINER_DLL_remove (h->ap_head, h->ap_tail, ap);
 705   len = sizeof (error);
 706   errno = 0;
 707   error = 0;
 708   if ( (0 == (tc->reason & GNUNET_SCHEDULER_REASON_WRITE_READY)) ||
 709        (GNUNET_OK !=
 710         GNUNET_NETWORK_socket_getsockopt (ap->sock, SOL_SOCKET, SO_ERROR,
 711                                           &error, &len)) ||
 712        (error != 0) )
 713     {
 714       GNUNET_break (GNUNET_OK == GNUNET_NETWORK_socket_close (ap->sock));
 715       GNUNET_free (ap);
 716       if ((NULL == h->ap_head) && (h->dns_active == GNUNET_NO))
 717         connect_fail_continuation (h);
 718       return;
 719     }
 720   GNUNET_assert (h->sock == NULL);
 721   h->sock = ap->sock;
 722   GNUNET_assert (h->addr == NULL);
 723   h->addr = GNUNET_malloc (ap->addrlen);
 724   memcpy (h->addr, ap->addr, ap->addrlen);
 725   h->addrlen = ap->addrlen;
 726   GNUNET_free (ap);
 727   /* cancel all other attempts */
 728   while (NULL != (pos = h->ap_head))
 729     {
 730       GNUNET_break (GNUNET_OK == GNUNET_NETWORK_socket_close (pos->sock));
 731       GNUNET_SCHEDULER_cancel (pos->task);
 732       GNUNET_CONTAINER_DLL_remove (h->ap_head, h->ap_tail, pos);
 733       GNUNET_free (pos);
 734     }
 735   connect_success_continuation (h);
 736 }
 737
 738
 739 /**
 740  * Try to establish a socket connection given the specified address.
 741  * This function is called by the resolver once we have a DNS reply.
 742  *
 743  * @param cls our "struct GNUNET_CONNECTION_Handle *"
 744  * @param addr address to try, NULL for "last call"
 745  * @param addrlen length of addr
 746  */
 747 static void
 748 try_connect_using_address (void *cls,
 749                            const struct sockaddr *addr, socklen_t addrlen)
 750 {
 751   struct GNUNET_CONNECTION_Handle *h = cls;
 752   struct AddressProbe *ap;
 753   struct GNUNET_TIME_Relative delay;
 754
 755   if (addr == NULL)
 756     {
 757       h->dns_active = NULL;
 758       if (NULL == h->ap_head)
 759         connect_fail_continuation (h);
 760       return;
 761     }
 762   if (h->sock != NULL)
 763     return;                     /* already connected */
 764   GNUNET_assert (h->addr == NULL);
 765   /* try to connect */
 766 #if DEBUG_CONNECTION
 767   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 768               "Trying to connect using address `%s:%u/%s:%u'\n",
 769               h->hostname, h->port, GNUNET_a2s (addr, addrlen), h->port);
 770 #endif
 771   ap = GNUNET_malloc (sizeof (struct AddressProbe) + addrlen);
 772   ap->addr = (const struct sockaddr *) &ap[1];
 773   memcpy (&ap[1], addr, addrlen);
 774   ap->addrlen = addrlen;
 775   ap->h = h;
 776
 777   switch (ap->addr->sa_family)
 778     {
 779     case AF_INET:
 780       ((struct sockaddr_in *) ap->addr)->sin_port = htons (h->port);
 781       break;
 782     case AF_INET6:
 783       ((struct sockaddr_in6 *) ap->addr)->sin6_port = htons (h->port);
 784       break;
 785     default:
 786       GNUNET_break (0);
 787       GNUNET_free (ap);
 788       return;                   /* not supported by us */
 789     }
 790   ap->sock =
 791     GNUNET_NETWORK_socket_create (ap->addr->sa_family, SOCK_STREAM, 0);
 792   if (ap->sock == NULL)
 793     {
 794       GNUNET_free (ap);
 795       return;                   /* not supported by OS */
 796     }
 797 #if DEBUG_CONNECTION
 798   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 799               _("Trying to connect to `%s' (%p)\n"),
 800               GNUNET_a2s (ap->addr, ap->addrlen), h);
 801 #endif
 802   if ((GNUNET_OK != GNUNET_NETWORK_socket_connect (ap->sock,
 803                                                    ap->addr,
 804                                                    ap->addrlen)) &&
 805       (errno != EINPROGRESS))
 806     {
 807       /* maybe refused / unsupported address, try next */
 808       GNUNET_log_strerror (GNUNET_ERROR_TYPE_INFO, "connect");
 809       GNUNET_break (GNUNET_OK == GNUNET_NETWORK_socket_close (ap->sock));
 810       GNUNET_free (ap);
 811       return;
 812     }
 813   GNUNET_CONTAINER_DLL_insert (h->ap_head, h->ap_tail, ap);
 814   delay = GNUNET_CONNECTION_CONNECT_RETRY_TIMEOUT;
 815   if (h->nth.notify_ready != NULL)
 816     delay = GNUNET_TIME_relative_min (delay,
 817                                       GNUNET_TIME_absolute_get_remaining
 818                                       (h->nth.transmit_timeout));
 819   if (h->receiver != NULL)
 820     delay = GNUNET_TIME_relative_min (delay,
 821                                       GNUNET_TIME_absolute_get_remaining
 822                                       (h->receive_timeout));
 823   ap->task =
 824     GNUNET_SCHEDULER_add_write_net (delay, ap->sock,
 825                                     &connect_probe_continuation, ap);
 826 }
 827
 828
 829 /**
 830  * Create a socket handle by (asynchronously) connecting to a host.
 831  * This function returns immediately, even if the connection has not
 832  * yet been established.  This function only creates TCP connections.
 833  *
 834  * @param cfg configuration to use
 835  * @param hostname name of the host to connect to
 836  * @param port port to connect to
 837  * @return the socket handle
 838  */
 839 struct GNUNET_CONNECTION_Handle *
 840 GNUNET_CONNECTION_create_from_connect (const struct
 841                                        GNUNET_CONFIGURATION_Handle *cfg,
 842                                        const char *hostname, uint16_t port)
 843 {
 844   struct GNUNET_CONNECTION_Handle *ret;
 845
 846   GNUNET_assert (0 < strlen (hostname));        /* sanity check */
 847   ret = GNUNET_malloc (sizeof (struct GNUNET_CONNECTION_Handle));
 848   ret->cfg = cfg;
 849   ret->write_buffer_size = GNUNET_SERVER_MIN_BUFFER_SIZE;
 850   ret->write_buffer = GNUNET_malloc(ret->write_buffer_size);
 851   ret->port = port;
 852   ret->hostname = GNUNET_strdup (hostname);
 853   ret->dns_active = GNUNET_RESOLVER_ip_get (cfg,
 854                                             ret->hostname,
 855                                             AF_UNSPEC,
 856                                             GNUNET_CONNECTION_CONNECT_RETRY_TIMEOUT,
 857                                             &try_connect_using_address, ret);
 858   return ret;
 859 }
 860
 861
 862 /**
 863  * Create a socket handle by connecting to a UNIX domain service.
 864  * This function returns immediately, even if the connection has not
 865  * yet been established.  This function only creates UNIX connections.
 866  *
 867  * @param cfg configuration to use
 868  * @param unixpath path to connect to
 869  * @return the socket handle, NULL on systems without UNIX support
 870  */
 871 struct GNUNET_CONNECTION_Handle *
 872 GNUNET_CONNECTION_create_from_connect_to_unixpath (const struct
 873                                                    GNUNET_CONFIGURATION_Handle *cfg,
 874                                                    const char *unixpath)
 875 {
 876 #ifdef AF_UNIX
 877   struct GNUNET_CONNECTION_Handle *ret;
 878   struct sockaddr_un *un;
 879   size_t slen;
 880
 881   GNUNET_assert (0 < strlen (unixpath));        /* sanity check */
 882   un = GNUNET_malloc (sizeof (struct sockaddr_un));
 883   un->sun_family = AF_UNIX;
 884   slen = strlen (unixpath) + 1;
 885   if (slen >= sizeof (un->sun_path))
 886     slen = sizeof (un->sun_path) - 1;
 887   memcpy (un->sun_path,
 888           unixpath,
 889           slen);
 890   un->sun_path[slen] = '\0';
 891   slen += sizeof (sa_family_t);
 892 #if LINUX
 893   un->sun_path[0] = '\0';
 894   slen = sizeof (struct sockaddr_un);
 895 #endif
 896   ret = GNUNET_malloc (sizeof (struct GNUNET_CONNECTION_Handle));
 897   ret->cfg = cfg;
 898   ret->write_buffer_size = GNUNET_SERVER_MIN_BUFFER_SIZE;
 899   ret->write_buffer = GNUNET_malloc(ret->write_buffer_size);
 900   ret->port = 0;
 901   ret->hostname = NULL;
 902   ret->addr = (struct sockaddr*) un;
 903   ret->addrlen = slen;
 904   ret->sock = GNUNET_NETWORK_socket_create (AF_UNIX, SOCK_STREAM, 0);
 905   if (NULL == ret->sock)
 906     {
 907       GNUNET_free (ret->addr);
 908       GNUNET_free (ret->write_buffer);
 909       GNUNET_free (ret);
 910       return NULL;
 911     }
 912   if (GNUNET_OK != GNUNET_NETWORK_socket_connect (ret->sock,
 913                                                   ret->addr,
 914                                                   ret->addrlen))
 915     {
 916       GNUNET_break (GNUNET_OK == GNUNET_NETWORK_socket_close (ret->sock));
 917       GNUNET_free (ret->addr);
 918       GNUNET_free (ret->write_buffer);
 919       GNUNET_free (ret);
 920       return NULL;
 921     }
 922   connect_success_continuation (ret);
 923   return ret;
 924 #else
 925   return NULL;
 926 #endif
 927 }
 928
 929
 930 /**
 931  * Create a socket handle by (asynchronously) connecting to a host.
 932  * This function returns immediately, even if the connection has not
 933  * yet been established.  This function only creates TCP connections.
 934  *
 935  * @param af_family address family to use
 936  * @param serv_addr server address
 937  * @param addrlen length of server address
 938  * @return the socket handle
 939  */
 940 struct GNUNET_CONNECTION_Handle *
 941 GNUNET_CONNECTION_create_from_sockaddr (int af_family,
 942                                         const struct sockaddr *serv_addr,
 943                                         socklen_t addrlen)
 944 {
 945   struct GNUNET_NETWORK_Handle *s;
 946   struct GNUNET_CONNECTION_Handle *ret;
 947
 948   s = GNUNET_NETWORK_socket_create (af_family, SOCK_STREAM, 0);
 949   if (s == NULL)
 950     {
 951       GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING |
 952                            GNUNET_ERROR_TYPE_BULK, "socket");
 953       return NULL;
 954     }
 955   if ((GNUNET_OK != GNUNET_NETWORK_socket_connect (s, serv_addr, addrlen))
 956       && (errno != EINPROGRESS))
 957     {
 958       /* maybe refused / unsupported address, try next */
 959       GNUNET_log_strerror (GNUNET_ERROR_TYPE_INFO, "connect");
 960       GNUNET_break (GNUNET_OK == GNUNET_NETWORK_socket_close (s));
 961       return NULL;
 962     }
 963   ret = GNUNET_CONNECTION_create_from_existing (s);
 964   ret->addr = GNUNET_malloc (addrlen);
 965   memcpy (ret->addr, serv_addr, addrlen);
 966   ret->addrlen = addrlen;
 967 #if DEBUG_CONNECTION
 968   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 969               _("Trying to connect to `%s' (%p)\n"),
 970               GNUNET_a2s (serv_addr, addrlen), ret);
 971 #endif
 972   return ret;
 973 }
 974
 975
 976 /**
 977  * Check if socket is valid (no fatal errors have happened so far).
 978  * Note that a socket that is still trying to connect is considered
 979  * valid.
 980  *
 981  * @param sock socket to check
 982  * @return GNUNET_YES if valid, GNUNET_NO otherwise
 983  */
 984 int
 985 GNUNET_CONNECTION_check (struct GNUNET_CONNECTION_Handle *sock)
 986 {
 987   if ((sock->ap_head != NULL) || (sock->dns_active != NULL))
 988     return GNUNET_YES;          /* still trying to connect */
 989   return (sock->sock == NULL) ? GNUNET_NO : GNUNET_YES;
 990 }
 991
 992
 993 /**
 994  * Close the socket and free associated resources. Pending
 995  * transmissions may be completed or dropped depending on the
 996  * arguments.   If a receive call is pending and should
 997  * NOT be completed, 'GNUNET_CONNECTION_receive_cancel'
 998  * should be called explicitly first.
 999  *
1000  * @param sock socket to destroy
1001  * @param finish_pending_write should pending writes be completed or aborted?
1002  *        (this applies to transmissions where the data has already been
1003  *        read from the application; all other transmissions should be
1004  *        aborted using 'GNUNET_CONNECTION_notify_transmit_ready_cancel').
1005  */
1006 void
1007 GNUNET_CONNECTION_destroy (struct GNUNET_CONNECTION_Handle *sock,
1008                            int finish_pending_write)
1009 {
1010   if (GNUNET_NO == finish_pending_write)
1011     {
1012       if (sock->write_task != GNUNET_SCHEDULER_NO_TASK)
1013         {
1014           GNUNET_SCHEDULER_cancel (sock->write_task);
1015           sock->write_task = GNUNET_SCHEDULER_NO_TASK;
1016           sock->write_buffer_off = 0;
1017         }
1018     }
1019   if ((sock->write_buffer_off == 0) && (sock->dns_active != NULL))
1020     {
1021       GNUNET_RESOLVER_request_cancel (sock->dns_active);
1022       sock->dns_active = NULL;
1023     }
1024
1025   GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == sock->destroy_task);
1026   sock->destroy_task
1027     = GNUNET_SCHEDULER_add_now (&destroy_continuation, sock);
1028 }
1029
1030
1031 /**
1032  * Tell the receiver callback that a timeout was reached.
1033  */
1034 static void
1035 signal_timeout (struct GNUNET_CONNECTION_Handle *sh)
1036 {
1037   GNUNET_CONNECTION_Receiver receiver;
1038
1039 #if DEBUG_CONNECTION
1040   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1041               "Network signals time out to receiver (%p)!\n", sh);
1042 #endif
1043   GNUNET_assert (NULL != (receiver = sh->receiver));
1044   sh->receiver = NULL;
1045   receiver (sh->receiver_cls, NULL, 0, NULL, 0, 0);
1046 }
1047
1048
1049 /**
1050  * Tell the receiver callback that we had an IO error.
1051  */
1052 static void
1053 signal_error (struct GNUNET_CONNECTION_Handle *sh, int errcode)
1054 {
1055   GNUNET_CONNECTION_Receiver receiver;
1056   GNUNET_assert (NULL != (receiver = sh->receiver));
1057   sh->receiver = NULL;
1058   receiver (sh->receiver_cls, NULL, 0, sh->addr, sh->addrlen, errcode);
1059 }
1060
1061
1062 /**
1063  * This function is called once we either timeout
1064  * or have data ready to read.
1065  */
1066 static void
1067 receive_ready (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1068 {
1069   struct GNUNET_CONNECTION_Handle *sh = cls;
1070   struct GNUNET_TIME_Absolute now;
1071   char buffer[sh->max];
1072   ssize_t ret;
1073   GNUNET_CONNECTION_Receiver receiver;
1074
1075   sh->read_task = GNUNET_SCHEDULER_NO_TASK;
1076   if ( (GNUNET_YES == sh->ignore_shutdown) &&
1077        (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN)))
1078     {
1079       /* ignore shutdown request, go again immediately */
1080 #if DEBUG_CONNECTION
1081       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1082                   "Ignoring shutdown signal per configuration\n");
1083 #endif
1084       sh->read_task = GNUNET_SCHEDULER_add_read_net (GNUNET_TIME_absolute_get_remaining
1085                                                      (sh->receive_timeout),
1086                                                      sh->sock,
1087                                                      &receive_ready, sh);
1088       return;
1089     }
1090   now = GNUNET_TIME_absolute_get ();
1091   if ((now.abs_value > sh->receive_timeout.abs_value) ||
1092       (0 != (tc->reason & GNUNET_SCHEDULER_REASON_TIMEOUT)) ||
1093       (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN)))
1094     {
1095 #if DEBUG_CONNECTION
1096       if (0 == (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1097         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1098                     "Receive from `%s' encounters error: time out by %llums... (%p)\n",
1099                     GNUNET_a2s (sh->addr, sh->addrlen),
1100                     GNUNET_TIME_absolute_get_duration (sh->receive_timeout).
1101                     rel_value, sh);
1102 #endif
1103       signal_timeout (sh);
1104       return;
1105     }
1106   if (sh->sock == NULL)
1107     {
1108       /* connect failed for good */
1109 #if DEBUG_CONNECTION
1110       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1111                   "Receive encounters error, socket closed... (%p)\n", sh);
1112 #endif
1113       signal_error (sh, ECONNREFUSED);
1114       return;
1115     }
1116   GNUNET_assert (GNUNET_NETWORK_fdset_isset (tc->read_ready, sh->sock));
1117 RETRY:
1118   ret = GNUNET_NETWORK_socket_recv (sh->sock, buffer, sh->max);
1119   if (ret == -1)
1120     {
1121       if (errno == EINTR)
1122         goto RETRY;
1123 #if DEBUG_CONNECTION
1124       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1125                   "Error receiving: %s\n", STRERROR (errno));
1126 #endif
1127       signal_error (sh, errno);
1128       return;
1129     }
1130 #if DEBUG_CONNECTION
1131   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1132               "receive_ready read %u/%u bytes from `%s' (%p)!\n",
1133               (unsigned int) ret,
1134               sh->max, GNUNET_a2s (sh->addr, sh->addrlen), sh);
1135 #endif
1136   GNUNET_assert (NULL != (receiver = sh->receiver));
1137   sh->receiver = NULL;
1138   receiver (sh->receiver_cls, buffer, ret, sh->addr, sh->addrlen, 0);
1139 }
1140
1141
1142 /**
1143  * This function is called after establishing a connection either has
1144  * succeeded or timed out.  Note that it is possible that the attempt
1145  * timed out and that we're immediately retrying.  If we are retrying,
1146  * we need to wait again (or timeout); if we succeeded, we need to
1147  * wait for data (or timeout).
1148  *
1149  * @param cls our connection handle
1150  * @param tc task context describing why we are here
1151  */
1152 static void
1153 receive_again (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1154 {
1155   struct GNUNET_CONNECTION_Handle *sh = cls;
1156   struct GNUNET_TIME_Absolute now;
1157
1158   sh->read_task = GNUNET_SCHEDULER_NO_TASK;
1159   if (sh->sock == NULL)
1160     {
1161       /* not connected and no longer trying */
1162 #if DEBUG_CONNECTION
1163       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1164                   "Receive encounters error, socket closed (%p)...\n", sh);
1165 #endif
1166       signal_error (sh, ECONNREFUSED);
1167       return;
1168     }
1169   now = GNUNET_TIME_absolute_get ();
1170   if ((now.abs_value > sh->receive_timeout.abs_value) ||
1171       (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN)))
1172     {
1173 #if DEBUG_CONNECTION
1174       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1175                   "Receive encounters error: time out (%p)...\n", sh);
1176 #endif
1177       signal_timeout (sh);
1178       return;
1179     }
1180   GNUNET_assert (sh->sock != NULL);
1181   /* connect succeeded, wait for data! */
1182   sh->read_task = GNUNET_SCHEDULER_add_read_net (GNUNET_TIME_absolute_get_remaining
1183                                                  (sh->receive_timeout),
1184                                                  sh->sock,
1185                                                  &receive_ready, sh);
1186 }
1187
1188
1189 /**
1190  * Receive data from the given socket.  Note that this function will
1191  * call "receiver" asynchronously using the scheduler.  It will
1192  * "immediately" return.  Note that there MUST only be one active
1193  * receive call per socket at any given point in time (so do not
1194  * call receive again until the receiver callback has been invoked).
1195  *
1196  * @param sock socket handle
1197  * @param max maximum number of bytes to read
1198  * @param timeout maximum amount of time to wait (use -1 for "forever")
1199  * @param receiver function to call with received data
1200  * @param receiver_cls closure for receiver
1201  */
1202 void
1203 GNUNET_CONNECTION_receive (struct GNUNET_CONNECTION_Handle *sock,
1204                            size_t max,
1205                            struct GNUNET_TIME_Relative timeout,
1206                            GNUNET_CONNECTION_Receiver receiver,
1207                            void *receiver_cls)
1208 {
1209   struct GNUNET_SCHEDULER_TaskContext tc;
1210
1211   GNUNET_assert ((sock->read_task == GNUNET_SCHEDULER_NO_TASK) &&
1212                  (0 == (sock->ccs & COCO_RECEIVE_AGAIN)) &&
1213                  (sock->receiver == NULL));
1214   sock->receiver = receiver;
1215   sock->receiver_cls = receiver_cls;
1216   sock->receive_timeout = GNUNET_TIME_relative_to_absolute (timeout);
1217   sock->max = max;
1218   if (sock->sock != NULL)
1219     {
1220       memset (&tc, 0, sizeof (tc));
1221       tc.reason = GNUNET_SCHEDULER_REASON_PREREQ_DONE;
1222       receive_again (sock, &tc);
1223       return;
1224     }
1225   if ((sock->dns_active == NULL) && (sock->ap_head == NULL))
1226     {
1227       receiver (receiver_cls, NULL, 0, NULL, 0, ETIMEDOUT);
1228       return;
1229     }
1230   sock->ccs += COCO_RECEIVE_AGAIN;
1231 }
1232
1233
1234 /**
1235  * Configure this connection to ignore shutdown signals.
1236  *
1237  * @param sock socket handle
1238  * @param do_ignore GNUNET_YES to ignore, GNUNET_NO to restore default
1239  */
1240 void
1241 GNUNET_CONNECTION_ignore_shutdown (struct GNUNET_CONNECTION_Handle *sock,
1242                                    int do_ignore)
1243 {
1244   sock->ignore_shutdown = do_ignore;
1245 }
1246
1247
1248 /**
1249  * Cancel receive job on the given socket.  Note that the
1250  * receiver callback must not have been called yet in order
1251  * for the cancellation to be valid.
1252  *
1253  * @param sock socket handle
1254  * @return closure of the original receiver callback closure
1255  */
1256 void *
1257 GNUNET_CONNECTION_receive_cancel (struct GNUNET_CONNECTION_Handle *sock)
1258 {
1259   if (sock->read_task != GNUNET_SCHEDULER_NO_TASK)
1260     {
1261       GNUNET_assert (sock == GNUNET_SCHEDULER_cancel (sock->read_task));
1262       sock->read_task = GNUNET_SCHEDULER_NO_TASK;
1263     }
1264   else
1265     {
1266       GNUNET_assert (0 != (sock->ccs & COCO_RECEIVE_AGAIN));
1267       sock->ccs -= COCO_RECEIVE_AGAIN;
1268     }
1269   sock->receiver = NULL;
1270   return sock->receiver_cls;
1271 }
1272
1273
1274 /**
1275  * Try to call the transmit notify method (check if we do
1276  * have enough space available first)!
1277  *
1278  * @param sock socket for which we should do this processing
1279  * @return GNUNET_YES if we were able to call notify
1280  */
1281 static int
1282 process_notify (struct GNUNET_CONNECTION_Handle *sock)
1283 {
1284   size_t used;
1285   size_t avail;
1286   size_t size;
1287   GNUNET_CONNECTION_TransmitReadyNotify notify;
1288
1289   GNUNET_assert (sock->write_task == GNUNET_SCHEDULER_NO_TASK);
1290   if (NULL == (notify = sock->nth.notify_ready))
1291     return GNUNET_NO;
1292   used = sock->write_buffer_off - sock->write_buffer_pos;
1293   avail = sock->write_buffer_size - used;
1294   size = sock->nth.notify_size;
1295   if (size > avail)
1296     return GNUNET_NO;
1297   sock->nth.notify_ready = NULL;
1298   if (sock->write_buffer_size - sock->write_buffer_off < size)
1299     {
1300       /* need to compact */
1301       memmove (sock->write_buffer,
1302                &sock->write_buffer[sock->write_buffer_pos], used);
1303       sock->write_buffer_off -= sock->write_buffer_pos;
1304       sock->write_buffer_pos = 0;
1305     }
1306   avail = sock->write_buffer_size - sock->write_buffer_off;
1307   GNUNET_assert (avail >= size);
1308   size = notify (sock->nth.notify_ready_cls,
1309                  avail,
1310                  &sock->write_buffer[sock->write_buffer_off]);
1311   GNUNET_assert (size <= avail);
1312   sock->write_buffer_off += size;
1313   return GNUNET_YES;
1314 }
1315
1316
1317 /**
1318  * Task invoked by the scheduler when a call to transmit
1319  * is timing out (we never got enough buffer space to call
1320  * the callback function before the specified timeout
1321  * expired).
1322  *
1323  * This task notifies the client about the timeout.
1324  */
1325 static void
1326 transmit_timeout (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1327 {
1328   struct GNUNET_CONNECTION_Handle *sock = cls;
1329   GNUNET_CONNECTION_TransmitReadyNotify notify;
1330
1331 #if DEBUG_CONNECTION
1332   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1333               "transmit_timeout running (%p)\n", sock);
1334 #endif
1335   sock->nth.timeout_task = GNUNET_SCHEDULER_NO_TASK;
1336 #if DEBUG_CONNECTION
1337   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1338               "Transmit to `%s:%u/%s' fails, time out reached (%p).\n",
1339               sock->hostname,
1340               sock->port, GNUNET_a2s (sock->addr, sock->addrlen), sock);
1341 #endif
1342   GNUNET_assert (0 != (sock->ccs & COCO_TRANSMIT_READY));
1343   sock->ccs -= COCO_TRANSMIT_READY;     /* remove request */
1344   notify = sock->nth.notify_ready;
1345   sock->nth.notify_ready = NULL;
1346   notify (sock->nth.notify_ready_cls, 0, NULL);
1347 }
1348
1349
1350 /**
1351  * Task invoked by the scheduler when we failed to connect
1352  * at the time of being asked to transmit.
1353  *
1354  * This task notifies the client about the error.
1355  */
1356 static void
1357 connect_error (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1358 {
1359   struct GNUNET_CONNECTION_Handle *sock = cls;
1360   GNUNET_CONNECTION_TransmitReadyNotify notify;
1361
1362 #if DEBUG_CONNECTION
1363   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1364               "Transmission request of size %u fails, connection failed (%p).\n",
1365               sock->nth.notify_size, sock);
1366 #endif
1367   sock->write_task = GNUNET_SCHEDULER_NO_TASK;
1368   notify = sock->nth.notify_ready;
1369   sock->nth.notify_ready = NULL;
1370   notify (sock->nth.notify_ready_cls, 0, NULL);
1371 }
1372
1373
1374 static void
1375 transmit_error (struct GNUNET_CONNECTION_Handle *sock)
1376 {
1377   GNUNET_CONNECTION_TransmitReadyNotify notify;
1378
1379   if (NULL != sock->sock)
1380     {
1381       GNUNET_NETWORK_socket_shutdown (sock->sock, SHUT_RDWR);
1382       GNUNET_break (GNUNET_OK ==
1383                     GNUNET_NETWORK_socket_close (sock->sock));
1384       sock->sock = NULL;
1385     }
1386   if (sock->read_task != GNUNET_SCHEDULER_NO_TASK)
1387     {
1388       GNUNET_SCHEDULER_cancel (sock->read_task);
1389       sock->read_task = GNUNET_SCHEDULER_NO_TASK;
1390       signal_timeout (sock);
1391       return;
1392     }
1393   if (sock->nth.notify_ready == NULL)
1394     return;                     /* nobody to tell about it */
1395   notify = sock->nth.notify_ready;
1396   sock->nth.notify_ready = NULL;
1397   notify (sock->nth.notify_ready_cls, 0, NULL);
1398 }
1399
1400
1401 /**
1402  * See if we are now connected.  If not, wait longer for
1403  * connect to succeed.  If connected, we should be able
1404  * to write now as well, unless we timed out.
1405  *
1406  * @param cls our connection handle
1407  * @param tc task context describing why we are here
1408  */
1409 static void
1410 transmit_ready (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1411 {
1412   struct GNUNET_CONNECTION_Handle *sock = cls;
1413   GNUNET_CONNECTION_TransmitReadyNotify notify;
1414   ssize_t ret;
1415   size_t have;
1416
1417 #if DEBUG_CONNECTION
1418   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1419               "transmit_ready running (%p).\n", sock);
1420 #endif
1421   GNUNET_assert (sock->write_task != GNUNET_SCHEDULER_NO_TASK);
1422   sock->write_task = GNUNET_SCHEDULER_NO_TASK;
1423   GNUNET_assert (sock->nth.timeout_task == GNUNET_SCHEDULER_NO_TASK);
1424   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1425     {
1426       if (sock->ignore_shutdown == GNUNET_YES)
1427         goto SCHEDULE_WRITE;    /* ignore shutdown, go again immediately */
1428 #if DEBUG_CONNECTION
1429       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1430                   "Transmit to `%s' fails, shutdown happened (%p).\n",
1431                   GNUNET_a2s (sock->addr, sock->addrlen), sock);
1432 #endif
1433       notify = sock->nth.notify_ready;
1434       sock->nth.notify_ready = NULL;
1435       notify (sock->nth.notify_ready_cls, 0, NULL);
1436       return;
1437     }
1438   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_TIMEOUT))
1439     {
1440 #if DEBUG_CONNECTION
1441       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1442                   "Transmit to `%s' fails, time out reached (%p).\n",
1443                   GNUNET_a2s (sock->addr, sock->addrlen), sock);
1444 #endif
1445       notify = sock->nth.notify_ready;
1446       sock->nth.notify_ready = NULL;
1447       notify (sock->nth.notify_ready_cls, 0, NULL);
1448       return;
1449     }
1450   GNUNET_assert (NULL != sock->sock);
1451   if (tc->write_ready == NULL)
1452     {
1453       /* special circumstances (in particular,
1454          PREREQ_DONE after connect): not yet ready to write,
1455          but no "fatal" error either.  Hence retry.  */
1456       goto SCHEDULE_WRITE;
1457     }
1458   if (!GNUNET_NETWORK_fdset_isset (tc->write_ready, sock->sock))
1459     {
1460 #if DEBUG_CONNECTION
1461       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1462                   _
1463                   ("Could not satisfy pending transmission request, socket closed or connect failed (%p).\n"),
1464                   sock);
1465 #endif
1466       transmit_error (sock);
1467       return;                   /* connect failed for good, we're finished */
1468     }
1469   GNUNET_assert (sock->write_buffer_off >= sock->write_buffer_pos);
1470   if ( (sock->nth.notify_ready != NULL) &&
1471        (sock->write_buffer_size < sock->nth.notify_size) )
1472     {
1473       sock->write_buffer = GNUNET_realloc(sock->write_buffer,
1474                                           sock->nth.notify_size);
1475       sock->write_buffer_size = sock->nth.notify_size;
1476     }
1477   process_notify (sock);
1478   have = sock->write_buffer_off - sock->write_buffer_pos;
1479   if (have == 0)
1480     {
1481       /* no data ready for writing, terminate write loop */
1482       return;
1483     }
1484   GNUNET_assert (have <= sock->write_buffer_size);
1485   GNUNET_assert (have + sock->write_buffer_pos <= sock->write_buffer_size);
1486   GNUNET_assert (sock->write_buffer_pos <= sock->write_buffer_size);
1487 RETRY:
1488   ret = GNUNET_NETWORK_socket_send (sock->sock,
1489                                     &sock->write_buffer[sock->
1490                                                         write_buffer_pos],
1491                                     have);
1492   if (ret == -1)
1493     {
1494       if (errno == EINTR)
1495         goto RETRY;
1496 #if 0
1497       int en = errno;
1498       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1499                   _("Failed to send to `%s': %s\n"),
1500                   GNUNET_a2s (sock->addr,
1501                               sock->addrlen),
1502                   STRERROR (en));
1503 #endif
1504 #if DEBUG_CONNECTION
1505       GNUNET_log_strerror (GNUNET_ERROR_TYPE_DEBUG, "send");
1506 #endif
1507       transmit_error (sock);
1508       return;
1509     }
1510 #if DEBUG_CONNECTION
1511   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1512               "transmit_ready transmitted %u/%u bytes to `%s' (%p)\n",
1513               (unsigned int) ret,
1514               have, GNUNET_a2s (sock->addr, sock->addrlen), sock);
1515 #endif
1516   sock->write_buffer_pos += ret;
1517   if (sock->write_buffer_pos == sock->write_buffer_off)
1518     {
1519       /* transmitted all pending data */
1520       sock->write_buffer_pos = 0;
1521       sock->write_buffer_off = 0;
1522     }
1523   if ((sock->write_buffer_off == 0) && (NULL == sock->nth.notify_ready))
1524     return;                     /* all data sent! */
1525   /* not done writing, schedule more */
1526 SCHEDULE_WRITE:
1527 #if DEBUG_CONNECTION
1528   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1529               "Re-scheduling transmit_ready (more to do) (%p).\n", sock);
1530 #endif
1531   if (sock->write_task == GNUNET_SCHEDULER_NO_TASK)
1532     sock->write_task =
1533       GNUNET_SCHEDULER_add_write_net (GNUNET_TIME_absolute_get_remaining
1534                                       (sock->nth.transmit_timeout),
1535                                       sock->sock, &transmit_ready, sock);
1536 }
1537
1538
1539 /**
1540  * Ask the socket to call us once the specified number of bytes
1541  * are free in the transmission buffer.  May call the notify
1542  * method immediately if enough space is available.
1543  *
1544  * @param sock socket
1545  * @param size number of bytes to send
1546  * @param timeout after how long should we give up (and call
1547  *        notify with buf NULL and size 0)?
1548  * @param notify function to call
1549  * @param notify_cls closure for notify
1550  * @return non-NULL if the notify callback was queued,
1551  *         NULL if we are already going to notify someone else (busy)
1552  */
1553 struct GNUNET_CONNECTION_TransmitHandle *
1554 GNUNET_CONNECTION_notify_transmit_ready (struct GNUNET_CONNECTION_Handle
1555                                          *sock, size_t size,
1556                                          struct GNUNET_TIME_Relative timeout,
1557                                          GNUNET_CONNECTION_TransmitReadyNotify
1558                                          notify, void *notify_cls)
1559 {
1560   if (sock->nth.notify_ready != NULL)
1561     return NULL;
1562   GNUNET_assert (notify != NULL);
1563   GNUNET_assert (size < GNUNET_SERVER_MAX_MESSAGE_SIZE);
1564   GNUNET_assert (sock->write_buffer_off <= sock->write_buffer_size);
1565   GNUNET_assert (sock->write_buffer_pos <= sock->write_buffer_size);
1566   GNUNET_assert (sock->write_buffer_pos <= sock->write_buffer_off);
1567   sock->nth.notify_ready = notify;
1568   sock->nth.notify_ready_cls = notify_cls;
1569   sock->nth.sh = sock;
1570   sock->nth.notify_size = size;
1571   sock->nth.transmit_timeout = GNUNET_TIME_relative_to_absolute (timeout);
1572   GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == sock->nth.timeout_task);
1573   if ((sock->sock == NULL) &&
1574       (sock->ap_head == NULL) && (sock->dns_active == NULL))
1575     {
1576       if (sock->write_task != GNUNET_SCHEDULER_NO_TASK)
1577         GNUNET_SCHEDULER_cancel (sock->write_task);
1578       sock->write_task = GNUNET_SCHEDULER_add_now (&connect_error, sock);
1579       return &sock->nth;
1580     }
1581   if (GNUNET_SCHEDULER_NO_TASK != sock->write_task)
1582     return &sock->nth;
1583   if (sock->sock != NULL)
1584     {
1585 #if DEBUG_CONNECTION
1586       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1587                   "Scheduling transmit_ready (%p).\n", sock);
1588 #endif
1589       sock->write_task = GNUNET_SCHEDULER_add_write_net (GNUNET_TIME_absolute_get_remaining
1590                                                          (sock->nth.
1591                                                           transmit_timeout),
1592                                                          sock->sock,
1593                                                          &transmit_ready,
1594                                                          sock);
1595     }
1596   else
1597     {
1598 #if DEBUG_CONNECTION
1599       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1600                   "CCS-Scheduling transmit_ready, adding timeout task (%p).\n",
1601                   sock);
1602 #endif
1603       sock->ccs |= COCO_TRANSMIT_READY;
1604       sock->nth.timeout_task = GNUNET_SCHEDULER_add_delayed (timeout,
1605                                                              &transmit_timeout,
1606                                                              sock);
1607     }
1608   return &sock->nth;
1609 }
1610
1611
1612 /**
1613  * Cancel the specified transmission-ready
1614  * notification.
1615  */
1616 void
1617 GNUNET_CONNECTION_notify_transmit_ready_cancel (struct
1618                                                 GNUNET_CONNECTION_TransmitHandle
1619                                                 *h)
1620 {
1621   GNUNET_assert (h->notify_ready != NULL);
1622   if (0 != (h->sh->ccs & COCO_TRANSMIT_READY))
1623     {
1624 #if DEBUG_CONNECTION
1625       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1626                   "notify_transmit_ready_cancel cancels timeout_task (%p)\n",
1627                   h);
1628 #endif
1629       GNUNET_SCHEDULER_cancel (h->timeout_task);
1630       h->timeout_task = GNUNET_SCHEDULER_NO_TASK;
1631       h->sh->ccs -= COCO_TRANSMIT_READY;
1632     }
1633   else
1634     {
1635       if (h->sh->write_task != GNUNET_SCHEDULER_NO_TASK)
1636         {
1637           GNUNET_SCHEDULER_cancel (h->sh->write_task);
1638           h->sh->write_task = GNUNET_SCHEDULER_NO_TASK;
1639         }
1640     }
1641   h->notify_ready = NULL;
1642 }
1643
1644 /* end of connection.c */