src/fs/gnunet-service-fs_mesh_server.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2012, 2013 Christian Grothoff (and other contributing authors)
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 3, or (at your
   8      option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18      Boston, MA 02111-1307, USA.
  19 */
  20
  21 /**
  22  * @file fs/gnunet-service-fs_mesh.c
  23  * @brief non-anonymous file-transfer
  24  * @author Christian Grothoff
  25  *
  26  * TODO:
  27  * - PORT is set to old application type, unsure if we should keep
  28  *   it that way (fine for now)
  29  */
  30 #include "platform.h"
  31 #include "gnunet_constants.h"
  32 #include "gnunet_util_lib.h"
  33 #include "gnunet_mesh_service.h"
  34 #include "gnunet_protocols.h"
  35 #include "gnunet_applications.h"
  36 #include "gnunet-service-fs.h"
  37 #include "gnunet-service-fs_indexing.h"
  38 #include "gnunet-service-fs_mesh.h"
  39
  40 /**
  41  * After how long do we termiante idle connections?
  42  */
  43 #define IDLE_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 2)
  44
  45
  46 /**
  47  * A message in the queue to be written to the mesh.
  48  */
  49 struct WriteQueueItem
  50 {
  51   /**
  52    * Kept in a DLL.
  53    */
  54   struct WriteQueueItem *next;
  55
  56   /**
  57    * Kept in a DLL.
  58    */
  59   struct WriteQueueItem *prev;
  60
  61   /**
  62    * Number of bytes of payload, allocated at the end of this struct.
  63    */
  64   size_t msize;
  65 };
  66
  67
  68 /**
  69  * Information we keep around for each active meshing client.
  70  */
  71 struct MeshClient
  72 {
  73   /**
  74    * DLL
  75    */
  76   struct MeshClient *next;
  77
  78   /**
  79    * DLL
  80    */
  81   struct MeshClient *prev;
  82
  83   /**
  84    * Tunnel for communication.
  85    */
  86   struct GNUNET_MESH_Tunnel *tunnel;
  87
  88   /**
  89    * Handle for active write operation, or NULL.
  90    */
  91   struct GNUNET_MESH_TransmitHandle *wh;
  92
  93   /**
  94    * Head of write queue.
  95    */
  96   struct WriteQueueItem *wqi_head;
  97
  98   /**
  99    * Tail of write queue.
 100    */
 101   struct WriteQueueItem *wqi_tail;
 102
 103   /**
 104    * Current active request to the datastore, if we have one pending.
 105    */
 106   struct GNUNET_DATASTORE_QueueEntry *qe;
 107
 108   /**
 109    * Task that is scheduled to asynchronously terminate the connection.
 110    */
 111   GNUNET_SCHEDULER_TaskIdentifier terminate_task;
 112
 113   /**
 114    * Task that is scheduled to terminate idle connections.
 115    */
 116   GNUNET_SCHEDULER_TaskIdentifier timeout_task;
 117
 118   /**
 119    * Size of the last write that was initiated.
 120    */
 121   size_t reply_size;
 122
 123 };
 124
 125
 126 /**
 127  * Listen tunnel for incoming requests.
 128  */
 129 static struct GNUNET_MESH_Handle *listen_tunnel;
 130
 131 /**
 132  * Head of DLL of mesh clients.
 133  */
 134 static struct MeshClient *sc_head;
 135
 136 /**
 137  * Tail of DLL of mesh clients.
 138  */
 139 static struct MeshClient *sc_tail;
 140
 141 /**
 142  * Number of active mesh clients in the 'sc_*'-DLL.
 143  */
 144 static unsigned int sc_count;
 145
 146 /**
 147  * Maximum allowed number of mesh clients.
 148  */
 149 static unsigned long long sc_count_max;
 150
 151
 152
 153 /**
 154  * Task run to asynchronously terminate the mesh due to timeout.
 155  *
 156  * @param cls the 'struct MeshClient'
 157  * @param tc scheduler context
 158  */
 159 static void
 160 timeout_mesh_task (void *cls,
 161                      const struct GNUNET_SCHEDULER_TaskContext *tc)
 162 {
 163   struct MeshClient *sc = cls;
 164   struct GNUNET_MESH_Tunnel *tun;
 165
 166   sc->timeout_task = GNUNET_SCHEDULER_NO_TASK;
 167   tun = sc->tunnel;
 168   sc->tunnel = NULL;
 169   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 170               "Timeout for inactive mesh client %p\n",
 171               sc);
 172   GNUNET_MESH_tunnel_destroy (tun);
 173 }
 174
 175
 176 /**
 177  * Reset the timeout for the mesh client (due to activity).
 178  *
 179  * @param sc client handle to reset timeout for
 180  */
 181 static void
 182 refresh_timeout_task (struct MeshClient *sc)
 183 {
 184   if (GNUNET_SCHEDULER_NO_TASK != sc->timeout_task)
 185     GNUNET_SCHEDULER_cancel (sc->timeout_task);
 186   sc->timeout_task = GNUNET_SCHEDULER_add_delayed (IDLE_TIMEOUT,
 187                                                    &timeout_mesh_task,
 188                                                    sc);
 189 }
 190
 191
 192 /**
 193  * We're done handling a request from a client, read the next one.
 194  *
 195  * @param sc client to continue reading requests from
 196  */
 197 static void
 198 continue_reading (struct MeshClient *sc)
 199 {
 200   refresh_timeout_task (sc);
 201   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 202               "Finished processing mesh request from client %p, ready to receive the next one\n",
 203               sc);
 204   GNUNET_MESH_receive_done (sc->tunnel);
 205 }
 206
 207
 208 /**
 209  * Transmit the next entry from the write queue.
 210  *
 211  * @param sc where to process the write queue
 212  */
 213 static void
 214 continue_writing (struct MeshClient *sc);
 215
 216
 217 /**
 218  * Send a reply now, mesh is ready.
 219  *
 220  * @param cls closure with the struct MeshClient which sent the query
 221  * @param size number of bytes available in 'buf'
 222  * @param buf where to write the message
 223  * @return number of bytes written to 'buf'
 224  */
 225 static size_t
 226 write_continuation (void *cls,
 227                     size_t size,
 228                     void *buf)
 229 {
 230   struct MeshClient *sc = cls;
 231   struct GNUNET_MESH_Tunnel *tun;
 232   struct WriteQueueItem *wqi;
 233   size_t ret;
 234
 235   sc->wh = NULL;
 236   if (NULL == (wqi = sc->wqi_head))
 237   {
 238     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 239                 "Write queue empty, reading more requests\n");
 240     return 0;
 241   }
 242   if (0 == size)
 243   {
 244     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 245                 "Transmission of reply failed, terminating mesh\n");
 246     tun = sc->tunnel;
 247     sc->tunnel = NULL;
 248     GNUNET_MESH_tunnel_destroy (tun);
 249     return 0;
 250   }
 251   GNUNET_CONTAINER_DLL_remove (sc->wqi_head,
 252                                sc->wqi_tail,
 253                                wqi);
 254   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 255               "Transmitted %u byte reply via mesh to %p\n",
 256               (unsigned int) size,
 257               sc);
 258   GNUNET_STATISTICS_update (GSF_stats,
 259                             gettext_noop ("# Blocks transferred via mesh"), 1,
 260                             GNUNET_NO);
 261   memcpy (buf, &wqi[1], ret = wqi->msize);
 262   GNUNET_free (wqi);
 263   continue_writing (sc);
 264   return ret;
 265 }
 266
 267
 268 /**
 269  * Transmit the next entry from the write queue.
 270  *
 271  * @param sc where to process the write queue
 272  */
 273 static void
 274 continue_writing (struct MeshClient *sc)
 275 {
 276   struct WriteQueueItem *wqi;
 277   struct GNUNET_MESH_Tunnel *tun;
 278
 279   if (NULL != sc->wh)
 280   {
 281     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 282                 "Write pending, waiting for it to complete\n");
 283     return; /* write already pending */
 284   }
 285   if (NULL == (wqi = sc->wqi_head))
 286   {
 287     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 288                 "Write queue empty, reading more requests\n");
 289     continue_reading (sc);
 290     return;
 291   }
 292   sc->wh = GNUNET_MESH_notify_transmit_ready (sc->tunnel, GNUNET_NO,
 293                                               GNUNET_TIME_UNIT_FOREVER_REL,
 294                                               wqi->msize,
 295                                               &write_continuation,
 296                                               sc);
 297   if (NULL == sc->wh)
 298   {
 299     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 300                 "Write failed; terminating mesh\n");
 301     tun = sc->tunnel;
 302     sc->tunnel = NULL;
 303     GNUNET_MESH_tunnel_destroy (tun);
 304     return;
 305   }
 306 }
 307
 308
 309 /**
 310  * Process a datum that was stored in the datastore.
 311  *
 312  * @param cls closure with the struct MeshClient which sent the query
 313  * @param key key for the content
 314  * @param size number of bytes in data
 315  * @param data content stored
 316  * @param type type of the content
 317  * @param priority priority of the content
 318  * @param anonymity anonymity-level for the content
 319  * @param expiration expiration time for the content
 320  * @param uid unique identifier for the datum;
 321  *        maybe 0 if no unique identifier is available
 322  */
 323 static void
 324 handle_datastore_reply (void *cls,
 325                         const struct GNUNET_HashCode *key,
 326                         size_t size, const void *data,
 327                         enum GNUNET_BLOCK_Type type,
 328                         uint32_t priority,
 329                         uint32_t anonymity,
 330                         struct GNUNET_TIME_Absolute
 331                         expiration, uint64_t uid)
 332 {
 333   struct MeshClient *sc = cls;
 334   size_t msize = size + sizeof (struct MeshReplyMessage);
 335   struct WriteQueueItem *wqi;
 336   struct MeshReplyMessage *srm;
 337
 338   sc->qe = NULL;
 339   if (GNUNET_BLOCK_TYPE_FS_ONDEMAND == type)
 340   {
 341     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 342                 "Performing on-demand encoding for query %s\n",
 343                 GNUNET_h2s (key));
 344     if (GNUNET_OK !=
 345         GNUNET_FS_handle_on_demand_block (key,
 346                                           size, data, type,
 347                                           priority, anonymity,
 348                                           expiration, uid,
 349                                           &handle_datastore_reply,
 350                                           sc))
 351     {
 352       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 353                   "On-demand encoding request failed\n");
 354       continue_writing (sc);
 355     }
 356     return;
 357   }
 358   if (msize > GNUNET_SERVER_MAX_MESSAGE_SIZE)
 359   {
 360     GNUNET_break (0);
 361     continue_writing (sc);
 362     return;
 363   }
 364   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 365               "Starting transmission of %u byte reply for query `%s' via mesh to %p\n",
 366               (unsigned int) size,
 367               GNUNET_h2s (key),
 368               sc);
 369   wqi = GNUNET_malloc (sizeof (struct WriteQueueItem) + msize);
 370   wqi->msize = msize;
 371   srm = (struct MeshReplyMessage *) &wqi[1];
 372   srm->header.size = htons ((uint16_t) msize);
 373   srm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_MESH_REPLY);
 374   srm->type = htonl (type);
 375   srm->expiration = GNUNET_TIME_absolute_hton (expiration);
 376   memcpy (&srm[1], data, size);
 377   sc->reply_size = msize;
 378   GNUNET_CONTAINER_DLL_insert (sc->wqi_head,
 379                                sc->wqi_tail,
 380                                wqi);
 381   continue_writing (sc);
 382 }
 383
 384
 385 /**
 386  * Functions with this signature are called whenever a
 387  * complete query message is received.
 388  *
 389  * Do not call GNUNET_SERVER_mst_destroy in callback
 390  *
 391  * @param cls closure with the 'struct MeshClient'
 392  * @param tunnel tunnel handle
 393  * @param tunnel_ctx tunnel context
 394  * @param message the actual message
 395  * @return GNUNET_OK on success, GNUNET_SYSERR to stop further processing
 396  */
 397 static int
 398 request_cb (void *cls,
 399             struct GNUNET_MESH_Tunnel *tunnel,
 400             void **tunnel_ctx,
 401             const struct GNUNET_MessageHeader *message)
 402 {
 403   struct MeshClient *sc = *tunnel_ctx;
 404   const struct MeshQueryMessage *sqm;
 405
 406   sqm = (const struct MeshQueryMessage *) message;
 407   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 408               "Received query for `%s' via mesh from client %p\n",
 409               GNUNET_h2s (&sqm->query),
 410               sc);
 411   GNUNET_STATISTICS_update (GSF_stats,
 412                             gettext_noop ("# queries received via mesh"), 1,
 413                             GNUNET_NO);
 414   refresh_timeout_task (sc);
 415   sc->qe = GNUNET_DATASTORE_get_key (GSF_dsh,
 416                                      0,
 417                                      &sqm->query,
 418                                      ntohl (sqm->type),
 419                                      0 /* priority */,
 420                                      GSF_datastore_queue_size,
 421                                      GNUNET_TIME_UNIT_FOREVER_REL,
 422                                      &handle_datastore_reply, sc);
 423   if (NULL == sc->qe)
 424   {
 425     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 426                 "Queueing request with datastore failed (queue full?)\n");
 427     continue_writing (sc);
 428   }
 429   return GNUNET_OK;
 430 }
 431
 432
 433 /**
 434  * Functions of this type are called upon new mesh connection from other peers.
 435  *
 436  * @param cls the closure from GNUNET_MESH_connect
 437  * @param tunnel the tunnel representing the mesh
 438  * @param initiator the identity of the peer who wants to establish a mesh
 439  *            with us; NULL on binding error
 440  * @param port mesh port used for the incoming connection
 441  * @return initial tunnel context (our 'struct MeshClient')
 442  */
 443 static void *
 444 accept_cb (void *cls,
 445            struct GNUNET_MESH_Tunnel *tunnel,
 446            const struct GNUNET_PeerIdentity *initiator,
 447            uint32_t port)
 448 {
 449   struct MeshClient *sc;
 450
 451   GNUNET_assert (NULL != tunnel);
 452   if (sc_count >= sc_count_max)
 453   {
 454     GNUNET_STATISTICS_update (GSF_stats,
 455                               gettext_noop ("# mesh client connections rejected"), 1,
 456                               GNUNET_NO);
 457     GNUNET_MESH_tunnel_destroy (tunnel);
 458     return NULL;
 459   }
 460   GNUNET_STATISTICS_update (GSF_stats,
 461                             gettext_noop ("# mesh connections active"), 1,
 462                             GNUNET_NO);
 463   sc = GNUNET_new (struct MeshClient);
 464   sc->tunnel = tunnel;
 465   GNUNET_CONTAINER_DLL_insert (sc_head,
 466                                sc_tail,
 467                                sc);
 468   sc_count++;
 469   refresh_timeout_task (sc);
 470   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 471               "Accepting inbound mesh connection from `%s' as client %p\n",
 472               GNUNET_i2s (initiator),
 473               sc);
 474   return sc;
 475 }
 476
 477
 478 /**
 479  * Function called by mesh when a client disconnects.
 480  * Cleans up our 'struct MeshClient' of that tunnel.
 481  *
 482  * @param cls NULL
 483  * @param tunnel tunnel of the disconnecting client
 484  * @param tunnel_ctx our 'struct MeshClient'
 485  */
 486 static void
 487 cleaner_cb (void *cls,
 488             const struct GNUNET_MESH_Tunnel *tunnel,
 489             void *tunnel_ctx)
 490 {
 491   struct MeshClient *sc = tunnel_ctx;
 492   struct WriteQueueItem *wqi;
 493
 494   if (NULL == sc)
 495     return;
 496   sc->tunnel = NULL;
 497   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 498               "Terminating mesh connection with client %p\n",
 499               sc);
 500   GNUNET_STATISTICS_update (GSF_stats,
 501                             gettext_noop ("# mesh connections active"), -1,
 502                             GNUNET_NO);
 503   if (GNUNET_SCHEDULER_NO_TASK != sc->terminate_task)
 504     GNUNET_SCHEDULER_cancel (sc->terminate_task);
 505   if (GNUNET_SCHEDULER_NO_TASK != sc->timeout_task)
 506     GNUNET_SCHEDULER_cancel (sc->timeout_task);
 507   if (NULL != sc->wh)
 508     GNUNET_MESH_notify_transmit_ready_cancel (sc->wh);
 509   if (NULL != sc->qe)
 510     GNUNET_DATASTORE_cancel (sc->qe);
 511   while (NULL != (wqi = sc->wqi_head))
 512   {
 513     GNUNET_CONTAINER_DLL_remove (sc->wqi_head,
 514                                  sc->wqi_tail,
 515                                  wqi);
 516     GNUNET_free (wqi);
 517   }
 518   GNUNET_CONTAINER_DLL_remove (sc_head,
 519                                sc_tail,
 520                                sc);
 521   sc_count--;
 522   GNUNET_free (sc);
 523 }
 524
 525
 526 /**
 527  * Initialize subsystem for non-anonymous file-sharing.
 528  */
 529 void
 530 GSF_mesh_start_server ()
 531 {
 532   static const struct GNUNET_MESH_MessageHandler handlers[] = {
 533     { &request_cb, GNUNET_MESSAGE_TYPE_FS_MESH_QUERY, sizeof (struct MeshQueryMessage)},
 534     { NULL, 0, 0 }
 535   };
 536   static const uint32_t ports[] = {
 537     GNUNET_APPLICATION_TYPE_FS_BLOCK_TRANSFER,
 538     0
 539   };
 540
 541   if (GNUNET_YES !=
 542       GNUNET_CONFIGURATION_get_value_number (GSF_cfg,
 543                                              "fs",
 544                                              "MAX_MESH_CLIENTS",
 545                                              &sc_count_max))
 546     return;
 547   listen_tunnel = GNUNET_MESH_connect (GSF_cfg,
 548                                        NULL,
 549                                        &accept_cb,
 550                                        &cleaner_cb,
 551                                        handlers,
 552                                        ports);
 553 }
 554
 555
 556 /**
 557  * Shutdown subsystem for non-anonymous file-sharing.
 558  */
 559 void
 560 GSF_mesh_stop_server ()
 561 {
 562   if (NULL != listen_tunnel)
 563   {
 564     GNUNET_MESH_disconnect (listen_tunnel);
 565     listen_tunnel = NULL;
 566   }
 567   GNUNET_assert (NULL == sc_head);
 568   GNUNET_assert (0 == sc_count);
 569 }
 570
 571 /* end of gnunet-service-fs_mesh.c */