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