scheduler.cpp

Go to the documentation of this file.

#ifndef SCHEDULER_IMPLEMENTATION_FILE
#define SCHEDULER_IMPLEMENTATION_FILE

/*****************************************************************************\
*                                                                             *
*  Name   : scheduler                                                         *
*  Author : Chris Koeritz                                                     *
*                                                                             *
*******************************************************************************
* Copyright (c) 1991-$now By Author.  This program 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 of   *
* the License or (at your option) any later version.  This is online at:      *
*     http://www.fsf.org/copyleft/gpl.html                                    *
* Please send any updates to: fred@gruntose.com                               *
\*****************************************************************************/

// Notes:
//
//   _pending_sends must be touched only when the schedule is locked.
//
//   _weight_tracker likewise.

#include "implem_only.h"
#include "schedulable.h"
#include "schedule_actor.h"
#include "scheduler.h"

#include <basis/istring.h>
#include <basis/mutex.h>
#include <basis/portable.h>
#include <basis/set.cpp>
#include <mathematics/averager.cpp>
#include <data_struct/memory_limiter.h>
#include <mechanisms/ithread.h>
#include <nodes/safe_list.h>
#include <nodes/safe_node.h>
#include <post_office/letter.h>
#include <post_office/mailbox.h>

using namespace basis;
using namespace nodes;

//#define DEBUG_SCHEDULER
  // uncomment for noisy version.

const int DURATION_AVERAGER_SIZE = 20;
  // the number of entries to keep in the averager that tracks scheduling
  // durations.

//hmmm: make tunable.
const int ENFORCED_SCHEDULER_PAUSE = 42;
  // we will sleep this long if we chewed through everything we could in the
  // schedule but there's still more left.  we don't want the scheduler to
  // run continually.

const int SCHEDULE_HITS_ALLOWED = 1480;
  // the number of items we'll crank per scheduling run.

#define LIFT_WEIGHT_AND_VERIFY(id, size) \
  if (!_weight_tracker->record_deletion(id, size)) { \
    LOG(istring(istring::SPRINTF, "size for actor %d is botched!", id)); \
  }


class waiting_item : public letter
{
public:
  schedulable *_held;

  istring text_form() const
      { if (_held) return _held->schedulable_text_form(); else return ""; }

  waiting_item(schedulable *to_hold) : _held(to_hold) {}
};


// the scheduler's iterator relies on the scheduler to set its hidden
// iterator pointer correctly.

schedule_iterator::schedule_iterator(const scheduler &mgr)
: _manager(mgr),
  _hidden_iterator(NIL)
{}

schedule_iterator::~schedule_iterator() { _hidden_iterator = NIL; }

void schedule_iterator::next() { _hidden_iterator->next(); }

void schedule_iterator::previous() { _hidden_iterator->previous(); }

schedulable *schedule_iterator::access()
{ return dynamic_cast<schedulable *>(_hidden_iterator->access()); }

bool schedule_iterator::is_head() { return _hidden_iterator->is_head(); }

bool schedule_iterator::is_tail() { return _hidden_iterator->is_tail(); }

void schedule_iterator::jump_head() { _hidden_iterator->jump_head(); }

void schedule_iterator::jump_tail() { _hidden_iterator->jump_tail(); }


class schedule_driver : public ithread
{
public:
  schedule_driver(scheduler &to_drive);
  ~schedule_driver();

  IMPLEMENT_CLASS_NAME("schedule_driver");

private:
  scheduler &_to_drive;  // the scheduling object we push around.

  void perform_activity(void *thread_data);
    // main function of object; invoked by the ithread base.
};


schedule_driver::schedule_driver(scheduler &to_drive)
: ithread(to_drive.snooze_interval(), ithread::SLACK_INTERVAL),
  _to_drive(to_drive)
{}

schedule_driver::~schedule_driver() { stop(); }

void schedule_driver::perform_activity(void *formal(thread_data))
{
  FUNCDEF("perform_activity");
  outcome ret;  // used inside loop.
  // run until thread is told to quit.
  time_stamp start_time;  // click, stopwatch starts...
  try {
    // the scheduling happens for a while.
    ret = _to_drive.process_schedule(SCHEDULE_HITS_ALLOWED);
  } catch (...) {
    LOG("an exception has occurred during schedule driving!");
  }
  // click, stopwatch stops...
  time_stamp stop_time;
  int duration = int(stop_time.value() - start_time.value());
//#ifdef DEBUG_SCHEDULER
  if (duration > 500)
    LOG(istring(istring::SPRINTF, "scheduling took %d ms.", duration));  
//#endif
  if (ret == scheduler::MORE_LEFT) {
    // if there were more items to do, we re-energize the scheduler.
    reschedule(ENFORCED_SCHEDULER_PAUSE);
  }
}


scheduler::scheduler(actor_mapper &mapper, int time_per, int maximum_weight,
    int max_actor_weight, int snooze_interval)
: _mapper(mapper),
  _time_per_item(time_per),
  _schedule(new safe_list),
  _processing_durations(new int_averager(DURATION_AVERAGER_SIZE, false)),
  _activation_durations(new int_averager(DURATION_AVERAGER_SIZE, false)),
  _item_durations(new int_averager(DURATION_AVERAGER_SIZE, false)),
  _bandwidths(new int_averager(DURATION_AVERAGER_SIZE, false)),
  _active_item(NIL),
  _weight_tracker(new memory_limiter(maximum_weight, max_actor_weight)),
  _pending_sends(new mailbox),
  _timely_to_run(new int_set),
  _queued_actors(new int_set),
  _snooze_interval(snooze_interval),
  _driver(new schedule_driver(*this))
{}

scheduler::~scheduler()
{
  FUNCDEF("destructor");
  // shut down the scheduling thread before anything else happens.
  _driver->stop();

  // check if the schedule was clean on shutdown or not.
  safe_list_write_iterator *iter = _schedule->open_writer();
  if (!_schedule->empty(*iter)) {
#ifdef DEBUG_SCHEDULER
    LOG("Some items remain in the schedule!...");
    // close our iterator so the text_form function can do one.
    _schedule->close_writer(iter);
    string_array dump_list;
    schedule_text_form(dump_list, 0);
    istring dump;
    for (int i = 0; i < dump_list.length(); i++)
      dump += dump_list[i] + log_base::platform_ending();
    LOG(dump);  // dump the schedule's contents out.

    dump.reset();
    pending_text_form(dump, 0);
    for (int j = 0; j < dump_list.length(); j++)
      dump += dump_list[j] + log_base::platform_ending();
    LOG(dump);  // dump the pending items out.

    // re-open the iterator so we can clean the schedule and release all
    // the trapped memory.
    iter = _schedule->open_writer();
#endif
    while (!iter->is_tail()) {
      schedulable *found = dynamic_cast<schedulable *>(iter->access());
      if (found) LIFT_WEIGHT_AND_VERIFY(found->actor_id(), found->weight());
      _schedule->zap(*iter);
    }
  }
  _schedule->close_writer(iter);
  WHACK(_timely_to_run);
  WHACK(_queued_actors);
  WHACK(_schedule);
  WHACK(_processing_durations);
  WHACK(_activation_durations);
  WHACK(_item_durations);
  WHACK(_bandwidths);
  WHACK(_weight_tracker);
  WHACK(_pending_sends);
  WHACK(_driver);
}

const int_averager &scheduler::item_durations() const
{ return *_item_durations; }

const int_averager &scheduler::bandwidths() const
{ return *_bandwidths; }

const int_averager &scheduler::processing_durations() const
{ return *_processing_durations; }

const int_averager &scheduler::activation_durations() const
{ return *_activation_durations; }

const schedulable *scheduler::active_item(schedule_iterator &formal(to_use)) const
{ return _active_item; }

void scheduler::start_scheduling() { _driver->start(NIL); }

void scheduler::stop_scheduling() { _driver->stop(); }

schedule_iterator *scheduler::open_iterator(common::list_positions where)
{
  if ( (where != common::HEAD) && (where != common::TAIL) )
    return NIL;  // bad position.
  safe_list_write_iterator *iter = _schedule->open_writer(where);
  schedule_iterator *to_return = new schedule_iterator(*this);
  to_return->_hidden_iterator = iter;
  return to_return;
}

void scheduler::close_iterator(schedule_iterator * &to_close)
{
  if (!to_close) return;  // argh!  what the heck are they trying to pull?
  if (to_close->_hidden_iterator) {
    _schedule->close_writer(to_close->_hidden_iterator);
    to_close->_hidden_iterator = NIL;
  }
  WHACK(to_close);
}

void scheduler::zap(schedule_iterator &to_use)
{
  FUNCDEF("zap");
  schedulable *found = dynamic_cast<schedulable *>(to_use.access());
  LIFT_WEIGHT_AND_VERIFY(found->actor_id(), found->weight());
  _schedule->zap(*to_use._hidden_iterator); 
}

schedule_iterator *scheduler::find(scheduling_id to_find)
{
  safe_list_write_iterator *iter = _schedule->open_writer();
  while (!iter->is_tail()) {
    schedulable *examining = dynamic_cast<schedulable *>(iter->access());
    if (examining && (examining->item_id() == to_find)) {
      schedule_iterator *to_return = new schedule_iterator(*this);
      to_return->_hidden_iterator = iter;
      return to_return;
    }
    iter->next();
  }
  _schedule->close_writer(iter);
  return NIL;
}

bool scheduler::items_scheduled(schedule_iterator &iter, int actor)
{
  // check the active item first.
  const schedulable *active = active_item(iter);
  if (active && (active->actor_id() == actor))
    return true;

  // now scoot through and look at the other items.
  iter.jump_head();
  while (!iter.is_tail()) {
    schedulable *examining = dynamic_cast<schedulable *>(iter.access());
    if (examining && (examining->actor_id() == actor))
      return true;
    iter.next();
  }

  return false;
}

outcome scheduler::do_preprocess(schedulable &to_preprocess)
{
  FUNCDEF("do_preprocess");
  // now preprocess the item as it begins to slide into operation.
  actor_mapping *actor = _mapper.open_actor(to_preprocess.actor_id());
  if (!actor) {
    // toss this item because there's no actor for it.
    LOG(istring(istring::SPRINTF, "could not find actor for id %d; "
        "cancelling item.", to_preprocess.actor_id()));
    return CANCEL;
  }

  // reset the time stamp on the item to now, since we're just about to
  // preprocess.
  to_preprocess.stats().entry = time_stamp();

  // reset the expiration counter since we're starting the item now.
  to_preprocess.expiration().kabump();

  outcome ret = actor->_actor.preprocess(to_preprocess);
  _mapper.close_actor(actor);
  to_preprocess.stats().last_outcome = ret.value();
  if (ret != schedule_actor::OKAY) return CANCEL;
  return OKAY;
}

outcome scheduler::add(schedulable *to_add, bool single_file)
{
  FUNCDEF("add");
  if (!to_add) return BAD_INPUT;
  if (!to_add->item_id()) {
    WHACK(to_add);
    return BAD_INPUT;
  }

  // see if this item has already been scheduled somehow.  the list must not
  // be locked before this check, since the find method will also lock it.
  schedule_iterator *found = find(to_add->item_id());
  if (found) {
    LOG(istring(istring::SPRINTF, "item %d is already in schedule!",
        to_add->item_id().raw_id()));
    close_iterator(found);
    WHACK(to_add);
    return BAD_INPUT;
  }

  // create a write iterator to enable us to store the item in the schedule.
  safe_list_write_iterator *iter = _schedule->open_writer(safe_list::TAIL);

  // check that we have enough memory left.
  if (!_weight_tracker->okay_allocation(to_add->actor_id(), to_add->weight())) {
#ifdef DEBUG_SCHEDULER
    LOG(isprintf("we are too bloated for item %d!",
        to_add->item_id().raw_id()));
#endif
    _schedule->close_writer(iter);
    WHACK(to_add);
    return NO_SPACE;
  }

  // reset certain data bits appropriately...
  to_add->stats().last_outcome = int(schedule_actor::OKAY);

  if (single_file) {
    // instead of queueing this up for action immediately, we add it to the
    // mailbox.  then, once we've dropped our lock on the schedule, we push any
    // items that are ready into the real schedule.
    waiting_item *new_guy = new waiting_item(to_add);
    _pending_sends->drop_off(to_add->actor_id(), new_guy);
#ifdef DEBUG_SCHEDULER
    LOG(isprintf("adding item %d to the pending sends.",
        to_add->item_id().raw_id()));
#endif
    _schedule->close_writer(iter);
    promote_actor(to_add->actor_id());
    return OKAY;
  } 

  // stuff the item into the schedule now instead of later.
  _schedule->insert(*iter, to_add);

#ifdef DEBUG_SCHEDULER
  LOG(isprintf("added item %d directly to schedule.",
      to_add->item_id().raw_id()));
#endif

  // done editing the schedule.
  _schedule->close_writer(iter);
  return OKAY;
}

bool scheduler::time_to_schedule(schedulable &to_check)
{
  if (to_check.start_time <= time_stamp()) return true;
  return false;
}

bool scheduler::find_timely_items(int_set &timely)
{
  FUNCDEF("find_timely_items");
  timely.reset();  // clear out any prior contents.
  safe_list_write_iterator *iter;
  for (iter = _schedule->open_writer(); !iter->is_tail(); iter->next()) {
    if (!iter->access()) break;  // no elements, so scram.
    // look at the current item in the schedule.
    schedulable *to_examine = dynamic_cast<schedulable *>(iter->access());
    if (!to_examine) {
      // something wrong with this one.
      LOG(istring("found bad schedulable in list!"));
      _schedule->zap(*iter);
      continue;
    }
    // check if we already added this one.  we don't bother checking again.
    if (timely.member(to_examine->item_id().raw_id())) continue;
    // if it's time now, we add this one to our list of interesting items.
    if (time_to_schedule(*to_examine))
      timely += to_examine->item_id().raw_id();
  }
  _schedule->close_writer(iter);
  return !!timely.elements();
}

outcome scheduler::schedule_item(int &hits, schedulable &to_process)
{
  FUNCDEF("schedule_item");
  if (!to_process.preprocessed()) {
    // this item has not been preprocessed yet, so we'll do it now.
#ifdef DEBUG_SCHEDULER
    LOG(isprintf("preprocessing %d.", to_process.item_id().raw_id()));
#endif
    outcome ret = do_preprocess(to_process);
    if (ret != OKAY) {
      // if it wasn't okay, they don't want to schedule it after all.
      LOG(isprintf("preprocess disliked item %d, gave outcome ",
          to_process.item_id().raw_id()) + outcome_name(ret));
      return ret;
    }
    to_process.preprocessed(true);
#ifdef DEBUG_SCHEDULER
    LOG(isprintf("preprocessed %d.", to_process.item_id().raw_id()));
#endif
  }

  // find the actor on this item, for future reference.
  actor_mapping *actor = _mapper.open_actor(to_process.actor_id());
  if (!actor) {
    LOG(isprintf("could not find actor for id %d; now whacking item.",
        to_process.actor_id()));
    return schedule_actor::CANCEL;
  } 

  outcome current_outcome = schedule_actor::OKAY;
  if (to_process.expiration().dead()) {
    // this item is already deceased.
    LOG(isprintf("item %d is dead.", to_process.item_id().raw_id()));
    current_outcome = schedule_actor::CANCEL;
  } else if (to_process.expiration_interval) {
    // if an expiration interval was set, we check that the item has
    // not already expired.
    outcome ret = check_expiration(to_process, actor->_actor);
    if (ret != scheduler::OKAY) {
      // need to chuck this bugger out of the schedule now; it expired and
      // didn't get a reprieve.
      LOG(isprintf("item %d is cancelled!  expired and not renewed.",
          to_process.item_id().raw_id())
          + istring("  got outcome ") + outcome_name(ret));
      current_outcome = schedule_actor::CANCEL;
      to_process.expiration().reset(0, 0);
    }
  } else {
    // it's not dead, and it has no expiration interval, so we say it had
    // a healthy heartbeat and should not expire.
    to_process.expiration().kabump();
  }

  if (current_outcome != schedule_actor::OKAY) {
    return current_outcome;
  }

  // if nothing bad has happened so far, schedule the item.
  time_stamp activation_start;
#ifdef DEBUG_SCHEDULER
  LOG(isprintf("activating %d.", to_process.item_id().raw_id()));
#endif
  // give the item some time to operate.
  current_outcome = actor->_actor.activate(to_process, _time_per_item);
  if (current_outcome != schedule_actor::POSTPONE) {
    // we have now definitely activated something.
    hits++;
#ifdef DEBUG_SCHEDULER
    LOG(isprintf("activated %d (outcome=%s)", to_process.item_id().raw_id(),
        schedule_actor::outcome_name(current_outcome)));
#endif
  }

  int activation_duration = int(time_stamp().value()
      - activation_start.value());
  if (activation_duration)
    _activation_durations->add(activation_duration, 1);

  to_process.stats().last_outcome = current_outcome.value();
    // record what happened.

  // process the returned status from scheduling the item.  if the item
  // needs to be removed, it gets whacked here.
  if ( (current_outcome == schedule_actor::FINISHED)
        || (current_outcome == schedule_actor::CANCEL) ) {
#ifdef DEBUG_SCHEDULER
    LOG(isprintf("postprocessing %d.", to_process.item_id().raw_id()));
#endif
    // this item has had it; do its postprocessing.
    actor->_actor.postprocess(to_process);
      // we don't record the postprocessing outcome in the stats, because
      // the last outcome is supposed to be the last one encountered
      // _during_ scheduling.
#ifdef DEBUG_SCHEDULER
    LOG(isprintf("postprocessed %d.", to_process.item_id().raw_id()));
#endif
    _mapper.close_actor(actor);
      // close out the actor since we're done talking to it about this
      // item forever.

    // set the exit time.
    to_process.stats().exit = time_stamp();

#ifdef DEBUG_SCHEDULER
    LOG(isprintf("deleting %d.", to_process.item_id().raw_id()));
#endif

    // get its overall duration...
    if (to_process.stats().exit.initialized()
        && to_process.stats().entry.initialized()) {
      time_stamp exit = (const time_stamp &)to_process.stats().exit;
      time_stamp entry = (const time_stamp &)to_process.stats().entry;
      int overall_duration = int(exit.value() - entry.value());
      if (overall_duration) {
        _item_durations->add(overall_duration, 1);
#ifdef DEBUG_SCHEDULER
        if (!to_process.weight())
          LOG("invalid weight (zero) for transmission!!!");
#endif
        double raw_bandwidth = double(to_process.weight())
            / overall_duration;
          // compute bytes per millisecond.
        int bandwidth = int(raw_bandwidth / 1024.0 * 1000.0);
        _bandwidths->add(bandwidth, 1);
      }
    }

    return current_outcome;
  }

  // shut down the actor now that we're done with it.
  _mapper.close_actor(actor);

  // make sure, if it's periodic, that its start time gets bumped.
  if (to_process.entry_interval
      && (current_outcome == schedule_actor::POSTPONE))
    to_process.start_time = time_stamp(to_process.entry_interval);
  
  return current_outcome;
}

void scheduler::crunch_on_timely_items(int &hits, int hits_allowed,
    int_set &timely)
{
  FUNCDEF("crunch_on_timely_items");
  safe_list_write_iterator *iter;
  for (iter = _schedule->open_writer(); !iter->is_tail(); iter->next()) {
    if (hits > hits_allowed) break;  // we've used up our welcome.
    if (!iter->access()) break;  // no elements, so scram.
    if (!timely.elements()) break;  // we ran out of things to look at.

    // look at the current item in the schedule.
    schedulable *to_examine = dynamic_cast<schedulable *>(iter->access());
    if (!to_examine) {
      // something wrong with this one.
      LOG(istring("found bad schedulable in list!"));
      _schedule->zap(*iter);
      continue;
    }
    // make sure this is one that we're authorized to talk to.
    if (!timely.member(to_examine->item_id().raw_id()))
      continue;
    // re-check that this particular item is ready.
    if (!time_to_schedule(*to_examine)) {
      // this one is no longer relevant.
      timely -= to_examine->item_id().raw_id();
      continue;
    }

    to_examine = dynamic_cast<schedulable *>(_schedule->remove(*iter));
      // reset the current item by taking over control.
    if (!to_examine) {
      LOG("logic error!  schedule item removed was blank!");
      continue;
    }

    _active_item = to_examine;  // set the active object.
    _schedule->close_writer(iter);
      // relinquish control during the processing of this item.  an iterator
      // must be re-opened before continuing the loop.
  
    outcome sched_ret = schedule_item(hits, *to_examine);
      // activate the item now.

    iter = _schedule->open_writer(safe_list::TAIL);
      // re-open the iterator so we're protected.
    _active_item = NIL;  // nobody active now.
    // decide upon the fate of this item based on its scheduling outcome.
    if ( (sched_ret == schedule_actor::FINISHED)
          || (sched_ret == schedule_actor::CANCEL) ) {
      // this one is done, so we're not going to add it back.
      LIFT_WEIGHT_AND_VERIFY(to_examine->actor_id(), to_examine->weight());
      _schedule->close_writer(iter);  // close list so we can change it.
      // now that the item's not listed as active, move up any waiting items.
      promote_actor(to_examine->actor_id());
      timely -= to_examine->item_id().raw_id();  // remove from consideration.
      WHACK(to_examine);
      // re-open iterator to keep going.
      iter = _schedule->open_writer(safe_list::TAIL);
    } else {
      // this item is healthy still; add it to the tail of the schedule.
      _schedule->insert(*iter, to_examine);
      // check if we were told to snooze on this item.
      if (sched_ret == schedule_actor::POSTPONE)
        timely -= to_examine->item_id().raw_id();
    }
    iter->jump_head();
      // pop back up to the top.  our exit conditions should ensure we
      // won't get trapped in a vicious cycle.
  }
  _schedule->close_writer(iter);
}

outcome scheduler::process_schedule(int hits_allowed)
{
  FUNCDEF("process_schedule");
  time_stamp processing_began;  // when scheduling starts.
  int hits = 0;  // number of times we've activated an item.

  promote_queued_items();
    // move the entries that are queued if there's no other active entry
    // for the actor.  actors will be individually promoted during scheduling.

  // iterate through the schedule as long as we have time.
  while (hits < hits_allowed) {
    bool keep_running = find_timely_items(*_timely_to_run);
#ifdef DEBUG_SCHEDULER
    if (_timely_to_run->elements()) {
      istring list;
      for (int i = 0; i < _timely_to_run->elements(); i++) {
        if (i != 0) list += ", ";
        list += isprintf("%d", (*_timely_to_run)[i]);
      }
      LOG(istring("timely list: ") + list);
    }
#endif
    if (!keep_running) break;
    crunch_on_timely_items(hits, hits_allowed, *_timely_to_run);
  }

  // record the accounting information.
  int time_taken = int(time_stamp().value() - processing_began.value());
  if (time_taken) _processing_durations->add(time_taken);

//#ifdef DEBUG_SCHEDULER
  if (hits > 140)
    LOG(istring(istring::SPRINTF, "scheduler got in %d hits.", hits));
//#endif

  if (hits >= hits_allowed) return MORE_LEFT;
    // not everything was totally finished with.
  return OKAY;
}

outcome scheduler::check_expiration(schedulable &to_check,
    schedule_actor &actor)
{
  FUNCDEF("check_expiration");
  if (to_check.expiration().dead()) return CANCEL;
  if (!to_check.expiration().due()) return OKAY;

  // invoke the actor's expiration processing.
  outcome ret = actor.expiration(to_check);
  to_check.expiration().kabump();
    // reset the timer, regardless of the outcome.

  // deal with the expiration appropriately.
  switch (ret.value()) {
    // treated as fairly fresh.
    case schedule_actor::POSTPONE: case schedule_actor::OKAY:
      return OKAY;
    // treated as expired.
    case schedule_actor::FINISHED: case schedule_actor::CANCEL:
      return CANCEL;
    // unexpected, so expire it.
    default: return CANCEL;
  }
}

outcome scheduler::mark_dead(scheduling_id to_mark)
{
  FUNCDEF("mark_dead");
  schedule_iterator *to_zap = find(to_mark);
  if (!to_zap) return NOT_FOUND;
  to_zap->access()->expiration().reset(0, 0);
  close_iterator(to_zap);
  return OKAY;
}

outcome scheduler::remove_by_actor(int actor_id)
{
  FUNCDEF("remove_by_actor");
  safe_list_write_iterator *iter = _schedule->open_writer();
  while (!iter->is_tail()) {
    schedulable *found = dynamic_cast<schedulable *>(iter->access());
    if (found && (found->actor_id() == actor_id) ) {
      LIFT_WEIGHT_AND_VERIFY(found->actor_id(), found->weight());
      _schedule->zap(*iter);
      continue;
    }
    iter->next();
  }

  // clear the pending sends for this actor also.
  _pending_sends->close_out(actor_id);

  _schedule->close_writer(iter);
  return OKAY;
}

void scheduler::schedule_text_form(string_array &info, int formal(indent))
{
//hmmm: use the indent.
  info.reset();
  safe_list_read_iterator *iter = _schedule->open_reader();
  if (_active_item)
    info += istring("active => ") + _active_item->schedulable_text_form();
  while (!iter->is_tail()) {
    const schedulable *found = dynamic_cast<const schedulable *>
        (iter->observe());
    if (found) info += found->schedulable_text_form();
    iter->next();
  }
  _schedule->close_reader(iter);
}

void scheduler::pending_text_form(istring &info, int formal(indent))
{
//hmmm: use the indent.
  safe_list_read_iterator *iter = _schedule->open_reader();
  _pending_sends->show(info);
  _schedule->close_reader(iter);
}

void scheduler::memory_text_form(istring &info, int indent)
{ info = _weight_tracker->text_form(indent); }

void scheduler::promote_queued_items()
{
  FUNCDEF("promote_queued_items");
  safe_list_write_iterator *iter = _schedule->open_writer();

  _pending_sends->clean_up();
    // shake out any dead mailboxes so we're not promoting unnecessarily.

  _pending_sends->get_ids(*_queued_actors);
    // get those actors that have any contents.

  _schedule->close_writer(iter);

  // try to promote each actor with things that need handling.
  for (int i = 0; i < _queued_actors->elements(); i++) {
    int to_promote = (*_queued_actors)[i];
    promote_actor(to_promote);
  }
}

void scheduler::promote_actor(int actor_id)
{
  FUNCDEF("promote_actor");
  safe_list_write_iterator *iter = _schedule->open_writer();

  if (!_pending_sends->waiting(actor_id)) {
    // there wouldn't be anything to send, so why bother?
    _schedule->close_writer(iter);
    return;
  }

  if (_active_item && (_active_item->actor_id() == actor_id) ) {
    // the active guy actually belongs to this one, so we can't promote.
    _schedule->close_writer(iter);
    return;
  }

  while (!iter->is_tail()) {
    schedulable *examining = dynamic_cast<schedulable *>(iter->access());
    if (examining && (examining->actor_id() == actor_id)) {
      // this actor already has something in place.
      _schedule->close_writer(iter);
      return;
    }
    iter->next();
  }

  // we've guaranteed that nothing is in the scheduler for the actor.

  letter *next_letter = NIL;
  if (!_pending_sends->pick_up(actor_id, next_letter)) {
    // what?  nothing to deliver...
#ifdef DEBUG_SCHEDULER
    LOG(istring(istring::SPRINTF, "huh? actor %d had no pending sends, "
        "but we checked before and it did.", actor_id));
#endif
    _schedule->close_writer(iter);
    return;
  }

  waiting_item *ready_item = (waiting_item *)next_letter;
  // finally, add the item into the scheduler.
#ifdef DEBUG_SCHEDULER
  LOG(istring(istring::SPRINTF, "adding item %d for actor %d into "
      "schedule from pending.", ready_item->_held->item_id().raw_id(),
      actor_id));
  int last_sched_id = ready_item->_held->item_id().raw_id();
#endif

  _schedule->insert(*iter, ready_item->_held);
  WHACK(next_letter);

#ifdef DEBUG_SCHEDULER
  // test that it's not missing after the insert.
  iter->jump_head();
  bool found_it = false;
  for ( ; !iter->is_tail(); iter->next()) {
    schedulable *found = dynamic_cast<schedulable *>(iter->access());
    if (found->item_id().raw_id() == last_sched_id) found_it = true;
  }
  if (!found_it)
    deadly_error("scheduler", "promote actor",
        "missing schedule id we just added!");
#endif

  _schedule->close_writer(iter);
}

outcome removal_applier(letter &to_promote, int uid, void *datalink)
{
  #define static_class_name() "scheduler"
  FUNCDEF("removal_applier");
  if (uid) {}
  waiting_item *item = (waiting_item *)&to_promote;
  int to_whack = *(int *)datalink;

#ifdef DEBUG_SCHEDULER
  if (item->_held->item_id() == to_whack)
    LOG(istring(istring::SPRINTF, "whacking item %x for actor %d "
        "from pending.", to_whack, uid));
#endif

  if (item->_held->item_id() == to_whack) {
    WHACK(item->_held);   // clean the item for real; the husk is tossed after.
    return mailbox::APPLY_WHACK_STOP;
  }
  return mailbox::OKAY;
  #undef class_name
}

outcome scheduler::remove_by_id(scheduling_id to_remove)
{
  FUNCDEF("remove_by_id");
  safe_list_write_iterator *iter = _schedule->open_writer();
  bool found_it = false;
  while (!iter->is_tail()) {
    schedulable *found = dynamic_cast<schedulable *>(iter->access());
    if (found->item_id() == to_remove) {
      LIFT_WEIGHT_AND_VERIFY(found->actor_id(), found->weight());
#ifdef DEBUG_SCHEDULER
      LOG(istring(istring::SPRINTF, "found item %d for actor %d to whack.",
          to_remove.raw_id(), found->actor_id()));
#endif
      _schedule->zap(*iter);
      found_it = true;
      break;
    }
    iter->next();
  }

  if (!found_it) {
    // clean any with that id out of the pending items also.
    int id = to_remove.raw_id();
    _pending_sends->apply(removal_applier, (void *)&id);
  }

  _schedule->close_writer(iter);
  return OKAY;
}

outcome filter_actor_applier(letter &to_promote, int uid, void *datalink)
{
  #undef static_class_name
  #define static_class_name() "scheduler"
  FUNCDEF("filter_actor_applier");
  if (uid) {}
  waiting_item *item = (waiting_item *)&to_promote;
  int_set *healthy_actors = (int_set *)datalink;

  if (!healthy_actors->member(item->_held->actor_id())) {
#ifdef DEBUG_SCHEDULER
    LOG(istring(istring::SPRINTF, "whacking item %d for actor %d "
        "from pending.", item->_held->actor_id(), uid));
#endif
    WHACK(item->_held);   // clean the item for real; the husk is tossed after.
    return mailbox::APPLY_WHACK;
  }

  return mailbox::OKAY;
  #undef class_name
}

void scheduler::filter_missing_actors(const int_set &valid_actors)
{
  FUNCDEF("filter_missing_actors");
  safe_list_write_iterator *iter = _schedule->open_writer();
  while (!iter->is_tail()) {
    schedulable *found = dynamic_cast<schedulable *>(iter->access());
    if (found && !valid_actors.member(found->actor_id())) {
      // this one seems bad, since it's not in our list.
      LIFT_WEIGHT_AND_VERIFY(found->actor_id(), found->weight());
#ifdef DEBUG_SCHEDULER
      LOG(istring(istring::SPRINTF, "cleaning item %d for dead conn %d.",
          found->item_id().raw_id(), found->actor_id()));
#endif
      _schedule->zap(*iter);
    } else
      iter->next();  // only go to next node if we haven't changed the list.
  }

  // clean any out of the pending queue also.
  _pending_sends->apply