src/scalarproduct/scalarproduct_api.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2013, 2014, 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  * @file scalarproduct/scalarproduct_api.c
  17  * @brief API for the scalarproduct
  18  * @author Christian Fuchs
  19  * @author Gaurav Kukreja
  20  * @author Christian Grothoff
  21  */
  22 #include "platform.h"
  23 #include "gnunet_util_lib.h"
  24 #include "gnunet_statistics_service.h"
  25 #include "gnunet_scalarproduct_service.h"
  26 #include "gnunet_protocols.h"
  27 #include "scalarproduct.h"
  28
  29 #define LOG(kind,...) GNUNET_log_from (kind, "scalarproduct-api",__VA_ARGS__)
  30
  31
  32 /**
  33  * The abstraction function for our internal callback
  34  *
  35  * @param h computation handle
  36  * @param msg response we got, NULL on errors
  37  * @param status processing status code
  38  */
  39 typedef void
  40 (*GNUNET_SCALARPRODUCT_ResponseMessageHandler) (struct GNUNET_SCALARPRODUCT_ComputationHandle *h,
  41                                                 const struct ClientResponseMessage *msg,
  42                                                 enum GNUNET_SCALARPRODUCT_ResponseStatus status);
  43
  44
  45 /**
  46  * A handle returned for each computation
  47  */
  48 struct GNUNET_SCALARPRODUCT_ComputationHandle
  49 {
  50   /**
  51    * Our configuration.
  52    */
  53   const struct GNUNET_CONFIGURATION_Handle *cfg;
  54
  55   /**
  56    * Current connection to the scalarproduct service.
  57    */
  58   struct GNUNET_MQ_Handle *mq;
  59
  60   /**
  61    * Function to call after transmission of the request (Bob).
  62    */
  63   GNUNET_SCALARPRODUCT_ContinuationWithStatus cont_status;
  64
  65   /**
  66    * Function to call after transmission of the request (Alice).
  67    */
  68   GNUNET_SCALARPRODUCT_DatumProcessor cont_datum;
  69
  70   /**
  71    * Closure for @e cont_status or @e cont_datum.
  72    */
  73   void *cont_cls;
  74
  75   /**
  76    * API internal callback for results and failures to be forwarded to
  77    * the client.
  78    */
  79   GNUNET_SCALARPRODUCT_ResponseMessageHandler response_proc;
  80
  81   /**
  82    * The shared session key identifying this computation
  83    */
  84   struct GNUNET_HashCode key;
  85
  86 };
  87
  88
  89 /**
  90  * Called when a response is received from the service. Perform basic
  91  * check that the message is well-formed.
  92  *
  93  * @param cls Pointer to the Master Context
  94  * @param message Pointer to the data received in response
  95  * @return #GNUNET_OK if @a message is well-formed
  96  */
  97 static int
  98 check_response (void *cls,
  99                  const struct ClientResponseMessage *message)
 100 {
 101   if (ntohs (message->header.size) !=
 102       ntohl (message->product_length) + sizeof (struct ClientResponseMessage))
 103   {
 104     GNUNET_break (0);
 105     return GNUNET_SYSERR;
 106   }
 107   return GNUNET_OK;
 108 }
 109
 110
 111 /**
 112  * Handles the STATUS received from the service for a response, does
 113  * not contain a payload.  Called when we participate as "Bob" via
 114  * #GNUNET_SCALARPRODUCT_accept_computation().
 115  *
 116  * @param h our Handle
 117  * @param msg the response received
 118  * @param status the condition the request was terminated with (eg: disconnect)
 119  */
 120 static void
 121 process_status_message (struct GNUNET_SCALARPRODUCT_ComputationHandle *h,
 122                         const struct ClientResponseMessage *msg,
 123                         enum GNUNET_SCALARPRODUCT_ResponseStatus status)
 124 {
 125   if (NULL != h->cont_status)
 126     h->cont_status (h->cont_cls,
 127                     status);
 128   GNUNET_SCALARPRODUCT_cancel (h);
 129 }
 130
 131
 132 /**
 133  * Called when a response is received from the service. After basic
 134  * check, the handler in `h->response_proc` is called. This functions
 135  * handles the response to the client which used the API.
 136  *
 137  * @param cls Pointer to the Master Context
 138  * @param msg Pointer to the data received in response
 139  */
 140 static void
 141 handle_response (void *cls,
 142                  const struct ClientResponseMessage *message)
 143 {
 144   struct GNUNET_SCALARPRODUCT_ComputationHandle *h = cls;
 145   enum GNUNET_SCALARPRODUCT_ResponseStatus status;
 146
 147   status = (enum GNUNET_SCALARPRODUCT_ResponseStatus) ntohl (message->status);
 148   h->response_proc (h,
 149                     message,
 150                     status);
 151 }
 152
 153
 154 /**
 155  * Check if the keys for all given elements are unique.
 156  *
 157  * @param elements elements to check
 158  * @param element_count size of the @a elements array
 159  * @return #GNUNET_OK if all keys are unique
 160  */
 161 static int
 162 check_unique (const struct GNUNET_SCALARPRODUCT_Element *elements,
 163               uint32_t element_count)
 164 {
 165   struct GNUNET_CONTAINER_MultiHashMap *map;
 166   uint32_t i;
 167   int ok;
 168
 169   ok = GNUNET_OK;
 170   map = GNUNET_CONTAINER_multihashmap_create (2 * element_count,
 171                                               GNUNET_YES);
 172   for (i=0;i<element_count;i++)
 173     if (GNUNET_OK !=
 174         GNUNET_CONTAINER_multihashmap_put (map,
 175                                            &elements[i].key,
 176                                            map,
 177                                            GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
 178     {
 179       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 180                   _("Keys given to SCALARPRODUCT not unique!\n"));
 181       ok = GNUNET_SYSERR;
 182     }
 183   GNUNET_CONTAINER_multihashmap_destroy (map);
 184   return ok;
 185 }
 186
 187
 188 /**
 189  * We encountered an error communicating with the set service while
 190  * performing a set operation. Report to the application.
 191  *
 192  * @param cls the `struct GNUNET_SCALARPRODUCT_ComputationHandle`
 193  * @param error error code
 194  */
 195 static void
 196 mq_error_handler (void *cls,
 197                   enum GNUNET_MQ_Error error)
 198 {
 199   struct GNUNET_SCALARPRODUCT_ComputationHandle *h = cls;
 200
 201   LOG (GNUNET_ERROR_TYPE_INFO,
 202        "Disconnected from SCALARPRODUCT service.\n");
 203   h->response_proc (h,
 204                     NULL,
 205                     GNUNET_SCALARPRODUCT_STATUS_DISCONNECTED);
 206 }
 207
 208
 209 /**
 210  * Used by Bob's client to cooperate with Alice,
 211  *
 212  * @param cfg the gnunet configuration handle
 213  * @param key Session key unique to the requesting client
 214  * @param elements Array of elements of the vector
 215  * @param element_count Number of elements in the @a elements vector
 216  * @param cont Callback function
 217  * @param cont_cls Closure for @a cont
 218  * @return a new handle for this computation
 219  */
 220 struct GNUNET_SCALARPRODUCT_ComputationHandle *
 221 GNUNET_SCALARPRODUCT_accept_computation (const struct GNUNET_CONFIGURATION_Handle *cfg,
 222                                          const struct GNUNET_HashCode *session_key,
 223                                          const struct GNUNET_SCALARPRODUCT_Element *elements,
 224                                          uint32_t element_count,
 225                                          GNUNET_SCALARPRODUCT_ContinuationWithStatus cont,
 226                                          void *cont_cls)
 227 {
 228   struct GNUNET_SCALARPRODUCT_ComputationHandle *h
 229     = GNUNET_new (struct GNUNET_SCALARPRODUCT_ComputationHandle);
 230   struct GNUNET_MQ_MessageHandler handlers[] = {
 231     GNUNET_MQ_hd_var_size (response,
 232                            GNUNET_MESSAGE_TYPE_SCALARPRODUCT_RESULT,
 233                            struct ClientResponseMessage,
 234                            h),
 235     GNUNET_MQ_handler_end ()
 236   };
 237   struct GNUNET_MQ_Envelope *env;
 238   struct BobComputationMessage *msg;
 239   struct ComputationBobCryptodataMultipartMessage *mmsg;
 240   uint32_t size;
 241   uint16_t possible;
 242   uint16_t todo;
 243   uint32_t element_count_transfered;
 244
 245
 246   if (GNUNET_SYSERR == check_unique (elements,
 247                                      element_count))
 248     return NULL;
 249   h->cont_status = cont;
 250   h->cont_cls = cont_cls;
 251   h->response_proc = &process_status_message;
 252   h->cfg = cfg;
 253   h->key = *session_key;
 254   h->mq = GNUNET_CLIENT_connect (cfg,
 255                                  "scalarproduct-bob",
 256                                  handlers,
 257                                  &mq_error_handler,
 258                                  h);
 259   if (NULL == h->mq)
 260   {
 261     /* scalarproduct configuration error */
 262     GNUNET_break (0);
 263     GNUNET_free (h);
 264     return NULL;
 265   }
 266   possible = (GNUNET_MAX_MESSAGE_SIZE - 1 - sizeof (struct BobComputationMessage))
 267     / sizeof (struct GNUNET_SCALARPRODUCT_Element);
 268   todo = GNUNET_MIN (possible,
 269                      element_count);
 270   size = todo * sizeof (struct GNUNET_SCALARPRODUCT_Element);
 271   env = GNUNET_MQ_msg_extra (msg,
 272                              size,
 273                              GNUNET_MESSAGE_TYPE_SCALARPRODUCT_CLIENT_TO_BOB);
 274   msg->element_count_total = htonl (element_count);
 275   msg->element_count_contained = htonl (todo);
 276   msg->session_key = *session_key;
 277   GNUNET_memcpy (&msg[1],
 278           elements,
 279           size);
 280   element_count_transfered = todo;
 281   GNUNET_MQ_send (h->mq,
 282                   env);
 283   possible = (GNUNET_MAX_MESSAGE_SIZE - 1 - sizeof (*mmsg))
 284     / sizeof (struct GNUNET_SCALARPRODUCT_Element);
 285   while (element_count_transfered < element_count)
 286   {
 287     todo = GNUNET_MIN (possible,
 288                        element_count - element_count_transfered);
 289     size = todo * sizeof (struct GNUNET_SCALARPRODUCT_Element);
 290     env = GNUNET_MQ_msg_extra (mmsg,
 291                                size,
 292                                GNUNET_MESSAGE_TYPE_SCALARPRODUCT_CLIENT_MULTIPART_BOB);
 293     mmsg->element_count_contained = htonl (todo);
 294     GNUNET_memcpy (&mmsg[1],
 295             &elements[element_count_transfered],
 296             size);
 297     element_count_transfered += todo;
 298     GNUNET_MQ_send (h->mq,
 299                     env);
 300   }
 301   return h;
 302 }
 303
 304
 305 /**
 306  * Handles the RESULT received from the service for a request, should
 307  * contain a result MPI value.  Called when we participate as "Alice" via
 308  * #GNUNET_SCALARPRODUCT_start_computation().
 309  *
 310  * @param h our Handle
 311  * @param msg Pointer to the response received
 312  * @param status the condition the request was terminated with (eg: disconnect)
 313  */
 314 static void
 315 process_result_message (struct GNUNET_SCALARPRODUCT_ComputationHandle *h,
 316                         const struct ClientResponseMessage *msg,
 317                         enum GNUNET_SCALARPRODUCT_ResponseStatus status)
 318 {
 319   uint32_t product_len;
 320   gcry_mpi_t result = NULL;
 321   gcry_error_t rc;
 322   gcry_mpi_t num;
 323   size_t rsize;
 324
 325   if (GNUNET_SCALARPRODUCT_STATUS_SUCCESS == status)
 326   {
 327     result = gcry_mpi_new (0);
 328
 329     product_len = ntohl (msg->product_length);
 330     if (0 < product_len)
 331     {
 332       rsize = 0;
 333       if (0 != (rc = gcry_mpi_scan (&num, GCRYMPI_FMT_STD,
 334                                     &msg[1],
 335                                     product_len,
 336                                     &rsize)))
 337       {
 338         LOG_GCRY (GNUNET_ERROR_TYPE_ERROR,
 339                   "gcry_mpi_scan",
 340                   rc);
 341         gcry_mpi_release (result);
 342         result = NULL;
 343         status = GNUNET_SCALARPRODUCT_STATUS_INVALID_RESPONSE;
 344       }
 345       else
 346       {
 347         if (0 < (int32_t) ntohl (msg->range))
 348           gcry_mpi_add (result, result, num);
 349         else
 350           gcry_mpi_sub (result, result, num);
 351         gcry_mpi_release (num);
 352       }
 353     }
 354   }
 355   if (NULL != h->cont_datum)
 356     h->cont_datum (h->cont_cls,
 357                    status,
 358                    result);
 359   if (NULL != result)
 360     gcry_mpi_release (result);
 361   GNUNET_SCALARPRODUCT_cancel (h);
 362 }
 363
 364
 365 /**
 366  * Request by Alice's client for computing a scalar product
 367  *
 368  * @param cfg the gnunet configuration handle
 369  * @param session_key Session key should be unique to the requesting client
 370  * @param peer PeerID of the other peer
 371  * @param elements Array of elements of the vector
 372  * @param element_count Number of elements in the @a elements vector
 373  * @param cont Callback function
 374  * @param cont_cls Closure for @a cont
 375  * @return a new handle for this computation
 376  */
 377 struct GNUNET_SCALARPRODUCT_ComputationHandle *
 378 GNUNET_SCALARPRODUCT_start_computation (const struct GNUNET_CONFIGURATION_Handle *cfg,
 379                                         const struct GNUNET_HashCode *session_key,
 380                                         const struct GNUNET_PeerIdentity *peer,
 381                                         const struct GNUNET_SCALARPRODUCT_Element *elements,
 382                                         uint32_t element_count,
 383                                         GNUNET_SCALARPRODUCT_DatumProcessor cont,
 384                                         void *cont_cls)
 385 {
 386   struct GNUNET_SCALARPRODUCT_ComputationHandle *h
 387     = GNUNET_new (struct GNUNET_SCALARPRODUCT_ComputationHandle);
 388   struct GNUNET_MQ_MessageHandler handlers[] = {
 389     GNUNET_MQ_hd_var_size (response,
 390                            GNUNET_MESSAGE_TYPE_SCALARPRODUCT_RESULT,
 391                            struct ClientResponseMessage,
 392                            h),
 393     GNUNET_MQ_handler_end ()
 394   };
 395   struct GNUNET_MQ_Envelope *env;
 396   struct AliceComputationMessage *msg;
 397   struct ComputationBobCryptodataMultipartMessage *mmsg;
 398   uint32_t size;
 399   uint16_t possible;
 400   uint16_t todo;
 401   uint32_t element_count_transfered;
 402
 403   if (GNUNET_SYSERR == check_unique (elements,
 404                                      element_count))
 405     return NULL;
 406   h->mq = GNUNET_CLIENT_connect (cfg,
 407                                  "scalarproduct-alice",
 408                                  handlers,
 409                                  &mq_error_handler,
 410                                  h);
 411   if (NULL == h->mq)
 412   {
 413     /* missconfigured scalarproduct service */
 414     GNUNET_break (0);
 415     GNUNET_free (h);
 416     return NULL;
 417   }
 418   h->cont_datum = cont;
 419   h->cont_cls = cont_cls;
 420   h->response_proc = &process_result_message;
 421   h->cfg = cfg;
 422   h->key = *session_key;
 423
 424   possible = (GNUNET_MAX_MESSAGE_SIZE - 1 - sizeof (struct AliceComputationMessage))
 425       / sizeof (struct GNUNET_SCALARPRODUCT_Element);
 426   todo = GNUNET_MIN (possible,
 427                      element_count);
 428   size = todo * sizeof (struct GNUNET_SCALARPRODUCT_Element);
 429   env = GNUNET_MQ_msg_extra (msg,
 430                              size,
 431                              GNUNET_MESSAGE_TYPE_SCALARPRODUCT_CLIENT_TO_ALICE);
 432   msg->element_count_total = htonl (element_count);
 433   msg->element_count_contained = htonl (todo);
 434   msg->reserved = htonl (0);
 435   msg->peer = *peer;
 436   msg->session_key = *session_key;
 437   GNUNET_memcpy (&msg[1],
 438           elements,
 439           size);
 440   GNUNET_MQ_send (h->mq,
 441                   env);
 442   element_count_transfered = todo;
 443   possible = (GNUNET_MAX_MESSAGE_SIZE - 1 - sizeof (*mmsg))
 444     / sizeof (struct GNUNET_SCALARPRODUCT_Element);
 445   while (element_count_transfered < element_count)
 446   {
 447     todo = GNUNET_MIN (possible,
 448                        element_count - element_count_transfered);
 449     size = todo * sizeof (struct GNUNET_SCALARPRODUCT_Element);
 450     env = GNUNET_MQ_msg_extra (mmsg,
 451                                size,
 452                                GNUNET_MESSAGE_TYPE_SCALARPRODUCT_CLIENT_MULTIPART_ALICE);
 453     mmsg->element_count_contained = htonl (todo);
 454     GNUNET_memcpy (&mmsg[1],
 455             &elements[element_count_transfered],
 456             size);
 457     element_count_transfered += todo;
 458     GNUNET_MQ_send (h->mq,
 459                     env);
 460   }
 461   return h;
 462 }
 463
 464
 465 /**
 466  * Cancel an ongoing computation or revoke our collaboration offer.
 467  * Closes the connection to the service
 468  *
 469  * @param h computation handle to terminate
 470  */
 471 void
 472 GNUNET_SCALARPRODUCT_cancel (struct GNUNET_SCALARPRODUCT_ComputationHandle *h)
 473 {
 474   if (NULL != h->mq)
 475   {
 476     GNUNET_MQ_destroy (h->mq);
 477     h->mq = NULL;
 478   }
 479   GNUNET_free (h);
 480 }
 481
 482
 483 /* end of scalarproduct_api.c */