GVKun编程网logo

C++通过Swig垮线程回调Python函数（c++跨线程访问控件）

3

以上就是给各位分享C++通过Swig垮线程回调Python函数，其中也会对c++跨线程访问控件进行解释，同时本文还将给你拓展C++通过Swig跨线程回调Python代码、C扩展库中回调Python函数

以上就是给各位分享C++通过Swig垮线程回调Python函数，其中也会对c++跨线程访问控件进行解释，同时本文还将给你拓展C++通过Swig跨线程回调Python代码、C扩展库中回调Python函数、Python Swig-从ctypes指针创建swig包装的实例、python – Celery – 链接远程回调等相关知识，如果能碰巧解决你现在面临的问题，别忘了关注本站，现在开始吧！

本文目录一览：

C++通过Swig垮线程回调Python函数（c++跨线程访问控件）

C++通过Swig垮线程回调Python函数（c++跨线程访问控件）

C++ 定义 Callback 类. PyThreadStateLock 保证垮线程调用成功：

 

 

#include <Python/Python.h>
class Callback
{
public:
    Callback(){}
    
    virtual ~Callback(){}
    virtual void call(ObjWithPyCallback& object){}
};

class ObjWithPyCallback
{
public:
    
    ObjWithPyCallback();
    ~ObjWithPyCallback();
    void setCallback(Callback &callback);
    void call(); 
    
private:
    void NotifyThread();
    Callback* callback_;
};
class PyThreadStateLock
{
public:
    PyThreadStateLock(void)
    {
        state = PyGILState_Ensure( );
    }
    
    ~PyThreadStateLock(void)
    {
        PyGILState_Release( state );
    }
private:
    PyGILState_STATE state;
};

 

Callback.cpp

 

ObjWithPyCallback::ObjWithPyCallback() : callback_(NULL)
{
    Py_Initialize();
    PyEval_InitThreads();
    PyEval_ReleaseLock();
}

ObjWithPyCallback::~ObjWithPyCallback()
{    
}

void ObjWithPyCallback::setCallback(Callback &callback)
{
    callback_ = &callback;
}

void ObjWithPyCallback::call()
{
    if ( ! callback_ )
    {
        std::cerr << "No callback is set.\n";
    }
    else
    {
        thread(&ObjWithPyCallback::NotifyThread,this).detach();//call in another thread
        //callback_->call(*this);//call directly
        
    }
}

void ObjWithPyCallback::NotifyThread()
{
    class PyThreadStateLock PyThreadLock;//fix segmentation fault 
    callback_->call(*this);
}

 

.i 文件中

%module(directors="1") cb
%{
   #include "Callback.h"
%}
%feature("director") Callback;

%include "Callback.h"

 

在 python 中调用：

 

import cb

class CB(cb.Callback):
    def __init__(self):
        super(CB,self).__init__()
    def call(self,x):
        print("Hello from CB!")
        print(x)

o = cb.ObjWithPyCallback()
mycb=CB()
o.setCallback(mycb)
o.call()

C++通过Swig跨线程回调Python代码

C++通过Swig跨线程回调Python代码

C++ 定义 Callback 类. PyThreadStateLock 保证垮线程调用成功：

 

 

#include <Python/Python.h>
class Callback
{
public:
    Callback(){}
    
    virtual ~Callback(){}
    virtual void call(ObjWithPyCallback& object){}
};

class ObjWithPyCallback
{
public:
    
    ObjWithPyCallback();
    ~ObjWithPyCallback();
    void setCallback(Callback &callback);
    void call(); 
    
private:
    void NotifyThread();
    Callback* callback_;
};

class PyThreadStateLock
{
public:
    PyThreadStateLock(void)
    {
        state = PyGILState_Ensure( );
    }
    
    ~PyThreadStateLock(void)
    {
        PyGILState_Release( state );
    }
private:
    PyGILState_STATE state;
};

 

Callback.cpp

 

ObjWithPyCallback::ObjWithPyCallback() : callback_(NULL)
{
    Py_Initialize();
    PyEval_InitThreads();
    PyEval_ReleaseLock();
}

ObjWithPyCallback::~ObjWithPyCallback()
{    
}

void ObjWithPyCallback::setCallback(Callback &callback)
{
    callback_ = &callback;
}

void ObjWithPyCallback::call()
{
    if ( ! callback_ )
    {
        std::cerr << "No callback is set.\n";
    }
    else
    {
        thread(&ObjWithPyCallback::NotifyThread, this).detach();//call in another thread
        //callback_->call(*this);//call directly
        
    }
}

void ObjWithPyCallback::NotifyThread()
{
    class PyThreadStateLock PyThreadLock;//fix segmentation fault 
    callback_->call(*this);
}

 

.i 文件中

%module(directors="1") cb
%{
   #include "Callback.h"
%}
%feature("director") Callback;

%include "Callback.h"

 

在 python 中调用：

 

import cb

class CB(cb.Callback):
    def __init__(self):
        super(CB, self).__init__()
    def call(self, x):
        print("Hello from CB!")
        print(x)

o = cb.ObjWithPyCallback()
mycb=CB()
o.setCallback(mycb)
o.call()

 

C扩展库中回调Python函数

C扩展库中回调Python函数

下面是小编 jb51.cc 通过网络收集整理的代码片段。

小编小编现在分享给大家，也给大家做个参考。

// c 程序 段
static PyObject *gc_before_extract = NULL;

/// 设置回调
static PyObject *
SetBeforeCallbackFn(PyObject *dummy,PyObject *args)
{
PyObject *temp = NULL;

if (PyArg_ParseTuple(args,"O:set_callback",&temp)) {
   if (!PyCallable_Check(temp)) {
    PyErr_SetString(PyExc_TypeError,"parameter must be callable");
   }
   Py_XINCREF(temp);         /* Add a reference to new callback */
   Py_XDECREF(gc_before_extract); /* dispose of prevIoUs callback */
   gc_before_extract = temp;       /* Remember new callback */
}

return Py_BuildValue("l",(gc_before_extract == NULL) ? 0 : 1);
}

/// 调用上面函数设置的python脚本函数
int BeforeExt(char *pBeforeExtract)
{
PyObject* pArgs = NULL;
PyObject* pRetVal = NULL;
int    nRetVal = 0;

pArgs = Py_BuildValue("(s)",pFileName);
pRetVal = PyEval_CallObject(gc_before_extract,pArgs);
if (pRetVal)
{
   fprintf(stderr,"PyEval_CallObject : ok \r\n");
   nRetVal = PyInt_AsLong(pRetVal);
   fprintf(stderr,"PyEval_CallObject : return : %d \r\n",nRetVal);
}
Py_DECREF(pArgs);
Py_DECREF(pRetVal);
return nRetVal;
}

/// 测试函数
static 
PyObject* my_test_callback(PyObject *self,PyObject *args) 
{ 
PyObject * arglist; 
PyObject * result = NULL;

BeforeExt("good");

result = PyEval_CallObject(gc_before_extract,args); 
if (result)
{
   Py_DECREF(result); 
}
Py_INCREF(Py_None); 
return Py_None; 
}

/////// python 脚本
from pyArchive import *
from ctypes import *

# 回调函数必须有一个int型的返回值
def set_callback_fn(abc) :
    print "t_callback_fn say : {0}".format(abc)
    return 11

CMPFUNC = CFUNCTYPE(c_int,c_char_p) 
_callback = CMPFUNC(set_callback_fn)

    if SetBeforeCallbackFn(_callback):
        print "set call back ok"
        my_test_callback("script call : good luck")
    else :
        print "set call back fail"

以上是小编(jb51.cc)为你收集整理的全部代码内容，希望文章能够帮你解决所遇到的程序开发问题。

如果觉得小编网站内容还不错，欢迎将小编网站推荐给程序员好友。

Python Swig-从ctypes指针创建swig包装的实例

Python Swig-从ctypes指针创建swig包装的实例

我有用swig包装的类的C 代码。我无法修改代码或包装。在python中，我具有使用ctypes的指向所述C
类实例的指针。如何围绕该指针创建一个Swig包装器？

我知道swig对象拥有一个’this’属性，该属性在内部指向包装的对象，但是我找不到一种将其设置为我手头的指针的方法。

谢谢您的帮助！

答案1

小编典典

您 可以
执行此操作，但是这需要大量工作，并且解决使ctypes或SWIG接口完整且可用的根本问题比强制互换性要容易得多。（还值得注意的是，从SWIG创建一个ctypes对象比执行您要尝试的工作（从ctypes一个创建SWIG对象）要容易得多。

为了说明这一点，我创建了下面的头文件，将使用SWIG进行包装：

struct Foo {  double d;  char msg[20];};

然后，我用以下接口包装它：

%module test%{#include "test.h"%}// Prove I''m not cheating here - we can''t even instantiate this class%nodefaultctor;%include "test.h"

我还添加了一个测试函数供我们从ctypes调用，该函数未进行SWIG包装：

#include "test.h"// This function returns a Foo, but is only accessible to ctypesextern "C" Foo *fun1() {    static Foo f = { 1.234, "Hello" };    return &f;}

您this对SWIG封装类的此属性的猜测是一个很好的起点，但它并不只是更改它那么简单-
您插入的对象的类型必须与SWIG期望的匹配。它不仅仅是一个表示为int的指针：

repr(test.Foo().this) # You''ll need to drop the nodefaultctor directive to see this"<Swig Object of type ''Foo *'' at 0x7f621197d1e0>"

如果检查SWIG生成的源代码，我们可以看到有一个函数带有指针，一些类型信息并为我们创建了这些对象：

SWIGRUNTIME PyObject *SwigPyObject_New(void *ptr, swig_type_info *ty, int own);

让我们忽略目前默认情况下SWIGRUNTIME定义的事实static，为了让我们开始尝试运行时，我们可以将其重新定义为extern。稍后，我们将介绍无法解决问题的解决方法。

因此，我们的目标是获取“仅ctypes”函数的输出，并通过更多的ctypes调用传递它，SwigPyObject_New以创建可以与SWIG模块的this属性交换的东西。

为了进行调用，我们通常会调用SWIG_TypeQuery以查找swig_type_info要使用的正确方法。但是，实际上这是一个宏，它可以扩展以传递一些始终是静态的静态变量。因此，我们将使用以下功能：

SWIGRUNTIME swig_type_info *SWIG_Python_TypeQuery(const char *type)

（具有相同的SWIGRUNTIME条件）。

至此，我们已经足够可以交换代理对象的此属性，并且只要能够构造供体就可以完成。（尽管那会泄漏）。我们可以通过两种方法使它更好：

  1. __init__里面的猴子补丁test.Foo可以工作。如果您确实希望%nodefaultctor在SWIG界面中进行重新编译，那么这是最好的选择：

    def patched_init(self, ptr):self.this = ptr

    test.Foo.init = patched_init

  2. 创建一个仅具有的新类，然后在修改属性之前将__init__其设置为this属性__class__，而改用该类：

    class FooCtypesSwigInterop(object):def __init__(self, ptr):    self.this = ptr    self.__class__ = test.Foo

当您不想破坏test.Foo现有的__init__实现时，此选项最有意义。

这样，我们现在可以通过以下方式实现我们的初始目标：

import ctypesimport test# This *must* happen after the import of the real SWIG module# 0x4 is RTLD_NOLOAD which ensures that we get the same handle as the Python # import did, even if it was loaded with RTLD_LOCAL as Python is prone to.swig_module = ctypes.PyDLL(''./_test.so'',ctypes.RTLD_LOCAL|0x4)# Note that we used PyDLL instead of CDLL because we need to retain the GIL# Setup SWIG_Python_TypeQuery to have the right argument/return types# (Using void* as a substitute for swig_type_info*)SWIG_Python_TypeQuery = swig_module.SWIG_Python_TypeQuerySWIG_Python_TypeQuery.argtypes = [ctypes.c_char_p]SWIG_Python_TypeQuery.restype = ctypes.c_void_p# Likewise SwigPyObject_New, using ctypes.py_object though for returnSwigPyObject_New = swig_module.SwigPyObject_NewSwigPyObject_New.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_int]SwigPyObject_New.restype = ctypes.py_object# Actually do the type query for the type our ctypes function returns:SWIGTYPE_p_Foo = SWIG_Python_TypeQuery(''Foo*'')print(hex(SWIGTYPE_p_Foo))# Now the ctypes function itself that returns the type we want SWIG managedfun1 = swig_module.fun1fun1.argtypes = []fun1.restype = ctypes.c_void_p# Make the actual ctypes call we care about hereresult = fun1()print(hex(result))# And then create a SwigPyObject for it from the void* return type# Note that 0 means not owned by SWIGsresult = SwigPyObject_New(result, SWIGTYPE_p_Foo, 0)print(repr(sresult))# This is how we jimmy it back into the this attribute of a SWIG typeclass FooCtypesSwigInterop(object):    def __init__(self, ptr):        self.this = ptr        self.__class__ = test.Fooc = FooCtypesSwigInterop(sresult)# Finally a usable SWIG object from the ctypes call print(c.msg)

所有这些都可以编译并使用：

swig3.0 -python -c++ -Wall test.ig++ -shared -o _test.so test_wrap.cxx fun1.cc -Wall -Wextra -fPIC -I/usr/include/python2.7/ -std=c++11 -DSWIGRUNTIME=externLD_LIBRARY_PATH=.  python run.py

并给我们：

0x7fb6eccf29e00x7fb6eccf2640<Swig Object of type ''Foo *'' at 0x7fb6ee436720>Hello

要解决SWIGRUNTIME定义为的问题，static您需要再执行一步。请使用调试符号或对已获得的SWIG二进制模块进行反向工程，但无法进行修改以找到我们需要的两个函数的地址，这些地址相对于已导出的符号未导出。然后，您可以使用它们来构造ctypes函数指针，而不用按名称查找它们。当然，购买/查找/重写SWIG模块或将缺少的功能添加到ctypes接口可能会更容易。

（最后，值得注意的是，尽管SWIG运行-builtin时需要进行一些实质性更改，但似乎在这里并不适用），才能使此答案有效。

python – Celery – 链接远程回调

python – Celery – 链接远程回调

我有三个Celery任务分别在三个不同的服务器上运行.

> tasks.send_push_notification
> tasks.send_sms
> tasks.send_email

我想设置一个工作流程,如果发送推送通知失败,我应该尝试发送短信.如果发送短信失败,我应该发送电子邮件.

如果这3个任务及其代码库位于同一台服务器上,我会按照the example on chained tasks并完成类似的操作

from celery import chain
from tasks import send_push_notification,send_sms,send_email
import json
# some paylaod
payload = json.dumps({})
res = chain(
    send_push_notification.subtask(payload),send_sms.subtask(payload),send_email.subtask(payload)
)()

但任务保存在3个不同的服务器上！

我试过了

# 1
from celery import chain
from my_celery_app import app
res = chain(
    app.send_task(''tasks.send_push_notification'',payload),app.send_task(''tasks.send_sms'',app.send_task(''tasks.send_email'',payload)
)()
# Which fails because I am chaining tasks not subtasks

和

# 2
from celery import chain,subtask
res = chain(
    subtask(''tasks.send_push_notification'',subtask(''tasks.send_sms'',subtask(''tasks.send_email'',payload)
)()
# fails because I am not adding the tasks on the broker

如何才能做到这一点？

更新：
我可以使用链接NOT链来做到这一点.

from celery import subtask
res = app.send_task(
    ''tasks.send_push_notification'',(payload,),link=subtask(
        ''tasks.send_sms'',link=subtask(
            ''tasks.send_email'',)
    )
)

有很多筑巢.而且因为我实际上需要创建一个数据库驱动的工作流,所以用这种方式创建它会很复杂.

解决方法

为什么不在你的任务中处理它,

def push_notification_task(payload):
    if not send_push_notification(payload):
        sms_notification_task.delay(payload)

def sms_notification_task(payload):
    if not send_sms_notification(payload):
        email_notification_task.delay(payload)

def email_notification_task(payload):
    send_email_notification(payload)

此外,chain将按给定顺序执行所有任务,而您希望下一个任务仅在首次失败时运行.

今天的关于C++通过Swig垮线程回调Python函数和c++跨线程访问控件的分享已经结束，谢谢您的关注，如果想了解更多关于C++通过Swig跨线程回调Python代码、C扩展库中回调Python函数、Python Swig-从ctypes指针创建swig包装的实例、python – Celery – 链接远程回调的相关知识，请在本站进行查询。

本文标签：

上一篇C++11并行编程-条件变量(condition_variable)详细说明（c++条件变量使用）

下一篇[BZOJ2179]-大数乘法-FFT模板（大数乘法 fft）

相关文章