post_office.cpp

Go to the documentation of this file.

#ifndef POST_OFFICE_IMPLEMENTATION_FILE
#define POST_OFFICE_IMPLEMENTATION_FILE

/*****************************************************************************\
*                                                                             *
*  Name   : post_office                                                       *
*  Author : Chris Koeritz                                                     *
*                                                                             *
*******************************************************************************
* Copyright (c) 1998-$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                               *
\*****************************************************************************/

#include "letter.h"
#include "mailbox.h"
#include "post_office.h"

#include <basis/mutex.h>
#include <basis/log_base.h>
#include <basis/portable.h>
#include <basis/set.h>
#include <data_struct/amorph.cpp>
#include <data_struct/unique_id.h>
#include <mechanisms/ithread.h>
#include <mechanisms/thread_cabinet.h>
#include <mechanisms/time_stamp.h>
#include <nodes/catalogable.h>
#include <nodes/object_catalog.h>
#include <opsystem/path_configuration.h>

using namespace nodes;

//#define DEBUG_POST_OFFICE
  // uncomment if you want the noisy version.

#undef LOG
#define LOG(a) CLASS_EMERGENCY_LOG(program_wide_logger(), a)

const int CLEANING_INTERVAL = 14 * SECOND_ms;
  // the interval between cleaning of extra letters and dead mailboxes.

const int SNOOZE_TIME_FOR_POSTMAN = 42;
  // we'll snooze for this long if absolutely nothing happened during the
  // thread's activation.  if things are going on, our snooze time is reduced
  // by the length of time we were delivering items.

const int DELIVERIES_ALLOWED = 350;
  // the maximum number of deliveries we'll try to get done per thread run.


//hmmm: arrhhh--maybe we need to spawn a thread per postal route.

class postal_carrier : public ithread
{
public:
  postal_carrier(post_office &parent, const unique_int &route)
  : ithread(SNOOZE_TIME_FOR_POSTMAN, ithread::SLACK_INTERVAL),
    _parent(parent),
    _route(route)
  {}

  IMPLEMENT_CLASS_NAME("postal_carrier");

  void perform_activity(void *) {
    FUNCDEF("perform_activity");
    bool finished;
    try {
      finished = _parent.deliver_mail_on_route(_route, *this); 
    } catch(...) {
      LOG("caught exception during mail delivery!");
    }
    if (!finished) {
      // not finished delivering all items.
      reschedule();
    } else {
      reschedule(SNOOZE_TIME_FOR_POSTMAN);
    }
  }

private:
  post_office &_parent;
  unique_int _route;
};


class postal_cache : public mailbox {};


class tagged_mail_stop : public nodes::catalogable
{
public:
  mail_stop *_route;
  unique_int _thread_id;

  tagged_mail_stop(const unique_int &id = 0, mail_stop *route = NIL,
          const unique_int &thread_id = 0)
      : catalogable(id), _route(route), _thread_id(thread_id) {}

  IMPLEMENT_CLASS_NAME("tagged_mail_stop");

  virtual istring catalogable_name() const {
    return isprintf("%s: id=%d, addr=%08lx, thr_id=%d",
        static_class_name(), id().raw_id(), _route, _thread_id.raw_id());
  }
};

class route_map : public object_catalog {};


class letter_morph : public amorph<letter> {};


post_office::post_office()
: _post(new mailbox),
  _routes(new route_map),
  _stock_lock(new mutex),
  _next_cleaning(new time_stamp),
  _threads(new thread_cabinet)
{
}

post_office::~post_office()
{
  stop_serving();
  WHACK(_post);
  WHACK(_routes);
  WHACK(_next_cleaning);
  WHACK(_stock_lock);
  WHACK(_threads);
}

void post_office::show_routes(istring &to_fill)
{
//hmmm: simplify this; just use the int_set returning func and print that.
  objcat_iterator *iter = _routes->find_head();
  istring current_line;
  istring temp;
  if (_routes->elements(*iter))
    to_fill += istring("Mail Delivery Routes:") + log_base::platform_ending();

  for ( ; !iter->is_tail(); iter->next()) {
    const tagged_mail_stop *tag
        = dynamic_cast<const tagged_mail_stop *>(iter->cat());
    if (!tag) continue;

    temp = istring(istring::SPRINTF, "%d ", tag->id().raw_id());
    if (current_line.length() + temp.length() >= 80) {
      current_line += log_base::platform_ending();
      to_fill += current_line;
      current_line.reset();
    }
    current_line += temp;
  }
  _routes->iter_unlock(iter);
  // catch the last line we created.
  if (!!current_line) to_fill += current_line;
}

void post_office::stop_serving() { if (_threads) _threads->stop_all(); }

void post_office::show_mail(istring &output)
{
  output.reset();
  output += log_base::platform_ending();
  output += istring("Mailbox Contents at ") + utility::timestamp(true, true)
      + log_base::platform_ending() + log_base::platform_ending();
  istring box_state;
  _post->show(box_state); 
  if (box_state.t()) output += box_state;
  else
    output += istring("No items are awaiting delivery.")
        + log_base::platform_ending();
}

void post_office::drop_off(const unique_int &id, letter *package)
{
  FUNCDEF("drop_off");
#ifdef DEBUG_POST_OFFICE
  LOG(istring(istring::SPRINTF, "mailbox drop for %d: ", id)
      + package->text_form());
#endif
  _post->drop_off(id, package); 
#ifdef DEBUG_POST_OFFICE
  if (!route_listed(id)) {
    LOG(isprintf("letter for %d has no route!", id));
  }
#endif
}

bool post_office::pick_up(const unique_int &id, letter * &package)
{
  FUNCDEF("pick_up");
  bool to_return = _post->pick_up(id, package); 
#ifdef DEBUG_POST_OFFICE
  if (to_return)
    LOG(istring(istring::SPRINTF, "mailbox grab for %d: ", id)
        + package->text_form());
#endif
  return to_return;
}

bool post_office::route_listed(const unique_int &id)
{
  int_set route_set;
  get_route_list(route_set);
  return route_set.member(id.raw_id());
}

void post_office::get_route_list(int_set &route_set)
{
  // gather the set of routes that we should carry mail to.
  route_set.reset();

  objcat_iterator *iter = _routes->find_head();
  if (!_routes->elements(*iter)) {
    // if there are no elements, why bother iterating?
    _routes->iter_unlock(iter);
    return;
  }

  for ( ; !iter->is_tail(); iter->next()) {
    const tagged_mail_stop *tag
        = dynamic_cast<const tagged_mail_stop *>(iter->cat());
    if (!tag) continue;
    route_set.add(tag->id().raw_id());
  }
  _routes->iter_unlock(iter);
}

void post_office::clean_package_list(post_office &formal(post),
    letter_morph &to_clean)
{
  FUNCDEF("clean_package_list");
  // recycle all the stuff we had in the list.
  while (to_clean.elements()) {
    letter *package = to_clean.acquire(0);
    to_clean.zap(0, 0);
    if (!package) {
      LOG("saw empty package in list to clean!");
      continue;
    }
    WHACK(package);
  }
}

bool post_office::deliver_mail_on_route(const unique_int &route,
    ithread &carrier)
{
  FUNCDEF("deliver_mail_on_route");
#ifdef DEBUG_POST_OFFICE
  time_stamp enter;
#endif
  if (carrier.should_stop()) return true;  // get out if thread was told to.

  int deliveries = 0;  // number of items delivered so far.
  letter_morph items_for_route;
    // holds the items that need to be sent to this route.

  // pickup all of the mail that we can for this route.
  while (deliveries < DELIVERIES_ALLOWED) {
    if (carrier.should_stop())
      return true;  // get out if thread was told to.
    letter *package;
    if (!_post->pick_up(route, package)) {
      // there are no more letters for this route.
      break;  // skip out of the loop.
    }
    deliveries++;  // count this item as a delivery.
    items_for_route.append(package);
  }

  if (!items_for_route.elements()) return true;  // nothing to handle.

  // locate the destination for this route.
  objcat_iterator *iter = _routes->find(route);  // find the route.
  if (!iter) {
    // we failed to find the route we wanted...
    LOG(istring(istring::SPRINTF, "route %d disappeared!", route.raw_id()));
    clean_package_list(*this, items_for_route);
    return true;
  }

  objcat_isolater *iso = _routes->separate(iter);
  if (!iso) {
    // some kind of bad error has occurred.
    LOG(isprintf("failed to separate route %d!", route.raw_id()));
    clean_package_list(*this, items_for_route);
    _routes->iter_unlock(iter);
    return true;
  }
  tagged_mail_stop *real_route = dynamic_cast<tagged_mail_stop *>(&iso->cat());
  if (!real_route) {
    // another kind of bad error has occurred.
    LOG(isprintf("route was wrong type for %d!", route.raw_id()));
    _routes->rejoin(iso);  // return the lock on this route.
    clean_package_list(*this, items_for_route);
    return true;
  }

  // now deliver what we found for this route.
  for (int t = 0; t < items_for_route.elements(); t++) {
    if (carrier.should_stop()) {
      // get out if thread was told to.
      _routes->rejoin(iso);  // return the lock on this route.
      return true;
    }
    letter *package = items_for_route.acquire(t);
    // hand the package out on the route.
    mail_stop::items_to_deliver pack(route, package);
    real_route->_route->invoke_callback(pack);
      // the callee is responsible for cleaning up.
  }

  _routes->rejoin(iso);  // return the lock on this route.

  bool finished_all = (deliveries < DELIVERIES_ALLOWED);
    // true if we handled everything we could have.

  if (carrier.should_stop()) return true;  // get out if thread was told to.

  // this bit is for the post office at large, but we don't want an extra
  // thread when we've got all these others handy.
  bool cleaning_time = false;
  {
    // short lock to look at the time for next cleaning.
    auto_synchronizer l(*_stock_lock);
    cleaning_time = time_stamp() > *_next_cleaning;
  }
  if (cleaning_time) {
    _post->clean_up();  // get rid of dead mailboxes in main post office.
    _next_cleaning->reset(CLEANING_INTERVAL);
  }

  time_stamp exit;
#ifdef DEBUG_POST_OFFICE
  int duration = int(exit.value() - enter.value());
  if (duration > 20)
    LOG(isprintf("deliveries took %d ms.", duration));
#endif
  return finished_all;
}

bool post_office::register_route(const unique_int &id,
    mail_stop &carrier_path)
{
  objcat_iterator *iter = _routes->find(id);
  if (iter) {
    _routes->iter_unlock(iter);
    return false;  // already exists.
  }
  postal_carrier *po = new postal_carrier(*this, id);
  unique_int thread_id = _threads->add_thread(po, false, NIL);
    // add the thread so we can record its id.
  tagged_mail_stop *new_stop = new tagged_mail_stop(id, &carrier_path,
      thread_id);
  _routes->add(new_stop);
    // add the mail stop to our listings.
  po->start(NIL);
    // now start the thread so it can begin cranking.
  return true;
}

bool post_office::unregister_route(const unique_int &id)
{
  objcat_iterator *iter = _routes->find(id);
  if (!iter) return false;  // doesn't exist yet.
  unique_int thread_id = 0;
  const tagged_mail_stop *tag
      = dynamic_cast<const tagged_mail_stop *>(iter->cat());
  if (tag) thread_id = tag->_thread_id;
    // if the "tag" is nil, then something is badly wrong.
  _routes->iter_unlock(iter);
  _routes->zap_id(id);
  _threads->zap_thread(thread_id);
  return true;
}


#endif //POST_OFFICE_IMPLEMENTATION_FILE

---