src/fs/fs.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2001, 2002, 2003, 2004, 2005, 2006, 2008, 2009, 2010 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 2, 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/fs.c
  23  * @brief main FS functions
  24  * @author Christian Grothoff
  25  */
  26
  27 #include "platform.h"
  28 #include "gnunet_fs_service.h"
  29 #include "fs.h"
  30
  31
  32 /**
  33  * Start the given job (send signal, remove from pending queue, update
  34  * counters and state).
  35  *
  36  * @param qe job to start
  37  */
  38 static void
  39 start_job (struct GNUNET_FS_QueueEntry *qe)
  40 {
  41   qe->client = GNUNET_CLIENT_connect (qe->h->sched, "fs", qe->h->cfg);
  42   if (qe->client == NULL)
  43     {
  44       GNUNET_break (0);
  45       return;
  46     }
  47   qe->start (qe->cls, qe->client);
  48   qe->start_times++;
  49   qe->h->active_blocks += qe->blocks;
  50   qe->start_time = GNUNET_TIME_absolute_get ();
  51   GNUNET_CONTAINER_DLL_remove (qe->h->pending_head,
  52                                qe->h->pending_tail,
  53                                qe);
  54   GNUNET_CONTAINER_DLL_insert_after (qe->h->running_head,
  55                                      qe->h->running_tail,
  56                                      qe->h->running_tail,
  57                                      qe);
  58 }
  59
  60
  61 /**
  62  * Stop the given job (send signal, remove from active queue, update
  63  * counters and state).
  64  *
  65  * @param qe job to stop
  66  */
  67 static void
  68 stop_job (struct GNUNET_FS_QueueEntry *qe)
  69 {
  70   qe->client = NULL;
  71   qe->stop (qe->cls);
  72   qe->h->active_downloads--;
  73   qe->h->active_blocks -= qe->blocks;
  74   qe->run_time = GNUNET_TIME_relative_add (qe->run_time,
  75                                            GNUNET_TIME_absolute_get_duration (qe->start_time));
  76   GNUNET_CONTAINER_DLL_remove (qe->h->running_head,
  77                                qe->h->running_tail,
  78                                qe);
  79   GNUNET_CONTAINER_DLL_insert_after (qe->h->pending_head,
  80                                      qe->h->pending_tail,
  81                                      qe->h->pending_tail,
  82                                      qe);
  83 }
  84
  85
  86 /**
  87  * Process the jobs in the job queue, possibly starting some
  88  * and stopping others.
  89  *
  90  * @param cls the 'struct GNUNET_FS_Handle'
  91  * @param tc scheduler context
  92  */
  93 static void
  94 process_job_queue (void *cls,
  95                    const struct GNUNET_SCHEDULER_TaskContext *tc)
  96 {
  97   struct GNUNET_FS_Handle *h = cls;
  98   struct GNUNET_FS_QueueEntry *qe;
  99   struct GNUNET_FS_QueueEntry *next;
 100   struct GNUNET_TIME_Relative run_time;
 101   struct GNUNET_TIME_Relative restart_at;
 102   struct GNUNET_TIME_Relative rst;
 103   struct GNUNET_TIME_Absolute end_time;
 104
 105   h->queue_job = GNUNET_SCHEDULER_NO_TASK;
 106   next = h->pending_head;
 107   while (NULL != (qe = next))
 108     {
 109       next = qe->next;
 110       if (h->running_head == NULL)
 111         {
 112           start_job (qe);
 113           continue;
 114         }
 115       if ( (qe->blocks + h->active_blocks <= h->max_parallel_requests) &&
 116            (h->active_downloads + 1 <= h->max_parallel_downloads) )
 117         {
 118           start_job (qe);
 119           continue;
 120         }
 121     }
 122   if (h->pending_head == NULL)
 123     return; /* no need to stop anything */
 124   restart_at = GNUNET_TIME_UNIT_FOREVER_REL;
 125   next = h->running_head;
 126   while (NULL != (qe = next))
 127     {
 128       next = qe->next;
 129       /* FIXME: might be faster/simpler to do this calculation only once
 130          when we start a job (OTOH, this would allow us to dynamically
 131          and easily adjust qe->blocks over time, given the right API...) */
 132       run_time = GNUNET_TIME_relative_multiply (h->avg_block_latency,
 133                                                 qe->blocks * qe->start_times);
 134       end_time = GNUNET_TIME_absolute_add (qe->start_time,
 135                                            run_time);
 136       rst = GNUNET_TIME_absolute_get_remaining (end_time);
 137       restart_at = GNUNET_TIME_relative_min (rst, restart_at);
 138       if (rst.value > 0)
 139         continue;
 140       stop_job (qe);
 141     }
 142   h->queue_job = GNUNET_SCHEDULER_add_delayed (h->sched,
 143                                                restart_at,
 144                                                &process_job_queue,
 145                                                h);
 146 }
 147
 148 /**
 149  * Add a job to the queue.
 150  *
 151  * @param h handle to the overall FS state
 152  * @param start function to call to begin the job
 153  * @param stop function to call to pause the job, or on dequeue (if the job was running)
 154  * @param cls closure for start and stop
 155  * @param blocks number of blocks this jobs uses
 156  * @return queue handle
 157  */
 158 struct GNUNET_FS_QueueEntry *
 159 GNUNET_FS_queue_ (struct GNUNET_FS_Handle *h,
 160                   GNUNET_FS_QueueStart start,
 161                   GNUNET_FS_QueueStop stop,
 162                   void *cls,
 163                   unsigned int blocks)
 164 {
 165   struct GNUNET_FS_QueueEntry *qe;
 166
 167   qe = GNUNET_malloc (sizeof (struct GNUNET_FS_QueueEntry));
 168   qe->h = h;
 169   qe->start = start;
 170   qe->stop = stop;
 171   qe->cls = cls;
 172   qe->queue_time = GNUNET_TIME_absolute_get ();
 173   qe->blocks = blocks;
 174   GNUNET_CONTAINER_DLL_insert_after (h->pending_head,
 175                                      h->pending_tail,
 176                                      h->pending_tail,
 177                                      qe);
 178   if (h->queue_job != GNUNET_SCHEDULER_NO_TASK)
 179     GNUNET_SCHEDULER_cancel (h->sched,
 180                              h->queue_job);
 181   h->queue_job
 182     = GNUNET_SCHEDULER_add_now (h->sched,
 183                                 &process_job_queue,
 184                                 h);
 185   return qe;
 186 }
 187
 188
 189 /**
 190  * Dequeue a job from the queue.
 191  * @param qh handle for the job
 192  */
 193 void
 194 GNUNET_FS_dequeue_ (struct GNUNET_FS_QueueEntry *qh)
 195 {
 196   struct GNUNET_FS_Handle *h;
 197
 198   h = qh->h;
 199   if (qh->client != NULL)
 200     stop_job (qh);
 201   GNUNET_CONTAINER_DLL_remove (h->pending_head,
 202                                h->pending_tail,
 203                                qh);
 204   GNUNET_free (qh);
 205   if (h->queue_job != GNUNET_SCHEDULER_NO_TASK)
 206     GNUNET_SCHEDULER_cancel (h->sched,
 207                              h->queue_job);
 208   h->queue_job
 209     = GNUNET_SCHEDULER_add_now (h->sched,
 210                                 &process_job_queue,
 211                                 h);
 212 }
 213
 214
 215 /**
 216  * Setup a connection to the file-sharing service.
 217  *
 218  * @param sched scheduler to use
 219  * @param cfg configuration to use
 220  * @param client_name unique identifier for this client
 221  * @param upcb function to call to notify about FS actions
 222  * @param upcb_cls closure for upcb
 223  * @param flags specific attributes for fs-operations
 224  * @param ... list of optional options, terminated with GNUNET_FS_OPTIONS_END
 225  * @return NULL on error
 226  */
 227 struct GNUNET_FS_Handle *
 228 GNUNET_FS_start (struct GNUNET_SCHEDULER_Handle *sched,
 229                  const struct GNUNET_CONFIGURATION_Handle *cfg,
 230                  const char *client_name,
 231                  GNUNET_FS_ProgressCallback upcb,
 232                  void *upcb_cls,
 233                  enum GNUNET_FS_Flags flags,
 234                  ...)
 235 {
 236   struct GNUNET_FS_Handle *ret;
 237   struct GNUNET_CLIENT_Connection *client;
 238   enum GNUNET_FS_OPTIONS opt;
 239   va_list ap;
 240
 241   client = GNUNET_CLIENT_connect (sched,
 242                                   "fs",
 243                                   cfg);
 244   if (NULL == client)
 245     return NULL;
 246   ret = GNUNET_malloc (sizeof (struct GNUNET_FS_Handle));
 247   ret->sched = sched;
 248   ret->cfg = cfg;
 249   ret->client_name = GNUNET_strdup (client_name);
 250   ret->upcb = upcb;
 251   ret->upcb_cls = upcb_cls;
 252   ret->client = client;
 253   ret->flags = flags;
 254   ret->max_parallel_downloads = 1;
 255   ret->max_parallel_requests = 1;
 256   ret->avg_block_latency = GNUNET_TIME_UNIT_MINUTES; /* conservative starting point */
 257   va_start (ap, flags);
 258   while (GNUNET_FS_OPTIONS_END != (opt = va_arg (ap, enum GNUNET_FS_OPTIONS)))
 259     {
 260       switch (opt)
 261         {
 262         case GNUNET_FS_OPTIONS_DOWNLOAD_PARALLELISM:
 263           ret->max_parallel_downloads = va_arg (ap, unsigned int);
 264           break;
 265         case GNUNET_FS_OPTIONS_REQUEST_PARALLELISM:
 266           ret->max_parallel_requests = va_arg (ap, unsigned int);
 267           break;
 268         default:
 269           GNUNET_break (0);
 270           GNUNET_free (ret->client_name);
 271           GNUNET_free (ret);
 272           va_end (ap);
 273           return NULL;
 274         }
 275     }
 276   va_end (ap);
 277   // FIXME: setup receive-loop with client
 278
 279   // FIXME: deserialize state; use client-name to find master-directory!
 280   // Deserialize-Upload:
 281   // * read FNs for upload FIs, deserialize each
 282   // Deserialize Search:
 283   // * read search queries
 284   // * for each query, read file with search results
 285   // * for each search result with active download, deserialize download
 286   // * for each directory search result, check for active downloads of contents
 287   // Deserialize Download:
 288   // * always part of search???
 289   // Deserialize Unindex:
 290   // * read FNs for unindex with progress offset
 291   return ret;
 292 }
 293
 294
 295 /**
 296  * Close our connection with the file-sharing service.
 297  * The callback given to GNUNET_FS_start will no longer be
 298  * called after this function returns.
 299  *
 300  * @param h handle that was returned from GNUNET_FS_start
 301  */
 302 void
 303 GNUNET_FS_stop (struct GNUNET_FS_Handle *h)
 304 {
 305   // FIXME: serialize state!? (or is it always serialized???)
 306   // FIXME: terminate receive-loop with client
 307   if (h->queue_job != GNUNET_SCHEDULER_NO_TASK)
 308     GNUNET_SCHEDULER_cancel (h->sched,
 309                              h->queue_job);
 310   GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
 311   GNUNET_free (h->client_name);
 312   GNUNET_free (h);
 313 }
 314
 315
 316 /* end of fs.c */