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