src/peerstore/gnunet-service-peerstore.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2014, 2015, 2016 GNUnet e.V.
   4
   5      GNUnet is free software: you can redistribute it and/or modify it
   6      under the terms of the GNU Affero General Public License as published
   7      by the Free Software Foundation, either version 3 of the License,
   8      or (at your 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      Affero General Public License for more details.
  14
  15      You should have received a copy of the GNU Affero General Public License
  16      along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17 */
  18
  19 /**
  20  * @file peerstore/gnunet-service-peerstore.c
  21  * @brief peerstore service implementation
  22  * @author Omar Tarabai
  23  */
  24 #include "platform.h"
  25 #include "gnunet_util_lib.h"
  26 #include "peerstore.h"
  27 #include "gnunet_peerstore_plugin.h"
  28 #include "peerstore_common.h"
  29
  30
  31 /**
  32  * Interval for expired records cleanup (in seconds)
  33  */
  34 #define EXPIRED_RECORDS_CLEANUP_INTERVAL 300    /* 5mins */
  35
  36 /**
  37  * Our configuration.
  38  */
  39 static const struct GNUNET_CONFIGURATION_Handle *cfg;
  40
  41 /**
  42  * Database plugin library name
  43  */
  44 static char *db_lib_name;
  45
  46 /**
  47  * Database handle
  48  */
  49 static struct GNUNET_PEERSTORE_PluginFunctions *db;
  50
  51 /**
  52  * Hashmap with all watch requests
  53  */
  54 static struct GNUNET_CONTAINER_MultiHashMap *watchers;
  55
  56 /**
  57  * Task run to clean up expired records.
  58  */
  59 static struct GNUNET_SCHEDULER_Task *expire_task;
  60
  61 /**
  62  * Are we in the process of shutting down the service? #GNUNET_YES / #GNUNET_NO
  63  */
  64 static int in_shutdown;
  65
  66 /**
  67  * Number of connected clients.
  68  */
  69 static unsigned int num_clients;
  70
  71
  72 /**
  73  * Perform the actual shutdown operations
  74  */
  75 static void
  76 do_shutdown ()
  77 {
  78   if (NULL != db_lib_name)
  79   {
  80     GNUNET_break (NULL ==
  81                   GNUNET_PLUGIN_unload (db_lib_name,
  82                                         db));
  83     GNUNET_free (db_lib_name);
  84     db_lib_name = NULL;
  85   }
  86   if (NULL != watchers)
  87   {
  88     GNUNET_CONTAINER_multihashmap_destroy (watchers);
  89     watchers = NULL;
  90   }
  91   if (NULL != expire_task)
  92   {
  93     GNUNET_SCHEDULER_cancel (expire_task);
  94     expire_task = NULL;
  95   }
  96   GNUNET_SCHEDULER_shutdown ();
  97 }
  98
  99
 100 /**
 101  * Task run during shutdown.
 102  *
 103  * @param cls unused
 104  */
 105 static void
 106 shutdown_task (void *cls)
 107 {
 108   in_shutdown = GNUNET_YES;
 109   if (0 == num_clients)      /* Only when no connected clients. */
 110     do_shutdown ();
 111 }
 112
 113
 114 /* Forward declaration */
 115 static void
 116 expire_records_continuation (void *cls,
 117                              int success);
 118
 119
 120 /**
 121  * Deletes any expired records from storage
 122  */
 123 static void
 124 cleanup_expired_records (void *cls)
 125 {
 126   int ret;
 127
 128   expire_task = NULL;
 129   GNUNET_assert (NULL != db);
 130   ret = db->expire_records (db->cls,
 131                             GNUNET_TIME_absolute_get (),
 132                             &expire_records_continuation,
 133                             NULL);
 134   if (GNUNET_OK != ret)
 135   {
 136     GNUNET_assert (NULL == expire_task);
 137     expire_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
 138                                                 (GNUNET_TIME_UNIT_SECONDS,
 139                                                  EXPIRED_RECORDS_CLEANUP_INTERVAL),
 140                                                 &cleanup_expired_records,
 141                                                 NULL);
 142   }
 143 }
 144
 145
 146 /**
 147  * Continuation to expire_records called by the peerstore plugin
 148  *
 149  * @param cls unused
 150  * @param success count of records deleted or #GNUNET_SYSERR
 151  */
 152 static void
 153 expire_records_continuation (void *cls,
 154                              int success)
 155 {
 156   if (success > 0)
 157     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 158                 "%d records expired.\n",
 159                 success);
 160   GNUNET_assert (NULL == expire_task);
 161   expire_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
 162                                               (GNUNET_TIME_UNIT_SECONDS,
 163                                                EXPIRED_RECORDS_CLEANUP_INTERVAL),
 164                                               &cleanup_expired_records,
 165                                               NULL);
 166 }
 167
 168
 169 /**
 170  * A client disconnected.  Remove all of its data structure entries.
 171  *
 172  * @param cls closure, NULL
 173  * @param client identification of the client
 174  * @param mq the message queue
 175  * @return
 176  */
 177 static void *
 178 client_connect_cb (void *cls,
 179                    struct GNUNET_SERVICE_Client *client,
 180                    struct GNUNET_MQ_Handle *mq)
 181 {
 182   num_clients++;
 183   return client;
 184 }
 185
 186
 187 /**
 188  * Search for a disconnected client and remove it
 189  *
 190  * @param cls closuer, a `struct GNUNET_SERVICE_Client`
 191  * @param key hash of record key
 192  * @param value the watcher client, a `struct GNUNET_SERVICE_Client *`
 193  * @return #GNUNET_OK to continue iterating
 194  */
 195 static int
 196 client_disconnect_it (void *cls,
 197                       const struct GNUNET_HashCode *key,
 198                       void *value)
 199 {
 200   if (value == cls)
 201   {
 202     GNUNET_CONTAINER_multihashmap_remove (watchers,
 203                                           key,
 204                                           value);
 205     num_clients++;
 206   }
 207   return GNUNET_OK;
 208 }
 209
 210
 211 /**
 212  * A client disconnected.  Remove all of its data structure entries.
 213  *
 214  * @param cls closure, NULL
 215  * @param client identification of the client
 216  */
 217 static void
 218 client_disconnect_cb (void *cls,
 219                       struct GNUNET_SERVICE_Client *client,
 220                       void *app_cls)
 221 {
 222   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 223               "A client disconnected, cleaning up.\n");
 224   if (NULL != watchers)
 225     GNUNET_CONTAINER_multihashmap_iterate (watchers,
 226                                            &client_disconnect_it,
 227                                            client);
 228   num_clients--;
 229   if ( (0 == num_clients) &&
 230        in_shutdown)
 231     do_shutdown ();
 232 }
 233
 234
 235 /**
 236  * Function called by for each matching record.
 237  *
 238  * @param cls closure
 239  * @param record peerstore record found
 240  * @param emsg error message or NULL if no errors
 241  * @return #GNUNET_YES to continue iteration
 242  */
 243 static void
 244 record_iterator (void *cls,
 245                  const struct GNUNET_PEERSTORE_Record *record,
 246                  const char *emsg)
 247 {
 248   struct GNUNET_PEERSTORE_Record *cls_record = cls;
 249   struct GNUNET_MQ_Envelope *env;
 250
 251   if (NULL == record)
 252   {
 253     /* No more records */
 254     struct GNUNET_MessageHeader *endmsg;
 255
 256     env = GNUNET_MQ_msg (endmsg,
 257                          GNUNET_MESSAGE_TYPE_PEERSTORE_ITERATE_END);
 258     GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (cls_record->client),
 259                     env);
 260     if (NULL == emsg)
 261     {
 262       GNUNET_SERVICE_client_continue (cls_record->client);
 263     }
 264     else
 265     {
 266       GNUNET_break (0);
 267       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 268                   "Failed to iterate: %s\n",
 269                   emsg);
 270       GNUNET_SERVICE_client_drop (cls_record->client);
 271     }
 272     PEERSTORE_destroy_record (cls_record);
 273     return;
 274   }
 275
 276   env = PEERSTORE_create_record_mq_envelope (record->sub_system,
 277                                              &record->peer,
 278                                              record->key,
 279                                              record->value,
 280                                              record->value_size,
 281                                              record->expiry,
 282                                              0,
 283                                              GNUNET_MESSAGE_TYPE_PEERSTORE_ITERATE_RECORD);
 284   GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (cls_record->client),
 285                   env);
 286 }
 287
 288
 289 /**
 290  * Iterator over all watcher clients
 291  * to notify them of a new record
 292  *
 293  * @param cls closure, a `struct GNUNET_PEERSTORE_Record *`
 294  * @param key hash of record key
 295  * @param value the watcher client, a `struct GNUNET_SERVICE_Client *`
 296  * @return #GNUNET_YES to continue iterating
 297  */
 298 static int
 299 watch_notifier_it (void *cls,
 300                    const struct GNUNET_HashCode *key,
 301                    void *value)
 302 {
 303   struct GNUNET_PEERSTORE_Record *record = cls;
 304   struct GNUNET_SERVICE_Client *client = value;
 305   struct GNUNET_MQ_Envelope *env;
 306
 307   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 308               "Found a watcher to update.\n");
 309   env = PEERSTORE_create_record_mq_envelope (record->sub_system,
 310                                              &record->peer,
 311                                              record->key,
 312                                              record->value,
 313                                              record->value_size,
 314                                              record->expiry,
 315                                              0,
 316                                              GNUNET_MESSAGE_TYPE_PEERSTORE_WATCH_RECORD);
 317   GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (client),
 318                   env);
 319   return GNUNET_YES;
 320 }
 321
 322
 323 /**
 324  * Given a new record, notifies watchers
 325  *
 326  * @param record changed record to update watchers with
 327  */
 328 static void
 329 watch_notifier (struct GNUNET_PEERSTORE_Record *record)
 330 {
 331   struct GNUNET_HashCode keyhash;
 332
 333   PEERSTORE_hash_key (record->sub_system,
 334                       &record->peer,
 335                       record->key,
 336                       &keyhash);
 337   GNUNET_CONTAINER_multihashmap_get_multiple (watchers,
 338                                               &keyhash,
 339                                               &watch_notifier_it,
 340                                               record);
 341 }
 342
 343
 344 /**
 345  * Handle a watch cancel request from client
 346  *
 347  * @param cls identification of the client
 348  * @param hm the actual message
 349  */
 350 static void
 351 handle_watch_cancel (void *cls,
 352                      const struct StoreKeyHashMessage *hm)
 353 {
 354   struct GNUNET_SERVICE_Client *client = cls;
 355
 356   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 357               "Received a watch cancel request.\n");
 358   if (GNUNET_OK !=
 359       GNUNET_CONTAINER_multihashmap_remove (watchers,
 360                                             &hm->keyhash,
 361                                             client))
 362   {
 363     GNUNET_break (0);
 364     GNUNET_SERVICE_client_drop (client);
 365     return;
 366   }
 367   num_clients++;
 368   GNUNET_SERVICE_client_continue (client);
 369 }
 370
 371
 372 /**
 373  * Handle a watch request from client
 374  *
 375  * @param cls identification of the client
 376  * @param hm the actual message
 377  */
 378 static void
 379 handle_watch (void *cls,
 380               const struct StoreKeyHashMessage *hm)
 381 {
 382   struct GNUNET_SERVICE_Client *client = cls;
 383
 384   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 385               "Received a watch request.\n");
 386   num_clients--; /* do not count watchers */
 387   GNUNET_SERVICE_client_mark_monitor (client);
 388   GNUNET_CONTAINER_multihashmap_put (watchers,
 389                                      &hm->keyhash,
 390                                      client,
 391                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
 392   GNUNET_SERVICE_client_continue (client);
 393 }
 394
 395
 396 /**
 397  * Check an iterate request from client
 398  *
 399  * @param cls client identification of the client
 400  * @param srm the actual message
 401  * @return #GNUNET_OK if @a srm is well-formed
 402  */
 403 static int
 404 check_iterate (void *cls,
 405                const struct StoreRecordMessage *srm)
 406 {
 407   struct GNUNET_PEERSTORE_Record *record;
 408
 409   record = PEERSTORE_parse_record_message (srm);
 410   if (NULL == record)
 411   {
 412     GNUNET_break (0);
 413     return GNUNET_SYSERR;
 414   }
 415   if (NULL == record->sub_system)
 416   {
 417     GNUNET_break (0);
 418     PEERSTORE_destroy_record (record);
 419     return GNUNET_SYSERR;
 420   }
 421   PEERSTORE_destroy_record (record);
 422   return GNUNET_OK;
 423 }
 424
 425
 426 /**
 427  * Handle an iterate request from client
 428  *
 429  * @param cls identification of the client
 430  * @param srm the actual message
 431  */
 432 static void
 433 handle_iterate (void *cls,
 434                 const struct StoreRecordMessage *srm)
 435 {
 436   struct GNUNET_SERVICE_Client *client = cls;
 437   struct GNUNET_PEERSTORE_Record *record;
 438
 439   record = PEERSTORE_parse_record_message (srm);
 440   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 441               "Iterate request: ss `%s', peer `%s', key `%s'\n",
 442               record->sub_system,
 443               GNUNET_i2s (&record->peer),
 444               (NULL == record->key) ? "NULL" : record->key);
 445   record->client = client;
 446   if (GNUNET_OK !=
 447       db->iterate_records (db->cls,
 448                            record->sub_system,
 449                            (ntohs (srm->peer_set)) ? &record->peer : NULL,
 450                            record->key,
 451                            &record_iterator,
 452                            record))
 453   {
 454     GNUNET_break (0);
 455     GNUNET_SERVICE_client_drop (client);
 456     PEERSTORE_destroy_record (record);
 457   }
 458 }
 459
 460
 461 /**
 462  * Continuation of store_record called by the peerstore plugin
 463  *
 464  * @param cls closure
 465  * @param success result
 466  */
 467 static void
 468 store_record_continuation (void *cls,
 469                            int success)
 470 {
 471   struct GNUNET_PEERSTORE_Record *record = cls;
 472
 473   if (GNUNET_OK == success)
 474   {
 475     watch_notifier (record);
 476     GNUNET_SERVICE_client_continue (record->client);
 477   }
 478   else
 479   {
 480     GNUNET_break (0);
 481     GNUNET_SERVICE_client_drop (record->client);
 482   }
 483   PEERSTORE_destroy_record (record);
 484 }
 485
 486
 487 /**
 488  * Check a store request from client
 489  *
 490  * @param cls client identification of the client
 491  * @param srm the actual message
 492  * @return #GNUNET_OK if @a srm is well-formed
 493  */
 494 static int
 495 check_store (void *cls,
 496               const struct StoreRecordMessage *srm)
 497 {
 498   struct GNUNET_PEERSTORE_Record *record;
 499
 500   record = PEERSTORE_parse_record_message (srm);
 501   if (NULL == record)
 502   {
 503     GNUNET_break (0);
 504     return GNUNET_SYSERR;
 505   }
 506   if ( (NULL == record->sub_system) ||
 507        (NULL == record->key) )
 508   {
 509     GNUNET_break (0);
 510     PEERSTORE_destroy_record (record);
 511     return GNUNET_SYSERR;
 512   }
 513   PEERSTORE_destroy_record (record);
 514   return GNUNET_OK;
 515 }
 516
 517
 518 /**
 519  * Handle a store request from client
 520  *
 521  * @param cls client identification of the client
 522  * @param srm the actual message
 523  */
 524 static void
 525 handle_store (void *cls,
 526               const struct StoreRecordMessage *srm)
 527 {
 528   struct GNUNET_SERVICE_Client *client = cls;
 529   struct GNUNET_PEERSTORE_Record *record;
 530
 531   record = PEERSTORE_parse_record_message (srm);
 532   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 533               "Received a store request. Sub system `%s' Peer `%s Key `%s' Options: %u.\n",
 534               record->sub_system,
 535               GNUNET_i2s (&record->peer),
 536               record->key,
 537               (uint32_t) ntohl (srm->options));
 538   record->client = client;
 539   if (GNUNET_OK !=
 540       db->store_record (db->cls,
 541                         record->sub_system,
 542                         &record->peer,
 543                         record->key,
 544                         record->value,
 545                         record->value_size,
 546                         record->expiry,
 547                         ntohl (srm->options),
 548                         &store_record_continuation,
 549                         record))
 550   {
 551     GNUNET_break (0);
 552     PEERSTORE_destroy_record (record);
 553     GNUNET_SERVICE_client_drop (client);
 554     return;
 555   }
 556 }
 557
 558
 559 /**
 560  * Peerstore service runner.
 561  *
 562  * @param cls closure
 563  * @param c configuration to use
 564  * @param service the initialized service
 565  */
 566 static void
 567 run (void *cls,
 568      const struct GNUNET_CONFIGURATION_Handle *c,
 569      struct GNUNET_SERVICE_Handle *service)
 570 {
 571   char *database;
 572
 573   in_shutdown = GNUNET_NO;
 574   cfg = c;
 575   if (GNUNET_OK !=
 576       GNUNET_CONFIGURATION_get_value_string (cfg,
 577                                              "peerstore",
 578                                              "DATABASE",
 579                                              &database))
 580   {
 581     GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
 582                                "peerstore",
 583                                "DATABASE");
 584     GNUNET_SCHEDULER_shutdown ();
 585     return;
 586   }
 587   GNUNET_asprintf (&db_lib_name,
 588                    "libgnunet_plugin_peerstore_%s",
 589                    database);
 590   db = GNUNET_PLUGIN_load (db_lib_name,
 591                            (void *) cfg);
 592   GNUNET_free (database);
 593   if (NULL == db)
 594   {
 595     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 596                 _("Could not load database backend `%s'\n"),
 597                 db_lib_name);
 598     GNUNET_SCHEDULER_shutdown ();
 599     return;
 600   }
 601   watchers = GNUNET_CONTAINER_multihashmap_create (10,
 602                                                    GNUNET_NO);
 603   expire_task = GNUNET_SCHEDULER_add_now (&cleanup_expired_records,
 604                                           NULL);
 605   GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
 606                                  NULL);
 607 }
 608
 609
 610 /**
 611  * Define "main" method using service macro.
 612  */
 613 GNUNET_SERVICE_MAIN
 614 ("peerstore",
 615  GNUNET_SERVICE_OPTION_SOFT_SHUTDOWN,
 616  &run,
 617  &client_connect_cb,
 618  &client_disconnect_cb,
 619  NULL,
 620  GNUNET_MQ_hd_var_size (store,
 621                         GNUNET_MESSAGE_TYPE_PEERSTORE_STORE,
 622                         struct StoreRecordMessage,
 623                         NULL),
 624  GNUNET_MQ_hd_var_size (iterate,
 625                         GNUNET_MESSAGE_TYPE_PEERSTORE_ITERATE,
 626                         struct StoreRecordMessage,
 627                         NULL),
 628  GNUNET_MQ_hd_fixed_size (watch,
 629                           GNUNET_MESSAGE_TYPE_PEERSTORE_WATCH,
 630                           struct StoreKeyHashMessage,
 631                           NULL),
 632  GNUNET_MQ_hd_fixed_size (watch_cancel,
 633                           GNUNET_MESSAGE_TYPE_PEERSTORE_WATCH_CANCEL,
 634                           struct StoreKeyHashMessage,
 635                           NULL),
 636  GNUNET_MQ_handler_end ());
 637
 638
 639 /* end of gnunet-service-peerstore.c */