#include "tfarmcontroller.h"

#include "tfarmexecutor.h"

#include "tfarmserver.h"

#include "tsystem.h"

#include "tconvert.h"

#include "tfilepath_io.h"

#include "service.h"

#include "tcli.h"

#include "tthreadmessage.h"

#include "tthread.h"

#include "tstream.h"

#include "tlog.h"

#include <qobject></qobject>

#include <qcoreapplication></qcoreapplication>

#include <qeventloop></qeventloop>

#include "tthread.h"

#include <string></string>

using namespace std;

#ifndef WIN32

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

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

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

#endif

int inline STRICMP(const QString &a, const QString &b)

{

	return a.compare(b, Qt::CaseSensitive);

}

int inline STRICMP(const char *a, const char *b)

{

	QString str(a);

	return str.compare(QString(b), Qt::CaseSensitive);

}

/*

#ifdef WIN32

#define STRICMP stricmp

#else

#define STRICMP strcasecmp

#endif

*/

#ifndef WIN32

#define NO_ERROR 0

#endif

//==============================================================================

namespace

{

TFilePath getGlobalRoot()

{

	TFilePath rootDir;

#ifdef WIN32

	TFilePath name(L"SOFTWARE\\OpenToonz\\OpenToonz\\1.0\\FARMROOT");

	rootDir = TFilePath(TSystem::getSystemValue(name).toStdString());

#else

	// Leggo la globalRoot da File txt

	Tifstream is(TFilePath("./OpenToonz_1.0.app/Contents/Resources/configfarmroot.txt"));

	if (is) {

		char line[1024];

		is.getline(line, 80);

		char *s = line;

		while (*s == ' ' || *s == '\t' || *s == '\n' || *s == '\"')

			s++;

		if (*s != '\0') {

			char *t = s;

			while (*t)

				t++;

			string pathName(s, t - 1);

			rootDir = TFilePath(pathName);

		}

	}

#endif

	return rootDir;

}

//--------------------------------------------------------------------

TFilePath getLocalRoot()

{

	TFilePath lroot;

#ifdef WIN32

	TFilePath name(L"SOFTWARE\\OpenToonz\\OpenToonz\\1.0\\TOONZROOT");

	lroot = TFilePath(TSystem::getSystemValue(name).toStdString()) + TFilePath("toonzfarm");

#else

	// Leggo la localRoot da File txt

	Tifstream is(TFilePath("./OpenToonz_1.0.app/Contents/Resources/configfarmroot.txt"));

	if (is) {

		char line[1024];

		is.getline(line, 80);

		char *s = line;

		while (*s == ' ' || *s == '\t' || *s == '\n' || *s == '\"')

			s++;

		if (*s != '\0') {

			char *t = s;

			while (*t)

				t++;

			string pathName(s, t - 1);

			lroot = TFilePath(pathName);

		}

	}

//TFilePath name = TFilePath("TFARMLOCALROOT");

//char *s = getenv(toString(name.getWideString()).c_str());

//lroot = TFilePath(s?s:"");

#endif

	return lroot;

}

//--------------------------------------------------------------------

bool myDoesExists(const TFilePath &fp)

{

	bool exists = false;

#ifdef WIN32

	TFileStatus fs(fp);

	exists = fs.doesExist();

#else

	int acc = access(toString(fp.getWideString()).c_str(), 00); // 00 == solo esistenza

	exists = acc != -1;

#endif

	return exists;

}

//--------------------------------------------------------------------

bool dirExists(const TFilePath &dirFp)

{

	bool exists = false;

#ifdef WIN32

	TFileStatus fs(dirFp);

	exists = fs.isDirectory();

#else

	int acc = access(toString(dirFp.getWideString()).c_str(), 00); // 00 == solo esistenza

	exists = acc != -1;

#endif

	return exists;

}

//-------------------------------------------------------------------

bool loadControllerData(QString &hostName, QString &addr, int &port)

{

	TFilePath groot = getGlobalRoot();

	TFilePath fp = groot + "config" + "controller.txt";

	ControllerData controllerData;

	::loadControllerData(fp, controllerData);

	hostName = controllerData.m_hostName;

	addr = controllerData.m_ipAddress;

	port = controllerData.m_port;

	return true;

}

//-------------------------------------------------------------------

bool isAScript(TFarmTask *task)

{

	return false; //todo per gli script

				  /*

#ifdef WIN32

   return task->m_cmdline.contains(".bat");

#else

  return (task->m_cmdline.contains(".csh")|| 

          task->m_cmdline.contains(".sh")|| 

          task->m_cmdline.contains(".tcsh"))

#endif   

*/

}

} // anonymous namespace

//==============================================================================

class CtrlFarmTask : public TFarmTask

{

public:

	CtrlFarmTask() : m_toBeDeleted(false), m_failureCount(0) {}

	CtrlFarmTask(

		const QString &id, const QString &name, const QString &cmdline,

		const QString &user, const QString &host, int stepCount, int priority)

		: TFarmTask(id, name, cmdline, user, host, stepCount, priority), m_toBeDeleted(false), m_failureCount(0)

	{

		m_id = id;

		m_status = Waiting;

	}

	CtrlFarmTask(const CtrlFarmTask &rhs)

		: TFarmTask(rhs)

	{

		m_serverId = rhs.m_serverId;

		m_subTasks = rhs.m_subTasks;

		m_toBeDeleted = rhs.m_toBeDeleted;

	}

	// TPersist implementation

	void loadData(TIStream &is);

	void saveData(TOStream &os);

	const TPersistDeclaration *getDeclaration() const;

	QString m_serverId;

	vector<qstring> m_subTasks;</qstring>

	bool m_toBeDeleted;

	int m_failureCount;

	vector<qstring> m_failedOnServers;</qstring>

};

namespace

{

class TFarmTaskDeclaration : public TPersistDeclaration

{

public:

	TFarmTaskDeclaration(const string &id)

		: TPersistDeclaration(id) {}

	TPersist *create() const

	{

		return new CtrlFarmTask;

	}

} Declaration("tfarmtask");

}

//------------------------------------------------------------------------------

void CtrlFarmTask::loadData(TIStream &is)

{

	is >> m_name;

	QString cmdline;

	is >> cmdline;

	parseCommandLine(cmdline);

	is >> m_priority;

	is >> m_user;

	is >> m_hostName;

	is >> m_id;

	is >> m_parentId;

	int status;

	is >> status;

	m_status = (TaskState)status;

	is >> m_server;

	QString dateStr;

	is >> dateStr;

	m_submissionDate = QDateTime::fromString(dateStr);

	is >> dateStr;

	m_startDate = QDateTime::fromString(dateStr);

	is >> dateStr;

	m_completionDate = QDateTime::fromString(dateStr);

	is >> m_successfullSteps;

	is >> m_failedSteps;

	is >> m_stepCount;

	is >> m_serverId;

	string tagName;

	while (is.openChild(tagName)) {

		if (tagName == "platform") {

			int plat;

			is >> plat;

			switch (plat) {

			case NoPlatform:

				m_platform = NoPlatform;

				break;

			case Windows:

				m_platform = Windows;

				break;

			case Irix:

				m_platform = Irix;

				break;

			case Linux:

				m_platform = Linux;

				break;

			}

		} else if (tagName == "dependencies") {

			int depCount = 0;

			is >> depCount;

			if (depCount > 0)

				m_dependencies = new Dependencies();

			while (depCount > 0) {

				TFarmTask::Id id;

				is >> id;

				m_dependencies->add(id);

				depCount--;

			}

		}

		is.closeChild();

	}

}

//------------------------------------------------------------------------------

void CtrlFarmTask::saveData(TOStream &os)

{

	os << m_name;

	os << getCommandLine();

	os << m_priority;

	os << m_user;

	os << m_hostName;

	os << m_id;

	os << m_parentId;

	os << (int)m_status;

	os << m_server;

	os << m_submissionDate.toString();

	os << m_startDate.toString();

	os << m_completionDate.toString();

	os << m_successfullSteps;

	os << m_failedSteps;

	os << m_stepCount;

	os << m_serverId;

	os.openChild("platform");

	os << m_platform;

	os.closeChild();

	os.openChild("dependencies");

	if (m_dependencies) {

		int depCount = m_dependencies->getTaskCount();

		os << depCount;

		int i = 0;

		while (i < depCount) {

			TFarmTask::Id id = m_dependencies->getTaskId(i++);

			os << id;

		}

	} else

		os << 0;

	os.closeChild();

}

//------------------------------------------------------------------------------

const TPersistDeclaration *CtrlFarmTask::getDeclaration() const

{

	return &Declaration;

}

//==============================================================================

class FarmServerProxy

{

public:

	FarmServerProxy(const QString &hostName, const QString &addr, int port, int maxTaskCount = 1)

		: m_hostName(hostName), m_addr(addr), m_port(port), m_offline(false), m_attached(false), m_maxTaskCount(maxTaskCount), m_platform(NoPlatform)

	{

		TFarmServerFactory serverFactory;

		serverFactory.create(m_hostName, m_addr, m_port, &m_server);

	}

	~FarmServerProxy() {}

	QString getId() const

	{

		return getIpAddress();

	}

	QString getHostName() const

	{

		return m_hostName;

	}

	QString getIpAddress() const

	{

		return m_addr;

	}

	int getPort() const

	{

		return m_port;

	}

	const vector<qstring> &getTasks() const</qstring>

	{

		return m_tasks;

	}

	int addTask(const CtrlFarmTask *task);

	void terminateTask(const QString &taskId);

	// e' possibile rimuovere un task solo se non running

	void removeTask(const QString &taskId);

	bool testConnection(int timeout);

	void queryHwInfo(TFarmServer::HwInfo &hwInfo)

	{

		m_server->queryHwInfo(hwInfo);

	}

	void attachController(const QString &name, const QString &addr, int port)

	{

		m_server->attachController(name, addr, port);

	}

	void detachController(const QString &name, const QString &addr, int port)

	{

		m_server->detachController(name, addr, port);

	}

	QString m_hostName;

	QString m_addr;

	int m_port;

	bool m_offline;

	bool m_attached;

	int m_maxTaskCount;

	TFarmPlatform m_platform;

	// vettore dei taskId assegnato al server

	vector<qstring> m_tasks;</qstring>

	TFarmServer *m_server;

};

//------------------------------------------------------------------------------

int FarmServerProxy::addTask(const CtrlFarmTask *task)

{

	int rc = m_server->addTask(task->m_id, task->getCommandLine());

	if (rc == 0)

		m_tasks.push_back(task->m_id);

	return rc;

}

//------------------------------------------------------------------------------

void FarmServerProxy::terminateTask(const QString &taskId)

{

	m_server->terminateTask(taskId);

}

//------------------------------------------------------------------------------

void FarmServerProxy::removeTask(const QString &taskId)

{

	vector<qstring>::iterator it = find(m_tasks.begin(), m_tasks.end(), taskId);</qstring>

	if (it != m_tasks.end()) {

		m_tasks.erase(it);

	}

}

//------------------------------------------------------------------------------

bool doTestConnection(const QString &hostName, const QString &addr, int port)

{

	TTcpIpClient client;

	int sock;

	int ret = client.connect(hostName, addr, port, sock);

	if (ret == OK) {

#ifdef WIN32

		closesocket(sock);

#else

		close(sock);

#endif

		return true;

	}

	return false;

}

//------------------------------------------------------------------------------

#ifdef WIN32

class ConnectionTest : public TThread::Runnable

{

public:

	ConnectionTest(const FarmServerProxy *server, HANDLE hEvent)

		: m_server(server), m_hEvent(hEvent) {}

	void run();

	const FarmServerProxy *m_server;

	HANDLE m_hEvent;

};

void ConnectionTest::run()

{

	bool res = doTestConnection(

		m_server->m_hostName, m_server->m_addr, m_server->m_port);

	SetEvent(m_hEvent);

}

#endif

//------------------------------------------------------------------------------

bool FarmServerProxy::testConnection(int timeout)

{

#ifdef WIN32

	HANDLE hEvent = CreateEvent(NULL, TRUE, FALSE, NULL);

	if (!hEvent) {

		// se fallisce la creazione dell'evento ci provo comunque

		// senza timeout

		return doTestConnection(m_hostName, m_addr, m_port);

	}

	TThread::Executor executor;

	executor.addTask(new ConnectionTest(this, hEvent));

	DWORD rr = WaitForSingleObject(hEvent, timeout);

	CloseHandle(hEvent);

	if (rr == WAIT_TIMEOUT)

		return false;

	else

		return true;

#else

	return doTestConnection(m_hostName, m_addr, m_port);

#endif

	TTcpIpClient client;

	int sock;

	int ret = client.connect(m_hostName, m_addr, m_port, sock);

	if (ret == OK) {

#ifdef WIN32

		closesocket(sock);

#else

		close(sock);

#endif

		return true;

	}

	return false;

}

//==============================================================================

class TaskId

{

	int m_id;

	int m_subId;

public:

	TaskId(int id, int subId = -1) : m_id(id), m_subId(m_subId){};

	TaskId(const QString &id)

	{

		int pos = id.indexOf(".");

		if (pos != -1) {

			m_id = id.left(pos).toInt();

			m_subId = id.mid(pos + 1, id.length() - pos).toInt();

		} else {

			m_id = id.toInt();

			m_subId = -1;

		}

	}

	inline bool operator==(const TaskId &f) const { return f.m_id == m_id && f.m_subId == m_subId; };

	inline bool operator!=(const TaskId &f) const { return (m_id != f.m_id || m_subId != f.m_subId); };

	inline bool operator<(const TaskId &f) const { return (m_id < f.m_id || (m_id == f.m_id && m_subId < f.m_subId)); };

	inline bool operator>(const TaskId &f) const { return f < *this; }

	inline bool operator>=(const TaskId &f) const { return !operator<(f); }

	inline bool operator<=(const TaskId &f) const { return !operator>(f); }

	TaskId &operator=(const TaskId &f)

	{

		m_id = f.m_id;

		m_subId = f.m_subId;

		return *this;

	}

	//operator string() const;

	QString toString() const

	{

		QString id(QString::number(m_id));

		if (m_subId >= 0)

			id += "." + ::QString::number(m_subId);

		return id;

	}

};

//==============================================================================

class FarmController : public TFarmExecutor, public TFarmController

{

public:

	FarmController(const QString &hostName, const QString &addr, int port, TUserLog *log);

	void loadServersData(const TFilePath &globalRoot);

	// TFarmExecutor interface implementation

	QString execute(const vector<qstring> &argv);</qstring>

	// TFarmController interface methods implementation

	QString addTask(

		const QString &name, const QString &cmdline,

		const QString &user, const QString &host,

		bool suspended,

		int priority,

		TFarmPlatform platform);

	QString addTask(const TFarmTask &task, bool suspended);

	void removeTask(const QString &id);

	void suspendTask(const QString &id);

	void activateTask(const QString &id);

	void restartTask(const QString &id);

	void getTasks(vector<qstring> &tasks);</qstring>

	void getTasks(const QString &parentId, vector<qstring> &tasks);</qstring>

	void getTasks(const QString &parentId, vector<taskshortinfo> &tasks);</taskshortinfo>

	void queryTaskInfo(const QString &id, TFarmTask &task);

	void queryTaskShortInfo(

		const QString &id,

		QString &parentId,

		QString &name,

		TaskState &status);

	// used (by a server) to notify a server start

	void attachServer(const QString &name, const QString &addr, int port);

	// used (by a server) to notify a server stop

	void detachServer(const QString &name, const QString &addr, int port);

	// used (by a server) to notify a task submission error

	void taskSubmissionError(const QString &taskId, int errCode);

	// used by a server to notify a task progress

	void taskProgress(

		const QString &taskId,

		int step,

		int stepCount,

		int frameNumber,

		FrameState state);

	// used (by a server) to notify a task completion

	void taskCompleted(const QString &taskId, int exitCode);

	// fills the servers vector with the names of the servers

	void getServers(vector<serveridentity> &servers);</serveridentity>

	// returns the state of the server whose id has been specified

	ServerState queryServerState2(const QString &id);

	// fills info with the infoes about the server whose id is specified

	void queryServerInfo(const QString &id, ServerInfo &info);

	// activates the server whose id has been specified

	void activateServer(const QString &id);

	// deactivates the server whose id has been specified

	// once deactivated, a server is not available for task rendering

	void deactivateServer(const QString &id, bool completeRunningTasks);

	// FarmController specific methods

	CtrlFarmTask *doAddTask(

		const QString &id, const QString &parentId,

		const QString &name, const QString &cmdline,

		const QString &user, const QString &host,

		bool suspended, int stepCount, int priority,

		TFarmPlatform platform);

	void startTask(CtrlFarmTask *task, FarmServerProxy *server);

	CtrlFarmTask *getTaskToStart(FarmServerProxy *server = 0);

	CtrlFarmTask *getNextTaskToStart(CtrlFarmTask *task, FarmServerProxy *server);

	// looks for a ready server to which to assign the task

	// returns true iff the task has been started

	bool tryToStartTask(CtrlFarmTask *task);

	ServerState getServerState(FarmServerProxy *server, QString &taskId);

	void initServer(FarmServerProxy *server);

	void load(const TFilePath &fp);

	void save(const TFilePath &fp) const;

	void doRestartTask(const QString &id, bool fromClient, FarmServerProxy *server);

	void activateReadyServers();

	// controller name, address and port

	QString m_hostName;

	QString m_addr;

	int m_port;

	TUserLog *m_userLog;

	map<taskid, *="" ctrlfarmtask=""> m_tasks;</taskid,>

	map<qstring, *="" farmserverproxy=""> m_servers;</qstring,>

	TThread::Mutex m_mutex;

	static int NextTaskId;

};

int FarmController::NextTaskId = 0;

//------------------------------------------------------------------------------

FarmController::FarmController(const QString &hostName, const QString &addr, int port, TUserLog *log)

	: TFarmExecutor(port), m_hostName(hostName), m_addr(addr), m_port(port), m_userLog(log)

{

	TFilePath rootDir = getGlobalRoot();

	TFilePath lastUsedIdFilePath = rootDir + "config" + "id.txt";

	Tifstream is(lastUsedIdFilePath);

	if (is.good())

		is >> NextTaskId;

}

//------------------------------------------------------------------------------

void FarmController::loadServersData(const TFilePath &globalRoot)

{

	TFilePath fp = globalRoot + "config" + "servers.txt";

	Tifstream is(fp);

	while (!is.eof()) {

		char line[80];

		is.getline(line, 80);

		if (line[0] != '#' && QString(line) != "") {

			istrstream iss(line);

			char hostName[512];

			char ipAddr[80];

			int port;

			iss >> hostName >> ipAddr >> port;

			FarmServerProxy *server = new FarmServerProxy(hostName, ipAddr, port);

			m_servers.insert(make_pair(QString(ipAddr), server));

			if (server->testConnection(500)) {

				initServer(server);

				try {

					server->attachController(m_hostName, m_addr, m_port);

				} catch (TException &) {

				}

			}

		}

	}

}

//------------------------------------------------------------------------------

namespace

{

inline QString toString(const TFarmTask &task, int ver)

{

	QString ss = task.m_name + ",";

	ss += task.getCommandLine() + ",";

	ss += QString::number((int)(task.m_priority)) + ",";

	ss += task.m_user + ",";

	ss += task.m_hostName + ",";

	ss += task.m_id + ",";

	ss += task.m_parentId + ",";

	ss += QString::number((int)(task.m_status)) + ",";

	ss += task.m_server + ",";

	ss += task.m_submissionDate.toString() + ",";

	ss += task.m_startDate.toString() + ",";

	ss += task.m_completionDate.toString() + ",";

	ss += QString::number(task.m_successfullSteps) + ",";

	ss += QString::number(task.m_failedSteps) + ",";

	ss += QString::number(task.m_stepCount);

	if (ver == 2) {

		ss += ",";

		ss += QString::number(task.m_platform) + ",";

		int depCount = 0;

		if (task.m_dependencies)

			depCount = task.m_dependencies->getTaskCount();

		ss += QString::number(depCount);

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

			TFarmTask::Id id = task.m_dependencies->getTaskId(i);

			ss += "," + id;

		}

	}

	ss += '\0';

	return ss;

}

inline QString toString(const ServerInfo &info)

{

	QString ss = info.m_name + ",";

	ss += info.m_ipAddress + ",";

	ss += info.m_portNumber + ",";

	ss += QString::number((int)(info.m_state)) + ",";

	ss += info.m_platform + ",";

	ss += QString::number((int)info.m_cpuCount) + ",";

	ss += QString::number((int)info.m_totPhysMem) + ",";

	ss += QString::number((int)info.m_totVirtMem) + ",";

	ss += QString::number((int)info.m_availPhysMem) + ",";

	ss += QString::number((int)info.m_availVirtMem) + ",";

	ss += info.m_currentTaskId + '\0';

	return ss;

}

} // anonymous namespace

//------------------------------------------------------------------------------

QString FarmController::execute(const vector<qstring> &argv)</qstring>

{

	QMutexLocker sl(&m_mutex);

	if (argv.size() > 0) {

#ifdef TRACE

		for (int i = 0; i < argv.size(); i++) {

			m_userLog->info(argv[i]);

		}

		m_userLog->info('\n');

#endif

		if (argv[0] == "addTask@string@string" && argv.size() > 5) {

			// ordine degli argomenti:

			// name, cmdline, user, host, suspended, priority, platform

			bool suspended;

			fromStr((int &)suspended, argv[5]);

			int priority;

			fromStr(priority, argv[6]);

			TFarmPlatform platform;

			fromStr((int &)platform, argv[7]);

			return addTask(argv[1], argv[2], argv[3], argv[4], suspended, priority, platform);

		} else if (argv[0] == "addTask@TFarmTask" && argv.size() > 5) {

			// ordine degli argomenti:

			// name, cmdline, user, host, suspended, priority,

			bool suspended;

			fromStr((int &)suspended, argv[5]);