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