Changeset 51

Timestamp:

03/30/10 06:35:57 (3 years ago)

Author:

faltet

Message:

First version of threaded blosc. Performance increase has not been detected yet (rather the contrary :-/).

Location:

branches/threaded/src

Files:

• blosc.c (modified) (11 diffs)
• blosc.h (modified) (1 diff)

branches/threaded/src/blosc.c

@@ -14 +14 @@
 #include <sys/types.h>
 #include <sys/stat.h>
+#include <assert.h>
 #include "blosc.h"
 #include "blosclz.h"
@@ -22 +23 @@
   #include <stdint.h>
   #include <unistd.h>
+  #include <pthread.h>
 #endif  /* _WIN32 */
 
@@ -34 +36 @@
 #define MAXSPLITS 16         /* Cannot be larger than 128 */
 
+/* The number of threads */
+#define NUM_THREADS 2
 
 /* Global variables for compressing/shuffling actions */
@@ -120 +124 @@
   unsigned char *_dest=NULL;       /* alias for destination buffer */
   unsigned char *flags;            /* flags for header */
-  unsigned int *starts;            /* start pointers for each block */
+  unsigned int *bstarts;           /* start pointers for each block */
   size_t nblocks;                  /* number of complete blocks in buffer */
   size_t tblocks;                  /* number of total blocks in buffer */
@@ -202 +206 @@
   _dest += sizeof(int)*3;
   ctbytes += sizeof(int)*3;
-  starts = (unsigned int *)_dest;          /* starts for every block */
+  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 */
@@ -212 +216 @@
 
   for (j = 0; j < tblocks; j++) {
-    starts[j] = ctbytes;
+    bstarts[j] = ctbytes;
     bsize = blocksize;
     leftoverblock = 0;
@@ -318 +322 @@
 
   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;
+    }
+    /* printf("dounshuffle, typesize, bsize, leftoverblock-->%d,%d,%d,%d\n", */
+    /*        dounshuffle, typesize, bsize, leftoverblock); */
+    /* printf("src, dest, tmp, tmp2, thread-->%p,%p,%p,%p,%d,%d\n", */
+    /*        src+bstarts[j], dest+j*blocksize, tmp, tmp2, thread_id, j); */
+    cbytes = _blosc_d(dounshuffle, typesize, bsize, leftoverblock,
+                      src+bstarts[j], dest+j*blocksize, tmp, tmp2);
+    //printf("cbytes2-->%d\n", cbytes);
+    if (cbytes < 0) {
+      //printf("Ei!!");
+      ntbytes = cbytes;          /* _blosc_d failure */
+      break;
+      //goto out;
+    }
+    ntbytes += cbytes;
+    /* Break if we have reached the end of the blocks */
+    if ((thread_id*thblocks+j) == (tblocks - 1)) {
+      break;
+    }
+  }
+
+ out:
+  //printf("ntbytes-->%d\n", ntbytes);
+  my_params->cbytes = ntbytes;
+  pthread_exit((void *)my_params);
 }
 
@@ -331 +407 @@
   size_t nblocks;                    /* number of complete blocks in buffer */
   size_t tblocks;                    /* number of total blocks in buffer */
-  size_t j;
-  size_t nbytes, ntbytes = 0;
-  int cbytes;
-  unsigned char *tmp, *tmp2;
+  size_t thblocks;                   /* number of total blocks per thread */
+  unsigned int *bstarts;             /* start pointers for each block */
+  size_t nblocks2, leftover2;
+  size_t t;
+  int nbytes, cbytes, ntbytes = 0;
   int dounshuffle = 0;
-  unsigned int typesize, blocksize, bsize, ctbytes_;
-  int leftoverblock;               /* left over block? */
+  unsigned int typesize, blocksize, ctbytes;
+  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;
 
   _src = (unsigned char *)(src);
@@ -350 +432 @@
   nbytes = ((unsigned int *)_src)[0];       /* chunk size */
   blocksize = ((unsigned int *)_src)[1];    /* block size */
-  ctbytes_ = ((unsigned int *)_src)[2];     /* compressed chunk size */
+  ctbytes = ((unsigned int *)_src)[2];      /* compressed chunk size */
   _src += sizeof(int)*3;
   /* Compute some params */
+  /* Total blocks */
   nblocks = nbytes / blocksize;
   leftover = nbytes % blocksize;
   tblocks = (leftover>0)? nblocks+1: nblocks;
-  _src += sizeof(int)*tblocks;              /* skip starts of blocks */
+  //printf("tblocks-->%d\n", tblocks);
+  /* Blocks per thread */
+  nblocks2 = tblocks / NUM_THREADS;
+  leftover2 = tblocks % NUM_THREADS;
+  thblocks = (leftover2>0)? nblocks2+1: nblocks2;
+  //printf("thblocks-->%d\n", thblocks);
+  bstarts = (unsigned int *)_src;
+  _src += sizeof(int)*tblocks;
 
   /* Check zero typesizes */
@@ -364 +454 @@
 
   if (nbytes > dest_size) {
-    /* This should never happen, but just in case. */
+    /* This should never happen but just in case */
     return -1;
   }
@@ -373 +463 @@
   }
 
-  /* Create temporary area */
+  /* Create temporary area for each thread */
+  for(t=0; t<NUM_THREADS; t++){
 #ifdef _WIN32
-  tmp = (unsigned char*)_aligned_malloc(blocksize, 16);
-  tmp2 = (unsigned char*)_aligned_malloc(blocksize, 16);
+    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 = (unsigned char *)(malloc(blocksize));
-  tmp2 = (unsigned char *)(malloc(blocksize));
+    /* Mac OS X guarantees 16-byte alignment in small allocs */
+    tmp[t] = (unsigned char *)malloc(blocksize);
+    tmp2[t] = (unsigned char *)malloc(blocksize);
 #else
-  posix_memalign((void **)&tmp, 16, blocksize);
-  posix_memalign((void **)&tmp2, 16, blocksize);
+    posix_memalign((void **)&tmp[t], 16, blocksize);
+    posix_memalign((void **)&tmp2[t], 16, blocksize);
 #endif  /* _WIN32 */
-
-  for (j = 0; j < tblocks; j++) {
-    bsize = blocksize;
-    leftoverblock = 0;
-    if ((j == tblocks - 1) && (leftover > 0)) {
-      bsize = leftover;
-      leftoverblock = 1;
-    }
-    cbytes = _blosc_d(dounshuffle, typesize, bsize, leftoverblock,
-                      _src, _dest, tmp, tmp2);
+  }
+
+  /* Initialize and set thread detached attribute */
+  pthread_attr_init(&attr);
+  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
+
+  for(t=0; t < NUM_THREADS; 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 == (NUM_THREADS - 1) && leftover > 0) {
+      params[t].leftover = leftover;
+    }
+    else {
+      params[t].leftover = 0;
+    }
+    //printf("Thread: %d\tbstarts: %d\n", t, bstarts[t*thblocks]);
+    params[t].src = (unsigned char *)src;   // + bstarts[t*thblocks];
+    params[t].dest = (unsigned char *)dest + t*thblocks*blocksize;
+    params[t].tmp = tmp[t];
+    params[t].tmp2 = tmp2[t];
+    //printf("In main: creating thread %ld\n", t);
+    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);
+      exit(-1);
+    }
+  }
+
+  /* Wait for threads to finish (join threads) */
+  pthread_attr_destroy(&attr);    /* free attribute */
+  for(t=0; t < NUM_THREADS; t++){
+    rc = pthread_join(threads[t], &status);
+    if (rc) {
+      fprintf(stderr, "ERROR; return code from pthread_join() is %d\n", rc);
+      exit(-1);
+    }
+    cbytes = ((struct thread_data_d *)status)->cbytes;
+    //printf("Thread %d.  cbytes-->%d\n", t, cbytes);
     if (cbytes < 0) {
       ntbytes = cbytes;          /* _blosc_d failure */
       goto out;
     }
-    _src += cbytes;
-    _dest += blocksize;
-    ntbytes += blocksize;
+    ntbytes += cbytes;          /* update decompressed bytes */
   }
 
  out:
+  for(t=0; t<NUM_THREADS; t++){
 #ifdef _WIN32
-  _aligned_free(tmp);
-  _aligned_free(tmp2);
+    _aligned_free(tmp[t]);
+    _aligned_free(tmp2[t]);
 #else
-  free(tmp);
-  free(tmp2);
+    free(tmp[t]);
+    free(tmp2[t]);
 #endif  /* _WIN32 */
+  }
+
+  //printf("ntbytes, ctbytes-->%d, %d\n", ntbytes, ctbytes);
+  assert(ntbytes+bstarts[0] == ctbytes);
   return nbytes;
 }
-
-

branches/threaded/src/blosc.h

@@ -13 +13 @@
 /* Version numbers */
 #define BLOSC_VERSION_MAJOR    0    /* for major interface/format changes  */
-#define BLOSC_VERSION_MINOR    8    /* for minor interface/format changes  */
-#define BLOSC_VERSION_RELEASE  2    /* for tweaks, bug-fixes, or development */
+#define BLOSC_VERSION_MINOR    9    /* for minor interface/format changes  */
+#define BLOSC_VERSION_RELEASE  0    /* for tweaks, bug-fixes, or development */
 
-#define BLOSC_VERSION_STRING   "0.8.2"  /* string version.  Sync with above! */
+#define BLOSC_VERSION_STRING   "0.9.0.dev"  /* string version.  Sync with above! */
 #define BLOSC_VERSION_DATE     "2010-03-30"      /* date version */