Compression.cpp | Explore in Territory

/*
 * Copyright (c) Meta Platforms, Inc. and affiliates.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <folly/compression/Compression.h>

#if FOLLY_HAVE_LIBLZ4
#include <lz4.h>
#include <lz4hc.h>
#if LZ4_VERSION_NUMBER >= 10301
#include <lz4frame.h>
#endif
#endif

#include <glog/logging.h>

#if FOLLY_HAVE_LIBSNAPPY
#include <snappy-sinksource.h>
#include <snappy.h>
#endif

#if FOLLY_HAVE_LIBZ
#include <folly/compression/Zlib.h>
#endif

#if FOLLY_HAVE_LIBLZMA
#include <lzma.h>
#endif

#if FOLLY_HAVE_LIBZSTD
#include <folly/compression/Zstd.h>
#endif

#if FOLLY_HAVE_LIBBZ2
#include <folly/portability/Windows.h>

#include <bzlib.h>
#endif

#include <algorithm>
#include <unordered_set>

#include <folly/Conv.h>
#include <folly/Memory.h>
#include <folly/Portability.h>
#include <folly/Random.h>
#include <folly/ScopeGuard.h>
#include <folly/Varint.h>
#include <folly/compression/Utils.h>
#include <folly/io/Cursor.h>
#include <folly/lang/Bits.h>
#include <folly/stop_watch.h>

dataStartsWithLE;
prefixToStringLE;

namespace folly {
namespace compression {

Codec::Codec(CodecType type, Optional<int> /* level */, StringPiece /* name */)
    : … { … }

// Ensure consistent behavior in the nullptr case
std::unique_ptr<IOBuf> Codec::compress(const IOBuf* data) { … }

std::string Codec::compress(const StringPiece data) { … }

std::unique_ptr<IOBuf> Codec::uncompress(
    const IOBuf* data, Optional<uint64_t> uncompressedLength) { … }

std::string Codec::uncompress(
    const StringPiece data, Optional<uint64_t> uncompressedLength) { … }

bool Codec::needsUncompressedLength() const { … }

uint64_t Codec::maxUncompressedLength() const { … }

bool Codec::doNeedsUncompressedLength() const { … }

uint64_t Codec::doMaxUncompressedLength() const { … }

std::vector<std::string> Codec::validPrefixes() const { … }

bool Codec::canUncompress(const IOBuf*, Optional<uint64_t>) const { … }

bool Codec::canUncompress(
    StringPiece data, Optional<uint64_t> uncompressedLength) const { … }

std::string Codec::doCompressString(const StringPiece data) { … }

std::string Codec::doUncompressString(
    const StringPiece data, Optional<uint64_t> uncompressedLength) { … }

uint64_t Codec::maxCompressedLength(uint64_t uncompressedLength) const { … }

Optional<uint64_t> Codec::getUncompressedLength(
    const folly::IOBuf* data, Optional<uint64_t> uncompressedLength) const { … }

Optional<uint64_t> Codec::getUncompressedLength(
    StringPiece data, Optional<uint64_t> uncompressedLength) const { … }

Optional<uint64_t> Codec::doGetUncompressedLength(
    const folly::IOBuf*, Optional<uint64_t> uncompressedLength) const { … }

bool StreamCodec::needsDataLength() const { … }

bool StreamCodec::doNeedsDataLength() const { … }

void StreamCodec::assertStateIs(State expected) const { … }

void StreamCodec::resetStream(Optional<uint64_t> uncompressedLength) { … }

bool StreamCodec::compressStream(
    ByteRange& input, MutableByteRange& output, StreamCodec::FlushOp flushOp) { … }

bool StreamCodec::uncompressStream(
    ByteRange& input, MutableByteRange& output, StreamCodec::FlushOp flushOp) { … }

static std::unique_ptr<IOBuf> addOutputBuffer(
    MutableByteRange& output, uint64_t size) { … }

std::unique_ptr<IOBuf> StreamCodec::doCompress(IOBuf const* data) { … }

static uint64_t computeBufferLength(
    uint64_t const compressedLength, uint64_t const blockSize) { … }

std::unique_ptr<IOBuf> StreamCodec::doUncompress(
    IOBuf const* data, Optional<uint64_t> uncompressedLength) { … }

namespace {

/**
 * No compression
 */
class NoCompressionCodec final : public Codec { … };

std::unique_ptr<Codec> NoCompressionCodec::create(int level, CodecType type) { … }

NoCompressionCodec::NoCompressionCodec(int level, CodecType type)
    : … { … }

uint64_t NoCompressionCodec::doMaxCompressedLength(
    uint64_t uncompressedLength) const { … }

std::unique_ptr<IOBuf> NoCompressionCodec::doCompress(const IOBuf* data) { … }

std::unique_ptr<IOBuf> NoCompressionCodec::doUncompress(
    const IOBuf* data, Optional<uint64_t> uncompressedLength) { … }

#if (FOLLY_HAVE_LIBLZ4 || FOLLY_HAVE_LIBLZMA)

void encodeVarintToIOBuf(uint64_t val, folly::IOBuf* out) { … }

inline uint64_t decodeVarintFromCursor(folly::io::Cursor& cursor) { … }

#endif // FOLLY_HAVE_LIBLZ4 || FOLLY_HAVE_LIBLZMA

#if FOLLY_HAVE_LIBLZ4

#if LZ4_VERSION_NUMBER >= 10802 && defined(LZ4_STATIC_LINKING_ONLY) && \
    defined(LZ4_HC_STATIC_LINKING_ONLY) && !defined(FOLLY_USE_LZ4_FAST_RESET)
#define FOLLY_USE_LZ4_FAST_RESET
#endif

#ifdef FOLLY_USE_LZ4_FAST_RESET
void lz4_stream_t_deleter(LZ4_stream_t* ctx) {
  LZ4_freeStream(ctx);
}

void lz4_streamhc_t_deleter(LZ4_streamHC_t* ctx) {
  LZ4_freeStreamHC(ctx);
}
#endif

/**
 * LZ4 compression
 */
class LZ4Codec final : public Codec { … };

std::unique_ptr<Codec> LZ4Codec::create(int level, CodecType type) { … }

int lz4ConvertLevel(int level) { … }

LZ4Codec::LZ4Codec(int level, CodecType type)
    : … { … }

bool LZ4Codec::doNeedsUncompressedLength() const { … }

// The value comes from lz4.h in lz4-r117, but older versions of lz4 don't
// define LZ4_MAX_INPUT_SIZE (even though the max size is the same), so do it
// here.
#ifndef LZ4_MAX_INPUT_SIZE
#define LZ4_MAX_INPUT_SIZE …
#endif

uint64_t LZ4Codec::doMaxUncompressedLength() const { … }

uint64_t LZ4Codec::doMaxCompressedLength(uint64_t uncompressedLength) const { … }

std::unique_ptr<IOBuf> LZ4Codec::doCompress(const IOBuf* data) { … }

std::unique_ptr<IOBuf> LZ4Codec::doUncompress(
    const IOBuf* data, Optional<uint64_t> uncompressedLength) { … }

#if LZ4_VERSION_NUMBER >= 10301

class LZ4FrameCodec final : public Codec { … };

/* static */ std::unique_ptr<Codec> LZ4FrameCodec::create(
    int level, CodecType type) { … }

constexpr uint32_t kLZ4FrameMagicLE = …;

std::vector<std::string> LZ4FrameCodec::validPrefixes() const { … }

bool LZ4FrameCodec::canUncompress(const IOBuf* data, Optional<uint64_t>) const { … }

uint64_t LZ4FrameCodec::doMaxCompressedLength(
    uint64_t uncompressedLength) const { … }

size_t lz4FrameThrowOnError(size_t code) { … }

void LZ4FrameCodec::resetDCtx() { … }

int lz4fConvertLevel(int level) { … }

LZ4FrameCodec::LZ4FrameCodec(int level, CodecType type)
    : … { … }

LZ4FrameCodec::~LZ4FrameCodec() { … }

std::unique_ptr<IOBuf> LZ4FrameCodec::doCompress(const IOBuf* data) { … }

std::unique_ptr<IOBuf> LZ4FrameCodec::doUncompress(
    const IOBuf* data, Optional<uint64_t> uncompressedLength) { … }

#endif // LZ4_VERSION_NUMBER >= 10301
#endif // FOLLY_HAVE_LIBLZ4

#if FOLLY_HAVE_LIBSNAPPY

/**
 * Snappy compression
 */

/**
 * Implementation of snappy::Source that reads from a IOBuf chain.
 */
class IOBufSnappySource final : public snappy::Source { … };

IOBufSnappySource::IOBufSnappySource(const IOBuf* data)
    : … { … }

size_t IOBufSnappySource::Available() const { … }

const char* IOBufSnappySource::Peek(size_t* len) { … }

void IOBufSnappySource::Skip(size_t n) { … }

class SnappyCodec final : public Codec { … };

std::unique_ptr<Codec> SnappyCodec::create(int level, CodecType type) { … }

SnappyCodec::SnappyCodec(int level, CodecType type) : … { … }

uint64_t SnappyCodec::doMaxUncompressedLength() const { … }

uint64_t SnappyCodec::doMaxCompressedLength(uint64_t uncompressedLength) const { … }

std::unique_ptr<IOBuf> SnappyCodec::doCompress(const IOBuf* data) { … }

std::unique_ptr<IOBuf> SnappyCodec::doUncompress(
    const IOBuf* data, Optional<uint64_t> uncompressedLength) { … }

folly::Optional<uint64_t> SnappyCodec::doGetUncompressedLength(
    const folly::IOBuf* data,
    folly::Optional<uint64_t> uncompressedLength) const { … }

#endif // FOLLY_HAVE_LIBSNAPPY

#if FOLLY_HAVE_LIBLZMA

/**
 * LZMA2 compression
 */
class LZMA2StreamCodec final : public StreamCodec { … };

constexpr uint64_t kLZMA2MagicLE = …;
constexpr unsigned kLZMA2MagicBytes = …;

std::vector<std::string> LZMA2StreamCodec::validPrefixes() const { … }

bool LZMA2StreamCodec::doNeedsDataLength() const { … }

bool LZMA2StreamCodec::canUncompress(
    const IOBuf* data, Optional<uint64_t>) const { … }

std::unique_ptr<Codec> LZMA2StreamCodec::createCodec(
    int level, CodecType type) { … }

std::unique_ptr<StreamCodec> LZMA2StreamCodec::createStream(
    int level, CodecType type) { … }

LZMA2StreamCodec::LZMA2StreamCodec(int level, CodecType type)
    : … { … }

LZMA2StreamCodec::~LZMA2StreamCodec() { … }

uint64_t LZMA2StreamCodec::doMaxUncompressedLength() const { … }

uint64_t LZMA2StreamCodec::doMaxCompressedLength(
    uint64_t uncompressedLength) const { … }

void LZMA2StreamCodec::doResetStream() { … }

void LZMA2StreamCodec::resetCStream() { … }

void LZMA2StreamCodec::resetDStream() { … }

lzma_ret lzmaThrowOnError(lzma_ret const rc) { … }

lzma_action lzmaTranslateFlush(StreamCodec::FlushOp flush) { … }

/**
 * Flushes the varint buffer.
 * Advances output by the number of bytes written.
 * Returns true when flushing is complete.
 */
bool LZMA2StreamCodec::flushVarintBuffer(MutableByteRange& output) { … }

bool LZMA2StreamCodec::doCompressStream(
    ByteRange& input, MutableByteRange& output, StreamCodec::FlushOp flushOp) { … }

/**
 * Attempts to decode a varint from input.
 * The function advances input by the number of bytes read.
 *
 * If there are too many bytes and the varint is not valid, throw a
 * runtime_error.
 *
 * If the uncompressed length was provided and a decoded varint does not match
 * the provided length, throw a runtime_error.
 *
 * Returns true if the varint was successfully decoded and matches the
 * uncompressed length if provided, and false if more bytes are needed.
 */
bool LZMA2StreamCodec::decodeAndCheckVarint(ByteRange& input) { … }

bool LZMA2StreamCodec::doUncompressStream(
    ByteRange& input, MutableByteRange& output, StreamCodec::FlushOp flushOp) { … }
#endif // FOLLY_HAVE_LIBLZMA

#if FOLLY_HAVE_LIBZSTD

int zstdConvertLevel(int level) { … }

int zstdFastConvertLevel(int level) { … }

std::unique_ptr<Codec> getZstdCodec(int level, CodecType type) { … }

std::unique_ptr<StreamCodec> getZstdStreamCodec(int level, CodecType type) { … }

std::unique_ptr<Codec> getZstdFastCodec(int level, CodecType type) { … }

std::unique_ptr<StreamCodec> getZstdFastStreamCodec(int level, CodecType type) { … }

#endif // FOLLY_HAVE_LIBZSTD

#if FOLLY_HAVE_LIBBZ2

class Bzip2StreamCodec final : public StreamCodec { … };

/* static */ std::unique_ptr<Codec> Bzip2StreamCodec::createCodec(
    int level, CodecType type) { … }

/* static */ std::unique_ptr<StreamCodec> Bzip2StreamCodec::createStream(
    int level, CodecType type) { … }

Bzip2StreamCodec::Bzip2StreamCodec(int level, CodecType type)
    : … { … }

uint32_t constexpr kBzip2MagicLE = …;
uint64_t constexpr kBzip2MagicBytes = …;

std::vector<std::string> Bzip2StreamCodec::validPrefixes() const { … }

bool Bzip2StreamCodec::canUncompress(
    IOBuf const* data, Optional<uint64_t>) const { … }

uint64_t Bzip2StreamCodec::doMaxCompressedLength(
    uint64_t uncompressedLength) const { … }

bz_stream createBzStream() { … }

// Throws on error condition, otherwise returns the code.
int bzCheck(int const rc) { … }

Bzip2StreamCodec::~Bzip2StreamCodec() { … }

void Bzip2StreamCodec::doResetStream() { … }

void Bzip2StreamCodec::resetCStream() { … }

int bzip2TranslateFlush(StreamCodec::FlushOp flushOp) { … }

bool Bzip2StreamCodec::doCompressStream(
    ByteRange& input, MutableByteRange& output, StreamCodec::FlushOp flushOp) { … }

void Bzip2StreamCodec::resetDStream() { … }

bool Bzip2StreamCodec::doUncompressStream(
    ByteRange& input, MutableByteRange& output, StreamCodec::FlushOp flushOp) { … }

#endif // FOLLY_HAVE_LIBBZ2

#if FOLLY_HAVE_LIBZ

zlib::Options getZlibOptions(CodecType type) { … }

std::unique_ptr<Codec> getZlibCodec(int level, CodecType type) { … }

std::unique_ptr<StreamCodec> getZlibStreamCodec(int level, CodecType type) { … }

#endif // FOLLY_HAVE_LIBZ

/**
 * Automatic decompression
 */
class AutomaticCodec final : public Codec { … };

std::vector<std::string> AutomaticCodec::validPrefixes() const { … }

bool AutomaticCodec::canUncompress(
    const IOBuf* data, Optional<uint64_t> uncompressedLength) const { … }

void AutomaticCodec::addCodecIfSupported(CodecType type) { … }

/* static */ std::unique_ptr<Codec> AutomaticCodec::create(
    std::vector<std::unique_ptr<Codec>> customCodecs,
    std::unique_ptr<Codec> terminalCodec) { … }

AutomaticCodec::AutomaticCodec(
    std::vector<std::unique_ptr<Codec>> customCodecs,
    std::unique_ptr<Codec> terminalCodec)
    : … { … }

void AutomaticCodec::checkCompatibleCodecs() const { … }

bool AutomaticCodec::doNeedsUncompressedLength() const { … }

uint64_t AutomaticCodec::doMaxUncompressedLength() const { … }

std::unique_ptr<IOBuf> AutomaticCodec::doUncompress(
    const IOBuf* data, Optional<uint64_t> uncompressedLength) { … }

CodecFactory;
StreamCodecFactory;
struct Factory { … };

constexpr Factory
    codecFactories[static_cast<size_t>(CodecType::NUM_CODEC_TYPES)] = …;

Factory const& getFactory(CodecType type) { … }
} // namespace

bool hasCodec(CodecType type) { … }

std::unique_ptr<Codec> getCodec(CodecType type, int level) { … }

bool hasStreamCodec(CodecType type) { … }

std::unique_ptr<StreamCodec> getStreamCodec(CodecType type, int level) { … }

std::unique_ptr<Codec> getAutoUncompressionCodec(
    std::vector<std::unique_ptr<Codec>> customCodecs,
    std::unique_ptr<Codec> terminalCodec) { … }
} // namespace compression
} // namespace folly
folly/folly/compression/Compression.cpp
