#include "stdafx.h"
#include "apptasker.h"
#include <iostream>
#include <stdlib.h>
#include "util.h"
#include "config.h"
#include "data.h"
#include "analysis.h"
#include "DataCollector.h"
#include "assert.h"
#include "tinyxml/tinyxml.h"

CAppTasker::CAppTasker() : 
m_thread(ThreadMain, this), 
m_incoming(ThreadIncoming, this),
LastSessionId(0),
m_stats(10000),
m_reportMinMax(NULL)
{
	m_reportMinMax = fopen("minmax_report.txt", "w");
}

CAppTasker::~CAppTasker()
{
	fclose(m_reportMinMax);
}

void CAppTasker::ThreadMain(void * p)
{
	static_cast<CAppTasker*>(p)->Main();
}

void CAppTasker::Main()
{
	CThread::SetName("Tasker");

	std::set<uint32> filter;
	SConfig config = GetConfig();
	
	for(size_t i=0;i<config.hosts.size();i++)
	{
		sockaddr_in addr=AddrFromString(config.hosts[i]);
		filter.insert(addr.sin_addr.S_un.S_addr);
	}

	TiXmlHandle hdl(&config.policiesXml);
	TiXmlElement* policies = hdl.FirstChild("CompressionPolicy").ToElement();
	if (!policies)
		throw std::runtime_error("no policy nodes");

	for (TiXmlNode* policy = policies->FirstChildElement("Policy"); policy; policy = policy->NextSibling("Policy"))
	{
		const char* name = policy->ToElement()->Attribute("name");
		if (!name)
			continue;
		TPolicy key = StringToKey(name);
		SMinMax3 mm;

#define SET_MINMAX(mm_xyz) \
	const char* s = node->ToElement()->Attribute("min"); \
	mm_xyz.mn = s ? atof(s) : 0.0f; \
	s = node->ToElement()->Attribute("max"); \
	mm_xyz.mx = s ? atof(s) : 0.0f; \
	s = node->ToElement()->Attribute("nbits"); \
	mm_xyz.nbits = s ? atoi(s) : 0;

		TiXmlNode* node = policy->ToElement()->FirstChildElement("Range");
		if (node)
		{
			SET_MINMAX(mm.x);
		}
		else
		{
			node = policy->ToElement()->FirstChildElement("Params");
			if (node)
			{
				SET_MINMAX(mm.x);
			}
			else
			{
				node = policy->ToElement()->FirstChildElement("XParams");
				if (node)
				{
					SET_MINMAX(mm.x)
				}
				node = policy->ToElement()->FirstChildElement("YParams");
				if (node)
				{
					SET_MINMAX(mm.y);
				}
				node = policy->ToElement()->FirstChildElement("ZParams");
				if (node)
				{
					SET_MINMAX(mm.z);
				}
			}
		}

#undef SET_MINMAX

		m_policyMinMaxMap[key] = mm;

	}

	while (true)
	{
		std::vector<uint32> output;

		{
			SCOPED_GLOBAL_LOCK;
			std::cerr << "Starting tasker" << std::endl;
		}

		try
		{
			m_sock = INVALID_SOCKET;
			m_sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
			if (m_sock == INVALID_SOCKET)
				throw std::runtime_error("Failed creating socket");

			sockaddr_in addr;
			memset(&addr, 0, sizeof(addr));
			addr.sin_family = AF_INET;
			addr.sin_port = htons(8000);
			if (0 != bind(m_sock, (sockaddr*)&addr, sizeof(addr)))
				throw std::runtime_error("Failed binding socket");

			SetSockOpt( m_sock, SOL_SOCKET, SO_BROADCAST, 1 );
			//SetSockOpt( m_sock, IPPROTO_TCP, IP_RECEIVE_BROADCAST, 1 );

			//GetConfig();

			while (true)
			{
				static const int BUFSZ = 16;
				char buf[BUFSZ];
				int bufsz = BUFSZ;
				int addrlen = sizeof(addr);
				int r = recvfrom(m_sock, buf, bufsz, 0, (sockaddr*)&addr, &addrlen);
				if (r == SOCKET_ERROR)
				{
					//std::cerr << WSAGetLastError() << std::endl;
					throw std::runtime_error("Socket error");
				}
				if (r != 1)
					continue;

				//SConfig config = GetConfig();

				if(config.hostFilter && filter.find(addr.sin_addr.S_un.S_addr)==filter.end())
					continue;

				srand(GetTickCount());

				try
				{
					output.resize(0);

					output.push_back('ok');
					if (config.streamAnalysis)
						output.push_back('strm');
					if (config.logSync)
						output.push_back('log');
					if (config.statsCollector)
						output.push_back('stat');
					if (config.snapping)
						output.push_back('snap');
					if (config.sampleCompression)
					{
						for (TiXmlNode* policy = policies->FirstChildElement("Policy"); policy; policy = policy->NextSibling("Policy"))
						{
							const char* name = policy->ToElement()->Attribute("name");
							if (!name)
								continue;
							const char* s = policy->ToElement()->Attribute("samplingWeight");
							float samplingWeight = s ? atof(s) : 0.0f;
							samplingWeight = samplingWeight > 1.0 ? 1.0 : samplingWeight < 0.0f ? 0.0f : samplingWeight;
							if ( samplingWeight + (1.0f - samplingWeight)*rand()/RAND_MAX  > config.samplingThreshold )
								output.push_back( StringToKey(string(name)) );
						}
					}

					switch (buf[0])
					{
					case 's':
						break;
					case 'c':
						break;
					}
					sendto(m_sock, (const char *)&output[0], output.size() * sizeof(uint32), 0, (sockaddr*)&addr, addrlen);
				}
				catch (std::exception & e)
				{
					SCOPED_GLOBAL_LOCK;
					std::cerr << "CAppTasker: " << e.what() << std::endl;
				}
			}
		}
		catch (std::exception& e)
		{
			SCOPED_GLOBAL_LOCK;
			std::cerr << "CAppTasker: " << e.what() << std::endl;
		}

		if (m_sock != INVALID_SOCKET)
			closesocket(m_sock);
		
		Sleep(2000);
	}
}

void CAppTasker::ThreadIncoming( void * p )
{
	static_cast<CAppTasker*>(p)->Incoming();
}

void CAppTasker::Incoming()
{
	CThread::SetName("Incoming");

	bool errorFlag = true;

	SOCKET sock = socket( AF_INET, SOCK_STREAM, IPPROTO_TCP );
	if (sock == INVALID_SOCKET)
		goto error_nothing;
	sockaddr_in addr;
	memset(&addr, 0, sizeof(addr));
	addr.sin_family = AF_INET;
	addr.sin_port = htons(8001);
	if (0 != bind(sock, (sockaddr*)&addr, sizeof(addr)))
		goto error_withsock;
	if (SOCKET_ERROR == listen(sock, 1))
		goto error_withsock;
	while (true)
	{
		int n = sizeof(addr);
		SOCKET acc = accept(sock, (sockaddr*)&addr, &n);
		if (acc == SOCKET_ERROR)
			continue;
		if (n != sizeof(addr))
		{
			closesocket(acc);
			continue;
		}
		{
			SCOPED_GLOBAL_LOCK;
			fprintf(stderr, "talking to client from %s\n", inet_ntoa(addr.sin_addr));
			//std::cerr << "Talking to client %s" << std::endl;
		}
		ReceiverParams * pParams = new ReceiverParams;
		pParams->addr = addr;
		pParams->sock = acc;
		pParams->pTasker = this;
		_beginthread( ThreadReceiver, 0, pParams );
	}

	errorFlag = false;

error_withsock:
	if (errorFlag)
	{
		fprintf(stderr, "error: %d\n", WSAGetLastError());
	}

	closesocket(sock);
error_nothing:
	return;
}

void CAppTasker::ThreadReceiver( void * p )
{
	static_cast<ReceiverParams*>(p)->pTasker->Receiver( static_cast<ReceiverParams*>(p) );
}

void CAppTasker::Receiver( ReceiverParams * p )
{
	uint32 session_id=InterlockedIncrement(&LastSessionId);

	static int num = 0;
	char buffer[256];
	sprintf(buffer, "Receiver %d", num++);
	CThread::SetName(buffer);

	typedef std::map<uint8, void (CAppTasker::*)(uint32,Socket&, const sockaddr_in&)> Dispatchers;
	Dispatchers dispatchers;

	dispatchers['d'] = &CAppTasker::Dispatch_DataEnt;
	dispatchers['e'] = &CAppTasker::Dispatch_EncodingEnt;
	dispatchers['k'] = &CAppTasker::Dispatch_KeepAlive;
	dispatchers['s'] = &CAppTasker::Dispatch_Stats;
	dispatchers['i'] = &CAppTasker::Dispatch_SessionId;
	dispatchers['p'] = &CAppTasker::Dispatch_Snapping;


	Socket sock(p->sock, 120);
	try
	{
		while (true)
		{
			char type = sock.ReadByte();
			Dispatchers::iterator iter = dispatchers.find(type);
			if (iter == dispatchers.end())
				throw std::runtime_error("Illegal request type");
			(this->*(iter->second))(session_id,sock, p->addr);
		}
	}
	catch (std::exception& e)
	{
		SCOPED_GLOBAL_LOCK;
		std::cerr << e.what() << std::endl;
	}

	shutdown(p->sock, SD_BOTH);
	Sleep(1500);

	delete p;

	// notify that the session is terminated
	sockaddr_in none; none.sin_addr.s_addr = ADDR_ANY;
	for (Dispatchers::const_iterator itor = dispatchers.begin(); itor != dispatchers.end(); ++itor)
		(this->*(itor->second))(session_id, Socket(INVALID_SOCKET), none);
}

void CAppTasker::Dispatch_DataEnt(uint32 session_id, Socket& sk, const sockaddr_in& addr )
{
	SCOPED_GLOBAL_LOCK;

	std::map<uint32, CDataCollector*>::iterator itor = m_dataCollectors.find(session_id);
	if (!sk)
	{
		if (itor != m_dataCollectors.end())
		{
			delete itor->second;
			m_dataCollectors.erase(itor);
		}
		return;
	}

	uint32 eid;
	string name;
	uint32 key;
	SVec3 v;

	ReadSock(sk, eid);
	ReadSock(sk, name);
	ReadSock(sk, key);
	ReadSock(sk, v.x);
	ReadSock(sk, v.y);
	ReadSock(sk, v.z);

	uint32 cookie = 0;
	ReadSock(sk, cookie);
	assert(cookie == 0xdeadbeaf);

	TPolicyMinMaxMap::const_iterator itor2 = m_policyMinMaxMap.find(key);
	if (itor2 != m_policyMinMaxMap.end())
	{
		const SMinMax3& mm = itor2->second;
		string policy = KeyToString(key);
		if (0 != mm.x.nbits)
		{
			if (v.x < mm.x.mn)
				fprintf(m_reportMinMax, "policy[%s] XParams min[%f] actual[%f]\n", policy.c_str(), mm.x.mn, v.x);
			else if (v.x > mm.x.mx)
				fprintf(m_reportMinMax, "policy[%s] XParams max[%f] actual[%f]\n", policy.c_str(), mm.x.mx, v.x);
		}
		if (0 != mm.y.nbits)
		{
			if (v.y < mm.y.mn)
				fprintf(m_reportMinMax, "policy[%s] YParams min[%f] actual[%f]\n", policy.c_str(), mm.y.mn, v.y);
			else if (v.y > mm.y.mx)
				fprintf(m_reportMinMax, "policy[%s] YParams max[%f] actual[%f]\n", policy.c_str(), mm.y.mx, v.y);
		}
		if (0 != mm.z.nbits)
		{
			if (v.z < mm.z.mn)
				fprintf(m_reportMinMax, "policy[%s] ZParams min[%f] actual[%f]\n", policy.c_str(), mm.z.mn, v.z);
			else if (v.z > mm.z.mx)
				fprintf(m_reportMinMax, "policy[%s] ZParams max[%f] actual[%f]\n", policy.c_str(), mm.z.mx, v.z);
		}
	}

	if (itor == m_dataCollectors.end())
	{
		m_dataCollectors[session_id] = new CDataCollector(session_id);
		itor = m_dataCollectors.find(session_id);
	}
	itor->second->Collect(key, eid, v, name);
}

void CAppTasker::Dispatch_EncodingEnt(uint32 session_id, Socket& sk, const sockaddr_in& addr )
{
	if (!sk)
		return;

	SPacketPtr pPkt = new SPacket;
	ReadSock(sk, *pPkt);
	pPkt->senderAddress = *(uint32*)&addr.sin_addr;
	SCOPED_GLOBAL_LOCK;
	SPacketPtr pSafe = pPkt;
	pPkt = NULL;
	AddPacket(pSafe);
}


void CAppTasker::Dispatch_Stats(uint32 session_id, Socket& sk, const sockaddr_in& addr)
{
	if (!sk)
		return;

	uint32 type;
	ReadSock(sk,type);

	switch(type)
	{
	case 'bpkt':
		{
			string to;
			float time;
			uint32 id;
			ReadSock(sk,to);
			ReadSock(sk,time);
			ReadSock(sk,id);
		}
		break;
	case 'ovrh':
		{
			string desc;
			float bits;
			ReadSock(sk,desc);
			ReadSock(sk,bits);
		}
		break;
	case 'msg':
		{
			string mag;
			float bits;
			ReadSock(sk,mag);
			ReadSock(sk,bits);
		}
		break;
	case 'bgrp':
		{
			string name;
			ReadSock(sk,name);
		}
		break;
	case 'data':
		{
			string name;
			float bits;
			ReadSock(sk,name);
			ReadSock(sk,bits);

			m_stats.Data(session_id,name.c_str(),bits);
		}
		break;
	case 'egrp':
		break;
	case 'epkt':
		{
			uint32 size;
			ReadSock(sk,size);

			m_stats.EndPacket();
		}
		break;
	case 'lost':
		{
			uint32 id;
			ReadSock(sk,id);
		}
		break;
	default:
		throw std::exception("invalid stat packet type");
	}
}

void CAppTasker::Dispatch_SessionId(uint32 session_id, Socket& sk, const sockaddr_in& addr )
{
	if (!sk)
		return;

	string value;
	ReadSock(sk, value);
}

void CAppTasker::Dispatch_Snapping(uint32 session_id, Socket& sk, const sockaddr_in& addr)
{
	std::map<uint32, FILE*>::iterator itor = m_snappingReports.find(session_id);
	if (!sk)
	{
		if (itor != m_snappingReports.end())
		{
			fprintf(itor->second, "</table>\n");
			fclose(itor->second);
			m_snappingReports.erase(itor);
		}
		return;
	}

	SVec3 witnessPos(0.0f, 0.0f, 0.0f);
	SVec3 witnessDir(0.0f, 0.0f, 0.0f);
	SVec3 entityPos0(0.0f, 0.0f, 0.0f);
	SVec3 entityPos1(0.0f, 0.0f, 0.0f);
	string entityCls;

	ReadSock(sk, witnessPos);
	ReadSock(sk, witnessDir);
	ReadSock(sk, entityPos0);
	ReadSock(sk, entityPos1);
	ReadSock(sk, entityCls);

	if (itor == m_snappingReports.end())
	{
		SYSTEMTIME st; GetSystemTime(&st);
		std::stringstream filename;
		filename << "snapping_" << st.wYear << "_" << st.wMonth << "_" << st.wDay << "_" << st.wHour << "-" << st.wMinute << "-" << st.wSecond << ".xml";
		m_snappingReports[session_id] = fopen(filename.str().c_str(), "w");
		itor = m_snappingReports.find(session_id);
		fprintf(itor->second, "<table>\n");
	}

	FILE* snappingReport = itor->second;
	fprintf(snappingReport, "\t<record>\n");
	fprintf(snappingReport, "\t\t<witnessPos x=\"%f\" y=\"%f\" z=\"%f\"/>\n", witnessPos.x, witnessPos.y, witnessPos.z);
	fprintf(snappingReport, "\t\t<witnessDir x=\"%f\" y=\"%f\" z=\"%f\"/>\n", witnessDir.x, witnessDir.y, witnessDir.z);
	fprintf(snappingReport, "\t\t<entityPos0 x=\"%f\" y=\"%f\" z=\"%f\"/>\n", entityPos0.x, entityPos0.y, entityPos0.z);
	fprintf(snappingReport, "\t\t<entityPos1 x=\"%f\" y=\"%f\" z=\"%f\"/>\n", entityPos1.x, entityPos1.y, entityPos1.z);
	fprintf(snappingReport, "\t\t<entityClass name=\"%s\"/>\n", entityCls.c_str());
	fprintf(snappingReport, "\t</record>\n");
}

