extensions
@@ -319,6 +321,7 @@ class session: public boost::noncopyable
void set_ip_filter(ip_filter const& f);
session_status status() const;
+ cache_status get_cache_status() const;
bool is_listening() const;
unsigned short listen_port() const;
@@ -644,6 +647,60 @@ becomes unresponsive.
network.
+
+
+
+cache_status get_cache_status() const;
+
+
+
Returns status of the disk cache for this session.
+
+
+struct cache_status
+{
+ size_type blocks_written;
+ size_type writes;
+ int write_size;
+};
+
+
+
blocks_written is the total number of 16 KiB blocks written to disk
+since this session was started.
+
writes is the total number of write operations performed since this
+session was started.
+
The ratio (blocks_written - writes) / blocks_written represents
+the number of saved write operations per total write operations. i.e. a kind
+of cache hit ratio for the write cahe.
+
write_size is the number of 16 KiB blocks currently in the write cache.
+
+
+
+
+void get_cache_info(sha1_hash const& ih
+ , std::vector<cached_piece_info>& ret) const;
+
+
+
get_cache_info() fills out the supplied vector with information for
+each piece that is currently in the disk cache for the torrent with the
+specified info-hash (ih).
+
+
+struct cached_piece_info
+{
+ int piece;
+ std::vector<bool> blocks;
+ ptime last_write;
+};
+
+
+
piece is the piece index for this cache entry.
+
blocks has one entry for each block in this piece. true represents
+the data for that block being in the disk cache and false means it's not.
+
last_write is the time when a block was last written to this piece. The older
+a piece is, the more likely it is to be flushed to disk.
+
@@ -2529,6 +2586,7 @@ struct session_settings
bool upnp_ignore_nonrouters;
int send_buffer_watermark;
bool auto_upload_slots;
+ int cache_size;
};
user_agent this is the client identification to the tracker.
@@ -2651,6 +2709,8 @@ slot is opened. If the upload rate has been saturated for an extended period
of time, on upload slot is closed. The number of upload slots will never be
less than what has been set by session::set_max_uploads(). To query the
current number of upload slots, see session_status::allowed_upload_slots.
+cache_size is the disk write cache. It is specified in units of 16 KiB blocks.
+It defaults to 128 (= 2 MB).
diff --git a/docs/manual.rst b/docs/manual.rst
index f2e69f9e7..fc09653bc 100644
--- a/docs/manual.rst
+++ b/docs/manual.rst
@@ -133,6 +133,7 @@ The ``session`` class has the following synopsis::
void set_ip_filter(ip_filter const& f);
session_status status() const;
+ cache_status get_cache_status() const;
bool is_listening() const;
unsigned short listen_port() const;
@@ -490,6 +491,65 @@ becomes unresponsive.
``dht_global_nodes`` is an estimation of the total number of nodes in the DHT
network.
+get_cache_status()
+------------------
+
+ ::
+
+ cache_status get_cache_status() const;
+
+Returns status of the disk cache for this session.
+
+ ::
+
+ struct cache_status
+ {
+ size_type blocks_written;
+ size_type writes;
+ int write_size;
+ };
+
+``blocks_written`` is the total number of 16 KiB blocks written to disk
+since this session was started.
+
+``writes`` is the total number of write operations performed since this
+session was started.
+
+The ratio (``blocks_written`` - ``writes``) / ``blocks_written`` represents
+the number of saved write operations per total write operations. i.e. a kind
+of cache hit ratio for the write cahe.
+
+``write_size`` is the number of 16 KiB blocks currently in the write cache.
+
+get_cache_info()
+----------------
+
+ ::
+
+ void get_cache_info(sha1_hash const& ih
+ , std::vector
& ret) const;
+
+``get_cache_info()`` fills out the supplied vector with information for
+each piece that is currently in the disk cache for the torrent with the
+specified info-hash (``ih``).
+
+ ::
+
+ struct cached_piece_info
+ {
+ int piece;
+ std::vector blocks;
+ ptime last_write;
+ };
+
+``piece`` is the piece index for this cache entry.
+
+``blocks`` has one entry for each block in this piece. ``true`` represents
+the data for that block being in the disk cache and ``false`` means it's not.
+
+``last_write`` is the time when a block was last written to this piece. The older
+a piece is, the more likely it is to be flushed to disk.
+
is_listening() listen_port() listen_on()
----------------------------------------
@@ -2513,6 +2573,7 @@ that will be sent to the tracker. The user-agent is a good way to identify your
bool upnp_ignore_nonrouters;
int send_buffer_watermark;
bool auto_upload_slots;
+ int cache_size;
};
``user_agent`` this is the client identification to the tracker.
@@ -2663,6 +2724,9 @@ of time, on upload slot is closed. The number of upload slots will never be
less than what has been set by ``session::set_max_uploads()``. To query the
current number of upload slots, see ``session_status::allowed_upload_slots``.
+``cache_size`` is the disk write cache. It is specified in units of 16 KiB blocks.
+It defaults to 128 (= 2 MB).
+
pe_settings
===========
diff --git a/examples/client_test.cpp b/examples/client_test.cpp
index a70e17df9..0e322039b 100644
--- a/examples/client_test.cpp
+++ b/examples/client_test.cpp
@@ -896,6 +896,7 @@ int main(int ac, char* av[])
std::vector queue;
bool print_peers = false;
+ bool print_cache = false;
bool print_log = false;
bool print_downloads = false;
bool print_piece_bar = false;
@@ -962,6 +963,7 @@ int main(int ac, char* av[])
}
if (c == 'i') print_peers = !print_peers;
+ if (c == 'c') print_cache = !print_cache;
if (c == 'l') print_log = !print_log;
if (c == 'd') print_downloads = !print_downloads;
if (c == 'f') print_file_progress = !print_file_progress;
@@ -1150,9 +1152,18 @@ int main(int ac, char* av[])
h.get_download_queue(queue);
std::sort(queue.begin(), queue.end(), bind(&partial_piece_info::piece_index, _1)
< bind(&partial_piece_info::piece_index, _2));
+
+ std::vector pieces;
+ ses.get_cache_info(h.info_hash(), pieces);
+
for (std::vector::iterator i = queue.begin();
i != queue.end(); ++i)
{
+ cached_piece_info* cp = 0;
+ std::vector::iterator cpi = std::find_if(pieces.begin(), pieces.end()
+ , bind(&cached_piece_info::piece, _1) == i->piece_index);
+ if (cpi != pieces.end()) cp = &*cpi;
+
out << to_string(i->piece_index, 4) << ": [";
for (int j = 0; j < i->blocks_in_piece; ++j)
{
@@ -1162,7 +1173,8 @@ int main(int ac, char* av[])
str[0] = (index < 10)?'0' + index:'A' + index - 10;
#ifdef ANSI_TERMINAL_COLORS
- if (i->blocks[j].bytes_progress > 0
+ if (cp && cp->blocks[j]) out << esc("36;7") << str << esc("0");
+ else if (i->blocks[j].bytes_progress > 0
&& i->blocks[j].state == block_info::requested)
{
if (i->blocks[j].num_peers > 1)
@@ -1176,7 +1188,8 @@ int main(int ac, char* av[])
else if (i->blocks[j].state == block_info::requested) out << str;
else out << " ";
#else
- if (i->blocks[j].state == block_info::finished) out << "#";
+ if (cp && cp->blocks[j]) out << "c";
+ else if (i->blocks[j].state == block_info::finished) out << "#";
else if (i->blocks[j].state == block_info::writing) out << "+";
else if (i->blocks[j].state == block_info::requested) out << str;
else out << " ";
@@ -1211,6 +1224,8 @@ int main(int ac, char* av[])
}
+ cache_status cs = ses.get_cache_status();
+
out << "==== conns: " << sess_stat.num_peers
<< " down: " << esc("32") << add_suffix(sess_stat.download_rate) << "/s" << esc("0")
<< " (" << esc("32") << add_suffix(sess_stat.total_download) << esc("0") << ") "
@@ -1218,7 +1233,10 @@ int main(int ac, char* av[])
<< " (" << esc("31") << add_suffix(sess_stat.total_upload) << esc("0") << ")"
" unchoked: " << sess_stat.num_unchoked << " / " << sess_stat.allowed_upload_slots
<< " bw queues: (" << sess_stat.up_bandwidth_queue
- << " | " << sess_stat.down_bandwidth_queue << ") ====" << std::endl;
+ << " | " << sess_stat.down_bandwidth_queue << ") "
+ " write cache hits: " << ((cs.blocks_written - cs.writes) * 100 / cs.blocks_written) << "% "
+ " cache size: " << add_suffix(cs.write_size * 16 * 1024)
+ << " ====" << std::endl;
if (print_log)
{
diff --git a/include/libtorrent/aux_/session_impl.hpp b/include/libtorrent/aux_/session_impl.hpp
index 916d842a2..7b1edf712 100644
--- a/include/libtorrent/aux_/session_impl.hpp
+++ b/include/libtorrent/aux_/session_impl.hpp
@@ -399,19 +399,21 @@ namespace libtorrent
// when they are destructed.
file_pool m_files;
+ // this is where all active sockets are stored.
+ // the selector can sleep while there's no activity on
+ // them
+ io_service m_io_service;
+
// handles disk io requests asynchronously
// peers have pointers into the disk buffer
// pool, and must be destructed before this
// object. The disk thread relies on the file
// pool object, and must be destructed before
- // m_files.
+ // m_files. The disk io thread posts completion
+ // events to the io service, and needs to be
+ // constructed after it.
disk_io_thread m_disk_thread;
- // this is where all active sockets are stored.
- // the selector can sleep while there's no activity on
- // them
- io_service m_io_service;
-
// this is a list of half-open tcp connections
// (only outgoing connections)
// this has to be one of the last
diff --git a/include/libtorrent/disk_io_thread.hpp b/include/libtorrent/disk_io_thread.hpp
index 1d9d88534..17238ef3c 100644
--- a/include/libtorrent/disk_io_thread.hpp
+++ b/include/libtorrent/disk_io_thread.hpp
@@ -44,11 +44,19 @@ POSSIBILITY OF SUCH DAMAGE.
#include
#include
#include
+#include
#include "libtorrent/config.hpp"
namespace libtorrent
{
+ struct cached_piece_info
+ {
+ int piece;
+ std::vector blocks;
+ ptime last_write;
+ };
+
struct disk_io_job
{
disk_io_job()
@@ -91,11 +99,22 @@ namespace libtorrent
boost::function callback;
};
+ struct cache_status
+ {
+ // the number of 16kB blocks written
+ size_type blocks_written;
+ // the number of write operations used
+ size_type writes;
+ // (blocks_written - writes) / blocks_written represents the
+ // "cache hit" ratio in the write cache
+ int write_size;
+ };
+
// this is a singleton consisting of the thread and a queue
// of disk io jobs
struct disk_io_thread : boost::noncopyable
{
- disk_io_thread(int block_size = 16 * 1024);
+ disk_io_thread(asio::io_service& ios, int block_size = 16 * 1024);
~disk_io_thread();
#ifdef TORRENT_STATS
@@ -122,6 +141,12 @@ namespace libtorrent
size_type queue_buffer_size() const
{ return m_queue_buffer_size; }
+ void get_cache_info(sha1_hash const& ih
+ , std::vector& ret) const;
+
+ cache_status status() const;
+ void set_cache_size(int s);
+
void operator()();
char* allocate_buffer();
@@ -130,13 +155,43 @@ namespace libtorrent
private:
typedef boost::recursive_mutex mutex_t;
+
+ struct cached_piece_entry
+ {
+ int piece;
+ // storage this piece belongs to
+ boost::intrusive_ptr storage;
+ // the last time a block was writting to this piece
+ ptime last_write;
+ // the number of blocks in the cache for this piece
+ int num_blocks;
+ // the pointers to the block data
+ boost::shared_array blocks;
+ };
+
+ char* allocate_buffer(mutex_t::scoped_lock& l);
+ void free_buffer(char* buf, mutex_t::scoped_lock& l);
+
+ std::vector::iterator find_cached_piece(
+ disk_io_job const& j, mutex_t::scoped_lock& l);
+ void flush_oldest_piece(mutex_t::scoped_lock& l);
+ void flush_and_remove(std::vector::iterator i, mutex_t::scoped_lock& l);
+ void flush(std::vector::iterator i, mutex_t::scoped_lock& l);
+ void cache_block(disk_io_job& j, mutex_t::scoped_lock& l);
+
mutable mutex_t m_mutex;
boost::condition m_signal;
bool m_abort;
std::deque m_jobs;
size_type m_queue_buffer_size;
+ std::vector m_pieces;
+ int m_num_cached_blocks;
+ // in (16kB) blocks
+ int m_cache_size;
+
// memory pool for read and write operations
+ // and disk cache
boost::pool<> m_pool;
#ifndef NDEBUG
@@ -151,6 +206,11 @@ namespace libtorrent
int m_allocations;
#endif
+ size_type m_writes;
+ size_type m_blocks_written;
+
+ asio::io_service& m_ios;
+
// thread for performing blocking disk io operations
boost::thread m_disk_io_thread;
};
diff --git a/include/libtorrent/session.hpp b/include/libtorrent/session.hpp
index d2ab6ab2e..cf1c593a3 100755
--- a/include/libtorrent/session.hpp
+++ b/include/libtorrent/session.hpp
@@ -61,6 +61,7 @@ POSSIBILITY OF SUCH DAMAGE.
#include "libtorrent/version.hpp"
#include "libtorrent/fingerprint.hpp"
#include "libtorrent/time.hpp"
+#include "libtorrent/disk_io_thread.hpp"
#include "libtorrent/storage.hpp"
@@ -176,6 +177,10 @@ namespace libtorrent
session_proxy abort() { return session_proxy(m_impl); }
session_status status() const;
+ cache_status get_cache_status() const;
+
+ void get_cache_info(sha1_hash const& ih
+ , std::vector& ret) const;
#ifndef TORRENT_DISABLE_DHT
void start_dht(entry const& startup_state = entry());
diff --git a/include/libtorrent/session_settings.hpp b/include/libtorrent/session_settings.hpp
index f2abe6b3c..1b9ee7873 100644
--- a/include/libtorrent/session_settings.hpp
+++ b/include/libtorrent/session_settings.hpp
@@ -120,6 +120,7 @@ namespace libtorrent
, upnp_ignore_nonrouters(true)
, send_buffer_watermark(80 * 1024)
, auto_upload_slots(true)
+ , cache_size(128)
{}
// this is the user agent that will be sent to the tracker
@@ -317,6 +318,10 @@ namespace libtorrent
// upload slots are never automatically decreased below
// the manual settings, through max_uploads.
bool auto_upload_slots;
+
+ // the disk write cache, specified in 16 KiB blocks.
+ // defaul is 128 (= 2 MB)
+ int cache_size;
};
#ifndef TORRENT_DISABLE_DHT
diff --git a/include/libtorrent/storage.hpp b/include/libtorrent/storage.hpp
index 659e72717..ff206ad7b 100755
--- a/include/libtorrent/storage.hpp
+++ b/include/libtorrent/storage.hpp
@@ -191,6 +191,8 @@ namespace libtorrent
~piece_manager();
+ torrent_info const* info() const { return m_info.get(); }
+
bool check_fastresume(aux::piece_checker_data& d
, std::vector& pieces, int& num_pieces, storage_mode_t storage_mode
, std::string& error_msg);
diff --git a/src/disk_io_thread.cpp b/src/disk_io_thread.cpp
index e1cfbfe5f..7c9a8cacb 100644
--- a/src/disk_io_thread.cpp
+++ b/src/disk_io_thread.cpp
@@ -34,22 +34,30 @@ POSSIBILITY OF SUCH DAMAGE.
#include
#include "libtorrent/disk_io_thread.hpp"
+#ifdef _WIN32
+#include
+#define alloca(s) _alloca(s)
+#endif
+
#ifdef TORRENT_DISK_STATS
-
#include "libtorrent/time.hpp"
-
#endif
namespace libtorrent
{
- disk_io_thread::disk_io_thread(int block_size)
+ disk_io_thread::disk_io_thread(asio::io_service& ios, int block_size)
: m_abort(false)
, m_queue_buffer_size(0)
+ , m_num_cached_blocks(0)
+ , m_cache_size(128) // 128 * 16kB = 2MB
, m_pool(block_size)
#ifndef NDEBUG
, m_block_size(block_size)
#endif
+ , m_writes(0)
+ , m_blocks_written(0)
+ , m_ios(ios)
, m_disk_io_thread(boost::ref(*this))
{
#ifdef TORRENT_STATS
@@ -100,6 +108,44 @@ namespace libtorrent
m_disk_io_thread.join();
}
+ void disk_io_thread::get_cache_info(sha1_hash const& ih, std::vector& ret) const
+ {
+ mutex_t::scoped_lock l(m_mutex);
+ ret.clear();
+ ret.reserve(m_pieces.size());
+ for (std::vector::const_iterator i = m_pieces.begin()
+ , end(m_pieces.end()); i != end; ++i)
+ {
+ torrent_info const& ti = *i->storage->info();
+ if (ti.info_hash() != ih) continue;
+ cached_piece_info info;
+ info.piece = i->piece;
+ info.last_write = i->last_write;
+ int blocks_in_piece = (ti.piece_size(i->piece) + (16 * 1024) - 1) / (16 * 1024);
+ info.blocks.resize(blocks_in_piece);
+ for (int b = 0; b < blocks_in_piece; ++b)
+ if (i->blocks[b]) info.blocks[b] = true;
+ ret.push_back(info);
+ }
+ }
+
+ cache_status disk_io_thread::status() const
+ {
+ mutex_t::scoped_lock l(m_mutex);
+ cache_status st;
+ st.blocks_written = m_blocks_written;
+ st.writes = m_writes;
+ st.write_size = m_num_cached_blocks;
+ return st;
+ }
+
+ void disk_io_thread::set_cache_size(int s)
+ {
+ mutex_t::scoped_lock l(m_mutex);
+ TORRENT_ASSERT(s >= 0);
+ m_cache_size = s;
+ }
+
// aborts read operations
void disk_io_thread::stop(boost::intrusive_ptr s)
{
@@ -115,7 +161,7 @@ namespace libtorrent
}
if (i->action == disk_io_job::read)
{
- i->callback(-1, *i);
+ if (i->callback) m_ios.post(bind(i->callback, -1, *i));
m_jobs.erase(i++);
continue;
}
@@ -150,14 +196,130 @@ namespace libtorrent
return false;
}
}
+
+ std::vector::iterator disk_io_thread::find_cached_piece(
+ disk_io_job const& j, mutex_t::scoped_lock& l)
+ {
+ TORRENT_ASSERT(l.locked());
+ for (std::vector::iterator i = m_pieces.begin()
+ , end(m_pieces.end()); i != end; ++i)
+ {
+ if (i->storage != j.storage || i->piece != j.piece) continue;
+ return i;
+ }
+ return m_pieces.end();
+ }
+ void disk_io_thread::flush_oldest_piece(mutex_t::scoped_lock& l)
+ {
+ TORRENT_ASSERT(l.locked());
+ std::vector::iterator i = std::min_element(
+ m_pieces.begin(), m_pieces.end()
+ , bind(&cached_piece_entry::last_write, _1)
+ < bind(&cached_piece_entry::last_write, _1));
+ if (i == m_pieces.end()) return;
+ flush_and_remove(i, l);
+ }
+
+ void disk_io_thread::flush_and_remove(std::vector::iterator e
+ , mutex_t::scoped_lock& l)
+ {
+ flush(e, l);
+ m_pieces.erase(e);
+ }
+
+ void disk_io_thread::flush(std::vector::iterator e
+ , mutex_t::scoped_lock& l)
+ {
+ TORRENT_ASSERT(l.locked());
+ cached_piece_entry& p = *e;
+ int piece_size = p.storage->info()->piece_size(p.piece);
+ TORRENT_ASSERT(piece_size > 0);
+// char* buf = (char*)alloca(piece_size);
+ std::vector temp(piece_size);
+ char* buf = &temp[0];
+ TORRENT_ASSERT(buf != 0);
+
+ int blocks_in_piece = (piece_size + (16 * 1024) - 1) / (16 * 1024);
+ int buffer_size = 0;
+ int offset = 0;
+ for (int i = 0; i <= blocks_in_piece; ++i)
+ {
+ if (i == blocks_in_piece || p.blocks[i] == 0)
+ {
+ if (buffer_size == 0) continue;
+
+ TORRENT_ASSERT(buffer_size <= i * 16 * 1024);
+ l.unlock();
+ p.storage->write_impl(buf, p.piece, (std::min)(i * 16 * 1024, piece_size) - buffer_size, buffer_size);
+ l.lock();
+ ++m_writes;
+// std::cerr << " flushing p: " << p.piece << " bytes: " << buffer_size << std::endl;
+ buffer_size = 0;
+ offset = 0;
+ continue;
+ }
+ int block_size = (std::min)(piece_size - offset, 16 * 1024);
+ TORRENT_ASSERT(offset + block_size <= piece_size);
+ TORRENT_ASSERT(offset + block_size > 0);
+ std::memcpy(buf + offset, p.blocks[i], block_size);
+ offset += 16 * 1024;
+ free_buffer(p.blocks[i], l);
+ p.blocks[i] = 0;
+ buffer_size += block_size;
+ ++m_blocks_written;
+ --m_num_cached_blocks;
+ }
+ TORRENT_ASSERT(buffer_size == 0);
+// std::cerr << " flushing p: " << p.piece << " cached_blocks: " << m_num_cached_blocks << std::endl;
+#ifndef NDEBUG
+ for (int i = 0; i < blocks_in_piece; ++i)
+ TORRENT_ASSERT(p.blocks[i] == 0);
+#endif
+ }
+
+ void disk_io_thread::cache_block(disk_io_job& j, mutex_t::scoped_lock& l)
+ {
+ TORRENT_ASSERT(l.locked());
+ TORRENT_ASSERT(find_cached_piece(j, l) == m_pieces.end());
+ cached_piece_entry p;
+
+ int piece_size = j.storage->info()->piece_size(j.piece);
+ int blocks_in_piece = (piece_size + (16 * 1024) - 1) / (16 * 1024);
+
+ p.piece = j.piece;
+ p.storage = j.storage;
+ p.last_write = time_now();
+ p.num_blocks = 1;
+ p.blocks.reset(new char*[blocks_in_piece]);
+ std::memset(&p.blocks[0], 0, blocks_in_piece * sizeof(char*));
+ int block = j.offset / (16 * 1024);
+// std::cerr << " adding cache entry for p: " << j.piece << " block: " << block << " cached_blocks: " << m_num_cached_blocks << std::endl;
+ p.blocks[block] = j.buffer;
+ ++m_num_cached_blocks;
+ m_pieces.push_back(p);
+ }
+
void disk_io_thread::add_job(disk_io_job const& j
, boost::function const& f)
{
TORRENT_ASSERT(!j.callback);
TORRENT_ASSERT(j.storage);
+ TORRENT_ASSERT(j.buffer_size <= 16 * 1024);
mutex_t::scoped_lock l(m_mutex);
-
+#ifndef NDEBUG
+ if (j.action == disk_io_job::write)
+ {
+ std::vector::iterator p = find_cached_piece(j, l);
+ if (p != m_pieces.end())
+ {
+ int block = j.offset / (16 * 1024);
+ char const* buffer = p->blocks[block];
+ TORRENT_ASSERT(buffer == 0);
+ }
+ }
+#endif
+
std::deque::reverse_iterator i = m_jobs.rbegin();
if (j.action == disk_io_job::read)
{
@@ -212,15 +374,27 @@ namespace libtorrent
char* disk_io_thread::allocate_buffer()
{
mutex_t::scoped_lock l(m_mutex);
+ return allocate_buffer(l);
+ }
+
+ void disk_io_thread::free_buffer(char* buf)
+ {
+ mutex_t::scoped_lock l(m_mutex);
+ free_buffer(buf, l);
+ }
+
+ char* disk_io_thread::allocate_buffer(mutex_t::scoped_lock& l)
+ {
+ TORRENT_ASSERT(l.locked());
#ifdef TORRENT_STATS
++m_allocations;
#endif
return (char*)m_pool.ordered_malloc();
}
- void disk_io_thread::free_buffer(char* buf)
+ void disk_io_thread::free_buffer(char* buf, mutex_t::scoped_lock& l)
{
- mutex_t::scoped_lock l(m_mutex);
+ TORRENT_ASSERT(l.locked());
#ifdef TORRENT_STATS
--m_allocations;
#endif
@@ -288,27 +462,53 @@ namespace libtorrent
// usleep(300);
break;
case disk_io_job::write:
+ {
+ mutex_t::scoped_lock l(m_mutex);
#ifdef TORRENT_DISK_STATS
m_log << log_time() << " write " << j.buffer_size << std::endl;
#endif
+ std::vector::iterator p = find_cached_piece(j, l);
+ int block = j.offset / (16 * 1024);
TORRENT_ASSERT(j.buffer);
TORRENT_ASSERT(j.buffer_size <= m_block_size);
- j.storage->write_impl(j.buffer, j.piece, j.offset
- , j.buffer_size);
-
+ if (p != m_pieces.end())
+ {
+ TORRENT_ASSERT(p->blocks[block] == 0);
+ if (p->blocks[block]) free_buffer(p->blocks[block]);
+ p->blocks[block] = j.buffer;
+ ++m_num_cached_blocks;
+ ++p->num_blocks;
+ p->last_write = time_now();
+// std::cerr << " adding cache entry for p: " << j.piece
+// << " block: " << block
+// << " cached_blocks: " << m_num_cached_blocks << std::endl;
+ }
+ else
+ {
+ cache_block(j, l);
+ }
+ free_current_buffer = false;
+ if (m_num_cached_blocks >= m_cache_size)
+ flush_oldest_piece(l);
+
// simulates a slow drive
// usleep(300);
break;
+ }
case disk_io_job::hash:
- {
+ {
#ifdef TORRENT_DISK_STATS
- m_log << log_time() << " hash" << std::endl;
+ m_log << log_time() << " hash" << std::endl;
#endif
- sha1_hash h = j.storage->hash_for_piece_impl(j.piece);
- j.str.resize(20);
- std::memcpy(&j.str[0], &h[0], 20);
- }
+ mutex_t::scoped_lock l(m_mutex);
+ std::vector::iterator i = find_cached_piece(j, l);
+ if (i != m_pieces.end()) flush_and_remove(i, l);
+ l.unlock();
+ sha1_hash h = j.storage->hash_for_piece_impl(j.piece);
+ j.str.resize(20);
+ std::memcpy(&j.str[0], &h[0], 20);
break;
+ }
case disk_io_job::move_storage:
#ifdef TORRENT_DISK_STATS
m_log << log_time() << " move" << std::endl;
@@ -317,17 +517,50 @@ namespace libtorrent
j.str = j.storage->save_path().string();
break;
case disk_io_job::release_files:
+ {
#ifdef TORRENT_DISK_STATS
m_log << log_time() << " release" << std::endl;
#endif
+ mutex_t::scoped_lock l(m_mutex);
+
+ std::vector::iterator i = std::remove_if(
+ m_pieces.begin(), m_pieces.end(), bind(&cached_piece_entry::storage, _1) == j.storage);
+
+ for (std::vector::iterator k = i; k != m_pieces.end(); ++k)
+ flush(k, l);
+ m_pieces.erase(i, m_pieces.end());
+ m_pool.release_memory();
+ l.unlock();
j.storage->release_files_impl();
break;
+ }
case disk_io_job::delete_files:
+ {
#ifdef TORRENT_DISK_STATS
m_log << log_time() << " delete" << std::endl;
#endif
+ mutex_t::scoped_lock l(m_mutex);
+
+ std::vector::iterator i = std::remove_if(
+ m_pieces.begin(), m_pieces.end(), bind(&cached_piece_entry::storage, _1) == j.storage);
+
+ for (std::vector::iterator k = i; k != m_pieces.end(); ++k)
+ {
+ torrent_info const& ti = *k->storage->info();
+ int blocks_in_piece = (ti.piece_size(k->piece) + (16 * 1024) - 1) / (16 * 1024);
+ for (int j = 0; j < blocks_in_piece; ++j)
+ {
+ if (k->blocks[j] == 0) continue;
+ free_buffer(k->blocks[j], l);
+ k->blocks[j] = 0;
+ }
+ }
+ m_pieces.erase(i, m_pieces.end());
+ m_pool.release_memory();
+ l.unlock();
j.storage->delete_files_impl();
break;
+ }
}
}
catch (std::exception& e)
@@ -343,8 +576,8 @@ namespace libtorrent
// if (!handler) std::cerr << "DISK THREAD: no callback specified" << std::endl;
// else std::cerr << "DISK THREAD: invoking callback" << std::endl;
- try { if (handler) handler(ret, j); }
- catch (std::exception&) {}
+
+ if (handler) m_ios.post(bind(handler, ret, j));
#ifndef NDEBUG
m_current.storage = 0;
diff --git a/src/peer_connection.cpp b/src/peer_connection.cpp
index 9e815e8d1..989b22fcb 100755
--- a/src/peer_connection.cpp
+++ b/src/peer_connection.cpp
@@ -3111,6 +3111,9 @@ namespace libtorrent
complete = false;
break;
}
+/*
+// this invariant is not valid anymore since the completion event
+// might be queued in the io service
if (complete && !piece_failed)
{
disk_io_job ret = m_ses.m_disk_thread.find_job(
@@ -3118,6 +3121,7 @@ namespace libtorrent
TORRENT_ASSERT(ret.action == disk_io_job::hash || ret.action == disk_io_job::write);
TORRENT_ASSERT(ret.piece == i->index);
}
+*/
}
}
// expensive when using checked iterators
diff --git a/src/piece_picker.cpp b/src/piece_picker.cpp
index 0a60c94d4..ecd2cbd89 100755
--- a/src/piece_picker.cpp
+++ b/src/piece_picker.cpp
@@ -48,8 +48,8 @@ POSSIBILITY OF SUCH DAMAGE.
#include "libtorrent/torrent.hpp"
#endif
-#define TORRENT_PIECE_PICKER_INVARIANT_CHECK INVARIANT_CHECK
-//#define TORRENT_PIECE_PICKER_INVARIANT_CHECK
+//#define TORRENT_PIECE_PICKER_INVARIANT_CHECK INVARIANT_CHECK
+#define TORRENT_PIECE_PICKER_INVARIANT_CHECK
//#define TORRENT_PICKER_LOG
diff --git a/src/policy.cpp b/src/policy.cpp
index e452128bd..9626d2d24 100755
--- a/src/policy.cpp
+++ b/src/policy.cpp
@@ -224,7 +224,10 @@ namespace libtorrent
// the number of blocks we want, but it will try to make the picked
// blocks be from whole pieces, possibly by returning more blocks
// than we requested.
- TORRENT_ASSERT(c.remote() == c.get_socket()->remote_endpoint());
+#ifndef NDEBUG
+ asio::error_code ec;
+ TORRENT_ASSERT(c.remote() == c.get_socket()->remote_endpoint(ec) || ec);
+#endif
piece_picker::piece_state_t state;
peer_connection::peer_speed_t speed = c.peer_speed();
diff --git a/src/session.cpp b/src/session.cpp
index 9663d0900..9528c3d3c 100755
--- a/src/session.cpp
+++ b/src/session.cpp
@@ -277,6 +277,17 @@ namespace libtorrent
return m_impl->status();
}
+ void session::get_cache_info(sha1_hash const& ih
+ , std::vector& ret) const
+ {
+ m_impl->m_disk_thread.get_cache_info(ih, ret);
+ }
+
+ cache_status session::get_cache_status() const
+ {
+ return m_impl->m_disk_thread.status();
+ }
+
#ifndef TORRENT_DISABLE_DHT
void session::start_dht(entry const& startup_state)
diff --git a/src/session_impl.cpp b/src/session_impl.cpp
index 3fbcdcacc..6de357159 100755
--- a/src/session_impl.cpp
+++ b/src/session_impl.cpp
@@ -552,6 +552,8 @@ namespace detail
)
: m_send_buffers(send_buffer_size)
, m_files(40)
+ , m_io_service()
+ , m_disk_thread(m_io_service)
, m_half_open(m_io_service)
, m_download_channel(m_io_service, peer_connection::download_channel)
#ifdef TORRENT_VERBOSE_BANDWIDTH_LIMIT
@@ -807,6 +809,8 @@ namespace detail
// less than 5 seconds unchoke interval is insane
TORRENT_ASSERT(s.unchoke_interval >= 5);
+ if (m_settings.cache_size != s.cache_size)
+ m_disk_thread.set_cache_size(s.cache_size);
m_settings = s;
m_files.resize(m_settings.file_pool_size);
if (!s.auto_upload_slots) m_allowed_upload_slots = m_max_uploads;
diff --git a/src/torrent.cpp b/src/torrent.cpp
index 941b183af..a5c81d313 100755
--- a/src/torrent.cpp
+++ b/src/torrent.cpp
@@ -2765,6 +2765,9 @@ namespace libtorrent
complete = false;
break;
}
+ // this is no longer valid since the completion event
+ // may be queued in the io service
+/*
if (complete && m_files_checked)
{
disk_io_job ret = m_ses.m_disk_thread.find_job(
@@ -2772,6 +2775,7 @@ namespace libtorrent
TORRENT_ASSERT(ret.action == disk_io_job::hash || ret.action == disk_io_job::write);
TORRENT_ASSERT(ret.piece == i->index);
}
+*/
}
}
diff --git a/src/torrent_handle.cpp b/src/torrent_handle.cpp
index de3b23a07..ea96b5b1f 100755
--- a/src/torrent_handle.cpp
+++ b/src/torrent_handle.cpp
@@ -569,7 +569,7 @@ namespace libtorrent
TORRENT_ASSERT(bits == 8 || j == num_bitmask_bytes - 1);
}
piece_struct["bitmask"] = bitmask;
-
+/*
TORRENT_ASSERT(t->filesystem().slot_for(i->index) >= 0);
unsigned long adler
= t->filesystem().piece_crc(
@@ -578,7 +578,7 @@ namespace libtorrent
, i->info);
piece_struct["adler32"] = adler;
-
+*/
// push the struct onto the unfinished-piece list
up.push_back(piece_struct);
}
diff --git a/test/test_storage.cpp b/test/test_storage.cpp
index c1c210cfa..93a99bdda 100644
--- a/test/test_storage.cpp
+++ b/test/test_storage.cpp
@@ -85,7 +85,8 @@ void run_storage_tests(boost::intrusive_ptr info
// make sure the piece_manager can identify the pieces
{
file_pool fp;
- disk_io_thread io;
+ asio::io_service ios;
+ disk_io_thread io(ios);
boost::shared_ptr dummy(new int);
boost::intrusive_ptr pm = new piece_manager(dummy, info
, test_path, fp, io, default_storage_constructor);
@@ -130,6 +131,8 @@ void run_storage_tests(boost::intrusive_ptr info
pm->async_read(r, bind(&on_read_piece, _1, _2, piece2, piece_size));
pm->async_release_files(none);
+ ios.run();
+
io.join();
}
}