Tanl Linguistic Pipeline

search/ixe/Common/ThreadPool.cpp

/*
**      IXE C++ Library
**      Common/ThreadPool.cpp
** ----------------------------------------------------------------------
**      Copyright (c) 2000  Ideare SpA. All rights reserved.
**      Copyright (c) 2000  Giuseppe Attardi (attardi@di.unipi.it),
**                          Antonio Cisternino (cisterni@di.unipi.it)
** ----------------------------------------------------------------------
**
**  This file is part of IXE.
**
**  IXE is free software; you can redistribute it and/or modify it
**  under the terms of the GNU General Public License, version 3,
**  as published by the Free Software Foundation.
**
**  IXE is distributed in the hope that it will be useful,
**  but WITHOUT ANY WARRANTY; without even the implied warranty of
**  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
**  GNU General Public License for more details.
**
**  You should have received a copy of the GNU General Public License
**  along with this program.  If not, see <http://www.gnu.org/licenses/>.
**  ----------------------------------------------------------------------
*/

// Settings
#include "ThreadPool.h"

// standard
#include <cstdlib>              /* for exit(2) */
#include <ctime>
#ifdef _WIN32
#include <process.h>            // _beginthreadex
#endif
#include <iostream>

// local
#include "include/exit_codes.h"
#include "Common/util.h"
#include "server/server_util.h"

using namespace std;

namespace IXE {

#ifndef _WIN32
pthread_key_t   jmpbuf_key;
#endif

#define Notify(event)   event.notify()

/*
No need to use these. Cancel is deferred by deafult in pthreads. Test for
cancellation happens at pthread_cond_wait() and pthread_cond_timedwait().
#ifdef _WIN32
#define DeferCancelWhile(stat) stat
#else
#define DeferCancelWhile(stat) \
        { int cancel_type; \
        ::pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, &cancel_type); \
        stat; \
        ::pthread_setcanceltype(cancel_type, 0); \
        ::pthread_testcancel(); }
#endif
*/

//=============================================================================
//=============================================================================

extern "C" void ThreadPool::Thread::Destroy(void *p)
{
# ifdef DEBUG_threads
  cerr << "ThreadPool::Thread::Destroy()" << endl;
# endif

  // This is called either:
  // a. at thread termination, or
  // b. from delete, issuing an ExitThread().
  //    Since ExitThread() does not return, delete was not completed
  //    and we must reissue it.
  ThreadPool::Thread* const t = static_cast<ThreadPool::Thread *>(p);
  if (t) {
    bool deleted;
    Locking(t->pool.threads,
            deleted = (t->pool.threads->find(t) == t->pool.threads->end()));
    if (!deleted) {
      //
      // The thread's destructor has not been called.
      // We are being called after an ExitThread() or TerminateThread().
      // Set the destroying flag to skip thread termination within delete.
      //
      t->destroying = true;
      delete t;
    }
  }
}

//=============================================================================
//=============================================================================

TRESULT
ThreadPool::Thread::Main(void* p)
{
# ifndef _WIN32

  // key where to store jmp_buf for error recovery:
  ::pthread_key_create(&jmpbuf_key, 0);

  //
  // Add our clean-up function to this thread.  This will ensure
  // that we get removed from our thread pool's set of threads even if
  // the derived class author calls pthread_exit() directly.
  //
  pthread_cleanup_push(Destroy, p); // no "::" (it's a macro)

# endif // _WIN32

  ThreadPool::Thread* const thread = static_cast<ThreadPool::Thread*>(p);
  register ThreadPool& pool = thread->pool;
  ThreadPool::Thread::TaskDescr task((void*)0);

  while (true) {

#   ifdef DEBUG_threads
    cerr << "Thread::Main(): waiting for task" << endl;
#   endif

    { // mutex on task queue
      LockUp taskQueue(pool.tasks._lock);

      while (pool.tasks->empty()) {
        Count current_threads;
        Locking(pool.threads,
                current_threads = pool.threads->size());
        if (current_threads <= pool.minThreads) {
          //
          // There are no threads beyond those originally
          // created: wait indefinitely for a task.
          //
          Notify(pool.idle_thread);
          ConditionWait(pool.task_available, &pool.tasks._lock);
        } else {
          //
          // We have become idle
          //
          Notify(pool.idle_thread);
          //
          // Wait only a finite amount of time for a task to
          // become available.
          //
          int result = 0;
          ConditionWaitInterval(pool.task_available, &pool.tasks._lock,
                                pool.timeout * 1000, 0, result);
          if (result == WAIT_TIMEOUT) {
            //
            // Retest the condition that there are still more threads than
            // the minimum.
            // We want to be absolutely sure before we commit suicide.
            //
            Locking(pool.threads,
                    current_threads = pool.threads->size());
            if (current_threads > pool.minThreads) {
              //
              // No task became available: commit suicide by
              // deleting the instance of the thread.
              //
              LeaveCriticalSection(&pool.tasks._lock);
              delete thread; // exits the thread so it does not return
            }
          }
        }
      }

#     ifdef DEBUG_threads
      cerr << "Thread::Main(): got task" << endl;
#     endif

      task = pool.tasks->front();
      pool.tasks->pop();

    } // unlock pool.tasks

    Locking(pool.busyThreads,
            ++(*pool.busyThreads));

#   ifdef DEBUG_threads
    cerr << "Thread::Main(): performing task" << endl;
#   endif

    thread->Run(task);          // do the real work

#   ifdef DEBUG_threads
    cerr << "Thread::Main(): completed task" << endl;
#   endif

    Locking(pool.busyThreads,
            --(*pool.busyThreads));
  }
  //
  // Although we never get here due to the infinite loop above, the
  // Pthread standard requires a pthread_cleanup_pop() within
  // the same scope of a pthread_cleanup_push().
  //
# ifndef _WIN32
  pthread_cleanup_pop(0);       // no "::" (it's a macro)
# endif
  return 0;                     // just make the compiler happy
}

//=============================================================================
//=============================================================================

ThreadPool::Thread::Thread(ThreadPool& p)
  : pool(p), destroying(false)
{ }

//=============================================================================
//=============================================================================
void
ThreadPool::Thread::Start()
{
# ifdef DEBUG_threads
  cerr << "Thread::Start()" << endl;
# endif

# ifdef _WIN32
  unsigned int id;
  thread = (HANDLE)_beginthreadex(NULL, 0, Main, (LPVOID)this, 0, &id);
  bool result = !thread;
# else
  ::pthread_attr_t attr;
  ::pthread_attr_init(&attr);
  ::pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
  int const result = ::pthread_create(&thread, &attr, Main, this);
# endif
  if (result) {
    cerr << pool.process_name << "could not create thread: "
         << ::strerror(result) << endl;
    ::exit(Exit_No_Create_Thread);
  }
}

//=============================================================================
//=============================================================================

/* virtual */
ThreadPool::Thread::~Thread()
{
# ifdef DEBUG_threads
  cerr << "Thread::~Thread()" << endl;
# endif

  if (!destroying) {
    // a genuine call to delete: also terminate the thread.
    if (thread == GetCurrentThread()) {
      // committing suicide: e.g. no tasks are available and
      // threads->size() > min_thread.
      //
      // ExitThread() will cause Thread::Destroy() to be called, but
      // will not return, hence the delete will not complete.
      // Thread::Destroy() will call back here with (destryoing == true)
      // and complete the task.
      ExitThread(0);      // calls both cancellation and data destructors
    } else {
      // With pthreads this only signals the thread to terminate:
      // cancel is deferred until next cancellation point.
      // Thread::Destroy() will be called, but it will notice that this
      // is no longer in the pool.
      // FIXME: on WIN32 this kills the thread without freeing its stack.
      TerminateThread(thread, 0); // does not call data destructors.
    }
  }

  // do this here, since it is used to signal to Thread::Destroy() the
  // delete completion.
  if (!(*pool.destructing)) {
    //
    // Delete the pointer to us in our thread pool's set of threads.
    //
    Locking(pool.threads,
            pool.threads->erase(this));
  }
}

ThreadPool::ThreadPool(char const* process_name,
                       Thread const* prototype,
                       int min_threads, int max_threads, int timeout) :
  process_name(process_name), minThreads(min_threads),
  maxThreads(max_threads),
  destructing(false), timeout(timeout),
  busyThreads(0)
{
  init(prototype);
}

ThreadPool::ThreadPool(char const* process_name,
                       int min_threads, int max_threads, int timeout) :
  process_name(process_name), minThreads(min_threads),
  maxThreads(max_threads),
  destructing(false), timeout(timeout),
  busyThreads(0)
{ }

void ThreadPool::init(Thread const* prototype)
{
    Locking(threads,
            if (threads->empty()) { // can only be done once
              ((Thread*)prototype)->Start();
              threads->insert((Thread*)prototype);
              for (Count i = 1; i < minThreads; ++i)
                threads->insert(prototype->clone(*this));
            });
}

//=============================================================================
//=============================================================================

ThreadPool::~ThreadPool()
{
  //
  // Set the destructing flag to prevent the thread destructor from
  // removing itself from our set since we're deleting the whole set
  // anyway, and, more importantly, to make it call pthread_cancel() on
  // its thread rather than calling pthread_exit() on our thread.
  //
  Locking(destructing,
          destructing = true);

  Locking(threads, {
    for (ThreadSet::iterator t = threads->begin(); t != threads->end(); ++t)
      delete *t;
    });
}

//=============================================================================
//=============================================================================

#define MAX_PENDING 1

void ThreadPool::AddTask(Thread::TaskDescr task)
{
# ifdef DEBUG_threads
  cerr << "ThreadPool::AddTask()" << endl;
# endif

  // we add immediately to the queue, risking overbooking up to MAX_PENDING,
  // in the hope that some thread may take over the task without having to wait
  Locking(tasks, {
    tasks->push(task);
    Notify(task_available);
  });

  Locking(threads, {
    // no need to lock busyThreads, since we are only reading it
    Count accepted = (Count)busyThreads + tasks->size();
    // cerr << "Accepted: " << accepted << endl;
    if (accepted >= maxThreads + MAX_PENDING) {
      //
      // We've reached the max number of threads we can make,
      // so just wait until one becomes idle.
      //
      ConditionWait(idle_thread, &threads._lock);
    } else if (busyThreads == threads->size()
               && threads->size() < maxThreads) {
      //
      // All the existing threads are busy
      // but we haven't reached the max number of threads we can make,
      // so create another one to handle the request.
      //
      threads->insert((*threads->begin())->clone(*this));
    }
  });
}

} // namespace IXE

 

Copyright © 2005-2011 G. Attardi. Generated on 3 Mar 2011 by doxygen 1.6.1.