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_util_lib.h"
  29 #include "gnunet_fs_service.h"
  30 #include "fs.h"
  31
  32
  33 /**
  34  * Start the given job (send signal, remove from pending queue, update
  35  * counters and state).
  36  *
  37  * @param qe job to start
  38  */
  39 static void
  40 start_job (struct GNUNET_FS_QueueEntry *qe)
  41 {
  42   qe->client = GNUNET_CLIENT_connect (qe->h->sched, "fs", qe->h->cfg);
  43   if (qe->client == NULL)
  44     {
  45       GNUNET_break (0);
  46       return;
  47     }
  48   qe->start (qe->cls, qe->client);
  49   qe->start_times++;
  50   qe->h->active_blocks += qe->blocks;
  51   qe->start_time = GNUNET_TIME_absolute_get ();
  52   GNUNET_CONTAINER_DLL_remove (qe->h->pending_head,
  53                                qe->h->pending_tail,
  54                                qe);
  55   GNUNET_CONTAINER_DLL_insert_after (qe->h->running_head,
  56                                      qe->h->running_tail,
  57                                      qe->h->running_tail,
  58                                      qe);
  59 }
  60
  61
  62 /**
  63  * Stop the given job (send signal, remove from active queue, update
  64  * counters and state).
  65  *
  66  * @param qe job to stop
  67  */
  68 static void
  69 stop_job (struct GNUNET_FS_QueueEntry *qe)
  70 {
  71   qe->client = NULL;
  72   qe->stop (qe->cls);
  73   qe->h->active_downloads--;
  74   qe->h->active_blocks -= qe->blocks;
  75   qe->run_time = GNUNET_TIME_relative_add (qe->run_time,
  76                                            GNUNET_TIME_absolute_get_duration (qe->start_time));
  77   GNUNET_CONTAINER_DLL_remove (qe->h->running_head,
  78                                qe->h->running_tail,
  79                                qe);
  80   GNUNET_CONTAINER_DLL_insert_after (qe->h->pending_head,
  81                                      qe->h->pending_tail,
  82                                      qe->h->pending_tail,
  83                                      qe);
  84 }
  85
  86
  87 /**
  88  * Process the jobs in the job queue, possibly starting some
  89  * and stopping others.
  90  *
  91  * @param cls the 'struct GNUNET_FS_Handle'
  92  * @param tc scheduler context
  93  */
  94 static void
  95 process_job_queue (void *cls,
  96                    const struct GNUNET_SCHEDULER_TaskContext *tc)
  97 {
  98   struct GNUNET_FS_Handle *h = cls;
  99   struct GNUNET_FS_QueueEntry *qe;
 100   struct GNUNET_FS_QueueEntry *next;
 101   struct GNUNET_TIME_Relative run_time;
 102   struct GNUNET_TIME_Relative restart_at;
 103   struct GNUNET_TIME_Relative rst;
 104   struct GNUNET_TIME_Absolute end_time;
 105
 106   h->queue_job = GNUNET_SCHEDULER_NO_TASK;
 107   next = h->pending_head;
 108   while (NULL != (qe = next))
 109     {
 110       next = qe->next;
 111       if (h->running_head == NULL)
 112         {
 113           start_job (qe);
 114           continue;
 115         }
 116       if ( (qe->blocks + h->active_blocks <= h->max_parallel_requests) &&
 117            (h->active_downloads + 1 <= h->max_parallel_downloads) )
 118         {
 119           start_job (qe);
 120           continue;
 121         }
 122     }
 123   if (h->pending_head == NULL)
 124     return; /* no need to stop anything */
 125   restart_at = GNUNET_TIME_UNIT_FOREVER_REL;
 126   next = h->running_head;
 127   while (NULL != (qe = next))
 128     {
 129       next = qe->next;
 130       /* FIXME: might be faster/simpler to do this calculation only once
 131          when we start a job (OTOH, this would allow us to dynamically
 132          and easily adjust qe->blocks over time, given the right API...) */
 133       run_time = GNUNET_TIME_relative_multiply (h->avg_block_latency,
 134                                                 qe->blocks * qe->start_times);
 135       end_time = GNUNET_TIME_absolute_add (qe->start_time,
 136                                            run_time);
 137       rst = GNUNET_TIME_absolute_get_remaining (end_time);
 138       restart_at = GNUNET_TIME_relative_min (rst, restart_at);
 139       if (rst.value > 0)
 140         continue;
 141       stop_job (qe);
 142     }
 143   h->queue_job = GNUNET_SCHEDULER_add_delayed (h->sched,
 144                                                restart_at,
 145                                                &process_job_queue,
 146                                                h);
 147 }
 148
 149
 150 /**
 151  * Add a job to the queue.
 152  *
 153  * @param h handle to the overall FS state
 154  * @param start function to call to begin the job
 155  * @param stop function to call to pause the job, or on dequeue (if the job was running)
 156  * @param cls closure for start and stop
 157  * @param blocks number of blocks this jobs uses
 158  * @return queue handle
 159  */
 160 struct GNUNET_FS_QueueEntry *
 161 GNUNET_FS_queue_ (struct GNUNET_FS_Handle *h,
 162                   GNUNET_FS_QueueStart start,
 163                   GNUNET_FS_QueueStop stop,
 164                   void *cls,
 165                   unsigned int blocks)
 166 {
 167   struct GNUNET_FS_QueueEntry *qe;
 168
 169   qe = GNUNET_malloc (sizeof (struct GNUNET_FS_QueueEntry));
 170   qe->h = h;
 171   qe->start = start;
 172   qe->stop = stop;
 173   qe->cls = cls;
 174   qe->queue_time = GNUNET_TIME_absolute_get ();
 175   qe->blocks = blocks;
 176   GNUNET_CONTAINER_DLL_insert_after (h->pending_head,
 177                                      h->pending_tail,
 178                                      h->pending_tail,
 179                                      qe);
 180   if (h->queue_job != GNUNET_SCHEDULER_NO_TASK)
 181     GNUNET_SCHEDULER_cancel (h->sched,
 182                              h->queue_job);
 183   h->queue_job
 184     = GNUNET_SCHEDULER_add_now (h->sched,
 185                                 &process_job_queue,
 186                                 h);
 187   return qe;
 188 }
 189
 190
 191 /**
 192  * Dequeue a job from the queue.
 193  * @param qh handle for the job
 194  */
 195 void
 196 GNUNET_FS_dequeue_ (struct GNUNET_FS_QueueEntry *qh)
 197 {
 198   struct GNUNET_FS_Handle *h;
 199
 200   h = qh->h;
 201   if (qh->client != NULL)
 202     stop_job (qh);
 203   GNUNET_CONTAINER_DLL_remove (h->pending_head,
 204                                h->pending_tail,
 205                                qh);
 206   GNUNET_free (qh);
 207   if (h->queue_job != GNUNET_SCHEDULER_NO_TASK)
 208     GNUNET_SCHEDULER_cancel (h->sched,
 209                              h->queue_job);
 210   h->queue_job
 211     = GNUNET_SCHEDULER_add_now (h->sched,
 212                                 &process_job_queue,
 213                                 h);
 214 }
 215
 216
 217 /**
 218  * Return the full filename where we would store state information
 219  * (for serialization/deserialization).
 220  *
 221  * @param h master context
 222  * @param ext component of the path
 223  * @param ent entity identifier (or emtpy string for the directory)
 224  * @return NULL on error
 225  */
 226 static char *
 227 get_serialization_file_name (struct GNUNET_FS_Handle *h,
 228                              const char *ext,
 229                              const char *ent)
 230 {
 231   char *basename;
 232   char *ret;
 233   if (GNUNET_OK !=
 234       GNUNET_CONFIGURATION_get_value_filename (h->cfg,
 235                                                "fs",
 236                                                "STATE_DIR",
 237                                                &basename))
 238     return NULL;
 239   GNUNET_asprintf (&ret,
 240                    "%s%s%s-%s%s%s",
 241                    basename,
 242                    DIR_SEPARATOR_STR,
 243                    h->client_name,
 244                    ext,
 245                    DIR_SEPARATOR_STR,
 246                    ent);
 247   GNUNET_free (basename);
 248   return ret;
 249 }
 250
 251
 252 /**
 253  * Return a read handle for deserialization.
 254  *
 255  * @param h master context
 256  * @param ext component of the path
 257  * @param ent entity identifier (or emtpy string for the directory)
 258  * @return NULL on error
 259  */
 260 static struct GNUNET_BIO_ReadHandle *
 261 get_read_handle (struct GNUNET_FS_Handle *h,
 262                  const char *ext,
 263                  const char *ent)
 264 {
 265   char *fn;
 266   struct GNUNET_BIO_ReadHandle *ret;
 267
 268   fn = get_serialization_file_name (h, ext, ent);
 269   if (fn == NULL)
 270     return NULL;
 271   ret = GNUNET_BIO_read_open (fn);
 272   GNUNET_free (fn);
 273   return ret;
 274 }
 275
 276
 277 /**
 278  * Using the given serialization filename, try to deserialize
 279  * the file-information tree associated with it.
 280  *
 281  * @param h master context
 282  * @param filename name of the file (without directory) with
 283  *        the infromation
 284  * @return NULL on error
 285  */
 286 static struct GNUNET_FS_FileInformation *
 287 deserialize_fi_node (struct GNUNET_BIO_ReadHandle *rh)
 288 {
 289   return NULL;
 290 }
 291
 292
 293 /**
 294  * Using the given serialization filename, try to deserialize
 295  * the file-information tree associated with it.
 296  *
 297  * @param h master context
 298  * @param filename name of the file (without directory) with
 299  *        the infromation
 300  * @return NULL on error
 301  */
 302 static struct GNUNET_FS_FileInformation *
 303 deserialize_file_information (struct GNUNET_FS_Handle *h,
 304                               const char *filename)
 305 {
 306   struct GNUNET_FS_FileInformation *ret;
 307   struct GNUNET_BIO_ReadHandle *rh;
 308   char *emsg;
 309
 310   rh = get_read_handle (h, "publish-fi", filename);
 311   if (rh == NULL)
 312     return NULL;
 313   ret = deserialize_fi_node (rh);
 314   if (GNUNET_OK !=
 315       GNUNET_BIO_read_close (rh, &emsg))
 316     {
 317       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 318                   _("Failed to resume publishing information `%s': %s\n"),
 319                   filename,
 320                   emsg);
 321       GNUNET_free (emsg);
 322     }
 323   return ret;
 324 }
 325
 326
 327 /**
 328  * Find the entry in the file information struct where the
 329  * serialization filename matches the given name.
 330  *
 331  * @param pos file information to search
 332  * @param srch filename to search for
 333  * @return NULL if srch was not found in this subtree
 334  */
 335 static struct GNUNET_FS_FileInformation *
 336 find_file_position (struct GNUNET_FS_FileInformation *pos,
 337                     const char *srch)
 338 {
 339   struct GNUNET_FS_FileInformation *r;
 340
 341   while (pos != NULL)
 342     {
 343       if (0 == strcmp (srch,
 344                        pos->serialization_name))
 345         return pos;
 346       if (pos->is_directory)
 347         {
 348           r = find_file_position (pos->data.dir.entries,
 349                                   srch);
 350           if (r != NULL)
 351             return r;
 352         }
 353       pos = pos->next;
 354     }
 355   return NULL;
 356 }
 357
 358
 359 /**
 360  * Signal the FS's progress function that we are resuming
 361  * an upload.
 362  *
 363  * @param cls closure (of type "struct GNUNET_FS_PublishContext*")
 364  * @param fi the entry in the publish-structure
 365  * @param length length of the file or directory
 366  * @param meta metadata for the file or directory (can be modified)
 367  * @param uri pointer to the keywords that will be used for this entry (can be modified)
 368  * @param anonymity pointer to selected anonymity level (can be modified)
 369  * @param priority pointer to selected priority (can be modified)
 370  * @param expirationTime pointer to selected expiration time (can be modified)
 371  * @param client_info pointer to client context set upon creation (can be modified)
 372  * @return GNUNET_OK to continue (always)
 373  */
 374 static int
 375 fip_signal_resume(void *cls,
 376                   struct GNUNET_FS_FileInformation *fi,
 377                   uint64_t length,
 378                   struct GNUNET_CONTAINER_MetaData *meta,
 379                   struct GNUNET_FS_Uri **uri,
 380                   uint32_t *anonymity,
 381                   uint32_t *priority,
 382                   struct GNUNET_TIME_Absolute *expirationTime,
 383                   void **client_info)
 384 {
 385   struct GNUNET_FS_PublishContext *sc = cls;
 386   struct GNUNET_FS_ProgressInfo pi;
 387
 388   pi.status = GNUNET_FS_STATUS_PUBLISH_RESUME;
 389   pi.value.publish.specifics.resume.message = sc->fi->emsg;
 390   pi.value.publish.specifics.resume.chk_uri = sc->fi->chk_uri;
 391   *client_info = GNUNET_FS_publish_make_status_ (&pi, sc, fi, 0);
 392   return GNUNET_OK;
 393 }
 394
 395
 396 /**
 397  * Function called with a filename of serialized publishing operation
 398  * to deserialize.
 399  *
 400  * @param cls the 'struct GNUNET_FS_Handle*'
 401  * @param filename complete filename (absolute path)
 402  * @return GNUNET_OK (continue to iterate)
 403  */
 404 static int
 405 deserialize_publish_file (void *cls,
 406                           const char *filename)
 407 {
 408   struct GNUNET_FS_Handle *h = cls;
 409   struct GNUNET_BIO_ReadHandle *rh;
 410   struct GNUNET_FS_PublishContext *pc;
 411   int32_t options;
 412   int32_t all_done;
 413   char *fi_root;
 414   char *ns;
 415   char *fi_pos;
 416   char *emsg;
 417
 418   rh = GNUNET_BIO_read_open (filename);
 419   if (rh == NULL)
 420     {
 421       if (0 != UNLINK (filename))
 422         GNUNET_log_strerror_file (GNUNET_ERROR_TYPE_WARNING,
 423                                   "unlink",
 424                                   filename);
 425       return GNUNET_OK;
 426     }
 427   while (1)
 428     {
 429       fi_root = NULL;
 430       fi_pos = NULL;
 431       ns = NULL;
 432       pc = GNUNET_malloc (sizeof (struct GNUNET_FS_PublishContext));
 433       pc->h = h;
 434       if ( (GNUNET_OK !=
 435             GNUNET_BIO_read_string (rh, "publish-nid", &pc->nid, 1024)) ||
 436            (GNUNET_OK !=
 437             GNUNET_BIO_read_string (rh, "publish-nuid", &pc->nuid, 1024)) ||
 438            (GNUNET_OK !=
 439             GNUNET_BIO_read_int32 (rh, &options)) ||
 440            (GNUNET_OK !=
 441             GNUNET_BIO_read_int32 (rh, &all_done)) ||
 442            (GNUNET_OK !=
 443             GNUNET_BIO_read_string (rh, "publish-firoot", &fi_root, 128)) ||
 444            (GNUNET_OK !=
 445             GNUNET_BIO_read_string (rh, "publish-fipos", &fi_pos, 128)) ||
 446            (GNUNET_OK !=
 447             GNUNET_BIO_read_string (rh, "publish-ns", &ns, 1024)) )
 448         {
 449           GNUNET_free_non_null (pc->nid);
 450           GNUNET_free_non_null (pc->nuid);
 451           GNUNET_free_non_null (fi_root);
 452           GNUNET_free_non_null (ns);
 453           GNUNET_free (pc);
 454           break;
 455         }
 456        pc->options = options;
 457        pc->all_done = all_done;
 458        pc->fi = deserialize_file_information (h, fi_root);
 459        if (pc->fi == NULL)
 460          {
 461            GNUNET_free_non_null (pc->nid);
 462            GNUNET_free_non_null (pc->nuid);
 463            GNUNET_free_non_null (fi_root);
 464            GNUNET_free_non_null (ns);
 465            GNUNET_free (pc);
 466            continue;
 467          }
 468        if (ns != NULL)
 469          {
 470            pc->namespace = GNUNET_FS_namespace_create (h, ns);
 471            if (pc->namespace == NULL)
 472              {
 473                GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 474                            _("Failed to recover namespace `%s', cannot resume publishing operation.\n"),
 475                            ns);
 476                GNUNET_free_non_null (pc->nid);
 477                GNUNET_free_non_null (pc->nuid);
 478                GNUNET_free_non_null (fi_root);
 479                GNUNET_free_non_null (ns);
 480                GNUNET_free (pc);
 481                continue;
 482              }
 483          }
 484        if (fi_pos != NULL)
 485          {
 486            pc->fi_pos = find_file_position (pc->fi,
 487                                             fi_pos);
 488            GNUNET_free (fi_pos);
 489            if (pc->fi_pos == NULL)
 490              {
 491                /* failed to find position for resuming, outch! Will start from root! */
 492                GNUNET_break (0);
 493                if (pc->all_done != GNUNET_YES)
 494                  pc->fi_pos = pc->fi;
 495              }
 496          }
 497        pc->serialization = GNUNET_strdup (filename);
 498        /* generate RESUME event(s) */
 499        GNUNET_FS_file_information_inspect (pc->fi,
 500                                            &fip_signal_resume,
 501                                            pc);
 502
 503        /* re-start publishing (if needed)... */
 504        if (pc->all_done != GNUNET_YES)
 505          pc->upload_task
 506            = GNUNET_SCHEDULER_add_with_priority (h->sched,
 507                                                  GNUNET_SCHEDULER_PRIORITY_BACKGROUND,
 508                                                  &GNUNET_FS_publish_main_,
 509                                                  pc);
 510     }
 511   if (GNUNET_OK !=
 512       GNUNET_BIO_read_close (rh, &emsg))
 513     {
 514       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 515                   _("Failed to resume publishing operation `%s': %s\n"),
 516                   filename,
 517                   emsg);
 518       GNUNET_free (emsg);
 519     }
 520   return GNUNET_OK;
 521 }
 522
 523
 524 /**
 525  * Deserialize information about pending publish operations.
 526  *
 527  * @param h master context
 528  */
 529 static void
 530 deserialize_publish (struct GNUNET_FS_Handle *h)
 531 {
 532   char *dn;
 533
 534   dn = get_serialization_file_name (h, "publish", "");
 535   if (dn == NULL)
 536     return;
 537   GNUNET_DISK_directory_scan (dn, &deserialize_publish_file, h);
 538   GNUNET_free (dn);
 539 }
 540
 541
 542 /**
 543  * Setup a connection to the file-sharing service.
 544  *
 545  * @param sched scheduler to use
 546  * @param cfg configuration to use
 547  * @param client_name unique identifier for this client
 548  * @param upcb function to call to notify about FS actions
 549  * @param upcb_cls closure for upcb
 550  * @param flags specific attributes for fs-operations
 551  * @param ... list of optional options, terminated with GNUNET_FS_OPTIONS_END
 552  * @return NULL on error
 553  */
 554 struct GNUNET_FS_Handle *
 555 GNUNET_FS_start (struct GNUNET_SCHEDULER_Handle *sched,
 556                  const struct GNUNET_CONFIGURATION_Handle *cfg,
 557                  const char *client_name,
 558                  GNUNET_FS_ProgressCallback upcb,
 559                  void *upcb_cls,
 560                  enum GNUNET_FS_Flags flags,
 561                  ...)
 562 {
 563   struct GNUNET_FS_Handle *ret;
 564   struct GNUNET_CLIENT_Connection *client;
 565   enum GNUNET_FS_OPTIONS opt;
 566   va_list ap;
 567
 568   client = GNUNET_CLIENT_connect (sched,
 569                                   "fs",
 570                                   cfg);
 571   if (NULL == client)
 572     return NULL;
 573   ret = GNUNET_malloc (sizeof (struct GNUNET_FS_Handle));
 574   ret->sched = sched;
 575   ret->cfg = cfg;
 576   ret->client_name = GNUNET_strdup (client_name);
 577   ret->upcb = upcb;
 578   ret->upcb_cls = upcb_cls;
 579   ret->client = client;
 580   ret->flags = flags;
 581   ret->max_parallel_downloads = 1;
 582   ret->max_parallel_requests = 1;
 583   ret->avg_block_latency = GNUNET_TIME_UNIT_MINUTES; /* conservative starting point */
 584   va_start (ap, flags);
 585   while (GNUNET_FS_OPTIONS_END != (opt = va_arg (ap, enum GNUNET_FS_OPTIONS)))
 586     {
 587       switch (opt)
 588         {
 589         case GNUNET_FS_OPTIONS_DOWNLOAD_PARALLELISM:
 590           ret->max_parallel_downloads = va_arg (ap, unsigned int);
 591           break;
 592         case GNUNET_FS_OPTIONS_REQUEST_PARALLELISM:
 593           ret->max_parallel_requests = va_arg (ap, unsigned int);
 594           break;
 595         default:
 596           GNUNET_break (0);
 597           GNUNET_free (ret->client_name);
 598           GNUNET_free (ret);
 599           va_end (ap);
 600           return NULL;
 601         }
 602     }
 603   va_end (ap);
 604   // FIXME: setup receive-loop with client (do we need one?)
 605
 606   if (0 != (GNUNET_FS_FLAGS_PERSISTENCE & flags))
 607     {
 608       deserialize_publish (ret);
 609       /* FIXME: not implemented! */
 610       // Deserialize Search:
 611       // * read search queries
 612       // * for each query, read file with search results
 613       // * for each search result with active download, deserialize download
 614       // * for each directory search result, check for active downloads of contents
 615       // Deserialize Download:
 616       // * always part of search???
 617       // Deserialize Unindex:
 618       // * read FNs for unindex with progress offset
 619     }
 620   return ret;
 621 }
 622
 623
 624 /**
 625  * Close our connection with the file-sharing service.
 626  * The callback given to GNUNET_FS_start will no longer be
 627  * called after this function returns.
 628  *
 629  * @param h handle that was returned from GNUNET_FS_start
 630  */
 631 void
 632 GNUNET_FS_stop (struct GNUNET_FS_Handle *h)
 633 {
 634   if (0 != (GNUNET_FS_FLAGS_PERSISTENCE & h->flags))
 635     {
 636       // FIXME: generate SUSPEND events and clean up state!
 637     }
 638   // FIXME: terminate receive-loop with client  (do we need one?)
 639   if (h->queue_job != GNUNET_SCHEDULER_NO_TASK)
 640     GNUNET_SCHEDULER_cancel (h->sched,
 641                              h->queue_job);
 642   GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
 643   GNUNET_free (h->client_name);
 644   GNUNET_free (h);
 645 }
 646
 647
 648 /* end of fs.c */