Merge branch '2016-03-08-cache-writeback' into v0.7-devel

Makefile.in

@@ -36,11 +36,15 @@ SOURCE=\
 	base/rolling_hash.cc \
 	base/xml_utils.cc \
 	block-cache/block_cache.cc \
+	block-cache/copier.cc \
+	block-cache/io_engine.cc \
+	block-cache/mem_pool.cc \
 	caching/cache_check.cc \
 	caching/cache_dump.cc \
 	caching/cache_metadata_size.cc \
 	caching/cache_repair.cc \
 	caching/cache_restore.cc \
+	caching/cache_writeback.cc \
 	caching/commands.cc \
 	caching/hint_array.cc \
 	caching/mapping_array.cc \
@@ -180,8 +184,8 @@ INSTALL_DATA = $(INSTALL) -p -m 644
 
 ifeq ("@TESTING@", "yes")
 TEST_INCLUDES=\
-	-Igmock-1.6.0/include \
+	-Igoogletest/googlemock/include \
-	-Igmock-1.6.0/gtest/include
+	-Igoogletest/googletest/include
 else
 TEST_INCLUDES=
 endif
@@ -234,6 +238,7 @@ install: bin/pdata_tools
 	ln -s -f pdata_tools $(BINDIR)/cache_metadata_size
 	ln -s -f pdata_tools $(BINDIR)/cache_repair
 	ln -s -f pdata_tools $(BINDIR)/cache_restore
+	ln -s -f pdata_tools $(BINDIR)/cache_writeback
 	ln -s -f pdata_tools $(BINDIR)/thin_check
 	ln -s -f pdata_tools $(BINDIR)/thin_delta
 	ln -s -f pdata_tools $(BINDIR)/thin_dump
@@ -253,6 +258,7 @@ install: bin/pdata_tools
 	$(INSTALL_DATA) man8/cache_dump.8 $(MANPATH)/man8
 	$(INSTALL_DATA) man8/cache_repair.8 $(MANPATH)/man8
 	$(INSTALL_DATA) man8/cache_restore.8 $(MANPATH)/man8
+	$(INSTALL_DATA) man8/cache_writeback.8 $(MANPATH)/man8
 	$(INSTALL_DATA) man8/thin_check.8 $(MANPATH)/man8
 	$(INSTALL_DATA) man8/thin_delta.8 $(MANPATH)/man8
 	$(INSTALL_DATA) man8/thin_dump.8 $(MANPATH)/man8

base/unique_handle.h

@@ -0,0 +1,67 @@
+#ifndef BASE_UNIQUE_HANDLE_H
+#define BASE_UNIQUE_HANDLE_H
+
+#include <list>
+#include <memory>
+#include <unistd.h>
+
+//----------------------------------------------------------------
+
+namespace base {
+	template <typename T, T TNul = T()>
+	class unique_handle
+	{
+	public:
+		unique_handle(std::nullptr_t = nullptr)
+			: id_(TNul) {
+		}
+
+		unique_handle(T x)
+		: id_(x) {
+		}
+
+		explicit operator bool() const {
+			return id_ != TNul;
+		}
+
+		operator T&() {
+			return id_;
+		}
+
+		operator T() const {
+			return id_;
+		}
+
+		T *operator&() {
+			return &id_;
+		}
+
+		const T *operator&() const {
+			return &id_;
+		}
+
+		friend bool operator == (unique_handle a, unique_handle b) { return a.id_ == b.id_; }
+		friend bool operator != (unique_handle a, unique_handle b) { return a.id_ != b.id_; }
+		friend bool operator == (unique_handle a, std::nullptr_t) { return a.id_ == TNul; }
+		friend bool operator != (unique_handle a, std::nullptr_t) { return a.id_ != TNul; }
+		friend bool operator == (std::nullptr_t, unique_handle b) { return TNul == b.id_; }
+		friend bool operator != (std::nullptr_t, unique_handle b) { return TNul != b.id_; }
+
+	private:
+		T id_;
+	};
+
+	//--------------------------------
+
+	struct fd_deleter {
+		typedef unique_handle<int, -1> pointer;
+		void operator()(pointer p) {
+			::close(p);
+		}
+	};
+	typedef std::unique_ptr<int, fd_deleter> unique_fd;
+}
+
+//----------------------------------------------------------------
+
+#endif

block-cache/block_cache.cc

@@ -525,8 +525,11 @@ block_cache::get(block_address index, unsigned flags, validator::ptr v)
 	block *b = lookup_or_read_block(index, flags, v);
 
 	if (b) {
-		if (b->ref_count_ && flags & (GF_DIRTY | GF_ZERO))
+		if (b->ref_count_ && (flags & (GF_DIRTY | GF_ZERO))) {
-			throw std::runtime_error("attempt to write lock block concurrently");
+			std::ostringstream out;
+			out << "attempt to write lock block " << index << " concurrently";
+			throw std::runtime_error(out.str());
+		}
 
 		// FIXME: this gets called even for new blocks
 		hit(*b);

block-cache/copier.cc

@@ -0,0 +1,192 @@
+#include "block-cache/copier.h"
+
+#include <stdexcept>
+
+using namespace bcache;
+using namespace boost;
+using namespace std;
+
+//----------------------------------------------------------------
+
+copier::copier(io_engine &engine,
+	       string const &src, string const &dest,
+	       sector_t block_size, size_t mem)
+	: pool_(block_size * 512, mem, PAGE_SIZE),
+	  block_size_(block_size),
+	  nr_blocks_(mem / block_size),
+	  engine_(engine),
+	  src_handle_(engine_.open_file(src, io_engine::M_READ_ONLY)),
+	  dest_handle_(engine_.open_file(dest, io_engine::M_READ_WRITE)),
+	  genkey_count_(0)
+{
+}
+
+copier::~copier()
+{
+	engine_.close_file(src_handle_);
+	engine_.close_file(dest_handle_);
+}
+
+void
+copier::issue(copy_op const &op)
+{
+	void *data;
+
+	while (!(data = pool_.alloc())) {
+		wait_();
+
+		// data may still not be present because the wait_ could
+		// have completed a read and issued the corresponding
+		// write.
+	}
+
+	copy_job job(op, data);
+	job.op.read_complete = job.op.write_complete = false;
+	unsigned key = genkey(); // used as context for the io_engine
+
+	auto r = engine_.issue_io(src_handle_,
+				  io_engine::D_READ,
+				  to_sector(op.src_b),
+				  to_sector(op.src_e),
+				  data,
+				  key);
+
+	if (r)
+		jobs_.insert(make_pair(key, job));
+
+	else
+		complete(job);
+}
+
+unsigned
+copier::nr_pending() const
+{
+	return jobs_.size() + complete_.size();
+}
+
+boost::optional<copy_op>
+copier::wait()
+{
+	if (complete_.empty())
+		wait_();
+
+	return wait_complete();
+}
+
+boost::optional<copy_op>
+copier::wait(unsigned &micro)
+{
+	if (complete_.empty())
+		wait_(micro);
+	return wait_complete();
+}
+
+bool
+copier::pending() const
+{
+	return !jobs_.empty();
+}
+
+boost::optional<copy_op>
+copier::wait_complete()
+{
+	if (complete_.empty()) {
+		return optional<copy_op>();
+
+	} else {
+		auto op = complete_.front();
+		complete_.pop_front();
+		return optional<copy_op>(op);
+	}
+}
+
+void
+copier::wait_(unsigned &micro)
+{
+	optional<io_engine::wait_result> mp;
+
+	if (!pending())
+		return;
+
+
+	bool completed = false;
+	while (pending() && !completed) {
+		mp = engine_.wait(micro);
+		if (mp)
+			completed = wait_successful(*mp);
+
+		if (!micro)
+			break;
+	}
+}
+
+void
+copier::wait_()
+{
+	bool completed = false;
+
+	while (pending() && !completed) {
+		auto mp = engine_.wait();
+		if (mp)
+			completed = wait_successful(*mp);
+	}
+}
+
+bool
+copier::wait_successful(io_engine::wait_result const &p)
+{
+	auto it = jobs_.find(p.second);
+	if (it == jobs_.end())
+		throw runtime_error("Internal error.  Lost track of copy job.");
+
+	copy_job &j = it->second;
+	if (!p.first) {
+		// IO was unsuccessful
+		complete(j);
+		jobs_.erase(it);
+		return true;
+	}
+
+	// IO was successful
+	if (!j.op.read_complete) {
+		j.op.read_complete = true;
+		if (!engine_.issue_io(dest_handle_,
+				      io_engine::D_WRITE,
+				      to_sector(j.op.dest_b),
+				      to_sector(j.op.dest_b + (j.op.src_e - j.op.src_b)),
+				      j.data,
+				      it->first)) {
+			complete(j);
+			jobs_.erase(it);
+			return true;
+		}
+		return false;
+
+	} else {
+		j.op.write_complete = true;
+		complete(j);
+		jobs_.erase(it);
+		return true;
+	}
+}
+
+void
+copier::complete(copy_job const &j)
+{
+	pool_.free(j.data);
+	complete_.push_back(j.op);
+}
+
+sector_t
+copier::to_sector(block_address b) const
+{
+	return b * block_size_;
+}
+
+unsigned
+copier::genkey()
+{
+	return genkey_count_++;
+}
+
+//----------------------------------------------------------------

block-cache/copier.h

@@ -0,0 +1,106 @@
+#ifndef BLOCK_CACHE_COPIER_H
+#define BLOCK_CACHE_COPIER_H
+
+#include "block-cache/io_engine.h"
+#include "block-cache/mem_pool.h"
+
+#include <string>
+#include <list>
+#include <map>
+
+//----------------------------------------------------------------
+
+namespace bcache {
+	using block_address = uint64_t;
+
+	struct copy_op {
+		copy_op()
+			: src_b(0),
+			  src_e(0),
+			  dest_b(0),
+			  read_complete(false),
+			  write_complete(false) {
+		}
+
+		copy_op(block_address src_b_,
+			block_address src_e_,
+			block_address dest_b_)
+			: src_b(src_b_),
+			  src_e(src_e_),
+			  dest_b(dest_b_),
+			  read_complete(false),
+			  write_complete(false) {
+		}
+
+		bool operator <(copy_op const &rhs) const {
+			return dest_b < rhs.dest_b;
+		}
+
+		bool success() const {
+			return read_complete && write_complete;
+		}
+
+		block_address src_b, src_e;
+		block_address dest_b;
+
+		bool read_complete;
+		bool write_complete;
+	};
+
+	class copy_job {
+	public:
+		copy_job(copy_op const &op_, void *data_)
+			: op(op_), data(data_) {
+		}
+
+		copy_op op;
+		void *data;
+	};
+
+	class copier {
+	public:
+		copier(io_engine &engine,
+		       std::string const &src, std::string const &dest,
+		       sector_t block_size, size_t mem);
+		~copier();
+
+		sector_t get_block_size() const {
+			return block_size_;
+		}
+
+		// Blocks if out of memory.
+		void issue(copy_op const &op);
+
+		unsigned nr_pending() const;
+		boost::optional<copy_op> wait();
+		boost::optional<copy_op> wait(unsigned &micro);
+
+	private:
+		bool pending() const;
+		bool wait_successful(io_engine::wait_result const &p);
+		boost::optional<copy_op> wait_complete();
+		void wait_(unsigned &micro);
+		void wait_();
+		void complete(copy_job const &j);
+
+		sector_t to_sector(block_address b) const;
+		unsigned genkey();
+
+		mempool pool_;
+		sector_t block_size_;
+		unsigned nr_blocks_;
+		io_engine &engine_;
+		io_engine::handle src_handle_;
+		io_engine::handle dest_handle_;
+		unsigned genkey_count_;
+
+		using job_map = std::map<unsigned, copy_job>;
+		using op_list = std::list<copy_op>;
+		job_map jobs_;
+		op_list complete_;
+	};
+}
+
+//----------------------------------------------------------------
+
+#endif

block-cache/io_engine.cc

@@ -0,0 +1,199 @@
+#include "base/container_of.h"
+#include "block-cache/io_engine.h"
+
+#include <errno.h>
+#include <fcntl.h>
+#include <sstream>
+#include <stdexcept>
+#include <sys/stat.h>
+#include <sys/types.h>
+
+using namespace bcache;
+using namespace boost;
+using namespace std;
+
+//----------------------------------------------------------------
+
+control_block_set::control_block_set(unsigned nr)
+	: cbs_(nr)
+{
+	for (auto i = 0u; i < nr; i++)
+		free_cbs_.insert(i);
+}
+
+iocb *
+control_block_set::alloc(unsigned context)
+{
+	if (free_cbs_.empty())
+		return nullptr;
+
+	auto it = free_cbs_.begin();
+
+	cblock &cb = cbs_[*it];
+	cb.context = context;
+	free_cbs_.erase(it);
+
+	return &cb.cb;
+}
+
+void
+control_block_set::free(iocb *cb)
+{
+	cblock *b = base::container_of(cb, &cblock::cb);
+	unsigned index = b - &cbs_[0];
+	free_cbs_.insert(index);
+}
+
+unsigned
+control_block_set::context(iocb *cb) const
+{
+	cblock *b = base::container_of(cb, &cblock::cb);
+	return b->context;
+}
+
+//----------------------------------------------------------------
+
+aio_engine::aio_engine(unsigned max_io)
+	: aio_context_(0),
+	  cbs_(max_io)
+{
+	int r = io_setup(max_io, &aio_context_);
+	if (r < 0)
+		throw runtime_error("io_setup failed");
+}
+
+aio_engine::~aio_engine()
+{
+	io_destroy(aio_context_);
+}
+
+aio_engine::handle
+aio_engine::open_file(std::string const &path, mode m, sharing s)
+{
+	int flags = (m == M_READ_ONLY) ? O_RDONLY : O_RDWR;
+	if (s == EXCLUSIVE)
+		flags |= O_EXCL;
+	int fd = ::open(path.c_str(), O_DIRECT | flags);
+	if (fd < 0) {
+		ostringstream out;
+		out << "unable to open '" << path << "'";
+		throw runtime_error(out.str());
+	}
+
+	descriptors_.push_back(base::unique_fd(fd));
+
+	return static_cast<handle>(fd);
+}
+
+void
+aio_engine::close_file(handle h)
+{
+	for (auto it = descriptors_.begin(); it != descriptors_.end(); ++it) {
+		unsigned it_h = it->get();
+		if (it_h == h) {
+			descriptors_.erase(it);
+			return;
+		}
+	}
+
+	ostringstream out;
+	out << "unknown descriptor (" << h << ")";
+	throw runtime_error(out.str());
+}
+
+bool
+aio_engine::issue_io(handle h, dir d, sector_t b, sector_t e, void *data, unsigned context)
+{
+	if (reinterpret_cast<uint64_t>(data) & (PAGE_SIZE - 1))
+		throw runtime_error("Data passed to issue_io must be page aligned\n");
+
+	iocb *cb;
+
+	cb = cbs_.alloc(context);
+	if (!cb)
+		return false;
+
+	memset(cb, 0, sizeof(*cb));
+
+	cb->aio_fildes = static_cast<int>(h);
+	cb->u.c.buf = data;
+	cb->u.c.offset = b << SECTOR_SHIFT;
+	cb->u.c.nbytes = (e - b) << SECTOR_SHIFT;
+	cb->aio_lio_opcode = (d == D_READ) ? IO_CMD_PREAD : IO_CMD_PWRITE;
+
+	int r = io_submit(aio_context_, 1, &cb);
+	return r == 1;
+}
+
+optional<io_engine::wait_result>
+aio_engine::wait()
+{
+	return wait_(NULL);
+}
+
+optional<io_engine::wait_result>
+aio_engine::wait(unsigned &microsec)
+{
+	timespec start = micro_to_ts(microsec);
+	timespec stop = start;
+	auto r = wait_(&stop);
+	microsec = ts_to_micro(stop) - microsec;
+	return r;
+}
+
+boost::optional<io_engine::wait_result>
+aio_engine::wait_(timespec *ts)
+{
+	int r;
+	struct io_event event;
+
+	memset(&event, 0, sizeof(event));
+	r = io_getevents(aio_context_, 1, 1, &event, ts);
+	if (r < 0) {
+		std::ostringstream out;
+		out << "io_getevents failed: " << r;
+		throw std::runtime_error(out.str());
+	}
+
+	if (r == 0) {
+		return optional<wait_result>();
+	}
+
+	iocb *cb = reinterpret_cast<iocb *>(event.obj);
+	unsigned context = cbs_.context(cb);
+
+	if (event.res == cb->u.c.nbytes) {
+		cbs_.free(cb);
+		return optional<wait_result>(make_pair(true, context));
+
+	} else if (static_cast<int>(event.res) < 0) {
+		cbs_.free(cb);
+		return optional<wait_result>(make_pair(false, context));
+
+	} else {
+		cbs_.free(cb);
+		return optional<wait_result>(make_pair(false, context));
+	}
+
+	// shouldn't get here
+	return optional<wait_result>(make_pair(false, 0));
+}
+
+struct timespec
+aio_engine::micro_to_ts(unsigned micro)
+{
+	timespec ts;
+	ts.tv_sec = micro / 1000000u;
+	ts.tv_nsec = (micro % 1000000) * 1000;
+	return ts;
+}
+
+unsigned
+aio_engine::ts_to_micro(timespec const &ts)
+{
+	unsigned micro = ts.tv_sec * 1000000;
+	micro += ts.tv_nsec / 1000;
+	return micro;
+}
+
+//----------------------------------------------------------------

block-cache/io_engine.h

@@ -0,0 +1,117 @@
+#ifndef BLOCK_CACHE_IO_ENGINE_H
+#define BLOCK_CACHE_IO_ENGINE_H
+
+#include "base/unique_handle.h"
+
+#include <boost/optional.hpp>
+#include <ctype.h>
+#include <set>
+#include <string>
+#include <libaio.h>
+
+//----------------------------------------------------------------
+
+namespace bcache {
+	using sector_t = uint64_t;
+
+	unsigned const SECTOR_SHIFT = 9;
+	unsigned const PAGE_SIZE = 4096;
+
+	// Virtual base class to aid unit testing
+	class io_engine {
+	public:
+		enum mode {
+			M_READ_ONLY,
+			M_READ_WRITE
+		};
+
+		enum dir {
+			D_READ,
+			D_WRITE
+		};
+
+		enum sharing {
+			EXCLUSIVE,
+			SHARED
+		};
+
+		io_engine() {}
+		virtual ~io_engine() {}
+
+		using handle = unsigned;
+
+		virtual handle open_file(std::string const &path, mode m, sharing s = EXCLUSIVE) = 0;
+		virtual void close_file(handle h) = 0;
+
+		// returns false if there are insufficient resources to
+		// queue the IO
+		virtual bool issue_io(handle h, dir d, sector_t b, sector_t e, void *data, unsigned context) = 0;
+
+		// returns (success, context)
+		using wait_result = std::pair<bool, unsigned>;
+		virtual boost::optional<wait_result> wait() = 0;
+		virtual boost::optional<wait_result> wait(unsigned &microsec) = 0;
+
+	private:
+		io_engine(io_engine const &) = delete;
+		io_engine &operator =(io_engine const &) = delete;
+	};
+
+	//--------------------------------
+
+	class control_block_set {
+	public:
+		control_block_set(unsigned nr);
+
+		iocb *alloc(unsigned context);
+		void free(iocb *);
+
+		unsigned context(iocb *) const;
+
+	private:
+		struct cblock {
+			unsigned context;
+			struct iocb cb;
+		};
+
+		std::set<unsigned> free_cbs_;
+		std::vector<cblock> cbs_;
+	};
+
+	//----------------
+
+	class aio_engine : public io_engine {
+	public:
+		// max_io is the maximum nr of concurrent ios expected
+		aio_engine(unsigned max_io);
+		~aio_engine();
+
+		using handle = unsigned;
+
+		virtual handle open_file(std::string const &path, mode m, sharing s = EXCLUSIVE);
+		virtual void close_file(handle h);
+
+		// Returns false if queueing the io failed
+		virtual bool issue_io(handle h, dir d, sector_t b, sector_t e, void *data, unsigned context);
+
+		virtual boost::optional<wait_result> wait();
+		virtual boost::optional<wait_result> wait(unsigned &microsec);
+
+	private:
+		static struct timespec micro_to_ts(unsigned micro);
+		static unsigned ts_to_micro(timespec const &ts);
+		boost::optional<io_engine::wait_result> wait_(timespec *ts);
+
+		std::list<base::unique_fd> descriptors_;
+
+		io_context_t aio_context_;
+		control_block_set cbs_;
+
+		aio_engine(io_engine const &) = delete;
+		aio_engine &operator =(io_engine const &) = delete;
+	};
+}
+
+//----------------------------------------------------------------
+
+#endif

block-cache/mem_pool.cc

@@ -0,0 +1,62 @@
+#include "block-cache/mem_pool.h"
+
+#include <sstream>
+#include <stdexcept>
+#include <stdlib.h>
+
+using namespace bcache;
+using namespace boost;
+using namespace mempool_detail;
+using namespace std;
+
+//----------------------------------------------------------------
+
+mempool::mempool(size_t block_size, size_t total_mem, size_t alignment)
+{
+	mem_ = alloc_aligned(total_mem, alignment);
+
+	unsigned nr_blocks = total_mem / block_size;
+	for (auto i = 0u; i < nr_blocks; i++)
+		free(static_cast<unsigned char *>(mem_) + (block_size * i));
+}
+
+mempool::~mempool()
+{
+	free_.clear();
+	::free(mem_);
+}
+
+void *
+mempool::alloc()
+{
+	if (free_.empty())
+		return nullptr;
+
+	mempool_detail::alloc_block &b = free_.front();
+	free_.pop_front();
+	return reinterpret_cast<void *>(&b);
+}
+
+void
+mempool::free(void *data)
+{
+	mempool_detail::alloc_block *b = reinterpret_cast<mempool_detail::alloc_block *>(data);
+	free_.push_front(*b);
+}
+
+void *
+mempool::alloc_aligned(size_t len, size_t alignment)
+{
+	void *result = NULL;
+	int r = posix_memalign(&result, alignment, len);
+	if (r) {
+		ostringstream out;
+		out << "posix_memalign failed: len = " << len << ", alignment = " << alignment << ", r = " << r << "\n";
+		throw runtime_error(out.str());
+	}
+
+	return result;
+}
+
+//----------------------------------------------------------------
+

block-cache/mem_pool.h

@@ -0,0 +1,46 @@
+#ifndef BLOCK_CACHE_MEM_POOL_H
+#define BLOCK_CACHE_MEM_POOL_H
+
+#include <boost/intrusive/list.hpp>
+#include <boost/optional.hpp>
+#include <list>
+
+namespace bi = boost::intrusive;
+
+//----------------------------------------------------------------
+
+namespace bcache {
+	// FIXME: move to base?
+
+	namespace mempool_detail {
+		struct alloc_block : public bi::list_base_hook<bi::link_mode<bi::normal_link>> {
+		};
+	};
+
+	class mempool {
+	public:
+		// alignment must be a power of 2
+		mempool(size_t block_size, size_t total_mem, size_t alignment = 8);
+		~mempool();
+
+		void *alloc();
+		void free(void *data);
+
+	private:
+		static void *alloc_aligned(size_t len, size_t alignment);
+
+		using block_list = bi::list<mempool_detail::alloc_block>;
+
+		void *mem_;
+		block_list free_;
+
+		//----------------
+
+		mempool(mempool const &) = delete;
+		mempool &operator =(mempool const &) = delete;
+	};
+}
+
+//----------------------------------------------------------------
+
+#endif

caching/cache_writeback.cc

@@ -0,0 +1,459 @@
+#include "base/progress_monitor.h"
+#include "persistent-data/file_utils.h"
+#include "block-cache/copier.h"
+#include "caching/commands.h"
+#include "caching/mapping_array.h"
+#include "caching/metadata.h"
+#include "version.h"
+
+#include <boost/optional.hpp>
+#include <getopt.h>
+#include <string>
+#include <stdexcept>
+
+using namespace bcache;
+using namespace caching;
+using namespace boost;
+using namespace std;
+
+//----------------------------------------------------------------
+
+namespace {
+
+	template <typename T> T safe_div(T const n, T const d, T const def) {
+		return (d == T()) ? def : (n / d);
+	}
+
+	//--------------------------------
+
+	struct flags {
+		flags()
+			: cache_size(4 * 1024 * 1024),
+			  sort_buffers(16 * 1024),
+			  list_failed_blocks(false),
+			  update_metadata(true) {
+		}
+
+		// The sort buffers have a dramatic effect on the
+		// performance.  We give up 10% of the general buffer space
+		// for them.
+		void calc_sort_buffer_size() {
+			size_t sbs = cache_size / 10;
+			cache_size = cache_size - sbs;
+
+			sort_buffers = sbs / sizeof(copy_op);
+		}
+
+		using maybe_string = boost::optional<string>;
+
+		size_t cache_size;
+		unsigned sort_buffers;
+		maybe_string metadata_dev;
+		maybe_string origin_dev;
+		maybe_string fast_dev;
+		bool list_failed_blocks;
+		bool update_metadata;
+	};
+
+	//--------------------------------
+
+	class copy_batch {
+	public:
+		copy_batch(unsigned nr)
+		: max_(nr),
+		  count_(0),
+		  ops_(nr) {
+		}
+
+		bool space() const {
+			return count_ < max_;
+		}
+
+		void push_op(copy_op const &op) {
+			if (!space())
+				throw runtime_error("copy_batch out of space");
+
+			ops_[count_++] = op;
+		}
+
+		void reset() {
+			count_ = 0;
+		}
+
+		vector<copy_op>::iterator begin() {
+			return ops_.begin();
+		}
+
+		vector<copy_op>::iterator end() {
+			return ops_.begin() + count_;
+		}
+
+	private:
+		unsigned max_;
+		unsigned count_;
+		vector<copy_op> ops_;
+	};
+
+	class copy_visitor : public mapping_visitor {
+	public:
+		copy_visitor(copier &c, unsigned sort_buffer, bool only_dirty,
+			     bool list_failed_blocks,
+			     progress_monitor &monitor, unsigned cache_blocks)
+			: copier_(c),
+			  block_size_(c.get_block_size()),
+			  only_dirty_(only_dirty),
+			  list_failed_blocks_(list_failed_blocks),
+			  batch_(sort_buffer),
+			  monitor_(monitor),
+			  cache_blocks_(cache_blocks) {
+		}
+
+		virtual void visit(block_address cblock, mapping const &m) {
+			stats_.blocks_scanned = cblock;
+			update_monitor();
+
+			if (!(m.flags_ & M_VALID))
+				return;
+
+			if (only_dirty_ && !(m.flags_ & M_DIRTY))
+				return;
+
+			copy_op cop;
+			cop.src_b = cblock;
+			cop.src_e = cblock + 1ull;
+			cop.dest_b = m.oblock_;
+
+			// blocks
+			stats_.blocks_needed++;
+			batch_.push_op(cop);
+			if (!batch_.space())
+				issue();
+		}
+
+		void issue() {
+			auto compare_dest = [](copy_op const &lhs, copy_op const &rhs) {
+				return lhs.dest_b < rhs.dest_b;
+			};
+			sort(batch_.begin(), batch_.end(), compare_dest);
+
+			auto e = batch_.end();
+			for (auto it = batch_.begin(); it != e; ++it) {
+				copier_.issue(*it);
+				stats_.blocks_issued++;
+				update_monitor();
+
+				check_for_completed_copies();
+			}
+			check_for_completed_copies();
+
+			batch_.reset();
+		}
+
+		void check_for_completed_copies(bool block = false) {
+			optional<copy_op> mop;
+
+			do {
+				if (block)
+					mop = copier_.wait();
+
+				else {
+					unsigned micro = 0;
+					mop = copier_.wait(micro);
+				}
+
+				if (mop) {
+					inc_completed(*mop);
+					if (!mop->success()) {
+						failed_blocks_.insert(*mop);
+						failed_cblocks_.insert(mop->src_b);
+					}
+				}
+
+			} while (mop);
+		}
+
+		void complete() {
+			issue();
+
+			while (copier_.nr_pending())
+				check_for_completed_copies(true);
+
+			monitor_.update_percent(100);
+			cerr << "\n";
+		}
+
+		void inc_completed(copy_op const &op) {
+			stats_.blocks_completed++;
+			update_monitor();
+		}
+
+		void update_monitor() {
+			static unsigned call_count = 0;
+			if (call_count++ % 128)
+				return;
+
+			uint64_t scanned = stats_.blocks_scanned * 100 / cache_blocks_;
+			uint64_t copied = safe_div<block_address>(stats_.blocks_completed * 100,
+								  stats_.blocks_needed, 100ull);
+			uint64_t percent = min<uint64_t>(scanned, copied);
+			monitor_.update_percent(percent);
+		}
+
+		struct copy_stats {
+			copy_stats()
+				: blocks_scanned(0),
+				  blocks_needed(0),
+				  blocks_issued(0),
+				  blocks_completed(0),
+				  blocks_failed(0) {
+			}
+
+			block_address blocks_scanned;
+			block_address blocks_needed;
+			block_address blocks_issued;
+			block_address blocks_completed;
+			block_address blocks_failed;
+		};
+
+		copy_stats const &get_stats() const {
+			return stats_;
+		}
+
+		set<block_address> failed_writebacks() const {
+			return failed_cblocks_;
+		}
+
+	private:
+		copier &copier_;
+		unsigned block_size_;
+		bool only_dirty_;
+		bool list_failed_blocks_;
+
+		copy_stats stats_;
+		copy_batch batch_;
+		progress_monitor &monitor_;
+		unsigned cache_blocks_;
+
+		set<copy_op> failed_blocks_;
+		set<block_address> failed_cblocks_;
+	};
+
+	//--------------------------------
+
+	using namespace mapping_array_damage;
+
+	class ignore_damage_visitor : public damage_visitor {
+	public:
+		ignore_damage_visitor()
+			: corruption_(false) {
+		}
+
+		void visit(missing_mappings const &d) {
+			cerr << "missing mappings (" << d.keys_.begin_ << ", " << d.keys_.end_ << "]\n";
+			corruption_ = true;
+		}
+
+		void visit(invalid_mapping const &d) {
+			cerr << "invalid mapping cblock = " << d.cblock_ << ", oblock = " << d.m_.oblock_ << "\n";
+			corruption_ = true;
+		}
+
+		bool was_corruption() const {
+			return corruption_;
+		}
+
+	private:
+		bool corruption_;
+	};
+
+	bool clean_shutdown(metadata const &md) {
+		return md.sb_.flags.get_flag(superblock_flags::CLEAN_SHUTDOWN);
+	}
+
+	void update_metadata(metadata &md, set<block_address> const &failed_writebacks) {
+		cout << "Updating metadata ... ";
+
+		cout.flush();
+
+		auto &mappings = md.mappings_;
+		for (block_address cblock = 0; cblock < mappings->get_nr_entries(); cblock++) {
+			auto m = mappings->get(cblock);
+			if (!(m.flags_ & M_VALID))
+				continue;
+
+			if (!(m.flags_ & M_DIRTY))
+				continue;
+
+			if (failed_writebacks.count(cblock))
+				continue;
+
+			m.flags_ &= ~M_DIRTY;
+			cerr << "clearing dirty flag for block " << cblock << "\n";
+			mappings->set(cblock, m);
+		}
+		md.commit(true);
+		cout << "done\n";
+		cout.flush();
+	}
+
+	int writeback_(flags const &f) {
+		block_manager<>::ptr bm = open_bm(*f.metadata_dev, block_manager<>::READ_WRITE);
+		metadata md(bm, metadata::OPEN);
+
+		// FIXME: we're going to have to copy runs to get the through put with small block sizes
+		unsigned max_ios = f.cache_size / (md.sb_.data_block_size << SECTOR_SHIFT);
+		aio_engine engine(max_ios);
+		copier c(engine, *f.fast_dev, *f.origin_dev,
+			 md.sb_.data_block_size, f.cache_size);
+
+		auto bar = create_progress_bar("Copying data");
+		copy_visitor cv(c, f.sort_buffers, clean_shutdown(md), f.list_failed_blocks,
+				*bar, md.sb_.cache_blocks);
+
+		ignore_damage_visitor dv;
+
+		walk_mapping_array(*md.mappings_, cv, dv);
+		cv.complete();
+
+		auto stats = cv.get_stats();
+		cout << stats.blocks_issued - stats.blocks_failed << "/"
+		     << stats.blocks_issued << " blocks successfully copied.\n";
+
+		if (stats.blocks_failed)
+			cout << stats.blocks_failed << " blocks were not copied\n";
+
+		if (dv.was_corruption()) {
+			cout << "Metadata corruption was found, some data may not have been copied.\n";
+			if (f.update_metadata)
+				cout << "Unable to update metadata.\n";
+
+		} else if (f.update_metadata)
+			update_metadata(md, cv.failed_writebacks());
+
+		return (stats.blocks_failed || dv.was_corruption()) ? 1 : 0;
+	}
+
+	int writeback(flags const &f) {
+		int r;
+
+		try {
+			r = writeback_(f);
+
+		} catch (std::exception &e) {
+			cerr << e.what() << endl;
+			return 1;
+		}
+
+		return r;
+	}
+}
+
+//----------------------------------------------------------------
+
+cache_writeback_cmd::cache_writeback_cmd()
+	: command("cache_writeback")
+{
+}
+
+void
+cache_writeback_cmd::usage(std::ostream &out) const
+{
+	out << "Usage: " << get_name() << " [options]\n"
+	    << "\t\t--metadata-device <dev>\n"
+	    << "\t\t--origin-device <dev>\n"
+	    << "\t\t--fast-device <dev>\n"
+	    << "\t\t--buffer-size-meg <size>\n"
+	    << "\t\t--list-failed-blocks\n"
+	    << "\t\t--no-metadata-update\n"
+	    << "Options:\n"
+	    << "  {-h|--help}\n"
+	    << "  {-V|--version}" << endl;
+}
+
+int
+cache_writeback_cmd::run(int argc, char **argv)
+{
+	int c;
+	flags fs;
+	char const *short_opts = "hV";
+	option const long_opts[] = {
+		{ "metadata-device", required_argument, NULL, 0 },
+		{ "origin-device", required_argument, NULL, 1 },
+		{ "fast-device", required_argument, NULL, 2 },
+		{ "buffer-size-meg", required_argument, NULL, 3 },
+		{ "list-failed-blocks", no_argument, NULL, 4 },
+		{ "no-metadata-update", no_argument, NULL, 5 },
+		{ "help", no_argument, NULL, 'h'},
+		{ "version", no_argument, NULL, 'V'},
+		{ NULL, no_argument, NULL, 0 }
+	};
+
+	while ((c = getopt_long(argc, argv, short_opts, long_opts, NULL)) != -1) {
+		switch(c) {
+		case 0:
+			fs.metadata_dev = optarg;
+			break;
+
+		case 1:
+			fs.origin_dev = optarg;
+			break;
+
+		case 2:
+			fs.fast_dev = optarg;
+			break;
+
+		case 3:
+			fs.cache_size = parse_uint64(optarg, "buffer size") * 1024 * 1024;
+			break;
+
+		case 4:
+			fs.list_failed_blocks = true;
+			break;
+
+		case 5:
+			fs.update_metadata = false;
+			break;
+
+		case 'h':
+			usage(cout);
+			return 0;
+
+		case 'V':
+			cout << THIN_PROVISIONING_TOOLS_VERSION << endl;
+			return 0;
+
+		default:
+			usage(cerr);
+			return 1;
+		}
+	}
+
+	fs.calc_sort_buffer_size();
+
+	if (argc != optind) {
+		usage(cerr);
+		return 1;
+	}
+
+        if (!fs.metadata_dev) {
+		cerr << "No metadata device provided.\n\n";
+		usage(cerr);
+		return 1;
+	}
+
+	if (!fs.origin_dev) {
+		cerr << "No origin device provided.\n\n";
+		usage(cerr);
+		return 1;
+	}
+
+	if (!fs.fast_dev) {
+		cerr << "No fast device provided.\n\n";
+		usage(cerr);
+		return 1;
+	}
+
+	return writeback(fs);
+}
+
+//----------------------------------------------------------------

caching/commands.cc

@@ -13,6 +13,7 @@ caching::register_cache_commands(application &app)
 	app.add_cmd(command::ptr(new cache_metadata_size_cmd));
 	app.add_cmd(command::ptr(new cache_restore_cmd));
 	app.add_cmd(command::ptr(new cache_repair_cmd));
+	app.add_cmd(command::ptr(new cache_writeback_cmd));
 }
 
 //----------------------------------------------------------------

caching/commands.h

@@ -63,6 +63,13 @@ namespace caching {
 		virtual int run(int argc, char **argv);
 	};
 
+	class cache_writeback_cmd : public base::command {
+	public:
+		cache_writeback_cmd();
+		virtual void usage(std::ostream &out) const;
+		virtual int run(int argc, char **argv);
+	};
+
 	void register_cache_commands(base::application &app);
 }
 

get-gmock.sh

@@ -1,6 +1,3 @@
 #!/bin/sh -e
 
-wget https://googlemock.googlecode.com/files/gmock-1.6.0.zip
+git clone https://github.com/google/googletest
-unzip gmock-1.6.0.zip
-cd gmock-1.6.0
-./configure

man8/cache_writeback.8

@@ -0,0 +1,53 @@
+.TH CACHE_WRITEBACK 8 "Thin Provisioning Tools" "Red Hat, Inc." \" -*- nroff -*-
+.SH NAME
+cache_writeback \- writeback dirty blocks to the origin device.
+
+.SH SYNOPSIS
+.B cache_writeback
+.RB [ options ]
+.RB --metadata-device
+.I {device|file}
+.RB --origin-device
+.I {device|file}
+.RB --fast-device
+.I {device|file}
+
+.SH DESCRIPTION
+.B cache_writeback
+
+An offline tool that writesback dirty data to the data device
+(origin).  Intended for use in recovery scenarios when the SSD is
+giving IO errors.
+
+This tool cannot be run on a live cache.
+
+.SH OPTIONS
+
+.IP "\fB\\-\-metadata\-device\fP \fI{device|file}\fP"
+Location of cache metadata.
+
+.IP "\fB\-\-origin\-device\fP \fI{device|file}\fP"
+Slow device being cached.
+
+.IP "\fB\-\-fast\-device\fP \fI{device|file}\fP"
+Fast device containing the data that needs to be written back.
+
+.IP "\fB\-\-skip\-metadata\-update\fP"
+Do not update the metadata to clear the dirty flags for written back
+data.  You may not want to do this if you're decommissioning the
+cache.
+
+.IP "\fB\-h, \-\-help\fP"
+Print help and exit.
+
+.IP "\fB\-V, \-\-version\fP"
+Output version information and exit.
+
+.SH SEE ALSO
+.B cache_dump(8)
+.B cache_check(8)
+.B cache_repair(8)
+.B cache_restore(8)
+
+.SH AUTHOR
+Joe Thornber <ejt@redhat.com>

persistent-data/block.tcc

@@ -19,6 +19,7 @@
 #include "block.h"
 
 #include "base/error_string.h"
+#include "block-cache/io_engine.h"
 
 #include <errno.h>
 #include <fcntl.h>
@@ -38,8 +39,6 @@ namespace {
 	using namespace std;
 
 	int const DEFAULT_MODE = 0666;
-	unsigned const SECTOR_SHIFT = 9;
-
 	int const OPEN_FLAGS = O_DIRECT;
 
 	// FIXME: introduce a new exception for this, or at least lift this

thin-provisioning/commands.h

@@ -66,6 +66,7 @@ namespace thin_provisioning {
 	class thin_trim_cmd : public base::command {
 	public:
 		thin_trim_cmd();
+
 		virtual void usage(std::ostream &out) const;
 		virtual int run(int argc, char **argv);
 	};
@@ -74,6 +75,7 @@ namespace thin_provisioning {
 	class thin_ll_dump_cmd : public base::command {
 	public:
 		thin_ll_dump_cmd();
+
 		virtual void usage(std::ostream &out) const;
 		virtual int run(int argc, char **argv);
 	};
@@ -81,6 +83,7 @@ namespace thin_provisioning {
 	class thin_ll_restore_cmd : public base::command {
 	public:
 		thin_ll_restore_cmd();
+
 		virtual void usage(std::ostream &out) const;
 		virtual int run(int argc, char **argv);
 	};
@@ -88,6 +91,7 @@ namespace thin_provisioning {
 	class thin_scan_cmd : public base::command {
 	public:
 		thin_scan_cmd();
+
 		virtual void usage(std::ostream &out) const;
 		virtual int run(int argc, char **argv);
 	};

unit-tests/Makefile.in

@@ -16,10 +16,10 @@
 # with thin-provisioning-tools.  If not, see
 # <http://www.gnu.org/licenses/>.
 
-GMOCK_DIR=gmock-1.6.0/
+GMOCK_DIR=googletest
 GMOCK_INCLUDES=\
-	-Igmock-1.6.0/include \
+	-I$(GMOCK_DIR)/googlemock/include \
-	-Igmock-1.6.0/gtest/include
+	-I$(GMOCK_DIR)/googletest/include
 
 GMOCK_FLAGS=\
 	-Wno-unused-local-typedefs
@@ -28,16 +28,16 @@ GMOCK_LIBS=\
 	-Llib -lpdata -lgmock -lpthread -laio
 
 GMOCK_DEPS=\
-	$(wildcard $(GMOCK_DIR)/include/*.h) \
+	$(wildcard $(GMOCK_DIR)/googlemock/include/*.h) \
-	$(wildcard $(GMOCK_DIR)/src/*.cc) \
+	$(wildcard $(GMOCK_DIR)/googlemock/src/*.cc) \
-	$(wildcard $(GMOCK_DIR)/gtest/include/*.h) \
+	$(wildcard $(GMOCK_DIR)/googletest/include/*.h) \
-	$(wildcard $(GMOCK_DIR)/gtest/src/*.cc)
+	$(wildcard $(GMOCK_DIR)/googletest/src/*.cc)
 
 lib/libgmock.a: $(GMOCK_DEPS)
 	@echo "    [CXX] gtest"
-	$(V)g++ $(GMOCK_INCLUDES) -I$(GMOCK_DIR)/gtest -c $(GMOCK_DIR)/gtest/src/gtest-all.cc
+	$(V)g++ $(GMOCK_INCLUDES) -I$(GMOCK_DIR)/googletest -c $(GMOCK_DIR)/googletest/src/gtest-all.cc
 	@echo "    [CXX] gmock"
-	$(V)g++ $(GMOCK_INCLUDES) -I$(GMOCK_DIR) -c $(GMOCK_DIR)/src/gmock-all.cc
+	$(V)g++ $(GMOCK_INCLUDES) -I$(GMOCK_DIR)/googlemock -c $(GMOCK_DIR)/googlemock/src/gmock-all.cc
 	@echo "    [AR]  $<"
 	$(V)ar -rv lib/libgmock.a gtest-all.o gmock-all.o > /dev/null 2>&1
 
@@ -55,9 +55,12 @@ TEST_SOURCE=\
 	unit-tests/btree_counter_t.cc \
 	unit-tests/btree_damage_visitor_t.cc \
 	unit-tests/cache_superblock_t.cc \
+	unit-tests/copier_t.cc \
 	unit-tests/damage_tracker_t.cc \
 	unit-tests/endian_t.cc \
 	unit-tests/error_state_t.cc \
+	unit-tests/io_engine_t.cc \
+	unit-tests/mem_pool_t.cc \
 	unit-tests/rmap_visitor_t.cc \
 	unit-tests/rolling_hash_t.cc \
 	unit-tests/run_set_t.cc \

unit-tests/block_t.cc

@@ -64,6 +64,7 @@ namespace {
 		typedef boost::shared_ptr<validator_mock> ptr;
 
 		MOCK_CONST_METHOD2(check, void(void const *, block_address));
+		MOCK_CONST_METHOD1(check_raw, bool(void const *data));
 		MOCK_CONST_METHOD2(prepare, void(void *, block_address));
 	};
 

unit-tests/copier_t.cc

@@ -0,0 +1,273 @@
+// Copyright (C) 2016 Red Hat, Inc. All rights reserved.
+//
+// This file is part of the thin-provisioning-tools source.
+//
+// thin-provisioning-tools is free software: you can redistribute it
+// and/or modify it under the terms of the GNU General Public License
+// as published by the Free Software Foundation, either version 3 of
+// the License, or (at your option) any later version.
+//
+// thin-provisioning-tools is distributed in the hope that it will be
+// useful, but WITHOUT ANY WARRANTY; without even the implied warranty
+// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along
+// with thin-provisioning-tools.  If not, see
+// <http://www.gnu.org/licenses/>.
+
+#include "gmock/gmock.h"
+#include "block-cache/copier.h"
+#include "test_utils.h"
+
+#include <fcntl.h>
+
+using namespace boost;
+using namespace std;
+using namespace test;
+using namespace testing;
+
+//----------------------------------------------------------------
+
+namespace {
+	unsigned const BLOCK_SIZE = 64u;
+	using wait_result = io_engine::wait_result;
+
+	ostream &operator <<(ostream &out, wait_result const &wr) {
+		out << "wait_result[" << wr.first << ", " << wr.second << "]";
+		return out;
+	}
+
+	ostream &operator <<(ostream &out, optional<wait_result> const &mwr) {
+		if (mwr) {
+			out << "Just[wait_result[" << mwr->first << ", " << mwr->second << "]]";
+		} else
+			out << "Nothing";
+		return out;
+	}
+
+	class io_engine_mock : public io_engine {
+	public:
+		MOCK_METHOD3(open_file, handle(string const &, mode, sharing));
+		MOCK_METHOD1(close_file, void(handle));
+		MOCK_METHOD6(issue_io, bool(handle, dir, sector_t, sector_t, void *, unsigned));
+
+		MOCK_METHOD0(wait, optional<wait_result>());
+		MOCK_METHOD1(wait, optional<wait_result>(unsigned &));
+	};
+
+	class CopierTests : public Test {
+	public:
+		CopierTests()
+			: src_file_("copy_src"),
+			  dest_file_("copy_dest") {
+		}
+
+		unique_ptr<copier> make_copier() {
+			EXPECT_CALL(engine_, open_file(src_file_, io_engine::M_READ_ONLY, io_engine::EXCLUSIVE)).
+				WillOnce(Return(SRC_HANDLE));
+			EXPECT_CALL(engine_, open_file(dest_file_, io_engine::M_READ_WRITE, io_engine::EXCLUSIVE)).
+				WillOnce(Return(DEST_HANDLE));
+
+			EXPECT_CALL(engine_, close_file(SRC_HANDLE)).Times(1);
+			EXPECT_CALL(engine_, close_file(DEST_HANDLE)).Times(1);
+
+			return unique_ptr<copier>(new copier(engine_, src_file_,
+							     dest_file_,
+							     BLOCK_SIZE, 1 * 1024 * 1024));
+		}
+
+		static optional<wait_result> make_wr(bool success, unsigned context) {
+			return optional<wait_result>(wait_result(success, context));
+		}
+
+		void issue_successful_op(copier &c, copy_op &op, unsigned context) {
+			InSequence dummy;
+
+			unsigned nr_pending = c.nr_pending();
+			EXPECT_CALL(engine_, issue_io(SRC_HANDLE, io_engine::D_READ,
+						      op.src_b * BLOCK_SIZE,
+						      op.src_e * BLOCK_SIZE, _, context)).
+				WillOnce(Return(true));
+			c.issue(op);
+
+			ASSERT_TRUE(c.nr_pending() == nr_pending + 1);
+
+			EXPECT_CALL(engine_, wait()).
+				WillOnce(Return(make_wr(true, context)));
+
+			EXPECT_CALL(engine_, issue_io(DEST_HANDLE, io_engine::D_WRITE,
+						      op.dest_b * BLOCK_SIZE,
+						      (op.dest_b + (op.src_e - op.src_b)) * BLOCK_SIZE, _, context)).
+				WillOnce(Return(true));
+
+			EXPECT_CALL(engine_, wait()).
+				WillOnce(Return(make_wr(true, context)));
+
+			auto mop = c.wait();
+			ASSERT_EQ(c.nr_pending(), nr_pending);
+
+			ASSERT_TRUE(mop->success());
+		}
+
+		unsigned const SRC_HANDLE = 10;
+		unsigned const DEST_HANDLE = 11;
+
+		string src_file_;
+		string dest_file_;
+		StrictMock<io_engine_mock> engine_;
+	};
+}
+
+//----------------------------------------------------------------
+
+TEST_F(CopierTests, empty_test)
+{
+	auto c = make_copier();
+}
+
+TEST_F(CopierTests, successful_copy)
+{
+ 	// Copy first block
+	copy_op op1(0, 1, 0);
+
+	auto c = make_copier();
+	issue_successful_op(*c, op1, 0);
+}
+
+TEST_F(CopierTests, unsuccessful_issue_read)
+{
+	copy_op op1(0, 1, 0);
+	auto c = make_copier();
+
+	InSequence dummy;
+	EXPECT_CALL(engine_, issue_io(SRC_HANDLE, io_engine::D_READ, 0, BLOCK_SIZE, _, 0)).
+		WillOnce(Return(false));
+	c->issue(op1);
+
+	ASSERT_EQ(c->nr_pending(), 1u);
+
+	auto mop = c->wait();
+	ASSERT_EQ(c->nr_pending(), 0u);
+	ASSERT_FALSE(mop->success());
+}
+
+TEST_F(CopierTests, unsuccessful_read)
+{
+	copy_op op1(0, 1, 0);
+	auto c = make_copier();
+
+	InSequence dummy;
+	EXPECT_CALL(engine_, issue_io(SRC_HANDLE, io_engine::D_READ, 0, BLOCK_SIZE, _, 0)).
+		WillOnce(Return(true));
+	c->issue(op1);
+
+	ASSERT_EQ(c->nr_pending(), 1u);
+	EXPECT_CALL(engine_, wait()).
+		WillOnce(Return(make_wr(false, 0u)));
+	ASSERT_EQ(c->nr_pending(), 1u);
+
+	auto mop = c->wait();
+	ASSERT_EQ(c->nr_pending(), 0u);
+	ASSERT_FALSE(mop->success());
+}
+
+TEST_F(CopierTests, unsuccessful_issue_write)
+{
+	copy_op op1(0, 1, 0);
+	auto c = make_copier();
+
+	InSequence dummy;
+	EXPECT_CALL(engine_, issue_io(SRC_HANDLE, io_engine::D_READ, 0, BLOCK_SIZE, _, 0)).
+		WillOnce(Return(true));
+	c->issue(op1);
+
+	ASSERT_EQ(c->nr_pending(), 1u);
+
+	EXPECT_CALL(engine_, wait()).
+		WillOnce(Return(make_wr(true, 0u)));
+	ASSERT_EQ(c->nr_pending(), 1u);
+
+	EXPECT_CALL(engine_, issue_io(DEST_HANDLE, io_engine::D_WRITE, 0, BLOCK_SIZE, _, 0)).
+		WillOnce(Return(false));
+
+	auto mop = c->wait();
+	ASSERT_EQ(c->nr_pending(), 0u);
+	ASSERT_FALSE(mop->success());
+}
+
+TEST_F(CopierTests, unsuccessful_write)
+{
+	// Copy first block
+	copy_op op1(0, 1, 0);
+
+	auto c = make_copier();
+
+	InSequence dummy;
+	EXPECT_CALL(engine_, issue_io(SRC_HANDLE, io_engine::D_READ, 0, BLOCK_SIZE, _, 0)).
+		WillOnce(Return(true));
+	c->issue(op1);
+	ASSERT_EQ(c->nr_pending(), 1u);
+
+	EXPECT_CALL(engine_, wait()).
+		WillOnce(Return(make_wr(true, 0u)));
+
+	EXPECT_CALL(engine_, issue_io(DEST_HANDLE, io_engine::D_WRITE, 0, BLOCK_SIZE, _, 0)).
+		WillOnce(Return(true));
+
+	EXPECT_CALL(engine_, wait()).
+		WillOnce(Return(make_wr(false, 0u)));
+
+	auto mop = c->wait();
+	ASSERT_EQ(c->nr_pending(), 0u);
+
+	ASSERT_FALSE(mop->success());
+}
+
+TEST_F(CopierTests, copy_the_same_block_many_times)
+{
+	auto c = make_copier();
+	copy_op op1(0, 1, 0);
+
+	for (unsigned i = 0; i < 50000; i++)
+		issue_successful_op(*c, op1, i);
+}
+
+TEST_F(CopierTests, copy_different_blocks)
+{
+	auto c = make_copier();
+	for (unsigned i = 0; i < 5000; i++) {
+		copy_op op(i, i + 1, i);
+		issue_successful_op(*c, op, i);
+	}
+}
+
+TEST_F(CopierTests, wait_can_timeout)
+{
+	copy_op op1(0, 1, 0);
+	auto c = make_copier();
+
+	InSequence dummy;
+	EXPECT_CALL(engine_, issue_io(SRC_HANDLE, io_engine::D_READ, 0, BLOCK_SIZE, _, 0)).
+		WillOnce(Return(true));
+	c->issue(op1);
+
+	ASSERT_EQ(c->nr_pending(), 1u);
+
+	unsigned micro = 10000;
+	EXPECT_CALL(engine_, wait(micro)).
+		WillOnce(Return(make_wr(true, 0u)));
+	ASSERT_EQ(c->nr_pending(), 1u);
+
+	EXPECT_CALL(engine_, issue_io(DEST_HANDLE, io_engine::D_WRITE, 0, BLOCK_SIZE, _, 0)).
+		WillOnce(Return(true));
+
+	EXPECT_CALL(engine_, wait(micro)).
+		WillOnce(DoAll(SetArgReferee<0>(0u), Return(optional<wait_result>())));
+
+	auto mop = c->wait(micro);
+	ASSERT_FALSE(mop);
+	ASSERT_EQ(c->nr_pending(), 1u);
+}
+
+//----------------------------------------------------------------

unit-tests/io_engine_t.cc

@@ -0,0 +1,206 @@
+// Copyright (C) 2016 Red Hat, Inc. All rights reserved.
+//
+// This file is part of the thin-provisioning-tools source.
+//
+// thin-provisioning-tools is free software: you can redistribute it
+// and/or modify it under the terms of the GNU General Public License
+// as published by the Free Software Foundation, either version 3 of
+// the License, or (at your option) any later version.
+//
+// thin-provisioning-tools is distributed in the hope that it will be
+// useful, but WITHOUT ANY WARRANTY; without even the implied warranty
+// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along
+// with thin-provisioning-tools.  If not, see
+// <http://www.gnu.org/licenses/>.
+
+#include "gmock/gmock.h"
+#include "block-cache/mem_pool.h"
+#include "block-cache/io_engine.h"
+#include "test_utils.h"
+
+
+#include <fcntl.h>
+
+using namespace boost;
+using namespace std;
+using namespace test;
+using namespace testing;
+
+//----------------------------------------------------------------
+
+namespace {
+	unsigned const MAX_IO = 64;
+	unsigned const PAGE_SIZE = 4096;
+
+	class IOEngineTests : public Test {
+	public:
+		IOEngineTests()
+			: pool_(64 * 512, 128 * 512, PAGE_SIZE),
+			  src_file_("copy_src", 32),
+			  dest_file_("copy_dest", 32),
+			  engine_(new aio_engine(MAX_IO)) {
+		}
+
+		// in sectors
+		static uint64_t meg(unsigned n) {
+			return 2 * 1024 * n;
+		}
+
+		mempool pool_;
+		temp_file src_file_;
+		temp_file dest_file_;
+		unique_ptr<io_engine> engine_;
+	};
+}
+
+//----------------------------------------------------------------
+
+TEST_F(IOEngineTests, empty_test)
+{
+}
+
+TEST_F(IOEngineTests, open_and_close)
+{
+	auto src_handle = engine_->open_file(src_file_.get_path(), io_engine::M_READ_ONLY);
+	auto dest_handle = engine_->open_file(dest_file_.get_path(), io_engine::M_READ_WRITE);
+	ASSERT_TRUE(src_handle != dest_handle);
+	engine_->close_file(src_handle);
+	engine_->close_file(dest_handle);
+}
+
+TEST_F(IOEngineTests, you_can_read_a_read_only_handle)
+{
+	unsigned nr_sectors = 8;
+	auto src_handle = engine_->open_file(src_file_.get_path(), io_engine::M_READ_ONLY);
+	void *data = pool_.alloc();
+	bool r = engine_->issue_io(src_handle,
+				   io_engine::D_READ,
+				   0, nr_sectors,
+				   data,
+				   123);
+	ASSERT_TRUE(r);
+	auto wr = engine_->wait();
+	ASSERT_TRUE(wr->first);
+	ASSERT_TRUE(wr->second == 123);
+
+	engine_->close_file(src_handle);
+	pool_.free(data);
+}
+
+
+TEST_F(IOEngineTests, you_cannot_write_to_a_read_only_handle)
+{
+	unsigned nr_sectors = 8;
+	auto src_handle = engine_->open_file(src_file_.get_path(), io_engine::M_READ_ONLY);
+	void *data = pool_.alloc();
+	bool r = engine_->issue_io(src_handle,
+				   io_engine::D_WRITE,
+				   0, nr_sectors,
+				   data,
+				   0);
+	ASSERT_FALSE(r);
+	engine_->close_file(src_handle);
+	pool_.free(data);
+}
+
+TEST_F(IOEngineTests, you_can_write_to_a_read_write_handle)
+{
+	unsigned nr_sectors = 8;
+	auto src_handle = engine_->open_file(src_file_.get_path(), io_engine::M_READ_ONLY);
+	void *data = pool_.alloc();
+	bool r = engine_->issue_io(src_handle,
+				   io_engine::D_READ,
+				   0, nr_sectors,
+				   data,
+				   123);
+	ASSERT_TRUE(r);
+	auto wr = engine_->wait();
+	ASSERT_TRUE(wr->first);
+	ASSERT_TRUE(wr->second == 123);
+
+	engine_->close_file(src_handle);
+	pool_.free(data);
+}
+
+TEST_F(IOEngineTests, final_block_read_succeeds)
+{
+	unsigned nr_sectors = 8;
+	auto src_handle = engine_->open_file(src_file_.get_path(), io_engine::M_READ_ONLY);
+	void *data = pool_.alloc();
+	bool r = engine_->issue_io(src_handle,
+				   io_engine::D_READ,
+				   meg(32) - nr_sectors, meg(32),
+				   data,
+				   123);
+	ASSERT_TRUE(r);
+	auto wr = engine_->wait();
+	ASSERT_TRUE(wr->first);
+
+	engine_->close_file(src_handle);
+	pool_.free(data);
+
+}
+
+TEST_F(IOEngineTests, out_of_bounds_read_fails)
+{
+	unsigned nr_sectors = 8;
+	auto src_handle = engine_->open_file(src_file_.get_path(), io_engine::M_READ_ONLY);
+	void *data = pool_.alloc();
+	bool r = engine_->issue_io(src_handle,
+				   io_engine::D_READ,
+				   meg(32), meg(32) + nr_sectors,
+				   data,
+				   123);
+	ASSERT_TRUE(r);
+	auto wr = engine_->wait();
+	ASSERT_FALSE(wr->first);
+
+	engine_->close_file(src_handle);
+	pool_.free(data);
+
+}
+
+TEST_F(IOEngineTests, out_of_bounds_write_succeeds)
+{
+	unsigned nr_sectors = 8;
+	auto handle = engine_->open_file(dest_file_.get_path(), io_engine::M_READ_WRITE);
+	void *data = pool_.alloc();
+	bool r = engine_->issue_io(handle,
+				   io_engine::D_WRITE,
+				   meg(32), meg(32) + nr_sectors,
+				   data,
+				   123);
+	ASSERT_TRUE(r);
+	auto wr = engine_->wait();
+	ASSERT_TRUE(wr->first);
+
+	engine_->close_file(handle);
+	pool_.free(data);
+
+}
+
+TEST_F(IOEngineTests, succeed_with_timeout)
+{
+	unsigned nr_sectors = 8;
+	auto src_handle = engine_->open_file(src_file_.get_path(), io_engine::M_READ_ONLY);
+	void *data = pool_.alloc();
+	bool r = engine_->issue_io(src_handle,
+				   io_engine::D_READ,
+				   0, nr_sectors,
+				   data,
+				   123);
+	ASSERT_TRUE(r);
+	unsigned micro = 10;
+	auto wr = engine_->wait(micro);
+	ASSERT_TRUE(wr->first);
+	ASSERT_TRUE(wr->second == 123);
+
+	engine_->close_file(src_handle);
+	pool_.free(data);
+}
+
+
+//----------------------------------------------------------------

unit-tests/mem_pool_t.cc

@@ -0,0 +1,112 @@
+// Copyright (C) 2016 Red Hat, Inc. All rights reserved.
+//
+// This file is part of the thin-provisioning-tools source.
+//
+// thin-provisioning-tools is free software: you can redistribute it
+// and/or modify it under the terms of the GNU General Public License
+// as published by the Free Software Foundation, either version 3 of
+// the License, or (at your option) any later version.
+//
+// thin-provisioning-tools is distributed in the hope that it will be
+// useful, but WITHOUT ANY WARRANTY; without even the implied warranty
+// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along
+// with thin-provisioning-tools.  If not, see
+// <http://www.gnu.org/licenses/>.
+
+#include "gmock/gmock.h"
+#include "block-cache/mem_pool.h"
+#include "test_utils.h"
+
+#include <fcntl.h>
+
+using namespace boost;
+using namespace std;
+using namespace test;
+using namespace testing;
+
+//----------------------------------------------------------------
+
+namespace {
+	class MempoolTests : public Test {
+	public:
+		bool aligned(void *data, size_t alignment) {
+			return (reinterpret_cast<size_t>(data) % alignment) == 0;
+		}
+
+	private:
+
+	};
+}
+
+//----------------------------------------------------------------
+
+TEST_F(MempoolTests, empty_test)
+{
+}
+
+TEST_F(MempoolTests, create_destroy_cycle)
+{
+	for (size_t bs = 64; bs <= 512; bs *= 2) {
+		mempool mp(bs, 4 * 1024 * 1024, bs);
+	}
+}
+
+TEST_F(MempoolTests, alignments_observed)
+{
+	for (size_t bs = 64; bs <= 512; bs *= 2) {
+		mempool mp(bs, 512 * 1024, bs);
+
+		for (unsigned i = 0; i < 100; i++) {
+			auto md = mp.alloc();
+
+			if (!md)
+				throw runtime_error("couldn't alloc");
+
+			ASSERT_TRUE(aligned(md, bs));
+		}
+	}
+}
+
+TEST_F(MempoolTests, alloc_free_cycle)
+{
+	mempool mp(512, 512 * 1024, 512);
+
+	for (unsigned i = 0; i < 10000; i++) {
+		auto md = mp.alloc();
+		mp.free(md);
+	}
+}
+
+TEST_F(MempoolTests, exhaust_pool)
+{
+	mempool mp(512, 100 * 512, 512);
+
+	for (unsigned i = 0; i < 100; i++) {
+		auto md = mp.alloc();
+		ASSERT_NE(md, nullptr);
+	}
+
+	auto md = mp.alloc();
+	ASSERT_EQ(md, nullptr);
+}
+
+// Use valgrind
+TEST_F(MempoolTests, data_can_be_written)
+{
+	mempool mp(512, 100 * 512, 512);
+
+	for (unsigned i = 0; i < 100; i++) {
+		auto md = mp.alloc();
+		ASSERT_NE(md, nullptr);
+
+		memset(md, 0, 512);
+	}
+
+	auto md = mp.alloc();
+	ASSERT_EQ(md, nullptr);
+}
+
+//----------------------------------------------------------------

unit-tests/test_utils.cc

@@ -3,6 +3,8 @@
 #include "persistent-data/space-maps/core.h"
 
 using namespace persistent_data;
+using namespace std;
+using namespace test;
 
 //----------------------------------------------------------------
 
@@ -21,3 +23,35 @@ test::open_temporary_tm(block_manager<>::ptr bm)
 }
 
 //----------------------------------------------------------------
+
+temp_file::temp_file(string const &name_base, unsigned meg_size)
+	: path_(gen_path(name_base))
+{
+	int fd = ::open(path_.c_str(), O_CREAT | O_RDWR, 0666);
+	if (fd < 0)
+		throw runtime_error("couldn't open file");
+
+	if (::fallocate(fd, 0, 0, 1024 * 1024 * meg_size))
+		throw runtime_error("couldn't fallocate");
+
+	::close(fd);
+}
+
+temp_file::~temp_file()
+{
+//	::unlink(path_.c_str());
+}
+
+string const &
+temp_file::get_path() const
+{
+	return path_;
+}
+
+string
+temp_file::gen_path(string const &base)
+{
+	return string("./") + base + string(".tmp");
+}
+
+//----------------------------------------------------------------

unit-tests/test_utils.h

@@ -114,6 +114,18 @@ namespace test {
 		std::unique_ptr<with_directory> dir_;
 	};
 
+	class temp_file {
+	public:
+		temp_file(std::string const &name_base, unsigned meg_size);
+		~temp_file();
+		std::string const &get_path() const;
+
+	private:
+		static string gen_path(string const &base);
+
+		string path_;
+	};
+
 	//--------------------------------
 
 	template <typename T>