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 * @file stream/stream_api.c
23 * @brief Implementation of the stream library
24 * @author Sree Harsha Totakura
27 #include "gnunet_common.h"
28 #include "gnunet_stream_lib.h"
31 * states in the Protocol
36 * Client initialization state
41 * Listener initialization state
46 * Pre-connection establishment state
51 * State where a connection has been established
56 * State where the socket is closed on our side and waiting to be ACK'ed
58 STATE_RECEIVE_CLOSE_WAIT,
61 * State where the socket is closed for reading
66 * State where the socket is closed on our side and waiting to be ACK'ed
68 STATE_TRANSMIT_CLOSE_WAIT,
71 * State where the socket is closed for writing
73 STATE_TRANSMIT_CLOSED,
76 * State where the socket is closed on our side and waiting to be ACK'ed
81 * State where the socket is closed
88 * The STREAM Socket Handler
90 struct GNUNET_STREAM_Socket
95 struct GNUNET_MESH_Handle *mesh;
98 * The mesh tunnel handle
100 struct GNUNET_MESH_Tunnel *tunnel;
103 * The session id associated with this stream connection
108 * The peer identity of the peer at the other end of the stream
110 GNUNET_PeerIdentity other_peer;
113 * Stream open closure
118 * Stream open callback
120 GNUNET_STREAM_OpenCallback open_cb;
123 * Retransmission timeout
125 struct GNUNET_TIME_Relative retransmit_timeout;
128 * The state of the protocol associated with this socket
133 * The status of the socket
135 enum GNUNET_STREAM_Status status;
138 * The current transmit handle (if a pending transmit request exists)
140 struct GNUNET_MESH_TransmitHandle *transmit_handle;
143 * The current message associated with the transmit handle
145 struct GNUNET_MessageHeader *message;
150 * A socket for listening
152 struct GNUNET_STREAM_ListenSocket
158 struct GNUNET_MESH_Handle *mesh;
163 GNUNET_MESH_ApplicationType port;
166 * The callback function which is called after successful opening socket
168 GNUNET_STREAM_ListenCallback listen_cb;
171 * The call back closure
178 * Default value in seconds for various timeouts
180 static unsigned int default_timeout = 300;
184 * Callback function from send_message
186 * @param cls closure the socket on which the send message was called
187 * @param size number of bytes available in buf
188 * @param buf where the callee should write the message
189 * @return number of bytes written to buf
192 send_message_notify (void *cls, size_t size, void *buf)
194 struct GNUNET_STREAM_Socket *socket = cls;
197 socket->transmit_handle = NULL; /* Remove the transmit handle */
198 if (0 == size) /* Socket closed? */
200 // statistics ("message timeout")
203 GNUNET_log (GNUNET_ERROR_TYPE_INFO,
204 "Message not sent as tunnel was closed \n");
207 else /* Size is more or equal to what was requested */
209 ret = ntohs (socket->message->size);
210 GNUNET_assert (size >= ret);
211 memcpy (buf, socket->message, ret);
213 GNUNET_free (socket->message); /* Free the message memory */
214 socket->message = NULL;
220 * Sends a message using the mesh connection of a socket
222 * @param socket the socket whose mesh connection is used
223 * @param message the message to be sent
226 send_message (struct GNUNET_STREAM_Socket *socket,
227 struct GNUNET_MessageHeader *message)
229 socket->message = message;
230 socket->transmit_handle =
231 GNUNET_MESH_notify_transmit_ready (socket->tunnel,
233 timeout, /* FIXME: Maxdelay */
235 ntohs (message->size),
236 &send_message_notify,
241 * Makes state transition dependending on the given state
243 * @param socket the socket whose state has to be transitioned
246 make_state_transition (struct GNUNET_STREAM_Socket *socket)
253 * Message Handler for mesh
255 * @param cls closure (set from GNUNET_MESH_connect)
256 * @param tunnel connection to the other end
257 * @param tunnel_ctx place to store local state associated with the tunnel
258 * @param sender who sent the message
259 * @param message the actual message
260 * @param atsi performance data for the connection
261 * @return GNUNET_OK to keep the connection open,
262 * GNUNET_SYSERR to close it (signal serious error)
265 handle_data (void *cls,
266 struct GNUNET_MESH_Tunnel *tunnel,
268 const struct GNUNET_PeerIdentity *sender,
269 const struct GNUNET_MessageHeader *message,
270 const struct GNUNET_ATS_Information*atsi)
272 struct GNUNET_STREAM_Socket *socket = cls;
274 const struct GNUNET_STREAM_DataMessage *data_msg;
277 size = ntohs (message->size);
278 if (size < sizeof (struct GNUNET_STREAM_DataMessage))
281 return GNUNET_SYSERR;
283 data_msg = (const struct GNUNET_STREAM_DataMessage *) message;
284 size -= sizeof (Struct GNUNET_STREAM_DataMessage);
285 payload = &data_msg[1];
293 * Message Handler for mesh
295 * @param cls closure (set from GNUNET_MESH_connect)
296 * @param tunnel connection to the other end
297 * @param tunnel_ctx place to store local state associated with the tunnel
298 * @param sender who sent the message
299 * @param ack the actual message
300 * @param atsi performance data for the connection
301 * @return GNUNET_OK to keep the connection open,
302 * GNUNET_SYSERR to close it (signal serious error)
305 handle_ack (struct STREAM_Socket *socket,
306 struct GNUNET_MESH_Tunnel *tunnel,
307 const struct GNUNET_PeerIdentity *sender,
308 const struct GNUNET_STREAM_AckMessage *ack,
309 const struct GNUNET_ATS_Information*atsi)
316 * Message Handler for mesh
318 * @param cls the 'struct GNUNET_STREAM_Socket'
319 * @param tunnel connection to the other end
320 * @param tunnel_ctx unused
321 * @param sender who sent the message
322 * @param message the actual message
323 * @param atsi performance data for the connection
324 * @return GNUNET_OK to keep the connection open,
325 * GNUNET_SYSERR to close it (signal serious error)
328 client_handle_ack (void *cls,
329 struct GNUNET_MESH_Tunnel *tunnel,
331 const struct GNUNET_PeerIdentity *sender,
332 const struct GNUNET_MessageHeader *message,
333 const struct GNUNET_ATS_Information*atsi)
335 struct GNUNET_STREAM_Socket *socket = cls;
336 const struct GNUNET_STREAM_AckMessage *ack = (const struct GNUNET_STREAM_AckMessage *) message;
338 return handle_ack (socket, tunnel, sender, ack, atsi);
343 * Message Handler for mesh
345 * @param cls the server's listen socket
346 * @param tunnel connection to the other end
347 * @param tunnel_ctx pointer to the 'struct GNUNET_STREAM_Socket*'
348 * @param sender who sent the message
349 * @param message the actual message
350 * @param atsi performance data for the connection
351 * @return GNUNET_OK to keep the connection open,
352 * GNUNET_SYSERR to close it (signal serious error)
355 server_handle_ack (void *cls,
356 struct GNUNET_MESH_Tunnel *tunnel,
358 const struct GNUNET_PeerIdentity *sender,
359 const struct GNUNET_MessageHeader *message,
360 const struct GNUNET_ATS_Information*atsi)
362 struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
363 const struct GNUNET_STREAM_AckMessage *ack = (const struct GNUNET_STREAM_AckMessage *) message;
365 return handle_ack (socket, tunnel, sender, ack, atsi);
370 * For client message handlers, the stream socket is in the
373 static struct GNUNET_MESH_MessageHandler client_message_handlers[] = {
374 {&client_handle_data, GNUNET_MESSAGE_TYPE_STREAM_DATA, 0},
375 {&client_handle_ack, GNUNET_MESSAGE_TYPE_STREAM_ACK, sizeof (struct GNUNET_STREAM_AckMessage) },
376 {&client_handle_hello, GNUNET_MESSAGE_TYPE_STREAM_HELLO, 0},
377 {&client_handle_hello_ack, GNUNET_MESSAGE_TYPE_STREAM_HELLO_ACK, 0},
378 {&client_handle_reset, GNUNET_MESSAGE_TYPE_STREAM_RESET, 0},
379 {&client_handle_data, GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE, 0},
380 {&client_handle_data, GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE_ACK, 0},
381 {&client_handle_data, GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE, 0},
382 {&client_handle_data, GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE_ACK, 0},
383 {&client_handle_data, GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE, 0},
384 {&client_handle_data, GNUNET_MESSAGE_TYPE_STREAM_CLOSE, 0},
385 {&client_handle_data, GNUNET_MESSAGE_TYPE_STREAM_CLOSE_ACK, 0},
391 * For server message handlers, the stream socket is in the
392 * tunnel context, and the listen socket in the closure argument.
394 static struct GNUNET_MESH_MessageHandler server_message_handlers[] = {
395 {&server_handle_data, GNUNET_MESSAGE_TYPE_STREAM_DATA, 0},
396 {&server_handle_ack, GNUNET_MESSAGE_TYPE_STREAM_ACK, sizeof (struct GNUNET_STREAM_AckMessage) },
397 {&server_handle_hello, GNUNET_MESSAGE_TYPE_STREAM_HELLO, 0},
398 {&server_handle_hello_ack, GNUNET_MESSAGE_TYPE_STREAM_HELLO_ACK, 0},
399 {&server_handle_reset, GNUNET_MESSAGE_TYPE_STREAM_RESET, 0},
400 {&server_handle_data, GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE, 0},
401 {&server_handle_data, GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE_ACK, 0},
402 {&server_handle_data, GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE, 0},
403 {&server_handle_data, GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE_ACK, 0},
404 {&server_handle_data, GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE, 0},
405 {&server_handle_data, GNUNET_MESSAGE_TYPE_STREAM_CLOSE, 0},
406 {&server_handle_data, GNUNET_MESSAGE_TYPE_STREAM_CLOSE_ACK, 0},
412 * Function called when our target peer is connected to our tunnel
414 * @param peer the peer identity of the target
415 * @param atsi performance data for the connection
418 mesh_peer_connect_callback (void *cls,
419 const struct GNUNET_PeerIdentity *peer,
420 const struct GNUNET_ATS_Information * atsi)
422 const struct GNUNET_STREAM_Socket *socket = cls;
424 if (0 != memcmp (socket->other_peer,
426 sizeof (struct GNUNET_PeerIdentity)))
428 GNUNET_log (GNUNET_ERROR_TYPE_INFO,
429 "A peer (%s) which is not our target has\
430 connected to our tunnel", GNUNET_i2s (peer));
434 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
435 "Target peer %s connected\n", GNUNET_i2s (peer));
437 /* Set state to INIT */
438 socket->state = STATE_INIT;
440 /* Try to achieve ESTABLISHED state */
441 make_state_transition (socket);
443 /* Call open callback */
444 if (NULL == open_cls)
446 GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
447 "STREAM_open callback is NULL\n");
449 if (NULL != socket->open_cb)
451 socket->open_cb (socket->open_cls, socket);
457 * Function called when our target peer is disconnected from our tunnel
459 * @param peer the peer identity of the target
462 mesh_peer_disconnect_callback (void *cls,
463 const struct GNUNET_PeerIdentity *peer)
470 * Function to find the mapped socket of a tunnel
472 * @param tunnel the tunnel whose associated socket has to be retrieved
473 * @return the socket corresponding to the tunnel
475 static struct GNUNET_STREAM_Socket *
476 find_socket (const struct GNUNET_MESH_Tunnel *tunnel)
478 /* Search tunnel in a list or hashtable and retrieve the socket */
487 * Tries to open a stream to the target peer
489 * @param cfg configuration to use
490 * @param target the target peer to which the stream has to be opened
491 * @param app_port the application port number which uniquely identifies this
493 * @param open_cb this function will be called after stream has be established
494 * @param open_cb_cls the closure for open_cb
495 * @param ... options to the stream, terminated by GNUNET_STREAM_OPTION_END
496 * @return if successful it returns the stream socket; NULL if stream cannot be
499 struct GNUNET_STREAM_Socket *
500 GNUNET_STREAM_open (const struct GNUNET_CONFIGURATION_Handle *cfg,
501 const struct GNUNET_PeerIdentity *target,
502 GNUNET_MESH_ApplicationType app_port,
503 GNUNET_STREAM_OpenCallback open_cb,
507 struct GNUNET_STREAM_Socket *socket;
508 enum GNUNET_STREAM_Option option;
509 va_list vargs; /* Variable arguments */
511 socket = GNUNET_malloc (sizeof (struct GNUNET_STREAM_Socket));
512 socket->other_peer = *target;
513 socket->open_cb = open_cb;
514 socket->open_cls = open_cb_cls;
517 socket->retransmit_timeout =
518 GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, default_timeout);
520 va_start (vargs, open_cb_cls); /* Parse variable args */
522 option = va_arg (vargs, enum GNUNET_STREAM_Option);
525 case GNUNET_STREAM_OPTION_INITIAL_RETRANSMIT_TIMEOUT:
526 /* Expect struct GNUNET_TIME_Relative */
527 socket->retransmit_timeout = va_arg (vargs,
528 struct GNUNET_TIME_Relative);
530 case GNUNET_STREAM_OPTION_END:
534 } while (0 != option);
535 va_end (vargs); /* End of variable args parsing */
537 socket->mesh = GNUNET_MESH_connect (cfg, /* the configuration handle */
538 1, /* QUEUE size as parameter? */
540 NULL, /* No inbound tunnel handler */
541 NULL, /* No inbound tunnel cleaner */
543 NULL); /* We don't get inbound tunnels */
544 // FIXME: if (NULL == socket->mesh) ...
546 /* Now create the mesh tunnel to target */
547 socket->tunnel = GNUNET_MESH_tunnel_create (socket->mesh,
548 NULL, /* Tunnel context */
549 &mesh_peer_connect_callback,
550 &mesh_peer_disconnect_callback,
552 // FIXME: if (NULL == socket->tunnel) ...
561 * @param socket the stream socket
564 GNUNET_STREAM_close (struct GNUNET_STREAM_Socket *socket)
566 /* Clear Transmit handles */
567 if (NULL != socket->transmit_handle)
569 GNUNET_MESH_notify_transmit_ready_cancel (socket->transmit_handle);
571 /* Clear existing message queue message */
572 if (NULL != socket->message)
574 GNUNET_free (socket->message);
576 /* Close associated tunnel */
577 if (NULL != socket->tunnel)
579 GNUNET_MESH_tunnel_destroy (socket->tunnel);
581 /* Close mesh connection */
582 if (NULL != socket->mesh)
584 GNUNET_MESH_disconnect (socket->mesh);
586 GNUNET_free (socket);
591 * Method called whenever a peer creates a tunnel to us
594 * @param tunnel new handle to the tunnel
595 * @param initiator peer that started the tunnel
596 * @param atsi performance information for the tunnel
597 * @return initial tunnel context for the tunnel
598 * (can be NULL -- that's not an error)
601 new_tunnel_notify (void *cls,
602 struct GNUNET_MESH_Tunnel *tunnel,
603 const struct GNUNET_PeerIdentity *initiator,
604 const struct GNUNET_ATS_Information *atsi)
606 struct GNUNET_STREAM_ListenSocket *lsocket = cls;
607 struct GNUNET_STREAM_Socket *socket;
609 socket = GNUNET_malloc (sizeof (struct GNUNET_STREAM_Socket));
610 socket->tunnel = tunnel;
611 socket->session_id = 0; /* FIXME */
612 socket->other_peer = GNUNET_malloc (sizeof (struct GNUNET_PeerIdentity));
613 memcpy (socket->other_peer, initiator, sizeof (struct GNUNET_PeerIdentity));
614 socket->state = STATE_LISTEN;
616 if (GNUNET_SYSERR == lsocket->listen_cb (lsocket->listen_cb_cls,
620 socket->state = STATE_CLOSED;
621 make_state_transition (socket);
622 GNUNET_free (socket->other_peer);
623 GNUNET_free (socket);
624 GNUNET_MESH_tunnel_destroy (tunnel); /* Destroy the tunnel */
628 make_state_transition (socket);
635 * Function called whenever an inbound tunnel is destroyed. Should clean up
636 * any associated state. This function is NOT called if the client has
637 * explicitly asked for the tunnel to be destroyed using
638 * GNUNET_MESH_tunnel_destroy. It must NOT call GNUNET_MESH_tunnel_destroy on
641 * @param cls closure (set from GNUNET_MESH_connect)
642 * @param tunnel connection to the other end (henceforth invalid)
643 * @param tunnel_ctx place where local state associated
644 * with the tunnel is stored
647 tunnel_cleaner (void *cls,
648 const struct GNUNET_MESH_Tunnel *tunnel,
651 struct GNUNET_STREAM_ListenSocket *lsocket = cls;
652 struct GNUNET_STREAM_Socket *socket = tunnel_ctx;
654 socket = find_socket (tunnel);
655 GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
656 "Peer %s has terminated connection abruptly\n",
657 GNUNET_i2s (socket->other_peer));
659 socket->status = GNUNET_STREAM_SHUTDOWN;
660 /* Clear Transmit handles */
661 if (NULL != socket->transmit_handle)
663 GNUNET_MESH_notify_transmit_ready_cancel (socket->transmit_handle);
664 socket->transmit_handle = NULL;
667 /* Clear existing message queue message */
668 if (NULL != socket->message)
670 GNUNET_free (socket->message);
671 socket->message = NULL;
677 * Listens for stream connections for a specific application ports
679 * @param cfg the configuration to use
680 * @param app_port the application port for which new streams will be accepted
681 * @param listen_cb this function will be called when a peer tries to establish
683 * @param listen_cb_cls closure for listen_cb
684 * @return listen socket, NULL for any error
686 struct GNUNET_STREAM_ListenSocket *
687 GNUNET_STREAM_listen (const struct GNUNET_CONFIGURATION_Handle *cfg,
688 GNUNET_MESH_ApplicationType app_port,
689 GNUNET_STREAM_ListenCallback listen_cb,
692 /* FIXME: Add variable args for passing configration options? */
693 struct GNUNET_STREAM_ListenSocket *lsocket;
695 lsocket = GNUNET_malloc (sizeof (struct GNUNET_STREAM_ListenSocket));
696 lsocket->port = app_port;
697 lsocket->listen_cb = listen_cb;
698 lsocket->listen_cb_cls = listen_cb_cls;
699 lsocket->mesh = GNUNET_MESH_connect (cfg,
700 10, /* FIXME: QUEUE size as parameter? */
701 lsocket, /* Closure */
711 * Closes the listen socket
713 * @param socket the listen socket
716 GNUNET_STREAM_listen_close (struct GNUNET_STREAM_ListenSocket *lsocket)
718 /* Do house keeping */
720 /* Close MESH connection */
721 GNUNET_MESH_disconnect (lsocket->mesh);
723 GNUNET_free (lsocket);