实现signal和slot机制(2)-跨线程投递

前言

在1中，我们实现了一个基础的signal+slot的模块件调用机制，不过那个direct调用，在这篇中，我们将支持夸线程调用，即在对象的线程上下文中，调用对象的函数。

对象线程绑定

在qt中，有一个很重要的概念，对象线程绑定，即在每个对象实例时候，其保存其线程上下文信息，其中就有一个事件循环，跨线程的信号就是将这个一个task丢到这个事件循环中，在对象绑定的线程中完成slot响应。

实现

对slot调用的打包-Task

我们需要将一个跨线程的slot调用打包成一个task，然后丢到想相应的事件循环中。在事件循环中需要维护一个task队列，那第一还是需要做容器。

struct TaskBase
{
    virtual ~TaskBase() = 0;
    virtual void run() = 0;
};

TaskBase::~TaskBase()
{
}

TaskBase为我们Task基类，在事件循环中，维护其一个链表。
我们继承它，实现我们第一个Task，一个一般的Task。

template <typename Method>
struct NormalTask : public TaskBase
{
    typedef typename ParameterTupleTraits<ParameterTraits<Method>>::Parameters Parameters;
    typedef typename TypeTraits<typename ParameterTraits<Method>::classType_value>::pointerType_value type_value;

    NormalTask(type_value object, Method m, Parameters paras) : TaskBase(),
                                                                m_object(object),
                                                                m_method(m),
                                                                m_paras(paras)
    {
    }

    virtual ~NormalTask() {}

    void run()
    {
        printf("run:%lu\n", pthread_self());
        impleDoTask(m_object, m_method, m_paras);
    }

    type_value m_object;
    Method m_method;
    Parameters m_paras;
};

其实一个模板，于具体的对象成员函数相关，接受，对象指针，对象成员函数指针和参数。pthread_self可以打印当前线程的标识，其实一个unsigned long类型的数，这里为了看其的执行线程。

事件循环和对象绑定

为了实现对象线程绑定，我们必须提供一个在这套系统上的基类，正如qt中的QObject。

struct Box
{
    Box()
    {
        m_loop = EventLoop::currentEventLoop();
    }

    EventLoop *m_loop;
};

下面开始我们的事件循环，其实这只有一个任务队列，实际中其还必须能够监听系统各种事件，比如按键，鼠标，加其发送给合适的对象，就是我的一篇博客介绍的事件监听分发，这里不做讨论。

 static EventLoop *currentEventLoop()
    {
        EventLoop *re = NULL;
        pthread_t pid = pthread_self();
        std::map<pthread_t, EventLoop *>::iterator ite = s_eventloops.find(pid);
        if (ite != s_eventloops.end())
        {
            re = ite->second;
        }
        else
        {
            re = new EventLoop(pid);
            s_eventloops.insert(std::pair<pthread_t, EventLoop *>(pid, re));
        }
        return re;
    }

    static void exit()
    {
        for (auto ite = s_eventloops.begin(); ite != s_eventloops.end(); ++ite)
        {
            ite->second->quit();
            delete (ite->second);
        }
        s_eventloops.clear();
    }

    int exec()
    {
        while (run)
        {
            for (auto ite = m_tasks.begin(); ite != m_tasks.end();)
            {
                (*ite)->run();
                delete (*ite);
                ite = m_tasks.erase(ite);
            }
            //sleep(10);
        }
        return 1;
    }

    void quit()
    {
        run = false;
    }
};

std::map<pthread_t, EventLoop *> EventLoop::s_eventloops = std::map<pthread_t, EventLoop *>();

怎么实现投递

提供了一个枚举，来设置连接类型

enum CONNECT_TYPE
{
    SYNC,
    AYNC,
    AUTO
};

对应：同步，异步和系统决定。
在原来signaleemit时，我们必须实现区别发送

 void eemit()
    {
        printf("eemit:%lu\n", pthread_self());
        std::tuple<> para;
        for (typename listValue_type::iterator ite = sloters.begin(); ite != sloters.end(); ++ite)
        {
            CONNECT_TYPE type = (*ite)->m_type;
            switch (type)
            {
            case SYNC:
            {
                if (EventLoop::currentEventLoop() == ((*ite)->eventloop()))
                {
                    (*ite)->dotask(para);
                }
                else
                {
                    (*ite)->eventloop()->postNormalTask((*ite)->convertAnormalTask(para));
                }
            };
            break;
            case AYNC:
            {
                (*ite)->eventloop()->postNormalTask((*ite)->convertAnormalTask(para));
            };
            break;
            case AUTO:
            {
                if (EventLoop::currentEventLoop() == ((*ite)->eventloop()))
                {
                    (*ite)->dotask(para);
                }
                else
                {
                    (*ite)->eventloop()->postNormalTask((*ite)->convertAnormalTask(para));
                }
            };
            break;
            default:
                break;
            }
        }
    }

为了实现在eemit中根据对象的线程绑定来区分行为，但是我们得到的对象，只是一个基类指针，所以对原来的基类进行了改造。

template <typename Paras>
struct SlotBase
{
    SlotBase(CONNECT_TYPE type) : m_type(type)
    {
    }

    virtual TaskBase *convertAnormalTask(Paras paras) = 0;

    virtual EventLoop *eventloop() = 0;

    virtual ~SlotBase() = 0;
    virtual void dotask(Paras paras) = 0;
    CONNECT_TYPE m_type;
};

template <typename Paras>
SlotBase<Paras>::~SlotBase()
{
}

在slot实现这连个纯虚函数，来得到打包好的Task和得到槽对象的事件循环指针。

  TaskBase *convertAnormalTask(Parameters paras)
    {
        return new NormalTask<M>(m_object, m_method, paras);
    }

    EventLoop *eventloop()
    {
        return m_object->m_loop;
    }

这样保证了在eemit时获得正确的task和事件循环，当槽对象的eventloop和eemit的对象不在一个线程中，我们就将task投递到槽对象的eventloop中。
为了实现想下列的使用

    EventLoop *loop = EventLoop::currentEventLoop();
    loop->postNormalTask(NewNormalTask(&object, &A::func_b, 1314));

我们提供了NewNormalTask各版本

template <typename Method>
TaskBase *NewNormalTask(typename TypeTraits<typename ParameterTraits<Method>::classType_value>::pointerType_value object,
                        Method m, typename ParameterTraits<Method>::P0 p0,
                        typename ParameterTraits<Method>::P1 p1,
                        typename ParameterTraits<Method>::P2 p2,
                        typename ParameterTraits<Method>::P3 p3,
                        typename ParameterTraits<Method>::P4 p4,
                        typename ParameterTraits<Method>::P5 p5)
{
    StaticCheck<ParameterTraits<Method>::size == 6> a;
    UNUSE(a);
    typename ParameterTupleTraits<ParameterTraits<Method>>::Parameters paras = std::make_tuple(p0, p1, p2, p3, p4, p5);
    return new NormalTask<Method>(object, m, paras);
}

运用实例

#include <iostream>
#include "signal.h"
using namespace std;

struct TTT
{
    int a;
    int b;
    int c;
};

class N : public Box
{
public:
    virtual void func_c(int a, int b) = 0;

};

class A : public N
{
public:
    A()
    {
        s1.connect(this, &A::func_ii);
    }
    void func_b(int a) {cout << a << "aaaaaa\n";}
    void func_c(int a, int b) {cout << a << "+" << b << "=" << a+b << std::endl;}
    void func_a()
    {
        TTT t = {1, ‘s‘, ‘t‘};
        s.eemit(t, t, t);
    }


    void func_i()
    {
        int *i = new int(1);
        s1.eemit(i);
    }

    void func_ii(int *i)
    {
        cout << *i;
        (*i)++;
        delete i;
    }

    void func_z()
    {
        cout << "zhou xiang  ";

    }

    Signal<void (*)(TTT, TTT, TTT)> s;
    Signal<void (*)(int*)> s1;

};

class B : public Box
{
public:
    B(A *a):m_a(a){}
    B(){}
    void func_b(int a) {cout << a << "bbbbbbb\n";}
    void func_slot(TTT t, TTT t2, TTT t3)
    {
        cout << t.a + t2.b + t3.c << "-==-=-=-==-=\n";
    }

    void func_z()
    {
        cout << "love chenchen\n";
    }

    A *m_a;
    void func_e()
    {
        m_a->s.connect(this, &B::func_slot);
    }
};

void* pt_callback(void * arg)
{
    A aa;
    B bb;
    Signal<void (*)()> *s = (Signal<void (*)()>*)(arg);
    s->connect(&aa, &A::func_z);
    s->connect(&bb, &B::func_z);
    EventLoop *loop = EventLoop::currentEventLoop();
    loop->exec();
    return NULL;
}

int main()
{
    A object;
    B object1(&object);

    Signal<void (*)(int, int)> s;

    Signal<void (*)()> ssss;

    // ssss.connect(&object, &A::func_z);
    // ssss.connect(&object1, &B::func_z);
    // ssss.connect(&object1, &B::func_z);

//    s.connect(&object ,&A::func_c);
//    s.connect(&object ,&A::func_c);
//    s.connect(&object ,&A::func_c);
//    s.connect(&object ,&A::func_c);
//    s.connect(&object ,&A::func_c);
//    s.connect(&object ,&A::func_c);
//    s.connect(&object ,&A::func_c);
//    object1.func_e();
//    object1.func_e();

//    object.func_a();

    //ssss.eemit();
    // ssss.disconnect(&object, &A::func_z);
    // ssss.disconnect(&object1, &B::func_z);


    //object.func_i();


    //cout << "---------------------------\n";

    //ssss.eemit();


    //s.eemit(4, 456);


    // EventLoop *loop = EventLoop::currentEventLoop();
    // loop->postNormalTask(NewNormalTask(&object, &A::func_b, 1314));
    //std::cout << "+++++++++++++++++++++\n";

    pthread_t ptid;

    int ptre = pthread_create(&ptid, NULL, pt_callback, (void *)(&ssss));
    if (ptre != 0)
    {
        printf("dispatcher thread create fail!\n");
    }

    ssss.eemit();
    // ssss.disconnect(&object, &A::func_z);
    // ssss.disconnect(&object1, &B::func_z);


    //object.func_i();


    cout << "---------------------------\n";

    ssss.eemit();

    EventLoop *loop = EventLoop::currentEventLoop();
    loop->postNormalTask(NewNormalTask(&object, &A::func_b, 1314));
    loop->exec();

    return 0;
}

运行结果

在操作队列的时候，没有加锁。还有就是线程之间执行不可控，在eemit时，在pt_callback中，connect还没有完成，导致，有时有两个响应有时一个，线程向标准输出写入的时候也是不可控的，所有log有点难看，但还是说明了问题，操作在对象的线程上下问中进行。

项目代码可以在1中找到地址。