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