blob: ab4fb5f6b9ff1b6d46c45e84440a1d900aaafc8b (
plain)
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
|
#pragma once
#include "swoole.h"
#include "context.h"
#include "coroutine.h"
#include <string>
#include <iostream>
#include <list>
#include <queue>
#include <sys/stat.h>
namespace swoole {
enum channel_op
{
PRODUCER = 1,
CONSUMER = 2,
};
class Channel;
struct notify_msg_t
{
Channel *chan;
enum channel_op type;
};
struct timeout_msg_t
{
Channel *chan;
coroutine_t *co;
bool error;
swTimer_node *timer;
};
class Channel
{
private:
std::list<coroutine_t *> producer_queue;
std::list<coroutine_t *> consumer_queue;
std::queue<void *> data_queue;
size_t capacity;
uint32_t notify_producer_count;
uint32_t notify_consumer_count;
public:
bool closed;
inline bool is_empty()
{
return data_queue.size() == 0;
}
inline bool is_full()
{
return data_queue.size() == capacity;
}
inline size_t length()
{
return data_queue.size();
}
inline size_t consumer_num()
{
return consumer_queue.size();
}
inline size_t producer_num()
{
return producer_queue.size();
}
inline void remove(coroutine_t *co)
{
consumer_queue.remove(co);
}
inline coroutine_t* pop_coroutine(enum channel_op type)
{
coroutine_t* co;
if (type == PRODUCER)
{
co = producer_queue.front();
producer_queue.pop_front();
notify_producer_count--;
swDebug("resume producer[%d]", coroutine_get_cid(co));
}
else
{
co = consumer_queue.front();
consumer_queue.pop_front();
notify_consumer_count--;
swDebug("resume consumer[%d]", coroutine_get_cid(co));
}
return co;
}
Channel(size_t _capacity);
void yield(enum channel_op type);
void notify(enum channel_op type);
void* pop(double timeout = 0);
bool push(void *data);
bool close();
};
};
|
