C2Tester/Thread.cpp at main · uplife1980/C2Tester · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
#include "Thread.hpp"
#include <unistd.h>

#include <sys/sysinfo.h>
#include <sys/syscall.h>
#include "socket.hpp"
#include <ctime>

void ThreadFactory::startThreadFactory(const config& cfg)
{
    m_cfg = cfg;
}

void ThreadFactory::makeThread(vector<shared_ptr<thread>> &vecThread)
{

    int socketPerThread = m_cfg.m_vecUrl.size()/m_cfg.m_iThreadNum;
    int socketRemain = m_cfg.m_vecUrl.size() % m_cfg.m_iThreadNum;
    LOG_INFO("per is "<< socketPerThread<< " and remain is"<< socketRemain);
    auto itr = m_cfg.m_vecUrl.begin();
    int i = 0;
    int j = 0;
    //start threads
    for (; i < m_cfg.m_iThreadNum; i++)
    {
        auto itrBegin = itr;

        //iterate to find last socket that this thread needs.
        if (i < socketRemain)
        {
            for (auto j = 0; j < socketPerThread; j++)
                itr++;
            itr++;
        }
        else{
            for (auto j = 0; j < socketPerThread; j++)
                itr++;
        }

        shared_ptr<thread> pThread(new thread(Thread::ThreadWork, std::ref(m_cfg.m_vecUrl), itrBegin, itr, j++));
        usleep(100*1000);
        vecThread.push_back(pThread);
    }

    //wait all threads to stop.
    for (auto i = vecThread.begin(); i != vecThread.end();)
    {
        (*i)->join();
        vecThread.erase(i);
        i = vecThread.begin();
    }


}
/*
shared_ptr<ThreadFactory> ThreadFactory::getThreadFactorySingleton()
{
    if (m_pThreadFactory == NULL)
    {
        m_pThreadFactory.reset(new ThreadFactory());
    }

    return m_pThreadFactory;

}*/

ThreadFactory::ThreadFactory()
{

}

//end is included
void Thread::ThreadWork(vector<Endpoint> &vecEndpoint, const vector<Endpoint>::iterator begin, const vector<Endpoint>::iterator end, const int cpuNum)
{
        			cpu_set_t mask;

				CPU_ZERO(&mask);
				CPU_SET(cpuNum, &mask);
				pthread_setaffinity_np(pthread_self(), sizeof(mask), &mask);

    int tid = syscall(SYS_gettid);


    int epollfd = epoll_create(1);
    if(epollfd < 0)
    {
        LOG_ERROR("epoll create error");
        exit(0);
    }

    vector<shared_ptr<Socket>> vecSocket;
    //connect socket
    for(auto itr = begin; itr != end; itr++)
    {
        shared_ptr<Socket> s(new Socket(*itr));
        vecSocket.push_back(s);
        updateEvents(epollfd, s->getFD(), EPOLLIN|EPOLLET, EPOLL_CTL_ADD);
    }
    LOG_INFO("tid " << tid << " is handling "<< vecSocket.size()<<" sockets.");
    //loop epoll
    for (;;)
    { //实际应用应当注册信号处理函数，退出时清理资源
        loop_once(epollfd, 10000);
    }

}