cron.c

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

#if DEBUG_CRON
  GNUNET_GE_LOG (NULL,
                 GNUNET_GE_STATUS | GNUNET_GE_DEVELOPER | GNUNET_GE_BULK,
                 "Advancing job %p-%p\n", method, data);
#endif
  GNUNET_mutex_lock (cron->deltaListLock_);
  jobId = cron->firstUsed_;
  if (jobId == -1)
    {
      /* not in queue - do nothing! */
      GNUNET_mutex_unlock (cron->deltaListLock_);
      return;
    }
  last = NULL;
  job = &cron->deltaList_[jobId];
  while ((job->method != method) ||
         (job->data != data) || (job->deltaRepeat != deltaRepeat))
    {
      last = job;
      if (job->next == -1)
        {
          /* not in queue; add if not running */
          if ((method != cron->runningJob_) ||
              (data != cron->runningData_) ||
              (deltaRepeat != cron->runningRepeat_))
            {
              GNUNET_cron_add_job (cron, method, 0, deltaRepeat, data);
            }
          GNUNET_mutex_unlock (cron->deltaListLock_);
          return;
        }
      jobId = job->next;
      job = &cron->deltaList_[jobId];
    }
  /* ok, found it; remove, re-add with time 0 */
  GNUNET_cron_del_job (cron, method, deltaRepeat, data);
  GNUNET_cron_add_job (cron, method, 0, deltaRepeat, data);
  GNUNET_mutex_unlock (cron->deltaListLock_);
}

void
GNUNET_cron_add_job (struct GNUNET_CronManager *cron,
                     GNUNET_CronJob method,
                     unsigned int delta, unsigned int deltaRepeat, void *data)
{
  UTIL_cron_DeltaListEntry *entry;
  UTIL_cron_DeltaListEntry *pos;
  int last;
  int current;

#if DEBUG_CRON
  GNUNET_GE_LOG (cron->ectx,
                 GNUNET_GE_STATUS | GNUNET_GE_DEVELOPER | GNUNET_GE_BULK,
                 "Adding job %p-%p to fire in %d CU\n", method, data, delta);
#endif

  GNUNET_mutex_lock (cron->deltaListLock_);
  if (cron->firstFree_ == -1)
    {                           /* need to grow */
      unsigned int i;

      GNUNET_array_grow (cron->deltaList_, cron->deltaListSize_,
                         cron->deltaListSize_ * 2);
      for (i = cron->deltaListSize_ / 2; i < cron->deltaListSize_; i++)
        cron->deltaList_[i].next = i - 1;
      cron->deltaList_[cron->deltaListSize_ / 2].next = -1;
      cron->firstFree_ = cron->deltaListSize_ - 1;
    }
  entry = &cron->deltaList_[cron->firstFree_];
  entry->method = method;
  entry->data = data;
  entry->deltaRepeat = deltaRepeat;
  entry->delta = GNUNET_get_time () + delta;
  if (cron->firstUsed_ == -1)
    {
      cron->firstUsed_ = cron->firstFree_;
      cron->firstFree_ = entry->next;
      entry->next = -1;         /* end of list */
      GNUNET_mutex_unlock (cron->deltaListLock_);
      /* interrupt sleeping cron-thread! */
      abortSleep (cron);
      return;
    }
  /* no, there are jobs waiting */
  last = -1;
  current = cron->firstUsed_;
  pos = &cron->deltaList_[current];

  while (entry->delta > pos->delta)
    {
      if (pos->next != -1)
        {
          last = current;
          current = pos->next;
          pos = &cron->deltaList_[current];
        }
      else
        {                       /* append */
          pos->next = cron->firstFree_;
          cron->firstFree_ = entry->next;
          entry->next = -1;
          GNUNET_mutex_unlock (cron->deltaListLock_);
#if HAVE_PRINT_CRON_TAB
          printCronTab ();
#endif
          return;
        }
    }
  /* insert before pos */
  if (last == -1)
    {
      cron->firstUsed_ = cron->firstFree_;
      abortSleep (cron);
    }
  else
    {
      cron->deltaList_[last].next = cron->firstFree_;
#if HAVE_PRINT_CRON_TAB
      printCronTab ();
#endif
    }
  cron->firstFree_ = entry->next;
  entry->next = current;
  GNUNET_mutex_unlock (cron->deltaListLock_);
}

/**
 * Process the cron-job at the beginning of the waiting queue, that
 * is, remove, invoke, and re-insert if it is a periodical job. Make
 * sure the cron job is held when calling this method, but
 * note that it will be released briefly for the time
 * where the job is running (the job to run may add other
 * jobs!)
 */
static void
runJob (struct GNUNET_CronManager *cron)
{
  UTIL_cron_DeltaListEntry *job;
  int jobId;
  GNUNET_CronJob method;
  void *data;
  unsigned int repeat;

  jobId = cron->firstUsed_;
  if (jobId == -1)
    return;                     /* no job to be done */
  job = &cron->deltaList_[jobId];
  method = job->method;
  cron->runningJob_ = method;
  data = job->data;
  cron->runningData_ = data;
  repeat = job->deltaRepeat;
  cron->runningRepeat_ = repeat;
  /* remove from queue */
  cron->firstUsed_ = job->next;
  job->next = cron->firstFree_;
  cron->firstFree_ = jobId;
  GNUNET_mutex_unlock (cron->deltaListLock_);
  /* re-insert */
  if (repeat > 0)
    {
#if DEBUG_CRON
      GNUNET_GE_LOG (cron->ectx,
                     GNUNET_GE_STATUS | GNUNET_GE_DEVELOPER | GNUNET_GE_BULK,
                     "adding periodic job %p-%p to run again in %u\n",
                     method, data, repeat);
#endif
      GNUNET_cron_add_job (cron, method, repeat, repeat, data);
    }
  /* run */
#if DEBUG_CRON
  GNUNET_GE_LOG (cron->ectx,
                 GNUNET_GE_STATUS | GNUNET_GE_DEVELOPER | GNUNET_GE_BULK,
                 "running job %p-%p\n", method, data);
#endif
  method (data);
  GNUNET_mutex_lock (cron->deltaListLock_);
  cron->runningJob_ = NULL;
#if DEBUG_CRON
  GNUNET_GE_LOG (cron->ectx,
                 GNUNET_GE_STATUS | GNUNET_GE_DEVELOPER | GNUNET_GE_BULK,
                 "job %p-%p done\n", method, data);
#endif
}

/**
 * The main-method of cron.
 */
static void *
cron_main_method (void *ctx)
{
  struct GNUNET_CronManager *cron = ctx;
  GNUNET_CronTime now;
  GNUNET_CronTime next;

  while (cron->cron_shutdown == GNUNET_NO)
    {
#if HAVE_PRINT_CRON_TAB
      printCronTab (cron);
#endif
      now = GNUNET_get_time ();
      next = now + 0xFFFFFFFF;
      GNUNET_mutex_lock (cron->deltaListLock_);
      while ((cron->cron_shutdown == GNUNET_NO) && (cron->firstUsed_ != -1))
        {
          now = GNUNET_get_time ();
          next = cron->deltaList_[cron->firstUsed_].delta;
          if (next <= now)
            {
#if DEBUG_CRON
              GNUNET_GE_LOG (cron->ectx,
                             GNUNET_GE_STATUS | GNUNET_GE_DEVELOPER |
                             GNUNET_GE_BULK, "running cron job, table is\n");
              printCronTab (cron);
#endif
              runJob (cron);
#if DEBUG_CRON
              GNUNET_GE_LOG (cron->ectx,
                             GNUNET_GE_STATUS | GNUNET_GE_DEVELOPER |
                             GNUNET_GE_BULK, "job run, new table is\n");
              printCronTab (cron);
#endif
            }
          else
            break;
        }
      GNUNET_mutex_unlock (cron->deltaListLock_);
      next = next - now;        /* how long to sleep */
#if DEBUG_CRON
      GNUNET_GE_LOG (cron->ectx,
                     GNUNET_GE_STATUS | GNUNET_GE_DEVELOPER | GNUNET_GE_BULK,
                     "Sleeping at %llu for %llu CU (%llu s, %llu CU)\n",
                     now, next, next / GNUNET_CRON_SECONDS, next);
#endif
      if (next > MAXSLEEP)
        next = MAXSLEEP;
      if (cron->cron_shutdown == GNUNET_NO)
        GNUNET_thread_sleep (next);
#if DEBUG_CRON
      GNUNET_GE_LOG (cron->ectx,
                     GNUNET_GE_STATUS | GNUNET_GE_DEVELOPER | GNUNET_GE_BULK,
                     "woke up at  %llu - %lld CS late\n",
                     GNUNET_get_time (), GNUNET_get_time () - (now + next));
#endif
    }
  GNUNET_semaphore_up (cron->cron_signal);
#if DEBUG_CRON
  GNUNET_GE_LOG (cron->ectx,
                 GNUNET_GE_STATUS | GNUNET_GE_DEVELOPER | GNUNET_GE_BULK,
                 "Cron thread exits.\n");
  printCronTab (cron);
#endif
  return NULL;
}


/**
 * Make sure to call stopCron before calling this method!
 */
void
GNUNET_cron_destroy (struct GNUNET_CronManager *cron)
{
  int i;

  GNUNET_GE_ASSERT (cron->ectx, cron->cron_signal == NULL);
  i = cron->firstUsed_;
  while (i != -1)
    {
      GNUNET_free_non_null (cron->deltaList_[i].data);
      i = cron->deltaList_[i].next;
    }
  GNUNET_mutex_destroy (cron->deltaListLock_);
  GNUNET_mutex_destroy (cron->inBlockLock_);
  GNUNET_free (cron->deltaList_);
  GNUNET_semaphore_destroy (cron->cron_signal_up);
  GNUNET_free (cron);
}

/**
 * Start the cron jobs.
 */
void
GNUNET_cron_start (struct GNUNET_CronManager *cron)
{
  GNUNET_GE_ASSERT (cron->ectx, cron->cron_signal == NULL);
  cron->cron_shutdown = GNUNET_NO;
  cron->cron_signal = GNUNET_semaphore_create (0);
  /* large stack, we don't know for sure
     what the cron jobs may be doing */
  cron->cron_handle =
    GNUNET_thread_create (&cron_main_method, cron, 256 * 1024);
  if (cron->cron_handle == 0)
    GNUNET_GE_DIE_STRERROR (cron->ectx,
                            GNUNET_GE_FATAL | GNUNET_GE_ADMIN | GNUNET_GE_USER
                            | GNUNET_GE_BULK, "pthread_create");
}

int
GNUNET_cron_del_job (struct GNUNET_CronManager *cron,
                     GNUNET_CronJob method, unsigned int repeat, void *data)
{
  UTIL_cron_DeltaListEntry *job;
  UTIL_cron_DeltaListEntry *last;
  int jobId;

#if DEBUG_CRON
  GNUNET_GE_LOG (cron->ectx,
                 GNUNET_GE_STATUS | GNUNET_GE_DEVELOPER | GNUNET_GE_BULK,
                 "deleting job %p-%p\n", method, data);
#endif
  GNUNET_mutex_lock (cron->deltaListLock_);
  jobId = cron->firstUsed_;
  if (jobId == -1)
    {
      GNUNET_mutex_unlock (cron->deltaListLock_);
      return 0;
    }
  last = NULL;
  job = &cron->deltaList_[jobId];
  while ((job->method != method) ||
         (job->data != data) || (job->deltaRepeat != repeat))
    {
      last = job;
      if (job->next == -1)
        {
          GNUNET_mutex_unlock (cron->deltaListLock_);
          return 0;
        }
      jobId = job->next;
      job = &cron->deltaList_[jobId];
    }
  if (last != NULL)
    last->next = job->next;
  else
    cron->firstUsed_ = job->next;
  job->next = cron->firstFree_;
  cron->firstFree_ = jobId;
  job->method = NULL;
  job->data = NULL;
  job->deltaRepeat = 0;
  GNUNET_mutex_unlock (cron->deltaListLock_);
  return 1;
}





/* end of cron.c */