fsui.c

GNUnet是一个安全的点对点网络框架

/*
     This file is part of GNUnet.
     (C) 2001, 2002, 2003, 2004, 2005, 2006, 2008 Christian Grothoff (and other contributing authors)

     GNUnet is free software; you can redistribute it and/or modify
     it under the terms of the GNU General Public License as published
     by the Free Software Foundation; either version 2, or (at your
     option) any later version.

     GNUnet is distributed in the hope that it will be useful, but
     WITHOUT ANY WARRANTY; without even the implied warranty of
     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
     General Public License for more details.

     You should have received a copy of the GNU General Public License
     along with GNUnet; see the file COPYING.  If not, write to the
     Free Software Foundation, Inc., 59 Temple Place - Suite 330,
     Boston, MA 02111-1307, USA.
*/

/**
 * @file applications/fs/fsui/fsui.c
 * @brief main FSUI functions
 * @author Christian Grothoff
 */

#include "platform.h"
#include "gnunet_fsui_lib.h"
#include "gnunet_directories.h"
#include "fsui.h"
#include "fs.h"

#define DEBUG_PERSISTENCE GNUNET_NO

/* ***************** CRON code ***************** */

/**
 * How often should cron run?
 */
#define GNUNET_FSUI_UDT_FREQUENCY (2 * GNUNET_CRON_SECONDS)

#define SQUARE(x) ((x)*(x))

/**
 * We made progress on a test download.  Since
 * a test download only contains a single block,
 * any progress means that the test succeeded.
 * We just set the flag to notify the cron in
 * the next iteration.
 */
static void
test_download_progress (unsigned long long totalBytes,
                        unsigned long long completedBytes,
                        GNUNET_CronTime eta,
                        unsigned long long lastBlockOffset,
                        const char *lastBlock, unsigned int lastBlockSize,
                        void *closure)
{
  struct SearchResultList *srl = closure;
  if (lastBlockSize > 0)        /* check against IBlock events */
    srl->test_download_start_time = 0;
}


/**
 * Cron job for download load management.
 */
static void
updateDownloadThreads (void *c)
{
  GNUNET_FSUI_Context *ctx = c;
  GNUNET_FSUI_DownloadList *dpos;
  struct SearchResultList *srl;
  struct GNUNET_FSUI_SearchList *sl;
  unsigned long long off;
  unsigned long long len;
  GNUNET_CronTime now;
  GNUNET_FSUI_Event event;

  GNUNET_mutex_lock (ctx->lock);
  dpos = ctx->activeDownloads.child;
#if DEBUG_PERSISTENCE
  if (dpos != NULL)
    GNUNET_GE_LOG (ctx->ectx,
                   GNUNET_GE_DEBUG | GNUNET_GE_REQUEST | GNUNET_GE_USER,
                   "Download thread manager schedules pending downloads...\n");
#endif
  ctx->next_min_block_resume = -1;
  while (dpos != NULL)
    {
      GNUNET_FSUI_updateDownloadThread (dpos);
      dpos = dpos->next;
    }
  ctx->min_block_resume = ctx->next_min_block_resume;
  now = GNUNET_get_time ();
  sl = ctx->activeSearches;
  while (sl != NULL)
    {
      srl = sl->resultsReceived;
      while (srl != NULL)
        {
          if (srl->test_download != NULL)
            {
              if (srl->test_download_start_time == 0)
                {
                  /* probe was successful, kill */
                  GNUNET_ECRS_file_download_partial_stop (srl->test_download);
                  srl->test_download = NULL;
                  srl->probeSuccess++;
                  event.type = GNUNET_FSUI_search_update;
                  event.data.SearchUpdate.sc.pos = sl;
                  event.data.SearchUpdate.sc.cctx = sl->cctx;
                  event.data.SearchUpdate.fi = srl->fi;
                  event.data.SearchUpdate.searchURI = sl->uri;
                  event.data.SearchUpdate.availability_rank =
                    srl->probeSuccess - srl->probeFailure;
                  event.data.SearchUpdate.availability_certainty =
                    srl->probeSuccess + srl->probeFailure;
                  event.data.SearchUpdate.applicability_rank =
                    srl->matchingSearchCount;
                  ctx->ecb (ctx->ecbClosure, &event);
                  ctx->active_probes--;
                  srl->last_probe_time = now;
                }
              else
                {
                  /* consider stopping */
                  if ((now - srl->test_download_start_time)
                      >
                      SQUARE (srl->probeSuccess + srl->probeFailure +
                              1) * GNUNET_FSUI_PROBE_TIME_FACTOR)
                    {
                      /* timeout hit! */
                      GNUNET_ECRS_file_download_partial_stop
                        (srl->test_download);
                      srl->test_download = NULL;
                      srl->probeFailure++;
                      event.type = GNUNET_FSUI_search_update;
                      event.data.SearchUpdate.sc.pos = sl;
                      event.data.SearchUpdate.sc.cctx = sl->cctx;
                      event.data.SearchUpdate.fi = srl->fi;
                      event.data.SearchUpdate.searchURI = sl->uri;
                      event.data.SearchUpdate.availability_rank =
                        srl->probeSuccess - srl->probeFailure;
                      event.data.SearchUpdate.availability_certainty =
                        srl->probeSuccess + srl->probeFailure;
                      event.data.SearchUpdate.applicability_rank =
                        srl->matchingSearchCount;
                      ctx->ecb (ctx->ecbClosure, &event);
                      ctx->active_probes--;
                      srl->last_probe_time = now;
                    }
                }
            }
          else
            {
              len = GNUNET_ECRS_uri_get_file_size (srl->fi.uri);
              if (len == 0)
                srl->probeSuccess = -1; /* MAX */
              /* consider starting */
              if (((srl->probeSuccess + srl->probeFailure) <
                   GNUNET_FSUI_MAX_PROBES)
                  &&
                  ((srl->last_probe_time <
                    now +
                    GNUNET_FSUI_PROBE_DELAY * SQUARE (ctx->active_probes) +
                    GNUNET_random_u64 (GNUNET_RANDOM_QUALITY_WEAK,
                                       GNUNET_FSUI_PROBE_DELAY)))
                  && (ctx->active_probes < GNUNET_FSUI_HARD_PROBE_LIMIT))
                {
                  off = len / GNUNET_ECRS_DBLOCK_SIZE;
                  if (off > 0)
                    off = GNUNET_random_u32 (GNUNET_RANDOM_QUALITY_WEAK, off);
                  off *= GNUNET_ECRS_DBLOCK_SIZE;
                  if (len - off < GNUNET_ECRS_DBLOCK_SIZE)
                    len = len - off;
                  else
                    len = GNUNET_ECRS_DBLOCK_SIZE;
                  srl->test_download
                    = GNUNET_ECRS_file_download_partial_start (ctx->ectx,
                                                               ctx->cfg,
                                                               sl->probe_context,
                                                               srl->fi.uri,
                                                               NULL, off, len,
                                                               1, GNUNET_YES,
                                                               &test_download_progress,
                                                               srl);
                  if (srl->test_download != NULL)
                    {
                      srl->test_download_start_time = now;
                      ctx->active_probes++;
                    }
                }
            }

          srl = srl->next;
        }
      sl = sl->next;
    }
  GNUNET_mutex_unlock (ctx->lock);
}

/* ******************* START code *********************** */

static void
signalDownloadResume (struct GNUNET_FSUI_DownloadList *ret,
                      GNUNET_FSUI_Context * ctx)
{
  GNUNET_FSUI_Event event;
  GNUNET_CronTime now;
  GNUNET_CronTime eta;

  while (ret != NULL)
    {
      event.type = GNUNET_FSUI_download_resumed;
      event.data.DownloadResumed.dc.pos = ret;
      event.data.DownloadResumed.dc.cctx = ret->cctx;
      event.data.DownloadResumed.dc.ppos =
        ret->parent == &ctx->activeDownloads ? NULL : ret->parent;
      event.data.DownloadResumed.dc.pcctx = ret->parent->cctx;
      event.data.DownloadResumed.dc.spos = ret->search;
      event.data.DownloadResumed.dc.sctx =
        ret->search == NULL ? NULL : ret->search->cctx;
      event.data.DownloadResumed.completed = ret->completed;
      event.data.DownloadResumed.total = ret->total;
      event.data.DownloadResumed.state = ret->state;
      now = GNUNET_get_time ();
      if ((ret->total == 0) || (ret->completed == 0))
        {
          eta = now;
        }
      else
        {
          eta = (GNUNET_CronTime) (now - ret->runTime +
                                   (((double) (ret->runTime) /
                                     (double) ret->completed)) *
                                   (double) ret->total);
          if (eta < now)
            eta = now;
        }
      event.data.DownloadResumed.eta = eta;
      event.data.DownloadResumed.filename = ret->filename;
      event.data.DownloadResumed.fi.uri = ret->fi.uri;
      event.data.DownloadResumed.fi.meta = ret->fi.meta;
      event.data.DownloadResumed.anonymityLevel = ret->anonymityLevel;
      ret->cctx = ctx->ecb (ctx->ecbClosure, &event);
      if (ret->child != NULL)
        signalDownloadResume (ret->child, ctx);
      ret = ret->next;
    }
}

static void
signalUploadResume (struct GNUNET_FSUI_UploadList *ret,
                    GNUNET_FSUI_Context * ctx)
{
  GNUNET_FSUI_Event event;
  GNUNET_CronTime now;
  GNUNET_CronTime eta;

  while (ret != NULL)
    {
      event.type = GNUNET_FSUI_upload_resumed;
      event.data.UploadResumed.uc.pos = ret;
      event.data.UploadResumed.uc.cctx = NULL;
      event.data.UploadResumed.uc.ppos = ret->parent;
      event.data.UploadResumed.uc.pcctx = ret->parent->cctx;
      event.data.UploadResumed.completed = ret->completed;
      event.data.UploadResumed.total = ret->total;
      event.data.UploadResumed.uri = ret->uri;
      event.data.UploadResumed.state = ret->state;
      now = GNUNET_get_time ();
      if ((ret->total == 0) || (ret->completed == 0))
        {
          eta = now;
        }
      else
        {
          eta = (GNUNET_CronTime) (ret->start_time +
                                   (((double) (now - ret->start_time) /
                                     (double) ret->completed)) *
                                   (double) ret->total);
          if (eta < now)
            eta = now;
        }
      event.data.UploadResumed.eta = eta;
      event.data.UploadResumed.anonymityLevel = ret->shared->anonymityLevel;
      event.data.UploadResumed.filename = ret->filename;
      ret->cctx = ctx->ecb (ctx->ecbClosure, &event);
      if (ret->child != NULL)
        signalUploadResume (ret->child, ctx);
      ret = ret->next;
    }
}

/**
 * Resume uploads.
 * Only re-starts top-level upload threads;
 * threads below are controlled by the parent.
 */
static void
doResumeUploads (struct GNUNET_FSUI_UploadList *ret,
                 GNUNET_FSUI_Context * ctx)
{
  while (ret != NULL)
    {
      if (ret->state == GNUNET_FSUI_ACTIVE)
        {
          ret->shared->handle =
            GNUNET_thread_create (&GNUNET_FSUI_uploadThread, ret, 128 * 1024);
          if (ret->shared->handle == NULL)
            GNUNET_GE_DIE_STRERROR (ctx->ectx,
                                    GNUNET_GE_FATAL | GNUNET_GE_ADMIN |
                                    GNUNET_GE_IMMEDIATE, "pthread_create");
        }
      ret = ret->next;
    }
}

/**
 * Start FSUI manager.  Use the given progress callback to notify the
 * UI about events.  Start processing pending activities that were
 * running when GNUNET_FSUI_stop was called previously.
 *
 * @param name name of the context, must not be NULL
 * @return NULL on error
 */
struct GNUNET_FSUI_Context *
GNUNET_FSUI_start (struct GNUNET_GE_Context *ectx,
                   struct GNUNET_GC_Configuration *cfg,
                   const char *name,
                   unsigned int threadPoolSize,
                   int doResume, GNUNET_FSUI_EventProcessor cb, void *closure)
{
  GNUNET_FSUI_Event event;
  GNUNET_FSUI_Context *ret;
  GNUNET_FSUI_SearchList *list;
  GNUNET_FSUI_UnindexList *xlist;
  struct SearchResultList *pos;
  struct SearchRecordList *rec;
  unsigned int valid;
  unsigned int i;
  GNUNET_ECRS_FileInfo *fis;
  int *av_ranks;
  unsigned int *av_certs;
  unsigned int *ap_ranks;
  char *fn;
  unsigned long long size;

  GNUNET_GE_ASSERT (ectx, cfg != NULL);
  ret = GNUNET_malloc (sizeof (GNUNET_FSUI_Context));
  memset (ret, 0, sizeof (GNUNET_FSUI_Context));
  ret->activeDownloads.state = GNUNET_FSUI_PENDING;     /* !? */
  ret->activeDownloads.ctx = ret;
  ret->cfg = cfg;
  ret->ecb = cb;
  ret->ecbClosure = closure;
  ret->threadPoolSize = threadPoolSize;
  if (ret->threadPoolSize == 0)
    ret->threadPoolSize = 32;
  ret->activeDownloadThreads = 0;
  ret->name = GNUNET_get_home_filename (ectx,
                                        cfg, GNUNET_NO, "fsui", name, NULL);
  /* 1) read state  in */
  if (doResume)
    {
      fn = GNUNET_get_home_filename (ectx,
                                     cfg,
                                     GNUNET_NO, "fsui-locks", name, NULL);
      ret->ipc = GNUNET_IPC_semaphore_create (ectx, fn, 1);
#if DEBUG_PERSISTENCE
      GNUNET_GE_LOG (ectx,
                     GNUNET_GE_INFO | GNUNET_GE_REQUEST | GNUNET_GE_USER,
                     "Getting IPC lock for FSUI (%s).\n", fn);
#endif
      GNUNET_IPC_semaphore_down (ret->ipc, GNUNET_YES);
#if DEBUG_PERSISTENCE
      GNUNET_GE_LOG (ectx,
                     GNUNET_GE_INFO | GNUNET_GE_REQUEST | GNUNET_GE_USER,
                     "Aquired IPC lock.\n");
#endif
      GNUNET_free (fn);
      GNUNET_FSUI_deserialize (ret);
    }
  else
    {
      ret->ipc = NULL;
    }
  ret->lock = GNUNET_mutex_create (GNUNET_YES);

  /* 2) do resume events */
  /* 2a) signal search restarts */
  list = ret->activeSearches;
  while (list != NULL)
    {
      valid = 0;
      pos = list->resultsReceived;
      while (pos != NULL)
        {
          if (pos->mandatoryMatchesRemaining == 0)
            valid++;
          pos = pos->next;
        }
      fis = NULL;
      av_ranks = NULL;
      av_certs = NULL;
      ap_ranks = NULL;
      if (valid > 0)
        {
          fis = GNUNET_malloc (sizeof (GNUNET_ECRS_FileInfo) * valid);
          av_ranks = GNUNET_malloc (sizeof (int) * valid);
          av_certs = GNUNET_malloc (sizeof (unsigned int) * valid);
          ap_ranks = GNUNET_malloc (sizeof (unsigned int) * valid);
          pos = list->resultsReceived;
          i = 0;
          while (pos != NULL)
            {
              if (pos->mandatoryMatchesRemaining == 0)
                {
                  fis[i] = pos->fi;
                  av_ranks[i] = pos->probeSuccess - pos->probeFailure;
                  av_certs[i] = pos->probeSuccess + pos->probeFailure;
                  ap_ranks[i] = pos->matchingSearchCount;
                  i++;
                }
              pos = pos->next;
            }
        }
      event.type = GNUNET_FSUI_search_resumed;
      event.data.SearchResumed.sc.pos = list;
      event.data.SearchResumed.sc.cctx = NULL;
      event.data.SearchResumed.fis = fis;
      event.data.SearchResumed.fisSize = valid;
      event.data.SearchResumed.anonymityLevel = list->anonymityLevel;