Changeset 8856791

Timestamp:

Dec 7, 2018, 1:28:55 AM (6 years ago)

Author:

Paul Brossier <piem@piem.org>

Branches:

feature/cnn, feature/crepe, feature/timestretch, fix/ffmpeg5, master

Children:

bdb249b

Parents:

9bd769a

git-author:

Paul Brossier <piem@piem.org> (09/30/16 01:54:50)

git-committer:

Paul Brossier <piem@piem.org> (12/07/18 01:28:55)

Message:

src/effects/timestretch_rubberband.c: add initial pthread support

File:

• src/effects/timestretch_rubberband.c (modified) (11 diffs)

src/effects/timestretch_rubberband.c

@@ -34 +34 @@
 #define MAX_STRETCH_RATIO 40.
 
+#define HAVE_THREADS 1
+#if 0
+#undef HAVE_THREADS
+#endif
+
+#ifdef HAVE_THREADS
+#include <pthread.h>
+#endif
+
 /** generic time stretching structure */
 struct _aubio_timestretch_t
@@ -49 +58 @@
   RubberBandState rb;
   RubberBandOptions rboptions;
+
+#ifdef HAVE_THREADS
+  pthread_t read_thread;
+  pthread_mutex_t read_mutex;
+  pthread_cond_t read_avail;
+  pthread_cond_t read_request;
+  sint_t available;
+#endif
 };
 
@@ -55 +72 @@
 static void aubio_timestretch_warmup (aubio_timestretch_t * p);
 static sint_t aubio_timestretch_fetch(aubio_timestretch_t *p, uint_t fetch);
+#ifdef HAVE_THREADS
+static void *aubio_timestretch_readfn(void *p);
+#endif
 
 aubio_timestretch_t *
@@ -63 +83 @@
   p->samplerate = samplerate;
   p->hopsize = hopsize;
+  //p->source_hopsize = 2048;
   p->source_hopsize = hopsize;
   p->pitchscale = 1.;
@@ -91 +112 @@
   //rubberband_set_debug_level(p->rb, 10);
 
+#ifdef HAVE_THREADS
+  pthread_mutex_init(&p->read_mutex, 0);
+  pthread_cond_init (&p->read_avail, 0);
+  pthread_cond_init (&p->read_request, 0);
+  pthread_create(&p->read_thread, 0, aubio_timestretch_readfn, p);
+  //AUBIO_DBG("timestretch: new_ waiting for warmup, got %d available\n", p->available);
+  pthread_mutex_lock(&p->read_mutex);
+  pthread_cond_wait(&p->read_avail, &p->read_mutex);
+  pthread_mutex_unlock(&p->read_mutex);
+  //AUBIO_DBG("timestretch: new_ warm up success, got %d available\n", p->available);
+#else
   aubio_timestretch_warmup(p);
+#endif
 
   return p;
@@ -100 +133 @@
 }
 
+#ifdef HAVE_THREADS
+void *
+aubio_timestretch_readfn(void *z)
+{
+  aubio_timestretch_t *p = z;
+  // signal main-thread when we are done
+  //AUBIO_WRN("timestretch: read_thread locking, got %d available\n", p->available);
+  pthread_mutex_lock(&p->read_mutex);
+  aubio_timestretch_warmup(p);
+  //AUBIO_WRN("timestretch: signaling warmup\n");
+  pthread_cond_signal(&p->read_avail);
+  //AUBIO_WRN("timestretch: unlocking in readfn\n");
+  pthread_mutex_unlock(&p->read_mutex);
+  AUBIO_WRN("timestretch: entering readfn loop\n");
+  while(1) { //p->available < (int)p->hopsize && p->eof != 1) {
+    //AUBIO_WRN("timestretch: locking in readfn\n");
+    pthread_mutex_lock(&p->read_mutex);
+    p->available = aubio_timestretch_fetch(p, p->hopsize);
+    //AUBIO_WRN("timestretch: read_thread read %d\n", p->available);
+    // signal main-thread when we are done
+    //AUBIO_WRN("timestretch: signaling new read\n");
+    pthread_cond_signal(&p->read_avail);
+    if (p->eof != 1) {
+      pthread_cond_wait(&p->read_request, &p->read_mutex);
+    }
+    if (p->eof == 1) {
+      AUBIO_WRN("timestretch: read_thread eof reached %d, %d/%d\n", p->available,
+        p->hopsize, p->source_hopsize);
+      pthread_mutex_unlock(&p->read_mutex);
+      break;
+    }
+    //AUBIO_WRN("timestretch: unlocking in readfn\n");
+    pthread_mutex_unlock(&p->read_mutex);
+  }
+#if 1
+  pthread_mutex_lock(&p->read_mutex);
+  //AUBIO_WRN("timestretch: signaling end\n");
+  pthread_cond_signal(&p->read_avail);
+  pthread_mutex_unlock(&p->read_mutex);
+#endif
+  //AUBIO_WRN("timestretch: exiting readfn\n");
+  pthread_exit(NULL);
+}
+#endif
+
 static void
 aubio_timestretch_warmup (aubio_timestretch_t * p)
@@ -105 +183 @@
   // warm up rubber band
   unsigned int latency = MAX(p->hopsize, rubberband_get_latency(p->rb));
+#ifdef HAVE_THREADS
+  p->available = aubio_timestretch_fetch(p, latency);
+#else
   aubio_timestretch_fetch(p, latency);
+#endif
 }
 
@@ -111 +193 @@
 del_aubio_timestretch (aubio_timestretch_t * p)
 {
+#ifdef HAVE_THREADS
+  pthread_mutex_lock(&p->read_mutex);
+  pthread_cond_signal(&p->read_request);
+  //pthread_cond_wait(&p->read_avail, &p->read_mutex);
+  pthread_mutex_unlock(&p->read_mutex);
+#if 1
+  void *threadfn;
+  if ((p->eof == 0) && (pthread_cancel(p->read_thread))) {
+      AUBIO_WRN("timestretch: cancelling thread failed\n");
+  }
+  if (pthread_join(p->read_thread, &threadfn)) {
+      AUBIO_WRN("timestretch: joining thread failed\n");
+  }
+#endif
+  pthread_mutex_destroy(&p->read_mutex);
+  pthread_cond_destroy(&p->read_avail);
+  pthread_cond_destroy(&p->read_request);
+#endif
   if (p->in) del_fvec(p->in);
   if (p->source) del_aubio_source(p->source);
@@ -205 +305 @@
 aubio_timestretch_do (aubio_timestretch_t * p, fvec_t * out, uint_t * read)
 {
+#ifndef HAVE_THREADS
   int available = aubio_timestretch_fetch(p, p->hopsize);
+#else /* HAVE_THREADS */
+  int available;
+  pthread_mutex_lock(&p->read_mutex);
+  if (p->eof != 1) {
+    // signal a read request
+    pthread_cond_signal(&p->read_request);
+    // wait for an available signal
+    pthread_cond_wait(&p->read_avail, &p->read_mutex);
+  } else {
+    available = rubberband_available(p->rb);
+  }
+#endif /* HAVE_THREADS */
   // now retrieve the samples and write them into out->data
   if (available >= (int)p->hopsize) {
@@ -217 +330 @@
     *read = 0;
   }
+#ifdef HAVE_THREADS
+  pthread_mutex_unlock(&p->read_mutex);
+#endif
 }
 
@@ -222 +338 @@
 aubio_timestretch_seek (aubio_timestretch_t *p, uint_t pos)
 {
+  uint_t err = AUBIO_OK;
+#if HAVE_THREADS
+  AUBIO_WRN("timestretch: seek_ waiting for warmup, got %d available\n", p->available);
+  pthread_mutex_lock(&p->read_mutex);
+#endif
   p->eof = 0;
   rubberband_reset(p->rb);
-  return aubio_source_seek(p->source, pos);
-}
-
-#endif
+  err = aubio_source_seek(p->source, pos);
+#if HAVE_THREADS
+  p->available = 0;
+  void *threadfn;
+  if ((p->eof == 0) && (pthread_cancel(p->read_thread) == 0)) {
+      AUBIO_WRN("timestretch: cancelling thread failed\n");
+  }
+  if (pthread_join(p->read_thread, &threadfn)) {
+      AUBIO_WRN("timestretch: joining thread failed\n");
+  }
+  pthread_create(&p->read_thread, 0, aubio_timestretch_readfn, p);
+  pthread_cond_wait(&p->read_avail, &p->read_mutex);
+  pthread_mutex_unlock(&p->read_mutex);
+  //AUBIO_WRN("timestretch: seek_ warm up success, got %d available\n", p->available);
+#endif
+  return err;
+}
+
+#endif