path: root/server/src/muse_server_connection.c

/*
 * muse-server
 *
 * Copyright (c) 2017 Samsung Electronics Co., Ltd. All rights reserved.
 *
 * Contact: YoungHun Kim <yh8004.kim@samsung.com>
 *
 * 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.
 *
 */

#include "muse_server_private.h"

#define EPOLL_ERR -1

static const char *connection_cmd[API_MAX] = {
	"connect",
	"disconnect"
};

static void _ms_connection_module_instance_info(muse_module_h m, ms_connection_t *connection, GQueue *queue, int cmd);

static void _ms_connection_module_instance_info(muse_module_h m, ms_connection_t *connection, GQueue *queue, int cmd)
{
	muse_module_h connecting_m;
	char pids[MUSE_MSG_LEN_MAX] = {'\0',};
	char pid[MUSE_PARAM_MAX] = {'\0',};
	int idx, len, caution_instance;

	muse_return_if_fail(m);
	muse_return_if_fail(connection);
	muse_return_if_fail(queue);

	len = g_queue_get_length(queue);

	for (idx = 0; idx < len; idx++) {
		connecting_m = (muse_module_h)g_queue_peek_nth(queue, idx);
		if (!connecting_m) {
			LOGW("[%d] Make sure if the queue length is changed (%d = %d), which means that it was destroyed somewhere",
				idx, len, g_queue_get_length(queue));
			continue;
		}
		snprintf(pid, MUSE_PARAM_MAX, "%d ", connecting_m->pid);
		strncat(pids, pid, MUSE_MSG_LEN_MAX - strlen(pids) - 1);
	}

	LOGW("total number of module = %d ( %s) - %s %p from pid %d %s client (count %d)",
		len, pids, connection_cmd[cmd], m, m->pid, ms_config_get_host_name(m->idx), connection->instance_count[m->idx]);

	caution_instance = ms_config_get_caution_instance(m->idx);

	if (caution_instance != DEFAULT_CAUTION_INSTANCE && connection->instance_count[m->idx] > caution_instance) {
		LOGI("[MUSE_MODULE_COMMAND_CREATE_CAUTION] player # %d | camera # %d | recorder # %d > %d",
			connection->instance_count[MUSE_PLAYER], connection->instance_count[MUSE_CAMERA],
			connection->instance_count[MUSE_RECORDER], caution_instance);
		ms_cmd_dispatch(m, MUSE_MODULE_COMMAND_CREATE_CAUTION);
	}
}

int ms_connection_register(muse_module_h m)
{
	int fd;
	GQueue *queue;
	struct epoll_event event = { 0 };
	char err_msg[MUSE_MSG_LEN_MAX] = {'\0',};
	ms_connection_t *connection = NULL;

	LOGD("Enter");

	muse_return_val_if_fail(ms_get_instance(), MM_ERROR_UNKNOWN);

	connection = ms_get_instance()->connection;

	muse_return_val_if_fail(connection, MM_ERROR_INVALID_ARGUMENT);
	muse_return_val_if_fail(connection->instance_queue, MM_ERROR_INVALID_ARGUMENT);
	muse_return_val_if_fail(m, MM_ERROR_INVALID_ARGUMENT);

	ms_connection_lock(connection);

	queue = connection->instance_queue;

	fd = m->ch[MUSE_CHANNEL_MSG].sock_fd;

	event.events = EPOLLIN;
	event.data.ptr = GINT_TO_POINTER(fd);

	if (epoll_ctl(connection->epfd, EPOLL_CTL_ADD, fd, &event) == EPOLL_ERR) {
		strerror_r(errno, err_msg, MUSE_MSG_LEN_MAX);
		LOGE("epoll ctl error - %s", err_msg);
	}

	g_queue_push_tail(queue, (gpointer)m);

	connection->instance_count[m->idx]++;

	_ms_connection_module_instance_info(m, connection, queue, API_CREATE);

	ms_connection_unlock(connection);

	LOGD("Leave");

	return MM_ERROR_NONE;
}

int ms_connection_unregister(muse_module_h m)
{
	int fd;
	GQueue *queue;
	ms_connection_t *connection = NULL;

	LOGD("Enter");

	muse_return_val_if_fail(m, MM_ERROR_INVALID_ARGUMENT);
	muse_return_val_if_fail(ms_get_instance(), MM_ERROR_UNKNOWN);
	muse_return_val_if_fail(ms_is_server_ready(), MM_ERROR_UNKNOWN);

	connection = ms_get_instance()->connection;

	muse_return_val_if_fail(connection, MM_ERROR_INVALID_ARGUMENT);
	muse_return_val_if_fail(connection->instance_queue, MM_ERROR_INVALID_ARGUMENT);

	ms_connection_lock(connection);

	queue = connection->instance_queue;

	fd = m->ch[MUSE_CHANNEL_MSG].sock_fd;

	if (g_queue_remove(queue, (gpointer)m))
		connection->instance_count[m->idx]--;

	_ms_connection_module_instance_info(m, connection, queue, API_DESTROY);

	ms_connection_unlock(connection);

	LOGD("Leave");

	return MM_ERROR_NONE;
}

gboolean ms_connection_get_state(ms_connection_state_e *state)
{
	int idx, fd, ep_fd, fd_count, errsv;
	struct epoll_event *p_event;
	struct timeval tv_c, tv_s, tv_r;
	char err_msg[MUSE_MSG_LEN_MAX] = {'\0',};
	ms_connection_t *connection = NULL;

	muse_return_val_if_fail(ms_get_instance(), MM_ERROR_UNKNOWN);

	connection = ms_get_instance()->connection;

	muse_return_val_if_fail(connection, FALSE);

	ep_fd = connection->epfd;

	do { /* We have to execute epoll_wait again in case of error of EINTR  */
		gettimeofday(&tv_s, NULL);
		fd_count = epoll_wait(ep_fd, connection->events, MS_EVENT_MAX, MS_TIMEOUT_MSEC);
		if (fd_count < 0) {
			errsv = errno;
			strerror_r(errsv, err_msg, MUSE_MSG_LEN_MAX);
			LOGE("[%d : %d] errno : %d (%s)", ep_fd, muse_core_fd_is_valid(ep_fd), errsv, err_msg);
		}
		gettimeofday(&tv_c, NULL);
	} while (fd_count < 0 && errsv == EINTR);

	if (fd_count == 0) { /* There is no muse instance during timeout, which is not error case */
		timersub(&tv_c, &tv_s, &tv_r);
		if (tv_r.tv_sec != MS_TIMEOUT_SEC) {
			LOGW("sleep (10 ms) because epoll_wait does not wait for events during timeout, which would occur high cpu usage");
			usleep(10000); /* 10ms */
		}
		return FALSE;
	} else if (fd_count < 0) { /* When an error occurs, epoll_wait() returns -1 */
		strerror_r(errno, err_msg, MUSE_MSG_LEN_MAX);
		LOGE("[%d : %d] %s", ep_fd, muse_core_fd_is_valid(connection->epfd), err_msg);
		return FALSE;
	}

	for (idx = 0; idx < fd_count; idx++) {
		p_event = &connection->events[idx];
		fd = GPOINTER_TO_INT(p_event->data.ptr);

		if (p_event->events == EPOLLIN) {
			LOGI("CONNECTED [epoll fd : %d] [client fd : %d]", ep_fd, fd);

			p_event->events = EPOLLRDHUP;
			if (muse_core_fd_is_valid(fd) && epoll_ctl(ep_fd, EPOLL_CTL_MOD, fd, p_event) == EPOLL_ERR) {
				strerror_r(errno, err_msg, MUSE_MSG_LEN_MAX);
				LOGE("epoll ctl error - %s [%d] [fd : %d]", err_msg, errno, fd);
			}
			*state = MUSE_CONNECTION_STATE_CONNECTED;
		} else if (p_event->events & EPOLLRDHUP) {
			LOGI("DISCONNECTED [epoll fd : %d] [client fd : %d]", ep_fd, fd);

			if (muse_core_fd_is_valid(fd) && epoll_ctl(ep_fd, EPOLL_CTL_DEL, fd, p_event) == EPOLL_ERR) {
				strerror_r(errno, err_msg, MUSE_MSG_LEN_MAX);
				LOGE("epoll ctl error - %s [%d] [fd : %d]", err_msg, errno, fd);
			}
			*state = MUSE_CONNECTION_STATE_DISCONNECTED;
		}
	}

	return TRUE;
}

void ms_connection_lock(ms_connection_t *connection)
{
	muse_return_if_fail(connection);
	g_mutex_lock(&connection->lock);
}

void ms_connection_unlock(ms_connection_t *connection)
{
	muse_return_if_fail(connection);
	g_mutex_unlock(&connection->lock);
}

void ms_connection_deinit(ms_connection_t *connection)
{
	muse_return_if_fail(connection);

	g_queue_free(connection->instance_queue);
	connection->instance_queue = NULL;

	g_mutex_clear(&connection->lock);

	close(connection->epfd);

	free(connection);
}

void ms_connection_init(ms_connection_t *connection)
{
	muse_return_if_fail(connection);

	connection->epfd = epoll_create(MS_EVENT_MAX);
	muse_return_if_fail(muse_core_fd_is_valid(connection->epfd));

	connection->instance_queue = g_queue_new();
	muse_return_if_fail(connection->instance_queue);

	g_mutex_init(&connection->lock);
}