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