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