2 This file is part of GNUnet.
3 (C) 2012 Christian Grothoff (and other contributing authors)
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.
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.
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.
22 * Copy MESH handle from lsocket to socket
24 * Checks for matching the sender and socket->other_peer in server
28 * @file stream/stream_api.c
29 * @brief Implementation of the stream library
30 * @author Sree Harsha Totakura
33 #include "gnunet_common.h"
34 #include "gnunet_crypto_lib.h"
35 #include "gnunet_stream_lib.h"
36 #include "gnunet_testing_lib.h"
37 #include "stream_protocol.h"
41 * The maximum packet size of a stream packet
43 #define MAX_PACKET_SIZE 64000
46 * The maximum payload a data message packet can carry
48 static size_t max_payload_size =
49 MAX_PACKET_SIZE - sizeof (struct GNUNET_STREAM_DataMessage);
54 #define RECEIVE_BUFFER_SIZE 4096000
57 * states in the Protocol
62 * Client initialization state
67 * Listener initialization state
72 * Pre-connection establishment state
77 * State where a connection has been established
82 * State where the socket is closed on our side and waiting to be ACK'ed
84 STATE_RECEIVE_CLOSE_WAIT,
87 * State where the socket is closed for reading
92 * State where the socket is closed on our side and waiting to be ACK'ed
94 STATE_TRANSMIT_CLOSE_WAIT,
97 * State where the socket is closed for writing
99 STATE_TRANSMIT_CLOSED,
102 * State where the socket is closed on our side and waiting to be ACK'ed
107 * State where the socket is closed
114 * Functions of this type are called when a message is written
116 * @param cls the closure from queue_message
117 * @param socket the socket the written message was bound to
119 typedef void (*SendFinishCallback) (void *cls,
120 struct GNUNET_STREAM_Socket *socket);
124 * The send message queue
131 struct GNUNET_STREAM_MessageHeader *message;
134 * Callback to be called when the message is sent
136 SendFinishCallback finish_cb;
139 * The closure for finish_cb
144 * The next message in queue. Should be NULL in the last message
146 struct MessageQueue *next;
149 * The next message in queue. Should be NULL in the first message
151 struct MessageQueue *prev;
156 * The STREAM Socket Handler
158 struct GNUNET_STREAM_Socket
162 * The peer identity of the peer at the other end of the stream
164 struct GNUNET_PeerIdentity other_peer;
167 * Our Peer Identity (for debugging)
169 struct GNUNET_PeerIdentity our_id;
172 * Retransmission timeout
174 struct GNUNET_TIME_Relative retransmit_timeout;
177 * The Acknowledgement Bitmap
179 GNUNET_STREAM_AckBitmap ack_bitmap;
182 * Time when the Acknowledgement was queued
184 struct GNUNET_TIME_Absolute ack_time_registered;
187 * Queued Acknowledgement deadline
189 struct GNUNET_TIME_Relative ack_time_deadline;
192 * The task for sending timely Acks
194 GNUNET_SCHEDULER_TaskIdentifier ack_task_id;
197 * Task scheduled to continue a read operation.
199 GNUNET_SCHEDULER_TaskIdentifier read_task;
204 struct GNUNET_MESH_Handle *mesh;
207 * The mesh tunnel handle
209 struct GNUNET_MESH_Tunnel *tunnel;
212 * Stream open closure
217 * Stream open callback
219 GNUNET_STREAM_OpenCallback open_cb;
222 * The current transmit handle (if a pending transmit request exists)
224 struct GNUNET_MESH_TransmitHandle *transmit_handle;
227 * The current message associated with the transmit handle
229 struct MessageQueue *queue_head;
232 * The queue tail, should always point to the last message in queue
234 struct MessageQueue *queue_tail;
237 * The write IO_handle associated with this socket
239 struct GNUNET_STREAM_IOWriteHandle *write_handle;
242 * The read IO_handle associated with this socket
244 struct GNUNET_STREAM_IOReadHandle *read_handle;
247 * Buffer for storing received messages
249 void *receive_buffer;
252 * Task identifier for the read io timeout task
254 GNUNET_SCHEDULER_TaskIdentifier read_io_timeout_task;
257 * The state of the protocol associated with this socket
262 * The status of the socket
264 enum GNUNET_STREAM_Status status;
267 * The number of previous timeouts; FIXME: currently not used
269 unsigned int retries;
272 * Is this socket derived from listen socket?
274 unsigned int derived;
277 * The application port number (type: uint32_t)
279 GNUNET_MESH_ApplicationType app_port;
282 * The session id associated with this stream connection
283 * FIXME: Not used currently, may be removed
288 * Write sequence number. Set to random when sending HELLO(client) and
291 uint32_t write_sequence_number;
294 * Read sequence number. This number's value is determined during handshake
296 uint32_t read_sequence_number;
299 * The receiver buffer size
301 uint32_t receive_buffer_size;
304 * The receiver buffer boundaries
306 uint32_t receive_buffer_boundaries[GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH];
309 * receiver's available buffer after the last acknowledged packet
311 uint32_t receive_window_available;
314 * The offset pointer used during write operation
316 uint32_t write_offset;
319 * The offset after which we are expecting data
321 uint32_t read_offset;
324 * The offset upto which user has read from the received buffer
326 uint32_t copy_offset;
331 * A socket for listening
333 struct GNUNET_STREAM_ListenSocket
337 * Our Peer's identity
339 struct GNUNET_PeerIdentity our_id;
344 struct GNUNET_MESH_Handle *mesh;
347 * The callback function which is called after successful opening socket
349 GNUNET_STREAM_ListenCallback listen_cb;
352 * The call back closure
358 * FIXME: Remove if not required!
360 GNUNET_MESH_ApplicationType port;
365 * The IO Write Handle
367 struct GNUNET_STREAM_IOWriteHandle
370 * The packet_buffers associated with this Handle
372 struct GNUNET_STREAM_DataMessage *messages[64];
375 * The bitmap of this IOHandle; Corresponding bit for a message is set when
376 * it has been acknowledged by the receiver
378 GNUNET_STREAM_AckBitmap ack_bitmap;
381 * Number of packets sent before waiting for an ack
383 * FIXME: Do we need this?
385 unsigned int sent_packets;
392 struct GNUNET_STREAM_IOReadHandle
395 * Callback for the read processor
397 GNUNET_STREAM_DataProcessor proc;
400 * The closure pointer for the read processor callback
407 * Default value in seconds for various timeouts
409 static unsigned int default_timeout = 300;
413 * Callback function for sending hello message
415 * @param cls closure the socket
416 * @param size number of bytes available in buf
417 * @param buf where the callee should write the message
418 * @return number of bytes written to buf
421 send_message_notify (void *cls, size_t size, void *buf)
423 struct GNUNET_STREAM_Socket *socket = cls;
424 struct MessageQueue *head;
427 socket->transmit_handle = NULL; /* Remove the transmit handle */
428 head = socket->queue_head;
430 return 0; /* just to be safe */
431 if (0 == size) /* request timed out */
434 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
435 "Message sending timed out. Retry %d \n",
437 socket->transmit_handle =
438 GNUNET_MESH_notify_transmit_ready (socket->tunnel,
441 /* FIXME: exponential backoff */
442 socket->retransmit_timeout,
444 ntohs (head->message->header.size),
445 &send_message_notify,
450 ret = ntohs (head->message->header.size);
451 GNUNET_assert (size >= ret);
452 memcpy (buf, head->message, ret);
453 if (NULL != head->finish_cb)
455 head->finish_cb (head->finish_cb_cls, socket);
457 GNUNET_CONTAINER_DLL_remove (socket->queue_head,
460 GNUNET_free (head->message);
462 head = socket->queue_head;
463 if (NULL != head) /* more pending messages to send */
466 socket->transmit_handle =
467 GNUNET_MESH_notify_transmit_ready (socket->tunnel,
470 /* FIXME: exponential backoff */
471 socket->retransmit_timeout,
473 ntohs (head->message->header.size),
474 &send_message_notify,
482 * Queues a message for sending using the mesh connection of a socket
484 * @param socket the socket whose mesh connection is used
485 * @param message the message to be sent
486 * @param finish_cb the callback to be called when the message is sent
487 * @param finish_cb_cls the closure for the callback
490 queue_message (struct GNUNET_STREAM_Socket *socket,
491 struct GNUNET_STREAM_MessageHeader *message,
492 SendFinishCallback finish_cb,
495 struct MessageQueue *queue_entity;
497 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
498 "Queueing message of type %d and size %d\n",
499 ntohs (message->header.type),
500 ntohs (message->header.size));
501 queue_entity = GNUNET_malloc (sizeof (struct MessageQueue));
502 queue_entity->message = message;
503 queue_entity->finish_cb = finish_cb;
504 queue_entity->finish_cb_cls = finish_cb_cls;
505 GNUNET_CONTAINER_DLL_insert_tail (socket->queue_head,
508 if (NULL == socket->transmit_handle)
511 socket->transmit_handle =
512 GNUNET_MESH_notify_transmit_ready (socket->tunnel,
515 socket->retransmit_timeout,
517 ntohs (message->header.size),
518 &send_message_notify,
525 * Callback function for sending ack message
527 * @param cls closure the ACK message created in ack_task
528 * @param size number of bytes available in buffer
529 * @param buf where the callee should write the message
530 * @return number of bytes written to buf
533 send_ack_notify (void *cls, size_t size, void *buf)
535 struct GNUNET_STREAM_AckMessage *ack_msg = cls;
539 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
540 "%s called with size 0\n", __func__);
543 GNUNET_assert (ack_msg->header.header.size <= size);
545 size = ack_msg->header.header.size;
546 memcpy (buf, ack_msg, size);
552 * Task for sending ACK message
554 * @param cls the socket
555 * @param tc the Task context
559 const struct GNUNET_SCHEDULER_TaskContext *tc)
561 struct GNUNET_STREAM_Socket *socket = cls;
562 struct GNUNET_STREAM_AckMessage *ack_msg;
564 if (GNUNET_SCHEDULER_REASON_SHUTDOWN == tc->reason)
569 socket->ack_task_id = 0;
571 /* Create the ACK Message */
572 ack_msg = GNUNET_malloc (sizeof (struct GNUNET_STREAM_AckMessage));
573 ack_msg->header.header.size = htons (sizeof (struct
574 GNUNET_STREAM_AckMessage));
575 ack_msg->header.header.type = htons (GNUNET_MESSAGE_TYPE_STREAM_ACK);
576 ack_msg->bitmap = GNUNET_htonll (socket->ack_bitmap);
577 ack_msg->base_sequence_number = htonl (socket->read_sequence_number);
578 ack_msg->receive_window_remaining =
579 htonl (RECEIVE_BUFFER_SIZE - socket->receive_buffer_size);
581 /* Request MESH for sending ACK */
582 GNUNET_MESH_notify_transmit_ready (socket->tunnel,
585 socket->retransmit_timeout,
587 ntohs (ack_msg->header.header.size),
596 * Function to modify a bit in GNUNET_STREAM_AckBitmap
598 * @param bitmap the bitmap to modify
599 * @param bit the bit number to modify
600 * @param value GNUNET_YES to on, GNUNET_NO to off
603 ackbitmap_modify_bit (GNUNET_STREAM_AckBitmap *bitmap,
607 GNUNET_assert (bit < 64);
608 if (GNUNET_YES == value)
609 *bitmap |= (1LL << bit);
611 *bitmap &= ~(1LL << bit);
616 * Function to check if a bit is set in the GNUNET_STREAM_AckBitmap
618 * @param bitmap address of the bitmap that has to be checked
619 * @param bit the bit number to check
620 * @return GNUNET_YES if the bit is set; GNUNET_NO if not
623 ackbitmap_is_bit_set (const GNUNET_STREAM_AckBitmap *bitmap,
626 GNUNET_assert (bit < 64);
627 return 0 != (*bitmap & (1LL << bit));
633 * Function called when Data Message is sent
635 * @param cls the io_handle corresponding to the Data Message
636 * @param socket the socket which was used
639 write_data_finish_cb (void *cls,
640 struct GNUNET_STREAM_Socket *socket)
642 struct GNUNET_STREAM_IOWriteHandle *io_handle = cls;
644 io_handle->sent_packets++;
649 * Writes data using the given socket. The amount of data written is limited by
650 * the receive_window_size
652 * @param socket the socket to use
655 write_data (struct GNUNET_STREAM_Socket *socket)
657 struct GNUNET_STREAM_IOWriteHandle *io_handle = socket->write_handle;
658 int packet; /* Although an int, should never be negative */
662 /* Find the last acknowledged packet */
663 for (packet=0; packet < 64; packet++)
665 if (GNUNET_YES == ackbitmap_is_bit_set (&io_handle->ack_bitmap,
668 else if (NULL == io_handle->messages[packet])
671 /* Resend packets which weren't ack'ed */
672 for (packet=0; packet < ack_packet; packet++)
674 if (GNUNET_NO == ackbitmap_is_bit_set (&io_handle->ack_bitmap,
677 queue_message (socket,
678 &io_handle->messages[packet]->header,
683 packet = ack_packet + 1;
684 /* Now send new packets if there is enough buffer space */
685 while ( (NULL != io_handle->messages[packet]) &&
686 (socket->receive_window_available >= ntohs (io_handle->messages[packet]->header.header.size)) )
688 socket->receive_window_available -= ntohs (io_handle->messages[packet]->header.header.size);
689 queue_message (socket,
690 &io_handle->messages[packet]->header,
691 &write_data_finish_cb,
699 * Task for calling the read processor
701 * @param cls the socket
702 * @param tc the task context
705 call_read_processor (void *cls,
706 const struct GNUNET_SCHEDULER_TaskContext *tc)
708 struct GNUNET_STREAM_Socket *socket = cls;
710 size_t valid_read_size;
712 uint32_t sequence_increase;
713 uint32_t offset_increase;
715 socket->read_task = GNUNET_SCHEDULER_NO_TASK;
716 if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
719 GNUNET_assert (NULL != socket->read_handle);
720 GNUNET_assert (NULL != socket->read_handle->proc);
722 /* Check the bitmap for any holes */
723 for (packet=0; packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH; packet++)
725 if (GNUNET_NO == ackbitmap_is_bit_set (&socket->ack_bitmap,
729 /* We only call read processor if we have the first packet */
730 GNUNET_assert (0 < packet);
733 socket->receive_buffer_boundaries[packet-1] - socket->copy_offset;
735 GNUNET_assert (0 != valid_read_size);
737 /* Cancel the read_io_timeout_task */
738 GNUNET_SCHEDULER_cancel (socket->read_io_timeout_task);
739 socket->read_io_timeout_task = GNUNET_SCHEDULER_NO_TASK;
741 /* Call the data processor */
743 socket->read_handle->proc (socket->read_handle->proc_cls,
745 socket->receive_buffer + socket->copy_offset,
747 /* Free the read handle */
748 GNUNET_free (socket->read_handle);
749 socket->read_handle = NULL;
751 GNUNET_assert (read_size <= valid_read_size);
752 socket->copy_offset += read_size;
754 /* Determine upto which packet we can remove from the buffer */
755 for (packet = 0; packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH; packet++)
756 if (socket->copy_offset < socket->receive_buffer_boundaries[packet])
759 /* If no packets can be removed we can't move the buffer */
760 if (0 == packet) return;
762 sequence_increase = packet;
764 /* Shift the data in the receive buffer */
765 memmove (socket->receive_buffer,
766 socket->receive_buffer
767 + socket->receive_buffer_boundaries[sequence_increase-1],
768 socket->receive_buffer_size - socket->receive_buffer_boundaries[sequence_increase-1]);
770 /* Shift the bitmap */
771 socket->ack_bitmap = socket->ack_bitmap >> sequence_increase;
773 /* Set read_sequence_number */
774 socket->read_sequence_number += sequence_increase;
776 /* Set read_offset */
777 offset_increase = socket->receive_buffer_boundaries[sequence_increase-1];
778 socket->read_offset += offset_increase;
780 /* Fix copy_offset */
781 GNUNET_assert (offset_increase <= socket->copy_offset);
782 socket->copy_offset -= offset_increase;
784 /* Fix relative boundaries */
785 for (packet=0; packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH; packet++)
787 if (packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH - sequence_increase)
789 socket->receive_buffer_boundaries[packet] =
790 socket->receive_buffer_boundaries[packet + sequence_increase]
794 socket->receive_buffer_boundaries[packet] = 0;
800 * Cancels the existing read io handle
802 * @param cls the closure from the SCHEDULER call
803 * @param tc the task context
806 read_io_timeout (void *cls,
807 const struct GNUNET_SCHEDULER_TaskContext *tc)
809 struct GNUNET_STREAM_Socket *socket = cls;
811 socket->read_io_timeout_task = GNUNET_SCHEDULER_NO_TASK;
812 if (socket->read_task != GNUNET_SCHEDULER_NO_TASK)
814 GNUNET_SCHEDULER_cancel (socket->read_task);
815 socket->read_task = GNUNET_SCHEDULER_NO_TASK;
817 GNUNET_assert (NULL != socket->read_handle);
819 GNUNET_free (socket->read_handle);
820 socket->read_handle = NULL;
825 * Handler for DATA messages; Same for both client and server
827 * @param socket the socket through which the ack was received
828 * @param tunnel connection to the other end
829 * @param sender who sent the message
830 * @param msg the data message
831 * @param atsi performance data for the connection
832 * @return GNUNET_OK to keep the connection open,
833 * GNUNET_SYSERR to close it (signal serious error)
836 handle_data (struct GNUNET_STREAM_Socket *socket,
837 struct GNUNET_MESH_Tunnel *tunnel,
838 const struct GNUNET_PeerIdentity *sender,
839 const struct GNUNET_STREAM_DataMessage *msg,
840 const struct GNUNET_ATS_Information*atsi)
843 uint32_t bytes_needed;
844 uint32_t relative_offset;
845 uint32_t relative_sequence_number;
848 size = htons (msg->header.header.size);
849 if (size < sizeof (struct GNUNET_STREAM_DataMessage))
852 return GNUNET_SYSERR;
855 switch (socket->state)
857 case STATE_ESTABLISHED:
858 case STATE_TRANSMIT_CLOSED:
859 case STATE_TRANSMIT_CLOSE_WAIT:
861 /* check if the message's sequence number is in the range we are
863 relative_sequence_number =
864 ntohl (msg->sequence_number) - socket->read_sequence_number;
865 if ( relative_sequence_number > 64)
867 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
868 "Ignoring received message with sequence number %d",
869 ntohl (msg->sequence_number));
873 /* Check if we have to allocate the buffer */
874 size -= sizeof (struct GNUNET_STREAM_DataMessage);
875 relative_offset = ntohl (msg->offset) - socket->read_offset;
876 bytes_needed = relative_offset + size;
878 if (bytes_needed > socket->receive_buffer_size)
880 if (bytes_needed <= RECEIVE_BUFFER_SIZE)
882 socket->receive_buffer = GNUNET_realloc (socket->receive_buffer,
884 socket->receive_buffer_size = bytes_needed;
888 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
889 "Cannot accommodate packet %d as buffer is full\n",
890 ntohl (msg->sequence_number));
895 /* Copy Data to buffer */
897 GNUNET_assert (relative_offset + size <= socket->receive_buffer_size);
898 memcpy (socket->receive_buffer + relative_offset,
901 socket->receive_buffer_boundaries[relative_sequence_number] =
902 relative_offset + size;
904 /* Modify the ACK bitmap */
905 ackbitmap_modify_bit (&socket->ack_bitmap,
906 relative_sequence_number,
909 /* Start ACK sending task if one is not already present */
910 if (0 == socket->ack_task_id)
912 socket->ack_task_id =
913 GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_ntoh
919 if ((NULL != socket->read_handle) /* A read handle is waiting */
920 /* There is no current read task */
921 && (GNUNET_SCHEDULER_NO_TASK == socket->read_task)
922 /* We have the first packet */
923 && (GNUNET_YES == ackbitmap_is_bit_set(&socket->ack_bitmap,
927 GNUNET_SCHEDULER_add_now (&call_read_processor,
934 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
935 "%s: Received data message when it cannot be handled\n",
936 GNUNET_i2s (&socket->our_id));
943 * Client's message Handler for GNUNET_MESSAGE_TYPE_STREAM_DATA
945 * @param cls the socket (set from GNUNET_MESH_connect)
946 * @param tunnel connection to the other end
947 * @param tunnel_ctx place to store local state associated with the tunnel
948 * @param sender who sent the message
949 * @param message the actual message
950 * @param atsi performance data for the connection
951 * @return GNUNET_OK to keep the connection open,
952 * GNUNET_SYSERR to close it (signal serious error)
955 client_handle_data (void *cls,
956 struct GNUNET_MESH_Tunnel *tunnel,
958 const struct GNUNET_PeerIdentity *sender,
959 const struct GNUNET_MessageHeader *message,
960 const struct GNUNET_ATS_Information*atsi)
962 struct GNUNET_STREAM_Socket *socket = cls;
964 return handle_data (socket,
967 (const struct GNUNET_STREAM_DataMessage *) message,
973 * Callback to set state to ESTABLISHED
975 * @param cls the closure from queue_message FIXME: document
976 * @param socket the socket to requiring state change
979 set_state_established (void *cls,
980 struct GNUNET_STREAM_Socket *socket)
982 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
983 "%s: Attaining ESTABLISHED state\n",
984 GNUNET_i2s (&socket->our_id));
985 socket->write_offset = 0;
986 socket->read_offset = 0;
987 socket->state = STATE_ESTABLISHED;
989 socket->open_cb (socket->open_cls, socket);
994 * Callback to set state to HELLO_WAIT
996 * @param cls the closure from queue_message
997 * @param socket the socket to requiring state change
1000 set_state_hello_wait (void *cls,
1001 struct GNUNET_STREAM_Socket *socket)
1003 GNUNET_assert (STATE_INIT == socket->state);
1004 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1005 "%s: Attaining HELLO_WAIT state\n",
1006 GNUNET_i2s (&socket->our_id));
1007 socket->state = STATE_HELLO_WAIT;
1012 * Client's message handler for GNUNET_MESSAGE_TYPE_STREAM_HELLO_ACK
1014 * @param cls the socket (set from GNUNET_MESH_connect)
1015 * @param tunnel connection to the other end
1016 * @param tunnel_ctx this is NULL
1017 * @param sender who sent the message
1018 * @param message the actual message
1019 * @param atsi performance data for the connection
1020 * @return GNUNET_OK to keep the connection open,
1021 * GNUNET_SYSERR to close it (signal serious error)
1024 client_handle_hello_ack (void *cls,
1025 struct GNUNET_MESH_Tunnel *tunnel,
1027 const struct GNUNET_PeerIdentity *sender,
1028 const struct GNUNET_MessageHeader *message,
1029 const struct GNUNET_ATS_Information*atsi)
1031 struct GNUNET_STREAM_Socket *socket = cls;
1032 const struct GNUNET_STREAM_HelloAckMessage *ack_msg;
1033 struct GNUNET_STREAM_HelloAckMessage *reply;
1035 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1036 "%s: Received HELLO_ACK from %s\n",
1037 GNUNET_i2s (&socket->our_id),
1038 GNUNET_i2s (sender));
1039 ack_msg = (const struct GNUNET_STREAM_HelloAckMessage *) message;
1040 GNUNET_assert (socket->tunnel == tunnel);
1041 switch (socket->state)
1043 case STATE_HELLO_WAIT:
1044 socket->read_sequence_number = ntohl (ack_msg->sequence_number);
1045 socket->receive_window_available = ntohl (ack_msg->receive_window_size);
1046 /* Get the random sequence number */
1047 socket->write_sequence_number =
1048 GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_NONCE, UINT32_MAX);
1050 GNUNET_malloc (sizeof (struct GNUNET_STREAM_HelloAckMessage));
1051 reply->header.header.size =
1052 htons (sizeof (struct GNUNET_STREAM_MessageHeader));
1053 reply->header.header.type =
1054 htons (GNUNET_MESSAGE_TYPE_STREAM_HELLO_ACK);
1055 reply->sequence_number = htonl (socket->write_sequence_number);
1056 reply->receive_window_size = htonl (RECEIVE_BUFFER_SIZE);
1057 queue_message (socket,
1059 &set_state_established,
1062 case STATE_ESTABLISHED:
1063 case STATE_RECEIVE_CLOSE_WAIT:
1064 // call statistics (# ACKs ignored++)
1068 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1069 "%s: Server sent HELLO_ACK when in state %d\n",
1070 GNUNET_i2s (&socket->our_id),
1072 socket->state = STATE_CLOSED; // introduce STATE_ERROR?
1073 return GNUNET_SYSERR;
1080 * Client's message handler for GNUNET_MESSAGE_TYPE_STREAM_RESET
1082 * @param cls the socket (set from GNUNET_MESH_connect)
1083 * @param tunnel connection to the other end
1084 * @param tunnel_ctx this is NULL
1085 * @param sender who sent the message
1086 * @param message the actual message
1087 * @param atsi performance data for the connection
1088 * @return GNUNET_OK to keep the connection open,
1089 * GNUNET_SYSERR to close it (signal serious error)
1092 client_handle_reset (void *cls,
1093 struct GNUNET_MESH_Tunnel *tunnel,
1095 const struct GNUNET_PeerIdentity *sender,
1096 const struct GNUNET_MessageHeader *message,
1097 const struct GNUNET_ATS_Information*atsi)
1099 struct GNUNET_STREAM_Socket *socket = cls;
1106 * Common message handler for handling TRANSMIT_CLOSE messages
1108 * @param socket the socket through which the ack was received
1109 * @param tunnel connection to the other end
1110 * @param sender who sent the message
1111 * @param msg the transmit close message
1112 * @param atsi performance data for the connection
1113 * @return GNUNET_OK to keep the connection open,
1114 * GNUNET_SYSERR to close it (signal serious error)
1117 handle_transmit_close (struct GNUNET_STREAM_Socket *socket,
1118 struct GNUNET_MESH_Tunnel *tunnel,
1119 const struct GNUNET_PeerIdentity *sender,
1120 const struct GNUNET_STREAM_MessageHeader *msg,
1121 const struct GNUNET_ATS_Information*atsi)
1123 struct GNUNET_STREAM_MessageHeader *reply;
1125 switch (socket->state)
1127 case STATE_ESTABLISHED:
1128 socket->state = STATE_RECEIVE_CLOSED;
1130 /* Send TRANSMIT_CLOSE_ACK */
1131 reply = GNUNET_malloc (sizeof (struct GNUNET_STREAM_MessageHeader));
1132 reply->header.type =
1133 htons (GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE_ACK);
1134 reply->header.size = htons (sizeof (struct GNUNET_STREAM_MessageHeader));
1135 queue_message (socket, reply, NULL, NULL);
1139 /* FIXME: Call statistics? */
1147 * Client's message handler for GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE
1149 * @param cls the socket (set from GNUNET_MESH_connect)
1150 * @param tunnel connection to the other end
1151 * @param tunnel_ctx this is NULL
1152 * @param sender who sent the message
1153 * @param message the actual message
1154 * @param atsi performance data for the connection
1155 * @return GNUNET_OK to keep the connection open,
1156 * GNUNET_SYSERR to close it (signal serious error)
1159 client_handle_transmit_close (void *cls,
1160 struct GNUNET_MESH_Tunnel *tunnel,
1162 const struct GNUNET_PeerIdentity *sender,
1163 const struct GNUNET_MessageHeader *message,
1164 const struct GNUNET_ATS_Information*atsi)
1166 struct GNUNET_STREAM_Socket *socket = cls;
1168 return handle_transmit_close (socket,
1171 (struct GNUNET_STREAM_MessageHeader *)message,
1177 * Client's message handler for GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE_ACK
1179 * @param cls the socket (set from GNUNET_MESH_connect)
1180 * @param tunnel connection to the other end
1181 * @param tunnel_ctx this is NULL
1182 * @param sender who sent the message
1183 * @param message the actual message
1184 * @param atsi performance data for the connection
1185 * @return GNUNET_OK to keep the connection open,
1186 * GNUNET_SYSERR to close it (signal serious error)
1189 client_handle_transmit_close_ack (void *cls,
1190 struct GNUNET_MESH_Tunnel *tunnel,
1192 const struct GNUNET_PeerIdentity *sender,
1193 const struct GNUNET_MessageHeader *message,
1194 const struct GNUNET_ATS_Information*atsi)
1196 struct GNUNET_STREAM_Socket *socket = cls;
1203 * Client's message handler for GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE
1205 * @param cls the socket (set from GNUNET_MESH_connect)
1206 * @param tunnel connection to the other end
1207 * @param tunnel_ctx this is NULL
1208 * @param sender who sent the message
1209 * @param message the actual message
1210 * @param atsi performance data for the connection
1211 * @return GNUNET_OK to keep the connection open,
1212 * GNUNET_SYSERR to close it (signal serious error)
1215 client_handle_receive_close (void *cls,
1216 struct GNUNET_MESH_Tunnel *tunnel,
1218 const struct GNUNET_PeerIdentity *sender,
1219 const struct GNUNET_MessageHeader *message,
1220 const struct GNUNET_ATS_Information*atsi)
1222 struct GNUNET_STREAM_Socket *socket = cls;
1229 * Client's message handler for GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE_ACK
1231 * @param cls the socket (set from GNUNET_MESH_connect)
1232 * @param tunnel connection to the other end
1233 * @param tunnel_ctx this is NULL
1234 * @param sender who sent the message
1235 * @param message the actual message
1236 * @param atsi performance data for the connection
1237 * @return GNUNET_OK to keep the connection open,
1238 * GNUNET_SYSERR to close it (signal serious error)
1241 client_handle_receive_close_ack (void *cls,
1242 struct GNUNET_MESH_Tunnel *tunnel,
1244 const struct GNUNET_PeerIdentity *sender,
1245 const struct GNUNET_MessageHeader *message,
1246 const struct GNUNET_ATS_Information*atsi)
1248 struct GNUNET_STREAM_Socket *socket = cls;
1255 * Client's message handler for GNUNET_MESSAGE_TYPE_STREAM_CLOSE
1257 * @param cls the socket (set from GNUNET_MESH_connect)
1258 * @param tunnel connection to the other end
1259 * @param tunnel_ctx this is NULL
1260 * @param sender who sent the message
1261 * @param message the actual message
1262 * @param atsi performance data for the connection
1263 * @return GNUNET_OK to keep the connection open,
1264 * GNUNET_SYSERR to close it (signal serious error)
1267 client_handle_close (void *cls,
1268 struct GNUNET_MESH_Tunnel *tunnel,
1270 const struct GNUNET_PeerIdentity *sender,
1271 const struct GNUNET_MessageHeader *message,
1272 const struct GNUNET_ATS_Information*atsi)
1274 struct GNUNET_STREAM_Socket *socket = cls;
1281 * Client's message handler for GNUNET_MESSAGE_TYPE_STREAM_CLOSE_ACK
1283 * @param cls the socket (set from GNUNET_MESH_connect)
1284 * @param tunnel connection to the other end
1285 * @param tunnel_ctx this is NULL
1286 * @param sender who sent the message
1287 * @param message the actual message
1288 * @param atsi performance data for the connection
1289 * @return GNUNET_OK to keep the connection open,
1290 * GNUNET_SYSERR to close it (signal serious error)
1293 client_handle_close_ack (void *cls,
1294 struct GNUNET_MESH_Tunnel *tunnel,
1296 const struct GNUNET_PeerIdentity *sender,
1297 const struct GNUNET_MessageHeader *message,
1298 const struct GNUNET_ATS_Information*atsi)
1300 struct GNUNET_STREAM_Socket *socket = cls;
1305 /*****************************/
1306 /* Server's Message Handlers */
1307 /*****************************/
1310 * Server's message Handler for GNUNET_MESSAGE_TYPE_STREAM_DATA
1312 * @param cls the closure
1313 * @param tunnel connection to the other end
1314 * @param tunnel_ctx the socket
1315 * @param sender who sent the message
1316 * @param message the actual message
1317 * @param atsi performance data for the connection
1318 * @return GNUNET_OK to keep the connection open,
1319 * GNUNET_SYSERR to close it (signal serious error)
1322 server_handle_data (void *cls,
1323 struct GNUNET_MESH_Tunnel *tunnel,
1325 const struct GNUNET_PeerIdentity *sender,
1326 const struct GNUNET_MessageHeader *message,
1327 const struct GNUNET_ATS_Information*atsi)
1329 struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
1331 return handle_data (socket,
1334 (const struct GNUNET_STREAM_DataMessage *)message,
1340 * Server's message handler for GNUNET_MESSAGE_TYPE_STREAM_HELLO
1342 * @param cls the closure
1343 * @param tunnel connection to the other end
1344 * @param tunnel_ctx the socket
1345 * @param sender who sent the message
1346 * @param message the actual message
1347 * @param atsi performance data for the connection
1348 * @return GNUNET_OK to keep the connection open,
1349 * GNUNET_SYSERR to close it (signal serious error)
1352 server_handle_hello (void *cls,
1353 struct GNUNET_MESH_Tunnel *tunnel,
1355 const struct GNUNET_PeerIdentity *sender,
1356 const struct GNUNET_MessageHeader *message,
1357 const struct GNUNET_ATS_Information*atsi)
1359 struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
1360 struct GNUNET_STREAM_HelloAckMessage *reply;
1362 GNUNET_assert (socket->tunnel == tunnel);
1363 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1364 "Received HELLO from %s\n", GNUNET_i2s(sender));
1366 /* Catch possible protocol breaks */
1367 GNUNET_break_op (0 == memcmp (&socket->other_peer,
1369 sizeof (struct GNUNET_PeerIdentity)));
1371 if (STATE_INIT == socket->state)
1373 /* Get the random sequence number */
1374 socket->write_sequence_number =
1375 GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_NONCE, UINT32_MAX);
1377 GNUNET_malloc (sizeof (struct GNUNET_STREAM_HelloAckMessage));
1378 reply->header.header.size =
1379 htons (sizeof (struct GNUNET_STREAM_MessageHeader));
1380 reply->header.header.type =
1381 htons (GNUNET_MESSAGE_TYPE_STREAM_HELLO_ACK);
1382 reply->sequence_number = htonl (socket->write_sequence_number);
1383 queue_message (socket,
1385 &set_state_hello_wait,
1390 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1391 "Client sent HELLO when in state %d\n", socket->state);
1392 /* FIXME: Send RESET? */
1400 * Server's message handler for GNUNET_MESSAGE_TYPE_STREAM_HELLO_ACK
1402 * @param cls the closure
1403 * @param tunnel connection to the other end
1404 * @param tunnel_ctx the socket
1405 * @param sender who sent the message
1406 * @param message the actual message
1407 * @param atsi performance data for the connection
1408 * @return GNUNET_OK to keep the connection open,
1409 * GNUNET_SYSERR to close it (signal serious error)
1412 server_handle_hello_ack (void *cls,
1413 struct GNUNET_MESH_Tunnel *tunnel,
1415 const struct GNUNET_PeerIdentity *sender,
1416 const struct GNUNET_MessageHeader *message,
1417 const struct GNUNET_ATS_Information*atsi)
1419 struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
1420 const struct GNUNET_STREAM_HelloAckMessage *ack_message;
1422 ack_message = (struct GNUNET_STREAM_HelloAckMessage *) message;
1423 GNUNET_assert (socket->tunnel == tunnel);
1424 if (STATE_HELLO_WAIT == socket->state)
1426 socket->read_sequence_number = ntohl (ack_message->sequence_number);
1427 socket->receive_window_available =
1428 ntohl (ack_message->receive_window_size);
1429 /* Attain ESTABLISHED state */
1430 set_state_established (NULL, socket);
1434 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1435 "Client sent HELLO_ACK when in state %d\n", socket->state);
1436 /* FIXME: Send RESET? */
1444 * Server's message handler for GNUNET_MESSAGE_TYPE_STREAM_RESET
1446 * @param cls the closure
1447 * @param tunnel connection to the other end
1448 * @param tunnel_ctx the socket
1449 * @param sender who sent the message
1450 * @param message the actual message
1451 * @param atsi performance data for the connection
1452 * @return GNUNET_OK to keep the connection open,
1453 * GNUNET_SYSERR to close it (signal serious error)
1456 server_handle_reset (void *cls,
1457 struct GNUNET_MESH_Tunnel *tunnel,
1459 const struct GNUNET_PeerIdentity *sender,
1460 const struct GNUNET_MessageHeader *message,
1461 const struct GNUNET_ATS_Information*atsi)
1463 struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
1470 * Server's message handler for GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE
1472 * @param cls the closure
1473 * @param tunnel connection to the other end
1474 * @param tunnel_ctx the socket
1475 * @param sender who sent the message
1476 * @param message the actual message
1477 * @param atsi performance data for the connection
1478 * @return GNUNET_OK to keep the connection open,
1479 * GNUNET_SYSERR to close it (signal serious error)
1482 server_handle_transmit_close (void *cls,
1483 struct GNUNET_MESH_Tunnel *tunnel,
1485 const struct GNUNET_PeerIdentity *sender,
1486 const struct GNUNET_MessageHeader *message,
1487 const struct GNUNET_ATS_Information*atsi)
1489 struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
1491 return handle_transmit_close (socket,
1494 (struct GNUNET_STREAM_MessageHeader *)message,
1500 * Server's message handler for GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE_ACK
1502 * @param cls the closure
1503 * @param tunnel connection to the other end
1504 * @param tunnel_ctx the socket
1505 * @param sender who sent the message
1506 * @param message the actual message
1507 * @param atsi performance data for the connection
1508 * @return GNUNET_OK to keep the connection open,
1509 * GNUNET_SYSERR to close it (signal serious error)
1512 server_handle_transmit_close_ack (void *cls,
1513 struct GNUNET_MESH_Tunnel *tunnel,
1515 const struct GNUNET_PeerIdentity *sender,
1516 const struct GNUNET_MessageHeader *message,
1517 const struct GNUNET_ATS_Information*atsi)
1519 struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
1526 * Server's message handler for GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE
1528 * @param cls the closure
1529 * @param tunnel connection to the other end
1530 * @param tunnel_ctx the socket
1531 * @param sender who sent the message
1532 * @param message the actual message
1533 * @param atsi performance data for the connection
1534 * @return GNUNET_OK to keep the connection open,
1535 * GNUNET_SYSERR to close it (signal serious error)
1538 server_handle_receive_close (void *cls,
1539 struct GNUNET_MESH_Tunnel *tunnel,
1541 const struct GNUNET_PeerIdentity *sender,
1542 const struct GNUNET_MessageHeader *message,
1543 const struct GNUNET_ATS_Information*atsi)
1545 struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
1552 * Server's message handler for GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE_ACK
1554 * @param cls the closure
1555 * @param tunnel connection to the other end
1556 * @param tunnel_ctx the socket
1557 * @param sender who sent the message
1558 * @param message the actual message
1559 * @param atsi performance data for the connection
1560 * @return GNUNET_OK to keep the connection open,
1561 * GNUNET_SYSERR to close it (signal serious error)
1564 server_handle_receive_close_ack (void *cls,
1565 struct GNUNET_MESH_Tunnel *tunnel,
1567 const struct GNUNET_PeerIdentity *sender,
1568 const struct GNUNET_MessageHeader *message,
1569 const struct GNUNET_ATS_Information*atsi)
1571 struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
1578 * Server's message handler for GNUNET_MESSAGE_TYPE_STREAM_CLOSE
1580 * @param cls the closure
1581 * @param tunnel connection to the other end
1582 * @param tunnel_ctx the socket
1583 * @param sender who sent the message
1584 * @param message the actual message
1585 * @param atsi performance data for the connection
1586 * @return GNUNET_OK to keep the connection open,
1587 * GNUNET_SYSERR to close it (signal serious error)
1590 server_handle_close (void *cls,
1591 struct GNUNET_MESH_Tunnel *tunnel,
1593 const struct GNUNET_PeerIdentity *sender,
1594 const struct GNUNET_MessageHeader *message,
1595 const struct GNUNET_ATS_Information*atsi)
1597 struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
1604 * Server's message handler for GNUNET_MESSAGE_TYPE_STREAM_CLOSE_ACK
1606 * @param cls the closure
1607 * @param tunnel connection to the other end
1608 * @param tunnel_ctx the socket
1609 * @param sender who sent the message
1610 * @param message the actual message
1611 * @param atsi performance data for the connection
1612 * @return GNUNET_OK to keep the connection open,
1613 * GNUNET_SYSERR to close it (signal serious error)
1616 server_handle_close_ack (void *cls,
1617 struct GNUNET_MESH_Tunnel *tunnel,
1619 const struct GNUNET_PeerIdentity *sender,
1620 const struct GNUNET_MessageHeader *message,
1621 const struct GNUNET_ATS_Information*atsi)
1623 struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
1630 * Message Handler for mesh
1632 * @param socket the socket through which the ack was received
1633 * @param tunnel connection to the other end
1634 * @param sender who sent the message
1635 * @param ack the acknowledgment message
1636 * @param atsi performance data for the connection
1637 * @return GNUNET_OK to keep the connection open,
1638 * GNUNET_SYSERR to close it (signal serious error)
1641 handle_ack (struct GNUNET_STREAM_Socket *socket,
1642 struct GNUNET_MESH_Tunnel *tunnel,
1643 const struct GNUNET_PeerIdentity *sender,
1644 const struct GNUNET_STREAM_AckMessage *ack,
1645 const struct GNUNET_ATS_Information*atsi)
1647 switch (socket->state)
1649 case (STATE_ESTABLISHED):
1650 if (NULL == socket->write_handle)
1652 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1653 "Received DATA ACK when write_handle is NULL\n");
1657 socket->write_handle->ack_bitmap = GNUNET_ntohll (ack->bitmap);
1658 socket->receive_window_available =
1659 ntohl (ack->receive_window_remaining);
1660 write_data (socket);
1670 * Message Handler for mesh
1672 * @param cls the 'struct GNUNET_STREAM_Socket'
1673 * @param tunnel connection to the other end
1674 * @param tunnel_ctx unused
1675 * @param sender who sent the message
1676 * @param message the actual message
1677 * @param atsi performance data for the connection
1678 * @return GNUNET_OK to keep the connection open,
1679 * GNUNET_SYSERR to close it (signal serious error)
1682 client_handle_ack (void *cls,
1683 struct GNUNET_MESH_Tunnel *tunnel,
1685 const struct GNUNET_PeerIdentity *sender,
1686 const struct GNUNET_MessageHeader *message,
1687 const struct GNUNET_ATS_Information*atsi)
1689 struct GNUNET_STREAM_Socket *socket = cls;
1690 const struct GNUNET_STREAM_AckMessage *ack = (const struct GNUNET_STREAM_AckMessage *) message;
1692 return handle_ack (socket, tunnel, sender, ack, atsi);
1697 * Message Handler for mesh
1699 * @param cls the server's listen socket
1700 * @param tunnel connection to the other end
1701 * @param tunnel_ctx pointer to the 'struct GNUNET_STREAM_Socket*'
1702 * @param sender who sent the message
1703 * @param message the actual message
1704 * @param atsi performance data for the connection
1705 * @return GNUNET_OK to keep the connection open,
1706 * GNUNET_SYSERR to close it (signal serious error)
1709 server_handle_ack (void *cls,
1710 struct GNUNET_MESH_Tunnel *tunnel,
1712 const struct GNUNET_PeerIdentity *sender,
1713 const struct GNUNET_MessageHeader *message,
1714 const struct GNUNET_ATS_Information*atsi)
1716 struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
1717 const struct GNUNET_STREAM_AckMessage *ack = (const struct GNUNET_STREAM_AckMessage *) message;
1719 return handle_ack (socket, tunnel, sender, ack, atsi);
1724 * For client message handlers, the stream socket is in the
1727 static struct GNUNET_MESH_MessageHandler client_message_handlers[] = {
1728 {&client_handle_data, GNUNET_MESSAGE_TYPE_STREAM_DATA, 0},
1729 {&client_handle_ack, GNUNET_MESSAGE_TYPE_STREAM_ACK,
1730 sizeof (struct GNUNET_STREAM_AckMessage) },
1731 {&client_handle_hello_ack, GNUNET_MESSAGE_TYPE_STREAM_HELLO_ACK,
1732 sizeof (struct GNUNET_STREAM_HelloAckMessage)},
1733 {&client_handle_reset, GNUNET_MESSAGE_TYPE_STREAM_RESET,
1734 sizeof (struct GNUNET_STREAM_MessageHeader)},
1735 {&client_handle_transmit_close, GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE,
1736 sizeof (struct GNUNET_STREAM_MessageHeader)},
1737 {&client_handle_transmit_close_ack, GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE_ACK,
1738 sizeof (struct GNUNET_STREAM_MessageHeader)},
1739 {&client_handle_receive_close, GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE,
1740 sizeof (struct GNUNET_STREAM_MessageHeader)},
1741 {&client_handle_receive_close_ack, GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE_ACK,
1742 sizeof (struct GNUNET_STREAM_MessageHeader)},
1743 {&client_handle_close, GNUNET_MESSAGE_TYPE_STREAM_CLOSE,
1744 sizeof (struct GNUNET_STREAM_MessageHeader)},
1745 {&client_handle_close_ack, GNUNET_MESSAGE_TYPE_STREAM_CLOSE_ACK,
1746 sizeof (struct GNUNET_STREAM_MessageHeader)},
1752 * For server message handlers, the stream socket is in the
1753 * tunnel context, and the listen socket in the closure argument.
1755 static struct GNUNET_MESH_MessageHandler server_message_handlers[] = {
1756 {&server_handle_data, GNUNET_MESSAGE_TYPE_STREAM_DATA, 0},
1757 {&server_handle_ack, GNUNET_MESSAGE_TYPE_STREAM_ACK,
1758 sizeof (struct GNUNET_STREAM_AckMessage) },
1759 {&server_handle_hello, GNUNET_MESSAGE_TYPE_STREAM_HELLO,
1760 sizeof (struct GNUNET_STREAM_MessageHeader)},
1761 {&server_handle_hello_ack, GNUNET_MESSAGE_TYPE_STREAM_HELLO_ACK,
1762 sizeof (struct GNUNET_STREAM_HelloAckMessage)},
1763 {&server_handle_reset, GNUNET_MESSAGE_TYPE_STREAM_RESET,
1764 sizeof (struct GNUNET_STREAM_MessageHeader)},
1765 {&server_handle_transmit_close, GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE,
1766 sizeof (struct GNUNET_STREAM_MessageHeader)},
1767 {&server_handle_transmit_close_ack, GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE_ACK,
1768 sizeof (struct GNUNET_STREAM_MessageHeader)},
1769 {&server_handle_receive_close, GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE,
1770 sizeof (struct GNUNET_STREAM_MessageHeader)},
1771 {&server_handle_receive_close_ack, GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE_ACK,
1772 sizeof (struct GNUNET_STREAM_MessageHeader)},
1773 {&server_handle_close, GNUNET_MESSAGE_TYPE_STREAM_CLOSE,
1774 sizeof (struct GNUNET_STREAM_MessageHeader)},
1775 {&server_handle_close_ack, GNUNET_MESSAGE_TYPE_STREAM_CLOSE_ACK,
1776 sizeof (struct GNUNET_STREAM_MessageHeader)},
1782 * Function called when our target peer is connected to our tunnel
1784 * @param cls the socket for which this tunnel is created
1785 * @param peer the peer identity of the target
1786 * @param atsi performance data for the connection
1789 mesh_peer_connect_callback (void *cls,
1790 const struct GNUNET_PeerIdentity *peer,
1791 const struct GNUNET_ATS_Information * atsi)
1793 struct GNUNET_STREAM_Socket *socket = cls;
1794 struct GNUNET_STREAM_MessageHeader *message;
1796 if (0 != memcmp (&socket->other_peer,
1798 sizeof (struct GNUNET_PeerIdentity)))
1800 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1801 "%s: A peer (%s) which is not our target has connected",
1803 GNUNET_i2s (&socket->our_id),
1808 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1809 "Target peer %s connected\n", GNUNET_i2s (peer));
1811 /* Set state to INIT */
1812 socket->state = STATE_INIT;
1814 /* Send HELLO message */
1815 message = GNUNET_malloc (sizeof (struct GNUNET_STREAM_MessageHeader));
1816 message->header.type = htons (GNUNET_MESSAGE_TYPE_STREAM_HELLO);
1817 message->header.size = htons (sizeof (struct GNUNET_STREAM_MessageHeader));
1818 queue_message (socket,
1820 &set_state_hello_wait,
1823 /* Call open callback */
1824 if (NULL == socket->open_cb)
1826 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1827 "STREAM_open callback is NULL\n");
1833 * Function called when our target peer is disconnected from our tunnel
1835 * @param cls the socket associated which this tunnel
1836 * @param peer the peer identity of the target
1839 mesh_peer_disconnect_callback (void *cls,
1840 const struct GNUNET_PeerIdentity *peer)
1847 * Method called whenever a peer creates a tunnel to us
1849 * @param cls closure
1850 * @param tunnel new handle to the tunnel
1851 * @param initiator peer that started the tunnel
1852 * @param atsi performance information for the tunnel
1853 * @return initial tunnel context for the tunnel
1854 * (can be NULL -- that's not an error)
1857 new_tunnel_notify (void *cls,
1858 struct GNUNET_MESH_Tunnel *tunnel,
1859 const struct GNUNET_PeerIdentity *initiator,
1860 const struct GNUNET_ATS_Information *atsi)
1862 struct GNUNET_STREAM_ListenSocket *lsocket = cls;
1863 struct GNUNET_STREAM_Socket *socket;
1865 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1866 "%s: Peer %s initiated tunnel to us\n",
1867 GNUNET_i2s (&lsocket->our_id),
1868 GNUNET_i2s (initiator));
1870 socket = GNUNET_malloc (sizeof (struct GNUNET_STREAM_Socket));
1871 socket->tunnel = tunnel;
1872 socket->session_id = 0; /* FIXME */
1873 socket->other_peer = *initiator;
1874 socket->state = STATE_INIT;
1875 socket->derived = GNUNET_YES;
1876 socket->our_id = lsocket->our_id;
1878 /* FIXME: Copy MESH handle from lsocket to socket */
1880 if (GNUNET_SYSERR == lsocket->listen_cb (lsocket->listen_cb_cls,
1882 &socket->other_peer))
1884 socket->state = STATE_CLOSED;
1885 /* FIXME: Send CLOSE message and then free */
1886 GNUNET_free (socket);
1887 GNUNET_MESH_tunnel_destroy (tunnel); /* Destroy the tunnel */
1894 * Function called whenever an inbound tunnel is destroyed. Should clean up
1895 * any associated state. This function is NOT called if the client has
1896 * explicitly asked for the tunnel to be destroyed using
1897 * GNUNET_MESH_tunnel_destroy. It must NOT call GNUNET_MESH_tunnel_destroy on
1900 * @param cls closure (set from GNUNET_MESH_connect)
1901 * @param tunnel connection to the other end (henceforth invalid)
1902 * @param tunnel_ctx place where local state associated
1903 * with the tunnel is stored
1906 tunnel_cleaner (void *cls,
1907 const struct GNUNET_MESH_Tunnel *tunnel,
1910 struct GNUNET_STREAM_Socket *socket = tunnel_ctx;
1912 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1913 "%s: Peer %s has terminated connection abruptly\n",
1914 GNUNET_i2s (&socket->our_id),
1915 GNUNET_i2s (&socket->other_peer));
1917 socket->status = GNUNET_STREAM_SHUTDOWN;
1918 /* Clear Transmit handles */
1919 if (NULL != socket->transmit_handle)
1921 GNUNET_MESH_notify_transmit_ready_cancel (socket->transmit_handle);
1922 socket->transmit_handle = NULL;
1924 socket->tunnel = NULL;
1934 * Tries to open a stream to the target peer
1936 * @param cfg configuration to use
1937 * @param target the target peer to which the stream has to be opened
1938 * @param app_port the application port number which uniquely identifies this
1940 * @param open_cb this function will be called after stream has be established
1941 * @param open_cb_cls the closure for open_cb
1942 * @param ... options to the stream, terminated by GNUNET_STREAM_OPTION_END
1943 * @return if successful it returns the stream socket; NULL if stream cannot be
1946 struct GNUNET_STREAM_Socket *
1947 GNUNET_STREAM_open (const struct GNUNET_CONFIGURATION_Handle *cfg,
1948 const struct GNUNET_PeerIdentity *target,
1949 GNUNET_MESH_ApplicationType app_port,
1950 GNUNET_STREAM_OpenCallback open_cb,
1954 struct GNUNET_STREAM_Socket *socket;
1955 enum GNUNET_STREAM_Option option;
1956 va_list vargs; /* Variable arguments */
1958 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1961 socket = GNUNET_malloc (sizeof (struct GNUNET_STREAM_Socket));
1962 socket->other_peer = *target;
1963 socket->open_cb = open_cb;
1964 socket->open_cls = open_cb_cls;
1965 GNUNET_TESTING_get_peer_identity (cfg, &socket->our_id);
1968 socket->retransmit_timeout =
1969 GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, default_timeout);
1971 va_start (vargs, open_cb_cls); /* Parse variable args */
1973 option = va_arg (vargs, enum GNUNET_STREAM_Option);
1976 case GNUNET_STREAM_OPTION_INITIAL_RETRANSMIT_TIMEOUT:
1977 /* Expect struct GNUNET_TIME_Relative */
1978 socket->retransmit_timeout = va_arg (vargs,
1979 struct GNUNET_TIME_Relative);
1981 case GNUNET_STREAM_OPTION_END:
1984 } while (GNUNET_STREAM_OPTION_END != option);
1985 va_end (vargs); /* End of variable args parsing */
1986 socket->mesh = GNUNET_MESH_connect (cfg, /* the configuration handle */
1987 10, /* QUEUE size as parameter? */
1989 NULL, /* No inbound tunnel handler */
1991 client_message_handlers,
1992 &app_port); /* We don't get inbound tunnels */
1993 if (NULL == socket->mesh) /* Fail if we cannot connect to mesh */
1995 GNUNET_free (socket);
1999 /* Now create the mesh tunnel to target */
2000 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2001 "Creating MESH Tunnel\n");
2002 socket->tunnel = GNUNET_MESH_tunnel_create (socket->mesh,
2003 NULL, /* Tunnel context */
2004 &mesh_peer_connect_callback,
2005 &mesh_peer_disconnect_callback,
2007 GNUNET_assert (NULL != socket->tunnel);
2008 GNUNET_MESH_peer_request_connect_add (socket->tunnel,
2011 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2012 "%s() END\n", __func__);
2018 * Shutdown the stream for reading or writing (man 2 shutdown).
2020 * @param socket the stream socket
2021 * @param how SHUT_RD, SHUT_WR or SHUT_RDWR
2024 GNUNET_STREAM_shutdown (struct GNUNET_STREAM_Socket *socket,
2034 * @param socket the stream socket
2037 GNUNET_STREAM_close (struct GNUNET_STREAM_Socket *socket)
2039 struct MessageQueue *head;
2041 if (socket->read_task != GNUNET_SCHEDULER_NO_TASK)
2043 /* socket closed with read task pending!? */
2045 GNUNET_SCHEDULER_cancel (socket->read_task);
2046 socket->read_task = GNUNET_SCHEDULER_NO_TASK;
2049 /* Clear Transmit handles */
2050 if (NULL != socket->transmit_handle)
2052 GNUNET_MESH_notify_transmit_ready_cancel (socket->transmit_handle);
2053 socket->transmit_handle = NULL;
2056 /* Clear existing message queue */
2057 while (NULL != (head = socket->queue_head)) {
2058 GNUNET_CONTAINER_DLL_remove (socket->queue_head,
2061 GNUNET_free (head->message);
2065 /* Close associated tunnel */
2066 if (NULL != socket->tunnel)
2068 GNUNET_MESH_tunnel_destroy (socket->tunnel);
2069 socket->tunnel = NULL;
2072 /* Close mesh connection */
2073 if (NULL != socket->mesh && GNUNET_YES != socket->derived)
2075 GNUNET_MESH_disconnect (socket->mesh);
2076 socket->mesh = NULL;
2079 /* Release receive buffer */
2080 if (NULL != socket->receive_buffer)
2082 GNUNET_free (socket->receive_buffer);
2085 GNUNET_free (socket);
2090 * Listens for stream connections for a specific application ports
2092 * @param cfg the configuration to use
2093 * @param app_port the application port for which new streams will be accepted
2094 * @param listen_cb this function will be called when a peer tries to establish
2096 * @param listen_cb_cls closure for listen_cb
2097 * @return listen socket, NULL for any error
2099 struct GNUNET_STREAM_ListenSocket *
2100 GNUNET_STREAM_listen (const struct GNUNET_CONFIGURATION_Handle *cfg,
2101 GNUNET_MESH_ApplicationType app_port,
2102 GNUNET_STREAM_ListenCallback listen_cb,
2103 void *listen_cb_cls)
2105 /* FIXME: Add variable args for passing configration options? */
2106 struct GNUNET_STREAM_ListenSocket *lsocket;
2108 lsocket = GNUNET_malloc (sizeof (struct GNUNET_STREAM_ListenSocket));
2109 lsocket->port = app_port;
2110 lsocket->listen_cb = listen_cb;
2111 lsocket->listen_cb_cls = listen_cb_cls;
2112 GNUNET_TESTING_get_peer_identity (cfg, &lsocket->our_id);
2113 lsocket->mesh = GNUNET_MESH_connect (cfg,
2114 10, /* FIXME: QUEUE size as parameter? */
2115 lsocket, /* Closure */
2118 server_message_handlers,
2120 GNUNET_assert (NULL != lsocket->mesh);
2126 * Closes the listen socket
2128 * @param lsocket the listen socket
2131 GNUNET_STREAM_listen_close (struct GNUNET_STREAM_ListenSocket *lsocket)
2133 /* Close MESH connection */
2134 GNUNET_assert (NULL != lsocket->mesh);
2135 GNUNET_MESH_disconnect (lsocket->mesh);
2137 GNUNET_free (lsocket);
2142 * Tries to write the given data to the stream
2144 * @param socket the socket representing a stream
2145 * @param data the data buffer from where the data is written into the stream
2146 * @param size the number of bytes to be written from the data buffer
2147 * @param timeout the timeout period
2148 * @param write_cont the function to call upon writing some bytes into the stream
2149 * @param write_cont_cls the closure
2150 * @return handle to cancel the operation
2152 struct GNUNET_STREAM_IOWriteHandle *
2153 GNUNET_STREAM_write (struct GNUNET_STREAM_Socket *socket,
2156 struct GNUNET_TIME_Relative timeout,
2157 GNUNET_STREAM_CompletionContinuation write_cont,
2158 void *write_cont_cls)
2160 unsigned int num_needed_packets;
2161 unsigned int packet;
2162 struct GNUNET_STREAM_IOWriteHandle *io_handle;
2163 uint32_t packet_size;
2164 uint32_t payload_size;
2165 struct GNUNET_STREAM_DataMessage *data_msg;
2168 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2171 /* Return NULL if there is already a write request pending */
2172 if (NULL != socket->write_handle)
2177 if (!((STATE_ESTABLISHED == socket->state)
2178 || (STATE_RECEIVE_CLOSE_WAIT == socket->state)
2179 || (STATE_RECEIVE_CLOSED == socket->state)))
2181 GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
2182 "%s: Attempting to write on a closed (OR) not-yet-established"
2184 GNUNET_i2s (&socket->our_id));
2187 if (GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH * max_payload_size < size)
2188 size = GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH * max_payload_size;
2189 num_needed_packets = (size + (max_payload_size - 1)) / max_payload_size;
2190 io_handle = GNUNET_malloc (sizeof (struct GNUNET_STREAM_IOWriteHandle));
2192 /* Divide the given buffer into packets for sending */
2193 for (packet=0; packet < num_needed_packets; packet++)
2195 if ((packet + 1) * max_payload_size < size)
2197 payload_size = max_payload_size;
2198 packet_size = MAX_PACKET_SIZE;
2202 payload_size = size - packet * max_payload_size;
2203 packet_size = payload_size + sizeof (struct
2204 GNUNET_STREAM_DataMessage);
2206 io_handle->messages[packet] = GNUNET_malloc (packet_size);
2207 io_handle->messages[packet]->header.header.size = htons (packet_size);
2208 io_handle->messages[packet]->header.header.type =
2209 htons (GNUNET_MESSAGE_TYPE_STREAM_DATA);
2210 io_handle->messages[packet]->sequence_number =
2211 htons (socket->write_sequence_number++);
2212 io_handle->messages[packet]->offset = htons (socket->write_offset);
2214 /* FIXME: Remove the fixed delay for ack deadline; Set it to the value
2215 determined from RTT */
2216 io_handle->messages[packet]->ack_deadline =
2217 GNUNET_TIME_relative_hton (GNUNET_TIME_relative_multiply
2218 (GNUNET_TIME_UNIT_SECONDS, 5));
2219 data_msg = io_handle->messages[packet];
2220 /* Copy data from given buffer to the packet */
2221 memcpy (&data_msg[1],
2224 sweep += payload_size;
2225 socket->write_offset += payload_size;
2227 socket->write_handle = io_handle;
2228 write_data (socket);
2232 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2233 "%s() END\n", __func__);
2238 * Tries to read data from the stream
2240 * @param socket the socket representing a stream
2241 * @param timeout the timeout period
2242 * @param proc function to call with data (once only)
2243 * @param proc_cls the closure for proc
2244 * @return handle to cancel the operation
2246 struct GNUNET_STREAM_IOReadHandle *
2247 GNUNET_STREAM_read (struct GNUNET_STREAM_Socket *socket,
2248 struct GNUNET_TIME_Relative timeout,
2249 GNUNET_STREAM_DataProcessor proc,
2252 struct GNUNET_STREAM_IOReadHandle *read_handle;
2254 /* Return NULL if there is already a read handle; the user has to cancel that
2255 first before continuing or has to wait until it is completed */
2256 if (NULL != socket->read_handle) return NULL;
2258 read_handle = GNUNET_malloc (sizeof (struct GNUNET_STREAM_IOReadHandle));
2259 read_handle->proc = proc;
2260 socket->read_handle = read_handle;
2262 /* Check if we have a packet at bitmap 0 */
2263 if (GNUNET_YES == ackbitmap_is_bit_set (&socket->ack_bitmap,
2266 socket->read_task = GNUNET_SCHEDULER_add_now (&call_read_processor,
2271 /* Setup the read timeout task */
2272 socket->read_io_timeout_task = GNUNET_SCHEDULER_add_delayed (timeout,
2280 * Cancel pending write operation.
2282 * @param ioh handle to operation to cancel
2285 GNUNET_STREAM_io_write_cancel (struct GNUNET_STREAM_IOWriteHandle *ioh)
2292 * Cancel pending read operation.
2294 * @param ioh handle to operation to cancel
2297 GNUNET_STREAM_io_read_cancel (struct GNUNET_STREAM_IOReadHandle *ioh)