src/stream/stream_api.c

   1 /*
   2   This file is part of GNUnet.
   3   (C) 2012 Christian Grothoff (and other contributing authors)
   4
   5   GNUnet is free software; you can redistribute it and/or modify
   6   it under the terms of the GNU General Public License as published
   7   by the Free Software Foundation; either version 3, 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 /* TODO:
  22  *
  23  * Checks for matching the sender and socket->other_peer in server
  24  * message handlers
  25  *
  26  * Add code for write io timeout
  27  *
  28  * Include retransmission for control messages
  29  **/
  30
  31 /**
  32  * @file stream/stream_api.c
  33  * @brief Implementation of the stream library
  34  * @author Sree Harsha Totakura
  35  */
  36
  37
  38 #include "platform.h"
  39 #include "gnunet_common.h"
  40 #include "gnunet_crypto_lib.h"
  41 #include "gnunet_lockmanager_service.h"
  42 #include "gnunet_stream_lib.h"
  43 #include "stream_protocol.h"
  44
  45 /**
  46  * Generic logging shorthand
  47  */
  48 #define LOG(kind,...)                                   \
  49   GNUNET_log_from (kind, "stream-api", __VA_ARGS__)
  50
  51 /**
  52  * Debug logging shorthand
  53  */
  54 #define LOG_DEBUG(...)                          \
  55   LOG (GNUNET_ERROR_TYPE_DEBUG, __VA_ARGS__)
  56
  57 /**
  58  * Time in relative seconds shorthand
  59  */
  60 #define TIME_REL_SECS(sec) \
  61   GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, sec)
  62
  63 /**
  64  * The maximum packet size of a stream packet
  65  */
  66 #define DEFAULT_MAX_PAYLOAD_SIZE 64000
  67
  68 /**
  69  * Receive buffer
  70  */
  71 #define RECEIVE_BUFFER_SIZE 4096000
  72
  73 /**
  74  * states in the Protocol
  75  */
  76 enum State
  77   {
  78     /**
  79      * Client initialization state
  80      */
  81     STATE_INIT,
  82
  83     /**
  84      * Listener initialization state
  85      */
  86     STATE_LISTEN,
  87
  88     /**
  89      * Pre-connection establishment state
  90      */
  91     STATE_HELLO_WAIT,
  92
  93     /**
  94      * State where a connection has been established
  95      */
  96     STATE_ESTABLISHED,
  97
  98     /**
  99      * State where the socket is closed on our side and waiting to be ACK'ed
 100      */
 101     STATE_RECEIVE_CLOSE_WAIT,
 102
 103     /**
 104      * State where the socket is closed for reading
 105      */
 106     STATE_RECEIVE_CLOSED,
 107
 108     /**
 109      * State where the socket is closed on our side and waiting to be ACK'ed
 110      */
 111     STATE_TRANSMIT_CLOSE_WAIT,
 112
 113     /**
 114      * State where the socket is closed for writing
 115      */
 116     STATE_TRANSMIT_CLOSED,
 117
 118     /**
 119      * State where the socket is closed on our side and waiting to be ACK'ed
 120      */
 121     STATE_CLOSE_WAIT,
 122
 123     /**
 124      * State where the socket is closed
 125      */
 126     STATE_CLOSED
 127   };
 128
 129
 130 /**
 131  * Functions of this type are called when a message is written
 132  *
 133  * @param cls the closure from queue_message
 134  * @param socket the socket the written message was bound to
 135  */
 136 typedef void (*SendFinishCallback) (void *cls,
 137                                     struct GNUNET_STREAM_Socket *socket);
 138
 139
 140 /**
 141  * The send message queue
 142  */
 143 struct MessageQueue
 144 {
 145   /**
 146    * The message
 147    */
 148   struct GNUNET_STREAM_MessageHeader *message;
 149
 150   /**
 151    * Callback to be called when the message is sent
 152    */
 153   SendFinishCallback finish_cb;
 154
 155   /**
 156    * The closure for finish_cb
 157    */
 158   void *finish_cb_cls;
 159
 160   /**
 161    * The next message in queue. Should be NULL in the last message
 162    */
 163   struct MessageQueue *next;
 164
 165   /**
 166    * The next message in queue. Should be NULL in the first message
 167    */
 168   struct MessageQueue *prev;
 169 };
 170
 171
 172 /**
 173  * The STREAM Socket Handler
 174  */
 175 struct GNUNET_STREAM_Socket
 176 {
 177   /**
 178    * Retransmission timeout
 179    */
 180   struct GNUNET_TIME_Relative retransmit_timeout;
 181
 182   /**
 183    * The Acknowledgement Bitmap
 184    */
 185   GNUNET_STREAM_AckBitmap ack_bitmap;
 186
 187   /**
 188    * Time when the Acknowledgement was queued
 189    */
 190   struct GNUNET_TIME_Absolute ack_time_registered;
 191
 192   /**
 193    * Queued Acknowledgement deadline
 194    */
 195   struct GNUNET_TIME_Relative ack_time_deadline;
 196
 197   /**
 198    * The mesh handle
 199    */
 200   struct GNUNET_MESH_Handle *mesh;
 201
 202   /**
 203    * The mesh tunnel handle
 204    */
 205   struct GNUNET_MESH_Tunnel *tunnel;
 206
 207   /**
 208    * Stream open closure
 209    */
 210   void *open_cls;
 211
 212   /**
 213    * Stream open callback
 214    */
 215   GNUNET_STREAM_OpenCallback open_cb;
 216
 217   /**
 218    * The current transmit handle (if a pending transmit request exists)
 219    */
 220   struct GNUNET_MESH_TransmitHandle *transmit_handle;
 221
 222   /**
 223    * The current message associated with the transmit handle
 224    */
 225   struct MessageQueue *queue_head;
 226
 227   /**
 228    * The queue tail, should always point to the last message in queue
 229    */
 230   struct MessageQueue *queue_tail;
 231
 232   /**
 233    * The write IO_handle associated with this socket
 234    */
 235   struct GNUNET_STREAM_IOWriteHandle *write_handle;
 236
 237   /**
 238    * The read IO_handle associated with this socket
 239    */
 240   struct GNUNET_STREAM_IOReadHandle *read_handle;
 241
 242   /**
 243    * The shutdown handle associated with this socket
 244    */
 245   struct GNUNET_STREAM_ShutdownHandle *shutdown_handle;
 246
 247   /**
 248    * Buffer for storing received messages
 249    */
 250   void *receive_buffer;
 251
 252   /**
 253    * The listen socket from which this socket is derived. Should be NULL if it
 254    * is not a derived socket
 255    */
 256   struct GNUNET_STREAM_ListenSocket *lsocket;
 257
 258   /**
 259    * The peer identity of the peer at the other end of the stream
 260    */
 261   struct GNUNET_PeerIdentity other_peer;
 262
 263   /**
 264    * Task identifier for the read io timeout task
 265    */
 266   GNUNET_SCHEDULER_TaskIdentifier read_io_timeout_task_id;
 267
 268   /**
 269    * Task identifier for retransmission task after timeout
 270    */
 271   GNUNET_SCHEDULER_TaskIdentifier data_retransmission_task_id;
 272
 273   /**
 274    * Task identifier for retransmission of control messages
 275    */
 276   GNUNET_SCHEDULER_TaskIdentifier control_retransmission_task_id;
 277
 278   /**
 279    * The task for sending timely Acks
 280    */
 281   GNUNET_SCHEDULER_TaskIdentifier ack_task_id;
 282
 283   /**
 284    * Task scheduled to continue a read operation.
 285    */
 286   GNUNET_SCHEDULER_TaskIdentifier read_task_id;
 287
 288   /**
 289    * The state of the protocol associated with this socket
 290    */
 291   enum State state;
 292
 293   /**
 294    * The status of the socket
 295    */
 296   enum GNUNET_STREAM_Status status;
 297
 298   /**
 299    * The number of previous timeouts; FIXME: currently not used
 300    */
 301   unsigned int retries;
 302
 303   /**
 304    * The application port number (type: uint32_t)
 305    */
 306   GNUNET_MESH_ApplicationType app_port;
 307
 308   /**
 309    * Whether testing mode is active or not
 310    */
 311   int testing_active;
 312
 313   /**
 314    * The write sequence number to be set incase of testing
 315    */
 316   uint32_t testing_set_write_sequence_number_value;
 317
 318   /**
 319    * The session id associated with this stream connection
 320    * FIXME: Not used currently, may be removed
 321    */
 322   uint32_t session_id;
 323
 324   /**
 325    * Write sequence number. Set to random when sending HELLO(client) and
 326    * HELLO_ACK(server)
 327    */
 328   uint32_t write_sequence_number;
 329
 330   /**
 331    * Read sequence number. This number's value is determined during handshake
 332    */
 333   uint32_t read_sequence_number;
 334
 335   /**
 336    * The receiver buffer size
 337    */
 338   uint32_t receive_buffer_size;
 339
 340   /**
 341    * The receiver buffer boundaries
 342    */
 343   uint32_t receive_buffer_boundaries[GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH];
 344
 345   /**
 346    * receiver's available buffer after the last acknowledged packet
 347    */
 348   uint32_t receiver_window_available;
 349
 350   /**
 351    * The offset pointer used during write operation
 352    */
 353   uint32_t write_offset;
 354
 355   /**
 356    * The offset after which we are expecting data
 357    */
 358   uint32_t read_offset;
 359
 360   /**
 361    * The offset upto which user has read from the received buffer
 362    */
 363   uint32_t copy_offset;
 364
 365   /**
 366    * The maximum size of the data message payload this stream handle can send
 367    */
 368   uint16_t max_payload_size;
 369 };
 370
 371
 372 /**
 373  * A socket for listening
 374  */
 375 struct GNUNET_STREAM_ListenSocket
 376 {
 377   /**
 378    * The mesh handle
 379    */
 380   struct GNUNET_MESH_Handle *mesh;
 381
 382   /**
 383    * Our configuration
 384    */
 385   struct GNUNET_CONFIGURATION_Handle *cfg;
 386
 387   /**
 388    * Handle to the lock manager service
 389    */
 390   struct GNUNET_LOCKMANAGER_Handle *lockmanager;
 391
 392   /**
 393    * The active LockingRequest from lockmanager
 394    */
 395   struct GNUNET_LOCKMANAGER_LockingRequest *locking_request;
 396
 397   /**
 398    * Callback to call after acquring a lock and listening
 399    */
 400   GNUNET_STREAM_ListenSuccessCallback listen_ok_cb;
 401
 402   /**
 403    * The callback function which is called after successful opening socket
 404    */
 405   GNUNET_STREAM_ListenCallback listen_cb;
 406
 407   /**
 408    * The call back closure
 409    */
 410   void *listen_cb_cls;
 411
 412   /**
 413    * The service port
 414    */
 415   GNUNET_MESH_ApplicationType port;
 416
 417   /**
 418    * The id of the lockmanager timeout task
 419    */
 420   GNUNET_SCHEDULER_TaskIdentifier lockmanager_acquire_timeout_task;
 421
 422   /**
 423    * The retransmit timeout
 424    */
 425   struct GNUNET_TIME_Relative retransmit_timeout;
 426
 427   /**
 428    * Listen enabled?
 429    */
 430   int listening;
 431
 432   /**
 433    * Whether testing mode is active or not
 434    */
 435   int testing_active;
 436
 437   /**
 438    * The write sequence number to be set incase of testing
 439    */
 440   uint32_t testing_set_write_sequence_number_value;
 441
 442   /**
 443    * The maximum size of the data message payload this stream handle can send
 444    */
 445   uint16_t max_payload_size;
 446
 447 };
 448
 449
 450 /**
 451  * The IO Write Handle
 452  */
 453 struct GNUNET_STREAM_IOWriteHandle
 454 {
 455   /**
 456    * The socket to which this write handle is associated
 457    */
 458   struct GNUNET_STREAM_Socket *socket;
 459
 460   /**
 461    * The packet_buffers associated with this Handle
 462    */
 463   struct GNUNET_STREAM_DataMessage *messages[GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH];
 464
 465   /**
 466    * The write continuation callback
 467    */
 468   GNUNET_STREAM_CompletionContinuation write_cont;
 469
 470   /**
 471    * Write continuation closure
 472    */
 473   void *write_cont_cls;
 474
 475   /**
 476    * The bitmap of this IOHandle; Corresponding bit for a message is set when
 477    * it has been acknowledged by the receiver
 478    */
 479   GNUNET_STREAM_AckBitmap ack_bitmap;
 480
 481   /**
 482    * Number of bytes in this write handle
 483    */
 484   size_t size;
 485
 486   /**
 487    * Number of packets already transmitted from this IO handle. Retransmitted
 488    * packets are not taken into account here. This is used to determine which
 489    * packets account for retransmission and which packets occupy buffer space at
 490    * the receiver.
 491    */
 492   unsigned int packets_sent;
 493 };
 494
 495
 496 /**
 497  * The IO Read Handle
 498  */
 499 struct GNUNET_STREAM_IOReadHandle
 500 {
 501   /**
 502    * The socket to which this read handle is associated
 503    */
 504   struct GNUNET_STREAM_Socket *socket;
 505
 506   /**
 507    * Callback for the read processor
 508    */
 509   GNUNET_STREAM_DataProcessor proc;
 510
 511   /**
 512    * The closure pointer for the read processor callback
 513    */
 514   void *proc_cls;
 515 };
 516
 517
 518 /**
 519  * Handle for Shutdown
 520  */
 521 struct GNUNET_STREAM_ShutdownHandle
 522 {
 523   /**
 524    * The socket associated with this shutdown handle
 525    */
 526   struct GNUNET_STREAM_Socket *socket;
 527
 528   /**
 529    * Shutdown completion callback
 530    */
 531   GNUNET_STREAM_ShutdownCompletion completion_cb;
 532
 533   /**
 534    * Closure for completion callback
 535    */
 536   void *completion_cls;
 537
 538   /**
 539    * Close message retransmission task id
 540    */
 541   GNUNET_SCHEDULER_TaskIdentifier close_msg_retransmission_task_id;
 542
 543   /**
 544    * Which operation to shutdown? SHUT_RD, SHUT_WR or SHUT_RDWR
 545    */
 546   int operation;
 547 };
 548
 549
 550 /**
 551  * Default value in seconds for various timeouts
 552  */
 553 static const unsigned int default_timeout = 10;
 554
 555 /**
 556  * The domain name for locks we use here
 557  */
 558 static const char *locking_domain = "GNUNET_STREAM_APPLOCK";
 559
 560
 561 /**
 562  * Callback function for sending queued message
 563  *
 564  * @param cls closure the socket
 565  * @param size number of bytes available in buf
 566  * @param buf where the callee should write the message
 567  * @return number of bytes written to buf
 568  */
 569 static size_t
 570 send_message_notify (void *cls, size_t size, void *buf)
 571 {
 572   struct GNUNET_STREAM_Socket *socket = cls;
 573   struct MessageQueue *head;
 574   size_t ret;
 575
 576   socket->transmit_handle = NULL; /* Remove the transmit handle */
 577   head = socket->queue_head;
 578   if (NULL == head)
 579     return 0; /* just to be safe */
 580   if (0 == size)                /* request timed out */
 581   {
 582     socket->retries++;
 583     LOG (GNUNET_ERROR_TYPE_DEBUG,
 584          "%s: Message sending timed out. Retry %d \n",
 585          GNUNET_i2s (&socket->other_peer),
 586          socket->retries);
 587     socket->transmit_handle =
 588       GNUNET_MESH_notify_transmit_ready (socket->tunnel,
 589                                          GNUNET_NO, /* Corking */
 590                                          /* FIXME: exponential backoff */
 591                                          socket->retransmit_timeout,
 592                                          &socket->other_peer,
 593                                          ntohs (head->message->header.size),
 594                                          &send_message_notify,
 595                                          socket);
 596     return 0;
 597   }
 598   ret = ntohs (head->message->header.size);
 599   GNUNET_assert (size >= ret);
 600   memcpy (buf, head->message, ret);
 601   if (NULL != head->finish_cb)
 602   {
 603     head->finish_cb (head->finish_cb_cls, socket);
 604   }
 605   GNUNET_CONTAINER_DLL_remove (socket->queue_head,
 606                                socket->queue_tail,
 607                                head);
 608   GNUNET_free (head->message);
 609   GNUNET_free (head);
 610   head = socket->queue_head;
 611   if (NULL != head)    /* more pending messages to send */
 612   {
 613     socket->retries = 0;
 614     socket->transmit_handle =
 615       GNUNET_MESH_notify_transmit_ready (socket->tunnel,
 616                                          GNUNET_NO, /* Corking */
 617                                          /* FIXME: exponential backoff */
 618                                          socket->retransmit_timeout,
 619                                          &socket->other_peer,
 620                                          ntohs (head->message->header.size),
 621                                          &send_message_notify,
 622                                          socket);
 623   }
 624   return ret;
 625 }
 626
 627
 628 /**
 629  * Queues a message for sending using the mesh connection of a socket
 630  *
 631  * @param socket the socket whose mesh connection is used
 632  * @param message the message to be sent
 633  * @param finish_cb the callback to be called when the message is sent
 634  * @param finish_cb_cls the closure for the callback
 635  * @param urgent set to GNUNET_YES to add the message to the beginning of the
 636  *          queue; GNUNET_NO to add at the tail
 637  */
 638 static void
 639 queue_message (struct GNUNET_STREAM_Socket *socket,
 640                struct GNUNET_STREAM_MessageHeader *message,
 641                SendFinishCallback finish_cb,
 642                void *finish_cb_cls,
 643                int urgent)
 644 {
 645   struct MessageQueue *queue_entity;
 646
 647   GNUNET_assert
 648     ((ntohs (message->header.type) >= GNUNET_MESSAGE_TYPE_STREAM_DATA)
 649      && (ntohs (message->header.type) <= GNUNET_MESSAGE_TYPE_STREAM_CLOSE_ACK));
 650   LOG (GNUNET_ERROR_TYPE_DEBUG,
 651        "%s: Queueing message of type %d and size %d\n",
 652        GNUNET_i2s (&socket->other_peer),
 653        ntohs (message->header.type),
 654        ntohs (message->header.size));
 655   GNUNET_assert (NULL != message);
 656   queue_entity = GNUNET_malloc (sizeof (struct MessageQueue));
 657   queue_entity->message = message;
 658   queue_entity->finish_cb = finish_cb;
 659   queue_entity->finish_cb_cls = finish_cb_cls;
 660   if (GNUNET_YES == urgent)
 661   {
 662     GNUNET_CONTAINER_DLL_insert (socket->queue_head, socket->queue_tail,
 663                                  queue_entity);
 664     if (NULL != socket->transmit_handle)
 665     {
 666       GNUNET_MESH_notify_transmit_ready_cancel (socket->transmit_handle);
 667       socket->transmit_handle = NULL;
 668     }
 669   }
 670   else
 671     GNUNET_CONTAINER_DLL_insert_tail (socket->queue_head,
 672                                       socket->queue_tail,
 673                                       queue_entity);
 674   if (NULL == socket->transmit_handle)
 675   {
 676     socket->retries = 0;
 677     socket->transmit_handle =
 678       GNUNET_MESH_notify_transmit_ready (socket->tunnel,
 679                                          GNUNET_NO, /* Corking */
 680                                          socket->retransmit_timeout,
 681                                          &socket->other_peer,
 682                                          ntohs (message->header.size),
 683                                          &send_message_notify,
 684                                          socket);
 685   }
 686 }
 687
 688
 689 /**
 690  * Copies a message and queues it for sending using the mesh connection of
 691  * given socket
 692  *
 693  * @param socket the socket whose mesh connection is used
 694  * @param message the message to be sent
 695  * @param finish_cb the callback to be called when the message is sent
 696  * @param finish_cb_cls the closure for the callback
 697  */
 698 static void
 699 copy_and_queue_message (struct GNUNET_STREAM_Socket *socket,
 700                         const struct GNUNET_STREAM_MessageHeader *message,
 701                         SendFinishCallback finish_cb,
 702                         void *finish_cb_cls)
 703 {
 704   struct GNUNET_STREAM_MessageHeader *msg_copy;
 705   uint16_t size;
 706
 707   size = ntohs (message->header.size);
 708   msg_copy = GNUNET_malloc (size);
 709   memcpy (msg_copy, message, size);
 710   queue_message (socket, msg_copy, finish_cb, finish_cb_cls, GNUNET_NO);
 711 }
 712
 713
 714 /**
 715  * Writes data using the given socket. The amount of data written is limited by
 716  * the receiver_window_size
 717  *
 718  * @param socket the socket to use
 719  */
 720 static void
 721 write_data (struct GNUNET_STREAM_Socket *socket);
 722
 723
 724 /**
 725  * Task for retransmitting data messages if they aren't ACK before their ack
 726  * deadline
 727  *
 728  * @param cls the socket
 729  * @param tc the Task context
 730  */
 731 static void
 732 data_retransmission_task (void *cls,
 733                           const struct GNUNET_SCHEDULER_TaskContext *tc)
 734 {
 735   struct GNUNET_STREAM_Socket *socket = cls;
 736
 737   if (GNUNET_SCHEDULER_REASON_SHUTDOWN == tc->reason)
 738     return;
 739   LOG (GNUNET_ERROR_TYPE_DEBUG,
 740        "%s: Retransmitting DATA...\n", GNUNET_i2s (&socket->other_peer));
 741   socket->data_retransmission_task_id = GNUNET_SCHEDULER_NO_TASK;
 742   write_data (socket);
 743 }
 744
 745
 746 /**
 747  * Task for sending ACK message
 748  *
 749  * @param cls the socket
 750  * @param tc the Task context
 751  */
 752 static void
 753 ack_task (void *cls,
 754           const struct GNUNET_SCHEDULER_TaskContext *tc)
 755 {
 756   struct GNUNET_STREAM_Socket *socket = cls;
 757   struct GNUNET_STREAM_AckMessage *ack_msg;
 758
 759   if (GNUNET_SCHEDULER_REASON_SHUTDOWN == tc->reason)
 760   {
 761     return;
 762   }
 763   socket->ack_task_id = GNUNET_SCHEDULER_NO_TASK;
 764   /* Create the ACK Message */
 765   ack_msg = GNUNET_malloc (sizeof (struct GNUNET_STREAM_AckMessage));
 766   ack_msg->header.header.size = htons (sizeof (struct
 767                                                GNUNET_STREAM_AckMessage));
 768   ack_msg->header.header.type = htons (GNUNET_MESSAGE_TYPE_STREAM_ACK);
 769   ack_msg->bitmap = GNUNET_htonll (socket->ack_bitmap);
 770   ack_msg->base_sequence_number = htonl (socket->read_sequence_number);
 771   ack_msg->receive_window_remaining =
 772     htonl (RECEIVE_BUFFER_SIZE - socket->receive_buffer_size);
 773   /* Queue up ACK for immediate sending */
 774   queue_message (socket, &ack_msg->header, NULL, NULL, GNUNET_YES);
 775 }
 776
 777
 778 /**
 779  * Retransmission task for shutdown messages
 780  *
 781  * @param cls the shutdown handle
 782  * @param tc the Task Context
 783  */
 784 static void
 785 close_msg_retransmission_task (void *cls,
 786                                const struct GNUNET_SCHEDULER_TaskContext *tc)
 787 {
 788   struct GNUNET_STREAM_ShutdownHandle *shutdown_handle = cls;
 789   struct GNUNET_STREAM_MessageHeader *msg;
 790   struct GNUNET_STREAM_Socket *socket;
 791
 792   GNUNET_assert (NULL != shutdown_handle);
 793   socket = shutdown_handle->socket;
 794
 795   msg = GNUNET_malloc (sizeof (struct GNUNET_STREAM_MessageHeader));
 796   msg->header.size = htons (sizeof (struct GNUNET_STREAM_MessageHeader));
 797   switch (shutdown_handle->operation)
 798   {
 799   case SHUT_RDWR:
 800     msg->header.type = htons (GNUNET_MESSAGE_TYPE_STREAM_CLOSE);
 801     break;
 802   case SHUT_RD:
 803     msg->header.type = htons (GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE);
 804     break;
 805   case SHUT_WR:
 806     msg->header.type = htons (GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE);
 807     break;
 808   default:
 809     GNUNET_free (msg);
 810     shutdown_handle->close_msg_retransmission_task_id =
 811       GNUNET_SCHEDULER_NO_TASK;
 812     return;
 813   }
 814   queue_message (socket, msg, NULL, NULL, GNUNET_NO);
 815   shutdown_handle->close_msg_retransmission_task_id =
 816     GNUNET_SCHEDULER_add_delayed (socket->retransmit_timeout,
 817                                   &close_msg_retransmission_task,
 818                                   shutdown_handle);
 819 }
 820
 821
 822 /**
 823  * Function to modify a bit in GNUNET_STREAM_AckBitmap
 824  *
 825  * @param bitmap the bitmap to modify
 826  * @param bit the bit number to modify
 827  * @param value GNUNET_YES to on, GNUNET_NO to off
 828  */
 829 static void
 830 ackbitmap_modify_bit (GNUNET_STREAM_AckBitmap *bitmap,
 831                       unsigned int bit,
 832                       int value)
 833 {
 834   GNUNET_assert (bit < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH);
 835   if (GNUNET_YES == value)
 836     *bitmap |= (1LL << bit);
 837   else
 838     *bitmap &= ~(1LL << bit);
 839 }
 840
 841
 842 /**
 843  * Function to check if a bit is set in the GNUNET_STREAM_AckBitmap
 844  *
 845  * @param bitmap address of the bitmap that has to be checked
 846  * @param bit the bit number to check
 847  * @return GNUNET_YES if the bit is set; GNUNET_NO if not
 848  */
 849 static uint8_t
 850 ackbitmap_is_bit_set (const GNUNET_STREAM_AckBitmap *bitmap,
 851                       unsigned int bit)
 852 {
 853   GNUNET_assert (bit < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH);
 854   return 0 != (*bitmap & (1LL << bit));
 855 }
 856
 857
 858 /**
 859  * Writes data using the given socket. The amount of data written is limited by
 860  * the receiver_window_size
 861  *
 862  * @param socket the socket to use
 863  */
 864 static void
 865 write_data (struct GNUNET_STREAM_Socket *socket)
 866 {
 867   struct GNUNET_STREAM_IOWriteHandle *io_handle = socket->write_handle;
 868   unsigned int packet;
 869
 870   for (packet=0; packet < io_handle->packets_sent; packet++)
 871   {
 872     if (GNUNET_NO == ackbitmap_is_bit_set (&io_handle->ack_bitmap,
 873                                            packet))
 874     {
 875       LOG (GNUNET_ERROR_TYPE_DEBUG,
 876            "%s: Retransmitting DATA message with sequence %u\n",
 877            GNUNET_i2s (&socket->other_peer),
 878            ntohl (io_handle->messages[packet]->sequence_number));
 879       copy_and_queue_message (socket,
 880                               &io_handle->messages[packet]->header,
 881                               NULL,
 882                               NULL);
 883     }
 884   }
 885   /* Now send new packets if there is enough buffer space */
 886   while ( (NULL != io_handle->messages[packet]) &&
 887           (socket->receiver_window_available
 888            >= ntohs (io_handle->messages[packet]->header.header.size)) &&
 889           (packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH))
 890   {
 891     socket->receiver_window_available -=
 892       ntohs (io_handle->messages[packet]->header.header.size);
 893     LOG (GNUNET_ERROR_TYPE_DEBUG,
 894          "%s: Placing DATA message with sequence %u in send queue\n",
 895          GNUNET_i2s (&socket->other_peer),
 896          ntohl (io_handle->messages[packet]->sequence_number));
 897     copy_and_queue_message (socket,
 898                             &io_handle->messages[packet]->header,
 899                             NULL,
 900                             NULL);
 901     packet++;
 902   }
 903   io_handle->packets_sent = packet;
 904   if (GNUNET_SCHEDULER_NO_TASK == socket->data_retransmission_task_id)
 905     socket->data_retransmission_task_id =
 906       GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
 907                                     (GNUNET_TIME_UNIT_SECONDS, 8),
 908                                     &data_retransmission_task,
 909                                     socket);
 910 }
 911
 912
 913 /**
 914  * Task for calling the read processor
 915  *
 916  * @param cls the socket
 917  * @param tc the task context
 918  */
 919 static void
 920 call_read_processor (void *cls,
 921                      const struct GNUNET_SCHEDULER_TaskContext *tc)
 922 {
 923   struct GNUNET_STREAM_Socket *socket = cls;
 924   size_t read_size;
 925   size_t valid_read_size;
 926   unsigned int packet;
 927   uint32_t sequence_increase;
 928   uint32_t offset_increase;
 929
 930   socket->read_task_id = GNUNET_SCHEDULER_NO_TASK;
 931   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
 932     return;
 933
 934   if (NULL == socket->receive_buffer)
 935     return;
 936
 937   GNUNET_assert (NULL != socket->read_handle);
 938   GNUNET_assert (NULL != socket->read_handle->proc);
 939
 940   /* Check the bitmap for any holes */
 941   for (packet=0; packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH; packet++)
 942   {
 943     if (GNUNET_NO == ackbitmap_is_bit_set (&socket->ack_bitmap,
 944                                            packet))
 945       break;
 946   }
 947   /* We only call read processor if we have the first packet */
 948   GNUNET_assert (0 < packet);
 949   valid_read_size =
 950     socket->receive_buffer_boundaries[packet-1] - socket->copy_offset;
 951   GNUNET_assert (0 != valid_read_size);
 952   /* Cancel the read_io_timeout_task */
 953   GNUNET_SCHEDULER_cancel (socket->read_io_timeout_task_id);
 954   socket->read_io_timeout_task_id = GNUNET_SCHEDULER_NO_TASK;
 955   /* Call the data processor */
 956   LOG (GNUNET_ERROR_TYPE_DEBUG,
 957        "%s: Calling read processor\n",
 958        GNUNET_i2s (&socket->other_peer));
 959   read_size =
 960     socket->read_handle->proc (socket->read_handle->proc_cls,
 961                                socket->status,
 962                                socket->receive_buffer + socket->copy_offset,
 963                                valid_read_size);
 964   LOG (GNUNET_ERROR_TYPE_DEBUG,
 965        "%s: Read processor read %d bytes\n",
 966        GNUNET_i2s (&socket->other_peer), read_size);
 967   LOG (GNUNET_ERROR_TYPE_DEBUG,
 968        "%s: Read processor completed successfully\n",
 969        GNUNET_i2s (&socket->other_peer));
 970   /* Free the read handle */
 971   GNUNET_free (socket->read_handle);
 972   socket->read_handle = NULL;
 973   GNUNET_assert (read_size <= valid_read_size);
 974   socket->copy_offset += read_size;
 975   /* Determine upto which packet we can remove from the buffer */
 976   for (packet = 0; packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH; packet++)
 977   {
 978     if (socket->copy_offset == socket->receive_buffer_boundaries[packet])
 979     { packet++; break; }
 980     if (socket->copy_offset < socket->receive_buffer_boundaries[packet])
 981       break;
 982   }
 983
 984   /* If no packets can be removed we can't move the buffer */
 985   if (0 == packet) return;
 986   sequence_increase = packet;
 987   LOG (GNUNET_ERROR_TYPE_DEBUG,
 988        "%s: Sequence increase after read processor completion: %u\n",
 989        GNUNET_i2s (&socket->other_peer), sequence_increase);
 990
 991   /* Shift the data in the receive buffer */
 992   socket->receive_buffer =
 993     memmove (socket->receive_buffer,
 994              socket->receive_buffer
 995              + socket->receive_buffer_boundaries[sequence_increase-1],
 996              socket->receive_buffer_size
 997              - socket->receive_buffer_boundaries[sequence_increase-1]);
 998   /* Shift the bitmap */
 999   socket->ack_bitmap = socket->ack_bitmap >> sequence_increase;
1000   /* Set read_sequence_number */
1001   socket->read_sequence_number += sequence_increase;
1002   /* Set read_offset */
1003   offset_increase = socket->receive_buffer_boundaries[sequence_increase-1];
1004   socket->read_offset += offset_increase;
1005   /* Fix copy_offset */
1006   GNUNET_assert (offset_increase <= socket->copy_offset);
1007   socket->copy_offset -= offset_increase;
1008   /* Fix relative boundaries */
1009   for (packet=0; packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH; packet++)
1010   {
1011     if (packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH - sequence_increase)
1012     {
1013       uint32_t ahead_buffer_boundary;
1014
1015       ahead_buffer_boundary =
1016         socket->receive_buffer_boundaries[packet + sequence_increase];
1017       if (0 == ahead_buffer_boundary)
1018         socket->receive_buffer_boundaries[packet] = 0;
1019       else
1020       {
1021         GNUNET_assert (offset_increase < ahead_buffer_boundary);
1022         socket->receive_buffer_boundaries[packet] =
1023           ahead_buffer_boundary - offset_increase;
1024       }
1025     }
1026     else
1027       socket->receive_buffer_boundaries[packet] = 0;
1028   }
1029 }
1030
1031
1032 /**
1033  * Cancels the existing read io handle
1034  *
1035  * @param cls the closure from the SCHEDULER call
1036  * @param tc the task context
1037  */
1038 static void
1039 read_io_timeout (void *cls,
1040                  const struct GNUNET_SCHEDULER_TaskContext *tc)
1041 {
1042   struct GNUNET_STREAM_Socket *socket = cls;
1043   GNUNET_STREAM_DataProcessor proc;
1044   void *proc_cls;
1045
1046   socket->read_io_timeout_task_id = GNUNET_SCHEDULER_NO_TASK;
1047   if (socket->read_task_id != GNUNET_SCHEDULER_NO_TASK)
1048   {
1049     LOG (GNUNET_ERROR_TYPE_DEBUG,
1050          "%s: Read task timedout - Cancelling it\n",
1051          GNUNET_i2s (&socket->other_peer));
1052     GNUNET_SCHEDULER_cancel (socket->read_task_id);
1053     socket->read_task_id = GNUNET_SCHEDULER_NO_TASK;
1054   }
1055   GNUNET_assert (NULL != socket->read_handle);
1056   proc = socket->read_handle->proc;
1057   proc_cls = socket->read_handle->proc_cls;
1058   GNUNET_free (socket->read_handle);
1059   socket->read_handle = NULL;
1060   /* Call the read processor to signal timeout */
1061   proc (proc_cls,
1062         GNUNET_STREAM_TIMEOUT,
1063         NULL,
1064         0);
1065 }
1066
1067
1068 /**
1069  * Handler for DATA messages; Same for both client and server
1070  *
1071  * @param socket the socket through which the ack was received
1072  * @param tunnel connection to the other end
1073  * @param sender who sent the message
1074  * @param msg the data message
1075  * @param atsi performance data for the connection
1076  * @return GNUNET_OK to keep the connection open,
1077  *         GNUNET_SYSERR to close it (signal serious error)
1078  */
1079 static int
1080 handle_data (struct GNUNET_STREAM_Socket *socket,
1081              struct GNUNET_MESH_Tunnel *tunnel,
1082              const struct GNUNET_PeerIdentity *sender,
1083              const struct GNUNET_STREAM_DataMessage *msg,
1084              const struct GNUNET_ATS_Information*atsi)
1085 {
1086   const void *payload;
1087   struct GNUNET_TIME_Relative ack_deadline_rel;
1088   uint32_t bytes_needed;
1089   uint32_t relative_offset;
1090   uint32_t relative_sequence_number;
1091   uint16_t size;
1092
1093   size = htons (msg->header.header.size);
1094   if (size < sizeof (struct GNUNET_STREAM_DataMessage))
1095   {
1096     GNUNET_break_op (0);
1097     return GNUNET_SYSERR;
1098   }
1099   if (0 != memcmp (sender, &socket->other_peer,
1100                    sizeof (struct GNUNET_PeerIdentity)))
1101   {
1102     LOG (GNUNET_ERROR_TYPE_DEBUG,
1103          "%s: Received DATA from non-confirming peer\n",
1104          GNUNET_i2s (&socket->other_peer));
1105     return GNUNET_YES;
1106   }
1107   switch (socket->state)
1108   {
1109   case STATE_ESTABLISHED:
1110   case STATE_TRANSMIT_CLOSED:
1111   case STATE_TRANSMIT_CLOSE_WAIT:
1112     /* check if the message's sequence number is in the range we are
1113        expecting */
1114     relative_sequence_number =
1115       ntohl (msg->sequence_number) - socket->read_sequence_number;
1116     if ( relative_sequence_number > GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH)
1117     {
1118       LOG (GNUNET_ERROR_TYPE_DEBUG,
1119            "%s: Ignoring received message with sequence number %u\n",
1120            GNUNET_i2s (&socket->other_peer),
1121            ntohl (msg->sequence_number));
1122       /* Start ACK sending task if one is not already present */
1123       if (GNUNET_SCHEDULER_NO_TASK == socket->ack_task_id)
1124       {
1125         socket->ack_task_id =
1126           GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_ntoh
1127                                         (msg->ack_deadline),
1128                                         &ack_task,
1129                                         socket);
1130       }
1131       return GNUNET_YES;
1132     }
1133     /* Check if we have already seen this message */
1134     if (GNUNET_YES == ackbitmap_is_bit_set (&socket->ack_bitmap,
1135                                             relative_sequence_number))
1136     {
1137       LOG (GNUNET_ERROR_TYPE_DEBUG,
1138            "%s: Ignoring already received message with sequence number %u\n",
1139            GNUNET_i2s (&socket->other_peer),
1140            ntohl (msg->sequence_number));
1141       /* Start ACK sending task if one is not already present */
1142       if (GNUNET_SCHEDULER_NO_TASK == socket->ack_task_id)
1143       {
1144         socket->ack_task_id =
1145           GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_ntoh
1146                                         (msg->ack_deadline), &ack_task, socket);
1147       }
1148       return GNUNET_YES;
1149     }
1150     LOG (GNUNET_ERROR_TYPE_DEBUG,
1151          "%s: Receiving DATA with sequence number: %u and size: %d from %s\n",
1152          GNUNET_i2s (&socket->other_peer), ntohl (msg->sequence_number),
1153          ntohs (msg->header.header.size), GNUNET_i2s (&socket->other_peer));
1154     /* Check if we have to allocate the buffer */
1155     size -= sizeof (struct GNUNET_STREAM_DataMessage);
1156     relative_offset = ntohl (msg->offset) - socket->read_offset;
1157     bytes_needed = relative_offset + size;
1158     if (bytes_needed > socket->receive_buffer_size)
1159     {
1160       if (bytes_needed <= RECEIVE_BUFFER_SIZE)
1161       {
1162         socket->receive_buffer = GNUNET_realloc (socket->receive_buffer,
1163                                                  bytes_needed);
1164         socket->receive_buffer_size = bytes_needed;
1165       }
1166       else
1167       {
1168         LOG (GNUNET_ERROR_TYPE_DEBUG,
1169              "%s: Cannot accommodate packet %d as buffer is full\n",
1170              GNUNET_i2s (&socket->other_peer), ntohl (msg->sequence_number));
1171         return GNUNET_YES;
1172       }
1173     }
1174     /* Copy Data to buffer */
1175     payload = &msg[1];
1176     GNUNET_assert (relative_offset + size <= socket->receive_buffer_size);
1177     memcpy (socket->receive_buffer + relative_offset, payload, size);
1178     socket->receive_buffer_boundaries[relative_sequence_number] =
1179         relative_offset + size;
1180     /* Modify the ACK bitmap */
1181     ackbitmap_modify_bit (&socket->ack_bitmap, relative_sequence_number,
1182                           GNUNET_YES);
1183     /* Start ACK sending task if one is not already present */
1184     ack_deadline_rel = GNUNET_TIME_relative_ntoh (msg->ack_deadline);
1185     if (GNUNET_SCHEDULER_NO_TASK == socket->ack_task_id)
1186     {
1187       ack_deadline_rel =
1188           GNUNET_TIME_relative_min (ack_deadline_rel,
1189                                     GNUNET_TIME_relative_multiply
1190                                     (GNUNET_TIME_UNIT_SECONDS, 300));
1191       socket->ack_task_id =
1192           GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_ntoh
1193                                         (msg->ack_deadline), &ack_task, socket);
1194       socket->ack_time_registered = GNUNET_TIME_absolute_get ();
1195       socket->ack_time_deadline = ack_deadline_rel;
1196     }
1197     else
1198     {
1199       struct GNUNET_TIME_Relative ack_time_past;
1200       struct GNUNET_TIME_Relative ack_time_remaining;
1201       struct GNUNET_TIME_Relative ack_time_min;
1202       ack_time_past =
1203           GNUNET_TIME_absolute_get_duration (socket->ack_time_registered);
1204       ack_time_remaining = GNUNET_TIME_relative_subtract
1205           (socket->ack_time_deadline, ack_time_past);
1206       ack_time_min = GNUNET_TIME_relative_min (ack_time_remaining,
1207                                                ack_deadline_rel);
1208       if (0 == memcmp(&ack_deadline_rel, &ack_time_min,
1209                       sizeof (struct GNUNET_TIME_Relative)))
1210       {
1211         ack_deadline_rel = ack_time_min;
1212         GNUNET_SCHEDULER_cancel (socket->ack_task_id);
1213         socket->ack_task_id = GNUNET_SCHEDULER_add_delayed (ack_deadline_rel,
1214                                                             &ack_task, socket);
1215         socket->ack_time_registered = GNUNET_TIME_absolute_get ();
1216         socket->ack_time_deadline = ack_deadline_rel;
1217       }
1218     }
1219     if ((NULL != socket->read_handle) /* A read handle is waiting */
1220         /* There is no current read task */
1221         && (GNUNET_SCHEDULER_NO_TASK == socket->read_task_id)
1222         /* We have the first packet */
1223         && (GNUNET_YES == ackbitmap_is_bit_set(&socket->ack_bitmap, 0)))
1224     {
1225       LOG (GNUNET_ERROR_TYPE_DEBUG, "%s: Scheduling read processor\n",
1226            GNUNET_i2s (&socket->other_peer));
1227       socket->read_task_id = GNUNET_SCHEDULER_add_now (&call_read_processor,
1228                                                        socket);
1229     }
1230     break;
1231   default:
1232     LOG (GNUNET_ERROR_TYPE_DEBUG,
1233          "%s: Received data message when it cannot be handled\n",
1234          GNUNET_i2s (&socket->other_peer));
1235     break;
1236   }
1237   return GNUNET_YES;
1238 }
1239
1240
1241 /**
1242  * Client's message Handler for GNUNET_MESSAGE_TYPE_STREAM_DATA
1243  *
1244  * @param cls the socket (set from GNUNET_MESH_connect)
1245  * @param tunnel connection to the other end
1246  * @param tunnel_ctx place to store local state associated with the tunnel
1247  * @param sender who sent the message
1248  * @param message the actual message
1249  * @param atsi performance data for the connection
1250  * @return GNUNET_OK to keep the connection open,
1251  *         GNUNET_SYSERR to close it (signal serious error)
1252  */
1253 static int
1254 client_handle_data (void *cls,
1255                     struct GNUNET_MESH_Tunnel *tunnel,
1256                     void **tunnel_ctx,
1257                     const struct GNUNET_PeerIdentity *sender,
1258                     const struct GNUNET_MessageHeader *message,
1259                     const struct GNUNET_ATS_Information*atsi)
1260 {
1261   struct GNUNET_STREAM_Socket *socket = cls;
1262
1263   return handle_data (socket, tunnel, sender,
1264                       (const struct GNUNET_STREAM_DataMessage *) message, atsi);
1265 }
1266
1267
1268 /**
1269  * Callback to set state to ESTABLISHED
1270  *
1271  * @param cls the closure NULL;
1272  * @param socket the socket to requiring state change
1273  */
1274 static void
1275 set_state_established (void *cls,
1276                        struct GNUNET_STREAM_Socket *socket)
1277 {
1278   LOG (GNUNET_ERROR_TYPE_DEBUG,
1279        "%s: Attaining ESTABLISHED state\n",
1280        GNUNET_i2s (&socket->other_peer));
1281   socket->write_offset = 0;
1282   socket->read_offset = 0;
1283   socket->state = STATE_ESTABLISHED;
1284   GNUNET_assert (GNUNET_SCHEDULER_NO_TASK !=
1285                  socket->control_retransmission_task_id);
1286   GNUNET_SCHEDULER_cancel (socket->control_retransmission_task_id);
1287   socket->control_retransmission_task_id = GNUNET_SCHEDULER_NO_TASK;
1288   if (NULL != socket->lsocket)
1289   {
1290     LOG (GNUNET_ERROR_TYPE_DEBUG,
1291          "%s: Calling listen callback\n",
1292          GNUNET_i2s (&socket->other_peer));
1293     if (GNUNET_SYSERR ==
1294         socket->lsocket->listen_cb (socket->lsocket->listen_cb_cls,
1295                                     socket,
1296                                     &socket->other_peer))
1297     {
1298       socket->state = STATE_CLOSED;
1299       /* FIXME: We should close in a decent way (send RST) */
1300       GNUNET_MESH_tunnel_destroy (socket->tunnel); /* Destroy the tunnel */
1301       GNUNET_free (socket);
1302     }
1303   }
1304   else
1305     socket->open_cb (socket->open_cls, socket);
1306 }
1307
1308
1309 /**
1310  * Callback to set state to HELLO_WAIT
1311  *
1312  * @param cls the closure from queue_message
1313  * @param socket the socket to requiring state change
1314  */
1315 static void
1316 set_state_hello_wait (void *cls,
1317                       struct GNUNET_STREAM_Socket *socket)
1318 {
1319   GNUNET_assert (STATE_INIT == socket->state);
1320   LOG (GNUNET_ERROR_TYPE_DEBUG,
1321        "%s: Attaining HELLO_WAIT state\n",
1322        GNUNET_i2s (&socket->other_peer));
1323   socket->state = STATE_HELLO_WAIT;
1324 }
1325
1326
1327 /**
1328  * Callback to set state to CLOSE_WAIT
1329  *
1330  * @param cls the closure from queue_message
1331  * @param socket the socket requiring state change
1332  */
1333 static void
1334 set_state_close_wait (void *cls,
1335                       struct GNUNET_STREAM_Socket *socket)
1336 {
1337   LOG (GNUNET_ERROR_TYPE_DEBUG,
1338        "%s: Attaing CLOSE_WAIT state\n",
1339        GNUNET_i2s (&socket->other_peer));
1340   socket->state = STATE_CLOSE_WAIT;
1341   GNUNET_free_non_null (socket->receive_buffer); /* Free the receive buffer */
1342   socket->receive_buffer = NULL;
1343   socket->receive_buffer_size = 0;
1344 }
1345
1346
1347 /**
1348  * Callback to set state to RECEIVE_CLOSE_WAIT
1349  *
1350  * @param cls the closure from queue_message
1351  * @param socket the socket requiring state change
1352  */
1353 static void
1354 set_state_receive_close_wait (void *cls,
1355                               struct GNUNET_STREAM_Socket *socket)
1356 {
1357   LOG (GNUNET_ERROR_TYPE_DEBUG,
1358        "%s: Attaing RECEIVE_CLOSE_WAIT state\n",
1359        GNUNET_i2s (&socket->other_peer));
1360   socket->state = STATE_RECEIVE_CLOSE_WAIT;
1361   GNUNET_free_non_null (socket->receive_buffer); /* Free the receive buffer */
1362   socket->receive_buffer = NULL;
1363   socket->receive_buffer_size = 0;
1364 }
1365
1366
1367 /**
1368  * Callback to set state to TRANSMIT_CLOSE_WAIT
1369  *
1370  * @param cls the closure from queue_message
1371  * @param socket the socket requiring state change
1372  */
1373 static void
1374 set_state_transmit_close_wait (void *cls,
1375                                struct GNUNET_STREAM_Socket *socket)
1376 {
1377   LOG (GNUNET_ERROR_TYPE_DEBUG,
1378        "%s: Attaing TRANSMIT_CLOSE_WAIT state\n",
1379        GNUNET_i2s (&socket->other_peer));
1380   socket->state = STATE_TRANSMIT_CLOSE_WAIT;
1381 }
1382
1383
1384 /**
1385  * Callback to set state to CLOSED
1386  *
1387  * @param cls the closure from queue_message
1388  * @param socket the socket requiring state change
1389  */
1390 static void
1391 set_state_closed (void *cls,
1392                   struct GNUNET_STREAM_Socket *socket)
1393 {
1394   socket->state = STATE_CLOSED;
1395 }
1396
1397
1398 /**
1399  * Returns GNUNET_MESSAGE_TYPE_STREAM_HELLO
1400  *
1401  * @return the generate hello message
1402  */
1403 static struct GNUNET_STREAM_MessageHeader *
1404 generate_hello (void)
1405 {
1406   struct GNUNET_STREAM_MessageHeader *msg;
1407
1408   msg = GNUNET_malloc (sizeof (struct GNUNET_STREAM_MessageHeader));
1409   msg->header.type = htons (GNUNET_MESSAGE_TYPE_STREAM_HELLO);
1410   msg->header.size = htons (sizeof (struct GNUNET_STREAM_MessageHeader));
1411   return msg;
1412 }
1413
1414
1415 /**
1416  * Returns a new HelloAckMessage. Also sets the write sequence number for the
1417  * socket
1418  *
1419  * @param socket the socket for which this HelloAckMessage has to be generated
1420  * @param generate_seq GNUNET_YES to generate the write sequence number,
1421  *          GNUNET_NO to use the existing sequence number
1422  * @return the HelloAckMessage
1423  */
1424 static struct GNUNET_STREAM_HelloAckMessage *
1425 generate_hello_ack (struct GNUNET_STREAM_Socket *socket,
1426                     int generate_seq)
1427 {
1428   struct GNUNET_STREAM_HelloAckMessage *msg;
1429
1430   if (GNUNET_YES == generate_seq)
1431   {
1432     if (GNUNET_YES == socket->testing_active)
1433       socket->write_sequence_number =
1434         socket->testing_set_write_sequence_number_value;
1435     else
1436       socket->write_sequence_number =
1437         GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_NONCE, UINT32_MAX);
1438     LOG_DEBUG ("%s: write sequence number %u\n",
1439                GNUNET_i2s (&socket->other_peer),
1440                (unsigned int) socket->write_sequence_number);
1441   }
1442   msg = GNUNET_malloc (sizeof (struct GNUNET_STREAM_HelloAckMessage));
1443   msg->header.header.size =
1444     htons (sizeof (struct GNUNET_STREAM_HelloAckMessage));
1445   msg->header.header.type = htons (GNUNET_MESSAGE_TYPE_STREAM_HELLO_ACK);
1446   msg->sequence_number = htonl (socket->write_sequence_number);
1447   msg->receiver_window_size = htonl (RECEIVE_BUFFER_SIZE);
1448   return msg;
1449 }
1450
1451
1452 /**
1453  * Task for retransmitting control messages if they aren't ACK'ed before a
1454  * deadline
1455  *
1456  * @param cls the socket
1457  * @param tc the Task context
1458  */
1459 static void
1460 control_retransmission_task (void *cls,
1461                              const struct GNUNET_SCHEDULER_TaskContext *tc)
1462 {
1463   struct GNUNET_STREAM_Socket *socket = cls;
1464
1465   if (GNUNET_SCHEDULER_REASON_SHUTDOWN == tc->reason)
1466     return;
1467   socket->control_retransmission_task_id = GNUNET_SCHEDULER_NO_TASK;
1468   LOG_DEBUG ("%s: Retransmitting a control message\n",
1469                  GNUNET_i2s (&socket->other_peer));
1470   switch (socket->state)
1471   {
1472   case STATE_INIT:
1473     GNUNET_break (0);
1474     break;
1475   case STATE_LISTEN:
1476     GNUNET_break (0);
1477     break;
1478   case STATE_HELLO_WAIT:
1479     if (NULL == socket->lsocket) /* We are client */
1480       queue_message (socket, generate_hello (), NULL, NULL, GNUNET_NO);
1481     else
1482       queue_message (socket,
1483                      (struct GNUNET_STREAM_MessageHeader *)
1484                      generate_hello_ack (socket, GNUNET_NO), NULL, NULL,
1485                      GNUNET_NO);
1486     socket->control_retransmission_task_id =
1487     GNUNET_SCHEDULER_add_delayed (socket->retransmit_timeout,
1488                                   &control_retransmission_task, socket);
1489     break;
1490   case STATE_ESTABLISHED:
1491     if (NULL == socket->lsocket)
1492       queue_message (socket,
1493                      (struct GNUNET_STREAM_MessageHeader *)
1494                      generate_hello_ack (socket, GNUNET_NO), NULL, NULL,
1495                      GNUNET_NO);
1496     else
1497       GNUNET_break (0);
1498   default:
1499     GNUNET_break (0);
1500   }
1501 }
1502
1503
1504 /**
1505  * Client's message handler for GNUNET_MESSAGE_TYPE_STREAM_HELLO_ACK
1506  *
1507  * @param cls the socket (set from GNUNET_MESH_connect)
1508  * @param tunnel connection to the other end
1509  * @param tunnel_ctx this is NULL
1510  * @param sender who sent the message
1511  * @param message the actual message
1512  * @param atsi performance data for the connection
1513  * @return GNUNET_OK to keep the connection open,
1514  *         GNUNET_SYSERR to close it (signal serious error)
1515  */
1516 static int
1517 client_handle_hello_ack (void *cls,
1518                          struct GNUNET_MESH_Tunnel *tunnel,
1519                          void **tunnel_ctx,
1520                          const struct GNUNET_PeerIdentity *sender,
1521                          const struct GNUNET_MessageHeader *message,
1522                          const struct GNUNET_ATS_Information*atsi)
1523 {
1524   struct GNUNET_STREAM_Socket *socket = cls;
1525   const struct GNUNET_STREAM_HelloAckMessage *ack_msg;
1526   struct GNUNET_STREAM_HelloAckMessage *reply;
1527
1528   if (0 != memcmp (sender, &socket->other_peer,
1529                    sizeof (struct GNUNET_PeerIdentity)))
1530   {
1531     LOG (GNUNET_ERROR_TYPE_DEBUG,
1532          "%s: Received HELLO_ACK from non-confirming peer\n",
1533          GNUNET_i2s (&socket->other_peer));
1534     return GNUNET_YES;
1535   }
1536   ack_msg = (const struct GNUNET_STREAM_HelloAckMessage *) message;
1537   LOG (GNUNET_ERROR_TYPE_DEBUG, "%s: Received HELLO_ACK from %s\n",
1538        GNUNET_i2s (&socket->other_peer), GNUNET_i2s (&socket->other_peer));
1539   GNUNET_assert (socket->tunnel == tunnel);
1540   switch (socket->state)
1541   {
1542   case STATE_HELLO_WAIT:
1543     socket->read_sequence_number = ntohl (ack_msg->sequence_number);
1544     LOG (GNUNET_ERROR_TYPE_DEBUG,
1545          "%s: Read sequence number %u\n",
1546          GNUNET_i2s (&socket->other_peer),
1547          (unsigned int) socket->read_sequence_number);
1548     socket->receiver_window_available = ntohl (ack_msg->receiver_window_size);
1549     reply = generate_hello_ack (socket, GNUNET_YES);
1550     queue_message (socket, &reply->header, &set_state_established,
1551                    NULL, GNUNET_NO);
1552     return GNUNET_OK;
1553   case STATE_ESTABLISHED:
1554     // call statistics (# ACKs ignored++)
1555     GNUNET_assert (GNUNET_SCHEDULER_NO_TASK ==
1556                    socket->control_retransmission_task_id);
1557     socket->control_retransmission_task_id =
1558       GNUNET_SCHEDULER_add_now (&control_retransmission_task, socket);
1559     return GNUNET_OK;
1560   default:
1561     LOG_DEBUG ("%s: Server %s sent HELLO_ACK when in state %d\n",
1562                GNUNET_i2s (&socket->other_peer),
1563                GNUNET_i2s (&socket->other_peer), socket->state);
1564     socket->state = STATE_CLOSED; // introduce STATE_ERROR?
1565     return GNUNET_SYSERR;
1566   }
1567 }
1568
1569
1570 /**
1571  * Client's message handler for GNUNET_MESSAGE_TYPE_STREAM_RESET
1572  *
1573  * @param cls the socket (set from GNUNET_MESH_connect)
1574  * @param tunnel connection to the other end
1575  * @param tunnel_ctx this is NULL
1576  * @param sender who sent the message
1577  * @param message the actual message
1578  * @param atsi performance data for the connection
1579  * @return GNUNET_OK to keep the connection open,
1580  *         GNUNET_SYSERR to close it (signal serious error)
1581  */
1582 static int
1583 client_handle_reset (void *cls,
1584                      struct GNUNET_MESH_Tunnel *tunnel,
1585                      void **tunnel_ctx,
1586                      const struct GNUNET_PeerIdentity *sender,
1587                      const struct GNUNET_MessageHeader *message,
1588                      const struct GNUNET_ATS_Information*atsi)
1589 {
1590   // struct GNUNET_STREAM_Socket *socket = cls;
1591
1592   return GNUNET_OK;
1593 }
1594
1595
1596 /**
1597  * Common message handler for handling TRANSMIT_CLOSE messages
1598  *
1599  * @param socket the socket through which the ack was received
1600  * @param tunnel connection to the other end
1601  * @param sender who sent the message
1602  * @param msg the transmit close message
1603  * @param atsi performance data for the connection
1604  * @return GNUNET_OK to keep the connection open,
1605  *         GNUNET_SYSERR to close it (signal serious error)
1606  */
1607 static int
1608 handle_transmit_close (struct GNUNET_STREAM_Socket *socket,
1609                        struct GNUNET_MESH_Tunnel *tunnel,
1610                        const struct GNUNET_PeerIdentity *sender,
1611                        const struct GNUNET_STREAM_MessageHeader *msg,
1612                        const struct GNUNET_ATS_Information*atsi)
1613 {
1614   struct GNUNET_STREAM_MessageHeader *reply;
1615
1616   switch (socket->state)
1617   {
1618   case STATE_ESTABLISHED:
1619     socket->state = STATE_RECEIVE_CLOSED;
1620     /* Send TRANSMIT_CLOSE_ACK */
1621     reply = GNUNET_malloc (sizeof (struct GNUNET_STREAM_MessageHeader));
1622     reply->header.type =
1623       htons (GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE_ACK);
1624     reply->header.size = htons (sizeof (struct GNUNET_STREAM_MessageHeader));
1625     queue_message (socket, reply, NULL, NULL, GNUNET_NO);
1626     break;
1627   default:
1628     /* FIXME: Call statistics? */
1629     break;
1630   }
1631   return GNUNET_YES;
1632 }
1633
1634
1635 /**
1636  * Client's message handler for GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE
1637  *
1638  * @param cls the socket (set from GNUNET_MESH_connect)
1639  * @param tunnel connection to the other end
1640  * @param tunnel_ctx this is NULL
1641  * @param sender who sent the message
1642  * @param message the actual message
1643  * @param atsi performance data for the connection
1644  * @return GNUNET_OK to keep the connection open,
1645  *         GNUNET_SYSERR to close it (signal serious error)
1646  */
1647 static int
1648 client_handle_transmit_close (void *cls,
1649                               struct GNUNET_MESH_Tunnel *tunnel,
1650                               void **tunnel_ctx,
1651                               const struct GNUNET_PeerIdentity *sender,
1652                               const struct GNUNET_MessageHeader *message,
1653                               const struct GNUNET_ATS_Information*atsi)
1654 {
1655   struct GNUNET_STREAM_Socket *socket = cls;
1656
1657   return handle_transmit_close (socket,
1658                                 tunnel,
1659                                 sender,
1660                                 (struct GNUNET_STREAM_MessageHeader *)message,
1661                                 atsi);
1662 }
1663
1664
1665 /**
1666  * Generic handler for GNUNET_MESSAGE_TYPE_STREAM_*_CLOSE_ACK messages
1667  *
1668  * @param socket the socket
1669  * @param tunnel connection to the other end
1670  * @param sender who sent the message
1671  * @param message the actual message
1672  * @param atsi performance data for the connection
1673  * @param operation the close operation which is being ACK'ed
1674  * @return GNUNET_OK to keep the connection open,
1675  *         GNUNET_SYSERR to close it (signal serious error)
1676  */
1677 static int
1678 handle_generic_close_ack (struct GNUNET_STREAM_Socket *socket,
1679                           struct GNUNET_MESH_Tunnel *tunnel,
1680                           const struct GNUNET_PeerIdentity *sender,
1681                           const struct GNUNET_STREAM_MessageHeader *message,
1682                           const struct GNUNET_ATS_Information *atsi,
1683                           int operation)
1684 {
1685   struct GNUNET_STREAM_ShutdownHandle *shutdown_handle;
1686
1687   shutdown_handle = socket->shutdown_handle;
1688   if (NULL == shutdown_handle)
1689   {
1690     LOG (GNUNET_ERROR_TYPE_DEBUG,
1691          "%s: Received CLOSE_ACK when shutdown handle is NULL\n",
1692          GNUNET_i2s (&socket->other_peer));
1693     return GNUNET_OK;
1694   }
1695   switch (operation)
1696   {
1697   case SHUT_RDWR:
1698     switch (socket->state)
1699     {
1700     case STATE_CLOSE_WAIT:
1701       if (SHUT_RDWR != shutdown_handle->operation)
1702       {
1703         LOG (GNUNET_ERROR_TYPE_DEBUG,
1704              "%s: Received CLOSE_ACK when shutdown handle is not for "
1705              "SHUT_RDWR\n", GNUNET_i2s (&socket->other_peer));
1706         return GNUNET_OK;
1707       }
1708       LOG (GNUNET_ERROR_TYPE_DEBUG, "%s: Received CLOSE_ACK from %s\n",
1709            GNUNET_i2s (&socket->other_peer), GNUNET_i2s (&socket->other_peer));
1710       socket->state = STATE_CLOSED;
1711       break;
1712     default:
1713       LOG (GNUNET_ERROR_TYPE_DEBUG,
1714            "%s: Received CLOSE_ACK when in it not expected\n",
1715            GNUNET_i2s (&socket->other_peer));
1716       return GNUNET_OK;
1717     }
1718     break;
1719   case SHUT_RD:
1720     switch (socket->state)
1721     {
1722     case STATE_RECEIVE_CLOSE_WAIT:
1723       if (SHUT_RD != shutdown_handle->operation)
1724       {
1725         LOG (GNUNET_ERROR_TYPE_DEBUG,
1726              "%s: Received RECEIVE_CLOSE_ACK when shutdown handle "
1727              "is not for SHUT_RD\n", GNUNET_i2s (&socket->other_peer));
1728         return GNUNET_OK;
1729       }
1730       LOG (GNUNET_ERROR_TYPE_DEBUG, "%s: Received RECEIVE_CLOSE_ACK from %s\n",
1731            GNUNET_i2s (&socket->other_peer), GNUNET_i2s (&socket->other_peer));
1732       socket->state = STATE_RECEIVE_CLOSED;
1733       break;
1734     default:
1735       LOG (GNUNET_ERROR_TYPE_DEBUG,
1736            "%s: Received RECEIVE_CLOSE_ACK when in it not expected\n",
1737            GNUNET_i2s (&socket->other_peer));
1738       return GNUNET_OK;
1739     }
1740     break;
1741   case SHUT_WR:
1742     switch (socket->state)
1743     {
1744     case STATE_TRANSMIT_CLOSE_WAIT:
1745       if (SHUT_WR != shutdown_handle->operation)
1746       {
1747         LOG (GNUNET_ERROR_TYPE_DEBUG,
1748              "%s: Received TRANSMIT_CLOSE_ACK when shutdown handle "
1749              "is not for SHUT_WR\n",
1750              GNUNET_i2s (&socket->other_peer));
1751         return GNUNET_OK;
1752       }
1753       LOG (GNUNET_ERROR_TYPE_DEBUG, "%s: Received TRANSMIT_CLOSE_ACK from %s\n",
1754            GNUNET_i2s (&socket->other_peer), GNUNET_i2s (&socket->other_peer));
1755       socket->state = STATE_TRANSMIT_CLOSED;
1756       break;
1757     default:
1758       LOG (GNUNET_ERROR_TYPE_DEBUG,
1759            "%s: Received TRANSMIT_CLOSE_ACK when in it not expected\n",
1760            GNUNET_i2s (&socket->other_peer));
1761       return GNUNET_OK;
1762     }
1763     break;
1764   default:
1765     GNUNET_assert (0);
1766   }
1767   if (NULL != shutdown_handle->completion_cb) /* Shutdown completion */
1768     shutdown_handle->completion_cb(shutdown_handle->completion_cls,
1769                                    operation);
1770   if (GNUNET_SCHEDULER_NO_TASK
1771       != shutdown_handle->close_msg_retransmission_task_id)
1772   {
1773     GNUNET_SCHEDULER_cancel
1774       (shutdown_handle->close_msg_retransmission_task_id);
1775     shutdown_handle->close_msg_retransmission_task_id =
1776         GNUNET_SCHEDULER_NO_TASK;
1777   }
1778   GNUNET_free (shutdown_handle); /* Free shutdown handle */
1779   socket->shutdown_handle = NULL;
1780   return GNUNET_OK;
1781 }
1782
1783
1784 /**
1785  * Client's message handler for GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE_ACK
1786  *
1787  * @param cls the socket (set from GNUNET_MESH_connect)
1788  * @param tunnel connection to the other end
1789  * @param tunnel_ctx this is NULL
1790  * @param sender who sent the message
1791  * @param message the actual message
1792  * @param atsi performance data for the connection
1793  * @return GNUNET_OK to keep the connection open,
1794  *         GNUNET_SYSERR to close it (signal serious error)
1795  */
1796 static int
1797 client_handle_transmit_close_ack (void *cls,
1798                                   struct GNUNET_MESH_Tunnel *tunnel,
1799                                   void **tunnel_ctx,
1800                                   const struct GNUNET_PeerIdentity *sender,
1801                                   const struct GNUNET_MessageHeader *message,
1802                                   const struct GNUNET_ATS_Information*atsi)
1803 {
1804   struct GNUNET_STREAM_Socket *socket = cls;
1805
1806   return handle_generic_close_ack (socket,
1807                                    tunnel,
1808                                    sender,
1809                                    (const struct GNUNET_STREAM_MessageHeader *)
1810                                    message,
1811                                    atsi,
1812                                    SHUT_WR);
1813 }
1814
1815
1816 /**
1817  * Generic handler for GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE
1818  *
1819  * @param socket the socket
1820  * @param tunnel connection to the other end
1821  * @param sender who sent the message
1822  * @param message the actual message
1823  * @param atsi performance data for the connection
1824  * @return GNUNET_OK to keep the connection open,
1825  *         GNUNET_SYSERR to close it (signal serious error)
1826  */
1827 static int
1828 handle_receive_close (struct GNUNET_STREAM_Socket *socket,
1829                       struct GNUNET_MESH_Tunnel *tunnel,
1830                       const struct GNUNET_PeerIdentity *sender,
1831                       const struct GNUNET_STREAM_MessageHeader *message,
1832                       const struct GNUNET_ATS_Information *atsi)
1833 {
1834   struct GNUNET_STREAM_MessageHeader *receive_close_ack;
1835
1836   switch (socket->state)
1837   {
1838   case STATE_INIT:
1839   case STATE_LISTEN:
1840   case STATE_HELLO_WAIT:
1841     LOG (GNUNET_ERROR_TYPE_DEBUG,
1842          "%s: Ignoring RECEIVE_CLOSE as it cannot be handled now\n",
1843          GNUNET_i2s (&socket->other_peer));
1844     return GNUNET_OK;
1845   default:
1846     break;
1847   }
1848
1849   LOG (GNUNET_ERROR_TYPE_DEBUG,
1850        "%s: Received RECEIVE_CLOSE from %s\n",
1851        GNUNET_i2s (&socket->other_peer),
1852        GNUNET_i2s (&socket->other_peer));
1853   receive_close_ack =
1854     GNUNET_malloc (sizeof (struct GNUNET_STREAM_MessageHeader));
1855   receive_close_ack->header.size =
1856     htons (sizeof (struct GNUNET_STREAM_MessageHeader));
1857   receive_close_ack->header.type =
1858     htons (GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE_ACK);
1859   queue_message (socket, receive_close_ack, &set_state_closed,
1860                  NULL, GNUNET_NO);
1861   /* FIXME: Handle the case where write handle is present; the write operation
1862      should be deemed as finised and the write continuation callback
1863      has to be called with the stream status GNUNET_STREAM_SHUTDOWN */
1864   return GNUNET_OK;
1865 }
1866
1867
1868 /**
1869  * Client's message handler for GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE
1870  *
1871  * @param cls the socket (set from GNUNET_MESH_connect)
1872  * @param tunnel connection to the other end
1873  * @param tunnel_ctx this is NULL
1874  * @param sender who sent the message
1875  * @param message the actual message
1876  * @param atsi performance data for the connection
1877  * @return GNUNET_OK to keep the connection open,
1878  *         GNUNET_SYSERR to close it (signal serious error)
1879  */
1880 static int
1881 client_handle_receive_close (void *cls,
1882                              struct GNUNET_MESH_Tunnel *tunnel,
1883                              void **tunnel_ctx,
1884                              const struct GNUNET_PeerIdentity *sender,
1885                              const struct GNUNET_MessageHeader *message,
1886                              const struct GNUNET_ATS_Information*atsi)
1887 {
1888   struct GNUNET_STREAM_Socket *socket = cls;
1889
1890   return
1891     handle_receive_close (socket,
1892                           tunnel,
1893                           sender,
1894                           (const struct GNUNET_STREAM_MessageHeader *) message,
1895                           atsi);
1896 }
1897
1898
1899 /**
1900  * Client's message handler for GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE_ACK
1901  *
1902  * @param cls the socket (set from GNUNET_MESH_connect)
1903  * @param tunnel connection to the other end
1904  * @param tunnel_ctx this is NULL
1905  * @param sender who sent the message
1906  * @param message the actual message
1907  * @param atsi performance data for the connection
1908  * @return GNUNET_OK to keep the connection open,
1909  *         GNUNET_SYSERR to close it (signal serious error)
1910  */
1911 static int
1912 client_handle_receive_close_ack (void *cls,
1913                                  struct GNUNET_MESH_Tunnel *tunnel,
1914                                  void **tunnel_ctx,
1915                                  const struct GNUNET_PeerIdentity *sender,
1916                                  const struct GNUNET_MessageHeader *message,
1917                                  const struct GNUNET_ATS_Information*atsi)
1918 {
1919   struct GNUNET_STREAM_Socket *socket = cls;
1920
1921   return handle_generic_close_ack (socket,
1922                                    tunnel,
1923                                    sender,
1924                                    (const struct GNUNET_STREAM_MessageHeader *)
1925                                    message,
1926                                    atsi,
1927                                    SHUT_RD);
1928 }
1929
1930
1931 /**
1932  * Generic handler for GNUNET_MESSAGE_TYPE_STREAM_CLOSE
1933  *
1934  * @param socket the socket
1935  * @param tunnel connection to the other end
1936  * @param sender who sent the message
1937  * @param message the actual message
1938  * @param atsi performance data for the connection
1939  * @return GNUNET_OK to keep the connection open,
1940  *         GNUNET_SYSERR to close it (signal serious error)
1941  */
1942 static int
1943 handle_close (struct GNUNET_STREAM_Socket *socket,
1944               struct GNUNET_MESH_Tunnel *tunnel,
1945               const struct GNUNET_PeerIdentity *sender,
1946               const struct GNUNET_STREAM_MessageHeader *message,
1947               const struct GNUNET_ATS_Information*atsi)
1948 {
1949   struct GNUNET_STREAM_MessageHeader *close_ack;
1950
1951   switch (socket->state)
1952   {
1953   case STATE_INIT:
1954   case STATE_LISTEN:
1955   case STATE_HELLO_WAIT:
1956     LOG (GNUNET_ERROR_TYPE_DEBUG,
1957          "%s: Ignoring RECEIVE_CLOSE as it cannot be handled now\n",
1958          GNUNET_i2s (&socket->other_peer));
1959     return GNUNET_OK;
1960   default:
1961     break;
1962   }
1963
1964   LOG (GNUNET_ERROR_TYPE_DEBUG,
1965        "%s: Received CLOSE from %s\n",
1966        GNUNET_i2s (&socket->other_peer),
1967        GNUNET_i2s (&socket->other_peer));
1968   close_ack = GNUNET_malloc (sizeof (struct GNUNET_STREAM_MessageHeader));
1969   close_ack->header.size = htons (sizeof (struct GNUNET_STREAM_MessageHeader));
1970   close_ack->header.type = htons (GNUNET_MESSAGE_TYPE_STREAM_CLOSE_ACK);
1971   queue_message (socket, close_ack, &set_state_closed, NULL, GNUNET_NO);
1972   if (socket->state == STATE_CLOSED)
1973     return GNUNET_OK;
1974
1975   GNUNET_free_non_null (socket->receive_buffer); /* Free the receive buffer */
1976   socket->receive_buffer = NULL;
1977   socket->receive_buffer_size = 0;
1978   return GNUNET_OK;
1979 }
1980
1981
1982 /**
1983  * Client's message handler for GNUNET_MESSAGE_TYPE_STREAM_CLOSE
1984  *
1985  * @param cls the socket (set from GNUNET_MESH_connect)
1986  * @param tunnel connection to the other end
1987  * @param tunnel_ctx this is NULL
1988  * @param sender who sent the message
1989  * @param message the actual message
1990  * @param atsi performance data for the connection
1991  * @return GNUNET_OK to keep the connection open,
1992  *         GNUNET_SYSERR to close it (signal serious error)
1993  */
1994 static int
1995 client_handle_close (void *cls,
1996                      struct GNUNET_MESH_Tunnel *tunnel,
1997                      void **tunnel_ctx,
1998                      const struct GNUNET_PeerIdentity *sender,
1999                      const struct GNUNET_MessageHeader *message,
2000                      const struct GNUNET_ATS_Information*atsi)
2001 {
2002   struct GNUNET_STREAM_Socket *socket = cls;
2003
2004   return handle_close (socket,
2005                        tunnel,
2006                        sender,
2007                        (const struct GNUNET_STREAM_MessageHeader *) message,
2008                        atsi);
2009 }
2010
2011
2012 /**
2013  * Client's message handler for GNUNET_MESSAGE_TYPE_STREAM_CLOSE_ACK
2014  *
2015  * @param cls the socket (set from GNUNET_MESH_connect)
2016  * @param tunnel connection to the other end
2017  * @param tunnel_ctx this is NULL
2018  * @param sender who sent the message
2019  * @param message the actual message
2020  * @param atsi performance data for the connection
2021  * @return GNUNET_OK to keep the connection open,
2022  *         GNUNET_SYSERR to close it (signal serious error)
2023  */
2024 static int
2025 client_handle_close_ack (void *cls,
2026                          struct GNUNET_MESH_Tunnel *tunnel,
2027                          void **tunnel_ctx,
2028                          const struct GNUNET_PeerIdentity *sender,
2029                          const struct GNUNET_MessageHeader *message,
2030                          const struct GNUNET_ATS_Information *atsi)
2031 {
2032   struct GNUNET_STREAM_Socket *socket = cls;
2033
2034   return handle_generic_close_ack (socket,
2035                                    tunnel,
2036                                    sender,
2037                                    (const struct GNUNET_STREAM_MessageHeader *)
2038                                    message,
2039                                    atsi,
2040                                    SHUT_RDWR);
2041 }
2042
2043 /*****************************/
2044 /* Server's Message Handlers */
2045 /*****************************/
2046
2047 /**
2048  * Server's message Handler for GNUNET_MESSAGE_TYPE_STREAM_DATA
2049  *
2050  * @param cls the closure
2051  * @param tunnel connection to the other end
2052  * @param tunnel_ctx the socket
2053  * @param sender who sent the message
2054  * @param message the actual message
2055  * @param atsi performance data for the connection
2056  * @return GNUNET_OK to keep the connection open,
2057  *         GNUNET_SYSERR to close it (signal serious error)
2058  */
2059 static int
2060 server_handle_data (void *cls,
2061                     struct GNUNET_MESH_Tunnel *tunnel,
2062                     void **tunnel_ctx,
2063                     const struct GNUNET_PeerIdentity *sender,
2064                     const struct GNUNET_MessageHeader *message,
2065                     const struct GNUNET_ATS_Information*atsi)
2066 {
2067   struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
2068
2069   return handle_data (socket,
2070                       tunnel,
2071                       sender,
2072                       (const struct GNUNET_STREAM_DataMessage *)message,
2073                       atsi);
2074 }
2075
2076
2077 /**
2078  * Server's message handler for GNUNET_MESSAGE_TYPE_STREAM_HELLO
2079  *
2080  * @param cls the closure
2081  * @param tunnel connection to the other end
2082  * @param tunnel_ctx the socket
2083  * @param sender who sent the message
2084  * @param message the actual message
2085  * @param atsi performance data for the connection
2086  * @return GNUNET_OK to keep the connection open,
2087  *         GNUNET_SYSERR to close it (signal serious error)
2088  */
2089 static int
2090 server_handle_hello (void *cls,
2091                      struct GNUNET_MESH_Tunnel *tunnel,
2092                      void **tunnel_ctx,
2093                      const struct GNUNET_PeerIdentity *sender,
2094                      const struct GNUNET_MessageHeader *message,
2095                      const struct GNUNET_ATS_Information*atsi)
2096 {
2097   struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
2098   struct GNUNET_STREAM_HelloAckMessage *reply;
2099
2100   if (0 != memcmp (sender,
2101                    &socket->other_peer,
2102                    sizeof (struct GNUNET_PeerIdentity)))
2103   {
2104     LOG_DEBUG ("%s: Received HELLO from non-confirming peer\n",
2105                GNUNET_i2s (&socket->other_peer));
2106     return GNUNET_YES;
2107   }
2108   GNUNET_assert (GNUNET_MESSAGE_TYPE_STREAM_HELLO == ntohs (message->type));
2109   GNUNET_assert (socket->tunnel == tunnel);
2110   LOG_DEBUG ("%s: Received HELLO from %s\n", GNUNET_i2s (&socket->other_peer),
2111              GNUNET_i2s (&socket->other_peer));
2112   switch (socket->status)
2113   {
2114   case STATE_INIT:
2115     reply = generate_hello_ack (socket, GNUNET_YES);
2116     queue_message (socket, &reply->header, &set_state_hello_wait, NULL,
2117                    GNUNET_NO);
2118     GNUNET_assert (GNUNET_SCHEDULER_NO_TASK ==
2119                    socket->control_retransmission_task_id);
2120     socket->control_retransmission_task_id =
2121       GNUNET_SCHEDULER_add_delayed (socket->retransmit_timeout,
2122                                     &control_retransmission_task, socket);
2123     break;
2124   case STATE_HELLO_WAIT:
2125     /* Perhaps our HELLO_ACK was lost */
2126     GNUNET_assert (GNUNET_SCHEDULER_NO_TASK !=
2127                    socket->control_retransmission_task_id);
2128     GNUNET_SCHEDULER_cancel (socket->control_retransmission_task_id);
2129     socket->control_retransmission_task_id =
2130       GNUNET_SCHEDULER_add_now (&control_retransmission_task, socket);
2131     break;
2132   default:
2133     LOG_DEBUG( "%s: Client sent HELLO when in state %d\n",
2134                GNUNET_i2s (&socket->other_peer), socket->state);
2135     /* FIXME: Send RESET? */
2136   }
2137   return GNUNET_OK;
2138 }
2139
2140
2141 /**
2142  * Server's message handler for GNUNET_MESSAGE_TYPE_STREAM_HELLO_ACK
2143  *
2144  * @param cls the closure
2145  * @param tunnel connection to the other end
2146  * @param tunnel_ctx the socket
2147  * @param sender who sent the message
2148  * @param message the actual message
2149  * @param atsi performance data for the connection
2150  * @return GNUNET_OK to keep the connection open,
2151  *         GNUNET_SYSERR to close it (signal serious error)
2152  */
2153 static int
2154 server_handle_hello_ack (void *cls,
2155                          struct GNUNET_MESH_Tunnel *tunnel,
2156                          void **tunnel_ctx,
2157                          const struct GNUNET_PeerIdentity *sender,
2158                          const struct GNUNET_MessageHeader *message,
2159                          const struct GNUNET_ATS_Information*atsi)
2160 {
2161   struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
2162   const struct GNUNET_STREAM_HelloAckMessage *ack_message;
2163
2164   GNUNET_assert (GNUNET_MESSAGE_TYPE_STREAM_HELLO_ACK ==
2165                  ntohs (message->type));
2166   GNUNET_assert (socket->tunnel == tunnel);
2167   ack_message = (struct GNUNET_STREAM_HelloAckMessage *) message;
2168   switch (socket->state)
2169   {
2170   case STATE_HELLO_WAIT:
2171     LOG (GNUNET_ERROR_TYPE_DEBUG,
2172          "%s: Received HELLO_ACK from %s\n",
2173          GNUNET_i2s (&socket->other_peer),
2174          GNUNET_i2s (&socket->other_peer));
2175     socket->read_sequence_number = ntohl (ack_message->sequence_number);
2176     LOG (GNUNET_ERROR_TYPE_DEBUG,
2177          "%s: Read sequence number %u\n",
2178          GNUNET_i2s (&socket->other_peer),
2179          (unsigned int) socket->read_sequence_number);
2180     socket->receiver_window_available =
2181       ntohl (ack_message->receiver_window_size);
2182     set_state_established (NULL, socket);
2183     break;
2184   default:
2185     LOG (GNUNET_ERROR_TYPE_DEBUG,
2186          "Client sent HELLO_ACK when in state %d\n", socket->state);
2187   }
2188   return GNUNET_OK;
2189 }
2190
2191
2192 /**
2193  * Server's message handler for GNUNET_MESSAGE_TYPE_STREAM_RESET
2194  *
2195  * @param cls the closure
2196  * @param tunnel connection to the other end
2197  * @param tunnel_ctx the socket
2198  * @param sender who sent the message
2199  * @param message the actual message
2200  * @param atsi performance data for the connection
2201  * @return GNUNET_OK to keep the connection open,
2202  *         GNUNET_SYSERR to close it (signal serious error)
2203  */
2204 static int
2205 server_handle_reset (void *cls,
2206                      struct GNUNET_MESH_Tunnel *tunnel,
2207                      void **tunnel_ctx,
2208                      const struct GNUNET_PeerIdentity *sender,
2209                      const struct GNUNET_MessageHeader *message,
2210                      const struct GNUNET_ATS_Information*atsi)
2211 {
2212   // struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
2213
2214   return GNUNET_OK;
2215 }
2216
2217
2218 /**
2219  * Server's message handler for GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE
2220  *
2221  * @param cls the closure
2222  * @param tunnel connection to the other end
2223  * @param tunnel_ctx the socket
2224  * @param sender who sent the message
2225  * @param message the actual message
2226  * @param atsi performance data for the connection
2227  * @return GNUNET_OK to keep the connection open,
2228  *         GNUNET_SYSERR to close it (signal serious error)
2229  */
2230 static int
2231 server_handle_transmit_close (void *cls,
2232                               struct GNUNET_MESH_Tunnel *tunnel,
2233                               void **tunnel_ctx,
2234                               const struct GNUNET_PeerIdentity *sender,
2235                               const struct GNUNET_MessageHeader *message,
2236                               const struct GNUNET_ATS_Information*atsi)
2237 {
2238   struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
2239
2240   return handle_transmit_close (socket,
2241                                 tunnel,
2242                                 sender,
2243                                 (struct GNUNET_STREAM_MessageHeader *)message,
2244                                 atsi);
2245 }
2246
2247
2248 /**
2249  * Server's message handler for GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE_ACK
2250  *
2251  * @param cls the closure
2252  * @param tunnel connection to the other end
2253  * @param tunnel_ctx the socket
2254  * @param sender who sent the message
2255  * @param message the actual message
2256  * @param atsi performance data for the connection
2257  * @return GNUNET_OK to keep the connection open,
2258  *         GNUNET_SYSERR to close it (signal serious error)
2259  */
2260 static int
2261 server_handle_transmit_close_ack (void *cls,
2262                                   struct GNUNET_MESH_Tunnel *tunnel,
2263                                   void **tunnel_ctx,
2264                                   const struct GNUNET_PeerIdentity *sender,
2265                                   const struct GNUNET_MessageHeader *message,
2266                                   const struct GNUNET_ATS_Information*atsi)
2267 {
2268   struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
2269
2270   return handle_generic_close_ack (socket,
2271                                    tunnel,
2272                                    sender,
2273                                    (const struct GNUNET_STREAM_MessageHeader *)
2274                                    message,
2275                                    atsi,
2276                                    SHUT_WR);
2277 }
2278
2279
2280 /**
2281  * Server's message handler for GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE
2282  *
2283  * @param cls the closure
2284  * @param tunnel connection to the other end
2285  * @param tunnel_ctx the socket
2286  * @param sender who sent the message
2287  * @param message the actual message
2288  * @param atsi performance data for the connection
2289  * @return GNUNET_OK to keep the connection open,
2290  *         GNUNET_SYSERR to close it (signal serious error)
2291  */
2292 static int
2293 server_handle_receive_close (void *cls,
2294                              struct GNUNET_MESH_Tunnel *tunnel,
2295                              void **tunnel_ctx,
2296                              const struct GNUNET_PeerIdentity *sender,
2297                              const struct GNUNET_MessageHeader *message,
2298                              const struct GNUNET_ATS_Information*atsi)
2299 {
2300   struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
2301
2302   return
2303     handle_receive_close (socket,
2304                           tunnel,
2305                           sender,
2306                           (const struct GNUNET_STREAM_MessageHeader *) message,
2307                           atsi);
2308 }
2309
2310
2311 /**
2312  * Server's message handler for GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE_ACK
2313  *
2314  * @param cls the closure
2315  * @param tunnel connection to the other end
2316  * @param tunnel_ctx the socket
2317  * @param sender who sent the message
2318  * @param message the actual message
2319  * @param atsi performance data for the connection
2320  * @return GNUNET_OK to keep the connection open,
2321  *         GNUNET_SYSERR to close it (signal serious error)
2322  */
2323 static int
2324 server_handle_receive_close_ack (void *cls,
2325                                  struct GNUNET_MESH_Tunnel *tunnel,
2326                                  void **tunnel_ctx,
2327                                  const struct GNUNET_PeerIdentity *sender,
2328                                  const struct GNUNET_MessageHeader *message,
2329                                  const struct GNUNET_ATS_Information*atsi)
2330 {
2331   struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
2332
2333   return handle_generic_close_ack (socket,
2334                                    tunnel,
2335                                    sender,
2336                                    (const struct GNUNET_STREAM_MessageHeader *)
2337                                    message,
2338                                    atsi,
2339                                    SHUT_RD);
2340 }
2341
2342
2343 /**
2344  * Server's message handler for GNUNET_MESSAGE_TYPE_STREAM_CLOSE
2345  *
2346  * @param cls the listen socket (from GNUNET_MESH_connect in
2347  *          GNUNET_STREAM_listen)
2348  * @param tunnel connection to the other end
2349  * @param tunnel_ctx the socket
2350  * @param sender who sent the message
2351  * @param message the actual message
2352  * @param atsi performance data for the connection
2353  * @return GNUNET_OK to keep the connection open,
2354  *         GNUNET_SYSERR to close it (signal serious error)
2355  */
2356 static int
2357 server_handle_close (void *cls,
2358                      struct GNUNET_MESH_Tunnel *tunnel,
2359                      void **tunnel_ctx,
2360                      const struct GNUNET_PeerIdentity *sender,
2361                      const struct GNUNET_MessageHeader *message,
2362                      const struct GNUNET_ATS_Information*atsi)
2363 {
2364   struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
2365
2366   return handle_close (socket,
2367                        tunnel,
2368                        sender,
2369                        (const struct GNUNET_STREAM_MessageHeader *) message,
2370                        atsi);
2371 }
2372
2373
2374 /**
2375  * Server's message handler for GNUNET_MESSAGE_TYPE_STREAM_CLOSE_ACK
2376  *
2377  * @param cls the closure
2378  * @param tunnel connection to the other end
2379  * @param tunnel_ctx the socket
2380  * @param sender who sent the message
2381  * @param message the actual message
2382  * @param atsi performance data for the connection
2383  * @return GNUNET_OK to keep the connection open,
2384  *         GNUNET_SYSERR to close it (signal serious error)
2385  */
2386 static int
2387 server_handle_close_ack (void *cls,
2388                          struct GNUNET_MESH_Tunnel *tunnel,
2389                          void **tunnel_ctx,
2390                          const struct GNUNET_PeerIdentity *sender,
2391                          const struct GNUNET_MessageHeader *message,
2392                          const struct GNUNET_ATS_Information*atsi)
2393 {
2394   struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
2395
2396   return handle_generic_close_ack (socket,
2397                                    tunnel,
2398                                    sender,
2399                                    (const struct GNUNET_STREAM_MessageHeader *)
2400                                    message,
2401                                    atsi,
2402                                    SHUT_RDWR);
2403 }
2404
2405
2406 /**
2407  * Handler for DATA_ACK messages
2408  *
2409  * @param socket the socket through which the ack was received
2410  * @param tunnel connection to the other end
2411  * @param sender who sent the message
2412  * @param ack the acknowledgment message
2413  * @param atsi performance data for the connection
2414  * @return GNUNET_OK to keep the connection open,
2415  *         GNUNET_SYSERR to close it (signal serious error)
2416  */
2417 static int
2418 handle_ack (struct GNUNET_STREAM_Socket *socket,
2419             struct GNUNET_MESH_Tunnel *tunnel,
2420             const struct GNUNET_PeerIdentity *sender,
2421             const struct GNUNET_STREAM_AckMessage *ack,
2422             const struct GNUNET_ATS_Information*atsi)
2423 {
2424   unsigned int packet;
2425   int need_retransmission;
2426   uint32_t sequence_difference;
2427
2428   if (0 != memcmp (sender,
2429                    &socket->other_peer,
2430                    sizeof (struct GNUNET_PeerIdentity)))
2431   {
2432     LOG (GNUNET_ERROR_TYPE_DEBUG,
2433          "%s: Received ACK from non-confirming peer\n",
2434          GNUNET_i2s (&socket->other_peer));
2435     return GNUNET_YES;
2436   }
2437   switch (socket->state)
2438   {
2439   case (STATE_ESTABLISHED):
2440   case (STATE_RECEIVE_CLOSED):
2441   case (STATE_RECEIVE_CLOSE_WAIT):
2442     if (NULL == socket->write_handle)
2443     {
2444       LOG (GNUNET_ERROR_TYPE_DEBUG,
2445            "%s: Received DATA_ACK when write_handle is NULL\n",
2446            GNUNET_i2s (&socket->other_peer));
2447       return GNUNET_OK;
2448     }
2449     sequence_difference =
2450         socket->write_sequence_number - ntohl (ack->base_sequence_number);
2451     if (!(sequence_difference <= GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH))
2452     {
2453       LOG (GNUNET_ERROR_TYPE_DEBUG,
2454            "%s: Received DATA_ACK with unexpected base sequence number\n",
2455            GNUNET_i2s (&socket->other_peer));
2456       LOG (GNUNET_ERROR_TYPE_DEBUG,
2457            "%s: Current write sequence: %u; Ack's base sequence: %u\n",
2458            GNUNET_i2s (&socket->other_peer),
2459            socket->write_sequence_number,
2460            ntohl (ack->base_sequence_number));
2461       return GNUNET_OK;
2462     }
2463     /* FIXME: include the case when write_handle is cancelled - ignore the
2464        acks */
2465     LOG (GNUNET_ERROR_TYPE_DEBUG, "%s: Received DATA_ACK from %s\n",
2466          GNUNET_i2s (&socket->other_peer), GNUNET_i2s (&socket->other_peer));
2467     /* Cancel the retransmission task */
2468     if (GNUNET_SCHEDULER_NO_TASK != socket->data_retransmission_task_id)
2469     {
2470       GNUNET_SCHEDULER_cancel (socket->data_retransmission_task_id);
2471       socket->data_retransmission_task_id = GNUNET_SCHEDULER_NO_TASK;
2472     }
2473     for (packet=0; packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH; packet++)
2474     {
2475       if (NULL == socket->write_handle->messages[packet]) break;
2476       /* BS: Base sequence from ack; PS: sequence num of current packet */
2477       sequence_difference = ntohl (ack->base_sequence_number)
2478         - ntohl (socket->write_handle->messages[packet]->sequence_number);
2479       if ((0 == sequence_difference) ||
2480           (GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH < sequence_difference))
2481         continue; /* The message in our handle is not yet received */
2482       /* case where BS = PS + GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH */
2483       /* sequence_difference <= GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH */
2484       ackbitmap_modify_bit (&socket->write_handle->ack_bitmap,
2485                             packet, GNUNET_YES);
2486     }
2487     /* Update the receive window remaining
2488        FIXME : Should update with the value from a data ack with greater
2489        sequence number */
2490     socket->receiver_window_available =
2491       ntohl (ack->receive_window_remaining);
2492     /* Check if we have received all acknowledgements */
2493     need_retransmission = GNUNET_NO;
2494     for (packet=0; packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH; packet++)
2495     {
2496       if (NULL == socket->write_handle->messages[packet]) break;
2497       if (GNUNET_YES != ackbitmap_is_bit_set
2498           (&socket->write_handle->ack_bitmap,packet))
2499       {
2500         need_retransmission = GNUNET_YES;
2501         break;
2502       }
2503     }
2504     if (GNUNET_YES == need_retransmission)
2505     {
2506       write_data (socket);
2507     }
2508     else      /* We have to call the write continuation callback now */
2509     {
2510       struct GNUNET_STREAM_IOWriteHandle *write_handle;
2511
2512       /* Free the packets */
2513       for (packet=0; packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH; packet++)
2514       {
2515         GNUNET_free_non_null (socket->write_handle->messages[packet]);
2516       }
2517       write_handle = socket->write_handle;
2518       socket->write_handle = NULL;
2519       if (NULL != write_handle->write_cont)
2520         write_handle->write_cont (write_handle->write_cont_cls,
2521                                   socket->status,
2522                                   write_handle->size);
2523       /* We are done with the write handle - Freeing it */
2524       GNUNET_free (write_handle);
2525       LOG (GNUNET_ERROR_TYPE_DEBUG,
2526            "%s: Write completion callback completed\n",
2527            GNUNET_i2s (&socket->other_peer));
2528     }
2529     break;
2530   default:
2531     break;
2532   }
2533   return GNUNET_OK;
2534 }
2535
2536
2537 /**
2538  * Handler for DATA_ACK messages
2539  *
2540  * @param cls the 'struct GNUNET_STREAM_Socket'
2541  * @param tunnel connection to the other end
2542  * @param tunnel_ctx unused
2543  * @param sender who sent the message
2544  * @param message the actual message
2545  * @param atsi performance data for the connection
2546  * @return GNUNET_OK to keep the connection open,
2547  *         GNUNET_SYSERR to close it (signal serious error)
2548  */
2549 static int
2550 client_handle_ack (void *cls,
2551                    struct GNUNET_MESH_Tunnel *tunnel,
2552                    void **tunnel_ctx,
2553                    const struct GNUNET_PeerIdentity *sender,
2554                    const struct GNUNET_MessageHeader *message,
2555                    const struct GNUNET_ATS_Information*atsi)
2556 {
2557   struct GNUNET_STREAM_Socket *socket = cls;
2558   const struct GNUNET_STREAM_AckMessage *ack = (const struct GNUNET_STREAM_AckMessage *) message;
2559
2560   return handle_ack (socket, tunnel, sender, ack, atsi);
2561 }
2562
2563
2564 /**
2565  * Handler for DATA_ACK messages
2566  *
2567  * @param cls the server's listen socket
2568  * @param tunnel connection to the other end
2569  * @param tunnel_ctx pointer to the 'struct GNUNET_STREAM_Socket*'
2570  * @param sender who sent the message
2571  * @param message the actual message
2572  * @param atsi performance data for the connection
2573  * @return GNUNET_OK to keep the connection open,
2574  *         GNUNET_SYSERR to close it (signal serious error)
2575  */
2576 static int
2577 server_handle_ack (void *cls,
2578                    struct GNUNET_MESH_Tunnel *tunnel,
2579                    void **tunnel_ctx,
2580                    const struct GNUNET_PeerIdentity *sender,
2581                    const struct GNUNET_MessageHeader *message,
2582                    const struct GNUNET_ATS_Information*atsi)
2583 {
2584   struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
2585   const struct GNUNET_STREAM_AckMessage *ack = (const struct GNUNET_STREAM_AckMessage *) message;
2586
2587   return handle_ack (socket, tunnel, sender, ack, atsi);
2588 }
2589
2590
2591 /**
2592  * For client message handlers, the stream socket is in the
2593  * closure argument.
2594  */
2595 static struct GNUNET_MESH_MessageHandler client_message_handlers[] = {
2596   {&client_handle_data, GNUNET_MESSAGE_TYPE_STREAM_DATA, 0},
2597   {&client_handle_ack, GNUNET_MESSAGE_TYPE_STREAM_ACK,
2598    sizeof (struct GNUNET_STREAM_AckMessage) },
2599   {&client_handle_hello_ack, GNUNET_MESSAGE_TYPE_STREAM_HELLO_ACK,
2600    sizeof (struct GNUNET_STREAM_HelloAckMessage)},
2601   {&client_handle_reset, GNUNET_MESSAGE_TYPE_STREAM_RESET,
2602    sizeof (struct GNUNET_STREAM_MessageHeader)},
2603   {&client_handle_transmit_close, GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE,
2604    sizeof (struct GNUNET_STREAM_MessageHeader)},
2605   {&client_handle_transmit_close_ack, GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE_ACK,
2606    sizeof (struct GNUNET_STREAM_MessageHeader)},
2607   {&client_handle_receive_close, GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE,
2608    sizeof (struct GNUNET_STREAM_MessageHeader)},
2609   {&client_handle_receive_close_ack, GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE_ACK,
2610    sizeof (struct GNUNET_STREAM_MessageHeader)},
2611   {&client_handle_close, GNUNET_MESSAGE_TYPE_STREAM_CLOSE,
2612    sizeof (struct GNUNET_STREAM_MessageHeader)},
2613   {&client_handle_close_ack, GNUNET_MESSAGE_TYPE_STREAM_CLOSE_ACK,
2614    sizeof (struct GNUNET_STREAM_MessageHeader)},
2615   {NULL, 0, 0}
2616 };
2617
2618
2619 /**
2620  * For server message handlers, the stream socket is in the
2621  * tunnel context, and the listen socket in the closure argument.
2622  */
2623 static struct GNUNET_MESH_MessageHandler server_message_handlers[] = {
2624   {&server_handle_data, GNUNET_MESSAGE_TYPE_STREAM_DATA, 0},
2625   {&server_handle_ack, GNUNET_MESSAGE_TYPE_STREAM_ACK,
2626    sizeof (struct GNUNET_STREAM_AckMessage) },
2627   {&server_handle_hello, GNUNET_MESSAGE_TYPE_STREAM_HELLO,
2628    sizeof (struct GNUNET_STREAM_MessageHeader)},
2629   {&server_handle_hello_ack, GNUNET_MESSAGE_TYPE_STREAM_HELLO_ACK,
2630    sizeof (struct GNUNET_STREAM_HelloAckMessage)},
2631   {&server_handle_reset, GNUNET_MESSAGE_TYPE_STREAM_RESET,
2632    sizeof (struct GNUNET_STREAM_MessageHeader)},
2633   {&server_handle_transmit_close, GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE,
2634    sizeof (struct GNUNET_STREAM_MessageHeader)},
2635   {&server_handle_transmit_close_ack, GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE_ACK,
2636    sizeof (struct GNUNET_STREAM_MessageHeader)},
2637   {&server_handle_receive_close, GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE,
2638    sizeof (struct GNUNET_STREAM_MessageHeader)},
2639   {&server_handle_receive_close_ack, GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE_ACK,
2640    sizeof (struct GNUNET_STREAM_MessageHeader)},
2641   {&server_handle_close, GNUNET_MESSAGE_TYPE_STREAM_CLOSE,
2642    sizeof (struct GNUNET_STREAM_MessageHeader)},
2643   {&server_handle_close_ack, GNUNET_MESSAGE_TYPE_STREAM_CLOSE_ACK,
2644    sizeof (struct GNUNET_STREAM_MessageHeader)},
2645   {NULL, 0, 0}
2646 };
2647
2648
2649 /**
2650  * Function called when our target peer is connected to our tunnel
2651  *
2652  * @param cls the socket for which this tunnel is created
2653  * @param peer the peer identity of the target
2654  * @param atsi performance data for the connection
2655  */
2656 static void
2657 mesh_peer_connect_callback (void *cls,
2658                             const struct GNUNET_PeerIdentity *peer,
2659                             const struct GNUNET_ATS_Information * atsi)
2660 {
2661   struct GNUNET_STREAM_Socket *socket = cls;
2662   struct GNUNET_STREAM_MessageHeader *message;
2663
2664   if (0 != memcmp (peer,
2665                    &socket->other_peer,
2666                    sizeof (struct GNUNET_PeerIdentity)))
2667   {
2668     LOG (GNUNET_ERROR_TYPE_DEBUG,
2669          "%s: A peer which is not our target has connected to our tunnel\n",
2670          GNUNET_i2s(peer));
2671     return;
2672   }
2673   LOG (GNUNET_ERROR_TYPE_DEBUG,
2674        "%s: Target peer %s connected\n",
2675        GNUNET_i2s (&socket->other_peer),
2676        GNUNET_i2s (&socket->other_peer));
2677   /* Set state to INIT */
2678   socket->state = STATE_INIT;
2679   /* Send HELLO message */
2680   message = generate_hello ();
2681   queue_message (socket, message, &set_state_hello_wait, NULL, GNUNET_NO);
2682   GNUNET_assert (GNUNET_SCHEDULER_NO_TASK ==
2683                  socket->control_retransmission_task_id);
2684   socket->control_retransmission_task_id =
2685     GNUNET_SCHEDULER_add_delayed (socket->retransmit_timeout,
2686                                   &control_retransmission_task, socket);
2687 }
2688
2689
2690 /**
2691  * Function called when our target peer is disconnected from our tunnel
2692  *
2693  * @param cls the socket associated which this tunnel
2694  * @param peer the peer identity of the target
2695  */
2696 static void
2697 mesh_peer_disconnect_callback (void *cls,
2698                                const struct GNUNET_PeerIdentity *peer)
2699 {
2700   struct GNUNET_STREAM_Socket *socket=cls;
2701
2702   /* If the state is SHUTDOWN its ok; else set the state of the socket to SYSERR */
2703   LOG (GNUNET_ERROR_TYPE_DEBUG,
2704        "%s: Other peer %s disconnected \n",
2705        GNUNET_i2s (&socket->other_peer),
2706        GNUNET_i2s (&socket->other_peer));
2707 }
2708
2709
2710 /**
2711  * Method called whenever a peer creates a tunnel to us
2712  *
2713  * @param cls closure
2714  * @param tunnel new handle to the tunnel
2715  * @param initiator peer that started the tunnel
2716  * @param atsi performance information for the tunnel
2717  * @return initial tunnel context for the tunnel
2718  *         (can be NULL -- that's not an error)
2719  */
2720 static void *
2721 new_tunnel_notify (void *cls,
2722                    struct GNUNET_MESH_Tunnel *tunnel,
2723                    const struct GNUNET_PeerIdentity *initiator,
2724                    const struct GNUNET_ATS_Information *atsi)
2725 {
2726   struct GNUNET_STREAM_ListenSocket *lsocket = cls;
2727   struct GNUNET_STREAM_Socket *socket;
2728
2729   /* FIXME: If a tunnel is already created, we should not accept new tunnels
2730      from the same peer again until the socket is closed */
2731
2732   if (GNUNET_NO == lsocket->listening)
2733   {
2734     GNUNET_MESH_tunnel_destroy (tunnel);
2735     return NULL;
2736   }
2737   socket = GNUNET_malloc (sizeof (struct GNUNET_STREAM_Socket));
2738   socket->other_peer = *initiator;
2739   socket->tunnel = tunnel;
2740   socket->session_id = 0;       /* FIXME */
2741   socket->state = STATE_INIT;
2742   socket->lsocket = lsocket;
2743   socket->retransmit_timeout = lsocket->retransmit_timeout;
2744   socket->testing_active = lsocket->testing_active;
2745   socket->testing_set_write_sequence_number_value =
2746       lsocket->testing_set_write_sequence_number_value;
2747   socket->max_payload_size = lsocket->max_payload_size;
2748   LOG (GNUNET_ERROR_TYPE_DEBUG,
2749        "%s: Peer %s initiated tunnel to us\n",
2750        GNUNET_i2s (&socket->other_peer),
2751        GNUNET_i2s (&socket->other_peer));
2752   /* FIXME: Copy MESH handle from lsocket to socket */
2753   return socket;
2754 }
2755
2756
2757 /**
2758  * Function called whenever an inbound tunnel is destroyed.  Should clean up
2759  * any associated state.  This function is NOT called if the client has
2760  * explicitly asked for the tunnel to be destroyed using
2761  * GNUNET_MESH_tunnel_destroy. It must NOT call GNUNET_MESH_tunnel_destroy on
2762  * the tunnel.
2763  *
2764  * @param cls closure (set from GNUNET_MESH_connect)
2765  * @param tunnel connection to the other end (henceforth invalid)
2766  * @param tunnel_ctx place where local state associated
2767  *                   with the tunnel is stored
2768  */
2769 static void
2770 tunnel_cleaner (void *cls,
2771                 const struct GNUNET_MESH_Tunnel *tunnel,
2772                 void *tunnel_ctx)
2773 {
2774   struct GNUNET_STREAM_Socket *socket = tunnel_ctx;
2775
2776   if (tunnel != socket->tunnel)
2777     return;
2778
2779   GNUNET_break_op(0);
2780   LOG (GNUNET_ERROR_TYPE_DEBUG,
2781        "%s: Peer %s has terminated connection abruptly\n",
2782        GNUNET_i2s (&socket->other_peer),
2783        GNUNET_i2s (&socket->other_peer));
2784
2785   socket->status = GNUNET_STREAM_SHUTDOWN;
2786
2787   /* Clear Transmit handles */
2788   if (NULL != socket->transmit_handle)
2789   {
2790     GNUNET_MESH_notify_transmit_ready_cancel (socket->transmit_handle);
2791     socket->transmit_handle = NULL;
2792   }
2793   /* Stop Tasks using socket->tunnel */
2794   if (GNUNET_SCHEDULER_NO_TASK != socket->ack_task_id)
2795   {
2796     GNUNET_SCHEDULER_cancel (socket->ack_task_id);
2797     socket->ack_task_id = GNUNET_SCHEDULER_NO_TASK;
2798   }
2799   if (GNUNET_SCHEDULER_NO_TASK != socket->data_retransmission_task_id)
2800   {
2801     GNUNET_SCHEDULER_cancel (socket->data_retransmission_task_id);
2802     socket->data_retransmission_task_id = GNUNET_SCHEDULER_NO_TASK;
2803   }
2804   /* FIXME: Cancel all other tasks using socket->tunnel */
2805   socket->tunnel = NULL;
2806 }
2807
2808
2809 /**
2810  * Callback to signal timeout on lockmanager lock acquire
2811  *
2812  * @param cls the ListenSocket
2813  * @param tc the scheduler task context
2814  */
2815 static void
2816 lockmanager_acquire_timeout (void *cls,
2817                              const struct GNUNET_SCHEDULER_TaskContext *tc)
2818 {
2819   struct GNUNET_STREAM_ListenSocket *lsocket = cls;
2820   GNUNET_STREAM_ListenCallback listen_cb;
2821   void *listen_cb_cls;
2822
2823   lsocket->lockmanager_acquire_timeout_task = GNUNET_SCHEDULER_NO_TASK;
2824   listen_cb = lsocket->listen_cb;
2825   listen_cb_cls = lsocket->listen_cb_cls;
2826   if (NULL != listen_cb)
2827     listen_cb (listen_cb_cls, NULL, NULL);
2828 }
2829
2830
2831 /**
2832  * Callback to notify us on the status changes on app_port lock
2833  *
2834  * @param cls the ListenSocket
2835  * @param domain the domain name of the lock
2836  * @param lock the app_port
2837  * @param status the current status of the lock
2838  */
2839 static void
2840 lock_status_change_cb (void *cls, const char *domain, uint32_t lock,
2841                        enum GNUNET_LOCKMANAGER_Status status)
2842 {
2843   struct GNUNET_STREAM_ListenSocket *lsocket = cls;
2844
2845   GNUNET_assert (lock == (uint32_t) lsocket->port);
2846   if (GNUNET_LOCKMANAGER_SUCCESS == status)
2847   {
2848     lsocket->listening = GNUNET_YES;
2849     if (GNUNET_SCHEDULER_NO_TASK != lsocket->lockmanager_acquire_timeout_task)
2850     {
2851       GNUNET_SCHEDULER_cancel (lsocket->lockmanager_acquire_timeout_task);
2852       lsocket->lockmanager_acquire_timeout_task = GNUNET_SCHEDULER_NO_TASK;
2853     }
2854     if (NULL == lsocket->mesh)
2855     {
2856       GNUNET_MESH_ApplicationType ports[] = {lsocket->port, 0};
2857
2858       lsocket->mesh = GNUNET_MESH_connect (lsocket->cfg,
2859                                            lsocket, /* Closure */
2860                                            &new_tunnel_notify,
2861                                            &tunnel_cleaner,
2862                                            server_message_handlers,
2863                                            ports);
2864       GNUNET_assert (NULL != lsocket->mesh);
2865       if (NULL != lsocket->listen_ok_cb)
2866       {
2867         (void) lsocket->listen_ok_cb ();
2868       }
2869     }
2870   }
2871   if (GNUNET_LOCKMANAGER_RELEASE == status)
2872     lsocket->listening = GNUNET_NO;
2873 }
2874
2875
2876 /*****************/
2877 /* API functions */
2878 /*****************/
2879
2880
2881 /**
2882  * Tries to open a stream to the target peer
2883  *
2884  * @param cfg configuration to use
2885  * @param target the target peer to which the stream has to be opened
2886  * @param app_port the application port number which uniquely identifies this
2887  *            stream
2888  * @param open_cb this function will be called after stream has be established;
2889  *          cannot be NULL
2890  * @param open_cb_cls the closure for open_cb
2891  * @param ... options to the stream, terminated by GNUNET_STREAM_OPTION_END
2892  * @return if successful it returns the stream socket; NULL if stream cannot be
2893  *         opened
2894  */
2895 struct GNUNET_STREAM_Socket *
2896 GNUNET_STREAM_open (const struct GNUNET_CONFIGURATION_Handle *cfg,
2897                     const struct GNUNET_PeerIdentity *target,
2898                     GNUNET_MESH_ApplicationType app_port,
2899                     GNUNET_STREAM_OpenCallback open_cb,
2900                     void *open_cb_cls,
2901                     ...)
2902 {
2903   struct GNUNET_STREAM_Socket *socket;
2904   enum GNUNET_STREAM_Option option;
2905   GNUNET_MESH_ApplicationType ports[] = {app_port, 0};
2906   va_list vargs;
2907   uint16_t payload_size;
2908
2909   LOG (GNUNET_ERROR_TYPE_DEBUG,
2910        "%s\n", __func__);
2911   GNUNET_assert (NULL != open_cb);
2912   socket = GNUNET_malloc (sizeof (struct GNUNET_STREAM_Socket));
2913   socket->other_peer = *target;
2914   socket->open_cb = open_cb;
2915   socket->open_cls = open_cb_cls;
2916   /* Set defaults */
2917   socket->retransmit_timeout = TIME_REL_SECS (default_timeout);
2918   socket->testing_active = GNUNET_NO;
2919   socket->max_payload_size = DEFAULT_MAX_PAYLOAD_SIZE;
2920   va_start (vargs, open_cb_cls); /* Parse variable args */
2921   do {
2922     option = va_arg (vargs, enum GNUNET_STREAM_Option);
2923     switch (option)
2924     {
2925     case GNUNET_STREAM_OPTION_INITIAL_RETRANSMIT_TIMEOUT:
2926       /* Expect struct GNUNET_TIME_Relative */
2927       socket->retransmit_timeout = va_arg (vargs,
2928                                            struct GNUNET_TIME_Relative);
2929       break;
2930     case GNUNET_STREAM_OPTION_TESTING_SET_WRITE_SEQUENCE_NUMBER:
2931       socket->testing_active = GNUNET_YES;
2932       socket->testing_set_write_sequence_number_value = va_arg (vargs,
2933                                                                 uint32_t);
2934       break;
2935     case GNUNET_STREAM_OPTION_LISTEN_TIMEOUT:
2936       GNUNET_break (0);          /* Option irrelevant in STREAM_open */
2937       break;
2938     case GNUNET_STREAM_OPTION_SIGNAL_LISTEN_SUCCESS:
2939       GNUNET_break (0);          /* Option irrelevant in STREAM_open */
2940       break;
2941     case GNUNET_STREAM_OPTION_MAX_PAYLOAD_SIZE:
2942       payload_size = (uint16_t) va_arg (vargs, unsigned int);
2943       GNUNET_assert (0 != payload_size);
2944       if (payload_size < socket->max_payload_size)
2945         socket->max_payload_size = payload_size;
2946       break;
2947     case GNUNET_STREAM_OPTION_END:
2948       break;
2949     }
2950   } while (GNUNET_STREAM_OPTION_END != option);
2951   va_end (vargs);               /* End of variable args parsing */
2952   socket->mesh = GNUNET_MESH_connect (cfg, /* the configuration handle */
2953                                       socket, /* cls */
2954                                       NULL, /* No inbound tunnel handler */
2955                                       NULL, /* No in-tunnel cleaner */
2956                                       client_message_handlers,
2957                                       ports); /* We don't get inbound tunnels */
2958   if (NULL == socket->mesh)   /* Fail if we cannot connect to mesh */
2959   {
2960     GNUNET_free (socket);
2961     return NULL;
2962   }
2963   /* Now create the mesh tunnel to target */
2964   LOG (GNUNET_ERROR_TYPE_DEBUG, "Creating MESH Tunnel\n");
2965   socket->tunnel = GNUNET_MESH_tunnel_create (socket->mesh,
2966                                               NULL, /* Tunnel context */
2967                                               &mesh_peer_connect_callback,
2968                                               &mesh_peer_disconnect_callback,
2969                                               socket);
2970   GNUNET_assert (NULL != socket->tunnel);
2971   GNUNET_MESH_peer_request_connect_add (socket->tunnel,
2972                                         &socket->other_peer);
2973   LOG (GNUNET_ERROR_TYPE_DEBUG, "%s() END\n", __func__);
2974   return socket;
2975 }
2976
2977
2978 /**
2979  * Shutdown the stream for reading or writing (similar to man 2 shutdown).
2980  *
2981  * @param socket the stream socket
2982  * @param operation SHUT_RD, SHUT_WR or SHUT_RDWR
2983  * @param completion_cb the callback that will be called upon successful
2984  *          shutdown of given operation
2985  * @param completion_cls the closure for the completion callback
2986  * @return the shutdown handle
2987  */
2988 struct GNUNET_STREAM_ShutdownHandle *
2989 GNUNET_STREAM_shutdown (struct GNUNET_STREAM_Socket *socket,
2990                         int operation,
2991                         GNUNET_STREAM_ShutdownCompletion completion_cb,
2992                         void *completion_cls)
2993 {
2994   struct GNUNET_STREAM_ShutdownHandle *handle;
2995   struct GNUNET_STREAM_MessageHeader *msg;
2996
2997   GNUNET_assert (NULL == socket->shutdown_handle);
2998
2999   handle = GNUNET_malloc (sizeof (struct GNUNET_STREAM_ShutdownHandle));
3000   handle->socket = socket;
3001   handle->completion_cb = completion_cb;
3002   handle->completion_cls = completion_cls;
3003   socket->shutdown_handle = handle;
3004
3005   msg = GNUNET_malloc (sizeof (struct GNUNET_STREAM_MessageHeader));
3006   msg->header.size = htons (sizeof (struct GNUNET_STREAM_MessageHeader));
3007   switch (operation)
3008   {
3009   case SHUT_RD:
3010     handle->operation = SHUT_RD;
3011     if (NULL != socket->read_handle)
3012       LOG (GNUNET_ERROR_TYPE_WARNING,
3013            "Existing read handle should be cancelled before shutting"
3014            " down reading\n");
3015     msg->header.type = htons (GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE);
3016     queue_message (socket, msg, &set_state_receive_close_wait, NULL,
3017                    GNUNET_NO);
3018     break;
3019   case SHUT_WR:
3020     handle->operation = SHUT_WR;
3021     if (NULL != socket->write_handle)
3022       LOG (GNUNET_ERROR_TYPE_WARNING,
3023            "Existing write handle should be cancelled before shutting"
3024            " down writing\n");
3025     msg->header.type = htons (GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE);
3026     queue_message (socket, msg, &set_state_transmit_close_wait, NULL,
3027                    GNUNET_NO);
3028     break;
3029   case SHUT_RDWR:
3030     handle->operation = SHUT_RDWR;
3031     if (NULL != socket->write_handle)
3032       LOG (GNUNET_ERROR_TYPE_WARNING,
3033            "Existing write handle should be cancelled before shutting"
3034            " down writing\n");
3035     if (NULL != socket->read_handle)
3036       LOG (GNUNET_ERROR_TYPE_WARNING,
3037            "Existing read handle should be cancelled before shutting"
3038            " down reading\n");
3039     msg->header.type = htons (GNUNET_MESSAGE_TYPE_STREAM_CLOSE);
3040     queue_message (socket, msg, &set_state_close_wait, NULL, GNUNET_NO);
3041     break;
3042   default:
3043     LOG (GNUNET_ERROR_TYPE_WARNING,
3044          "GNUNET_STREAM_shutdown called with invalid value for "
3045          "parameter operation -- Ignoring\n");
3046     GNUNET_free (msg);
3047     GNUNET_free (handle);
3048     return NULL;
3049   }
3050   handle->close_msg_retransmission_task_id =
3051     GNUNET_SCHEDULER_add_delayed (socket->retransmit_timeout,
3052                                   &close_msg_retransmission_task,
3053                                   handle);
3054   return handle;
3055 }
3056
3057
3058 /**
3059  * Cancels a pending shutdown
3060  *
3061  * @param handle the shutdown handle returned from GNUNET_STREAM_shutdown
3062  */
3063 void
3064 GNUNET_STREAM_shutdown_cancel (struct GNUNET_STREAM_ShutdownHandle *handle)
3065 {
3066   if (GNUNET_SCHEDULER_NO_TASK != handle->close_msg_retransmission_task_id)
3067     GNUNET_SCHEDULER_cancel (handle->close_msg_retransmission_task_id);
3068   handle->socket->shutdown_handle = NULL;
3069   GNUNET_free (handle);
3070 }
3071
3072
3073 /**
3074  * Closes the stream
3075  *
3076  * @param socket the stream socket
3077  */
3078 void
3079 GNUNET_STREAM_close (struct GNUNET_STREAM_Socket *socket)
3080 {
3081   struct MessageQueue *head;
3082
3083   if (NULL != socket->read_handle)
3084   {
3085     LOG (GNUNET_ERROR_TYPE_WARNING,
3086          "Closing STREAM socket when a read handle is pending\n");
3087     GNUNET_STREAM_io_read_cancel (socket->read_handle);
3088   }
3089   if (NULL != socket->write_handle)
3090   {
3091     LOG (GNUNET_ERROR_TYPE_WARNING,
3092          "Closing STREAM socket when a write handle is pending\n");
3093     GNUNET_STREAM_io_write_cancel (socket->write_handle);
3094     //socket->write_handle = NULL;
3095   }
3096   if (socket->read_task_id != GNUNET_SCHEDULER_NO_TASK)
3097   {
3098     /* socket closed with read task pending!? */
3099     GNUNET_break (0);
3100     GNUNET_SCHEDULER_cancel (socket->read_task_id);
3101     socket->read_task_id = GNUNET_SCHEDULER_NO_TASK;
3102   }
3103   /* Terminate the ack'ing tasks if they are still present */
3104   if (socket->ack_task_id != GNUNET_SCHEDULER_NO_TASK)
3105   {
3106     GNUNET_SCHEDULER_cancel (socket->ack_task_id);
3107     socket->ack_task_id = GNUNET_SCHEDULER_NO_TASK;
3108   }
3109   /* Terminate the control retransmission tasks */
3110   if (GNUNET_SCHEDULER_NO_TASK != socket->control_retransmission_task_id)
3111   {
3112     GNUNET_SCHEDULER_cancel (socket->control_retransmission_task_id);
3113   }
3114   /* Clear Transmit handles */
3115   if (NULL != socket->transmit_handle)
3116   {
3117     GNUNET_MESH_notify_transmit_ready_cancel (socket->transmit_handle);
3118     socket->transmit_handle = NULL;
3119   }
3120   /* Clear existing message queue */
3121   while (NULL != (head = socket->queue_head)) {
3122     GNUNET_CONTAINER_DLL_remove (socket->queue_head,
3123                                  socket->queue_tail,
3124                                  head);
3125     GNUNET_free (head->message);
3126     GNUNET_free (head);
3127   }
3128   /* Close associated tunnel */
3129   if (NULL != socket->tunnel)
3130   {
3131     GNUNET_MESH_tunnel_destroy (socket->tunnel);
3132     socket->tunnel = NULL;
3133   }
3134   /* Close mesh connection */
3135   if (NULL != socket->mesh && NULL == socket->lsocket)
3136   {
3137     GNUNET_MESH_disconnect (socket->mesh);
3138     socket->mesh = NULL;
3139   }
3140   /* Release receive buffer */
3141   if (NULL != socket->receive_buffer)
3142   {
3143     GNUNET_free (socket->receive_buffer);
3144   }
3145   GNUNET_free (socket);
3146 }
3147
3148
3149 /**
3150  * Listens for stream connections for a specific application ports
3151  *
3152  * @param cfg the configuration to use
3153  * @param app_port the application port for which new streams will be accepted
3154  * @param listen_cb this function will be called when a peer tries to establish
3155  *            a stream with us
3156  * @param listen_cb_cls closure for listen_cb
3157  * @param ... options to the stream, terminated by GNUNET_STREAM_OPTION_END
3158  * @return listen socket, NULL for any error
3159  */
3160 struct GNUNET_STREAM_ListenSocket *
3161 GNUNET_STREAM_listen (const struct GNUNET_CONFIGURATION_Handle *cfg,
3162                       GNUNET_MESH_ApplicationType app_port,
3163                       GNUNET_STREAM_ListenCallback listen_cb,
3164                       void *listen_cb_cls,
3165                       ...)
3166 {
3167   /* FIXME: Add variable args for passing configration options? */
3168   struct GNUNET_STREAM_ListenSocket *lsocket;
3169   struct GNUNET_TIME_Relative listen_timeout;
3170   enum GNUNET_STREAM_Option option;
3171   va_list vargs;
3172   uint16_t payload_size;
3173
3174   GNUNET_assert (NULL != listen_cb);
3175   lsocket = GNUNET_malloc (sizeof (struct GNUNET_STREAM_ListenSocket));
3176   lsocket->cfg = GNUNET_CONFIGURATION_dup (cfg);
3177   lsocket->lockmanager = GNUNET_LOCKMANAGER_connect (lsocket->cfg);
3178   if (NULL == lsocket->lockmanager)
3179   {
3180     GNUNET_CONFIGURATION_destroy (lsocket->cfg);
3181     GNUNET_free (lsocket);
3182     return NULL;
3183   }
3184   lsocket->listening = GNUNET_NO;/* We listen when we get a lock on app_port */
3185   /* Set defaults */
3186   lsocket->retransmit_timeout = TIME_REL_SECS (default_timeout);
3187   lsocket->testing_active = GNUNET_NO;
3188   lsocket->listen_ok_cb = NULL;
3189   lsocket->max_payload_size = DEFAULT_MAX_PAYLOAD_SIZE;
3190   listen_timeout = TIME_REL_SECS (60); /* A minute for listen timeout */
3191   va_start (vargs, listen_cb_cls);
3192   do {
3193     option = va_arg (vargs, enum GNUNET_STREAM_Option);
3194     switch (option)
3195     {
3196     case GNUNET_STREAM_OPTION_INITIAL_RETRANSMIT_TIMEOUT:
3197       lsocket->retransmit_timeout = va_arg (vargs,
3198                                             struct GNUNET_TIME_Relative);
3199       break;
3200     case GNUNET_STREAM_OPTION_TESTING_SET_WRITE_SEQUENCE_NUMBER:
3201       lsocket->testing_active = GNUNET_YES;
3202       lsocket->testing_set_write_sequence_number_value = va_arg (vargs,
3203                                                                  uint32_t);
3204       break;
3205     case GNUNET_STREAM_OPTION_LISTEN_TIMEOUT:
3206       listen_timeout = GNUNET_TIME_relative_multiply
3207         (GNUNET_TIME_UNIT_MILLISECONDS, va_arg (vargs, uint32_t));
3208       break;
3209     case GNUNET_STREAM_OPTION_SIGNAL_LISTEN_SUCCESS:
3210       lsocket->listen_ok_cb = va_arg (vargs,
3211                                       GNUNET_STREAM_ListenSuccessCallback);
3212       break;
3213     case GNUNET_STREAM_OPTION_MAX_PAYLOAD_SIZE:
3214       payload_size = (uint16_t) va_arg (vargs, unsigned int);
3215       GNUNET_assert (0 != payload_size);
3216       if (payload_size < lsocket->max_payload_size)
3217         lsocket->max_payload_size = payload_size;
3218       break;
3219     case GNUNET_STREAM_OPTION_END:
3220       break;
3221     }
3222   } while (GNUNET_STREAM_OPTION_END != option);
3223   va_end (vargs);
3224   lsocket->port = app_port;
3225   lsocket->listen_cb = listen_cb;
3226   lsocket->listen_cb_cls = listen_cb_cls;
3227   lsocket->locking_request =
3228     GNUNET_LOCKMANAGER_acquire_lock (lsocket->lockmanager, locking_domain,
3229                                      (uint32_t) lsocket->port,
3230                                      &lock_status_change_cb, lsocket);
3231   lsocket->lockmanager_acquire_timeout_task =
3232     GNUNET_SCHEDULER_add_delayed (listen_timeout,
3233                                   &lockmanager_acquire_timeout, lsocket);
3234   return lsocket;
3235 }
3236
3237
3238 /**
3239  * Closes the listen socket
3240  *
3241  * @param lsocket the listen socket
3242  */
3243 void
3244 GNUNET_STREAM_listen_close (struct GNUNET_STREAM_ListenSocket *lsocket)
3245 {
3246   /* Close MESH connection */
3247   if (NULL != lsocket->mesh)
3248     GNUNET_MESH_disconnect (lsocket->mesh);
3249   GNUNET_CONFIGURATION_destroy (lsocket->cfg);
3250   if (GNUNET_SCHEDULER_NO_TASK != lsocket->lockmanager_acquire_timeout_task)
3251     GNUNET_SCHEDULER_cancel (lsocket->lockmanager_acquire_timeout_task);
3252   if (NULL != lsocket->locking_request)
3253     GNUNET_LOCKMANAGER_cancel_request (lsocket->locking_request);
3254   if (NULL != lsocket->lockmanager)
3255     GNUNET_LOCKMANAGER_disconnect (lsocket->lockmanager);
3256   GNUNET_free (lsocket);
3257 }
3258
3259
3260 /**
3261  * Tries to write the given data to the stream. The maximum size of data that
3262  * can be written as part of a write operation is (64 * (64000 - sizeof (struct
3263  * GNUNET_STREAM_DataMessage))). If size is greater than this it is not an API
3264  * violation, however only the said number of maximum bytes will be written.
3265  *
3266  * @param socket the socket representing a stream
3267  * @param data the data buffer from where the data is written into the stream
3268  * @param size the number of bytes to be written from the data buffer
3269  * @param timeout the timeout period
3270  * @param write_cont the function to call upon writing some bytes into the
3271  *          stream
3272  * @param write_cont_cls the closure
3273  *
3274  * @return handle to cancel the operation; if a previous write is pending or
3275  *           the stream has been shutdown for this operation then write_cont is
3276  *           immediately called and NULL is returned.
3277  */
3278 struct GNUNET_STREAM_IOWriteHandle *
3279 GNUNET_STREAM_write (struct GNUNET_STREAM_Socket *socket,
3280                      const void *data,
3281                      size_t size,
3282                      struct GNUNET_TIME_Relative timeout,
3283                      GNUNET_STREAM_CompletionContinuation write_cont,
3284                      void *write_cont_cls)
3285 {
3286   struct GNUNET_STREAM_IOWriteHandle *io_handle;
3287   struct GNUNET_STREAM_DataMessage *data_msg;
3288   const void *sweep;
3289   struct GNUNET_TIME_Relative ack_deadline;
3290   unsigned int num_needed_packets;
3291   unsigned int packet;
3292   uint32_t packet_size;
3293   uint32_t payload_size;
3294   uint16_t max_data_packet_size;
3295
3296   LOG (GNUNET_ERROR_TYPE_DEBUG,
3297        "%s\n", __func__);
3298   if (NULL != socket->write_handle)
3299   {
3300     GNUNET_break (0);
3301     return NULL;
3302   }
3303   switch (socket->state)
3304   {
3305   case STATE_TRANSMIT_CLOSED:
3306   case STATE_TRANSMIT_CLOSE_WAIT:
3307   case STATE_CLOSED:
3308   case STATE_CLOSE_WAIT:
3309     if (NULL != write_cont)
3310       write_cont (write_cont_cls, GNUNET_STREAM_SHUTDOWN, 0);
3311     LOG (GNUNET_ERROR_TYPE_DEBUG,
3312          "%s() END\n", __func__);
3313     return NULL;
3314   case STATE_INIT:
3315   case STATE_LISTEN:
3316   case STATE_HELLO_WAIT:
3317     if (NULL != write_cont)
3318       /* FIXME: GNUNET_STREAM_SYSERR?? */
3319       write_cont (write_cont_cls, GNUNET_STREAM_SYSERR, 0);
3320     LOG (GNUNET_ERROR_TYPE_DEBUG,
3321          "%s() END\n", __func__);
3322     return NULL;
3323   case STATE_ESTABLISHED:
3324   case STATE_RECEIVE_CLOSED:
3325   case STATE_RECEIVE_CLOSE_WAIT:
3326     break;
3327   }
3328   if (GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH * socket->max_payload_size < size)
3329     size = GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH  * socket->max_payload_size;
3330   num_needed_packets =
3331       (size + (socket->max_payload_size - 1)) / socket->max_payload_size;
3332   io_handle = GNUNET_malloc (sizeof (struct GNUNET_STREAM_IOWriteHandle));
3333   io_handle->socket = socket;
3334   io_handle->write_cont = write_cont;
3335   io_handle->write_cont_cls = write_cont_cls;
3336   io_handle->size = size;
3337   io_handle->packets_sent = 0;
3338   sweep = data;
3339   /* FIXME: Remove the fixed delay for ack deadline; Set it to the value
3340      determined from RTT */
3341   ack_deadline = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5);
3342   /* Divide the given buffer into packets for sending */
3343   max_data_packet_size =
3344       socket->max_payload_size + sizeof (struct GNUNET_STREAM_DataMessage);
3345   for (packet=0; packet < num_needed_packets; packet++)
3346   {
3347     if ((packet + 1) * socket->max_payload_size < size)
3348     {
3349       payload_size = socket->max_payload_size;
3350       packet_size = max_data_packet_size;
3351     }
3352     else
3353     {
3354       payload_size = size - packet * socket->max_payload_size;
3355       packet_size =
3356           payload_size + sizeof (struct GNUNET_STREAM_DataMessage);
3357     }
3358     io_handle->messages[packet] = GNUNET_malloc (packet_size);
3359     io_handle->messages[packet]->header.header.size = htons (packet_size);
3360     io_handle->messages[packet]->header.header.type =
3361       htons (GNUNET_MESSAGE_TYPE_STREAM_DATA);
3362     io_handle->messages[packet]->sequence_number =
3363       htonl (socket->write_sequence_number++);
3364     io_handle->messages[packet]->offset = htonl (socket->write_offset);
3365     /* FIXME: Remove the fixed delay for ack deadline; Set it to the value
3366        determined from RTT */
3367     io_handle->messages[packet]->ack_deadline =
3368       GNUNET_TIME_relative_hton (ack_deadline);
3369     data_msg = io_handle->messages[packet];
3370     /* Copy data from given buffer to the packet */
3371     memcpy (&data_msg[1], sweep, payload_size);
3372     sweep += payload_size;
3373     socket->write_offset += payload_size;
3374   }
3375   /* ack the last data message. FIXME: remove when we figure out how to do this
3376      using RTT */
3377   io_handle->messages[num_needed_packets - 1]->ack_deadline =
3378       GNUNET_TIME_relative_hton (GNUNET_TIME_UNIT_ZERO);
3379   socket->write_handle = io_handle;
3380   write_data (socket);
3381   LOG (GNUNET_ERROR_TYPE_DEBUG,
3382        "%s() END\n", __func__);
3383   return io_handle;
3384 }
3385
3386
3387 /**
3388  * Tries to read data from the stream.
3389  *
3390  * @param socket the socket representing a stream
3391  * @param timeout the timeout period
3392  * @param proc function to call with data (once only)
3393  * @param proc_cls the closure for proc
3394  *
3395  * @return handle to cancel the operation; if the stream has been shutdown for
3396  *           this type of opeartion then the DataProcessor is immediately
3397  *           called with GNUNET_STREAM_SHUTDOWN as status and NULL if returned
3398  */
3399 struct GNUNET_STREAM_IOReadHandle *
3400 GNUNET_STREAM_read (struct GNUNET_STREAM_Socket *socket,
3401                     struct GNUNET_TIME_Relative timeout,
3402                     GNUNET_STREAM_DataProcessor proc,
3403                     void *proc_cls)
3404 {
3405   struct GNUNET_STREAM_IOReadHandle *read_handle;
3406
3407   LOG (GNUNET_ERROR_TYPE_DEBUG,
3408        "%s: %s()\n",
3409        GNUNET_i2s (&socket->other_peer),
3410        __func__);
3411   /* Return NULL if there is already a read handle; the user has to cancel that
3412      first before continuing or has to wait until it is completed */
3413   if (NULL != socket->read_handle)
3414     return NULL;
3415   GNUNET_assert (NULL != proc);
3416   switch (socket->state)
3417   {
3418   case STATE_RECEIVE_CLOSED:
3419   case STATE_RECEIVE_CLOSE_WAIT:
3420   case STATE_CLOSED:
3421   case STATE_CLOSE_WAIT:
3422     proc (proc_cls, GNUNET_STREAM_SHUTDOWN, NULL, 0);
3423     LOG (GNUNET_ERROR_TYPE_DEBUG,
3424          "%s: %s() END\n",
3425          GNUNET_i2s (&socket->other_peer),
3426          __func__);
3427     return NULL;
3428   default:
3429     break;
3430   }
3431   read_handle = GNUNET_malloc (sizeof (struct GNUNET_STREAM_IOReadHandle));
3432   read_handle->proc = proc;
3433   read_handle->proc_cls = proc_cls;
3434   read_handle->socket = socket;
3435   socket->read_handle = read_handle;
3436   /* Check if we have a packet at bitmap 0 */
3437   if (GNUNET_YES == ackbitmap_is_bit_set (&socket->ack_bitmap,
3438                                           0))
3439   {
3440     socket->read_task_id = GNUNET_SCHEDULER_add_now (&call_read_processor,
3441                                                      socket);
3442   }
3443   /* Setup the read timeout task */
3444   socket->read_io_timeout_task_id =
3445     GNUNET_SCHEDULER_add_delayed (timeout,
3446                                   &read_io_timeout,
3447                                   socket);
3448   LOG (GNUNET_ERROR_TYPE_DEBUG,
3449        "%s: %s() END\n",
3450        GNUNET_i2s (&socket->other_peer),
3451        __func__);
3452   return read_handle;
3453 }
3454
3455
3456 /**
3457  * Cancel pending write operation.
3458  *
3459  * @param ioh handle to operation to cancel
3460  */
3461 void
3462 GNUNET_STREAM_io_write_cancel (struct GNUNET_STREAM_IOWriteHandle *ioh)
3463 {
3464   struct GNUNET_STREAM_Socket *socket = ioh->socket;
3465   unsigned int packet;
3466
3467   GNUNET_assert (NULL != socket->write_handle);
3468   GNUNET_assert (socket->write_handle == ioh);
3469
3470   if (GNUNET_SCHEDULER_NO_TASK != socket->data_retransmission_task_id)
3471   {
3472     GNUNET_SCHEDULER_cancel (socket->data_retransmission_task_id);
3473     socket->data_retransmission_task_id = GNUNET_SCHEDULER_NO_TASK;
3474   }
3475
3476   for (packet=0; packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH; packet++)
3477   {
3478     if (NULL == ioh->messages[packet]) break;
3479     GNUNET_free (ioh->messages[packet]);
3480   }
3481
3482   GNUNET_free (socket->write_handle);
3483   socket->write_handle = NULL;
3484 }
3485
3486
3487 /**
3488  * Cancel pending read operation.
3489  *
3490  * @param ioh handle to operation to cancel
3491  */
3492 void
3493 GNUNET_STREAM_io_read_cancel (struct GNUNET_STREAM_IOReadHandle *ioh)
3494 {
3495   struct GNUNET_STREAM_Socket *socket;
3496
3497   socket = ioh->socket;
3498   GNUNET_assert (NULL != socket->read_handle);
3499   GNUNET_assert (ioh == socket->read_handle);
3500   /* Read io time task should be there; if it is already executed then this
3501   read handle is not valid */
3502   GNUNET_assert (GNUNET_SCHEDULER_NO_TASK != socket->read_io_timeout_task_id);
3503   GNUNET_SCHEDULER_cancel (socket->read_io_timeout_task_id);
3504   socket->read_io_timeout_task_id = GNUNET_SCHEDULER_NO_TASK;
3505   /* reading task may be present; if so we have to stop it */
3506   if (GNUNET_SCHEDULER_NO_TASK != socket->read_task_id)
3507   {
3508     GNUNET_SCHEDULER_cancel (socket->read_task_id);
3509     socket->read_task_id = GNUNET_SCHEDULER_NO_TASK;
3510   }
3511   GNUNET_free (ioh);
3512   socket->read_handle = NULL;
3513 }
3514
3515 /* end of stream_api.c */