| 2682 | } |
| 2683 | |
| 2684 | void readQueryFromClient(connection *conn) { |
| 2685 | client *c = (client*)connGetPrivateData(conn); |
| 2686 | serverAssert(conn == c->conn); |
| 2687 | int nread, readlen; |
| 2688 | size_t qblen; |
| 2689 | |
| 2690 | serverAssertDebug(FCorrectThread(c)); |
| 2691 | serverAssertDebug(!GlobalLocksAcquired()); |
| 2692 | |
| 2693 | AeLocker aelock; |
| 2694 | AssertCorrectThread(c); |
| 2695 | std::unique_lock<decltype(c->lock)> lock(c->lock, std::defer_lock); |
| 2696 | if (!lock.try_lock()) |
| 2697 | return; // Process something else while we wait |
| 2698 | |
| 2699 | /* Update total number of reads on server */ |
| 2700 | g_pserver->stat_total_reads_processed.fetch_add(1, std::memory_order_relaxed); |
| 2701 | |
| 2702 | readlen = PROTO_IOBUF_LEN; |
| 2703 | /* If this is a multi bulk request, and we are processing a bulk reply |
| 2704 | * that is large enough, try to maximize the probability that the query |
| 2705 | * buffer contains exactly the SDS string representing the object, even |
| 2706 | * at the risk of requiring more read(2) calls. This way the function |
| 2707 | * processMultiBulkBuffer() can avoid copying buffers to create the |
| 2708 | * Redis Object representing the argument. */ |
| 2709 | if (c->reqtype == PROTO_REQ_MULTIBULK && c->multibulklen && c->bulklen != -1 |
| 2710 | && c->bulklen >= PROTO_MBULK_BIG_ARG) |
| 2711 | { |
| 2712 | ssize_t remaining = (size_t)(c->bulklen+2)-sdslen(c->querybuf); |
| 2713 | |
| 2714 | /* Note that the 'remaining' variable may be zero in some edge case, |
| 2715 | * for example once we resume a blocked client after CLIENT PAUSE. */ |
| 2716 | if (remaining > 0 && remaining < readlen) readlen = remaining; |
| 2717 | } |
| 2718 | |
| 2719 | qblen = sdslen(c->querybuf); |
| 2720 | if (c->querybuf_peak < qblen) c->querybuf_peak = qblen; |
| 2721 | c->querybuf = sdsMakeRoomFor(c->querybuf, readlen); |
| 2722 | |
| 2723 | nread = connRead(c->conn, c->querybuf+qblen, readlen); |
| 2724 | |
| 2725 | if (nread == -1) { |
| 2726 | if (connGetState(conn) == CONN_STATE_CONNECTED) { |
| 2727 | return; |
| 2728 | } else { |
| 2729 | serverLog(LL_VERBOSE, "Reading from client: %s",connGetLastError(c->conn)); |
| 2730 | aelock.arm(c); |
| 2731 | freeClientAsync(c); |
| 2732 | return; |
| 2733 | } |
| 2734 | } else if (nread == 0) { |
| 2735 | serverLog(LL_VERBOSE, "Client closed connection"); |
| 2736 | aelock.arm(c); |
| 2737 | freeClientAsync(c); |
| 2738 | return; |
| 2739 | } else if (c->flags & CLIENT_MASTER) { |
| 2740 | /* Append the query buffer to the pending (not applied) buffer |
| 2741 | * of the master. We'll use this buffer later in order to have a |
no test coverage detected