| // Copyright (c) 2019 The Chromium Authors. All rights reserved. |
| // Use of this source code is governed by a BSD-style license that can be |
| // found in the LICENSE file. |
| |
| #include "net/third_party/quiche/src/quic/core/batch_writer/quic_batch_writer_base.h" |
| #include <cstdint> |
| |
| #include "net/third_party/quiche/src/quic/platform/api/quic_export.h" |
| #include "net/third_party/quiche/src/quic/platform/api/quic_flags.h" |
| #include "net/third_party/quiche/src/quic/platform/api/quic_ptr_util.h" |
| #include "net/third_party/quiche/src/quic/platform/api/quic_server_stats.h" |
| |
| namespace quic { |
| |
| QuicBatchWriterBase::QuicBatchWriterBase( |
| std::unique_ptr<QuicBatchWriterBuffer> batch_buffer) |
| : write_blocked_(false), batch_buffer_(std::move(batch_buffer)) {} |
| |
| WriteResult QuicBatchWriterBase::WritePacket( |
| const char* buffer, |
| size_t buf_len, |
| const QuicIpAddress& self_address, |
| const QuicSocketAddress& peer_address, |
| PerPacketOptions* options) { |
| const WriteResult result = |
| InternalWritePacket(buffer, buf_len, self_address, peer_address, options); |
| if (result.status == WRITE_STATUS_BLOCKED) { |
| write_blocked_ = true; |
| } |
| return result; |
| } |
| |
| QuicBatchWriterBase::ReleaseTime QuicBatchWriterBase::GetReleaseTime( |
| const PerPacketOptions* options) const { |
| DCHECK(SupportsReleaseTime()); |
| |
| if (options == nullptr) { |
| return {0, QuicTime::Delta::Zero()}; |
| } |
| |
| const uint64_t now = NowInNanosForReleaseTime(); |
| const uint64_t ideal_release_time = |
| now + options->release_time_delay.ToMicroseconds() * 1000; |
| |
| if ((options->release_time_delay.IsZero() || options->allow_burst) && |
| !buffered_writes().empty() && |
| // If release time of buffered packets is in the past, flush buffered |
| // packets and buffer this packet at the ideal release time. |
| (buffered_writes().back().release_time >= now)) { |
| // Send as soon as possible, but no sooner than the last buffered packet. |
| const uint64_t actual_release_time = buffered_writes().back().release_time; |
| |
| const int64_t offset_ns = actual_release_time - ideal_release_time; |
| if (offset_ns >= 0) { |
| QUIC_SERVER_HISTOGRAM_TIMES("batch_writer_positive_release_time_offset", |
| offset_ns / 1000, 1, 100000, 50, |
| "Duration from ideal release time to actual " |
| "release time, in microseconds."); |
| } else { |
| QUIC_SERVER_HISTOGRAM_TIMES("batch_writer_negative_release_time_offset", |
| -offset_ns / 1000, 1, 100000, 50, |
| "Duration from actual release time to ideal " |
| "release time, in microseconds."); |
| } |
| |
| ReleaseTime result{actual_release_time, |
| QuicTime::Delta::FromMicroseconds(offset_ns / 1000)}; |
| |
| QUIC_DVLOG(1) << "ideal_release_time:" << ideal_release_time |
| << ", actual_release_time:" << actual_release_time |
| << ", offset:" << result.release_time_offset; |
| return result; |
| } |
| |
| // Send according to the release time delay. |
| return {ideal_release_time, QuicTime::Delta::Zero()}; |
| } |
| |
| WriteResult QuicBatchWriterBase::InternalWritePacket( |
| const char* buffer, |
| size_t buf_len, |
| const QuicIpAddress& self_address, |
| const QuicSocketAddress& peer_address, |
| PerPacketOptions* options) { |
| if (buf_len > kMaxOutgoingPacketSize) { |
| return WriteResult(WRITE_STATUS_MSG_TOO_BIG, EMSGSIZE); |
| } |
| |
| ReleaseTime release_time = SupportsReleaseTime() |
| ? GetReleaseTime(options) |
| : ReleaseTime{0, QuicTime::Delta::Zero()}; |
| |
| const CanBatchResult can_batch_result = |
| CanBatch(buffer, buf_len, self_address, peer_address, options, |
| release_time.actual_release_time); |
| |
| bool buffered = false; |
| bool flush = can_batch_result.must_flush; |
| |
| if (can_batch_result.can_batch) { |
| QuicBatchWriterBuffer::PushResult push_result = |
| batch_buffer_->PushBufferedWrite(buffer, buf_len, self_address, |
| peer_address, options, |
| release_time.actual_release_time); |
| if (push_result.succeeded) { |
| buffered = true; |
| // If there's no space left after the packet is buffered, force a flush. |
| flush = flush || (batch_buffer_->GetNextWriteLocation() == nullptr); |
| } else { |
| // If there's no space without this packet, force a flush. |
| flush = true; |
| } |
| } |
| |
| if (!flush) { |
| WriteResult result(WRITE_STATUS_OK, 0); |
| result.send_time_offset = release_time.release_time_offset; |
| return result; |
| } |
| |
| size_t num_buffered_packets = buffered_writes().size(); |
| const FlushImplResult flush_result = CheckedFlush(); |
| WriteResult result = flush_result.write_result; |
| QUIC_DVLOG(1) << "Internally flushed " << flush_result.num_packets_sent |
| << " out of " << num_buffered_packets |
| << " packets. WriteResult=" << result; |
| |
| if (result.status != WRITE_STATUS_OK) { |
| if (IsWriteBlockedStatus(result.status)) { |
| return WriteResult( |
| buffered ? WRITE_STATUS_BLOCKED_DATA_BUFFERED : WRITE_STATUS_BLOCKED, |
| result.error_code); |
| } |
| |
| // Drop all packets, including the one being written. |
| size_t dropped_packets = |
| buffered ? buffered_writes().size() : buffered_writes().size() + 1; |
| |
| batch_buffer().Clear(); |
| result.dropped_packets = |
| dropped_packets > std::numeric_limits<uint16_t>::max() |
| ? std::numeric_limits<uint16_t>::max() |
| : static_cast<uint16_t>(dropped_packets); |
| return result; |
| } |
| |
| if (!buffered) { |
| QuicBatchWriterBuffer::PushResult push_result = |
| batch_buffer_->PushBufferedWrite(buffer, buf_len, self_address, |
| peer_address, options, |
| release_time.actual_release_time); |
| buffered = push_result.succeeded; |
| |
| // Since buffered_writes has been emptied, this write must have been |
| // buffered successfully. |
| QUIC_BUG_IF(!buffered) << "Failed to push to an empty batch buffer." |
| << " self_addr:" << self_address.ToString() |
| << ", peer_addr:" << peer_address.ToString() |
| << ", buf_len:" << buf_len; |
| } |
| |
| result.send_time_offset = release_time.release_time_offset; |
| return result; |
| } |
| |
| QuicBatchWriterBase::FlushImplResult QuicBatchWriterBase::CheckedFlush() { |
| if (buffered_writes().empty()) { |
| return FlushImplResult{WriteResult(WRITE_STATUS_OK, 0), |
| /*num_packets_sent=*/0, /*bytes_written=*/0}; |
| } |
| |
| const FlushImplResult flush_result = FlushImpl(); |
| |
| // Either flush_result.write_result.status is not WRITE_STATUS_OK, or it is |
| // WRITE_STATUS_OK and batch_buffer is empty. |
| DCHECK(flush_result.write_result.status != WRITE_STATUS_OK || |
| buffered_writes().empty()); |
| |
| // Flush should never return WRITE_STATUS_BLOCKED_DATA_BUFFERED. |
| DCHECK(flush_result.write_result.status != |
| WRITE_STATUS_BLOCKED_DATA_BUFFERED); |
| |
| return flush_result; |
| } |
| |
| WriteResult QuicBatchWriterBase::Flush() { |
| size_t num_buffered_packets = buffered_writes().size(); |
| FlushImplResult flush_result = CheckedFlush(); |
| QUIC_DVLOG(1) << "Externally flushed " << flush_result.num_packets_sent |
| << " out of " << num_buffered_packets |
| << " packets. WriteResult=" << flush_result.write_result; |
| |
| if (IsWriteError(flush_result.write_result.status)) { |
| if (buffered_writes().size() > std::numeric_limits<uint16_t>::max()) { |
| flush_result.write_result.dropped_packets = |
| std::numeric_limits<uint16_t>::max(); |
| } else { |
| flush_result.write_result.dropped_packets = |
| static_cast<uint16_t>(buffered_writes().size()); |
| } |
| // Treat all errors as non-retryable fatal errors. Drop all buffered packets |
| // to avoid sending them and getting the same error again. |
| batch_buffer().Clear(); |
| } |
| |
| if (flush_result.write_result.status == WRITE_STATUS_BLOCKED) { |
| write_blocked_ = true; |
| } |
| return flush_result.write_result; |
| } |
| |
| } // namespace quic |
