myvtm/libtoolkit/process_monitor-unix.c

#include "precompile.h"
#include "process_monitor.h"
#include "list.h"
#include "spinlock.h"
#include <assert.h>
#include <winpr/synch.h>
#include <winpr/thread.h>
#include <winpr/wlog.h>

#define TAG TOOLKIT_TAG("process_monitor")

#define CMD_REMOVE	0
#define CMD_ADD		    1
#define CMD_EXIT	        2

#define EVT_IDX             0
#define INIT_CNT           1

typedef struct cmd_entry_t 
{
	struct list_head entry;
	int cmd;
	HANDLE hproc;
	tk_process_t* proc;
	HANDLE wait_evt;
	int result;
}cmd_entry_t;

typedef struct monitor_t 
{
	struct list_head entry;
	HANDLE thread;
	/*[evt, process_handle1, process_handle2,..., process_handleN] */
	HANDLE arr_handle[MAXIMUM_WAIT_OBJECTS];
	int arr_cnt;
	struct list_head cmd_list;
	process_monitor_t *parent;
}monitor_t;

struct process_monitor_t
{
	struct list_head full_monitor_list;
	struct list_head avail_monitor_list;
	CRITICAL_SECTION lock;
	process_monitor_on_detect_process_end cb;
	void *user_data;
};

static void process_monitor_lock(process_monitor_t *pm)
{
	EnterCriticalSection(&pm->lock);
}

static void process_monitor_unlock(process_monitor_t *pm)
{
	LeaveCriticalSection(&pm->lock);
}

static unsigned int __stdcall monitor_thread_proc(void *arg)
{
	monitor_t *monitor = (monitor_t*)arg;

	for (;;) {
		DWORD dwRet = WaitForSingleObject(monitor->arr_handle[0], 500);
		int i;
		if (dwRet == WAIT_OBJECT_0) {
			cmd_entry_t* cmd = list_first_entry(&monitor->cmd_list, cmd_entry_t, entry);
			list_del(&cmd->entry);
			if (cmd->cmd == CMD_ADD) {
				monitor->arr_handle[monitor->arr_cnt++] = cmd->hproc;
				cmd->result = 0;
				if (cmd->wait_evt)
					SetEvent(cmd->wait_evt);
			}
			else if (cmd->cmd == CMD_REMOVE) {
				int i;
				cmd->result = -1;
				for (i = 0; i < monitor->arr_cnt; ++i) {
					if (monitor->arr_handle[i] == cmd->hproc) {
						if (i != monitor->arr_cnt - 1) {
							monitor->arr_handle[i] = monitor->arr_handle[monitor->arr_cnt - 1];
						}
						monitor->arr_cnt--;
						cmd->result = 0;
						break;
					}
				}
				if (cmd->wait_evt)
					SetEvent(cmd->wait_evt);
			}
			else if (cmd->cmd == CMD_EXIT) {
				cmd->result = 0;
				if (cmd->wait_evt)
					SetEvent(cmd->wait_evt);
				break;
			}
			else {
				cmd->result = -1;
				if (cmd->wait_evt)
					SetEvent(cmd->wait_evt);
			}
		}
		else if (dwRet != WAIT_TIMEOUT) {
			WLog_ERR(TAG, "WaitForSingleObject failed!");
			abort();
		}
		for (i = 1; i < monitor->arr_cnt; ++i) {
			dwRet = WaitForSingleObject(monitor->arr_handle[i], 0);
			if (dwRet == WAIT_OBJECT_0) {
				int idx = i;
				HANDLE hProc = monitor->arr_handle[idx];
				WLog_DBG(TAG, "process exit, execute callback!");
				if ((*monitor->parent->cb)(monitor->parent, hProc, monitor->parent->user_data)) {
					if (monitor->arr_handle[idx] == hProc) {
						if (idx != monitor->arr_cnt - 1)
							monitor->arr_handle[idx] = monitor->arr_handle[monitor->arr_cnt - 1];
						monitor->arr_cnt--;
					}
					else {
						assert(false);
					}
				}
			}
		}
	}

	return 0;
}

TOOLKIT_API int process_monitor_add(process_monitor_t *monitor, tk_process_t* process)
{
	cmd_entry_t ce;
	monitor_t *m = NULL;

	assert(monitor);
	assert(process != NULL);

	ce.wait_evt = CreateEventA(NULL, FALSE, FALSE, NULL);
	if (!ce.wait_evt)
		return -1;
	ce.cmd = CMD_ADD;
	ce.hproc = process->handle;

	process_monitor_lock(monitor);
	if (list_empty(&monitor->avail_monitor_list)) {
		m = MALLOC_T(monitor_t);
		if(m == NULL) {
			return -1;
		}
		m->arr_cnt = INIT_CNT;
		/*
		 *The state of a semaphore is signaled when its count is greater than zero
		 *the count is increased by a specified amount by calling ReleaseSemaphore.
		 */
		m->arr_handle[EVT_IDX] = CreateSemaphoreA(NULL, 0, 0x7fffffff, NULL);
		if (!m->arr_handle[EVT_IDX]) {
			free(m);
			process_monitor_unlock(monitor);
			return -1;
		}
		m->parent = monitor;
		m->thread = (HANDLE)_beginthreadex(NULL, 0, &monitor_thread_proc, m, 0, NULL);
		if (!m->thread) {
			CloseHandle(m->arr_handle[EVT_IDX]);
			free(m);
			process_monitor_unlock(monitor);
			return -1;
		}
		INIT_LIST_HEAD(&m->cmd_list);
		/*only one elem at current time*/
		list_add_tail(&m->entry, &monitor->avail_monitor_list);
	} else {
		/*get the existed elem*/
		m = list_first_entry(&monitor->avail_monitor_list, monitor_t, entry);
	}
	list_add_tail(&ce.entry, &m->cmd_list);
	ReleaseSemaphore(m->arr_handle[EVT_IDX], 1, NULL);
	/*wait CMD_ADD cmd result.*/
	WaitForSingleObject(ce.wait_evt, INFINITE);
	if (m->arr_cnt == MAXIMUM_WAIT_OBJECTS) {
		list_del(&m->entry);
		list_add_tail(&m->entry, &monitor->full_monitor_list);
	}
	process_monitor_unlock(monitor);
	CloseHandle(ce.wait_evt);
	WLog_DBG(TAG, "process monitor add: %d", ce.result);
	return ce.result;
}

TOOLKIT_API int process_monitor_remove(process_monitor_t *monitor, tk_process_t* process)
{
	cmd_entry_t ce;

	assert(monitor);
	assert(process);

	ce.wait_evt = CreateEventA(NULL, FALSE, FALSE, NULL);
	if (!ce.wait_evt)
		return -1;
	ce.cmd = CMD_REMOVE;
	ce.hproc = process->handle;
	ce.result = -1;

	process_monitor_lock(monitor);
	{
		monitor_t *pos;
		list_for_each_entry(pos, &monitor->full_monitor_list, monitor_t, entry) {
			list_add_tail(&ce.entry, &pos->cmd_list);
			ReleaseSemaphore(pos->arr_handle[EVT_IDX], 1, NULL);
			WaitForSingleObject(ce.wait_evt, INFINITE);
			if (ce.result == 0) {
				list_del(&pos->entry);
				list_add_tail(&pos->entry, &monitor->avail_monitor_list);
				break;
			}
		}
		if (ce.result == -1) {
			list_for_each_entry(pos, &monitor->avail_monitor_list, monitor_t, entry) {
				list_add_tail(&ce.entry, &pos->cmd_list);
				ReleaseSemaphore(pos->arr_handle[EVT_IDX], 1, NULL);
				WaitForSingleObject(ce.wait_evt, INFINITE);
				if (ce.result == 0) {
					break;
				}
			}
		}
	}
	process_monitor_unlock(monitor);
	CloseHandle(ce.wait_evt);
	WLog_DBG(TAG, "process monitor remove: %d", ce.result);
	return ce.result;
}

TOOLKIT_API int process_monitor_create(process_monitor_t **p_monitor)
{
	process_monitor_t *pm = MALLOC_T(process_monitor_t);
	INIT_LIST_HEAD(&pm->full_monitor_list);
	INIT_LIST_HEAD(&pm->avail_monitor_list);
	InitializeCriticalSection(&pm->lock);
	pm->cb = NULL;
	*p_monitor = pm;
	return 0;
}

TOOLKIT_API int process_monitor_destroy(process_monitor_t *monitor)
{
	int i;
	struct list_head *lists[] = {&monitor->avail_monitor_list, &monitor->full_monitor_list};
	for (i = 0; i < array_size(lists); ++i) {
		monitor_t *pos, *n;
		list_for_each_entry_safe(pos, n, lists[i], monitor_t, entry) {
			cmd_entry_t ce;
			ce.cmd = CMD_EXIT;
			ce.wait_evt = NULL;
			list_add_tail(&ce.entry, &pos->cmd_list);
			ReleaseSemaphore(pos->arr_handle[EVT_IDX], 1, NULL);
			WaitForSingleObject(pos->thread, INFINITE);
			list_del(&pos->entry);
			CloseHandle(pos->thread);
			CloseHandle(pos->arr_handle[EVT_IDX]);
			free(pos);
		}
	}
	DeleteCriticalSection(&monitor->lock);
	free(monitor);
	return 0;
}

TOOLKIT_API int process_monitor_set_cb(process_monitor_t *monitor, process_monitor_on_detect_process_end cb, void *user_data)
{
	monitor->cb = cb;
	monitor->user_data = user_data;
	return 0;
}

TOOLKIT_API int process_monitor_start(process_monitor_t *monitor)
{
	return 0;
}

TOOLKIT_API int process_monitor_stop(process_monitor_t *monitor)
{
	return 0;
}