| OLD | NEW |
| 1 // Copyright (c) 2012 The Chromium Authors. All rights reserved. | 1 // Copyright (c) 2012 The Chromium Authors. All rights reserved. |
| 2 // Use of this source code is governed by a BSD-style license that can be | 2 // Use of this source code is governed by a BSD-style license that can be |
| 3 // found in the LICENSE file. | 3 // found in the LICENSE file. |
| 4 | 4 |
| 5 #include "net/quic/quic_stream_sequencer.h" | 5 #include "net/quic/quic_stream_sequencer.h" |
| 6 | 6 |
| 7 #include <algorithm> | 7 #include <algorithm> |
| 8 #include <limits> | 8 #include <limits> |
| 9 | 9 |
| 10 #include "base/logging.h" | 10 #include "base/logging.h" |
| (...skipping 72 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 83 data_len -= bytes_consumed; | 83 data_len -= bytes_consumed; |
| 84 data.Consume(bytes_consumed); | 84 data.Consume(bytes_consumed); |
| 85 byte_offset += bytes_consumed; | 85 byte_offset += bytes_consumed; |
| 86 } | 86 } |
| 87 } | 87 } |
| 88 | 88 |
| 89 // Buffer any remaining data to be consumed by the stream when ready. | 89 // Buffer any remaining data to be consumed by the stream when ready. |
| 90 for (size_t i = 0; i < data.Size(); ++i) { | 90 for (size_t i = 0; i < data.Size(); ++i) { |
| 91 DVLOG(1) << "Buffering stream data at offset " << byte_offset; | 91 DVLOG(1) << "Buffering stream data at offset " << byte_offset; |
| 92 const iovec& iov = data.iovec()[i]; | 92 const iovec& iov = data.iovec()[i]; |
| 93 frames_.insert(make_pair( | 93 buffered_frames_.insert(make_pair( |
| 94 byte_offset, string(static_cast<char*>(iov.iov_base), iov.iov_len))); | 94 byte_offset, string(static_cast<char*>(iov.iov_base), iov.iov_len))); |
| 95 byte_offset += iov.iov_len; | 95 byte_offset += iov.iov_len; |
| 96 num_bytes_buffered_ += iov.iov_len; | 96 num_bytes_buffered_ += iov.iov_len; |
| 97 } | 97 } |
| 98 return true; | 98 return true; |
| 99 } | 99 } |
| 100 | 100 |
| 101 void QuicStreamSequencer::CloseStreamAtOffset(QuicStreamOffset offset) { | 101 void QuicStreamSequencer::CloseStreamAtOffset(QuicStreamOffset offset) { |
| 102 const QuicStreamOffset kMaxOffset = numeric_limits<QuicStreamOffset>::max(); | 102 const QuicStreamOffset kMaxOffset = numeric_limits<QuicStreamOffset>::max(); |
| 103 | 103 |
| (...skipping 10 matching lines...) Expand all Loading... |
| 114 } | 114 } |
| 115 | 115 |
| 116 bool QuicStreamSequencer::MaybeCloseStream() { | 116 bool QuicStreamSequencer::MaybeCloseStream() { |
| 117 if (!blocked_ && IsClosed()) { | 117 if (!blocked_ && IsClosed()) { |
| 118 DVLOG(1) << "Passing up termination, as we've processed " | 118 DVLOG(1) << "Passing up termination, as we've processed " |
| 119 << num_bytes_consumed_ << " of " << close_offset_ | 119 << num_bytes_consumed_ << " of " << close_offset_ |
| 120 << " bytes."; | 120 << " bytes."; |
| 121 // Technically it's an error if num_bytes_consumed isn't exactly | 121 // Technically it's an error if num_bytes_consumed isn't exactly |
| 122 // equal, but error handling seems silly at this point. | 122 // equal, but error handling seems silly at this point. |
| 123 stream_->OnFinRead(); | 123 stream_->OnFinRead(); |
| 124 frames_.clear(); | 124 buffered_frames_.clear(); |
| 125 num_bytes_buffered_ = 0; | 125 num_bytes_buffered_ = 0; |
| 126 return true; | 126 return true; |
| 127 } | 127 } |
| 128 return false; | 128 return false; |
| 129 } | 129 } |
| 130 | 130 |
| 131 int QuicStreamSequencer::GetReadableRegions(iovec* iov, size_t iov_len) { | 131 int QuicStreamSequencer::GetReadableRegions(iovec* iov, size_t iov_len) { |
| 132 DCHECK(!blocked_); | 132 DCHECK(!blocked_); |
| 133 FrameMap::iterator it = frames_.begin(); | 133 FrameMap::iterator it = buffered_frames_.begin(); |
| 134 size_t index = 0; | 134 size_t index = 0; |
| 135 QuicStreamOffset offset = num_bytes_consumed_; | 135 QuicStreamOffset offset = num_bytes_consumed_; |
| 136 while (it != frames_.end() && index < iov_len) { | 136 while (it != buffered_frames_.end() && index < iov_len) { |
| 137 if (it->first != offset) return index; | 137 if (it->first != offset) return index; |
| 138 | 138 |
| 139 iov[index].iov_base = static_cast<void*>( | 139 iov[index].iov_base = static_cast<void*>( |
| 140 const_cast<char*>(it->second.data())); | 140 const_cast<char*>(it->second.data())); |
| 141 iov[index].iov_len = it->second.size(); | 141 iov[index].iov_len = it->second.size(); |
| 142 offset += it->second.size(); | 142 offset += it->second.size(); |
| 143 | 143 |
| 144 ++index; | 144 ++index; |
| 145 ++it; | 145 ++it; |
| 146 } | 146 } |
| 147 return index; | 147 return index; |
| 148 } | 148 } |
| 149 | 149 |
| 150 int QuicStreamSequencer::Readv(const struct iovec* iov, size_t iov_len) { | 150 int QuicStreamSequencer::Readv(const struct iovec* iov, size_t iov_len) { |
| 151 DCHECK(!blocked_); | 151 DCHECK(!blocked_); |
| 152 FrameMap::iterator it = frames_.begin(); | 152 FrameMap::iterator it = buffered_frames_.begin(); |
| 153 size_t iov_index = 0; | 153 size_t iov_index = 0; |
| 154 size_t iov_offset = 0; | 154 size_t iov_offset = 0; |
| 155 size_t frame_offset = 0; | 155 size_t frame_offset = 0; |
| 156 size_t initial_bytes_consumed = num_bytes_consumed_; | 156 size_t initial_bytes_consumed = num_bytes_consumed_; |
| 157 | 157 |
| 158 while (iov_index < iov_len && | 158 while (iov_index < iov_len && |
| 159 it != frames_.end() && | 159 it != buffered_frames_.end() && |
| 160 it->first == num_bytes_consumed_) { | 160 it->first == num_bytes_consumed_) { |
| 161 int bytes_to_read = min(iov[iov_index].iov_len - iov_offset, | 161 int bytes_to_read = min(iov[iov_index].iov_len - iov_offset, |
| 162 it->second.size() - frame_offset); | 162 it->second.size() - frame_offset); |
| 163 | 163 |
| 164 char* iov_ptr = static_cast<char*>(iov[iov_index].iov_base) + iov_offset; | 164 char* iov_ptr = static_cast<char*>(iov[iov_index].iov_base) + iov_offset; |
| 165 memcpy(iov_ptr, | 165 memcpy(iov_ptr, |
| 166 it->second.data() + frame_offset, bytes_to_read); | 166 it->second.data() + frame_offset, bytes_to_read); |
| 167 frame_offset += bytes_to_read; | 167 frame_offset += bytes_to_read; |
| 168 iov_offset += bytes_to_read; | 168 iov_offset += bytes_to_read; |
| 169 | 169 |
| 170 if (iov[iov_index].iov_len == iov_offset) { | 170 if (iov[iov_index].iov_len == iov_offset) { |
| 171 // We've filled this buffer. | 171 // We've filled this buffer. |
| 172 iov_offset = 0; | 172 iov_offset = 0; |
| 173 ++iov_index; | 173 ++iov_index; |
| 174 } | 174 } |
| 175 if (it->second.size() == frame_offset) { | 175 if (it->second.size() == frame_offset) { |
| 176 // We've copied this whole frame | 176 // We've copied this whole frame |
| 177 RecordBytesConsumed(it->second.size()); | 177 RecordBytesConsumed(it->second.size()); |
| 178 frames_.erase(it); | 178 buffered_frames_.erase(it); |
| 179 it = frames_.begin(); | 179 it = buffered_frames_.begin(); |
| 180 frame_offset = 0; | 180 frame_offset = 0; |
| 181 } | 181 } |
| 182 } | 182 } |
| 183 // We've finished copying. If we have a partial frame, update it. | 183 // We've finished copying. If we have a partial frame, update it. |
| 184 if (frame_offset != 0) { | 184 if (frame_offset != 0) { |
| 185 frames_.insert(make_pair(it->first + frame_offset, | 185 buffered_frames_.insert(make_pair(it->first + frame_offset, |
| 186 it->second.substr(frame_offset))); | 186 it->second.substr(frame_offset))); |
| 187 frames_.erase(frames_.begin()); | 187 buffered_frames_.erase(buffered_frames_.begin()); |
| 188 RecordBytesConsumed(frame_offset); | 188 RecordBytesConsumed(frame_offset); |
| 189 } | 189 } |
| 190 return num_bytes_consumed_ - initial_bytes_consumed; | 190 return num_bytes_consumed_ - initial_bytes_consumed; |
| 191 } | 191 } |
| 192 | 192 |
| 193 bool QuicStreamSequencer::HasBytesToRead() const { | 193 bool QuicStreamSequencer::HasBytesToRead() const { |
| 194 FrameMap::const_iterator it = frames_.begin(); | 194 FrameMap::const_iterator it = buffered_frames_.begin(); |
| 195 | 195 |
| 196 return it != frames_.end() && it->first == num_bytes_consumed_; | 196 return it != buffered_frames_.end() && it->first == num_bytes_consumed_; |
| 197 } | 197 } |
| 198 | 198 |
| 199 bool QuicStreamSequencer::IsClosed() const { | 199 bool QuicStreamSequencer::IsClosed() const { |
| 200 return num_bytes_consumed_ >= close_offset_; | 200 return num_bytes_consumed_ >= close_offset_; |
| 201 } | 201 } |
| 202 | 202 |
| 203 bool QuicStreamSequencer::IsDuplicate(const QuicStreamFrame& frame) const { | 203 bool QuicStreamSequencer::IsDuplicate(const QuicStreamFrame& frame) const { |
| 204 // A frame is duplicate if the frame offset is smaller than our bytes consumed | 204 // A frame is duplicate if the frame offset is smaller than our bytes consumed |
| 205 // or we have stored the frame in our map. | 205 // or we have stored the frame in our map. |
| 206 // TODO(pwestin): Is it possible that a new frame contain more data even if | 206 // TODO(pwestin): Is it possible that a new frame contain more data even if |
| 207 // the offset is the same? | 207 // the offset is the same? |
| 208 return frame.offset < num_bytes_consumed_ || | 208 return frame.offset < num_bytes_consumed_ || |
| 209 frames_.find(frame.offset) != frames_.end(); | 209 buffered_frames_.find(frame.offset) != buffered_frames_.end(); |
| 210 } | 210 } |
| 211 | 211 |
| 212 void QuicStreamSequencer::SetBlockedUntilFlush() { | 212 void QuicStreamSequencer::SetBlockedUntilFlush() { |
| 213 blocked_ = true; | 213 blocked_ = true; |
| 214 } | 214 } |
| 215 | 215 |
| 216 void QuicStreamSequencer::FlushBufferedFrames() { | 216 void QuicStreamSequencer::FlushBufferedFrames() { |
| 217 blocked_ = false; | 217 blocked_ = false; |
| 218 FrameMap::iterator it = frames_.find(num_bytes_consumed_); | 218 FrameMap::iterator it = buffered_frames_.find(num_bytes_consumed_); |
| 219 while (it != frames_.end()) { | 219 while (it != buffered_frames_.end()) { |
| 220 DVLOG(1) << "Flushing buffered packet at offset " << it->first; | 220 DVLOG(1) << "Flushing buffered packet at offset " << it->first; |
| 221 string* data = &it->second; | 221 string* data = &it->second; |
| 222 size_t bytes_consumed = stream_->ProcessRawData(data->c_str(), | 222 size_t bytes_consumed = stream_->ProcessRawData(data->c_str(), |
| 223 data->size()); | 223 data->size()); |
| 224 RecordBytesConsumed(bytes_consumed); | 224 RecordBytesConsumed(bytes_consumed); |
| 225 if (MaybeCloseStream()) { | 225 if (MaybeCloseStream()) { |
| 226 return; | 226 return; |
| 227 } | 227 } |
| 228 if (bytes_consumed > data->size()) { | 228 if (bytes_consumed > data->size()) { |
| 229 stream_->Reset(QUIC_ERROR_PROCESSING_STREAM); // Programming error | 229 stream_->Reset(QUIC_ERROR_PROCESSING_STREAM); // Programming error |
| 230 return; | 230 return; |
| 231 } else if (bytes_consumed == data->size()) { | 231 } else if (bytes_consumed == data->size()) { |
| 232 frames_.erase(it); | 232 buffered_frames_.erase(it); |
| 233 it = frames_.find(num_bytes_consumed_); | 233 it = buffered_frames_.find(num_bytes_consumed_); |
| 234 } else { | 234 } else { |
| 235 string new_data = it->second.substr(bytes_consumed); | 235 string new_data = it->second.substr(bytes_consumed); |
| 236 frames_.erase(it); | 236 buffered_frames_.erase(it); |
| 237 frames_.insert(make_pair(num_bytes_consumed_, new_data)); | 237 buffered_frames_.insert(make_pair(num_bytes_consumed_, new_data)); |
| 238 return; | 238 return; |
| 239 } | 239 } |
| 240 } | 240 } |
| 241 MaybeCloseStream(); | 241 MaybeCloseStream(); |
| 242 } | 242 } |
| 243 | 243 |
| 244 void QuicStreamSequencer::RecordBytesConsumed(size_t bytes_consumed) { | 244 void QuicStreamSequencer::RecordBytesConsumed(size_t bytes_consumed) { |
| 245 num_bytes_consumed_ += bytes_consumed; | 245 num_bytes_consumed_ += bytes_consumed; |
| 246 num_bytes_buffered_ -= bytes_consumed; | 246 num_bytes_buffered_ -= bytes_consumed; |
| 247 | 247 |
| 248 stream_->AddBytesConsumed(bytes_consumed); | 248 stream_->AddBytesConsumed(bytes_consumed); |
| 249 } | 249 } |
| 250 | 250 |
| 251 } // namespace net | 251 } // namespace net |
| OLD | NEW |