LibThread::Scheduler Class Reference

Public Member Functions
	Scheduler (int n)
void	set_maxconcurrency (int n)
int	get_maxconcurrency ()
int	threadpool_size (ThreadPool *pool)
virtual	~Scheduler ()
ThreadState *	getThread (int i)
void	shutdown (bool wait)
void	addThread (ThreadPool *owner, ThreadState *thread)
void	attachJob (ThreadPool *pool, Job *job)
void	detachJob (Job *job)
void	queueJob (Job *job)
void	broadcastJob (ThreadPool *pool, Job *job)
void	cancelDeps (Job *job)
void	cancelJob (Job *job)
void	waitJob (Job *job)
void	clearThreadState ()
Public Member Functions inherited from LibThread::SharedObject
	SharedObject ()
virtual	~SharedObject ()
void	set_type (int type_init)
int	get_type ()
void	set_name (std::string &name_init)
void	set_name (const char *s)
std::string &	get_name ()
void	incref (int by=1)
long	decref ()
long	getref ()
virtual BOOLEAN	op2 (int op, leftv res, leftv a1, leftv a2)
virtual BOOLEAN	op3 (int op, leftv res, leftv a1, leftv a2, leftv a3)

Static Public Member Functions
static void	notifyDeps (Scheduler *scheduler, Job *job)
static void *	main (ThreadState *ts, void *arg)

Data Fields
Lock	lock

Private Attributes
bool	single_threaded
size_t	jobid
int	nthreads
int	maxconcurrency
int	running
bool	shutting_down
int	shutdown_counter
vector< ThreadState * >	threads
vector< ThreadPool * >	thread_owners
priority_queue< Job *, vector< Job * >, JobCompare >	global_queue
vector< JobQueue * >	thread_queues
vector< Job * >	pending
ConditionVariable	cond
ConditionVariable	response

Friends
class	Job

Detailed Description

Definition at line 1654 of file shared.cc.

Constructor & Destructor Documentation

Scheduler()

LibThread::Scheduler::Scheduler ( int  n )

inline

Definition at line 1673 of file shared.cc.

                   :
    SharedObject(), threads(), thread_owners(), global_queue(), thread_queues(),
    single_threaded(n==0), nthreads(n == 0 ? 1 : n),
    lock(true), cond(&lock), response(&lock),
    shutting_down(false), shutdown_counter(0), jobid(0),
    maxconcurrency(n), running(0)
  {
    thread_queues.push_back(new JobQueue());
  }

~Scheduler()

virtual LibThread::Scheduler::~Scheduler ( )

inlinevirtual

Definition at line 1696 of file shared.cc.

                       {
    for (unsigned i = 0; i < thread_queues.size(); i++) {
      JobQueue *q = thread_queues[i];
      while (!q->empty()) {
        Job *job = q->front();
        q->pop();
        releaseShared(job);
      }
    }
    thread_queues.clear();
    threads.clear();
  }

Member Function Documentation

addThread()

void LibThread::Scheduler::addThread ( ThreadPool *  owner, ThreadState *  thread  )

inline

Definition at line 1735 of file shared.cc.

                                                         {
    lock.lock();
    thread_owners.push_back(owner);
    threads.push_back(thread);
    thread_queues.push_back(new JobQueue());
    lock.unlock();
  }

attachJob()

void LibThread::Scheduler::attachJob ( ThreadPool *  pool, Job *  job  )

inline

Definition at line 1742 of file shared.cc.

                                             {
    lock.lock();
    job->pool = pool;
    job->id = jobid++;
    acquireShared(job);
    if (job->ready()) {
      global_queue.push(job);
      cond.signal();
    }
    else if (job->pending_index < 0) {
      job->pool = pool;
      job->pending_index = pending.size();
      pending.push_back(job);
    }
    lock.unlock();
  }

broadcastJob()

void LibThread::Scheduler::broadcastJob ( ThreadPool *  pool, Job *  job  )

inline

Definition at line 1776 of file shared.cc.

                                                {
    lock.lock();
    for (unsigned i = 0; i <thread_queues.size(); i++) {
      if (thread_owners[i] == pool) {
        acquireShared(job);
        thread_queues[i]->push(job);
      }
    }
    lock.unlock();
  }

cancelDeps()

void LibThread::Scheduler::cancelDeps ( Job *  job )

inline

Definition at line 1786 of file shared.cc.

                             {
    vector<Job *> &notify = job->notify;
    for (unsigned i = 0; i <notify.size(); i++) {
      Job *next = notify[i];
      if (!next->cancelled) {
        cancelJob(next);
      }
    }
  }

cancelJob()

void LibThread::Scheduler::cancelJob ( Job *  job )

inline

Definition at line 1795 of file shared.cc.

                           {
    lock.lock();
    if (!job->cancelled) {
      job->cancelled = true;
      if (!job->running && !job->done) {
        job->done = true;
        cancelDeps(job);
      }
    }
    lock.unlock();
  }

clearThreadState()

void LibThread::Scheduler::clearThreadState ( )

inline

Definition at line 1826 of file shared.cc.

                          {
    threads.clear();
  }

detachJob()

void LibThread::Scheduler::detachJob ( Job *  job )

inline

Definition at line 1758 of file shared.cc.

                           {
    lock.lock();
    long i = job->pending_index;
    job->pending_index = -1;
    if (i >= 0) {
      job = pending.back();
      pending.resize(pending.size()-1);
      pending[i] = job;
      job->pending_index = i;
    }
    lock.unlock();
  }

get_maxconcurrency()

int LibThread::Scheduler::get_maxconcurrency ( )

inline

Definition at line 1685 of file shared.cc.

                           {
    return maxconcurrency;
  }

getThread()

ThreadState * LibThread::Scheduler::getThread ( int  i )

inline

Definition at line 1708 of file shared.cc.

{ return threads[i]; }

main()

static void * LibThread::Scheduler::main ( ThreadState *  ts, void *  arg  )

inlinestatic

Definition at line 1856 of file shared.cc.

                                                {
    SchedInfo *info = (SchedInfo *) arg;
    Scheduler *scheduler = info->scheduler;
    ThreadPool *oldThreadPool = currentThreadPoolRef;
    // TODO: set current thread pool
    // currentThreadPoolRef = pool;
    Lock &lock = scheduler->lock;
    ConditionVariable &cond = scheduler->cond;
    ConditionVariable &response = scheduler->response;
    JobQueue *my_queue = scheduler->thread_queues[info->num];
    if (!scheduler->single_threaded)
      thread_init();
    lock.lock();
    for (;;) {
      if (info->job && info->job->done)
        break;
      if (scheduler->shutting_down) {
        scheduler->shutdown_counter++;
        scheduler->response.signal();
        break;
      }
      if (!my_queue->empty()) {
       Job *job = my_queue->front();
       my_queue->pop();
       if (!scheduler->global_queue.empty())
         cond.signal();
       currentJobRef = job;
       job->run();
       currentJobRef = NULL;
       notifyDeps(scheduler, job);
       releaseShared(job);
       scheduler->response.signal();
       continue;
      } else if (!scheduler->global_queue.empty()) {
       Job *job = scheduler->global_queue.top();
       scheduler->global_queue.pop();
       if (!scheduler->global_queue.empty())
         cond.signal();
       currentJobRef = job;
       job->run();
       currentJobRef = NULL;
       notifyDeps(scheduler, job);
       releaseShared(job);
       scheduler->response.signal();
       continue;
      } else {
        if (scheduler->single_threaded) {
          break;
        }
        cond.wait();
      }
    }
    // TODO: correct current thread pool
    // releaseShared(currentThreadPoolRef);
    currentThreadPoolRef = oldThreadPool;
    scheduler->lock.unlock();
    delete info;
    return NULL;
  }

notifyDeps()

static void LibThread::Scheduler::notifyDeps ( Scheduler *  scheduler, Job *  job  )

inlinestatic

Definition at line 1829 of file shared.cc.

                                                         {
    vector<Job *> &notify = job->notify;
    job->incref(notify.size());
    for (unsigned i = 0; i <notify.size(); i++) {
      Job *next = notify[i];
      if (!next->queued && next->ready() && !next->cancelled) {
        next->queued = true;
        scheduler->queueJob(next);
      }
    }
    vector<Trigger *> &triggers = job->triggers;
    leftv arg = NULL;
    if (triggers.size() > 0 && job->result.size() > 0)
      arg = LinTree::from_string(job->result);
    for (unsigned i = 0; i < triggers.size(); i++) {
      Trigger *trigger = triggers[i];
      if (trigger->accept(arg)) {
        trigger->activate(arg);
        if (trigger->ready())
           scheduler->queueJob(trigger);
      }
    }
    if (arg) {
      arg->CleanUp();
      omFreeBin(arg, sleftv_bin);
    }
  }

queueJob()

void LibThread::Scheduler::queueJob ( Job *  job )

inline

Definition at line 1770 of file shared.cc.

                          {
    lock.lock();
    global_queue.push(job);
    cond.signal();
    lock.unlock();
  }

set_maxconcurrency()

void LibThread::Scheduler::set_maxconcurrency ( int  n )

inline

Definition at line 1682 of file shared.cc.

                                 {
    maxconcurrency = n;
  }

shutdown()

void LibThread::Scheduler::shutdown ( bool  wait )

inline

Definition at line 1709 of file shared.cc.

                           {
    if (single_threaded) {
      SchedInfo *info = new SchedInfo();
      info->num = 0;
      info->scheduler = this;
      acquireShared(this);
      info->job = NULL;
      Scheduler::main(NULL, info);
      return;
    }
    lock.lock();
    if (wait) {
      while (!global_queue.empty()) {
        response.wait();
      }
    }
    shutting_down = true;
    while (shutdown_counter < nthreads) {
      cond.broadcast();
      response.wait();
    }
    lock.unlock();
    for (unsigned i = 0; i <threads.size(); i++) {
      joinThread(threads[i]);
    }
  }

threadpool_size()

int LibThread::Scheduler::threadpool_size ( ThreadPool *  pool )

inline

Definition at line 1688 of file shared.cc.

                                        {
    int n;
    for (unsigned i = 0; i <thread_owners.size(); i++) {
      if (thread_owners[i] == pool)
        n++;
    }
    return n;
  }

waitJob()

void LibThread::Scheduler::waitJob ( Job *  job )

inline

Definition at line 1806 of file shared.cc.

                         {
    if (single_threaded) {
      SchedInfo *info = new SchedInfo();
      info->num = 0;
      info->scheduler = this;
      acquireShared(this);
      info->job = job;
      Scheduler::main(NULL, info);
    } else {
      lock.lock();
      for (;;) {
        if (job->done || job->cancelled) {
          break;
        }
        response.wait();
      }
      response.signal(); // forward signal
      lock.unlock();
    }
  }

Friends And Related Function Documentation

Job

friend class Job

friend

Definition at line 1670 of file shared.cc.

Field Documentation

cond

ConditionVariable LibThread::Scheduler::cond

private

Definition at line 1668 of file shared.cc.

global_queue

priority_queue<Job *, vector<Job *>, JobCompare> LibThread::Scheduler::global_queue

private

Definition at line 1665 of file shared.cc.

jobid

size_t LibThread::Scheduler::jobid

private

Definition at line 1657 of file shared.cc.

lock

Lock LibThread::Scheduler::lock

Definition at line 1672 of file shared.cc.

maxconcurrency

int LibThread::Scheduler::maxconcurrency

private

Definition at line 1659 of file shared.cc.

nthreads

int LibThread::Scheduler::nthreads

private

Definition at line 1658 of file shared.cc.

pending

vector<Job *> LibThread::Scheduler::pending

private

Definition at line 1667 of file shared.cc.

response

ConditionVariable LibThread::Scheduler::response

private

Definition at line 1669 of file shared.cc.

running

int LibThread::Scheduler::running

private

Definition at line 1660 of file shared.cc.

shutdown_counter

int LibThread::Scheduler::shutdown_counter

private

Definition at line 1662 of file shared.cc.

shutting_down

bool LibThread::Scheduler::shutting_down

private

Definition at line 1661 of file shared.cc.

single_threaded

bool LibThread::Scheduler::single_threaded

private

Definition at line 1656 of file shared.cc.

thread_owners

vector<ThreadPool *> LibThread::Scheduler::thread_owners

private

Definition at line 1664 of file shared.cc.

thread_queues

vector<JobQueue *> LibThread::Scheduler::thread_queues

private

Definition at line 1666 of file shared.cc.

threads

vector<ThreadState *> LibThread::Scheduler::threads

private

Definition at line 1663 of file shared.cc.

---

The documentation for this class was generated from the following file:

• Singular/dyn_modules/systhreads/shared.cc