add support for delays
[oweals/gnunet.git] / src / fs / gnunet-service-fs_lc.c
1 /*
2      This file is part of GNUnet.
3      (C) 2011 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 fs/gnunet-service-fs_lc.c
23  * @brief API to handle 'connected peers'
24  * @author Christian Grothoff
25  */
26
27 #include "platform.h"
28 #include "gnunet-service-fs.h"
29 #include "gnunet-service-fs_lc.h"
30 #include "gnunet-service-fs_cp.h"
31 #include "gnunet-service-fs_pr.h"
32
33
34 /**
35  * Doubly-linked list of requests we are performing
36  * on behalf of the same client.
37  */
38 struct ClientRequest
39 {
40
41   /**
42    * This is a doubly-linked list.
43    */
44   struct ClientRequest *next;
45
46   /**
47    * This is a doubly-linked list.
48    */
49   struct ClientRequest *prev;
50
51   /**
52    * Request this entry represents.
53    */
54   struct GSF_PendingRequest *pr;
55
56   /**
57    * Client list this request belongs to.
58    */
59   struct GSF_LocalClient *lc;
60
61   /**
62    * Task scheduled to destroy the request.
63    */
64   GNUNET_SCHEDULER_TaskIdentifier kill_task;
65
66 };
67
68
69 /**
70  * Replies to be transmitted to the client.  The actual
71  * response message is allocated after this struct.
72  */
73 struct ClientResponse
74 {
75   /**
76    * This is a doubly-linked list.
77    */
78   struct ClientResponse *next;
79
80   /**
81    * This is a doubly-linked list.
82    */
83   struct ClientResponse *prev;
84
85   /**
86    * Client list entry this response belongs to.
87    */
88   struct GSF_LocalClient *lc;
89
90   /**
91    * Number of bytes in the response.
92    */
93   size_t msize;
94 };
95
96
97 /**
98  * A local client.
99  */
100 struct GSF_LocalClient
101 {
102
103   /**
104    * We keep clients in a DLL.
105    */
106   struct GSF_LocalClient *next;
107
108   /**
109    * We keep clients in a DLL.
110    */
111   struct GSF_LocalClient *prev;
112
113   /**
114    * ID of the client.
115    */
116   struct GNUNET_SERVER_Client *client;
117
118   /**
119    * Head of list of requests performed on behalf
120    * of this client right now.
121    */
122   struct ClientRequest *cr_head;
123
124   /**
125    * Tail of list of requests performed on behalf
126    * of this client right now.
127    */
128   struct ClientRequest *cr_tail;
129
130   /**
131    * Head of linked list of responses.
132    */
133   struct ClientResponse *res_head;
134
135   /**
136    * Tail of linked list of responses.
137    */
138   struct ClientResponse *res_tail;
139
140   /**
141    * Context for sending replies.
142    */
143   struct GNUNET_CONNECTION_TransmitHandle *th;
144
145 };
146
147
148 /**
149  * Head of linked list of our local clients.
150  */
151 static struct GSF_LocalClient *client_head;
152
153
154 /**
155  * Head of linked list of our local clients.
156  */
157 static struct GSF_LocalClient *client_tail;
158
159
160 /**
161  * Look up a local client record or create one if it
162  * doesn't exist yet.
163  *
164  * @param client handle of the client
165  * @return handle to local client entry
166  */
167 struct GSF_LocalClient *
168 GSF_local_client_lookup_ (struct GNUNET_SERVER_Client *client)
169 {
170   struct GSF_LocalClient *pos;
171
172   pos = client_head;
173   while ( (pos != NULL) &&
174           (pos->client != client) )
175     pos = pos->next;
176   if (pos != NULL)
177     return pos;
178   pos = GNUNET_malloc (sizeof (struct GSF_LocalClient));
179   pos->client = client;
180   GNUNET_CONTAINER_DLL_insert (client_head,
181                                client_tail,
182                                pos);
183   return pos;
184 }
185
186
187 /**
188  * Free the given client request.
189  *
190  * @param cls the client request to free
191  * @param tc task context
192  */ 
193 static void
194 client_request_destroy (void *cls,
195                         const struct GNUNET_SCHEDULER_TaskContext *tc)
196 {
197   struct ClientRequest *cr = cls;
198   struct GSF_LocalClient *lc;
199
200   cr->kill_task = GNUNET_SCHEDULER_NO_TASK;
201   lc = cr->lc;
202   GNUNET_CONTAINER_DLL_remove (lc->cr_head,
203                                lc->cr_tail,
204                                cr);
205   GSF_pending_request_cancel_ (cr->pr);
206   GNUNET_STATISTICS_update (GSF_stats,
207                             gettext_noop ("# client searches active"),
208                             - 1,
209                             GNUNET_NO);
210   GNUNET_free (cr);
211 }
212
213
214 /**
215  * Handle a reply to a pending request.  Also called if a request
216  * expires (then with data == NULL).  The handler may be called
217  * many times (depending on the request type), but will not be
218  * called during or after a call to GSF_pending_request_cancel 
219  * and will also not be called anymore after a call signalling
220  * expiration.
221  *
222  * @param cls user-specified closure
223  * @param eval evaluation of the result
224  * @param pr handle to the original pending request
225  * @param reply_anonymity_level anonymity level for the reply, UINT32_MAX for "unknown"
226  * @param expiration when does 'data' expire? 
227  * @param type type of the block
228  * @param data response data, NULL on request expiration
229  * @param data_len number of bytes in data
230  */
231 static void
232 client_response_handler (void *cls,
233                          enum GNUNET_BLOCK_EvaluationResult eval,
234                          struct GSF_PendingRequest *pr,
235                          uint32_t reply_anonymity_level,
236                          struct GNUNET_TIME_Absolute expiration,
237                          enum GNUNET_BLOCK_Type type,
238                          const void *data,
239                          size_t data_len)
240 {
241   struct ClientRequest *cr = cls;
242   struct GSF_LocalClient *lc;
243   struct PutMessage *pm;
244   const struct GSF_PendingRequestData *prd;
245   size_t msize;
246
247   if (NULL == data)
248     {
249       /* ugh, request 'timed out' -- how can this be? */
250       GNUNET_break (0);
251       return;
252     }
253   prd = GSF_pending_request_get_data_ (pr);
254   GNUNET_break (type != GNUNET_BLOCK_TYPE_ANY);
255   if ( (prd->type != type) &&
256        (prd->type != GNUNET_BLOCK_TYPE_ANY) )
257     {
258       GNUNET_break (0);
259       return;
260     }
261   GNUNET_STATISTICS_update (GSF_stats,
262                             gettext_noop ("# replies received for local clients"),
263                             1,
264                             GNUNET_NO);
265   GNUNET_assert (pr == cr->pr);
266   lc = cr->lc;
267   msize = sizeof (struct PutMessage) + data_len;
268   {
269     char buf[msize];
270     
271     pm = (struct PutMessage*) buf;
272     pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
273     pm->header.size = htons (msize);
274     pm->type = htonl (type);
275     pm->expiration = GNUNET_TIME_absolute_hton (expiration);
276     memcpy (&pm[1], data, data_len);      
277     GSF_local_client_transmit_ (lc, &pm->header);
278   }
279 #if DEBUG_FS_CLIENT
280   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
281               "Queued reply to query `%s' for local client\n",
282               GNUNET_h2s (&prd->query),
283               (unsigned int) prd->type);
284 #endif
285   if (eval != GNUNET_BLOCK_EVALUATION_OK_LAST)
286     return;
287   cr->kill_task = GNUNET_SCHEDULER_add_now (&client_request_destroy,
288                                             cr);
289 }
290
291
292 /**
293  * Handle START_SEARCH-message (search request from local client).
294  *
295  * @param client identification of the client
296  * @param message the actual message
297  * @return pending request handle for the request, NULL on error
298  */
299 struct GSF_PendingRequest *
300 GSF_local_client_start_search_handler_ (struct GNUNET_SERVER_Client *client,
301                                         const struct GNUNET_MessageHeader *message)
302 {
303   static GNUNET_HashCode all_zeros;
304   const struct SearchMessage *sm;
305   struct GSF_LocalClient *lc;
306   struct ClientRequest *cr;
307   struct GSF_PendingRequestData *prd;
308   uint16_t msize;
309   unsigned int sc;
310   enum GNUNET_BLOCK_Type type;
311   enum GSF_PendingRequestOptions options;
312
313   msize = ntohs (message->size);
314   if ( (msize < sizeof (struct SearchMessage)) ||
315        (0 != (msize - sizeof (struct SearchMessage)) % sizeof (GNUNET_HashCode)) )
316     {
317       GNUNET_break (0);
318       GNUNET_SERVER_receive_done (client,
319                                   GNUNET_SYSERR);
320       return NULL;
321     }
322   GNUNET_STATISTICS_update (GSF_stats,
323                             gettext_noop ("# client searches received"),
324                             1,
325                             GNUNET_NO);
326   sc = (msize - sizeof (struct SearchMessage)) / sizeof (GNUNET_HashCode);
327   sm = (const struct SearchMessage*) message;
328   type = ntohl (sm->type);
329 #if DEBUG_FS_CLIENT
330   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
331               "Received request for `%s' of type %u from local client\n",
332               GNUNET_h2s (&sm->query),
333               (unsigned int) type);
334 #endif
335   lc = GSF_local_client_lookup_ (client);
336
337   /* detect duplicate KBLOCK requests */
338   if ( (type == GNUNET_BLOCK_TYPE_FS_KBLOCK) ||
339        (type == GNUNET_BLOCK_TYPE_FS_NBLOCK) ||
340        (type == GNUNET_BLOCK_TYPE_ANY) )
341     {
342       /* FIXME: this does currently not work to filter duplicate
343          results from *local* datastore since the local store is
344          queried before we continue to process additional
345          messages from the client! -- fix protocol? */
346       cr = lc->cr_head;
347       while (cr != NULL) 
348         {
349           prd = GSF_pending_request_get_data_ (cr->pr);
350           if ( (0 != memcmp (&prd->query,
351                              &sm->query,
352                              sizeof (GNUNET_HashCode))) &&
353                (prd->type == type) )
354             break;
355           cr = cr->next;
356         }
357       if (cr != NULL)   
358         { 
359 #if DEBUG_FS_CLIENT
360           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
361                       "Have existing request, merging content-seen lists.\n");
362 #endif
363           GSF_pending_request_update_ (cr->pr,
364                                        (const GNUNET_HashCode*) &sm[1],
365                                        sc);
366           GNUNET_STATISTICS_update (GSF_stats,
367                                     gettext_noop ("# client searches updated (merged content seen list)"),
368                                     1,
369                                     GNUNET_NO);
370           GNUNET_SERVER_receive_done (client,
371                                       GNUNET_OK);
372           return NULL;
373         }
374     }
375
376   GNUNET_STATISTICS_update (GSF_stats,
377                             gettext_noop ("# client searches active"),
378                             1,
379                             GNUNET_NO);
380   cr = GNUNET_malloc (sizeof (struct ClientRequest));
381   cr->lc = lc;
382   GNUNET_CONTAINER_DLL_insert (lc->cr_head,
383                                lc->cr_tail,
384                                cr);
385   options = GSF_PRO_LOCAL_REQUEST;  
386   if (0 != (1 & ntohl (sm->options)))
387     options |= GSF_PRO_LOCAL_ONLY;
388   cr->pr = GSF_pending_request_create_ (options,
389                                         type,
390                                         &sm->query,
391                                         (type == GNUNET_BLOCK_TYPE_FS_SBLOCK)
392                                         ? &sm->target /* namespace */
393                                         : NULL,
394                                         (0 != memcmp (&sm->target,
395                                                       &all_zeros,
396                                                       sizeof (GNUNET_HashCode)))
397                                         ? (const struct GNUNET_PeerIdentity*) &sm->target
398                                         : NULL,
399                                         NULL, 0, 0 /* bf */, 
400                                         ntohl (sm->anonymity_level),
401                                         0 /* priority */,
402                                         0 /* ttl */,
403                                         0 /* sender PID */,
404                                         (const GNUNET_HashCode*) &sm[1], sc,
405                                         &client_response_handler,
406                                         cr);
407   return cr->pr;
408 }
409
410
411 /**
412  * Transmit the given message by copying it to the target buffer
413  * "buf".  "buf" will be NULL and "size" zero if the socket was closed
414  * for writing in the meantime.  In that case, do nothing
415  * (the disconnect or shutdown handler will take care of the rest).
416  * If we were able to transmit messages and there are still more
417  * pending, ask core again for further calls to this function.
418  *
419  * @param cls closure, pointer to the 'struct GSF_LocalClient'
420  * @param size number of bytes available in buf
421  * @param buf where the callee should write the message
422  * @return number of bytes written to buf
423  */
424 static size_t
425 transmit_to_client (void *cls,
426                     size_t size,
427                     void *buf)
428 {
429   struct GSF_LocalClient *lc = cls;
430   char *cbuf = buf;
431   struct ClientResponse *res;
432   size_t msize;
433   
434   lc->th = NULL;
435   if (NULL == buf)
436     return 0;
437   msize = 0;
438   while ( (NULL != (res = lc->res_head) ) &&
439           (res->msize <= size) )
440     {
441       memcpy (&cbuf[msize], &res[1], res->msize);
442       msize += res->msize;
443       size -= res->msize;
444       GNUNET_CONTAINER_DLL_remove (lc->res_head,
445                                    lc->res_tail,
446                                    res);
447       GNUNET_free (res);
448     }
449   if (NULL != res)
450     lc->th = GNUNET_SERVER_notify_transmit_ready (lc->client,
451                                                   res->msize,
452                                                   GNUNET_TIME_UNIT_FOREVER_REL,
453                                                   &transmit_to_client,
454                                                   lc);
455   return msize;
456 }
457
458
459 /**
460  * Transmit a message to the given local client as soon as possible.
461  * If the client disconnects before transmission, the message is
462  * simply discarded.
463  *
464  * @param lc recipient
465  * @param msg message to transmit to client
466  */
467 void
468 GSF_local_client_transmit_ (struct GSF_LocalClient *lc,
469                             const struct GNUNET_MessageHeader *msg)
470 {
471   struct ClientResponse *res;
472   size_t msize;
473
474   msize = ntohs (msg->size);
475   res = GNUNET_malloc (sizeof (struct ClientResponse) + msize);
476   res->lc = lc;
477   res->msize = msize;
478   memcpy (&res[1], msg, msize);
479   GNUNET_CONTAINER_DLL_insert_tail (lc->res_head,
480                                     lc->res_tail,
481                                     res);
482   if (NULL == lc->th)
483     lc->th = GNUNET_SERVER_notify_transmit_ready (lc->client,
484                                                   msize,
485                                                   GNUNET_TIME_UNIT_FOREVER_REL,
486                                                   &transmit_to_client,
487                                                   lc);
488 }
489
490
491 /**
492  * A client disconnected from us.  Tear down the local client
493  * record.
494  *
495  * @param cls unused
496  * @param client handle of the client
497  */
498 void
499 GSF_client_disconnect_handler_ (void *cls,
500                                 struct GNUNET_SERVER_Client *client)
501 {
502   struct GSF_LocalClient *pos;
503   struct ClientRequest *cr;
504   struct ClientResponse *res;
505
506   pos = client_head;
507   while ( (pos != NULL) &&
508           (pos->client != client) )
509     pos = pos->next;
510   if (pos == NULL)
511     return;
512   while (NULL != (cr = pos->cr_head))
513     {      
514       GNUNET_CONTAINER_DLL_remove (pos->cr_head,
515                                    pos->cr_tail,
516                                    cr);
517       GSF_pending_request_cancel_ (cr->pr);
518       GNUNET_STATISTICS_update (GSF_stats,
519                                 gettext_noop ("# client searches active"),
520                                 - 1,
521                                 GNUNET_NO);
522       if (GNUNET_SCHEDULER_NO_TASK != cr->kill_task)
523         GNUNET_SCHEDULER_cancel (cr->kill_task);
524       GNUNET_free (cr);
525     }
526   while (NULL != (res = pos->res_head))
527     {
528       GNUNET_CONTAINER_DLL_remove (pos->res_head,
529                                    pos->res_tail,
530                                    res);
531       GNUNET_free (res);
532     }
533   if (pos->th != NULL)
534     {
535       GNUNET_CONNECTION_notify_transmit_ready_cancel (pos->th);
536       pos->th = NULL;
537     }
538   GSF_handle_local_client_disconnect_ (pos);
539   GNUNET_CONTAINER_DLL_remove (client_head,
540                                client_tail,
541                                pos);
542   GNUNET_free (pos);
543 }
544
545
546 /* end of gnunet-service-fs_lc.c */