job.C

/*
 * This file is part of the "Archon" framework.
 * (http://files3d.sourceforge.net)
 *
 * Copyright © 2002 by Kristian Spangsege and Brian Kristiansen.
 *
 * Permission to use, copy, modify, and distribute this software and
 * its documentation under the terms of the GNU General Public License is
 * hereby granted. No representations are made about the suitability of
 * this software for any purpose. It is provided "as is" without express
 * or implied warranty. See the GNU General Public License
 * (http://www.gnu.org/copyleft/gpl.html) for more details.
 *
 * The characters in this file are ISO8859-1 encoded.
 *
 * The documentation in this file is in "Doxygen" style
 * (http://www.doxygen.org).
 */

#include <iostream>
#include <algorithm>
#include <archon/util/logger.H>
#include <archon/util/text.H>

#include <archon/util/job.H>

namespace Archon
{
  namespace Utilities
  {
    using namespace std;

    static inline void log(string m)
    {
      Logger::get()->log(m);
    }

    void Job::Queue::worker()
    {
      unsigned long id;

      {
        Mutex::Lock l(mutex);
        id = nextThreadId++;
      }

      //log((string)"Job::Queue::worker(" + Text::toString(id) + "): started");
    
      for(;;)
      {
        Mutex::Lock l(mutex);

        if(pendingJobs.empty())
        {
          --activeThreads;
          idleThreads.push_back(id);

          if(activeThreads == 0) allDone.notifyAll();

          Time wakeup = Time::now();
          wakeup += maxIdleTime;

          for(;;)
          {
            //log((string)"Job::Queue::worker(" + Text::toString(id) + "): going to sleep");

            const bool timedOut = newJob.timedWait(wakeup);

            if(shutdown)
            {
              /*
               * Note that the contents of 'idleThreads' is undefined after
               * 'shutdown' is set to true. The only thing that can be relied on
               * is that the size of the list equals the number of threads that
               * still have not ended their duty.
               */
              idleThreads.pop_back();
              //log((string)"Job::Queue::worker(" + Text::toString(id) + "): terminated");
              threadQuit.notifyAll();
              return;
            }

            if(timedOut)
            {
              idleThreads.erase(find(idleThreads.begin(), idleThreads.end(), id));
              if(!pendingJobs.empty()) break;
              //log((string)"Job::Queue::worker(" + Text::toString(id) + "): timed out");
              threadQuit.notifyAll();
              return;
            }

            if(pendingJobs.empty() || idleThreads.back() != id) continue;
            idleThreads.pop_back();
            break;
          }

          ++activeThreads;
        }

        Ref<Job> job = pendingJobs.front();
        pendingJobs.pop_front();

        job->thread = Thread::self().get();

        l.release();

        //log((string)"Job::Queue::worker(" + Text::toString(id) + "): running new job");
        job->main();

        {
          Mutex::Lock l(mutex);
          job->thread = 0;
        }

        Thread::selfResurrect();
      }
    }

    Job::Queue::Queue(int maxThreads, Time maxIdleTime):
      maxThreads(maxThreads), maxIdleTime(maxIdleTime),
      newJob(mutex), threadQuit(mutex), allDone(mutex),
      nextThreadId(0), activeThreads(0), shutdown(false)
    {
    }

    Job::Queue::~Queue()
    {{ // The extra scope is needed to work around gcc3.2 bug #8287
      Mutex::Lock l(mutex);

      // Wait for all thread to become idle
      while(!pendingJobs.empty() || activeThreads) allDone.wait();

      // Shutdown all threads
      shutdown = true;
      newJob.notifyAll();

      /*
       * Note that the contents of 'idleThreads' is undefined after
       * 'shutdown' is set to true. The only thing that can be relied on
       * is that the size of the list equals the number of threads that
       * still have not ended their duty.
       */
      while(!idleThreads.empty()) threadQuit.wait();
    }}

    void Job::Queue::workerEntry(Job::Queue *q)
    {
      q->worker();
    }

    void Job::Queue::add(Ref<Job> j)
    {
      Mutex::Lock l(mutex);
      if(shutdown) ARCHON_THROW1(StateException, "Shutting down");
      pendingJobs.push_back(j);
      if(idleThreads.empty() && activeThreads < maxThreads)
      {
        ++activeThreads;
        Thread::run(workerEntry, this);
      }
      newJob.notifyAll();
    }

    void Job::Queue::cancel(Ref<Job> j)
    {
      Mutex::Lock l(mutex);
      if(j->thread)
      {
        Ref<Thread> t(j->thread, RefSafeIncTag());
        l.release();
        if(!t) return;
        t->terminate();
      }
      else pendingJobs.remove(j);
    }

    void Job::Queue::wait()
    {
      Mutex::Lock l(mutex);
      while(!pendingJobs.empty() || activeThreads > 0) allDone.wait();
    }
  }
}

---