path: root/src/aul_worker.c

/*
 * Copyright (c) 2019 Samsung Electronics Co., Ltd. All rights reserved.
 *
 * Licensed 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.
 */

#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/syscall.h>
#include <fcntl.h>
#include <glib.h>
#include <gio/gio.h>

#include "aul.h"
#include "aul_util.h"
#include "aul_sock.h"
#include "aul_worker.h"

#define DEFAULT_INTERVAL 5000

struct job_s {
	char *name;
	GIOChannel *channel;
	guint tag;
	void *callback;
	void *user_data;
};

struct anr_timer_s {
	guint tag;
	gint64 start_time;
	GQueue *cmd_queue;
};

struct aul_worker_s {
	char *name;
	GThread *thread;
	GMutex mutex;
	GCond cond;
	GMainContext *context;
	GMainLoop *loop;
	GList *jobs;
	struct anr_timer_s *anr_timer;
};

static void __destroy_anr_timer(struct anr_timer_s *timer)
{
	g_queue_free(timer->cmd_queue);
	free(timer);
}

static struct anr_timer_s *__create_anr_timer(void)
{
	struct anr_timer_s *timer;

	timer = calloc(1, sizeof(struct anr_timer_s));
	if (!timer) {
		_E("Out of memory");
		return NULL;
	}

	timer->cmd_queue = g_queue_new();
	if (!timer->cmd_queue) {
		_E("Failed to create GQueue");
		__destroy_anr_timer(timer);
		return NULL;
	}

	return timer;
}

static void __destroy_job(gpointer data)
{
	struct job_s *job = (struct job_s *)data;
	GSource *source;
	GMainContext *context;
	GError *error = NULL;

	if (job->channel) {
		g_io_channel_shutdown(job->channel, TRUE, &error);
		if (error) {
			_E("g_io_channel_shutdown() is failed. error(%s)",
					error->message);
			g_error_free(error);
		}
		g_io_channel_unref(job->channel);
	}

	if (job->tag) {
		context = g_main_context_get_thread_default();
		source = g_main_context_find_source_by_id(context, job->tag);
		_W("GMainContext(%p), GSource(%p)", context, source);
		if (source && !g_source_is_destroyed(source)) {
			_W("Destroy GSource(%p)", source);
			g_source_destroy(source);
		}
	}

	free(job->name);
	free(job);
}

static struct job_s *__create_job(const char *name,
		GIOChannel *channel,
		void *callback,
		void *user_data)
{
	struct job_s *job;

	job = calloc(1, sizeof(struct job_s));
	if (!job) {
		_E("Failed to create job");
		return NULL;
	}

	job->name = strdup(name);
	if (!job->name) {
		_E("Failed to duplicate job name");
		__destroy_job(job);
		return NULL;
	}

	job->channel = channel;
	job->callback = callback;
	job->user_data = user_data;

	return job;
}

static int __set_comm(const char *name)
{
	int fd;
	ssize_t bytes_written;
	char path[PATH_MAX];
	pid_t tid = syscall(__NR_gettid);

	_I("[%s] TID(%d)", name, tid);
	snprintf(path, sizeof(path), "/proc/%d/comm", tid);
	fd = open(path, O_WRONLY);
	if (fd < 0) {
		_E("Failed to open %s. error(%d)", path, errno);
		return -1;
	}

	bytes_written = write(fd, name, strlen(name) + 1);
	if (bytes_written < 0) {
		_E("Failed to write name(%s)", name);
		close(fd);
		return -1;
	}

	close(fd);

	return 0;
}

static GIOCondition __convert_aul_io_condition(int condition)
{
	GIOCondition cond = 0;

	if (condition & AUL_IO_IN)
		cond |= G_IO_IN;
	if (condition & AUL_IO_OUT)
		cond |= G_IO_OUT;
	if (condition & AUL_IO_PRI)
		cond |= G_IO_PRI;
	if (condition & AUL_IO_HUP)
		cond |= G_IO_HUP;
	if (condition & AUL_IO_ERR)
		cond |= G_IO_ERR;
	if (condition & AUL_IO_NVAL)
		cond |= G_IO_NVAL;

	return cond;
}

static int __convert_g_io_condition(GIOCondition condition)
{
	int cond = 0;

	if (condition & G_IO_IN)
		cond |= AUL_IO_IN;
	if (condition & G_IO_OUT)
		cond |= AUL_IO_OUT;
	if (condition & G_IO_PRI)
		cond |= AUL_IO_PRI;
	if (condition & G_IO_HUP)
		cond |= AUL_IO_HUP;
	if (condition & G_IO_ERR)
		cond |= AUL_IO_ERR;
	if (condition & G_IO_NVAL)
		cond |= AUL_IO_NVAL;

	return cond;
}

static gboolean __io_job_handler(GIOChannel *io, GIOCondition condition,
		gpointer data)
{
	int fd = g_io_channel_unix_get_fd(io);
	struct job_s *job = (struct job_s *)data;
	aul_worker_io_job_cb callback;
	GSource *source;
	int cond;

	source = g_main_current_source();
	if (!source || g_source_is_destroyed(source)) {
		_E("[__JOB__] GSource(%p) is destroyed", source);
		return G_SOURCE_REMOVE;
	}

	cond = __convert_g_io_condition(condition);
	callback = (aul_worker_io_job_cb)job->callback;
	if (callback(fd, cond, job->user_data))
		return G_SOURCE_CONTINUE;

	_I("[__JOB__] name(%s)", job->name);
	job->tag = 0;

	return G_SOURCE_REMOVE;
}

int aul_worker_add_io_job(aul_worker_h handle, const char *job_name,
		int fd, int condition, aul_worker_io_job_cb callback,
		void *user_data)
{
	GIOCondition cond = __convert_aul_io_condition(condition);
	struct aul_worker_s *worker = (struct aul_worker_s *)handle;
	struct job_s *job;
	GIOChannel *channel;
	GSource *source;

	if (!worker || !job_name || fd < 0 || !callback) {
		_E("Invalid parameter");
		return AUL_R_EINVAL;
	}

	channel = g_io_channel_unix_new(fd);
	if (!channel) {
		_E("Failed to create GIOChannel");
		return AUL_R_ENOMEM;
	}

	source = g_io_create_watch(channel, cond);
	if (!source) {
		_E("Failed to create GSource");
		g_io_channel_unref(channel);
		return AUL_R_ENOMEM;
	}

	job = __create_job(job_name, channel, callback, user_data);
	if (!job) {
		_E("Failed to create job(%s)", job_name);
		g_source_unref(source);
		g_io_channel_unref(channel);
		return AUL_R_ENOMEM;
	}

	g_source_set_callback(source, (GSourceFunc)__io_job_handler, job, NULL);
	g_source_set_priority(source, G_PRIORITY_DEFAULT);

	g_mutex_lock(&worker->mutex);
	worker->jobs = g_list_append(worker->jobs, job);
	job->tag = g_source_attach(source, worker->context);
	_W("GMainContext(%p), GSource(%p)", worker->context, source);
	g_mutex_unlock(&worker->mutex);

	g_source_unref(source);

	return AUL_R_OK;
}

void aul_worker_destroy(aul_worker_h handle)
{
	struct aul_worker_s *worker = (struct aul_worker_s *)handle;

	if (!worker)
		return;

	if (worker->thread) {
		if (g_main_loop_is_running(worker->loop))
			g_main_loop_quit(worker->loop);
		else
			_E("GMainLoop is not running");

		g_thread_join(worker->thread);
	}

	if (worker->loop)
		g_main_loop_unref(worker->loop);

	if (worker->context)
		g_main_context_unref(worker->context);

	g_cond_clear(&worker->cond);
	g_mutex_clear(&worker->mutex);

	__destroy_anr_timer(worker->anr_timer);
	free(worker->name);
	free(worker);
}

static gboolean __notify_cb(gpointer data)
{
	struct aul_worker_s *worker = (struct aul_worker_s *)data;

	_W("GMainLoop is started");
	g_mutex_lock(&worker->mutex);
	g_cond_signal(&worker->cond);
	g_mutex_unlock(&worker->mutex);

	return G_SOURCE_REMOVE;
}

static gpointer __worker_thread_loop(gpointer data)
{
	struct aul_worker_s *worker = (struct aul_worker_s *)data;
	struct anr_timer_s *anr_timer = worker->anr_timer;
	GSource *source;

	g_mutex_lock(&worker->mutex);
	__set_comm(worker->name);

	source = g_idle_source_new();
	if (!source) {
		_E("Failed to create GSource");
		g_cond_signal(&worker->cond);
		g_mutex_unlock(&worker->mutex);
		return NULL;
	}

	g_source_set_callback(source, (GSourceFunc)__notify_cb, worker, NULL);
	g_source_set_priority(source, G_PRIORITY_HIGH);
	g_source_attach(source, worker->context);
	g_source_unref(source);

	g_main_context_push_thread_default(worker->context);
	g_mutex_unlock(&worker->mutex);
	g_main_loop_run(worker->loop);

	g_mutex_lock(&worker->mutex);
	if (anr_timer->tag)
		g_source_remove(anr_timer->tag);

	g_list_free_full(worker->jobs, __destroy_job);
	g_main_context_pop_thread_default(worker->context);
	g_mutex_unlock(&worker->mutex);
	_W("Shut down worker");

	return NULL;
}

aul_worker_h aul_worker_create(const char *name)
{
	struct aul_worker_s *worker;

	if (!name) {
		_E("Invalid parameter");
		return NULL;
	}

	worker = calloc(1, sizeof(struct aul_worker_s));
	if (!worker) {
		_E("Out of memory");
		return NULL;
	}

	g_mutex_init(&worker->mutex);
	g_cond_init(&worker->cond);

	worker->name = strdup(name);
	if (!worker->name) {
		_E("Failed to duplicate name");
		aul_worker_destroy(worker);
		return NULL;
	}

	worker->anr_timer = __create_anr_timer();
	if (!worker->anr_timer) {
		_E("Failed to create ANR timer");
		aul_worker_destroy(worker);
		return NULL;
	}

	worker->context = g_main_context_new();
	if (!worker->context) {
		_E("Failed to create GMainContext");
		aul_worker_destroy(worker);
		return NULL;
	}
	_W("GMainContext(%p)", worker->context);

	worker->loop = g_main_loop_new(worker->context, FALSE);
	if (!worker->loop) {
		_E("Failed to create GMainLoop");
		aul_worker_destroy(worker);
		return NULL;
	}

	g_mutex_lock(&worker->mutex);

	worker->thread = g_thread_new(name, __worker_thread_loop, worker);
	if (!worker->thread) {
		_E("Failed to create worker thread");
		g_mutex_unlock(&worker->mutex);
		aul_worker_destroy(worker);
		return NULL;
	}

	while (!g_main_loop_is_running(worker->loop))
		g_cond_wait(&worker->cond, &worker->mutex);

	g_mutex_unlock(&worker->mutex);

	return worker;
}

static gboolean __timeout_cb(gpointer data)
{
	struct aul_worker_s *worker = data;
	struct anr_timer_s *anr_timer;
	char buf[12];
	bundle *b;
	int cmd;

	_E("Application Not Responding");
	g_mutex_lock(&worker->mutex);
	anr_timer = worker->anr_timer;
	anr_timer->tag = 0;
	anr_timer->start_time = 0;

	b = bundle_create();
	while (!g_queue_is_empty(anr_timer->cmd_queue)) {
		cmd = GPOINTER_TO_INT(g_queue_pop_head(anr_timer->cmd_queue));
		snprintf(buf, sizeof(buf), "%d", cmd);
		bundle_del(b, AUL_K_COMMAND);
		bundle_add_str(b, AUL_K_COMMAND, buf);
		aul_sock_send_bundle(AUL_UTIL_PID, getuid(), ANR_NOTIFY, b,
				AUL_SOCK_NOREPLY);
	}
	bundle_free(b);
	g_mutex_unlock(&worker->mutex);

	return G_SOURCE_REMOVE;
}

int aul_worker_add_anr_timer(aul_worker_h handle, int cmd)
{
	struct aul_worker_s *worker = handle;
	struct anr_timer_s *anr_timer;
	unsigned int elapsed_time;
	unsigned int interval;
	gint64 current_time;
	GSource *source;

	if (!worker) {
		_E("Invalid parameter");
		return AUL_R_EINVAL;
	}

	g_autoptr(GMutexLocker) locker = g_mutex_locker_new(&worker->mutex);

	anr_timer = worker->anr_timer;
	if (anr_timer->tag) {
		source = g_main_context_find_source_by_id(worker->context,
				anr_timer->tag);
		g_source_destroy(source);
		anr_timer->tag = 0;
	}

	g_queue_push_tail(anr_timer->cmd_queue, GINT_TO_POINTER(cmd));

	current_time = g_get_monotonic_time();
	if (anr_timer->start_time == 0) {
		interval = DEFAULT_INTERVAL;
		anr_timer->start_time = current_time;
	} else {
		elapsed_time = (current_time - anr_timer->start_time) / 1000;
		interval = DEFAULT_INTERVAL + (DEFAULT_INTERVAL - elapsed_time);
		anr_timer->start_time = current_time;
	}

	source = g_timeout_source_new(interval);
	g_source_set_callback(source, (GSourceFunc)__timeout_cb, worker, NULL);
	anr_timer->tag = g_source_attach(source, worker->context);
	g_source_unref(source);

	return AUL_R_OK;
}

int aul_worker_remove_anr_timer(aul_worker_h handle)
{
	struct aul_worker_s *worker = handle;
	struct anr_timer_s *anr_timer;
	GSource *source;

	if (!worker) {
		_E("Invalid parameter");
		return AUL_R_EINVAL;
	}

	g_autoptr(GMutexLocker) locker = g_mutex_locker_new(&worker->mutex);

	anr_timer = worker->anr_timer;
	if (g_queue_is_empty(anr_timer->cmd_queue)) {
		_E("Queue is empty");
		return AUL_R_ERROR;
	}

	g_queue_pop_head(anr_timer->cmd_queue);
	if (!g_queue_is_empty(anr_timer->cmd_queue))
		return AUL_R_OK;

	if (anr_timer->tag) {
		source = g_main_context_find_source_by_id(worker->context,
				anr_timer->tag);
		g_source_destroy(source);
		anr_timer->tag = 0;
		anr_timer->start_time = 0;
	}

	return AUL_R_OK;
}