Work on making diskstreams follow the transport.

Timeline/Audio_File.C

@@ -41,10 +41,10 @@ Audio_File::from_file ( const char * filename )
 
 done:
 
-    a->_peaks = new Peaks;
+/*     a->_peaks = new Peaks; */
 
-    a->_peaks->clip( a );
-    a->_peaks->open();
+/*     a->_peaks->clip( a ); */
+/*     a->_peaks->open(); */
 
     _open_files[ string( filename ) ] = a;
 

Timeline/Disk_Stream.C

@@ -21,33 +21,42 @@
 #include "Track_Header.H"
 #include "Audio_Track.H"
 #include "Port.H"
+#include "Engine.H" // for locking.
 
-// float Disk_Stream::seconds_to_buffer = 5.0f;
-float Disk_Stream::seconds_to_buffer = 5.0f;
-// size_t Disk_Stream::disk_block_frames = 2048;
+/**********/
+/* Engine */
+/**********/
 
 /* A Disk_Stream uses a separate I/O thread to stream a track's
    regions from disk into a ringbuffer, to be processed by the RT
-   thread (or vice-versa). */
+   thread (or vice-versa). The I/O thread syncronizes access with the
+   user thread via the Timeline mutex. The size of the buffer (in
+   seconds) must be set before any Disk_Stream objects are created;
+   that is, at startup time. The default is 5 seconds, which may or
+   may not be excessive depending on various external factors. */
+
 /* FIXME: handle termination of IO thread in destructor */
-/* FIXME: could all of this not simply be included in the Track_Header
-   class? */
 /* FIXME: deal with (jack) buffer size changes */
-/* FIXME: can this be made to actually handle capture? */
 /* FIXME: needs error handling everywhere! */
+/* TODO: handle capture too */
+
+float Disk_Stream::seconds_to_buffer = 1.0f;
+// size_t Disk_Stream::disk_block_frames = 2048;
 
 Disk_Stream::Disk_Stream ( Track_Header *th, float frame_rate, nframes_t nframes, int channels ) : _th( th )
 {
     _frame = 0;
     _thread = 0;
+    _terminate = false;
+    _pending_seek = -1;
 
     printf( "nframes %lu\n", nframes );
 
-    const int blocks = frame_rate * seconds_to_buffer / nframes;
+    _total_blocks = frame_rate * seconds_to_buffer / nframes;
 
     _nframes = nframes;
 
-    size_t bufsize = blocks * nframes * sizeof( sample_t );
+    size_t bufsize = _total_blocks * nframes * sizeof( sample_t );
 
 /*     const int blocks = 64; */
 /*     const size_t bufsize = (blocks * (nframes * sizeof( sample_t ))) + sizeof( sample_t ); */
@@ -55,19 +64,29 @@ Disk_Stream::Disk_Stream ( Track_Header *th, float frame_rate, nframes_t nframes
     for ( int i = channels; i--; )
         _rb.push_back( jack_ringbuffer_create( bufsize ) );
 
-    sem_init( &_blocks, 0, blocks );
+    sem_init( &_blocks, 0, _total_blocks );
 
     run();
 }
 
 Disk_Stream::~Disk_Stream ( )
 {
+
+    /* stop the IO thread */
+    _terminate = true;
+    pthread_join( _thread, NULL );
+
+    /* it isn't safe to do all this with the RT thread running */
+    engine->lock();
+
     _th = NULL;
 
     sem_destroy( &_blocks );
 
     for ( int i = channels(); i--; )
         jack_ringbuffer_free( _rb[ i ] );
+
+    engine->unlock();
 }
 
 Audio_Track *
@@ -84,6 +103,56 @@ Disk_Stream::run ( void )
         /* error */;
 }
 
+/* to be called when the JACK buffer size changes. */
+void
+Disk_Stream::resize ( nframes_t nframes )
+{
+    if ( nframes != _nframes )
+        /* FIXME: to something here! */;
+}
+
+bool
+Disk_Stream::seek_pending ( void )
+{
+    return _pending_seek != (nframes_t)-1;
+}
+
+/* THREAD: RT */
+/** request that the IO thread perform a seek and rebuffer.  This is
+ called for each Disk_Stream whenever the RT thread determines that
+ the transport has jumped to a new position. This is called *before*
+ process. */
+void
+Disk_Stream::seek ( nframes_t frame )
+{
+    printf( "requesting seek\n" );
+
+    if ( seek_pending() )
+        printf( "seek error, attempt to seek while seek is pending\n" );
+
+    _pending_seek = frame;
+
+    /* flush buffers */
+    for ( int i = channels(); i--; )
+        jack_ringbuffer_read_advance( _rb[ i ], jack_ringbuffer_read_space( _rb[ i ] ) );
+
+    /* dirty hack... reset the semaphore. Should we just call sem_init
+     * again instead? */
+
+/*     sem_init( &_blocks, 0, _total_blocks ); */
+
+    int n;
+    sem_getvalue( &_blocks, &n );
+
+    n = _total_blocks - n;
+
+    while ( n-- )
+        sem_post( &_blocks );
+
+}
+
+
+
 /* void */
 /* DIsk_Stream::shutdown ( void ) */
 /* { */
@@ -100,9 +169,9 @@ Disk_Stream::io_thread ( void *arg )
 }
 
 /* THREAD: IO */
-/** read a block of data from the track into /buf/ */
+/** read /nframes/ from the attached track into /buf/ */
 void
-Disk_Stream::read_block ( sample_t *buf )
+Disk_Stream::read_block ( sample_t *buf, nframes_t nframes )
 {
 
     /* stupid chicken/egg */
@@ -119,14 +188,24 @@ Disk_Stream::read_block ( sample_t *buf )
 
     timeline->rdlock();
 
-    if ( track()->play( buf, _frame, _nframes, channels() ) )
-        _frame += _nframes;
+    if ( track()->play( buf, _frame, nframes, channels() ) )
+        _frame += nframes;
     else
         /* error */;
 
     timeline->unlock();
 }
 
+int
+Disk_Stream::output_buffer_percent ( void )
+{
+    int n;
+
+    sem_getvalue( &_blocks, &n );
+
+    return 100 - (n * 100 / _total_blocks);
+}
+
 /* THREAD: IO */
 void
 Disk_Stream::io_thread ( void )
@@ -145,7 +224,26 @@ Disk_Stream::io_thread ( void )
     {
 //        printf( "IO: RT thread is ready for more data...\n" );
 
-        read_block( buf );
+        printf( "IO: disk buffer is %3d%% full\r", output_buffer_percent() );
+
+//        lock(); // for seeking
+
+        if ( seek_pending() )
+        {
+            printf( "performing seek\n" );
+            _frame = _pending_seek;
+            _pending_seek = -1;
+            /* finish flushing the buffer */
+
+/*             for ( int i = channels(); i-- ) */
+/*                 jack_ringbuffer_write_advance( _rb[ i ], jack_ringbuffer_write_space( _rb[ i ] ) ); */
+
+        }
+
+        /* FIXME: should we not read from disk in larger-than-JACK-buffer blocks? */
+        read_block( buf, _nframes );
+
+//        unlock(); // for seeking
 
         /* deinterleave the buffer and stuff it into the per-channel ringbuffers */
 
@@ -158,6 +256,7 @@ Disk_Stream::io_thread ( void )
             while ( jack_ringbuffer_write_space( _rb[ i ] ) < block_size )
             {
                 printf( "IO: disk buffer overrun!\n" );
+                /* FIXME: is this *really* the right thing to do?  */
                 usleep( 2000 );
             }
 
@@ -165,13 +264,15 @@ Disk_Stream::io_thread ( void )
         }
     }
 
+    printf( "IO thread terminating.\n" );
+
     delete[] buf;
     delete[] cbuf;
 }
 
 /* THREAD: RT */
-/** take a block from the ringbuffers and send it out the track's
- *  ports */
+/** take a single block from the ringbuffers and send it out the
+ *  attached track's ports */
 nframes_t
 Disk_Stream::process ( nframes_t nframes )
 {
@@ -184,19 +285,11 @@ Disk_Stream::process ( nframes_t nframes )
 
         void *buf = _th->output[ i ].buffer( nframes );
 
-        /* FIXME: handle underrun */
-
-/*         if ( jack_ringbuffer_read_space( _rb[ i ] ) < block_size ) */
-/*         { */
-/*             printf( "disktream (rt): buffer underrun!\n" ); */
-/*             memset( buf, 0, block_size ); */
-/*         } */
-/*         else */
-
         if ( jack_ringbuffer_read( _rb[ i ], (char*)buf, block_size ) < block_size )
         {
             printf( "RT: buffer underrun (disk can't keep up).\n" );
             memset( buf, 0, block_size );
+            /* FIXME: we need to resync somehow */
         }
 
 /*             /\* testing. *\/ */

Timeline/Disk_Stream.H

@@ -27,13 +27,15 @@
 
 #include <pthread.h>
 
+#include "Mutex.H"
+
 #include <vector>
 using std::vector;
 
 class Track_Header;
 class Audio_Track;
 
-class Disk_Stream
+class Disk_Stream : public Mutex
 {
 
     pthread_t _thread;
@@ -49,7 +51,11 @@ class Disk_Stream
 
     sem_t _blocks;                                                /* semaphore to wake the IO thread with */
 
-//   volatile  nframes_t _seek_request;
+    int _total_blocks;
+
+    volatile nframes_t _pending_seek;                           /* absolute transport position to seek to */
+
+    volatile int _terminate;
 
     int channels ( void ) const { return _rb.size(); }
 
@@ -60,7 +66,19 @@ class Disk_Stream
 protected:
 
     void block_processed ( void ) { sem_post( &_blocks ); }
-    bool wait_for_block ( void ) { while ( sem_wait( &_blocks ) == EINTR ); return true; }
+    bool wait_for_block ( void )
+        {
+            if ( _terminate )
+                return false;
+            else
+            {
+                while ( sem_wait( &_blocks ) == EINTR );
+                return true;
+            }
+        }
+
+    void read_block ( sample_t *buf, nframes_t nframes );
+    void io_thread ( void );
 
 public:
 
@@ -71,23 +89,12 @@ public:
 
     virtual ~Disk_Stream ( );
 
-    void
-    resize ( nframes_t nframes )
-        {
-            if ( nframes != _nframes )
-                /* FIXME: to something here! */;
-        }
-
-    void
-    seek ( nframes_t frame )
-        {
-            _frame = frame;
-            /* FIXME: need to signal the IO thread somehow? */
-        }
+    void resize ( nframes_t nframes );
+    void seek ( nframes_t frame );
+    bool seek_pending ( void );
 
     void run ( void );
-    void read_block ( sample_t *buf );
-    void io_thread ( void );
     nframes_t process ( nframes_t nframes );
 
+    int output_buffer_percent ( void );
 };

Timeline/Engine.C

@@ -40,6 +40,63 @@ Engine::process ( nframes_t nframes, void *arg )
     return ((Engine*)arg)->process( nframes );
 }
 
+/* static wrapper */
+int
+Engine::sync ( jack_transport_state_t state, jack_position_t *pos, void *arg )
+{
+    return ((Engine*)arg)->sync( state, pos );
+}
+
+void
+Engine::request_locate ( nframes_t frame )
+{
+    if ( timeline )
+        timeline->seek( frame );
+}
+
+/* THREAD: RT */
+/** This is the jack slow-sync callback. */
+int
+Engine::sync ( jack_transport_state_t state, jack_position_t *pos )
+{
+    static bool seeking = false;
+
+    switch ( state )
+    {
+        case JackTransportStopped:           /* new position requested */
+            /* JACK docs lie. This is only called when the transport
+               is *really* stopped, not when starting a slow-sync
+               cycle */
+            request_locate( pos->frame );
+            return 1;
+         case JackTransportStarting:          /* this means JACK is polling slow-sync clients */
+        {
+            if ( ! seeking )
+            {
+                request_locate( pos->frame );
+                seeking = true;
+            }
+
+            int r = timeline->seek_pending();
+
+            if ( ! r )
+                seeking = false;
+
+            return ! seeking;
+        }
+        case JackTransportRolling:           /* JACK's timeout has expired */
+            /* FIXME: what's the right thing to do here? */
+//            request_locate( pos->frame );
+            return 1;
+//            return transport.frame == pos->frame;
+            break;
+        default:
+            printf( "unknown transport state.\n" );
+    }
+
+    return 0;
+}
+
 /* THREAD: RT */
 int
 Engine::process ( nframes_t nframes )
@@ -83,6 +140,10 @@ Engine::init ( void )
 
     jack_set_process_callback( _client, &Engine::process, this );
 
+    /* FIXME: should we wait to register this until after the session
+     has been loaded (and we have disk threads running)? */
+    jack_set_sync_callback( _client, &Engine::sync, this );
+
     jack_activate( _client );
 
     /* we don't need to create any ports until tracks are created */

Timeline/Engine.H

@@ -44,6 +44,8 @@ class Engine : public Mutex
 
     static int process ( nframes_t nframes, void *arg );
     int process ( nframes_t nframes );
+    static int sync ( jack_transport_state_t state, jack_position_t *pos, void *arg );
+    int sync ( jack_transport_state_t state, jack_position_t *pos );
 
 private:
 
@@ -52,12 +54,12 @@ private:
 
 public:
 
-
-
     Engine ( );
 
     int init ( void );
 
+    void request_locate ( nframes_t frame );
+
     nframes_t nframes ( void ) const { return jack_get_buffer_size( _client ); }
     float frame_rate ( void ) const { return jack_get_sample_rate( _client ); }
 

Timeline/Peaks.C

@@ -38,6 +38,8 @@
 
 #include <math.h>
 
+#include <FL/Fl.H> // for Fl::check();
+
 Peaks::peakbuffer Peaks::_peakbuf;
 
 
@@ -305,6 +307,8 @@ Peaks::current ( void ) const
 }
 
 
+/* FIXME: we need to work out a way to run this in another thread and
+ possibly stream back the data to the GUI */
 /** build peaks file for /filename/ if necessary */
 bool
 Peaks::make_peaks ( int chunksize )
@@ -334,6 +338,8 @@ Peaks::make_peaks ( int chunksize )
     do {
         len = read_source_peaks( peaks, 1, chunksize );
         fwrite( peaks, sizeof( peaks ), 1, fp );
+        /* FIXME: GUI code shouldn't be here! */
+        Fl::check();
     }
     while ( len );
 

Timeline/Timeline.C

@@ -29,6 +29,9 @@
 
 #include "Track_Header.H"
 
+
+#include "Disk_Stream.H"
+
 void
 Timeline::cb_scroll ( Fl_Widget *w, void *v )
 {
@@ -153,7 +156,7 @@ Timeline::Timeline ( int X, int Y, int W, int H, const char* L ) : Fl_Overlay_Wi
             o->type( Fl_Pack::VERTICAL );
             o->spacing( 0 );
 
-            for ( int i = 1; i--;  )
+            for ( int i = 2; i--;  )
             {
 //                Track_Header *t = new Track_Header( 0, 0, W, 75 );
                 Track_Header *t = new Track_Header( 0, 0, W, 30 );
@@ -616,3 +619,30 @@ Timeline::process ( nframes_t nframes )
     /* FIXME: BOGUS */
     return nframes;
 }
+
+/* THREAD: RT */
+void
+Timeline::seek ( nframes_t frame )
+{
+    for ( int i = tracks->children(); i-- ; )
+    {
+        Track_Header *t = (Track_Header*)tracks->child( i );
+
+        t->seek( frame );
+    }
+}
+
+/* THREAD: RT */
+int
+Timeline::seek_pending ( void )
+{
+    int r = 0;
+
+    for ( int i = tracks->children(); i-- ; )
+    {
+        Track_Header *t = (Track_Header*)tracks->child( i );
+
+        if ( t->diskstream )
+            r += t->diskstream->output_buffer_percent() < 50;
+    }
+}

Timeline/Timeline.H

@@ -148,5 +148,8 @@ private:
 
     friend class Engine; // FIXME: only Engine::process() needs to be friended.x
 
+    /* Engine */
     nframes_t process ( nframes_t nframes );
+    void seek ( nframes_t frame );
+    int seek_pending ( void );
 };

Timeline/Track_Header.C

@@ -100,7 +100,13 @@ Track_Header::Track_Header ( int X, int Y, int W, int H, const char *L ) :
     _show_all_takes = false;
     _size = 1;
 
-    output.push_back( Port( "foo" ) );
+    {
+        char pname[40];
+        static int n = 0;
+        snprintf( pname, sizeof( pname ), "out-%d", n++ );
+
+        output.push_back( Port( strdup( pname ) ) );
+    }
 
     diskstream = new Disk_Stream( this, engine->frame_rate(), engine->nframes(), 1 );
 
@@ -299,3 +305,11 @@ Track_Header::process ( nframes_t nframes )
     else
         return 0;
 }
+
+/* THREAD: RT */
+void
+Track_Header::seek ( nframes_t frame )
+{
+    if ( diskstream )
+        return diskstream->seek( frame );
+}

Timeline/Track_Header.H

@@ -264,6 +264,8 @@ public:
         }
 
 
+    /* Engine */
     nframes_t process ( nframes_t nframes );
+    void seek ( nframes_t frame );
 };
 #endif