src/or/scheduler_vanilla.c

   1 /* Copyright (c) 2017, The Tor Project, Inc. */
   2 /* See LICENSE for licensing information */
   3
   4 #include <event2/event.h>
   5
   6 #include "or.h"
   7 #include "config.h"
   8 #define TOR_CHANNEL_INTERNAL_
   9 #include "channel.h"
  10 #define SCHEDULER_PRIVATE_
  11 #include "scheduler.h"
  12
  13 /*****************************************************************************
  14  * Other internal data
  15  *****************************************************************************/
  16
  17 /* Maximum cells to flush in a single call to channel_flush_some_cells(); */
  18 #define MAX_FLUSH_CELLS 1000
  19
  20 /*****************************************************************************
  21  * Externally called function implementations
  22  *****************************************************************************/
  23
  24 /* Return true iff the scheduler has work to perform. */
  25 static int
  26 have_work(void)
  27 {
  28   smartlist_t *cp = get_channels_pending();
  29   IF_BUG_ONCE(!cp) {
  30     return 0; // channels_pending doesn't exist so... no work?
  31   }
  32   return smartlist_len(cp) > 0;
  33 }
  34
  35 /** Re-trigger the scheduler in a way safe to use from the callback */
  36
  37 static void
  38 vanilla_scheduler_schedule(void)
  39 {
  40   if (!have_work()) {
  41     return;
  42   }
  43
  44   /* Activate our event so it can process channels. */
  45   scheduler_ev_active(EV_TIMEOUT);
  46 }
  47
  48 static void
  49 vanilla_scheduler_run(void)
  50 {
  51   int n_cells, n_chans_before, n_chans_after;
  52   ssize_t flushed, flushed_this_time;
  53   smartlist_t *cp = get_channels_pending();
  54   smartlist_t *to_readd = NULL;
  55   channel_t *chan = NULL;
  56
  57   log_debug(LD_SCHED, "We have a chance to run the scheduler");
  58
  59   n_chans_before = smartlist_len(cp);
  60
  61   while (smartlist_len(cp) > 0) {
  62     /* Pop off a channel */
  63     chan = smartlist_pqueue_pop(cp,
  64                                 scheduler_compare_channels,
  65                                 offsetof(channel_t, sched_heap_idx));
  66     IF_BUG_ONCE(!chan) {
  67       /* Some-freaking-how a NULL got into the channels_pending. That should
  68        * never happen, but it should be harmless to ignore it and keep looping.
  69        */
  70       continue;
  71     }
  72
  73     /* Figure out how many cells we can write */
  74     n_cells = channel_num_cells_writeable(chan);
  75     if (n_cells > 0) {
  76       log_debug(LD_SCHED,
  77                 "Scheduler saw pending channel " U64_FORMAT " at %p with "
  78                 "%d cells writeable",
  79                 U64_PRINTF_ARG(chan->global_identifier), chan, n_cells);
  80
  81       flushed = 0;
  82       while (flushed < n_cells) {
  83         flushed_this_time =
  84           channel_flush_some_cells(chan,
  85                         MIN(MAX_FLUSH_CELLS, (size_t) n_cells - flushed));
  86         if (flushed_this_time <= 0) break;
  87         flushed += flushed_this_time;
  88       }
  89
  90       if (flushed < n_cells) {
  91         /* We ran out of cells to flush */
  92         scheduler_set_channel_state(chan, SCHED_CHAN_WAITING_FOR_CELLS);
  93       } else {
  94         /* The channel may still have some cells */
  95         if (channel_more_to_flush(chan)) {
  96         /* The channel goes to either pending or waiting_to_write */
  97           if (channel_num_cells_writeable(chan) > 0) {
  98             /* Add it back to pending later */
  99             if (!to_readd) to_readd = smartlist_new();
 100             smartlist_add(to_readd, chan);
 101             log_debug(LD_SCHED,
 102                       "Channel " U64_FORMAT " at %p "
 103                       "is still pending",
 104                       U64_PRINTF_ARG(chan->global_identifier),
 105                       chan);
 106           } else {
 107             /* It's waiting to be able to write more */
 108             scheduler_set_channel_state(chan, SCHED_CHAN_WAITING_TO_WRITE);
 109           }
 110         } else {
 111           /* No cells left; it can go to idle or waiting_for_cells */
 112           if (channel_num_cells_writeable(chan) > 0) {
 113             /*
 114              * It can still accept writes, so it goes to
 115              * waiting_for_cells
 116              */
 117             scheduler_set_channel_state(chan, SCHED_CHAN_WAITING_FOR_CELLS);
 118           } else {
 119             /*
 120              * We exactly filled up the output queue with all available
 121              * cells; go to idle.
 122              */
 123             scheduler_set_channel_state(chan, SCHED_CHAN_IDLE);
 124           }
 125         }
 126       }
 127
 128       log_debug(LD_SCHED,
 129                 "Scheduler flushed %d cells onto pending channel "
 130                 U64_FORMAT " at %p",
 131                 (int)flushed, U64_PRINTF_ARG(chan->global_identifier),
 132                 chan);
 133     } else {
 134       log_info(LD_SCHED,
 135                "Scheduler saw pending channel " U64_FORMAT " at %p with "
 136                "no cells writeable",
 137                U64_PRINTF_ARG(chan->global_identifier), chan);
 138       /* Put it back to WAITING_TO_WRITE */
 139       scheduler_set_channel_state(chan, SCHED_CHAN_WAITING_TO_WRITE);
 140     }
 141   }
 142
 143   /* Readd any channels we need to */
 144   if (to_readd) {
 145     SMARTLIST_FOREACH_BEGIN(to_readd, channel_t *, readd_chan) {
 146       scheduler_set_channel_state(readd_chan, SCHED_CHAN_PENDING);
 147       smartlist_pqueue_add(cp,
 148                            scheduler_compare_channels,
 149                            offsetof(channel_t, sched_heap_idx),
 150                            readd_chan);
 151     } SMARTLIST_FOREACH_END(readd_chan);
 152     smartlist_free(to_readd);
 153   }
 154
 155   n_chans_after = smartlist_len(cp);
 156   log_debug(LD_SCHED, "Scheduler handled %d of %d pending channels",
 157             n_chans_before - n_chans_after, n_chans_before);
 158 }
 159
 160 /* Stores the vanilla scheduler function pointers. */
 161 static scheduler_t vanilla_scheduler = {
 162   .type = SCHEDULER_VANILLA,
 163   .free_all = NULL,
 164   .on_channel_free = NULL,
 165   .init = NULL,
 166   .on_new_consensus = NULL,
 167   .schedule = vanilla_scheduler_schedule,
 168   .run = vanilla_scheduler_run,
 169   .on_new_options = NULL,
 170 };
 171
 172 scheduler_t *
 173 get_vanilla_scheduler(void)
 174 {
 175   return &vanilla_scheduler;
 176 }
 177