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 fs/gnunet-service-fs_stream.c
23 * @brief non-anonymous file-transfer
24 * @author Christian Grothoff
27 #include "gnunet_constants.h"
28 #include "gnunet_util_lib.h"
29 #include "gnunet_stream_lib.h"
30 #include "gnunet_protocols.h"
31 #include "gnunet_applications.h"
32 #include "gnunet-service-fs.h"
33 #include "gnunet-service-fs_indexing.h"
34 #include "gnunet-service-fs_stream.h"
37 * After how long do we termiante idle connections?
39 #define IDLE_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 2)
43 * A message in the queue to be written to the stream.
50 struct WriteQueueItem *next;
55 struct WriteQueueItem *prev;
58 * Number of bytes of payload, allocated at the end of this struct.
65 * Information we keep around for each active streaming client.
72 struct StreamClient *next;
77 struct StreamClient *prev;
80 * Socket for communication.
82 struct GNUNET_STREAM_Socket *socket;
85 * Handle for active read operation, or NULL.
87 struct GNUNET_STREAM_IOReadHandle *rh;
90 * Handle for active write operation, or NULL.
92 struct GNUNET_STREAM_IOWriteHandle *wh;
95 * Head of write queue.
97 struct WriteQueueItem *wqi_head;
100 * Tail of write queue.
102 struct WriteQueueItem *wqi_tail;
105 * Tokenizer for requests.
107 struct GNUNET_SERVER_MessageStreamTokenizer *mst;
110 * Current active request to the datastore, if we have one pending.
112 struct GNUNET_DATASTORE_QueueEntry *qe;
115 * Task that is scheduled to asynchronously terminate the connection.
117 GNUNET_SCHEDULER_TaskIdentifier terminate_task;
120 * Task that is scheduled to terminate idle connections.
122 GNUNET_SCHEDULER_TaskIdentifier timeout_task;
125 * Size of the last write that was initiated.
133 * Query from one peer, asking the other for CHK-data.
135 struct StreamQueryMessage
139 * Type is GNUNET_MESSAGE_TYPE_FS_STREAM_QUERY.
141 struct GNUNET_MessageHeader header;
144 * Block type must be DBLOCK or IBLOCK.
149 * Query hash from CHK (hash of encrypted block).
151 struct GNUNET_HashCode query;
157 * Reply to a StreamQueryMessage.
159 struct StreamReplyMessage
163 * Type is GNUNET_MESSAGE_TYPE_FS_STREAM_REPLY.
165 struct GNUNET_MessageHeader header;
168 * Block type must be DBLOCK or IBLOCK.
173 * Expiration time for the block.
175 struct GNUNET_TIME_AbsoluteNBO expiration;
177 /* followed by the encrypted block */
183 * Handle for a stream to another peer.
189 * Handle for a request that is going out via stream API.
191 struct GSF_StreamRequest
197 struct GSF_StreamRequest *next;
202 struct GSF_StreamRequest *prev;
205 * Which stream is this request associated with?
207 struct StreamHandle *sh;
210 * Function to call with the result.
212 GSF_StreamReplyProcessor proc;
220 * Query to transmit to the other peer.
222 struct GNUNET_HashCode query;
225 * Desired type for the reply.
227 enum GNUNET_BLOCK_Type type;
230 * Did we transmit this request already? YES if we are
231 * in the 'waiting' DLL, NO if we are in the 'pending' DLL.
238 * Handle for a stream to another peer.
243 * Head of DLL of pending requests on this stream.
245 struct GSF_StreamRequest *pending_head;
248 * Tail of DLL of pending requests on this stream.
250 struct GSF_StreamRequest *pending_tail;
253 * Map from query to 'struct GSF_StreamRequest's waiting for
256 struct GNUNET_CONTAINER_MultiHashMap *waiting_map;
259 * Connection to the other peer.
261 struct GNUNET_STREAM_Socket *stream;
264 * Handle for active read operation, or NULL.
266 struct GNUNET_STREAM_IOReadHandle *rh;
269 * Handle for active write operation, or NULL.
271 struct GNUNET_STREAM_IOWriteHandle *wh;
274 * Tokenizer for replies.
276 struct GNUNET_SERVER_MessageStreamTokenizer *mst;
279 * Which peer does this stream go to?
281 struct GNUNET_PeerIdentity target;
284 * Task to kill inactive streams (we keep them around for
285 * a few seconds to give the application a chance to give
288 GNUNET_SCHEDULER_TaskIdentifier timeout_task;
291 * Task to reset streams that had errors (asynchronously,
292 * as we may not be able to do it immediately during a
293 * callback from the stream API).
295 GNUNET_SCHEDULER_TaskIdentifier reset_task;
298 * Is this stream ready for transmission?
306 * Listen socket for incoming requests.
308 static struct GNUNET_STREAM_ListenSocket *listen_socket;
311 * Head of DLL of stream clients.
313 static struct StreamClient *sc_head;
316 * Tail of DLL of stream clients.
318 static struct StreamClient *sc_tail;
321 * Number of active stream clients in the 'sc_*'-DLL.
323 static unsigned int sc_count;
326 * Maximum allowed number of stream clients.
328 static unsigned long long sc_count_max;
331 * Map from peer identities to 'struct StreamHandles' with streams to
334 static struct GNUNET_CONTAINER_MultiHashMap *stream_map;
337 /* ********************* client-side code ************************* */
340 * Iterator called on each entry in a waiting map to
341 * call the 'proc' continuation and release associated
344 * @param cls the 'struct StreamHandle'
345 * @param key the key of the entry in the map (the query)
346 * @param value the 'struct GSF_StreamRequest' to clean up
347 * @return GNUNET_YES (continue to iterate)
350 free_waiting_entry (void *cls,
351 const struct GNUNET_HashCode *key,
354 struct GSF_StreamRequest *sr = value;
356 sr->proc (sr->proc_cls, GNUNET_BLOCK_TYPE_ANY,
357 GNUNET_TIME_UNIT_FOREVER_ABS,
359 GSF_stream_query_cancel (sr);
365 * Destroy a stream handle.
367 * @param sh stream to process
370 destroy_stream_handle (struct StreamHandle *sh)
372 struct GSF_StreamRequest *sr;
374 while (NULL != (sr = sh->pending_head))
376 sr->proc (sr->proc_cls, GNUNET_BLOCK_TYPE_ANY,
377 GNUNET_TIME_UNIT_FOREVER_ABS,
379 GSF_stream_query_cancel (sr);
381 GNUNET_CONTAINER_multihashmap_iterate (sh->waiting_map,
385 GNUNET_STREAM_io_write_cancel (sh->wh);
387 GNUNET_STREAM_io_read_cancel (sh->rh);
388 if (GNUNET_SCHEDULER_NO_TASK != sh->timeout_task)
389 GNUNET_SCHEDULER_cancel (sh->timeout_task);
390 GNUNET_STREAM_close (sh->stream);
391 GNUNET_assert (GNUNET_OK ==
392 GNUNET_CONTAINER_multihashmap_remove (stream_map,
393 &sh->target.hashPubKey,
395 GNUNET_CONTAINER_multihashmap_destroy (sh->waiting_map);
401 * Transmit pending requests via the stream.
403 * @param sh stream to process
406 transmit_pending (struct StreamHandle *sh);
410 * Function called once the stream is ready for transmission.
412 * @param cls the 'struct StreamHandle'
413 * @param socket stream socket handle
416 stream_ready_cb (void *cls,
417 struct GNUNET_STREAM_Socket *socket)
419 struct StreamHandle *sh = cls;
421 sh->is_ready = GNUNET_YES;
422 transmit_pending (sh);
427 * Iterator called on each entry in a waiting map to
428 * move it back to the pending list.
430 * @param cls the 'struct StreamHandle'
431 * @param key the key of the entry in the map (the query)
432 * @param value the 'struct GSF_StreamRequest' to move to pending
433 * @return GNUNET_YES (continue to iterate)
436 move_to_pending (void *cls,
437 const struct GNUNET_HashCode *key,
440 struct StreamHandle *sh = cls;
441 struct GSF_StreamRequest *sr = value;
443 GNUNET_assert (GNUNET_YES ==
444 GNUNET_CONTAINER_multihashmap_remove (sh->waiting_map,
447 GNUNET_CONTAINER_DLL_insert (sh->pending_head,
450 sr->was_transmitted = GNUNET_NO;
456 * We had a serious error, tear down and re-create stream from scratch.
458 * @param sh stream to reset
461 reset_stream (struct StreamHandle *sh)
463 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
464 "Resetting stream to %s\n",
465 GNUNET_i2s (&sh->target));
467 GNUNET_STREAM_io_read_cancel (sh->rh);
468 GNUNET_STREAM_close (sh->stream);
469 sh->is_ready = GNUNET_NO;
470 GNUNET_CONTAINER_multihashmap_iterate (sh->waiting_map,
473 sh->stream = GNUNET_STREAM_open (GSF_cfg,
475 GNUNET_APPLICATION_TYPE_FS_BLOCK_TRANSFER,
476 &stream_ready_cb, sh,
477 GNUNET_STREAM_OPTION_END);
482 * Task called when it is time to destroy an inactive stream.
484 * @param cls the 'struct StreamHandle' to tear down
485 * @param tc scheduler context, unused
488 stream_timeout (void *cls,
489 const struct GNUNET_SCHEDULER_TaskContext *tc)
491 struct StreamHandle *sh = cls;
493 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
494 "Timeout on stream to %s\n",
495 GNUNET_i2s (&sh->target));
496 sh->timeout_task = GNUNET_SCHEDULER_NO_TASK;
497 destroy_stream_handle (sh);
502 * Task called when it is time to reset an stream.
504 * @param cls the 'struct StreamHandle' to tear down
505 * @param tc scheduler context, unused
508 reset_stream_task (void *cls,
509 const struct GNUNET_SCHEDULER_TaskContext *tc)
511 struct StreamHandle *sh = cls;
513 sh->reset_task = GNUNET_SCHEDULER_NO_TASK;
519 * We had a serious error, tear down and re-create stream from scratch,
520 * but do so asynchronously.
522 * @param sh stream to reset
525 reset_stream_async (struct StreamHandle *sh)
527 if (GNUNET_SCHEDULER_NO_TASK == sh->reset_task)
528 sh->reset_task = GNUNET_SCHEDULER_add_now (&reset_stream_task,
534 * We got a reply from the stream. Process it.
536 * @param cls the struct StreamHandle
537 * @param status the status of the stream at the time this function is called
538 * @param data traffic from the other side
539 * @param size the number of bytes available in data read; will be 0 on timeout
540 * @return number of bytes of processed from 'data' (any data remaining should be
541 * given to the next time the read processor is called).
544 handle_stream_reply (void *cls,
545 enum GNUNET_STREAM_Status status,
549 struct StreamHandle *sh = cls;
552 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
553 "Received %u bytes from stream to %s\n",
555 GNUNET_i2s (&sh->target));
557 GNUNET_SERVER_mst_receive (sh->mst,
560 GNUNET_NO, GNUNET_NO))
563 reset_stream_async (sh);
566 sh->rh = GNUNET_STREAM_read (sh->stream,
567 GNUNET_TIME_UNIT_FOREVER_REL,
568 &handle_stream_reply,
575 * Functions of this signature are called whenever we transmitted a
576 * query via a stream.
578 * @param cls the struct StreamHandle for which we did the write call
579 * @param status the status of the stream at the time this function is called;
580 * GNUNET_OK if writing to stream was completed successfully,
581 * GNUNET_STREAM_SHUTDOWN if the stream is shutdown for writing in the
583 * @param size the number of bytes written
586 query_write_continuation (void *cls,
587 enum GNUNET_STREAM_Status status,
590 struct StreamHandle *sh = cls;
593 if ( (GNUNET_STREAM_OK != status) ||
594 (sizeof (struct StreamQueryMessage) != size) )
599 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
600 "Successfully transmitted %u bytes via stream to %s\n",
602 GNUNET_i2s (&sh->target));
604 sh->rh = GNUNET_STREAM_read (sh->stream,
605 GNUNET_TIME_UNIT_FOREVER_REL,
606 &handle_stream_reply,
608 transmit_pending (sh);
613 * Transmit pending requests via the stream.
615 * @param sh stream to process
618 transmit_pending (struct StreamHandle *sh)
620 struct StreamQueryMessage sqm;
621 struct GSF_StreamRequest *sr;
625 sr = sh->pending_head;
628 GNUNET_CONTAINER_DLL_remove (sh->pending_head,
631 GNUNET_CONTAINER_multihashmap_put (sh->waiting_map,
634 GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
635 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
636 "Sending query via stream to %s\n",
637 GNUNET_i2s (&sh->target));
638 sr->was_transmitted = GNUNET_YES;
639 sqm.header.size = htons (sizeof (sqm));
640 sqm.header.type = htons (GNUNET_MESSAGE_TYPE_FS_STREAM_QUERY);
641 sqm.type = htonl (sr->type);
642 sqm.query = sr->query;
643 sh->wh = GNUNET_STREAM_write (sh->stream,
645 GNUNET_TIME_UNIT_FOREVER_REL,
646 &query_write_continuation,
652 * Closure for 'handle_reply'.
654 struct HandleReplyClosure
663 * Expiration time for the block.
665 struct GNUNET_TIME_Absolute expiration;
668 * Number of bytes in 'data'.
675 enum GNUNET_BLOCK_Type type;
678 * Did we have a matching query?
685 * Iterator called on each entry in a waiting map to
688 * @param cls the 'struct HandleReplyClosure'
689 * @param key the key of the entry in the map (the query)
690 * @param value the 'struct GSF_StreamRequest' to handle result for
691 * @return GNUNET_YES (continue to iterate)
694 handle_reply (void *cls,
695 const struct GNUNET_HashCode *key,
698 struct HandleReplyClosure *hrc = cls;
699 struct GSF_StreamRequest *sr = value;
701 sr->proc (sr->proc_cls,
706 GSF_stream_query_cancel (sr);
707 hrc->found = GNUNET_YES;
713 * Functions with this signature are called whenever a
714 * complete reply is received.
716 * Do not call GNUNET_SERVER_mst_destroy in callback
718 * @param cls closure with the 'struct StreamHandle'
719 * @param client identification of the client, NULL
720 * @param message the actual message
721 * @return GNUNET_OK on success, GNUNET_SYSERR to stop further processing
726 const struct GNUNET_MessageHeader *message)
728 struct StreamHandle *sh = cls;
729 const struct StreamReplyMessage *srm;
730 struct HandleReplyClosure hrc;
732 enum GNUNET_BLOCK_Type type;
733 struct GNUNET_HashCode query;
735 msize = ntohs (message->size);
736 switch (ntohs (message->type))
738 case GNUNET_MESSAGE_TYPE_FS_STREAM_REPLY:
739 if (sizeof (struct StreamReplyMessage) > msize)
742 reset_stream_async (sh);
743 return GNUNET_SYSERR;
745 srm = (const struct StreamReplyMessage *) message;
746 msize -= sizeof (struct StreamReplyMessage);
747 type = (enum GNUNET_BLOCK_Type) ntohl (srm->type);
749 GNUNET_BLOCK_get_key (GSF_block_ctx,
751 &srm[1], msize, &query))
754 reset_stream_async (sh);
755 return GNUNET_SYSERR;
757 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
758 "Received reply `%s' via stream\n",
759 GNUNET_h2s (&query));
760 GNUNET_STATISTICS_update (GSF_stats,
761 gettext_noop ("# replies received via stream"), 1,
764 hrc.data_size = msize;
765 hrc.expiration = GNUNET_TIME_absolute_ntoh (srm->expiration);
767 hrc.found = GNUNET_NO;
768 GNUNET_CONTAINER_multihashmap_get_multiple (sh->waiting_map,
772 if (GNUNET_NO == hrc.found)
774 GNUNET_STATISTICS_update (GSF_stats,
775 gettext_noop ("# replies received via stream dropped"), 1,
782 reset_stream_async (sh);
783 return GNUNET_SYSERR;
789 * Get (or create) a stream to talk to the given peer.
791 * @param target peer we want to communicate with
793 static struct StreamHandle *
794 get_stream (const struct GNUNET_PeerIdentity *target)
796 struct StreamHandle *sh;
798 sh = GNUNET_CONTAINER_multihashmap_get (stream_map,
799 &target->hashPubKey);
802 if (GNUNET_SCHEDULER_NO_TASK != sh->timeout_task)
804 GNUNET_SCHEDULER_cancel (sh->timeout_task);
805 sh->timeout_task = GNUNET_SCHEDULER_NO_TASK;
809 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
810 "Creating stream to %s\n",
811 GNUNET_i2s (target));
812 sh = GNUNET_malloc (sizeof (struct StreamHandle));
813 sh->mst = GNUNET_SERVER_mst_create (&reply_cb,
815 sh->waiting_map = GNUNET_CONTAINER_multihashmap_create (512, GNUNET_YES);
816 sh->target = *target;
817 sh->stream = GNUNET_STREAM_open (GSF_cfg,
819 GNUNET_APPLICATION_TYPE_FS_BLOCK_TRANSFER,
820 &stream_ready_cb, sh,
821 GNUNET_STREAM_OPTION_END);
822 GNUNET_assert (GNUNET_OK ==
823 GNUNET_CONTAINER_multihashmap_put (stream_map,
824 &sh->target.hashPubKey,
826 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
832 * Look for a block by directly contacting a particular peer.
834 * @param target peer that should have the block
835 * @param query hash to query for the block
836 * @param type desired type for the block
837 * @param proc function to call with result
838 * @param proc_cls closure for 'proc'
839 * @return handle to cancel the operation
841 struct GSF_StreamRequest *
842 GSF_stream_query (const struct GNUNET_PeerIdentity *target,
843 const struct GNUNET_HashCode *query,
844 enum GNUNET_BLOCK_Type type,
845 GSF_StreamReplyProcessor proc, void *proc_cls)
847 struct StreamHandle *sh;
848 struct GSF_StreamRequest *sr;
850 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
851 "Preparing to send query for %s via stream to %s\n",
853 GNUNET_i2s (target));
854 sh = get_stream (target);
855 sr = GNUNET_malloc (sizeof (struct GSF_StreamRequest));
858 sr->proc_cls = proc_cls;
861 GNUNET_CONTAINER_DLL_insert (sh->pending_head,
864 if (GNUNET_YES == sh->is_ready)
865 transmit_pending (sh);
871 * Cancel an active request; must not be called after 'proc'
874 * @param sr request to cancel
877 GSF_stream_query_cancel (struct GSF_StreamRequest *sr)
879 struct StreamHandle *sh = sr->sh;
881 if (GNUNET_YES == sr->was_transmitted)
882 GNUNET_assert (GNUNET_OK ==
883 GNUNET_CONTAINER_multihashmap_remove (sh->waiting_map,
887 GNUNET_CONTAINER_DLL_remove (sh->pending_head,
891 if ( (0 == GNUNET_CONTAINER_multihashmap_size (sh->waiting_map)) &&
892 (NULL == sh->pending_head) )
893 sh->timeout_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
899 /* ********************* server-side code ************************* */
903 * We're done with a particular client, clean up.
905 * @param sc client to clean up
908 terminate_stream (struct StreamClient *sc)
910 GNUNET_STATISTICS_update (GSF_stats,
911 gettext_noop ("# stream connections active"), -1,
913 if (GNUNET_SCHEDULER_NO_TASK != sc->terminate_task)
914 GNUNET_SCHEDULER_cancel (sc->terminate_task);
915 if (GNUNET_SCHEDULER_NO_TASK != sc->timeout_task)
916 GNUNET_SCHEDULER_cancel (sc->timeout_task);
918 GNUNET_STREAM_io_read_cancel (sc->rh);
920 GNUNET_STREAM_io_write_cancel (sc->wh);
922 GNUNET_DATASTORE_cancel (sc->qe);
923 GNUNET_SERVER_mst_destroy (sc->mst);
924 GNUNET_STREAM_close (sc->socket);
925 struct WriteQueueItem *wqi;
926 while (NULL != (wqi = sc->wqi_head))
928 GNUNET_CONTAINER_DLL_remove (sc->wqi_head,
935 GNUNET_CONTAINER_DLL_remove (sc_head,
944 * Task run to asynchronously terminate the stream.
946 * @param cls the 'struct StreamClient'
947 * @param tc scheduler context
950 terminate_stream_task (void *cls,
951 const struct GNUNET_SCHEDULER_TaskContext *tc)
953 struct StreamClient *sc = cls;
955 sc->terminate_task = GNUNET_SCHEDULER_NO_TASK;
956 terminate_stream (sc);
961 * Task run to asynchronously terminate the stream due to timeout.
963 * @param cls the 'struct StreamClient'
964 * @param tc scheduler context
967 timeout_stream_task (void *cls,
968 const struct GNUNET_SCHEDULER_TaskContext *tc)
970 struct StreamClient *sc = cls;
972 sc->timeout_task = GNUNET_SCHEDULER_NO_TASK;
973 terminate_stream (sc);
978 * Reset the timeout for the stream client (due to activity).
980 * @param sc client handle to reset timeout for
983 refresh_timeout_task (struct StreamClient *sc)
985 if (GNUNET_SCHEDULER_NO_TASK != sc->timeout_task)
986 GNUNET_SCHEDULER_cancel (sc->timeout_task);
987 sc->timeout_task = GNUNET_SCHEDULER_add_delayed (IDLE_TIMEOUT,
988 &timeout_stream_task,
994 * We had a serious error, termiante stream,
995 * but do so asynchronously.
997 * @param sc stream to reset
1000 terminate_stream_async (struct StreamClient *sc)
1002 if (GNUNET_SCHEDULER_NO_TASK == sc->terminate_task)
1003 sc->terminate_task = GNUNET_SCHEDULER_add_now (&terminate_stream_task,
1009 * Functions of this signature are called whenever data is available from the
1012 * @param cls the closure from GNUNET_STREAM_read
1013 * @param status the status of the stream at the time this function is called
1014 * @param data traffic from the other side
1015 * @param size the number of bytes available in data read; will be 0 on timeout
1016 * @return number of bytes of processed from 'data' (any data remaining should be
1017 * given to the next time the read processor is called).
1020 process_request (void *cls,
1021 enum GNUNET_STREAM_Status status,
1027 * We're done handling a request from a client, read the next one.
1029 * @param sc client to continue reading requests from
1032 continue_reading (struct StreamClient *sc)
1037 GNUNET_SERVER_mst_receive (sc->mst,
1040 GNUNET_NO, GNUNET_YES);
1041 if (GNUNET_NO == ret)
1043 refresh_timeout_task (sc);
1044 sc->rh = GNUNET_STREAM_read (sc->socket,
1045 GNUNET_TIME_UNIT_FOREVER_REL,
1052 * Transmit the next entry from the write queue.
1054 * @param sc where to process the write queue
1057 continue_writing (struct StreamClient *sc);
1061 * Functions of this signature are called whenever data is available from the
1064 * @param cls the closure from GNUNET_STREAM_read
1065 * @param status the status of the stream at the time this function is called
1066 * @param data traffic from the other side
1067 * @param size the number of bytes available in data read; will be 0 on timeout
1068 * @return number of bytes of processed from 'data' (any data remaining should be
1069 * given to the next time the read processor is called).
1072 process_request (void *cls,
1073 enum GNUNET_STREAM_Status status,
1077 struct StreamClient *sc = cls;
1081 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1082 "Received %u byte query via stream\n",
1083 (unsigned int) size);
1086 case GNUNET_STREAM_OK:
1088 GNUNET_SERVER_mst_receive (sc->mst,
1091 GNUNET_NO, GNUNET_YES);
1092 if (GNUNET_NO == ret)
1093 return size; /* more messages in MST */
1094 if (GNUNET_SYSERR == ret)
1096 GNUNET_break_op (0);
1097 terminate_stream_async (sc);
1101 case GNUNET_STREAM_TIMEOUT:
1102 case GNUNET_STREAM_SHUTDOWN:
1103 case GNUNET_STREAM_SYSERR:
1104 terminate_stream_async (sc);
1110 continue_writing (sc);
1116 * Sending a reply was completed, continue processing.
1118 * @param cls closure with the struct StreamClient which sent the query
1119 * @param status result code for the operation
1120 * @param size number of bytes that were transmitted
1123 write_continuation (void *cls,
1124 enum GNUNET_STREAM_Status status,
1127 struct StreamClient *sc = cls;
1130 if ( (GNUNET_STREAM_OK != status) ||
1131 (size != sc->reply_size) )
1133 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1134 "Transmission of reply failed, terminating stream\n");
1135 terminate_stream (sc);
1138 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1139 "Transmitted %u byte reply via stream\n",
1140 (unsigned int) size);
1141 GNUNET_STATISTICS_update (GSF_stats,
1142 gettext_noop ("# Blocks transferred via stream"), 1,
1144 continue_writing (sc);
1149 * Transmit the next entry from the write queue.
1151 * @param sc where to process the write queue
1154 continue_writing (struct StreamClient *sc)
1156 struct WriteQueueItem *wqi;
1159 return; /* write already pending */
1160 if (NULL == (wqi = sc->wqi_head))
1162 continue_reading (sc);
1165 GNUNET_CONTAINER_DLL_remove (sc->wqi_head,
1168 sc->wh = GNUNET_STREAM_write (sc->socket,
1169 &wqi[1], wqi->msize,
1170 GNUNET_TIME_UNIT_FOREVER_REL,
1171 &write_continuation,
1176 terminate_stream (sc);
1183 * Process a datum that was stored in the datastore.
1185 * @param cls closure with the struct StreamClient which sent the query
1186 * @param key key for the content
1187 * @param size number of bytes in data
1188 * @param data content stored
1189 * @param type type of the content
1190 * @param priority priority of the content
1191 * @param anonymity anonymity-level for the content
1192 * @param expiration expiration time for the content
1193 * @param uid unique identifier for the datum;
1194 * maybe 0 if no unique identifier is available
1197 handle_datastore_reply (void *cls,
1198 const struct GNUNET_HashCode * key,
1199 size_t size, const void *data,
1200 enum GNUNET_BLOCK_Type type,
1203 struct GNUNET_TIME_Absolute
1204 expiration, uint64_t uid)
1206 struct StreamClient *sc = cls;
1207 size_t msize = size + sizeof (struct StreamReplyMessage);
1208 struct WriteQueueItem *wqi;
1209 struct StreamReplyMessage *srm;
1212 if (GNUNET_BLOCK_TYPE_FS_ONDEMAND == type)
1215 GNUNET_FS_handle_on_demand_block (key,
1217 priority, anonymity,
1219 &handle_datastore_reply,
1222 continue_writing (sc);
1226 if (msize > GNUNET_SERVER_MAX_MESSAGE_SIZE)
1229 continue_writing (sc);
1232 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1233 "Starting transmission of %u byte reply via stream\n",
1234 (unsigned int) size);
1235 wqi = GNUNET_malloc (sizeof (struct WriteQueueItem) + msize);
1236 srm = (struct StreamReplyMessage *) &wqi[1];
1237 srm->header.size = htons ((uint16_t) msize);
1238 srm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_STREAM_REPLY);
1239 srm->type = htonl (type);
1240 srm->expiration = GNUNET_TIME_absolute_hton (expiration);
1241 memcpy (&srm[1], data, size);
1242 sc->reply_size = msize;
1243 GNUNET_CONTAINER_DLL_insert (sc->wqi_head,
1246 continue_writing (sc);
1251 * Functions with this signature are called whenever a
1252 * complete query message is received.
1254 * Do not call GNUNET_SERVER_mst_destroy in callback
1256 * @param cls closure with the 'struct StreamClient'
1257 * @param client identification of the client, NULL
1258 * @param message the actual message
1259 * @return GNUNET_OK on success, GNUNET_SYSERR to stop further processing
1262 request_cb (void *cls,
1264 const struct GNUNET_MessageHeader *message)
1266 struct StreamClient *sc = cls;
1267 const struct StreamQueryMessage *sqm;
1269 switch (ntohs (message->type))
1271 case GNUNET_MESSAGE_TYPE_FS_STREAM_QUERY:
1272 if (sizeof (struct StreamQueryMessage) !=
1273 ntohs (message->size))
1275 GNUNET_break_op (0);
1276 terminate_stream_async (sc);
1277 return GNUNET_SYSERR;
1279 sqm = (const struct StreamQueryMessage *) message;
1280 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1281 "Received query for `%s' via stream\n",
1282 GNUNET_h2s (&sqm->query));
1283 GNUNET_STATISTICS_update (GSF_stats,
1284 gettext_noop ("# queries received via stream"), 1,
1286 refresh_timeout_task (sc);
1287 sc->qe = GNUNET_DATASTORE_get_key (GSF_dsh,
1292 GSF_datastore_queue_size,
1293 GNUNET_TIME_UNIT_FOREVER_REL,
1294 &handle_datastore_reply, sc);
1296 continue_writing (sc);
1299 GNUNET_break_op (0);
1300 terminate_stream_async (sc);
1301 return GNUNET_SYSERR;
1307 * Functions of this type are called upon new stream connection from other peers
1308 * or upon binding error which happen when the app_port given in
1309 * GNUNET_STREAM_listen() is already taken.
1311 * @param cls the closure from GNUNET_STREAM_listen
1312 * @param socket the socket representing the stream; NULL on binding error
1313 * @param initiator the identity of the peer who wants to establish a stream
1314 * with us; NULL on binding error
1315 * @return GNUNET_OK to keep the socket open, GNUNET_SYSERR to close the
1316 * stream (the socket will be invalid after the call)
1319 accept_cb (void *cls,
1320 struct GNUNET_STREAM_Socket *socket,
1321 const struct GNUNET_PeerIdentity *initiator)
1323 struct StreamClient *sc;
1326 return GNUNET_SYSERR;
1327 if (sc_count >= sc_count_max)
1329 GNUNET_STATISTICS_update (GSF_stats,
1330 gettext_noop ("# stream client connections rejected"), 1,
1332 return GNUNET_SYSERR;
1334 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1335 "Accepting inbound stream connection from `%s'\n",
1336 GNUNET_i2s (initiator));
1337 GNUNET_STATISTICS_update (GSF_stats,
1338 gettext_noop ("# stream connections active"), 1,
1340 sc = GNUNET_malloc (sizeof (struct StreamClient));
1341 sc->socket = socket;
1342 sc->mst = GNUNET_SERVER_mst_create (&request_cb,
1344 sc->rh = GNUNET_STREAM_read (sc->socket,
1345 GNUNET_TIME_UNIT_FOREVER_REL,
1348 GNUNET_CONTAINER_DLL_insert (sc_head,
1352 refresh_timeout_task (sc);
1358 * Initialize subsystem for non-anonymous file-sharing.
1363 stream_map = GNUNET_CONTAINER_multihashmap_create (16, GNUNET_YES);
1365 GNUNET_CONFIGURATION_get_value_number (GSF_cfg,
1367 "MAX_STREAM_CLIENTS",
1370 listen_socket = GNUNET_STREAM_listen (GSF_cfg,
1371 GNUNET_APPLICATION_TYPE_FS_BLOCK_TRANSFER,
1373 GNUNET_STREAM_OPTION_END);
1379 * Function called on each active streams to shut them down.
1382 * @param key target peer, unused
1383 * @param value the 'struct StreamHandle' to destroy
1384 * @return GNUNET_YES (continue to iterate)
1387 release_streams (void *cls,
1388 const struct GNUNET_HashCode *key,
1391 struct StreamHandle *sh = value;
1393 destroy_stream_handle (sh);
1399 * Shutdown subsystem for non-anonymous file-sharing.
1404 struct StreamClient *sc;
1406 while (NULL != (sc = sc_head))
1407 terminate_stream (sc);
1408 if (NULL != listen_socket)
1410 GNUNET_STREAM_listen_close (listen_socket);
1411 listen_socket = NULL;
1413 GNUNET_CONTAINER_multihashmap_iterate (stream_map,
1416 GNUNET_CONTAINER_multihashmap_destroy (stream_map);
1420 /* end of gnunet-service-fs_stream.c */