HDDS-6615. EC: Improve write performance by pipelining encode and flush

[kaijchen]

What changes were proposed in this pull request?

Introduced a new flush thread in ECKeyOutputStream, pipelining the encode stage and flush stage.
The producer thread does the buffering and encoding, and the consumer thread does the flushing.
They are connected by a ArrayBlockingQueue, whose size is "ec.stripe.queue.size", which defaults to 2.
Necessary refactors has been done to seperate the two stages.

EOFDummyStripe is used to denote the end of the key, CheckpointDummyStripe is used for syncing in tests.
Synchronization was added in tests.

NOTE: Currently the exception handling needs another look. If exception were thrown in the flush thread,
the main thread will not get it until calling Future.get() in close(). This has been fixed in d710aff.

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/HDDS-6615

How was this patch tested?

It's covered by existing test. Synchronization was added before checking state.
Mannual test of writing 1 key by freon shows ~20% performance improvement.

• Before (63s): https://paste.ubuntu.com/p/pZJ9BB5ttF/
• After (50s): https://paste.ubuntu.com/p/tRRYW3tmqG/

[sodonnel]

This looks like a good change. I need to review in more detail next week (out of the office on Friday).

Out of interest, have you benchmarked Ratis writes vs EC writes? How do they compare?

I also wonder if it would be possible to do the writes in writeDataCells and writeParityCells in parallel? Looks like we are writing to one DN, waiting, write to the next, etc. If this is possible to parallel, it would be better in a followup Jira. This one has enough in it already.

[kaijchen]

I also wonder if it would be possible to do the writes in writeDataCells and writeParityCells in parallel? Looks like we are writing to one DN, waiting, write to the next, etc. If this is possible to parallel, it would be better in a followup Jira. This one has enough in it already.

ozone/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ECBlockOutputStream.java

Lines 84 to 89 in 6abee8a

	@Override
	public void write(byte[] b, int off, int len) throws IOException {
	this.currentChunkRspFuture =
	writeChunkToContainer(ChunkBuffer.wrap(ByteBuffer.wrap(b, off, len)));
	updateWrittenDataLength(len);
	}

BlockOutputStream#writeChunkToContainer() is async, and the result of future is checked in ECBlockOutputStreamEntry#getFailedStreams(). So we are syncing at each stripe instead of each cell.

[kaijchen]

Out of interest, have you benchmarked Ratis writes vs EC writes? How do they compare?

I have tested RATIS/THREE vs EC/RS-10-4-1024K before, and here is the rough result:

Writing many keys in EC is slightly faster (~10%) than Ratis, but not as fast as in theory (because EC writes less data).
Reading many keys in EC is slightly slower (~8%) than Ratis, probably because of more connections and smaller blocks.
Comparing single key is unfair, because EC is utilizing more machines than Ratis.

When writing many large keys (1000x 5GB / 100 threads), EC works fine but Ratis cannot complete the write.

[sodonnel]

LGTM - thanks for working on this, as it seems a good change.

[kaijchen]

Thanks @sodonnel for the review.