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