source: branches/threaded/src/blosc.c @ 55

/*********************************************************************
  Blosc - Blocked Suffling and Compression Library

  Author: Francesc Alted (faltet@pytables.org)
  Creation date: 2009-05-20

  See LICENSES/BLOSC.txt for details about copyright and rights to use.
**********************************************************************/


#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <assert.h>
#include "blosc.h"
#include "blosclz.h"
#include "shuffle.h"
#ifdef _WIN32
  #include <windows.h>
#else
  #include <stdint.h>
  #include <unistd.h>
  #include <pthread.h>
#endif  /* _WIN32 */


/* Starting point for the blocksize computation */
#define BLOCKSIZE (4*1024)      /* 4 KB (page size) */

/* Maximum typesize before considering buffer as a stream of bytes. */
#define MAXTYPESIZE 256         /* Cannot be larger than 256 */

/* The maximum number of splits in a block for compression */
#define MAXSPLITS 16         /* Cannot be larger than 128 */

/* The number of threads */
#define NUM_THREADS 4

/* Global variables for compressing/shuffling actions */
int clevel;                     /* Compression level */
int do_shuffle;                 /* Shuffle is active? */



/* Shuffle & Compress a single block */
static size_t
_blosc_c(int clevel, int doshuffle, size_t typesize, size_t blocksize,
         int leftoverblock, unsigned int ctbytes, unsigned int nbytes,
         unsigned char* _src, unsigned char* _dest, unsigned char *tmp)
{
  size_t j, neblock, nsplits;
  int cbytes, maxout;
  unsigned int btbytes = 0;
  unsigned char* _tmp;

  if (doshuffle && (typesize > 1)) {
    /* Shuffle this block (this makes sense only if typesize > 1) */
    shuffle(typesize, blocksize, _src, tmp);
    _tmp = tmp;
  }
  else {
    _tmp = _src;
  }

  /* Compress for each shuffled slice split for this block. */
  /* If the number of bytes of type is too large, or we are in a leftover
     block, do not split at all. */
  if ((typesize <= MAXSPLITS) && (!leftoverblock)) {
    nsplits = typesize;
  }
  else {
    nsplits = 1;
  }
  neblock = blocksize / nsplits;
  for (j = 0; j < nsplits; j++) {
    _dest += sizeof(int);
    btbytes += sizeof(int);
    ctbytes += sizeof(int);
    maxout = neblock;
    if (ctbytes+maxout > nbytes) {
      maxout = nbytes - ctbytes;   /* avoid buffer overrun */
      if (maxout <= 0) {
        return 0;                  /* non-compressible block */
      }
    }
    cbytes = blosclz_compress(clevel, _tmp+j*neblock, neblock,
                              _dest, maxout-1);
    if (cbytes >= maxout) {
      /* Buffer overrun caused by blosclz_compress (should never happen) */
      return -1;
    }
    else if (cbytes < 0) {
      /* cbytes should never be negative */
      return -2;
    }
    else if (cbytes == 0) {
      /* The compressor has been unable to compress data significantly. */
      /* Before doing the copy, check that we are not running into a
         buffer overflow. */
      if ((ctbytes+neblock) > nbytes) {
        return 0;    /* Non-compressible data */
      }
      memcpy(_dest, _tmp+j*neblock, neblock);
      cbytes = neblock;
    }
    ((unsigned int *)(_dest))[-1] = cbytes;
    _dest += cbytes;
    btbytes += cbytes;
    ctbytes += cbytes;
  }  /* Closes j < nsplits */

  return btbytes;
}


/* Structure for parameters in compression routines */
struct thread_data_c{
  size_t typesize;
  size_t blocksize;
  int thread_id;
  int clevel;
  int doshuffle;
  int cbytes;
  unsigned int tblocks;
  unsigned int thblocks;
  unsigned int leftover;
  unsigned char *src;
  unsigned char *tmp;
  unsigned char *tmp2;
  unsigned int *thstarts;
};


/* Compress & shuffle several blocks in a single thread */
void *t_blosc_c(void *params)
{
  /* Parameters */
  struct thread_data_c *my_params = (struct thread_data_c *)params;
  int thread_id = my_params->thread_id;
  int clevel = my_params->clevel;
  int doshuffle = my_params->doshuffle;
  size_t typesize = my_params->typesize;
  size_t blocksize = my_params->blocksize;
  unsigned int tblocks = my_params->tblocks;
  unsigned int thblocks = my_params->thblocks;
  unsigned int leftover = my_params->leftover;
  unsigned char *src = my_params->src;
  unsigned char *tmp = my_params->tmp;
  unsigned char *tmp2 = my_params->tmp2;
  unsigned int *thstarts = my_params->thstarts;
  /* Other local vars */
  int j, cbytes, ntbytes = 0;
  unsigned int bsize = blocksize;
  unsigned int leftoverblock = 0;
  unsigned int sstart;


  for (j = 0; j < thblocks; j++) {
    if (((thread_id*thblocks+j) == (tblocks - 1)) && (leftover > 0)) {
      bsize = leftover;
      leftoverblock = 1;
    }
    sstart = (thread_id*thblocks+j)*blocksize;
    cbytes = _blosc_c(clevel, doshuffle, typesize, bsize, leftoverblock,
                      (unsigned int)ntbytes, thblocks*blocksize,
                      src+sstart, tmp2, tmp);
    if (cbytes <= 0) {
      ntbytes = cbytes;          /* _blosc_c failure or uncompressible data */
      break;
    }
    thstarts[j] = ntbytes;
    ntbytes += cbytes;
    tmp2 += cbytes;
    /* Break if we have reached the end of the blocks */
    if ((thread_id*thblocks+j) == (tblocks - 1)) {
      break;
    }
  }

  my_params->cbytes = ntbytes;
  pthread_exit((void *)my_params);
}


unsigned int
blosc_compress(int clevel, int doshuffle, size_t typesize, size_t nbytes,
               const void *src, void *dest)
{
  unsigned char *_dest=NULL;       /* current pos for destination buffer */
  unsigned char *flags;            /* flags for header */
  unsigned int *bstarts;           /* start pointers for each block */
  size_t tblocks;                  /* number of total blocks in buffer */
  size_t leftover;                 /* extra bytes at end of buffer */
  size_t thblocks;                 /* number of total blocks per thread */
  size_t leftover2, leftover3;     /* helpers */
  size_t blocksize;                /* length of the block in bytes */
  size_t bsize;                    /* corrected blocksize for last block */
  unsigned int ctbytes = 0;        /* the number of bytes in output buffer */
  unsigned int *ctbytes_;          /* placeholder for bytes in output buffer */
  int cbytes;                      /* temporary compressed buffer length */
  int leftoverblock;               /* left over block? */
  unsigned int i, j;               /* local index variables */
  char *too_long_message = "The impossible happened: buffer overflow!";
  size_t t;
  pthread_t threads[NUM_THREADS];
  unsigned char *tmp[NUM_THREADS], *tmp2[NUM_THREADS];
  struct thread_data_c params[NUM_THREADS];
  unsigned int *thstarts[NUM_THREADS];   /* start pointers for each thread */
  int rc;
  void *status;
  pthread_attr_t attr;
  int nthreads = NUM_THREADS;
  int giveup = 0;                  /* sentinel */

  /* Compression level */
  if (clevel < 0 || clevel > 9) {
    /* If clevel not in 0..9, print an error */
    fprintf(stderr, "`clevel` parameter must be between 0 and 9!\n");
    return -10;
  }
  else if (clevel == 0) {
    /* No compression wanted.  Just return without doing anything else. */
    return 0;
  }

  /* Shuffle */
  if (doshuffle != 0 && doshuffle != 1) {
    fprintf(stderr, "`doshuffle` parameter must be either 0 or 1!\n");
    return -10;
  }

  /* Compute a blocksize depending on the optimization level */
  blocksize = BLOCKSIZE;
  /* 3 first optimization levels will not change blocksize */
  for (i=4; i<=(unsigned int)clevel; i++) {
    /* Escape if blocksize grows more than nbytes */
    if (blocksize*2 > nbytes) break;
    blocksize *= 2;
  }
  /* blocksize must be a multiple of the typesize */
  blocksize = blocksize / typesize * typesize;

  /* Compute number of blocks in buffer */
  tblocks = nbytes / blocksize;
  leftover = nbytes % blocksize;
  tblocks = (leftover>0)? tblocks+1: tblocks;
  /* Blocks per thread */
  thblocks = tblocks / nthreads;
  leftover2 = tblocks % nthreads;
  thblocks = (leftover2>0)? thblocks+1: thblocks;
  /* Final correction for the number of threads */
  if (nthreads > (tblocks / thblocks)) {
    nthreads = tblocks / thblocks;
    leftover3 = tblocks % thblocks;
    nthreads = (leftover3>0)? nthreads+1: nthreads;
  }

  /* Check typesize limits */
  if (typesize == MAXTYPESIZE) {
    typesize = 0;               /* zero means MAXTYPESIZE */
  }
  else if (typesize > MAXTYPESIZE) {
    /* If typesize is too large, treat buffer as an 1-byte stream. */
    typesize = 1;
  }

  _dest = (unsigned char *)(dest);
  /* Write header for this block */
  _dest[0] = BLOSC_VERSION_FORMAT;         /* blosc format version */
  _dest[1] = BLOSCLZ_VERSION_FORMAT;       /* blosclz format version */
  flags = _dest+2;                         /* flags */
  _dest[2] = 0;                            /* zeroes flags */
  _dest[3] = (unsigned char)typesize;      /* type size */
  _dest += 4;
  ctbytes += 4;
  ((unsigned int *)_dest)[0] = nbytes;     /* size of the chunk */
  ((unsigned int *)_dest)[1] = blocksize;  /* block size */
  ctbytes_ = (unsigned int *)(_dest+8);    /* compressed chunk size (pointer) */
  _dest += sizeof(int)*3;
  ctbytes += sizeof(int)*3;
  bstarts = (unsigned int *)_dest;         /* starts for every block */
  _dest += sizeof(int)*tblocks;            /* book space for pointers to */
  ctbytes += sizeof(int)*tblocks;          /* every block in output */

  if (doshuffle == 1) {
    /* Shuffle is active */
    *flags |= 0x1;                         /* bit 0 set to one in flags */
  }

  /* Initialize and set thread detached attribute */
  pthread_attr_init(&attr);
  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

  for(t=0; t < nthreads; t++){
  /* Create temporary area for each thread */
#ifdef _WIN32
    tmp[t] = (unsigned char*)_aligned_malloc(blocksize, 16);
    tmp2[t] = (unsigned char*)_aligned_malloc(blocksize*thblocks, 16);
    thstarts[t] = (unsigned int*)_aligned_malloc(sizeof(int)*thblocks, 16);
#elif defined __APPLE__
    /* Mac OS X guarantees 16-byte alignment in small allocs */
    tmp[t] = (unsigned char *)malloc(blocksize);
    tmp2[t] = (unsigned char *)malloc(blocksize*thblocks);
    thstarts[t] = (unsigned int*)malloc(sizeof(int)*thblocks);
#else
    rc = posix_memalign((void **)&tmp[t], 16, blocksize);
    rc = posix_memalign((void **)&tmp2[t], 16, blocksize*thblocks);
    rc = posix_memalign((void **)&thstarts[t], 16, sizeof(int)*thblocks);
#endif  /* _WIN32 */
    /* Populate parameters for thread */
    params[t].thread_id = t;
    params[t].clevel = clevel;
    params[t].doshuffle = doshuffle;
    params[t].typesize = typesize;
    params[t].blocksize = blocksize;
    params[t].tblocks = tblocks;
    params[t].thblocks = thblocks;
    if (t == (nthreads - 1) && leftover > 0) {
      params[t].leftover = leftover;
    }
    else {
      params[t].leftover = 0;
    }
    params[t].src = (unsigned char *)src;
    params[t].tmp = tmp[t];
    params[t].tmp2 = tmp2[t];
    params[t].thstarts = thstarts[t];
    rc = pthread_create(&threads[t], &attr, t_blosc_c, (void *)&params[t]);
    if (rc){
      fprintf(stderr, "ERROR; return code from pthread_create() is %d\n", rc);
      fprintf(stderr, "\tError detail: %s\n", strerror(rc));
      exit(-1);
    }
  }  /* Close t < nthreads */

  /* Wait for threads to finish (join threads) */
  pthread_attr_destroy(&attr);    /* free attribute */
  for(t=0; t<nthreads; t++){
    rc = pthread_join(threads[t], &status);
    if (rc) {
      fprintf(stderr, "ERROR; return code from pthread_join() is %d\n", rc);
      fprintf(stderr, "\tError detail: %s\n", strerror(rc));
      exit(-1);
    }
    if (giveup) {
      /* We have been signaled to give up.  Wait for all threads to finish. */
      continue;
    }
    cbytes = ((struct thread_data_c *)status)->cbytes;
    /* Check result of compression */
    if (cbytes < 0) {
      fprintf(stderr, "%s\n", too_long_message);
      ctbytes = cbytes;         /* error in _blosc_c */
      giveup = 1;
    }
    else if ((cbytes == 0) | (ctbytes+cbytes >= nbytes)) {
      ctbytes = 0;              /* uncompressible data */
      giveup = 1;
    }
    else {
      /* Compression process seems right */
      /* Copy block starts */
      for (j=0; j<thblocks; j++) {
        /* Check that we do not exceed the end of the blocks */
        if ((t*thblocks+j) < tblocks) {
          bstarts[t*thblocks+j] = thstarts[t][j]+ctbytes;
        }
      }
      /* Copy thread's private buffer to destination buffer */
      memcpy(_dest, tmp2[t], cbytes);
      _dest += cbytes;
      ctbytes += cbytes;          /* update compressed bytes */
    }
  }  /* Close t < nthreads */

  /* Release buffers */
  for(t=0; t < nthreads; t++){
#ifdef _WIN32
    _aligned_free(tmp[t]);
    _aligned_free(tmp2[t]);
    _aligned_free(thstarts[t]);
#else
    free(tmp[t]);
    free(tmp2[t]);
    free(thstarts[t]);
#endif  /* _WIN32 */
  }

  assert(ctbytes < nbytes);
  *ctbytes_ = ctbytes;   /* set the number of compressed bytes in header */
  return ctbytes;
}


/* Decompress & unshuffle a single block */
static size_t
_blosc_d(int dounshuffle, size_t typesize, size_t blocksize, int leftoverblock,
         unsigned char* _src, unsigned char* _dest,
         unsigned char *tmp, unsigned char *tmp2)
{
  size_t j, neblock, nsplits;
  size_t nbytes, cbytes, ctbytes = 0, ntbytes = 0;
  unsigned char* _tmp;

  if (dounshuffle && (typesize > 1)) {
    _tmp = tmp;
  }
  else {
    _tmp = _dest;
  }

  /* Compress for each shuffled slice split for this block. */
  /* If the number of bytes is too large, do not split all. */
  if ((typesize <= MAXSPLITS) && (!leftoverblock)) {
    nsplits = typesize;
  }
  else {
    nsplits = 1;
  }
  neblock = blocksize / nsplits;
  for (j = 0; j < nsplits; j++) {
    cbytes = ((unsigned int *)(_src))[0]; /* amount of compressed bytes */
    _src += sizeof(int);
    ctbytes += sizeof(int);
    /* Uncompress */
    if (cbytes == neblock) {
      memcpy(_tmp, _src, neblock);
      nbytes = neblock;
    }
    else {
      nbytes = blosclz_decompress(_src, cbytes, _tmp, neblock);
      if (nbytes != neblock) {
        return -2;
      }
    }
    _src += cbytes;
    ctbytes += cbytes;
    _tmp += neblock;
    ntbytes += nbytes;
  } /* Closes j < nsplits */

  if (dounshuffle && (typesize > 1)) {
    if ((uintptr_t)_dest % 16 == 0) {
      /* 16-bytes aligned _dest.  SSE2 unshuffle will work. */
      unshuffle(typesize, blocksize, tmp, _dest);
    }
    else {
      /* _dest is not aligned.  Use tmp2, which is aligned, and copy. */
      unshuffle(typesize, blocksize, tmp, tmp2);
      memcpy(_dest, tmp2, blocksize);
    }
  }

  return ctbytes;
}


/* Structure for parameters in decompression routines */
struct thread_data_d{
  size_t typesize;
  size_t blocksize;
  int thread_id;
  int dounshuffle;
  int cbytes;
  unsigned int tblocks;
  unsigned int thblocks;
  unsigned int leftover;
  unsigned int *bstarts;             /* start pointers for each block */
  unsigned char *src;
  unsigned char *dest;
  unsigned char *tmp;
  unsigned char *tmp2;
};


/* Decompress & unshuffle several blocks in a single thread */
void *t_blosc_d(void *params)
{
  /* Parameters */
  struct thread_data_d *my_params = (struct thread_data_d *)params;
  int thread_id = my_params->thread_id;
  int dounshuffle = my_params->dounshuffle;
  size_t typesize = my_params->typesize;
  size_t blocksize = my_params->blocksize;
  unsigned int tblocks = my_params->tblocks;
  unsigned int thblocks = my_params->thblocks;
  unsigned int *bstarts = my_params->bstarts;
  unsigned int leftover = my_params->leftover;
  unsigned char *src = my_params->src;
  unsigned char *dest = my_params->dest;
  unsigned char *tmp = my_params->tmp;
  unsigned char *tmp2 = my_params->tmp2;
  /* Other local vars */
  int j, cbytes, ntbytes = 0;
  unsigned int bsize = blocksize;
  unsigned int leftoverblock = 0;

  for (j = 0; j < thblocks; j++) {
    if (((thread_id*thblocks+j) == (tblocks - 1)) && (leftover > 0)) {
      bsize = leftover;
      leftoverblock = 1;
    }
    cbytes = _blosc_d(dounshuffle, typesize, bsize, leftoverblock,
                      src+bstarts[j], dest+j*blocksize, tmp, tmp2);
    if (cbytes < 0) {
      ntbytes = cbytes;          /* _blosc_d failure */
      goto out;
    }
    ntbytes += cbytes;
    /* Break if we have reached the end of the blocks */
    if ((thread_id*thblocks+j) == (tblocks - 1)) {
      goto out;
    }
  }

 out:
  my_params->cbytes = ntbytes;
  if (NUM_THREADS > 1) {
    pthread_exit((void *)my_params);
  }
  else {
    return (void *)my_params;
  }
}


unsigned int
blosc_decompress(const void *src, void *dest, size_t dest_size)
{
  unsigned char *_src=NULL;          /* alias for source buffer */
  unsigned char *_dest=NULL;         /* alias for destination buffer */
  unsigned char version, versionlz;  /* versions for compressed header */
  unsigned char flags;               /* flags for header */
  size_t leftover;                   /* extra bytes at end of buffer */
  size_t tblocks;                    /* number of total blocks in buffer */
  size_t thblocks;                   /* number of total blocks per thread */
  size_t leftover2, leftover3;       /* helpers */
  unsigned int *bstarts;             /* start pointers for each block */
  int nbytes, cbytes, ntbytes = 0;
  int dounshuffle = 0;
  unsigned int typesize, blocksize, ctbytes;
  size_t t;
  pthread_t threads[NUM_THREADS];
  unsigned char *tmp[NUM_THREADS], *tmp2[NUM_THREADS];
  struct thread_data_d params[NUM_THREADS];
  int rc;
  void *status;
  pthread_attr_t attr;
  int nthreads = NUM_THREADS;
  int giveup;

  _src = (unsigned char *)(src);
  _dest = (unsigned char *)(dest);

  /* Read the header block */
  version = _src[0];                        /* blosc format version */
  versionlz = _src[1];                      /* blosclz format version */
  flags = _src[2];                          /* flags */
  typesize = (unsigned int)_src[3];         /* typesize */
  _src += 4;
  nbytes = ((unsigned int *)_src)[0];       /* chunk size */
  blocksize = ((unsigned int *)_src)[1];    /* block size */
  ctbytes = ((unsigned int *)_src)[2];      /* compressed chunk size */
  _src += sizeof(int)*3;
  /* Compute some params */
  /* Total blocks */
  tblocks = nbytes / blocksize;
  leftover = nbytes % blocksize;
  tblocks = (leftover>0)? tblocks+1: tblocks;
  /* Blocks per thread */
  thblocks = tblocks / nthreads;
  leftover2 = tblocks % nthreads;
  thblocks = (leftover2>0)? thblocks+1: thblocks;
  /* Final correction for the number of threads */
  if (nthreads > (tblocks / thblocks)) {
    nthreads = tblocks / thblocks;
    leftover3 = tblocks % thblocks;
    nthreads = (leftover3>0)? nthreads+1: nthreads;
  }
  bstarts = (unsigned int *)_src;
  _src += sizeof(int)*tblocks;

  /* Check zero typesizes */
  if (typesize == 0) {
    typesize = MAXTYPESIZE;
  }

  if (nbytes > dest_size) {
    /* This should never happen but just in case */
    return -1;
  }

  if ((flags & 0x1) == 1) {
    /* Input is shuffled.  Unshuffle it. */
    dounshuffle = 1;
  }

  /* Create temporary area for each thread */
  for(t=0; t<nthreads; t++){
#ifdef _WIN32
    tmp[t] = (unsigned char*)_aligned_malloc(blocksize, 16);
    tmp2[t] = (unsigned char*)_aligned_malloc(blocksize, 16);
#elif defined __APPLE__
    /* Mac OS X guarantees 16-byte alignment in small allocs */
    tmp[t] = (unsigned char *)malloc(blocksize);
    tmp2[t] = (unsigned char *)malloc(blocksize);
#else
    rc = posix_memalign((void **)&tmp[t], 16, blocksize);
    rc = posix_memalign((void **)&tmp2[t], 16, blocksize);
#endif  /* _WIN32 */
  }

  /* Initialize and set thread detached attribute */
  pthread_attr_init(&attr);
  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

  for(t=0; t < nthreads; t++){
    params[t].thread_id = t;
    params[t].dounshuffle = dounshuffle;
    params[t].typesize = typesize;
    params[t].blocksize = blocksize;
    params[t].tblocks = tblocks;
    params[t].thblocks = thblocks;
    params[t].bstarts = bstarts + t*thblocks;
    if (t == (nthreads - 1) && leftover > 0) {
      params[t].leftover = leftover;
    }
    else {
      params[t].leftover = 0;
    }
    params[t].src = (unsigned char *)src;
    params[t].dest = (unsigned char *)dest + t*thblocks*blocksize;
    params[t].tmp = tmp[t];
    params[t].tmp2 = tmp2[t];
    if (nthreads > 1) {
      rc = pthread_create(&threads[t], &attr, t_blosc_d, (void *)&params[t]);
      if (rc){
        fprintf(stderr, "ERROR; return code from pthread_create() is %d\n", rc);
        fprintf(stderr, "\tError detail: %s\n", strerror(rc));
        exit(-1);
      }
    }
    else {
      status = t_blosc_d((void *)&params[t]);
      cbytes = ((struct thread_data_d *)status)->cbytes;
      if (cbytes < 0) {
        nbytes = cbytes;          /* _blosc_d failure */
        goto out;
      }
      ntbytes += cbytes;          /* update decompressed bytes */
    }
  }

  /* Wait for threads to finish (join threads) */
  pthread_attr_destroy(&attr);    /* free attribute */
  if (nthreads > 1) {
    for(t=0; t < nthreads; t++){
      rc = pthread_join(threads[t], &status);
      if (rc) {
        fprintf(stderr, "ERROR; return code from pthread_join() is %d\n", rc);
        fprintf(stderr, "\tError detail: %s\n", strerror(rc));
        exit(-1);
      }
      if (giveup) {
        continue;
      }
      cbytes = ((struct thread_data_d *)status)->cbytes;
      if (cbytes < 0) {
        nbytes = cbytes;          /* _blosc_d failure */
        giveup = 1;
      }
      ntbytes += cbytes;          /* update decompressed bytes */
    }
  }

 out:
  /* Release buffers */
  for(t=0; t < nthreads; t++){
#ifdef _WIN32
    _aligned_free(tmp[t]);
    _aligned_free(tmp2[t]);
#else
    free(tmp[t]);
    free(tmp2[t]);
#endif  /* _WIN32 */
  }

  /* assert(ntbytes+bstarts[0] == ctbytes); */
  return nbytes;
}