| 115 | |
| 116 | |
| 117 | ParseResult ParseRedisMessage(butil::IOBuf* source, Socket* socket, |
| 118 | bool read_eof, const void* arg) { |
| 119 | if (read_eof || source->empty()) { |
| 120 | return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA); |
| 121 | } |
| 122 | const Server* server = static_cast<const Server*>(arg); |
| 123 | if (server) { |
| 124 | const RedisService* const rs = server->options().redis_service; |
| 125 | if (!rs) { |
| 126 | return MakeParseError(PARSE_ERROR_TRY_OTHERS); |
| 127 | } |
| 128 | RedisConnContext* ctx = static_cast<RedisConnContext*>(socket->parsing_context()); |
| 129 | if (ctx == NULL) { |
| 130 | ctx = new RedisConnContext(rs); |
| 131 | socket->reset_parsing_context(ctx); |
| 132 | } |
| 133 | std::vector<butil::StringPiece> current_args; |
| 134 | butil::IOBufAppender appender; |
| 135 | ParseError err = PARSE_OK; |
| 136 | |
| 137 | err = ctx->parser.Consume(*source, ¤t_args, &ctx->arena); |
| 138 | if (err != PARSE_OK) { |
| 139 | return MakeParseError(err); |
| 140 | } |
| 141 | while (true) { |
| 142 | std::vector<butil::StringPiece> next_args; |
| 143 | err = ctx->parser.Consume(*source, &next_args, &ctx->arena); |
| 144 | if (err != PARSE_OK) { |
| 145 | break; |
| 146 | } |
| 147 | if (ConsumeCommand(ctx, current_args, false, &appender) != 0) { |
| 148 | return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG); |
| 149 | } |
| 150 | current_args.swap(next_args); |
| 151 | } |
| 152 | if (ConsumeCommand(ctx, current_args, |
| 153 | true /*must be the last message*/, &appender) != 0) { |
| 154 | return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG); |
| 155 | } |
| 156 | butil::IOBuf sendbuf; |
| 157 | appender.move_to(sendbuf); |
| 158 | CHECK(!sendbuf.empty()); |
| 159 | Socket::WriteOptions wopt; |
| 160 | wopt.ignore_eovercrowded = true; |
| 161 | LOG_IF(WARNING, socket->Write(&sendbuf, &wopt) != 0) |
| 162 | << "Fail to send redis reply"; |
| 163 | if(ctx->parser.ParsedArgsSize() == 0) { |
| 164 | ctx->arena.clear(); |
| 165 | } |
| 166 | return MakeParseError(err); |
| 167 | } else { |
| 168 | // NOTE(gejun): PopPipelinedInfo() is actually more contended than what |
| 169 | // I thought before. The Socket._pipeline_q is a SPSC queue pushed before |
| 170 | // sending and popped when response comes back, being protected by a |
| 171 | // mutex. Previously the mutex is shared with Socket._id_wait_list. When |
| 172 | // 200 bthreads access one redis-server, ~1.5s in total is spent on |
| 173 | // contention in 10-second duration. If the mutex is separated, the time |
| 174 | // drops to ~0.25s. I further replaced PeekPipelinedInfo() with |
nothing calls this directly
no test coverage detected