Changeset 61

Timestamp:

04/27/10 06:27:17 (3 years ago)

Author:

faltet

Message:

Temporary buffers for compression/decompression only reset when needed.

This optimization makes the decompression quite a bit faster, specially
for moderately large number of threads (4) and buffer sizes (1 MB).

Location:

branches/threaded/src

Files:

• bench.c (modified) (1 diff)
• blosc.c (modified) (6 diffs)

branches/threaded/src/bench.c

@@ -144 +144 @@
   unsigned int elsize = 8;        /* Datatype size */
   int rshift = 12;                /* For random data */
-  int nthreads = 4;               /* The number of threads */
+  int nthreads = 2;               /* The number of threads */
   int doshuffle = 1;              /* Shuffle? */
   unsigned char *orig, *round;

branches/threaded/src/blosc.c

@@ -44 +44 @@
 int init_threads_done = 0;      /* pool of threads initialized? */
 int end_threads = 0;            /* should exisiting threads end? */
-int init_mem = 1;               /* shoudl memory be initialized? */
+int init_temps_done = 0;        /* temporaries in threads initialized? */
 int giveup;                     /* should (de-)compression give up? */
 int nblock = -1;                /* block counter */
 pthread_t threads[MAX_THREADS]; /* opaque structure for threads */
-int tids[MAX_THREADS];          /* ID per each threads */
+int tids[MAX_THREADS];          /* ID per each thread */
 pthread_attr_t ct_attr;         /* creation time attributes for threads */
 
@@ -76 +76 @@
 
 
-/* Convenience functions for creating and releasing temporaries */
-void
-create_temporaries(void)
-{
-  int tid;
-  size_t blocksize = params.blocksize;
-  /* Extended blocksize for temporary destination.  Extended blocksize
-   is only useful for compression in parallel mode, but it doesn't
-   hurt other modes either. */
-  size_t ebsize = blocksize + params.typesize*sizeof(int);
-  unsigned char *tmp, *tmp2;
-
-  /* Create temporary area for each thread */
-  for (tid = 0; tid < nthreads; tid++) {
-#ifdef _WIN32
-    tmp = (unsigned char *)_aligned_malloc(blocksize, 16);
-    tmp2 = (unsigned char *)_aligned_malloc(ebsize, 16);
-#elif defined __APPLE__
-    /* Mac OS X guarantees 16-byte alignment in small allocs */
-    tmp = (unsigned char *)malloc(blocksize);
-    tmp2 = (unsigned char *)malloc(ebsize);
-#else
-    posix_memalign((void **)&tmp, 16, blocksize);
-    posix_memalign((void **)&tmp2, 16, ebsize);
-#endif  /* _WIN32 */
-    params.tmp[tid] = tmp;
-    params.tmp2[tid] = tmp2;
-  }
-}
-
-
-void
-release_temporaries(void)
-{
-  int tid;
-  unsigned char *tmp, *tmp2;
-
-  /* Release buffers */
-  for (tid = 0; tid < nthreads; tid++) {
-    tmp = params.tmp[tid];
-    tmp2 = params.tmp2[tid];
-#ifdef _WIN32
-    _aligned_free(tmp);
-    _aligned_free(tmp2);
-#else
-    free(tmp);
-    free(tmp2);
-#endif  /* _WIN32 */
-  }
-}
+/* Structure for parameters meant for keeping track of current temporaries */
+struct temp_data {
+  int nthreads;
+  size_t typesize;
+  size_t blocksize;
+} current_temp;
+
 
 
@@ -447 +404 @@
 
 
+/* Convenience functions for creating and releasing temporaries */
+void
+create_temporaries(void)
+{
+  int tid;
+  size_t typesize = params.typesize;
+  size_t blocksize = params.blocksize;
+  /* Extended blocksize for temporary destination.  Extended blocksize
+   is only useful for compression in parallel mode, but it doesn't
+   hurt other modes either. */
+  size_t ebsize = blocksize + typesize*sizeof(int);
+  unsigned char *tmp, *tmp2;
+
+  /* Create temporary area for each thread */
+  for (tid = 0; tid < nthreads; tid++) {
+#ifdef _WIN32
+    tmp = (unsigned char *)_aligned_malloc(blocksize, 16);
+    tmp2 = (unsigned char *)_aligned_malloc(ebsize, 16);
+#elif defined __APPLE__
+    /* Mac OS X guarantees 16-byte alignment in small allocs */
+    tmp = (unsigned char *)malloc(blocksize);
+    tmp2 = (unsigned char *)malloc(ebsize);
+#else
+    posix_memalign((void **)&tmp, 16, blocksize);
+    posix_memalign((void **)&tmp2, 16, ebsize);
+#endif  /* _WIN32 */
+    params.tmp[tid] = tmp;
+    params.tmp2[tid] = tmp2;
+  }
+
+  init_temps_done = 1;
+  /* Update params for current temporaries */
+  current_temp.nthreads = nthreads;
+  current_temp.typesize = typesize;
+  current_temp.blocksize = blocksize;
+
+}
+
+
+void
+release_temporaries(void)
+{
+  int tid;
+  unsigned char *tmp, *tmp2;
+
+  /* Release buffers */
+  for (tid = 0; tid < nthreads; tid++) {
+    tmp = params.tmp[tid];
+    tmp2 = params.tmp2[tid];
+#ifdef _WIN32
+    _aligned_free(tmp);
+    _aligned_free(tmp2);
+#else
+    free(tmp);
+    free(tmp2);
+#endif  /* _WIN32 */
+  }
+
+  init_temps_done = 0;
+
+}
+
+
 /* Do the compression or decompression of the buffer depending on the
    global params. */
@@ -453 +473 @@
   int ntbytes;
 
-  create_temporaries();
+  /* Initialize/reset temporaries if needed */
+  if (!init_temps_done) {
+    create_temporaries();
+  }
+  else if (current_temp.nthreads != nthreads ||
+           current_temp.typesize != params.typesize ||
+           current_temp.blocksize != params.blocksize) {
+    release_temporaries();
+    create_temporaries();
+  }
 
   if (nthreads == 1) {
@@ -462 +491 @@
   }
 
-  release_temporaries();
 
   return ntbytes;
@@ -784 +812 @@
       init_threads_done = 0;
       end_threads = 0;
-      printf("Pool de %d threads acabat!\n", nthreads);
     }
     nthreads = nthreads_new;
-    init_mem = 1;
     if (nthreads > 1) {
       /* Launch a new pool of threads */
       init_threads();
-      printf("Pool de %d threads començat!\n", nthreads);
     }
     return nthreads_old;