Replication Message Serialization Speedup

[jkosh44]

Summary

Currently, we serialize all messages related to replication using JSON. The implementation can be found here:

noisepage/src/include/replication/replication_messages.h

Lines 36 to 108 in 97eb7ec

	/** Abstraction over the underlying format used to send replication messages over the network */
	class MessageWrapper {
	public:
	/** The underlying format of messages used in replication */
	using MessageFormat = common::json;
	/** Default constructor */
	MessageWrapper();
	/** Constructor which parses a string */
	explicit MessageWrapper(std::string_view str);
	/**
	* Adds a value to the message with a specific key
	*
	* @tparam T type of value to add
	* @param key key of value in message
	* @param value value to add to message
	*/
	template <typename T>
	void Put(const char *key, T value);
	/**
	* Get a value from the message with specific key
	*
	* @tparam T type of value to get
	* @param key key of value
	* @return value from message with specified key
	*/
	template <typename T>
	T Get(const char *key) const;
	/**
	* Deserializes a given input to a string using the CBOR (Concise Binary Object Representation) serialization
	* format.
	*
	* @param cbor cbor input
	* @return parsed string
	*
	*/
	static std::string FromCbor(const std::vector<uint8_t> &cbor);
	/**
	* Serializes a given input string to CBOR format
	* @param str string to serialize
	* @return CBOR format of string
	*/
	static std::vector<uint8_t> ToCbor(std::string_view str);
	/**
	* Serialize the message
	*
	* @return serialized version of message
	*/
	std::string Serialize() const;
	/**
	* Converts MessageWrapper to JSON
	* @return JSON version of MessageWrapper
	*/
	common::json ToJson() const;
	/**
	* Converts JSON to MessageWrapper
	* @param j JSON to convert
	*/
	void FromJson(const common::json &j);
	private:
	std::unique_ptr<MessageFormat> underlying_message_;
	};
	DEFINE_JSON_HEADER_DECLARATIONS(MessageWrapper);

noisepage/src/replication/replication_messages.cpp

Lines 19 to 56 in 97eb7ec

	// MessageWrapper
	MessageWrapper::MessageWrapper() : underlying_message_(std::make_unique<MessageFormat>()) {}
	MessageWrapper::MessageWrapper(std::string_view str)
	: underlying_message_(std::make_unique<MessageFormat>(common::json::parse(str))) {}
	template <typename T>
	void MessageWrapper::Put(const char *key, T value) {
	(*underlying_message_)[key] = value;
	}
	template void MessageWrapper::Put<std::string>(const char *key, std::string value);
	template void MessageWrapper::Put<std::vector<uint8_t>>(const char *key, std::vector<uint8_t> value);
	template void MessageWrapper::Put<MessageWrapper>(const char *key, MessageWrapper value);
	template void MessageWrapper::Put<record_batch_id_t>(const char *key, record_batch_id_t value);
	template void MessageWrapper::Put<msg_id_t>(const char *key, msg_id_t value);
	template void MessageWrapper::Put<transaction::timestamp_t>(const char *key, transaction::timestamp_t value);
	template <typename T>
	T MessageWrapper::Get(const char *key) const {
	return underlying_message_->at(key).get<T>();
	}
	template std::string MessageWrapper::Get<std::string>(const char *key) const;
	template std::vector<uint8_t> MessageWrapper::Get<std::vector<uint8_t>>(const char *key) const;
	template MessageWrapper MessageWrapper::Get<MessageWrapper>(const char *key) const;
	template record_batch_id_t MessageWrapper::Get<record_batch_id_t>(const char *key) const;
	template msg_id_t MessageWrapper::Get<msg_id_t>(const char *key) const;
	template transaction::timestamp_t MessageWrapper::Get<transaction::timestamp_t>(const char *key) const;
	std::string MessageWrapper::FromCbor(const std::vector<uint8_t> &cbor) { return common::json::from_cbor(cbor); }
	std::vector<uint8_t> MessageWrapper::ToCbor(std::string_view str) { return common::json::to_cbor(str); }
	std::string MessageWrapper::Serialize() const { return underlying_message_->dump(); }
	common::json MessageWrapper::ToJson() const { return *underlying_message_; }
	void MessageWrapper::FromJson(const common::json &j) { *(this->underlying_message_) = j; }

Turning replication on causes a significant slowdown to the database and one of the primary causes is the JSON serialization of messages. Below are some performance results of running TPCC on dev10 with 8 threads with various database configurations:

Replication	Durability	Modifications	Request Throughput (request per second)
DISABLED	Sync	None	208.891796443148
Async	Sync	None	96.2971597537919
Async	Sync	Remove JSON serialization of messages (all messages replaced with empty string)	181.1826548771372

Below are some metrics on log throughput for the primary node with various database configurations:

Replication	Durability	Modifications	Log Throughput (records per millisecond)
DISABLED	Sync	None	98.80787541
DISABLED	Async	None	102.9403467
Sync	Sync	None	1.130251208
Async	Sync	None	1.273598347
Async	Async	None	1.235403819
Async	Async	Remove JSON serialization of messages (all messages replaced with empty string)	88.39578983

Just for reference below are some metrics on log throughput for the replica node

Durability	Log Throughput (records per millisecond)
Sync	1.009166671
Async	1.007238045

Solution

A solution to this is to switch to a different message format than JSON and I plan on investigating a handful of alternatives and their impact on log throughput and request throughput.

Nlohmann

We use the Nlohmann JSON package to implement JSON in NoisePage. This package comes with a bunch of other binary formats built into the package. It's probably worth trying all of these since they can each be implemented with a couple of changed lines. Some require you to first convert your data to JSON before converting to a different binary format, and it's unclear to me if this has a significant performance penalty compared to converting directly to the message format.

• BSON
• CBOR
• MessagePack
• UBJSON

Alternatives

Below are a handful of message formats I have found from some brief research. I plan on narrowing this down to roughly 4 after some more research.

• Protobuf
• Cap'n Proto
• FlatBuffer
• Apache Thrift
• Apache Avro
• Apache Parquet

Dependency Bloat

One of the considerations when implementing a new message format will be dependency bloat. I don't plan on coming up with my own implementation for any of these formats so we'll have to bring in third-party libraries. It will be important to make sure we don't bring in more than necessary to avoid dependency bloat.

[jkosh44]

If anyone has a particular format that I left out that they think would be good, please let me know.

[jkosh44]

BSON

branch: https://github.com/jkosh44/noisepage/tree/bson

UPDATE: The original JSON implementation was converting the record contents itself to and from CBOR. The original numbers used for BSON kept that conversion in. The updated numbers remove that conversion.

Log Throughput Primary

Replication	Durability	Modifications	Log Throughput (records per millisecond)
Sync	Sync	None	0.839491378
Async	Sync	None	0.9220486988
Async	Async	None	0.9210122535

Log Throughput Replica

Durability	Log Throughput (records per millisecond)
Sync	0.7778571493
Async	0.7783025631

##Log Throughput Primary

Replication	Durability	Modifications	Log Throughput (records per millisecond)
Sync	Sync	None	3.8700566481216665
Async	Sync	None	30.736948475102366
Async	Async	None	29.028352465595663

Log Throughput Replica

Durability	Log Throughput (records per millisecond)
Sync	3.8883037697513125
Async	3.874525911

[jkosh44]

Message Pack

branch: https://github.com/jkosh44/noisepage/tree/messagepack

Log Throughput Primary

Replication	Durability	Modifications	Log Throughput (records per millisecond)
Sync	Sync	None	3.8606188627282036
Async	Sync	None	29.38279984543937
Async	Async	None	28.471362156671457

Log Throughput Replica

Durability	Log Throughput (records per millisecond)
Sync	3.887471136
Async	3.91150189

[jkosh44]

UBJSON

branch: https://github.com/jkosh44/noisepage/tree/ubjson

Log Throughput Primary

Replication	Durability	Modifications	Log Throughput (records per millisecond)
Sync	Sync	None	3.864162520678762
Async	Sync	None	30.148788480999627
Async	Async	None	28.387447086724443

Log Throughput Replica

Durability	Log Throughput (records per millisecond)
Sync	3.887471136
Async	3.91150189

[jkosh44]

CBOR

branch: https://github.com/jkosh44/noisepage/tree/cbor

Log Throughput Primary

Replication	Durability	Modifications	Log Throughput (records per millisecond)
Sync	Sync	None	3.861907674897514
Async	Sync	None	30.714806087560053
Async	Async	None	28.913739773546347

Log Throughput Replica

Durability	Log Throughput (records per millisecond)
Sync	3.918585055891593
Async	3.9006568057238713