folly::io::StreamCodec Class Reference

#include <Compression.h>

Inheritance diagram for folly::io::StreamCodec:

Public Types
enum	FlushOp { FlushOp::NONE, FlushOp::FLUSH, FlushOp::END }

Public Member Functions
	~StreamCodec () override
bool	needsDataLength () const
void	resetStream (folly::Optional< uint64_t > uncompressedLength=folly::none)
bool	compressStream (folly::ByteRange &input, folly::MutableByteRange &output, FlushOp flushOp=StreamCodec::FlushOp::NONE)
bool	uncompressStream (folly::ByteRange &input, folly::MutableByteRange &output, FlushOp flushOp=StreamCodec::FlushOp::NONE)
Public Member Functions inherited from folly::io::Codec
virtual	~Codec ()
uint64_t	maxUncompressedLength () const
CodecType	type () const
bool	needsUncompressedLength () const
std::unique_ptr< IOBuf >	compress (const folly::IOBuf *data)
std::string	compress (StringPiece data)
std::unique_ptr< IOBuf >	uncompress (const IOBuf *data, folly::Optional< uint64_t > uncompressedLength=folly::none)
std::string	uncompress (StringPiece data, folly::Optional< uint64_t > uncompressedLength=folly::none)
uint64_t	maxCompressedLength (uint64_t uncompressedLength) const
folly::Optional< uint64_t >	getUncompressedLength (const folly::IOBuf *data, folly::Optional< uint64_t > uncompressedLength=folly::none) const
virtual std::vector< std::string >	validPrefixes () const
virtual bool	canUncompress (const folly::IOBuf *data, folly::Optional< uint64_t > uncompressedLength=folly::none) const

Protected Member Functions
	StreamCodec (CodecType type, folly::Optional< int > level=folly::none, folly::StringPiece name={}, bool counters=true)
folly::Optional< uint64_t >	uncompressedLength () const
Protected Member Functions inherited from folly::io::Codec
	Codec (CodecType type, folly::Optional< int > level=folly::none, folly::StringPiece name={}, bool counters=true)

Private Types
enum	State {   State::RESET, State::COMPRESS, State::COMPRESS_FLUSH, State::COMPRESS_END,   State::UNCOMPRESS, State::END }

Private Member Functions
std::unique_ptr< IOBuf >	doCompress (const folly::IOBuf *data) override
std::unique_ptr< IOBuf >	doUncompress (const folly::IOBuf *data, folly::Optional< uint64_t > uncompressedLength) override
virtual bool	doNeedsDataLength () const
virtual void	doResetStream ()=0
virtual bool	doCompressStream (folly::ByteRange &input, folly::MutableByteRange &output, FlushOp flushOp)=0
virtual bool	doUncompressStream (folly::ByteRange &input, folly::MutableByteRange &output, FlushOp flushOp)=0
void	assertStateIs (State expected) const

Private Attributes
State	state_ {State::RESET}
ByteRange	previousInput_ {}
folly::Optional< uint64_t >	uncompressedLength_ {}
bool	progressMade_ {true}

Additional Inherited Members
Static Public Attributes inherited from folly::io::Codec
static constexpr uint64_t	UNLIMITED_UNCOMPRESSED_LENGTH = uint64_t(-1)

Detailed Description

Definition at line 274 of file Compression.h.

Member Enumeration Documentation

enum folly::io::StreamCodec::FlushOp

strong

Enumerator
NONE
FLUSH
END

Definition at line 311 of file Compression.h.

{ NONE, FLUSH, END };

enum folly::io::StreamCodec::State

strongprivate

Enumerator
RESET
COMPRESS
COMPRESS_FLUSH
COMPRESS_END
UNCOMPRESS
END

Definition at line 424 of file Compression.h.

                   {
    RESET,
    COMPRESS,
    COMPRESS_FLUSH,
    COMPRESS_END,
    UNCOMPRESS,
    END,
  };

Constructor & Destructor Documentation

folly::io::StreamCodec::~StreamCodec ( )

inlineoverride

Definition at line 276 of file Compression.h.

References folly::none.

{}

folly::io::StreamCodec::StreamCodec ( CodecType  type, folly::Optional< int >  level = folly::none, folly::StringPiece  name = {}, bool  counters = true  )

inlineprotected

Definition at line 392 of file Compression.h.

References folly::gen::move, and name.

                              {},
      bool counters = true)
      : Codec(type, std::move(level), name, counters) {}

Member Function Documentation

void folly::io::StreamCodec::assertStateIs ( State  expected ) const

private

Definition at line 322 of file Compression.cpp.

                                                    {
  if (state_ != expected) {
    throw std::logic_error(folly::to<std::string>(
        "Codec: state is ", state_, "; expected state ", expected));
  }
}

bool folly::io::StreamCodec::compressStream	(	folly::ByteRange &	input,
		folly::MutableByteRange &	output,
		StreamCodec::FlushOp	flushOp = StreamCodec::FlushOp::NONE
	)

Compresses some data from the input buffer and writes the compressed data into the output buffer. It may read input without producing any output, except when forced to flush.

The input buffer is advanced to point to the range of data that hasn't yet been read. Compression will resume at this point for the next call to compressStream(). The output buffer is advanced one byte past the last byte written.

The default flushOp is NONE, which allows compressStream() complete discretion in how much data to gather before writing any output.

If flushOp is END, all pending and input data is flushed to the output buffer, and the frame is ended. compressStream() must be called with the same input and flushOp END until it returns true. At this point the caller must call resetStream() to use the codec again.

If flushOp is FLUSH, all pending and input data is flushed to the output buffer, but the frame is not ended. compressStream() must be called with the same input and flushOp END until it returns true. At this point the caller can continue to compressStream() with any input data and flushOp. The uncompressor, if passed all the produced output data, will be able to uncompress all the input data passed to compressStream() so far. Excessive use of flushOp FLUSH will deteriorate compression ratio. This is useful for stateful streaming across a network. Most users don't need to use this flushOp.

A std::logic_error is thrown on incorrect usage of the API. A std::runtime_error is thrown upon error conditions or if no forward progress could be made twice in a row.

Definition at line 336 of file Compression.cpp.

References folly::Range< Iter >::empty(), END, FLUSH, NONE, folly::Range< Iter >::size(), and uint64_t.

Referenced by folly::io::test::compressSome().

                                {
  if (state_ == State::RESET && input.empty() &&
      flushOp == StreamCodec::FlushOp::END &&
      uncompressedLength().value_or(0) != 0) {
    throw std::runtime_error("Codec: invalid uncompressed length");
  }

  if (!uncompressedLength() && needsDataLength()) {
    throw std::runtime_error("Codec: uncompressed length required");
  }
  if (state_ == State::RESET && !input.empty() &&
      uncompressedLength() == uint64_t(0)) {
    throw std::runtime_error("Codec: invalid uncompressed length");
  }
  // Handle input state transitions
  switch (flushOp) {
    case StreamCodec::FlushOp::NONE:
      if (state_ == State::RESET) {
        state_ = State::COMPRESS;
      }
      assertStateIs(State::COMPRESS);
      break;
    case StreamCodec::FlushOp::FLUSH:
      if (state_ == State::RESET || state_ == State::COMPRESS) {
        state_ = State::COMPRESS_FLUSH;
      }
      assertStateIs(State::COMPRESS_FLUSH);
      break;
    case StreamCodec::FlushOp::END:
      if (state_ == State::RESET || state_ == State::COMPRESS) {
        state_ = State::COMPRESS_END;
      }
      assertStateIs(State::COMPRESS_END);
      break;
  }
  size_t const inputSize = input.size();
  size_t const outputSize = output.size();
  bool const done = doCompressStream(input, output, flushOp);
  if (!done && inputSize == input.size() && outputSize == output.size()) {
    if (!progressMade_) {
      throw std::runtime_error("Codec: No forward progress made");
    }
    // Throw an exception if there is no progress again next time
    progressMade_ = false;
  } else {
    progressMade_ = true;
  }
  // Handle output state transitions
  if (done) {
    if (state_ == State::COMPRESS_FLUSH) {
      state_ = State::COMPRESS;
    } else if (state_ == State::COMPRESS_END) {
      state_ = State::END;
    }
    // Check internal invariants
    DCHECK(input.empty());
    DCHECK(flushOp != StreamCodec::FlushOp::NONE);
  }
  return done;
}

std::unique_ptr< IOBuf > folly::io::StreamCodec::doCompress ( const folly::IOBuf *  data )

overrideprivatevirtual

Implements folly::io::Codec.

Definition at line 444 of file Compression.cpp.

References folly::io::addOutputBuffer(), buffer(), folly::IOBuf::computeChainDataLength(), current, folly::data(), folly::IOBuf::data(), folly::Range< Iter >::empty(), END, folly::IOBuf::length(), folly::io::Codec::maxCompressedLength(), folly::IOBuf::next(), NONE, gmock_output_test::output, folly::Range< Iter >::size(), and uint64_t.

                                                              {
  uint64_t const uncompressedLength = data->computeChainDataLength();
  resetStream(uncompressedLength);
  uint64_t const maxCompressedLen = maxCompressedLength(uncompressedLength);

  auto constexpr kMaxSingleStepLength = uint64_t(64) << 20; // 64 MB
  auto constexpr kDefaultBufferLength = uint64_t(4) << 20; // 4 MB

  MutableByteRange output;
  auto buffer = addOutputBuffer(
      output,
      maxCompressedLen <= kMaxSingleStepLength ? maxCompressedLen
                                               : kDefaultBufferLength);

  // Compress the entire IOBuf chain into the IOBuf chain pointed to by buffer
  IOBuf const* current = data;
  ByteRange input{current->data(), current->length()};
  StreamCodec::FlushOp flushOp = StreamCodec::FlushOp::NONE;
  bool done = false;
  while (!done) {
    while (input.empty() && current->next() != data) {
      current = current->next();
      input = {current->data(), current->length()};
    }
    if (current->next() == data) {
      // This is the last input buffer so end the stream
      flushOp = StreamCodec::FlushOp::END;
    }
    if (output.empty()) {
      buffer->prependChain(addOutputBuffer(output, kDefaultBufferLength));
    }
    done = compressStream(input, output, flushOp);
    if (done) {
      DCHECK(input.empty());
      DCHECK(flushOp == StreamCodec::FlushOp::END);
      DCHECK_EQ(current->next(), data);
    }
  }
  buffer->prev()->trimEnd(output.size());
  return buffer;
}

virtual bool folly::io::StreamCodec::doCompressStream ( folly::ByteRange &  input, folly::MutableByteRange &  output, FlushOp  flushOp  )

privatepure virtual

bool folly::io::StreamCodec::doNeedsDataLength ( ) const

privatevirtual

Definition at line 318 of file Compression.cpp.

                                          {
  return false;
}

virtual void folly::io::StreamCodec::doResetStream ( )

privatepure virtual

std::unique_ptr< IOBuf > folly::io::StreamCodec::doUncompress ( const folly::IOBuf *  data, folly::Optional< uint64_t >  uncompressedLength  )

overrideprivatevirtual

Implements folly::io::Codec.

Definition at line 494 of file Compression.cpp.

References folly::io::addOutputBuffer(), folly::Range< Iter >::advance(), folly::IOBuf::append(), b, folly::IOBuf::begin(), buffer(), folly::io::BZIP2, folly::IOBufQueue::cacheChainLength(), folly::io::Codec::canUncompress(), folly::IOBufQueue::chainLength(), folly::IOBuf::clone(), folly::IOBuf::cloneCoalescedAsValue(), folly::IOBuf::coalesce(), folly::io::Codec::Codec(), folly::io::COMPRESSION_LEVEL_BEST, folly::io::COMPRESSION_LEVEL_DEFAULT, folly::io::COMPRESSION_LEVEL_FASTEST, folly::io::computeBufferLength(), folly::IOBuf::computeChainDataLength(), folly::IOBuf::create(), current, folly::data(), folly::Range< Iter >::data(), folly::IOBuf::data(), folly::io::compression::detail::dataStartsWithLE(), folly::io::Codec::doCompress(), folly::io::Codec::doMaxCompressedLength(), folly::io::Codec::doMaxUncompressedLength(), folly::io::Codec::doNeedsUncompressedLength(), folly::io::Codec::doUncompress(), folly::Range< Iter >::empty(), folly::encodeVarint(), END, FLUSH, folly::io::getCodec(), folly::io::getStreamCodec(), folly::io::Codec::getUncompressedLength(), folly::io::GZIP, folly::Optional< Value >::hasValue(), int8_t, folly::IOBuf::isChained(), folly::kMaxVarintLength64, folly::IOBuf::length(), folly::io::LZ4, folly::io::LZ4_FRAME, folly::io::LZ4_VARINT_SIZE, folly::io::LZMA2, folly::io::LZMA2_VARINT_SIZE, max, folly::io::Codec::maxCompressedLength(), folly::io::Codec::maxUncompressedLength(), min, folly::IOBufQueue::move(), folly::IOBuf::next(), folly::io::NO_COMPRESSION, NONE, gmock_output_test::output, folly::io::detail::CursorBase< Derived, BufType >::peekBytes(), folly::IOBufQueue::postallocate(), folly::IOBufQueue::preallocate(), folly::io::compression::detail::prefixToStringLE(), folly::range(), folly::io::detail::CursorBase< Derived, BufType >::read(), SCOPE_EXIT, folly::Range< Iter >::size(), folly::Skip, folly::io::SNAPPY, stream, folly::IOBuf::tailroom(), folly::TooManyBytes, folly::tryDecodeVarint(), folly::io::Codec::type(), uint32_t, uint64_t, uint8_t, folly::Range< Iter >::uncheckedAdvance(), val, folly::io::Codec::validPrefixes(), folly::IOBuf::writableTail(), folly::io::ZLIB, folly::io::ZSTD, and folly::io::ZSTD_FAST.

                                           {
  auto constexpr kMaxSingleStepLength = uint64_t(64) << 20; // 64 MB
  auto constexpr kBlockSize = uint64_t(128) << 10;
  auto const defaultBufferLength =
      computeBufferLength(data->computeChainDataLength(), kBlockSize);

  uncompressedLength = getUncompressedLength(data, uncompressedLength);
  resetStream(uncompressedLength);

  MutableByteRange output;
  auto buffer = addOutputBuffer(
      output,
      (uncompressedLength && *uncompressedLength <= kMaxSingleStepLength
           ? *uncompressedLength
           : defaultBufferLength));

  // Uncompress the entire IOBuf chain into the IOBuf chain pointed to by buffer
  IOBuf const* current = data;
  ByteRange input{current->data(), current->length()};
  StreamCodec::FlushOp flushOp = StreamCodec::FlushOp::NONE;
  bool done = false;
  while (!done) {
    while (input.empty() && current->next() != data) {
      current = current->next();
      input = {current->data(), current->length()};
    }
    if (current->next() == data) {
      // Tell the uncompressor there is no more input (it may optimize)
      flushOp = StreamCodec::FlushOp::END;
    }
    if (output.empty()) {
      buffer->prependChain(addOutputBuffer(output, defaultBufferLength));
    }
    done = uncompressStream(input, output, flushOp);
  }
  if (!input.empty()) {
    throw std::runtime_error("Codec: Junk after end of data");
  }

  buffer->prev()->trimEnd(output.size());
  if (uncompressedLength &&
      *uncompressedLength != buffer->computeChainDataLength()) {
    throw std::runtime_error("Codec: invalid uncompressed length");
  }

  return buffer;
}

virtual bool folly::io::StreamCodec::doUncompressStream ( folly::ByteRange &  input, folly::MutableByteRange &  output, FlushOp  flushOp  )

privatepure virtual

bool folly::io::StreamCodec::needsDataLength

(

)

const

Does the codec need the data length before compression streaming?

Definition at line 314 of file Compression.cpp.

                                        {
  return doNeedsDataLength();
}

void folly::io::StreamCodec::resetStream

(

folly::Optional< uint64_t >

uncompressedLength = folly::none

)

Reset the state of the codec, and set the uncompressed length for the next streaming operation. If uncompressedLength is not none it must be exactly the uncompressed length. compressStream() must be passed exactly uncompressedLength input bytes before the stream is ended. uncompressStream() must be passed a compressed frame that uncompresses to uncompressedLength.

Definition at line 329 of file Compression.cpp.

                                                                   {
  state_ = State::RESET;
  uncompressedLength_ = uncompressedLength;
  progressMade_ = true;
  doResetStream();
}

folly::Optional<uint64_t> folly::io::StreamCodec::uncompressedLength ( ) const

inlineprotected

Definition at line 401 of file Compression.h.

References folly::data().

                                                     {
    return uncompressedLength_;
  }

bool folly::io::StreamCodec::uncompressStream	(	folly::ByteRange &	input,
		folly::MutableByteRange &	output,
		StreamCodec::FlushOp	flushOp = StreamCodec::FlushOp::NONE
	)

Uncompresses some data from the input buffer and writes the uncompressed data into the output buffer. It may read input without producing any output.

The input buffer is advanced to point to the range of data that hasn't yet been read. Uncompression will resume at this point for the next call to uncompressStream(). The output buffer is advanced one byte past the last byte written.

The default flushOp is NONE, which allows uncompressStream() complete discretion in how much output data to flush. The uncompressor may not make maximum forward progress, but will make some forward progress when possible.

If flushOp is END, the caller guarantees that no more input will be presented to uncompressStream(). uncompressStream() must be called with the same input and flushOp END until it returns true. This is not mandatory, but if the input is all available in one buffer, and there is enough output space to write the entire frame, codecs can uncompress faster.

If flushOp is FLUSH, uncompressStream() is guaranteed to make the maximum amount of forward progress possible. When using this flushOp and uncompressStream() returns with !output.empty() the caller knows that all pending output has been flushed. This is useful for stateful streaming across a network, and it should be used in conjunction with compressStream() with flushOp FLUSH. Most users don't need to use this flushOp.

A std::runtime_error is thrown upon error conditions or if no forward progress could be made upon two consecutive calls to the function (only the second call will throw an exception).

Returns true at the end of a frame. At this point resetStream() must be called to reuse the codec.

Definition at line 400 of file Compression.cpp.

References folly::Range< Iter >::empty(), and folly::Range< Iter >::size().

Referenced by folly::io::test::uncompressSome().

                                {
  if (state_ == State::RESET && input.empty()) {
    if (uncompressedLength().value_or(0) == 0) {
      return true;
    }
    return false;
  }
  // Handle input state transitions
  if (state_ == State::RESET) {
    state_ = State::UNCOMPRESS;
  }
  assertStateIs(State::UNCOMPRESS);
  size_t const inputSize = input.size();
  size_t const outputSize = output.size();
  bool const done = doUncompressStream(input, output, flushOp);
  if (!done && inputSize == input.size() && outputSize == output.size()) {
    if (!progressMade_) {
      throw std::runtime_error("Codec: no forward progress made");
    }
    // Throw an exception if there is no progress again next time
    progressMade_ = false;
  } else {
    progressMade_ = true;
  }
  // Handle output state transitions
  if (done) {
    state_ = State::END;
  }
  return done;
}

Member Data Documentation

ByteRange folly::io::StreamCodec::previousInput_ {}

private

Definition at line 435 of file Compression.h.

bool folly::io::StreamCodec::progressMade_ {true}

private

Definition at line 437 of file Compression.h.

State folly::io::StreamCodec::state_ {State::RESET}

private

Definition at line 434 of file Compression.h.

folly::Optional<uint64_t> folly::io::StreamCodec::uncompressedLength_ {}

private

Definition at line 436 of file Compression.h.

---

The documentation for this class was generated from the following files:

• proxygen/folly/folly/compression/Compression.h
• proxygen/folly/folly/compression/Compression.cpp