src/stream/stream_api.c

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