starpu-max/starpupy/src/starpu_task_wrapper.c

/* StarPU --- Runtime system for heterogeneous multicore architectures.
 *
 * Copyright (C) 2020-2021 Université de Bordeaux, CNRS (LaBRI UMR 5800), Inria
 *
 * StarPU is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation; either version 2.1 of the License, or (at
 * your option) any later version.
 *
 * StarPU is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 *
 * See the GNU Lesser General Public License in COPYING.LGPL for more details.
 */
#undef NDEBUG
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <starpu.h>

#define PY_SSIZE_T_CLEAN
#include <Python.h>

#ifdef STARPU_PYTHON_HAVE_NUMPY
#include <numpy/arrayobject.h>
#endif

/*********************Functions passed in task_submit wrapper***********************/

static PyObject *StarpupyError; /*starpupy error exception*/
static PyObject *asyncio_module; /*python asyncio module*/

static char* starpu_cloudpickle_dumps(PyObject *obj, PyObject **obj_bytes, Py_ssize_t *obj_data_size)
{
	PyObject *cloudpickle_module = PyImport_ImportModule("cloudpickle");
	if (cloudpickle_module == NULL)
	{
		printf("can't find cloudpickle module\n");
		exit(1);
	}
	PyObject *dumps = PyObject_GetAttrString(cloudpickle_module, "dumps");
	*obj_bytes= PyObject_CallFunctionObjArgs(dumps, obj, NULL);

	char* obj_data;
	PyBytes_AsStringAndSize(*obj_bytes, &obj_data, obj_data_size);

	return obj_data;
}

static PyObject* starpu_cloudpickle_loads(char* pyString, Py_ssize_t pyString_size)
{
	PyObject *pickle_module = PyImport_ImportModule("pickle");
	if (pickle_module == NULL)
	{
		printf("can't find pickle module\n");
		exit(1);
	}
	PyObject *loads = PyObject_GetAttrString(pickle_module, "loads");
	PyObject *obj_bytes_str = PyBytes_FromStringAndSize(pyString, pyString_size);
	PyObject *obj = PyObject_CallFunctionObjArgs(loads, obj_bytes_str, NULL);

	Py_DECREF(obj_bytes_str);

	return obj;
}

/* prologue_callback_func*/
void prologue_cb_func(void *cl_arg)
{
	PyObject *func_data;
	size_t func_data_size;
	PyObject *argList;
	PyObject *fut;
	PyObject *loop;
	int sb;

	/*make sure we own the GIL*/
	PyGILState_STATE state = PyGILState_Ensure();

	struct starpu_task *task = starpu_task_get_current();
	/*Initialize struct starpu_codelet_unpack_arg_data*/
	struct starpu_codelet_pack_arg_data data_org;
	starpu_codelet_unpack_arg_init(&data_org, &task->cl_arg, &task->cl_arg_size);

	/*get func_py char**/
	starpu_codelet_pick_arg(&data_org, (void**)&func_data, &func_data_size);
	/*get argList*/
	starpu_codelet_unpack_arg(&data_org, &argList, sizeof(argList));
	/*get fut*/
	starpu_codelet_unpack_arg(&data_org, &fut, sizeof(fut));
	/*get loop*/
	starpu_codelet_unpack_arg(&data_org, &loop, sizeof(loop));
	/*get sb*/
	starpu_codelet_unpack_arg(&data_org, &sb, sizeof(sb));

	/*Repack the data*/
	/*Initialize struct starpu_codelet_pack_arg_data*/
	struct starpu_codelet_pack_arg_data data;
	starpu_codelet_pack_arg_init(&data);

	/*repack func_data*/
	starpu_codelet_pack_arg(&data, func_data, func_data_size);

	/*check if there is Future in argList, if so, get the Future result*/
	int i;
	for(i=0; i < PyTuple_Size(argList); i++)
	{
		PyObject *obj=PyTuple_GetItem(argList, i);
		const char* tp = Py_TYPE(obj)->tp_name;
		if(strcmp(tp, "_asyncio.Future") == 0)
		{
			PyObject *done =  PyObject_CallMethod(obj, "done", NULL);
			/*if the future object is not finished, we will await it for the result*/
			if (!PyObject_IsTrue(done))
			{
				PyObject *pModule = PyImport_ImportModule("starpu");
				PyObject *pDict = PyModule_GetDict(pModule);
				/*call the method wait_for_fut to await obj*/
				PyObject *wait_method=PyDict_GetItemString(pDict, "wait_for_fut");
				/*call wait_for_fut(obj)*/
				PyObject *wait_obj = PyObject_CallFunctionObjArgs(wait_method, obj, NULL);
				/*call obj = asyncio.run_coroutine_threadsafe(wait_for_fut(obj), loop)*/
				obj = PyObject_CallMethod(asyncio_module, "run_coroutine_threadsafe", "O,O", wait_obj, loop);
			}
		    
			/*if one of arguments is Future, get its result*/
			PyObject *fut_result = PyObject_CallMethod(obj, "result", NULL);
			/*replace the Future argument to its result*/
			PyTuple_SetItem(argList, i, fut_result);
		}
	}

	/*use cloudpickle to dump dumps argList*/
	Py_ssize_t arg_data_size;
	PyObject *arg_bytes;
	char* arg_data = starpu_cloudpickle_dumps(argList, &arg_bytes, &arg_data_size);
	starpu_codelet_pack_arg(&data, arg_data, arg_data_size);
	Py_DECREF(arg_bytes);
	/*repack fut*/
	starpu_codelet_pack_arg(&data, &fut, sizeof(fut));
	/*repack loop*/
	starpu_codelet_pack_arg(&data, &loop, sizeof(loop));
	/*repack sb*/
	starpu_codelet_pack_arg(&data, &sb, sizeof(sb));
	/*finish repacking data and store the struct in cl_arg*/
	starpu_codelet_pack_arg_fini(&data, &task->cl_arg, &task->cl_arg_size);

	/*restore previous GIL state*/
	PyGILState_Release(state);
}

/*function passed to starpu_codelet.cpu_func*/
void starpupy_codelet_func(void *buffers[], void *cl_arg)
{
	char* func_data;
	size_t func_data_size;
	PyObject *func_py; /*the python function passed in*/
	char* arg_data;
	size_t arg_data_size;
	PyObject *argList; /*argument list of python function passed in*/

	/*make sure we own the GIL*/
	PyGILState_STATE state = PyGILState_Ensure();

	//struct codelet_args *cst = (struct codelet_args*) cl_arg;

	struct starpu_task *task = starpu_task_get_current();
	/*Initialize struct starpu_codelet_unpack_arg_data*/
	struct starpu_codelet_pack_arg_data data;
	starpu_codelet_unpack_arg_init(&data, &task->cl_arg, &task->cl_arg_size);

	/*get func_py char**/
	starpu_codelet_pick_arg(&data, (void**)&func_data, &func_data_size);
	/*get argList char**/
	starpu_codelet_pick_arg(&data, (void**)&arg_data, &arg_data_size);
	/*skip fut*/
	starpu_codelet_unpack_discard_arg(&data);
	/*skip loop*/
	starpu_codelet_unpack_discard_arg(&data);
	/*skip sb*/
	starpu_codelet_unpack_discard_arg(&data);

	/*use cloudpickle to load function (maybe only function name)*/
	PyObject *pFunc=starpu_cloudpickle_loads(func_data, func_data_size);

	/* if the function name is passed in*/
	const char* tp_func = Py_TYPE(pFunc)->tp_name;
	if (strcmp(tp_func, "str")==0)
	{
		/*getattr(sys.modules[__name__], "<functionname>")*/
		/*get sys.modules*/
		PyObject *sys_modules = PyImport_GetModuleDict();
		/*get sys.modules[__name__]*/
		PyObject *sys_modules_name=PyDict_GetItemString(sys_modules,"__main__");
		/*get function object*/
		func_py=PyObject_GetAttr(sys_modules_name,pFunc);
	}
	else
	{
		func_py=pFunc;
	}

	/*use cloudpickle to load argList*/
	argList=starpu_cloudpickle_loads(arg_data, arg_data_size);

	/*verify that the function is a proper callable*/
	if (!PyCallable_Check(func_py))
	{
		printf("py_callback: expected a callable function\n");
		exit(1);
	}

	/*call the python function get the return value rv*/
	PyObject *rv = PyObject_CallObject(func_py, argList);

	/*Initialize struct starpu_codelet_pack_arg_data for return value*/
	struct starpu_codelet_pack_arg_data data_ret;
	starpu_codelet_pack_arg_init(&data_ret);

	/*if the result is None type, pack NULL without using cloudpickle*/
	if (rv==Py_None)
	{
		char* rv_data=NULL;
		Py_ssize_t rv_data_size=0;
		starpu_codelet_pack_arg(&data_ret, &rv_data_size, sizeof(rv_data_size));
	    starpu_codelet_pack_arg(&data_ret, &rv_data, sizeof(rv_data));
	}
	/*else use cloudpickle to dump rv*/
	else
	{
		Py_ssize_t rv_data_size;
		PyObject *rv_bytes;
		char* rv_data = starpu_cloudpickle_dumps(rv, &rv_bytes, &rv_data_size);
		starpu_codelet_pack_arg(&data_ret, &rv_data_size, sizeof(rv_data_size));
		starpu_codelet_pack_arg(&data_ret, rv_data, rv_data_size);
		Py_DECREF(rv_bytes);
	}

	/*store the return value in task_>cl_ret*/
	starpu_codelet_pack_arg_fini(&data_ret, &task->cl_ret, &task->cl_ret_size);

	Py_DECREF(func_py);
	Py_DECREF(argList);

	/*restore previous GIL state*/
	PyGILState_Release(state);
}

/*function passed to starpu_task.epilogue_callback_func*/
void epilogue_cb_func(void *v)
{
	PyObject *fut; /*asyncio.Future*/
	PyObject *loop; /*asyncio.Eventloop*/
	char* rv_data;
	size_t rv_data_size;
	PyObject *rv; /*return value when using PyObject_CallObject call the function f*/

	/*make sure we own the GIL*/
	PyGILState_STATE state = PyGILState_Ensure();

	struct starpu_task *task = starpu_task_get_current();

	/*Initialize struct starpu_codelet_unpack_arg_data data*/
	struct starpu_codelet_pack_arg_data data;
	starpu_codelet_unpack_arg_init(&data, &task->cl_arg, &task->cl_arg_size);

	/*skip func_py*/
	starpu_codelet_unpack_discard_arg(&data);
	/*skip argList*/
	starpu_codelet_unpack_discard_arg(&data);
	/*get fut*/
	starpu_codelet_unpack_arg(&data, &fut, sizeof(fut));
	/*get loop*/
	starpu_codelet_unpack_arg(&data, &loop, sizeof(loop));
	/*skip sb*/
	starpu_codelet_unpack_discard_arg(&data);

	/*Initialize struct starpu_codelet_unpack_arg_data data*/
	struct starpu_codelet_pack_arg_data data_ret;
	starpu_codelet_unpack_arg_init(&data_ret, &task->cl_ret, &task->cl_ret_size);
	/*get rv_data_size*/
	starpu_codelet_unpack_arg(&data_ret, &rv_data_size, sizeof(rv_data_size));

	/*if the rv_data_size is 0, the result is None type*/
	if (rv_data_size==0)
	{
		starpu_codelet_unpack_discard_arg(&data_ret);
		rv=Py_None;
	}
	/*else use cloudpickle to load rv*/
	else
	{
		starpu_codelet_pick_arg(&data_ret, (void**)&rv_data, &rv_data_size);
		rv=starpu_cloudpickle_loads(rv_data, rv_data_size);
	}

	/*set the Future result and mark the Future as done*/
	PyObject *set_result = PyObject_GetAttrString(fut, "set_result");
	PyObject *loop_callback = PyObject_CallMethod(loop, "call_soon_threadsafe", "(O,O)", set_result, rv);

	Py_DECREF(loop_callback);
	Py_DECREF(set_result);
	Py_DECREF(rv);
	Py_DECREF(fut);
	Py_DECREF(loop);

	struct starpu_codelet *func_cl=(struct starpu_codelet *) task->cl;
	if (func_cl->model != NULL)
	{
		struct starpu_perfmodel *perf =(struct starpu_perfmodel *) func_cl->model;
		PyObject *perfmodel=PyCapsule_New(perf, "Perf", 0);
		Py_DECREF(perfmodel);
	}

	/*restore previous GIL state*/
	PyGILState_Release(state);
}

void cb_func(void *v)
{
	struct starpu_task *task = starpu_task_get_current();

	/*deallocate task*/
	free(task->cl);
	free(task->cl_arg);
}

/***********************************************************************************/
/*PyObject*->struct starpu_task**/
static struct starpu_task *PyTask_AsTask(PyObject *obj)
{
	return (struct starpu_task *) PyCapsule_GetPointer(obj, "Task");
}

/* destructor function for task */
static void del_Task(PyObject *obj)
{
	struct starpu_task *obj_task=PyTask_AsTask(obj);
	obj_task->destroy=1; /*XXX we should call starpu task destroy*/
}

/*struct starpu_task*->PyObject**/
static PyObject *PyTask_FromTask(struct starpu_task *task)
{
	return PyCapsule_New(task, "Task", del_Task);
}

/***********************************************************************************/
static size_t sizebase (struct starpu_task *task, unsigned nimpl)
{
	int sb;

	/*Initialize struct starpu_codelet_unpack_arg_data*/
	struct starpu_codelet_pack_arg_data data;
	starpu_codelet_unpack_arg_init(&data, &task->cl_arg, &task->cl_arg_size);

	/*skip func_py*/
	//starpu_codelet_unpack_discard_arg(&data);
	starpu_codelet_unpack_discard_arg(&data);
	/*skip argList*/
	//starpu_codelet_unpack_discard_arg(&data);
	starpu_codelet_unpack_discard_arg(&data);
	/*skip fut*/
	starpu_codelet_unpack_discard_arg(&data);
	/*skip loop*/
	starpu_codelet_unpack_discard_arg(&data);
	/*get sb*/
	starpu_codelet_unpack_arg(&data, &sb, sizeof(sb));

	return sb;
}

/*initialization of perfmodel*/
static PyObject* init_perfmodel(PyObject *self, PyObject *args)
{
	char *sym;

	if (!PyArg_ParseTuple(args, "s", &sym))
		return NULL;

	/*allocate a perfmodel structure*/
	struct starpu_perfmodel *perf=(struct starpu_perfmodel*)calloc(1, sizeof(struct starpu_perfmodel));

	/*get the perfmodel symbol*/
	char *p =strdup(sym);
	perf->symbol=p;
	perf->type=STARPU_HISTORY_BASED;

	/*struct perfmodel*->PyObject**/
	PyObject *perfmodel=PyCapsule_New(perf, "Perf", NULL);

	return perfmodel;
}

/*free perfmodel*/
static PyObject* free_perfmodel(PyObject *self, PyObject *args)
{
	PyObject *perfmodel;
	if (!PyArg_ParseTuple(args, "O", &perfmodel))
		return NULL;

	/*PyObject*->struct perfmodel**/
	struct starpu_perfmodel *perf=PyCapsule_GetPointer(perfmodel, "Perf");

	starpu_save_history_based_model(perf);
	//starpu_perfmodel_unload_model(perf);
	//free(perf->symbol);
	starpu_perfmodel_deinit(perf);
	free(perf);

	/*return type is void*/
	Py_INCREF(Py_None);
	return Py_None;
}

static PyObject* starpu_save_history_based_model_wrapper(PyObject *self, PyObject *args)
{
	PyObject *perfmodel;
	if (!PyArg_ParseTuple(args, "O", &perfmodel))
		return NULL;

	/*PyObject*->struct perfmodel**/
	struct starpu_perfmodel *perf=PyCapsule_GetPointer(perfmodel, "Perf");

	starpu_save_history_based_model(perf);

	/*return type is void*/
	Py_INCREF(Py_None);
	return Py_None;
}

/*****************************Wrappers of StarPU methods****************************/
/*wrapper submit method*/
static PyObject* starpu_task_submit_wrapper(PyObject *self, PyObject *args)
{
	/*get the running Event loop*/
	PyObject *loop = PyObject_CallMethod(asyncio_module, "get_running_loop", NULL);
	/*create a asyncio.Future object*/
	PyObject *fut = PyObject_CallMethod(loop, "create_future", NULL);

	/*first argument in args is always the python function passed in*/
	PyObject *func_py = PyTuple_GetItem(args, 0);

	Py_INCREF(fut);
	Py_INCREF(loop);
	Py_INCREF(func_py);

	/*allocate a task structure and initialize it with default values*/
	struct starpu_task *task=starpu_task_create();
	task->destroy=0;

	PyObject *PyTask=PyTask_FromTask(task);

	/*set one of fut attribute to the task pointer*/
	PyObject_SetAttrString(fut, "starpu_task", PyTask);
	/*check the arguments of python function passed in*/
	int i;
	for(i=1; i < PyTuple_Size(args)-1; i++)
	{
		PyObject *obj=PyTuple_GetItem(args, i);
		const char* tp = Py_TYPE(obj)->tp_name;
		if(strcmp(tp, "_asyncio.Future") == 0)
		{
			/*if one of arguments is Future, get its corresponding task*/
			PyObject *fut_task=PyObject_GetAttrString(obj, "starpu_task");
			/*declare task dependencies between the current task and the corresponding task of Future argument*/
			starpu_task_declare_deps(task, 1, PyTask_AsTask(fut_task));

			Py_DECREF(fut_task);
		}
	}

	/*allocate a codelet structure*/
	struct starpu_codelet *func_cl=(struct starpu_codelet*)malloc(sizeof(struct starpu_codelet));
	/*initialize func_cl with default values*/
	starpu_codelet_init(func_cl);
	func_cl->cpu_funcs[0]=&starpupy_codelet_func;
	func_cl->cpu_funcs_name[0]="starpupy_codelet_func";

	/*check whether the option perfmodel is None*/
	PyObject *dict_option = PyTuple_GetItem(args, PyTuple_Size(args)-1);/*the last argument is the option dictionary*/
	PyObject *perfmodel = PyDict_GetItemString(dict_option, "perfmodel");
	const char *tp_perf = Py_TYPE(perfmodel)->tp_name;
	if (strcmp(tp_perf, "PyCapsule")==0)
	{
		/*PyObject*->struct perfmodel**/
		struct starpu_perfmodel *perf=PyCapsule_GetPointer(perfmodel, "Perf");
		func_cl->model=perf;
		Py_INCREF(perfmodel);
	}

	/*Initialize struct starpu_codelet_pack_arg_data*/
	struct starpu_codelet_pack_arg_data data;
	starpu_codelet_pack_arg_init(&data);

	/*argument list of python function passed in*/
	PyObject *argList;

	/*pass args in argList*/
	if (PyTuple_Size(args)==2)/*function no arguments*/
		argList = PyTuple_New(0);
	else
	{
		/*function has arguments*/
		argList = PyTuple_New(PyTuple_Size(args)-2);
		int i;
		for(i=0; i < PyTuple_Size(args)-2; i++)
		{
			PyObject *tmp=PyTuple_GetItem(args, i+1);
			PyTuple_SetItem(argList, i, tmp);
			Py_INCREF(PyTuple_GetItem(argList, i));
		}
	}

	/*use cloudpickle to dump func_py*/
	Py_ssize_t func_data_size;
	PyObject *func_bytes;
	char* func_data = starpu_cloudpickle_dumps(func_py, &func_bytes, &func_data_size);
    starpu_codelet_pack_arg(&data, func_data, func_data_size);
    Py_DECREF(func_bytes);
    /*pack argList*/
	starpu_codelet_pack_arg(&data, &argList, sizeof(argList));
	/*pack fut*/
	starpu_codelet_pack_arg(&data, &fut, sizeof(fut));
	/*pack loop*/
	starpu_codelet_pack_arg(&data, &loop, sizeof(loop));

	task->cl=func_cl;

	/*pass optional values name=None, synchronous=1, priority=0, color=None, flops=None, perfmodel=None, sizebase=0*/
	/*const char * name*/
	PyObject *PyName = PyDict_GetItemString(dict_option, "name");
	if (PyName!=Py_None)
	{
		const char* name_str = PyUnicode_AsUTF8(PyName);
		char* name = strdup(name_str);
		//printf("name is %s\n", name);
		task->name=name;
	}

	/*unsigned synchronous:1*/
	PyObject *PySync = PyDict_GetItemString(dict_option, "synchronous");
	unsigned sync=PyLong_AsUnsignedLong(PySync);
	//printf("sync is %u\n", sync);
	task->synchronous=sync;

	/*int priority*/
	PyObject *PyPrio = PyDict_GetItemString(dict_option, "priority");
	int prio=PyLong_AsLong(PyPrio);
	//printf("prio is %d\n", prio);
	task->priority=prio;

	/*unsigned color*/
	PyObject *PyColor = PyDict_GetItemString(dict_option, "color");
	if (PyColor!=Py_None)
	{
		unsigned color=PyLong_AsUnsignedLong(PyColor);
		//printf("color is %u\n", color);
		task->color=color;
	}

	/*double flops*/
	PyObject *PyFlops = PyDict_GetItemString(dict_option, "flops");
	if (PyFlops!=Py_None)
	{
		double flops=PyFloat_AsDouble(PyFlops);
		//printf("flops is %f\n", flops);
		task->flops=flops;
	}

	/*int sizebase*/
	PyObject *PySB = PyDict_GetItemString(dict_option, "sizebase");
	int sb=PyLong_AsLong(PySB);
	//printf("pack sizebase is %d\n", sb);
	/*pack sb*/
	starpu_codelet_pack_arg(&data, &sb, sizeof(sb));

	/*finish packing data and store the struct in cl_arg*/
	starpu_codelet_pack_arg_fini(&data, &task->cl_arg, &task->cl_arg_size);

	task->prologue_callback_func=&prologue_cb_func;
	task->epilogue_callback_func=&epilogue_cb_func;
	task->callback_func=&cb_func;

	/*call starpu_task_submit method*/
	int ret;
	Py_BEGIN_ALLOW_THREADS
	ret = starpu_task_submit(task);
	Py_END_ALLOW_THREADS
	if (ret!=0)
	{
		PyErr_Format(StarpupyError, "Unexpected value %d returned for starpu_task_submit", ret);
		return NULL;
	}

	if (strcmp(tp_perf, "PyCapsule")==0)
	{
		struct starpu_perfmodel *perf =(struct starpu_perfmodel *) func_cl->model;
		perf->size_base=&sizebase;
	}

	//printf("the number of reference is %ld\n", Py_REFCNT(func_py));
	//_Py_PrintReferences(stderr);
	//COUNTREFS();
	return fut;
}

/*wrapper wait for all method*/
static PyObject* starpu_task_wait_for_all_wrapper(PyObject *self, PyObject *args)
{
	/*call starpu_task_wait_for_all method*/
	Py_BEGIN_ALLOW_THREADS
		starpu_task_wait_for_all();
	Py_END_ALLOW_THREADS

	/*return type is void*/
	Py_INCREF(Py_None);
	return Py_None;
}

/*wrapper pause method*/
static PyObject* starpu_pause_wrapper(PyObject *self, PyObject *args)
{
	/*call starpu_pause method*/
	starpu_pause();

	/*return type is void*/
	Py_INCREF(Py_None);
	return Py_None;
}

/*wrapper resume method*/
static PyObject* starpu_resume_wrapper(PyObject *self, PyObject *args)
{
	/*call starpu_resume method*/
	starpu_resume();

	/*return type is void*/
	Py_INCREF(Py_None);
	return Py_None;
}

/*wrapper get count cpu method*/
static PyObject* starpu_cpu_worker_get_count_wrapper(PyObject *self, PyObject *args)
{
	/*call starpu_cpu_worker_get_count method*/
	int num_cpu=starpu_cpu_worker_get_count();

	/*return type is unsigned*/
	return Py_BuildValue("I", num_cpu);
}

/*wrapper get min priority method*/
static PyObject* starpu_sched_get_min_priority_wrapper(PyObject *self, PyObject *args)
{
	/*call starpu_sched_get_min_priority*/
	int min_prio=starpu_sched_get_min_priority();

	/*return type is int*/
	return Py_BuildValue("i", min_prio);
}

/*wrapper get max priority method*/
static PyObject* starpu_sched_get_max_priority_wrapper(PyObject *self, PyObject *args)
{
	/*call starpu_sched_get_max_priority*/
	int max_prio=starpu_sched_get_max_priority();

	/*return type is int*/
	return Py_BuildValue("i", max_prio);
}

/*wrapper get the number of no completed submitted tasks method*/
static PyObject* starpu_task_nsubmitted_wrapper(PyObject *self, PyObject *args)
{
	/*call starpu_task_nsubmitted*/
	int num_task=starpu_task_nsubmitted();

	/*Return the number of submitted tasks which have not completed yet */
	return Py_BuildValue("i", num_task);
}

/*wrapper shutdown method*/
static PyObject* starpu_shutdown_wrapper(PyObject *self, PyObject *args)
{
	/*call starpu_shutdown method*/
	Py_BEGIN_ALLOW_THREADS
	starpu_shutdown();
	Py_END_ALLOW_THREADS

	/*return type is void*/
	Py_INCREF(Py_None);
	return Py_None;
}
/***********************************************************************************/

/***************The module’s method table and initialization function**************/
/*method table*/
static PyMethodDef starpupyMethods[] =
{
	{"_task_submit", starpu_task_submit_wrapper, METH_VARARGS, "submit the task"}, /*submit method*/
	{"task_wait_for_all", starpu_task_wait_for_all_wrapper, METH_VARARGS, "wait the task"}, /*wait for all method*/
	{"pause", starpu_pause_wrapper, METH_VARARGS, "suspend the processing of new tasks by workers"}, /*pause method*/
	{"resume", starpu_resume_wrapper, METH_VARARGS, "resume the workers polling for new tasks"}, /*resume method*/
	{"cpu_worker_get_count", starpu_cpu_worker_get_count_wrapper, METH_VARARGS, "return the number of CPUs controlled by StarPU"}, /*get count cpu method*/
	{"init_perfmodel", init_perfmodel, METH_VARARGS, "initialize struct starpu_perfmodel"}, /*initialize perfmodel*/
	{"free_perfmodel", free_perfmodel, METH_VARARGS, "free struct starpu_perfmodel"}, /*free perfmodel*/
	{"save_history_based_model", starpu_save_history_based_model_wrapper, METH_VARARGS, "save the performance model"}, /*save the performance model*/
	{"sched_get_min_priority", starpu_sched_get_min_priority_wrapper, METH_VARARGS, "get the number of min priority"}, /*get the number of min priority*/
	{"sched_get_max_priority", starpu_sched_get_max_priority_wrapper, METH_VARARGS, "get the number of max priority"}, /*get the number of max priority*/
	{"task_nsubmitted", starpu_task_nsubmitted_wrapper, METH_VARARGS, "get the number of submitted tasks which have not completed yet"}, /*get the number of submitted tasks which have not completed yet*/
	{"shutdown", starpu_shutdown_wrapper, METH_VARARGS, "shutdown starpu"}, /*shutdown starpu*/
	{NULL, NULL}
};

/*deallocation function*/
static void starpupyFree(void *self)
{
	starpu_shutdown();
	Py_DECREF(asyncio_module);
	//COUNTREFS();
}

/*module definition structure*/
static struct PyModuleDef starpupymodule =
{
	PyModuleDef_HEAD_INIT,
	"starpupy", /*name of module*/
	NULL,
	-1,
	starpupyMethods, /*method table*/
	NULL,
	NULL,
	NULL,
	starpupyFree /*deallocation function*/
};

/*initialization function*/
PyMODINIT_FUNC
PyInit_starpupy(void)
{
	PyObject *m;

	/*module import initialization*/
	m = PyModule_Create(&starpupymodule);
	if (m == NULL)
		return NULL;

	StarpupyError = PyErr_NewException("starpupy.error", NULL, NULL);
	Py_XINCREF(StarpupyError);
	if (PyModule_AddObject(m, "error", StarpupyError) < 0)
	{
		Py_XDECREF(StarpupyError);
		Py_CLEAR(StarpupyError);
		Py_DECREF(m);
		return NULL;
	}

#if PY_MAJOR_VERSION < 3 || (PY_MAJOR_VERSION == 3 && PY_MINOR_VERSION < 9)
	PyEval_InitThreads();
#endif
	/*starpu initialization*/
	int ret;
	struct starpu_conf conf;
	starpu_conf_init(&conf);

	Py_BEGIN_ALLOW_THREADS
	ret = starpu_init(&conf);
	Py_END_ALLOW_THREADS
	if (ret!=0)
	{
		PyErr_Format(StarpupyError, "Unexpected value %d returned for starpu_init", ret);
		return NULL;
	}

	if (conf.sched_policy_name && !strcmp(conf.sched_policy_name, "graph_test"))
	{
		/* FIXME: should call starpu_do_schedule when appropriate, the graph_test scheduler needs it. */
		fprintf(stderr,"TODO: The graph_test scheduler needs starpu_do_schedule calls\n");
		exit(77);
	}

	/*python asysncio import*/
	asyncio_module = PyImport_ImportModule("asyncio");

#ifdef STARPU_PYTHON_HAVE_NUMPY
	/*numpy import array*/
	import_array();
#endif
	
	return m;
}
/***********************************************************************************/