/*
* Author: Manoj Ampalam <manoj.ampalam@microsoft.com>
*/

#include "includes.h"
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/select.h>
#include <unistd.h>
#include <fcntl.h>
#include "../test_helper/test_helper.h"

#define SMALL_RECV_BUF_SIZE 128

#pragma warning(disable:4267)

fd_set read_set, write_set, except_set;
struct timeval time_val;
char *send_buf, *recv_buf;
int ret, r, w;

int unset_nonblock(int fd);

int set_nonblock(int fd);

void prep_input_buffer(char* buf, int size, int seed);

void 
file_blocking_io_tests()
{
	char* small_send_buf = "sample payload";
	char small_recv_buf[SMALL_RECV_BUF_SIZE];

	TEST_START("Basic pipe()");
	int pipeio[2];
	ret = pipe(pipeio);
	ASSERT_INT_EQ(ret, 0);
	TEST_DONE();

	TEST_START("pipe read and write");
	r = pipeio[0];
	w = pipeio[1];
	ret = write(r, small_send_buf, strlen(small_send_buf));
	ASSERT_INT_EQ(ret, -1);
	ASSERT_INT_EQ(errno, EACCES);
	ret = read(w, small_recv_buf, SMALL_RECV_BUF_SIZE);
	ASSERT_INT_EQ(ret, -1);
	ASSERT_INT_EQ(errno, EACCES);
	ret = write(w, small_send_buf, strlen(small_send_buf));
	ASSERT_INT_EQ(ret, strlen(small_send_buf));
	ret = read(r, small_recv_buf, SMALL_RECV_BUF_SIZE);
	ASSERT_INT_EQ(ret, strlen(small_send_buf));
	small_recv_buf[ret] = '\0';
	ASSERT_STRING_EQ(small_send_buf, small_recv_buf);
	memset(small_recv_buf, 0, sizeof(small_recv_buf));
	TEST_DONE();

	TEST_START("close pipe fds");
	ret = close(w);
	ASSERT_INT_EQ(ret, 0);
	ret = read(r, small_recv_buf, SMALL_RECV_BUF_SIZE); /* send on other side is closed*/
	ASSERT_INT_EQ(ret, 0);
	ret = close(r);
	ASSERT_INT_EQ(ret, 0);
	TEST_DONE();
}

void file_simple_fileio()
{
	char* small_write_buf = "sample payload";
	char small_read_buf[SMALL_RECV_BUF_SIZE];

	int f;
	TEST_START("file io");
	f = open("tmp.txt", O_WRONLY | O_CREAT | O_TRUNC);
	ASSERT_INT_NE(f, -1);
	close(f);
	f = open("tmp.txt", O_RDONLY);
	ASSERT_INT_NE(f, -1);
	struct stat st;
	ret = fstat(f, &st);
	ASSERT_INT_EQ(ret, 0);
	ASSERT_INT_EQ(st.st_size, 0);
	ret = read(f, small_read_buf, SMALL_RECV_BUF_SIZE);
	ASSERT_INT_EQ(ret, 0);
	close(f);
	f = open("tmp.txt", O_WRONLY | O_CREAT | O_TRUNC);
	ASSERT_INT_NE(f, -1);
	ret = write(f, small_write_buf, strlen(small_write_buf));
	ASSERT_INT_EQ(ret, strlen(small_write_buf));
	close(f);
	f = open("tmp.txt", O_RDONLY);
	ASSERT_INT_NE(f, -1);
	ret = fstat(f, &st);
	ASSERT_INT_EQ(ret, 0);
	ASSERT_INT_EQ(st.st_size, strlen(small_write_buf));
	ret = read(f, small_read_buf, SMALL_RECV_BUF_SIZE);
	ASSERT_INT_EQ(ret, strlen(small_write_buf));
	small_read_buf[ret] = '\0';
	ASSERT_STRING_EQ(small_write_buf, small_read_buf);
	ret = read(f, small_read_buf, SMALL_RECV_BUF_SIZE);
	ASSERT_INT_EQ(ret, 0);
	close(f);
	TEST_DONE();
}

void file_simple_fileio_mode()
{
	char * small_write_buf = "sample payload", *c, small_read_buf[SMALL_RECV_BUF_SIZE];
	int ret;
	FILE* f;
	struct stat st;

	TEST_START("file mode");
	f = fopen("tmp.txt", "w");
	ASSERT_PTR_NE(f, NULL);
	fclose(f);
	f = fopen("tmp.txt", "r");
	ASSERT_PTR_NE(f, NULL);
	c = fgets(small_read_buf, sizeof(small_read_buf), f);
	ASSERT_PTR_EQ(c, NULL);
	fclose(f);
	
	ret = stat("tmp.txt", &st);
	ASSERT_INT_EQ(ret, 0);
	ASSERT_INT_EQ(st.st_size, 0);
	
	f = fopen("tmp.txt", "w");
	ASSERT_PTR_NE(f, NULL);
	ret = fputs(small_write_buf, f);
	ASSERT_INT_EQ(ret, 0);
	fclose(f);

	ret = stat("tmp.txt", &st);
	ASSERT_INT_EQ(ret, 0);
	ASSERT_INT_EQ(st.st_size, strlen(small_write_buf));

	f = fopen("tmp.txt", "r");
	ASSERT_PTR_NE(f, NULL);
	c = fgets(small_read_buf, sizeof(small_read_buf), f);
	ASSERT_PTR_NE(c, NULL);	
	ASSERT_STRING_EQ(small_write_buf, small_read_buf);

	c = fgets(small_read_buf, sizeof(small_read_buf), f);
	ASSERT_PTR_EQ(c, NULL);	
	fclose(f);
	TEST_DONE();
}

void 
file_nonblocking_io_tests()
{
	char* small_send_buf = "sample payload";
	char small_recv_buf[SMALL_RECV_BUF_SIZE];

	TEST_START("non blocking file io");
	int pipeio[2];
	ret = pipe(pipeio);
	ASSERT_INT_EQ(ret, 0);
	r = pipeio[0];
	w = pipeio[1];
	ret = set_nonblock(r);
	ASSERT_INT_EQ(ret, 0);
	ret = read(r, small_recv_buf, SMALL_RECV_BUF_SIZE);
	ASSERT_INT_EQ(ret, -1);
	ASSERT_INT_EQ(errno, EAGAIN);
	ret = unset_nonblock(w);
	ASSERT_INT_EQ(ret, 0);
	ret = write(w, small_send_buf, strlen(small_send_buf));
	ASSERT_INT_EQ(ret, strlen(small_send_buf));
	ret = unset_nonblock(r);
	ASSERT_INT_EQ(ret, 0);
	ret = read(r, small_recv_buf, SMALL_RECV_BUF_SIZE);
	ASSERT_INT_EQ(ret, strlen(small_send_buf));
	small_recv_buf[ret] = '\0';
	ASSERT_STRING_EQ(small_send_buf, small_recv_buf);
	memset(small_recv_buf, 0, sizeof(small_recv_buf));
	send_buf = malloc(10 * 1024);
	ASSERT_PTR_NE(send_buf, NULL);
	ret = set_nonblock(w);
	ASSERT_INT_EQ(ret, 0);
	ret = 1;
	while (ret > 0) {
		ret = write(w, send_buf, 10 * 1024);
	}
	ASSERT_INT_EQ(ret, -1);
	ASSERT_INT_EQ(errno, EAGAIN);
	ret = close(r);
	ASSERT_INT_EQ(ret, 0);
	ret = close(w);
	ASSERT_INT_EQ(ret, 0);
	TEST_DONE();

	free(send_buf);
}

void 
file_select_tests() {
	int num_bytes = 1024 * 700; //700KB
	int bytes_sent = 0;
	int bytes_received = 0;
	int seed = 326;
	int eagain_results = 0;

	TEST_START("select on file fds");
	int pipeio[2];
	ret = pipe(pipeio);
	ASSERT_INT_EQ(ret, 0);
	r = pipeio[0];
	w = pipeio[1];
	ret = set_nonblock(w);
	ASSERT_INT_EQ(ret, 0);
	ret = set_nonblock(r);
	ASSERT_INT_EQ(ret, 0);
	send_buf = malloc(num_bytes);
	recv_buf = malloc(num_bytes + 1);
	ASSERT_PTR_NE(send_buf, NULL);
	ASSERT_PTR_NE(recv_buf, NULL);
	prep_input_buffer(send_buf, num_bytes, 17);
	FD_ZERO(&read_set);
	FD_ZERO(&write_set);
	FD_SET(w, &write_set);
	FD_SET(r, &read_set);
	while (-1 != select(max(r, w) + 1, &read_set, &write_set, NULL, &time_val)) {
		if (FD_ISSET(w, &write_set)) {
			while ((bytes_sent < num_bytes) && ((ret = write(w, send_buf + bytes_sent, num_bytes - bytes_sent)) > 0))
				bytes_sent += ret;
			if (bytes_sent < num_bytes) {
				ASSERT_INT_EQ(ret, -1);
				ASSERT_INT_EQ(errno, EAGAIN);
				eagain_results++;
			}
		}

		if (FD_ISSET(r, &read_set)) {
			while ((ret = read(r, recv_buf + bytes_received, num_bytes - bytes_received + 1)) > 0)
				bytes_received += ret;
			if (ret == 0)
				break;
			ASSERT_INT_EQ(ret, -1);
			ASSERT_INT_EQ(errno, EAGAIN);
			eagain_results++;
		}

		if (bytes_sent < num_bytes)
			FD_SET(w, &write_set);
		else {
			FD_CLR(w, &write_set);
			ret = close(w);
			ASSERT_INT_EQ(ret, 0);
		}
		FD_SET(r, &read_set);
	}

	/*ensure that we hit send and recv paths that returned EAGAIN. Else it would not have touched the async paths*/
	/*if this assert is being hit, then num_bytes is too small. up it*/
	ASSERT_INT_GT(eagain_results, 0);
	ASSERT_INT_EQ(bytes_sent, bytes_received);
	ASSERT_INT_EQ(memcmp(send_buf, recv_buf, num_bytes), 0);
	ret = close(r);
	ASSERT_INT_EQ(ret, 0);

	free(send_buf);
	free(recv_buf);
	TEST_DONE();

}

void console_io_test()
{
	char tmp[10];
	TEST_START("console io test");
	ret = read(STDIN_FILENO, tmp, 10);
	ret = write(STDOUT_FILENO, "sample output", 13);
	ASSERT_INT_EQ(errno, 0);
	ASSERT_INT_EQ(ret, 13);
	TEST_DONE();
}

void 
file_tests()
{
	//console_io_test();
	file_simple_fileio();
	file_simple_fileio_mode();
	file_blocking_io_tests();
	file_nonblocking_io_tests();
	file_select_tests();
}