Commit | Line | Data |
---|---|---|
e2641e09 | 1 | #include "redis.h" |
e2641e09 | 2 | #include <sys/uio.h> |
3 | ||
4 | void *dupClientReplyValue(void *o) { | |
5 | incrRefCount((robj*)o); | |
6 | return o; | |
7 | } | |
8 | ||
9 | int listMatchObjects(void *a, void *b) { | |
10 | return equalStringObjects(a,b); | |
11 | } | |
12 | ||
13 | redisClient *createClient(int fd) { | |
f3357792 | 14 | redisClient *c = zmalloc(sizeof(redisClient)); |
834ef78e | 15 | c->bufpos = 0; |
e2641e09 | 16 | |
17 | anetNonBlock(NULL,fd); | |
18 | anetTcpNoDelay(NULL,fd); | |
19 | if (!c) return NULL; | |
106bd87a PN |
20 | if (aeCreateFileEvent(server.el,fd,AE_READABLE, |
21 | readQueryFromClient, c) == AE_ERR) | |
22 | { | |
23 | close(fd); | |
24 | zfree(c); | |
25 | return NULL; | |
26 | } | |
27 | ||
e2641e09 | 28 | selectDb(c,0); |
29 | c->fd = fd; | |
30 | c->querybuf = sdsempty(); | |
cd8788f2 | 31 | c->reqtype = 0; |
e2641e09 | 32 | c->argc = 0; |
33 | c->argv = NULL; | |
cd8788f2 | 34 | c->multibulklen = 0; |
e2641e09 | 35 | c->bulklen = -1; |
e2641e09 | 36 | c->sentlen = 0; |
37 | c->flags = 0; | |
38 | c->lastinteraction = time(NULL); | |
39 | c->authenticated = 0; | |
40 | c->replstate = REDIS_REPL_NONE; | |
41 | c->reply = listCreate(); | |
42 | listSetFreeMethod(c->reply,decrRefCount); | |
43 | listSetDupMethod(c->reply,dupClientReplyValue); | |
e3c51c4b DJMM |
44 | c->bpop.keys = NULL; |
45 | c->bpop.count = 0; | |
46 | c->bpop.timeout = 0; | |
47 | c->bpop.target = NULL; | |
e2641e09 | 48 | c->io_keys = listCreate(); |
49 | c->watched_keys = listCreate(); | |
50 | listSetFreeMethod(c->io_keys,decrRefCount); | |
51 | c->pubsub_channels = dictCreate(&setDictType,NULL); | |
52 | c->pubsub_patterns = listCreate(); | |
53 | listSetFreeMethod(c->pubsub_patterns,decrRefCount); | |
54 | listSetMatchMethod(c->pubsub_patterns,listMatchObjects); | |
e2641e09 | 55 | listAddNodeTail(server.clients,c); |
56 | initClientMultiState(c); | |
57 | return c; | |
58 | } | |
59 | ||
a3a323e0 PN |
60 | /* Set the event loop to listen for write events on the client's socket. |
61 | * Typically gets called every time a reply is built. */ | |
4c2e506a | 62 | int _installWriteEvent(redisClient *c) { |
a3a323e0 PN |
63 | /* When CLOSE_AFTER_REPLY is set, no more replies may be added! */ |
64 | redisAssert(!(c->flags & REDIS_CLOSE_AFTER_REPLY)); | |
65 | ||
57b07380 | 66 | if (c->fd <= 0) return REDIS_ERR; |
834ef78e | 67 | if (c->bufpos == 0 && listLength(c->reply) == 0 && |
e2641e09 | 68 | (c->replstate == REDIS_REPL_NONE || |
69 | c->replstate == REDIS_REPL_ONLINE) && | |
70 | aeCreateFileEvent(server.el, c->fd, AE_WRITABLE, | |
834ef78e PN |
71 | sendReplyToClient, c) == AE_ERR) return REDIS_ERR; |
72 | return REDIS_OK; | |
73 | } | |
74 | ||
36c19d03 PN |
75 | /* Create a duplicate of the last object in the reply list when |
76 | * it is not exclusively owned by the reply list. */ | |
77 | robj *dupLastObjectIfNeeded(list *reply) { | |
78 | robj *new, *cur; | |
79 | listNode *ln; | |
80 | redisAssert(listLength(reply) > 0); | |
81 | ln = listLast(reply); | |
82 | cur = listNodeValue(ln); | |
83 | if (cur->refcount > 1) { | |
84 | new = dupStringObject(cur); | |
85 | decrRefCount(cur); | |
86 | listNodeValue(ln) = new; | |
87 | } | |
88 | return listNodeValue(ln); | |
834ef78e PN |
89 | } |
90 | ||
36c19d03 | 91 | int _addReplyToBuffer(redisClient *c, char *s, size_t len) { |
f3357792 | 92 | size_t available = sizeof(c->buf)-c->bufpos; |
36c19d03 PN |
93 | |
94 | /* If there already are entries in the reply list, we cannot | |
95 | * add anything more to the static buffer. */ | |
96 | if (listLength(c->reply) > 0) return REDIS_ERR; | |
97 | ||
98 | /* Check that the buffer has enough space available for this string. */ | |
99 | if (len > available) return REDIS_ERR; | |
e2641e09 | 100 | |
36c19d03 PN |
101 | memcpy(c->buf+c->bufpos,s,len); |
102 | c->bufpos+=len; | |
103 | return REDIS_OK; | |
834ef78e PN |
104 | } |
105 | ||
36c19d03 PN |
106 | void _addReplyObjectToList(redisClient *c, robj *o) { |
107 | robj *tail; | |
108 | if (listLength(c->reply) == 0) { | |
109 | incrRefCount(o); | |
110 | listAddNodeTail(c->reply,o); | |
111 | } else { | |
112 | tail = listNodeValue(listLast(c->reply)); | |
113 | ||
114 | /* Append to this object when possible. */ | |
115 | if (tail->ptr != NULL && | |
116 | sdslen(tail->ptr)+sdslen(o->ptr) <= REDIS_REPLY_CHUNK_BYTES) | |
117 | { | |
118 | tail = dupLastObjectIfNeeded(c->reply); | |
119 | tail->ptr = sdscatlen(tail->ptr,o->ptr,sdslen(o->ptr)); | |
120 | } else { | |
121 | incrRefCount(o); | |
122 | listAddNodeTail(c->reply,o); | |
123 | } | |
124 | } | |
125 | } | |
834ef78e | 126 | |
36c19d03 PN |
127 | /* This method takes responsibility over the sds. When it is no longer |
128 | * needed it will be free'd, otherwise it ends up in a robj. */ | |
129 | void _addReplySdsToList(redisClient *c, sds s) { | |
130 | robj *tail; | |
131 | if (listLength(c->reply) == 0) { | |
132 | listAddNodeTail(c->reply,createObject(REDIS_STRING,s)); | |
133 | } else { | |
134 | tail = listNodeValue(listLast(c->reply)); | |
135 | ||
136 | /* Append to this object when possible. */ | |
137 | if (tail->ptr != NULL && | |
138 | sdslen(tail->ptr)+sdslen(s) <= REDIS_REPLY_CHUNK_BYTES) | |
139 | { | |
140 | tail = dupLastObjectIfNeeded(c->reply); | |
141 | tail->ptr = sdscatlen(tail->ptr,s,sdslen(s)); | |
142 | sdsfree(s); | |
834ef78e | 143 | } else { |
36c19d03 | 144 | listAddNodeTail(c->reply,createObject(REDIS_STRING,s)); |
834ef78e | 145 | } |
36c19d03 PN |
146 | } |
147 | } | |
148 | ||
149 | void _addReplyStringToList(redisClient *c, char *s, size_t len) { | |
150 | robj *tail; | |
151 | if (listLength(c->reply) == 0) { | |
152 | listAddNodeTail(c->reply,createStringObject(s,len)); | |
834ef78e | 153 | } else { |
36c19d03 PN |
154 | tail = listNodeValue(listLast(c->reply)); |
155 | ||
156 | /* Append to this object when possible. */ | |
157 | if (tail->ptr != NULL && | |
158 | sdslen(tail->ptr)+len <= REDIS_REPLY_CHUNK_BYTES) | |
159 | { | |
160 | tail = dupLastObjectIfNeeded(c->reply); | |
161 | tail->ptr = sdscatlen(tail->ptr,s,len); | |
834ef78e | 162 | } else { |
36c19d03 | 163 | listAddNodeTail(c->reply,createStringObject(s,len)); |
834ef78e PN |
164 | } |
165 | } | |
166 | } | |
e2641e09 | 167 | |
834ef78e | 168 | void addReply(redisClient *c, robj *obj) { |
4c2e506a | 169 | if (_installWriteEvent(c) != REDIS_OK) return; |
170 | redisAssert(!server.vm_enabled || obj->storage == REDIS_VM_MEMORY); | |
171 | ||
172 | /* This is an important place where we can avoid copy-on-write | |
173 | * when there is a saving child running, avoiding touching the | |
174 | * refcount field of the object if it's not needed. | |
175 | * | |
176 | * If the encoding is RAW and there is room in the static buffer | |
177 | * we'll be able to send the object to the client without | |
178 | * messing with its page. */ | |
179 | if (obj->encoding == REDIS_ENCODING_RAW) { | |
180 | if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK) | |
181 | _addReplyObjectToList(c,obj); | |
834ef78e | 182 | } else { |
d51ebef5 | 183 | /* FIXME: convert the long into string and use _addReplyToBuffer() |
184 | * instead of calling getDecodedObject. As this place in the | |
185 | * code is too performance critical. */ | |
834ef78e | 186 | obj = getDecodedObject(obj); |
4c2e506a | 187 | if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK) |
188 | _addReplyObjectToList(c,obj); | |
189 | decrRefCount(obj); | |
e2641e09 | 190 | } |
e2641e09 | 191 | } |
192 | ||
193 | void addReplySds(redisClient *c, sds s) { | |
4c2e506a | 194 | if (_installWriteEvent(c) != REDIS_OK) { |
cd76bb65 PN |
195 | /* The caller expects the sds to be free'd. */ |
196 | sdsfree(s); | |
197 | return; | |
198 | } | |
36c19d03 | 199 | if (_addReplyToBuffer(c,s,sdslen(s)) == REDIS_OK) { |
834ef78e PN |
200 | sdsfree(s); |
201 | } else { | |
36c19d03 PN |
202 | /* This method free's the sds when it is no longer needed. */ |
203 | _addReplySdsToList(c,s); | |
834ef78e | 204 | } |
e2641e09 | 205 | } |
206 | ||
834ef78e | 207 | void addReplyString(redisClient *c, char *s, size_t len) { |
4c2e506a | 208 | if (_installWriteEvent(c) != REDIS_OK) return; |
36c19d03 PN |
209 | if (_addReplyToBuffer(c,s,len) != REDIS_OK) |
210 | _addReplyStringToList(c,s,len); | |
834ef78e | 211 | } |
e2641e09 | 212 | |
3ab20376 PN |
213 | void _addReplyError(redisClient *c, char *s, size_t len) { |
214 | addReplyString(c,"-ERR ",5); | |
215 | addReplyString(c,s,len); | |
216 | addReplyString(c,"\r\n",2); | |
e2641e09 | 217 | } |
218 | ||
3ab20376 PN |
219 | void addReplyError(redisClient *c, char *err) { |
220 | _addReplyError(c,err,strlen(err)); | |
221 | } | |
e2641e09 | 222 | |
3ab20376 PN |
223 | void addReplyErrorFormat(redisClient *c, const char *fmt, ...) { |
224 | va_list ap; | |
225 | va_start(ap,fmt); | |
226 | sds s = sdscatvprintf(sdsempty(),fmt,ap); | |
227 | va_end(ap); | |
228 | _addReplyError(c,s,sdslen(s)); | |
229 | sdsfree(s); | |
230 | } | |
231 | ||
232 | void _addReplyStatus(redisClient *c, char *s, size_t len) { | |
233 | addReplyString(c,"+",1); | |
234 | addReplyString(c,s,len); | |
235 | addReplyString(c,"\r\n",2); | |
236 | } | |
237 | ||
238 | void addReplyStatus(redisClient *c, char *status) { | |
239 | _addReplyStatus(c,status,strlen(status)); | |
240 | } | |
241 | ||
242 | void addReplyStatusFormat(redisClient *c, const char *fmt, ...) { | |
243 | va_list ap; | |
244 | va_start(ap,fmt); | |
245 | sds s = sdscatvprintf(sdsempty(),fmt,ap); | |
246 | va_end(ap); | |
247 | _addReplyStatus(c,s,sdslen(s)); | |
248 | sdsfree(s); | |
249 | } | |
250 | ||
b301c1fc PN |
251 | /* Adds an empty object to the reply list that will contain the multi bulk |
252 | * length, which is not known when this function is called. */ | |
253 | void *addDeferredMultiBulkLength(redisClient *c) { | |
4c2e506a | 254 | /* Note that we install the write event here even if the object is not |
255 | * ready to be sent, since we are sure that before returning to the | |
256 | * event loop setDeferredMultiBulkLength() will be called. */ | |
257 | if (_installWriteEvent(c) != REDIS_OK) return NULL; | |
36c19d03 | 258 | listAddNodeTail(c->reply,createObject(REDIS_STRING,NULL)); |
b301c1fc PN |
259 | return listLast(c->reply); |
260 | } | |
261 | ||
262 | /* Populate the length object and try glueing it to the next chunk. */ | |
263 | void setDeferredMultiBulkLength(redisClient *c, void *node, long length) { | |
264 | listNode *ln = (listNode*)node; | |
265 | robj *len, *next; | |
266 | ||
267 | /* Abort when *node is NULL (see addDeferredMultiBulkLength). */ | |
268 | if (node == NULL) return; | |
269 | ||
270 | len = listNodeValue(ln); | |
271 | len->ptr = sdscatprintf(sdsempty(),"*%ld\r\n",length); | |
272 | if (ln->next != NULL) { | |
273 | next = listNodeValue(ln->next); | |
36c19d03 | 274 | |
49128f0b | 275 | /* Only glue when the next node is non-NULL (an sds in this case) */ |
36c19d03 | 276 | if (next->ptr != NULL) { |
49128f0b | 277 | len->ptr = sdscatlen(len->ptr,next->ptr,sdslen(next->ptr)); |
b301c1fc PN |
278 | listDelNode(c->reply,ln->next); |
279 | } | |
e2641e09 | 280 | } |
b301c1fc PN |
281 | } |
282 | ||
d51ebef5 | 283 | /* Add a duble as a bulk reply */ |
834ef78e PN |
284 | void addReplyDouble(redisClient *c, double d) { |
285 | char dbuf[128], sbuf[128]; | |
286 | int dlen, slen; | |
287 | dlen = snprintf(dbuf,sizeof(dbuf),"%.17g",d); | |
288 | slen = snprintf(sbuf,sizeof(sbuf),"$%d\r\n%s\r\n",dlen,dbuf); | |
289 | addReplyString(c,sbuf,slen); | |
e2641e09 | 290 | } |
291 | ||
d51ebef5 | 292 | /* Add a long long as integer reply or bulk len / multi bulk count. |
293 | * Basically this is used to output <prefix><long long><crlf>. */ | |
834ef78e | 294 | void _addReplyLongLong(redisClient *c, long long ll, char prefix) { |
e2641e09 | 295 | char buf[128]; |
834ef78e PN |
296 | int len; |
297 | buf[0] = prefix; | |
e2641e09 | 298 | len = ll2string(buf+1,sizeof(buf)-1,ll); |
299 | buf[len+1] = '\r'; | |
300 | buf[len+2] = '\n'; | |
834ef78e | 301 | addReplyString(c,buf,len+3); |
e2641e09 | 302 | } |
303 | ||
834ef78e PN |
304 | void addReplyLongLong(redisClient *c, long long ll) { |
305 | _addReplyLongLong(c,ll,':'); | |
306 | } | |
e2641e09 | 307 | |
0537e7bf PN |
308 | void addReplyMultiBulkLen(redisClient *c, long length) { |
309 | _addReplyLongLong(c,length,'*'); | |
e2641e09 | 310 | } |
311 | ||
d51ebef5 | 312 | /* Create the length prefix of a bulk reply, example: $2234 */ |
e2641e09 | 313 | void addReplyBulkLen(redisClient *c, robj *obj) { |
834ef78e | 314 | size_t len; |
e2641e09 | 315 | |
316 | if (obj->encoding == REDIS_ENCODING_RAW) { | |
317 | len = sdslen(obj->ptr); | |
318 | } else { | |
319 | long n = (long)obj->ptr; | |
320 | ||
321 | /* Compute how many bytes will take this integer as a radix 10 string */ | |
322 | len = 1; | |
323 | if (n < 0) { | |
324 | len++; | |
325 | n = -n; | |
326 | } | |
327 | while((n = n/10) != 0) { | |
328 | len++; | |
329 | } | |
330 | } | |
834ef78e | 331 | _addReplyLongLong(c,len,'$'); |
e2641e09 | 332 | } |
333 | ||
d51ebef5 | 334 | /* Add a Redis Object as a bulk reply */ |
e2641e09 | 335 | void addReplyBulk(redisClient *c, robj *obj) { |
336 | addReplyBulkLen(c,obj); | |
337 | addReply(c,obj); | |
338 | addReply(c,shared.crlf); | |
339 | } | |
340 | ||
d51ebef5 | 341 | /* Add a C buffer as bulk reply */ |
342 | void addReplyBulkCBuffer(redisClient *c, void *p, size_t len) { | |
343 | _addReplyLongLong(c,len,'$'); | |
344 | addReplyString(c,p,len); | |
345 | addReply(c,shared.crlf); | |
346 | } | |
347 | ||
348 | /* Add a C nul term string as bulk reply */ | |
e2641e09 | 349 | void addReplyBulkCString(redisClient *c, char *s) { |
350 | if (s == NULL) { | |
351 | addReply(c,shared.nullbulk); | |
352 | } else { | |
d51ebef5 | 353 | addReplyBulkCBuffer(c,s,strlen(s)); |
e2641e09 | 354 | } |
355 | } | |
356 | ||
d51ebef5 | 357 | /* Add a long long as a bulk reply */ |
358 | void addReplyBulkLongLong(redisClient *c, long long ll) { | |
359 | char buf[64]; | |
360 | int len; | |
361 | ||
362 | len = ll2string(buf,64,ll); | |
363 | addReplyBulkCBuffer(c,buf,len); | |
364 | } | |
365 | ||
ab17b909 | 366 | static void acceptCommonHandler(int fd) { |
e2641e09 | 367 | redisClient *c; |
ab17b909 | 368 | if ((c = createClient(fd)) == NULL) { |
e2641e09 | 369 | redisLog(REDIS_WARNING,"Error allocating resoures for the client"); |
ab17b909 | 370 | close(fd); /* May be already closed, just ingore errors */ |
e2641e09 | 371 | return; |
372 | } | |
373 | /* If maxclient directive is set and this is one client more... close the | |
374 | * connection. Note that we create the client instead to check before | |
375 | * for this condition, since now the socket is already set in nonblocking | |
376 | * mode and we can send an error for free using the Kernel I/O */ | |
377 | if (server.maxclients && listLength(server.clients) > server.maxclients) { | |
378 | char *err = "-ERR max number of clients reached\r\n"; | |
379 | ||
380 | /* That's a best effort error message, don't check write errors */ | |
381 | if (write(c->fd,err,strlen(err)) == -1) { | |
382 | /* Nothing to do, Just to avoid the warning... */ | |
383 | } | |
384 | freeClient(c); | |
385 | return; | |
386 | } | |
387 | server.stat_numconnections++; | |
388 | } | |
389 | ||
ab17b909 PN |
390 | void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) { |
391 | int cport, cfd; | |
392 | char cip[128]; | |
393 | REDIS_NOTUSED(el); | |
394 | REDIS_NOTUSED(mask); | |
395 | REDIS_NOTUSED(privdata); | |
396 | ||
397 | cfd = anetTcpAccept(server.neterr, fd, cip, &cport); | |
398 | if (cfd == AE_ERR) { | |
399 | redisLog(REDIS_VERBOSE,"Accepting client connection: %s", server.neterr); | |
400 | return; | |
401 | } | |
402 | redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport); | |
403 | acceptCommonHandler(cfd); | |
404 | } | |
405 | ||
406 | void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) { | |
407 | int cfd; | |
ab17b909 PN |
408 | REDIS_NOTUSED(el); |
409 | REDIS_NOTUSED(mask); | |
410 | REDIS_NOTUSED(privdata); | |
411 | ||
4fe83b55 | 412 | cfd = anetUnixAccept(server.neterr, fd); |
ab17b909 PN |
413 | if (cfd == AE_ERR) { |
414 | redisLog(REDIS_VERBOSE,"Accepting client connection: %s", server.neterr); | |
415 | return; | |
416 | } | |
417 | redisLog(REDIS_VERBOSE,"Accepted connection to %s", server.unixsocket); | |
418 | acceptCommonHandler(cfd); | |
419 | } | |
420 | ||
421 | ||
e2641e09 | 422 | static void freeClientArgv(redisClient *c) { |
423 | int j; | |
e2641e09 | 424 | for (j = 0; j < c->argc; j++) |
425 | decrRefCount(c->argv[j]); | |
e2641e09 | 426 | c->argc = 0; |
e2641e09 | 427 | } |
428 | ||
429 | void freeClient(redisClient *c) { | |
430 | listNode *ln; | |
431 | ||
432 | /* Note that if the client we are freeing is blocked into a blocking | |
433 | * call, we have to set querybuf to NULL *before* to call | |
434 | * unblockClientWaitingData() to avoid processInputBuffer() will get | |
435 | * called. Also it is important to remove the file events after | |
436 | * this, because this call adds the READABLE event. */ | |
437 | sdsfree(c->querybuf); | |
438 | c->querybuf = NULL; | |
439 | if (c->flags & REDIS_BLOCKED) | |
440 | unblockClientWaitingData(c); | |
441 | ||
442 | /* UNWATCH all the keys */ | |
443 | unwatchAllKeys(c); | |
444 | listRelease(c->watched_keys); | |
445 | /* Unsubscribe from all the pubsub channels */ | |
446 | pubsubUnsubscribeAllChannels(c,0); | |
447 | pubsubUnsubscribeAllPatterns(c,0); | |
448 | dictRelease(c->pubsub_channels); | |
449 | listRelease(c->pubsub_patterns); | |
450 | /* Obvious cleanup */ | |
451 | aeDeleteFileEvent(server.el,c->fd,AE_READABLE); | |
452 | aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE); | |
453 | listRelease(c->reply); | |
454 | freeClientArgv(c); | |
455 | close(c->fd); | |
456 | /* Remove from the list of clients */ | |
457 | ln = listSearchKey(server.clients,c); | |
458 | redisAssert(ln != NULL); | |
459 | listDelNode(server.clients,ln); | |
1a71fb96 | 460 | /* Remove from the list of clients waiting for swapped keys, or ready |
461 | * to be restarted, but not yet woken up again. */ | |
462 | if (c->flags & REDIS_IO_WAIT) { | |
463 | redisAssert(server.vm_enabled); | |
464 | if (listLength(c->io_keys) == 0) { | |
465 | ln = listSearchKey(server.io_ready_clients,c); | |
466 | ||
467 | /* When this client is waiting to be woken up (REDIS_IO_WAIT), | |
468 | * it should be present in the list io_ready_clients */ | |
469 | redisAssert(ln != NULL); | |
e2641e09 | 470 | listDelNode(server.io_ready_clients,ln); |
1a71fb96 | 471 | } else { |
472 | while (listLength(c->io_keys)) { | |
473 | ln = listFirst(c->io_keys); | |
474 | dontWaitForSwappedKey(c,ln->value); | |
475 | } | |
e2641e09 | 476 | } |
1a71fb96 | 477 | server.vm_blocked_clients--; |
e2641e09 | 478 | } |
479 | listRelease(c->io_keys); | |
778b2210 | 480 | /* Master/slave cleanup. |
481 | * Case 1: we lost the connection with a slave. */ | |
e2641e09 | 482 | if (c->flags & REDIS_SLAVE) { |
483 | if (c->replstate == REDIS_REPL_SEND_BULK && c->repldbfd != -1) | |
484 | close(c->repldbfd); | |
485 | list *l = (c->flags & REDIS_MONITOR) ? server.monitors : server.slaves; | |
486 | ln = listSearchKey(l,c); | |
487 | redisAssert(ln != NULL); | |
488 | listDelNode(l,ln); | |
489 | } | |
778b2210 | 490 | |
491 | /* Case 2: we lost the connection with the master. */ | |
e2641e09 | 492 | if (c->flags & REDIS_MASTER) { |
493 | server.master = NULL; | |
f4aa600b | 494 | /* FIXME */ |
e2641e09 | 495 | server.replstate = REDIS_REPL_CONNECT; |
778b2210 | 496 | /* Since we lost the connection with the master, we should also |
497 | * close the connection with all our slaves if we have any, so | |
498 | * when we'll resync with the master the other slaves will sync again | |
499 | * with us as well. Note that also when the slave is not connected | |
500 | * to the master it will keep refusing connections by other slaves. */ | |
501 | while (listLength(server.slaves)) { | |
502 | ln = listFirst(server.slaves); | |
503 | freeClient((redisClient*)ln->value); | |
504 | } | |
e2641e09 | 505 | } |
506 | /* Release memory */ | |
507 | zfree(c->argv); | |
e2641e09 | 508 | freeClientMultiState(c); |
509 | zfree(c); | |
510 | } | |
511 | ||
e2641e09 | 512 | void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) { |
513 | redisClient *c = privdata; | |
514 | int nwritten = 0, totwritten = 0, objlen; | |
515 | robj *o; | |
516 | REDIS_NOTUSED(el); | |
517 | REDIS_NOTUSED(mask); | |
518 | ||
519 | /* Use writev() if we have enough buffers to send */ | |
520 | if (!server.glueoutputbuf && | |
521 | listLength(c->reply) > REDIS_WRITEV_THRESHOLD && | |
522 | !(c->flags & REDIS_MASTER)) | |
523 | { | |
524 | sendReplyToClientWritev(el, fd, privdata, mask); | |
525 | return; | |
526 | } | |
527 | ||
834ef78e PN |
528 | while(c->bufpos > 0 || listLength(c->reply)) { |
529 | if (c->bufpos > 0) { | |
530 | if (c->flags & REDIS_MASTER) { | |
531 | /* Don't reply to a master */ | |
532 | nwritten = c->bufpos - c->sentlen; | |
533 | } else { | |
534 | nwritten = write(fd,c->buf+c->sentlen,c->bufpos-c->sentlen); | |
535 | if (nwritten <= 0) break; | |
536 | } | |
537 | c->sentlen += nwritten; | |
538 | totwritten += nwritten; | |
539 | ||
540 | /* If the buffer was sent, set bufpos to zero to continue with | |
541 | * the remainder of the reply. */ | |
542 | if (c->sentlen == c->bufpos) { | |
543 | c->bufpos = 0; | |
544 | c->sentlen = 0; | |
545 | } | |
546 | } else { | |
547 | o = listNodeValue(listFirst(c->reply)); | |
548 | objlen = sdslen(o->ptr); | |
e2641e09 | 549 | |
834ef78e PN |
550 | if (objlen == 0) { |
551 | listDelNode(c->reply,listFirst(c->reply)); | |
552 | continue; | |
553 | } | |
e2641e09 | 554 | |
834ef78e PN |
555 | if (c->flags & REDIS_MASTER) { |
556 | /* Don't reply to a master */ | |
557 | nwritten = objlen - c->sentlen; | |
558 | } else { | |
559 | nwritten = write(fd, ((char*)o->ptr)+c->sentlen,objlen-c->sentlen); | |
560 | if (nwritten <= 0) break; | |
561 | } | |
562 | c->sentlen += nwritten; | |
563 | totwritten += nwritten; | |
e2641e09 | 564 | |
834ef78e PN |
565 | /* If we fully sent the object on head go to the next one */ |
566 | if (c->sentlen == objlen) { | |
567 | listDelNode(c->reply,listFirst(c->reply)); | |
568 | c->sentlen = 0; | |
569 | } | |
e2641e09 | 570 | } |
571 | /* Note that we avoid to send more thank REDIS_MAX_WRITE_PER_EVENT | |
572 | * bytes, in a single threaded server it's a good idea to serve | |
573 | * other clients as well, even if a very large request comes from | |
574 | * super fast link that is always able to accept data (in real world | |
575 | * scenario think about 'KEYS *' against the loopback interfae) */ | |
576 | if (totwritten > REDIS_MAX_WRITE_PER_EVENT) break; | |
577 | } | |
578 | if (nwritten == -1) { | |
579 | if (errno == EAGAIN) { | |
580 | nwritten = 0; | |
581 | } else { | |
582 | redisLog(REDIS_VERBOSE, | |
583 | "Error writing to client: %s", strerror(errno)); | |
584 | freeClient(c); | |
585 | return; | |
586 | } | |
587 | } | |
588 | if (totwritten > 0) c->lastinteraction = time(NULL); | |
589 | if (listLength(c->reply) == 0) { | |
590 | c->sentlen = 0; | |
591 | aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE); | |
941c9fa2 PN |
592 | |
593 | /* Close connection after entire reply has been sent. */ | |
cd8788f2 | 594 | if (c->flags & REDIS_CLOSE_AFTER_REPLY) freeClient(c); |
e2641e09 | 595 | } |
596 | } | |
597 | ||
598 | void sendReplyToClientWritev(aeEventLoop *el, int fd, void *privdata, int mask) | |
599 | { | |
600 | redisClient *c = privdata; | |
601 | int nwritten = 0, totwritten = 0, objlen, willwrite; | |
602 | robj *o; | |
603 | struct iovec iov[REDIS_WRITEV_IOVEC_COUNT]; | |
604 | int offset, ion = 0; | |
605 | REDIS_NOTUSED(el); | |
606 | REDIS_NOTUSED(mask); | |
607 | ||
608 | listNode *node; | |
609 | while (listLength(c->reply)) { | |
610 | offset = c->sentlen; | |
611 | ion = 0; | |
612 | willwrite = 0; | |
613 | ||
614 | /* fill-in the iov[] array */ | |
615 | for(node = listFirst(c->reply); node; node = listNextNode(node)) { | |
616 | o = listNodeValue(node); | |
617 | objlen = sdslen(o->ptr); | |
618 | ||
619 | if (totwritten + objlen - offset > REDIS_MAX_WRITE_PER_EVENT) | |
620 | break; | |
621 | ||
622 | if(ion == REDIS_WRITEV_IOVEC_COUNT) | |
623 | break; /* no more iovecs */ | |
624 | ||
625 | iov[ion].iov_base = ((char*)o->ptr) + offset; | |
626 | iov[ion].iov_len = objlen - offset; | |
627 | willwrite += objlen - offset; | |
628 | offset = 0; /* just for the first item */ | |
629 | ion++; | |
630 | } | |
631 | ||
632 | if(willwrite == 0) | |
633 | break; | |
634 | ||
635 | /* write all collected blocks at once */ | |
636 | if((nwritten = writev(fd, iov, ion)) < 0) { | |
637 | if (errno != EAGAIN) { | |
638 | redisLog(REDIS_VERBOSE, | |
639 | "Error writing to client: %s", strerror(errno)); | |
640 | freeClient(c); | |
641 | return; | |
642 | } | |
643 | break; | |
644 | } | |
645 | ||
646 | totwritten += nwritten; | |
647 | offset = c->sentlen; | |
648 | ||
649 | /* remove written robjs from c->reply */ | |
650 | while (nwritten && listLength(c->reply)) { | |
651 | o = listNodeValue(listFirst(c->reply)); | |
652 | objlen = sdslen(o->ptr); | |
653 | ||
654 | if(nwritten >= objlen - offset) { | |
655 | listDelNode(c->reply, listFirst(c->reply)); | |
656 | nwritten -= objlen - offset; | |
657 | c->sentlen = 0; | |
658 | } else { | |
659 | /* partial write */ | |
660 | c->sentlen += nwritten; | |
661 | break; | |
662 | } | |
663 | offset = 0; | |
664 | } | |
665 | } | |
666 | ||
667 | if (totwritten > 0) | |
668 | c->lastinteraction = time(NULL); | |
669 | ||
670 | if (listLength(c->reply) == 0) { | |
671 | c->sentlen = 0; | |
672 | aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE); | |
673 | } | |
674 | } | |
675 | ||
676 | /* resetClient prepare the client to process the next command */ | |
677 | void resetClient(redisClient *c) { | |
678 | freeClientArgv(c); | |
cd8788f2 PN |
679 | c->reqtype = 0; |
680 | c->multibulklen = 0; | |
e2641e09 | 681 | c->bulklen = -1; |
e2641e09 | 682 | } |
683 | ||
684 | void closeTimedoutClients(void) { | |
685 | redisClient *c; | |
686 | listNode *ln; | |
687 | time_t now = time(NULL); | |
688 | listIter li; | |
689 | ||
690 | listRewind(server.clients,&li); | |
691 | while ((ln = listNext(&li)) != NULL) { | |
692 | c = listNodeValue(ln); | |
693 | if (server.maxidletime && | |
694 | !(c->flags & REDIS_SLAVE) && /* no timeout for slaves */ | |
695 | !(c->flags & REDIS_MASTER) && /* no timeout for masters */ | |
e452436a | 696 | !(c->flags & REDIS_BLOCKED) && /* no timeout for BLPOP */ |
e2641e09 | 697 | dictSize(c->pubsub_channels) == 0 && /* no timeout for pubsub */ |
698 | listLength(c->pubsub_patterns) == 0 && | |
699 | (now - c->lastinteraction > server.maxidletime)) | |
700 | { | |
701 | redisLog(REDIS_VERBOSE,"Closing idle client"); | |
702 | freeClient(c); | |
703 | } else if (c->flags & REDIS_BLOCKED) { | |
e3c51c4b | 704 | if (c->bpop.timeout != 0 && c->bpop.timeout < now) { |
e2641e09 | 705 | addReply(c,shared.nullmultibulk); |
706 | unblockClientWaitingData(c); | |
707 | } | |
708 | } | |
709 | } | |
710 | } | |
711 | ||
cd8788f2 PN |
712 | int processInlineBuffer(redisClient *c) { |
713 | char *newline = strstr(c->querybuf,"\r\n"); | |
714 | int argc, j; | |
715 | sds *argv; | |
716 | size_t querylen; | |
717 | ||
718 | /* Nothing to do without a \r\n */ | |
719 | if (newline == NULL) | |
720 | return REDIS_ERR; | |
721 | ||
722 | /* Split the input buffer up to the \r\n */ | |
723 | querylen = newline-(c->querybuf); | |
724 | argv = sdssplitlen(c->querybuf,querylen," ",1,&argc); | |
725 | ||
726 | /* Leave data after the first line of the query in the buffer */ | |
727 | c->querybuf = sdsrange(c->querybuf,querylen+2,-1); | |
728 | ||
729 | /* Setup argv array on client structure */ | |
730 | if (c->argv) zfree(c->argv); | |
731 | c->argv = zmalloc(sizeof(robj*)*argc); | |
732 | ||
733 | /* Create redis objects for all arguments. */ | |
734 | for (c->argc = 0, j = 0; j < argc; j++) { | |
735 | if (sdslen(argv[j])) { | |
736 | c->argv[c->argc] = createObject(REDIS_STRING,argv[j]); | |
737 | c->argc++; | |
738 | } else { | |
739 | sdsfree(argv[j]); | |
740 | } | |
741 | } | |
742 | zfree(argv); | |
743 | return REDIS_OK; | |
744 | } | |
745 | ||
746 | /* Helper function. Trims query buffer to make the function that processes | |
747 | * multi bulk requests idempotent. */ | |
748 | static void setProtocolError(redisClient *c, int pos) { | |
749 | c->flags |= REDIS_CLOSE_AFTER_REPLY; | |
750 | c->querybuf = sdsrange(c->querybuf,pos,-1); | |
751 | } | |
752 | ||
753 | int processMultibulkBuffer(redisClient *c) { | |
754 | char *newline = NULL; | |
755 | char *eptr; | |
756 | int pos = 0, tolerr; | |
757 | long bulklen; | |
758 | ||
759 | if (c->multibulklen == 0) { | |
760 | /* The client should have been reset */ | |
761 | redisAssert(c->argc == 0); | |
762 | ||
763 | /* Multi bulk length cannot be read without a \r\n */ | |
764 | newline = strstr(c->querybuf,"\r\n"); | |
765 | if (newline == NULL) | |
766 | return REDIS_ERR; | |
767 | ||
768 | /* We know for sure there is a whole line since newline != NULL, | |
769 | * so go ahead and find out the multi bulk length. */ | |
770 | redisAssert(c->querybuf[0] == '*'); | |
771 | c->multibulklen = strtol(c->querybuf+1,&eptr,10); | |
772 | pos = (newline-c->querybuf)+2; | |
773 | if (c->multibulklen <= 0) { | |
774 | c->querybuf = sdsrange(c->querybuf,pos,-1); | |
775 | return REDIS_OK; | |
b19c33d4 PN |
776 | } else if (c->multibulklen > 1024*1024) { |
777 | addReplyError(c,"Protocol error: invalid multibulk length"); | |
778 | setProtocolError(c,pos); | |
779 | return REDIS_ERR; | |
cd8788f2 PN |
780 | } |
781 | ||
782 | /* Setup argv array on client structure */ | |
783 | if (c->argv) zfree(c->argv); | |
784 | c->argv = zmalloc(sizeof(robj*)*c->multibulklen); | |
785 | ||
786 | /* Search new newline */ | |
787 | newline = strstr(c->querybuf+pos,"\r\n"); | |
788 | } | |
789 | ||
790 | redisAssert(c->multibulklen > 0); | |
791 | while(c->multibulklen) { | |
792 | /* Read bulk length if unknown */ | |
793 | if (c->bulklen == -1) { | |
794 | newline = strstr(c->querybuf+pos,"\r\n"); | |
795 | if (newline != NULL) { | |
796 | if (c->querybuf[pos] != '$') { | |
797 | addReplyErrorFormat(c, | |
798 | "Protocol error: expected '$', got '%c'", | |
799 | c->querybuf[pos]); | |
800 | setProtocolError(c,pos); | |
801 | return REDIS_ERR; | |
e2641e09 | 802 | } |
cd8788f2 PN |
803 | |
804 | bulklen = strtol(c->querybuf+pos+1,&eptr,10); | |
805 | tolerr = (eptr[0] != '\r'); | |
806 | if (tolerr || bulklen == LONG_MIN || bulklen == LONG_MAX || | |
401c3e21 | 807 | bulklen < 0 || bulklen > 512*1024*1024) |
cd8788f2 PN |
808 | { |
809 | addReplyError(c,"Protocol error: invalid bulk length"); | |
810 | setProtocolError(c,pos); | |
811 | return REDIS_ERR; | |
812 | } | |
813 | pos += eptr-(c->querybuf+pos)+2; | |
814 | c->bulklen = bulklen; | |
815 | } else { | |
816 | /* No newline in current buffer, so wait for more data */ | |
817 | break; | |
e2641e09 | 818 | } |
cd8788f2 PN |
819 | } |
820 | ||
821 | /* Read bulk argument */ | |
822 | if (sdslen(c->querybuf)-pos < (unsigned)(c->bulklen+2)) { | |
823 | /* Not enough data (+2 == trailing \r\n) */ | |
824 | break; | |
825 | } else { | |
826 | c->argv[c->argc++] = createStringObject(c->querybuf+pos,c->bulklen); | |
827 | pos += c->bulklen+2; | |
828 | c->bulklen = -1; | |
829 | c->multibulklen--; | |
830 | } | |
831 | } | |
832 | ||
833 | /* Trim to pos */ | |
834 | c->querybuf = sdsrange(c->querybuf,pos,-1); | |
835 | ||
836 | /* We're done when c->multibulk == 0 */ | |
837 | if (c->multibulklen == 0) { | |
838 | return REDIS_OK; | |
839 | } | |
840 | return REDIS_ERR; | |
841 | } | |
842 | ||
843 | void processInputBuffer(redisClient *c) { | |
844 | /* Keep processing while there is something in the input buffer */ | |
845 | while(sdslen(c->querybuf)) { | |
4794d88f | 846 | /* Immediately abort if the client is in the middle of something. */ |
cd8788f2 PN |
847 | if (c->flags & REDIS_BLOCKED || c->flags & REDIS_IO_WAIT) return; |
848 | ||
5e78edb3 PN |
849 | /* REDIS_CLOSE_AFTER_REPLY closes the connection once the reply is |
850 | * written to the client. Make sure to not let the reply grow after | |
851 | * this flag has been set (i.e. don't process more commands). */ | |
852 | if (c->flags & REDIS_CLOSE_AFTER_REPLY) return; | |
cd8788f2 PN |
853 | |
854 | /* Determine request type when unknown. */ | |
855 | if (!c->reqtype) { | |
856 | if (c->querybuf[0] == '*') { | |
857 | c->reqtype = REDIS_REQ_MULTIBULK; | |
e2641e09 | 858 | } else { |
cd8788f2 | 859 | c->reqtype = REDIS_REQ_INLINE; |
e2641e09 | 860 | } |
e2641e09 | 861 | } |
cd8788f2 PN |
862 | |
863 | if (c->reqtype == REDIS_REQ_INLINE) { | |
864 | if (processInlineBuffer(c) != REDIS_OK) break; | |
865 | } else if (c->reqtype == REDIS_REQ_MULTIBULK) { | |
866 | if (processMultibulkBuffer(c) != REDIS_OK) break; | |
867 | } else { | |
868 | redisPanic("Unknown request type"); | |
e2641e09 | 869 | } |
cd8788f2 PN |
870 | |
871 | /* Multibulk processing could see a <= 0 length. */ | |
9da6caac PN |
872 | if (c->argc == 0) { |
873 | resetClient(c); | |
874 | } else { | |
875 | /* Only reset the client when the command was executed. */ | |
876 | if (processCommand(c) == REDIS_OK) | |
877 | resetClient(c); | |
878 | } | |
e2641e09 | 879 | } |
880 | } | |
881 | ||
882 | void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) { | |
883 | redisClient *c = (redisClient*) privdata; | |
884 | char buf[REDIS_IOBUF_LEN]; | |
885 | int nread; | |
886 | REDIS_NOTUSED(el); | |
887 | REDIS_NOTUSED(mask); | |
888 | ||
889 | nread = read(fd, buf, REDIS_IOBUF_LEN); | |
890 | if (nread == -1) { | |
891 | if (errno == EAGAIN) { | |
892 | nread = 0; | |
893 | } else { | |
894 | redisLog(REDIS_VERBOSE, "Reading from client: %s",strerror(errno)); | |
895 | freeClient(c); | |
896 | return; | |
897 | } | |
898 | } else if (nread == 0) { | |
899 | redisLog(REDIS_VERBOSE, "Client closed connection"); | |
900 | freeClient(c); | |
901 | return; | |
902 | } | |
903 | if (nread) { | |
cd8788f2 | 904 | c->querybuf = sdscatlen(c->querybuf,buf,nread); |
e2641e09 | 905 | c->lastinteraction = time(NULL); |
906 | } else { | |
907 | return; | |
908 | } | |
909 | processInputBuffer(c); | |
910 | } |