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