#include <climits></climits>

#include <cstring></cstring>

#include <fcntl.h></fcntl.h>

#include <unistd.h></unistd.h>

#include <sys types.h=""></sys>

#include <sys socket.h=""></sys>

#include <sys epoll.h=""></sys>

#include <sys eventfd.h=""></sys>

#include <netinet in.h=""></netinet>

#include <netinet tcp.h=""></netinet>

#include <netdb.h></netdb.h>

#include "protocol.h"

#define DebugMSG     HideDebugMSG

#define SrvDebugMSG  HideDebugMSG

#define ConnDebugMSG ShowDebugMSG

Protocol::Protocol():

	thread(),

	epollFd(-1),

	eventFd(-1),

	state(STATE_NONE),

	stateWanted(STATE_NONE),

	closeWaiters(0),

	finishWaiters(0),

	connFirst(), connLast(),

	srvFirst(), srvLast()

	{ }

Protocol::~Protocol() {

	assert(state == STATE_NONE || state == STATE_FINISHED);

	closeWait();

	while(finishWaiters > 0) std::this_thread::yield();

	DebugMSG();

	if (thread) thread->join();

	DebugMSG("deleted");

}

bool Protocol::open() {

	DebugMSG();

	State s = state;

	do {

		if (s != STATE_NONE && s != STATE_FINISHED) return false;

	} while(!state.compare_exchange_weak(s, STATE_OPENING));

	if (s == STATE_FINISHED) while(finishWaiters > 0) std::this_thread::yield();

	if (thread) thread->join();

	thread = new std::thread(&Protocol::threadRun, this);

	state = STATE_INITIALIZING;

	DebugMSG();

	return true;

}

void Protocol::wantState(State state) {

	ReadLock lock(stateProtector);

	State s = this->state;

	DebugMSG("%d, %d", s, state);

	if (s <= STATE_OPENING || s >= STATE_CLOSING || s >= state) return;

	DebugMSG("apply");

	State sw = stateWanted;

	do {

		if (sw >= state) return;

	} while(!stateWanted.compare_exchange_weak(sw, state));

	wakeup();

}

void Protocol::closeReq()

	{ DebugMSG(); wantState(STATE_CLOSE_REQ); }

void Protocol::close()

	{ DebugMSG(); wantState(STATE_CLOSING); }

void Protocol::closeWait(unsigned long long timeoutUs, bool withReq) {

	DebugMSG();

	stateProtector.lock();

	State s = state;

	if (s <= STATE_OPENING || s >= STATE_FINISHED)

		{ stateProtector.unlock(); return; }

	if (s == STATE_CLOSED) {

		finishWaiters++;

		stateProtector.unlock();

	} else {

		if (withReq && s < STATE_CLOSE_REQ) closeReq();

		closeWaiters++;

		stateProtector.unlock();

		std::mutex fakeMutex;

		{

			std::unique_lock<std::mutex> fakeLock(fakeMutex);</std::mutex>

			if (timeoutUs && s < STATE_CLOSING) {

				std::chrono::steady_clock::time_point t

					= std::chrono::steady_clock::now()

						+ std::chrono::microseconds(timeoutUs);

				while(s < STATE_CLOSED) {

					if (t <= std::chrono::steady_clock::now()) { close(); break; }

					DebugMSG("wait_until, %d", s);

					closeCondition.wait_until(fakeLock, t);

					s = state;

				}

			}

			if (s < STATE_CLOSING) close();

			while(s < STATE_CLOSED) {

				DebugMSG("wait, %d", s);

				closeCondition.wait(fakeLock);

				s = state;

			}

			DebugMSG("awaiten");

		}

		finishWaiters++;

		closeWaiters--;

	}

	while(s != STATE_FINISHED) { std::this_thread::yield(); s = state; }

	finishWaiters--;

	DebugMSG();

}

void Protocol::wakeup() {

	unsigned long long num = 0;

	::write(eventFd, &num, sizeof(num));

}

void Protocol::updateEvents(Connection &connection) {

	unsigned int events = EPOLLERR | EPOLLRDHUP;

	if (connection.readQueue.count()) events |= EPOLLIN;

	if (connection.writeQueue.count()) events |= EPOLLOUT;

	ConnDebugMSG("events: %08x", events);

	if (connection.events.exchange(events) == events) return;

	ConnDebugMSG("apply");

	struct epoll_event event = {};

	event.data.ptr = &connection;

	event.events = events;

	epoll_ctl(epollFd, EPOLL_CTL_MOD, connection.sockId, &event);

}

void Protocol::initSocket(int sockId) {

	int tcp_nodelay = 1;

	setsockopt(sockId, IPPROTO_TCP, TCP_NODELAY, &tcp_nodelay, sizeof(int));

	fcntl(sockId, F_SETFL, fcntl(sockId, F_GETFL, 0) | O_NONBLOCK);

}

void Protocol::threadConnQueue() {

	while(connQueue.peek()) {

		ConnDebugMSG();

		Connection &connection = connQueue.pop();

		connection.enqueued = false;

		if (connection.stateLocal == Connection::STATE_NONE) {

			ConnDebugMSG("none");

			int sockId = connection.sockId;

			bool connected = false;

			if (sockId >= 0) {

				ConnDebugMSG("accepting");

				initSocket(sockId);

			} else

			if (connection.addressSize >= sizeof(sa_family_t)) {

				ConnDebugMSG("connecting");

				struct sockaddr *addr = (struct sockaddr*)connection.address;

				sockId = ::socket(addr->sa_family, SOCK_STREAM, 0);

				if (sockId >= 0) {

					initSocket(sockId);

					if (0 == ::connect(sockId, addr, (int)connection.addressSize))

						connected = true;

				} else {

					ConnDebugMSG("none -> cannot create socket");

				}

			}

			struct epoll_event event = {};

			if (sockId >= 0) {

				event.data.ptr = &connection;

				event.events = EPOLLERR | EPOLLRDHUP;

				if (!connected) event.events |= EPOLLOUT;

				if (0 != epoll_ctl(epollFd, EPOLL_CTL_ADD, sockId, &event)) {

					ConnDebugMSG("epoll_ctl(%d, ADD, %d, %08x) errno: %d", epollFd, sockId, event.events, errno);

					::close(sockId);

					connection.sockId = sockId = -1;

					assert(false);

				}

			}

			if (sockId >= 0) {

				ConnDebugMSG("opening");

				connection.protocol = this;

				connection.prev = connLast;

				assert(!connection.next);

				(connection.prev ? connection.prev->next : connFirst) = &connection;

				if (connection.server) {

					connection.srvPrev = connection.server->connLast;

					assert(!connection.srvNext);

					(connection.srvPrev ? connection.srvPrev->srvNext : connection.server->connFirst) = &connection;

				}

				connection.sockId = sockId;

				if (connected) {

					ConnDebugMSG("to open");

					connection.events = event.events;

					connection.state = connection.stateLocal = Connection::STATE_OPEN;

					connection.stateProtector.wait();

					connection.onOpen(connection.address, connection.addressSize);

				} else {

					assert(!connection.server);

					ConnDebugMSG("to connecting");

					connection.state = connection.stateLocal = Connection::STATE_CONNECTING;

					connection.stateProtector.wait();

				}

			} else {

				ConnDebugMSG("none -> to closed");

				Server *server = connection.server;

				assert(connected || !server);

				connection.server = nullptr;

				connection.sockId = -1;

				connection.onOpeningError();

				connection.state = Connection::STATE_CLOSED;

				while(connection.closeWaiters > 0)

					{ connection.closeCondition.notify_all(); std::this_thread::yield(); }

				connection.state = Connection::STATE_FINISHED;

				ConnDebugMSG("none -> to finished");

				if (server) threadSrvDisconnect(*server, connection, true);

			}

		} else {

			Connection::State stateWanted = connection.stateWanted;

			if ( connection.stateLocal == Connection::STATE_CONNECTING

			  && stateWanted == Connection::STATE_CLOSING )

			{

				ConnDebugMSG("connecting -> to closing");

				connection.onOpeningError();

				connection.state = connection.stateLocal = Connection::STATE_CLOSING;

				connection.stateProtector.wait();

			}

			if ( connection.stateLocal == Connection::STATE_OPEN

			  && stateWanted == Connection::STATE_CLOSE_REQ )

			{

				ConnDebugMSG("to close req");

				connection.state = connection.stateLocal = Connection::STATE_CLOSE_REQ;

				connection.stateProtector.wait();

				connection.onCloseReqested();

			} else

			if ( connection.stateLocal >= Connection::STATE_OPEN

			  && connection.stateLocal <= Connection::STATE_CLOSE_REQ

			  && stateWanted == Connection::STATE_CLOSING )

			{

				ConnDebugMSG("to closing");

				connection.state = connection.stateLocal = Connection::STATE_CLOSING;

				connection.stateProtector.wait();

				while(Connection::Task *task = connection.readQueue.peek()) {

					connection.readQueue.pop();

					if (!connection.onReadReady(*task, true))

						assert(false);

				}

				while(Connection::Task *task = connection.writeQueue.peek()) {

					connection.writeQueue.pop();

					if (!connection.onWriteReady(*task, true))

						assert(false);

				}

				connection.errorLocal = connection.error;

				connection.onClose(connection.errorLocal);

			}

			if (connection.stateLocal == Connection::STATE_CLOSING && !connection.enqueued) {

				ConnDebugMSG("to closed");

				Server *server = connection.server;

				if (server) {

					(connection.srvPrev ? connection.srvPrev->srvNext : connection.server->connFirst) = connection.srvNext;

					(connection.srvNext ? connection.srvNext->srvPrev : connection.server->connLast) = connection.srvPrev;

					connection.server = nullptr;

				}

				bool error = connection.errorLocal;

				connection.errorLocal = false;

				(connection.prev ? connection.prev->next : connFirst) = connection.next;

				(connection.next ? connection.next->prev : connLast) = connection.prev;

				if (connection.sockId >= 0) {

					struct epoll_event event = {};

					epoll_ctl(epollFd, EPOLL_CTL_DEL, connection.sockId, &event);

					::close(connection.sockId);

					connection.sockId = -1;

				}

				connection.protocol = nullptr;

				memset(connection.address, 0, sizeof(connection.address));

				connection.addressSize = 0;

				connection.stateWanted = Connection::STATE_NONE;

				connection.stateLocal = Connection::STATE_NONE;

				connection.state = Connection::STATE_CLOSED;

				while(connection.closeWaiters > 0)

					{ connection.closeCondition.notify_all(); std::this_thread::yield(); }

				connection.state = Connection::STATE_FINISHED;

				ConnDebugMSG("to finished");

				if (server) threadSrvDisconnect(*server, connection, error);

			}

		}

	}

}

void Protocol::threadConnEvents(Connection &connection, unsigned int events) {

	bool needUpdate = false;

	if (connection.stateLocal == Connection::STATE_CONNECTING) {

		ConnDebugMSG("connecting");

		if (events & (EPOLLHUP | EPOLLERR)) {

			connection.close(true);

			return;

		}

		if (events & EPOLLOUT) {

			ConnDebugMSG("connecting -> to open");

			connection.state = connection.stateLocal = Connection::STATE_OPEN;

			connection.stateProtector.wait();

			connection.onOpen(connection.address, connection.addressSize);

			needUpdate = true;

		}

	}

	if ( connection.stateLocal < Connection::STATE_OPEN

	  || connection.stateLocal >= Connection::STATE_CLOSING )

		return;

	bool closing = (bool)(events & (EPOLLHUP | EPOLLRDHUP | EPOLLERR));

	if (events & EPOLLIN) {

		ConnDebugMSG("input");

		bool ready = true;

		int pops = 0;

		int count = (connection.readQueue.count() + 2)*2;

		while(ready) {

			Connection::Task *task = connection.readQueue.peek();

			if (!task) {

				if (pops) needUpdate = true;

				break;

			}

			size_t completion = 0;

			while(task->size > task->completion) {

				size_t size = task->size - task->completion;

				if (size > INT_MAX) size = INT_MAX;

				int res = ::recv(

					connection.sockId,

					(unsigned char*)task->data + task->completion,

					(int)size, MSG_DONTWAIT );

				if (res == EAGAIN || res == 0) {

					ready = false;

					break;

				} else

				if (res < 0) {

					connection.close(true);

					return;

				} else {

					completion += res;

					task->completion += res;

				}

			}

			if (task->size <= task->completion || (completion && task->watch)) {

				connection.readQueue.pop();

				if (!connection.onReadReady(*task, false))

					connection.readQueue.unpop(*task); else ++pops;

				if (!closing && --count < 0) ready = false;

			}

		}

	}

	if (closing) {

		ConnDebugMSG("hup");

		connection.close((bool)(events & EPOLLERR));

		return;

	}

	if (events & EPOLLOUT) {

		ConnDebugMSG("output");

		bool ready = true;

		int pops = 0;

		int count = connection.writeQueue.count() + 2;

		while(ready) {

			Connection::Task *task = connection.writeQueue.peek();

			if (!task) {

				if (pops) needUpdate = true;

				break;

			}

			size_t completion = 0;

			while(task->size > task->completion) {

				size_t size = task->size - task->completion;

				if (size > INT_MAX) size = INT_MAX;

				int res = ::send(

					connection.sockId,

					(const unsigned char*)task->data + task->completion,

					(int)size, MSG_DONTWAIT );

				if (res == EAGAIN || res == 0) {

					ready = false;

					break;

				} else

				if (res < 0) {

					connection.close(true);

					return;

				} else {

					completion += res;

					task->completion += res;

				}

			}

			if (task->size <= task->completion || (completion && task->watch)) {

				connection.writeQueue.pop();

				if (!connection.onWriteReady(*task, false))

					connection.writeQueue.unpop(*task); else ++pops;

				if (--count <= 0) ready = false;

			}

		}

	}

	if (needUpdate) updateEvents(connection);

}

void Protocol::threadSrvDisconnect(Server& server, Connection& connection, bool error) {

	server.onDisconnect(&connection, error);

	if (!server.connFirst && server.stateLocal == Server::STATE_CLOSING_CONNECTIONS) {

		SrvDebugMSG("to closing");

		server.state = server.stateLocal = Server::STATE_CLOSING;

		server.stateProtector.wait();

		server.onClose(server.error);

		if (!server.enqueued.exchange(true)) srvQueue.push(server);

	}

}

void Protocol::threadSrvQueue() {

	while(srvQueue.peek()) {

		SrvDebugMSG();

		Server &server = srvQueue.pop();

		server.enqueued = false;

		if (server.stateLocal == Server::STATE_NONE) {

			SrvDebugMSG("none");

			int sockId = -1;

			if (server.addressSize >= sizeof(sa_family_t)) {

				struct sockaddr *addr = (struct sockaddr*)server.address;

				sockId = ::socket(addr->sa_family, SOCK_STREAM, 0);

				if (sockId >= 0) {

					initSocket(sockId);

					if ( 0 != ::bind(sockId, addr, (int)server.addressSize)

					  || 0 != ::listen(sockId, 128) )

					{

						SrvDebugMSG("cannot bind/listen");

						::close(sockId);

						sockId = -1;

					}

				} else {

					SrvDebugMSG("cannot create socket");

				}

			}

			if (sockId >= 0) {

				SrvDebugMSG("to open");

				server.protocol = this;

				server.prev = srvLast;

				assert(!server.next);

				(server.prev ? server.prev->next : srvFirst) = &server;

				server.sockId = sockId;

				struct epoll_event event = {};

				event.data.ptr = &server;

				event.events = EPOLLIN | EPOLLERR;

				epoll_ctl(epollFd, EPOLL_CTL_ADD, sockId, &event);

				server.state = server.stateLocal = Server::STATE_OPEN;

				server.stateProtector.wait();

				server.onOpen();

			} else {

				SrvDebugMSG("opening error -> to closed");

				server.onOpeningError();

				server.state = Server::STATE_CLOSED;

				while(server.closeWaiters > 0)

					{ server.closeCondition.notify_all(); std::this_thread::yield(); }

				server.state = Server::STATE_FINISHED;

				SrvDebugMSG("opening error -> finished");

			}

		} else {

			Server::State stateWanted = server.stateWanted;

			if ( server.stateLocal == Server::STATE_OPEN

			  && stateWanted == Server::STATE_CLOSE_REQ )

			{

				SrvDebugMSG("to close req");

				for(Connection *conn = server.connFirst; conn; conn = conn->srvNext)

					conn->closeReq();

				struct epoll_event event = {};

				epoll_ctl(epollFd, EPOLL_CTL_DEL, server.sockId, &event);

				server.state = server.stateLocal = Server::STATE_CLOSE_REQ;

				server.stateProtector.wait();

				server.onCloseReqested();

			} else

			if ( ( server.stateLocal == Server::STATE_OPEN

				|| server.stateLocal == Server::STATE_CLOSE_REQ )

			  && stateWanted == Server::STATE_CLOSING_CONNECTIONS )

			{

				SrvDebugMSG("to closing connections");

				if (server.stateLocal != Server::STATE_CLOSE_REQ) {

					struct epoll_event event = {};

					epoll_ctl(epollFd, EPOLL_CTL_DEL, server.sockId, &event);

				}

				if (server.connFirst) {

					for(Connection *conn = server.connFirst; conn; conn = conn->srvNext)

						conn->close(true);

					server.state = server.stateLocal = Server::STATE_CLOSING_CONNECTIONS;

				} else {

					SrvDebugMSG("to closing");

					server.state = server.stateLocal = Server::STATE_CLOSING;

					server.stateProtector.wait();

					server.onClose(server.error);

				}

			}

			if (server.stateLocal == Server::STATE_CLOSING && !server.enqueued) {

				SrvDebugMSG("to closed");

				if (server.sockId >= 0) {

					struct epoll_event event = {};

					epoll_ctl(epollFd, EPOLL_CTL_DEL, server.sockId, &event);

					::close(server.sockId);

					server.sockId = -1;

				}

				(server.prev ? server.prev->next : srvFirst) = server.next;

				(server.next ? server.next->prev : srvLast) = server.prev;

				server.protocol = nullptr;

				memset(server.address, 0, sizeof(server.address));

				server.addressSize = 0;

				server.stateWanted = Server::STATE_NONE;

				server.stateLocal = Server::STATE_NONE;

				SrvDebugMSG("closeWaiters");

				server.state = Server::STATE_CLOSED;

				while(server.closeWaiters > 0)

					{ server.closeCondition.notify_all(); std::this_thread::yield(); }

				server.state = Server::STATE_FINISHED;

				SrvDebugMSG("finished");

			}

		}

	}

}

void Protocol::threadSrvEvents(Server &server, unsigned int events) {

	if (events & (EPOLLHUP | EPOLLERR)) {

		if ( server.stateLocal >= Server::STATE_OPEN

		  && server.stateLocal < Server::STATE_CLOSING )

			server.close((bool)(events & EPOLLERR));

		return;

	}

	if (events & EPOLLIN) {

		if ( server.stateLocal >= Server::STATE_OPEN

		  && server.stateLocal < Server::STATE_CLOSE_REQ)

		{

			unsigned char address[MAX_ADDR_SIZE] = {};

			socklen_t len = sizeof(address);

			int res = ::accept(server.sockId, (struct sockaddr*)address, &len);

			assert(len < sizeof(address));

			if (res >= 0) {

				if (Connection *conn = server.onConnect(address, len))

					if (!conn->open(*this, address, len, res))

						server.onDisconnect(conn, true);

			} else

			if (res != EAGAIN) {

				server.close(true);

				return;

			}

		}

	}

}

void Protocol::threadRun() {

	DebugMSG();

	epollFd = ::epoll_create(32);

	eventFd = ::eventfd(0, EFD_NONBLOCK);

	assert(epollFd >= 0);

	assert(eventFd >= 0);

	struct epoll_event event = {};

	event.events = EPOLLIN;

	epoll_ctl(epollFd, EPOLL_CTL_ADD, eventFd, &event);

	const int maxCount = 16;

	struct epoll_event events[maxCount] = {};

	int count = 0;

	State stateLocal = STATE_OPEN;

	state = stateLocal;

	while(true) {

		for(int i = 0; i < count; ++i) {

			Root *pointer = (Root*)events[i].data.ptr;

			if (!pointer) {

				unsigned long long num = 0;

				while(::read(eventFd, &num, sizeof(num)) > 0 && !num);

			} else

			if (Connection *connection = dynamic_cast<connection*>(pointer)) {</connection*>

				threadConnEvents(*connection, events[i].events);

			} else

			if (Server *server = dynamic_cast<server*>(pointer)) {</server*>

				threadSrvEvents(*server, events[i].events);

			}

		}

		State sw = stateWanted;

		if (stateLocal == STATE_OPEN && sw == STATE_CLOSE_REQ) {

			DebugMSG("to closse req");

			state = stateLocal = STATE_CLOSE_REQ;

			stateProtector.wait();

			for(Server *server = srvFirst; server; server = server->next)

				server->closeReq();

			for(Connection *conn = connFirst; conn; conn = conn->next)

				if (!conn->server)

					conn->closeReq();

		} else

		if (stateLocal >= STATE_OPEN && stateLocal <= STATE_CLOSE_REQ && sw == STATE_CLOSING) {

			DebugMSG("to closing");

			state = stateLocal = STATE_CLOSING;

			stateProtector.wait();

			for(Server *server = srvFirst; server; server = server->next)

				server->close(true);

			for(Connection *conn = connFirst; conn; conn = conn->next)

				if (!conn->server)

					conn->close(true);

		}

		while(connQueue.peek() || srvQueue.peek()) {

			threadConnQueue();

			threadSrvQueue();

		}

		if (stateLocal == STATE_CLOSING) {

			assert(!connFirst && !srvFirst);

			break;

		}

		count = epoll_wait(epollFd, events, maxCount, 1000);

		DebugMSG("epoll count: %d", count);

	}

	DebugMSG("closeWaiters");

	state = STATE_CLOSED;

	while(closeWaiters > 0)

		{ closeCondition.notify_all(); std::this_thread::yield(); }

	::close(epollFd);

	::close(eventFd);

	epollFd = -1;

	eventFd = -1;

	assert(!connFirst && !srvFirst);

	assert(!connQueue.peek() && !srvQueue.peek());

	stateWanted = STATE_NONE;

	stateLocal = STATE_NONE;

	state = STATE_FINISHED;

	DebugMSG("finished");

}