The Tor Browser: nsprpub/pr/src/misc/prtpool.c@6474c204b198 (annotated)

nsprpub/pr/src/misc/prtpool.c@6474c204b198 (annotated)

nsprpub/pr/src/misc/prtpool.c

Wed, 31 Dec 2014 06:09:35 +0100

author: Michael Schloh von Bennewitz <michael@schloh.com>
date: Wed, 31 Dec 2014 06:09:35 +0100
changeset 0: 6474c204b198
permissions: -rw-r--r--

Cloned upstream origin tor-browser at tor-browser-31.3.0esr-4.5-1-build1
revision ID fc1c9ff7c1b2defdbc039f12214767608f46423f for hacking purpose.

 /* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
 /* This Source Code Form is subject to the terms of the Mozilla Public
  * License, v. 2.0. If a copy of the MPL was not distributed with this
  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
 #include "nspr.h"
 /*
  * Thread pools
  *	Thread pools create and manage threads to provide support for
  *	scheduling jobs onto one or more threads.
  *
  */
 #ifdef OPT_WINNT
 #include <windows.h>
 #endif
 /*
  * worker thread
  */
 typedef struct wthread {
 	PRCList		links;
 	PRThread	*thread;
 } wthread;
 /*
  * queue of timer jobs
  */
 typedef struct timer_jobq {
 	PRCList		list;
 	PRLock		*lock;
 	PRCondVar	*cv;
 	PRInt32		cnt;
 	PRCList 	wthreads;
 } timer_jobq;
 /*
  * queue of jobs
  */
 typedef struct tp_jobq {
 	PRCList		list;
 	PRInt32		cnt;
 	PRLock		*lock;
 	PRCondVar	*cv;
 	PRCList 	wthreads;
 #ifdef OPT_WINNT
 	HANDLE		nt_completion_port;
 #endif
 } tp_jobq;
 /*
  * queue of IO jobs
  */
 typedef struct io_jobq {
 	PRCList		list;
 	PRPollDesc  *pollfds;
 	PRInt32  	npollfds;
 	PRJob		**polljobs;
 	PRLock		*lock;
 	PRInt32		cnt;
 	PRFileDesc	*notify_fd;
 	PRCList 	wthreads;
 } io_jobq;
 /*
  * Threadpool
  */
 struct PRThreadPool {
 	PRInt32		init_threads;
 	PRInt32		max_threads;
 	PRInt32		current_threads;
 	PRInt32		idle_threads;
 	PRUint32	stacksize;
 	tp_jobq		jobq;
 	io_jobq		ioq;
 	timer_jobq	timerq;
 	PRLock		*join_lock;		/* used with jobp->join_cv */
 	PRCondVar	*shutdown_cv;
 	PRBool		shutdown;
 };
 typedef enum io_op_type
 	{ JOB_IO_READ, JOB_IO_WRITE, JOB_IO_CONNECT, JOB_IO_ACCEPT } io_op_type;
 #ifdef OPT_WINNT
 typedef struct NT_notifier {
 	OVERLAPPED overlapped;		/* must be first */
 	PRJob	*jobp;
 } NT_notifier;
 #endif
 struct PRJob {
 	PRCList			links;		/* 	for linking jobs */
 	PRBool			on_ioq;		/* job on ioq */
 	PRBool			on_timerq;	/* job on timerq */
 	PRJobFn			job_func;
 	void 			*job_arg;
 	PRCondVar		*join_cv;
 	PRBool			join_wait;	/* == PR_TRUE, when waiting to join */
 	PRCondVar		*cancel_cv;	/* for cancelling IO jobs */
 	PRBool			cancel_io;	/* for cancelling IO jobs */
 	PRThreadPool	*tpool;		/* back pointer to thread pool */
 	PRJobIoDesc		*iod;
 	io_op_type		io_op;
 	PRInt16			io_poll_flags;
 	PRNetAddr		*netaddr;
 	PRIntervalTime	timeout;	/* relative value */
 	PRIntervalTime	absolute;
 #ifdef OPT_WINNT
 	NT_notifier		nt_notifier;
 #endif
 };
 #define JOB_LINKS_PTR(_qp) \
     ((PRJob *) ((char *) (_qp) - offsetof(PRJob, links)))
 #define WTHREAD_LINKS_PTR(_qp) \
     ((wthread *) ((char *) (_qp) - offsetof(wthread, links)))
 #define JOINABLE_JOB(_jobp) (NULL != (_jobp)->join_cv)
 #define JOIN_NOTIFY(_jobp)								\
 				PR_BEGIN_MACRO							\
 				PR_Lock(_jobp->tpool->join_lock);		\
 				_jobp->join_wait = PR_FALSE;			\
 				PR_NotifyCondVar(_jobp->join_cv);		\
 				PR_Unlock(_jobp->tpool->join_lock);		\
 				PR_END_MACRO
 #define CANCEL_IO_JOB(jobp)								\
 				PR_BEGIN_MACRO							\
 				jobp->cancel_io = PR_FALSE;				\
 				jobp->on_ioq = PR_FALSE;				\
 				PR_REMOVE_AND_INIT_LINK(&jobp->links);	\
 				tp->ioq.cnt--;							\
 				PR_NotifyCondVar(jobp->cancel_cv);		\
 				PR_END_MACRO
 static void delete_job(PRJob *jobp);
 static PRThreadPool * alloc_threadpool(void);
 static PRJob * alloc_job(PRBool joinable, PRThreadPool *tp);
 static void notify_ioq(PRThreadPool *tp);
 static void notify_timerq(PRThreadPool *tp);
 /*
  * locks are acquired in the following order
  *
  *	tp->ioq.lock,tp->timerq.lock
  *			|
  *			V
  *		tp->jobq->lock
  */
 /*
  * worker thread function
  */
 static void wstart(void *arg)
 {
 PRThreadPool *tp = (PRThreadPool *) arg;
 PRCList *head;
 	/*
 	 * execute jobs until shutdown
 	 */
 	while (!tp->shutdown) {
 		PRJob *jobp;
 #ifdef OPT_WINNT
 		BOOL rv;
 		DWORD unused, shutdown;
 		LPOVERLAPPED olp;
 		PR_Lock(tp->jobq.lock);
 		tp->idle_threads++;
 		PR_Unlock(tp->jobq.lock);
 		rv = GetQueuedCompletionStatus(tp->jobq.nt_completion_port,
 					&unused, &shutdown, &olp, INFINITE);
 		PR_ASSERT(rv);
 		if (shutdown)
 			break;
 		jobp = ((NT_notifier *) olp)->jobp;
 		PR_Lock(tp->jobq.lock);
 		tp->idle_threads--;
 		tp->jobq.cnt--;
 		PR_Unlock(tp->jobq.lock);
 #else
 		PR_Lock(tp->jobq.lock);
 		while (PR_CLIST_IS_EMPTY(&tp->jobq.list) && (!tp->shutdown)) {
 			tp->idle_threads++;
 			PR_WaitCondVar(tp->jobq.cv, PR_INTERVAL_NO_TIMEOUT);
 			tp->idle_threads--;
 		}
 		if (tp->shutdown) {
 			PR_Unlock(tp->jobq.lock);
 			break;
 		}
 		head = PR_LIST_HEAD(&tp->jobq.list);
 		/*
 		 * remove job from queue
 		 */
 		PR_REMOVE_AND_INIT_LINK(head);
 		tp->jobq.cnt--;
 		jobp = JOB_LINKS_PTR(head);
 		PR_Unlock(tp->jobq.lock);
 #endif
 		jobp->job_func(jobp->job_arg);
 		if (!JOINABLE_JOB(jobp)) {
 			delete_job(jobp);
 		} else {
 			JOIN_NOTIFY(jobp);
 		}
 	}
 	PR_Lock(tp->jobq.lock);
 	tp->current_threads--;
 	PR_Unlock(tp->jobq.lock);
 }
 /*
  * add a job to the work queue
  */
 static void
 add_to_jobq(PRThreadPool *tp, PRJob *jobp)
 {
 	/*
 	 * add to jobq
 	 */
 #ifdef OPT_WINNT
 	PR_Lock(tp->jobq.lock);
 	tp->jobq.cnt++;
 	PR_Unlock(tp->jobq.lock);
 	/*
 	 * notify worker thread(s)
 	 */
 	PostQueuedCompletionStatus(tp->jobq.nt_completion_port, 0,
             FALSE, &jobp->nt_notifier.overlapped);
 #else
 	PR_Lock(tp->jobq.lock);
 	PR_APPEND_LINK(&jobp->links,&tp->jobq.list);
 	tp->jobq.cnt++;
 	if ((tp->idle_threads < tp->jobq.cnt) &&
 					(tp->current_threads < tp->max_threads)) {
 		wthread *wthrp;
 		/*
 		 * increment thread count and unlock the jobq lock
 		 */
 		tp->current_threads++;
 		PR_Unlock(tp->jobq.lock);
 		/* create new worker thread */
 		wthrp = PR_NEWZAP(wthread);
 		if (wthrp) {
 			wthrp->thread = PR_CreateThread(PR_USER_THREAD, wstart,
 						tp, PR_PRIORITY_NORMAL,
 						PR_GLOBAL_THREAD,PR_JOINABLE_THREAD,tp->stacksize);
 			if (NULL == wthrp->thread) {
 				PR_DELETE(wthrp);  /* this sets wthrp to NULL */
 			}
 		}
 		PR_Lock(tp->jobq.lock);
 		if (NULL == wthrp) {
 			tp->current_threads--;
 		} else {
 			PR_APPEND_LINK(&wthrp->links, &tp->jobq.wthreads);
 		}
 	}
 	/*
 	 * wakeup a worker thread
 	 */
 	PR_NotifyCondVar(tp->jobq.cv);
 	PR_Unlock(tp->jobq.lock);
 #endif
 }
 /*
  * io worker thread function
  */
 static void io_wstart(void *arg)
 {
 PRThreadPool *tp = (PRThreadPool *) arg;
 int pollfd_cnt, pollfds_used;
 int rv;
 PRCList *qp, *nextqp;
 PRPollDesc *pollfds;
 PRJob **polljobs;
 int poll_timeout;
 PRIntervalTime now;
 	/*
 	 * scan io_jobq
 	 * construct poll list
 	 * call PR_Poll
 	 * for all fds, for which poll returns true, move the job to
 	 * jobq and wakeup worker thread.
 	 */
 	while (!tp->shutdown) {
 		PRJob *jobp;
 		pollfd_cnt = tp->ioq.cnt + 10;
 		if (pollfd_cnt > tp->ioq.npollfds) {
 			/*
 			 * re-allocate pollfd array if the current one is not large
 			 * enough
 			 */
 			if (NULL != tp->ioq.pollfds)
 				PR_Free(tp->ioq.pollfds);
 			tp->ioq.pollfds = (PRPollDesc *) PR_Malloc(pollfd_cnt *
 						(sizeof(PRPollDesc) + sizeof(PRJob *)));
 			PR_ASSERT(NULL != tp->ioq.pollfds);
 			/*
 			 * array of pollfds
 			 */
 			pollfds = tp->ioq.pollfds;
 			tp->ioq.polljobs = (PRJob **) (&tp->ioq.pollfds[pollfd_cnt]);
 			/*
 			 * parallel array of jobs
 			 */
 			polljobs = tp->ioq.polljobs;
 			tp->ioq.npollfds = pollfd_cnt;
 		}
 		pollfds_used = 0;
 		/*
 		 * add the notify fd; used for unblocking io thread(s)
 		 */
 		pollfds[pollfds_used].fd = tp->ioq.notify_fd;
 		pollfds[pollfds_used].in_flags = PR_POLL_READ;
 		pollfds[pollfds_used].out_flags = 0;
 		polljobs[pollfds_used] = NULL;
 		pollfds_used++;
 		/*
 		 * fill in the pollfd array
 		 */
 		PR_Lock(tp->ioq.lock);
 		for (qp = tp->ioq.list.next; qp != &tp->ioq.list; qp = nextqp) {
 			nextqp = qp->next;
 			jobp = JOB_LINKS_PTR(qp);
 			if (jobp->cancel_io) {
 				CANCEL_IO_JOB(jobp);
 				continue;
 			}
 			if (pollfds_used == (pollfd_cnt))
 				break;
 			pollfds[pollfds_used].fd = jobp->iod->socket;
 			pollfds[pollfds_used].in_flags = jobp->io_poll_flags;
 			pollfds[pollfds_used].out_flags = 0;
 			polljobs[pollfds_used] = jobp;
 			pollfds_used++;
 		}
 		if (!PR_CLIST_IS_EMPTY(&tp->ioq.list)) {
 			qp = tp->ioq.list.next;
 			jobp = JOB_LINKS_PTR(qp);
 			if (PR_INTERVAL_NO_TIMEOUT == jobp->timeout)
 				poll_timeout = PR_INTERVAL_NO_TIMEOUT;
 			else if (PR_INTERVAL_NO_WAIT == jobp->timeout)
 				poll_timeout = PR_INTERVAL_NO_WAIT;
 			else {
 				poll_timeout = jobp->absolute - PR_IntervalNow();
 				if (poll_timeout <= 0) /* already timed out */
 					poll_timeout = PR_INTERVAL_NO_WAIT;
 			}
 		} else {
 			poll_timeout = PR_INTERVAL_NO_TIMEOUT;
 		}
 		PR_Unlock(tp->ioq.lock);
 		/*
 		 * XXXX
 		 * should retry if more jobs have been added to the queue?
 		 *
 		 */
 		PR_ASSERT(pollfds_used <= pollfd_cnt);
 		rv = PR_Poll(tp->ioq.pollfds, pollfds_used, poll_timeout);
 		if (tp->shutdown) {
 			break;
 		}
 		if (rv > 0) {
 			/*
 			 * at least one io event is set
 			 */
 			PRStatus rval_status;
 			PRInt32 index;
 			PR_ASSERT(pollfds[0].fd == tp->ioq.notify_fd);
 			/*
 			 * reset the pollable event, if notified
 			 */
 			if (pollfds[0].out_flags & PR_POLL_READ) {
 				rval_status = PR_WaitForPollableEvent(tp->ioq.notify_fd);
 				PR_ASSERT(PR_SUCCESS == rval_status);
 			}
 			for(index = 1; index < (pollfds_used); index++) {
                 PRInt16 events = pollfds[index].in_flags;
                 PRInt16 revents = pollfds[index].out_flags;
 				jobp = polljobs[index];
                 if ((revents & PR_POLL_NVAL) ||  /* busted in all cases */
                 	(revents & PR_POLL_ERR) ||
                 			((events & PR_POLL_WRITE) &&
 							(revents & PR_POLL_HUP))) { /* write op & hup */
 					PR_Lock(tp->ioq.lock);
 					if (jobp->cancel_io) {
 						CANCEL_IO_JOB(jobp);
 						PR_Unlock(tp->ioq.lock);
 						continue;
 					}
 					PR_REMOVE_AND_INIT_LINK(&jobp->links);
 					tp->ioq.cnt--;
 					jobp->on_ioq = PR_FALSE;
 					PR_Unlock(tp->ioq.lock);
 					/* set error */
                     if (PR_POLL_NVAL & revents)
 						jobp->iod->error = PR_BAD_DESCRIPTOR_ERROR;
                     else if (PR_POLL_HUP & revents)
 						jobp->iod->error = PR_CONNECT_RESET_ERROR;
                     else
 						jobp->iod->error = PR_IO_ERROR;
 					/*
 					 * add to jobq
 					 */
 					add_to_jobq(tp, jobp);
 				} else if (revents) {
 					/*
 					 * add to jobq
 					 */
 					PR_Lock(tp->ioq.lock);
 					if (jobp->cancel_io) {
 						CANCEL_IO_JOB(jobp);
 						PR_Unlock(tp->ioq.lock);
 						continue;
 					}
 					PR_REMOVE_AND_INIT_LINK(&jobp->links);
 					tp->ioq.cnt--;
 					jobp->on_ioq = PR_FALSE;
 					PR_Unlock(tp->ioq.lock);
 					if (jobp->io_op == JOB_IO_CONNECT) {
 						if (PR_GetConnectStatus(&pollfds[index]) == PR_SUCCESS)
 							jobp->iod->error = 0;
 						else
 							jobp->iod->error = PR_GetError();
 					} else
 						jobp->iod->error = 0;
 					add_to_jobq(tp, jobp);
 				}
 			}
 		}
 		/*
 		 * timeout processing
 		 */
 		now = PR_IntervalNow();
 		PR_Lock(tp->ioq.lock);
 		for (qp = tp->ioq.list.next; qp != &tp->ioq.list; qp = nextqp) {
 			nextqp = qp->next;
 			jobp = JOB_LINKS_PTR(qp);
 			if (jobp->cancel_io) {
 				CANCEL_IO_JOB(jobp);
 				continue;
 			}
 			if (PR_INTERVAL_NO_TIMEOUT == jobp->timeout)
 				break;
 			if ((PR_INTERVAL_NO_WAIT != jobp->timeout) &&
 								((PRInt32)(jobp->absolute - now) > 0))
 				break;
 			PR_REMOVE_AND_INIT_LINK(&jobp->links);
 			tp->ioq.cnt--;
 			jobp->on_ioq = PR_FALSE;
 			jobp->iod->error = PR_IO_TIMEOUT_ERROR;
 			add_to_jobq(tp, jobp);
 		}
 		PR_Unlock(tp->ioq.lock);
 	}
 }
 /*
  * timer worker thread function
  */
 static void timer_wstart(void *arg)
 {
 PRThreadPool *tp = (PRThreadPool *) arg;
 PRCList *qp;
 PRIntervalTime timeout;
 PRIntervalTime now;
 	/*
 	 * call PR_WaitCondVar with minimum value of all timeouts
 	 */
 	while (!tp->shutdown) {
 		PRJob *jobp;
 		PR_Lock(tp->timerq.lock);
 		if (PR_CLIST_IS_EMPTY(&tp->timerq.list)) {
 			timeout = PR_INTERVAL_NO_TIMEOUT;
 		} else {
 			PRCList *qp;
 			qp = tp->timerq.list.next;
 			jobp = JOB_LINKS_PTR(qp);
 			timeout = jobp->absolute - PR_IntervalNow();
             if (timeout <= 0)
 				timeout = PR_INTERVAL_NO_WAIT;  /* already timed out */
 		}
 		if (PR_INTERVAL_NO_WAIT != timeout)
 			PR_WaitCondVar(tp->timerq.cv, timeout);
 		if (tp->shutdown) {
 			PR_Unlock(tp->timerq.lock);
 			break;
 		}
 		/*
 		 * move expired-timer jobs to jobq
 		 */
 		now = PR_IntervalNow();
 		while (!PR_CLIST_IS_EMPTY(&tp->timerq.list)) {
 			qp = tp->timerq.list.next;
 			jobp = JOB_LINKS_PTR(qp);
 			if ((PRInt32)(jobp->absolute - now) > 0) {
 				break;
 			}
 			/*
 			 * job timed out
 			 */
 			PR_REMOVE_AND_INIT_LINK(&jobp->links);
 			tp->timerq.cnt--;
 			jobp->on_timerq = PR_FALSE;
 			add_to_jobq(tp, jobp);
 		}
 		PR_Unlock(tp->timerq.lock);
 	}
 }
 static void
 delete_threadpool(PRThreadPool *tp)
 {
 	if (NULL != tp) {
 		if (NULL != tp->shutdown_cv)
 			PR_DestroyCondVar(tp->shutdown_cv);
 		if (NULL != tp->jobq.cv)
 			PR_DestroyCondVar(tp->jobq.cv);
 		if (NULL != tp->jobq.lock)
 			PR_DestroyLock(tp->jobq.lock);
 		if (NULL != tp->join_lock)
 			PR_DestroyLock(tp->join_lock);
 #ifdef OPT_WINNT
 		if (NULL != tp->jobq.nt_completion_port)
 			CloseHandle(tp->jobq.nt_completion_port);
 #endif
 		/* Timer queue */
 		if (NULL != tp->timerq.cv)
 			PR_DestroyCondVar(tp->timerq.cv);
 		if (NULL != tp->timerq.lock)
 			PR_DestroyLock(tp->timerq.lock);
 		if (NULL != tp->ioq.lock)
 			PR_DestroyLock(tp->ioq.lock);
 		if (NULL != tp->ioq.pollfds)
 			PR_Free(tp->ioq.pollfds);
 		if (NULL != tp->ioq.notify_fd)
 			PR_DestroyPollableEvent(tp->ioq.notify_fd);
 		PR_Free(tp);
 	}
 	return;
 }
 static PRThreadPool *
 alloc_threadpool(void)
 {
 PRThreadPool *tp;
 	tp = (PRThreadPool *) PR_CALLOC(sizeof(*tp));
 	if (NULL == tp)
 		goto failed;
 	tp->jobq.lock = PR_NewLock();
 	if (NULL == tp->jobq.lock)
 		goto failed;
 	tp->jobq.cv = PR_NewCondVar(tp->jobq.lock);
 	if (NULL == tp->jobq.cv)
 		goto failed;
 	tp->join_lock = PR_NewLock();
 	if (NULL == tp->join_lock)
 		goto failed;
 #ifdef OPT_WINNT
 	tp->jobq.nt_completion_port = CreateIoCompletionPort(INVALID_HANDLE_VALUE,
 									NULL, 0, 0);
 	if (NULL == tp->jobq.nt_completion_port)
 		goto failed;
 #endif
 	tp->ioq.lock = PR_NewLock();
 	if (NULL == tp->ioq.lock)
 		goto failed;
 	/* Timer queue */
 	tp->timerq.lock = PR_NewLock();
 	if (NULL == tp->timerq.lock)
 		goto failed;
 	tp->timerq.cv = PR_NewCondVar(tp->timerq.lock);
 	if (NULL == tp->timerq.cv)
 		goto failed;
 	tp->shutdown_cv = PR_NewCondVar(tp->jobq.lock);
 	if (NULL == tp->shutdown_cv)
 		goto failed;
 	tp->ioq.notify_fd = PR_NewPollableEvent();
 	if (NULL == tp->ioq.notify_fd)
 		goto failed;
 	return tp;
 failed:
 	delete_threadpool(tp);
 	PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);
 	return NULL;
 }
 /* Create thread pool */
 PR_IMPLEMENT(PRThreadPool *)
 PR_CreateThreadPool(PRInt32 initial_threads, PRInt32 max_threads,
                                 PRUint32 stacksize)
 {
 PRThreadPool *tp;
 PRThread *thr;
 int i;
 wthread *wthrp;
 	tp = alloc_threadpool();
 	if (NULL == tp)
 		return NULL;
 	tp->init_threads = initial_threads;
 	tp->max_threads = max_threads;
 	tp->stacksize = stacksize;
 	PR_INIT_CLIST(&tp->jobq.list);
 	PR_INIT_CLIST(&tp->ioq.list);
 	PR_INIT_CLIST(&tp->timerq.list);
 	PR_INIT_CLIST(&tp->jobq.wthreads);
 	PR_INIT_CLIST(&tp->ioq.wthreads);
 	PR_INIT_CLIST(&tp->timerq.wthreads);
 	tp->shutdown = PR_FALSE;
 	PR_Lock(tp->jobq.lock);
 	for(i=0; i < initial_threads; ++i) {
 		thr = PR_CreateThread(PR_USER_THREAD, wstart,
 						tp, PR_PRIORITY_NORMAL,
 						PR_GLOBAL_THREAD, PR_JOINABLE_THREAD,stacksize);
 		PR_ASSERT(thr);
 		wthrp = PR_NEWZAP(wthread);
 		PR_ASSERT(wthrp);
 		wthrp->thread = thr;
 		PR_APPEND_LINK(&wthrp->links, &tp->jobq.wthreads);
 	}
 	tp->current_threads = initial_threads;
 	thr = PR_CreateThread(PR_USER_THREAD, io_wstart,
 					tp, PR_PRIORITY_NORMAL,
 					PR_GLOBAL_THREAD,PR_JOINABLE_THREAD,stacksize);
 	PR_ASSERT(thr);
 	wthrp = PR_NEWZAP(wthread);
 	PR_ASSERT(wthrp);
 	wthrp->thread = thr;
 	PR_APPEND_LINK(&wthrp->links, &tp->ioq.wthreads);
 	thr = PR_CreateThread(PR_USER_THREAD, timer_wstart,
 					tp, PR_PRIORITY_NORMAL,
 					PR_GLOBAL_THREAD,PR_JOINABLE_THREAD,stacksize);
 	PR_ASSERT(thr);
 	wthrp = PR_NEWZAP(wthread);
 	PR_ASSERT(wthrp);
 	wthrp->thread = thr;
 	PR_APPEND_LINK(&wthrp->links, &tp->timerq.wthreads);
 	PR_Unlock(tp->jobq.lock);
 	return tp;
 }
 static void
 delete_job(PRJob *jobp)
 {
 	if (NULL != jobp) {
 		if (NULL != jobp->join_cv) {
 			PR_DestroyCondVar(jobp->join_cv);
 			jobp->join_cv = NULL;
 		}
 		if (NULL != jobp->cancel_cv) {
 			PR_DestroyCondVar(jobp->cancel_cv);
 			jobp->cancel_cv = NULL;
 		}
 		PR_DELETE(jobp);
 	}
 }
 static PRJob *
 alloc_job(PRBool joinable, PRThreadPool *tp)
 {
 	PRJob *jobp;
 	jobp = PR_NEWZAP(PRJob);
 	if (NULL == jobp)
 		goto failed;
 	if (joinable) {
 		jobp->join_cv = PR_NewCondVar(tp->join_lock);
 		jobp->join_wait = PR_TRUE;
 		if (NULL == jobp->join_cv)
 			goto failed;
 	} else {
 		jobp->join_cv = NULL;
 	}
 #ifdef OPT_WINNT
 	jobp->nt_notifier.jobp = jobp;
 #endif
 	return jobp;
 failed:
 	delete_job(jobp);
 	PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);
 	return NULL;
 }
 /* queue a job */
 PR_IMPLEMENT(PRJob *)
 PR_QueueJob(PRThreadPool *tpool, PRJobFn fn, void *arg, PRBool joinable)
 {
 	PRJob *jobp;
 	jobp = alloc_job(joinable, tpool);
 	if (NULL == jobp)
 		return NULL;
 	jobp->job_func = fn;
 	jobp->job_arg = arg;
 	jobp->tpool = tpool;
 	add_to_jobq(tpool, jobp);
 	return jobp;
 }
 /* queue a job, when a socket is readable or writeable */
 static PRJob *
 queue_io_job(PRThreadPool *tpool, PRJobIoDesc *iod, PRJobFn fn, void * arg,
 				PRBool joinable, io_op_type op)
 {
 	PRJob *jobp;
 	PRIntervalTime now;
 	jobp = alloc_job(joinable, tpool);
 	if (NULL == jobp) {
 		return NULL;
 	}
 	/*
 	 * Add a new job to io_jobq
 	 * wakeup io worker thread
 	 */
 	jobp->job_func = fn;
 	jobp->job_arg = arg;
 	jobp->tpool = tpool;
 	jobp->iod = iod;
 	if (JOB_IO_READ == op) {
 		jobp->io_op = JOB_IO_READ;
 		jobp->io_poll_flags = PR_POLL_READ;
 	} else if (JOB_IO_WRITE == op) {
 		jobp->io_op = JOB_IO_WRITE;
 		jobp->io_poll_flags = PR_POLL_WRITE;
 	} else if (JOB_IO_ACCEPT == op) {
 		jobp->io_op = JOB_IO_ACCEPT;
 		jobp->io_poll_flags = PR_POLL_READ;
 	} else if (JOB_IO_CONNECT == op) {
 		jobp->io_op = JOB_IO_CONNECT;
 		jobp->io_poll_flags = PR_POLL_WRITE|PR_POLL_EXCEPT;
 	} else {
 		delete_job(jobp);
 		PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);
 		return NULL;
 	}
 	jobp->timeout = iod->timeout;
 	if ((PR_INTERVAL_NO_TIMEOUT == iod->timeout) ||
 			(PR_INTERVAL_NO_WAIT == iod->timeout)) {
 		jobp->absolute = iod->timeout;
 	} else {
 		now = PR_IntervalNow();
 		jobp->absolute = now + iod->timeout;
 	}
 	PR_Lock(tpool->ioq.lock);
 	if (PR_CLIST_IS_EMPTY(&tpool->ioq.list) ||
 			(PR_INTERVAL_NO_TIMEOUT == iod->timeout)) {
 		PR_APPEND_LINK(&jobp->links,&tpool->ioq.list);
 	} else if (PR_INTERVAL_NO_WAIT == iod->timeout) {
 		PR_INSERT_LINK(&jobp->links,&tpool->ioq.list);
 	} else {
 		PRCList *qp;
 		PRJob *tmp_jobp;
 		/*
 		 * insert into the timeout-sorted ioq
 		 */
 		for (qp = tpool->ioq.list.prev; qp != &tpool->ioq.list;
 							qp = qp->prev) {
 			tmp_jobp = JOB_LINKS_PTR(qp);
 			if ((PRInt32)(jobp->absolute - tmp_jobp->absolute) >= 0) {
 				break;
 			}
 		}
 		PR_INSERT_AFTER(&jobp->links,qp);
 	}
 	jobp->on_ioq = PR_TRUE;
 	tpool->ioq.cnt++;
 	/*
 	 * notify io worker thread(s)
 	 */
 	PR_Unlock(tpool->ioq.lock);
 	notify_ioq(tpool);
 	return jobp;
 }
 /* queue a job, when a socket is readable */
 PR_IMPLEMENT(PRJob *)
 PR_QueueJob_Read(PRThreadPool *tpool, PRJobIoDesc *iod, PRJobFn fn, void * arg,
 											PRBool joinable)
 {
 	return (queue_io_job(tpool, iod, fn, arg, joinable, JOB_IO_READ));
 }
 /* queue a job, when a socket is writeable */
 PR_IMPLEMENT(PRJob *)
 PR_QueueJob_Write(PRThreadPool *tpool, PRJobIoDesc *iod, PRJobFn fn,void * arg,
 										PRBool joinable)
 {
 	return (queue_io_job(tpool, iod, fn, arg, joinable, JOB_IO_WRITE));
 }
 /* queue a job, when a socket has a pending connection */
 PR_IMPLEMENT(PRJob *)
 PR_QueueJob_Accept(PRThreadPool *tpool, PRJobIoDesc *iod, PRJobFn fn,
 								void * arg, PRBool joinable)
 {
 	return (queue_io_job(tpool, iod, fn, arg, joinable, JOB_IO_ACCEPT));
 }
 /* queue a job, when a socket can be connected */
 PR_IMPLEMENT(PRJob *)
 PR_QueueJob_Connect(PRThreadPool *tpool, PRJobIoDesc *iod,
 			const PRNetAddr *addr, PRJobFn fn, void * arg, PRBool joinable)
 {
 	PRStatus rv;
 	PRErrorCode err;
 	rv = PR_Connect(iod->socket, addr, PR_INTERVAL_NO_WAIT);
 	if ((rv == PR_FAILURE) && ((err = PR_GetError()) == PR_IN_PROGRESS_ERROR)){
 		/* connection pending */
 		return(queue_io_job(tpool, iod, fn, arg, joinable, JOB_IO_CONNECT));
 	} else {
 		/*
 		 * connection succeeded or failed; add to jobq right away
 		 */
 		if (rv == PR_FAILURE)
 			iod->error = err;
 		else
 			iod->error = 0;
 		return(PR_QueueJob(tpool, fn, arg, joinable));
 	}
 }
 /* queue a job, when a timer expires */
 PR_IMPLEMENT(PRJob *)
 PR_QueueJob_Timer(PRThreadPool *tpool, PRIntervalTime timeout,
 							PRJobFn fn, void * arg, PRBool joinable)
 {
 	PRIntervalTime now;
 	PRJob *jobp;
 	if (PR_INTERVAL_NO_TIMEOUT == timeout) {
 		PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);
 		return NULL;
 	}
 	if (PR_INTERVAL_NO_WAIT == timeout) {
 		/*
 		 * no waiting; add to jobq right away
 		 */
 		return(PR_QueueJob(tpool, fn, arg, joinable));
 	}
 	jobp = alloc_job(joinable, tpool);
 	if (NULL == jobp) {
 		return NULL;
 	}
 	/*
 	 * Add a new job to timer_jobq
 	 * wakeup timer worker thread
 	 */
 	jobp->job_func = fn;
 	jobp->job_arg = arg;
 	jobp->tpool = tpool;
 	jobp->timeout = timeout;
 	now = PR_IntervalNow();
 	jobp->absolute = now + timeout;
 	PR_Lock(tpool->timerq.lock);
 	jobp->on_timerq = PR_TRUE;
 	if (PR_CLIST_IS_EMPTY(&tpool->timerq.list))
 		PR_APPEND_LINK(&jobp->links,&tpool->timerq.list);
 	else {
 		PRCList *qp;
 		PRJob *tmp_jobp;
 		/*
 		 * insert into the sorted timer jobq
 		 */
 		for (qp = tpool->timerq.list.prev; qp != &tpool->timerq.list;
 							qp = qp->prev) {
 			tmp_jobp = JOB_LINKS_PTR(qp);
 			if ((PRInt32)(jobp->absolute - tmp_jobp->absolute) >= 0) {
 				break;
 			}
 		}
 		PR_INSERT_AFTER(&jobp->links,qp);
 	}
 	tpool->timerq.cnt++;
 	/*
 	 * notify timer worker thread(s)
 	 */
 	notify_timerq(tpool);
 	PR_Unlock(tpool->timerq.lock);
 	return jobp;
 }
 static void
 notify_timerq(PRThreadPool *tp)
 {
 	/*
 	 * wakeup the timer thread(s)
 	 */
 	PR_NotifyCondVar(tp->timerq.cv);
 }
 static void
 notify_ioq(PRThreadPool *tp)
 {
 PRStatus rval_status;
 	/*
 	 * wakeup the io thread(s)
 	 */
 	rval_status = PR_SetPollableEvent(tp->ioq.notify_fd);
 	PR_ASSERT(PR_SUCCESS == rval_status);
 }
 /*
  * cancel a job
  *
  *	XXXX: is this needed? likely to be removed
  */
 PR_IMPLEMENT(PRStatus)
 PR_CancelJob(PRJob *jobp) {
 	PRStatus rval = PR_FAILURE;
 	PRThreadPool *tp;
 	if (jobp->on_timerq) {
 		/*
 		 * now, check again while holding the timerq lock
 		 */
 		tp = jobp->tpool;
 		PR_Lock(tp->timerq.lock);
 		if (jobp->on_timerq) {
 			jobp->on_timerq = PR_FALSE;
 			PR_REMOVE_AND_INIT_LINK(&jobp->links);
 			tp->timerq.cnt--;
 			PR_Unlock(tp->timerq.lock);
 			if (!JOINABLE_JOB(jobp)) {
 				delete_job(jobp);
 			} else {
 				JOIN_NOTIFY(jobp);
 			}
 			rval = PR_SUCCESS;
 		} else
 			PR_Unlock(tp->timerq.lock);
 	} else if (jobp->on_ioq) {
 		/*
 		 * now, check again while holding the ioq lock
 		 */
 		tp = jobp->tpool;
 		PR_Lock(tp->ioq.lock);
 		if (jobp->on_ioq) {
 			jobp->cancel_cv = PR_NewCondVar(tp->ioq.lock);
 			if (NULL == jobp->cancel_cv) {
 				PR_Unlock(tp->ioq.lock);
 				PR_SetError(PR_INSUFFICIENT_RESOURCES_ERROR, 0);
 				return PR_FAILURE;
 			}
 			/*
 			 * mark job 'cancelled' and notify io thread(s)
 			 * XXXX:
 			 *		this assumes there is only one io thread; when there
 			 * 		are multiple threads, the io thread processing this job
 			 * 		must be notified.
 			 */
 			jobp->cancel_io = PR_TRUE;
 			PR_Unlock(tp->ioq.lock);	/* release, reacquire ioq lock */
 			notify_ioq(tp);
 			PR_Lock(tp->ioq.lock);
 			while (jobp->cancel_io)
 				PR_WaitCondVar(jobp->cancel_cv, PR_INTERVAL_NO_TIMEOUT);
 			PR_Unlock(tp->ioq.lock);
 			PR_ASSERT(!jobp->on_ioq);
 			if (!JOINABLE_JOB(jobp)) {
 				delete_job(jobp);
 			} else {
 				JOIN_NOTIFY(jobp);
 			}
 			rval = PR_SUCCESS;
 		} else
 			PR_Unlock(tp->ioq.lock);
 	}
 	if (PR_FAILURE == rval)
 		PR_SetError(PR_INVALID_STATE_ERROR, 0);
 	return rval;
 }
 /* join a job, wait until completion */
 PR_IMPLEMENT(PRStatus)
 PR_JoinJob(PRJob *jobp)
 {
 	if (!JOINABLE_JOB(jobp)) {
 		PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);
 		return PR_FAILURE;
 	}
 	PR_Lock(jobp->tpool->join_lock);
 	while(jobp->join_wait)
 		PR_WaitCondVar(jobp->join_cv, PR_INTERVAL_NO_TIMEOUT);
 	PR_Unlock(jobp->tpool->join_lock);
 	delete_job(jobp);
 	return PR_SUCCESS;
 }
 /* shutdown threadpool */
 PR_IMPLEMENT(PRStatus)
 PR_ShutdownThreadPool(PRThreadPool *tpool)
 {
 PRStatus rval = PR_SUCCESS;
 	PR_Lock(tpool->jobq.lock);
 	tpool->shutdown = PR_TRUE;
 	PR_NotifyAllCondVar(tpool->shutdown_cv);
 	PR_Unlock(tpool->jobq.lock);
 	return rval;
 }
 /*
  * join thread pool
  * 	wait for termination of worker threads
  *	reclaim threadpool resources
  */
 PR_IMPLEMENT(PRStatus)
 PR_JoinThreadPool(PRThreadPool *tpool)
 {
 PRStatus rval = PR_SUCCESS;
 PRCList *head;
 PRStatus rval_status;
 	PR_Lock(tpool->jobq.lock);
 	while (!tpool->shutdown)
 		PR_WaitCondVar(tpool->shutdown_cv, PR_INTERVAL_NO_TIMEOUT);
 	/*
 	 * wakeup worker threads
 	 */
 #ifdef OPT_WINNT
 	/*
 	 * post shutdown notification for all threads
 	 */
 	{
 		int i;
 		for(i=0; i < tpool->current_threads; i++) {
 			PostQueuedCompletionStatus(tpool->jobq.nt_completion_port, 0,
 												TRUE, NULL);
 		}
 	}
 #else
 	PR_NotifyAllCondVar(tpool->jobq.cv);
 #endif
 	/*
 	 * wakeup io thread(s)
 	 */
 	notify_ioq(tpool);
 	/*
 	 * wakeup timer thread(s)
 	 */
 	PR_Lock(tpool->timerq.lock);
 	notify_timerq(tpool);
 	PR_Unlock(tpool->timerq.lock);
 	while (!PR_CLIST_IS_EMPTY(&tpool->jobq.wthreads)) {
 		wthread *wthrp;
 		head = PR_LIST_HEAD(&tpool->jobq.wthreads);
 		PR_REMOVE_AND_INIT_LINK(head);
 		PR_Unlock(tpool->jobq.lock);
 		wthrp = WTHREAD_LINKS_PTR(head);
 		rval_status = PR_JoinThread(wthrp->thread);
 		PR_ASSERT(PR_SUCCESS == rval_status);
 		PR_DELETE(wthrp);
 		PR_Lock(tpool->jobq.lock);
 	}
 	PR_Unlock(tpool->jobq.lock);
 	while (!PR_CLIST_IS_EMPTY(&tpool->ioq.wthreads)) {
 		wthread *wthrp;
 		head = PR_LIST_HEAD(&tpool->ioq.wthreads);
 		PR_REMOVE_AND_INIT_LINK(head);
 		wthrp = WTHREAD_LINKS_PTR(head);
 		rval_status = PR_JoinThread(wthrp->thread);
 		PR_ASSERT(PR_SUCCESS == rval_status);
 		PR_DELETE(wthrp);
 	}
 	while (!PR_CLIST_IS_EMPTY(&tpool->timerq.wthreads)) {
 		wthread *wthrp;
 		head = PR_LIST_HEAD(&tpool->timerq.wthreads);
 		PR_REMOVE_AND_INIT_LINK(head);
 		wthrp = WTHREAD_LINKS_PTR(head);
 		rval_status = PR_JoinThread(wthrp->thread);
 		PR_ASSERT(PR_SUCCESS == rval_status);
 		PR_DELETE(wthrp);
 	}
 	/*
 	 * Delete queued jobs
 	 */
 	while (!PR_CLIST_IS_EMPTY(&tpool->jobq.list)) {
 		PRJob *jobp;
 		head = PR_LIST_HEAD(&tpool->jobq.list);
 		PR_REMOVE_AND_INIT_LINK(head);
 		jobp = JOB_LINKS_PTR(head);
 		tpool->jobq.cnt--;
 		delete_job(jobp);
 	}
 	/* delete io jobs */
 	while (!PR_CLIST_IS_EMPTY(&tpool->ioq.list)) {
 		PRJob *jobp;
 		head = PR_LIST_HEAD(&tpool->ioq.list);
 		PR_REMOVE_AND_INIT_LINK(head);
 		tpool->ioq.cnt--;
 		jobp = JOB_LINKS_PTR(head);
 		delete_job(jobp);
 	}
 	/* delete timer jobs */
 	while (!PR_CLIST_IS_EMPTY(&tpool->timerq.list)) {
 		PRJob *jobp;
 		head = PR_LIST_HEAD(&tpool->timerq.list);
 		PR_REMOVE_AND_INIT_LINK(head);
 		tpool->timerq.cnt--;
 		jobp = JOB_LINKS_PTR(head);
 		delete_job(jobp);
 	}
 	PR_ASSERT(0 == tpool->jobq.cnt);
 	PR_ASSERT(0 == tpool->ioq.cnt);
 	PR_ASSERT(0 == tpool->timerq.cnt);
 	delete_threadpool(tpool);
 	return rval;
 }

The Tor Browser / annotate

nsprpub/pr/src/misc/prtpool.c@6474c204b198 (annotated)

nsprpub/pr/src/misc/prtpool.c