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