ETM-Schurig/powersync-core
Archived
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix thread syncronisation

Browse Source
This commit is contained in:
Michael Zanetti 2020-06-26 10:23:53 +02:00
parent 3c47e59811
commit 5d0751ae27
4 changed files with 134 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

104
libnymea-core/integrations/python/pything.h
View File

@ -11,6 +11,7 @@
#include <QPointer> #include <QPointer>
#include <QThread> #include <QThread>
#include <QMutexLocker>
#pragma GCC diagnostic push #pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Winvalid-offsetof" #pragma GCC diagnostic ignored "-Winvalid-offsetof"
@ -19,28 +20,34 @@
typedef struct _thing { typedef struct _thing {
PyObject_HEAD PyObject_HEAD
Thing *thing; Thing *thing;
QMutex *mutex;
} PyThing; } PyThing;
static int PyThing_init(PyThing */*self*/, PyObject */*args*/, PyObject */*kwds*/) { static PyObject* PyThing_new(PyTypeObject *type, PyObject */*args*/, PyObject */*kwds*/) {
return 0; PyThing *self = (PyThing*)type->tp_alloc(type, 0);
if (self == NULL) {
return nullptr;
}
self->mutex = new QMutex();
return (PyObject*)self;
} }
static void PyThing_dealloc(PyThing * self) { static void PyThing_dealloc(PyThing * self) {
Py_TYPE(self)->tp_free(self); Py_TYPE(self)->tp_free(self);
delete self->mutex;
} }
static PyObject *PyThing_getName(PyThing *self, void */*closure*/) static PyObject *PyThing_getName(PyThing *self, void */*closure*/)
{ {
QMutexLocker(self->mutex);
if (!self->thing) { if (!self->thing) {
PyErr_SetString(PyExc_ValueError, "Thing has been removed from the system."); PyErr_SetString(PyExc_ValueError, "Thing has been removed from the system.");
return nullptr; return nullptr;
} }
QString name; QString name;
// FIXME: Should not be a direct connection! QMetaObject::invokeMethod(self->thing, "name", Qt::BlockingQueuedConnection, Q_RETURN_ARG(QString, name));
qWarning() << "name thread" << QThread::currentThread();
QMetaObject::invokeMethod(self->thing, "name", Qt::DirectConnection, Q_RETURN_ARG(QString, name));
PyObject *ret = PyUnicode_FromString(name.toUtf8().data()); PyObject *ret = PyUnicode_FromString(name.toUtf8().data());
Py_INCREF(ret); Py_INCREF(ret);
return ret; return ret;
@ -48,12 +55,14 @@ static PyObject *PyThing_getName(PyThing *self, void */*closure*/)
static int PyThing_setName(PyThing *self, PyObject *value, void */*closure*/){ static int PyThing_setName(PyThing *self, PyObject *value, void */*closure*/){
QString name = QString(PyUnicode_AsUTF8(value)); QString name = QString(PyUnicode_AsUTF8(value));
QMutexLocker(self->mutex);
QMetaObject::invokeMethod(self->thing, "setName", Qt::QueuedConnection, Q_ARG(QString, name)); QMetaObject::invokeMethod(self->thing, "setName", Qt::QueuedConnection, Q_ARG(QString, name));
return 0; return 0;
} }
static PyObject *PyThing_getSettings(PyThing *self, void */*closure*/) static PyObject *PyThing_getSettings(PyThing *self, void */*closure*/)
{ {
QMutexLocker(self->mutex);
if (!self->thing) { if (!self->thing) {
PyErr_SetString(PyExc_ValueError, "Thing has been removed from the system."); PyErr_SetString(PyExc_ValueError, "Thing has been removed from the system.");
return nullptr; return nullptr;
@ -90,6 +99,7 @@ static PyObject * PyThing_setStateValue(PyThing* self, PyObject* args)
QVariant value(bytes); QVariant value(bytes);
QMutexLocker(self->mutex);
if (self->thing != nullptr) { if (self->thing != nullptr) {
QMetaObject::invokeMethod(self->thing, "setStateValue", Qt::QueuedConnection, Q_ARG(StateTypeId, stateTypeId), Q_ARG(QVariant, value)); QMetaObject::invokeMethod(self->thing, "setStateValue", Qt::QueuedConnection, Q_ARG(StateTypeId, stateTypeId), Q_ARG(QVariant, value));
} }
@ -102,6 +112,7 @@ static PyObject * PyThing_setStateValue(PyThing* self, PyObject* args)
static PyObject *PyThing_settingChanged(PyThing *self, void */*closure*/) static PyObject *PyThing_settingChanged(PyThing *self, void */*closure*/)
{ {
QMutexLocker(self->mutex);
if (!self->thing) { if (!self->thing) {
PyErr_SetString(PyExc_ValueError, "Thing has been removed from the system."); PyErr_SetString(PyExc_ValueError, "Thing has been removed from the system.");
return nullptr; return nullptr;
@ -154,6 +165,7 @@ static PyObject * PyThing_emitEvent(PyThing* self, PyObject* args)
} }
} }
QMutexLocker(self->mutex);
if (self->thing != nullptr) { if (self->thing != nullptr) {
QMetaObject::invokeMethod(self->thing, "emitEvent", Qt::QueuedConnection, Q_ARG(EventTypeId, eventTypeId), Q_ARG(ParamList, params)); QMetaObject::invokeMethod(self->thing, "emitEvent", Qt::QueuedConnection, Q_ARG(EventTypeId, eventTypeId), Q_ARG(ParamList, params));
} }
@ -161,7 +173,7 @@ static PyObject * PyThing_emitEvent(PyThing* self, PyObject* args)
Py_RETURN_NONE; Py_RETURN_NONE;
} }
static PyGetSetDef PyThing_getseters[] = { static PyGetSetDef PyThing_getset[] = {
{"name", (getter)PyThing_getName, (setter)PyThing_setName, "Thing name", nullptr}, {"name", (getter)PyThing_getName, (setter)PyThing_setName, "Thing name", nullptr},
{"settings", (getter)PyThing_getSettings, (setter)PyThing_setSettings, "Thing settings", nullptr}, {"settings", (getter)PyThing_getSettings, (setter)PyThing_setSettings, "Thing settings", nullptr},
{"settingChanged", (getter)PyThing_settingChanged, nullptr, "Signal for changed settings", nullptr}, {"settingChanged", (getter)PyThing_settingChanged, nullptr, "Signal for changed settings", nullptr},
@ -176,41 +188,59 @@ static PyMethodDef PyThing_methods[] = {
static PyTypeObject PyThingType = { static PyTypeObject PyThingType = {
PyVarObject_HEAD_INIT(NULL, 0) PyVarObject_HEAD_INIT(NULL, 0)
"nymea.Thing", /* tp_name */ "nymea.Thing", /* tp_name */
sizeof(PyThing), /* tp_basicsize */ sizeof(PyThing), /* tp_basicsize */
0, /* tp_itemsize */ 0, /* tp_itemsize */
0, /* tp_dealloc */ (destructor)PyThing_dealloc, /* tp_dealloc */
0, /* tp_print */ 0, /* tp_print */
0, /* tp_getattr */ 0, /* tp_getattr */
0, /* tp_setattr */ 0, /* tp_setattr */
0, /* tp_reserved */ 0, /* tp_reserved */
0, /* tp_repr */ 0, /* tp_repr */
0, /* tp_as_number */ 0, /* tp_as_number */
0, /* tp_as_sequence */ 0, /* tp_as_sequence */
0, /* tp_as_mapping */ 0, /* tp_as_mapping */
0, /* tp_hash */ 0, /* tp_hash */
0, /* tp_call */ 0, /* tp_call */
0, /* tp_str */ 0, /* tp_str */
0, /* tp_getattro */ 0, /* tp_getattro */
0, /* tp_setattro */ 0, /* tp_setattro */
0, /* tp_as_buffer */ 0, /* tp_as_buffer */
Py_TPFLAGS_DEFAULT, /* tp_flags */ Py_TPFLAGS_DEFAULT, /* tp_flags */
"Noddy objects", /* tp_doc */ "Thing", /* tp_doc */
0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0 0, /* tp_traverse */
0, /* tp_clear */
0, /* tp_richcompare */
0, /* tp_weaklistoffset */
0, /* tp_iter */
0, /* tp_iternext */
PyThing_methods, /* tp_methods */
0, /* tp_members */
PyThing_getset, /* tp_getset */
0, /* tp_base */
0, /* tp_dict */
0, /* tp_descr_get */
0, /* tp_descr_set */
0, /* tp_dictoffset */
0, /* tp_init */
0, /* tp_alloc */
(newfunc)PyThing_new, /* tp_new */
0, /* tp_free */
0, /* tp_is_gc */
0, /* tp_bases */
0, /* tp_mro */
0, /* tp_cache */
0, /* tp_subclasses */
0, /* tp_weaklist */
0, /* tp_del */
0, /* tp_version_tag */
0, /* tp_finalize */
0, /* tp_vectorcall */
0, /* tp_print DEPRECATED*/
}; };
static void registerThingType(PyObject *module) static void registerThingType(PyObject *module)
{ {
PyThingType.tp_new = PyType_GenericNew;
PyThingType.tp_dealloc= reinterpret_cast<destructor>(PyThing_dealloc);
PyThingType.tp_basicsize = sizeof(PyThing);
PyThingType.tp_flags = Py_TPFLAGS_DEFAULT;
PyThingType.tp_doc = "Thing class";
PyThingType.tp_methods = PyThing_methods;
PyThingType.tp_getset = PyThing_getseters;
// PyThingType.tp_members = PyThingSetupInfo_members;
PyThingType.tp_init = reinterpret_cast<initproc>(PyThing_init);
if (PyType_Ready(&PyThingType) < 0) { if (PyType_Ready(&PyThingType) < 0) {
return; return;
} }

92
libnymea-core/integrations/python/pythingsetupinfo.h
View File

@ -16,16 +16,22 @@ typedef struct {
PyObject_HEAD PyObject_HEAD
ThingSetupInfo* info; ThingSetupInfo* info;
PyThing *pyThing; PyThing *pyThing;
QMutex *mutex;
} PyThingSetupInfo; } PyThingSetupInfo;
static int PyThingSetupInfo_init(PyThingSetupInfo */*self*/, PyObject */*args*/, PyObject */*kwds*/) { static PyObject* PyThingSetupInfo_new(PyTypeObject *type, PyObject */*args*/, PyObject */*kwds*/) {
return 0; PyThingSetupInfo *self = (PyThingSetupInfo*)type->tp_alloc(type, 0);
if (self == NULL) {
return nullptr;
}
self->mutex = new QMutex();
return (PyObject*)self;
} }
static void PyThingSetupInfo_dealloc(PyThingSetupInfo * self) { static void PyThingSetupInfo_dealloc(PyThingSetupInfo * self) {
Py_TYPE(self)->tp_free(self); Py_TYPE(self)->tp_free(self);
delete self->mutex;
} }
static PyObject * PyThingSetupInfo_finish(PyThingSetupInfo* self, PyObject* args) { static PyObject * PyThingSetupInfo_finish(PyThingSetupInfo* self, PyObject* args) {
@ -40,6 +46,7 @@ static PyObject * PyThingSetupInfo_finish(PyThingSetupInfo* self, PyObject* args
Thing::ThingError thingError = static_cast<Thing::ThingError>(status); Thing::ThingError thingError = static_cast<Thing::ThingError>(status);
QString displayMessage = message != nullptr ? QString(message) : QString(); QString displayMessage = message != nullptr ? QString(message) : QString();
QMutexLocker(self->mutex);
if (self->info) { if (self->info) {
QMetaObject::invokeMethod(self->info, "finish", Qt::QueuedConnection, Q_ARG(Thing::ThingError, thingError), Q_ARG(QString, displayMessage)); QMetaObject::invokeMethod(self->info, "finish", Qt::QueuedConnection, Q_ARG(Thing::ThingError, thingError), Q_ARG(QString, displayMessage));
} }
@ -59,39 +66,58 @@ static PyMethodDef PyThingSetupInfo_methods[] = {
static PyTypeObject PyThingSetupInfoType = { static PyTypeObject PyThingSetupInfoType = {
PyVarObject_HEAD_INIT(NULL, 0) PyVarObject_HEAD_INIT(NULL, 0)
"nymea.ThingSetupInfo", /* tp_name */ "nymea.ThingSetupInfo", /* tp_name */
sizeof(PyThingSetupInfo), /* tp_basicsize */ sizeof(PyThingSetupInfo), /* tp_basicsize */
0, /* tp_itemsize */ 0, /* tp_itemsize */
0, /* tp_dealloc */ (destructor)PyThingSetupInfo_dealloc, /* tp_dealloc */
0, /* tp_vectorcall_offset */ 0, /* tp_print */
0, /* tp_getattr */ 0, /* tp_getattr */
0, /* tp_setattr */ 0, /* tp_setattr */
0, /* tp_as_async */ 0, /* tp_reserved */
0, /* tp_repr */ 0, /* tp_repr */
0, /* tp_as_number */ 0, /* tp_as_number */
0, /* tp_as_sequence */ 0, /* tp_as_sequence */
0, /* tp_as_mapping */ 0, /* tp_as_mapping */
0, /* tp_hash */ 0, /* tp_hash */
0, /* tp_call */ 0, /* tp_call */
0, /* tp_str */ 0, /* tp_str */
0, /* tp_getattro */ 0, /* tp_getattro */
0, /* tp_setattro */ 0, /* tp_setattro */
0, /* tp_as_buffer */ 0, /* tp_as_buffer */
Py_TPFLAGS_DEFAULT, /* tp_flags */ Py_TPFLAGS_DEFAULT, /* tp_flags */
"Noddy objects", /* tp_doc */ "ThingSetupInfo", /* tp_doc */
0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0 0, /* tp_traverse */
0, /* tp_clear */
0, /* tp_richcompare */
0, /* tp_weaklistoffset */
0, /* tp_iter */
0, /* tp_iternext */
PyThingSetupInfo_methods, /* tp_methods */
PyThingSetupInfo_members, /* tp_members */
0, /* tp_getset */
0, /* tp_base */
0, /* tp_dict */
0, /* tp_descr_get */
0, /* tp_descr_set */
0, /* tp_dictoffset */
0, /* tp_init */
0, /* tp_alloc */
(newfunc)PyThingSetupInfo_new, /* tp_new */
0, /* tp_free */
0, /* tp_is_gc */
0, /* tp_bases */
0, /* tp_mro */
0, /* tp_cache */
0, /* tp_subclasses */
0, /* tp_weaklist */
0, /* tp_del */
0, /* tp_version_tag */
0, /* tp_finalize */
0, /* tp_vectorcall */
0, /* tp_print DEPRECATED*/
}; };
static void registerThingSetupInfoType(PyObject *module) { static void registerThingSetupInfoType(PyObject *module) {
PyThingSetupInfoType.tp_new = PyType_GenericNew;
PyThingSetupInfoType.tp_dealloc=(destructor) PyThingSetupInfo_dealloc;
PyThingSetupInfoType.tp_basicsize = sizeof(PyThingSetupInfo);
PyThingSetupInfoType.tp_flags = Py_TPFLAGS_DEFAULT;
PyThingSetupInfoType.tp_doc = "ThingSetupInfo class";
PyThingSetupInfoType.tp_methods = PyThingSetupInfo_methods;
PyThingSetupInfoType.tp_members = PyThingSetupInfo_members;
PyThingSetupInfoType.tp_init = (initproc)PyThingSetupInfo_init;
if (PyType_Ready(&PyThingSetupInfoType) < 0) { if (PyType_Ready(&PyThingSetupInfoType) < 0) {
return; return;
} }

34
libnymea-core/integrations/pythonintegrationplugin.cpp
View File

@ -208,36 +208,32 @@ void PythonIntegrationPlugin::discoverThings(ThingDiscoveryInfo *info)
void PythonIntegrationPlugin::setupThing(ThingSetupInfo *info) void PythonIntegrationPlugin::setupThing(ThingSetupInfo *info)
{ {
PyThing *pyThing = PyObject_New(PyThing, &PyThingType); PyGILState_STATE s = PyGILState_Ensure();
PyThing *pyThing = (PyThing*)PyObject_CallObject((PyObject*)&PyThingType, NULL);
pyThing->thing = info->thing(); pyThing->thing = info->thing();
PyThingSetupInfo *pyInfo = PyObject_New(PyThingSetupInfo, &PyThingSetupInfoType); PyThingSetupInfo *pyInfo = (PyThingSetupInfo*)PyObject_CallObject((PyObject*)&PyThingSetupInfoType, NULL);
pyInfo->info = info; pyInfo->info = info;
pyInfo->pyThing = pyThing; pyInfo->pyThing = pyThing;
PyGILState_Release(s);
connect(info, &ThingSetupInfo::finished, this, [=](){ connect(info, &ThingSetupInfo::finished, this, [=](){
if (info->status() == Thing::ThingErrorNoError) { if (info->status() == Thing::ThingErrorNoError) {
m_things.insert(info->thing(), pyThing); m_things.insert(info->thing(), pyThing);
} else { } else {
PyGILState_STATE s = PyGILState_Ensure();
Py_DECREF(pyThing); Py_DECREF(pyThing);
PyGILState_Release(s);
} }
}); });
connect(info, &ThingSetupInfo::aborted, this, [=](){ connect(info, &ThingSetupInfo::aborted, this, [=](){
PyGILState_STATE s = PyGILState_Ensure();
Py_DECREF(pyThing); Py_DECREF(pyThing);
PyGILState_Release(s);
}); });
connect(info, &ThingSetupInfo::destroyed, this, [=](){ connect(info, &ThingSetupInfo::destroyed, this, [=](){
PyEval_RestoreThread(s_mainThread); QMutexLocker(pyInfo->mutex);
// PyGILState_STATE s = PyGILState_Ensure();
pyInfo->info = nullptr; pyInfo->info = nullptr;
Py_DECREF(pyInfo); Py_DECREF(pyInfo);
// PyGILState_Release(s);
s_mainThread = PyEval_SaveThread();
}); });
@ -282,13 +278,10 @@ void PythonIntegrationPlugin::thingRemoved(Thing *thing)
callPluginFunction("thingRemoved", reinterpret_cast<PyObject*>(pyThing)); callPluginFunction("thingRemoved", reinterpret_cast<PyObject*>(pyThing));
PyGILState_STATE s = PyGILState_Ensure(); QMutexLocker(pyThing->mutex);
pyThing->thing = nullptr; pyThing->thing = nullptr;
Py_DECREF(pyThing); Py_DECREF(pyThing);
PyGILState_Release(s);
m_things.remove(thing); m_things.remove(thing);
} }
@ -456,16 +449,6 @@ void PythonIntegrationPlugin::callPluginFunction(const QString &function, PyObje
PyGILState_Release(s); PyGILState_Release(s);
return; return;
} }
// PyObject *coro = result;
// PyObject *get_running_loop = PyObject_GetAttrString(s_asyncio, "get_event_loop");
// PyObject *loop = PyObject_CallFunctionObjArgs(get_running_loop, nullptr);
// Py_DECREF(get_running_loop);
// PyObject *run_in_executor = PyObject_GetAttrString(loop, "run_in_executor");
// result = PyObject_CallFunctionObjArgs(run_in_executor, Py_None, coro);
// Py_DECREF(result);
// Py_DECREF(coro);
// Spawn a event loop for python // Spawn a event loop for python
PyObject *new_event_loop = PyObject_GetAttrString(s_asyncio, "new_event_loop"); PyObject *new_event_loop = PyObject_GetAttrString(s_asyncio, "new_event_loop");
@ -488,7 +471,6 @@ void PythonIntegrationPlugin::callPluginFunction(const QString &function, PyObje
PyObject *run_until_complete = PyObject_GetAttrString(loop, "run_until_complete"); PyObject *run_until_complete = PyObject_GetAttrString(loop, "run_until_complete");
QtConcurrent::run([=](){ QtConcurrent::run([=](){
qWarning() << "new thread for func" << function << QThread::currentThread();
PyGILState_STATE s = PyGILState_Ensure(); PyGILState_STATE s = PyGILState_Ensure();
PyObject_CallFunctionObjArgs(run_until_complete, task, nullptr); PyObject_CallFunctionObjArgs(run_until_complete, task, nullptr);
PyGILState_Release(s); PyGILState_Release(s);

2
libnymea/integrations/thing.cpp
View File

@ -183,7 +183,6 @@ ThingClass Thing::thingClass() const
/*! Returns the name of this Thing. This is visible to the user. */ /*! Returns the name of this Thing. This is visible to the user. */
QString Thing::name() const QString Thing::name() const
{ {
qWarning() << "thing name called";
return m_name; return m_name;
} }
@ -234,7 +233,6 @@ void Thing::setParamValue(const ParamTypeId &paramTypeId, const QVariant &value)
ParamList Thing::settings() const ParamList Thing::settings() const
{ {
qWarning() << "thing settings called";
return m_settings; return m_settings;
} }