#include <sys/param.h>
#include "ThreadPool.h"
#include "FreeList.h"
#include <assert.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>

Include dependency graph for ThreadPool.c:

Functions
static long	DiffMillis (struct timeval time1, struct timeval time2)
	Returns the difference in milliseconds between two timeval structures. More...

static UPNP_INLINE void	StatsInit (ThreadPoolStats *stats)

static UPNP_INLINE void	StatsAccountLQ (ThreadPool *tp, long diffTime)

static UPNP_INLINE void	StatsAccountMQ (ThreadPool *tp, long diffTime)

static UPNP_INLINE void	StatsAccountHQ (ThreadPool *tp, long diffTime)

static UPNP_INLINE void	CalcWaitTime (ThreadPool tp, ThreadPriority p, ThreadPoolJob job)

static UPNP_INLINE time_t	StatsTime (time_t *t)

static int	CmpThreadPoolJob (void jobA, void jobB)
	Compares thread pool jobs. More...

static void	FreeThreadPoolJob (ThreadPool tp, ThreadPoolJob tpj)
	Deallocates a dynamically allocated ThreadPoolJob. More...

static int	SetPolicyType (PolicyType in)
	Sets the scheduling policy of the current process. More...

static int	SetPriority (ThreadPriority priority)
	Sets the priority of the currently running thread. More...

static void	BumpPriority (ThreadPool *tp)
	Determines whether any jobs need to be bumped to a higher priority Q and bumps them. More...

static void	SetRelTimeout (struct timespec *time, int relMillis)
	Sets the fields of the passed in timespec to be relMillis milliseconds in the future. More...

static void	SetSeed (void)
	Sets seed for random number generator. Each thread sets the seed random number generator. More...

static void *	WorkerThread (void *arg)
	Implements a thread pool worker. Worker waits for a job to become available. Worker picks up persistent jobs first, high priority, med priority, then low priority. More...

static ThreadPoolJob *	CreateThreadPoolJob (ThreadPoolJob job, int id, ThreadPool tp)
	Creates a Thread Pool Job. (Dynamically allocated) More...

static int	CreateWorker (ThreadPool *tp)
	Creates a worker thread, if the thread pool does not already have max threads. More...

static void	AddWorker (ThreadPool *tp)
	Determines whether or not a thread should be added based on the jobsPerThread ratio. Adds a thread if appropriate. More...

int	ThreadPoolInit (ThreadPool tp, ThreadPoolAttr attr)
	Initializes and starts ThreadPool. Must be called first and only once for ThreadPool. More...

int	ThreadPoolAddPersistent (ThreadPool tp, ThreadPoolJob job, int *jobId)
	Adds a persistent job to the thread pool. More...

int	ThreadPoolAdd (ThreadPool tp, ThreadPoolJob job, int *jobId)
	Adds a job to the thread pool. Job will be run as soon as possible. More...

int	ThreadPoolRemove (ThreadPool tp, int jobId, ThreadPoolJob out)
	Removes a job from the thread pool. Can only remove jobs which are not currently running. More...

int	ThreadPoolGetAttr (ThreadPool tp, ThreadPoolAttr out)
	Gets the current set of attributes associated with the thread pool. More...

int	ThreadPoolSetAttr (ThreadPool tp, ThreadPoolAttr attr)
	Sets the attributes for the thread pool. Only affects future calculations. More...

int	ThreadPoolShutdown (ThreadPool *tp)
	Shuts the thread pool down. Waits for all threads to finish. May block indefinitely if jobs do not exit. More...

int	TPAttrInit (ThreadPoolAttr *attr)
	Initializes thread pool attributes. Sets values to defaults defined in ThreadPool.h. More...

int	TPJobInit (ThreadPoolJob job, start_routine func, void arg)
	Initializes thread pool job. Sets the priority to default defined in ThreadPool.h. Sets the free_routine to default defined in ThreadPool.h. More...

int	TPJobSetPriority (ThreadPoolJob *job, ThreadPriority priority)
	Sets the max threads for the thread pool attributes. More...

int	TPJobSetFreeFunction (ThreadPoolJob *job, free_routine func)
	Sets the max threads for the thread pool attributes. More...

int	TPAttrSetMaxThreads (ThreadPoolAttr *attr, int maxThreads)
	Sets the max threads for the thread pool attributes. More...

int	TPAttrSetMinThreads (ThreadPoolAttr *attr, int minThreads)
	Sets the min threads for the thread pool attributes. More...

int	TPAttrSetStackSize (ThreadPoolAttr *attr, size_t stackSize)
	Sets the stack size for the thread pool attributes. More...

int	TPAttrSetIdleTime (ThreadPoolAttr *attr, int idleTime)
	Sets the idle time for the thread pool attributes. More...

int	TPAttrSetJobsPerThread (ThreadPoolAttr *attr, int jobsPerThread)
	Sets the jobs per thread ratio. More...

int	TPAttrSetStarvationTime (ThreadPoolAttr *attr, int starvationTime)
	Sets the starvation time for the thread pool attributes. More...

int	TPAttrSetSchedPolicy (ThreadPoolAttr *attr, PolicyType schedPolicy)
	Sets the scheduling policy for the thread pool attributes. More...

int	TPAttrSetMaxJobsTotal (ThreadPoolAttr *attr, int maxJobsTotal)
	Sets the maximum number jobs that can be qeued totally. More...

Function Documentation

static void AddWorker ( ThreadPool * tp )

static

Determines whether or not a thread should be added based on the jobsPerThread ratio. Adds a thread if appropriate.

Remarks: The ThreadPool object mutex must be locked prior to calling this function.

Parameters

tp	A pointer to the ThreadPool object.

References THREADPOOL::attr, THREADPOOL::busyThreads, CreateWorker(), THREADPOOL::highJobQ, THREADPOOLATTR::jobsPerThread, THREADPOOL::lowJobQ, THREADPOOL::medJobQ, THREADPOOL::persistentThreads, LINKEDLIST::size, and THREADPOOL::totalThreads.

Referenced by ThreadPoolAdd().

static void BumpPriority ( ThreadPool * tp )

static

Determines whether any jobs need to be bumped to a higher priority Q and bumps them.

tp->mutex must be locked.

Returns

Parameters

tp .

References THREADPOOL::attr, DiffMillis(), LINKEDLIST::head, THREADPOOL::highJobQ, ListAddTail(), ListDelNode(), THREADPOOL::lowJobQ, THREADPOOLATTR::maxIdleTime, THREADPOOL::medJobQ, LINKEDLIST::size, and THREADPOOLATTR::starvationTime.

Referenced by WorkerThread().

static int CmpThreadPoolJob	(	void *	jobA,
		void *	jobB
	)

static

Compares thread pool jobs.

Referenced by ThreadPoolInit().

static ThreadPoolJob* CreateThreadPoolJob	(	ThreadPoolJob *	job,
		int	id,
		ThreadPool *	tp
	)

static

Creates a Thread Pool Job. (Dynamically allocated)

Returns: ThreadPoolJob *on success, NULL on failure.

Parameters

job	job is copied.
id	id of job.
tp	.

References FreeListAlloc(), and THREADPOOL::jobFreeList.

Referenced by ThreadPoolAdd(), and ThreadPoolAddPersistent().

static int CreateWorker ( ThreadPool * tp )

static

Creates a worker thread, if the thread pool does not already have max threads.

Remarks: The ThreadPool object mutex must be locked prior to calling this function.

Returns

0 on success, < 0 on failure.
EMAXTHREADS if already max threads reached.
EAGAIN if system can not create thread.

Parameters

tp	A pointer to the ThreadPool object.

References THREADPOOL::attr, THREADPOOLATTR::maxThreads, THREADPOOL::mutex, THREADPOOL::pendingWorkerThreadStart, THREADPOOLATTR::stackSize, THREADPOOL::start_and_shutdown, THREADPOOL::stats, THREADPOOL::totalThreads, and WorkerThread().

Referenced by AddWorker(), ThreadPoolAddPersistent(), ThreadPoolInit(), and ThreadPoolSetAttr().

static long DiffMillis	(	struct timeval *	time1,
		struct timeval *	time2
	)

static

Returns the difference in milliseconds between two timeval structures.

Returns: The difference in milliseconds, time1-time2.

Parameters

time1	.
time2	.

Referenced by BumpPriority().

static void FreeThreadPoolJob	(	ThreadPool *	tp,
		ThreadPoolJob *	tpj
	)

static

Deallocates a dynamically allocated ThreadPoolJob.

Parameters

tp	.
tpj	Must be allocated with CreateThreadPoolJob.

References FreeListFree(), and THREADPOOL::jobFreeList.

Referenced by ThreadPoolAdd(), ThreadPoolRemove(), ThreadPoolShutdown(), and WorkerThread().

static int SetPolicyType ( PolicyType in )

static

Sets the scheduling policy of the current process.

Returns

0 on success.
result of GetLastError() on failure.

Parameters

in .

Referenced by ThreadPoolInit(), and ThreadPoolSetAttr().

static int SetPriority ( ThreadPriority priority )

static

Sets the priority of the currently running thread.

Returns

0 on success.
EINVAL invalid priority or the result of GerLastError.

Parameters

priority .

Referenced by WorkerThread().

static void SetRelTimeout	(	struct timespec *	time,
		int	relMillis
	)

static

Sets the fields of the passed in timespec to be relMillis milliseconds in the future.

Parameters

time	.
relMillis	milliseconds in the future.

Referenced by WorkerThread().

static void SetSeed ( void )

static

Sets seed for random number generator. Each thread sets the seed random number generator.

Referenced by WorkerThread().

int ThreadPoolAdd	(	ThreadPool *	tp,
		ThreadPoolJob *	job,
		int *	jobId
	)

Adds a job to the thread pool. Job will be run as soon as possible.

Returns

0 on success, nonzero on failure.
EOUTOFMEM if not enough memory to add job.

Parameters

tp	valid thread pool pointer.
job	.
jobId	id of job.

References AddWorker(), THREADPOOL::attr, THREADPOOL::condition, CreateThreadPoolJob(), FreeThreadPoolJob(), THREADPOOL::highJobQ, INVALID_JOB_ID, THREADPOOL::lastJobId, ListAddTail(), THREADPOOL::lowJobQ, THREADPOOLATTR::maxJobsTotal, THREADPOOL::medJobQ, THREADPOOL::mutex, and LINKEDLIST::size.

Referenced by genaInitNotify(), genaInitNotifyExt(), genaNotifyAll(), genaNotifyAllExt(), genaNotifyThread(), readFromSSDPSocket(), ssdp_handle_ctrlpt_msg(), TimerThreadWorker(), UpnpGetServiceVarStatusAsync(), UpnpRenewSubscriptionAsync(), UpnpSendActionAsync(), UpnpSendActionExAsync(), UpnpSubscribeAsync(), and UpnpUnSubscribeAsync().

int ThreadPoolAddPersistent	(	ThreadPool *	tp,
		ThreadPoolJob *	job,
		int *	jobId
	)

Adds a persistent job to the thread pool.

Job will be run as soon as possible. Call will block until job is scheduled.

Returns

0 on success.
EOUTOFMEM not enough memory to add job.
EMAXTHREADS not enough threads to add persistent job.

Parameters

tp	Valid thread pool pointer.
job	Valid thread pool job.
jobId	.

References THREADPOOL::attr, THREADPOOL::condition, CreateThreadPoolJob(), CreateWorker(), INVALID_JOB_ID, THREADPOOL::lastJobId, THREADPOOLATTR::maxThreads, THREADPOOL::mutex, THREADPOOL::persistentJob, THREADPOOL::persistentThreads, THREADPOOL::start_and_shutdown, and THREADPOOL::totalThreads.

Referenced by StartMiniServer(), TimerThreadInit(), and TimerThreadWorker().

int ThreadPoolGetAttr	(	ThreadPool *	tp,
		ThreadPoolAttr *	out
	)

Gets the current set of attributes associated with the thread pool.

Returns

0 on success, nonzero on failure.

Parameters

tp	valid thread pool pointer.
out	non null pointer to store attributes.

References THREADPOOL::attr, THREADPOOL::mutex, and THREADPOOL::shutdown.

int ThreadPoolInit	(	ThreadPool *	tp,
		ThreadPoolAttr *	attr
	)

Initializes and starts ThreadPool. Must be called first and only once for ThreadPool.

Returns

0 on success.
EAGAIN if not enough system resources to create minimum threads.
INVALID_POLICY if schedPolicy can't be set.
EMAXTHREADS if minimum threads is greater than maximum threads.

Parameters

tp Must be valid, non null, pointer to ThreadPool.

attr

Can be null. if not null then attr contains the following fields:

minWorkerThreads - minimum number of worker threads thread pool will never have less than this number of threads.
maxWorkerThreads - maximum number of worker threads thread pool will never have more than this number of threads.
maxIdleTime - maximum time that a worker thread will spend idle. If a worker is idle longer than this time and there are more than the min number of workers running, then the worker thread exits.
jobsPerThread - ratio of jobs to thread to try and maintain if a job is scheduled and the number of jobs per thread is greater than this number,and if less than the maximum number of workers are running then a new thread is started to help out with efficiency.
schedPolicy - scheduling policy to try and set (OS dependent).

References THREADPOOL::attr, THREADPOOL::busyThreads, CmpThreadPoolJob(), THREADPOOL::condition, CreateWorker(), FreeListInit(), THREADPOOL::highJobQ, INVALID_POLICY, THREADPOOL::jobFreeList, JOBFREELISTSIZE, THREADPOOL::lastJobId, ListInit(), THREADPOOL::lowJobQ, THREADPOOL::medJobQ, THREADPOOLATTR::minThreads, THREADPOOL::mutex, THREADPOOL::pendingWorkerThreadStart, THREADPOOL::persistentJob, THREADPOOL::persistentThreads, THREADPOOLATTR::schedPolicy, SetPolicyType(), THREADPOOL::shutdown, THREADPOOL::start_and_shutdown, THREADPOOL::stats, ThreadPoolShutdown(), THREADPOOL::totalThreads, and TPAttrInit().

Referenced by UpnpInitThreadPools().

int ThreadPoolRemove	(	ThreadPool *	tp,
		int	jobId,
		ThreadPoolJob *	out
	)

Removes a job from the thread pool. Can only remove jobs which are not currently running.

Returns

0 on success, nonzero on failure.
INVALID_JOB_ID if job not found.

Parameters

tp	valid thread pool pointer.
jobId	id of job.
out	space for removed job.

References FreeThreadPoolJob(), THREADPOOL::highJobQ, INVALID_JOB_ID, ListDelNode(), ListFind(), THREADPOOL::lowJobQ, THREADPOOL::medJobQ, THREADPOOL::mutex, and THREADPOOL::persistentJob.

int ThreadPoolSetAttr	(	ThreadPool *	tp,
		ThreadPoolAttr *	attr
	)

Sets the attributes for the thread pool. Only affects future calculations.

Returns

0 on success, nonzero on failure.
INVALID_POLICY if policy can not be set.

Parameters

tp	valid thread pool pointer.
attr	pointer to attributes, null sets attributes to default.

References THREADPOOL::attr, THREADPOOL::condition, CreateWorker(), INVALID_POLICY, THREADPOOLATTR::minThreads, THREADPOOL::mutex, THREADPOOLATTR::schedPolicy, SetPolicyType(), ThreadPoolShutdown(), THREADPOOL::totalThreads, and TPAttrInit().

int ThreadPoolShutdown ( ThreadPool * tp )

Shuts the thread pool down. Waits for all threads to finish. May block indefinitely if jobs do not exit.

Returns: 0 on success, nonzero on failure

Parameters

tp	must be valid tp.

References THREADPOOL::condition, FreeListDestroy(), FreeThreadPoolJob(), THREADPOOL::highJobQ, THREADPOOL::jobFreeList, ListDelNode(), ListDestroy(), ListHead(), THREADPOOL::lowJobQ, THREADPOOL::medJobQ, THREADPOOL::mutex, THREADPOOL::persistentJob, THREADPOOL::shutdown, LINKEDLIST::size, THREADPOOL::start_and_shutdown, and THREADPOOL::totalThreads.

Referenced by ThreadPoolInit(), ThreadPoolSetAttr(), and UpnpFinish().

int TPAttrInit ( ThreadPoolAttr * attr )

Initializes thread pool attributes. Sets values to defaults defined in ThreadPool.h.

Returns: Always returns 0.

Parameters

attr	must be valid thread pool attributes.

References DEFAULT_IDLE_TIME, DEFAULT_JOBS_PER_THREAD, DEFAULT_MAX_JOBS_TOTAL, DEFAULT_MAX_THREADS, DEFAULT_MIN_THREADS, DEFAULT_STACK_SIZE, DEFAULT_STARVATION_TIME, THREADPOOLATTR::jobsPerThread, THREADPOOLATTR::maxIdleTime, THREADPOOLATTR::maxJobsTotal, THREADPOOLATTR::maxThreads, THREADPOOLATTR::minThreads, THREADPOOLATTR::schedPolicy, THREADPOOLATTR::stackSize, and THREADPOOLATTR::starvationTime.

Referenced by ThreadPoolInit(), ThreadPoolSetAttr(), and UpnpInitThreadPools().

int TPAttrSetIdleTime	(	ThreadPoolAttr *	attr,
		int	idleTime
	)

Sets the idle time for the thread pool attributes.

Returns: Always returns 0.

Parameters

attr	must be valid thread pool attributes.
idleTime	.

References THREADPOOLATTR::maxIdleTime.

Referenced by UpnpInitThreadPools().

int TPAttrSetJobsPerThread	(	ThreadPoolAttr *	attr,
		int	jobsPerThread
	)

Sets the jobs per thread ratio.

Returns: Always returns 0.

Parameters

attr	must be valid thread pool attributes.
jobsPerThread	number of jobs per thread to maintain.

References THREADPOOLATTR::jobsPerThread.

Referenced by UpnpInitThreadPools().

int TPAttrSetMaxJobsTotal	(	ThreadPoolAttr *	attr,
		int	maxJobsTotal
	)

Sets the maximum number jobs that can be qeued totally.

Returns: Always returns 0.

Parameters

attr	must be valid thread pool attributes.
maxJobsTotal	maximum number of jobs.

References THREADPOOLATTR::maxJobsTotal.

Referenced by UpnpInitThreadPools().

int TPAttrSetMaxThreads	(	ThreadPoolAttr *	attr,
		int	maxThreads
	)

Sets the max threads for the thread pool attributes.

Returns: Always returns 0.

Parameters

attr	must be valid thread pool attributes.
maxThreads	value to set.

References THREADPOOLATTR::maxThreads.

Referenced by UpnpInitThreadPools().

int TPAttrSetMinThreads	(	ThreadPoolAttr *	attr,
		int	minThreads
	)

Sets the min threads for the thread pool attributes.

Returns: Always returns 0.

Parameters

attr	must be valid thread pool attributes.
minThreads	value to set.

References THREADPOOLATTR::minThreads.

Referenced by UpnpInitThreadPools().

int TPAttrSetSchedPolicy	(	ThreadPoolAttr *	attr,
		PolicyType	schedPolicy
	)

Sets the scheduling policy for the thread pool attributes.

Returns: Always returns 0.

Parameters

attr	must be valid thread pool attributes.
schedPolicy	must be a valid policy type.

References THREADPOOLATTR::schedPolicy.

int TPAttrSetStackSize	(	ThreadPoolAttr *	attr,
		size_t	stackSize
	)

Sets the stack size for the thread pool attributes.

Returns: Always returns 0.

Parameters

attr	must be valid thread pool attributes.
stackSize	value to set.

References THREADPOOLATTR::stackSize.

Referenced by UpnpInitThreadPools().

int TPAttrSetStarvationTime	(	ThreadPoolAttr *	attr,
		int	starvationTime
	)

Sets the starvation time for the thread pool attributes.

Returns: Always returns 0.

Parameters

attr	must be valid thread pool attributes.
starvationTime	milliseconds.

References THREADPOOLATTR::starvationTime.

int TPJobInit	(	ThreadPoolJob *	job,
		start_routine	func,
		void *	arg
	)

Initializes thread pool job. Sets the priority to default defined in ThreadPool.h. Sets the free_routine to default defined in ThreadPool.h.

Returns: Always returns 0.

Parameters

job	must be valid thread pool attributes.
func	function to run, must be valid.
arg	argument to pass to function.

References DEFAULT_FREE_ROUTINE, and DEFAULT_PRIORITY.

Referenced by genaInitNotify(), genaInitNotifyExt(), genaNotifyAll(), genaNotifyAllExt(), genaNotifyThread(), readFromSSDPSocket(), ScheduleGenaAutoRenew(), SearchByTarget(), ssdp_handle_ctrlpt_msg(), ssdp_handle_device_request(), StartMiniServer(), TimerThreadInit(), UpnpGetServiceVarStatusAsync(), UpnpRenewSubscriptionAsync(), UpnpSendActionAsync(), UpnpSendActionExAsync(), UpnpSendAdvertisementLowPower(), UpnpSubscribeAsync(), and UpnpUnSubscribeAsync().

int TPJobSetFreeFunction	(	ThreadPoolJob *	job,
		free_routine	func
	)

Sets the max threads for the thread pool attributes.

Returns: Always returns 0.

Parameters

job	must be valid thread pool attributes.
func	value to set.

Referenced by genaInitNotify(), genaInitNotifyExt(), genaNotifyAll(), genaNotifyAllExt(), genaNotifyThread(), readFromSSDPSocket(), ScheduleGenaAutoRenew(), SearchByTarget(), ssdp_handle_ctrlpt_msg(), ssdp_handle_device_request(), StartMiniServer(), UpnpGetServiceVarStatusAsync(), UpnpRenewSubscriptionAsync(), UpnpSendActionAsync(), UpnpSendActionExAsync(), UpnpSendAdvertisementLowPower(), UpnpSubscribeAsync(), and UpnpUnSubscribeAsync().

int TPJobSetPriority	(	ThreadPoolJob *	job,
		ThreadPriority	priority
	)

Sets the max threads for the thread pool attributes.

Returns: Always returns 0.

Parameters

job	must be valid thread pool attributes.
priority	value to set.

Referenced by genaInitNotify(), genaInitNotifyExt(), genaNotifyAll(), genaNotifyAllExt(), genaNotifyThread(), readFromSSDPSocket(), ScheduleGenaAutoRenew(), SearchByTarget(), ssdp_handle_ctrlpt_msg(), StartMiniServer(), TimerThreadInit(), UpnpGetServiceVarStatusAsync(), UpnpRenewSubscriptionAsync(), UpnpSendActionAsync(), UpnpSendActionExAsync(), UpnpSendAdvertisementLowPower(), UpnpSubscribeAsync(), and UpnpUnSubscribeAsync().

static void* WorkerThread ( void * arg )

static

Implements a thread pool worker. Worker waits for a job to become available. Worker picks up persistent jobs first, high priority, med priority, then low priority.

If worker remains idle for more than specified max, the worker is released.

Parameters

arg	arg -> is cast to (ThreadPool *).

References THREADPOOL::attr, BumpPriority(), THREADPOOL::busyThreads, THREADPOOL::condition, DEFAULT_PRIORITY, FreeThreadPoolJob(), THREADPOOL::highJobQ, ListDelNode(), ListHead(), THREADPOOL::lowJobQ, THREADPOOLATTR::maxIdleTime, THREADPOOLATTR::maxThreads, THREADPOOL::medJobQ, THREADPOOLATTR::minThreads, THREADPOOL::mutex, THREADPOOL::pendingWorkerThreadStart, THREADPOOL::persistentJob, THREADPOOL::persistentThreads, SetPriority(), SetRelTimeout(), SetSeed(), THREADPOOL::shutdown, LINKEDLIST::size, THREADPOOL::start_and_shutdown, THREADPOOL::stats, and THREADPOOL::totalThreads.

Referenced by CreateWorker().

Functions

Function Documentation