diff --git a/include/libtorrent/aux_/disk_cache.hpp b/include/libtorrent/aux_/disk_cache.hpp index 29ed579bf..493920b16 100644 --- a/include/libtorrent/aux_/disk_cache.hpp +++ b/include/libtorrent/aux_/disk_cache.hpp @@ -12,9 +12,11 @@ see LICENSE file. #include #include +#include #include #include #include +#include #include "libtorrent/storage_defs.hpp" #include "libtorrent/disk_interface.hpp" // for default_block_size @@ -51,28 +53,32 @@ namespace libtorrent { namespace libtorrent::aux { -namespace mi = boost::multi_index; + struct pread_storage; -using cached_piece_flags = libtorrent::flags::bitfield_flag; -// uniquely identifies a torrent and piece -struct piece_location -{ - piece_location(storage_index_t const t, piece_index_t const p) - : torrent(t), piece(p) {} - storage_index_t torrent; - piece_index_t piece; - bool operator==(piece_location const& rhs) const + namespace mi = boost::multi_index; + + using cached_piece_flags = + libtorrent::flags::bitfield_flag; + + // uniquely identifies a torrent and piece + struct piece_location { - return std::tie(torrent, piece) - == std::tie(rhs.torrent, rhs.piece); - } + piece_location(storage_index_t const t, piece_index_t const p) + : torrent(t) + , piece(p) + {} + storage_index_t torrent; + piece_index_t piece; + bool operator==(piece_location const& rhs) const + { + return std::tie(torrent, piece) == std::tie(rhs.torrent, rhs.piece); + } - bool operator<(piece_location const& rhs) const - { - return std::tie(torrent, piece) - < std::tie(rhs.torrent, rhs.piece); - } + bool operator<(piece_location const& rhs) const + { + return std::tie(torrent, piece) < std::tie(rhs.torrent, rhs.piece); + } }; inline size_t hash_value(piece_location const& l) @@ -133,32 +139,30 @@ struct TORRENT_EXTRA_EXPORT cached_block_entry write_state_t write_state; }; -// returns the buffer attached to the write job, or an empty span if j is null. -// found via ADL by visit_block_iovecs when iterating a span. +// ADL hook for visit_block_iovecs over span. Reads only +// borrowed_buf (not wj.buf) because flush_piece_impl invokes this without +// holding the cache mutex. inline span write_buf(disk_job const* j) { if (j == nullptr) return {}; TORRENT_ASSERT(j->get_type() == aux::job_action_t::write); auto const& w = std::get(j->action); - return {w.buf.data(), w.buffer_size}; + return {w.borrowed_buf, w.borrowed_buf ? w.buffer_size : std::uint16_t{0}}; } struct TORRENT_EXTRA_EXPORT cached_piece_entry { - cached_piece_entry(piece_location const& loc - , int const piece_size_v2 - , int const piece_size - , int const num_blocks - , bool v1 - , bool v2); + cached_piece_entry(piece_location const& loc, + int const piece_size_v2, + int const piece_size, + int const num_blocks, + bool v1); span get_blocks() const; piece_location piece; unique_ptr blocks; - // allocated only for v2 torrents (null for v1-only) - unique_ptr block_hashes; // allocated only for v1 torrents (null for v2-only) std::unique_ptr ph; @@ -199,6 +203,15 @@ struct TORRENT_EXTRA_EXPORT cached_piece_entry // the number of blocks that have a write job associated with them std::uint16_t num_jobs = 0; + // number of this piece's blocks that are either in m_v2_hash_queue or + // currently being processed by a drain batch. Decremented by + // drain_v2_hash_queue once a batch has stored its hashes on + // pread_storage. While > 0, a clear_piece must wait (try_clear_piece + // parks it on clear_piece) and a hash_job hung via wait_for_v2_queue or + // try_hash_piece's deferred path cannot post yet. drain_v2_hash_queue + // dispatches both when this hits zero. + std::uint16_t v2_pending = 0; + // this is set when the piece has been populated with all blocks; // it will make it prioritized for flushing to disk. // it will be cleared once all blocks have been flushed. @@ -220,10 +233,6 @@ struct TORRENT_EXTRA_EXPORT cached_piece_entry // set if this piece requires v1 SHA1 hashing (ph member is allocated). static constexpr cached_piece_flags v1_hashes_flag = 4_bit; - // set if this piece requires v2 SHA256 block hashing (block_hashes - // member is allocated). - static constexpr cached_piece_flags v2_hashes_flag = 5_bit; - // set when a block is inserted into this piece and the hasher thread // should be woken to make hashing progress. Cleared when the hasher // thread picks it up. Used to coalesce multiple insertions into a @@ -264,6 +273,40 @@ struct TORRENT_EXTRA_EXPORT cached_piece_entry using insert_result_flags = libtorrent::flags::bitfield_flag; +// one entry per v2 block waiting for a hasher thread. The entry owns the +// buffer until the hash is computed and stored on the storage; if the cbe +// is flushed while the entry is still queued, the queue keeps the buffer +// alive so the hasher can still consume it without a disk read-back. (If +// the cpe is removed entirely -- free_piece / clear_piece_impl -- the +// matching queue entries are dropped along with their buffers.) +struct v2_hash_queue_entry +{ + v2_hash_queue_entry(piece_location p, + std::shared_ptr s, + std::uint16_t b, + std::uint16_t hl, + disk_buffer_holder buf_) + : piece(p) + , storage(std::move(s)) + , block(b) + , hash_len(hl) + , buf(std::move(buf_)) + {} + + v2_hash_queue_entry(v2_hash_queue_entry&&) = default; + v2_hash_queue_entry& operator=(v2_hash_queue_entry&&) = default; + + piece_location piece; + // keeps the storage alive past torrent removal, so the precomputed + // hash can still be stored on it. + std::shared_ptr storage; + std::uint16_t block = 0; + // blocks straddling the v2 piece boundary in a hybrid torrent are + // hashed only up to piece_size2. + std::uint16_t hash_len = 0; + disk_buffer_holder buf; +}; + struct TORRENT_EXTRA_EXPORT disk_cache { disk_cache(io_context& ios); @@ -343,12 +386,9 @@ struct TORRENT_EXTRA_EXPORT disk_cache } } auto const& cbe = i->blocks[block_idx]; - // There's nothing stopping the hash threads from hashing the blocks in - // parallel. This should not depend on the hasher_cursor. That's a v1 - // concept - TORRENT_ASSERT(i->block_hashes); - if (!i->block_hashes[block_idx].is_all_zeros()) - return i->block_hashes[block_idx]; + // independent of hasher_cursor: any v2 block may be hashed in any + // order. Precomputed v2 hashes live on pread_storage and are + // consulted by the caller before reaching here. if (cbe.data()) { int const blk_size = std::min(default_block_size @@ -375,19 +415,20 @@ struct TORRENT_EXTRA_EXPORT disk_cache // the mutex, sets hashing=true (which pins all buffers at index >= hasher_cursor // so flush_piece_impl cannot free them), then releases the lock and calls: // - // f(ph, hasher_cursor, blocks, v2_hashes) + // f(ph, hasher_cursor, blocks) // // ph: SHA1 piece hasher already fed blocks [0, hasher_cursor). // hasher_cursor: index of the first block not yet incorporated in ph. // f() must continue feeding blocks from this index onward. // blocks: per-block buffer pointers. A null entry means the block is not // in the cache and must be read from disk. - // v2_hashes: per-block SHA256 hash snapshots. all-zeros means not-yet-computed. - // a non-zero entry is a pre-computed hash that f() may use directly without re-hashing. // // After f() returns hasher_cursor is advanced and buffers that have // already been flushed to disk are freed. If all blocks are flushed, the // piece entry is removed from the cache entirely. + // v2 block hashes (when applicable) live in pread_storage's + // precomputed_block_hashes; the caller is expected to consult that + // directly. template hash_piece_result hash_piece(piece_location const loc, disk_job* j, Fun f) { @@ -415,20 +456,11 @@ struct TORRENT_EXTRA_EXPORT disk_cache std::uint16_t const hasher_cursor = piece_iter->hasher_cursor; TORRENT_ALLOCA(blocks, char const*, blocks_in_piece); - TORRENT_ALLOCA(v2_hashes, sha256_hash, (piece_iter->flags & cached_piece_entry::v2_hashes_flag) ? blocks_in_piece : 0); - int num_unhashed = 0; for (int i = 0; i < blocks_in_piece; ++i) { char const* buf = piece_iter->blocks[i].data(); blocks[i] = buf; - if (buf && i >= hasher_cursor) - ++num_unhashed; - } - if (piece_iter->block_hashes) - { - for (int i = 0; i < blocks_in_piece; ++i) - v2_hashes[i] = piece_iter->block_hashes[i]; } view.modify(piece_iter, [](cached_piece_entry& e) { e.flags |= cached_piece_entry::hashing_flag; }); @@ -436,13 +468,26 @@ struct TORRENT_EXTRA_EXPORT disk_cache auto se = scope_end([&] { l.lock(); + // recount rather than relying on the pre-unlock snapshot: + // blocks flushed during the hashing window had their + // m_num_unhashed contribution removed by flush_piece_impl. + bool const is_v1 = bool(piece_iter->flags & cached_piece_entry::v1_hashes_flag); + int actual_unhashed = 0; + if (is_v1) + { + for (int i = hasher_cursor; i < blocks_in_piece; ++i) + { + auto const& cbe = piece_iter->blocks[i]; + if (cbe.has_buf() || cbe.get_write_job()) ++actual_unhashed; + } + } view.modify(piece_iter, [&](cached_piece_entry& e) { e.flags |= cached_piece_entry::force_flush_flag | cached_piece_entry::piece_hash_returned_flag; e.flags &= ~cached_piece_entry::hashing_flag; e.hasher_cursor = static_cast(blocks_in_piece); }); - TORRENT_ASSERT(m_num_unhashed >= num_unhashed); - m_num_unhashed -= num_unhashed; + TORRENT_ASSERT(m_num_unhashed >= actual_unhashed); + m_num_unhashed -= actual_unhashed; { TORRENT_ASSERT(m_allocator); bulk_free_buffer to_free(*m_allocator); @@ -462,7 +507,7 @@ struct TORRENT_EXTRA_EXPORT disk_cache view.erase(piece_iter); } }); - f(piece_iter->ph.get(), hasher_cursor, blocks, v2_hashes); + f(piece_iter->ph.get(), hasher_cursor, blocks); return hash_piece_result::completed; } @@ -511,6 +556,10 @@ struct TORRENT_EXTRA_EXPORT disk_cache int piece_size; bool v1; // piece requires SHA-1 hashing bool v2; // piece requires per-block SHA-256 hashing + // owning reference to the storage; used to keep it alive while v2 + // blocks for this piece sit in m_v2_hash_queue. May be null when v2 + // is false or when the caller (e.g. tests) doesn't need the queue. + std::shared_ptr storage; }; // the return value indicates whether the piece needs its hasher kicked or @@ -534,6 +583,14 @@ struct TORRENT_EXTRA_EXPORT disk_cache hash_result try_hash_piece(piece_location loc, disk_job* hash_job); + // Hangs hash_job on the piece if there are pending v2 hash queue + // entries for it (either still in m_v2_hash_queue or currently being + // hashed by a drain). drain_v2_hash_queue will post the job when the + // queue empties for the piece. Returns true if hung; false if there's + // nothing to wait for (piece gone, no pending entries) or another + // path already owns hash_job. + bool wait_for_v2_queue(piece_location loc, disk_job* hash_job); + // Called from a hasher thread when woken by interrupt. Processes all // pieces marked with needs_hasher_kick_flag, calling kick_hasher on each. // Returns true if force_flush_flag was set on any piece, meaning a generic @@ -542,6 +599,18 @@ struct TORRENT_EXTRA_EXPORT disk_cache // retry_jobs so the caller can re-submit them. bool kick_pending_hashers(jobqueue_t& completed_jobs, jobqueue_t& retry_jobs); + // Drains the queue: pops entries in batches, hashes each batch outside + // the cache lock, and forwards results via + // `store(storage, piece, block, hash)`. Hash jobs parked on a piece + // (via try_hash_piece's v2-pending defer branch or wait_for_v2_queue) + // are appended to `posted` once that piece's v2_pending reaches zero. + // `clear_piece_fun` is invoked for clear_piece jobs deferred for the + // same reason (see try_clear_piece). Returns when the queue is empty. + template + void drain_v2_hash_queue(Fun store, + jobqueue_t& posted, + std::function const& clear_piece_fun); + // this should be called by a disk thread // the callback should return the number of blocks it successfully flushed // to disk. The flush callback receives a snapshot of the pending write @@ -557,7 +626,6 @@ struct TORRENT_EXTRA_EXPORT disk_cache std::function clear_piece_fun); std::size_t size() const; - std::size_t num_flushing() const; std::tuple stats() const; @@ -579,6 +647,10 @@ private: void free_piece(cached_piece_entry const& cpe); + // remove every m_v2_hash_queue entry whose piece matches loc, releasing + // the buffers they own. Returns the count removed. Mutex must be held. + int drop_v2_queue_entries(piece_location loc); + // this requires the mutex to be locked void clear_piece_impl(cached_piece_entry& cpe, jobqueue_t& aborted); @@ -603,19 +675,131 @@ private: // while finishing hashing blocks. int m_blocks = 0; - // the number of blocks currently being flushed by a disk thread - // we use this to avoid over-shooting flushing blocks - int m_flushing_blocks = 0; - - // the number of blocks in the cache that have not yet been passed throught - // the piece hasher. i.e. where the hasher_cursor is <= the block index + // number of blocks in the cache belonging to v1 (or hybrid) pieces that + // have not yet been passed through the v1 piece hasher, i.e. where + // hasher_cursor <= block_idx. Pure-v2 pieces have no v1 hasher cursor + // concept and don't contribute to this counter. int m_num_unhashed = 0; // record disk_observers that we've signalled back-pressure to. Once the // cache size drop below the low watermark, we'll signal them they can resume back_pressure m_back_pressure; + + // FIFO queue of v2 block hashing work. Each entry owns its buffer (moved + // out of the originating write_job in insert()); the cbe reads through + // borrowed_buf while the entry is queued. Drained by drain_v2_hash_queue() + // from hasher threads: on success the buffer moves back into the cbe's + // write_job for the normal flush path; if the cbe is gone (clear_piece, + // flush completed) the buffer drops with the entry. cpe.v2_pending counts + // this piece's outstanding entries (queued + in flight inside a drain + // batch) so a clear_piece can wait for them. + std::deque m_v2_hash_queue; }; +template +void disk_cache::drain_v2_hash_queue(Fun store, + jobqueue_t& posted, + std::function const& clear_piece_fun) +{ + // Batched so the cache mutex is acquired once per batch_size entries + // instead of per entry. The bound also caps the stack footprint and + // lets other hasher threads grab their own batches in parallel. + constexpr int batch_size = 16; + std::vector batch; + batch.reserve(batch_size); + + for (;;) + { + batch.clear(); + { + std::unique_lock l(m_mutex); + INVARIANT_CHECK; + int const take = std::min(int(m_v2_hash_queue.size()), batch_size); + if (take == 0) return; + for (int i = 0; i < take; ++i) + { + batch.push_back(std::move(m_v2_hash_queue.front())); + m_v2_hash_queue.pop_front(); + } + } + + for (auto& entry : batch) + { + hasher256 h; + h.update({entry.buf.data(), entry.hash_len}); + store(entry.storage, entry.piece.piece, int(entry.block), h.final()); + } + + { + std::unique_lock l(m_mutex); + INVARIANT_CHECK; + // If the cbe still references our buffer, hand it back so the + // regular flush path can take it. Otherwise the block was already + // flushed (or the piece evicted) and the buffer just drops. + auto& view = m_pieces.template get<0>(); + // piece_iters is kept aligned with batch so the loop below + // (which gates on v2_pending == 0 after all decrements have + // landed) can reuse the lookups. v2_pending isn't an indexed + // key, so view.modify doesn't invalidate cached iterators. + // Every slot must be assigned exactly once before any + // `continue` -- a default-constructed multi_index iterator is + // singular and comparing it is undefined. + TORRENT_ALLOCA(piece_iters, + typename piece_container::nth_index<0>::type::iterator, + int(batch.size())); + int i = 0; + for (auto& entry : batch) + { + auto piece_iter = view.find(entry.piece); + piece_iters[i++] = piece_iter; + if (piece_iter == view.end()) continue; + view.modify(piece_iter, [](cached_piece_entry& e) { + TORRENT_ASSERT(e.v2_pending > 0); + --e.v2_pending; + }); + if (entry.block >= piece_iter->blocks_in_piece()) continue; + auto& cbe = piece_iter->blocks[entry.block]; + auto* j = cbe.get_write_job(); + if (j == nullptr) continue; + auto& wj = std::get(j->action); + // cbe may have been cleared and re-inserted with a new buffer + // while we were hashing; an address mismatch means this + // write_job isn't ours. + if (wj.borrowed_buf != entry.buf.data()) continue; + wj.buf = std::move(entry.buf); + ++m_blocks; + } + // post any hash_job parked on the cpe by try_hash_piece's + // v2-pending defer branch or wait_for_v2_queue, and finish any + // clear_piece job try_clear_piece (or flush_piece_impl's deferred- + // clear branch) parked on the cpe -- all gated on v2_pending == 0. + for (auto piece_iter : piece_iters) + { + if (piece_iter == view.end()) continue; + if (piece_iter->v2_pending > 0) continue; + if (piece_iter->hash_job != nullptr) + { + disk_job* j = nullptr; + view.modify(piece_iter, + [&](cached_piece_entry& e) { j = std::exchange(e.hash_job, nullptr); }); + posted.push_back(j); + } + if (piece_iter->clear_piece != nullptr) + { + disk_job* clear_piece = nullptr; + view.modify(piece_iter, [&](cached_piece_entry& e) { + clear_piece = std::exchange(e.clear_piece, nullptr); + }); + // clear_piece_impl already ran when the clear was parked; + // any aborted writes were dispatched at that point. We + // only need to post the clear job's completion now. + clear_piece_fun({}, clear_piece); + } + } + m_back_pressure.check_buffer_level(m_blocks + int(m_v2_hash_queue.size())); + } + } +} } #endif diff --git a/include/libtorrent/aux_/disk_job.hpp b/include/libtorrent/aux_/disk_job.hpp index b4b3df967..33446ec09 100644 --- a/include/libtorrent/aux_/disk_job.hpp +++ b/include/libtorrent/aux_/disk_job.hpp @@ -127,6 +127,12 @@ namespace job { // passed in/out // number of bytes 'buf' points to std::uint16_t buffer_size; + + // non-owning view of the bytes this write covers. Written once by + // disk_cache::insert() under the cache mutex; the flushing thread + // reads it without the mutex, so we need it to be stable across + // buffer moves between owners (wjob.buf <-> v2 hash queue entry). + char const* borrowed_buf = nullptr; }; // the hash jobs computes the SHA-1 hash of a whole piece. If the diff --git a/include/libtorrent/aux_/pread_storage.hpp b/include/libtorrent/aux_/pread_storage.hpp index fca1b26b4..7241d118e 100644 --- a/include/libtorrent/aux_/pread_storage.hpp +++ b/include/libtorrent/aux_/pread_storage.hpp @@ -25,6 +25,7 @@ see LICENSE file. #include "libtorrent/span.hpp" #include "libtorrent/aux_/vector.hpp" #include "libtorrent/aux_/open_mode.hpp" // for aux::open_mode_t +#include "libtorrent/aux_/precomputed_block_hashes.hpp" #include "libtorrent/disk_interface.hpp" // for disk_job_flags_t namespace libtorrent::aux { @@ -109,6 +110,18 @@ namespace libtorrent::aux { bool v1() const { return m_v1; } bool v2() const { return m_v2; } + // SHA-256 block hashes deposited by the v2 hash queue + // (disk_cache::drain_v2_hash_queue) for later consumption by hash and + // hash2 jobs, avoiding a read-back from disk. A block whose hash + // hasn't been computed reads as an all-zero hash (a SHA-256 over a + // non-empty block is never all zeros), so no separate "present" + // bitfield is needed. See m_precomputed_v2. + void store_precomputed_v2(piece_index_t piece, int block, sha256_hash const& h); + // the piece's block hashes (indexed by block); empty if none. + aux::vector take_precomputed_v2(piece_index_t piece); + std::optional take_precomputed_v2_block(piece_index_t piece, int block); + void drop_precomputed_v2(piece_index_t piece); + private: bool m_need_tick = false; @@ -180,6 +193,17 @@ namespace libtorrent::aux { // this is a v2 torrent. If both v1 and v2 are set, it's a hybrid // torrent bool m_v2; + + // SHA-256 block hashes deposited by the v2 hash queue + // (disk_cache::drain_v2_hash_queue) as each v2 block is hashed in + // memory. An entry is added the first time store_precomputed_v2() is + // called for a piece, and removed by take_precomputed_v2() (full-piece + // hash), take_precomputed_v2_block() (single-block hash2) or + // drop_precomputed_v2() (clear_piece). An entry for a piece that has + // blocks hashed but is never consumed by hash/hash2/clear lives until + // the storage is destroyed; it is therefore bounded by the number of + // in-flight pieces the piece picker keeps open. + precomputed_block_hashes m_precomputed_v2; }; } diff --git a/include/libtorrent/aux_/precomputed_block_hashes.hpp b/include/libtorrent/aux_/precomputed_block_hashes.hpp index 3d9b27115..267c60be7 100644 --- a/include/libtorrent/aux_/precomputed_block_hashes.hpp +++ b/include/libtorrent/aux_/precomputed_block_hashes.hpp @@ -21,9 +21,11 @@ see LICENSE file. namespace libtorrent::aux { - // A thread-safe cache of SHA-256 block hashes computed inline while writing - // blocks to disk, so a later hash/hash2 job doesn't have to re-read the block - // just to hash it. Keyed by piece, then indexed by block within the piece. + // A thread-safe cache of SHA-256 block hashes computed by the v2 hash + // queue (disk_cache::drain_v2_hash_queue) while the block buffer is still + // in memory, so a later hash/hash2 job doesn't have to re-read the block + // from disk just to hash it. Keyed by piece, then indexed by block within + // the piece. // // A block whose hash has not been computed reads as an all-zero hash. A // SHA-256 over a (non-empty) block is never all zeros, so this doubles as the diff --git a/src/disk_cache.cpp b/src/disk_cache.cpp index ca1f6ddd0..fdd936ac7 100644 --- a/src/disk_cache.cpp +++ b/src/disk_cache.cpp @@ -12,6 +12,10 @@ see LICENSE file. #include "libtorrent/aux_/debug_disk_thread.hpp" #include "libtorrent/bitfield.hpp" +#if TORRENT_USE_INVARIANT_CHECKS +#include +#endif + namespace libtorrent::aux { namespace mi = boost::multi_index; @@ -67,7 +71,7 @@ char const* cached_block_entry::data() const noexcept if (auto const* j = std::get_if(&write_state)) { TORRENT_ASSERT((*j)->get_type() == aux::job_action_t::write); - return std::get((*j)->action).buf.data(); + return std::get((*j)->action).borrowed_buf; } return nullptr; } @@ -84,7 +88,7 @@ span cached_block_entry::buf(int const block_size) const TORRENT_ASSERT((*j)->get_type() == aux::job_action_t::write); auto const& job = std::get((*j)->action); TORRENT_ASSERT(block_size == job.buffer_size); - return {job.buf.data(), job.buffer_size}; + return {job.borrowed_buf, job.buffer_size}; } return {nullptr, 0}; } @@ -95,7 +99,7 @@ span cached_block_entry::write_buf() const { TORRENT_ASSERT((*j)->get_type() == aux::job_action_t::write); auto const& job = std::get((*j)->action); - return {job.buf.data(), job.buffer_size}; + return {job.borrowed_buf, job.borrowed_buf ? job.buffer_size : std::uint16_t{0}}; } return {nullptr, 0}; } @@ -165,16 +169,17 @@ lt::hasher& piece_hasher::ctx() return *ctx; } -cached_piece_entry::cached_piece_entry(piece_location const& loc - , int const piece_size_v2, int const piece_size_arg, int const num_blocks, bool const v1, bool const v2) +cached_piece_entry::cached_piece_entry(piece_location const& loc, + int const piece_size_v2, + int const piece_size_arg, + int const num_blocks, + bool const v1) : piece(loc) , blocks(aux::make_unique(num_blocks)) - , block_hashes(v2 ? aux::make_unique(num_blocks) : aux::unique_ptr()) , ph(v1 ? std::make_unique() : nullptr) , piece_size2(piece_size_v2) , piece_size(piece_size_arg) - , flags((v1 ? v1_hashes_flag : cached_piece_flags{}) - | (v2 ? v2_hashes_flag : cached_piece_flags{})) + , flags(v1 ? v1_hashes_flag : cached_piece_flags{}) { TORRENT_ASSERT(piece_size_arg > 0); TORRENT_ASSERT(num_blocks == blocks_in_piece()); @@ -189,11 +194,14 @@ disk_cache::disk_cache(io_context& ios) : m_back_pressure(ios) {} -// If the specified piece exists in the cache, and it's unlocked, clear all -// write jobs (return them in "aborted"). Returns true if the clear_piece -// job should be posted as complete. Returns false if the piece is locked by -// another thread, and the clear_piece job has been queued to be issued once -// the piece is unlocked. +// If the specified piece exists in the cache and is not in use by another +// thread, clear it: abort all write jobs (returned in `aborted`), drop any +// v2 hash queue entries, and reset cursors/flags. Returns true when the +// caller can post the clear_piece job as complete now. Returns false if the +// clear must be deferred -- the clear_piece job is parked on the cpe and +// will be dispatched by whichever path eventually releases the gate that +// blocked us (flush_piece_impl's scope_end for flushing_flag, or +// drain_v2_hash_queue for v2_pending). bool disk_cache::try_clear_piece(piece_location const loc, disk_job* j, jobqueue_t& aborted) { std::unique_lock l(m_mutex); @@ -203,6 +211,11 @@ bool disk_cache::try_clear_piece(piece_location const loc, disk_job* j, jobqueue auto& view = m_pieces.template get<0>(); auto i = view.find(loc); if (i == view.end()) return true; + + // we clear a piece after it fails the hash check. It doesn't make sense + // to be hashing still + TORRENT_ASSERT(!(i->flags & cached_piece_entry::hashing_flag)); + if (i->flags & cached_piece_entry::flushing_flag) { // postpone the clearing until we're done flushing @@ -210,19 +223,23 @@ bool disk_cache::try_clear_piece(piece_location const loc, disk_job* j, jobqueue return false; } - // we clear a piece after it fails the hash check. It doesn't make sense - // to be hashing still - TORRENT_ASSERT(!(i->flags & cached_piece_entry::hashing_flag)); - if (i->flags & cached_piece_entry::hashing_flag) - { - // postpone the clearing until we're done hashing - view.modify(i, [&](cached_piece_entry& e) { e.clear_piece = j; }); - return false; - } - view.modify(i, [&](cached_piece_entry& e) { clear_piece_impl(e, aborted); }); + + // In-flight drain batches still own entries for this piece -- their + // store_precomputed_v2() calls land on pread_storage *after* we return. + // Defer the completion (and the drop_precomputed_v2 the caller will + // issue) until the drain decrements v2_pending to zero, so the late + // store doesn't race the drop. The caller's per-storage fence keeps new + // async_writes blocked across this wait; v2 hashing 16 KiB is much + // faster than the disk I/O that drove the request, so this branch is + // rarely taken. + if (i->v2_pending > 0) + { + view.modify(i, [&](cached_piece_entry& e) { e.clear_piece = j; }); + return false; + } return true; } @@ -251,8 +268,8 @@ insert_result_flags disk_cache::insert(piece_location const loc if (i == view.end()) { int const num_blocks = (params.piece_size + default_block_size - 1) / default_block_size; - i = m_pieces.emplace(loc, params.piece_size2 - , params.piece_size, num_blocks, params.v1, params.v2).first; + i = m_pieces.emplace(loc, params.piece_size2, params.piece_size, num_blocks, params.v1) + .first; } TORRENT_ASSERT(!(i->flags & cached_piece_entry::piece_hash_returned_flag)); @@ -271,17 +288,43 @@ insert_result_flags disk_cache::insert(piece_location const loc TORRENT_ASSERT(block_idx >= i->hasher_cursor); TORRENT_ASSERT(write_job->get_type() == aux::job_action_t::write); - TORRENT_ASSERT(std::get(write_job->action).buffer_size + auto& wjob = std::get(write_job->action); + TORRENT_ASSERT(wjob.buffer_size == std::min(default_block_size, params.piece_size - block_idx * default_block_size)); // All write jobs must use the same disk buffer allocator if (!m_allocator) - m_allocator = std::get(write_job->action).buf.m_allocator; + m_allocator = wjob.buf.m_allocator; else - TORRENT_ASSERT(m_allocator == std::get(write_job->action).buf.m_allocator); + TORRENT_ASSERT(m_allocator == wjob.buf.m_allocator); + + // captured while wjob.buf is in scope, before any move that may empty + // it. flush reads through this without the cache mutex (see write_buf()). + TORRENT_ASSERT(wjob.borrowed_buf == nullptr); + wjob.borrowed_buf = wjob.buf.data(); + + // move v2 blocks whose data falls inside piece_size2 onto the hash + // queue. The queue owns the buffer; the cache reaches the bytes through + // wjob.borrowed_buf. + if (params.v2) + { + int const block_offset = block_idx * default_block_size; + int const piece_size2 = params.piece_size2; + int const v2_len = std::min(int(wjob.buffer_size), piece_size2 - block_offset); + if (v2_len > 0) + { + m_v2_hash_queue.emplace_back(loc, + params.storage, + static_cast(block_idx), + static_cast(v2_len), + std::move(wjob.buf)); + view.modify(i, [](cached_piece_entry& e) { ++e.v2_pending; }); + } + } blk.write_state = write_job; - ++m_blocks; - ++m_num_unhashed; + // queue-owned buffers contribute to the level via m_v2_hash_queue.size(). + if (wjob.buf) ++m_blocks; + if (params.v1) ++m_num_unhashed; bool const effective_force_flush = force_flush || compute_force_flush(*i); view.modify(i, [effective_force_flush](cached_piece_entry& e) { @@ -291,11 +334,12 @@ insert_result_flags disk_cache::insert(piece_location const loc insert_result_flags ret{}; - if (m_back_pressure.has_back_pressure(m_blocks, std::move(o))) + if (m_back_pressure.has_back_pressure(m_blocks + int(m_v2_hash_queue.size()), std::move(o))) ret |= exceeded_limit; - if ((i->hasher_cursor == block_idx - || bool(i->flags & cached_piece_entry::v2_hashes_flag)) + // need_hasher_kick covers v1 hasher progress only; the caller wakes the + // hasher for v2 queue work itself (it knows storage->v2()). + if (params.v1 && i->hasher_cursor == block_idx && !(i->flags & cached_piece_entry::piece_hash_returned_flag) && !(i->flags & cached_piece_entry::needs_hasher_kick_flag)) { @@ -315,7 +359,7 @@ void disk_cache::set_max_size(int const max_size) std::optional disk_cache::flush_request() const { std::unique_lock l(m_mutex); - return m_back_pressure.should_flush(m_blocks); + return m_back_pressure.should_flush(m_blocks + int(m_v2_hash_queue.size())); } // this call can have 3 outcomes: @@ -338,30 +382,42 @@ disk_cache::hash_result disk_cache::try_hash_piece(piece_location const loc, dis if (!(i->flags & cached_piece_entry::hashing_flag) && i->hasher_cursor == i->blocks_in_piece()) { + // v1 hash is computed. If the v2 drain still has entries for + // this piece, async_hash's contract to deliver v1 + v2 hashes + // together means we can't post yet -- hang and let + // drain_v2_hash_queue post the job through the retry queue when + // v2_pending hits zero. do_job(hash) -> hash_piece scope_end + // will then set piece_hash_returned_flag and force_flush_flag. + if (i->v2_pending > 0) + { + TORRENT_ASSERT(i->hash_job == nullptr); + view.modify(i, [&](cached_piece_entry& e) { e.hash_job = hash_job; }); + return hash_result::job_queued; + } + view.modify(i, [&](cached_piece_entry& e) { // mark for flush so a generic thread will write any pending // write_jobs that are still in the cache for this piece - e.flags |= cached_piece_entry::piece_hash_returned_flag | cached_piece_entry::force_flush_flag; + e.flags |= + cached_piece_entry::piece_hash_returned_flag | cached_piece_entry::force_flush_flag; auto& job = std::get(hash_job->action); TORRENT_ASSERT(bool(e.ph) == bool(e.flags & cached_piece_entry::v1_hashes_flag)); job.piece_hash = e.ph ? e.ph->final_hash() : sha1_hash{}; - if (!job.block_hashes.empty()) - { - TORRENT_ASSERT(bool(i->flags & cached_piece_entry::v2_hashes_flag)); - TORRENT_ASSERT(e.block_hashes); - TORRENT_ASSERT(job.block_hashes.size() <= e.blocks_in_piece()); - int const to_copy = std::min(int(job.block_hashes.size()), int(e.blocks_in_piece())); - for (int idx = 0; idx < to_copy; ++idx) - job.block_hashes[idx] = e.block_hashes[idx]; - } }); return hash_result::job_completed; } - if ((i->flags & cached_piece_entry::hashing_flag) - && i->hasher_cursor < i->blocks_in_piece() - ) + // v2-only with v2 still draining: same defer, just reached through a + // different gate -- there's no v1 hasher cursor to wait on. + if (!(i->flags & cached_piece_entry::v1_hashes_flag) && i->v2_pending > 0) + { + TORRENT_ASSERT(i->hash_job == nullptr); + view.modify(i, [&](cached_piece_entry& e) { e.hash_job = hash_job; }); + return hash_result::job_queued; + } + + if ((i->flags & cached_piece_entry::hashing_flag) && i->hasher_cursor < i->blocks_in_piece()) { // We're not done hashing yet, let the hashing thread post the // completion once it's done @@ -376,6 +432,23 @@ disk_cache::hash_result disk_cache::try_hash_piece(piece_location const loc, dis return hash_result::post_job; } +bool disk_cache::wait_for_v2_queue(piece_location const loc, disk_job* hash_job) +{ + std::unique_lock l(m_mutex); + INVARIANT_CHECK; + + auto& view = m_pieces.template get<0>(); + auto pi = view.find(loc); + if (pi == view.end()) return false; + // hash_piece's deferred path will own hash_job in this case + if (pi->flags & cached_piece_entry::hashing_flag) return false; + if (pi->hash_job != nullptr) return false; + if (pi->v2_pending == 0) return false; + + view.modify(pi, [hash_job](cached_piece_entry& e) { e.hash_job = hash_job; }); + return true; +} + // this should be called from a hasher thread, with m_mutex held. // hashing_flag must already be set on the piece by the caller. // Returns true if force_flush_flag was set on the piece. @@ -395,8 +468,6 @@ bool disk_cache::kick_hasher(piece_container::nth_index<4>::type::iterator piece // allocating a slot for every block, just the ones in front of the cursor TORRENT_ALLOCA(blocks_storage, span, piece_iter->blocks_in_piece() - int(cursor)); - int count_hashed = 0; - // hashing_flag is already set by caller. We need to clear it before returning. TORRENT_ASSERT(piece_iter->flags & cached_piece_entry::hashing_flag); @@ -419,52 +490,25 @@ keep_going: while (end < piece_iter->blocks_in_piece() && blocks_storage[end - cursor].data()) ++end; - bool const need_v1 = bool(piece_iter->flags & cached_piece_entry::v1_hashes_flag); - bool const need_v2 = bool(piece_iter->flags & cached_piece_entry::v2_hashes_flag); + // insert() only sets needs_hasher_kick_flag for v1/hybrid pieces. + TORRENT_ASSERT(piece_iter->flags & cached_piece_entry::v1_hashes_flag); + TORRENT_ASSERT(piece_iter->ph); - DLOG("kick_hasher: piece: %d hashed_cursor: [%d, %d] v1: %d v2: %d ctx: %p\n" - , static_cast(piece_iter->piece.piece) - , cursor, end - , need_v1, need_v2 - , piece_iter->ph.get()); + DLOG("kick_hasher: piece: %d hashed_cursor: [%d, %d] ctx: %p\n", + static_cast(piece_iter->piece.piece), + cursor, + end, + piece_iter->ph.get()); l.unlock(); - std::uint16_t const cursor_start = cursor; - int bytes_left = piece_iter->piece_size2 - int(cursor_start) * default_block_size; - bool contiguous = true; - - // NOTE: i is in the blocks_storage space. It's offset by "cursor_start" to - // get back to block index - for (int i = 0; i < n; ++i, bytes_left -= default_block_size) + for (span const buf : blocks_storage) { - span const buf = blocks_storage[i]; - if (buf.data() == nullptr) - { - contiguous = false; - if (!need_v2) break; - continue; - } + if (buf.data() == nullptr) break; - if (contiguous) - { - if (need_v1) - { - TORRENT_ASSERT(piece_iter->ph); - auto& ctx = const_cast(*piece_iter->ph); - ctx.update(buf); - } - ++cursor; - ++count_hashed; - } - - if (need_v2) - { - TORRENT_ASSERT(piece_iter->block_hashes); - int const blk_idx = int(cursor_start) + i; - if (bytes_left > 0 && piece_iter->block_hashes[blk_idx].is_all_zeros()) - piece_iter->block_hashes[blk_idx] = hasher256(buf.first(std::min(bytes_left, default_block_size))).final(); - } + auto& ctx = const_cast(*piece_iter->ph); + ctx.update(buf); + ++cursor; } l.lock(); @@ -476,11 +520,19 @@ keep_going: goto keep_going; } - TORRENT_ASSERT(m_num_unhashed >= count_hashed); - m_num_unhashed -= count_hashed; + // recount rather than relying on (cursor - old_cursor): blocks flushed + // during the unlocked hashing window have already had their + // m_num_unhashed contribution removed by flush_piece_impl. + int const count = cursor - piece_iter->hasher_cursor; + { + int actual_hashed = 0; + for (auto const& cbe : piece_iter->get_blocks().subspan(piece_iter->hasher_cursor, count)) + if (cbe.has_buf() || cbe.get_write_job()) ++actual_hashed; + TORRENT_ASSERT(m_num_unhashed >= actual_hashed); + m_num_unhashed -= actual_hashed; + } // blocks that have been flushed and hashed can be removed from the cache immediately - int const count = cursor - piece_iter->hasher_cursor; TORRENT_ASSERT(m_allocator); bulk_free_buffer to_free(*m_allocator); for (auto& cbe : piece_iter->get_blocks().subspan(piece_iter->hasher_cursor, count)) @@ -492,7 +544,7 @@ keep_going: } TORRENT_ASSERT(l.owns_lock()); - m_back_pressure.check_buffer_level(m_blocks); + m_back_pressure.check_buffer_level(m_blocks + int(m_v2_hash_queue.size())); auto& view = m_pieces.template get<4>(); @@ -541,40 +593,50 @@ keep_going: } // there's a hash job hung on this piece, post it now - disk_job* j = nullptr; - - sha1_hash piece_hash; + disk_job* const j = piece_iter->hash_job; + TORRENT_ASSERT(j != nullptr); TORRENT_ASSERT(!(piece_iter->flags & cached_piece_entry::piece_hash_returned_flag)); + // peek at the hung job to decide between inline completion and a retry + // through the disk thread pool. A non-empty block_hashes array means + // the caller wants v2 block hashes too -- do_job(hash) will pull those + // from pread_storage::precomputed_block_hashes (populated by the v2 + // drain) with a disk fallback for any still missing. + auto& job = std::get(j->action); + bool const will_retry = !job.block_hashes.empty(); + + sha1_hash piece_hash; bool const force_flush = cursor == piece_iter->blocks_in_piece() || bool(piece_iter->flags & cached_piece_entry::piece_hash_returned_flag); - view.modify(piece_iter, [&j, &piece_hash, force_flush, cursor](cached_piece_entry& e) { - j = std::exchange(e.hash_job, nullptr); + view.modify(piece_iter, [&piece_hash, force_flush, cursor, will_retry](cached_piece_entry& e) { + e.hash_job = nullptr; if (force_flush) e.flags |= cached_piece_entry::force_flush_flag; e.hasher_cursor = cursor; e.flags &= ~cached_piece_entry::hashing_flag; - e.flags |= cached_piece_entry::piece_hash_returned_flag; - // we've hashed all blocks, and there's a hash job associated with - // this piece, post it. + // only mark the hash as returned when we're delivering it inline. + // In the retry case do_job(hash) -> hash_piece's scope_end sets + // the flag once the v2 block hashes are actually in the job. If we + // set it eagerly, flush_to_disk could evict the cpe between here + // and the retry, forcing do_job(hash) to read everything back + // from disk instead of reusing the cached state. + if (!will_retry) e.flags |= cached_piece_entry::piece_hash_returned_flag; TORRENT_ASSERT(bool(e.ph) == bool(e.flags & cached_piece_entry::v1_hashes_flag)); piece_hash = e.ph ? e.ph->final_hash() : sha1_hash{}; }); - auto& job = std::get(j->action); - if (need_v1) job.piece_hash = piece_hash; - if (!job.block_hashes.empty()) + job.piece_hash = piece_hash; + if (will_retry) { - TORRENT_ASSERT(need_v2); - TORRENT_ASSERT(piece_iter->block_hashes); - int const to_copy = std::min( - piece_iter->blocks_in_piece(), - int(job.block_hashes.size())); - for (int i = 0; i < to_copy; ++i) - job.block_hashes[i] = piece_iter->block_hashes[i]; + DLOG("kick_hasher: retrying attached v2 job piece: %d\n", + static_cast(piece_iter->piece.piece)); + retry_jobs.push_back(j); + } + else + { + DLOG("kick_hasher: posting attached job piece: %d\n", + static_cast(piece_iter->piece.piece)); + completed_jobs.push_back(j); } - DLOG("kick_hasher: posting attached job piece: %d\n" - , static_cast(piece_iter->piece.piece)); - completed_jobs.push_back(j); if (piece_iter->flags & cached_piece_entry::pending_free_flag) { @@ -643,8 +705,10 @@ Iter disk_cache::flush_piece_impl(View& view, // TODO: pass the block offset as a parameter instead of computing it like this int const block_offset = static_cast(blocks.data() - piece_iter->get_blocks().data()); - view.modify(piece_iter, [](cached_piece_entry& e) { TORRENT_ASSERT(!(e.flags & cached_piece_entry::flushing_flag)); e.flags |= cached_piece_entry::flushing_flag; }); - m_flushing_blocks += num_blocks; + view.modify(piece_iter, [](cached_piece_entry& e) { + TORRENT_ASSERT(!(e.flags & cached_piece_entry::flushing_flag)); + e.flags |= cached_piece_entry::flushing_flag; + }); // Snapshot the pending write_job pointer for each block while we still // hold the mutex. flushing_flag prevents other threads from flushing @@ -676,8 +740,6 @@ Iter disk_cache::flush_piece_impl(View& view, // until flush_storage clears it after waking up e.flags &= ~cached_piece_entry::flushing_flag; }); - TORRENT_ASSERT(m_flushing_blocks >= num_blocks); - m_flushing_blocks -= num_blocks; if (notify) m_flushing_cv.notify_all(); }); flushed_blocks.resize(int(blocks.size())); @@ -708,14 +770,14 @@ Iter disk_cache::flush_piece_impl(View& view, TORRENT_ASSERT(j); TORRENT_ASSERT(j->get_type() == aux::job_action_t::write); auto& job = std::get(j->action); + // borrowed_buf is write-once and stays valid for the buffer's + // lifetime; leave it alone for the unlocked completed-job path. disk_buffer_ref ref(std::move(job.buf)); - // if a hasher thread has captured a pointer into this buffer (it - // snapshotted blocks_storage before we re-acquired the lock), we - // must keep the buffer alive until hashing completes; kick_hasher - // will free it. In every other case — including write errors — - // release the buffer immediately. - bool const needed_by_hasher = (piece_iter->flags & cached_piece_entry::hashing_flag) + // ref empty means the v2 hash queue owns the buffer; the cache + // didn't count it in m_blocks and can't hold-alive for the hasher. + bool const needed_by_hasher = bool(ref) + && (piece_iter->flags & cached_piece_entry::hashing_flag) && block_index >= hasher_cursor; if (needed_by_hasher) { @@ -723,17 +785,22 @@ Iter disk_cache::flush_piece_impl(View& view, } else { + bool const cache_owned = bool(ref); to_free.add(std::move(ref)); // mark as flushed (with null buffer) so compute_flushed_cursor can // advance past this block even though the buffer has been freed blk.write_state = disk_buffer_ref{}; - if (block_index >= hasher_cursor) + if ((piece_iter->flags & cached_piece_entry::v1_hashes_flag) + && block_index >= hasher_cursor) { TORRENT_ASSERT(m_num_unhashed > 0); --m_num_unhashed; } - TORRENT_ASSERT(m_blocks > 0); - --m_blocks; + if (cache_owned) + { + TORRENT_ASSERT(m_blocks > 0); + --m_blocks; + } } ++jobs; @@ -759,14 +826,32 @@ Iter disk_cache::flush_piece_impl(View& view, disk_job* clear_piece = nullptr; view.modify(piece_iter, [&](cached_piece_entry& e) { clear_piece_impl(e, aborted); - clear_piece = std::exchange(e.clear_piece, nullptr); + // if a v2 drain still owes us a callback, leave e.clear_piece + // armed so drain_v2_hash_queue finishes the clear once + // v2_pending hits zero (the late store_precomputed_v2() must + // land before drop_precomputed_v2()). + if (e.v2_pending == 0) clear_piece = std::exchange(e.clear_piece, nullptr); }); - clear_piece_fun(std::move(aborted), clear_piece); + if (clear_piece != nullptr || !aborted.empty()) + clear_piece_fun(std::move(aborted), clear_piece); } return next_iter; } +int disk_cache::drop_v2_queue_entries(piece_location const loc) +{ + int dropped = 0; + auto new_end = std::remove_if( + m_v2_hash_queue.begin(), m_v2_hash_queue.end(), [&](v2_hash_queue_entry const& e) { + if (!(e.piece == loc)) return false; + ++dropped; + return true; + }); + m_v2_hash_queue.erase(new_end, m_v2_hash_queue.end()); + return dropped; +} + void disk_cache::free_piece(cached_piece_entry const& cpe) { #if TORRENT_USE_ASSERTS @@ -785,13 +870,22 @@ void disk_cache::free_piece(cached_piece_entry const& cpe) TORRENT_ASSERT(cpe.hash_job == nullptr); #endif TORRENT_ASSERT(m_allocator); + + // drop any queued v2 hash entries for this piece. The queue owns those + // buffers and they're released when the entries go out of scope. Any + // drain batch already in flight for this piece keeps its own buffer; on + // dispose the drain will look up the piece, find it gone, and drop the + // buffer naturally. + drop_v2_queue_entries(cpe.piece); + bulk_free_buffer to_free(*m_allocator); + bool const is_v1 = bool(cpe.flags & cached_piece_entry::v1_hashes_flag); int idx = 0; for (auto& blk : cpe.get_blocks()) { if (blk.has_buf()) { - if (idx >= cpe.hasher_cursor) + if (is_v1 && idx >= cpe.hasher_cursor) { TORRENT_ASSERT(m_num_unhashed > 0); --m_num_unhashed; @@ -825,7 +919,7 @@ void disk_cache::flush_to_disk(std::functionflushed_cursor == piece_iter->blocks_in_piece() && bool(piece_iter->flags & cached_piece_entry::piece_hash_returned_flag) && !(piece_iter->flags & cached_piece_entry::flushing_flag) - && !(piece_iter->flags & cached_piece_entry::notify_flushed_flag)) + && !(piece_iter->flags & cached_piece_entry::notify_flushed_flag) + && !(piece_iter->flags & cached_piece_entry::hashing_flag)) { - TORRENT_ASSERT(!(piece_iter->flags & cached_piece_entry::hashing_flag)); + // piece_hash_returned_flag is set at the same modify() that + // extracts hash_job (try_hash_piece's job_completed path, or + // kick_hasher / hash_piece scope_end), so flag-set implies + // hash_job == nullptr -- the cpe is safe to evict. free_piece(*piece_iter); view.erase(piece_iter); } @@ -878,10 +976,15 @@ void disk_cache::flush_to_disk(std::function(); for (auto piece_iter = view2.begin(); piece_iter != view2.end();) { - // We avoid flushing if other threads have already initiated sufficient - // amount of flushing - if (m_blocks - m_flushing_blocks <= target_blocks) - return; + // exit only on the actual level. Predicting the level after in-flight + // flushes finish (subtracting concurrent flushing count) creates a + // race where each thread can exit assuming the others will finish + // their work, but if the last thread also takes that shortcut, no + // one drives the level down to target and back-pressure stays on. + // Cheap flushing is the preferred path (no read-back later), so we + // want to exhaust it here rather than fall through to the expensive + // pass. + if (m_blocks + int(m_v2_hash_queue.size()) <= target_blocks) return; int const num_eligible_blocks = piece_iter->hasher_cursor - piece_iter->flushed_cursor; @@ -917,7 +1020,10 @@ void disk_cache::flush_to_disk(std::function(); for (auto piece_iter = view3.begin(); piece_iter != view3.end();) { - if (m_blocks - m_flushing_blocks <= target_blocks) return; + // safety net pass: exit only on the actual level. See the comment + // in the cheap pass above for why we don't subtract the concurrent + // flushing count from this check. + if (m_blocks + int(m_v2_hash_queue.size()) <= target_blocks) return; // skip pieces a hasher or another flush is currently using if (piece_iter->flags @@ -930,6 +1036,7 @@ void disk_cache::flush_to_disk(std::functionhasher_cursor; + bool const is_v1 = bool(piece_iter->flags & cached_piece_entry::v1_hashes_flag); span const blocks = piece_iter->get_blocks(); for (int i = 0; i < int(blocks.size()); ++i) { @@ -938,7 +1045,7 @@ void disk_cache::flush_to_disk(std::function 0); --m_blocks; - if (i >= hasher_cursor) + if (is_v1 && i >= hasher_cursor) { TORRENT_ASSERT(m_num_unhashed > 0); --m_num_unhashed; @@ -952,10 +1059,9 @@ void disk_cache::flush_to_disk(std::function(); for (auto piece_iter = view4.begin(); piece_iter != view4.end();) { - // We avoid flushing if other threads have already initiated sufficient - // amount of flushing - if (m_blocks - m_flushing_blocks <= target_blocks) - return; + // safety-net pass: exit only on the actual level. See the comment + // in pass 3. + if (m_blocks + int(m_v2_hash_queue.size()) <= target_blocks) return; if (piece_iter->flags & cached_piece_entry::flushing_flag) { @@ -1045,6 +1151,23 @@ void disk_cache::flush_storage(std::functionhash_job != nullptr) + { + jobqueue_t aborted; + view.modify(piece_iter, [&](cached_piece_entry& e) { + aborted.push_back(std::exchange(e.hash_job, nullptr)); + }); + clear_piece_fun(std::move(aborted), nullptr); + } + + // In-flight drain batches keep their own copy of the buffer; freeing + // the piece here just removes the cbe. When the drain finishes and + // looks up the piece it'll find it gone, store the precomputed hash + // on the (still-alive via shared_ptr) storage, and drop the buffer + // naturally. The stored hash is harmless: nobody will read it back. free_piece(*piece_iter); view.erase(piece_iter); } @@ -1054,21 +1177,14 @@ std::size_t disk_cache::size() const { std::unique_lock l(m_mutex); INVARIANT_CHECK; - return static_cast(m_blocks); -} - -std::size_t disk_cache::num_flushing() const -{ - std::unique_lock l(m_mutex); - INVARIANT_CHECK; - return static_cast(m_flushing_blocks); + return static_cast(m_blocks) + m_v2_hash_queue.size(); } std::tuple disk_cache::stats() const { std::unique_lock l(m_mutex); INVARIANT_CHECK; - return {m_blocks, m_num_unhashed}; + return {std::int64_t(m_blocks) + std::int64_t(m_v2_hash_queue.size()), m_num_unhashed}; } #if TORRENT_USE_INVARIANT_CHECKS @@ -1077,32 +1193,79 @@ void disk_cache::check_invariant() const // mutex must be held by caller int dirty_blocks = 0; int flushed_blocks = 0; - int flushing_blocks = 0; int unhashed_blocks = 0; + // v2_pending on each cpe counts that piece's blocks in m_v2_hash_queue + // plus any blocks held in a drain batch (popped from the queue but not + // yet decremented). We can directly observe only the queue half; we + // derive bounds against the cpe by counting queue entries per piece. + std::map queue_counts; + for (auto const& entry : m_v2_hash_queue) + ++queue_counts[entry.piece]; + + // every queue entry must reference a cpe in the cache: clear_piece_impl + // and free_piece drop matching entries when a cpe is removed, so a + // queue entry without a cpe would mean a leak. + { + auto const& view0 = m_pieces.template get<0>(); + for (auto const& entry : m_v2_hash_queue) + { + auto const it = view0.find(entry.piece); + TORRENT_ASSERT(it != view0.end()); + TORRENT_ASSERT(int(entry.block) < it->blocks_in_piece()); + } + } + + int total_v2_pending = 0; + auto& view = m_pieces.template get<2>(); for (auto const& piece_entry : view) { int const num_blocks = piece_entry.blocks_in_piece(); - - if (piece_entry.flags & cached_piece_entry::flushing_flag) - flushing_blocks += num_blocks; + bool const is_v1 = bool(piece_entry.flags & cached_piece_entry::v1_hashes_flag); span const blocks = piece_entry.get_blocks(); TORRENT_ASSERT(piece_entry.flushed_cursor <= num_blocks); TORRENT_ASSERT(piece_entry.hasher_cursor <= num_blocks); + // each block can contribute at most one v2 hash queue entry (insert() + // pushes once per write, and a block can only be written once per + // cycle), so v2_pending is bounded by blocks_in_piece. + TORRENT_ASSERT(int(piece_entry.v2_pending) <= num_blocks); + + // v2_pending >= (queue entries for this piece). The slack is the + // number of entries this piece has in flight inside drain batches. + { + auto const it = queue_counts.find(piece_entry.piece); + int const queued = (it != queue_counts.end()) ? it->second : 0; + TORRENT_ASSERT(queued <= int(piece_entry.v2_pending)); + } + total_v2_pending += int(piece_entry.v2_pending); + // pending_free_flag is set by flush_storage() when it wants to erase a // piece but hashing_flag is active. The hasher will erase it when done. // It must never outlive the hashing window. if (piece_entry.flags & cached_piece_entry::pending_free_flag) TORRENT_ASSERT(piece_entry.flags & cached_piece_entry::hashing_flag); + if (piece_entry.clear_piece != nullptr) + { + TORRENT_ASSERT(!(piece_entry.flags & cached_piece_entry::hashing_flag)); + } + int idx = 0; for (auto& be : blocks) { - if (be.get_write_job()) ++dirty_blocks; + if (auto* j = be.get_write_job()) + { + auto const& wj = std::get(j->action); + // If wj.buf is empty, the buffer for this block lives in a + // v2 hash queue entry (counted by queue_blocks below), not + // in this write job, so this slot does not hold a buffer + // in cache. + if (bool(wj.buf)) ++dirty_blocks; + } if (be.has_buf()) ++flushed_blocks; if (!(piece_entry.flags & cached_piece_entry::flushing_flag)) @@ -1132,7 +1295,7 @@ void disk_cache::check_invariant() const || piece_entry.hasher_cursor == piece_entry.blocks_in_piece()); } - if (idx >= piece_entry.hasher_cursor && (be.has_buf() || be.get_write_job())) + if (is_v1 && idx >= piece_entry.hasher_cursor && (be.has_buf() || be.get_write_job())) ++unhashed_blocks; ++idx; @@ -1142,8 +1305,12 @@ void disk_cache::check_invariant() const // are in flight. We just know the limit TORRENT_ASSERT(dirty_blocks <= m_blocks); TORRENT_ASSERT(dirty_blocks + flushed_blocks == m_blocks); - TORRENT_ASSERT(flushing_blocks >= m_flushing_blocks); TORRENT_ASSERT(unhashed_blocks == m_num_unhashed); + + // Sum of v2_pending across cpes equals (queue size) + (in-flight drain + // batches' size). Without drain activity Σ v2_pending == queue size; + // with a drain mid-execution the sum is strictly greater. + TORRENT_ASSERT(int(m_v2_hash_queue.size()) <= total_v2_pending); } #endif @@ -1153,24 +1320,46 @@ void disk_cache::clear_piece_impl(cached_piece_entry& cpe, jobqueue_t& aborted) INVARIANT_CHECK; TORRENT_ASSERT(!(cpe.flags & cached_piece_entry::flushing_flag)); TORRENT_ASSERT(!(cpe.flags & cached_piece_entry::hashing_flag)); + std::uint16_t jobs = 0; int const hasher_cursor = cpe.hasher_cursor; TORRENT_ASSERT(m_allocator); bulk_free_buffer to_free(*m_allocator); + + // hash_job may be hung via wait_for_v2_queue (hash request waiting for + // the hasher to finish). Abort it -- the caller is throwing the piece + // away. + if (cpe.hash_job) aborted.push_back(std::exchange(cpe.hash_job, nullptr)); + + // drop any queued v2 hash entries for this piece. The queue owns those + // buffers and they're released when the entries go out of scope. v2_pending + // is updated by the drain when it disposes in-flight entries, which is + // safe because those entries own their own buffer copy. + { + int const dropped = drop_v2_queue_entries(cpe.piece); + TORRENT_ASSERT(cpe.v2_pending >= dropped); + cpe.v2_pending -= static_cast(dropped); + } + + bool const is_v1 = bool(cpe.flags & cached_piece_entry::v1_hashes_flag); for (int idx = 0; idx < cpe.blocks_in_piece(); ++idx) { auto& cbe = cpe.blocks[idx]; - if (cbe.data() && idx >= hasher_cursor) + if (is_v1 && cbe.data() && idx >= hasher_cursor) { TORRENT_ASSERT(m_num_unhashed > 0); --m_num_unhashed; } - if (cbe.get_write_job()) + if (auto* j = cbe.get_write_job()) { + // only decrement m_blocks if the cache owned the buffer (i.e. + // wj.buf is non-empty). Queue-owned buffers were never counted. + auto const& job = std::get(j->action); + bool const cache_owned = bool(job.buf); aborted.push_back(cbe.take_write_job()); ++jobs; - --m_blocks; + if (cache_owned) --m_blocks; } if (cbe.has_buf()) @@ -1189,8 +1378,6 @@ void disk_cache::clear_piece_impl(cached_piece_entry& cpe, jobqueue_t& aborted) | cached_piece_entry::needs_hasher_kick_flag); cpe.hasher_cursor = 0; cpe.flushed_cursor = 0; - if (cpe.block_hashes) - std::fill_n(cpe.block_hashes.get(), cpe.blocks_in_piece(), sha256_hash{}); TORRENT_ASSERT(cpe.num_jobs >= jobs); cpe.num_jobs -= jobs; if (cpe.ph) *cpe.ph = piece_hasher{}; diff --git a/src/pread_disk_io.cpp b/src/pread_disk_io.cpp index 7375012bb..6c4c8c3fd 100644 --- a/src/pread_disk_io.cpp +++ b/src/pread_disk_io.cpp @@ -59,6 +59,16 @@ bool valid_flags(disk_job_flags_t const flags) } #endif +// Copy SHA-256 block hashes that the v2 hash queue precomputed for this +// piece into block_hashes, filling only slots that are still all-zero. +void fill_precomputed_v2( + span block_hashes, aux::vector const& pc, int const blocks_in_piece2) +{ + int const to_copy = std::min({int(pc.size()), int(block_hashes.size()), blocks_in_piece2}); + for (int i = 0; i < to_copy; ++i) + if (block_hashes[i].is_all_zeros()) block_hashes[i] = pc[i]; +} + template status_t translate_error(aux::disk_job* j, Fun f) { @@ -614,6 +624,30 @@ bool pread_disk_io::async_write(storage_index_t const storage, peer_request cons , disk_job_flags_t const flags) { TORRENT_ASSERT(valid_flags(flags)); + + auto storage_ptr = m_torrents[storage]->shared_from_this(); + + // a clear_piece (or other fence job) is in flight for this storage. For + // clear_piece the cpe is about to be reset; for storage-wide fences (e.g. + // release/delete_files) the storage is about to go away. Either way, + // inserting now would race with the teardown and could leave a block in + // the cache that the picker has just been told to re-pick. Abort the + // write so peer_connection sees operation_aborted and calls + // mark_as_canceled, which puts the block back in state_none ready to be + // re-requested after the fence lowers. + if (storage_ptr->has_fence()) + { + aux::pread_disk_job* j = m_job_pool.allocate_job(flags, + std::move(storage_ptr), + std::move(handler), + disk_buffer_holder{}, + r.piece, + r.start, + std::uint16_t(r.length)); + m_completed_jobs.abort_job(m_ios, j); + return false; + } + disk_buffer_holder buffer(m_buffer_pool, m_buffer_pool.allocate_buffer("receive buffer")); if (!buffer) aux::throw_ex(); std::memcpy(buffer.data(), buf, aux::numeric_cast(r.length)); @@ -621,15 +655,13 @@ bool pread_disk_io::async_write(storage_index_t const storage, peer_request cons TORRENT_ASSERT(r.start % default_block_size == 0); TORRENT_ASSERT(r.length <= default_block_size); - aux::pread_disk_job* j = m_job_pool.allocate_job( - flags, - m_torrents[storage]->shared_from_this(), + aux::pread_disk_job* j = m_job_pool.allocate_job(flags, + std::move(storage_ptr), std::move(handler), std::move(buffer), r.piece, r.start, - std::uint16_t(r.length) - ); + std::uint16_t(r.length)); DLOG("async_write: piece: %d offset: %d flags: %x\n" , int(r.piece), int(r.start) @@ -641,18 +673,41 @@ bool pread_disk_io::async_write(storage_index_t const storage, peer_request cons int const piece_size = j->storage->v1() ? fs.piece_size(r.piece) : fs.piece_size2(r.piece); TORRENT_ASSERT(r.length == std::min(piece_size - r.start, default_block_size)); aux::disk_cache::piece_entry_params const piece_params{ - fs.piece_size2(r.piece) - , piece_size - , j->storage->v1() - , j->storage->v2() - }; + fs.piece_size2(r.piece), piece_size, j->storage->v1(), j->storage->v2(), j->storage}; auto const result = m_cache.insert( {j->storage->storage_index(), r.piece} , r.start / default_block_size , force_flush, std::move(o), j, piece_params); - if (result & aux::disk_cache::need_hasher_kick) + // v1 wake-up signal comes from the cache; for v2 the insert may have + // pushed a queue entry the cache has no way to flag. + if ((result & aux::disk_cache::need_hasher_kick) || j->storage->v2()) + { m_hash_threads.interrupt(); + // no thread available to process the interrupt -- drain inline so + // the v2 hash queue can't grow without bound (and so destruction + // sees an empty cache). + if (m_hash_threads.max_threads() == 0) + { + jobqueue_t completed; + jobqueue_t retry; + m_cache.kick_pending_hashers(completed, retry); + m_cache.drain_v2_hash_queue( + [](std::shared_ptr const& st, + piece_index_t const piece, + int const block, + sha256_hash const& h) { + if (st) st->store_precomputed_v2(piece, block, h); + }, + retry, + [this](jobqueue_t aborted, aux::disk_job* clear) { + clear_piece_jobs(std::move(aborted), static_cast(clear)); + }); + if (!completed.empty()) add_completed_jobs(std::move(completed)); + while (!retry.empty()) + add_job(static_cast(retry.pop_front())); + } + } std::unique_lock l(m_job_mutex); if (!m_flush_target) @@ -708,6 +763,33 @@ void pread_disk_io::async_hash(storage_index_t const storage // immediately if (ret == aux::disk_cache::job_completed) { + // any v2 hash missing from storage means we have to fall back to + // do_job(hash), which can read the block from disk. + auto& a = std::get(j->action); + bool need_disk_fallback = false; + if (!a.block_hashes.empty()) + { + int const blocks_in_piece2 = j->storage->files().blocks_in_piece2(piece); + fill_precomputed_v2( + a.block_hashes, j->storage->take_precomputed_v2(piece), blocks_in_piece2); + + for (int i = 0; i < blocks_in_piece2; ++i) + { + if (i >= int(a.block_hashes.size()) || a.block_hashes[i].is_all_zeros()) + { + need_disk_fallback = true; + break; + } + } + } + + if (need_disk_fallback) + { + DLOG("async_hash: v2 hashes incomplete, falling back to do_job(hash)\n"); + add_job(j); + return; + } + // TODO: we may not need to do this, the cache could tell us // this piece should be flushed to disk now. m_generic_threads.interrupt(); @@ -880,25 +962,14 @@ void pread_disk_io::async_clear_piece(storage_index_t const storage ); DLOG("async_clear_piece: piece: %d\n", int(index)); - // regular jobs are not executed in-order. - // clear piece must wait for all write jobs issued to the piece finish - // before it completes. - jobqueue_t aborted_jobs; - bool const immediate_completion = m_cache.try_clear_piece( - {j->storage->storage_index(), index}, j, aborted_jobs); - - m_completed_jobs.abort_jobs(m_ios, std::move(aborted_jobs)); - if (immediate_completion) - { - DLOG("immediate clear\n"); - jobqueue_t jobs; - jobs.push_back(j); - add_completed_jobs(std::move(jobs)); - } - else - { - DLOG("deferred clear\n"); - } + // clear_piece is a fence job: raise the per-storage fence so any + // async_write issued from the network thread after this point is + // aborted instead of racing with the cpe reset (see async_write). + // The fence stays up until the clear job itself completes -- if + // try_clear_piece defers (flushing/hashing/v2_pending), do_job + // returns job_deferred and the fence is only lowered when the + // deferred path eventually posts the completion. + add_fence_job(j); } status_t pread_disk_io::do_job(aux::job::hash& a, aux::pread_disk_job* j) @@ -917,68 +988,98 @@ status_t pread_disk_io::do_job(aux::job::hash& a, aux::pread_disk_job* j) int const blocks_to_read = std::max(blocks_in_piece, blocks_in_piece2); - // Callable matching the hash_piece() protocol (see disk_cache.hpp). - // Completes piece hash computation using the cache snapshot passed by hash_piece(), - // reading any missing blocks from disk. - // * Copies all non-zero v2_hashes entries into a.block_hashes (capturing both - // in-order hashes and any out-of-order hashes computed by kick_hasher's second - // pass). - // * Iterates blocks [hasher_cursor, blocks_to_read): for each block, either reads - // it from disk (buf == nullptr) or hashes it from the in-memory buffer. Skips - // the SHA256 work for any block whose v2 hash is already pre-computed (v2_done). - // * Finalises the SHA1 hash into a.piece_hash. - // - // Also invoked directly as a fallback (all-null blocks, hasher_cursor=0) when the - // piece is not in the cache at all, in which case every block is read from disk. - auto hash_partial_piece = [&] (lt::aux::piece_hasher* ph - , int const hasher_cursor - , span const blocks - , span const v2_hashes) + // async_hash's fast path may have partially populated a.block_hashes + // already; preserve those entries (only fill all-zero slots). + if (v2) { - // ph: SHA1 hasher already fed blocks [0, hasher_cursor) - // hasher_cursor: first block index that still needs processing for SHA1. - // blocks: per-block buffer pointers from the cache snapshot. - // v2_hashes: SHA256 block hashes from the cache snapshot. + fill_precomputed_v2( + a.block_hashes, j->storage->take_precomputed_v2(a.piece), blocks_in_piece2); + + // If any v2 block is still missing, rendezvous with the hasher + // queue rather than reading the block back from disk. The block + // is sitting in m_v2_hash_queue waiting to be hashed; doing it + // here too would duplicate the hash work and leak the second copy + // into pread_storage::m_precomputed_v2 (nobody else will consume + // it). Only worth waiting when there's actually a hasher thread + // that can drain the queue. + if (m_hash_threads.max_threads() > 0) + { + bool any_missing = false; + for (int i = 0; i < blocks_in_piece2; ++i) + { + if (a.block_hashes[i].is_all_zeros()) + { + any_missing = true; + break; + } + } + if (any_missing && m_cache.wait_for_v2_queue({j->storage->storage_index(), a.piece}, j)) + { + m_hash_threads.interrupt(); + return disk_status::job_deferred; + } + } + } + + // hash_piece() callback. Also reused as a fallback (all-null blocks, + // hasher_cursor=0) for pieces not in the cache. + auto hash_partial_piece = [&](lt::aux::piece_hasher* ph, + int const hasher_cursor, + span const blocks) { time_point const start_time = clock_type::now(); - // copy all pre-computed v2 hashes, including out-of-order ones + // v1 hashing is contiguous from hasher_cursor; v2 may need any + // earlier block that's still missing its hash. + int start = hasher_cursor; if (v2) { for (int i = 0; i < blocks_in_piece2; ++i) - if (!v2_hashes[i].is_all_zeros()) - a.block_hashes[i] = v2_hashes[i]; + { + if (a.block_hashes[i].is_all_zeros()) + { + start = std::min(start, i); + break; + } + } } - int offset = hasher_cursor * default_block_size; + int offset = start * default_block_size; int blocks_read_from_disk = 0; - for (int i = hasher_cursor; i < blocks_to_read; ++i) + for (int i = start; i < blocks_to_read; ++i) { bool const v2_block = i < blocks_in_piece2; - bool const v2_done = v2_block && !v2_hashes[i].is_all_zeros(); + bool const v2_done = v2_block && !a.block_hashes[i].is_all_zeros(); + bool const v1_done = i < hasher_cursor; - std::ptrdiff_t const len = v1 ? std::min(default_block_size, piece_size - offset) : 0; + std::ptrdiff_t const len = + (v1 && !v1_done) ? std::min(default_block_size, piece_size - offset) : 0; std::ptrdiff_t const len2 = (v2_block && !v2_done) ? std::min(default_block_size, piece_size2 - offset) : 0; + if (len == 0 && len2 == 0) + { + offset += default_block_size; + continue; + } + hasher256 ph2; - char const* buf = blocks[i]; + char const* buf = (i < int(blocks.size())) ? blocks[i] : nullptr; if (buf == nullptr) { DLOG("do_hash: reading (piece: %d block: %d)\n", int(a.piece), i); j->error.ec.clear(); - if (v1) + if (len > 0) { TORRENT_ASSERT(ph); - auto const flags = v2_block - ? (j->flags & ~disk_interface::flush_piece) - : j->flags; + auto const flags = + v2_block && !v2_done ? (j->flags & ~disk_interface::flush_piece) : j->flags; j->storage->hash(m_settings, ph->ctx(), len, a.piece , offset, file_mode, flags, j->error); ++blocks_read_from_disk; } - if (v2_block && !v2_done) + if (len2 > 0) { j->storage->hash2(m_settings, ph2, len2, a.piece, offset , file_mode, j->flags, j->error); @@ -988,18 +1089,16 @@ status_t pread_disk_io::do_job(aux::job::hash& a, aux::pread_disk_job* j) } else { - if (v1) + if (len > 0) { TORRENT_ASSERT(ph); ph->update({ buf, len }); } - if (v2_block && !v2_done) - ph2.update({buf, len2}); + if (len2 > 0) ph2.update({buf, len2}); } offset += default_block_size; - if (v2_block && !v2_done) - a.block_hashes[i] = ph2.final(); + if (len2 > 0) a.block_hashes[i] = ph2.final(); } if (v1) @@ -1030,10 +1129,9 @@ status_t pread_disk_io::do_job(aux::job::hash& a, aux::pread_disk_job* j) // fall back to reading everything from disk TORRENT_ALLOCA(blocks, char const*, blocks_to_read); - TORRENT_ALLOCA(v2_hashes, sha256_hash, v2 ? blocks_in_piece2 : 0); for (char const*& b : blocks) b = nullptr; lt::aux::piece_hasher ph_storage; - hash_partial_piece(v1 ? &ph_storage : nullptr, 0, blocks, v2_hashes); + hash_partial_piece(v1 ? &ph_storage : nullptr, 0, blocks); } return j->error ? disk_status::fatal_disk_error : status_t{}; @@ -1051,10 +1149,20 @@ status_t pread_disk_io::do_job(aux::job::hash2& a, aux::pread_disk_job* j) TORRENT_ASSERT(piece_size > a.offset); std::ptrdiff_t const len = std::min(default_block_size, piece_size - a.offset); + // fast path: SHA-256 was already computed by drain_v2_hash_queue() while + // the block was held in the v2 hash queue, and stashed on the storage. + int const blk = a.offset / default_block_size; + if (auto pre = j->storage->take_precomputed_v2_block(a.piece, blk)) + { + a.piece_hash2 = *pre; + std::int64_t const read_time = total_microseconds(clock_type::now() - start_time); + m_stats_counters.inc_stats_counter(counters::disk_hash_time, read_time); + m_stats_counters.inc_stats_counter(counters::disk_job_time, read_time); + return {}; + } + int ret = 0; - a.piece_hash2 = m_cache.hash2({ j->storage->storage_index(), a.piece } - , a.offset / default_block_size - , [&] { + a.piece_hash2 = m_cache.hash2({j->storage->storage_index(), a.piece}, blk, [&] { hasher256 h; ret = j->storage->hash2(m_settings, h, len, a.piece, a.offset , file_mode, j->flags, j->error); @@ -1218,9 +1326,34 @@ status_t pread_disk_io::do_job(aux::job::file_priority& a, aux::pread_disk_job* return status_t{}; } -status_t pread_disk_io::do_job(aux::job::clear_piece&, aux::pread_disk_job*) +status_t pread_disk_io::do_job(aux::job::clear_piece& a, aux::pread_disk_job* j) { - TORRENT_ASSERT_FAIL(); + // raise_fence ensured the previous outstanding jobs for this storage all + // completed before we got here, and the fence keeps new ones blocked. + // async_write checks has_fence() and aborts itself, so no fresh writes + // can race with the cpe reset either. + jobqueue_t aborted; + bool const immediate = + m_cache.try_clear_piece({j->storage->storage_index(), a.piece}, j, aborted); + + if (!aborted.empty()) m_completed_jobs.abort_jobs(m_ios, std::move(aborted)); + + if (!immediate) + { + // try_clear_piece parked j on the cpe waiting for flushing_flag or + // v2_pending to clear. The deferred path will dispatch the + // completion via clear_piece_jobs -> add_completed_jobs, which runs + // job_complete and lowers the fence. + DLOG("do_job(clear_piece): piece: %d deferred\n", int(a.piece)); + return disk_status::job_deferred; + } + + DLOG("do_job(clear_piece): piece: %d immediate\n", int(a.piece)); + // In the hash-failure caller take_precomputed_v2() in async_hash's path + // has already drained the map and this is a no-op; in the write-failure + // caller the v2 drain typically completed before the write error + // surfaced and the entries it stored are removed here. + j->storage->drop_precomputed_v2(a.piece); return {}; } @@ -1469,17 +1602,26 @@ int pread_disk_io::flush_cache_blocks( void pread_disk_io::clear_piece_jobs(jobqueue_t aborted, aux::pread_disk_job* clear) { m_completed_jobs.abort_jobs(m_ios, std::move(aborted)); - jobqueue_t jobs; - jobs.push_back(clear); - add_completed_jobs(std::move(jobs)); + // drop any precomputed v2 hashes for the cleared piece. We hold off on + // this drop until m_cache has confirmed v2_pending == 0 for the piece; + // otherwise an in-flight drain batch could call store_precomputed_v2() + // after we drop, leaving a stale hash to be consumed by a later + // hash/hash2 on the re-downloaded piece. + if (clear) + { + auto const& a = std::get(clear->action); + clear->storage->drop_precomputed_v2(a.piece); + jobqueue_t jobs; + jobs.push_back(clear); + add_completed_jobs(std::move(jobs)); + } } void pread_disk_io::try_flush_cache(int const target_cache_size , bool const optimistic , std::unique_lock& l) { - DLOG("flushing, cache target: %d (current size: %d currently flushing: %d)\n" - , target_cache_size, m_cache.size(), m_cache.num_flushing()); + DLOG("flushing, cache target: %d (current size: %d)\n", target_cache_size, m_cache.size()); l.unlock(); jobqueue_t completed_jobs; m_cache.flush_to_disk( @@ -1565,6 +1707,18 @@ void pread_disk_io::thread_fun(aux::disk_io_thread_pool& pool jobqueue_t completed; jobqueue_t retry; bool const needs_flush = m_cache.kick_pending_hashers(completed, retry); + + m_cache.drain_v2_hash_queue( + [](std::shared_ptr const& st, + piece_index_t const piece, + int const block, + sha256_hash const& h) { + if (st) st->store_precomputed_v2(piece, block, h); + }, + retry, + [this](jobqueue_t aborted, aux::disk_job* clear) { + clear_piece_jobs(std::move(aborted), static_cast(clear)); + }); add_completed_jobs(std::move(completed)); l.lock(); diff --git a/src/pread_storage.cpp b/src/pread_storage.cpp index e1716f89a..c6e7fa977 100644 --- a/src/pread_storage.cpp +++ b/src/pread_storage.cpp @@ -72,6 +72,30 @@ namespace libtorrent::aux { return {m_files, m_renamed_files}; } + // these forward to the precomputed_block_hashes data structure, adding the + // storage's knowledge of the per-piece v2 block count. + void pread_storage::store_precomputed_v2( + piece_index_t const piece, int const block, sha256_hash const& h) + { + m_precomputed_v2.store(piece, block, files().blocks_in_piece2(piece), h); + } + + aux::vector pread_storage::take_precomputed_v2(piece_index_t const piece) + { + return m_precomputed_v2.take(piece); + } + + std::optional pread_storage::take_precomputed_v2_block( + piece_index_t const piece, int const block) + { + return m_precomputed_v2.take_block(piece, block); + } + + void pread_storage::drop_precomputed_v2(piece_index_t const piece) + { + m_precomputed_v2.drop(piece); + } + void pread_storage::need_partfile() { if (m_part_file) return; diff --git a/test/disk_cache_test_utils.hpp b/test/disk_cache_test_utils.hpp index dd05b15b8..7a6ae72de 100644 --- a/test/disk_cache_test_utils.hpp +++ b/test/disk_cache_test_utils.hpp @@ -20,8 +20,10 @@ see LICENSE file. #include "libtorrent/io_context.hpp" #include "libtorrent/bitfield.hpp" #include "libtorrent/span.hpp" +#include "libtorrent/sha1_hash.hpp" #include "libtorrent/units.hpp" #include +#include #include #include @@ -88,17 +90,17 @@ struct cache_fixture lt::aux::disk_cache::piece_entry_params piece_params() const { return { - piece_size2, - piece_size, - bool(mode & test_mode::v1), bool(mode & test_mode::v2) - }; + piece_size2, piece_size, bool(mode & test_mode::v1), bool(mode & test_mode::v2), {}}; } // Allocate a write-job whose buffer is filled with fill_char. // The buffer is sized to the actual block size (which may be less than // default_block_size for the last block of a piece with unaligned piece_size). // Ownership is transferred to live_jobs; the raw pointer is returned. - // write_job->storage is intentionally null: disk_cache no longer needs it. + // write_job->storage is left null; disk_cache uses it only to keep the + // storage alive while a v2 hash queue entry is outstanding, and the test + // harness instead captures the v2 hashes directly in v2_hashes (see + // kick_hashers()). lt::aux::pread_disk_job* make_write_job( lt::piece_index_t const piece , int const block @@ -168,10 +170,28 @@ struct cache_fixture } // Advance the hasher for all pending pieces, discarding completed jobs. + // Also drains the v2 hash queue, capturing computed SHA-256s in v2_hashes + // (indexed by piece, then block) so tests can verify them without a real + // pread_storage. + std::map, lt::sha256_hash> v2_hashes; + // captured by the kick_hashers() clear-piece callback so tests can + // observe deferred clear_piece completions dispatched by the drain. + std::vector cleared_jobs; + std::vector aborted_jobs; void kick_hashers() { lt::jobqueue_t completed, retry; cache.kick_pending_hashers(completed, retry); + cache.drain_v2_hash_queue([this](std::shared_ptr const&, + lt::piece_index_t const piece, + int const block, + lt::sha256_hash const& h) { v2_hashes[{piece, block}] = h; }, + retry, + [this](lt::jobqueue_t aborted, lt::aux::disk_job* clear) { + cleared_jobs.push_back(clear); + while (!aborted.empty()) + aborted_jobs.push_back(aborted.pop_front()); + }); } // Simulate flush_storage(): marks every block with a write_job as flushed. diff --git a/test/test_disk_cache.cpp b/test/test_disk_cache.cpp index ae5f59c6a..c701d7b8a 100644 --- a/test/test_disk_cache.cpp +++ b/test/test_disk_cache.cpp @@ -182,17 +182,25 @@ void test_disk_bottleneck(test_mode_t const mode) cache_fixture f(2, mode); auto r0 = f.insert(0_piece, 0); - TEST_CHECK(bool(r0 & disk_cache::need_hasher_kick)); + // need_hasher_kick is set for v1/hybrid pieces when the v1 piece hasher + // can advance (this insert filled hasher_cursor's slot). Pure-v2 pieces + // don't drive the cache hasher; pread_disk_io wakes the hasher + // independently for v2 storages. + if (mode & test_mode::v1) + TEST_CHECK(bool(r0 & disk_cache::need_hasher_kick)); + else + TEST_CHECK(!(r0 & disk_cache::need_hasher_kick)); auto r1 = f.insert(0_piece, 1); - // needs_hasher_kick_flag already set from block 0 — not raised again. + // needs_hasher_kick_flag already set from block 0 (v1/hybrid case) or + // never set (pure v2 case) — either way not raised here. TEST_CHECK(!(r1 & disk_cache::need_hasher_kick)); TEST_EQUAL(int(f.cache.size()), 2); - jobqueue_t completed, retry; - TEST_CHECK(f.cache.kick_pending_hashers(completed, retry)); - TEST_CHECK(completed.empty()); + // kick_hashers() advances v1 and also drains the v2 hash queue, + // capturing computed v2 hashes in f.v2_hashes. + f.kick_hashers(); std::vector block_hashes(mode & test_mode::v2 ? 2 : 0); auto hash_job = std::make_unique(); @@ -202,12 +210,30 @@ void test_disk_bottleneck(test_mode_t const mode) sha1_hash{} }; - TEST_EQUAL(f.cache.try_hash_piece(f.loc(0_piece), hash_job.get()) - , disk_cache::hash_result::job_completed); + auto const hpr = f.cache.try_hash_piece(f.loc(0_piece), hash_job.get()); if (mode & test_mode::v1) + { + // v1 (or hybrid) -- the piece hasher advanced to blocks_in_piece in + // kick_hashers(), so the piece hash is ready. + TEST_EQUAL(hpr, disk_cache::hash_result::job_completed); TEST_CHECK(!std::get(hash_job->action).piece_hash.is_all_zeros()); - for (auto const& h : block_hashes) - TEST_CHECK(!h.is_all_zeros()); + } + else + { + // Pure-v2 -- no v1 piece hasher, so hasher_cursor never reaches + // blocks_in_piece and try_hash_piece returns post_job. The actual + // path in production is pread_disk_io::do_job(hash), which pulls + // the v2 hashes from precomputed_block_hashes. + TEST_EQUAL(hpr, disk_cache::hash_result::post_job); + } + if (mode & test_mode::v2) + { + // v2 block hashes live on the storage's precomputed_block_hashes + // (captured into f.v2_hashes by the test fixture's kick_hashers()). + TEST_EQUAL(int(f.v2_hashes.size()), 2); + TEST_CHECK(!f.v2_hashes.at({0_piece, 0}).is_all_zeros()); + TEST_CHECK(!f.v2_hashes.at({0_piece, 1}).is_all_zeros()); + } TEST_EQUAL(f.flush(), 2); TEST_EQUAL(int(f.cache.size()), 0); @@ -230,12 +256,18 @@ void test_hashing_bottleneck(test_mode_t const mode) sha1_hash{} }; - TEST_EQUAL(f.cache.try_hash_piece(f.loc(0_piece), hash_job.get()) - , disk_cache::hash_result::post_job); + // v2-only with v2_pending > 0: try_hash_piece hangs the hash_job on the + // cpe so drain_v2_hash_queue can post it once the queue drains, and + // clear_piece_impl extracts the hung hash_job alongside block 1's + // write_job. v1/hybrid don't take that gate -- kick_hasher hasn't run + // so the v1_done check sends them down post_job. + bool const v2_only = (mode & test_mode::v2) && !(mode & test_mode::v1); + TEST_EQUAL(f.cache.try_hash_piece(f.loc(0_piece), hash_job.get()), + v2_only ? disk_cache::hash_result::job_queued : disk_cache::hash_result::post_job); jobqueue_t aborted; TEST_CHECK(f.cache.try_clear_piece(f.loc(0_piece), nullptr, aborted)); - TEST_EQUAL(aborted.size(), 1); + TEST_EQUAL(aborted.size(), v2_only ? 2 : 1); TEST_EQUAL(int(f.cache.size()), 0); } @@ -249,8 +281,7 @@ void test_multi_piece(test_mode_t const mode) f.insert(2_piece, 0); TEST_EQUAL(int(f.cache.size()), 3); - jobqueue_t completed, retry; - f.cache.kick_pending_hashers(completed, retry); + f.kick_hashers(); for (auto p : {0_piece, 1_piece, 2_piece}) { @@ -261,12 +292,25 @@ void test_multi_piece(test_mode_t const mode) (mode & test_mode::v2) ? span{&bh, 1} : span{}, sha1_hash{} }; - TEST_EQUAL(f.cache.try_hash_piece(f.loc(p), hash_job.get()) - , disk_cache::hash_result::job_completed); + auto const hpr = f.cache.try_hash_piece(f.loc(p), hash_job.get()); if (mode & test_mode::v1) + { + TEST_EQUAL(hpr, disk_cache::hash_result::job_completed); TEST_CHECK(!std::get(hash_job->action).piece_hash.is_all_zeros()); + } + else + { + // Pure-v2: hasher_cursor doesn't advance, so try_hash_piece + // returns post_job (handled by pread_disk_io::do_job(hash) in + // production). + TEST_EQUAL(hpr, disk_cache::hash_result::post_job); + } if (mode & test_mode::v2) - TEST_CHECK(!bh.is_all_zeros()); + { + // v2 hashes are stashed on storage via drain_v2_hash_queue; + // the test fixture captured them in v2_hashes. + TEST_CHECK(!f.v2_hashes.at({p, 0}).is_all_zeros()); + } } TEST_EQUAL(f.flush(), 3); @@ -295,7 +339,7 @@ TORRENT_TEST(v2_hashing_bottleneck) cache_fixture f(1, test_mode::v2); f.insert(0_piece, 0); - // Flush before the hasher runs — block_hashes[0] stays all-zeros. + // Flush before the hasher runs -- block_hashes[0] stays all-zeros. TEST_EQUAL(f.flush(0), 1); // The precomputed hash is absent and the buffer is gone; fallback fires. @@ -417,6 +461,166 @@ TORRENT_TEST(clear_piece_partially_flushed) TEST_EQUAL(aborted.size(), 1); // only block 1's write_job } +namespace { + + // Pure-v2 / hybrid variant of clear_piece_v1: verifies that + // clear_piece_impl's m_v2_hash_queue dedup loop drops every queued entry + // for the piece and decrements v2_pending to zero. Both inserts left their + // buffers in m_v2_hash_queue, so size() before clear is queue-driven. + void test_clear_piece_queued_v2(test_mode_t const mode) + { + cache_fixture f(2, mode); + + f.insert(0_piece, 0); + f.insert(0_piece, 1); + TEST_EQUAL(int(f.cache.size()), 2); // both entries in m_v2_hash_queue + + jobqueue_t aborted; + TEST_CHECK(f.cache.try_clear_piece(f.loc(0_piece), nullptr, aborted)); + TEST_EQUAL(int(f.cache.size()), 0); // queue dedup'd both entries + TEST_EQUAL(aborted.size(), 2); // both write_jobs aborted + TEST_EQUAL(f.alloc.live, 0); + } + +} + +TORRENT_TEST(clear_piece_queued_v2) { test_clear_piece_queued_v2(test_mode::v2); } +TORRENT_TEST(clear_piece_queued_hybrid) +{ + test_clear_piece_queued_v2(test_mode::v1 | test_mode::v2); +} + +namespace { + + // Branch 2: clear_piece arrives while flushing_flag is set. The flush + // callback runs with the cache mutex released; try_clear_piece must + // park the clear_piece job on the cpe and return false. flush_piece_impl's + // post-scope_end block then dispatches the parked clear via + // clear_piece_fun once flushing_flag is cleared. + void test_clear_piece_during_flush(test_mode_t const mode) + { + cache_fixture f(2, mode); + f.insert(0_piece, 0); + f.insert(0_piece, 1); + + auto clear_job = std::make_unique(); + clear_job->action = job::clear_piece{{}, 0_piece}; + + jobqueue_t cb_aborted; + bool cb_immediate = true; + int callback_calls = 0; + + jobqueue_t deferred_aborted; + disk_job* deferred_clear = nullptr; + + f.cache.flush_to_disk( + [&](bitfield&, span) -> int { + ++callback_calls; + // cache mutex is released while we're in here; the cpe has + // flushing_flag set, so try_clear_piece parks our clear_job + // rather than running clear_piece_impl inline. + cb_immediate = f.cache.try_clear_piece(f.loc(0_piece), clear_job.get(), cb_aborted); + // Don't flush anything: leave the write_jobs to be aborted by + // the deferred clear. + return 0; + }, + 0, // target=0, optimistic=false -> reaches the expensive pass + [&](jobqueue_t aborted, disk_job* clear) { + while (!aborted.empty()) + deferred_aborted.push_back(aborted.pop_front()); + if (clear) deferred_clear = clear; + }, + false); + + TEST_EQUAL(callback_calls, 1); + TEST_CHECK(!cb_immediate); // try_clear_piece deferred + TEST_CHECK(cb_aborted.empty()); // nothing returned through the inline path + TEST_EQUAL(deferred_aborted.size(), 2); // both write_jobs aborted via clear_piece_fun + TEST_EQUAL(deferred_clear, clear_job.get()); + TEST_EQUAL(int(f.cache.size()), 0); + // alloc.live is mode-dependent here: v1 keeps the buffers attached to + // the aborted write_jobs (released only when the caller processes them); + // v2 moved the buffers into m_v2_hash_queue at insert time and clear_piece_impl's + // queue dedup drops them. So we don't assert on it. + } + +} + +TORRENT_TEST(clear_piece_during_flush_v1) { test_clear_piece_during_flush(test_mode::v1); } +TORRENT_TEST(clear_piece_during_flush_v2) { test_clear_piece_during_flush(test_mode::v2); } +TORRENT_TEST(clear_piece_during_flush_hybrid) +{ + test_clear_piece_during_flush(test_mode::v1 | test_mode::v2); +} + +namespace { + + // Branch 5: clear_piece arrives while a drain batch holds the only + // outstanding v2_pending decrement. clear_piece_impl drops queued entries + // (none -- already popped by drain) without decrementing v2_pending; the + // post-clear_piece_impl check in try_clear_piece sees v2_pending > 0 and + // parks the clear_job. When the drain re-acquires the lock and decrements + // v2_pending to zero, its second batch loop dispatches the parked clear + // via clear_piece_fun. + void test_clear_piece_v2_drain_in_flight(test_mode_t const mode) + { + cache_fixture f(1, mode); + f.insert(0_piece, 0); + TEST_EQUAL(int(f.cache.size()), 1); // one queue entry + + auto clear_job = std::make_unique(); + clear_job->action = job::clear_piece{{}, 0_piece}; + + jobqueue_t cb_aborted; + bool cb_immediate = true; + int store_calls = 0; + + jobqueue_t deferred_aborted; + disk_job* deferred_clear = nullptr; + + jobqueue_t posted; + f.cache.drain_v2_hash_queue( + [&](std::shared_ptr const&, + piece_index_t, + int, + sha256_hash const&) { + ++store_calls; + // We're inside the unlocked window of drain_v2_hash_queue: + // the entry has been popped from m_v2_hash_queue but + // v2_pending has not yet been decremented. try_clear_piece + // must defer -- v2_pending stays > 0 until drain re-acquires + // the lock. + cb_immediate = f.cache.try_clear_piece(f.loc(0_piece), clear_job.get(), cb_aborted); + }, + posted, + [&](jobqueue_t aborted, disk_job* clear) { + while (!aborted.empty()) + deferred_aborted.push_back(aborted.pop_front()); + if (clear) deferred_clear = clear; + }); + + TEST_EQUAL(store_calls, 1); + TEST_CHECK(!cb_immediate); + // clear_piece_impl ran inside try_clear_piece, so the write_job came + // back through the inline aborted queue, not the deferred one. + TEST_EQUAL(cb_aborted.size(), 1); + TEST_CHECK(deferred_aborted.empty()); + TEST_EQUAL(deferred_clear, clear_job.get()); + TEST_EQUAL(int(f.cache.size()), 0); + TEST_EQUAL(f.alloc.live, 0); + } + +} + +TORRENT_TEST(clear_piece_v2_drain_in_flight_v2) +{ + test_clear_piece_v2_drain_in_flight(test_mode::v2); +} +TORRENT_TEST(clear_piece_v2_drain_in_flight_hybrid) +{ + test_clear_piece_v2_drain_in_flight(test_mode::v1 | test_mode::v2); +} + // flush_piece_impl releases the cache mutex during the flush callback. // Inserts into previously-empty slots of the piece being flushed are legal // at that point (insert() only requires block_idx >= hasher_cursor). The @@ -498,10 +702,7 @@ TORRENT_TEST(flush_storage_after_hash_piece_with_monostate_block) bool callback_invoked = false; auto const res = f.cache.hash_piece(f.loc(0_piece), hash_job.get(), - [&](aux::piece_hasher*, - std::uint16_t const hasher_cursor, - span const blocks, - span) { + [&](aux::piece_hasher*, std::uint16_t const hasher_cursor, span const blocks) { callback_invoked = true; TEST_EQUAL(hasher_cursor, 0); TEST_EQUAL(blocks.size(), 2); @@ -729,11 +930,7 @@ void test_piece_size2_smaller_than_piece_size(test_mode_t const mode) cache_fixture f(blocks_in_piece, mode); disk_cache::piece_entry_params const params{ - piece_size2, - effective_piece_size, - need_v1, - need_v2 - }; + piece_size2, effective_piece_size, need_v1, need_v2, {}}; for (int blk = 0; blk < blocks_in_piece; ++blk) { @@ -742,8 +939,8 @@ void test_piece_size2_smaller_than_piece_size(test_mode_t const mode) f.insert(0_piece, blk, params, buf_size); } - jobqueue_t completed, retry; - f.cache.kick_pending_hashers(completed, retry); + // kick_hashers() also drains the v2 hash queue, populating f.v2_hashes. + f.kick_hashers(); int const v2_blocks = (piece_size2 + default_block_size - 1) / default_block_size; std::vector block_hashes(need_v2 ? v2_blocks : 0); @@ -753,12 +950,25 @@ void test_piece_size2_smaller_than_piece_size(test_mode_t const mode) span{block_hashes.data(), int(block_hashes.size())}, sha1_hash{} }; - TEST_EQUAL(f.cache.try_hash_piece(f.loc(0_piece), hash_job.get()) - , disk_cache::hash_result::job_completed); + auto const hpr = f.cache.try_hash_piece(f.loc(0_piece), hash_job.get()); if (need_v1) + { + TEST_EQUAL(hpr, disk_cache::hash_result::job_completed); TEST_CHECK(!std::get(hash_job->action).piece_hash.is_all_zeros()); - for (auto const& h : block_hashes) - TEST_CHECK(!h.is_all_zeros()); + } + else + { + // Pure-v2: try_hash_piece returns post_job (see comment in + // test_disk_bottleneck). + TEST_EQUAL(hpr, disk_cache::hash_result::post_job); + } + if (need_v2) + { + // v2 hashes were stashed via drain_v2_hash_queue into f.v2_hashes. + TEST_EQUAL(int(f.v2_hashes.size()), v2_blocks); + for (int b = 0; b < v2_blocks; ++b) + TEST_CHECK(!f.v2_hashes.at({0_piece, b}).is_all_zeros()); + } TEST_EQUAL(f.flush(), blocks_in_piece); TEST_EQUAL(int(f.cache.size()), 0); diff --git a/test/test_slow_hash.cpp b/test/test_slow_hash.cpp index f3029b235..86b576bf9 100644 --- a/test/test_slow_hash.cpp +++ b/test/test_slow_hash.cpp @@ -186,26 +186,52 @@ void test_hash_job_dispatched_by_hasher(test_mode_t const mode) jobqueue_t completed, retry; - // With slow hashing each block takes ~1.6 s. Run the hasher in a thread - // and call try_hash_piece after a short sleep, reliably catching the - // window where hashing_flag is set. + // With slow hashing each block takes ~1.6 s. Run both the v1 hasher + // (kick_pending_hashers) and the v2 drain on the same thread, mirroring + // pread_disk_io::thread_fun, so a v2-only piece has a hashing window + // too. Call try_hash_piece after a short sleep so the hasher is + // reliably mid-run. std::thread t([&]() { f.cache.kick_pending_hashers(completed, retry); + f.cache.drain_v2_hash_queue( + [&block_hashes](std::shared_ptr const&, + lt::piece_index_t, + int const block, + lt::sha256_hash const& h) { + if (block < int(block_hashes.size())) block_hashes[std::size_t(block)] = h; + }, + retry, + [](jobqueue_t, disk_job*) {}); }); std::this_thread::sleep_for(std::chrono::milliseconds(50)); auto const result = f.cache.try_hash_piece(f.loc(0_piece), hash_job.get()); t.join(); + // try_hash_piece parks the job in all three modes: v1/hybrid via the + // hashing_flag branch (kick_hasher mid-run), v2-only via the v2_pending + // branch (drain mid-run). The dispatched job lands in completed for + // v1-only (kick_hasher's main path) and in retry for v2 and hybrid + // (kick_hasher routes hybrid through retry; the v2-only drain posts + // the hung job there too). TEST_EQUAL(result, disk_cache::hash_result::job_queued); - // The hasher filled in the hash values and posted the job. - TEST_EQUAL(completed.size(), 1); + auto& dispatched = bool(mode & test_mode::v2) ? retry : completed; + TEST_EQUAL(dispatched.size(), 1); if (mode & test_mode::v1) TEST_CHECK(!std::get(hash_job->action).piece_hash.is_all_zeros()); for (auto const& h : block_hashes) TEST_CHECK(!h.is_all_zeros()); + if (!(mode & test_mode::v1)) + { + // v2-only: the dispatched job goes to retry; in production + // pread_disk_io re-runs do_job(hash) which calls hash_piece, + // whose scope_end sets piece_hash_returned_flag and lets pass-1 + // evict. The fixture doesn't drive that path, so the cpe stays + // in the cache without the flag set -- skip the eviction check. + return; + } TEST_EQUAL(f.flush(), 1); TEST_EQUAL(int(f.cache.size()), 0); } @@ -231,23 +257,38 @@ void test_hash_job_retry_when_piece_incomplete(test_mode_t const mode) jobqueue_t completed, retry; - // Slow hashing gives us a wide window. Run the hasher in a thread, park - // the hash job while hashing_flag is set, then let the hasher finish. + // Slow hashing gives us a wide window. Run the hasher and the v2 drain + // on the same thread, park the hash job while hashing_flag is set, then + // let the hasher finish. std::thread t([&]() { f.cache.kick_pending_hashers(completed, retry); + f.cache.drain_v2_hash_queue( + [&block_hashes](std::shared_ptr const&, + lt::piece_index_t, + int const block, + lt::sha256_hash const& h) { + if (block < int(block_hashes.size())) block_hashes[std::size_t(block)] = h; + }, + retry, + [](jobqueue_t, disk_job*) {}); }); std::this_thread::sleep_for(std::chrono::milliseconds(50)); - // hashing_flag is set; block 1 is absent so the hasher will stall at - // cursor=1. try_hash_piece parks the job (job_queued) now that the - // have_buffers guard in try_hash_piece has been removed — the retry - // mechanism handles the case where the hasher can't complete. + // For v1/hybrid the hasher has set hashing_flag and block 1 is absent, + // so it will stall at cursor=1; try_hash_piece parks the job + // (job_queued) and the retry mechanism handles the case where the + // hasher can't complete. For v2-only the drain is mid-run and + // v2_pending > 0 parks the job through the same job_queued path. auto const result = f.cache.try_hash_piece(f.loc(0_piece), hash_job.get()); + // v1/hybrid: parked via hashing_flag (kick_hasher mid-run). + // v2-only: parked via v2_pending (drain mid-run). TEST_EQUAL(result, disk_cache::hash_result::job_queued); t.join(); - // kick_hasher hashed block 0 (cursor=1) but stopped because block 1 is - // absent. The hung hash job must be in retry, not completed. + // v1/hybrid: kick_hasher hashed block 0 (cursor=1) but stopped because + // block 1 is absent, and posted the hung job to retry. + // v2-only: drain processed block 0's queue entry, found the hung + // hash_job, and posted it to retry. Either way the job lands in retry. TEST_EQUAL(completed.size(), 0); TEST_EQUAL(retry.size(), 1); TEST_CHECK(retry.pop_front() == hash_job.get()); @@ -283,6 +324,13 @@ void test_hash_job_retry_when_piece_incomplete(test_mode_t const mode) // disk I/O - the data has already been written. void test_drop_held_alive_buffers(test_mode_t const mode) { + // The held-alive mechanism only applies to v1 pieces: kick_hasher pins + // block buffers until the SHA-1 hasher reaches them. For pieces with + // any v2 component (v2-only or hybrid) the insert path moves wjob.buf + // into m_v2_hash_queue, so cbe never owns the buffer; the data is + // owned by the queue entry and freed by drain_v2_hash_queue. + if (mode & test_mode::v2) return; + cache_fixture f(3, mode); f.insert(0_piece, 0); f.insert(0_piece, 2); @@ -329,8 +377,14 @@ void test_drop_held_alive_buffers(test_mode_t const mode) // block_hashes; instead it sets pending_free_flag and defers the erasure. // When kick_hasher() subsequently clears hashing_flag it sees // pending_free_flag and erases the piece itself. +// +// hashing_flag is set by kick_pending_hashers, which only kicks v1/hybrid +// pieces. v2-only pieces have no hashing_flag (their work happens in +// drain_v2_hash_queue), so they don't exercise the pending_free_flag race. void test_flush_storage_during_hashing(test_mode_t const mode) { + if (!(mode & test_mode::v1)) return; + // 2-block piece so the hasher has real work to do. cache_fixture f(2, mode); f.insert(0_piece, 0);