Draft: Resolve "Add compression on demand on data" (!277) · Merge requests · hpc / gekkofs

include/client/rpc/forward_data.hpp

+10 −0

Changes for include/client/rpc/forward_data.hpp: 10 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -59,6 +59,16 @@ forward_read(const std::string& path, void* buf, off64_t offset,
		size_t read_size, const int8_t num_copies,
		std::set<int8_t>& failed);

		std::pair<int, ssize_t>
		forward_write_compressed(const std::string& path, const void* buf,
		off64_t offset, size_t write_size,
		const int8_t num_copy = 0);

		std::pair<int, ssize_t>
		forward_read_compressed(const std::string& path, void* buf, off64_t offset,
		size_t read_size, const int8_t num_copies,
		std::set<int8_t>& failed);

		int
		forward_truncate(const std::string& path, size_t current_size, size_t new_size,
		const int8_t num_copies);

Conflict: This file was modified in both the source and target branches. Ask someone with write access to resolve it.

include/client/env.hpp

+6 −0

Changes for include/client/env.hpp: 6 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -77,9 +77,15 @@ static constexpr auto RANGE_FD = ADD_PREFIX("RANGE_FD");
		static constexpr auto DIRENTS_BUFF_SIZE = ADD_PREFIX("DIRENTS_BUFF_SIZE");
		static constexpr auto USE_DIRENTS_COMPRESSION =
		ADD_PREFIX("USE_DIRENTS_COMPRESSION");
		static constexpr auto DATA_COMPRESSION = ADD_PREFIX("DATA_COMPRESSION");
		static constexpr auto DATA_DISTRIBUTION_HOSTS =
		ADD_PREFIX("DATA_DISTRIBUTION_HOSTS");


		static constexpr auto NUM_REPL = ADD_PREFIX("NUM_REPL");
		static constexpr auto PROXY_PID_FILE = ADD_PREFIX("PROXY_PID_FILE");
		static constexpr auto FUZZY_DIRECT_READ = ADD_PREFIX("FUZZY_DIRECT_READ");
		static constexpr auto DAEMON_ROOTDIR = ADD_PREFIX("DAEMON_ROOTDIR");
		namespace cache {
		static constexpr auto DENTRY = ADD_PREFIX("DENTRY_CACHE");
		static constexpr auto WRITE_SIZE = ADD_PREFIX("WRITE_SIZE_CACHE");

Conflict: This file was modified in both the source and target branches. Ask someone with write access to resolve it.

include/client/preload_context.hpp

+7 −0

Changes for include/client/preload_context.hpp: 7 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -138,6 +138,7 @@ private:
		std::bitset<MAX_USER_FDS> protected_fds_;
		std::string hostname;
		int replicas_;
		bool use_compression_{false};

		bool protect_fds_{false};
		bool protect_files_generator_{false};
		@@ -326,6 +327,12 @@ public:
		int
		get_replicas();

		bool
		use_compression() const;

		void
		enable_compression(bool enable);

		bool
		protect_fds() const;

Conflict: This file was modified in both the source and target branches. Ask someone with write access to resolve it.

include/common/rpc/distributor.hpp

+38 −3

Changes for include/common/rpc/distributor.hpp: 38 added lines, 3 removed lines.
Original line number	Diff line number	Diff line
		@@ -60,8 +60,6 @@ public:
		virtual host_t
		locate_data(const std::string& path, const chunkid_t& chnk_id,
		const int num_copy) const = 0;
		// TODO: We need to pass hosts_size in the server side, because the number
		// of servers are not defined (in startup)

		virtual unsigned int
		hosts_size() const = 0;
		@@ -69,6 +67,12 @@ public:
		virtual void
		hosts_size(unsigned int size) = 0;

		virtual void
		data_hosts_size(unsigned int size) = 0;

		virtual unsigned int
		data_hosts_size() const = 0;

		virtual host_t
		locate_data(const std::string& path, const chunkid_t& chnk_id,
		unsigned int hosts_size, const int num_copy) = 0;
		@@ -85,6 +89,7 @@ class SimpleHashDistributor : public Distributor {
		private:
		host_t localhost_;
		unsigned int hosts_size_{0};
		unsigned int data_hosts_size_{0};
		std::vector<host_t> all_hosts_;
		std::hash<std::string> str_hash;

		@@ -93,12 +98,21 @@ public:

		SimpleHashDistributor(host_t localhost, unsigned int hosts_size);

		SimpleHashDistributor(host_t localhost, unsigned int hosts_size,
		unsigned int data_hosts_size);

		unsigned int
		hosts_size() const override;

		void
		hosts_size(unsigned int size) override;

		unsigned int
		data_hosts_size() const override;

		void
		data_hosts_size(unsigned int size) override;

		host_t
		localhost() const override;

		@@ -108,7 +122,7 @@ public:

		host_t
		locate_data(const std::string& path, const chunkid_t& chnk_id,
		unsigned int host_size, const int num_copy);
		unsigned int host_size, const int num_copy) override;

		host_t
		locate_file_metadata(const std::string& path,
		@@ -122,6 +136,7 @@ class LocalOnlyDistributor : public Distributor {
		private:
		host_t localhost_;
		unsigned int hosts_size_{0};
		unsigned int data_hosts_size_{0};

		public:
		explicit LocalOnlyDistributor(host_t localhost);
		@@ -135,6 +150,12 @@ public:
		void
		hosts_size(unsigned int size) override;

		unsigned int
		data_hosts_size() const override;

		void
		data_hosts_size(unsigned int size) override;

		host_t
		locate_data(const std::string& path, const chunkid_t& chnk_id,
		const int num_copy) const override;
		@@ -155,6 +176,7 @@ class ForwarderDistributor : public Distributor {
		private:
		host_t fwd_host_;
		unsigned int hosts_size_{0};
		unsigned int data_hosts_size_{0};
		std::vector<host_t> all_hosts_;
		std::hash<std::string> str_hash;

		@@ -170,6 +192,12 @@ public:
		void
		hosts_size(unsigned int size) override;

		void
		data_hosts_size(unsigned int size) override;

		unsigned int
		data_hosts_size() const override;

		host_t
		locate_data(const std::string& path, const chunkid_t& chnk_id,
		const int num_copy) const override final;
		@@ -204,6 +232,7 @@ class GuidedDistributor : public Distributor {
		private:
		host_t localhost_;
		unsigned int hosts_size_{0};
		unsigned int data_hosts_size_{0};
		std::vector<host_t> all_hosts_;
		std::hash<std::string> str_hash;
		std::unordered_map<std::string, std::pair<IntervalSet, unsigned int>>
		@@ -226,6 +255,12 @@ public:
		void
		hosts_size(unsigned int size) override;

		void
		data_hosts_size(unsigned int size) override;

		unsigned int
		data_hosts_size() const override;

		host_t
		locate_data(const std::string& path, const chunkid_t& chnk_id,
		const int num_copy) const override;

Conflict: This file was modified in both the source and target branches. Ask someone with write access to resolve it.

include/common/rpc/rpc_types_thallium.hpp

+57 −7

Changes for include/common/rpc/rpc_types_thallium.hpp: 57 added lines, 7 removed lines.
Original line number	Diff line number	Diff line
		@@ -238,18 +238,14 @@ struct rpc_read_data_in_t {
		uint64_t chunk_start;
		uint64_t chunk_end;
		uint64_t total_chunk_size;
		tl::bulk bulk_handle; // SERIALIZATION OF BULK HANDLE?
		// Thallium bulk handles generally need to be exposed.
		// But here we are defining the input struct.
		// Thallium handles bulk separately or as part of
		// args? serialize function for bulk handle exists in
		// Thallium.
		tl::bulk bulk_handle;
		bool fuzzy_relo = false;

		template <class Archive>
		void
		serialize(Archive& ar) {
		ar(path, offset, host_id, host_size, wbitset, chunk_n, chunk_start,
		chunk_end, total_chunk_size, bulk_handle);
		chunk_end, total_chunk_size, bulk_handle, fuzzy_relo);
		}
		};

		@@ -336,6 +332,60 @@ struct rpc_get_dirents_filtered_out_t {
		}
		};


		struct rpc_write_data_compressed_in_t {
		std::string path;
		int64_t offset;
		uint64_t host_id;
		uint64_t host_size;
		std::string wbitset;
		uint64_t chunk_n;
		uint64_t chunk_start;
		uint64_t chunk_end;
		uint64_t total_chunk_size;
		uint64_t compressed_size;
		tl::bulk bulk_handle;

		template <class Archive>
		void
		serialize(Archive& ar) {
		ar(path, offset, host_id, host_size, wbitset, chunk_n, chunk_start,
		chunk_end, total_chunk_size, compressed_size, bulk_handle);
		}
		};

		struct rpc_read_data_compressed_in_t {
		std::string path;
		int64_t offset;
		uint64_t host_id;
		uint64_t host_size;
		std::string wbitset;
		uint64_t chunk_n;
		uint64_t chunk_start;
		uint64_t chunk_end;
		uint64_t total_chunk_size;
		tl::bulk bulk_handle;

		template <class Archive>
		void
		serialize(Archive& ar) {
		ar(path, offset, host_id, host_size, wbitset, chunk_n, chunk_start,
		chunk_end, total_chunk_size, bulk_handle);
		}
		};

		struct rpc_data_compressed_out_t {
		int32_t err;
		size_t io_size;
		size_t compressed_size;

		template <class Archive>
		void
		serialize(Archive& ar) {
		ar(err, io_size, compressed_size);
		}
		};

		struct rpc_config_out_t {
		std::string mountdir;
		std::string rootdir;

Draft: Resolve "Add compression on demand on data"

Merge request reports