/*
 * Copyright 2021-2022, Andrew Lindesay <apl@lindesay.co.nz>.
 * All rights reserved. Distributed under the terms of the MIT License.
 */


#include "ThreadedProcessNode.h"

#include <unistd.h>

#include "AbstractProcess.h"
#include "Logger.h"
#include "ProcessListener.h"


#define TIMEOUT_UNTIL_STARTED_SECS_DEFAULT 10
#define TIMEOUT_UNTIL_STOPPED_SECS_DEFAULT 10


ThreadedProcessNode::ThreadedProcessNode(AbstractProcess* process,
		int32 startTimeoutSeconds)
	:
	AbstractProcessNode(process),
	fWorker(B_BAD_THREAD_ID),
	fStartTimeoutSeconds(startTimeoutSeconds)
{
}


ThreadedProcessNode::ThreadedProcessNode(AbstractProcess* process)
	:
	AbstractProcessNode(process),
	fWorker(B_BAD_THREAD_ID),
	fStartTimeoutSeconds(TIMEOUT_UNTIL_STARTED_SECS_DEFAULT)
{
}


ThreadedProcessNode::~ThreadedProcessNode()
{
	if (IsRunning()) {
		HDFATAL("the process node is being deleted while the thread is"
			"still running");
	}
}


bool
ThreadedProcessNode::IsRunning()
{
	if (!AbstractProcessNode::IsRunning()) {
		AutoLocker<BLocker> locker(fLock);
		if (fWorker == B_BAD_THREAD_ID)
			return false;
		thread_info ti;
		status_t status = get_thread_info(fWorker, &ti);
		if (status != B_OK)
			// implies that the thread has stopped
			return false;
		HDTRACE("[Node<%s>] thread still running...", Process()->Name());
	}
	return true;
}


/*! Considered to be protected from concurrent access by the ProcessCoordinator
*/

status_t
ThreadedProcessNode::Start()
{
	AutoLocker<BLocker> locker(fLock);
	if (fWorker != B_BAD_THREAD_ID)
		return B_BUSY;

	HDINFO("[Node<%s>] initiating threaded", Process()->Name());

	fWorker = spawn_thread(&_RunProcessThreadEntry, Process()->Name(),
		B_NORMAL_PRIORITY, this);

	if (fWorker >= 0) {
		resume_thread(fWorker);
		return _SpinUntilProcessState(PROCESS_RUNNING | PROCESS_COMPLETE,
			fStartTimeoutSeconds);
	}

	return B_ERROR;
}


status_t
ThreadedProcessNode::RequestStop()
{
	return Process()->Stop();
}


void
ThreadedProcessNode::_RunProcessStart()
{
	if (fListener != NULL) {
		if (on_exit_thread(&_RunProcessThreadExit, this) != B_OK) {
			HDFATAL("unable to setup 'on exit' for thread");
		}
	}

	AbstractProcess* process = Process();

	if (process == NULL)
		HDFATAL("the process node must have a process defined");

	bigtime_t start = system_time();
	HDINFO("[Node<%s>] starting process in thread", process->Name());
	process->Run();
	HDINFO("[Node<%s>] finished process in thread %f seconds", process->Name(),
		(system_time() - start) / 1000000.0);
}


/*! This method is the initial function that is invoked on starting a new
	thread.  It will start a process that is part of the bulk-load.
 */

/*static*/ status_t
ThreadedProcessNode::_RunProcessThreadEntry(void* cookie)
{
	static_cast<ThreadedProcessNode*>(cookie)->_RunProcessStart();
	return B_OK;
}


void
ThreadedProcessNode::_RunProcessExit()
{
	AutoLocker<BLocker> locker(fLock);
	fWorker = B_BAD_THREAD_ID;
	HDTRACE("[Node<%s>] compute complete", Process()->Name());
	if (fListener != NULL)
		fListener->ProcessChanged();
}


/*static*/ void
ThreadedProcessNode::_RunProcessThreadExit(void* cookie)
{
	static_cast<ThreadedProcessNode*>(cookie)->_RunProcessExit();
}