BloscCompressor Class Reference

#include <Compressor.h>

Collaboration diagram for BloscCompressor:

Public Member Functions
size_t	getScratchSpaceSize (const size_t len) const
int64_t	compress (const uint8_t *buffer, const size_t buffer_size, uint8_t *compressed_buffer, const size_t compressed_buffer_size, const size_t min_compressor_bytes)
std::string	compress (const std::string &buffer)
size_t	decompress (const uint8_t *compressed_buffer, uint8_t *decompressed_buffer, const size_t decompressed_size)
std::string	decompress (const std::string &buffer, const size_t decompressed_size)
size_t	compressOrMemcpy (const uint8_t *input_buffer, uint8_t *output_buffer, const size_t uncompressed_size, const size_t min_compressor_bytes)
bool	decompressOrMemcpy (const uint8_t *compressed_buffer, const size_t compressed_buffer_size, uint8_t *decompressed_buffer, const size_t decompressed_size)
void	getBloscBufferSizes (const uint8_t *data_ptr, size_t *num_bytes_compressed, size_t *num_bytes_uncompressed, size_t *block_size)
int	setThreads (size_t num_threads)
int	setCompressor (std::string &compressor)
	~BloscCompressor ()

Static Public Member Functions
static BloscCompressor *	getCompressor ()

Private Member Functions
	BloscCompressor ()

Private Attributes
std::mutex	compressor_lock

Static Private Attributes
static BloscCompressor *	instance = NULL

Detailed Description

Definition at line 38 of file Compressor.h.

Constructor & Destructor Documentation

BloscCompressor::~BloscCompressor

(

)

Definition at line 51 of file Compressor.cpp.

References compressor_lock.

                                  {
  std::lock_guard<std::mutex> compressor_lock_(compressor_lock);
  blosc_destroy();
}

BloscCompressor::BloscCompressor ( )

private

Definition at line 37 of file Compressor.cpp.

References compressor_lock.

Referenced by getCompressor().

                                 {
  std::lock_guard<std::mutex> compressor_lock_(compressor_lock);
  blosc_init();
  // We use maximum number of threads here since with tests we found that compression
  // speed gets lear scalling with corresponding to the number of threads being used.

  blosc_set_nthreads(std::thread::hardware_concurrency());

  // We chosse faster compressor, accepting slightly lower compression ratio
  // https://lz4.github.io/lz4/

  blosc_set_compressor(BLOSC_LZ4HC_COMPNAME);
}

Here is the caller graph for this function:

Member Function Documentation

int64_t BloscCompressor::compress	(	const uint8_t *	buffer,
		const size_t	buffer_size,
		uint8_t *	compressed_buffer,
		const size_t	compressed_buffer_size,
		const size_t	min_compressor_bytes = g_compression_limit_bytes
	)

Definition at line 56 of file Compressor.cpp.

References compressor_lock, and to_string().

Referenced by compress(), and compressOrMemcpy().

                                                                   {
  if (compressed_buffer_size < BLOSC_MIN_HEADER_LENGTH) {
    // Blosc compressor checks this condition during the initialization
    // and throw "Output buffer size should be larger than 16 bytes" error
    // if compressed_buffer_size < 16 (BLOSC_MIN_HEADER_LENGTH)
    // but after sending interrupt signal, blosc compress function hangs until
    // thrift timed out error and could not check this code.
    // here, we can early return by explicitly checking this condition
    // so as to avoid hangs in query runtime
    return 0;
  }

  if (buffer_size < min_compressor_bytes && min_compressor_bytes != 0) {
    return 0;
  }
  std::lock_guard<std::mutex> compressor_lock_(compressor_lock);
  const auto compressed_len = blosc_compress(5,
                                             1,
                                             sizeof(unsigned char),
                                             buffer_size,
                                             buffer,
                                             &compressed_buffer[0],
                                             compressed_buffer_size);

  if (compressed_len <= 0) {
    // something went wrong. blosc retrun codes simply don't provide enough information
    // for us to decide what.
    throw CompressionFailedError(std::string("failed to compress result set of length ") +
                                 std::to_string(buffer_size));
  }
  // we need to tell the other endpoint the size of the acctual data so it can
  // decide whether it should decompress data or not. So we pass the original
  // data length. and only send the compressed result if the output of the
  // compressed result is smaller than the original
  return compressed_len;
}

Here is the call graph for this function:

Here is the caller graph for this function:

std::string BloscCompressor::compress

(

const std::string &

buffer

)

Definition at line 98 of file Compressor.cpp.

References compress(), and getScratchSpaceSize().

                                                           {
  const auto buffer_size = buffer.size();
  std::vector<uint8_t> compressed_buffer(getScratchSpaceSize(buffer_size));
  try {
    const size_t compressed_len = compress((uint8_t*)buffer.c_str(),
                                           buffer_size,
                                           &compressed_buffer[0],
                                           getScratchSpaceSize(buffer_size));
    if (compressed_len > 0 && compressed_len < buffer_size) {
      // we need to tell the other endpoint the size of the acctual data so it can
      // decide whether it should decompress data or not. So we pass the original
      // data length. and only send the compressed result if the output of the
      // compressed result is smaller than the original
      compressed_buffer.resize(compressed_len);
      return {compressed_buffer.begin(), compressed_buffer.end()};
    }
  } catch (const CompressionFailedError&) {
  }
  return buffer;
}

Here is the call graph for this function:

size_t BloscCompressor::compressOrMemcpy	(	const uint8_t *	input_buffer,
		uint8_t *	output_buffer,
		const size_t	uncompressed_size,
		const size_t	min_compressor_bytes
	)

Definition at line 160 of file Compressor.cpp.

References compress(), LOG, and logger::WARNING.

                                                                            {
  try {
    const auto compressed_size = compress(input_buffer,
                                          uncompressed_size,
                                          output_buffer,
                                          uncompressed_size,
                                          min_compressor_bytes);
    if (compressed_size > 0) {
      return compressed_size;
    }
  } catch (const CompressionFailedError&) {
    // catch exceptions from blosc
    // we copy regardless what happens in compressor
    if (uncompressed_size > min_compressor_bytes) {
      LOG(WARNING) << "Compressor failed for byte size of " << uncompressed_size;
    }
  }
  memcpy(output_buffer, input_buffer, uncompressed_size);
  return uncompressed_size;
}

Here is the call graph for this function:

size_t BloscCompressor::decompress	(	const uint8_t *	compressed_buffer,
		uint8_t *	decompressed_buffer,
		const size_t	decompressed_size
	)

Definition at line 119 of file Compressor.cpp.

References compressor_lock, getBloscBufferSizes(), and to_string().

Referenced by decompress(), and decompressOrMemcpy().

                                                                   {
  size_t decompressed_buf_len, compressed_buf_len, block_size, decompressed_len = 0;
  getBloscBufferSizes(
      &compressed_buffer[0], &compressed_buf_len, &decompressed_buf_len, &block_size);
  // check compressed buffer is a blosc compressed buffer.
  if (compressed_buf_len > 0 && decompressed_size == decompressed_buf_len) {
    std::lock_guard<std::mutex> compressor_lock_(compressor_lock);
    decompressed_len =
        blosc_decompress(&compressed_buffer[0], decompressed_buffer, decompressed_size);
  }

  if (decompressed_len == 0) {
    throw CompressionFailedError(
        std::string("failed to decompress buffer for compressed size: ") +
        std::to_string(compressed_buf_len));
  }
  if (decompressed_len != decompressed_size) {
    throw CompressionFailedError(
        std::string("decompression buffer size mismatch. Decompressed buffer length: ") +
        std::to_string(decompressed_len));
  }
  return decompressed_len;
}

Here is the call graph for this function:

Here is the caller graph for this function:

std::string BloscCompressor::decompress	(	const std::string &	buffer,
		const size_t	decompressed_size
	)

Definition at line 145 of file Compressor.cpp.

References decompress().

                                                                        {
  std::vector<uint8_t> decompressed_buffer(decompressed_size);
  if (buffer.size() == decompressed_size) {
    return buffer;
  }
  try {
    decompress(
        (uint8_t*)&buffer[0], (uint8_t*)&decompressed_buffer[0], decompressed_size);
    return {decompressed_buffer.begin(), decompressed_buffer.end()};
  } catch (const CompressionFailedError&) {
  }
  return buffer;
}

Here is the call graph for this function:

bool BloscCompressor::decompressOrMemcpy	(	const uint8_t *	compressed_buffer,
		const size_t	compressed_buffer_size,
		uint8_t *	decompressed_buffer,
		const size_t	decompressed_size
	)

Definition at line 184 of file Compressor.cpp.

References decompress().

                                                                         {
  try {
    decompress(compressed_buffer, decompressed_buffer, decompressed_size);
    return true;
  } catch (const CompressionFailedError&) {
    // we will memcpy if we find that the buffer is not compressed

    if (compressed_size > decompressed_size) {
      throw std::runtime_error(
          "compressed buffer size is greater than decompressed buffer size.");
    }
  }
  memcpy(decompressed_buffer, compressed_buffer, decompressed_size);
  return false;
}

Here is the call graph for this function:

void BloscCompressor::getBloscBufferSizes	(	const uint8_t *	data_ptr,
		size_t *	num_bytes_compressed,
		size_t *	num_bytes_uncompressed,
		size_t *	block_size
	)

Definition at line 203 of file Compressor.cpp.

Referenced by decompress().

                                                              {
  blosc_cbuffer_sizes(data_ptr, num_bytes_uncompressed, num_bytes_compressed, block_size);
}

Here is the caller graph for this function:

BloscCompressor * BloscCompressor::getCompressor ( )

static

Definition at line 212 of file Compressor.cpp.

References BloscCompressor(), and instance.

                                                {
  static std::mutex compressor_singleton_lock;
  std::lock_guard<std::mutex> singleton_lock(compressor_singleton_lock);
  if (instance == NULL) {
    instance = new BloscCompressor();
  }

  return instance;
}

Here is the call graph for this function:

size_t BloscCompressor::getScratchSpaceSize ( const size_t  len ) const

inline

Definition at line 46 of file Compressor.h.

Referenced by compress().

                                                            {
    return static_cast<size_t>(len * 1.1);
  }

Here is the caller graph for this function:

int BloscCompressor::setCompressor

(

std::string &

compressor

)

Definition at line 227 of file Compressor.cpp.

References compressor_lock.

                                                             {
  std::lock_guard<std::mutex> compressor_lock_(compressor_lock);
  // Blosc is resilent enough to detect that the comprressor that was provided to it was
  // supported or not. If the compressor is invalid or not supported it will simply keep
  // current compressor.
  return blosc_set_compressor(compressor_name.c_str());
}

int BloscCompressor::setThreads

(

size_t

num_threads

)

Definition at line 222 of file Compressor.cpp.

References compressor_lock.

                                                  {
  std::lock_guard<std::mutex> compressor_lock_(compressor_lock);
  return blosc_set_nthreads(static_cast<int>(num_threads));
}

Member Data Documentation

std::mutex BloscCompressor::compressor_lock

private

Definition at line 87 of file Compressor.h.

Referenced by BloscCompressor(), compress(), decompress(), setCompressor(), setThreads(), and ~BloscCompressor().

BloscCompressor * BloscCompressor::instance = NULL

staticprivate

Definition at line 88 of file Compressor.h.

Referenced by getCompressor().

---

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

• /home/jenkins-slave/workspace/core-os-doxygen/Shared/Compressor.h
• /home/jenkins-slave/workspace/core-os-doxygen/Shared/Compressor.cpp