| 324 | } |
| 325 | |
| 326 | int Stream::AppendIfNotFull(const butil::IOBuf &data, |
| 327 | const StreamWriteOptions* options) { |
| 328 | if (_cur_buf_size > 0) { |
| 329 | std::unique_lock<bthread_mutex_t> lck(_congestion_control_mutex); |
| 330 | if (_produced >= _remote_consumed + _cur_buf_size) { |
| 331 | const size_t saved_produced = _produced; |
| 332 | const size_t saved_remote_consumed = _remote_consumed; |
| 333 | lck.unlock(); |
| 334 | RPC_VLOG << "Stream=" << _id << " is full" |
| 335 | << "_produced=" << saved_produced |
| 336 | << " _remote_consumed=" << saved_remote_consumed |
| 337 | << " gap=" << saved_produced - saved_remote_consumed |
| 338 | << " max_buf_size=" << _cur_buf_size; |
| 339 | return 1; |
| 340 | } |
| 341 | _produced += data.length(); |
| 342 | } |
| 343 | |
| 344 | size_t data_length = data.length(); |
| 345 | butil::IOBuf copied_data(data); |
| 346 | Socket::WriteOptions wopt; |
| 347 | wopt.write_in_background = options != NULL && options->write_in_background; |
| 348 | const int rc = _fake_socket_weak_ref->Write(&copied_data, &wopt); |
| 349 | if (rc != 0) { |
| 350 | // Stream may be closed by peer before |
| 351 | LOG(WARNING) << "Fail to write to _fake_socket, " << berror(); |
| 352 | BAIDU_SCOPED_LOCK(_congestion_control_mutex); |
| 353 | _produced -= data_length; |
| 354 | return -1; |
| 355 | } |
| 356 | if (FLAGS_socket_max_streams_unconsumed_bytes > 0) { |
| 357 | _host_socket->_total_streams_unconsumed_size += data_length; |
| 358 | } |
| 359 | return 0; |
| 360 | } |
| 361 | |
| 362 | void Stream::SetRemoteConsumed(size_t new_remote_consumed) { |
| 363 | CHECK(_cur_buf_size > 0); |
no test coverage detected