Timeline: Fix numerous issues with diskstream.

timeline/src/Engine/Disk_Stream.C

@@ -61,13 +61,14 @@ Disk_Stream::Disk_Stream ( Track *track, float frame_rate, nframes_t nframes, in
 
     _frame = 0;
     _terminate = false;
-    _pending_seek = -1;
+    _pending_seek = false;
+    _seek_frame = 0;
     _xruns = 0;
     _frame_rate = frame_rate;
 
+    sem_init( &_blocks, 0, 0 );
+        
     _resize_buffers( nframes, channels );
-
-    sem_init( &_blocks, 0, _total_blocks );
 }
 
 Disk_Stream::~Disk_Stream ( )
@@ -94,37 +95,18 @@ Disk_Stream::~Disk_Stream ( )
 void
 Disk_Stream::base_flush ( bool is_output )
 {
-    THREAD_ASSERT( RT );
+//    THREAD_ASSERT( RT );
 
     /* flush buffers */
-    for ( int i = _rb.size(); i--; )
-        jack_ringbuffer_read_advance( _rb[ i ], jack_ringbuffer_read_space( _rb[ i ] ) );
-
-/*  sem_destroy( &_blocks ); */
-
-/*     if ( is_output ) */
-/*         sem_init( &_blocks, 0, _total_blocks ); */
-/*     else */
-/*         sem_init( &_blocks, 0, 0 ); */
+    for ( unsigned int i = _rb.size(); i--; )
+        jack_ringbuffer_reset( _rb[ i ] );
 
+    sem_destroy( &_blocks );
+    
     if ( is_output )
-    {
-        int n;
-        sem_getvalue( &_blocks, &n );
-
-        n = _total_blocks - n;
-
-        while ( n-- )
-            sem_post( &_blocks );
-    }
+        sem_init( &_blocks, 0, _total_blocks );
     else
-    {
-        sem_destroy( &_blocks );
-
         sem_init( &_blocks, 0, 0 );
-    }
-
-
 }
 
 /** signal thread to terminate, then detach it */
@@ -156,10 +138,6 @@ Disk_Stream::shutdown ( void )
         }
         
         _thread.join();
-
-        sem_destroy( &_blocks );
-
-        sem_init( &_blocks, 0, 0 );
     }
 }
 
@@ -195,7 +173,7 @@ Disk_Stream::_resize_buffers ( nframes_t nframes, int channels )
 
     _nframes = nframes;
 
-    _total_blocks = _frame_rate * seconds_to_buffer / nframes;
+    _total_blocks = ( _frame_rate * seconds_to_buffer ) / nframes;
 
     size_t bufsize = _total_blocks * nframes * sizeof( sample_t );
 
@@ -222,10 +200,10 @@ Disk_Stream::resize_buffers ( nframes_t nframes )
         if ( was_running )
             shutdown();
 
-        flush();
-
         _resize_buffers( nframes, channels() );
 
+        flush();
+
         if ( was_running )
             run();
     }

timeline/src/Engine/Disk_Stream.H

@@ -56,14 +56,15 @@ protected:
 
     sem_t _blocks;          /* semaphore to wake the IO thread with */
 
-    int _total_blocks; /* total number of blocks that we can  buffer */
-    int _disk_io_blocks; /* the number of blocks to read/write to/from disk at once */
+    nframes_t _total_blocks; /* total number of blocks that we can  buffer */
+    nframes_t _disk_io_blocks; /* the number of blocks to read/write to/from disk at once */
 
 
     nframes_t _frame_rate;      /* used for buffer size calculations */
 
     volatile nframes_t _frame;             /* location of disk read */
-    volatile nframes_t _pending_seek; /* absolute transport position to seek to */
+    volatile nframes_t _seek_frame; /* absolute transport position to seek to */
+    volatile bool _pending_seek; /* absolute transport position to seek to */
     volatile int _terminate;
 
     volatile int _xruns;
@@ -82,6 +83,9 @@ protected:
     void block_processed ( void ) { sem_post( &_blocks ); }
     bool wait_for_block ( void )
         {
+            if ( _terminate )
+                return false;
+
             while ( ! sem_wait( &_blocks ) && errno == EINTR )
             {}
 
@@ -117,6 +121,6 @@ public:
 
     virtual nframes_t process ( nframes_t nframes ) = 0;
 
-    int buffer_percent ( void );
+    virtual int buffer_percent ( void );
 
 };

timeline/src/Engine/Engine.C

@@ -174,10 +174,6 @@ Engine::process ( nframes_t nframes )
 
     if ( freewheeling() )
     {
-        /* freewheeling mode/export. We're actually running
-           non-RT. Assume that everything is quiescent. do I/O
-           synchronously */
-        
         if ( timeline )
         {
             timeline->rdlock();
@@ -185,9 +181,6 @@ Engine::process ( nframes_t nframes )
             timeline->process( nframes );
 
             timeline->unlock();
-
-            /* because we're going faster than realtime. */
-            timeline->wait_for_buffers();
         }
     }
     else

timeline/src/Engine/Playback_DS.C

@@ -38,7 +38,7 @@
 bool
 Playback_DS::seek_pending ( void )
 {
-    return _pending_seek != (nframes_t)-1;
+    return _pending_seek || buffer_percent() < 50;
 }
 
 /** request that the IO thread perform a seek and rebuffer.  This is
@@ -56,9 +56,11 @@ Playback_DS::seek ( nframes_t frame )
     if ( seek_pending() )
         printf( "seek error, attempt to seek while seek is pending\n" );
 
-    _pending_seek = frame;
+    _seek_frame = frame;
+    _pending_seek = true;
 
-    flush();
+    /* wake the IO thread */
+    block_processed();
 }
 
 /** set the playback delay to /frames/ frames. This be called prior to
@@ -82,30 +84,28 @@ Playback_DS::read_block ( sample_t *buf, nframes_t nframes )
     if ( !timeline )
         return;
 
-    while ( ! _terminate )
+    while ( timeline->tryrdlock() )
     {
-        if ( ! timeline->tryrdlock() )
-        {
-            if ( sequence() )
-            {
-                /* FIXME: how does this work if _delay is not a multiple of bufsize? */
-                
-                if ( _frame >= _delay )
-                {
-                    if ( ! sequence()->play( buf, _frame - _delay, nframes, channels() ) )
-                        WARNING( "Programming error?" );
-                }
-                
-                _frame += nframes;
-            }
-            
-            timeline->unlock();
-            
+        if ( _terminate )
             return;
-        }
-
+        
         usleep( 1000 * 10 );
     }
+    
+    if ( sequence() )
+    {
+        /* FIXME: how does this work if _delay is not a multiple of bufsize? */
+        
+        if ( _frame >= _delay )
+        {
+            if ( ! sequence()->play( buf, _frame - _delay, nframes, channels() ) )
+                WARNING( "Programming error?" );
+        }
+        
+        _frame += nframes;
+    }
+    
+    timeline->unlock();
 }
 
 void
@@ -117,61 +117,63 @@ Playback_DS::disk_thread ( void )
 
     /* buffer to hold the interleaved data returned by the track reader */
     sample_t *buf = buffer_alloc( _nframes * channels() * _disk_io_blocks );
-    sample_t *cbuf = buffer_alloc( _nframes * _disk_io_blocks );
-
-    int blocks_ready = 0;
+    sample_t *cbuf = buffer_alloc( _nframes );
 
-    const nframes_t nframes = _nframes * _disk_io_blocks;
+    const nframes_t nframes = _nframes;
+    nframes_t blocks_written;
 
-    while ( wait_for_block() )
+    while ( ! _terminate )
     {
 
-//        lock(); // for seeking
+    seek:
 
-        if ( seek_pending() )
-        {
-            /* FIXME: non-RT-safe IO */
-            DMESSAGE( "performing seek to frame %lu", (unsigned long)_pending_seek );
-
-            _frame = _pending_seek;
-            _pending_seek = -1;
-            blocks_ready = 0;
-        }
+        blocks_written = 0;
+        read_block( buf, nframes * _disk_io_blocks );
 
-        if ( ++blocks_ready < _disk_io_blocks )
+        while ( blocks_written < _disk_io_blocks &&
+                wait_for_block() )
         {
-            /* wait for more space */
-            continue;
-        }
-
-        /* reset */
-        blocks_ready = 0;
-
-        read_block( buf, nframes );
+//        lock(); // for seeking
+            
+            if ( _pending_seek )
+            {
+                /* FIXME: non-RT-safe IO */
+                DMESSAGE( "performing seek to frame %lu", (unsigned long)_seek_frame );
+                
+                _frame = _seek_frame;
+                _pending_seek = false;
 
-        /* might have received terminate signal while waiting for block */
-        if ( _terminate )
-            goto done;
+                flush();
+                
+                goto seek;
+            }
 
+            /* might have received terminate signal while waiting for block */
+            if ( _terminate )
+                goto done;
+        
 //        unlock(); // for seeking
+        
+            /* deinterleave the buffer and stuff it into the per-channel ringbuffers */
 
-        /* deinterleave the buffer and stuff it into the per-channel ringbuffers */
-
-        const size_t block_size = nframes * sizeof( sample_t );
+            const size_t block_size = nframes * sizeof( sample_t );
 
-        for ( int i = channels(); i--; )
-        {
-            buffer_deinterleave_one_channel( cbuf, buf, i, channels(), nframes );
+            for ( int i = 0; i < channels(); i++ )
+            {
+                buffer_deinterleave_one_channel( cbuf,
+                                                 buf + ( blocks_written * nframes * channels() ),
+                                                 i,
+                                                 channels(), 
+                                                 nframes );
 
-            size_t wr = 0;
+                while ( jack_ringbuffer_write_space( _rb[ i ] ) < block_size )
+                    usleep( 100 * 1000 );
 
-            while ( wr < block_size )
-            {
-                wr += jack_ringbuffer_write( _rb[ i ], ((char*)cbuf) + wr, block_size - wr );
-//                usleep( 10 * 1000 );
+                jack_ringbuffer_write( _rb[ i ], ((char*)cbuf), block_size );
             }
-         }
 
+            blocks_written++;
+        }
     }
 
 done:
@@ -203,22 +205,25 @@ Playback_DS::process ( nframes_t nframes )
 
         if ( engine->freewheeling() )
         {
-            size_t rd = 0;
-
-            while ( rd < block_size )
-            {
-                rd += jack_ringbuffer_read( _rb[ i ], ((char*)buf) + rd, block_size - rd );
-//                usleep( 10 * 1000 );
-            }
+            /* only ever read nframes at a time */
+            while ( jack_ringbuffer_read_space( _rb[i] ) < block_size )
+                usleep( 10 * 1000 );
+            
+            jack_ringbuffer_read( _rb[ i ], ((char*)buf), block_size );
         }
         else
-        {
-            if ( jack_ringbuffer_read( _rb[ i ], (char*)buf, block_size ) < block_size )
+       {
+            /* only ever read nframes at a time */
+            if ( jack_ringbuffer_read_space( _rb[i] ) < block_size )
             {
                 ++_xruns;
                 memset( buf, 0, block_size );
                 /* FIXME: we need to resync somehow */
             }
+            else
+            {
+                jack_ringbuffer_read( _rb[ i ], (char*)buf, block_size );
+            }
         }
 
         /* TODO: figure out a way to stop IO while muted without losing sync */

timeline/src/Engine/Record_DS.C

@@ -74,39 +74,40 @@ Record_DS::disk_thread ( void )
 
     track()->record( _capture, _frame );
 
-    const nframes_t nframes = _nframes * _disk_io_blocks;
+    const nframes_t nframes = _nframes;
 
     /* buffer to hold the interleaved data returned by the track reader */
-    sample_t *buf = buffer_alloc( nframes * channels() );
+    sample_t *buf = buffer_alloc( nframes * channels() * _disk_io_blocks );
     sample_t *cbuf = buffer_alloc( nframes );
 
     const size_t block_size = nframes * sizeof( sample_t );
 
-    int blocks_ready = 0;
+    nframes_t blocks_read = 0;
 
     while ( wait_for_block() )
     {
-        if ( ++blocks_ready < _disk_io_blocks )
-            continue;
-        else
-            blocks_ready = 0;
-
         /* pull data from the per-channel ringbuffers and interlace it */
         for ( int i = channels(); i--; )
         {
-            size_t rd = 0;
-
-            while ( rd < block_size )
-            {
-                rd += jack_ringbuffer_read( _rb[ i ], ((char*)cbuf) + rd, block_size - rd );
-//                usleep( 10 * 1000 );
-            }
-
-            buffer_interleave_one_channel( buf, cbuf, i, channels(), nframes );
+            while ( jack_ringbuffer_read_space( _rb[ i ] ) < block_size )
+                usleep( 10 * 1000 );
+
+            jack_ringbuffer_read( _rb[ i ], ((char*)cbuf), block_size );
+            
+            buffer_interleave_one_channel( buf + ( blocks_read * nframes * channels() ),
+                                           cbuf, 
+                                           i,
+                                           channels(),
+                                           nframes );
         }
 
-        write_block( buf, nframes );
+        blocks_read++;
 
+        if ( blocks_read == _disk_io_blocks )
+        {
+            write_block( buf, nframes * _disk_io_blocks );
+            blocks_read = 0;
+        }
     }
 
     DMESSAGE( "capture thread terminating" );
@@ -114,11 +115,7 @@ Record_DS::disk_thread ( void )
     /* flush what remains in the buffer out to disk */
 
     {
-        /* use JACk sized blocks for this last bit */
-        const nframes_t nframes = _nframes;
-        const size_t block_size = _nframes * sizeof( sample_t );
-
-        while ( blocks_ready-- > 0 || ( ! sem_trywait( &_blocks ) && errno != EAGAIN ) )
+        while ( blocks_read-- > 0 || ( ! sem_trywait( &_blocks ) && errno != EAGAIN ) )
         {
             for ( int i = channels(); i--; )
             {
@@ -138,8 +135,6 @@ Record_DS::disk_thread ( void )
             else
                 write_block( buf, nframes );
         }
-
-
     }
 
     free(buf);
@@ -259,21 +254,22 @@ Record_DS::process ( nframes_t nframes )
 
         if ( engine->freewheeling() )
         {
-            size_t wr = 0;
+            while ( jack_ringbuffer_write_space( _rb[i] ) < block_size )
+                usleep( 10 * 1000 );
 
-            while ( wr < block_size )
-            {
-                wr += jack_ringbuffer_write( _rb[ i ], ((char*)buf) + offset_size + wr, block_size - wr );
-//                usleep( 10 * 1000 );
-            }
+            jack_ringbuffer_write( _rb[ i ], ((char*)buf) + offset_size, block_size );
         }
         else
         {
-            if ( jack_ringbuffer_write( _rb[ i ], ((char*)buf) + offset_size, block_size ) < block_size )
+            if ( jack_ringbuffer_write_space( _rb[i] ) < block_size )
             {
-                ++_xruns;
                 memset( buf, 0, block_size );
                 /* FIXME: we need to resync somehow */
+                ++_xruns;
+            }
+            else
+            {
+                jack_ringbuffer_write( _rb[ i ], ((char*)buf) + offset_size, block_size );
             }
         }
     }

timeline/src/Engine/Timeline.C

@@ -225,7 +225,7 @@ Timeline::seek_pending ( void )
         Track *t = (Track*)tracks->child( i );
 
         if ( t->playback_ds )
-            if ( t->playback_ds->buffer_percent() < 50 )
+            if ( t->playback_ds->seek_pending() )
                 return true;
     }
 
@@ -283,15 +283,6 @@ Timeline::total_output_buffer_percent ( void )
     return r / cnt;
 }
 
-/** wait for I/O threads to fill their buffers */
-void
-Timeline::wait_for_buffers ( void )
-{
-    while ( total_output_buffer_percent() + total_input_buffer_percent() < 200 )
-        usleep( 5000 );
-}
-
-
 int
 Timeline::total_playback_xruns ( void )
 {