| 95 | } |
| 96 | |
| 97 | ChainedBuffers ReaderExecutor::readNextWindow() |
| 98 | { |
| 99 | StatTimer work_timer(stats, Stats::WorkMicroseconds); |
| 100 | |
| 101 | if (atEnd()) |
| 102 | return {}; |
| 103 | |
| 104 | size_t object_logical_start_offset = 0; |
| 105 | const StoredObject * object = offset_map.findObjectAt(position, &object_logical_start_offset); |
| 106 | if (!object) |
| 107 | { |
| 108 | reached_eof = true; |
| 109 | return {}; |
| 110 | } |
| 111 | |
| 112 | const size_t object_offset = position - object_logical_start_offset; |
| 113 | |
| 114 | /// Clamp the block to the object boundary so a window never straddles two |
| 115 | /// objects; the next call continues in the next object. Unknown total size |
| 116 | /// means stream a full block and let a short read mark EOF. |
| 117 | size_t want = block_size; |
| 118 | if (!offset_map.hasUnknownSize()) |
| 119 | { |
| 120 | const size_t remaining_in_object = object->bytes_size - object_offset; |
| 121 | want = std::min(block_size, remaining_in_object); |
| 122 | if (want == 0) |
| 123 | { |
| 124 | reached_eof = true; |
| 125 | return {}; |
| 126 | } |
| 127 | } |
| 128 | |
| 129 | /// `atEnd` already returned at the bound, so `position < *read_until` here. |
| 130 | chassert(!read_until || *read_until >= position); |
| 131 | if (read_until && *read_until - position < want) |
| 132 | want = *read_until - position; |
| 133 | |
| 134 | auto buffer = source->open(*object); |
| 135 | |
| 136 | /// Bound the request to the window so a remote source fetches exactly `want` |
| 137 | /// bytes rather than an open-ended tail that is then cancelled. Set before |
| 138 | /// the seek so the bound applies to the connection opened on the first read. |
| 139 | if (buffer->supportsRightBoundedReads()) |
| 140 | buffer->setReadUntilPosition(object_offset + want); |
| 141 | |
| 142 | if (object_offset > 0) |
| 143 | buffer->seek(static_cast<off_t>(object_offset), SEEK_SET); |
| 144 | |
| 145 | auto block = std::make_shared<OwnedChainedBuffer>(want); |
| 146 | const size_t got = buffer->read(block->data(), want); |
| 147 | |
| 148 | /// One open+read per window; requested bytes equal source bytes until caches/over-read land. |
| 149 | stats.add(Stats::SourceRequests); |
| 150 | stats.add(Stats::BytesFromSource, got); |
| 151 | stats.add(Stats::RequestedBytes, got); |
| 152 | |
| 153 | if (offset_map.hasUnknownSize()) |
| 154 | { |