src/backend/cdb/cdbthreadwork.c - hawq - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */

 /*-------------------------------------------------------------------------
  *
  * cdbbufferedread.c
  *	  Manages a work thread.
  *
  * (See .h file for usage comments)
  *
  *-------------------------------------------------------------------------
  */
 #include "postgres.h"
 #include <pthread.h>
 #include <limits.h>
 #include "cdb/cdbthreadwork.h"
 #include "cdb/cdbvars.h"

 #ifndef _WIN32
 #define mythread() ((unsigned long) pthread_self())
 #else
 #define mythread() ((unsigned long) pthread_self().p)
 #endif

 static void ThreadWork_Main_create_thread(
     ThreadWork         *threadWork);
 static void ThreadWork_Process_pthread_join(
     ThreadWork         *threadWork);
 static void ThreadWork_mutex_lock(
     ThreadWork         *threadWork,
     pthread_mutex_t    *mutex,
     bool               isProcess);
 static void ThreadWork_mutex_unlock(
     ThreadWork         *threadWork,
     pthread_mutex_t    *mutex,
     bool               isProcess);
 static void ThreadWork_cond_signal(
     ThreadWork         *threadWork,
     pthread_cond_t     *cond,
     bool               isProcess);
 static void ThreadWork_cond_wait(
     ThreadWork         *threadWork,
     pthread_cond_t     *cond,
     pthread_mutex_t    *mutex,
     bool               isProcess);
 /*
 static bool ThreadWork_cond_timedwait(
     ThreadWork         *threadWork,
     pthread_cond_t     *cond,
     pthread_mutex_t    *mutex,
 	struct timespec    *timeout,
     bool               isProcess);
 */
 static void* ThreadWork_Thread(void *arg);


 static void ThreadWork_Main_create_thread(
     ThreadWork         *threadWork)
 {
 	int pthread_err = 0;
 	pthread_attr_t t_atts;

 	Assert(threadWork != NULL);

 	/* save ourselves some memory: the defaults for thread stack
 	 * size are large (1M+) */
 	pthread_err = pthread_attr_init(&t_atts);
 	if (pthread_err != 0)
 	{
 //		elog(ERROR, "ThreadWork: pthread_attr_init failed with error %d", pthread_err);
 		fprintf(stderr, "ThreadWork: pthread_attr_init failed with error %d", pthread_err);
 		exit(1);	// UNDONE
 	}

 #ifdef pg_on_solaris
 	/* Solaris doesn't have PTHREAD_STACK_MIN ? */
 	pthread_err = pthread_attr_setstacksize(&t_atts, (256*1024));
 #else
 	pthread_err = pthread_attr_setstacksize(&t_atts, Max(PTHREAD_STACK_MIN, (256*1024)));
 #endif
 	if (pthread_err != 0)
 	{
 //		elog(ERROR, "ThreadWork: pthread_attr_setstacksize failed with error %d", pthread_err);
 		fprintf(stderr, "ThreadWork: pthread_attr_setstacksize failed with error %d", pthread_err);
 		pthread_attr_destroy(&t_atts);
 		exit(1);	// UNDONE
 	}

 	pthread_err = pthread_create(&threadWork->thread, &t_atts, ThreadWork_Thread, threadWork);
 	if (pthread_err != 0)
 	{
 //		elog(ERROR, "ThreadWork: pthread_create failed with error %d", pthread_err);
 		fprintf(stderr, "ThreadWork: pthread_create failed with error %d", pthread_err);
 		pthread_attr_destroy(&t_atts);
 		exit(1);	// UNDONE
 	}

 	pthread_attr_destroy(&t_atts);
 }

 /*
  * Start the work thread.
  */
 void ThreadWorkStart(
     ThreadWork         *threadWork,
     ThreadWorkProc     workProc,
     void			   *passThru)
 {
 	Assert(threadWork != NULL);
 	Assert(workProc != NULL);

 	memset(threadWork, 0, sizeof(ThreadWork));
 	threadWork->workProc = workProc;
 	threadWork->passThru = passThru;

 	pthread_mutex_init(&threadWork->mutex, NULL);
 	pthread_cond_init(&threadWork->commandCond, NULL);
 	pthread_cond_init(&threadWork->answerCond, NULL);

 	threadWork->command = No_WorkCommand;
 	threadWork->answer = No_WorkAnswer;

 	threadWork->waitingForCommand = false;
 	threadWork->waitingForAnswer = false;

 	ThreadWork_Main_create_thread(threadWork);
 }

 static void ThreadWork_Process_pthread_join(
     ThreadWork         *threadWork)
 {
 	int pthread_err;

 	Assert(threadWork != NULL);

 	pthread_err = pthread_join(threadWork->thread, NULL);
 	if (pthread_err != 0)
 	{
 //		elog(ERROR, "ThreadWork: pthread_join failed on thread %lu returned %d attempting to join to %lu)",
 		fprintf(stderr, "ThreadWork: pthread_join failed on thread %lu returned %d attempting to join to %lu)",
 #ifndef _WIN32
 			 (unsigned long) threadWork->thread,
 #else
 			 (unsigned long) threadWork->thread.p,
 #endif
 			 pthread_err, (unsigned long)mythread());
 		exit(1);	// UNDONE
 	}
 }

 static void ThreadWork_mutex_lock(
     ThreadWork         *threadWork,
     pthread_mutex_t    *mutex,
     bool               isProcess)
 {
 	int pthread_err;

 	Assert(threadWork != NULL);

 	pthread_err = pthread_mutex_lock(mutex);
 	if (pthread_err != 0)
 	{
 //		if (isProcess)
 //			elog(ERROR,"ThreadWork: pthread_mutex_lock failed with error %d",
 //				 pthread_err);
 //		else
 //		{
 //			write_log("ThreadWork: pthread_mutex_lock failed with error %d",
 //				      pthread_err);
 			fprintf(stderr, "ThreadWork: pthread_mutex_lock failed with error %d",
 				      pthread_err);
 			exit(1);	// UNDONE.
 //		}
 	}
 }

 static void ThreadWork_mutex_unlock(
     ThreadWork         *threadWork,
     pthread_mutex_t    *mutex,
     bool               isProcess)
 {
 	int pthread_err;

 	Assert(threadWork != NULL);

 	pthread_err = pthread_mutex_unlock(mutex);
 	if (pthread_err != 0)
 	{
 //		if (isProcess)
 //			elog(ERROR,"ThreadWork: pthread_mutex_unlock failed with error %d",
 //				 pthread_err);
 //		else
 //		{
 //			write_log("ThreadWork: pthread_mutex_unlock failed with error %d",
 //				      pthread_err);
 			fprintf(stderr,"ThreadWork: pthread_mutex_unlock failed with error %d",
 				      pthread_err);
 			exit(1);	// UNDONE.
 //		}
 	}
 }

 static void ThreadWork_cond_signal(
     ThreadWork         *threadWork,
     pthread_cond_t     *cond,
     bool               isProcess)
 {
 	int pthread_err;

 	Assert(threadWork != NULL);

 	pthread_err = pthread_cond_signal(cond);
 	if (pthread_err != 0)
 	{
 //		if (isProcess)
 //			elog(ERROR,"ThreadWork: pthread_cond_signal failed with error %d",
 //				 pthread_err);
 //		else
 //		{
 //			write_log("ThreadWork: pthread_cond_signal failed with error %d",
 //				      pthread_err);
 			fprintf(stderr,"ThreadWork: pthread_cond_signal failed with error %d",
 				      pthread_err);
 			exit(1);	// UNDONE.
 //		}
 	}
 }

 static void ThreadWork_cond_wait(
     ThreadWork         *threadWork,
     pthread_cond_t     *cond,
     pthread_mutex_t    *mutex,
     bool               isProcess)
 {
 	int pthread_err;

 	Assert(threadWork != NULL);
 	Assert(cond != NULL);

 	pthread_err = pthread_cond_wait(
 								cond,
 								mutex);
 	if (pthread_err != 0)
 	{
 //		if (isProcess)
 //			elog(ERROR,"ThreadWork: pthread_cond_wait failed with error %d",
 //				 pthread_err);
 //		else
 //		{
 //			write_log("ThreadWork: pthread_cond_wait failed with error %d",
 //				      pthread_err);
 			fprintf(stderr,"ThreadWork: pthread_cond_wait failed with error %d",
 				      pthread_err);
 			exit(1);	// UNDONE.
 //		}
 	}
 }

 /*
 static bool ThreadWork_cond_timedwait(
     ThreadWork         *threadWork,
     pthread_cond_t     *cond,
     pthread_mutex_t    *mutex,
 	struct timespec    *timeout,
     bool               isProcess)
 {
 	int pthread_err;

 	Assert(threadWork != NULL);
 	Assert(cond != NULL);
 	Assert(timeout != NULL);

 	pthread_err = pthread_cond_timedwait(
 									cond,
 									mutex,
 									timeout);
 	if (pthread_err != 0)
 	{
 		if (pthread_err == ETIMEDOUT)
 			return false;

 //		if (isProcess)
 //			elog(ERROR,"ThreadWork: pthread_cond_wait failed with error %d",
 //				 pthread_err);
 //		else
 //		{
 //			write_log("ThreadWork: pthread_cond_wait failed with error %d",
 //				      pthread_err);
 			fprintf(stderr,"ThreadWork: pthread_cond_wait failed with error %d",
 				      pthread_err);
 			exit(1);	// UNDONE.
 //		}
 	}

 	return true;
 }
 */

 /*
  * Give command to the work thread.
  */
 void ThreadWorkGiveCommand(
     ThreadWork          *threadWork,
     int  				command)
 {
 	Assert(threadWork != NULL);

 	ThreadWork_mutex_lock(
 					threadWork,
 					&threadWork->mutex,
 					/* isProcess */ true);

 	Assert(threadWork->command == No_WorkCommand);
 	Assert(threadWork->answer == No_WorkAnswer);
 	threadWork->command = command;

 	if (threadWork->waitingForCommand)
 		ThreadWork_cond_signal(
 						threadWork,
 						&threadWork->commandCond,
 						/* isProcess */ true);

 	ThreadWork_mutex_unlock(
             		threadWork,
 					&threadWork->mutex,
 					/* isProcess */ true);
 }

 /*
  * Get answer from the work thread.
  */
 void ThreadWorkGetAnswer(
     ThreadWork         *threadWork,
     int  			   *answer)
 {
 	Assert(threadWork != NULL);
 	Assert(answer != NULL);

 	ThreadWork_mutex_lock(
 					threadWork,
 					&threadWork->mutex,
 					/* isProcess */ true);

 	while (threadWork->answer == No_WorkAnswer)
 	{
 		/*
 		 * The condition is we are waiting for is an answer.
 		 */
 		threadWork->waitingForAnswer = true;
 		ThreadWork_cond_wait(
 						threadWork,
 						&threadWork->answerCond,
 						&threadWork->mutex,
 						/* isProcess */ false);
 		threadWork->waitingForAnswer = false;
 	}

 	*answer = threadWork->answer;

 	/*
 	 * Clear for next command.
 	 */
 	threadWork->answer = No_WorkAnswer;

 	ThreadWork_mutex_unlock(
             		threadWork,
 					&threadWork->mutex,
 					/* isProcess */ true);
 }

 /*
  * Try (i.e. poll) to get answer from the work thread.
  */
 bool ThreadWorkTryAnswer(
     ThreadWork         *threadWork,
     int  			   *answer)
 {
 	Assert(threadWork != NULL);
 	Assert(answer != NULL);

 	ThreadWork_mutex_lock(
 					threadWork,
 					&threadWork->mutex,
 					/* isProcess */ true);

 	*answer = threadWork->answer;
 	if (*answer != No_WorkAnswer)
 	{
 		/*
 		 * Clear for next command.
 		 */
 		threadWork->answer = No_WorkAnswer;
 	}

 	ThreadWork_mutex_unlock(
             		threadWork,
 					&threadWork->mutex,
 					/* isProcess */ true);


 	return (*answer != No_WorkAnswer);
 }


 /*
  * Give quit to the work thread and wait for it to finish.
  */
 void ThreadWorkQuit(
     ThreadWork         *threadWork)
 {
 	Assert(threadWork != NULL);

 	ThreadWorkGiveCommand(
 					threadWork,
 					Quit_WorkCommand);

 	/*
 	 * Wait for thread to finish.
 	 */
 	ThreadWork_Process_pthread_join(threadWork);

 	pthread_mutex_destroy(&threadWork->mutex);
 	pthread_cond_destroy(&threadWork->commandCond);
 	pthread_cond_destroy(&threadWork->answerCond);
 }

 static void* ThreadWork_Thread(void *arg)
 {
 	ThreadWork *threadWork = (ThreadWork*)arg;

 	int command;
 	int answer;

 	while (true)
 	{
 		ThreadWork_mutex_lock(
 						threadWork,
 						&threadWork->mutex,
 						/* isProcess */ false);

 		while (threadWork->command == No_WorkCommand)
 		{
 			/*
 			 * The condition is we are waiting for is a command.
 			 */
 			threadWork->waitingForCommand = true;
 			ThreadWork_cond_wait(
 							threadWork,
 							&threadWork->commandCond,
 							&threadWork->mutex,
 							/* isProcess */ false);
 			threadWork->waitingForCommand = false;
 		}

 		command = threadWork->command;

 		ThreadWork_mutex_unlock(
 						threadWork,
 						&threadWork->mutex,
 						/* isProcess */ false);

 		if (command == Quit_WorkCommand)
 			break;

 		answer = threadWork->workProc(threadWork->passThru, command);
 		Assert(answer != No_WorkAnswer);

 		/*
 		 * Give answer to main thread.
 		 */
 		ThreadWork_mutex_lock(
 						threadWork,
 						&threadWork->mutex,
 						/* isProcess */ false);

 		Assert(threadWork->command == command);
 		threadWork->command = No_WorkCommand;

 		Assert(threadWork->answer == No_WorkAnswer);
 		threadWork->answer = answer;

 		if (threadWork->waitingForAnswer)
 			ThreadWork_cond_signal(
 							threadWork,
 							&threadWork->answerCond,
 							/* isProcess */ false);

 		ThreadWork_mutex_unlock(
 						threadWork,
 						&threadWork->mutex,
 						/* isProcess */ false);
 	}

 	return NULL;
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/

	/*-------------------------------------------------------------------------
	*
	* cdbbufferedread.c
	* Manages a work thread.
	*
	* (See .h file for usage comments)
	*
	*-------------------------------------------------------------------------
	*/
	#include "postgres.h"
	#include <pthread.h>
	#include <limits.h>
	#include "cdb/cdbthreadwork.h"
	#include "cdb/cdbvars.h"

	#ifndef _WIN32
	#define mythread() ((unsigned long) pthread_self())
	#else
	#define mythread() ((unsigned long) pthread_self().p)
	#endif

	static void ThreadWork_Main_create_thread(
	ThreadWork *threadWork);
	static void ThreadWork_Process_pthread_join(
	ThreadWork *threadWork);
	static void ThreadWork_mutex_lock(
	ThreadWork *threadWork,
	pthread_mutex_t *mutex,
	bool isProcess);
	static void ThreadWork_mutex_unlock(
	ThreadWork *threadWork,
	pthread_mutex_t *mutex,
	bool isProcess);
	static void ThreadWork_cond_signal(
	ThreadWork *threadWork,
	pthread_cond_t *cond,
	bool isProcess);
	static void ThreadWork_cond_wait(
	ThreadWork *threadWork,
	pthread_cond_t *cond,
	pthread_mutex_t *mutex,
	bool isProcess);
	/*
	static bool ThreadWork_cond_timedwait(
	ThreadWork *threadWork,
	pthread_cond_t *cond,
	pthread_mutex_t *mutex,
	struct timespec *timeout,
	bool isProcess);
	*/
	static void* ThreadWork_Thread(void *arg);


	static void ThreadWork_Main_create_thread(
	ThreadWork *threadWork)
	{
	int pthread_err = 0;
	pthread_attr_t t_atts;

	Assert(threadWork != NULL);

	/* save ourselves some memory: the defaults for thread stack
	* size are large (1M+) */
	pthread_err = pthread_attr_init(&t_atts);
	if (pthread_err != 0)
	{
	// elog(ERROR, "ThreadWork: pthread_attr_init failed with error %d", pthread_err);
	fprintf(stderr, "ThreadWork: pthread_attr_init failed with error %d", pthread_err);
	exit(1); // UNDONE
	}

	#ifdef pg_on_solaris
	/* Solaris doesn't have PTHREAD_STACK_MIN ? */
	pthread_err = pthread_attr_setstacksize(&t_atts, (256*1024));
	#else
	pthread_err = pthread_attr_setstacksize(&t_atts, Max(PTHREAD_STACK_MIN, (256*1024)));
	#endif
	if (pthread_err != 0)
	{
	// elog(ERROR, "ThreadWork: pthread_attr_setstacksize failed with error %d", pthread_err);
	fprintf(stderr, "ThreadWork: pthread_attr_setstacksize failed with error %d", pthread_err);
	pthread_attr_destroy(&t_atts);
	exit(1); // UNDONE
	}

	pthread_err = pthread_create(&threadWork->thread, &t_atts, ThreadWork_Thread, threadWork);
	if (pthread_err != 0)
	{
	// elog(ERROR, "ThreadWork: pthread_create failed with error %d", pthread_err);
	fprintf(stderr, "ThreadWork: pthread_create failed with error %d", pthread_err);
	pthread_attr_destroy(&t_atts);
	exit(1); // UNDONE
	}

	pthread_attr_destroy(&t_atts);
	}

	/*
	* Start the work thread.
	*/
	void ThreadWorkStart(
	ThreadWork *threadWork,
	ThreadWorkProc workProc,
	void *passThru)
	{
	Assert(threadWork != NULL);
	Assert(workProc != NULL);

	memset(threadWork, 0, sizeof(ThreadWork));
	threadWork->workProc = workProc;
	threadWork->passThru = passThru;

	pthread_mutex_init(&threadWork->mutex, NULL);
	pthread_cond_init(&threadWork->commandCond, NULL);
	pthread_cond_init(&threadWork->answerCond, NULL);

	threadWork->command = No_WorkCommand;
	threadWork->answer = No_WorkAnswer;

	threadWork->waitingForCommand = false;
	threadWork->waitingForAnswer = false;

	ThreadWork_Main_create_thread(threadWork);
	}

	static void ThreadWork_Process_pthread_join(
	ThreadWork *threadWork)
	{
	int pthread_err;

	Assert(threadWork != NULL);

	pthread_err = pthread_join(threadWork->thread, NULL);
	if (pthread_err != 0)
	{
	// elog(ERROR, "ThreadWork: pthread_join failed on thread %lu returned %d attempting to join to %lu)",
	fprintf(stderr, "ThreadWork: pthread_join failed on thread %lu returned %d attempting to join to %lu)",
	#ifndef _WIN32
	(unsigned long) threadWork->thread,
	#else
	(unsigned long) threadWork->thread.p,
	#endif
	pthread_err, (unsigned long)mythread());
	exit(1); // UNDONE
	}
	}

	static void ThreadWork_mutex_lock(
	ThreadWork *threadWork,
	pthread_mutex_t *mutex,
	bool isProcess)
	{
	int pthread_err;

	Assert(threadWork != NULL);

	pthread_err = pthread_mutex_lock(mutex);
	if (pthread_err != 0)
	{
	// if (isProcess)
	// elog(ERROR,"ThreadWork: pthread_mutex_lock failed with error %d",
	// pthread_err);
	// else
	// {
	// write_log("ThreadWork: pthread_mutex_lock failed with error %d",
	// pthread_err);
	fprintf(stderr, "ThreadWork: pthread_mutex_lock failed with error %d",
	pthread_err);
	exit(1); // UNDONE.
	// }
	}
	}

	static void ThreadWork_mutex_unlock(
	ThreadWork *threadWork,
	pthread_mutex_t *mutex,
	bool isProcess)
	{
	int pthread_err;

	Assert(threadWork != NULL);

	pthread_err = pthread_mutex_unlock(mutex);
	if (pthread_err != 0)
	{
	// if (isProcess)
	// elog(ERROR,"ThreadWork: pthread_mutex_unlock failed with error %d",
	// pthread_err);
	// else
	// {
	// write_log("ThreadWork: pthread_mutex_unlock failed with error %d",
	// pthread_err);
	fprintf(stderr,"ThreadWork: pthread_mutex_unlock failed with error %d",
	pthread_err);
	exit(1); // UNDONE.
	// }
	}
	}

	static void ThreadWork_cond_signal(
	ThreadWork *threadWork,
	pthread_cond_t *cond,
	bool isProcess)
	{
	int pthread_err;

	Assert(threadWork != NULL);

	pthread_err = pthread_cond_signal(cond);
	if (pthread_err != 0)
	{
	// if (isProcess)
	// elog(ERROR,"ThreadWork: pthread_cond_signal failed with error %d",
	// pthread_err);
	// else
	// {
	// write_log("ThreadWork: pthread_cond_signal failed with error %d",
	// pthread_err);
	fprintf(stderr,"ThreadWork: pthread_cond_signal failed with error %d",
	pthread_err);
	exit(1); // UNDONE.
	// }
	}
	}

	static void ThreadWork_cond_wait(
	ThreadWork *threadWork,
	pthread_cond_t *cond,
	pthread_mutex_t *mutex,
	bool isProcess)
	{
	int pthread_err;

	Assert(threadWork != NULL);
	Assert(cond != NULL);

	pthread_err = pthread_cond_wait(
	cond,
	mutex);
	if (pthread_err != 0)
	{
	// if (isProcess)
	// elog(ERROR,"ThreadWork: pthread_cond_wait failed with error %d",
	// pthread_err);
	// else
	// {
	// write_log("ThreadWork: pthread_cond_wait failed with error %d",
	// pthread_err);
	fprintf(stderr,"ThreadWork: pthread_cond_wait failed with error %d",
	pthread_err);
	exit(1); // UNDONE.
	// }
	}
	}

	/*
	static bool ThreadWork_cond_timedwait(
	ThreadWork *threadWork,
	pthread_cond_t *cond,
	pthread_mutex_t *mutex,
	struct timespec *timeout,
	bool isProcess)
	{
	int pthread_err;

	Assert(threadWork != NULL);
	Assert(cond != NULL);
	Assert(timeout != NULL);

	pthread_err = pthread_cond_timedwait(
	cond,
	mutex,
	timeout);
	if (pthread_err != 0)
	{
	if (pthread_err == ETIMEDOUT)
	return false;

	// if (isProcess)
	// elog(ERROR,"ThreadWork: pthread_cond_wait failed with error %d",
	// pthread_err);
	// else
	// {
	// write_log("ThreadWork: pthread_cond_wait failed with error %d",
	// pthread_err);
	fprintf(stderr,"ThreadWork: pthread_cond_wait failed with error %d",
	pthread_err);
	exit(1); // UNDONE.
	// }
	}

	return true;
	}
	*/

	/*
	* Give command to the work thread.
	*/
	void ThreadWorkGiveCommand(
	ThreadWork *threadWork,
	int command)
	{
	Assert(threadWork != NULL);

	ThreadWork_mutex_lock(
	threadWork,
	&threadWork->mutex,
	/* isProcess */ true);

	Assert(threadWork->command == No_WorkCommand);
	Assert(threadWork->answer == No_WorkAnswer);
	threadWork->command = command;

	if (threadWork->waitingForCommand)
	ThreadWork_cond_signal(
	threadWork,
	&threadWork->commandCond,
	/* isProcess */ true);

	ThreadWork_mutex_unlock(
	threadWork,
	&threadWork->mutex,
	/* isProcess */ true);
	}

	/*
	* Get answer from the work thread.
	*/
	void ThreadWorkGetAnswer(
	ThreadWork *threadWork,
	int *answer)
	{
	Assert(threadWork != NULL);
	Assert(answer != NULL);

	ThreadWork_mutex_lock(
	threadWork,
	&threadWork->mutex,
	/* isProcess */ true);

	while (threadWork->answer == No_WorkAnswer)
	{
	/*
	* The condition is we are waiting for is an answer.
	*/
	threadWork->waitingForAnswer = true;
	ThreadWork_cond_wait(
	threadWork,
	&threadWork->answerCond,
	&threadWork->mutex,
	/* isProcess */ false);
	threadWork->waitingForAnswer = false;
	}

	*answer = threadWork->answer;

	/*
	* Clear for next command.
	*/
	threadWork->answer = No_WorkAnswer;

	ThreadWork_mutex_unlock(
	threadWork,
	&threadWork->mutex,
	/* isProcess */ true);
	}

	/*
	* Try (i.e. poll) to get answer from the work thread.
	*/
	bool ThreadWorkTryAnswer(
	ThreadWork *threadWork,
	int *answer)
	{
	Assert(threadWork != NULL);
	Assert(answer != NULL);

	ThreadWork_mutex_lock(
	threadWork,
	&threadWork->mutex,
	/* isProcess */ true);

	*answer = threadWork->answer;
	if (*answer != No_WorkAnswer)
	{
	/*
	* Clear for next command.
	*/
	threadWork->answer = No_WorkAnswer;
	}

	ThreadWork_mutex_unlock(
	threadWork,
	&threadWork->mutex,
	/* isProcess */ true);


	return (*answer != No_WorkAnswer);
	}


	/*
	* Give quit to the work thread and wait for it to finish.
	*/
	void ThreadWorkQuit(
	ThreadWork *threadWork)
	{
	Assert(threadWork != NULL);

	ThreadWorkGiveCommand(
	threadWork,
	Quit_WorkCommand);

	/*
	* Wait for thread to finish.
	*/
	ThreadWork_Process_pthread_join(threadWork);

	pthread_mutex_destroy(&threadWork->mutex);
	pthread_cond_destroy(&threadWork->commandCond);
	pthread_cond_destroy(&threadWork->answerCond);
	}

	static void* ThreadWork_Thread(void *arg)
	{
	ThreadWork threadWork = (ThreadWork)arg;

	int command;
	int answer;

	while (true)
	{
	ThreadWork_mutex_lock(
	threadWork,
	&threadWork->mutex,
	/* isProcess */ false);

	while (threadWork->command == No_WorkCommand)
	{
	/*
	* The condition is we are waiting for is a command.
	*/
	threadWork->waitingForCommand = true;
	ThreadWork_cond_wait(
	threadWork,
	&threadWork->commandCond,
	&threadWork->mutex,
	/* isProcess */ false);
	threadWork->waitingForCommand = false;
	}

	command = threadWork->command;

	ThreadWork_mutex_unlock(
	threadWork,
	&threadWork->mutex,
	/* isProcess */ false);

	if (command == Quit_WorkCommand)
	break;

	answer = threadWork->workProc(threadWork->passThru, command);
	Assert(answer != No_WorkAnswer);

	/*
	* Give answer to main thread.
	*/
	ThreadWork_mutex_lock(
	threadWork,
	&threadWork->mutex,
	/* isProcess */ false);

	Assert(threadWork->command == command);
	threadWork->command = No_WorkCommand;

	Assert(threadWork->answer == No_WorkAnswer);
	threadWork->answer = answer;

	if (threadWork->waitingForAnswer)
	ThreadWork_cond_signal(
	threadWork,
	&threadWork->answerCond,
	/* isProcess */ false);

	ThreadWork_mutex_unlock(
	threadWork,
	&threadWork->mutex,
	/* isProcess */ false);
	}

	return NULL;
	}