ThreadPool.c File Reference

#include "Config.h"
#include <assert.h>
#include <stdlib.h>
#include "ThreadPool.h"

Go to the source code of this file.

Data Structures
struct	ThreadPoolWork
	An opaque handle to a ThreadFunction submitted with ThreadPool::execute. More...

Macros
#define	_Class   _ThreadPool

Enumerations
enum	ThreadPoolWorkState { ThreadPoolWorkPending , ThreadPoolWorkRunning , ThreadPoolWorkFinished }
	The states of a ThreadPoolWork. More...

Functions
Class *	_ThreadPool (void)
static ident	_wait (ThreadPool *self, ThreadPoolWork *work)
static Object *	copy (const Object *self)
static void	dealloc (Object *self)
static void	dequeue (ThreadPool *self, ThreadPoolWork *work)
	Removes work from the queue.
static void	detach (ThreadPool *self, ThreadFunction function, ident data)
static ThreadPoolWork *	execute (ThreadPool *self, ThreadFunction function, ident data)
static ThreadPool *	init (ThreadPool *self)
static void	initialize (Class *clazz)
static ThreadPool *	initWithSize (ThreadPool *self, size_t size)
static ident	run (Thread *thread)
	ThreadFunction for the ThreadPool Threads.
static void	runOnCallingThread (ThreadPoolWork *work)
	Runs work on the calling thread, with a Thread that is never started.
static ThreadPoolWork *	submit (ThreadPool *self, ThreadFunction function, ident data, bool isDetached)
	Queues function, or runs it on the calling thread if this ThreadPool has no Threads.

Macro Definition Documentation

_Class

#define _Class   _ThreadPool

Definition at line 39 of file ThreadPool.c.

Enumeration Type Documentation

ThreadPoolWorkState

enum ThreadPoolWorkState

The states of a ThreadPoolWork.

Enumerator
ThreadPoolWorkPending
ThreadPoolWorkRunning
ThreadPoolWorkFinished

Definition at line 44 of file ThreadPool.c.

             {
  ThreadPoolWorkPending,
  ThreadPoolWorkRunning,
  ThreadPoolWorkFinished,
} ThreadPoolWorkState;

Function Documentation

_ThreadPool()

Class * _ThreadPool

(

void

)

Definition at line 375 of file ThreadPool.c.

                         {
  static Class *clazz;
  static Once once;
 
  do_once(&once, {
    clazz = _initialize(&(const ClassDef) {
      .name = "ThreadPool",
      .superclass = _Object(),
      .instanceSize = sizeof(ThreadPool),
      .interfaceSize = sizeof(ThreadPoolInterface),
      .initialize = initialize,
    });
  });
 
  return clazz;
}

_wait()

static ident _wait ( ThreadPool *  self, ThreadPoolWork *  work  )

static

Definition at line 321 of file ThreadPool.c.

                                                           {
 
  if (work == NULL) {
    return NULL;
  }
 
  assert(work->pool == self);
  assert(work->isDetached == false);
 
  bool isPending = false;
 
  synchronized(self->condition, {
    if (work->state == ThreadPoolWorkPending) {
      dequeue(self, work);
      work->state = ThreadPoolWorkRunning;
      isPending = true;
    } else {
      while (work->state != ThreadPoolWorkFinished) {
        $(self->condition, wait);
      }
    }
  });
 
  if (isPending) {
    runOnCallingThread(work);
  }
 
  const ident result = work->result;
  free(work);
 
  return result;
}

copy()

static Object * copy ( const Object *  self )

static

See also

Object::copy(const Object *)

Definition at line 92 of file ThreadPool.c.

                                        {
  return NULL;
}

dealloc()

static void dealloc ( Object *  self )

static

See also

Object::dealloc(Object *)

Definition at line 99 of file ThreadPool.c.

                                  {
 
  ThreadPool *this = (ThreadPool *) self;
 
  const Array *threads = this->threads;
 
  synchronized(this->condition, {
    for (size_t i = 0; i < threads->count; i++) {
      $((Thread *) $(threads, objectAtIndex, i), cancel);
    }
    $(this->condition, broadcast);
  });
 
  for (size_t i = 0; i < threads->count; i++) {
    $((Thread *) $(threads, objectAtIndex, i), join, NULL);
  }
 
  assert(this->head == NULL);
 
  release(this->threads);
 
  release(this->condition);
 
  super(Object, self, dealloc);
}

dequeue()

static void dequeue ( ThreadPool *  self, ThreadPoolWork *  work  )

static

Removes work from the queue.

Remarks

The caller MUST hold the condition.

Definition at line 218 of file ThreadPool.c.

                                                            {
 
  ThreadPoolWork *previous = NULL;
  for (ThreadPoolWork *w = self->head; w; previous = w, w = w->next) {
    if (w == work) {
      if (previous) {
        previous->next = work->next;
      } else {
        self->head = work->next;
      }
      if (self->tail == work) {
        self->tail = previous;
      }
      work->next = NULL;
      return;
    }
  }
 
  assert(false);
}

detach()

static void detach ( ThreadPool *  self, ThreadFunction  function, ident  data  )

static

Definition at line 184 of file ThreadPool.c.

                                                                          {
  submit(self, function, data, true);
}

execute()

static ThreadPoolWork * execute ( ThreadPool *  self, ThreadFunction  function, ident  data  )

static

Definition at line 192 of file ThreadPool.c.

                                                                                      {
  return submit(self, function, data, false);
}

init()

static ThreadPool * init ( ThreadPool *  self )

static

Definition at line 200 of file ThreadPool.c.

                                          {
 
#if defined(_WIN32)
  const size_t size = GetActiveProcessorCount(ALL_PROCESSOR_GROUPS);
#elif defined(_SC_NPROCESSORS_ONLN)
  const long processors = sysconf(_SC_NPROCESSORS_ONLN);
  const size_t size = processors > 0 ? (size_t) processors : 1;
#else
  const size_t size = 1;
#endif
 
  return $(self, initWithSize, size);
}

initialize()

static void initialize ( Class *  clazz )

static

See also

Class::initialize(Class *)

Definition at line 359 of file ThreadPool.c.

                                     {
 
  ((ObjectInterface *) clazz->interface)->copy = copy;
  ((ObjectInterface *) clazz->interface)->dealloc = dealloc;
 
  ((ThreadPoolInterface *) clazz->interface)->detach = detach;
  ((ThreadPoolInterface *) clazz->interface)->execute = execute;
  ((ThreadPoolInterface *) clazz->interface)->init = init;
  ((ThreadPoolInterface *) clazz->interface)->initWithSize = initWithSize;
  ((ThreadPoolInterface *) clazz->interface)->wait = _wait;
}

initWithSize()

static ThreadPool * initWithSize ( ThreadPool *  self, size_t  size  )

static

Definition at line 290 of file ThreadPool.c.

                                                               {
 
  self = (ThreadPool *) super(Object, self, init);
  if (self) {
 
    self->condition = $(alloc(Condition), init);
    assert(self->condition);
 
    self->size = size;
 
    self->threads = $(alloc(Array), initWithCapacity, size);
    assert(self->threads);
 
    for (size_t i = 0; i < self->size; i++) {
      Thread *thread = $(alloc(Thread), initWithFunction, run, self);
      assert(thread);
 
      $(self->threads, addObject, thread);
      $(thread, start);
 
      release(thread);
    }
  }
 
  return self;
}

run()

static ident run ( Thread *  thread )

static

ThreadFunction for the ThreadPool Threads.

Remarks

thread->data is the user data of the current ThreadPoolWork, so the ThreadPool is kept in a local.

Definition at line 244 of file ThreadPool.c.

                                 {
 
  ThreadPool *self = thread->data;
  Condition *condition = self->condition;
 
  $((Lock *) condition, lock);
 
  while (true) {
 
    while (self->head == NULL && $(thread, isCancelled) == false) {
      $(condition, wait);
    }
 
    ThreadPoolWork *work = self->head;
    if (work == NULL) {
      break;
    }
 
    dequeue(self, work);
    work->state = ThreadPoolWorkRunning;
 
    $((Lock *) condition, unlock);
 
    thread->data = work->data;
    const ident result = work->function(thread);
 
    $((Lock *) condition, lock);
 
    if (work->isDetached) {
      free(work);
    } else {
      work->result = result;
      work->state = ThreadPoolWorkFinished;
      $(condition, broadcast);
    }
  }
 
  $((Lock *) condition, unlock);
 
  return NULL;
}

runOnCallingThread()

static void runOnCallingThread ( ThreadPoolWork *  work )

static

Runs work on the calling thread, with a Thread that is never started.

Definition at line 130 of file ThreadPool.c.

                                                     {
 
  Thread *thread = $(alloc(Thread), initWithFunction, work->function, work->data);
  assert(thread);
 
  work->result = work->function(thread);
  work->state = ThreadPoolWorkFinished;
 
  release(thread);
}

submit()

static ThreadPoolWork * submit ( ThreadPool *  self, ThreadFunction  function, ident  data, bool  isDetached  )

static

Queues function, or runs it on the calling thread if this ThreadPool has no Threads.

Returns

The ThreadPoolWork, or NULL if isDetached.

Definition at line 145 of file ThreadPool.c.

                                                                                                      {
 
  assert(function);
 
  ThreadPoolWork *work = calloc(1, sizeof(ThreadPoolWork));
  assert(work);
 
  work->pool = self;
  work->state = ThreadPoolWorkPending;
  work->isDetached = isDetached;
  work->function = function;
  work->data = data;
 
  if (self->size == 0) {
    runOnCallingThread(work);
    if (isDetached) {
      free(work);
      return NULL;
    }
    return work;
  }
 
  synchronized(self->condition, {
    if (self->tail) {
      self->tail->next = work;
    } else {
      self->head = work;
    }
    self->tail = work;
    $(self->condition, broadcast);
  });
 
  return isDetached ? NULL : work;
}