src/psycstore/gnunet-service-psycstore.c

   1 /**
   2  * This file is part of GNUnet
   3  * Copyright (C) 2013 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
  16 /**
  17  * @file psycstore/gnunet-service-psycstore.c
  18  * @brief PSYCstore service
  19  * @author Gabor X Toth
  20  * @author Christian Grothoff
  21  */
  22
  23 #include <inttypes.h>
  24
  25 #include "platform.h"
  26 #include "gnunet_util_lib.h"
  27 #include "gnunet_constants.h"
  28 #include "gnunet_protocols.h"
  29 #include "gnunet_statistics_service.h"
  30 #include "gnunet_psyc_util_lib.h"
  31 #include "gnunet_psycstore_service.h"
  32 #include "gnunet_psycstore_plugin.h"
  33 #include "psycstore.h"
  34
  35
  36 /**
  37  * Handle to our current configuration.
  38  */
  39 static const struct GNUNET_CONFIGURATION_Handle *cfg;
  40
  41 /**
  42  * Service handle.
  43  */
  44 static struct GNUNET_SERVICE_Handle *service;
  45
  46 /**
  47  * Handle to the statistics service.
  48  */
  49 static struct GNUNET_STATISTICS_Handle *stats;
  50
  51 /**
  52  * Database handle
  53  */
  54 static struct GNUNET_PSYCSTORE_PluginFunctions *db;
  55
  56 /**
  57  * Name of the database plugin
  58  */
  59 static char *db_lib_name;
  60
  61
  62 /**
  63  * Task run during shutdown.
  64  *
  65  * @param cls unused
  66  */
  67 static void
  68 shutdown_task (void *cls)
  69 {
  70   if (NULL != stats)
  71   {
  72     GNUNET_STATISTICS_destroy (stats, GNUNET_NO);
  73     stats = NULL;
  74   }
  75   GNUNET_break (NULL == GNUNET_PLUGIN_unload (db_lib_name, db));
  76   GNUNET_free (db_lib_name);
  77   db_lib_name = NULL;
  78 }
  79
  80
  81 /**
  82  * Send a result code back to the client.
  83  *
  84  * @param client
  85  *        Client that should receive the result code.
  86  * @param result_code
  87  *        Code to transmit.
  88  * @param op_id
  89  *        Operation ID in network byte order.
  90  * @param err_msg
  91  *        Error message to include (or NULL for none).
  92  */
  93 static void
  94 send_result_code (struct GNUNET_SERVICE_Client *client,
  95                   uint64_t op_id,
  96                   int64_t result_code,
  97                   const char *err_msg)
  98 {
  99   struct OperationResult *res;
 100   size_t err_size = 0;
 101
 102   if (NULL != err_msg)
 103     err_size = strnlen (err_msg,
 104                         GNUNET_MAX_MESSAGE_SIZE - sizeof (*res) - 1) + 1;
 105   struct GNUNET_MQ_Envelope *
 106     env = GNUNET_MQ_msg_extra (res, err_size,
 107                                GNUNET_MESSAGE_TYPE_PSYCSTORE_RESULT_CODE);
 108   res->result_code = GNUNET_htonll (result_code - INT64_MIN);
 109   res->op_id = op_id;
 110   if (0 < err_size)
 111   {
 112     GNUNET_memcpy (&res[1], err_msg, err_size);
 113     ((char *) &res[1])[err_size - 1] = '\0';
 114   }
 115
 116   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 117               "Sending result to client: %" PRId64 " (%s)\n",
 118               result_code, err_msg);
 119   GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (client), env);
 120 }
 121
 122
 123 enum
 124 {
 125   MEMBERSHIP_TEST_NOT_NEEDED = 0,
 126   MEMBERSHIP_TEST_NEEDED = 1,
 127   MEMBERSHIP_TEST_DONE = 2,
 128 } MessageMembershipTest;
 129
 130
 131 struct SendClosure
 132 {
 133   struct GNUNET_SERVICE_Client *client;
 134
 135   /**
 136    * Channel's public key.
 137    */
 138   struct GNUNET_CRYPTO_EddsaPublicKey channel_key;
 139
 140   /**
 141    * Slave's public key.
 142    */
 143   struct GNUNET_CRYPTO_EcdsaPublicKey slave_key;
 144
 145   /**
 146    * Operation ID.
 147    */
 148   uint64_t op_id;
 149
 150   /**
 151    * Membership test result.
 152    */
 153   int membership_test_result;
 154
 155   /**
 156    * Do membership test with @a slave_key before returning fragment?
 157    * @see enum MessageMembershipTest
 158    */
 159   uint8_t membership_test;
 160 };
 161
 162
 163 static int
 164 send_fragment (void *cls, struct GNUNET_MULTICAST_MessageHeader *msg,
 165                enum GNUNET_PSYCSTORE_MessageFlags flags)
 166 {
 167   struct SendClosure *sc = cls;
 168   struct FragmentResult *res;
 169
 170   if (MEMBERSHIP_TEST_NEEDED == sc->membership_test)
 171   {
 172     sc->membership_test = MEMBERSHIP_TEST_DONE;
 173     sc->membership_test_result
 174       = db->membership_test (db->cls, &sc->channel_key, &sc->slave_key,
 175                              GNUNET_ntohll (msg->message_id));
 176     switch (sc->membership_test_result)
 177     {
 178     case GNUNET_YES:
 179       break;
 180
 181     case GNUNET_NO:
 182     case GNUNET_SYSERR:
 183       return GNUNET_NO;
 184     }
 185   }
 186
 187   size_t msg_size = ntohs (msg->header.size);
 188
 189   struct GNUNET_MQ_Envelope *
 190     env = GNUNET_MQ_msg_extra (res, msg_size,
 191                                GNUNET_MESSAGE_TYPE_PSYCSTORE_RESULT_FRAGMENT);
 192   res->op_id = sc->op_id;
 193   res->psycstore_flags = htonl (flags);
 194   GNUNET_memcpy (&res[1], msg, msg_size);
 195   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 196               "Sending fragment %llu to client\n",
 197               (unsigned long long) GNUNET_ntohll (msg->fragment_id));
 198   GNUNET_free (msg);
 199
 200   GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (sc->client), env);
 201   return GNUNET_YES;
 202 }
 203
 204
 205 static int
 206 send_state_var (void *cls, const char *name,
 207                 const void *value, uint32_t value_size)
 208 {
 209   struct SendClosure *sc = cls;
 210   struct StateResult *res;
 211   size_t name_size = strlen (name) + 1;
 212
 213   /** @todo FIXME: split up value into 64k chunks */
 214
 215   struct GNUNET_MQ_Envelope *
 216     env = GNUNET_MQ_msg_extra (res, name_size + value_size,
 217                                GNUNET_MESSAGE_TYPE_PSYCSTORE_RESULT_STATE);
 218   res->op_id = sc->op_id;
 219   res->name_size = htons (name_size);
 220   GNUNET_memcpy (&res[1], name, name_size);
 221   GNUNET_memcpy ((char *) &res[1] + name_size, value, value_size);
 222   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 223               "Sending state variable %s to client\n", name);
 224
 225   GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (sc->client), env);
 226   return GNUNET_OK;
 227 }
 228
 229
 230 static void
 231 handle_client_membership_store (void *cls,
 232                                 const struct MembershipStoreRequest *req)
 233 {
 234   struct GNUNET_SERVICE_Client *client = cls;
 235
 236   int ret = db->membership_store (db->cls, &req->channel_key, &req->slave_key,
 237                                   req->did_join,
 238                                   GNUNET_ntohll (req->announced_at),
 239                                   GNUNET_ntohll (req->effective_since),
 240                                   GNUNET_ntohll (req->group_generation));
 241
 242   if (ret != GNUNET_OK)
 243     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 244                 _("Failed to store membership information!\n"));
 245
 246   send_result_code (client, req->op_id, ret, NULL);
 247   GNUNET_SERVICE_client_continue (client);
 248 }
 249
 250
 251 static void
 252 handle_client_membership_test (void *cls,
 253                                const struct MembershipTestRequest *req)
 254 {
 255   struct GNUNET_SERVICE_Client *client = cls;
 256
 257   int ret = db->membership_test (db->cls, &req->channel_key, &req->slave_key,
 258                                  GNUNET_ntohll (req->message_id));
 259   switch (ret)
 260   {
 261   case GNUNET_YES:
 262   case GNUNET_NO:
 263     break;
 264   default:
 265     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 266                 _("Failed to test membership!\n"));
 267   }
 268
 269   send_result_code (client, req->op_id, ret, NULL);
 270   GNUNET_SERVICE_client_continue (client);
 271 }
 272
 273
 274 static int
 275 check_client_fragment_store (void *cls,
 276                              const struct FragmentStoreRequest *req)
 277 {
 278   return GNUNET_OK;
 279 }
 280
 281
 282 static void
 283 handle_client_fragment_store (void *cls,
 284                               const struct FragmentStoreRequest *req)
 285 {
 286   struct GNUNET_SERVICE_Client *client = cls;
 287
 288   const struct GNUNET_MessageHeader *
 289     msg = GNUNET_MQ_extract_nested_mh (req);
 290   if (NULL == msg
 291       || ntohs (msg->size) < sizeof (struct GNUNET_MULTICAST_MessageHeader))
 292   {
 293     GNUNET_break (0);
 294     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 295                 _("Dropping invalid fragment\n"));
 296     GNUNET_SERVICE_client_drop (client);
 297     return;
 298   }
 299
 300   int ret = db->fragment_store (db->cls, &req->channel_key,
 301                                 (const struct GNUNET_MULTICAST_MessageHeader *)
 302                                 msg, ntohl (req->psycstore_flags));
 303
 304   if (ret != GNUNET_OK)
 305     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 306                 _("Failed to store fragment\n"));
 307
 308   send_result_code (client, req->op_id, ret, NULL);
 309   GNUNET_SERVICE_client_continue (client);
 310 }
 311
 312
 313 static void
 314 handle_client_fragment_get (void *cls,
 315                             const struct FragmentGetRequest *req)
 316 {
 317   struct GNUNET_SERVICE_Client *client = cls;
 318
 319   struct SendClosure
 320     sc = { .op_id = req->op_id,
 321            .client = client,
 322            .channel_key = req->channel_key,
 323            .slave_key = req->slave_key,
 324            .membership_test = req->do_membership_test };
 325
 326   int64_t ret;
 327   uint64_t ret_frags = 0;
 328   uint64_t first_fragment_id = GNUNET_ntohll (req->first_fragment_id);
 329   uint64_t last_fragment_id = GNUNET_ntohll (req->last_fragment_id);
 330   uint64_t limit = GNUNET_ntohll (req->fragment_limit);
 331
 332   if (0 == limit)
 333     ret = db->fragment_get (db->cls, &req->channel_key,
 334                             first_fragment_id, last_fragment_id,
 335                             &ret_frags, send_fragment, &sc);
 336   else
 337     ret = db->fragment_get_latest (db->cls, &req->channel_key, limit,
 338                                    &ret_frags, send_fragment, &sc);
 339
 340   switch (ret)
 341   {
 342   case GNUNET_YES:
 343   case GNUNET_NO:
 344     if (MEMBERSHIP_TEST_DONE == sc.membership_test)
 345     {
 346       switch (sc.membership_test_result)
 347       {
 348       case GNUNET_YES:
 349         break;
 350
 351       case GNUNET_NO:
 352         ret = GNUNET_PSYCSTORE_MEMBERSHIP_TEST_FAILED;
 353         break;
 354
 355       case GNUNET_SYSERR:
 356         ret = GNUNET_SYSERR;
 357         break;
 358       }
 359     }
 360     break;
 361   default:
 362     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 363                 _("Failed to get fragment!\n"));
 364   }
 365   send_result_code (client, req->op_id, (ret < 0) ? ret : ret_frags, NULL);
 366   GNUNET_SERVICE_client_continue (client);
 367 }
 368
 369
 370 static int
 371 check_client_message_get (void *cls,
 372                           const struct MessageGetRequest *req)
 373 {
 374   return GNUNET_OK;
 375 }
 376
 377
 378 static void
 379 handle_client_message_get (void *cls,
 380                            const struct MessageGetRequest *req)
 381 {
 382   struct GNUNET_SERVICE_Client *client = cls;
 383
 384   uint16_t size = ntohs (req->header.size);
 385   const char *method_prefix = (const char *) &req[1];
 386
 387   if (size < sizeof (*req) + 1
 388       || '\0' != method_prefix[size - sizeof (*req) - 1])
 389   {
 390     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 391                 "Message get: invalid method prefix. size: %u < %u?\n",
 392                 size,
 393                 (unsigned int) (sizeof (*req) + 1));
 394     GNUNET_break (0);
 395     GNUNET_SERVICE_client_drop (client);
 396     return;
 397   }
 398
 399   struct SendClosure
 400     sc = { .op_id = req->op_id,
 401            .client = client,
 402            .channel_key = req->channel_key,
 403            .slave_key = req->slave_key,
 404            .membership_test = req->do_membership_test };
 405
 406   int64_t ret;
 407   uint64_t ret_frags = 0;
 408   uint64_t first_message_id = GNUNET_ntohll (req->first_message_id);
 409   uint64_t last_message_id = GNUNET_ntohll (req->last_message_id);
 410   uint64_t msg_limit = GNUNET_ntohll (req->message_limit);
 411   uint64_t frag_limit = GNUNET_ntohll (req->fragment_limit);
 412
 413   /** @todo method_prefix */
 414   if (0 == msg_limit)
 415     ret = db->message_get (db->cls, &req->channel_key,
 416                            first_message_id, last_message_id, frag_limit,
 417                            &ret_frags, send_fragment, &sc);
 418   else
 419     ret = db->message_get_latest (db->cls, &req->channel_key, msg_limit,
 420                                   &ret_frags, send_fragment, &sc);
 421
 422   switch (ret)
 423   {
 424   case GNUNET_YES:
 425   case GNUNET_NO:
 426     break;
 427   default:
 428     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 429                 _("Failed to get message!\n"));
 430   }
 431
 432   send_result_code (client, req->op_id, (ret < 0) ? ret : ret_frags, NULL);
 433   GNUNET_SERVICE_client_continue (client);
 434 }
 435
 436
 437 static void
 438 handle_client_message_get_fragment (void *cls,
 439                                     const struct MessageGetFragmentRequest *req)
 440 {
 441   struct GNUNET_SERVICE_Client *client = cls;
 442
 443   struct SendClosure
 444     sc = { .op_id = req->op_id, .client = client,
 445            .channel_key = req->channel_key, .slave_key = req->slave_key,
 446            .membership_test = req->do_membership_test };
 447
 448   int ret = db->message_get_fragment (db->cls, &req->channel_key,
 449                                       GNUNET_ntohll (req->message_id),
 450                                       GNUNET_ntohll (req->fragment_offset),
 451                                       &send_fragment, &sc);
 452   switch (ret)
 453   {
 454   case GNUNET_YES:
 455   case GNUNET_NO:
 456     break;
 457   default:
 458     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 459                 _("Failed to get message fragment!\n"));
 460   }
 461
 462   send_result_code (client, req->op_id, ret, NULL);
 463   GNUNET_SERVICE_client_continue (client);
 464 }
 465
 466
 467 static void
 468 handle_client_counters_get (void *cls,
 469                             const struct OperationRequest *req)
 470 {
 471   struct GNUNET_SERVICE_Client *client = cls;
 472
 473   struct CountersResult *res;
 474   struct GNUNET_MQ_Envelope *
 475     env = GNUNET_MQ_msg (res, GNUNET_MESSAGE_TYPE_PSYCSTORE_RESULT_COUNTERS);
 476
 477   int ret = db->counters_message_get (db->cls, &req->channel_key,
 478                                       &res->max_fragment_id, &res->max_message_id,
 479                                       &res->max_group_generation);
 480   switch (ret)
 481   {
 482   case GNUNET_OK:
 483     ret = db->counters_state_get (db->cls, &req->channel_key,
 484                                   &res->max_state_message_id);
 485   case GNUNET_NO:
 486     break;
 487   default:
 488     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 489                 _("Failed to get master counters!\n"));
 490   }
 491
 492   res->result_code = htonl (ret);
 493   res->op_id = req->op_id;
 494   res->max_fragment_id = GNUNET_htonll (res->max_fragment_id);
 495   res->max_message_id = GNUNET_htonll (res->max_message_id);
 496   res->max_group_generation = GNUNET_htonll (res->max_group_generation);
 497   res->max_state_message_id = GNUNET_htonll (res->max_state_message_id);
 498
 499   GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (client), env);
 500   GNUNET_SERVICE_client_continue (client);
 501 }
 502
 503
 504 struct StateModifyClosure
 505 {
 506   const struct GNUNET_CRYPTO_EddsaPublicKey channel_key;
 507   struct GNUNET_PSYC_ReceiveHandle *recv;
 508   enum GNUNET_PSYC_MessageState msg_state;
 509   char mod_oper;
 510   char *mod_name;
 511   char *mod_value;
 512   uint32_t mod_value_size;
 513   uint32_t mod_value_remaining;
 514 };
 515
 516
 517 static void
 518 recv_state_message_part (void *cls,
 519                          const struct GNUNET_PSYC_MessageHeader *msg,
 520                          const struct GNUNET_MessageHeader *pmsg)
 521 {
 522   struct StateModifyClosure *scls = cls;
 523   uint16_t psize;
 524
 525   if (NULL == msg)
 526   { // FIXME: error on unknown message
 527     return;
 528   }
 529
 530   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 531               "recv_state_message_part()  message_id: %" PRIu64
 532               ", fragment_offset: %" PRIu64 ", flags: %u\n",
 533               GNUNET_ntohll (msg->message_id),
 534               GNUNET_ntohll (msg->fragment_offset),
 535               ntohl (msg->flags));
 536
 537   if (NULL == pmsg)
 538   {
 539     scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_ERROR;
 540     return;
 541   }
 542
 543   switch (ntohs (pmsg->type))
 544   {
 545   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD:
 546   {
 547     scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_METHOD;
 548     break;
 549   }
 550
 551   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER:
 552   {
 553     struct GNUNET_PSYC_MessageModifier *
 554       pmod = (struct GNUNET_PSYC_MessageModifier *) pmsg;
 555     psize = ntohs (pmod->header.size);
 556     uint16_t name_size = ntohs (pmod->name_size);
 557     uint32_t value_size = ntohl (pmod->value_size);
 558
 559     const char *name = (const char *) &pmod[1];
 560     const void *value = name + name_size;
 561
 562     if (GNUNET_PSYC_OP_SET != pmod->oper)
 563     { // Apply non-transient operation.
 564       if (psize == sizeof (*pmod) + name_size + value_size)
 565       {
 566         db->state_modify_op (db->cls, &scls->channel_key,
 567                              pmod->oper, name, value, value_size);
 568       }
 569       else
 570       {
 571         scls->mod_oper = pmod->oper;
 572         scls->mod_name = GNUNET_malloc (name_size);
 573         GNUNET_memcpy (scls->mod_name, name, name_size);
 574
 575         scls->mod_value_size = value_size;
 576         scls->mod_value = GNUNET_malloc (scls->mod_value_size);
 577         scls->mod_value_remaining
 578           = scls->mod_value_size - (psize - sizeof (*pmod) - name_size);
 579         GNUNET_memcpy (scls->mod_value, value, value_size - scls->mod_value_remaining);
 580       }
 581     }
 582     scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_MODIFIER;
 583     break;
 584   }
 585
 586   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT:
 587     if (GNUNET_PSYC_OP_SET != scls->mod_oper)
 588     {
 589       if (scls->mod_value_remaining == 0)
 590       {
 591         GNUNET_break_op (0);
 592         scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_ERROR;
 593       }
 594       psize = ntohs (pmsg->size);
 595       GNUNET_memcpy (scls->mod_value + (scls->mod_value_size - scls->mod_value_remaining),
 596               &pmsg[1], psize - sizeof (*pmsg));
 597       scls->mod_value_remaining -= psize - sizeof (*pmsg);
 598       if (0 == scls->mod_value_remaining)
 599       {
 600         db->state_modify_op (db->cls, &scls->channel_key,
 601                              scls->mod_oper, scls->mod_name,
 602                              scls->mod_value, scls->mod_value_size);
 603         GNUNET_free (scls->mod_name);
 604         GNUNET_free (scls->mod_value);
 605         scls->mod_oper = 0;
 606         scls->mod_name = NULL;
 607         scls->mod_value = NULL;
 608         scls->mod_value_size = 0;
 609       }
 610     }
 611     scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_MOD_CONT;
 612     break;
 613
 614   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA:
 615     scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_DATA;
 616     break;
 617
 618   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END:
 619     scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_END;
 620     break;
 621
 622   default:
 623     scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_ERROR;
 624   }
 625 }
 626
 627
 628 static int
 629 recv_state_fragment (void *cls, struct GNUNET_MULTICAST_MessageHeader *msg,
 630                      enum GNUNET_PSYCSTORE_MessageFlags flags)
 631 {
 632   struct StateModifyClosure *scls = cls;
 633
 634   if (NULL == scls->recv)
 635   {
 636     scls->recv = GNUNET_PSYC_receive_create (NULL, recv_state_message_part,
 637                                              scls);
 638   }
 639
 640   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 641               "recv_state_fragment: %" PRIu64 "\n", GNUNET_ntohll (msg->fragment_id));
 642
 643   struct GNUNET_PSYC_MessageHeader *
 644     pmsg = GNUNET_PSYC_message_header_create (msg, flags);
 645   GNUNET_PSYC_receive_message (scls->recv, pmsg);
 646   GNUNET_free (pmsg);
 647
 648   return GNUNET_YES;
 649 }
 650
 651
 652 static void
 653 handle_client_state_modify (void *cls,
 654                             const struct StateModifyRequest *req)
 655 {
 656   struct GNUNET_SERVICE_Client *client = cls;
 657
 658   uint64_t message_id = GNUNET_ntohll (req->message_id);
 659   uint64_t state_delta = GNUNET_ntohll (req->state_delta);
 660   uint64_t ret_frags = 0;
 661   struct StateModifyClosure
 662     scls = { .channel_key = req->channel_key };
 663
 664   int ret = db->state_modify_begin (db->cls, &req->channel_key,
 665                                     message_id, state_delta);
 666
 667   if (GNUNET_OK != ret)
 668   {
 669     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 670                 _("Failed to begin modifying state: %d\n"), ret);
 671   }
 672   else
 673   {
 674     ret = db->message_get (db->cls, &req->channel_key,
 675                            message_id, message_id, 0,
 676                            &ret_frags, recv_state_fragment, &scls);
 677     if (GNUNET_OK != ret)
 678     {
 679       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 680                   _("Failed to modify state: %d\n"), ret);
 681       GNUNET_break (0);
 682     }
 683     else
 684     {
 685       if (GNUNET_OK != db->state_modify_end (db->cls, &req->channel_key, message_id))
 686       {
 687         GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 688                     _("Failed to end modifying state!\n"));
 689         GNUNET_break (0);
 690       }
 691     }
 692     if (NULL != scls.recv)
 693     {
 694       GNUNET_PSYC_receive_destroy (scls.recv);
 695     }
 696   }
 697
 698   send_result_code (client, req->op_id, ret, NULL);
 699   GNUNET_SERVICE_client_continue (client);
 700 }
 701
 702
 703 static int
 704 check_client_state_sync (void *cls,
 705                          const struct StateSyncRequest *req)
 706 {
 707   return GNUNET_OK;
 708 }
 709
 710
 711 /** @todo FIXME: stop processing further state sync messages after an error */
 712 static void
 713 handle_client_state_sync (void *cls,
 714                           const struct StateSyncRequest *req)
 715 {
 716   struct GNUNET_SERVICE_Client *client = cls;
 717
 718   int ret = GNUNET_SYSERR;
 719   const char *name = (const char *) &req[1];
 720   uint16_t name_size = ntohs (req->name_size);
 721
 722   if (name_size <= 2 || '\0' != name[name_size - 1])
 723   {
 724     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 725                 _("Tried to set invalid state variable name!\n"));
 726     GNUNET_break_op (0);
 727   }
 728   else
 729   {
 730     ret = GNUNET_OK;
 731
 732     if (req->flags & STATE_OP_FIRST)
 733     {
 734       ret = db->state_sync_begin (db->cls, &req->channel_key);
 735     }
 736     if (ret != GNUNET_OK)
 737     {
 738       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 739                   _("Failed to begin synchronizing state!\n"));
 740     }
 741     else
 742     {
 743       ret = db->state_sync_assign (db->cls, &req->channel_key, name,
 744                                    name + ntohs (req->name_size),
 745                                    ntohs (req->header.size) - sizeof (*req)
 746                                    - ntohs (req->name_size));
 747     }
 748
 749     if (GNUNET_OK == ret && req->flags & STATE_OP_LAST)
 750     {
 751       ret = db->state_sync_end (db->cls, &req->channel_key,
 752                                 GNUNET_ntohll (req->max_state_message_id),
 753                                 GNUNET_ntohll (req->state_hash_message_id));
 754       if (ret != GNUNET_OK)
 755         GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 756                     _("Failed to end synchronizing state!\n"));
 757     }
 758   }
 759   send_result_code (client, req->op_id, ret, NULL);
 760   GNUNET_SERVICE_client_continue (client);
 761 }
 762
 763
 764 static void
 765 handle_client_state_reset (void *cls,
 766                            const struct OperationRequest *req)
 767 {
 768   struct GNUNET_SERVICE_Client *client = cls;
 769
 770   int ret = db->state_reset (db->cls, &req->channel_key);
 771
 772   if (ret != GNUNET_OK)
 773     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 774                 _("Failed to reset state!\n"));
 775
 776   send_result_code (client, req->op_id, ret, NULL);
 777   GNUNET_SERVICE_client_continue (client);
 778 }
 779
 780
 781 static void
 782 handle_client_state_hash_update (void *cls,
 783                                  const struct StateHashUpdateRequest *req)
 784 {
 785   struct GNUNET_SERVICE_Client *client = cls;
 786
 787   int ret = db->state_reset (db->cls, &req->channel_key);
 788   if (ret != GNUNET_OK)
 789     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 790                 _("Failed to reset state!\n"));
 791
 792   send_result_code (client, req->op_id, ret, NULL);
 793   GNUNET_SERVICE_client_continue (client);
 794 }
 795
 796
 797 static int
 798 check_client_state_get (void *cls,
 799                         const struct OperationRequest *req)
 800 {
 801   return GNUNET_OK;
 802 }
 803
 804
 805 static void
 806 handle_client_state_get (void *cls,
 807                          const struct OperationRequest *req)
 808 {
 809   struct GNUNET_SERVICE_Client *client = cls;
 810
 811   struct SendClosure sc = { .op_id = req->op_id, .client = client };
 812   int64_t ret = GNUNET_SYSERR;
 813   const char *name = (const char *) &req[1];
 814   uint16_t name_size = ntohs (req->header.size) - sizeof (*req);
 815
 816   if (name_size <= 2 || '\0' != name[name_size - 1])
 817   {
 818     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 819                 _("Tried to get invalid state variable name!\n"));
 820     GNUNET_break (0);
 821   }
 822   else
 823   {
 824     ret = db->state_get (db->cls, &req->channel_key, name,
 825                          &send_state_var, &sc);
 826     if (GNUNET_NO == ret && name_size >= 5) /* min: _a_b\0 */
 827     {
 828       char *p, *n = GNUNET_malloc (name_size);
 829       GNUNET_memcpy (n, name, name_size);
 830       while (&n[1] < (p = strrchr (n, '_')) && GNUNET_NO == ret)
 831       {
 832         *p = '\0';
 833         ret = db->state_get (db->cls, &req->channel_key, n,
 834                              &send_state_var, &sc);
 835       }
 836       GNUNET_free (n);
 837     }
 838   }
 839   switch (ret)
 840   {
 841   case GNUNET_OK:
 842   case GNUNET_NO:
 843     break;
 844   default:
 845     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 846                 _("Failed to get state variable!\n"));
 847   }
 848
 849   send_result_code (client, req->op_id, ret, NULL);
 850   GNUNET_SERVICE_client_continue (client);
 851 }
 852
 853
 854 static int
 855 check_client_state_get_prefix (void *cls,
 856                                const struct OperationRequest *req)
 857 {
 858   return GNUNET_OK;
 859 }
 860
 861
 862 static void
 863 handle_client_state_get_prefix (void *cls,
 864                                 const struct OperationRequest *req)
 865 {
 866   struct GNUNET_SERVICE_Client *client = cls;
 867
 868   struct SendClosure sc = { .op_id = req->op_id, .client = client };
 869   int64_t ret = GNUNET_SYSERR;
 870   const char *name = (const char *) &req[1];
 871   uint16_t name_size = ntohs (req->header.size) - sizeof (*req);
 872
 873   if (name_size <= 1 || '\0' != name[name_size - 1])
 874   {
 875     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 876                 _("Tried to get invalid state variable name!\n"));
 877     GNUNET_break (0);
 878   }
 879   else
 880   {
 881     ret = db->state_get_prefix (db->cls, &req->channel_key, name,
 882                                 &send_state_var, &sc);
 883   }
 884   switch (ret)
 885   {
 886   case GNUNET_OK:
 887   case GNUNET_NO:
 888     break;
 889   default:
 890     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 891                 _("Failed to get state variable!\n"));
 892   }
 893
 894   send_result_code (client, req->op_id, ret, NULL);
 895   GNUNET_SERVICE_client_continue (client);
 896 }
 897
 898
 899 /**
 900  * A new client connected.
 901  *
 902  * @param cls NULL
 903  * @param client client to add
 904  * @param mq message queue for @a client
 905  * @return @a client
 906  */
 907 static void *
 908 client_notify_connect (void *cls,
 909                        struct GNUNET_SERVICE_Client *client,
 910                        struct GNUNET_MQ_Handle *mq)
 911 {
 912   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Client connected: %p\n", client);
 913
 914   return client;
 915 }
 916
 917
 918 /**
 919  * Called whenever a client is disconnected.
 920  * Frees our resources associated with that client.
 921  *
 922  * @param cls closure
 923  * @param client identification of the client
 924  * @param app_ctx must match @a client
 925  */
 926 static void
 927 client_notify_disconnect (void *cls,
 928                           struct GNUNET_SERVICE_Client *client,
 929                           void *app_ctx)
 930 {
 931 }
 932
 933
 934 /**
 935  * Initialize the PSYCstore service.
 936  *
 937  * @param cls Closure.
 938  * @param server The initialized server.
 939  * @param c Configuration to use.
 940  */
 941 static void
 942 run (void *cls,
 943      const struct GNUNET_CONFIGURATION_Handle *c,
 944      struct GNUNET_SERVICE_Handle *svc)
 945 {
 946   cfg = c;
 947   service = svc;
 948
 949   /* Loading database plugin */
 950   char *database;
 951   if (GNUNET_OK !=
 952       GNUNET_CONFIGURATION_get_value_string (cfg, "psycstore", "database",
 953                                              &database))
 954   {
 955     GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
 956                                "psycstore",
 957                                "database");
 958   }
 959   else
 960   {
 961     GNUNET_asprintf (&db_lib_name,
 962                      "libgnunet_plugin_psycstore_%s",
 963                      database);
 964     db = GNUNET_PLUGIN_load (db_lib_name, (void *) cfg);
 965     GNUNET_free (database);
 966   }
 967   if (NULL == db)
 968   {
 969     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 970                 "Could not load database backend `%s'\n",
 971                 db_lib_name);
 972     GNUNET_SCHEDULER_add_now (&shutdown_task, NULL);
 973     return;
 974   }
 975
 976   stats = GNUNET_STATISTICS_create ("psycstore", cfg);
 977   GNUNET_SCHEDULER_add_shutdown (shutdown_task,
 978                                  NULL);
 979 }
 980
 981 /**
 982  * Define "main" method using service macro.
 983  */
 984 GNUNET_SERVICE_MAIN
 985 ("psycstore",
 986  GNUNET_SERVICE_OPTION_NONE,
 987  run,
 988  client_notify_connect,
 989  client_notify_disconnect,
 990  NULL,
 991  GNUNET_MQ_hd_fixed_size (client_membership_store,
 992                           GNUNET_MESSAGE_TYPE_PSYCSTORE_MEMBERSHIP_STORE,
 993                           struct MembershipStoreRequest,
 994                           NULL),
 995  GNUNET_MQ_hd_fixed_size (client_membership_test,
 996                           GNUNET_MESSAGE_TYPE_PSYCSTORE_MEMBERSHIP_TEST,
 997                           struct MembershipTestRequest,
 998                           NULL),
 999  GNUNET_MQ_hd_var_size (client_fragment_store,
1000                         GNUNET_MESSAGE_TYPE_PSYCSTORE_FRAGMENT_STORE,
1001                         struct FragmentStoreRequest,
1002                         NULL),
1003  GNUNET_MQ_hd_fixed_size (client_fragment_get,
1004                           GNUNET_MESSAGE_TYPE_PSYCSTORE_FRAGMENT_GET,
1005                           struct FragmentGetRequest,
1006                           NULL),
1007  GNUNET_MQ_hd_var_size (client_message_get,
1008                         GNUNET_MESSAGE_TYPE_PSYCSTORE_MESSAGE_GET,
1009                         struct MessageGetRequest,
1010                         NULL),
1011  GNUNET_MQ_hd_fixed_size (client_message_get_fragment,
1012                           GNUNET_MESSAGE_TYPE_PSYCSTORE_MESSAGE_GET_FRAGMENT,
1013                           struct MessageGetFragmentRequest,
1014                           NULL),
1015  GNUNET_MQ_hd_fixed_size (client_counters_get,
1016                           GNUNET_MESSAGE_TYPE_PSYCSTORE_COUNTERS_GET,
1017                           struct OperationRequest,
1018                           NULL),
1019  GNUNET_MQ_hd_fixed_size (client_state_modify,
1020                           GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_MODIFY,
1021                           struct StateModifyRequest,
1022                           NULL),
1023  GNUNET_MQ_hd_var_size (client_state_sync,
1024                         GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_SYNC,
1025                         struct StateSyncRequest,
1026                         NULL),
1027  GNUNET_MQ_hd_fixed_size (client_state_reset,
1028                           GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_RESET,
1029                           struct OperationRequest,
1030                           NULL),
1031  GNUNET_MQ_hd_fixed_size (client_state_hash_update,
1032                           GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_HASH_UPDATE,
1033                           struct StateHashUpdateRequest,
1034                           NULL),
1035  GNUNET_MQ_hd_var_size (client_state_get,
1036                         GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_GET,
1037                         struct OperationRequest,
1038                         NULL),
1039  GNUNET_MQ_hd_var_size (client_state_get_prefix,
1040                         GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_GET_PREFIX,
1041                         struct OperationRequest,
1042                         NULL));
1043
1044 /* end of gnunet-service-psycstore.c */