spdy_write_queue.cc 7.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224
  1. // Copyright (c) 2013 The Chromium Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file.
  4. #include "net/spdy/spdy_write_queue.h"
  5. #include <cstddef>
  6. #include <utility>
  7. #include <vector>
  8. #include "base/check_op.h"
  9. #include "base/containers/circular_deque.h"
  10. #include "base/trace_event/memory_usage_estimator.h"
  11. #include "net/spdy/spdy_buffer.h"
  12. #include "net/spdy/spdy_buffer_producer.h"
  13. #include "net/spdy/spdy_stream.h"
  14. namespace net {
  15. bool IsSpdyFrameTypeWriteCapped(spdy::SpdyFrameType frame_type) {
  16. return frame_type == spdy::SpdyFrameType::RST_STREAM ||
  17. frame_type == spdy::SpdyFrameType::SETTINGS ||
  18. frame_type == spdy::SpdyFrameType::WINDOW_UPDATE ||
  19. frame_type == spdy::SpdyFrameType::PING ||
  20. frame_type == spdy::SpdyFrameType::GOAWAY;
  21. }
  22. SpdyWriteQueue::PendingWrite::PendingWrite() = default;
  23. SpdyWriteQueue::PendingWrite::PendingWrite(
  24. spdy::SpdyFrameType frame_type,
  25. std::unique_ptr<SpdyBufferProducer> frame_producer,
  26. const base::WeakPtr<SpdyStream>& stream,
  27. const MutableNetworkTrafficAnnotationTag& traffic_annotation)
  28. : frame_type(frame_type),
  29. frame_producer(std::move(frame_producer)),
  30. stream(stream),
  31. traffic_annotation(traffic_annotation),
  32. has_stream(stream.get() != nullptr) {}
  33. SpdyWriteQueue::PendingWrite::~PendingWrite() = default;
  34. SpdyWriteQueue::PendingWrite::PendingWrite(PendingWrite&& other) = default;
  35. SpdyWriteQueue::PendingWrite& SpdyWriteQueue::PendingWrite::operator=(
  36. PendingWrite&& other) = default;
  37. SpdyWriteQueue::SpdyWriteQueue() = default;
  38. SpdyWriteQueue::~SpdyWriteQueue() {
  39. DCHECK_GE(num_queued_capped_frames_, 0);
  40. Clear();
  41. }
  42. bool SpdyWriteQueue::IsEmpty() const {
  43. for (int i = MINIMUM_PRIORITY; i <= MAXIMUM_PRIORITY; i++) {
  44. if (!queue_[i].empty())
  45. return false;
  46. }
  47. return true;
  48. }
  49. void SpdyWriteQueue::Enqueue(
  50. RequestPriority priority,
  51. spdy::SpdyFrameType frame_type,
  52. std::unique_ptr<SpdyBufferProducer> frame_producer,
  53. const base::WeakPtr<SpdyStream>& stream,
  54. const NetworkTrafficAnnotationTag& traffic_annotation) {
  55. CHECK(!removing_writes_);
  56. CHECK_GE(priority, MINIMUM_PRIORITY);
  57. CHECK_LE(priority, MAXIMUM_PRIORITY);
  58. if (stream.get())
  59. DCHECK_EQ(stream->priority(), priority);
  60. queue_[priority].push_back(
  61. {frame_type, std::move(frame_producer), stream,
  62. MutableNetworkTrafficAnnotationTag(traffic_annotation)});
  63. if (IsSpdyFrameTypeWriteCapped(frame_type)) {
  64. DCHECK_GE(num_queued_capped_frames_, 0);
  65. num_queued_capped_frames_++;
  66. }
  67. }
  68. bool SpdyWriteQueue::Dequeue(
  69. spdy::SpdyFrameType* frame_type,
  70. std::unique_ptr<SpdyBufferProducer>* frame_producer,
  71. base::WeakPtr<SpdyStream>* stream,
  72. MutableNetworkTrafficAnnotationTag* traffic_annotation) {
  73. CHECK(!removing_writes_);
  74. for (int i = MAXIMUM_PRIORITY; i >= MINIMUM_PRIORITY; --i) {
  75. if (!queue_[i].empty()) {
  76. PendingWrite pending_write = std::move(queue_[i].front());
  77. queue_[i].pop_front();
  78. *frame_type = pending_write.frame_type;
  79. *frame_producer = std::move(pending_write.frame_producer);
  80. *stream = pending_write.stream;
  81. *traffic_annotation = pending_write.traffic_annotation;
  82. if (pending_write.has_stream)
  83. DCHECK(stream->get());
  84. if (IsSpdyFrameTypeWriteCapped(*frame_type)) {
  85. num_queued_capped_frames_--;
  86. DCHECK_GE(num_queued_capped_frames_, 0);
  87. }
  88. return true;
  89. }
  90. }
  91. return false;
  92. }
  93. void SpdyWriteQueue::RemovePendingWritesForStream(SpdyStream* stream) {
  94. CHECK(!removing_writes_);
  95. removing_writes_ = true;
  96. RequestPriority priority = stream->priority();
  97. CHECK_GE(priority, MINIMUM_PRIORITY);
  98. CHECK_LE(priority, MAXIMUM_PRIORITY);
  99. #if DCHECK_IS_ON()
  100. // |stream| should not have pending writes in a queue not matching
  101. // its priority.
  102. for (int i = MINIMUM_PRIORITY; i <= MAXIMUM_PRIORITY; ++i) {
  103. if (priority == i)
  104. continue;
  105. for (auto it = queue_[i].begin(); it != queue_[i].end(); ++it)
  106. DCHECK_NE(it->stream.get(), stream);
  107. }
  108. #endif
  109. // Defer deletion until queue iteration is complete, as
  110. // SpdyBuffer::~SpdyBuffer() can result in callbacks into SpdyWriteQueue.
  111. std::vector<std::unique_ptr<SpdyBufferProducer>> erased_buffer_producers;
  112. base::circular_deque<PendingWrite>& queue = queue_[priority];
  113. for (auto it = queue.begin(); it != queue.end();) {
  114. if (it->stream.get() == stream) {
  115. if (IsSpdyFrameTypeWriteCapped(it->frame_type)) {
  116. num_queued_capped_frames_--;
  117. DCHECK_GE(num_queued_capped_frames_, 0);
  118. }
  119. erased_buffer_producers.push_back(std::move(it->frame_producer));
  120. it = queue.erase(it);
  121. } else {
  122. ++it;
  123. }
  124. }
  125. removing_writes_ = false;
  126. // Iteration on |queue| is completed. Now |erased_buffer_producers| goes out
  127. // of scope, SpdyBufferProducers are destroyed.
  128. }
  129. void SpdyWriteQueue::RemovePendingWritesForStreamsAfter(
  130. spdy::SpdyStreamId last_good_stream_id) {
  131. CHECK(!removing_writes_);
  132. removing_writes_ = true;
  133. // Defer deletion until queue iteration is complete, as
  134. // SpdyBuffer::~SpdyBuffer() can result in callbacks into SpdyWriteQueue.
  135. std::vector<std::unique_ptr<SpdyBufferProducer>> erased_buffer_producers;
  136. for (int i = MINIMUM_PRIORITY; i <= MAXIMUM_PRIORITY; ++i) {
  137. base::circular_deque<PendingWrite>& queue = queue_[i];
  138. for (auto it = queue.begin(); it != queue.end();) {
  139. if (it->stream.get() && (it->stream->stream_id() > last_good_stream_id ||
  140. it->stream->stream_id() == 0)) {
  141. if (IsSpdyFrameTypeWriteCapped(it->frame_type)) {
  142. num_queued_capped_frames_--;
  143. DCHECK_GE(num_queued_capped_frames_, 0);
  144. }
  145. erased_buffer_producers.push_back(std::move(it->frame_producer));
  146. it = queue.erase(it);
  147. } else {
  148. ++it;
  149. }
  150. }
  151. }
  152. removing_writes_ = false;
  153. // Iteration on each |queue| is completed. Now |erased_buffer_producers| goes
  154. // out of scope, SpdyBufferProducers are destroyed.
  155. }
  156. void SpdyWriteQueue::ChangePriorityOfWritesForStream(
  157. SpdyStream* stream,
  158. RequestPriority old_priority,
  159. RequestPriority new_priority) {
  160. CHECK(!removing_writes_);
  161. DCHECK(stream);
  162. #if DCHECK_IS_ON()
  163. // |stream| should not have pending writes in a queue not matching
  164. // |old_priority|.
  165. for (int i = MINIMUM_PRIORITY; i <= MAXIMUM_PRIORITY; ++i) {
  166. if (i == old_priority)
  167. continue;
  168. for (auto it = queue_[i].begin(); it != queue_[i].end(); ++it)
  169. DCHECK_NE(it->stream.get(), stream);
  170. }
  171. #endif
  172. base::circular_deque<PendingWrite>& old_queue = queue_[old_priority];
  173. base::circular_deque<PendingWrite>& new_queue = queue_[new_priority];
  174. for (auto it = old_queue.begin(); it != old_queue.end();) {
  175. if (it->stream.get() == stream) {
  176. new_queue.push_back(std::move(*it));
  177. it = old_queue.erase(it);
  178. } else {
  179. ++it;
  180. }
  181. }
  182. }
  183. void SpdyWriteQueue::Clear() {
  184. CHECK(!removing_writes_);
  185. removing_writes_ = true;
  186. std::vector<std::unique_ptr<SpdyBufferProducer>> erased_buffer_producers;
  187. for (int i = MINIMUM_PRIORITY; i <= MAXIMUM_PRIORITY; ++i) {
  188. for (auto& pending_write : queue_[i]) {
  189. erased_buffer_producers.push_back(
  190. std::move(pending_write.frame_producer));
  191. }
  192. queue_[i].clear();
  193. }
  194. removing_writes_ = false;
  195. num_queued_capped_frames_ = 0;
  196. }
  197. } // namespace net