add libevent based daemon

houxiao

2017-01-09 0b1de1fddd889cf2ebbe578bfad83862f5ebdf5a

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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
#include "PipeLinePool.h" 
#include "logger.h" 
#include <pthread.h> 
 
#define PLP_MUTEX_LOCK(mut,_ret) if (mut != nullptr) {\ 
    int ret = pthread_mutex_lock((pthread_mutex_t*)mut); \ 
    if(ret != 0) \ 
    { \ 
        LOG_ERROR << "pthread_mutex_lock " << #mut <<  ": " << ret; \ 
        return _ret; \ 
    } \ 
} 
 
#define PLP_MUTEX_UNLOCK(mut,_ret) if (mut != nullptr) {\ 
    int ret = pthread_mutex_unlock((pthread_mutex_t*)mut); \ 
    if(ret != 0) \ 
    { \ 
        LOG_ERROR << "pthread_mutex_unlock " << #mut <<  ": " << ret; \ 
        return _ret; \ 
    } \ 
} 
 
PipeLinePool::PipeLinePool(bool _multithread_safe) :  
    multithread_safe(_multithread_safe), tsafe_mutex(nullptr), pl_mutex(nullptr),  
    pipelines(), pipelines_free() 
{ 
    if (multithread_safe) 
    { 
        tsafe_mutex = new pthread_mutex_t; 
        pthread_mutex_init((pthread_mutex_t*)tsafe_mutex, NULL); 
         
        pl_mutex = new pthread_mutex_t; 
        pthread_mutex_init((pthread_mutex_t*)pl_mutex, NULL); 
         
        PLP_MUTEX_LOCK(pl_mutex,); 
    } 
} 
 
PipeLinePool::~PipeLinePool() 
{ 
    if (multithread_safe) 
    { 
        PLP_MUTEX_UNLOCK(pl_mutex,); 
    } 
     
    pthread_mutex_destroy((pthread_mutex_t*)tsafe_mutex); 
    delete (pthread_mutex_t*)tsafe_mutex; 
    tsafe_mutex = nullptr; 
     
    pthread_mutex_destroy((pthread_mutex_t*)pl_mutex); 
    delete (pthread_mutex_t*)pl_mutex; 
    pl_mutex = nullptr; 
     
    pipelines_free.clear(); 
     
    for (pl_set_t::iterator iter = pipelines.begin(); iter != pipelines.end(); ++iter) 
        delete *iter; 
     
    pipelines.clear(); 
} 
 
void PipeLinePool::manage(PipeLine* pl) 
{ 
    if (pl == nullptr) 
        return; 
 
    PLP_MUTEX_LOCK(tsafe_mutex,); 
     
    if (pipelines.find(pl) != pipelines.end()) 
        return; 
     
    pipelines.insert(pl); 
    pipelines_free.insert(pl); 
     
    PLP_MUTEX_UNLOCK(tsafe_mutex,); 
} 
 
void PipeLinePool::unmanage(PipeLine* pl) 
{ 
    PLP_MUTEX_LOCK(tsafe_mutex,); 
 
    pipelines.erase(pl); 
    pipelines_free.erase(pl); 
     
    PLP_MUTEX_UNLOCK(tsafe_mutex,); 
} 
 
PipeLine* PipeLinePool::get_free() 
{ 
    if (pipelines_free.empty()) 
    { 
        PLP_MUTEX_LOCK(pl_mutex, nullptr); 
    } 
     
    PLP_MUTEX_LOCK(tsafe_mutex, nullptr); 
     
    if (pipelines_free.empty()) 
        return nullptr; 
     
    pl_set_t::iterator iter = pipelines_free.begin(); 
    PipeLine* pl = *iter; 
    pipelines_free.erase(iter); 
     
    PLP_MUTEX_UNLOCK(tsafe_mutex, nullptr); 
     
    return pl; 
} 
 
void PipeLinePool::release(PipeLine* pl) 
{ 
    if (pipelines.find(pl) == pipelines.end()) 
        return; 
    if (pipelines_free.find(pl) != pipelines.end()) 
        return; 
 
    PLP_MUTEX_LOCK(tsafe_mutex,); 
     
    pipelines_free.insert(pl); 
 
    PLP_MUTEX_UNLOCK(tsafe_mutex,); 
     
    PLP_MUTEX_UNLOCK(pl_mutex,); 
}