src/scalarproduct/scalarproduct_api.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2013 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  *
  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  ***  Datatype Declarations                          **********
  39  **************************************************************/
  40
  41 /**
  42  * the abstraction function for our internal callback
  43  */
  44 typedef void (*GNUNET_SCALARPRODUCT_ResponseMessageHandler) (void *cls,
  45                                                              const struct GNUNET_MessageHeader *msg,
  46                                                              enum GNUNET_SCALARPRODUCT_ResponseStatus status);
  47
  48 /**
  49  * Entry in the request queue per client
  50  */
  51 struct GNUNET_SCALARPRODUCT_ComputationHandle
  52 {
  53   /**
  54    * This is a linked list.
  55    */
  56   struct GNUNET_SCALARPRODUCT_ComputationHandle *next;
  57
  58   /**
  59    * This is a linked list.
  60    */
  61   struct GNUNET_SCALARPRODUCT_ComputationHandle *prev;
  62
  63   /**
  64    * Our configuration.
  65    */
  66   const struct GNUNET_CONFIGURATION_Handle *cfg;
  67
  68   /**
  69    * Current connection to the scalarproduct service.
  70    */
  71   struct GNUNET_CLIENT_Connection *client;
  72
  73   /**
  74    * Handle for statistics.
  75    */
  76   struct GNUNET_STATISTICS_Handle *stats;
  77
  78   /**
  79    * The shared session key identifying this computation
  80    */
  81   struct GNUNET_HashCode key;
  82
  83   /**
  84    * Current transmit handle.
  85    */
  86   struct GNUNET_CLIENT_TransmitHandle *th;
  87
  88   /**
  89    * Size of the message
  90    */
  91   uint16_t message_size;
  92
  93   /**
  94    * Message to be sent to the scalarproduct service
  95    */
  96   struct GNUNET_SCALARPRODUCT_client_request * msg;
  97
  98   /**
  99    * The msg handler callback
 100    */
 101   union
 102   {
 103   /**
 104    * Function to call after transmission of the request.
 105    */
 106   GNUNET_SCALARPRODUCT_ContinuationWithStatus cont_status;
 107
 108   /**
 109    * Function to call after transmission of the request.
 110    */
 111   GNUNET_SCALARPRODUCT_DatumProcessor cont_datum;
 112   };
 113
 114   /**
 115    * Closure for 'cont'.
 116    */
 117   void *cont_cls;
 118
 119   /**
 120    * Response Processor for response from the service. This function calls the
 121    * continuation function provided by the client.
 122    */
 123   GNUNET_SCALARPRODUCT_ResponseMessageHandler response_proc;
 124 };
 125
 126 /**************************************************************
 127  ***  Global Variables                               **********
 128  **************************************************************/
 129 /**
 130  * Head of the active sessions queue
 131  */
 132 static struct GNUNET_SCALARPRODUCT_ComputationHandle *head;
 133 /**
 134  * Tail of the active sessions queue
 135  */
 136 static struct GNUNET_SCALARPRODUCT_ComputationHandle *tail;
 137
 138 /**************************************************************
 139  ***  Function Declarations                          **********
 140  **************************************************************/
 141
 142 void
 143 GNUNET_SCALARPRODUCT_cancel (struct GNUNET_SCALARPRODUCT_ComputationHandle * h);
 144
 145 /**************************************************************
 146  ***  Static Function Declarations                   **********
 147  **************************************************************/
 148
 149
 150 /**
 151  * Handles the STATUS received from the service for a response, does not contain a payload
 152  *
 153  * @param cls our Handle
 154  * @param msg Pointer to the response received
 155  * @param status the condition the request was terminated with (eg: disconnect)
 156  */
 157 static void
 158 process_status_message (void *cls,
 159                         const struct GNUNET_MessageHeader *msg,
 160                         enum GNUNET_SCALARPRODUCT_ResponseStatus status)
 161 {
 162   struct GNUNET_SCALARPRODUCT_ComputationHandle *qe = cls;
 163
 164   qe->cont_status (qe->cont_cls, status);
 165 }
 166
 167
 168 /**
 169  * Handles the RESULT received from the service for a request, should contain a result MPI value
 170  *
 171  * @param cls our Handle
 172  * @param msg Pointer to the response received
 173  * @param status the condition the request was terminated with (eg: disconnect)
 174  */
 175 static void
 176 process_result_message (void *cls,
 177                         const struct GNUNET_MessageHeader *msg,
 178                         enum GNUNET_SCALARPRODUCT_ResponseStatus status)
 179 {
 180   struct GNUNET_SCALARPRODUCT_ComputationHandle *qe = cls;
 181   const struct GNUNET_SCALARPRODUCT_client_response *message =
 182           (const struct GNUNET_SCALARPRODUCT_client_response *) msg;
 183   gcry_mpi_t result = NULL;
 184   gcry_error_t rc;
 185
 186   if (GNUNET_SCALARPRODUCT_Status_Success == status)
 187     {
 188       size_t product_len = ntohl (message->product_length);
 189       result = gcry_mpi_new (0);
 190
 191       if (0 < product_len)
 192         {
 193           gcry_mpi_t num;
 194           size_t read = 0;
 195
 196           if (0 != (rc = gcry_mpi_scan (&num, GCRYMPI_FMT_STD, &message[1], product_len, &read)))
 197             {
 198               LOG_GCRY(GNUNET_ERROR_TYPE_ERROR, "gcry_mpi_scan", rc);
 199               gcry_mpi_release (result);
 200               result = NULL;
 201               status = GNUNET_SCALARPRODUCT_Status_InvalidResponse;
 202             }
 203           else
 204             {
 205               if (0 < message->range)
 206                 gcry_mpi_add (result, result, num);
 207               else if (0 > message->range)
 208                 gcry_mpi_sub (result, result, num);
 209               gcry_mpi_release (num);
 210             }
 211         }
 212     }
 213   qe->cont_datum (qe->cont_cls, status, result);
 214 }
 215
 216
 217 /**
 218  * Called when a response is received from the service. After basic check, the
 219  * handler in qe->response_proc is called. This functions handles the response
 220  * to the client which used the API.
 221  *
 222  * @param cls Pointer to the Master Context
 223  * @param msg Pointer to the data received in response
 224  */
 225 static void
 226 receive_cb (void *cls, const struct GNUNET_MessageHeader *msg)
 227 {
 228   struct GNUNET_SCALARPRODUCT_ComputationHandle *qe = cls;
 229   const struct GNUNET_SCALARPRODUCT_client_response *message =
 230           (const struct GNUNET_SCALARPRODUCT_client_response *) msg;
 231   enum GNUNET_SCALARPRODUCT_ResponseStatus status = GNUNET_SCALARPRODUCT_Status_InvalidResponse;
 232
 233   if (NULL == msg)
 234     {
 235       LOG (GNUNET_ERROR_TYPE_WARNING, "Disconnected by Service.\n");
 236       status = GNUNET_SCALARPRODUCT_Status_ServiceDisconnected;
 237     }
 238   else if (GNUNET_MESSAGE_TYPE_SCALARPRODUCT_SERVICE_TO_CLIENT != ntohs (msg->type))
 239     {
 240       LOG (GNUNET_ERROR_TYPE_WARNING, "Invalid message type received\n");
 241     }
 242   else if (0 < ntohl (message->product_length) || (0 == message->range))
 243     {
 244       // response for the responder client, successful
 245       GNUNET_STATISTICS_update (qe->stats,
 246                                 gettext_noop ("# SUC responder result messages received"), 1,
 247                                 GNUNET_NO);
 248
 249       status = GNUNET_SCALARPRODUCT_Status_Success;
 250     }
 251
 252   if (qe->cont_datum != NULL)
 253     qe->response_proc (qe, msg, status);
 254
 255   GNUNET_CONTAINER_DLL_remove (head, tail, qe);
 256   GNUNET_free (qe);
 257 }
 258
 259
 260 /**
 261  * Transmits the request to the VectorProduct Service
 262  *
 263  * @param cls Closure
 264  * @param size Size of the buffer
 265  * @param buf Pointer to the buffer
 266  *
 267  * @return Size of the message sent
 268  */
 269 static size_t
 270 transmit_request (void *cls, size_t size,
 271                   void *buf)
 272 {
 273   struct GNUNET_SCALARPRODUCT_ComputationHandle *qe = cls;
 274
 275   if (NULL == buf)
 276     {
 277       LOG (GNUNET_ERROR_TYPE_DEBUG, "Failed to transmit request to SCALARPRODUCT.\n");
 278       GNUNET_STATISTICS_update (qe->stats,
 279                                 gettext_noop ("# transmission request failures"),
 280                                 1, GNUNET_NO);
 281
 282       // notify caller about the error, done here.
 283       if (qe->cont_datum != NULL)
 284         qe->response_proc (qe, NULL, GNUNET_SCALARPRODUCT_Status_Failure);
 285
 286       GNUNET_SCALARPRODUCT_cancel (cls);
 287       return 0;
 288     }
 289   memcpy (buf, qe->msg, size);
 290
 291   GNUNET_free (qe->msg);
 292   qe->msg = NULL;
 293   qe->th = NULL;
 294
 295   GNUNET_CLIENT_receive (qe->client, &receive_cb, qe,
 296                          GNUNET_TIME_UNIT_FOREVER_REL);
 297
 298 #if INSANE_STATISTICS
 299   GNUNET_STATISTICS_update (qe->stats,
 300                             gettext_noop ("# bytes sent to scalarproduct"), 1,
 301                             GNUNET_NO);
 302 #endif
 303   return size;
 304 }
 305
 306
 307 /**************************************************************
 308  ***  API                                            **********
 309  **************************************************************/
 310
 311
 312 /**
 313  * Used by Bob's client to cooperate with Alice,
 314  *
 315  * @param cfg the gnunet configuration handle
 316  * @param key Session key unique to the requesting client
 317  * @param elements Array of elements of the vector
 318  * @param element_count Number of elements in the vector
 319  * @param cont Callback function
 320  * @param cont_cls Closure for the callback function
 321  */
 322 struct GNUNET_SCALARPRODUCT_ComputationHandle *
 323 GNUNET_SCALARPRODUCT_response (const struct GNUNET_CONFIGURATION_Handle *cfg,
 324                                const struct GNUNET_HashCode * key,
 325                                const int32_t * elements,
 326                                uint32_t element_count,
 327                                GNUNET_SCALARPRODUCT_ContinuationWithStatus cont,
 328                                void *cont_cls)
 329 {
 330   struct GNUNET_SCALARPRODUCT_ComputationHandle *h;
 331   struct GNUNET_SCALARPRODUCT_client_request *msg;
 332   int32_t * vector;
 333   uint16_t size;
 334   uint64_t i;
 335
 336   GNUNET_assert (GNUNET_SERVER_MAX_MESSAGE_SIZE >= sizeof (struct GNUNET_SCALARPRODUCT_client_request)
 337                  + element_count * sizeof (int32_t));
 338   h = GNUNET_new (struct GNUNET_SCALARPRODUCT_ComputationHandle);
 339   h->client = GNUNET_CLIENT_connect ("scalarproduct", cfg);
 340   if (!h->client)
 341     {
 342       LOG (GNUNET_ERROR_TYPE_ERROR,
 343            _ ("Failed to connect to the scalarproduct service\n"));
 344       GNUNET_free (h);
 345       return NULL;
 346     }
 347   h->stats = GNUNET_STATISTICS_create ("scalarproduct-api", cfg);
 348   if (!h->stats)
 349     {
 350       LOG (GNUNET_ERROR_TYPE_ERROR,
 351            _ ("Failed to send a message to the statistics service\n"));
 352       GNUNET_CLIENT_disconnect (h->client);
 353       GNUNET_free (h);
 354       return NULL;
 355     }
 356
 357   size = sizeof (struct GNUNET_SCALARPRODUCT_client_request) + element_count * sizeof (int32_t);
 358
 359   h->cont_status = cont;
 360   h->cont_cls = cont_cls;
 361   h->response_proc = &process_status_message;
 362   h->cfg = cfg;
 363   memcpy (&h->key, key, sizeof (struct GNUNET_HashCode));
 364
 365   msg = (struct GNUNET_SCALARPRODUCT_client_request*) GNUNET_malloc (size);
 366   h->msg = msg;
 367   msg->header.size = htons (size);
 368   msg->header.type = htons (GNUNET_MESSAGE_TYPE_SCALARPRODUCT_CLIENT_TO_BOB);
 369   msg->element_count = htonl (element_count);
 370
 371   vector = (int32_t*) & msg[1];
 372   // copy each element over to the message
 373   for (i = 0; i < element_count; i++)
 374     vector[i] = htonl (elements[i]);
 375
 376   memcpy (&msg->key, key, sizeof (struct GNUNET_HashCode));
 377
 378   h->th = GNUNET_CLIENT_notify_transmit_ready (h->client, size,
 379                                                GNUNET_TIME_UNIT_FOREVER_REL,
 380                                                GNUNET_YES, // retry is OK in the initial stage
 381                                                &transmit_request, h);
 382   if (!h->th)
 383     {
 384       LOG (GNUNET_ERROR_TYPE_ERROR,
 385            _ ("Failed to send a message to the scalarproduct service\n"));
 386       GNUNET_STATISTICS_destroy (h->stats, GNUNET_YES);
 387       GNUNET_CLIENT_disconnect (h->client);
 388       GNUNET_free (h->msg);
 389       GNUNET_free (h);
 390       return NULL;
 391     }
 392   GNUNET_CONTAINER_DLL_insert (head, tail, h);
 393   return h;
 394 }
 395
 396
 397 /**
 398  * Request by Alice's client for computing a scalar product
 399  *
 400  * @param cfg the gnunet configuration handle
 401  * @param key Session key should be unique to the requesting client
 402  * @param peer PeerID of the other peer
 403  * @param elements Array of elements of the vector
 404  * @param element_count Number of elements in the vector
 405  * @param mask Array of the mask
 406  * @param mask_bytes number of bytes in the mask
 407  * @param cont Callback function
 408  * @param cont_cls Closure for the callback function
 409  */
 410 struct GNUNET_SCALARPRODUCT_ComputationHandle *
 411 GNUNET_SCALARPRODUCT_request (const struct GNUNET_CONFIGURATION_Handle *cfg,
 412                               const struct GNUNET_HashCode * key,
 413                               const struct GNUNET_PeerIdentity *peer,
 414                               const int32_t * elements,
 415                               uint32_t element_count,
 416                               const unsigned char * mask,
 417                               uint32_t mask_bytes,
 418                               GNUNET_SCALARPRODUCT_DatumProcessor cont,
 419                               void *cont_cls)
 420 {
 421   struct GNUNET_SCALARPRODUCT_ComputationHandle *h;
 422   struct GNUNET_SCALARPRODUCT_client_request *msg;
 423   int32_t * vector;
 424   uint16_t size;
 425   uint64_t i;
 426
 427   GNUNET_assert (GNUNET_SERVER_MAX_MESSAGE_SIZE >= sizeof (struct GNUNET_SCALARPRODUCT_client_request)
 428                  +element_count * sizeof (int32_t)
 429                  + mask_bytes);
 430
 431   h = GNUNET_new (struct GNUNET_SCALARPRODUCT_ComputationHandle);
 432   h->client = GNUNET_CLIENT_connect ("scalarproduct", cfg);
 433   if (!h->client)
 434     {
 435       LOG (GNUNET_ERROR_TYPE_ERROR,
 436            _ ("Failed to connect to the scalarproduct service\n"));
 437       GNUNET_free (h);
 438       return NULL;
 439     }
 440   h->stats = GNUNET_STATISTICS_create ("scalarproduct-api", cfg);
 441   if (!h->stats)
 442     {
 443       LOG (GNUNET_ERROR_TYPE_ERROR,
 444            _ ("Failed to send a message to the statistics service\n"));
 445       GNUNET_CLIENT_disconnect (h->client);
 446       GNUNET_free (h);
 447       return NULL;
 448     }
 449
 450   size = sizeof (struct GNUNET_SCALARPRODUCT_client_request) + element_count * sizeof (int32_t) + mask_bytes;
 451
 452   h->cont_datum = cont;
 453   h->cont_cls = cont_cls;
 454   h->response_proc = &process_result_message;
 455   h->cfg = cfg;
 456   memcpy (&h->key, key, sizeof (struct GNUNET_HashCode));
 457
 458   msg = (struct GNUNET_SCALARPRODUCT_client_request*) GNUNET_malloc (size);
 459   h->msg = msg;
 460   msg->header.size = htons (size);
 461   msg->header.type = htons (GNUNET_MESSAGE_TYPE_SCALARPRODUCT_CLIENT_TO_ALICE);
 462   msg->element_count = htonl (element_count);
 463   msg->mask_length = htonl (mask_bytes);
 464
 465   vector = (int32_t*) & msg[1];
 466   // copy each element over to the message
 467   for (i = 0; i < element_count; i++)
 468     vector[i] = htonl (elements[i]);
 469
 470   memcpy (&msg->peer, peer, sizeof (struct GNUNET_PeerIdentity));
 471   memcpy (&msg->key, key, sizeof (struct GNUNET_HashCode));
 472   memcpy (&vector[element_count], mask, mask_bytes);
 473
 474   h->th = GNUNET_CLIENT_notify_transmit_ready (h->client, size,
 475                                                GNUNET_TIME_UNIT_FOREVER_REL,
 476                                                GNUNET_YES, // retry is OK in the initial stage
 477                                                &transmit_request, h);
 478   if (!h->th)
 479     {
 480       LOG (GNUNET_ERROR_TYPE_ERROR,
 481            _ ("Failed to send a message to the scalarproduct service\n"));
 482       GNUNET_STATISTICS_destroy (h->stats, GNUNET_YES);
 483       GNUNET_CLIENT_disconnect (h->client);
 484       GNUNET_free (h->msg);
 485       GNUNET_free (h);
 486       return NULL;
 487     }
 488   GNUNET_CONTAINER_DLL_insert (head, tail, h);
 489   return h;
 490 }
 491
 492
 493 /**
 494  * Cancel an ongoing computation or revoke our collaboration offer.
 495  * Closes the connection to the service
 496  *
 497  * @param h computation handle to terminate
 498  */
 499 void
 500 GNUNET_SCALARPRODUCT_cancel (struct GNUNET_SCALARPRODUCT_ComputationHandle * h)
 501 {
 502   struct GNUNET_SCALARPRODUCT_ComputationHandle * qe;
 503
 504   for (qe = head; head != NULL; qe = head)
 505     {
 506       if (qe == h)
 507         {
 508           GNUNET_CONTAINER_DLL_remove (head, tail, qe);
 509           if (NULL != qe->th)
 510             GNUNET_CLIENT_notify_transmit_ready_cancel (qe->th);
 511           GNUNET_CLIENT_disconnect (qe->client);
 512           GNUNET_STATISTICS_destroy (qe->stats, GNUNET_YES);
 513           GNUNET_free_non_null (qe->msg);
 514           GNUNET_free (qe);
 515           break;
 516         }
 517     }
 518 }
 519 /**
 520  * Cancel ALL ongoing computation or revoke our collaboration offer.
 521  * Closes ALL connections to the service
 522  */
 523 void
 524 GNUNET_SCALARPRODUCT_disconnect ()
 525 {
 526     struct GNUNET_SCALARPRODUCT_ComputationHandle * qe;
 527
 528     LOG (GNUNET_ERROR_TYPE_INFO, "Disconnecting from VectorProduct\n");
 529     for (qe = head; head != NULL; qe = head)
 530     {
 531         GNUNET_CONTAINER_DLL_remove (head, tail, qe);
 532         if (NULL != qe->th)
 533             GNUNET_CLIENT_notify_transmit_ready_cancel (qe->th);
 534         GNUNET_CLIENT_disconnect (qe->client);
 535         GNUNET_STATISTICS_destroy (qe->stats, GNUNET_YES);
 536         GNUNET_free_non_null (qe->msg);
 537         GNUNET_free (qe);
 538     }
 539 }
 540
 541 /* end of ext_api.c */