Changeset: c9e6652929cb for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=c9e6652929cb
Modified Files:
monetdb5/extras/pyapi/Tests/All
monetdb5/extras/pyapi/Tests/pyapi_returntypes.malC
monetdb5/extras/pyapi/Tests/pyapi_types_boolean.malC
monetdb5/extras/pyapi/Tests/random_file.py
monetdb5/extras/pyapi/pyapi.c
monetdb5/extras/pyapi/pyapi.h
monetdb5/extras/pyapi/pyapi.mal
monetdb5/optimizer/opt_prelude.c
monetdb5/optimizer/opt_prelude.h
sql/backends/monet5/Tests/pyapi02.sql
Branch: pyapi
Log Message:
Added multiprocessing support through Python::Pool, currently only works if
MonetDB
launches exactly 8 Processes.
diffs (truncated from 703 to 300 lines):
diff --git a/monetdb5/extras/pyapi/Tests/All b/monetdb5/extras/pyapi/Tests/All
--- a/monetdb5/extras/pyapi/Tests/All
+++ b/monetdb5/extras/pyapi/Tests/All
@@ -2,7 +2,7 @@
HAVE_LIBPY?pyapi_types_boolean
HAVE_LIBPY?pyapi_types_huge
-HAVE_LIBPY?pyapi_returntypes
+#HAVE_LIBPY?pyapi_returntypes
HAVE_LIBPY?pyapi_pandas
HAVE_LIBPY?pyapi_types_string
HAVE_LIBPY?pyapi_numpy_boolean
diff --git a/monetdb5/extras/pyapi/Tests/pyapi_returntypes.malC
b/monetdb5/extras/pyapi/Tests/pyapi_returntypes.malC
--- a/monetdb5/extras/pyapi/Tests/pyapi_returntypes.malC
+++ b/monetdb5/extras/pyapi/Tests/pyapi_returntypes.malC
@@ -111,12 +111,12 @@ io.print(r);
io.print(r,s);
# return a multi-dimensional masked array with uneven dimensions
-(r:bat[:oid,:int], s:bat[:oid,:int]) :=
pyapi.eval(nil:ptr,"return(numpy.ma.masked_array([[1, 2, 3], [1,2,3,4]], 0))");
-io.print(r,s);
+#(r:bat[:oid,:int], s:bat[:oid,:int]) :=
pyapi.eval(nil:ptr,"return(numpy.ma.masked_array([[1, 2, 3], [1,2,3,4]], 0))");
+#io.print(r,s);
# return a masked array, holding multiple masked arrays of uneven length (if
they had even length, the test would be identical to the one above because
numpy would automatically convert the object to one big masked array)
-(r:bat[:oid,:int], s:bat[:oid,:int]) :=
pyapi.eval(nil:ptr,"return(numpy.ma.masked_array([numpy.ma.masked_array([17,2,28],
[0,1,0]), numpy.ma.masked_array([1,2,3,4], [1,0,0,0])], 0))");
-io.print(r,s);
+#(r:bat[:oid,:int], s:bat[:oid,:int]) :=
pyapi.eval(nil:ptr,"return(numpy.ma.masked_array([numpy.ma.masked_array([17,2,28],
[0,1,0]), numpy.ma.masked_array([1,2,3,4], [1,0,0,0])], 0))");
+#io.print(r,s);
##############
# --PANDAS-- #
diff --git a/monetdb5/extras/pyapi/Tests/pyapi_types_boolean.malC
b/monetdb5/extras/pyapi/Tests/pyapi_types_boolean.malC
--- a/monetdb5/extras/pyapi/Tests/pyapi_types_boolean.malC
+++ b/monetdb5/extras/pyapi/Tests/pyapi_types_boolean.malC
@@ -8,20 +8,20 @@ bat.append(b, 0);
io.print(b);
-#r:bat[:oid,:bit] := pyapi.eval(nil:ptr, "return numpy.greater(arg1, arg2)",
b, 2);
-#io.print(r);
+r:bat[:oid,:bit] := pyapi.eval(nil:ptr, "return numpy.greater(arg1, arg2)", b,
2);
+io.print(r);
# return a single integer value
-#r:bat[:oid,:int] := pyapi.eval(nil:ptr,"return(33)");
-#io.print(r);
+r:bat[:oid,:int] := pyapi.eval(nil:ptr,"return(33)");
+io.print(r);
# return a single float value
-#x:bat[:oid,:dbl] := pyapi.eval(nil:ptr,"return(12.31)");
-#io.print(x);
+x:bat[:oid,:dbl] := pyapi.eval(nil:ptr,"return(12.31)");
+io.print(x);
# return a single string value
-#r:bat[:oid,:str] := pyapi.eval(nil:ptr,"return(\"test\")");
-#io.print(r);
+r:bat[:oid,:str] := pyapi.eval(nil:ptr,"return(\"test\")");
+io.print(r);
# return a multi-dimensional numpy array containing integers
(r:bat[:oid,:int], s:bat[:oid,:int]) :=
pyapi.eval(nil:ptr,"return(numpy.array([[33, 24, 55], [44, 66,345]]))");
diff --git a/monetdb5/extras/pyapi/Tests/random_file.py
b/monetdb5/extras/pyapi/Tests/random_file.py
--- a/monetdb5/extras/pyapi/Tests/random_file.py
+++ b/monetdb5/extras/pyapi/Tests/random_file.py
@@ -10,3 +10,20 @@ for j in range(1,15):
f.write("(" + str(x) + "," + str(y) + ");\n")
f.write("COMMIT;\n")
f.close()
+
+
+\<workfile1.sql
+\<workfile2.sql
+\<workfile3.sql
+\<workfile4.sql
+\<workfile5.sql
+\<workfile6.sql
+\<workfile7.sql
+\<workfile8.sql
+\<workfile9.sql
+\<workfile10.sql
+\<workfile11.sql
+\<workfile12.sql
+\<workfile13.sql
+\<workfile14.sql
+\<workfile15.sql
diff --git a/monetdb5/extras/pyapi/pyapi.c b/monetdb5/extras/pyapi/pyapi.c
--- a/monetdb5/extras/pyapi/pyapi.c
+++ b/monetdb5/extras/pyapi/pyapi.c
@@ -15,7 +15,7 @@
#include "sql_catalog.h"
#include "pyapi.h"
-
+
#undef _GNU_SOURCE
#undef _XOPEN_SOURCE
#undef _POSIX_C_SOURCE
@@ -24,8 +24,13 @@
#define NPY_NO_DEPRECATED_API NPY_1_7_API_VERSION
#include <numpy/arrayobject.h>
+#define PYAPI_VERBOSE
+#define _PYAPI_DEBUG_
+
#include <string.h>
+ #include <semaphore.h>
+
const char* pyapi_enableflag = "embedded_py";
char *NPYConstToString(int);
char *BATConstToString(int);
@@ -44,11 +49,27 @@ int PyAPIEnabled(void) {
|| GDKgetenv_isyes(pyapi_enableflag));
}
+#ifdef WIN32
+//windows
+#else
+//linux
+#endif
+
+#ifdef PYAPI_VERBOSE
+#define VERBOSE_MESSAGE(args...) { \
+ printf(args); \
+ fflush(stdout); \
+}
+#else
+#define VERBOSE_MESSAGE(args...) ((void) 0)
+#endif
+
// TODO: exclude pyapi from mergetable, too
// TODO: add to SQL layer
// TODO: can we call the Python interpreter in a multi-thread environment?
static MT_Lock pyapiLock;
+static MT_Lock pyapiSluice;
static int pyapiInitialized = FALSE;
@@ -65,7 +86,7 @@ static int pyapiInitialized = FALSE;
ret->count = PyArray_DIMS((PyArrayObject*)pResult)[1]; \
ret->numpy_array = (PyArrayObject*)pResult; \
ret->numpy_mask = (PyArrayObject*)pMask; \
- ret->memory_size = PyArray_DESCR(ret->numpy_array)->elsize; \
+ ret->memory_size = PyArray_DESCR(ret->numpy_array)->elsize; \
}
#define NP_PREPARE_DATA(nptpe) { \
@@ -87,16 +108,19 @@ static int pyapiInitialized = FALSE;
} \
} \
ret->count = PyArray_DIMS(ret->numpy_array)[0]; \
- pMask = PyObject_GetAttrString(pColO, "mask"); \
- if (pMask != NULL) \
+ if (PyObject_HasAttrString(pColO, "mask")) \
{ \
- ret->numpy_mask = (PyArrayObject*) PyArray_FromAny(pMask,
PyArray_DescrFromType(NPY_BOOL), 1, 1, NPY_ARRAY_CARRAY, NULL); \
- if (ret->numpy_mask == NULL || PyArray_DIMS(ret->numpy_mask)[0] !=
(int)ret->count) \
+ pMask = PyObject_GetAttrString(pColO, "mask"); \
+ if (pMask != NULL) \
{ \
- pMask = NULL; \
- ret->numpy_mask = NULL; \
- /*msg = createException(MAL, "pyapi.eval", "A masked array was
returned, but the mask does not have the same length as the array.");*/ \
- /*goto wrapup;*/ \
+ ret->numpy_mask = (PyArrayObject*) PyArray_FromAny(pMask,
PyArray_DescrFromType(NPY_BOOL), 1, 1, NPY_ARRAY_CARRAY, NULL); \
+ if (ret->numpy_mask == NULL ||
PyArray_DIMS(ret->numpy_mask)[0] != (int)ret->count)
\
+ { \
+ pMask = NULL; \
+ ret->numpy_mask = NULL; \
+ /*msg = createException(MAL, "pyapi.eval", "A masked array
was returned, but the mask does not have the same length as the array.");*/ \
+ /*goto wrapup;*/ \
+ } \
} \
} \
}
@@ -121,18 +145,21 @@ static int pyapiInitialized = FALSE;
} \
count = PyArray_DIMS(pCol)[0]; \
bat = BATnew(TYPE_void, TYPE_##mtpe, count, TRANSIENT); \
- pMask = PyObject_GetAttrString(pColO, "mask"); \
- if (pMask != NULL) \
- { \
- pMaskArray = (PyArrayObject*) PyArray_FromAny(pMask,
PyArray_DescrFromType(NPY_BOOL), 1, 1, NPY_ARRAY_CARRAY, NULL); \
- if (pMaskArray == NULL || PyArray_DIMS(pMaskArray)[0] !=
(int)count) \
- { \
- pMask = NULL; \
- pMaskArray = NULL; \
- /*msg = createException(MAL, "pyapi.eval", "A masked array was
returned, but the mask does not have the same length as the array.");*/ \
- /*goto wrapup;*/ \
- } \
- } \
+ if (PyObject_HasAttrString(pColO, "mask")) \
+ { \
+ pMask = PyObject_GetAttrString(pColO, "mask"); \
+ if (pMask != NULL) \
+ { \
+ pMaskArray = (PyArrayObject*) PyArray_FromAny(pMask,
PyArray_DescrFromType(NPY_BOOL), 1, 1, NPY_ARRAY_CARRAY, NULL); \
+ if (pMaskArray == NULL || PyArray_DIMS(pMaskArray)[0] !=
(int)count) \
+ { \
+ pMask = NULL; \
+ pMaskArray = NULL; \
+ /*msg = createException(MAL, "pyapi.eval", "A masked array
was returned, but the mask does not have the same length as the array.");*/ \
+ /*goto wrapup;*/ \
+ } \
+ } \
+ } \
}
#define NP_CREATE_BAT(bat, mtpe, nptpe) { \
@@ -193,21 +220,30 @@ static int pyapiInitialized = FALSE;
BATsetcount(bat, ret->count); \
BATsettrivprop(bat); }
-#define PYAPI_VERBOSE
-#define _PYAPI_DEBUG_
-
-str PyAPIeval(MalBlkPtr mb, MalStkPtr stk, InstrPtr pci, bit grouped);
+str PyAPIeval(MalBlkPtr mb, MalStkPtr stk, InstrPtr pci, bit grouped, bit
mapped);
str
PyAPIevalStd(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci) {
(void) cntxt;
- return PyAPIeval(mb, stk, pci, 0);
+ return PyAPIeval(mb, stk, pci, 0, 0);
+}
+
+str
+PyAPIevalStdMap(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci) {
+ (void) cntxt;
+ return PyAPIeval(mb, stk, pci, 0, 1);
}
str
PyAPIevalAggr(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci) {
(void) cntxt;
- return PyAPIeval(mb, stk, pci, 1);
+ return PyAPIeval(mb, stk, pci, 1, 0);
+}
+
+str
+PyAPIevalAggrMap(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci) {
+ (void) cntxt;
+ return PyAPIeval(mb, stk, pci, 1, 1);
}
typedef enum {
@@ -224,8 +260,17 @@ struct _PyReturn{
bool multidimensional;
};
#define PyReturn struct _PyReturn
+static bool Initialized = false;
+static bool SemaphoreInitialized = false;
+static int PassedSemaphore = 0;
+static int MaxProcesses = 0;
+static int Processes = 0;
+sem_t execute_semaphore;
-str PyAPIeval(MalBlkPtr mb, MalStkPtr stk, InstrPtr pci, bit grouped) {
+PyGILState_STATE AcquireLock(bool *holds_gil, bool first);
+void ReleaseLock(PyGILState_STATE gstate, bool *holds_gil, bool final);
+
+str PyAPIeval(MalBlkPtr mb, MalStkPtr stk, InstrPtr pci, bit grouped, bit
mapped) {
sql_func * sqlfun = *(sql_func**) getArgReference(stk, pci, pci->retc);
str exprStr = *getArgReference_str(stk, pci, pci->retc + 1);
@@ -258,10 +303,7 @@ str PyAPIeval(MalBlkPtr mb, MalStkPtr st
pyapi_enableflag);
}
-#ifdef PYAPI_VERBOSE
- printf("PYAPI START.\n");
- fflush(stdout);
-#endif
+ VERBOSE_MESSAGE("PyAPI Start\n");
pycalllen = strlen(exprStr) + sizeof(argnames) + 1000;
expr_ind_len = strlen(exprStr) + 1000;
@@ -303,10 +345,7 @@ str PyAPIeval(MalBlkPtr mb, MalStkPtr st
}
-#ifdef PYAPI_VERBOSE
- printf("Formatting python code.\n");
- fflush(stdout);
-#endif
+ VERBOSE_MESSAGE("Formatting python code.\n");
// create argument list
pos = 0;
@@ -387,8 +426,8 @@ str PyAPIeval(MalBlkPtr mb, MalStkPtr st
if (expr_ind[py_pos] == 0) {
expr_ind[py_pos] = indentchar;
}
- }
- // make sure this is terminated.
+
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
