#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 <gperftools/profiler.h>

#define DEFAULT_EPOLL_WAIT_TIME 5
#define MAX_URLS 50
#define MAX_EVENTS MAX_URLS
#define MAX_THREADS 300

const int conn_timeout = 30;
int run_epoll = 1;
int iteration = 1;

struct thread_params {
	int th_id; // thread id
	int epoll_fd; // epoll fd
	int waiting_time; // waiting time for epoll_wait()
	int timeouts; // collect number of timeouts
};

char* urls[MAX_URLS] = {
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50",
	"http://172.16.4.81:101/dsp?wt=50"
};


void error(const char* string)
{
	perror(string);
	exit(1);
}

int socket_callback(CURL* easy, curl_socket_t fd, int run_epoll, void* u, 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;
//	fprintf(stderr, "socket_callback\n");
	if (run_epoll == CURL_POLL_REMOVE) {
#ifdef DEBUG
		fprintf(stderr, ">>> %s: removing fd=%d\n", __func__, fd);
#endif

		int res = epoll_ctl(epollFd, EPOLL_CTL_DEL, fd, &event);
		if (res == -1 && errno != EBADF)
			fprintf(stderr, "ERROR epoll_ctl(DEL)");
		return 0;
	}

	if (run_epoll == CURL_POLL_IN || run_epoll == CURL_POLL_INOUT) {
		event.events |= EPOLLIN;
	}
	if (run_epoll == CURL_POLL_OUT || run_epoll == CURL_POLL_INOUT) {
		event.events |= EPOLLOUT;
	}

#ifdef DEBUG
	fprintf(stderr, ">>> %s: adding fd=%d run_epoll=%d\n", __func__, fd, run_epoll);
#endif
	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(stderr, "ERROR epoll_ctl(MOD)");
	}

	return 0;
}

// TODO: use provate timmer
int timer_callback(CURLM* multi, long timeout_ms, void* u)
{
	struct thread_params* params = (struct thread_params*)u;
	fprintf(stderr, "timer_callback:ThreadId:%d|timeout_ms:%ld\n", params->th_id, timeout_ms);
	return 0;
}

int response_writter(void *ptr, size_t size, size_t nmemb, void* userdata) {
	fprintf(stderr, "\nIn response_writter\n");
	int i = 0;
	int len = size * nmemb;
	char* cptr = (char*) ptr;

//#if DEBUG
	fprintf(stderr, "\n");
	for (i = 0; i < len; i++) {
		fprintf(stderr, "%c", cptr[i]);
	}
	fprintf(stderr, "\n");
//#endif
//	memcpy(userdata, ptr, len);
	return len;	
}

/*Note: POST request are not tested yet */
int set_curl_common_option(CURL *easy, char* resp_buff, int req_timeout) {
	int rc_multi = 0;
	rc_multi = curl_easy_setopt(easy, CURLOPT_HTTPGET, 1L);
	if (rc_multi != CURLM_OK) {
		fprintf(stderr,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(rc_multi), __FILE__, __LINE__);
	}
	rc_multi = curl_easy_setopt(easy, CURLOPT_WRITEFUNCTION, response_writter);
	if (rc_multi != CURLM_OK) {
		fprintf(stderr,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(rc_multi), __FILE__, __LINE__);
	}
	rc_multi = curl_easy_setopt(easy, CURLOPT_WRITEDATA, (void*)resp_buff);
	if (rc_multi != CURLM_OK) {
		fprintf(stderr,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(rc_multi), __FILE__, __LINE__);
	}
	rc_multi = curl_easy_setopt(easy, CURLOPT_CONNECTTIMEOUT_MS, conn_timeout);
	if (rc_multi != CURLM_OK) {
		fprintf(stderr,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(rc_multi), __FILE__, __LINE__);
	}
	rc_multi = curl_easy_setopt(easy, CURLOPT_TIMEOUT_MS, req_timeout);
	if (rc_multi != CURLM_OK) {
		fprintf(stderr,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(rc_multi), __FILE__, __LINE__);
	}
	rc_multi = curl_easy_setopt(easy, CURLOPT_NOSIGNAL, 1);
	if (rc_multi != CURLM_OK) {
		fprintf(stderr,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(rc_multi), __FILE__, __LINE__);
	}
	rc_multi = curl_easy_setopt(easy, CURLOPT_HEADERFUNCTION, NULL);
	if (rc_multi != CURLM_OK) {
		fprintf(stderr,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(rc_multi), __FILE__, __LINE__);
	}
	rc_multi = curl_easy_setopt(easy, CURLOPT_HEADERDATA, NULL);
	if (rc_multi != CURLM_OK) {
		fprintf(stderr,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(rc_multi), __FILE__, __LINE__);
	}
#ifdef DEBUG
	rc_multi = curl_easy_setopt(easy, CURLOPT_VERBOSE, 1);
	if (rc_multi != CURLM_OK) {
		fprintf(stderr,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(rc_multi), __FILE__, __LINE__);
	}
#endif
	/*
	rc_multi = curl_easy_setopt(easy, CURLOPT_FOLLOWLOCATION, 1);
	if (rc_multi != CURLM_OK) {
		fprintf(stderr,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(rc_multi), __FILE__, __LINE__);
	}
	*/

	return rc_multi;
}

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

	struct epoll_event events[MAX_EVENTS];
	fprintf(stderr, "INFO: running epoll() implementation\n");
	int running_handles = 1;
	int it = 0;
	int ev_bitmask = 0;
	int event_count;
	struct timespec t0;
        struct timespec t1;
        double t0_ms;
        double t1_ms;
        double t_diff;

	while((retval = curl_multi_perform(multi, &running_handles)) == CURLM_CALL_MULTI_PERFORM);
	retval = curl_multi_socket_action(multi, CURL_SOCKET_TIMEOUT, ev_bitmask, &running_handles);
	if ( retval != CURLM_OK) {
		fprintf(stderr, "%d:ERROR curl_multi_socket_action() return error = %d %s:%d\n", th_id, retval,__FILE__, __LINE__);
		return 1;
	}
	
	while (running_handles > 0 &&  params->waiting_time > 0) {
#ifdef DEBUG
		fprintf(stderr, ">>> calling epoll_wait\n");
#endif
		clock_gettime(CLOCK_MONOTONIC_RAW, &t0);
		event_count = epoll_wait(params->epoll_fd, events, MAX_EVENTS, params->waiting_time);
		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;
		params->waiting_time -= t_diff; 
		fprintf(stderr, "%d: epoll_wait: running_handles:%d event_counts:%d | params->waiting_time:%d\n", th_id, running_handles, event_count, params->waiting_time);
		
		if (event_count == -1) {
			fprintf(stderr, "%d:ERROR epoll_wait:%d|%s\n", th_id, errno, strerror(errno));
		}
		else if (event_count == 0) {
			retval = curl_multi_socket_action(multi, CURL_SOCKET_TIMEOUT, ev_bitmask, &running_handles);
			if ( retval != CURLM_OK) {
				fprintf(stderr, "%d:ERROR curl_multi_socket_action() return error = %d %s:%d\n", th_id, retval,__FILE__, __LINE__);
				break;
			}
		}
		else {
			for(it = 0; it < event_count; it++) {
				retval = curl_multi_socket_action(multi, events[it].data.fd, ev_bitmask, &running_handles);
				if ( retval != CURLM_OK) {
					fprintf(stderr, "%d:ERROR curl_multi_socket_action() return error = %d %s:%d\n", th_id, retval,__FILE__, __LINE__);
					break;
				}
			}
		}

		if (event_count == 0 || (running_handles > 0 && params->waiting_time <= 0)) {
			timeout_counter++;
		}
	}

	fprintf(stderr, "\nwhile end: running_handles:%d|params->waiting_time:%d\n", running_handles, params->waiting_time);

	return timeout_counter;
}

int execute_select(CURLM* multi, struct thread_params* params) {
	int th_id = params->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->waiting_time/1000;
	timeout_tv.tv_usec = (params->waiting_time%1000)*1000;

	while((retval = curl_multi_perform(multi, &still_running)) == CURLM_CALL_MULTI_PERFORM);
	if ( retval != CURLM_OK) {
		fprintf(stderr, "\nERROR curl_multi_perform() return error = %d %s:%d\n", retval,__FILE__, __LINE__);
		exit(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(stderr, "\nERROR curl_multi_fdset() return error = %d %s:%d\n", retval,  __FILE__, __LINE__);
			exit(1);
		}

		if(maxfd == -1) {
			fprintf(stderr, "\nERROR curl_multi_perform maxfd = -1 %s:%d\n", __FILE__, __LINE__);
			fprintf(stderr, "INFO: running select() implementation\n");
			exit(1);
		}else {
			if(timeout_tv.tv_sec == 0 && timeout_tv.tv_usec == 0) {
				fprintf(stderr, "\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(stderr,"\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(stderr, "\nERROR curl_multi_perform() return error = %d %s:%d\n", retval, __FILE__, __LINE__);
					exit(1);
				}
				break;
		}
	}

	return timeout_counter;
}

void* do_task(void* ptr) {
	struct thread_params* params = (struct thread_params*) ptr;
	int it, idx;
	int timeout_counter = 0;
	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(stderr,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(retval), __FILE__, __LINE__);
	}

	int req_timeout = params->waiting_time;

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

	for(it = 0; it < iteration; it++) {
		params->waiting_time = req_timeout;
		if(1 == run_epoll == 1) {
			/* SET socket callback function */
			retval = curl_multi_setopt(multi, CURLMOPT_SOCKETFUNCTION, socket_callback);
			if ( retval != CURLM_OK) {
				fprintf(stderr,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(retval), __FILE__, __LINE__);
				exit(1);
			}
			
			/* SET socket callback function parameters*/
			retval = curl_multi_setopt(multi, CURLMOPT_SOCKETDATA, ptr);
			if ( retval != CURLM_OK) {
				fprintf(stderr,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(retval), __FILE__, __LINE__);
				exit(1);
			}
		
			/* SET timmer function */	
			retval = curl_multi_setopt(multi, CURLMOPT_TIMERFUNCTION, timer_callback);
			if ( retval != CURLM_OK) {
				fprintf(stderr,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(retval), __FILE__, __LINE__);
				exit(1);
			}

			/* SET timmer function parameters */	
			retval = curl_multi_setopt(multi, CURLMOPT_TIMERDATA, ptr);
			if ( retval != CURLM_OK) {
				fprintf(stderr,"\nERROR curl_multi_setopt() failed : %s %s:%d\n", curl_multi_strerror(retval), __FILE__, __LINE__);
				exit(1);
			}
		}
		
		for(idx = 0; idx < MAX_URLS; idx++) {
			int url_idx = idx % MAX_URLS;
			retval = curl_easy_setopt(easy[idx], CURLOPT_URL, urls[url_idx]);
			if ( retval != CURLM_OK) {
				fprintf(stderr, "\nERROR curl_multi_perform() return error = %d %s:%d\n", retval,__FILE__, __LINE__);
				exit(1);
			}
			// set curl options
			//memset((void*)resp_buff[idx], 0, sizeof(resp_buff[idx]));
			set_curl_common_option(easy[idx], resp_buff[idx], req_timeout);
			retval = curl_multi_add_handle(multi, easy[idx]);
			if ( retval != CURLM_OK) {
				fprintf(stderr, "\nERROR curl_multi_perform() return error = %d %s:%d\n", retval,__FILE__, __LINE__);
				exit(1);
			}
		}

		if(1 == run_epoll) { // running epoll() implementation
			params->timeouts += execute_epoll(multi, params);
		}else { // running select() implementation
			params->timeouts += execute_select(multi, params);
		}
		
		for(idx = 0; idx < MAX_URLS; idx++) {
			retval = curl_multi_remove_handle(multi, easy[idx]);
			if (retval != CURLM_OK) {
				fprintf(stderr, "\nERROR curl_multi_remove_handle RTB failed with rc = %d %s:%d\n", retval, __FILE__, __LINE__);
				exit(1);
			}
		}
	}

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

/* 
 * 1. argv[1] number of threads needs to be created 
 * 2. argv[2] number of iteration per thread
 * 3. argv[3] run_epoll (0/1) (select/epoll) 
 * 4. argv[4] http request timeout
 */
int main(int argc, char** argv)
{
	if(5 != argc) {
		fprintf(stderr, "\nUsage: thread_count thread_iteration, run_poll(0/1) 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]);
	iteration = atoi(argv[2]);
	run_epoll = atoi(argv[3]);
	int req_timeout = atoi(argv[4]);

	if (run_epoll) {
		ProfilerStart("/tmp/epoll.prof");
	}else{
		ProfilerStart("/tmp/select.prof");
	}

	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(stderr, "\nERROR malloc() failed %s:%d\n", __FILE__, __LINE__);
		exit(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].waiting_time = req_timeout;
		params[idx].timeouts = 0;
		// create epoll_fd for thread;
		params[idx].epoll_fd = epoll_create(50);
		if (params[idx].epoll_fd == -1) {
			fprintf(stderr, "ERROR epoll_create");
			exit(1);
		}

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

	// Thread wait
	for (idx = 0; idx < nthreads; idx++) {
                ret = pthread_join(tid[idx], NULL);
                if (ret != 0) {
                        fprintf(stderr, "\nERROR pthread_join() ret = %d %s:%d\n", ret, __FILE__, __LINE__);
                        exit(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;
	for(idx = 0; idx < nthreads; idx++) {
		total_timeouts += params[idx].timeouts;
	}

        fprintf(stderr, "\nMain thread returning after time %lf ms and total timeouts = %d\n", t_diff, total_timeouts);
	ProfilerStop();
	return 0;
}