*/
/**
- * @file fs/gnunet-service-fs_mesh.c
+ * @file fs/gnunet-service-fs_mesh_server.c
* @brief non-anonymous file-transfer
* @author Christian Grothoff
*
* TODO:
- * - MESH2 API doesn't allow flow control for server yet (needed!)
- * - likely need to register clean up handler with mesh to handle
- * client disconnect (likely leaky right now)
- * - server is optional, currently client code will NPE if we have
- * no server, again MESH2 API requirement forcing this for now
- * - message handlers are symmetric for client/server, should be
- * separated (currently clients can get requests and servers can
- * handle answers, not good)
- * - code is entirely untested
- * - might have overlooked a few possible simplifications
* - PORT is set to old application type, unsure if we should keep
* it that way (fine for now)
*/
{
/**
* DLL
- */
+ */
struct MeshClient *next;
/**
* DLL
- */
+ */
struct MeshClient *prev;
/**
- * Socket for communication.
- */
- struct GNUNET_MESH_Tunnel *socket;
+ * Channel for communication.
+ */
+ struct GNUNET_MESH_Channel *channel;
/**
* Handle for active write operation, or NULL.
- */
+ */
struct GNUNET_MESH_TransmitHandle *wh;
/**
* Tail of write queue.
*/
struct WriteQueueItem *wqi_tail;
-
+
/**
* Current active request to the datastore, if we have one pending.
*/
/**
* Size of the last write that was initiated.
- */
+ */
size_t reply_size;
};
/**
- * Listen socket for incoming requests.
+ * Listen channel for incoming requests.
*/
-static struct GNUNET_MESH_Handle *listen_socket;
+static struct GNUNET_MESH_Handle *listen_channel;
/**
* Head of DLL of mesh clients.
- */
+ */
static struct MeshClient *sc_head;
/**
* Tail of DLL of mesh clients.
- */
+ */
static struct MeshClient *sc_tail;
/**
-/* ********************* server-side code ************************* */
-
-
-/**
- * We're done with a particular client, clean up.
- *
- * @param sc client to clean up
- */
-static void
-terminate_mesh (struct MeshClient *sc)
-{
- struct WriteQueueItem *wqi;
-
- fprintf (stderr,
- "terminate mesh called for %p\n",
- sc);
- GNUNET_STATISTICS_update (GSF_stats,
- gettext_noop ("# mesh connections active"), -1,
- GNUNET_NO);
- if (GNUNET_SCHEDULER_NO_TASK != sc->terminate_task)
- GNUNET_SCHEDULER_cancel (sc->terminate_task);
- if (GNUNET_SCHEDULER_NO_TASK != sc->timeout_task)
- GNUNET_SCHEDULER_cancel (sc->timeout_task);
- if (NULL != sc->wh)
- GNUNET_MESH_notify_transmit_ready_cancel (sc->wh);
- if (NULL != sc->qe)
- GNUNET_DATASTORE_cancel (sc->qe);
- while (NULL != (wqi = sc->wqi_head))
- {
- GNUNET_CONTAINER_DLL_remove (sc->wqi_head,
- sc->wqi_tail,
- wqi);
- GNUNET_free (wqi);
- }
- GNUNET_CONTAINER_DLL_remove (sc_head,
- sc_tail,
- sc);
- sc_count--;
- GNUNET_free (sc);
-}
-
-
/**
* Task run to asynchronously terminate the mesh due to timeout.
*
* @param cls the 'struct MeshClient'
* @param tc scheduler context
- */
+ */
static void
timeout_mesh_task (void *cls,
const struct GNUNET_SCHEDULER_TaskContext *tc)
{
struct MeshClient *sc = cls;
- struct GNUNET_MESH_Tunnel *tun;
+ struct GNUNET_MESH_Channel *tun;
sc->timeout_task = GNUNET_SCHEDULER_NO_TASK;
- tun = sc->socket;
- sc->socket = NULL;
- GNUNET_MESH_tunnel_destroy (tun);
+ tun = sc->channel;
+ sc->channel = NULL;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Timeout for inactive mesh client %p\n",
+ sc);
+ GNUNET_MESH_channel_destroy (tun);
}
refresh_timeout_task (struct MeshClient *sc)
{
if (GNUNET_SCHEDULER_NO_TASK != sc->timeout_task)
- GNUNET_SCHEDULER_cancel (sc->timeout_task);
+ GNUNET_SCHEDULER_cancel (sc->timeout_task);
sc->timeout_task = GNUNET_SCHEDULER_add_delayed (IDLE_TIMEOUT,
&timeout_mesh_task,
sc);
continue_reading (struct MeshClient *sc)
{
refresh_timeout_task (sc);
- GNUNET_MESH_receive_done (sc->socket);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Finished processing mesh request from client %p, ready to receive the next one\n",
+ sc);
+ GNUNET_MESH_receive_done (sc->channel);
}
/**
* Send a reply now, mesh is ready.
*
- * @param cls closure with the struct MeshClient which sent the query
- * @param size number of bytes available in 'buf'
+ * @param cls closure with the `struct MeshClient` which sent the query
+ * @param size number of bytes available in @a buf
* @param buf where to write the message
- * @return number of bytes written to 'buf'
+ * @return number of bytes written to @a buf
*/
static size_t
write_continuation (void *cls,
void *buf)
{
struct MeshClient *sc = cls;
+ struct GNUNET_MESH_Channel *tun;
struct WriteQueueItem *wqi;
size_t ret;
"Write queue empty, reading more requests\n");
return 0;
}
- if (0 == size)
+ if ( (0 == size) ||
+ (size < wqi->msize) )
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Transmission of reply failed, terminating mesh\n");
- terminate_mesh (sc);
+ tun = sc->channel;
+ sc->channel = NULL;
+ GNUNET_MESH_channel_destroy (tun);
return 0;
}
GNUNET_CONTAINER_DLL_remove (sc->wqi_head,
sc->wqi_tail,
wqi);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Transmitted %u byte reply via mesh\n",
- (unsigned int) size);
+ "Transmitted %u byte reply via mesh to %p\n",
+ (unsigned int) size,
+ sc);
GNUNET_STATISTICS_update (GSF_stats,
gettext_noop ("# Blocks transferred via mesh"), 1,
GNUNET_NO);
continue_writing (struct MeshClient *sc)
{
struct WriteQueueItem *wqi;
+ struct GNUNET_MESH_Channel *tun;
if (NULL != sc->wh)
{
continue_reading (sc);
return;
}
- sc->wh = GNUNET_MESH_notify_transmit_ready (sc->socket, GNUNET_NO,
+ sc->wh = GNUNET_MESH_notify_transmit_ready (sc->channel, GNUNET_NO,
GNUNET_TIME_UNIT_FOREVER_REL,
- wqi->msize,
+ wqi->msize,
&write_continuation,
sc);
if (NULL == sc->wh)
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Write failed; terminating mesh\n");
- terminate_mesh (sc);
+ tun = sc->channel;
+ sc->channel = NULL;
+ GNUNET_MESH_channel_destroy (tun);
return;
}
}
/**
* Process a datum that was stored in the datastore.
*
- * @param cls closure with the struct MeshClient which sent the query
+ * @param cls closure with the `struct MeshClient` which sent the query
* @param key key for the content
- * @param size number of bytes in data
+ * @param size number of bytes in @a data
* @param data content stored
* @param type type of the content
* @param priority priority of the content
* @param uid unique identifier for the datum;
* maybe 0 if no unique identifier is available
*/
-static void
+static void
handle_datastore_reply (void *cls,
- const struct GNUNET_HashCode * key,
+ const struct GNUNET_HashCode *key,
size_t size, const void *data,
enum GNUNET_BLOCK_Type type,
uint32_t priority,
uint32_t anonymity,
- struct GNUNET_TIME_Absolute
- expiration, uint64_t uid)
+ struct GNUNET_TIME_Absolute expiration,
+ uint64_t uid)
{
struct MeshClient *sc = cls;
size_t msize = size + sizeof (struct MeshReplyMessage);
struct MeshReplyMessage *srm;
sc->qe = NULL;
+ if (NULL == data)
+ {
+ /* no result, this should not really happen, as for
+ non-anonymous routing only peers that HAVE the
+ answers should be queried; OTOH, this is not a
+ hard error as we might have had the answer in the
+ past and the user might have unindexed it. Hence
+ we log at level "INFO" for now. */
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Have no answer for query `%s'\n",
+ GNUNET_h2s (key));
+ GNUNET_STATISTICS_update (GSF_stats,
+ gettext_noop ("# queries received via mesh not answered"), 1,
+ GNUNET_NO);
+ continue_writing (sc);
+ return;
+ }
if (GNUNET_BLOCK_TYPE_FS_ONDEMAND == type)
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Performing on-demand encoding\n");
+ "Performing on-demand encoding for query %s\n",
+ GNUNET_h2s (key));
if (GNUNET_OK !=
GNUNET_FS_handle_on_demand_block (key,
size, data, type,
continue_writing (sc);
return;
}
+ GNUNET_break (GNUNET_BLOCK_TYPE_ANY != type);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Starting transmission of %u byte reply for query `%s' via mesh\n",
+ "Starting transmission of %u byte reply of type %d for query `%s' via mesh to %p\n",
(unsigned int) size,
- GNUNET_h2s (key));
+ (unsigned int) type,
+ GNUNET_h2s (key),
+ sc);
wqi = GNUNET_malloc (sizeof (struct WriteQueueItem) + msize);
wqi->msize = msize;
srm = (struct MeshReplyMessage *) &wqi[1];
* Functions with this signature are called whenever a
* complete query message is received.
*
- * Do not call GNUNET_SERVER_mst_destroy in callback
+ * Do not call #GNUNET_SERVER_mst_destroy in callback
*
* @param cls closure with the 'struct MeshClient'
- * @param tunnel tunnel handle
- * @param tunnel_ctx tunnel context
+ * @param channel channel handle
+ * @param channel_ctx channel context
* @param message the actual message
- * @return GNUNET_OK on success, GNUNET_SYSERR to stop further processing
+ * @return #GNUNET_OK on success, #GNUNET_SYSERR to stop further processing
*/
static int
request_cb (void *cls,
- struct GNUNET_MESH_Tunnel *tunnel,
- void **tunnel_ctx,
+ struct GNUNET_MESH_Channel *channel,
+ void **channel_ctx,
const struct GNUNET_MessageHeader *message)
{
- struct MeshClient *sc = *tunnel_ctx;
+ struct MeshClient *sc = *channel_ctx;
const struct MeshQueryMessage *sqm;
- fprintf (stderr,
- "Request gets %p\n",
- sc);
sqm = (const struct MeshQueryMessage *) message;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Received query for `%s' via mesh\n",
- GNUNET_h2s (&sqm->query));
+ "Received query for `%s' via mesh from client %p\n",
+ GNUNET_h2s (&sqm->query),
+ sc);
GNUNET_STATISTICS_update (GSF_stats,
gettext_noop ("# queries received via mesh"), 1,
GNUNET_NO);
0,
&sqm->query,
ntohl (sqm->type),
- 0 /* priority */,
+ 0 /* priority */,
GSF_datastore_queue_size,
GNUNET_TIME_UNIT_FOREVER_REL,
&handle_datastore_reply, sc);
* Functions of this type are called upon new mesh connection from other peers.
*
* @param cls the closure from GNUNET_MESH_connect
- * @param socket the socket representing the mesh
+ * @param channel the channel representing the mesh
* @param initiator the identity of the peer who wants to establish a mesh
* with us; NULL on binding error
* @param port mesh port used for the incoming connection
- * @return initial tunnel context (our 'struct MeshClient')
+ * @param options channel option flags
+ * @return initial channel context (our 'struct MeshClient')
*/
static void *
accept_cb (void *cls,
- struct GNUNET_MESH_Tunnel *socket,
+ struct GNUNET_MESH_Channel *channel,
const struct GNUNET_PeerIdentity *initiator,
- uint32_t port)
+ uint32_t port, enum GNUNET_MESH_ChannelOption options)
{
struct MeshClient *sc;
- GNUNET_assert (NULL != socket);
+ GNUNET_assert (NULL != channel);
if (sc_count >= sc_count_max)
{
GNUNET_STATISTICS_update (GSF_stats,
gettext_noop ("# mesh client connections rejected"), 1,
GNUNET_NO);
- GNUNET_MESH_tunnel_destroy (socket);
+ GNUNET_MESH_channel_destroy (channel);
return NULL;
}
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Accepting inbound mesh connection from `%s'\n",
- GNUNET_i2s (initiator));
GNUNET_STATISTICS_update (GSF_stats,
gettext_noop ("# mesh connections active"), 1,
GNUNET_NO);
sc = GNUNET_new (struct MeshClient);
- sc->socket = socket;
+ sc->channel = channel;
GNUNET_CONTAINER_DLL_insert (sc_head,
sc_tail,
sc);
sc_count++;
refresh_timeout_task (sc);
- fprintf (stderr,
- "Accept returns %p\n",
- sc);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Accepting inbound mesh connection from `%s' as client %p\n",
+ GNUNET_i2s (initiator),
+ sc);
return sc;
}
/**
* Function called by mesh when a client disconnects.
- * Cleans up our 'struct MeshClient' of that tunnel.
+ * Cleans up our 'struct MeshClient' of that channel.
*
* @param cls NULL
- * @param tunnel tunnel of the disconnecting client
- * @param tunnel_ctx our 'struct MeshClient'
+ * @param channel channel of the disconnecting client
+ * @param channel_ctx our 'struct MeshClient'
*/
static void
cleaner_cb (void *cls,
- const struct GNUNET_MESH_Tunnel *tunnel,
- void *tunnel_ctx)
+ const struct GNUNET_MESH_Channel *channel,
+ void *channel_ctx)
{
- struct MeshClient *sc = tunnel_ctx;
+ struct MeshClient *sc = channel_ctx;
+ struct WriteQueueItem *wqi;
- fprintf (stderr,
- "Cleaner called with %p\n",
- sc);
- if (NULL != sc)
- terminate_mesh (sc);
+ if (NULL == sc)
+ return;
+ sc->channel = NULL;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Terminating mesh connection with client %p\n",
+ sc);
+ GNUNET_STATISTICS_update (GSF_stats,
+ gettext_noop ("# mesh connections active"), -1,
+ GNUNET_NO);
+ if (GNUNET_SCHEDULER_NO_TASK != sc->terminate_task)
+ GNUNET_SCHEDULER_cancel (sc->terminate_task);
+ if (GNUNET_SCHEDULER_NO_TASK != sc->timeout_task)
+ GNUNET_SCHEDULER_cancel (sc->timeout_task);
+ if (NULL != sc->wh)
+ GNUNET_MESH_notify_transmit_ready_cancel (sc->wh);
+ if (NULL != sc->qe)
+ GNUNET_DATASTORE_cancel (sc->qe);
+ while (NULL != (wqi = sc->wqi_head))
+ {
+ GNUNET_CONTAINER_DLL_remove (sc->wqi_head,
+ sc->wqi_tail,
+ wqi);
+ GNUNET_free (wqi);
+ }
+ GNUNET_CONTAINER_DLL_remove (sc_head,
+ sc_tail,
+ sc);
+ sc_count--;
+ GNUNET_free (sc);
}
"MAX_MESH_CLIENTS",
&sc_count_max))
return;
- listen_socket = GNUNET_MESH_connect (GSF_cfg,
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Initializing mesh FS server with a limit of %llu connections\n",
+ sc_count_max);
+ listen_channel = GNUNET_MESH_connect (GSF_cfg,
NULL,
&accept_cb,
&cleaner_cb,
void
GSF_mesh_stop_server ()
{
- struct MeshClient *sc;
-
- while (NULL != (sc = sc_head))
- terminate_mesh (sc);
- if (NULL != listen_socket)
+ if (NULL != listen_channel)
{
- GNUNET_MESH_disconnect (listen_socket);
- listen_socket = NULL;
+ GNUNET_MESH_disconnect (listen_channel);
+ listen_channel = NULL;
}
+ GNUNET_assert (NULL == sc_head);
+ GNUNET_assert (0 == sc_count);
}
/* end of gnunet-service-fs_mesh.c */
projects / oweals / gnunet.git / blobdiff