Changeset 54

Timestamp:

03/30/10 14:36:01 (3 years ago)

Author:

faltet

Message:

New version for accelerating the decompressor when NUM_THREADS=1.

Some code cleanup done too, and added a README-threaded.txt file for
telling my experience with threaded Blosc to those who are interested.

Location:

branches/threaded

Files:

• README-threaded.txt (added)
• src/bench.c (modified) (5 diffs)
• src/blosc.c (modified) (10 diffs)

branches/threaded/src/bench.c

@@ -36 +36 @@
 
 #define NITER  (20*1000)               /* Number of iterations */
-//#define NITER  (1)               /* Number of iterations */
 
 
@@ -109 +108 @@
   int *_srccpy;
   int rshift = 22;              /* For random data */
+  //int rshift = 20;              /* For random data */
   int clevel;
   int doshuffle = 1;            /* Shuffle? */
@@ -126 +126 @@
 
   /* Initialize the original buffer */
-  /* _src = (int *)src; */
-  /* _srccpy = (int *)srccpy; */
-  /* elsize = sizeof(int); */
-  /* for(i = 0; i < size/elsize; ++i) { */
-  /*   /\* Choose one below *\/ */
-  /*   /\* _src[i] = 1; *\/ */
-  /*   /\* _src[i] = 0x01010101; *\/ */
-  /*   /\* _src[i] = 0x01020304; *\/ */
-  /*   /\* _src[i] = i * 1/.3; *\/ */
-  /*   /\* _src[i] = i; *\/ */
-  /*   _src[i] = rand() >> rshift; */
+  _src = (int *)src;
+  _srccpy = (int *)srccpy;
+  elsize = sizeof(int);
+  for(i = 0; i < size/elsize; ++i) {
+    /* Choose one below */
+    /* _src[i] = 1; */
+    /* _src[i] = 0x01010101; */
+    /* _src[i] = 0x01020304; */
+    /* _src[i] = i * 1/.3; */
+    /* _src[i] = i; */
+    _src[i] = rand() >> rshift;
+  }
+
+  /* For data coming from a file */
+  /* fd = open(filename, 0); */
+  /* status = read(fd, src, size); */
+  /* if (status == -1) { */
+  /*   perror(NULL); */
   /* } */
-
-  /* For data coming from a file */
-  fd = open(filename, 0);
-  status = read(fd, src, size);
-  if (status == -1) {
-    perror(NULL);
-  }
-  close(fd);
+  /* close(fd); */
 
   printf("********************** Setup info *****************************\n");
@@ -166 +166 @@
 
   for (clevel=1; clevel<10; clevel++) {
-  //for (clevel=9; clevel<10; clevel++) {
 
     printf("Compression level: %d\n", clevel);
@@ -200 +199 @@
       printf("FAILED.  Error code: %d\n", nbytes);
     }
-    printf("Orig bytes: %d\tFinal bytes: %d\n", cbytes, nbytes);
+    /* printf("Orig bytes: %d\tFinal bytes: %d\n", cbytes, nbytes); */
 
     /* Check if data has had a good roundtrip */

branches/threaded/src/blosc.c

@@ -370 +370 @@
       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;
+      goto out;
     }
     ntbytes += cbytes;
     /* Break if we have reached the end of the blocks */
     if ((thread_id*thblocks+j) == (tblocks - 1)) {
-      break;
+      goto out;
     }
   }
 
  out:
-  //printf("ntbytes-->%d\n", ntbytes);
   my_params->cbytes = ntbytes;
-  pthread_exit((void *)my_params);
+  if (NUM_THREADS > 1) {
+    pthread_exit((void *)my_params);
+  }
+  else {
+    return (void *)my_params;
+  }
 }
 
@@ -405 +402 @@
   unsigned char flags;               /* flags for header */
   size_t leftover;                   /* extra bytes at end of buffer */
-  size_t nblocks;                    /* number of complete blocks in buffer */
   size_t tblocks;                    /* number of total blocks in buffer */
   size_t thblocks;                   /* number of total blocks per thread */
   unsigned int *bstarts;             /* start pointers for each block */
-  size_t nblocks2, leftover2;
+  size_t leftover2, leftover3;
   size_t t;
   int nbytes, cbytes, ntbytes = 0;
@@ -420 +416 @@
   void *status;
   pthread_attr_t attr;
+  int nthreads = NUM_THREADS;
 
   _src = (unsigned char *)(src);
@@ -436 +433 @@
   /* Compute some params */
   /* Total blocks */
-  nblocks = nbytes / blocksize;
+  tblocks = nbytes / blocksize;
   leftover = nbytes % blocksize;
-  tblocks = (leftover>0)? nblocks+1: nblocks;
-  //printf("tblocks-->%d\n", tblocks);
+  tblocks = (leftover>0)? tblocks+1: tblocks;
   /* Blocks per thread */
-  nblocks2 = tblocks / NUM_THREADS;
-  leftover2 = tblocks % NUM_THREADS;
-  thblocks = (leftover2>0)? nblocks2+1: nblocks2;
-  //printf("thblocks-->%d\n", thblocks);
+  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;
@@ -464 +465 @@
 
   /* Create temporary area for each thread */
-  for(t=0; t<NUM_THREADS; t++){
+  for(t=0; t<nthreads; t++){
 #ifdef _WIN32
     tmp[t] = (unsigned char*)_aligned_malloc(blocksize, 16);
@@ -482 +483 @@
   pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
 
-  for(t=0; t < NUM_THREADS; t++){
+  for(t=0; t < nthreads; t++){
     params[t].thread_id = t;
     params[t].dounshuffle = dounshuffle;
@@ -490 +491 @@
     params[t].thblocks = thblocks;
     params[t].bstarts = bstarts + t*thblocks;
-    if (t == (NUM_THREADS - 1) && leftover > 0) {
+    if (t == (nthreads - 1) && leftover > 0) {
       params[t].leftover = leftover;
     }
@@ -496 +497 @@
       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].src = (unsigned char *)src;
     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);
+    if (NUM_THREADS > 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 {
+      t_blosc_d((void *)&params[t]);
     }
   }
@@ -511 +516 @@
   /* Wait for threads to finish (join threads) */
   pthread_attr_destroy(&attr);    /* free attribute */
-  for(t=0; t < NUM_THREADS; t++){
+  if (NUM_THREADS > 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);
     }
     cbytes = ((struct thread_data_d *)status)->cbytes;
-    //printf("Thread %d.  cbytes-->%d\n", t, cbytes);
     if (cbytes < 0) {
-      ntbytes = cbytes;          /* _blosc_d failure */
+      nbytes = cbytes;          /* _blosc_d failure */
       goto out;
     }
     ntbytes += cbytes;          /* update decompressed bytes */
   }
+  }
 
  out:
-  for(t=0; t<NUM_THREADS; t++){
+  for(t=0; t < nthreads; t++){
 #ifdef _WIN32
     _aligned_free(tmp[t]);
@@ -537 +544 @@
   }
 
-  //printf("ntbytes, ctbytes-->%d, %d\n", ntbytes, ctbytes);
-  assert(ntbytes+bstarts[0] == ctbytes);
+  /* assert(ntbytes+bstarts[0] == ctbytes); */
   return nbytes;
 }