tango-tfe/plugin/protocol/http/src/http_entry.cpp

#include <MESA/MESA_list_queue.h>
#include <tfe_stream.h>
#include <tfe_utils.h>
#include <tfe_plugin.h>
#include <tfe_stream.h>
#include <http_parser.h>
#include <malloc.h>
#include <pthread.h>

#include <http_common.h>
#include <http_half.h>
#include <assert.h>
#include <event.h>
#include <tfe_proxy.h>

struct http_plugin __g_http_plugin;
struct http_plugin * g_http_plugin = &__g_http_plugin;

struct session_gc_cb_closure
{
#ifndef NDEBUG
	unsigned int __magic__;
#endif
	struct http_plugin * plugin;
	unsigned int thread_id;
};

static void http_plugin_session_gc_cb(evutil_socket_t fd, short what, void * arg)
{
	struct session_gc_cb_closure * closure = (struct session_gc_cb_closure *) arg;
	assert(closure->__magic__ == 0x8021);

	struct http_plugin * plugin_ctx = closure->plugin;
	struct hs_private_list * gc_list_hs_private = &plugin_ctx->gc_list_hs_private[closure->thread_id];

	struct http_session_private * hs_private_iter = NULL;
	struct http_session_private * hs_private_titer = NULL;

	TAILQ_FOREACH_SAFE(hs_private_iter, gc_list_hs_private, next, hs_private_titer)
	{
		assert(hs_private_iter->release_lock >= 0);
		if (hs_private_iter->release_lock > 0) continue;

		TAILQ_REMOVE(gc_list_hs_private, hs_private_iter, next);

		/* Call the http frame to raise END event */
		if (hs_private_iter->ht_frame)
		{
			struct tfe_http_session * hs_public = to_hs_public(hs_private_iter);
			http_frame_raise_session_end(hs_private_iter->ht_frame, NULL, hs_public, hs_private_iter->thread_id);
			hs_private_iter->ht_frame = NULL;
		}

		hs_private_destroy(hs_private_iter);
	}
}

#include <MESA/MESA_prof_load.h>
int http_plugin_config(struct tfe_proxy * proxy, struct http_plugin * ht_handle)
{
	MESA_load_profile_int_def("conf/tfe/tfe.conf", "http", "loglevel", &ht_handle->log_level, RLOG_LV_INFO);
	MESA_load_profile_string_def("conf/tfe/tfe.conf", "http", "logfile", ht_handle->log_file, sizeof(ht_handle->log_file), "log/http.log");
	return 0;
}

int http_plugin_init(struct tfe_proxy * proxy)
{
	unsigned int nr_work_thread = tfe_proxy_get_work_thread_count();
	struct http_plugin * plugin_ctx = g_http_plugin;

	http_plugin_config(proxy, plugin_ctx);

	for (unsigned int thread_id = 0; thread_id < nr_work_thread; thread_id++)
	{
#ifndef NDEBUG
		pthread_mutex_init(&plugin_ctx->lock_list_hs_private[thread_id], NULL);
#endif
		TAILQ_INIT(&plugin_ctx->gc_list_hs_private[thread_id]);
		struct event_base * ev_base = tfe_proxy_get_work_thread_evbase(thread_id);
		struct session_gc_cb_closure * closure = ALLOC(struct session_gc_cb_closure, 1);

#ifndef NDEBUG
		closure->__magic__ = 0x8021;
#endif
		closure->plugin = plugin_ctx;
		closure->thread_id = thread_id;

		/* TODO: Load GC delay from configure files */
		struct timeval gc_delay = {0, 500 * 1000};
		struct event * gc_event = event_new(ev_base, -1, EV_PERSIST, http_plugin_session_gc_cb, closure);

		if (unlikely(gc_event == NULL))
		{
			/* TODO: write a log */
			assert(0);
		}

		evtimer_add(gc_event, &gc_delay);
		plugin_ctx->gc_event_hs_private[thread_id] = gc_event;
	}

	plugin_ctx->logger = MESA_create_runtime_log_handle(plugin_ctx->log_file, plugin_ctx->log_level);
	assert(plugin_ctx->logger != NULL);

	return 0;
}

void http_plugin_deinit(struct tfe_proxy * proxy)
{
	return;
}

int http_connection_entry_open(const struct tfe_stream * stream, unsigned int thread_id,
	enum tfe_conn_dir dir, void ** pme)
{
	struct http_connection_private * ht_conn = ALLOC(
		struct http_connection_private, 1);
	TAILQ_INIT(&ht_conn->hs_private_list);
	TAILQ_INIT(&ht_conn->hs_private_orphan_list);
	ht_conn->stream = stream;
	*pme = (void *) ht_conn;
	return 0;
}

struct user_event_dispatch_closure
{
	const struct tfe_stream * stream;
	const struct tfe_http_session * session;
	unsigned int thread_id;
};

static int __user_event_dispatch(struct http_half_private * hf_private,
	enum tfe_http_event ev, const unsigned char * data, size_t len, void * user)
{
	struct user_event_dispatch_closure * __closure = (struct user_event_dispatch_closure *) user;
	struct http_frame_session_ctx * ht_frame = hf_private->session->ht_frame;
	//todo:
	http_frame_raise_event(ht_frame, __closure->stream, (struct tfe_http_session *) __closure->session,
		ev, data, len, __closure->thread_id);
	return 0;
}

static int __write_http_half(struct http_half_private * hf_private,
	const struct tfe_stream * stream, enum tfe_conn_dir dir)
{
	/* Construct, and write response immediately */
	hf_private_construct(hf_private);
	size_t __to_write_len = evbuffer_get_length(hf_private->evbuf_raw);
	unsigned char * __to_write = evbuffer_pullup(hf_private->evbuf_raw, __to_write_len);

	/* Write the data to stream, UPSTREAM is the incoming direction for response */
	int ret = 0;
	if (hf_private->write_ctx)
	{
		ret = tfe_stream_write_frag(hf_private->write_ctx, __to_write, __to_write_len);
	}
	else
	{
		ret = tfe_stream_write(stream, dir, __to_write, __to_write_len);
	}

	if (unlikely(ret < 0)) { assert(0); }
	evbuffer_drain(hf_private->evbuf_raw, __to_write_len);
	return ret;
}

static int __write_http_half_to_line(const struct tfe_stream * stream,
	enum tfe_conn_dir dir, struct http_half_private * hf_private)
{
	int ret = 0;

	if (hf_private->is_setup_by_stream)
	{
		/* By stream, first time to write http request/response,
		 * and has been call body_begin, construct the header */
		if (hf_private->evbuf_body != NULL && hf_private->write_ctx == NULL)
		{
//			printf("frag write start, stream = %p, hf_private = %p\n", stream, hf_private);

			/* Still need to send data(not call body_end()), need to write frag start
			 * Otherwise, means the body is ready, send out directly without frag */
			if (hf_private->message_status < STATUS_COMPLETE)
			{
				hf_private->write_ctx = tfe_stream_write_frag_start(stream, dir);
				assert(hf_private->write_ctx != NULL);
			}

			ret = __write_http_half(hf_private, stream, dir);
			if (unlikely(ret < 0)) return ret;
		}
	}
	else
	{
		/* Not stream, need to be complete, then construct all data and send out */
		if (hf_private->message_status == STATUS_COMPLETE)
		{
			ret = __write_http_half(hf_private, stream, dir);
			if (unlikely(ret < 0)) return ret;
		}
	}

	return 0;
}

static int __on_request_handle_user_req_or_resp(const tfe_stream * stream, struct http_session_private * hs_private,
	struct http_half_private * hf_private_req_in, bool & need_to_close_the_session)
{
	int ret = 0;

	/* Cannot setup user request and user response simultaneously */
	assert(!(hs_private->hf_private_req_user != NULL && hs_private->hf_private_resp_user != NULL));

	/* Only construct HTTP request or early answer response when incoming request is complete */
	if (hf_private_req_in->message_status != STATUS_COMPLETE)
	{
		return 0;
	}

	/* User's construct request */
	if (hs_private->hf_private_req_user != NULL)
	{
		struct http_half_private * hf_private_req_user = hs_private->hf_private_req_user;
		ret = __write_http_half_to_line(stream, CONN_DIR_UPSTREAM, hf_private_req_user);
		if (unlikely(ret < 0))
		{
			TFE_LOG_ERROR(g_http_plugin->logger, "Failed to write HTTP request setup by user. ");
			return ret;
		}

		if (hf_private_req_in->stream_action == ACTION_DROP_DATA ||
			hf_private_req_in->stream_action == ACTION_DEFER_DATA)
		{
			hf_private_req_in->stream_action = ACTION_DROP_DATA;
		}
	}

	/* User's construct response, send before real response arrive.
	 * Only early answer while the incoming request is send.
	 * Otherwise, we will construct user's response when real response arrived. */

	if (hs_private->hf_private_resp_user != NULL /* User response setup */
		&& hf_private_req_in->stream_action == ACTION_DROP_DATA /* Incoming request does not sent */)
	{
		struct http_half_private * hf_private_resp_user = hs_private->hf_private_resp_user;
		ret = __write_http_half_to_line(stream, CONN_DIR_DOWNSTREAM, hf_private_resp_user);

		if (unlikely(ret < 0))
		{
            TFE_LOG_ERROR(g_http_plugin->logger, "Failed to write HTTP request setup by user.");
			goto __errout;
		}

		need_to_close_the_session = true;
	}

	return 0;

__errout:
	return -1;
}

static int __on_response_handle_user_req_or_resp(const tfe_stream * stream, struct http_session_private * hs_private,
	struct http_half_private * hf_private_resp_in, bool & need_to_close_the_session)
{
	struct http_half_private * hf_private_resp_user = hs_private->hf_private_resp_user;
	if (hf_private_resp_user == NULL)
	{
		return 0;
	}

	int ret = __write_http_half_to_line(stream, CONN_DIR_DOWNSTREAM, hf_private_resp_user);
	if (unlikely(ret < 0))
	{
        TFE_LOG_ERROR(g_http_plugin->logger, "Failed to write HTTP response setup by user.");
		return -1;
	}

	if (hf_private_resp_in->stream_action == ACTION_DEFER_DATA)
	{
		hf_private_resp_in->stream_action = ACTION_DROP_DATA;
	}

	return 0;
}

static int __on_request_prepare_context(const struct tfe_stream * stream, unsigned int thread_id,
	struct http_connection_private * hc_private, struct http_session_private ** hs_private_out,
	struct http_half_private ** hf_private_out)
{
	struct http_session_private * hs_private = TAILQ_LAST(&hc_private->hs_private_list, hs_private_list);
	struct http_half_private * hf_private_req_in = to_hf_request_private(hs_private);

	if (hs_private == NULL || hf_private_req_in->message_status == STATUS_COMPLETE)
	{
		/* HTTP Request and Session */
		hf_private_req_in = hf_private_create(TFE_HTTP_REQUEST, 1, 0);
		hs_private = hs_private_create(hc_private, thread_id, hf_private_req_in, NULL);
		hf_private_set_session(hf_private_req_in, hs_private);

		/* Closure, catch stream, session and thread_id */
		struct user_event_dispatch_closure * __closure = ALLOC(struct user_event_dispatch_closure, 1);
		__closure->thread_id = thread_id;
		__closure->stream = stream;
		__closure->session = to_hs_public(hs_private);

		/* Set callback, this callback used to raise business event */
		hf_private_set_callback(hf_private_req_in, __user_event_dispatch, __closure, free);

		/* Call business plugin */
		hs_private->ht_frame = http_frame_alloc();
		assert(hs_private->ht_frame != NULL);

		http_frame_raise_session_begin(hs_private->ht_frame, stream, &hs_private->hs_public, thread_id);
		TAILQ_INSERT_TAIL(&hc_private->hs_private_list, hs_private, next);
	}

	*hs_private_out = hs_private;
	*hf_private_out = hf_private_req_in;
	return 0;
}

static int __on_response_prepare_context(const struct tfe_stream * stream, unsigned int thread_id,
	struct http_connection_private * hc_private, struct http_session_private ** hs_private_out,
	struct http_half_private ** hf_private_out)
{
	struct http_session_private * hs_private = TAILQ_FIRST(&hc_private->hs_private_list);

	/* Standalone response, it means missing something or malformed http protocol */
	if (hs_private == NULL)
	{
        TFE_LOG_ERROR(g_http_plugin->logger, "Standlone HTTP response emerged. Malformed HTTP Protocol, detached. ");
		return -1;
	}

	struct http_half_private * hf_private_req_in = to_hf_request_private(hs_private);

	/* The request is not completed, but response arrival.
	 * Something wrong happends to server, like 502 bad gateway */
	if(hf_private_req_in->message_status != STATUS_COMPLETE)
	{
		TFE_LOG_ERROR(g_http_plugin->logger, "Response arrival when request is not completed, detached. ");
		return -1;
	}

	struct http_half_private * hf_private_resp_in = to_hf_response_private(hs_private);
	/* First time parse http response */
	if (hf_private_resp_in == NULL)
	{
		/* HTTP Version */
		short resp_major = to_hf_request_private(hs_private)->major;
		short resp_minor = to_hf_request_private(hs_private)->minor;

		/* Response */
		hf_private_resp_in = hf_private_create(TFE_HTTP_RESPONSE, resp_major, resp_minor);
		hs_private_hf_private_set(hs_private, hf_private_resp_in, TFE_HTTP_RESPONSE);
		hf_private_set_session(hf_private_resp_in, hs_private);

		/* Closure, catch stream, session and thread_id */
		struct user_event_dispatch_closure * __closure = ALLOC(struct user_event_dispatch_closure, 1);
		__closure->thread_id = thread_id;
		__closure->stream = stream;
		__closure->session = to_hs_public(hs_private);

		/* Set callback, this callback used to raise business event */
		hf_private_set_callback(hf_private_resp_in, __user_event_dispatch, __closure, free);

		/* Setup user's action */
		if (hs_private->hf_private_resp_user != NULL)
		{
			hf_private_resp_in->is_user_stream_action_set = true;
			hf_private_resp_in->user_stream_action = ACTION_DROP_DATA;
		}
	}

	*hs_private_out = hs_private;
	*hf_private_out = hf_private_resp_in;

	return 0;
}

enum tfe_stream_action http_connection_entry(const struct tfe_stream * stream, enum tfe_conn_dir dir,
	struct http_connection_private * hc_private, unsigned int thread_id, const unsigned char * data, size_t len)
{
	struct http_session_private * hs_private = NULL;
	struct http_half_private * hf_private_in = NULL;

	bool __need_to_close_the_session = false;
	int ret = 0;

	enum tfe_stream_action __action = ACTION_FORWARD_DATA;
	size_t __action_args = 0;

	/* Prepare hs_private and hf_private_in */
	ret = (dir == CONN_DIR_DOWNSTREAM) ?
		  __on_request_prepare_context(stream, thread_id, hc_private, &hs_private, &hf_private_in) :
		  __on_response_prepare_context(stream, thread_id, hc_private, &hs_private, &hf_private_in);

	if (ret < 0)
	{
		goto __passthrough;
	}

	/* The session is suspended, and to resume */
	if (hs_private->suspend_tag_effective)
	{
		enum tfe_http_event __backup_event = hs_private->suspend_event;

		/* Clean up suspend tag, we can support user's call suspend in this callback */
		hs_private->suspend_event = (enum tfe_http_event) 0;
		hs_private->suspend_tag_effective = false;
		hs_private->release_lock--;
		hs_private->suspend_counter++;

		/* Retrieve the user's stream action, because the user may set request/response at resume's callback */
		if(hf_private_in->is_user_stream_action_set)
		{
			hf_private_in->stream_action = hf_private_in->user_stream_action;
		}
			
		/* Call user callback, tell user we resume from suspend */
		assert(hs_private->resume_tag_singal);
		hs_private->resume_tag_singal = false;
		hf_private_in->event_cb(hf_private_in, __backup_event, NULL, 0, hf_private_in->event_cb_user);
	}

	/* Parse the content, the data which in deferred state has been ignored. */
	ret = hf_private_parse(hf_private_in, data, len);

	/* Suspend, ask by user's callback */
	if (hs_private->suspend_tag_signal)
	{
		hs_private->suspend_tag_effective = true;
		hs_private->suspend_tag_signal = false;
		hs_private->release_lock++;
		hs_private->suspend_counter++;

		tfe_stream_suspend(stream, dir);
		return ACTION_DEFER_DATA;
	}

	if(hs_private->kill_signal)
	{
		tfe_stream_kill(stream);
		return ACTION_DROP_DATA;
	}

	if (hf_private_in->is_upgrade || hf_private_in->is_passthrough)
	{
		goto __passthrough;
	}

	/* Need more data, no boundary touched */
	if (ret == 0)
	{
		if (hf_private_in->stream_action == ACTION_DROP_DATA ||
			hf_private_in->stream_action == ACTION_FORWARD_DATA)
		{
			hf_private_in->parse_cursor = 0;
		}

		return hf_private_in->stream_action;
	}

	/* Some kind of error happened, write log and detach the stream */
	if (ret == -1)
	{
		TFE_LOG_ERROR(g_http_plugin->logger, "%s: Failed at parsing stream as HTTP, %u, %s, %s",
			stream->str_stream_info, hf_private_in->parse_errno, http_errno_name(hf_private_in->parse_errno),
			http_errno_description(hf_private_in->parse_errno));

		tfe_hexdump2file(stderr, "Failed at parsing stream as HTTP", data, (unsigned int)len);
		goto __passthrough;
	}

	ret = (dir == CONN_DIR_DOWNSTREAM) ?
		  __on_request_handle_user_req_or_resp(stream, hs_private, hf_private_in, __need_to_close_the_session) :
		  __on_response_handle_user_req_or_resp(stream, hs_private, hf_private_in, __need_to_close_the_session);

	if (ret < 0)
	{
		assert(0);
		goto __passthrough;
	}

	/* Touch a boundary, such as the end of HTTP headers, bodys, et al. */
	__action_args = hf_private_in->parse_cursor;
	hf_private_in->parse_cursor = 0;

	if (hf_private_in->stream_action == ACTION_FORWARD_DATA)
	{
		tfe_stream_action_set_opt(stream, ACTION_OPT_FOWARD_BYTES, &__action_args, sizeof(__action_args));
		__action = ACTION_FORWARD_DATA;
	}

	if (hf_private_in->stream_action == ACTION_DROP_DATA)
	{
		tfe_stream_action_set_opt(stream, ACTION_OPT_DROP_BYTES, &__action_args, sizeof(__action_args));
		__action = ACTION_DROP_DATA;
	}

	/* ON RESPONSE, and input message is complete, need to close the session */
	if (dir == CONN_DIR_UPSTREAM && hf_private_in->message_status == STATUS_COMPLETE)
	{
		__need_to_close_the_session = true;
	}

	/* There is nothing for this session, close the session */
	if (__need_to_close_the_session)
	{
		/* Try to close the session */
		if (hs_private_can_destroy(hs_private))
		{
			http_frame_raise_session_end(hs_private->ht_frame, stream, &hs_private->hs_public, thread_id);
			hs_private->ht_frame = NULL;
		}

		TAILQ_REMOVE(&hc_private->hs_private_list, hs_private, next);
		hs_private_gc_destroy(hs_private, &g_http_plugin->gc_list_hs_private[thread_id]);
	}

	return __action;

__passthrough:
	tfe_stream_detach(stream);
	return ACTION_FORWARD_DATA;
}

int __http_connection_identify(const struct tfe_stream * stream,
	struct http_connection_private * ht_conn, const unsigned char * data, size_t len)
{
	struct http_half_private * hf_private = hf_private_create(TFE_HTTP_REQUEST, 1, 0);
	int ret = hf_private_parse(hf_private, data, len);
	hf_private_destory(hf_private);
	return ret;
}

#define HTTP_INDENTIFY_LENGTH            4

enum tfe_stream_action http_connection_entry_data(const struct tfe_stream * stream, unsigned int thread_id,
	enum tfe_conn_dir dir, const unsigned char * data, size_t len, void ** pme)
{
	struct http_connection_private * ht_conn = (struct http_connection_private *) (*pme);

	if (ht_conn->is_preempted == 0)
	{
		/* If the server push response before client send request, this must not be HTTP, detach the stream */
		if (dir == CONN_DIR_UPSTREAM)
			goto __detach;

		/* Protocol Identification, we need 8 bytes at least to tell it is HTTP or not */
		if (len < HTTP_INDENTIFY_LENGTH) goto __detach;

		/* Now, we want to identify this stream */
		int ret = __http_connection_identify(stream, ht_conn, data, len);
		if (ret < 0) goto __detach;

		/* This is HTTP, try to preempt the stream
		 * It may be failed because other plugin preempted before us */
		ret = tfe_stream_preempt(stream);
		if (ret != 0) goto __detach;
		ht_conn->is_preempted = 1;
	}

	/* This stream has been preempt, this plugin try to parse it */
	return http_connection_entry(stream, dir, ht_conn, thread_id, data, len);

__detach:
	tfe_stream_detach(stream);
	return ACTION_FORWARD_DATA;
}

void http_connection_entry_close(const struct tfe_stream * stream, unsigned int thread_id,
	enum tfe_stream_close_reason reason, void ** pme)
{
	struct http_connection_private * ht_conn = (struct http_connection_private *) (*pme);
	struct http_plugin * plugin_ctx = g_http_plugin;

	struct http_session_private * hs_private_iter = NULL;
	struct http_session_private * hs_private_titer = NULL;

	TAILQ_FOREACH_SAFE(hs_private_iter, &ht_conn->hs_private_list, next, hs_private_titer)
	{
		TAILQ_REMOVE(&ht_conn->hs_private_list, hs_private_iter, next);

		/* Call the http frame to raise END event */
		if (hs_private_iter->ht_frame)
		{
			struct tfe_http_session * hs_public = to_hs_public(hs_private_iter);
			http_frame_raise_session_end(hs_private_iter->ht_frame, stream, hs_public, hs_private_iter->thread_id);
			hs_private_iter->ht_frame = NULL;
		}

		hs_private_gc_destroy(hs_private_iter, &plugin_ctx->gc_list_hs_private[thread_id]);
	}

	/* Clear session counter, and free ht_conn structure */
	ht_conn->session_id_counter = 0;
	free(ht_conn);

	*pme = NULL;
}

static struct tfe_plugin __http_plugin_info =
	{
		.symbol = "HTTP",
		.type = TFE_PLUGIN_TYPE_PROTOCOL,
		.on_init = http_plugin_init,
		.on_deinit = http_plugin_deinit,
		.on_open = http_connection_entry_open,
		.on_data = http_connection_entry_data,
		.on_close = http_connection_entry_close
	};

TFE_PLUGIN_REGISTER(HTTP, __http_plugin_info)