src/scalarproduct/scalarproduct_api.c

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