#include <string.h>
#include <assert.h>
#include <curl/curl.h>
#include <errno.h>
#include <poll.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/epoll.h>
#include <unistd.h>
#include <assert.h>
#include <pthread.h>

#define MAX_URLS 2
#define MAX_EVENTS 1000
#define MAX_THREADS 300


static int min(int a, int b) {
	return a<b?a:b;
}

// const int conn_timeout = 33;

struct thread_params {
	int th_id; // thread id
	int epoll_fd; // epoll fd
	int waiting_time; // timer time for next timeout
	int timeouts; // collect number of timeouts
	int iteration;
	int bytes_received;
	int time_left;// Total time out of 100 ms which is left
};

const char* urls[MAX_URLS] = {
	"https://127.0.0.1:8080/dsp?wt=50",
	"https://127.0.0.1:8080/dsp?wt=50"
};

int socket_callback(CURL* easy, curl_socket_t fd, int what, void* u, void* s)
{
	(void) easy;
	(void) s;
	struct thread_params* params = (struct thread_params*)u;
	int epollFd = params->epoll_fd;
	struct epoll_event event;
	event.events = 0;
	event.data.fd = fd;
	if (what == CURL_POLL_REMOVE) {
		int res = epoll_ctl(epollFd, EPOLL_CTL_DEL, fd, &event);
		(void) res;
		return 0;
	}
	if (what == CURL_POLL_IN || what == CURL_POLL_INOUT) {
		event.events |= EPOLLIN;
	}
	if (what == CURL_POLL_OUT || what == CURL_POLL_INOUT) {
		event.events |= EPOLLOUT;
	}

	if (event.events != 0) {
		int res = epoll_ctl(epollFd, EPOLL_CTL_ADD, fd, &event);
		if (res == -1)
			res = epoll_ctl(epollFd, EPOLL_CTL_MOD, fd, &event);
		if (res == -1)
			fprintf(stdout, "ERROR epoll_ctl(MOD)\n");
	} else {
		assert(0);
	}
	return 0;
}

int timer_callback(CURLM* multi, long timeout_ms, void* u)
{
	struct thread_params* params = (struct thread_params*)u;
	if (timeout_ms < 0) {
		return 0;
	}
	int retval;
	int running_handles;
	if (timeout_ms == 0) {
		retval = curl_multi_socket_action(multi, CURL_SOCKET_TIMEOUT, 0, &running_handles);
		if ( retval != CURLM_OK) {
			fprintf(stdout, "ERROR curl_multi_socket_action() return error = %d %s:%d\n", retval,__FILE__, __LINE__);
			assert(0);
		}
	} else {
		params->waiting_time = timeout_ms;
	}
	return 0;
}

int response_writer(void *ptr, size_t size, size_t nmemb, void* userdata) {
	(void) ptr;
	int len = size * nmemb;
	struct thread_params* params = (struct thread_params*)userdata;
	params->bytes_received +=len;
	return len;	
}

/*Note: POST request are not tested yet */
int set_curl_common_option(CURL *easy, void* resp_buff, int req_timeout) {
	(void) req_timeout;
	CURLcode rc_easy;
	rc_easy = curl_easy_setopt(easy, CURLOPT_HTTPGET, 1L);
	if (rc_easy != CURLE_OK) {
		fprintf(stdout,"\nERROR curl_easy_setopt() failed : %s %s:%d\n", curl_easy_strerror(rc_easy), __FILE__, __LINE__);
	}
	//
	rc_easy = curl_easy_setopt(easy, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2TLS);
	if (rc_easy != CURLE_OK) {
		fprintf(stdout,"\nERROR curl_easy_setopt() failed : %s %s:%d\n", curl_easy_strerror(rc_easy), __FILE__, __LINE__);
	}
	curl_easy_setopt(easy, CURLOPT_SSL_VERIFYPEER, 0L);
	curl_easy_setopt(easy, CURLOPT_SSL_VERIFYHOST, 0L);
	curl_easy_setopt(easy, CURLOPT_PIPEWAIT, 1L);
	//
	rc_easy = curl_easy_setopt(easy, CURLOPT_WRITEFUNCTION, response_writer);
	if (rc_easy != CURLE_OK) {
		fprintf(stdout,"\nERROR curl_easy_setopt() failed : %s %s:%d\n", curl_easy_strerror(rc_easy), __FILE__, __LINE__);
	}
	rc_easy = curl_easy_setopt(easy, CURLOPT_WRITEDATA, (void*)resp_buff);
	if (rc_easy != CURLE_OK) {
		fprintf(stdout,"\nERROR curl_easy_setopt() failed : %s %s:%d\n", curl_easy_strerror(rc_easy), __FILE__, __LINE__);
	}
	rc_easy = curl_easy_setopt(easy, CURLOPT_NOSIGNAL, 1);
	if (rc_easy != CURLE_OK) {
		fprintf(stdout,"\nERROR curl_easy_setopt() failed : %s %s:%d\n", curl_easy_strerror(rc_easy), __FILE__, __LINE__);
	}
	rc_easy = curl_easy_setopt(easy, CURLOPT_HEADERFUNCTION, NULL);
	if (rc_easy != CURLE_OK) {
		fprintf(stdout,"\nERROR curl_easy_setopt() failed : %s %s:%d\n", curl_easy_strerror(rc_easy), __FILE__, __LINE__);
	}
	rc_easy = curl_easy_setopt(easy, CURLOPT_HEADERDATA, NULL);
	if (rc_easy != CURLE_OK) {
		fprintf(stdout,"\nERROR curl_easy_setopt() failed : %s %s:%d\n", curl_easy_strerror(rc_easy), __FILE__, __LINE__);
	}
	rc_easy = curl_easy_setopt(easy, CURLOPT_VERBOSE, 1);
	if (rc_easy != CURLE_OK) {
		fprintf(stdout,"\nERROR curl_easy_setopt() failed : %s %s:%d\n", curl_easy_strerror(rc_easy), __FILE__, __LINE__);
	}
	return rc_easy;
}

int execute_epoll(CURLM* multi, struct thread_params* params) {
	int th_id = params->th_id;
	int timeout_counter = 0;
	CURLMcode retval;

	int running_handles = 100;
	int it = 0;
	int event_count;
	struct timespec t0;
	struct timespec t1;
	long long t0_ms;
	long long t1_ms;
	long long t_diff;

	retval = curl_multi_socket_action(multi, CURL_SOCKET_TIMEOUT, 0, &running_handles);
	if ( retval != CURLM_OK) {
		fprintf(stdout, "%d:ERROR curl_multi_socket_action() return error = %d %s:%d\n", th_id, retval,__FILE__, __LINE__);
		return 1;
	}
	fprintf(stdout, "# of running_handles = %d\n", running_handles);

	int ep_wait_time;
	while (running_handles > 0 &&  params->time_left > 0) {
		struct epoll_event events[MAX_EVENTS];
		clock_gettime(CLOCK_MONOTONIC_RAW, &t0);
		ep_wait_time = min(params->waiting_time, params->time_left);
		fprintf(stdout, "ep_wait_time = %d\n", ep_wait_time);
		event_count = epoll_wait(params->epoll_fd, events, MAX_EVENTS, ep_wait_time);
		clock_gettime(CLOCK_MONOTONIC_RAW, &t1);
		t0_ms = (t0.tv_sec * 1000 + (t0.tv_nsec / 1000000));
		t1_ms = (t1.tv_sec * 1000 + (t1.tv_nsec / 1000000));
		t_diff = t1_ms - t0_ms;
		fprintf(stdout, "t_diff = %lld\n", t_diff);

		if (event_count == -1) {
			fprintf(stdout, "%d:ERROR epoll_wait:%d|%s\n", th_id, errno, strerror(errno));
			assert(0);
		}
		else if (event_count == 0) {
			params->time_left -= ep_wait_time;
			retval = curl_multi_socket_action(multi, CURL_SOCKET_TIMEOUT, 0, &running_handles);
			if ( retval != CURLM_OK) {
				fprintf(stdout, "%d:ERROR curl_multi_socket_action() return error = %d %s:%d\n", th_id, retval,__FILE__, __LINE__);
				assert(0);
				break;
			}
		}
		else {
			for(it = 0; it < event_count; it++) {
				int ev_bitmask = 0;
				switch (events[it].events) {
					case EPOLLIN:
						ev_bitmask |= CURL_CSELECT_IN;
						break;
					case EPOLLOUT:
						ev_bitmask |= CURL_CSELECT_OUT;
						break;
					case EPOLLERR:
						ev_bitmask |= CURL_CSELECT_ERR;
						break;
					default:
						// More than one I/O event is detected on this fd, let curl figure out on its own what set of events has happened
						ev_bitmask = 0;
				} 
				retval = curl_multi_socket_action(multi, events[it].data.fd, ev_bitmask, &running_handles);
				if ( retval != CURLM_OK) {
					fprintf(stdout, "%d:ERROR curl_multi_socket_action() return error = %d %s:%d\n", th_id, retval,__FILE__, __LINE__);
					assert(0);
					break;
				}
			}
			params->time_left -= t_diff; 
		}

	}
	if (running_handles > 0) {
		timeout_counter += running_handles;
	}
	return timeout_counter;
}

int execute_select(CURLM* multi, struct thread_params* params) {
	int th_id = params->th_id;
	(void) th_id;
	int timeout_counter = 0;
	CURLMcode retval = CURLM_OK;
	int still_running = 0;
	int rc; /* select() return code */
	struct timeval timeout_tv;
	fd_set fdread;
	fd_set fdwrite;
	fd_set fdexcep;

	int maxfd;
	timeout_tv.tv_sec =  params->time_left/1000;
	timeout_tv.tv_usec = (params->time_left%1000)*1000;

	while((retval = curl_multi_perform(multi, &still_running)) == CURLM_CALL_MULTI_PERFORM);
	if ( retval != CURLM_OK) {
		fprintf(stdout, "\nERROR curl_multi_perform() return error = %d %s:%d\n", retval,__FILE__, __LINE__);
		return 1;
	}

	while(still_running) {
		FD_ZERO(&fdread);
		FD_ZERO(&fdwrite);
		FD_ZERO(&fdexcep);

		retval = curl_multi_fdset(multi, &fdread, &fdwrite, &fdexcep, &maxfd);
		if(retval != CURLM_OK) {
			fprintf(stdout, "\nERROR curl_multi_fdset() return error = %d %s:%d\n", retval,  __FILE__, __LINE__);
			return 1;
		}

		if(maxfd == -1) {
			fprintf(stdout, "\nERROR curl_multi_perform maxfd = -1 %s:%d\n", __FILE__, __LINE__);
			fprintf(stdout, "INFO: running select() implementation\n");
			return 1;
		}else {
			if(timeout_tv.tv_sec == 0 && timeout_tv.tv_usec == 0) {
				fprintf(stdout, "\nERROR tv_sec = 0 and tv_usec = 0 %s:%d\n", __FILE__, __LINE__);
				still_running = 0;
				rc = 0;
				break;
			}else {
				errno = 0;
				rc = select(maxfd+1, &fdread, &fdwrite, &fdexcep, &timeout_tv);
			}
		}

		switch(rc) {
			case -1:
				fprintf(stdout,"\nERROR:RTB select returned error %s:%d\n",__FILE__,__LINE__);
				perror("\nERROR SELECT RTB");
				still_running = 0;
				break;
			case 0:
				still_running = 0;
				timeout_counter++;
				break;
			default:
				while((retval=curl_multi_perform(multi, &still_running)) == CURLM_CALL_MULTI_PERFORM );
				if (retval != CURLM_OK) {
					fprintf(stdout, "\nERROR curl_multi_perform() return error = %d %s:%d\n", retval, __FILE__, __LINE__);
					return 1;
				}
				break;
		}
	}

	return timeout_counter;
}

void* do_task(void* ptr) {
	struct thread_params* params = (struct thread_params*) ptr;
	int it, idx;
	CURLMcode retval = CURLM_OK;
	CURL* easy[MAX_URLS];
	// char resp_buff[MAX_URLS][10240];
	CURLM* multi = curl_multi_init();

	retval = curl_multi_setopt(multi, CURLMOPT_MAXCONNECTS, 100);
	if ( retval != CURLM_OK) {
		fprintf(stdout,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(retval), __FILE__, __LINE__);
	}
	retval = curl_multi_setopt(multi, CURLMOPT_PIPELINING, CURLPIPE_MULTIPLEX);
	if ( retval != CURLM_OK) {
		fprintf(stdout,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(retval), __FILE__, __LINE__);
	}

	int req_timeout = params->time_left;

	for(idx = 0; idx < MAX_URLS; idx++) {
		easy[idx] = curl_easy_init();
	}

	for(it = 0; it < params->iteration; it++) {
		params->time_left = req_timeout;
		/* SET socket callback function */
		retval = curl_multi_setopt(multi, CURLMOPT_SOCKETFUNCTION, socket_callback);
		if ( retval != CURLM_OK) {
			fprintf(stdout,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(retval), __FILE__, __LINE__);
			return NULL;
		}

		/* SET socket callback function parameters*/
		retval = curl_multi_setopt(multi, CURLMOPT_SOCKETDATA, ptr);
		if ( retval != CURLM_OK) {
			fprintf(stdout,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(retval), __FILE__, __LINE__);
			return NULL;
		}

		/* SET timmer function */	
		retval = curl_multi_setopt(multi, CURLMOPT_TIMERFUNCTION, timer_callback);
		if ( retval != CURLM_OK) {
			fprintf(stdout,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(retval), __FILE__, __LINE__);
			return NULL;
		}

		/* SET timmer function parameters */	
		retval = curl_multi_setopt(multi, CURLMOPT_TIMERDATA, ptr);
		if ( retval != CURLM_OK) {
			fprintf(stdout,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(retval), __FILE__, __LINE__);
			return NULL;
		}

		for(idx = 0; idx < MAX_URLS; idx++) {
			int url_idx = idx % MAX_URLS;
			CURLcode rv = curl_easy_setopt(easy[idx], CURLOPT_URL, urls[url_idx]);
			if ( rv != CURLE_OK) {
				fprintf(stdout, "\nERROR curl_multi_perform() return error = %d %s:%d\n", retval,__FILE__, __LINE__);
				return NULL;
			}
			// set curl options
			//memset((void*)resp_buff[idx], 0, sizeof(resp_buff[idx]));
			set_curl_common_option(easy[idx], params, req_timeout);
			retval = curl_multi_add_handle(multi, easy[idx]);
			if ( retval != CURLM_OK) {
				fprintf(stdout, "\nERROR curl_multi_perform() return error = %d %s:%d\n", retval,__FILE__, __LINE__);
				return NULL;
			}
		}

		params->timeouts += execute_epoll(multi, params);

		for(idx = 0; idx < MAX_URLS; idx++) {
			retval = curl_multi_remove_handle(multi, easy[idx]);
			if (retval != CURLM_OK) {
				fprintf(stdout, "\nERROR curl_multi_remove_handle RTB failed with rc = %d %s:%d\n", retval, __FILE__, __LINE__);
				return NULL;
			}
		}
		fprintf(stdout, "xxxxxxxxxxxxxxxxxxxxxxxxxxxx All handles removed ***********\n");
	}

	// close the multi handle
	curl_multi_cleanup(multi);
	return NULL;
}

/* 
 * 1. argv[1] number of threads needs to be created 
 * 2. argv[2] number of iteration per thread
 * 3. argv[3] "timeout" for server to respond in milliseconds
 */
int main(int argc, char** argv)
{
	if(4 != argc) {
		fprintf(stdout, "\nUsage: thread_count thread_iteration, req_timeout_ms\n");
		return 1;
	}
	struct timespec t0;
	struct timespec t1;
	double t0_ms;
	double t1_ms;
	double t_diff;
	int idx = 0;
	int ret = 0;
	struct thread_params* params = NULL;

	int nthreads = atoi(argv[1]);
	int iteration = atoi(argv[2]);
	int req_timeout = atoi(argv[3]);


	curl_global_init(CURL_GLOBAL_ALL);
	printf("Using %s\n", curl_version());

	params = (struct thread_params *) malloc(sizeof(struct thread_params) * nthreads);
	if (NULL == params) {
		fprintf(stdout, "\nERROR malloc() failed %s:%d\n", __FILE__, __LINE__);
		return 1;
	}

	clock_gettime(CLOCK_MONOTONIC_RAW, &t0);
	// Thread start
	pthread_t tid[MAX_THREADS];
	for (idx = 0; idx < nthreads; idx++) {
		params[idx].th_id = idx;
		params[idx].time_left = req_timeout;
		params[idx].timeouts = 0;
		params[idx].iteration = iteration;
		params[idx].bytes_received = 0;
		// create epoll_fd for thread;
		params[idx].epoll_fd = epoll_create(50);
		if (params[idx].epoll_fd == -1) {
			fprintf(stdout, "ERROR epoll_create");
			return 1;
		}

		ret = pthread_create(&tid[idx], NULL, do_task, (void*) (&params[idx]));
		if (ret != 0) {
			fprintf(stdout, "\nERROR pthread_create() ret = %d %s:%d\n", ret, __FILE__, __LINE__);
			return 1;
		}
	}

	// Thread wait
	for (idx = 0; idx < nthreads; idx++) {
		ret = pthread_join(tid[idx], NULL);
		if (ret != 0) {
			fprintf(stdout, "\nERROR pthread_join() ret = %d %s:%d\n", ret, __FILE__, __LINE__);
			return 1;
		}
	}
	curl_global_cleanup();
	// cleanup all epoll_fds
	for(idx = 0; idx < nthreads; idx++) {
		close(params[idx].epoll_fd);
	}
	printf(">>> bye bye\n");

	clock_gettime(CLOCK_MONOTONIC_RAW, &t1);
	t0_ms = (t0.tv_sec * 1000.0 + (t0.tv_nsec / 1000000.0));
	t1_ms = (t1.tv_sec * 1000.0 + (t1.tv_nsec / 1000000.0));
	t_diff = t1_ms - t0_ms;
	curl_global_cleanup();

	int total_timeouts = 0;
	int total_bytes_received = 0;
	for(idx = 0; idx < nthreads; idx++) {
		total_timeouts += params[idx].timeouts;
		total_bytes_received  += params[idx].bytes_received;
	}
	fprintf(stdout, "\nMain thread returning after time %lf ms and total timeouts = %d, data_received = %d bytes\n", t_diff, total_timeouts, total_bytes_received);
	return 0;
}