This repository was archived by the owner on Sep 13, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 10
Expand file tree
/
Copy pathwritequeue.c
More file actions
278 lines (218 loc) · 7.34 KB
/
Copy pathwritequeue.c
File metadata and controls
278 lines (218 loc) · 7.34 KB
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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
#include "../blastbeat.h"
extern struct blastbeat_server blastbeat;
/*
The BlastBeat write queue
Non-blocking writes are pretty complex to manage
This implementation uses a queue of buffers associated to an ev_io structure
Whenever an item is put in the queue an ev_io writer will be start and it will try to
write datas until it fails (returning EAGAIN or an incomplete write)
The offset in each item is required for managing incomplete writes
Bandwidth limiting
limiting per-vhost bandwidth is vital for QoS
the writequeue uses a token bucket algorithm
->bandwidth -> is the bandiwdth limit (in bytes per second)
->bandwidth_bucket -> is initialized to 0 and incremented by ((->bandwidth*30)/1000) bytes every 30ms
(30ms has been choosen as a good compromise between performance and load, but could be tunable)
as soon as ->bandwidth_bucket == ->bandwidth the token-add hook is stopped (will be restarted as soon as it decrease)
when the underlying socket is ready to WRITE data, N tokens are removed from the bucket (where N is the size of the packet,
if the packet is bigger it will be split)
if a WRITE event happens when the token is 0, we have a non conformant packet, the write event will be stopped, and will be restarted
at the next token-add hook
*/
static int wq_push(struct bb_writer *bbw, char *buf, size_t len, int flags, struct bb_session *bbs) {
// check writequeue_buffer
if (bbw->len+len > blastbeat.writequeue_buffer) {
fprintf(stderr,"too much queued datas\n");
return -1;
}
struct bb_writer_item *bbwi = bb_alloc(sizeof(struct bb_writer_item));
if (!bbwi) {
bb_error("unable to allocate memory for a writequeue item: malloc()");
return -1;
}
bbwi->buf = buf;
bbwi->pos = 0;
bbwi->len = len;
bbwi->flags = flags;
bbwi->session = bbs;
bbwi->next = NULL;
if (!bbw->head) {
bbw->head = bbwi;
}
else {
bbw->tail->next = bbwi;
}
bbw->tail = bbwi;
bbw->len += len;
return 0;
}
static void wq_decapitate(struct bb_writer *bbw) {
struct bb_writer_item *head = bbw->head;
bbw->head = head->next;
// is it the last item ?
if (head == bbw->tail) {
bbw->tail = NULL;
bbw->head = NULL;
}
if ((head->flags & BB_WQ_FREE) && head->len > 0) {
bb_free(head->buf, head->len);
}
bb_free(head, sizeof(struct bb_writer_item));
}
static void bb_throttle(struct bb_virtualhost *vhost, struct bb_connection *bbc) {
// mark the vhost as throttled
vhost->throttled = 1;
// stop the writer
ev_io_stop(blastbeat.loop, &bbc->writer.writer);
bbc->throttle.vhost = vhost;
bbc->throttle.connection = bbc;
// wait for unthrottling
ev_prepare_start(blastbeat.loop, &bbc->throttle.throttle);
}
void bb_wq_callback(struct ev_loop *loop, struct ev_io *w, int revents) {
struct bb_writer *bbw = (struct bb_writer *) w;
struct bb_connection *bbc = bbw->connection;
struct bb_writer_item *bbwi = bbw->head;
while(bbwi) {
if (bbwi->flags & BB_WQ_CLOSE) goto end;
if (bbwi->flags & BB_WQ_EOS) goto end2;
if (bbwi->len == 0) goto next;
size_t bbw_len = bbwi->len-bbwi->pos;
if (bbwi->session) {
// bandwidth check
uint64_t bandwidth = bbwi->session->vhost->bandwidth;
if (bandwidth > 0) {
// full bucket detected throttle the connection
if (bbwi->session->vhost->bandwidth_bucket == 0) {
bb_throttle(bbwi->session->vhost, bbc);
return;
}
// if packet is bigger than bucket size, split it
if (bbw_len > bbwi->session->vhost->bandwidth_bucket) {
bbw_len = bbwi->session->vhost->bandwidth_bucket;
}
}
}
ssize_t wlen = bbc->acceptor->write(bbc, bbwi->buf+bbwi->pos, bbw_len);
if (wlen < 0) {
if (errno == EINPROGRESS || errno == EAGAIN || errno == EWOULDBLOCK) {
return ;
}
bb_error("unable to write to client: write()");
goto end;
}
if (wlen == 0) {
bb_error("client disconnected: write()");
goto end;
}
// reset the connection activity timer on successfully sent
bb_connection_reset_timer(bbc);
// account transferred bytes to the virtualhost
if (bbwi->session) {
bbwi->session->vhost->tx+=wlen;
if (bbwi->session->vhost->bandwidth > 0) {
bbwi->session->vhost->bandwidth_bucket -= wlen;
}
}
bbw->len -= wlen;
if (wlen < bbwi->len-bbwi->pos) {
bbwi->pos+=wlen;
return;
}
next:
bbwi = bbwi->next;
wq_decapitate(bbw);
}
ev_io_stop(blastbeat.loop, w);
return;
end:
bb_connection_close(bbc);
return;
end2:
// close the session (remember to decapitate, as other session will continue pushing)
wq_decapitate(bbw);
bb_session_close(bbwi->session);
}
static void bb_wq_start(struct bb_writer *bbw) {
ev_io_start(blastbeat.loop, &bbw->writer);
}
int bb_wq_push(struct bb_session *bbs, char *buf, size_t len, int flags) {
struct bb_connection *bbc = bbs->connection;
if (!bbc) return -1;
if (wq_push(&bbc->writer, buf, len, flags, bbs)) return -1;
// an item has been pushed, start the ev_io
bb_wq_start(&bbc->writer);
return 0;
}
int bb_wq_dumb_push(struct bb_connection *bbc, char *buf, size_t len, int flags) {
if (!bbc) return -1;
if (wq_push(&bbc->writer, buf, len, flags, NULL)) return -1;
// an item has been pushed, start the ev_io
bb_wq_start(&bbc->writer);
return 0;
}
int bb_wq_push_close(struct bb_session *bbs) {
struct bb_connection *bbc = bbs->connection;
if (!bbc) return -1;
if (wq_push(&bbc->writer, NULL, 0, BB_WQ_CLOSE, bbs)) return -1;
// an item has been pushed, start the ev_io
bb_wq_start(&bbc->writer);
return 0;
}
int bb_wq_push_eos(struct bb_session *bbs) {
struct bb_connection *bbc = bbs->connection;
if (!bbc) return -1;
// persistent session cannot defer close
if (bbs->persistent) return -1;
if (wq_push(&bbc->writer, NULL, 0, BB_WQ_EOS, bbs)) return -1;
// an item has been pushed, start the ev_io
bb_wq_start(&bbc->writer);
return 0;
}
int bb_wq_push_copy(struct bb_session *bbs, char *buf, size_t len, int flags) {
struct bb_connection *bbc = bbs->connection;
if (!bbc) return -1;
char *new_buf = bb_alloc(len);
if (!new_buf) {
bb_error("unable to allocate memory for writequeue item: malloc()");
return -1;
}
memcpy(new_buf, buf, len);
if (wq_push(&bbc->writer, new_buf, len, flags, bbs)) return -1;
// an item has been pushed, start the ev_io
bb_wq_start(&bbc->writer);
return 0;
}
void bb_connection_throttle_cb(struct ev_loop *loop, struct ev_prepare *w, int revents) {
struct bb_connection_throttle *bbct = (struct bb_connection_throttle *) w;
struct bb_virtualhost *vhost = bbct->vhost;
struct bb_connection *bbc = bbct->connection;
if (!vhost || !bbc) {
fprintf(stderr,"BUG in throttle system !!!\n");
return;
}
// no more throttled
if (vhost->throttled == 0) {
ev_prepare_stop(blastbeat.loop, w);
// just for safety (could be useful in future implementations)
bbct->vhost = NULL;
bbct->connection = NULL;
bb_wq_start(&bbc->writer);
}
}
void bb_throttle_cb(struct ev_loop *loop, struct ev_timer *w, int revents) {
struct bb_throttle *bbt = (struct bb_throttle *) w;
struct bb_virtualhost *vhost = bbt->vhost;
// bucket could be bigger if we decrease bandwidth from the dealer
// (will be possibile soon ;)
if (vhost->bandwidth_bucket >= vhost->bandwidth) return;
uint64_t token = ((vhost->bandwidth*30)/1000);
if (vhost->bandwidth_bucket + token > vhost->bandwidth) {
vhost->bandwidth_bucket = vhost->bandwidth;
}
else {
vhost->bandwidth_bucket += token;
}
// unthrottle
vhost->throttled = 0;
}