/*
 *
 *   (C) Copyright IBM Corp. 2001, 2003
 *
 *   This program is free software;  you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation; either version 2 of the License, or
 *   (at your option) any later version.
 *
 *   This program 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, write to the Free Software
 *   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
 *
 *   Module: daemon.c
 *
 */

#include <stdlib.h>
#include <stdio.h>
#include <stdarg.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>
#include <fcntl.h>
#include <pthread.h>
#include <signal.h>
#include <semaphore.h>
#include <sys/wait.h>

#include "fullengine.h"
#include "dmonmsg.h"
#include "cluster.h"
#include "engine.h"
#include "lists.h"
#include "memman.h"
#include "flatten.h"


sem_t msg_sem;

boolean worker_running = FALSE;

static unsigned char msg_buf[2048];

static inline void log_buffer(void * in_buff, uint len) {

	int i;
	int byte_count = 0;

	unsigned char * byte = (unsigned char *) in_buff;
	unsigned char * p = msg_buf;

	for (i = 0; i < len; i++) {
		sprintf(p, "%02x", *byte);
		p += strlen(p);	
		byte++;
		byte_count++;
		if (byte_count % 16 == 0) {
			LOG_DEBUG("%s\n", msg_buf);
			p = msg_buf;
		} else {
			if (byte_count % 4 == 0) {
				strcat(p, " ");
				p++;
			}
		}
	}

	if (p != msg_buf) {
		LOG_DEBUG("%s\n", msg_buf);
	}
}

#define SEND_MSG(msg)								\
{										\
	int rc;									\
	int retry = 5;								\
	LOG_DEBUG("Send message with command %#x of size %zu to node %s\n", 	\
		  (msg)->cmd, (msg)->size, nodeid_to_string(&((msg)->node)));	\
	do {									\
		rc = ece_funcs->send_msg(msg);					\
										\
		if (rc != 0) {							\
			if (rc == EAGAIN) {					\
				usleep(1000000);				\
				retry--;					\
				LOG_DEBUG("Retry count is %d.\n", retry);	\
			} else {						\
				LOG_SERIOUS("send_msg() to node %s returned error code %d: %s\n",	\
					    nodeid_to_string(&((msg)->node)), rc, evms_strerror(rc));	\
			}							\
		}								\
	} while ((rc == EAGAIN) && (retry > 0));				\
}

typedef struct worker_context_s {
	int pid;
        int input_pipe[2];
	int output_pipe[2];
        pthread_mutex_t input_pipe_mutex;
	pthread_mutex_t output_pipe_mutex;
} worker_context_t;

worker_context_t * worker = NULL;

typedef struct msg_pool_ent_s {
	element_t element;
	ece_msg_t msg;
} msg_pool_ent_t;

STATIC_LIST_DECL(msg_pool);
static pthread_mutex_t msg_pool_mutex = PTHREAD_MUTEX_INITIALIZER;
static sem_t msg_pool_sem;

/*
 * Get a message from the message pool or allocate a new one.
 * get_msg() is guaranteed to return a pointer to an ece_msg_t.
 * It may take a while or get blocked forever if there is a
 * deadlock.  But if it returns, the pointer will be valid.
 */
ece_msg_t * get_msg(const ece_msg_t * copy_msg) {

	msg_pool_ent_t * msg_pool_ent = NULL;
	ece_msg_t * msg = NULL;
	int rc;

	LOG_PROC_ENTRY();

	pthread_mutex_lock(&msg_pool_mutex);

	while (msg_pool_ent == NULL) {
		if (!list_empty(&msg_pool)) {
			msg_pool_ent = (msg_pool_ent_t *) msg_pool.links.next;
			remove_element(&msg_pool_ent->element);

		} else {
                        /* Try to allocate a new one. */
			msg_pool_ent = engine_alloc(sizeof(msg_pool_ent_t));

			if (msg_pool_ent == NULL) {
				/*
				 * Couldn't allocate a new one.
				 * Wait until one becomes available.
				 */
				sem_init(&msg_pool_sem, 0, 0);
				pthread_mutex_unlock(&msg_pool_mutex);
				do {
					rc = sem_wait(&msg_pool_sem);
				} while ((rc != 0) && (errno == EINTR));
				pthread_mutex_lock(&msg_pool_mutex);
			}
		}
	}

	msg = &(msg_pool_ent->msg);
	*msg = *copy_msg;

	pthread_mutex_unlock(&msg_pool_mutex);

	LOG_PROC_EXIT_PTR(msg);
	return msg;
}


/*
 * Return a message to the message pool.
 */
void free_msg(ece_msg_t * msg) {

	int sem_value = 0;
	msg_pool_ent_t * msg_pool_ent = (msg_pool_ent_t *) ((void *)msg - offsetof(msg_pool_ent_t, msg));

	LOG_PROC_ENTRY();
	LOG_DEBUG("Free %p\n", msg);

	pthread_mutex_lock(&msg_pool_mutex);

	insert_element(&msg_pool, &msg_pool_ent->element, INSERT_AFTER, NULL);

	sem_getvalue(&msg_pool_sem, &sem_value);
	if (sem_value != 0) {
		sem_post(&msg_pool_sem);
	}

	pthread_mutex_unlock(&msg_pool_mutex);

	LOG_PROC_EXIT_VOID();
}


static void msg_get_daemon_api_version(ece_msg_t * msg) {

	ece_msg_t * response_msg = get_msg(msg);
	evms_version_t net_version;

	LOG_PROC_ENTRY();

	/* It's OK to release the message. */
	sem_post(&msg_sem);

	evms_host_to_net(&net_version, evms_version_f,
			 engine_daemon_msg_version.major,
			 engine_daemon_msg_version.minor,
			 engine_daemon_msg_version.patchlevel);

	response_msg->cmd |= COMMAND_RESPONSE;
	response_msg->size = sizeof(net_version);
	response_msg->msg = &net_version;

	SEND_MSG(response_msg);

	free_msg(response_msg);
	LOG_PROC_EXIT_VOID();
}


static void msg_get_engine_api_version(ece_msg_t * msg) {

	ece_msg_t * response_msg = get_msg(msg);
	char response[sizeof(u_int32_t) + sizeof(evms_version_t)];

	LOG_PROC_ENTRY();

	/* Now it's OK to release the message. */
	sem_post(&msg_sem);

	evms_host_to_net(response, evms_get_api_version_rets_f,
			 0,
			 engine_api_version.major,
			 engine_api_version.minor,
			 engine_api_version.patchlevel);

	response_msg->cmd |= COMMAND_RESPONSE;
	response_msg->size = sizeof(response);
	response_msg->msg = &response;

	SEND_MSG(response_msg);

	free_msg(response_msg);
	LOG_PROC_EXIT_VOID();
}


static void msg_shutdown(ece_msg_t * msg) {

	ece_msg_t * response_msg = get_msg(msg);
	struct  flock lockinfo = {0};
	int status;
	u_int32_t net_rc = 0;

	LOG_PROC_ENTRY();

	/* Now it's OK to release the message. */
	sem_post(&msg_sem);


	/* Check if the Engine is open by trying to obtain the Engine lock. */
	lockinfo.l_type   = F_RDLCK;
	lockinfo.l_whence = SEEK_SET;
	lockinfo.l_start  = offsetof(lock_file_t, engine_lock);
	lockinfo.l_len    = sizeof(lock_file->engine_lock);

	status = fcntl(lock_file_fd, F_SETLK, &lockinfo);
	if (status != 0) {
		/*
		 * Reread the lock file to see which process owns the Engine
		 * lock.
		 */
                lseek(lock_file_fd, offsetof(lock_file_t, pid), SEEK_SET);
		read(lock_file_fd, &lock_file->pid, lock_file_size - offsetof(lock_file_t, pid));

		/*
		 * The shutdown signal is not sent to a worker.  It is only
		 * send to the console that opened the Engine.
		 */
		if (!(lock_file->mode & ENGINE_WORKER)) {
			send_shutdown(lock_file->pid);
		}

	} else {
		/*
		 * No one owns the Engine lock.  Therefore there is no one to
		 * shutdown.  Release the lock.
		 */
		lockinfo.l_type = F_UNLCK;
		fcntl(lock_file_fd, F_SETLK, &lockinfo);
	}

	response_msg->cmd |= COMMAND_RESPONSE;
	response_msg->size = sizeof(net_rc);
	response_msg->msg = &net_rc;
	SEND_MSG(response_msg);

	free_msg(response_msg);
	LOG_PROC_EXIT_VOID();
}


static int send_msg_to_worker(const ece_msg_t * msg) {

	msg_hdr_t input_msg_hdr;
	int bytes_written;

	LOG_PROC_ENTRY();

	input_msg_hdr.signature = MSG_HDR_SIG;
	input_msg_hdr.cmd       = msg->cmd;
	input_msg_hdr.size      = msg->size;

        pthread_mutex_lock(&worker->input_pipe_mutex);

	LOG_DEBUG("Request to write %zd bytes to fd %d.\n", sizeof(input_msg_hdr), worker->input_pipe[1]);
	bytes_written = write(worker->input_pipe[1], &input_msg_hdr,  sizeof(input_msg_hdr));
	if (bytes_written == -1) {
		LOG_DEBUG("Write of message header failed with errno %d: %s.\n", errno, strerror(errno));
		LOG_PROC_EXIT_INT(errno);
		return errno;
	} else {
		LOG_DEBUG("%d of %zd bytes written.\n", bytes_written, sizeof(input_msg_hdr));
	}

	if (msg->size > 0) {
		LOG_DEBUG("Request to write %zd bytes to fd %d.\n", msg->size, worker->input_pipe[1]);
		bytes_written = write(worker->input_pipe[1], msg->msg, msg->size);
		if (bytes_written == -1) {
			LOG_DEBUG("Write of message failed with errno %d: %s.\n", errno, strerror(errno));
			LOG_PROC_EXIT_INT(errno);
			return errno;
		} else {
			LOG_DEBUG("%d of %zd bytes written.\n", bytes_written, msg->size);
		}
	}

        pthread_mutex_unlock(&worker->input_pipe_mutex);

	LOG_PROC_EXIT_INT(0);
	return 0;
}


static int receive_from_worker(ece_msg_t * response_msg) {

	msg_hdr_t output_msg_hdr;
	int bytes_read;

	LOG_PROC_ENTRY();

	pthread_mutex_lock(&worker->output_pipe_mutex);

	LOG_DEBUG("Request to read %zd bytes from fd %d.\n", sizeof(output_msg_hdr), worker->output_pipe[0]);
	bytes_read = read(worker->output_pipe[0], &output_msg_hdr, sizeof(output_msg_hdr));
	if (bytes_read == -1) {
		LOG_SERIOUS("Read of message failed with errno %d: %s.\n", errno, strerror(errno));
		LOG_PROC_EXIT_INT(errno);
		return errno;
	} else {
		LOG_DEBUG("%d of %zd bytes read.\n", bytes_read, sizeof(output_msg_hdr));
	}

	response_msg->cmd  = output_msg_hdr.cmd;
	response_msg->size = output_msg_hdr.size;

	if (output_msg_hdr.size > 0) {
		LOG_DEBUG("Allocate %u bytes for response message.\n", output_msg_hdr.size);

		response_msg->msg = engine_alloc(output_msg_hdr.size);
		if (response_msg->msg == NULL) {
			LOG_DEBUG("Failed to allocate memory for a receive buffer.\n");
			response_msg->size = 0;
			LOG_PROC_EXIT_INT(ENOMEM);
			return ENOMEM;
		}

		LOG_DEBUG("Request to read %d bytes from fd %d.\n", output_msg_hdr.size, worker->output_pipe[0]);
		bytes_read = read(worker->output_pipe[0], response_msg->msg, output_msg_hdr.size);
		if (bytes_read == -1) {
			LOG_SERIOUS("Read of message failed with errno %d: %s.\n", errno, strerror(errno));
			LOG_PROC_EXIT_INT(errno);
			return errno;
		} else {
			LOG_DEBUG("%d of %d bytes read.\n", bytes_read, output_msg_hdr.size);
		}

	} else {
		response_msg->msg = NULL;
	}

	pthread_mutex_unlock(&worker->output_pipe_mutex);

	LOG_PROC_EXIT_INT(0);
	return 0;
}


static int receive_response_for_command(ece_msg_t * response_msg) {

	u_int32_t required_cmd = response_msg->cmd;
	int rc;

	LOG_PROC_ENTRY();

	do {
		rc = receive_from_worker(response_msg);

		if ((rc == 0) &&
		    (((response_msg->cmd & COMMAND_MASK) != (required_cmd & COMMAND_MASK)) ||
		     !(response_msg->cmd & COMMAND_RESPONSE))) {

			/*
			 * It's an out going request, such as a
			 * message or progress report.
			 * Send it back to the calling node.
			 * Leave the corrolator in the msg so that the caller
			 * can match it up with the thread that issued the call.
			 */
			if ((response_msg->cmd & COMMAND_MASK) != (required_cmd & COMMAND_MASK)) {
				LOG_DEBUG("Receive buffer command is %u which does not match the required command of %u.\n",
				       (response_msg->cmd & COMMAND_MASK), (required_cmd & COMMAND_MASK));
			}
			if (!(response_msg->cmd & COMMAND_RESPONSE)) {
				LOG_DEBUG("Received command %u is not a response.\n", response_msg->cmd);
			}

			LOG_DEBUG("Sending the message back to node %s.\n", nodeid_to_string(&response_msg->node));
			SEND_MSG(response_msg);
		}

		LOG_DEBUG("rc is %d.\n", rc);
		LOG_DEBUG("response cmd is %d.  msg cmd is %d.\n", (response_msg->cmd & COMMAND_MASK), (required_cmd & COMMAND_MASK));

	} while ((rc == 0) &&
		 (((response_msg->cmd & COMMAND_MASK) != (required_cmd & COMMAND_MASK)) ||
		  !(response_msg->cmd & COMMAND_RESPONSE)));

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


static void process_api(const ece_msg_t * msg) {

	ece_msg_t * response_msg = get_msg(msg);
	int rc;
	u_int32_t net_rc;

	LOG_PROC_ENTRY();

	send_msg_to_worker(msg);

	/* Now it's OK to release the message. */
	sem_post(&msg_sem);

	rc = receive_response_for_command(response_msg);

	if (rc == 0) {
		SEND_MSG(response_msg);

		/* engine_free() handles NULL pointers. */
		engine_free(response_msg->msg);

	} else {
		evms_host_to_net(&net_rc, int_f, rc);
		response_msg->cmd |= COMMAND_RESPONSE;
		response_msg->size = sizeof(net_rc);
		response_msg->msg  = &net_rc;
		SEND_MSG(response_msg);
	}

	free_msg(response_msg);
	LOG_PROC_EXIT_VOID();
}


static void shutdown_worker(worker_context_t * wrkr) {

	int status;
	pid_t pid;
	int timeout = 5;  /* seconds */
	
	LOG_PROC_ENTRY();

	LOG_DEBUG("Worker context: %p   worker pid: %d\n", wrkr, wrkr->pid);

	close(wrkr->input_pipe[0]);
	close(wrkr->input_pipe[1]);
	close(wrkr->output_pipe[0]);
	close(wrkr->output_pipe[1]);

	/* Update the node field in the lock file. */
	lock_file->node[0] = '\0';
	lseek(lock_file_fd, offsetof(lock_file_t, node), SEEK_SET);
	write(lock_file_fd, lock_file->node, 1);

	/* Nicely ask the worker to shutdown. */
	kill(wrkr->pid, SIGTERM);

	do {
		pid = waitpid(wrkr->pid, &status, WNOHANG);

		if (pid == 0) {
			LOG_DEBUG("Wait for worker pid %d to exit.\n", wrkr->pid);
			usleep(100000);
			timeout--;
		}

	} while ((pid == 0) && (timeout > 0));

	if (pid == 0) {
		/* Nice didn't work.  Kill it. */
		LOG_DEBUG("Kill worker pid %d.\n", wrkr->pid);
                kill(wrkr->pid, SIGKILL);
	}

	/* Wait for the worker to terminate and cleanup the defunct process. */
        pid = waitpid(wrkr->pid, &status, 0);
	LOG_DEBUG("Worker pid %d is dead.\n", wrkr->pid);

	LOG_DEBUG("Free worker context: %p\n", wrkr);
	engine_free(wrkr);

	LOG_PROC_EXIT_VOID();
}


#define WORKER_NAME	"evmsd_worker"

static void process_open_engine(const ece_msg_t * msg) {

	char * argv[] = {WORKER_NAME, NULL};
	const char * node_name;
	size_t req_lock_file_size;
	ece_msg_t * response_msg;
	u_int32_t net_rc;

	LOG_PROC_ENTRY();

	/*
	 * Check if the worker is already running.  If so, shut it down before
	 * launching a new worker.  Only one worker can be running at a time.
	 * the worker can be left running if the Engine on the console node gets
	 * terminated.  We don't have to worry about an EVMS_OPEN_ENGINE coming
	 * in while this worker should be running.  The protocol won't do that.
	 */
	if (worker_running) {
		shutdown_worker(worker);
	}

	worker = engine_alloc(sizeof(worker_context_t));

	if (worker == NULL) {
		LOG_CRITICAL("Failed to allocate memory for a worker context.\n");
		response_msg = get_msg(msg);

		evms_host_to_net(&net_rc, int_f, ENOMEM);

		response_msg->cmd |= COMMAND_RESPONSE;
		response_msg->size = sizeof(net_rc);
		response_msg->msg  = &net_rc;

		SEND_MSG(response_msg);

		free_msg(response_msg);

		LOG_PROC_EXIT_VOID();
		return;
	}

	LOG_DEBUG("new worker context: %p\n", worker);

	pthread_mutex_init(&worker->input_pipe_mutex, NULL);
	pthread_mutex_init(&worker->output_pipe_mutex, NULL);

	if (pipe(worker->input_pipe)) {
		LOG_SERIOUS("Failed to create an input pipe for the worker.  Error code is %d: %s\n",
			    errno, strerror(errno));

		response_msg = get_msg(msg);

		evms_host_to_net(&net_rc, int_f, errno);

		response_msg->cmd |= COMMAND_RESPONSE;
		response_msg->size = sizeof(net_rc);
		response_msg->msg  = &net_rc;

		SEND_MSG(response_msg);

		free_msg(response_msg);

		engine_free(worker);
		worker = NULL;

		LOG_PROC_EXIT_VOID();
		return;
	}

	LOG_DEBUG("worker->input_pipe handles are %d (read) and %d (write).\n", worker->input_pipe[0], worker->input_pipe[1]);

	if (pipe(worker->output_pipe)) {
		LOG_SERIOUS("Failed to create an output pipe for the worker.  Error code is %d: %s\n",
			    errno, strerror(errno));

		response_msg = get_msg(msg);

		evms_host_to_net(&net_rc, int_f, errno);

		response_msg->cmd |= COMMAND_RESPONSE;
		response_msg->size = sizeof(net_rc);
		response_msg->msg  = &net_rc;

		SEND_MSG(response_msg);

		free_msg(response_msg);

		close(worker->input_pipe[0]);
		close(worker->input_pipe[1]);

		engine_free(worker);
		worker = NULL;

		LOG_PROC_EXIT_VOID();
		return;
	}

	LOG_DEBUG("worker->output_pipe handles are %d (read) and %d (write).\n", worker->output_pipe[0], worker->output_pipe[1]);

	worker->pid = fork();
	
	switch (worker->pid) {
		case -1:
			/* Fork failed.  Send back an error code. */
			LOG_SERIOUS("Failed to fork to create the worker process.  Error code is %d: %s\n",
				    errno, strerror(errno));

			response_msg = get_msg(msg);

			evms_host_to_net(&net_rc, int_f, errno);

			response_msg->cmd |= COMMAND_RESPONSE;
			response_msg->size = sizeof(net_rc);
			response_msg->msg  = &net_rc;

			SEND_MSG(response_msg);

			free_msg(response_msg);

			close(worker->input_pipe[0]);
			close(worker->input_pipe[1]);
			close(worker->output_pipe[0]);
			close(worker->output_pipe[1]);
			engine_free(worker);
			worker = NULL;

			break;

		case 0:
			/* Child process */
			dup2(worker->input_pipe[0], 0);	/* Route stdin to worker->input_pipe's read fd */
			dup2(worker->output_pipe[1], 1);/* Route stdout to worker->output_pipe's write fd */

			LOG_DEFAULT("execvp(" WORKER_NAME ")\n");
			execvp(WORKER_NAME, argv);

			/* If the execvp() works it won't return here. */
			LOG_WARNING("execvp() failed.  errno is %d: %s\n", errno, strerror(errno));

			/* Using exit() can hang the GUI, use _exit */
			_exit(errno);
			break;

		default:
			/* Parent process */
			usleep(100000);

			worker_running = TRUE;
			LOG_DEBUG("New worker pid: %d\n", worker->pid);

			/* Update the node field in the lock file. */
			node_name = nodeid_to_string(&msg->node);
			req_lock_file_size = sizeof(lock_file_t) + strlen(node_name) + 1;

			if (lock_file_size < req_lock_file_size) {
				lock_file = engine_realloc(lock_file, req_lock_file_size);
			}
			if (lock_file != NULL) {
				lock_file_size = req_lock_file_size;
				strcpy(lock_file->node, node_name);
				lseek(lock_file_fd, offsetof(lock_file_t, node), SEEK_SET);
				write(lock_file_fd, lock_file->node, strlen(lock_file->node) + 1);
			}

			/* Send the evms_open_engine() API to the worker. */
			process_api(msg);
	}

	LOG_PROC_EXIT_VOID();
}


static void process_close_engine(ece_msg_t * msg) {

	worker_context_t * wrkr = worker;

	LOG_PROC_ENTRY();

	/*
	 * Mark the worker thread stopped before process_api() releases
	 * the daemon_router to handle more commands.  We don't want any
	 * subsequent incoming commands to think that the worker is running,
	 * since it will eventually be terminated.
	 */
	worker_running = FALSE;

	/* Send the evms_close_engine() API to the worker. */
	process_api(msg);

	shutdown_worker(wrkr);

	LOG_PROC_EXIT_VOID();
}


void * daemon_thread(void * arg) {

	ece_msg_t * msg = (ece_msg_t *) arg;

	LOG_PROC_ENTRY();

	log_buffer(msg->msg, msg->size);
	if (msg->cmd == EVMS_OPEN_ENGINE) {
		process_open_engine(msg);

	} else if (msg->cmd == EVMS_CLOSE_ENGINE) {
		process_close_engine(msg);

	} else {
		process_api(msg);
	}

	LOG_PROC_EXIT_VOID();
	return(NULL);
}


void daemon_router(ece_msg_t * msg) {

	u_int32_t cmd = msg->cmd;
	pthread_t tid;
	int rc;

	LOG_PROC_ENTRY();

	LOG_DEBUG("Message from node %s: command %#x (%s %s)  size: %zu\n",
		  nodeid_to_string(&msg->node), cmd, msg_cmd_name(cmd), (cmd & COMMAND_RESPONSE) ? "response" : "request", msg->size);

	/* Make sure the command is within range. */
	if ((cmd & COMMAND_MASK) >= INVALID_COMMAND) {
		ece_msg_t * response = get_msg(msg);

		LOG_DEBUG("%d is not a valid message command.\n", msg->cmd);
		response->cmd = MSG_INVALID_CMD | COMMAND_RESPONSE;
		response->msg = NULL;
		response->size = 0;
		SEND_MSG(response);
		free_msg(response);
		return;
	}

	/*
	 * The following commands are handled by the daemon directly.
	 */
	switch (cmd) {
		case MSG_GET_VERSION:
			msg_get_daemon_api_version(msg);
			return;

		case EVMS_GET_API_VERSION:
			msg_get_engine_api_version(msg);
			return;

		case SHUTDOWN:
			msg_shutdown(msg);
			return;
	}

	if (cmd & COMMAND_RESPONSE) {
		/* Forward incoming responses to callbacks on to the worker. */
		send_msg_to_worker(msg);

	} else {
		/*
		 * The EVMS_OPEN_ENGINE command will launch the worker thread.
		 * and open the Engine.  All other commands require the Engine
		 * to be open, which means the worker thread must be running.
		 * No sense sending the commands through the pipe to the worker
		 * thread if there is no worker thread.
		 */
		if ((cmd != EVMS_OPEN_ENGINE) && !worker_running) {
			ece_msg_t * response = get_msg(msg);
			u_int32_t flat_int = HOST_TO_NET32((u_int32_t) EINVAL);

			LOG_DEBUG("The Engine is not open.  The worker thread is not running.\n");
			response->cmd |= COMMAND_RESPONSE;
			response->msg = &flat_int;
			response->size = sizeof(flat_int);
			SEND_MSG(response);
			free_msg(response);
			return;
		}

		sem_init(&msg_sem, 0, 0);

		rc = pthread_create(&tid, &pthread_attr_detached, daemon_thread, msg);
		if (rc == 0) {
			/* Wait until it is OK to release the message. */
			do {
				rc = sem_wait(&msg_sem);
			} while ((rc != 0) && (errno == EINTR));

		} else {
			LOG_WARNING("Failed to launch the daemon thread to process the command.  "
				    "The command will be processed serially.\n");

			daemon_thread(msg);
		}
	}

	LOG_PROC_EXIT_VOID();
}