src/storage/blobfs/compression/external_decompressor.cc - fuchsia - Git at Google

 // Copyright 2020 The Fuchsia Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.

 #include "src/storage/blobfs/compression/external_decompressor.h"

 #include <fidl/fuchsia.blobfs.internal/cpp/wire.h>
 #include <lib/fdio/directory.h>
 #include <lib/fidl/cpp/wire/channel.h>
 #include <lib/syslog/cpp/macros.h>
 #include <lib/zx/fifo.h>
 #include <lib/zx/result.h>
 #include <lib/zx/time.h>
 #include <lib/zx/vmo.h>
 #include <zircon/assert.h>
 #include <zircon/errors.h>
 #include <zircon/rights.h>
 #include <zircon/status.h>
 #include <zircon/types.h>

 #include <cstddef>
 #include <memory>
 #include <optional>
 #include <utility>

 #include "src/storage/blobfs/compression_settings.h"

 namespace blobfs {

 DecompressorCreatorConnector& DecompressorCreatorConnector::DefaultServiceConnector() {
   class ServiceConnector final : public DecompressorCreatorConnector {
    public:
     // DecompressorCreatorConnector interface.
     zx_status_t ConnectToDecompressorCreator(
         fidl::ServerEnd<fuchsia_blobfs_internal::DecompressorCreator> remote_channel) final {
       return fdio_service_connect("/svc/fuchsia.blobfs.internal.DecompressorCreator",
                                   remote_channel.TakeChannel().release());
     }
   };
   static ServiceConnector singleton{};
   return singleton;
 }

 zx::result<std::unique_ptr<ExternalDecompressorClient>> ExternalDecompressorClient::Create(
     DecompressorCreatorConnector* connector, const zx::vmo& decompressed_vmo,
     const zx::vmo& compressed_vmo) {
   std::unique_ptr<ExternalDecompressorClient> client;
   client.reset(new ExternalDecompressorClient());
   client->connector_ = connector;

   zx_status_t status =
       decompressed_vmo.duplicate(ZX_DEFAULT_VMO_RIGHTS, &(client->decompressed_vmo_));
   if (status != ZX_OK) {
     FX_LOGS(ERROR) << "Failed to duplicate decompressed VMO: " << zx_status_get_string(status);
     return zx::error(status);
   }
   status = compressed_vmo.duplicate(ZX_DEFAULT_VMO_RIGHTS & (~ZX_RIGHT_WRITE),
                                     &(client->compressed_vmo_));
   if (status != ZX_OK) {
     FX_LOGS(ERROR) << "Failed to duplicate compressed VMO: " << zx_status_get_string(status);
     return zx::error(status);
   }
   status = client->ConnectToDecompressor();
   if (status != ZX_OK) {
     return zx::error(status);
   }
   return zx::ok(std::move(client));
 }

 zx_status_t ExternalDecompressorClient::ConnectToDecompressor() {
   if (zx_status_t status = PrepareDecompressorCreator(); status != ZX_OK) {
     return status;
   }

   zx::vmo remote_decompressed_vmo;
   if (zx_status_t status =
           decompressed_vmo_.duplicate(ZX_RIGHT_SAME_RIGHTS, &remote_decompressed_vmo);
       status != ZX_OK) {
     FX_PLOGS(ERROR, status) << "Failed to create remote duplicate of decompressed VMO";
     return status;
   }

   zx::vmo remote_compressed_vmo;
   if (zx_status_t status = compressed_vmo_.duplicate(ZX_RIGHT_SAME_RIGHTS, &remote_compressed_vmo);
       status != ZX_OK) {
     FX_PLOGS(ERROR, status) << "Failed to create remote duplicate of compressed VMO";
     return status;
   }

   zx::fifo remote_fifo;
   // Sized for 4 elements, allows enough pipelining to keep the remote process
   // from descheduling to have 2 in flight requests/response pairs.
   if (zx_status_t status = zx::fifo::create(
           4, sizeof(fuchsia_blobfs_internal::wire::DecompressRequest), 0, &fifo_, &remote_fifo);
       status != ZX_OK) {
     FX_PLOGS(ERROR, status) << "Failed create fifo for external decompressor";
     return status;
   }

   auto fidl_result = fidl::WireCall(decompressor_creator_)
                          ->Create(std::move(remote_fifo), std::move(remote_compressed_vmo),
                                   std::move(remote_decompressed_vmo));
   if (fidl_result.is_peer_closed()) {
     decompressor_creator_.reset();
   }
   if (!fidl_result.ok()) {
     FX_LOGS(ERROR) << "FIDL error communicating with external decompressor: "
                    << fidl_result.FormatDescription();
     return fidl_result.status();
   }
   if (fidl_result.value().status != ZX_OK) {
     FX_PLOGS(ERROR, fidl_result.value().status)
         << "Error calling Create on DecompressorCreator service";
   }
   return fidl_result.value().status;
 }

 zx_status_t ExternalDecompressorClient::PrepareDecompressorCreator() {
   if (decompressor_creator_.is_valid()) {
     zx_signals_t signal;
     zx_status_t status = decompressor_creator_.channel().wait_one(
         ZX_CHANNEL_WRITABLE | ZX_CHANNEL_PEER_CLOSED, zx::time::infinite_past(), &signal);
     if (status == ZX_OK && (signal & ZX_CHANNEL_PEER_CLOSED) == 0 &&
         (signal & ZX_CHANNEL_WRITABLE) != 0) {
       return ZX_OK;
     } else {
       decompressor_creator_.reset();
     }
   }

   auto remote_channel = fidl::CreateEndpoints(&decompressor_creator_);
   if (remote_channel.is_error()) {
     FX_PLOGS(ERROR, remote_channel.status_value())
         << "Failed to create channel pair for external decompressor.";
     return remote_channel.status_value();
   }

   if (zx_status_t status =
           connector_->ConnectToDecompressorCreator(std::move(remote_channel).value());
       status != ZX_OK) {
     FX_PLOGS(ERROR, status) << "Failed to connect to DecompressorCreator service";
     decompressor_creator_.reset();
     return status;
   }

   return ZX_OK;
 }

 zx_status_t ExternalDecompressorClient::SendRequest(
     const fuchsia_blobfs_internal::wire::DecompressRequest& request) {
   zx_status_t write_status = fifo_.write(sizeof(request), &request, 1, nullptr);
   if (write_status == ZX_OK) {
     return ZX_OK;
   }
   bool try_to_reconnect = false;
   if (write_status == ZX_ERR_SHOULD_WAIT) {
     // The fifo is full, wait for it to become writable.
     zx_signals_t signals = 0;
     if (zx_status_t status =
             fifo_.wait_one(ZX_FIFO_WRITABLE | ZX_FIFO_PEER_CLOSED, zx::time::infinite(), &signals);
         status != ZX_OK) {
       return status;
     }
     if ((signals & ZX_FIFO_PEER_CLOSED) != 0) {
       // The other end of the fifo was closed while waiting for the fifo to become writable. Try to
       // make a new connection.
       try_to_reconnect = true;
     }
   } else if (write_status == ZX_ERR_PEER_CLOSED || write_status == ZX_ERR_BAD_HANDLE) {
     try_to_reconnect = true;
   } else {
     FX_PLOGS(ERROR, write_status) << "Unexpected response when writing to fifo";
     return write_status;
   }
   if (try_to_reconnect) {
     fifo_.reset();
     if (zx_status_t status = ConnectToDecompressor(); status != ZX_OK) {
       return status;
     }
   }
   // The original fifo should now be writable or a new connection was established.
   return fifo_.write(sizeof(request), &request, 1, nullptr);
 }

 zx_status_t ExternalDecompressorClient::SendMessage(
     const fuchsia_blobfs_internal::wire::DecompressRequest& request) {
   if (zx_status_t status = SendRequest(request); status != ZX_OK) {
     FX_PLOGS(ERROR, status) << "Failed to write fifo request to decompressor";
     return status;
   }

   zx_signals_t signal;
   fifo_.wait_one(ZX_FIFO_READABLE | ZX_FIFO_PEER_CLOSED, zx::time::infinite(), &signal);
   if ((signal & ZX_FIFO_READABLE) == 0) {
     fifo_.reset();
     FX_LOGS(ERROR) << "External decompressor closed the fifo.";
     return ZX_ERR_INTERNAL;
   }

   fuchsia_blobfs_internal::wire::DecompressResponse response;
   if (zx_status_t status = fifo_.read(sizeof(response), &response, 1, nullptr); status != ZX_OK) {
     FX_PLOGS(ERROR, status) << "Failed to read from fifo";
     return status;
   }
   if (response.status != ZX_OK) {
     FX_PLOGS(ERROR, response.status) << "Error from external decompressor";
     return response.status;
   }
   if (response.size != request.decompressed.size) {
     FX_LOGS(ERROR) << "Decompressed size did not match. Expected: " << request.decompressed.size
                    << " Got: " << response.size;
     return ZX_ERR_IO_DATA_INTEGRITY;
   }
   return ZX_OK;
 }

 std::optional<CompressionAlgorithm> ExternalDecompressorClient::CompressionAlgorithmFidlToLocal(
     const fuchsia_blobfs_internal::wire::CompressionAlgorithm algorithm) {
   using Fidl = fuchsia_blobfs_internal::wire::CompressionAlgorithm;
   switch (algorithm) {
     case Fidl::kUncompressed:
       return CompressionAlgorithm::kUncompressed;
     case Fidl::kChunked:
     case Fidl::kChunkedPartial:
       return CompressionAlgorithm::kChunked;
     default:
       return std::nullopt;
   }
 }

 fuchsia_blobfs_internal::wire::CompressionAlgorithm
 ExternalDecompressorClient::CompressionAlgorithmLocalToFidl(CompressionAlgorithm algorithm) {
   using Fidl = fuchsia_blobfs_internal::wire::CompressionAlgorithm;
   switch (algorithm) {
     case CompressionAlgorithm::kUncompressed:
       return Fidl::kUncompressed;
     case CompressionAlgorithm::kChunked:
       return Fidl::kChunked;
   }

   ZX_DEBUG_ASSERT(false);
   return Fidl::kUncompressed;
 }

 zx::result<fuchsia_blobfs_internal::wire::CompressionAlgorithm>
 ExternalDecompressorClient::CompressionAlgorithmLocalToFidlForPartial(
     CompressionAlgorithm algorithm) {
   switch (algorithm) {
     case CompressionAlgorithm::kChunked:
       return zx::ok(fuchsia_blobfs_internal::wire::CompressionAlgorithm::kChunkedPartial);
     case CompressionAlgorithm::kUncompressed:
       return zx::error(ZX_ERR_NOT_SUPPORTED);
   }

   ZX_DEBUG_ASSERT(false);
   return zx::error(ZX_ERR_NOT_SUPPORTED);
 }

 ExternalSeekableDecompressor::ExternalSeekableDecompressor(ExternalDecompressorClient* client,
                                                            CompressionAlgorithm algorithm)
     : client_(client), algorithm_(algorithm) {}

 zx_status_t ExternalSeekableDecompressor::DecompressRange(size_t compressed_offset,
                                                           size_t compressed_size,
                                                           size_t uncompressed_size) {
   auto algorithm_or =
       ExternalDecompressorClient::CompressionAlgorithmLocalToFidlForPartial(algorithm_);
   if (!algorithm_or.is_ok()) {
     return algorithm_or.status_value();
   }
   fuchsia_blobfs_internal::wire::CompressionAlgorithm fidl_algorithm = algorithm_or.value();

   return client_->SendMessage({
       {0, uncompressed_size},
       {compressed_offset, compressed_size},
       fidl_algorithm,
   });
 }

 }  // namespace blobfs
	// Copyright 2020 The Fuchsia Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style license that can be
	// found in the LICENSE file.

	#include "src/storage/blobfs/compression/external_decompressor.h"

	#include <fidl/fuchsia.blobfs.internal/cpp/wire.h>
	#include <lib/fdio/directory.h>
	#include <lib/fidl/cpp/wire/channel.h>
	#include <lib/syslog/cpp/macros.h>
	#include <lib/zx/fifo.h>
	#include <lib/zx/result.h>
	#include <lib/zx/time.h>
	#include <lib/zx/vmo.h>
	#include <zircon/assert.h>
	#include <zircon/errors.h>
	#include <zircon/rights.h>
	#include <zircon/status.h>
	#include <zircon/types.h>

	#include <cstddef>
	#include <memory>
	#include <optional>
	#include <utility>

	#include "src/storage/blobfs/compression_settings.h"

	namespace blobfs {

	DecompressorCreatorConnector& DecompressorCreatorConnector::DefaultServiceConnector() {
	class ServiceConnector final : public DecompressorCreatorConnector {
	public:
	// DecompressorCreatorConnector interface.
	zx_status_t ConnectToDecompressorCreator(
	fidl::ServerEnd<fuchsia_blobfs_internal::DecompressorCreator> remote_channel) final {
	return fdio_service_connect("/svc/fuchsia.blobfs.internal.DecompressorCreator",
	remote_channel.TakeChannel().release());
	}
	};
	static ServiceConnector singleton{};
	return singleton;
	}

	zx::result<std::unique_ptr<ExternalDecompressorClient>> ExternalDecompressorClient::Create(
	DecompressorCreatorConnector* connector, const zx::vmo& decompressed_vmo,
	const zx::vmo& compressed_vmo) {
	std::unique_ptr<ExternalDecompressorClient> client;
	client.reset(new ExternalDecompressorClient());
	client->connector_ = connector;

	zx_status_t status =
	decompressed_vmo.duplicate(ZX_DEFAULT_VMO_RIGHTS, &(client->decompressed_vmo_));
	if (status != ZX_OK) {
	FX_LOGS(ERROR) << "Failed to duplicate decompressed VMO: " << zx_status_get_string(status);
	return zx::error(status);
	}
	status = compressed_vmo.duplicate(ZX_DEFAULT_VMO_RIGHTS & (~ZX_RIGHT_WRITE),
	&(client->compressed_vmo_));
	if (status != ZX_OK) {
	FX_LOGS(ERROR) << "Failed to duplicate compressed VMO: " << zx_status_get_string(status);
	return zx::error(status);
	}
	status = client->ConnectToDecompressor();
	if (status != ZX_OK) {
	return zx::error(status);
	}
	return zx::ok(std::move(client));
	}

	zx_status_t ExternalDecompressorClient::ConnectToDecompressor() {
	if (zx_status_t status = PrepareDecompressorCreator(); status != ZX_OK) {
	return status;
	}

	zx::vmo remote_decompressed_vmo;
	if (zx_status_t status =
	decompressed_vmo_.duplicate(ZX_RIGHT_SAME_RIGHTS, &remote_decompressed_vmo);
	status != ZX_OK) {
	FX_PLOGS(ERROR, status) << "Failed to create remote duplicate of decompressed VMO";
	return status;
	}

	zx::vmo remote_compressed_vmo;
	if (zx_status_t status = compressed_vmo_.duplicate(ZX_RIGHT_SAME_RIGHTS, &remote_compressed_vmo);
	status != ZX_OK) {
	FX_PLOGS(ERROR, status) << "Failed to create remote duplicate of compressed VMO";
	return status;
	}

	zx::fifo remote_fifo;
	// Sized for 4 elements, allows enough pipelining to keep the remote process
	// from descheduling to have 2 in flight requests/response pairs.
	if (zx_status_t status = zx::fifo::create(
	4, sizeof(fuchsia_blobfs_internal::wire::DecompressRequest), 0, &fifo_, &remote_fifo);
	status != ZX_OK) {
	FX_PLOGS(ERROR, status) << "Failed create fifo for external decompressor";
	return status;
	}

	auto fidl_result = fidl::WireCall(decompressor_creator_)
	->Create(std::move(remote_fifo), std::move(remote_compressed_vmo),
	std::move(remote_decompressed_vmo));
	if (fidl_result.is_peer_closed()) {
	decompressor_creator_.reset();
	}
	if (!fidl_result.ok()) {
	FX_LOGS(ERROR) << "FIDL error communicating with external decompressor: "
	<< fidl_result.FormatDescription();
	return fidl_result.status();
	}
	if (fidl_result.value().status != ZX_OK) {
	FX_PLOGS(ERROR, fidl_result.value().status)
	<< "Error calling Create on DecompressorCreator service";
	}
	return fidl_result.value().status;
	}

	zx_status_t ExternalDecompressorClient::PrepareDecompressorCreator() {
	if (decompressor_creator_.is_valid()) {
	zx_signals_t signal;
	zx_status_t status = decompressor_creator_.channel().wait_one(
	ZX_CHANNEL_WRITABLE \| ZX_CHANNEL_PEER_CLOSED, zx::time::infinite_past(), &signal);
	if (status == ZX_OK && (signal & ZX_CHANNEL_PEER_CLOSED) == 0 &&
	(signal & ZX_CHANNEL_WRITABLE) != 0) {
	return ZX_OK;
	} else {
	decompressor_creator_.reset();
	}
	}

	auto remote_channel = fidl::CreateEndpoints(&decompressor_creator_);
	if (remote_channel.is_error()) {
	FX_PLOGS(ERROR, remote_channel.status_value())
	<< "Failed to create channel pair for external decompressor.";
	return remote_channel.status_value();
	}

	if (zx_status_t status =
	connector_->ConnectToDecompressorCreator(std::move(remote_channel).value());
	status != ZX_OK) {
	FX_PLOGS(ERROR, status) << "Failed to connect to DecompressorCreator service";
	decompressor_creator_.reset();
	return status;
	}

	return ZX_OK;
	}

	zx_status_t ExternalDecompressorClient::SendRequest(
	const fuchsia_blobfs_internal::wire::DecompressRequest& request) {
	zx_status_t write_status = fifo_.write(sizeof(request), &request, 1, nullptr);
	if (write_status == ZX_OK) {
	return ZX_OK;
	}
	bool try_to_reconnect = false;
	if (write_status == ZX_ERR_SHOULD_WAIT) {
	// The fifo is full, wait for it to become writable.
	zx_signals_t signals = 0;
	if (zx_status_t status =
	fifo_.wait_one(ZX_FIFO_WRITABLE \| ZX_FIFO_PEER_CLOSED, zx::time::infinite(), &signals);
	status != ZX_OK) {
	return status;
	}
	if ((signals & ZX_FIFO_PEER_CLOSED) != 0) {
	// The other end of the fifo was closed while waiting for the fifo to become writable. Try to
	// make a new connection.
	try_to_reconnect = true;
	}
	} else if (write_status == ZX_ERR_PEER_CLOSED \|\| write_status == ZX_ERR_BAD_HANDLE) {
	try_to_reconnect = true;
	} else {
	FX_PLOGS(ERROR, write_status) << "Unexpected response when writing to fifo";
	return write_status;
	}
	if (try_to_reconnect) {
	fifo_.reset();
	if (zx_status_t status = ConnectToDecompressor(); status != ZX_OK) {
	return status;
	}
	}
	// The original fifo should now be writable or a new connection was established.
	return fifo_.write(sizeof(request), &request, 1, nullptr);
	}

	zx_status_t ExternalDecompressorClient::SendMessage(
	const fuchsia_blobfs_internal::wire::DecompressRequest& request) {
	if (zx_status_t status = SendRequest(request); status != ZX_OK) {
	FX_PLOGS(ERROR, status) << "Failed to write fifo request to decompressor";
	return status;
	}

	zx_signals_t signal;
	fifo_.wait_one(ZX_FIFO_READABLE \| ZX_FIFO_PEER_CLOSED, zx::time::infinite(), &signal);
	if ((signal & ZX_FIFO_READABLE) == 0) {
	fifo_.reset();
	FX_LOGS(ERROR) << "External decompressor closed the fifo.";
	return ZX_ERR_INTERNAL;
	}

	fuchsia_blobfs_internal::wire::DecompressResponse response;
	if (zx_status_t status = fifo_.read(sizeof(response), &response, 1, nullptr); status != ZX_OK) {
	FX_PLOGS(ERROR, status) << "Failed to read from fifo";
	return status;
	}
	if (response.status != ZX_OK) {
	FX_PLOGS(ERROR, response.status) << "Error from external decompressor";
	return response.status;
	}
	if (response.size != request.decompressed.size) {
	FX_LOGS(ERROR) << "Decompressed size did not match. Expected: " << request.decompressed.size
	<< " Got: " << response.size;
	return ZX_ERR_IO_DATA_INTEGRITY;
	}
	return ZX_OK;
	}

	std::optional<CompressionAlgorithm> ExternalDecompressorClient::CompressionAlgorithmFidlToLocal(
	const fuchsia_blobfs_internal::wire::CompressionAlgorithm algorithm) {
	using Fidl = fuchsia_blobfs_internal::wire::CompressionAlgorithm;
	switch (algorithm) {
	case Fidl::kUncompressed:
	return CompressionAlgorithm::kUncompressed;
	case Fidl::kChunked:
	case Fidl::kChunkedPartial:
	return CompressionAlgorithm::kChunked;
	default:
	return std::nullopt;
	}
	}

	fuchsia_blobfs_internal::wire::CompressionAlgorithm
	ExternalDecompressorClient::CompressionAlgorithmLocalToFidl(CompressionAlgorithm algorithm) {
	using Fidl = fuchsia_blobfs_internal::wire::CompressionAlgorithm;
	switch (algorithm) {
	case CompressionAlgorithm::kUncompressed:
	return Fidl::kUncompressed;
	case CompressionAlgorithm::kChunked:
	return Fidl::kChunked;
	}

	ZX_DEBUG_ASSERT(false);
	return Fidl::kUncompressed;
	}

	zx::result<fuchsia_blobfs_internal::wire::CompressionAlgorithm>
	ExternalDecompressorClient::CompressionAlgorithmLocalToFidlForPartial(
	CompressionAlgorithm algorithm) {
	switch (algorithm) {
	case CompressionAlgorithm::kChunked:
	return zx::ok(fuchsia_blobfs_internal::wire::CompressionAlgorithm::kChunkedPartial);
	case CompressionAlgorithm::kUncompressed:
	return zx::error(ZX_ERR_NOT_SUPPORTED);
	}

	ZX_DEBUG_ASSERT(false);
	return zx::error(ZX_ERR_NOT_SUPPORTED);
	}

	ExternalSeekableDecompressor::ExternalSeekableDecompressor(ExternalDecompressorClient* client,
	CompressionAlgorithm algorithm)
	: client_(client), algorithm_(algorithm) {}

	zx_status_t ExternalSeekableDecompressor::DecompressRange(size_t compressed_offset,
	size_t compressed_size,
	size_t uncompressed_size) {
	auto algorithm_or =
	ExternalDecompressorClient::CompressionAlgorithmLocalToFidlForPartial(algorithm_);
	if (!algorithm_or.is_ok()) {
	return algorithm_or.status_value();
	}
	fuchsia_blobfs_internal::wire::CompressionAlgorithm fidl_algorithm = algorithm_or.value();

	return client_->SendMessage({
	{0, uncompressed_size},
	{compressed_offset, compressed_size},
	fidl_algorithm,
	});
	}

	} // namespace blobfs