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