]>
Commit | Line | Data |
---|---|---|
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 | ||
30 | #include "redis.h" | |
31 | #include <sys/uio.h> | |
32 | ||
33 | static void setProtocolError(redisClient *c, int pos); | |
34 | ||
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 | ||
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) { | |
53 | redisClient *c = zmalloc(sizeof(redisClient)); | |
54 | ||
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 | } | |
69 | } | |
70 | ||
71 | selectDb(c,0); | |
72 | c->fd = fd; | |
73 | c->bufpos = 0; | |
74 | c->querybuf = sdsempty(); | |
75 | c->querybuf_peak = 0; | |
76 | c->reqtype = 0; | |
77 | c->argc = 0; | |
78 | c->argv = NULL; | |
79 | c->cmd = c->lastcmd = NULL; | |
80 | c->multibulklen = 0; | |
81 | c->bulklen = -1; | |
82 | c->sentlen = 0; | |
83 | c->flags = 0; | |
84 | c->ctime = c->lastinteraction = server.unixtime; | |
85 | c->authenticated = 0; | |
86 | c->replstate = REDIS_REPL_NONE; | |
87 | c->slave_listening_port = 0; | |
88 | c->reply = listCreate(); | |
89 | c->reply_bytes = 0; | |
90 | c->obuf_soft_limit_reached_time = 0; | |
91 | listSetFreeMethod(c->reply,decrRefCount); | |
92 | listSetDupMethod(c->reply,dupClientReplyValue); | |
93 | c->bpop.keys = dictCreate(&setDictType,NULL); | |
94 | c->bpop.timeout = 0; | |
95 | c->bpop.target = NULL; | |
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); | |
103 | if (fd != -1) listAddNodeTail(server.clients,c); | |
104 | initClientMultiState(c); | |
105 | return c; | |
106 | } | |
107 | ||
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) { | |
123 | if (c->flags & REDIS_LUA_CLIENT) return REDIS_OK; | |
124 | if (c->fd <= 0) return REDIS_ERR; /* Fake client */ | |
125 | if (c->bufpos == 0 && listLength(c->reply) == 0 && | |
126 | (c->replstate == REDIS_REPL_NONE || | |
127 | c->replstate == REDIS_REPL_ONLINE) && | |
128 | aeCreateFileEvent(server.el, c->fd, AE_WRITABLE, | |
129 | sendReplyToClient, c) == AE_ERR) return REDIS_ERR; | |
130 | return REDIS_OK; | |
131 | } | |
132 | ||
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); | |
147 | } | |
148 | ||
149 | /* ----------------------------------------------------------------------------- | |
150 | * Low level functions to add more data to output buffers. | |
151 | * -------------------------------------------------------------------------- */ | |
152 | ||
153 | int _addReplyToBuffer(redisClient *c, char *s, size_t len) { | |
154 | size_t available = sizeof(c->buf)-c->bufpos; | |
155 | ||
156 | if (c->flags & REDIS_CLOSE_AFTER_REPLY) return REDIS_OK; | |
157 | ||
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; | |
164 | ||
165 | memcpy(c->buf+c->bufpos,s,len); | |
166 | c->bufpos+=len; | |
167 | return REDIS_OK; | |
168 | } | |
169 | ||
170 | void _addReplyObjectToList(redisClient *c, robj *o) { | |
171 | robj *tail; | |
172 | ||
173 | if (c->flags & REDIS_CLOSE_AFTER_REPLY) return; | |
174 | ||
175 | if (listLength(c->reply) == 0) { | |
176 | incrRefCount(o); | |
177 | listAddNodeTail(c->reply,o); | |
178 | c->reply_bytes += zmalloc_size_sds(o->ptr); | |
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 | { | |
186 | c->reply_bytes -= zmalloc_size_sds(tail->ptr); | |
187 | tail = dupLastObjectIfNeeded(c->reply); | |
188 | tail->ptr = sdscatlen(tail->ptr,o->ptr,sdslen(o->ptr)); | |
189 | c->reply_bytes += zmalloc_size_sds(tail->ptr); | |
190 | } else { | |
191 | incrRefCount(o); | |
192 | listAddNodeTail(c->reply,o); | |
193 | c->reply_bytes += zmalloc_size_sds(o->ptr); | |
194 | } | |
195 | } | |
196 | asyncCloseClientOnOutputBufferLimitReached(c); | |
197 | } | |
198 | ||
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; | |
203 | ||
204 | if (c->flags & REDIS_CLOSE_AFTER_REPLY) { | |
205 | sdsfree(s); | |
206 | return; | |
207 | } | |
208 | ||
209 | if (listLength(c->reply) == 0) { | |
210 | listAddNodeTail(c->reply,createObject(REDIS_STRING,s)); | |
211 | c->reply_bytes += zmalloc_size_sds(s); | |
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 | { | |
219 | c->reply_bytes -= zmalloc_size_sds(tail->ptr); | |
220 | tail = dupLastObjectIfNeeded(c->reply); | |
221 | tail->ptr = sdscatlen(tail->ptr,s,sdslen(s)); | |
222 | c->reply_bytes += zmalloc_size_sds(tail->ptr); | |
223 | sdsfree(s); | |
224 | } else { | |
225 | listAddNodeTail(c->reply,createObject(REDIS_STRING,s)); | |
226 | c->reply_bytes += zmalloc_size_sds(s); | |
227 | } | |
228 | } | |
229 | asyncCloseClientOnOutputBufferLimitReached(c); | |
230 | } | |
231 | ||
232 | void _addReplyStringToList(redisClient *c, char *s, size_t len) { | |
233 | robj *tail; | |
234 | ||
235 | if (c->flags & REDIS_CLOSE_AFTER_REPLY) return; | |
236 | ||
237 | if (listLength(c->reply) == 0) { | |
238 | robj *o = createStringObject(s,len); | |
239 | ||
240 | listAddNodeTail(c->reply,o); | |
241 | c->reply_bytes += zmalloc_size_sds(o->ptr); | |
242 | } else { | |
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 | { | |
249 | c->reply_bytes -= zmalloc_size_sds(tail->ptr); | |
250 | tail = dupLastObjectIfNeeded(c->reply); | |
251 | tail->ptr = sdscatlen(tail->ptr,s,len); | |
252 | c->reply_bytes += zmalloc_size_sds(tail->ptr); | |
253 | } else { | |
254 | robj *o = createStringObject(s,len); | |
255 | ||
256 | listAddNodeTail(c->reply,o); | |
257 | c->reply_bytes += zmalloc_size_sds(o->ptr); | |
258 | } | |
259 | } | |
260 | asyncCloseClientOnOutputBufferLimitReached(c); | |
261 | } | |
262 | ||
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 | ||
268 | void addReply(redisClient *c, robj *obj) { | |
269 | if (prepareClientToWrite(c) != REDIS_OK) return; | |
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); | |
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 | } | |
295 | obj = getDecodedObject(obj); | |
296 | if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK) | |
297 | _addReplyObjectToList(c,obj); | |
298 | decrRefCount(obj); | |
299 | } else { | |
300 | redisPanic("Wrong obj->encoding in addReply()"); | |
301 | } | |
302 | } | |
303 | ||
304 | void addReplySds(redisClient *c, sds s) { | |
305 | if (prepareClientToWrite(c) != REDIS_OK) { | |
306 | /* The caller expects the sds to be free'd. */ | |
307 | sdsfree(s); | |
308 | return; | |
309 | } | |
310 | if (_addReplyToBuffer(c,s,sdslen(s)) == REDIS_OK) { | |
311 | sdsfree(s); | |
312 | } else { | |
313 | /* This method free's the sds when it is no longer needed. */ | |
314 | _addReplySdsToList(c,s); | |
315 | } | |
316 | } | |
317 | ||
318 | void addReplyString(redisClient *c, char *s, size_t len) { | |
319 | if (prepareClientToWrite(c) != REDIS_OK) return; | |
320 | if (_addReplyToBuffer(c,s,len) != REDIS_OK) | |
321 | _addReplyStringToList(c,s,len); | |
322 | } | |
323 | ||
324 | void addReplyErrorLength(redisClient *c, char *s, size_t len) { | |
325 | addReplyString(c,"-ERR ",5); | |
326 | addReplyString(c,s,len); | |
327 | addReplyString(c,"\r\n",2); | |
328 | } | |
329 | ||
330 | void addReplyError(redisClient *c, char *err) { | |
331 | addReplyErrorLength(c,err,strlen(err)); | |
332 | } | |
333 | ||
334 | void addReplyErrorFormat(redisClient *c, const char *fmt, ...) { | |
335 | size_t l, j; | |
336 | va_list ap; | |
337 | va_start(ap,fmt); | |
338 | sds s = sdscatvprintf(sdsempty(),fmt,ap); | |
339 | va_end(ap); | |
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 | } | |
346 | addReplyErrorLength(c,s,sdslen(s)); | |
347 | sdsfree(s); | |
348 | } | |
349 | ||
350 | void addReplyStatusLength(redisClient *c, char *s, size_t len) { | |
351 | addReplyString(c,"+",1); | |
352 | addReplyString(c,s,len); | |
353 | addReplyString(c,"\r\n",2); | |
354 | } | |
355 | ||
356 | void addReplyStatus(redisClient *c, char *status) { | |
357 | addReplyStatusLength(c,status,strlen(status)); | |
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); | |
365 | addReplyStatusLength(c,s,sdslen(s)); | |
366 | sdsfree(s); | |
367 | } | |
368 | ||
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) { | |
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. */ | |
375 | if (prepareClientToWrite(c) != REDIS_OK) return NULL; | |
376 | listAddNodeTail(c->reply,createObject(REDIS_STRING,NULL)); | |
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); | |
390 | c->reply_bytes += zmalloc_size_sds(len->ptr); | |
391 | if (ln->next != NULL) { | |
392 | next = listNodeValue(ln->next); | |
393 | ||
394 | /* Only glue when the next node is non-NULL (an sds in this case) */ | |
395 | if (next->ptr != NULL) { | |
396 | c->reply_bytes -= zmalloc_size_sds(len->ptr); | |
397 | c->reply_bytes -= zmalloc_size_sds(next->ptr); | |
398 | len->ptr = sdscatlen(len->ptr,next->ptr,sdslen(next->ptr)); | |
399 | c->reply_bytes += zmalloc_size_sds(len->ptr); | |
400 | listDelNode(c->reply,ln->next); | |
401 | } | |
402 | } | |
403 | asyncCloseClientOnOutputBufferLimitReached(c); | |
404 | } | |
405 | ||
406 | /* Add a duble as a bulk reply */ | |
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); | |
413 | } | |
414 | ||
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>. */ | |
417 | void addReplyLongLongWithPrefix(redisClient *c, long long ll, char prefix) { | |
418 | char buf[128]; | |
419 | int len; | |
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 | ||
432 | buf[0] = prefix; | |
433 | len = ll2string(buf+1,sizeof(buf)-1,ll); | |
434 | buf[len+1] = '\r'; | |
435 | buf[len+2] = '\n'; | |
436 | addReplyString(c,buf,len+3); | |
437 | } | |
438 | ||
439 | void addReplyLongLong(redisClient *c, long long ll) { | |
440 | if (ll == 0) | |
441 | addReply(c,shared.czero); | |
442 | else if (ll == 1) | |
443 | addReply(c,shared.cone); | |
444 | else | |
445 | addReplyLongLongWithPrefix(c,ll,':'); | |
446 | } | |
447 | ||
448 | void addReplyMultiBulkLen(redisClient *c, long length) { | |
449 | addReplyLongLongWithPrefix(c,length,'*'); | |
450 | } | |
451 | ||
452 | /* Create the length prefix of a bulk reply, example: $2234 */ | |
453 | void addReplyBulkLen(redisClient *c, robj *obj) { | |
454 | size_t len; | |
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 | } | |
471 | addReplyLongLongWithPrefix(c,len,'$'); | |
472 | } | |
473 | ||
474 | /* Add a Redis Object as a bulk reply */ | |
475 | void addReplyBulk(redisClient *c, robj *obj) { | |
476 | addReplyBulkLen(c,obj); | |
477 | addReply(c,obj); | |
478 | addReply(c,shared.crlf); | |
479 | } | |
480 | ||
481 | /* Add a C buffer as bulk reply */ | |
482 | void addReplyBulkCBuffer(redisClient *c, void *p, size_t len) { | |
483 | addReplyLongLongWithPrefix(c,len,'$'); | |
484 | addReplyString(c,p,len); | |
485 | addReply(c,shared.crlf); | |
486 | } | |
487 | ||
488 | /* Add a C nul term string as bulk reply */ | |
489 | void addReplyBulkCString(redisClient *c, char *s) { | |
490 | if (s == NULL) { | |
491 | addReply(c,shared.nullbulk); | |
492 | } else { | |
493 | addReplyBulkCBuffer(c,s,strlen(s)); | |
494 | } | |
495 | } | |
496 | ||
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 | ||
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; | |
514 | dst->reply_bytes = src->reply_bytes; | |
515 | } | |
516 | ||
517 | static void acceptCommonHandler(int fd, int flags) { | |
518 | redisClient *c; | |
519 | if ((c = createClient(fd)) == NULL) { | |
520 | redisLog(REDIS_WARNING,"Error allocating resources for the client"); | |
521 | close(fd); /* May be already closed, just ignore errors */ | |
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 */ | |
528 | if (listLength(server.clients) > server.maxclients) { | |
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 | } | |
535 | server.stat_rejected_conn++; | |
536 | freeClient(c); | |
537 | return; | |
538 | } | |
539 | server.stat_numconnections++; | |
540 | c->flags |= flags; | |
541 | } | |
542 | ||
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) { | |
552 | redisLog(REDIS_WARNING,"Accepting client connection: %s", server.neterr); | |
553 | return; | |
554 | } | |
555 | redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport); | |
556 | acceptCommonHandler(cfd,0); | |
557 | } | |
558 | ||
559 | void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) { | |
560 | int cfd; | |
561 | REDIS_NOTUSED(el); | |
562 | REDIS_NOTUSED(mask); | |
563 | REDIS_NOTUSED(privdata); | |
564 | ||
565 | cfd = anetUnixAccept(server.neterr, fd); | |
566 | if (cfd == AE_ERR) { | |
567 | redisLog(REDIS_WARNING,"Accepting client connection: %s", server.neterr); | |
568 | return; | |
569 | } | |
570 | redisLog(REDIS_VERBOSE,"Accepted connection to %s", server.unixsocket); | |
571 | acceptCommonHandler(cfd,REDIS_UNIX_SOCKET); | |
572 | } | |
573 | ||
574 | ||
575 | static void freeClientArgv(redisClient *c) { | |
576 | int j; | |
577 | for (j = 0; j < c->argc; j++) | |
578 | decrRefCount(c->argv[j]); | |
579 | c->argc = 0; | |
580 | c->cmd = NULL; | |
581 | } | |
582 | ||
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 | ||
593 | void freeClient(redisClient *c) { | |
594 | listNode *ln; | |
595 | ||
596 | /* If this is marked as current client unset it */ | |
597 | if (server.current_client == c) server.current_client = NULL; | |
598 | ||
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); | |
608 | dictRelease(c->bpop.keys); | |
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); | |
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 | } | |
635 | listRelease(c->io_keys); | |
636 | /* Master/slave cleanup. | |
637 | * Case 1: we lost the connection with a slave. */ | |
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 | } | |
646 | ||
647 | /* Case 2: we lost the connection with the master. */ | |
648 | if (c->flags & REDIS_MASTER) { | |
649 | server.master = NULL; | |
650 | server.repl_state = REDIS_REPL_CONNECT; | |
651 | server.repl_down_since = server.unixtime; | |
652 | /* We lost connection with our master, force our slaves to resync | |
653 | * with us as well to load the new data set. | |
654 | * | |
655 | * If server.masterhost is NULL the user called SLAVEOF NO ONE so | |
656 | * slave resync is not needed. */ | |
657 | if (server.masterhost != NULL) disconnectSlaves(); | |
658 | } | |
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 | ||
668 | /* Release memory */ | |
669 | zfree(c->argv); | |
670 | freeClientMultiState(c); | |
671 | zfree(c); | |
672 | } | |
673 | ||
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 | ||
695 | void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) { | |
696 | redisClient *c = privdata; | |
697 | int nwritten = 0, totwritten = 0, objlen; | |
698 | size_t objmem; | |
699 | robj *o; | |
700 | REDIS_NOTUSED(el); | |
701 | REDIS_NOTUSED(mask); | |
702 | ||
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); | |
724 | objmem = zmalloc_size_sds(o->ptr); | |
725 | ||
726 | if (objlen == 0) { | |
727 | listDelNode(c->reply,listFirst(c->reply)); | |
728 | continue; | |
729 | } | |
730 | ||
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; | |
740 | ||
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; | |
745 | c->reply_bytes -= objmem; | |
746 | } | |
747 | } | |
748 | /* Note that we avoid to send more than REDIS_MAX_WRITE_PER_EVENT | |
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 | |
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; | |
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 | } | |
770 | if (totwritten > 0) c->lastinteraction = server.unixtime; | |
771 | if (c->bufpos == 0 && listLength(c->reply) == 0) { | |
772 | c->sentlen = 0; | |
773 | aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE); | |
774 | ||
775 | /* Close connection after entire reply has been sent. */ | |
776 | if (c->flags & REDIS_CLOSE_AFTER_REPLY) freeClient(c); | |
777 | } | |
778 | } | |
779 | ||
780 | /* resetClient prepare the client to process the next command */ | |
781 | void resetClient(redisClient *c) { | |
782 | freeClientArgv(c); | |
783 | c->reqtype = 0; | |
784 | c->multibulklen = 0; | |
785 | c->bulklen = -1; | |
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); | |
788 | } | |
789 | ||
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 */ | |
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 | } | |
802 | return REDIS_ERR; | |
803 | } | |
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) { | |
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 | } | |
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; | |
844 | int pos = 0, ok; | |
845 | long long ll; | |
846 | ||
847 | if (c->multibulklen == 0) { | |
848 | /* The client should have been reset */ | |
849 | redisAssertWithInfo(c,NULL,c->argc == 0); | |
850 | ||
851 | /* Multi bulk length cannot be read without a \r\n */ | |
852 | newline = strchr(c->querybuf,'\r'); | |
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 | } | |
858 | return REDIS_ERR; | |
859 | } | |
860 | ||
861 | /* Buffer should also contain \n */ | |
862 | if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2)) | |
863 | return REDIS_ERR; | |
864 | ||
865 | /* We know for sure there is a whole line since newline != NULL, | |
866 | * so go ahead and find out the multi bulk length. */ | |
867 | redisAssertWithInfo(c,NULL,c->querybuf[0] == '*'); | |
868 | ok = string2ll(c->querybuf+1,newline-(c->querybuf+1),&ll); | |
869 | if (!ok || ll > 1024*1024) { | |
870 | addReplyError(c,"Protocol error: invalid multibulk length"); | |
871 | setProtocolError(c,pos); | |
872 | return REDIS_ERR; | |
873 | } | |
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 | ||
881 | c->multibulklen = ll; | |
882 | ||
883 | /* Setup argv array on client structure */ | |
884 | if (c->argv) zfree(c->argv); | |
885 | c->argv = zmalloc(sizeof(robj*)*c->multibulklen); | |
886 | } | |
887 | ||
888 | redisAssertWithInfo(c,NULL,c->multibulklen > 0); | |
889 | while(c->multibulklen) { | |
890 | /* Read bulk length if unknown */ | |
891 | if (c->bulklen == -1) { | |
892 | newline = strchr(c->querybuf+pos,'\r'); | |
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 | } | |
898 | break; | |
899 | } | |
900 | ||
901 | /* Buffer should also contain \n */ | |
902 | if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2)) | |
903 | break; | |
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; | |
911 | } | |
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; | |
921 | if (ll >= REDIS_MBULK_BIG_ARG) { | |
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; | |
928 | /* Hint the sds library about the amount of bytes this string is | |
929 | * going to contain. */ | |
930 | c->querybuf = sdsMakeRoomFor(c->querybuf,ll+2); | |
931 | } | |
932 | c->bulklen = ll; | |
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 { | |
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 && | |
944 | c->bulklen >= REDIS_MBULK_BIG_ARG && | |
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 | } | |
959 | c->bulklen = -1; | |
960 | c->multibulklen--; | |
961 | } | |
962 | } | |
963 | ||
964 | /* Trim to pos */ | |
965 | if (pos) c->querybuf = sdsrange(c->querybuf,pos,-1); | |
966 | ||
967 | /* We're done when c->multibulk == 0 */ | |
968 | if (c->multibulklen == 0) return REDIS_OK; | |
969 | ||
970 | /* Still not read to process the command */ | |
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)) { | |
977 | /* Immediately abort if the client is in the middle of something. */ | |
978 | if (c->flags & REDIS_BLOCKED) return; | |
979 | ||
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; | |
984 | ||
985 | /* Determine request type when unknown. */ | |
986 | if (!c->reqtype) { | |
987 | if (c->querybuf[0] == '*') { | |
988 | c->reqtype = REDIS_REQ_MULTIBULK; | |
989 | } else { | |
990 | c->reqtype = REDIS_REQ_INLINE; | |
991 | } | |
992 | } | |
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"); | |
1000 | } | |
1001 | ||
1002 | /* Multibulk processing could see a <= 0 length. */ | |
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 | } | |
1010 | } | |
1011 | } | |
1012 | ||
1013 | void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) { | |
1014 | redisClient *c = (redisClient*) privdata; | |
1015 | int nread, readlen; | |
1016 | size_t qblen; | |
1017 | REDIS_NOTUSED(el); | |
1018 | REDIS_NOTUSED(mask); | |
1019 | ||
1020 | server.current_client = c; | |
1021 | readlen = REDIS_IOBUF_LEN; | |
1022 | /* If this is a multi bulk request, and we are processing a bulk reply | |
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 | |
1026 | * processMultiBulkBuffer() can avoid copying buffers to create the | |
1027 | * Redis Object representing the argument. */ | |
1028 | if (c->reqtype == REDIS_REQ_MULTIBULK && c->multibulklen && c->bulklen != -1 | |
1029 | && c->bulklen >= REDIS_MBULK_BIG_ARG) | |
1030 | { | |
1031 | int remaining = (unsigned)(c->bulklen+2)-sdslen(c->querybuf); | |
1032 | ||
1033 | if (remaining < readlen) readlen = remaining; | |
1034 | } | |
1035 | ||
1036 | qblen = sdslen(c->querybuf); | |
1037 | if (c->querybuf_peak < qblen) c->querybuf_peak = qblen; | |
1038 | c->querybuf = sdsMakeRoomFor(c->querybuf, readlen); | |
1039 | nread = read(fd, c->querybuf+qblen, readlen); | |
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) { | |
1054 | sdsIncrLen(c->querybuf,nread); | |
1055 | c->lastinteraction = server.unixtime; | |
1056 | } else { | |
1057 | server.current_client = NULL; | |
1058 | return; | |
1059 | } | |
1060 | if (sdslen(c->querybuf) > server.client_max_querybuf_len) { | |
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); | |
1065 | sdsfree(ci); | |
1066 | sdsfree(bytes); | |
1067 | freeClient(c); | |
1068 | return; | |
1069 | } | |
1070 | processInputBuffer(c); | |
1071 | server.current_client = NULL; | |
1072 | } | |
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 | ||
1092 | /* Turn a Redis client into an sds string representing its state. */ | |
1093 | sds getClientInfoString(redisClient *client) { | |
1094 | char ip[32], flags[16], events[3], *p; | |
1095 | int port = 0; /* initialized to zero for the unix socket case. */ | |
1096 | int emask; | |
1097 | ||
1098 | if (!(client->flags & REDIS_UNIX_SOCKET)) | |
1099 | anetPeerToString(client->fd,ip,&port); | |
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'; | |
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'; | |
1113 | if (client->flags & REDIS_CLOSE_ASAP) *p++ = 'A'; | |
1114 | if (client->flags & REDIS_UNIX_SOCKET) *p++ = 'U'; | |
1115 | if (p == flags) *p++ = 'N'; | |
1116 | *p++ = '\0'; | |
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'; | |
1123 | return sdscatprintf(sdsempty(), | |
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", | |
1125 | (client->flags & REDIS_UNIX_SOCKET) ? server.unixsocket : ip, | |
1126 | port,client->fd, | |
1127 | (long)(server.unixtime - client->ctime), | |
1128 | (long)(server.unixtime - client->lastinteraction), | |
1129 | flags, | |
1130 | client->db->id, | |
1131 | (int) dictSize(client->pubsub_channels), | |
1132 | (int) listLength(client->pubsub_patterns), | |
1133 | (client->flags & REDIS_MULTI) ? client->mstate.count : -1, | |
1134 | (unsigned long) sdslen(client->querybuf), | |
1135 | (unsigned long) sdsavail(client->querybuf), | |
1136 | (unsigned long) client->bufpos, | |
1137 | (unsigned long) listLength(client->reply), | |
1138 | getClientOutputBufferMemoryUsage(client), | |
1139 | events, | |
1140 | client->lastcmd ? client->lastcmd->name : "NULL"); | |
1141 | } | |
1142 | ||
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) { | |
1151 | sds cs; | |
1152 | ||
1153 | client = listNodeValue(ln); | |
1154 | cs = getClientInfoString(client); | |
1155 | o = sdscatsds(o,cs); | |
1156 | sdsfree(cs); | |
1157 | o = sdscatlen(o,"\n",1); | |
1158 | } | |
1159 | return o; | |
1160 | } | |
1161 | ||
1162 | void clientCommand(redisClient *c) { | |
1163 | listNode *ln; | |
1164 | listIter li; | |
1165 | redisClient *client; | |
1166 | ||
1167 | if (!strcasecmp(c->argv[1]->ptr,"list") && c->argc == 2) { | |
1168 | sds o = getAllClientsInfoString(); | |
1169 | addReplyBulkCBuffer(c,o,sdslen(o)); | |
1170 | sdsfree(o); | |
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"); | |
1191 | } else { | |
1192 | addReplyError(c, "Syntax error, try CLIENT (LIST | KILL ip:port)"); | |
1193 | } | |
1194 | } | |
1195 | ||
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. */ | |
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; | |
1221 | c->cmd = lookupCommand(c->argv[0]->ptr); | |
1222 | redisAssertWithInfo(c,NULL,c->cmd != NULL); | |
1223 | va_end(ap); | |
1224 | } | |
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 | ||
1231 | redisAssertWithInfo(c,NULL,i < c->argc); | |
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); | |
1240 | redisAssertWithInfo(c,NULL,c->cmd != NULL); | |
1241 | } | |
1242 | } | |
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 | |
1256 | * enforcing the client output length limits. */ | |
1257 | unsigned long getClientOutputBufferMemoryUsage(redisClient *c) { | |
1258 | unsigned long list_item_size = sizeof(listNode)+sizeof(robj); | |
1259 | ||
1260 | return c->reply_bytes + (list_item_size*listLength(c->reply)); | |
1261 | } | |
1262 | ||
1263 | /* Get the class of a client, used in order to enforce limits to different | |
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 | } | |
1277 | ||
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; | |
1281 | else if (!strcasecmp(name,"pubsub")) return REDIS_CLIENT_LIMIT_CLASS_PUBSUB; | |
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 | ||
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 | |
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. | |
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. */ | |
1341 | void asyncCloseClientOnOutputBufferLimitReached(redisClient *c) { | |
1342 | redisAssert(c->reply_bytes < ULONG_MAX-(1024*64)); | |
1343 | if (c->reply_bytes == 0 || c->flags & REDIS_CLOSE_ASAP) return; | |
1344 | if (checkClientOutputBufferLimits(c)) { | |
1345 | sds client = getClientInfoString(c); | |
1346 | ||
1347 | freeClientAsync(c); | |
1348 | redisLog(REDIS_WARNING,"Client %s scheduled to be closed ASAP for overcoming of output buffer limits.", client); | |
1349 | sdsfree(client); | |
1350 | } | |
1351 | } | |
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 | } |