[cdc_rsync] Fix issue in UnzstdStream (#59)

Fixes an issue in UnzstdStream where the Read() method always tries to
read new input data if no input data is available, instead of first
trying to uncompress. Since zstd maintains internal buffers,
uncompression might succeed even without reading more input, so this
is faster. This bug can lead to pipeline stalls in cdc_rsync.

cdc_rsync/zstd_stream.cc

@@ -27,22 +27,19 @@ namespace {
 // trigger a flush. This happens when files with no changes are diff'ed (this
 // produces very low volume data). Flushing prevents that the server gets stale
 // and becomes overwhelmed later.
-constexpr absl::Duration kMinCompressPeriod = absl::Milliseconds(500);
+constexpr absl::Duration kDefaultAutoFlushPeriod = absl::Milliseconds(500);
 
 }  // namespace
 
 ZstdStream::ZstdStream(Socket* socket, int level, uint32_t num_threads)
-    : socket_(socket), cctx_(nullptr) {
+    : socket_(socket),
+      cctx_(nullptr),
+      auto_flush_period_(kDefaultAutoFlushPeriod) {
   status_ = WrapStatus(Initialize(level, num_threads),
                        "Failed to initialize stream compressor");
 }
 
 ZstdStream::~ZstdStream() {
-  if (cctx_) {
-    ZSTD_freeCCtx(cctx_);
-    cctx_ = nullptr;
-  }
-
   {
     absl::MutexLock lock(&mutex_);
     shutdown_ = true;
@@ -50,6 +47,11 @@ ZstdStream::~ZstdStream() {
   if (compressor_thread_.joinable()) {
     compressor_thread_.join();
   }
+
+  if (cctx_) {
+    ZSTD_freeCCtx(cctx_);
+    cctx_ = nullptr;
+  }
 }
 
 absl::Status ZstdStream::Write(const void* data, size_t size) {
@@ -79,7 +81,7 @@ absl::Status ZstdStream::Write(const void* data, size_t size) {
   return absl::OkStatus();
 }
 
-absl::Status ZstdStream::Flush() {
+absl::Status ZstdStream::Finish() {
   absl::MutexLock lock(&mutex_);
   if (!status_.ok()) return status_;
 
@@ -134,7 +136,7 @@ void ZstdStream::ThreadCompressorMain() {
              in_buffer_.size() == in_buffer_.capacity();
     };
     bool flush =
-        !mutex_.AwaitWithTimeout(absl::Condition(&cond), kMinCompressPeriod);
+        !mutex_.AwaitWithTimeout(absl::Condition(&cond), auto_flush_period_);
     if (shutdown_) {
       return;
     }