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