t_scheduler.cpp

Go to the documentation of this file.

/*****************************************************************************\
*                                                                             *
*  Name   : test_scheduler                                                    *
*  Author : Chris Koeritz                                                     *
*                                                                             *
*******************************************************************************
* Copyright (c) 2001-$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 <basis/chaos.h>
#include <basis/function.h>
#include <basis/guards.h>
#include <basis/istring.h>
#include <basis/mutex.h>
#include <basis/set.cpp>
#include <data_struct/amorph.cpp>
#include <mechanisms/ithread.h>
#include <mechanisms/safe_roller.h>
#include <mechanisms/time_stamp.h>
#include <opsystem/application_shell.h>
#include <loggers/console_logger.h>
#include <loggers/file_logger.h>
#include <opsystem/path_configuration.h>
#include <data_struct/static_memory_gremlin.h>
#include <scheduling/schedulable.h>
#include <scheduling/schedule_actor.h>
#include <scheduling/scheduler.h>

const int DEFAULT_FISH = 64;
  // the number of threads, by default.

const int DEFAULT_RUN_TIME = 142 * SECOND_ms;
  // the length of time to run the program.

int concurrent_biters = 0;
  // the number of threads that are currently active.

// thresholds for thread sleep.
const int LOWER_SLEEP = 0;
const int HIGHER_SLEEP = 82;

#define LOG(to_print) CLASS_EMERGENCY_LOG(program_wide_logger(), to_print)

chaos rando;

safe_roller &identor() { static safe_roller _ids; return _ids; }


// this class does something...

class lotus_blossom : public schedulable
{
public:
//add some record keeping?
  lotus_blossom(const scheduling_id &sched_id, int actor_id,
          const periodicity &when_to_schedule)
  : schedulable(sched_id, actor_id, when_to_schedule) {}

  virtual istring schedulable_text_form() const {
    return istring("lotus_blossom");
  }
};


// the actor thread is the base class for our test threads.

class actor_thread : public ithread, public schedule_actor
{
public:
  int _uid;  // the unique id for this thread.
  int_set _scheduled;  // these items are in the scheduler, we think.

  actor_thread(int uid) : ithread(0), schedule_actor(), _uid(uid),
      _scheduled() {}

  IMPLEMENT_CLASS_NAME("actor_thread");

  // inherited requirements.
  virtual int actor_id() const { return _uid; }
  virtual outcome preprocess(schedulable &to_process);
  virtual outcome expiration(schedulable &to_expire);
  virtual void postprocess(schedulable &to_process);
};

outcome actor_thread::preprocess(schedulable &formal(to_process))
{
//pick a random response.
// most of the time, allow it to be scheduled.
return schedule_actor::OKAY;
}

outcome actor_thread::expiration(schedulable &formal(to_expire))
{
//randomly decide whether to keep or not.
return schedule_actor::OKAY;
}

void actor_thread::postprocess(schedulable &formal(to_process))
{
//chuck this item out of the list of scheds in our thread.

//or maybe tell it to be postponed from deletion.
//later the thread can try to restart it.
// like move it to a "postponed" id list.
}


class land_shark : public actor_thread
{
public:
  land_shark(int uid) : actor_thread(uid)
  {
    FUNCDEF("constructor");
    safe_add(concurrent_biters, 1);
LOG(istring(istring::SPRINTF, "starting %d", _uid));
  }

  virtual ~land_shark() {
    FUNCDEF("destructor");
    safe_add(concurrent_biters, -1); 
LOG(istring(istring::SPRINTF, "closing %d", _uid));
  }

  IMPLEMENT_CLASS_NAME("land_shark");

  virtual outcome activate(schedulable &to_activate, int time_allowed);
    // a schedule item wants us to play with it now.  how cute.

  void perform_activity(void *) {
    FUNCDEF("perform_activity");
LOG(istring(istring::SPRINTF, "performing %d", _uid));
//uhhh?
    portable::sleep_ms(rando.inclusive(LOWER_SLEEP, HIGHER_SLEEP));
  }
};

outcome land_shark::activate(schedulable &formal(to_activate), int formal(time_allowed))
{
return schedule_actor::OKAY;
}


class minnow : public actor_thread
{
public:
  minnow(int uid) : actor_thread(uid)
  {
    FUNCDEF("constructor");
    safe_add(concurrent_biters, 1);
LOG(istring(istring::SPRINTF, "starting %d", _uid));
  }

  virtual ~minnow() {
    FUNCDEF("destructor");
    safe_add(concurrent_biters, -1);
LOG(istring(istring::SPRINTF, "closing %d", _uid));
  }

  IMPLEMENT_CLASS_NAME("minnow");

  virtual outcome activate(schedulable &to_activate, int time_allowed);
    // process the schedule item.

  void perform_activity(void *formal(data)) {
    FUNCDEF("perform_activity");
LOG(istring(istring::SPRINTF, "performing %d", _uid));
//what are we doing with these exactly?
  }
};

outcome minnow::activate(schedulable &formal(to_activate), int formal(time_allowed))
{
return schedule_actor::OKAY;
}


class thread_mapper : public actor_mapper
{
public:
  virtual ~thread_mapper() {}

  void add(actor_thread *to_add);
    // puts a new thread into the list.

  bool zap(int id);
    // zaps a thread in the list.  true is returned if it was there.

  actor_thread *lock_thread(int id);
    // this turns an id into a thread pointer, locking the list while the
    // pointer is out.

  void unlock_thread(actor_thread * &to_put);
    // it is crucial that lock_thread is always followed by unlock_thread and
    // that no more than one lock_thread is active at a time.

  void stop_all();
    // stops all threads and resets the list.

  // inherited responsibilities.
  virtual actor_mapping *open_actor(int actor_id);
  virtual void close_actor(actor_mapping *to_close);

private:
  mutex _protector;
  amorph<actor_thread> _thread_list;
};

void thread_mapper::add(actor_thread *to_add)
{
  auto_synchronizer l(_protector);
  _thread_list.append(to_add);
}

bool thread_mapper::zap(int id)
{
  auto_synchronizer l(_protector);
  for (int i = 0; i < _thread_list.elements(); i++)
    if (_thread_list.borrow(i)->_uid == id) {
      _thread_list.zap(i, i);
      return true;
    }
  return false;
}

actor_thread *thread_mapper::lock_thread(int id)
{
  _protector.lock();
  for (int i = 0; i < _thread_list.elements(); i++)
    if (_thread_list.borrow(i)->_uid == id) return _thread_list.borrow(i);
  // didn't find it so no lock.
  _protector.unlock();
  return NIL;
}

void thread_mapper::unlock_thread(actor_thread * &to_put)
{
  if (!to_put) return;  // do nothing; they didn't get a lock.
  to_put = NIL;
  _protector.unlock();
}

void thread_mapper::stop_all()
{
  for (int i = _thread_list.elements() - 1; i >= 0; i--) {
    _thread_list.borrow(i)->stop();
    _thread_list.zap(i, i);
  }
}

actor_mapping *thread_mapper::open_actor(int actor_id)
{
  actor_thread *at = lock_thread(actor_id); 
  if (!at) return NIL;
  return new actor_mapping(*at);
}

void thread_mapper::close_actor(actor_mapping *to_close)
{
  if (!to_close) return;  // not any good.
  actor_thread *real_actor = dynamic_cast<actor_thread *>(&to_close->_actor);
  unlock_thread(real_actor);
  WHACK(to_close);
}


class test_scheduler : public application_shell
{
public:
  test_scheduler() : application_shell(class_name()) {}
  IMPLEMENT_CLASS_NAME("t_scheduler");
  virtual int execute();
};

int test_scheduler::execute()
{
  FUNCDEF("main");
  // force these to be created before threads are started.
//hmmm: use safe static.
  identor();

  thread_mapper mapper;  // our thread management.

//hmmm: create scheduler

  LOG("starting addition of threads...");

  for (int i = 0; i < DEFAULT_FISH; i++) {
    actor_thread *t = NIL;
    int id = identor().next_id();
    if (i % 2)
      t = new land_shark(id);
    else
      t = new minnow(id);
    mapper.add(t);
    actor_thread *q = mapper.lock_thread(id);
    if (!q || (q != t))
      deadly_error(class_name(), func, "amorph has incorrect pointer!");
    t = NIL;  // reset since this is now owned by mapper.
    // start the thread we added.
    q->start(NIL);
    mapper.unlock_thread(q);
  }

  time_stamp when_to_leave(DEFAULT_RUN_TIME);
  while (when_to_leave > time_stamp()) {

    portable::sleep_ms(rando.inclusive(LOWER_SLEEP, HIGHER_SLEEP));
  }

  LOG("test done; now exiting from all threads...");

//hmmm: stop scheduler.

  // stop all the threads.
  mapper.stop_all();

  if (concurrent_biters != 0)
    deadly_error(class_name(), func, "threads were still active supposedly!");

  LOG("done exiting from all threads.");

  guards::alert_message("scheduler:: works for all functions tested.");
  return 0;
}

HOOPLE_MAIN(test_scheduler, )

---