]>
Commit | Line | Data |
---|---|---|
d288ee65 | 1 | /* |
2 | * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com> | |
3 | * All rights reserved. | |
4 | * | |
5 | * Redistribution and use in source and binary forms, with or without | |
6 | * modification, are permitted provided that the following conditions are met: | |
7 | * | |
8 | * * Redistributions of source code must retain the above copyright notice, | |
9 | * this list of conditions and the following disclaimer. | |
10 | * * Redistributions in binary form must reproduce the above copyright | |
11 | * notice, this list of conditions and the following disclaimer in the | |
12 | * documentation and/or other materials provided with the distribution. | |
13 | * * Neither the name of Redis nor the names of its contributors may be used | |
14 | * to endorse or promote products derived from this software without | |
15 | * specific prior written permission. | |
16 | * | |
17 | * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" | |
18 | * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE | |
19 | * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE | |
20 | * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE | |
21 | * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR | |
22 | * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF | |
23 | * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS | |
24 | * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN | |
25 | * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) | |
26 | * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE | |
27 | * POSSIBILITY OF SUCH DAMAGE. | |
28 | */ | |
29 | ||
e2641e09 | 30 | #include "redis.h" |
e2641e09 | 31 | #include <sys/uio.h> |
32 | ||
11e0c4c5 | 33 | static void setProtocolError(redisClient *c, int pos); |
34 | ||
609baba8 | 35 | /* To evaluate the output buffer size of a client we need to get size of |
36 | * allocated objects, however we can't used zmalloc_size() directly on sds | |
37 | * strings because of the trick they use to work (the header is before the | |
38 | * returned pointer), so we use this helper function. */ | |
39 | size_t zmalloc_size_sds(sds s) { | |
40 | return zmalloc_size(s-sizeof(struct sdshdr)); | |
41 | } | |
42 | ||
e2641e09 | 43 | void *dupClientReplyValue(void *o) { |
44 | incrRefCount((robj*)o); | |
45 | return o; | |
46 | } | |
47 | ||
48 | int listMatchObjects(void *a, void *b) { | |
49 | return equalStringObjects(a,b); | |
50 | } | |
51 | ||
52 | redisClient *createClient(int fd) { | |
f3357792 | 53 | redisClient *c = zmalloc(sizeof(redisClient)); |
e2641e09 | 54 | |
0f1d64ca | 55 | /* passing -1 as fd it is possible to create a non connected client. |
56 | * This is useful since all the Redis commands needs to be executed | |
57 | * in the context of a client. When commands are executed in other | |
58 | * contexts (for instance a Lua script) we need a non connected client. */ | |
59 | if (fd != -1) { | |
60 | anetNonBlock(NULL,fd); | |
61 | anetTcpNoDelay(NULL,fd); | |
62 | if (aeCreateFileEvent(server.el,fd,AE_READABLE, | |
63 | readQueryFromClient, c) == AE_ERR) | |
64 | { | |
65 | close(fd); | |
66 | zfree(c); | |
67 | return NULL; | |
68 | } | |
106bd87a PN |
69 | } |
70 | ||
e2641e09 | 71 | selectDb(c,0); |
72 | c->fd = fd; | |
56de4964 | 73 | c->bufpos = 0; |
e2641e09 | 74 | c->querybuf = sdsempty(); |
9fa9ccb0 | 75 | c->querybuf_peak = 0; |
cd8788f2 | 76 | c->reqtype = 0; |
e2641e09 | 77 | c->argc = 0; |
78 | c->argv = NULL; | |
2c74a9f9 | 79 | c->cmd = c->lastcmd = NULL; |
cd8788f2 | 80 | c->multibulklen = 0; |
e2641e09 | 81 | c->bulklen = -1; |
e2641e09 | 82 | c->sentlen = 0; |
83 | c->flags = 0; | |
56ff70f8 | 84 | c->ctime = c->lastinteraction = server.unixtime; |
e2641e09 | 85 | c->authenticated = 0; |
86 | c->replstate = REDIS_REPL_NONE; | |
dbd8c753 | 87 | c->slave_listening_port = 0; |
e2641e09 | 88 | c->reply = listCreate(); |
3853c168 | 89 | c->reply_bytes = 0; |
7eac2a75 | 90 | c->obuf_soft_limit_reached_time = 0; |
e2641e09 | 91 | listSetFreeMethod(c->reply,decrRefCount); |
92 | listSetDupMethod(c->reply,dupClientReplyValue); | |
54b08c86 | 93 | c->bpop.keys = dictCreate(&setDictType,NULL); |
e3c51c4b DJMM |
94 | c->bpop.timeout = 0; |
95 | c->bpop.target = NULL; | |
e2641e09 | 96 | c->io_keys = listCreate(); |
97 | c->watched_keys = listCreate(); | |
98 | listSetFreeMethod(c->io_keys,decrRefCount); | |
99 | c->pubsub_channels = dictCreate(&setDictType,NULL); | |
100 | c->pubsub_patterns = listCreate(); | |
101 | listSetFreeMethod(c->pubsub_patterns,decrRefCount); | |
102 | listSetMatchMethod(c->pubsub_patterns,listMatchObjects); | |
7b722727 | 103 | if (fd != -1) listAddNodeTail(server.clients,c); |
e2641e09 | 104 | initClientMultiState(c); |
105 | return c; | |
106 | } | |
107 | ||
51669c5a | 108 | /* This function is called every time we are going to transmit new data |
109 | * to the client. The behavior is the following: | |
110 | * | |
111 | * If the client should receive new data (normal clients will) the function | |
112 | * returns REDIS_OK, and make sure to install the write handler in our event | |
113 | * loop so that when the socket is writable new data gets written. | |
114 | * | |
115 | * If the client should not receive new data, because it is a fake client | |
116 | * or a slave, or because the setup of the write handler failed, the function | |
117 | * returns REDIS_ERR. | |
118 | * | |
119 | * Typically gets called every time a reply is built, before adding more | |
120 | * data to the clients output buffers. If the function returns REDIS_ERR no | |
121 | * data should be appended to the output buffers. */ | |
122 | int prepareClientToWrite(redisClient *c) { | |
7156f43c | 123 | if (c->flags & REDIS_LUA_CLIENT) return REDIS_OK; |
51669c5a | 124 | if (c->fd <= 0) return REDIS_ERR; /* Fake client */ |
834ef78e | 125 | if (c->bufpos == 0 && listLength(c->reply) == 0 && |
e2641e09 | 126 | (c->replstate == REDIS_REPL_NONE || |
127 | c->replstate == REDIS_REPL_ONLINE) && | |
128 | aeCreateFileEvent(server.el, c->fd, AE_WRITABLE, | |
834ef78e PN |
129 | sendReplyToClient, c) == AE_ERR) return REDIS_ERR; |
130 | return REDIS_OK; | |
131 | } | |
132 | ||
36c19d03 PN |
133 | /* Create a duplicate of the last object in the reply list when |
134 | * it is not exclusively owned by the reply list. */ | |
135 | robj *dupLastObjectIfNeeded(list *reply) { | |
136 | robj *new, *cur; | |
137 | listNode *ln; | |
138 | redisAssert(listLength(reply) > 0); | |
139 | ln = listLast(reply); | |
140 | cur = listNodeValue(ln); | |
141 | if (cur->refcount > 1) { | |
142 | new = dupStringObject(cur); | |
143 | decrRefCount(cur); | |
144 | listNodeValue(ln) = new; | |
145 | } | |
146 | return listNodeValue(ln); | |
834ef78e PN |
147 | } |
148 | ||
25ef3192 | 149 | /* ----------------------------------------------------------------------------- |
150 | * Low level functions to add more data to output buffers. | |
151 | * -------------------------------------------------------------------------- */ | |
152 | ||
36c19d03 | 153 | int _addReplyToBuffer(redisClient *c, char *s, size_t len) { |
f3357792 | 154 | size_t available = sizeof(c->buf)-c->bufpos; |
36c19d03 | 155 | |
25ef3192 | 156 | if (c->flags & REDIS_CLOSE_AFTER_REPLY) return REDIS_OK; |
157 | ||
36c19d03 PN |
158 | /* If there already are entries in the reply list, we cannot |
159 | * add anything more to the static buffer. */ | |
160 | if (listLength(c->reply) > 0) return REDIS_ERR; | |
161 | ||
162 | /* Check that the buffer has enough space available for this string. */ | |
163 | if (len > available) return REDIS_ERR; | |
e2641e09 | 164 | |
36c19d03 PN |
165 | memcpy(c->buf+c->bufpos,s,len); |
166 | c->bufpos+=len; | |
167 | return REDIS_OK; | |
834ef78e PN |
168 | } |
169 | ||
36c19d03 PN |
170 | void _addReplyObjectToList(redisClient *c, robj *o) { |
171 | robj *tail; | |
25ef3192 | 172 | |
173 | if (c->flags & REDIS_CLOSE_AFTER_REPLY) return; | |
174 | ||
36c19d03 PN |
175 | if (listLength(c->reply) == 0) { |
176 | incrRefCount(o); | |
177 | listAddNodeTail(c->reply,o); | |
609baba8 | 178 | c->reply_bytes += zmalloc_size_sds(o->ptr); |
36c19d03 PN |
179 | } else { |
180 | tail = listNodeValue(listLast(c->reply)); | |
181 | ||
182 | /* Append to this object when possible. */ | |
183 | if (tail->ptr != NULL && | |
184 | sdslen(tail->ptr)+sdslen(o->ptr) <= REDIS_REPLY_CHUNK_BYTES) | |
185 | { | |
609baba8 | 186 | c->reply_bytes -= zmalloc_size_sds(tail->ptr); |
36c19d03 PN |
187 | tail = dupLastObjectIfNeeded(c->reply); |
188 | tail->ptr = sdscatlen(tail->ptr,o->ptr,sdslen(o->ptr)); | |
609baba8 | 189 | c->reply_bytes += zmalloc_size_sds(tail->ptr); |
36c19d03 PN |
190 | } else { |
191 | incrRefCount(o); | |
192 | listAddNodeTail(c->reply,o); | |
609baba8 | 193 | c->reply_bytes += zmalloc_size_sds(o->ptr); |
36c19d03 PN |
194 | } |
195 | } | |
7eac2a75 | 196 | asyncCloseClientOnOutputBufferLimitReached(c); |
36c19d03 | 197 | } |
834ef78e | 198 | |
36c19d03 PN |
199 | /* This method takes responsibility over the sds. When it is no longer |
200 | * needed it will be free'd, otherwise it ends up in a robj. */ | |
201 | void _addReplySdsToList(redisClient *c, sds s) { | |
202 | robj *tail; | |
25ef3192 | 203 | |
5b94b8ac | 204 | if (c->flags & REDIS_CLOSE_AFTER_REPLY) { |
205 | sdsfree(s); | |
206 | return; | |
207 | } | |
25ef3192 | 208 | |
36c19d03 PN |
209 | if (listLength(c->reply) == 0) { |
210 | listAddNodeTail(c->reply,createObject(REDIS_STRING,s)); | |
609baba8 | 211 | c->reply_bytes += zmalloc_size_sds(s); |
36c19d03 PN |
212 | } else { |
213 | tail = listNodeValue(listLast(c->reply)); | |
214 | ||
215 | /* Append to this object when possible. */ | |
216 | if (tail->ptr != NULL && | |
217 | sdslen(tail->ptr)+sdslen(s) <= REDIS_REPLY_CHUNK_BYTES) | |
218 | { | |
609baba8 | 219 | c->reply_bytes -= zmalloc_size_sds(tail->ptr); |
36c19d03 PN |
220 | tail = dupLastObjectIfNeeded(c->reply); |
221 | tail->ptr = sdscatlen(tail->ptr,s,sdslen(s)); | |
609baba8 | 222 | c->reply_bytes += zmalloc_size_sds(tail->ptr); |
36c19d03 | 223 | sdsfree(s); |
834ef78e | 224 | } else { |
36c19d03 | 225 | listAddNodeTail(c->reply,createObject(REDIS_STRING,s)); |
609baba8 | 226 | c->reply_bytes += zmalloc_size_sds(s); |
834ef78e | 227 | } |
36c19d03 | 228 | } |
7eac2a75 | 229 | asyncCloseClientOnOutputBufferLimitReached(c); |
36c19d03 PN |
230 | } |
231 | ||
232 | void _addReplyStringToList(redisClient *c, char *s, size_t len) { | |
233 | robj *tail; | |
25ef3192 | 234 | |
235 | if (c->flags & REDIS_CLOSE_AFTER_REPLY) return; | |
236 | ||
36c19d03 | 237 | if (listLength(c->reply) == 0) { |
442246dd | 238 | robj *o = createStringObject(s,len); |
239 | ||
240 | listAddNodeTail(c->reply,o); | |
609baba8 | 241 | c->reply_bytes += zmalloc_size_sds(o->ptr); |
834ef78e | 242 | } else { |
36c19d03 PN |
243 | tail = listNodeValue(listLast(c->reply)); |
244 | ||
245 | /* Append to this object when possible. */ | |
246 | if (tail->ptr != NULL && | |
247 | sdslen(tail->ptr)+len <= REDIS_REPLY_CHUNK_BYTES) | |
248 | { | |
609baba8 | 249 | c->reply_bytes -= zmalloc_size_sds(tail->ptr); |
36c19d03 PN |
250 | tail = dupLastObjectIfNeeded(c->reply); |
251 | tail->ptr = sdscatlen(tail->ptr,s,len); | |
609baba8 | 252 | c->reply_bytes += zmalloc_size_sds(tail->ptr); |
834ef78e | 253 | } else { |
442246dd | 254 | robj *o = createStringObject(s,len); |
255 | ||
256 | listAddNodeTail(c->reply,o); | |
609baba8 | 257 | c->reply_bytes += zmalloc_size_sds(o->ptr); |
834ef78e PN |
258 | } |
259 | } | |
7eac2a75 | 260 | asyncCloseClientOnOutputBufferLimitReached(c); |
834ef78e | 261 | } |
e2641e09 | 262 | |
25ef3192 | 263 | /* ----------------------------------------------------------------------------- |
264 | * Higher level functions to queue data on the client output buffer. | |
265 | * The following functions are the ones that commands implementations will call. | |
266 | * -------------------------------------------------------------------------- */ | |
267 | ||
834ef78e | 268 | void addReply(redisClient *c, robj *obj) { |
51669c5a | 269 | if (prepareClientToWrite(c) != REDIS_OK) return; |
4c2e506a | 270 | |
271 | /* This is an important place where we can avoid copy-on-write | |
272 | * when there is a saving child running, avoiding touching the | |
273 | * refcount field of the object if it's not needed. | |
274 | * | |
275 | * If the encoding is RAW and there is room in the static buffer | |
276 | * we'll be able to send the object to the client without | |
277 | * messing with its page. */ | |
278 | if (obj->encoding == REDIS_ENCODING_RAW) { | |
279 | if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK) | |
280 | _addReplyObjectToList(c,obj); | |
51669c5a | 281 | } else if (obj->encoding == REDIS_ENCODING_INT) { |
282 | /* Optimization: if there is room in the static buffer for 32 bytes | |
283 | * (more than the max chars a 64 bit integer can take as string) we | |
284 | * avoid decoding the object and go for the lower level approach. */ | |
285 | if (listLength(c->reply) == 0 && (sizeof(c->buf) - c->bufpos) >= 32) { | |
286 | char buf[32]; | |
287 | int len; | |
288 | ||
289 | len = ll2string(buf,sizeof(buf),(long)obj->ptr); | |
290 | if (_addReplyToBuffer(c,buf,len) == REDIS_OK) | |
291 | return; | |
292 | /* else... continue with the normal code path, but should never | |
293 | * happen actually since we verified there is room. */ | |
294 | } | |
834ef78e | 295 | obj = getDecodedObject(obj); |
4c2e506a | 296 | if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK) |
297 | _addReplyObjectToList(c,obj); | |
298 | decrRefCount(obj); | |
51669c5a | 299 | } else { |
300 | redisPanic("Wrong obj->encoding in addReply()"); | |
e2641e09 | 301 | } |
e2641e09 | 302 | } |
303 | ||
304 | void addReplySds(redisClient *c, sds s) { | |
51669c5a | 305 | if (prepareClientToWrite(c) != REDIS_OK) { |
cd76bb65 PN |
306 | /* The caller expects the sds to be free'd. */ |
307 | sdsfree(s); | |
308 | return; | |
309 | } | |
36c19d03 | 310 | if (_addReplyToBuffer(c,s,sdslen(s)) == REDIS_OK) { |
834ef78e PN |
311 | sdsfree(s); |
312 | } else { | |
36c19d03 PN |
313 | /* This method free's the sds when it is no longer needed. */ |
314 | _addReplySdsToList(c,s); | |
834ef78e | 315 | } |
e2641e09 | 316 | } |
317 | ||
834ef78e | 318 | void addReplyString(redisClient *c, char *s, size_t len) { |
51669c5a | 319 | if (prepareClientToWrite(c) != REDIS_OK) return; |
36c19d03 PN |
320 | if (_addReplyToBuffer(c,s,len) != REDIS_OK) |
321 | _addReplyStringToList(c,s,len); | |
834ef78e | 322 | } |
e2641e09 | 323 | |
51669c5a | 324 | void addReplyErrorLength(redisClient *c, char *s, size_t len) { |
3ab20376 PN |
325 | addReplyString(c,"-ERR ",5); |
326 | addReplyString(c,s,len); | |
327 | addReplyString(c,"\r\n",2); | |
e2641e09 | 328 | } |
329 | ||
3ab20376 | 330 | void addReplyError(redisClient *c, char *err) { |
51669c5a | 331 | addReplyErrorLength(c,err,strlen(err)); |
3ab20376 | 332 | } |
e2641e09 | 333 | |
3ab20376 | 334 | void addReplyErrorFormat(redisClient *c, const char *fmt, ...) { |
3bb818df | 335 | size_t l, j; |
3ab20376 PN |
336 | va_list ap; |
337 | va_start(ap,fmt); | |
338 | sds s = sdscatvprintf(sdsempty(),fmt,ap); | |
339 | va_end(ap); | |
3bb818df | 340 | /* Make sure there are no newlines in the string, otherwise invalid protocol |
341 | * is emitted. */ | |
342 | l = sdslen(s); | |
343 | for (j = 0; j < l; j++) { | |
344 | if (s[j] == '\r' || s[j] == '\n') s[j] = ' '; | |
345 | } | |
51669c5a | 346 | addReplyErrorLength(c,s,sdslen(s)); |
3ab20376 PN |
347 | sdsfree(s); |
348 | } | |
349 | ||
51669c5a | 350 | void addReplyStatusLength(redisClient *c, char *s, size_t len) { |
3ab20376 PN |
351 | addReplyString(c,"+",1); |
352 | addReplyString(c,s,len); | |
353 | addReplyString(c,"\r\n",2); | |
354 | } | |
355 | ||
356 | void addReplyStatus(redisClient *c, char *status) { | |
51669c5a | 357 | addReplyStatusLength(c,status,strlen(status)); |
3ab20376 PN |
358 | } |
359 | ||
360 | void addReplyStatusFormat(redisClient *c, const char *fmt, ...) { | |
361 | va_list ap; | |
362 | va_start(ap,fmt); | |
363 | sds s = sdscatvprintf(sdsempty(),fmt,ap); | |
364 | va_end(ap); | |
51669c5a | 365 | addReplyStatusLength(c,s,sdslen(s)); |
3ab20376 PN |
366 | sdsfree(s); |
367 | } | |
368 | ||
b301c1fc PN |
369 | /* Adds an empty object to the reply list that will contain the multi bulk |
370 | * length, which is not known when this function is called. */ | |
371 | void *addDeferredMultiBulkLength(redisClient *c) { | |
4c2e506a | 372 | /* Note that we install the write event here even if the object is not |
373 | * ready to be sent, since we are sure that before returning to the | |
374 | * event loop setDeferredMultiBulkLength() will be called. */ | |
51669c5a | 375 | if (prepareClientToWrite(c) != REDIS_OK) return NULL; |
36c19d03 | 376 | listAddNodeTail(c->reply,createObject(REDIS_STRING,NULL)); |
b301c1fc PN |
377 | return listLast(c->reply); |
378 | } | |
379 | ||
380 | /* Populate the length object and try glueing it to the next chunk. */ | |
381 | void setDeferredMultiBulkLength(redisClient *c, void *node, long length) { | |
382 | listNode *ln = (listNode*)node; | |
383 | robj *len, *next; | |
384 | ||
385 | /* Abort when *node is NULL (see addDeferredMultiBulkLength). */ | |
386 | if (node == NULL) return; | |
387 | ||
388 | len = listNodeValue(ln); | |
389 | len->ptr = sdscatprintf(sdsempty(),"*%ld\r\n",length); | |
609baba8 | 390 | c->reply_bytes += zmalloc_size_sds(len->ptr); |
b301c1fc PN |
391 | if (ln->next != NULL) { |
392 | next = listNodeValue(ln->next); | |
36c19d03 | 393 | |
49128f0b | 394 | /* Only glue when the next node is non-NULL (an sds in this case) */ |
36c19d03 | 395 | if (next->ptr != NULL) { |
6fe9c402 | 396 | c->reply_bytes -= zmalloc_size_sds(len->ptr); |
397 | c->reply_bytes -= zmalloc_size_sds(next->ptr); | |
49128f0b | 398 | len->ptr = sdscatlen(len->ptr,next->ptr,sdslen(next->ptr)); |
6fe9c402 | 399 | c->reply_bytes += zmalloc_size_sds(len->ptr); |
b301c1fc PN |
400 | listDelNode(c->reply,ln->next); |
401 | } | |
e2641e09 | 402 | } |
7eac2a75 | 403 | asyncCloseClientOnOutputBufferLimitReached(c); |
b301c1fc PN |
404 | } |
405 | ||
d51ebef5 | 406 | /* Add a duble as a bulk reply */ |
834ef78e PN |
407 | void addReplyDouble(redisClient *c, double d) { |
408 | char dbuf[128], sbuf[128]; | |
409 | int dlen, slen; | |
410 | dlen = snprintf(dbuf,sizeof(dbuf),"%.17g",d); | |
411 | slen = snprintf(sbuf,sizeof(sbuf),"$%d\r\n%s\r\n",dlen,dbuf); | |
412 | addReplyString(c,sbuf,slen); | |
e2641e09 | 413 | } |
414 | ||
d51ebef5 | 415 | /* Add a long long as integer reply or bulk len / multi bulk count. |
416 | * Basically this is used to output <prefix><long long><crlf>. */ | |
51669c5a | 417 | void addReplyLongLongWithPrefix(redisClient *c, long long ll, char prefix) { |
e2641e09 | 418 | char buf[128]; |
834ef78e | 419 | int len; |
355f8591 | 420 | |
421 | /* Things like $3\r\n or *2\r\n are emitted very often by the protocol | |
422 | * so we have a few shared objects to use if the integer is small | |
423 | * like it is most of the times. */ | |
424 | if (prefix == '*' && ll < REDIS_SHARED_BULKHDR_LEN) { | |
425 | addReply(c,shared.mbulkhdr[ll]); | |
426 | return; | |
427 | } else if (prefix == '$' && ll < REDIS_SHARED_BULKHDR_LEN) { | |
428 | addReply(c,shared.bulkhdr[ll]); | |
429 | return; | |
430 | } | |
431 | ||
834ef78e | 432 | buf[0] = prefix; |
e2641e09 | 433 | len = ll2string(buf+1,sizeof(buf)-1,ll); |
434 | buf[len+1] = '\r'; | |
435 | buf[len+2] = '\n'; | |
834ef78e | 436 | addReplyString(c,buf,len+3); |
e2641e09 | 437 | } |
438 | ||
834ef78e | 439 | void addReplyLongLong(redisClient *c, long long ll) { |
009db676 | 440 | if (ll == 0) |
441 | addReply(c,shared.czero); | |
442 | else if (ll == 1) | |
443 | addReply(c,shared.cone); | |
444 | else | |
51669c5a | 445 | addReplyLongLongWithPrefix(c,ll,':'); |
834ef78e | 446 | } |
e2641e09 | 447 | |
0537e7bf | 448 | void addReplyMultiBulkLen(redisClient *c, long length) { |
51669c5a | 449 | addReplyLongLongWithPrefix(c,length,'*'); |
e2641e09 | 450 | } |
451 | ||
d51ebef5 | 452 | /* Create the length prefix of a bulk reply, example: $2234 */ |
e2641e09 | 453 | void addReplyBulkLen(redisClient *c, robj *obj) { |
834ef78e | 454 | size_t len; |
e2641e09 | 455 | |
456 | if (obj->encoding == REDIS_ENCODING_RAW) { | |
457 | len = sdslen(obj->ptr); | |
458 | } else { | |
459 | long n = (long)obj->ptr; | |
460 | ||
461 | /* Compute how many bytes will take this integer as a radix 10 string */ | |
462 | len = 1; | |
463 | if (n < 0) { | |
464 | len++; | |
465 | n = -n; | |
466 | } | |
467 | while((n = n/10) != 0) { | |
468 | len++; | |
469 | } | |
470 | } | |
51669c5a | 471 | addReplyLongLongWithPrefix(c,len,'$'); |
e2641e09 | 472 | } |
473 | ||
d51ebef5 | 474 | /* Add a Redis Object as a bulk reply */ |
e2641e09 | 475 | void addReplyBulk(redisClient *c, robj *obj) { |
476 | addReplyBulkLen(c,obj); | |
477 | addReply(c,obj); | |
478 | addReply(c,shared.crlf); | |
479 | } | |
480 | ||
d51ebef5 | 481 | /* Add a C buffer as bulk reply */ |
482 | void addReplyBulkCBuffer(redisClient *c, void *p, size_t len) { | |
51669c5a | 483 | addReplyLongLongWithPrefix(c,len,'$'); |
d51ebef5 | 484 | addReplyString(c,p,len); |
485 | addReply(c,shared.crlf); | |
486 | } | |
487 | ||
488 | /* Add a C nul term string as bulk reply */ | |
e2641e09 | 489 | void addReplyBulkCString(redisClient *c, char *s) { |
490 | if (s == NULL) { | |
491 | addReply(c,shared.nullbulk); | |
492 | } else { | |
d51ebef5 | 493 | addReplyBulkCBuffer(c,s,strlen(s)); |
e2641e09 | 494 | } |
495 | } | |
496 | ||
d51ebef5 | 497 | /* Add a long long as a bulk reply */ |
498 | void addReplyBulkLongLong(redisClient *c, long long ll) { | |
499 | char buf[64]; | |
500 | int len; | |
501 | ||
502 | len = ll2string(buf,64,ll); | |
503 | addReplyBulkCBuffer(c,buf,len); | |
504 | } | |
505 | ||
1824e3a3 | 506 | /* Copy 'src' client output buffers into 'dst' client output buffers. |
507 | * The function takes care of freeing the old output buffers of the | |
508 | * destination client. */ | |
509 | void copyClientOutputBuffer(redisClient *dst, redisClient *src) { | |
510 | listRelease(dst->reply); | |
511 | dst->reply = listDup(src->reply); | |
512 | memcpy(dst->buf,src->buf,src->bufpos); | |
513 | dst->bufpos = src->bufpos; | |
3853c168 | 514 | dst->reply_bytes = src->reply_bytes; |
1824e3a3 | 515 | } |
516 | ||
b6ffa85f | 517 | static void acceptCommonHandler(int fd, int flags) { |
e2641e09 | 518 | redisClient *c; |
ab17b909 | 519 | if ((c = createClient(fd)) == NULL) { |
ed44a74e YF |
520 | redisLog(REDIS_WARNING,"Error allocating resources for the client"); |
521 | close(fd); /* May be already closed, just ignore errors */ | |
e2641e09 | 522 | return; |
523 | } | |
524 | /* If maxclient directive is set and this is one client more... close the | |
525 | * connection. Note that we create the client instead to check before | |
526 | * for this condition, since now the socket is already set in nonblocking | |
527 | * mode and we can send an error for free using the Kernel I/O */ | |
58732c23 | 528 | if (listLength(server.clients) > server.maxclients) { |
e2641e09 | 529 | char *err = "-ERR max number of clients reached\r\n"; |
530 | ||
531 | /* That's a best effort error message, don't check write errors */ | |
532 | if (write(c->fd,err,strlen(err)) == -1) { | |
533 | /* Nothing to do, Just to avoid the warning... */ | |
534 | } | |
3c95e721 | 535 | server.stat_rejected_conn++; |
e2641e09 | 536 | freeClient(c); |
537 | return; | |
538 | } | |
539 | server.stat_numconnections++; | |
b6ffa85f | 540 | c->flags |= flags; |
e2641e09 | 541 | } |
542 | ||
ab17b909 PN |
543 | void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) { |
544 | int cport, cfd; | |
545 | char cip[128]; | |
546 | REDIS_NOTUSED(el); | |
547 | REDIS_NOTUSED(mask); | |
548 | REDIS_NOTUSED(privdata); | |
549 | ||
550 | cfd = anetTcpAccept(server.neterr, fd, cip, &cport); | |
551 | if (cfd == AE_ERR) { | |
df541bea | 552 | redisLog(REDIS_WARNING,"Accepting client connection: %s", server.neterr); |
ab17b909 PN |
553 | return; |
554 | } | |
555 | redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport); | |
b6ffa85f | 556 | acceptCommonHandler(cfd,0); |
ab17b909 PN |
557 | } |
558 | ||
559 | void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) { | |
560 | int cfd; | |
ab17b909 PN |
561 | REDIS_NOTUSED(el); |
562 | REDIS_NOTUSED(mask); | |
563 | REDIS_NOTUSED(privdata); | |
564 | ||
4fe83b55 | 565 | cfd = anetUnixAccept(server.neterr, fd); |
ab17b909 | 566 | if (cfd == AE_ERR) { |
df541bea | 567 | redisLog(REDIS_WARNING,"Accepting client connection: %s", server.neterr); |
ab17b909 PN |
568 | return; |
569 | } | |
570 | redisLog(REDIS_VERBOSE,"Accepted connection to %s", server.unixsocket); | |
b6ffa85f | 571 | acceptCommonHandler(cfd,REDIS_UNIX_SOCKET); |
ab17b909 PN |
572 | } |
573 | ||
574 | ||
e2641e09 | 575 | static void freeClientArgv(redisClient *c) { |
576 | int j; | |
e2641e09 | 577 | for (j = 0; j < c->argc; j++) |
578 | decrRefCount(c->argv[j]); | |
e2641e09 | 579 | c->argc = 0; |
09e2d9ee | 580 | c->cmd = NULL; |
e2641e09 | 581 | } |
582 | ||
ed4d4f11 | 583 | /* Close all the slaves connections. This is useful in chained replication |
584 | * when we resync with our own master and want to force all our slaves to | |
585 | * resync with us as well. */ | |
586 | void disconnectSlaves(void) { | |
587 | while (listLength(server.slaves)) { | |
588 | listNode *ln = listFirst(server.slaves); | |
589 | freeClient((redisClient*)ln->value); | |
590 | } | |
591 | } | |
592 | ||
e2641e09 | 593 | void freeClient(redisClient *c) { |
594 | listNode *ln; | |
595 | ||
00010fa9 | 596 | /* If this is marked as current client unset it */ |
597 | if (server.current_client == c) server.current_client = NULL; | |
598 | ||
e2641e09 | 599 | /* Note that if the client we are freeing is blocked into a blocking |
600 | * call, we have to set querybuf to NULL *before* to call | |
601 | * unblockClientWaitingData() to avoid processInputBuffer() will get | |
602 | * called. Also it is important to remove the file events after | |
603 | * this, because this call adds the READABLE event. */ | |
604 | sdsfree(c->querybuf); | |
605 | c->querybuf = NULL; | |
606 | if (c->flags & REDIS_BLOCKED) | |
607 | unblockClientWaitingData(c); | |
984f6edf | 608 | dictRelease(c->bpop.keys); |
e2641e09 | 609 | |
610 | /* UNWATCH all the keys */ | |
611 | unwatchAllKeys(c); | |
612 | listRelease(c->watched_keys); | |
613 | /* Unsubscribe from all the pubsub channels */ | |
614 | pubsubUnsubscribeAllChannels(c,0); | |
615 | pubsubUnsubscribeAllPatterns(c,0); | |
616 | dictRelease(c->pubsub_channels); | |
617 | listRelease(c->pubsub_patterns); | |
618 | /* Obvious cleanup */ | |
619 | aeDeleteFileEvent(server.el,c->fd,AE_READABLE); | |
620 | aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE); | |
621 | listRelease(c->reply); | |
622 | freeClientArgv(c); | |
623 | close(c->fd); | |
624 | /* Remove from the list of clients */ | |
625 | ln = listSearchKey(server.clients,c); | |
626 | redisAssert(ln != NULL); | |
627 | listDelNode(server.clients,ln); | |
3bcffcbe PN |
628 | /* When client was just unblocked because of a blocking operation, |
629 | * remove it from the list with unblocked clients. */ | |
630 | if (c->flags & REDIS_UNBLOCKED) { | |
631 | ln = listSearchKey(server.unblocked_clients,c); | |
632 | redisAssert(ln != NULL); | |
633 | listDelNode(server.unblocked_clients,ln); | |
634 | } | |
e2641e09 | 635 | listRelease(c->io_keys); |
778b2210 | 636 | /* Master/slave cleanup. |
637 | * Case 1: we lost the connection with a slave. */ | |
e2641e09 | 638 | if (c->flags & REDIS_SLAVE) { |
639 | if (c->replstate == REDIS_REPL_SEND_BULK && c->repldbfd != -1) | |
640 | close(c->repldbfd); | |
641 | list *l = (c->flags & REDIS_MONITOR) ? server.monitors : server.slaves; | |
642 | ln = listSearchKey(l,c); | |
643 | redisAssert(ln != NULL); | |
644 | listDelNode(l,ln); | |
645 | } | |
778b2210 | 646 | |
647 | /* Case 2: we lost the connection with the master. */ | |
e2641e09 | 648 | if (c->flags & REDIS_MASTER) { |
649 | server.master = NULL; | |
1844f990 | 650 | server.repl_state = REDIS_REPL_CONNECT; |
56ff70f8 | 651 | server.repl_down_since = server.unixtime; |
ed4d4f11 | 652 | /* We lost connection with our master, force our slaves to resync |
653 | * with us as well to load the new data set. | |
d37299e3 | 654 | * |
f1e38b35 | 655 | * If server.masterhost is NULL the user called SLAVEOF NO ONE so |
ed4d4f11 | 656 | * slave resync is not needed. */ |
657 | if (server.masterhost != NULL) disconnectSlaves(); | |
e2641e09 | 658 | } |
7eac2a75 | 659 | |
660 | /* If this client was scheduled for async freeing we need to remove it | |
661 | * from the queue. */ | |
662 | if (c->flags & REDIS_CLOSE_ASAP) { | |
663 | ln = listSearchKey(server.clients_to_close,c); | |
664 | redisAssert(ln != NULL); | |
665 | listDelNode(server.clients_to_close,ln); | |
666 | } | |
667 | ||
e2641e09 | 668 | /* Release memory */ |
669 | zfree(c->argv); | |
e2641e09 | 670 | freeClientMultiState(c); |
671 | zfree(c); | |
672 | } | |
673 | ||
7eac2a75 | 674 | /* Schedule a client to free it at a safe time in the serverCron() function. |
675 | * This function is useful when we need to terminate a client but we are in | |
676 | * a context where calling freeClient() is not possible, because the client | |
677 | * should be valid for the continuation of the flow of the program. */ | |
678 | void freeClientAsync(redisClient *c) { | |
679 | if (c->flags & REDIS_CLOSE_ASAP) return; | |
680 | c->flags |= REDIS_CLOSE_ASAP; | |
681 | listAddNodeTail(server.clients_to_close,c); | |
682 | } | |
683 | ||
684 | void freeClientsInAsyncFreeQueue(void) { | |
685 | while (listLength(server.clients_to_close)) { | |
686 | listNode *ln = listFirst(server.clients_to_close); | |
687 | redisClient *c = listNodeValue(ln); | |
688 | ||
689 | c->flags &= ~REDIS_CLOSE_ASAP; | |
690 | freeClient(c); | |
691 | listDelNode(server.clients_to_close,ln); | |
692 | } | |
693 | } | |
694 | ||
e2641e09 | 695 | void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) { |
696 | redisClient *c = privdata; | |
697 | int nwritten = 0, totwritten = 0, objlen; | |
442246dd | 698 | size_t objmem; |
e2641e09 | 699 | robj *o; |
700 | REDIS_NOTUSED(el); | |
701 | REDIS_NOTUSED(mask); | |
702 | ||
834ef78e PN |
703 | while(c->bufpos > 0 || listLength(c->reply)) { |
704 | if (c->bufpos > 0) { | |
705 | if (c->flags & REDIS_MASTER) { | |
706 | /* Don't reply to a master */ | |
707 | nwritten = c->bufpos - c->sentlen; | |
708 | } else { | |
709 | nwritten = write(fd,c->buf+c->sentlen,c->bufpos-c->sentlen); | |
710 | if (nwritten <= 0) break; | |
711 | } | |
712 | c->sentlen += nwritten; | |
713 | totwritten += nwritten; | |
714 | ||
715 | /* If the buffer was sent, set bufpos to zero to continue with | |
716 | * the remainder of the reply. */ | |
717 | if (c->sentlen == c->bufpos) { | |
718 | c->bufpos = 0; | |
719 | c->sentlen = 0; | |
720 | } | |
721 | } else { | |
722 | o = listNodeValue(listFirst(c->reply)); | |
723 | objlen = sdslen(o->ptr); | |
609baba8 | 724 | objmem = zmalloc_size_sds(o->ptr); |
e2641e09 | 725 | |
834ef78e PN |
726 | if (objlen == 0) { |
727 | listDelNode(c->reply,listFirst(c->reply)); | |
728 | continue; | |
729 | } | |
e2641e09 | 730 | |
834ef78e PN |
731 | if (c->flags & REDIS_MASTER) { |
732 | /* Don't reply to a master */ | |
733 | nwritten = objlen - c->sentlen; | |
734 | } else { | |
735 | nwritten = write(fd, ((char*)o->ptr)+c->sentlen,objlen-c->sentlen); | |
736 | if (nwritten <= 0) break; | |
737 | } | |
738 | c->sentlen += nwritten; | |
739 | totwritten += nwritten; | |
e2641e09 | 740 | |
834ef78e PN |
741 | /* If we fully sent the object on head go to the next one */ |
742 | if (c->sentlen == objlen) { | |
743 | listDelNode(c->reply,listFirst(c->reply)); | |
744 | c->sentlen = 0; | |
442246dd | 745 | c->reply_bytes -= objmem; |
834ef78e | 746 | } |
e2641e09 | 747 | } |
f6b32c14 | 748 | /* Note that we avoid to send more than REDIS_MAX_WRITE_PER_EVENT |
e2641e09 | 749 | * bytes, in a single threaded server it's a good idea to serve |
750 | * other clients as well, even if a very large request comes from | |
751 | * super fast link that is always able to accept data (in real world | |
f6b32c14 | 752 | * scenario think about 'KEYS *' against the loopback interface). |
753 | * | |
754 | * However if we are over the maxmemory limit we ignore that and | |
755 | * just deliver as much data as it is possible to deliver. */ | |
756 | if (totwritten > REDIS_MAX_WRITE_PER_EVENT && | |
757 | (server.maxmemory == 0 || | |
758 | zmalloc_used_memory() < server.maxmemory)) break; | |
e2641e09 | 759 | } |
760 | if (nwritten == -1) { | |
761 | if (errno == EAGAIN) { | |
762 | nwritten = 0; | |
763 | } else { | |
764 | redisLog(REDIS_VERBOSE, | |
765 | "Error writing to client: %s", strerror(errno)); | |
766 | freeClient(c); | |
767 | return; | |
768 | } | |
769 | } | |
56ff70f8 | 770 | if (totwritten > 0) c->lastinteraction = server.unixtime; |
3bc89500 | 771 | if (c->bufpos == 0 && listLength(c->reply) == 0) { |
e2641e09 | 772 | c->sentlen = 0; |
773 | aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE); | |
941c9fa2 PN |
774 | |
775 | /* Close connection after entire reply has been sent. */ | |
cd8788f2 | 776 | if (c->flags & REDIS_CLOSE_AFTER_REPLY) freeClient(c); |
e2641e09 | 777 | } |
778 | } | |
779 | ||
e2641e09 | 780 | /* resetClient prepare the client to process the next command */ |
781 | void resetClient(redisClient *c) { | |
782 | freeClientArgv(c); | |
cd8788f2 PN |
783 | c->reqtype = 0; |
784 | c->multibulklen = 0; | |
e2641e09 | 785 | c->bulklen = -1; |
6856c7b4 | 786 | /* We clear the ASKING flag as well if we are not inside a MULTI. */ |
787 | if (!(c->flags & REDIS_MULTI)) c->flags &= (~REDIS_ASKING); | |
e2641e09 | 788 | } |
789 | ||
cd8788f2 PN |
790 | int processInlineBuffer(redisClient *c) { |
791 | char *newline = strstr(c->querybuf,"\r\n"); | |
792 | int argc, j; | |
793 | sds *argv; | |
794 | size_t querylen; | |
795 | ||
796 | /* Nothing to do without a \r\n */ | |
11e0c4c5 | 797 | if (newline == NULL) { |
798 | if (sdslen(c->querybuf) > REDIS_INLINE_MAX_SIZE) { | |
799 | addReplyError(c,"Protocol error: too big inline request"); | |
800 | setProtocolError(c,0); | |
801 | } | |
cd8788f2 | 802 | return REDIS_ERR; |
11e0c4c5 | 803 | } |
cd8788f2 PN |
804 | |
805 | /* Split the input buffer up to the \r\n */ | |
806 | querylen = newline-(c->querybuf); | |
807 | argv = sdssplitlen(c->querybuf,querylen," ",1,&argc); | |
808 | ||
809 | /* Leave data after the first line of the query in the buffer */ | |
810 | c->querybuf = sdsrange(c->querybuf,querylen+2,-1); | |
811 | ||
812 | /* Setup argv array on client structure */ | |
813 | if (c->argv) zfree(c->argv); | |
814 | c->argv = zmalloc(sizeof(robj*)*argc); | |
815 | ||
816 | /* Create redis objects for all arguments. */ | |
817 | for (c->argc = 0, j = 0; j < argc; j++) { | |
818 | if (sdslen(argv[j])) { | |
819 | c->argv[c->argc] = createObject(REDIS_STRING,argv[j]); | |
820 | c->argc++; | |
821 | } else { | |
822 | sdsfree(argv[j]); | |
823 | } | |
824 | } | |
825 | zfree(argv); | |
826 | return REDIS_OK; | |
827 | } | |
828 | ||
829 | /* Helper function. Trims query buffer to make the function that processes | |
830 | * multi bulk requests idempotent. */ | |
831 | static void setProtocolError(redisClient *c, int pos) { | |
3e0a975e | 832 | if (server.verbosity >= REDIS_VERBOSE) { |
833 | sds client = getClientInfoString(c); | |
834 | redisLog(REDIS_VERBOSE, | |
835 | "Protocol error from client: %s", client); | |
836 | sdsfree(client); | |
837 | } | |
cd8788f2 PN |
838 | c->flags |= REDIS_CLOSE_AFTER_REPLY; |
839 | c->querybuf = sdsrange(c->querybuf,pos,-1); | |
840 | } | |
841 | ||
842 | int processMultibulkBuffer(redisClient *c) { | |
843 | char *newline = NULL; | |
5af30201 PN |
844 | int pos = 0, ok; |
845 | long long ll; | |
cd8788f2 PN |
846 | |
847 | if (c->multibulklen == 0) { | |
848 | /* The client should have been reset */ | |
eab0e26e | 849 | redisAssertWithInfo(c,NULL,c->argc == 0); |
cd8788f2 PN |
850 | |
851 | /* Multi bulk length cannot be read without a \r\n */ | |
5af30201 | 852 | newline = strchr(c->querybuf,'\r'); |
11e0c4c5 | 853 | if (newline == NULL) { |
854 | if (sdslen(c->querybuf) > REDIS_INLINE_MAX_SIZE) { | |
855 | addReplyError(c,"Protocol error: too big mbulk count string"); | |
856 | setProtocolError(c,0); | |
857 | } | |
cd8788f2 | 858 | return REDIS_ERR; |
11e0c4c5 | 859 | } |
cd8788f2 | 860 | |
bf9fd5ff PN |
861 | /* Buffer should also contain \n */ |
862 | if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2)) | |
863 | return REDIS_ERR; | |
864 | ||
cd8788f2 PN |
865 | /* We know for sure there is a whole line since newline != NULL, |
866 | * so go ahead and find out the multi bulk length. */ | |
eab0e26e | 867 | redisAssertWithInfo(c,NULL,c->querybuf[0] == '*'); |
5af30201 PN |
868 | ok = string2ll(c->querybuf+1,newline-(c->querybuf+1),&ll); |
869 | if (!ok || ll > 1024*1024) { | |
b19c33d4 PN |
870 | addReplyError(c,"Protocol error: invalid multibulk length"); |
871 | setProtocolError(c,pos); | |
872 | return REDIS_ERR; | |
cd8788f2 | 873 | } |
af0e51f2 PN |
874 | |
875 | pos = (newline-c->querybuf)+2; | |
876 | if (ll <= 0) { | |
877 | c->querybuf = sdsrange(c->querybuf,pos,-1); | |
878 | return REDIS_OK; | |
879 | } | |
880 | ||
5af30201 | 881 | c->multibulklen = ll; |
cd8788f2 PN |
882 | |
883 | /* Setup argv array on client structure */ | |
884 | if (c->argv) zfree(c->argv); | |
885 | c->argv = zmalloc(sizeof(robj*)*c->multibulklen); | |
cd8788f2 PN |
886 | } |
887 | ||
eab0e26e | 888 | redisAssertWithInfo(c,NULL,c->multibulklen > 0); |
cd8788f2 PN |
889 | while(c->multibulklen) { |
890 | /* Read bulk length if unknown */ | |
891 | if (c->bulklen == -1) { | |
5af30201 | 892 | newline = strchr(c->querybuf+pos,'\r'); |
11e0c4c5 | 893 | if (newline == NULL) { |
894 | if (sdslen(c->querybuf) > REDIS_INLINE_MAX_SIZE) { | |
895 | addReplyError(c,"Protocol error: too big bulk count string"); | |
896 | setProtocolError(c,0); | |
897 | } | |
bf9fd5ff | 898 | break; |
11e0c4c5 | 899 | } |
bf9fd5ff PN |
900 | |
901 | /* Buffer should also contain \n */ | |
902 | if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2)) | |
cd8788f2 | 903 | break; |
bf9fd5ff PN |
904 | |
905 | if (c->querybuf[pos] != '$') { | |
906 | addReplyErrorFormat(c, | |
907 | "Protocol error: expected '$', got '%c'", | |
908 | c->querybuf[pos]); | |
909 | setProtocolError(c,pos); | |
910 | return REDIS_ERR; | |
e2641e09 | 911 | } |
bf9fd5ff PN |
912 | |
913 | ok = string2ll(c->querybuf+pos+1,newline-(c->querybuf+pos+1),&ll); | |
914 | if (!ok || ll < 0 || ll > 512*1024*1024) { | |
915 | addReplyError(c,"Protocol error: invalid bulk length"); | |
916 | setProtocolError(c,pos); | |
917 | return REDIS_ERR; | |
918 | } | |
919 | ||
920 | pos += newline-(c->querybuf+pos)+2; | |
94d490b9 | 921 | if (ll >= REDIS_MBULK_BIG_ARG) { |
826b5beb | 922 | /* If we are going to read a large object from network |
923 | * try to make it likely that it will start at c->querybuf | |
924 | * boundary so that we can optimized object creation | |
925 | * avoiding a large copy of data. */ | |
926 | c->querybuf = sdsrange(c->querybuf,pos,-1); | |
927 | pos = 0; | |
b9031458 | 928 | /* Hint the sds library about the amount of bytes this string is |
929 | * going to contain. */ | |
94d490b9 | 930 | c->querybuf = sdsMakeRoomFor(c->querybuf,ll+2); |
b9031458 | 931 | } |
bf9fd5ff | 932 | c->bulklen = ll; |
cd8788f2 PN |
933 | } |
934 | ||
935 | /* Read bulk argument */ | |
936 | if (sdslen(c->querybuf)-pos < (unsigned)(c->bulklen+2)) { | |
937 | /* Not enough data (+2 == trailing \r\n) */ | |
938 | break; | |
939 | } else { | |
92170955 | 940 | /* Optimization: if the buffer contanins JUST our bulk element |
941 | * instead of creating a new object by *copying* the sds we | |
942 | * just use the current sds string. */ | |
943 | if (pos == 0 && | |
94d490b9 | 944 | c->bulklen >= REDIS_MBULK_BIG_ARG && |
92170955 | 945 | (signed) sdslen(c->querybuf) == c->bulklen+2) |
946 | { | |
947 | c->argv[c->argc++] = createObject(REDIS_STRING,c->querybuf); | |
948 | sdsIncrLen(c->querybuf,-2); /* remove CRLF */ | |
949 | c->querybuf = sdsempty(); | |
950 | /* Assume that if we saw a fat argument we'll see another one | |
951 | * likely... */ | |
952 | c->querybuf = sdsMakeRoomFor(c->querybuf,c->bulklen+2); | |
953 | pos = 0; | |
954 | } else { | |
955 | c->argv[c->argc++] = | |
956 | createStringObject(c->querybuf+pos,c->bulklen); | |
957 | pos += c->bulklen+2; | |
958 | } | |
cd8788f2 PN |
959 | c->bulklen = -1; |
960 | c->multibulklen--; | |
961 | } | |
962 | } | |
963 | ||
964 | /* Trim to pos */ | |
92170955 | 965 | if (pos) c->querybuf = sdsrange(c->querybuf,pos,-1); |
cd8788f2 PN |
966 | |
967 | /* We're done when c->multibulk == 0 */ | |
11e0c4c5 | 968 | if (c->multibulklen == 0) return REDIS_OK; |
969 | ||
970 | /* Still not read to process the command */ | |
cd8788f2 PN |
971 | return REDIS_ERR; |
972 | } | |
973 | ||
974 | void processInputBuffer(redisClient *c) { | |
975 | /* Keep processing while there is something in the input buffer */ | |
976 | while(sdslen(c->querybuf)) { | |
64f201c2 HW |
977 | /* Immediately abort if the client is in the middle of something. */ |
978 | if (c->flags & REDIS_BLOCKED) return; | |
979 | ||
5e78edb3 PN |
980 | /* REDIS_CLOSE_AFTER_REPLY closes the connection once the reply is |
981 | * written to the client. Make sure to not let the reply grow after | |
982 | * this flag has been set (i.e. don't process more commands). */ | |
983 | if (c->flags & REDIS_CLOSE_AFTER_REPLY) return; | |
cd8788f2 PN |
984 | |
985 | /* Determine request type when unknown. */ | |
986 | if (!c->reqtype) { | |
987 | if (c->querybuf[0] == '*') { | |
988 | c->reqtype = REDIS_REQ_MULTIBULK; | |
e2641e09 | 989 | } else { |
cd8788f2 | 990 | c->reqtype = REDIS_REQ_INLINE; |
e2641e09 | 991 | } |
e2641e09 | 992 | } |
cd8788f2 PN |
993 | |
994 | if (c->reqtype == REDIS_REQ_INLINE) { | |
995 | if (processInlineBuffer(c) != REDIS_OK) break; | |
996 | } else if (c->reqtype == REDIS_REQ_MULTIBULK) { | |
997 | if (processMultibulkBuffer(c) != REDIS_OK) break; | |
998 | } else { | |
999 | redisPanic("Unknown request type"); | |
e2641e09 | 1000 | } |
cd8788f2 PN |
1001 | |
1002 | /* Multibulk processing could see a <= 0 length. */ | |
9da6caac PN |
1003 | if (c->argc == 0) { |
1004 | resetClient(c); | |
1005 | } else { | |
1006 | /* Only reset the client when the command was executed. */ | |
1007 | if (processCommand(c) == REDIS_OK) | |
1008 | resetClient(c); | |
1009 | } | |
e2641e09 | 1010 | } |
1011 | } | |
1012 | ||
1013 | void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) { | |
1014 | redisClient *c = (redisClient*) privdata; | |
826b5beb | 1015 | int nread, readlen; |
b8d743e1 | 1016 | size_t qblen; |
e2641e09 | 1017 | REDIS_NOTUSED(el); |
1018 | REDIS_NOTUSED(mask); | |
1019 | ||
00010fa9 | 1020 | server.current_client = c; |
826b5beb | 1021 | readlen = REDIS_IOBUF_LEN; |
1022 | /* If this is a multi bulk request, and we are processing a bulk reply | |
ed44a74e YF |
1023 | * that is large enough, try to maximize the probability that the query |
1024 | * buffer contains exactly the SDS string representing the object, even | |
1025 | * at the risk of requiring more read(2) calls. This way the function | |
826b5beb | 1026 | * processMultiBulkBuffer() can avoid copying buffers to create the |
1027 | * Redis Object representing the argument. */ | |
826b5beb | 1028 | if (c->reqtype == REDIS_REQ_MULTIBULK && c->multibulklen && c->bulklen != -1 |
94d490b9 | 1029 | && c->bulklen >= REDIS_MBULK_BIG_ARG) |
826b5beb | 1030 | { |
1031 | int remaining = (unsigned)(c->bulklen+2)-sdslen(c->querybuf); | |
1032 | ||
1033 | if (remaining < readlen) readlen = remaining; | |
1034 | } | |
826b5beb | 1035 | |
b8d743e1 | 1036 | qblen = sdslen(c->querybuf); |
9fa9ccb0 | 1037 | if (c->querybuf_peak < qblen) c->querybuf_peak = qblen; |
826b5beb | 1038 | c->querybuf = sdsMakeRoomFor(c->querybuf, readlen); |
1039 | nread = read(fd, c->querybuf+qblen, readlen); | |
e2641e09 | 1040 | if (nread == -1) { |
1041 | if (errno == EAGAIN) { | |
1042 | nread = 0; | |
1043 | } else { | |
1044 | redisLog(REDIS_VERBOSE, "Reading from client: %s",strerror(errno)); | |
1045 | freeClient(c); | |
1046 | return; | |
1047 | } | |
1048 | } else if (nread == 0) { | |
1049 | redisLog(REDIS_VERBOSE, "Client closed connection"); | |
1050 | freeClient(c); | |
1051 | return; | |
1052 | } | |
1053 | if (nread) { | |
b8d743e1 | 1054 | sdsIncrLen(c->querybuf,nread); |
56ff70f8 | 1055 | c->lastinteraction = server.unixtime; |
e2641e09 | 1056 | } else { |
00010fa9 | 1057 | server.current_client = NULL; |
e2641e09 | 1058 | return; |
1059 | } | |
becf5fdb | 1060 | if (sdslen(c->querybuf) > server.client_max_querybuf_len) { |
63fd1399 | 1061 | sds ci = getClientInfoString(c), bytes = sdsempty(); |
1062 | ||
1063 | bytes = sdscatrepr(bytes,c->querybuf,64); | |
1064 | redisLog(REDIS_WARNING,"Closing client that reached max query buffer length: %s (qbuf initial bytes: %s)", ci, bytes); | |
becf5fdb | 1065 | sdsfree(ci); |
63fd1399 | 1066 | sdsfree(bytes); |
becf5fdb | 1067 | freeClient(c); |
1068 | return; | |
1069 | } | |
e2641e09 | 1070 | processInputBuffer(c); |
00010fa9 | 1071 | server.current_client = NULL; |
e2641e09 | 1072 | } |
7a1fd61e | 1073 | |
1074 | void getClientsMaxBuffers(unsigned long *longest_output_list, | |
1075 | unsigned long *biggest_input_buffer) { | |
1076 | redisClient *c; | |
1077 | listNode *ln; | |
1078 | listIter li; | |
1079 | unsigned long lol = 0, bib = 0; | |
1080 | ||
1081 | listRewind(server.clients,&li); | |
1082 | while ((ln = listNext(&li)) != NULL) { | |
1083 | c = listNodeValue(ln); | |
1084 | ||
1085 | if (listLength(c->reply) > lol) lol = listLength(c->reply); | |
1086 | if (sdslen(c->querybuf) > bib) bib = sdslen(c->querybuf); | |
1087 | } | |
1088 | *longest_output_list = lol; | |
1089 | *biggest_input_buffer = bib; | |
1090 | } | |
1091 | ||
17d25a33 | 1092 | /* Turn a Redis client into an sds string representing its state. */ |
1093 | sds getClientInfoString(redisClient *client) { | |
6621b8ff | 1094 | char ip[32], flags[16], events[3], *p; |
b6ffa85f | 1095 | int port = 0; /* initialized to zero for the unix socket case. */ |
6621b8ff | 1096 | int emask; |
17d25a33 | 1097 | |
b6ffa85f | 1098 | if (!(client->flags & REDIS_UNIX_SOCKET)) |
1099 | anetPeerToString(client->fd,ip,&port); | |
17d25a33 | 1100 | p = flags; |
1101 | if (client->flags & REDIS_SLAVE) { | |
1102 | if (client->flags & REDIS_MONITOR) | |
1103 | *p++ = 'O'; | |
1104 | else | |
1105 | *p++ = 'S'; | |
1106 | } | |
1107 | if (client->flags & REDIS_MASTER) *p++ = 'M'; | |
17d25a33 | 1108 | if (client->flags & REDIS_MULTI) *p++ = 'x'; |
1109 | if (client->flags & REDIS_BLOCKED) *p++ = 'b'; | |
1110 | if (client->flags & REDIS_DIRTY_CAS) *p++ = 'd'; | |
1111 | if (client->flags & REDIS_CLOSE_AFTER_REPLY) *p++ = 'c'; | |
1112 | if (client->flags & REDIS_UNBLOCKED) *p++ = 'u'; | |
7eac2a75 | 1113 | if (client->flags & REDIS_CLOSE_ASAP) *p++ = 'A'; |
b6ffa85f | 1114 | if (client->flags & REDIS_UNIX_SOCKET) *p++ = 'U'; |
afd0f06b | 1115 | if (p == flags) *p++ = 'N'; |
17d25a33 | 1116 | *p++ = '\0'; |
6621b8ff | 1117 | |
1118 | emask = client->fd == -1 ? 0 : aeGetFileEvents(server.el,client->fd); | |
1119 | p = events; | |
1120 | if (emask & AE_READABLE) *p++ = 'r'; | |
1121 | if (emask & AE_WRITABLE) *p++ = 'w'; | |
1122 | *p = '\0'; | |
17d25a33 | 1123 | return sdscatprintf(sdsempty(), |
b162e6f1 | 1124 | "addr=%s:%d fd=%d age=%ld idle=%ld flags=%s db=%d sub=%d psub=%d multi=%d qbuf=%lu qbuf-free=%lu obl=%lu oll=%lu omem=%lu events=%s cmd=%s", |
b6ffa85f | 1125 | (client->flags & REDIS_UNIX_SOCKET) ? server.unixsocket : ip, |
1126 | port,client->fd, | |
56ff70f8 PH |
1127 | (long)(server.unixtime - client->ctime), |
1128 | (long)(server.unixtime - client->lastinteraction), | |
17d25a33 | 1129 | flags, |
1130 | client->db->id, | |
1131 | (int) dictSize(client->pubsub_channels), | |
491c1c4e | 1132 | (int) listLength(client->pubsub_patterns), |
b162e6f1 | 1133 | (client->flags & REDIS_MULTI) ? client->mstate.count : -1, |
491c1c4e | 1134 | (unsigned long) sdslen(client->querybuf), |
57a5e54d | 1135 | (unsigned long) sdsavail(client->querybuf), |
491c1c4e | 1136 | (unsigned long) client->bufpos, |
6621b8ff | 1137 | (unsigned long) listLength(client->reply), |
3853c168 | 1138 | getClientOutputBufferMemoryUsage(client), |
2c74a9f9 | 1139 | events, |
1140 | client->lastcmd ? client->lastcmd->name : "NULL"); | |
17d25a33 | 1141 | } |
1142 | ||
45e7a1ce | 1143 | sds getAllClientsInfoString(void) { |
1144 | listNode *ln; | |
1145 | listIter li; | |
1146 | redisClient *client; | |
1147 | sds o = sdsempty(); | |
1148 | ||
1149 | listRewind(server.clients,&li); | |
1150 | while ((ln = listNext(&li)) != NULL) { | |
0a466a75 | 1151 | sds cs; |
1152 | ||
45e7a1ce | 1153 | client = listNodeValue(ln); |
0a466a75 | 1154 | cs = getClientInfoString(client); |
1155 | o = sdscatsds(o,cs); | |
1156 | sdsfree(cs); | |
45e7a1ce | 1157 | o = sdscatlen(o,"\n",1); |
1158 | } | |
1159 | return o; | |
1160 | } | |
1161 | ||
3cd12b56 | 1162 | void clientCommand(redisClient *c) { |
b93fdb7b | 1163 | listNode *ln; |
1164 | listIter li; | |
1165 | redisClient *client; | |
1166 | ||
3cd12b56 | 1167 | if (!strcasecmp(c->argv[1]->ptr,"list") && c->argc == 2) { |
45e7a1ce | 1168 | sds o = getAllClientsInfoString(); |
3cd12b56 | 1169 | addReplyBulkCBuffer(c,o,sdslen(o)); |
1170 | sdsfree(o); | |
b93fdb7b | 1171 | } else if (!strcasecmp(c->argv[1]->ptr,"kill") && c->argc == 3) { |
1172 | listRewind(server.clients,&li); | |
1173 | while ((ln = listNext(&li)) != NULL) { | |
1174 | char ip[32], addr[64]; | |
1175 | int port; | |
1176 | ||
1177 | client = listNodeValue(ln); | |
1178 | if (anetPeerToString(client->fd,ip,&port) == -1) continue; | |
1179 | snprintf(addr,sizeof(addr),"%s:%d",ip,port); | |
1180 | if (strcmp(addr,c->argv[2]->ptr) == 0) { | |
1181 | addReply(c,shared.ok); | |
1182 | if (c == client) { | |
1183 | client->flags |= REDIS_CLOSE_AFTER_REPLY; | |
1184 | } else { | |
1185 | freeClient(client); | |
1186 | } | |
1187 | return; | |
1188 | } | |
1189 | } | |
1190 | addReplyError(c,"No such client"); | |
3cd12b56 | 1191 | } else { |
1192 | addReplyError(c, "Syntax error, try CLIENT (LIST | KILL ip:port)"); | |
1193 | } | |
1194 | } | |
c1c9d551 | 1195 | |
4dd444bb | 1196 | /* Rewrite the command vector of the client. All the new objects ref count |
1197 | * is incremented. The old command vector is freed, and the old objects | |
1198 | * ref count is decremented. */ | |
c1c9d551 | 1199 | void rewriteClientCommandVector(redisClient *c, int argc, ...) { |
1200 | va_list ap; | |
1201 | int j; | |
1202 | robj **argv; /* The new argument vector */ | |
1203 | ||
1204 | argv = zmalloc(sizeof(robj*)*argc); | |
1205 | va_start(ap,argc); | |
1206 | for (j = 0; j < argc; j++) { | |
1207 | robj *a; | |
1208 | ||
1209 | a = va_arg(ap, robj*); | |
1210 | argv[j] = a; | |
1211 | incrRefCount(a); | |
1212 | } | |
1213 | /* We free the objects in the original vector at the end, so we are | |
1214 | * sure that if the same objects are reused in the new vector the | |
1215 | * refcount gets incremented before it gets decremented. */ | |
1216 | for (j = 0; j < c->argc; j++) decrRefCount(c->argv[j]); | |
1217 | zfree(c->argv); | |
1218 | /* Replace argv and argc with our new versions. */ | |
1219 | c->argv = argv; | |
1220 | c->argc = argc; | |
09e2d9ee | 1221 | c->cmd = lookupCommand(c->argv[0]->ptr); |
eab0e26e | 1222 | redisAssertWithInfo(c,NULL,c->cmd != NULL); |
c1c9d551 | 1223 | va_end(ap); |
1224 | } | |
4dd444bb | 1225 | |
1226 | /* Rewrite a single item in the command vector. | |
1227 | * The new val ref count is incremented, and the old decremented. */ | |
1228 | void rewriteClientCommandArgument(redisClient *c, int i, robj *newval) { | |
1229 | robj *oldval; | |
1230 | ||
eab0e26e | 1231 | redisAssertWithInfo(c,NULL,i < c->argc); |
4dd444bb | 1232 | oldval = c->argv[i]; |
1233 | c->argv[i] = newval; | |
1234 | incrRefCount(newval); | |
1235 | decrRefCount(oldval); | |
1236 | ||
1237 | /* If this is the command name make sure to fix c->cmd. */ | |
1238 | if (i == 0) { | |
1239 | c->cmd = lookupCommand(c->argv[0]->ptr); | |
eab0e26e | 1240 | redisAssertWithInfo(c,NULL,c->cmd != NULL); |
4dd444bb | 1241 | } |
1242 | } | |
3853c168 | 1243 | |
1244 | /* This function returns the number of bytes that Redis is virtually | |
1245 | * using to store the reply still not read by the client. | |
1246 | * It is "virtual" since the reply output list may contain objects that | |
1247 | * are shared and are not really using additional memory. | |
1248 | * | |
1249 | * The function returns the total sum of the length of all the objects | |
1250 | * stored in the output list, plus the memory used to allocate every | |
1251 | * list node. The static reply buffer is not taken into account since it | |
1252 | * is allocated anyway. | |
1253 | * | |
1254 | * Note: this function is very fast so can be called as many time as | |
1255 | * the caller wishes. The main usage of this function currently is | |
2f0f0d95 | 1256 | * enforcing the client output length limits. */ |
3853c168 | 1257 | unsigned long getClientOutputBufferMemoryUsage(redisClient *c) { |
442246dd | 1258 | unsigned long list_item_size = sizeof(listNode)+sizeof(robj); |
3853c168 | 1259 | |
1260 | return c->reply_bytes + (list_item_size*listLength(c->reply)); | |
1261 | } | |
498dc555 | 1262 | |
ed44a74e | 1263 | /* Get the class of a client, used in order to enforce limits to different |
498dc555 | 1264 | * classes of clients. |
1265 | * | |
1266 | * The function will return one of the following: | |
1267 | * REDIS_CLIENT_LIMIT_CLASS_NORMAL -> Normal client | |
1268 | * REDIS_CLIENT_LIMIT_CLASS_SLAVE -> Slave or client executing MONITOR command | |
1269 | * REDIS_CLIENT_LIMIT_CLASS_PUBSUB -> Client subscribed to Pub/Sub channels | |
1270 | */ | |
1271 | int getClientLimitClass(redisClient *c) { | |
1272 | if (c->flags & REDIS_SLAVE) return REDIS_CLIENT_LIMIT_CLASS_SLAVE; | |
1273 | if (dictSize(c->pubsub_channels) || listLength(c->pubsub_patterns)) | |
1274 | return REDIS_CLIENT_LIMIT_CLASS_PUBSUB; | |
1275 | return REDIS_CLIENT_LIMIT_CLASS_NORMAL; | |
1276 | } | |
7eac2a75 | 1277 | |
7fe8d49a | 1278 | int getClientLimitClassByName(char *name) { |
1279 | if (!strcasecmp(name,"normal")) return REDIS_CLIENT_LIMIT_CLASS_NORMAL; | |
1280 | else if (!strcasecmp(name,"slave")) return REDIS_CLIENT_LIMIT_CLASS_SLAVE; | |
c715c9b8 | 1281 | else if (!strcasecmp(name,"pubsub")) return REDIS_CLIENT_LIMIT_CLASS_PUBSUB; |
7fe8d49a | 1282 | else return -1; |
1283 | } | |
1284 | ||
1285 | char *getClientLimitClassName(int class) { | |
1286 | switch(class) { | |
1287 | case REDIS_CLIENT_LIMIT_CLASS_NORMAL: return "normal"; | |
1288 | case REDIS_CLIENT_LIMIT_CLASS_SLAVE: return "slave"; | |
1289 | case REDIS_CLIENT_LIMIT_CLASS_PUBSUB: return "pubsub"; | |
1290 | default: return NULL; | |
1291 | } | |
1292 | } | |
1293 | ||
7eac2a75 | 1294 | /* The function checks if the client reached output buffer soft or hard |
1295 | * limit, and also update the state needed to check the soft limit as | |
1296 | * a side effect. | |
1297 | * | |
1298 | * Return value: non-zero if the client reached the soft or the hard limit. | |
1299 | * Otherwise zero is returned. */ | |
1300 | int checkClientOutputBufferLimits(redisClient *c) { | |
1301 | int soft = 0, hard = 0, class; | |
1302 | unsigned long used_mem = getClientOutputBufferMemoryUsage(c); | |
1303 | ||
1304 | class = getClientLimitClass(c); | |
1305 | if (server.client_obuf_limits[class].hard_limit_bytes && | |
1306 | used_mem >= server.client_obuf_limits[class].hard_limit_bytes) | |
1307 | hard = 1; | |
1308 | if (server.client_obuf_limits[class].soft_limit_bytes && | |
1309 | used_mem >= server.client_obuf_limits[class].soft_limit_bytes) | |
1310 | soft = 1; | |
1311 | ||
1312 | /* We need to check if the soft limit is reached continuously for the | |
1313 | * specified amount of seconds. */ | |
1314 | if (soft) { | |
1315 | if (c->obuf_soft_limit_reached_time == 0) { | |
1316 | c->obuf_soft_limit_reached_time = server.unixtime; | |
1317 | soft = 0; /* First time we see the soft limit reached */ | |
1318 | } else { | |
1319 | time_t elapsed = server.unixtime - c->obuf_soft_limit_reached_time; | |
1320 | ||
1321 | if (elapsed <= | |
1322 | server.client_obuf_limits[class].soft_limit_seconds) { | |
1323 | soft = 0; /* The client still did not reached the max number of | |
1324 | seconds for the soft limit to be considered | |
1325 | reached. */ | |
1326 | } | |
1327 | } | |
1328 | } else { | |
1329 | c->obuf_soft_limit_reached_time = 0; | |
1330 | } | |
1331 | return soft || hard; | |
1332 | } | |
1333 | ||
1334 | /* Asynchronously close a client if soft or hard limit is reached on the | |
06b3dced | 1335 | * output buffer size. The caller can check if the client will be closed |
1336 | * checking if the client REDIS_CLOSE_ASAP flag is set. | |
7eac2a75 | 1337 | * |
1338 | * Note: we need to close the client asynchronously because this function is | |
1339 | * called from contexts where the client can't be freed safely, i.e. from the | |
1340 | * lower level functions pushing data inside the client output buffers. */ | |
06b3dced | 1341 | void asyncCloseClientOnOutputBufferLimitReached(redisClient *c) { |
6fe9c402 | 1342 | redisAssert(c->reply_bytes < ULONG_MAX-(1024*64)); |
6e09ad1c | 1343 | if (c->reply_bytes == 0 || c->flags & REDIS_CLOSE_ASAP) return; |
7eac2a75 | 1344 | if (checkClientOutputBufferLimits(c)) { |
1345 | sds client = getClientInfoString(c); | |
1346 | ||
1347 | freeClientAsync(c); | |
7957c676 | 1348 | redisLog(REDIS_WARNING,"Client %s scheduled to be closed ASAP for overcoming of output buffer limits.", client); |
7eac2a75 | 1349 | sdsfree(client); |
7eac2a75 | 1350 | } |
1351 | } | |
8b7c3455 | 1352 | |
1353 | /* Helper function used by freeMemoryIfNeeded() in order to flush slaves | |
1354 | * output buffers without returning control to the event loop. */ | |
1355 | void flushSlavesOutputBuffers(void) { | |
1356 | listIter li; | |
1357 | listNode *ln; | |
1358 | ||
1359 | listRewind(server.slaves,&li); | |
1360 | while((ln = listNext(&li))) { | |
1361 | redisClient *slave = listNodeValue(ln); | |
1362 | int events; | |
1363 | ||
1364 | events = aeGetFileEvents(server.el,slave->fd); | |
1365 | if (events & AE_WRITABLE && | |
1366 | slave->replstate == REDIS_REPL_ONLINE && | |
1367 | listLength(slave->reply)) | |
1368 | { | |
1369 | sendReplyToClient(server.el,slave->fd,slave,0); | |
1370 | } | |
1371 | } | |
1372 | } |