| 204 | } |
| 205 | |
| 206 | int InputMessenger::ProcessNewMessage( |
| 207 | Socket* m, ssize_t bytes, bool read_eof, |
| 208 | const uint64_t received_us, const uint64_t base_realtime, |
| 209 | InputMessageClosure& last_msg) { |
| 210 | m->AddInputBytes(bytes); |
| 211 | |
| 212 | // Avoid this socket to be closed due to idle_timeout_s |
| 213 | m->_last_readtime_us.store(received_us, butil::memory_order_relaxed); |
| 214 | |
| 215 | size_t last_size = m->_read_buf.length(); |
| 216 | int num_bthread_created = 0; |
| 217 | while (1) { |
| 218 | size_t index = 8888; |
| 219 | ParseResult pr = CutInputMessage(m, &index, read_eof); |
| 220 | if (!pr.is_ok()) { |
| 221 | if (pr.error() == PARSE_ERROR_NOT_ENOUGH_DATA) { |
| 222 | // incomplete message, re-read. |
| 223 | // However, some buffer may have been consumed |
| 224 | // under protocols like HTTP. Record this size |
| 225 | m->_last_msg_size += (last_size - m->_read_buf.length()); |
| 226 | break; |
| 227 | } else if (pr.error() == PARSE_ERROR_TRY_OTHERS) { |
| 228 | LOG(WARNING) |
| 229 | << "Close " << *m << " due to unknown message: " |
| 230 | << butil::ToPrintable(m->_read_buf); |
| 231 | m->SetFailed(EINVAL, "Close %s due to unknown message", |
| 232 | m->description().c_str()); |
| 233 | return -1; |
| 234 | } else { |
| 235 | LOG(WARNING) << "Close " << *m << ": " << pr.error_str(); |
| 236 | m->SetFailed(EINVAL, "Close %s: %s", |
| 237 | m->description().c_str(), pr.error_str()); |
| 238 | return -1; |
| 239 | } |
| 240 | } |
| 241 | |
| 242 | m->AddInputMessages(1); |
| 243 | // Calculate average size of messages |
| 244 | const size_t cur_size = m->_read_buf.length(); |
| 245 | if (cur_size == 0) { |
| 246 | // _read_buf is consumed, it's good timing to return blocks |
| 247 | // cached internally back to TLS, otherwise the memory is not |
| 248 | // reused until next message arrives which is quite uncertain |
| 249 | // in situations that most connections are idle. |
| 250 | m->_read_buf.return_cached_blocks(); |
| 251 | } |
| 252 | m->_last_msg_size += (last_size - cur_size); |
| 253 | last_size = cur_size; |
| 254 | const size_t old_avg = m->_avg_msg_size; |
| 255 | if (old_avg != 0) { |
| 256 | m->_avg_msg_size = (old_avg * (MSG_SIZE_WINDOW - 1) + m->_last_msg_size) |
| 257 | / MSG_SIZE_WINDOW; |
| 258 | } else { |
| 259 | m->_avg_msg_size = m->_last_msg_size; |
| 260 | } |
| 261 | m->_last_msg_size = 0; |
| 262 | |
| 263 | if (pr.message() == NULL) { // the Process() step can be skipped. |
no test coverage detected