#include "msg_cpp.h"

extern "C" {
	#include "xbt/log.h"
	#include "xbt/asserts.h"
};


MsgGlobal* MsgGlobal::_instance = NULL;



MsgGlobal::MsgGlobal()
{
}

MsgGlobal::~MsgGlobal()
{
	MSG_clean();
}

MsgGlobal* MsgGlobal::instance()
{
	if(_instance == NULL) {
		_instance = new MsgGlobal;
	}
	return _instance;
}

void MsgGlobal::destroy()
{
	if(_instance != NULL) {
		delete _instance;
		_instance = NULL;
	}
}

void MsgGlobal::createEnvironment(const char *file)
{
	MSG_create_environment(file);
}

void MsgGlobal::config(const char *name, ...)
{
	va_list va;

	va_start(va, name);
	MSG_config(name, va);
}

void MsgGlobal::globalInit(int *argc, char **argv)
{
	MSG_global_init(argc, argv);
}

void MsgGlobal::globalInitArgs(int *argc, char **argv)
{
	MSG_global_init_args(argc, argv);
}

MSG_error_t MsgGlobal::setChannelNumber(int number)
{
	return MSG_set_channel_number(number);
}

int MsgGlobal::getChannelNumber()
{
	return MSG_get_channel_number();
}

MSG_error_t MsgGlobal::main()
{
	return MSG_main();
}

MSG_error_t MsgGlobal::clean()
{
	return MSG_clean();
}

void MsgGlobal::functionRegister(const char *name, m_process_code_t code)
{
	MSG_function_register(name, code);
}

m_process_code_t MsgGlobal::getRegisteredFunction(const char *name)
{
	return MSG_get_registered_function(name);
}

void MsgGlobal::launchApplication(const char *file)
{
	MSG_launch_application(file);
}

void MsgGlobal::pajeOutput(const char *filename)
{
	MSG_paje_output(filename);
}

double MsgGlobal::getClock()
{
	return MSG_get_clock();
}

MSG_error_t MsgGlobal::getErrno()
{
	return MSG_get_errno();
}

int MsgGlobal::killAllProcesses(int resetPIDs)
{
	return MSG_process_killall(resetPIDs);
}

MsgProcessPtr MsgGlobal::getCurrentProcess()
{
	MsgProcessPtr ret;
	m_process_t proc_t;

	proc_t = MSG_process_self();
	
	if(proc_t != NULL) {
		ret = new MsgProcess(proc_t);
	}

	return ret;
}

MSG_error_t MsgGlobal::currentProcessSleep(double nb_sec)
{
	return MSG_process_sleep(nb_sec);
}

MsgHostPtr MsgGlobal::getHostByName(const char *name)
{
	MsgHostPtr host = NULL;
	m_host_t host_t = MSG_get_host_by_name(name);
	if(host_t)
		host = MsgGlobal::instance()->createHost(host_t);
	return host;
}

MsgProcessPtr MsgGlobal::processFromPID(int pid)
{
	MsgProcessPtr ret;
	m_process_t proc_t;

	proc_t = MSG_process_from_PID(pid);
	
	if(proc_t != NULL) {
		ret = new MsgProcess(proc_t);
	}

	return ret;
}

MsgTaskPtr MsgGlobal::getTaskFromCurrentProcess(m_channel_t channel, double max_duration)
{
	MsgTaskPtr ret;
	MSG_error_t e;
	m_task_t task_t;
	
	e = MSG_task_get(&task_t, channel);

	if(e == MSG_OK) {
		ret = new MsgTask(task_t);
	}

	return ret;
}

/**
\brief Test whether there is a pending communication on a channel.

It takes one parameter.
\param channel the channel on which the agent should be
listening. This value has to be >=0 and < than the maximal
number of channels fixed with MsgSimulation::setChannelNumber.
\return true if there is a pending communication and false otherwise
*/
bool MsgGlobal::currentHostPendingComm(m_channel_t channel)
{
	return (MSG_task_Iprobe(channel) != 0);
}

/**
\brief Test whether there is a pending communication on a channel, and who sent it.

It takes one parameter.
\param channel the channel on which the agent should be
listening. This value has to be >=0 and < than the maximal
number of channels fixed with MsgSimulation::setChannelNumber.
\return -1 if there is no pending communication and the PID of the process who sent it otherwise
*/
int MsgGlobal::currentHostPendingCommPID(m_channel_t channel)
{
	return MSG_task_probe_from(channel);
}

/**
\brief Wait for at most \a max_duration second for a task reception
on \a channel. *\a PID is updated with the PID of the first process
that triggered this event if any.

 It takes three parameters:
 \param channel the channel on which the agent should be
 listening. This value has to be >=0 and < than the maximal.
 number of channels fixed with MsgSimulation::setChannelNumber.
 \param PID a memory location for storing an int.
 \param max_duration the maximum time to wait for a task before
giving up. In the case of a reception, *\a PID will be updated
with the PID of the first process to send a task.
\return #MSG_HOST_FAILURE if the host is shut down in the meantime
and #MSG_OK otherwise.
*/
MSG_error_t MsgGlobal::currentHostPendingCommPID(m_channel_t channel, double max_duration, int *PID)
{
	return MSG_channel_select_from(channel, max_duration, PID);
}

/**
\brief Return the number of tasks waiting to be received on a \a
channel and sent by \a host.

It takes two parameters.
\param channel the channel on which the agent should be
listening. This value has to be >=0 and < than the maximal
number of channels fixed with MsgSimulation::setChannelNumber.
\param host the host that is to be watched.
\return the number of tasks waiting to be received on \a channel
and sent by \a host.
*/
int MsgGlobal::currentHostPendingCommFromHost(m_channel_t channel, MsgHostPtr host)
{
	return MSG_task_probe_from_host(channel, host->getObject());
}

MsgTaskPtr MsgGlobal::createTask(const char *name, double compute_duration, double message_size, void *data)
{
	MsgTaskPtr ret;
	m_task_t task_t;
	
	task_t = MSG_task_create(name, compute_duration, message_size, data);

	if(task_t != NULL) {
		ret = createTask(task_t);
	}

	return ret;
}

MsgTaskPtr MsgGlobal::createParallelTask(const char *name, int host_nb, const MsgHostPtr *host_list,
										 double *computation_amount, double *communication_amount,
										 void *data)
{
	MsgTaskPtr ret;
	m_task_t task_t;
	m_host_t *host_t_list;
	int i;
	
	xbt_assert(host_nb != 0);

	host_t_list = (m_host_t *) malloc(sizeof(m_host_t) * host_nb);
	
	for(i=0; i<host_nb; i++) {
		host_t_list[i] = host_list[i]->getObject();
	}

	task_t = MSG_parallel_task_create(name, host_nb, host_t_list, computation_amount, communication_amount, data);

	free(host_t_list);

	if(task_t != NULL) {
		ret = createTask(task_t);
	}

	return ret;
}

MsgProcessPtr MsgGlobal::createProcess(const char *name, m_process_code_t code, void *data,
	  									MsgHostPtr host, int argc, char **argv)
{
	MsgProcessPtr ret;
	m_process_t proc_t;
	
	if(argc == 0 || argv == NULL) {
		proc_t = MSG_process_create(name, code, data, host->getObject());
	}
	else {
		proc_t = MSG_process_create_with_arguments(name, code, data, host->getObject(), argc, argv);
	}

	if(proc_t != NULL) {
		ret = createProcess(proc_t);
	}

	return ret;
}

MsgHostPtr MsgGlobal::createHost(m_host_t host_t)
{
	MsgHostPtr ret = new MsgHost(host_t);
	return ret;
}

MsgProcessPtr MsgGlobal::createProcess(m_process_t proc_t)
{
	MsgProcessPtr ret = new MsgProcess(proc_t);
	return ret;
}

MsgTaskPtr MsgGlobal::createTask(m_task_t task_t)
{
	MsgTaskPtr ret = new MsgTask(task_t);
	return ret;
}


///////////////////////////////////////////////////////////////////////////////////////////////////////////
MsgHost::MsgHost(m_host_t host_t)
{
	_host_t = host_t;
}

MSG_error_t MsgHost::setData(void *data)
{
	return MSG_host_set_data(_host_t, data);
}

void *MsgHost::getData()
{
	return MSG_host_get_data(_host_t);
}

const char *MsgHost::getName()
{
	return MSG_host_get_name(_host_t);
}

int MsgHost::getMsgload()
{
	return MSG_get_host_msgload(_host_t);
}

double MsgHost::getSpeed()
{
	return MSG_get_host_speed(_host_t);
}

int MsgHost::isAvail()
{
	return MSG_host_is_avail(_host_t);
}

/**
\brief Listen on \a channel and waits for receiving a task from \a host.

It takes three parameters.
\param channel the channel on which the agent should be
listening. This value has to be >=0 and < than the maximal
number of channels fixed with MsgSimulation::setChannelNumber.
\return NULL on error.
*/
MsgTaskPtr MsgHost::getTask(m_channel_t channel)
{
	m_task_t task_t;
	MSG_error_t e;
	MsgTaskPtr ret;

	e = MSG_task_get(&task_t, channel);
	if(e == MSG_OK) {
		ret = MsgGlobal::instance()->createTask(task_t);
	}

	return ret;
}

MSG_error_t MsgHost::putTask(MsgTaskPtr task, m_channel_t channel)
{
	return MSG_task_put(task->getObject(), _host_t, channel);
}

MSG_error_t MsgHost::putTaskBounded(MsgTaskPtr task, m_channel_t channel, double max_rate)
{
	return MSG_task_put_bounded(task->getObject(), _host_t, channel, max_rate);
}

/////////////////////////////////////////////////////////////////////////////////////////////////////////
MsgProcess::MsgProcess(m_process_t proc_t)
{
	_proc_t = proc_t;
}

void MsgProcess::kill()
{
	MSG_process_kill(_proc_t);
}

MSG_error_t MsgProcess::changeHost(MsgHostPtr host)
{
	return MSG_process_change_host(_proc_t, host->getObject());
}

void* MsgProcess::getData()
{
	return MSG_process_get_data(_proc_t);
}

MSG_error_t MsgProcess::setData(void *data)
{
	return MSG_process_set_data(_proc_t, data);
}

MsgHostPtr MsgProcess::getHost()
{
	MsgHostPtr ret;
	m_host_t host_t;

	host_t = MSG_process_get_host(_proc_t);
	
	if(host_t != NULL) {
		ret = MsgGlobal::instance()->createHost(host_t);
	}

	return ret;
}

int MsgProcess::getPID()
{
	return MSG_process_get_PID(_proc_t);
}

int MsgProcess::getPPID()
{
	return MSG_process_get_PPID(_proc_t);
}

const char *MsgProcess::getName()
{
	return MSG_process_get_name(_proc_t);
}

MSG_error_t MsgProcess::suspend()
{
	return MSG_process_suspend(_proc_t);
}

MSG_error_t MsgProcess::resume()
{
	return MSG_process_resume(_proc_t);
}

bool MsgProcess::isSuspended()
{
	return (MSG_process_is_suspended(_proc_t) != 0);
}

////////////////////////////////////////////////////////////////////////////////////////////
MsgTask::MsgTask(m_task_t task_t)
{
	_task_t = task_t;
}

void *MsgTask::getData()
{
	return MSG_task_get_data(_task_t);
}

MsgProcessPtr MsgTask::getSender()
{
	MsgProcessPtr ret;
	m_process_t proc_t;

	proc_t = MSG_task_get_sender(_task_t);
	
	if(proc_t != NULL) {
		ret = new MsgProcess(proc_t);
	}

	return ret;
}

MsgHostPtr MsgTask::getSource()
{
	MsgHostPtr ret;
	m_host_t host_t;

	host_t = MSG_task_get_source(_task_t);
	
	if(host_t != NULL) {
		ret = new MsgHost(host_t);
	}

	return ret;
}

const char *MsgTask::getName()
{
	return MSG_task_get_name(_task_t);
}

MSG_error_t MsgTask::cancel()
{
	return MSG_task_cancel(_task_t);
}

MSG_error_t MsgTask::destroy()
{
	return MSG_task_destroy(_task_t);
}

MSG_error_t MsgTask::put(MsgHostPtr dest, m_channel_t channel, double max_duration)
{
	MSG_error_t ret;

	if(max_duration == MSG_INFINITE) {
		ret = MSG_task_put(_task_t, dest->getObject(), channel);
	}
	else {
		ret = MSG_task_put_with_timeout(_task_t, dest->getObject(), channel, max_duration);
	}

	return ret;
}

MSG_error_t MsgTask::putBounded(MsgHostPtr dest, m_channel_t channel, double max_rate)
{
	return MSG_task_put_bounded(_task_t, dest->getObject(), channel, max_rate);
}

MSG_error_t MsgTask::execute()
{
	return MSG_task_execute(_task_t);
}

MSG_error_t MsgTask::parallelExecute()
{
	return MSG_parallel_task_execute(_task_t);
}

void MsgTask::setPriority(double priority)
{
	MSG_task_set_priority(_task_t, priority);
}

double MsgTask::getComputeDuration()
{
	return MSG_task_get_compute_duration(_task_t);
}

double MsgTask::getRemainingComputation()
{
	return MSG_task_get_remaining_computation(_task_t);
}

double MsgTask::getDataSize()
{
	return MSG_task_get_data_size(_task_t);
}
