Changeset 391

Timestamp:

02/09/07 00:01:30 (16 years ago)

Author:

pieterpalmers

Message:

* Partially finished:

• Introduce TimestampedBuffer? util class
  
• replace interal ringbuffer of SP with timed ringbuffer
  

* Compiles & works

Files:

• branches/streaming-rework/src/libstreaming/AmdtpStreamProcessor.cpp (modified) (34 diffs)
• branches/streaming-rework/src/libstreaming/AmdtpStreamProcessor.h (modified) (4 diffs)
• branches/streaming-rework/src/libstreaming/IsoHandler.cpp (modified) (1 diff)
• branches/streaming-rework/src/libstreaming/IsoHandler.h (modified) (1 diff)
• branches/streaming-rework/src/libstreaming/IsoHandlerManager.cpp (modified) (2 diffs)
• branches/streaming-rework/src/libstreaming/StreamProcessor.cpp (modified) (10 diffs)
• branches/streaming-rework/src/libstreaming/StreamProcessor.h (modified) (9 diffs)
• branches/streaming-rework/src/libstreaming/StreamProcessorManager.cpp (modified) (3 diffs)
• branches/streaming-rework/src/libutil/TimestampedBuffer.cpp (added)
• branches/streaming-rework/src/libutil/TimestampedBuffer.h (added)
• branches/streaming-rework/src/Makefile.am (modified) (2 diffs)

branches/streaming-rework/src/libstreaming/AmdtpStreamProcessor.cpp

@@ -53 +53 @@
-/* transmit */
-AmdtpTransmitStreamProcessor::AmdtpTransmitStreamProcessor(int port, int framerate, int dimension)
-
-
-
-
+
+
+
-
-}
-
-AmdtpTransmitStreamProcessor::~AmdtpTransmitStreamProcessor() {
-
-
+
-}
-
@@ -95 +93 @@
-    unsigned int nevents=0;
-    
+    m_last_cycle=cycle;
+    
-    debugOutput(DEBUG_LEVEL_VERY_VERBOSE,"Xmit handler for cycle %d, (running=%d, enabled=%d,%d)\n",
-        cycle, m_running, m_disabled, m_is_disabled);
@@ -124 +124 @@
-    uint64_t fc;
-    
-
+m_data_buffer->
-    
-    int64_t timestamp = ts_tail;
@@ -134 +134 @@
-    
-    // FIXME: test
-
-
+
+
+
+
+
-    
-    // this happens if m_buffer_tail_timestamp wraps around while there are 
@@ -268 +271 @@
-            debugOutput(DEBUG_LEVEL_VERBOSE,"Preparing to enable...\n");
-            
-
-
+
+
+
+
+
+
+
+
+
+
+
+
-            // recalculate the buffer head timestamp
-            float ticks_per_frame=m_SyncSource->getTicksPerFrame();
@@ -278 +291 @@
-            // plus one frame
-            ts += (uint64_t)ticks_per_frame;
+            
+            // account for the cycle lag between sync SP and this SP
+            ts += sync_lag_cycles * TICKS_PER_CYCLE;
+            
-            if (ts >= TICKS_PER_SECOND * 128L) {
-                ts -= TICKS_PER_SECOND * 128L;
-            }
-
-            
+
+//             m_data_buffer->
-            int64_t timestamp = ts;
-        
@@ -289 +306 @@
-            // frames_in_buffer * rate 
-            // later
-
+m_data_buffer->
-            timestamp += (int64_t)((float)frames_in_buffer * ticks_per_frame);
-            
@@ -302 +319 @@
-            }
-        
-StreamProcessor::
+m_data_buffer->
-            
-            debugOutput(DEBUG_LEVEL_VERBOSE,"XMIT TS SET: TS=%10lld, TSTMP=%10llu, FC=%4d, %f\n",
@@ -349 +366 @@
-    *tag = IEC61883_TAG_WITH_CIP;
-    *sy = 0;
-
-
-
-
-
+
+
-    {
+        *length = nevents*sizeof(quadlet_t)*m_dimension + 8;
+
+        // process all ports that should be handled on a per-packet base
+        // this is MIDI for AMDTP (due to the need of DBC)
+        if (!encodePacketPorts((quadlet_t *)(data+8), nevents, packet->dbc)) {
+            debugWarning("Problem encoding Packet Ports\n");
+        }
+
+        packet->fdf = m_fdf;
+
+        // convert the timestamp to SYT format
+        uint64_t ts=timestamp + TRANSMIT_TRANSFER_DELAY;
+        
+        // check if it wrapped
+        if (ts >= TICKS_PER_SECOND * 128L) {
+            ts -= TICKS_PER_SECOND * 128L;
+        }
+        
+        unsigned int timestamp_SYT = TICKS_TO_SYT(ts);
+        packet->syt = ntohs(timestamp_SYT);
+        
+        // update the frame counter such that it reflects the new value
+        // done in the SP base class
+        if (!StreamProcessor::getFrames(nevents)) {
+            debugError("Could not do StreamProcessor::getFrames(%d)\n",nevents);
+             return RAW1394_ISO_ERROR;
+        }
+        
+        return RAW1394_ISO_OK;    
+    } else {
-        /* there is no more data in the ringbuffer */
-        // convert the timestamp to SYT format
@@ -392 +436 @@
-
-        return RAW1394_ISO_DEFER;
-    } else {
-        *length = read_size + 8;
-
-        // process all ports that should be handled on a per-packet base
-        // this is MIDI for AMDTP (due to the need of DBC)
-        if (!encodePacketPorts((quadlet_t *)(data+8), nevents, packet->dbc)) {
-            debugWarning("Problem encoding Packet Ports\n");
-        }
-
-        packet->fdf = m_fdf;
-
-        // convert the timestamp to SYT format
-        uint64_t ts=timestamp + TRANSMIT_TRANSFER_DELAY;
-        
-        // check if it wrapped
-        if (ts >= TICKS_PER_SECOND * 128L) {
-            ts -= TICKS_PER_SECOND * 128L;
-        }
-        
-        unsigned int timestamp_SYT = TICKS_TO_SYT(ts);
-        packet->syt = ntohs(timestamp_SYT);
-
-        // calculate the new buffer head timestamp. this is
-        // the previous buffer head timestamp plus
-        // the number of frames sent * ticks_per_frame
-        timestamp += (int64_t)((float)nevents * ticks_per_frame );
-        
-        // check if it wrapped
-        if (timestamp >= TICKS_PER_SECOND * 128L) {
-            timestamp -= TICKS_PER_SECOND * 128L;
-        }
-        
-        // update the frame counter such that it reflects the new value
-        // also update the buffer head timestamp
-        // done in the SP base class
-        if (!StreamProcessor::getFrames(nevents, timestamp)) {
-            debugError("Could not do StreamProcessor::getFrames(%d, %llu)\n",nevents, timestamp);
-             return RAW1394_ISO_ERROR;
-        }
-        
-        return RAW1394_ISO_OK;
-    }
-
@@ -470 +473 @@
-    uint64_t ts;
-    uint64_t fc;
-
+m_data_buffer->
-
-    // update the frame counter such that it reflects the buffer content,
@@ -476 +479 @@
-    // done in the SP base class
-    if (!StreamProcessor::putFrames(m_ringbuffer_size_frames, ts)) {
-
-
+11llu
+,ts
-        return false;
-    }
@@ -488 +491 @@
-    debugOutput( DEBUG_LEVEL_VERBOSE, "Resetting...\n");
-
-    // reset the event buffer, discard all content
-    freebob_ringbuffer_reset(m_event_buffer);
-    
-    // reset the statistics
-    m_PeriodStat.reset();
@@ -566 +566 @@
-        m_syt_interval);
-
+    // prepare the framerate estimate
+    m_ticks_per_frame = (TICKS_PER_SECOND*1.0) / ((float)m_framerate);
+    
-    // allocate the event buffer
-    m_ringbuffer_size_frames=m_nb_buffers * m_period;
-
-
-
-
+
-    // add the receive processing delay
-//     m_ringbuffer_size_frames+=(uint)(RECEIVE_PROCESSING_DELAY/m_ticks_per_frame);
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
-
-    // set the parameters of ports we can:
@@ -685 +682 @@
-    
-    // prefill the event buffer
-
-
-
-
+
+
+
+
+
-    
-    debugOutput( DEBUG_LEVEL_VERBOSE, "Prepared for:\n");
@@ -724 +722 @@
-}
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
-        debugWarning("Could not write to event buffer\n");
-
-
+
+
+
-    free(dummybuffer);
-    
-true
+retval
-}
-
-bool AmdtpTransmitStreamProcessor::canClientTransferFrames(unsigned int nbframes) {
-    // there has to be enough space to put the frames in
-
+m_data_buffer->
-}
-
-bool AmdtpTransmitStreamProcessor::putFrames(unsigned int nbframes, int64_t ts) {
-
-
-
-
-
+
-    
-    debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "AmdtpTransmitStreamProcessor::putFrames(%d, %llu)\n",nbframes, ts);
-    
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
-    
-    // recalculate the buffer tail timestamp
@@ -878 +795 @@
- */
-
-int AmdtpTransmitStreamProcessor::transmit
+bool AmdtpTransmitStreamProcessor::processWrite
-                       unsigned int nevents, unsigned int offset)
-{
-int problem=0
+bool no_problem=true
-
-    for ( PortVectorIterator it = m_PeriodPorts.begin();
@@ -899 +816 @@
-            if(encodePortToMBLAEvents(static_cast<AmdtpAudioPort *>(*it), (quadlet_t *)data, offset, nevents)) {
-                debugWarning("Could not encode port %s to MBLA events",(*it)->getName().c_str());
-problem=1
+no_problem=false
-            }
-            break;
@@ -908 +825 @@
-        }
-    }
-
+no_
-
-}
@@ -1090 +1007 @@
-    : ReceiveStreamProcessor(port, framerate), m_dimension(dimension), m_last_timestamp(0), m_last_timestamp2(0) {
-
-
-}
-
-AmdtpReceiveStreamProcessor::~AmdtpReceiveStreamProcessor() {
-    freebob_ringbuffer_free(m_event_buffer);
-    free(m_cluster_buffer);
-
-}
-
-bool AmdtpReceiveStreamProcessor::init() {
+
-    // call the parent init
-    // this has to be done before allocating the buffers, 
@@ -1117 +1032 @@
-    
-    enum raw1394_iso_disposition retval=RAW1394_ISO_OK;
+    m_last_cycle=cycle;
-    
-    struct iec61883_packet *packet = (struct iec61883_packet *) data;
@@ -1144 +1060 @@
-            channel, cycle,syt_timestamp,  
-            CYCLE_TIMER_GET_CYCLES(syt_timestamp), CYCLE_TIMER_GET_OFFSET(syt_timestamp),
-
+m_data_buffer->
-        
-        // reconstruct the full cycle
@@ -1199 +1115 @@
-        // we have to keep in mind that there are also
-        // some packets buffered by the ISO layer
-NbBuffers
+WakeupInterval
-        // these contain at most x*syt_interval
-        // frames, meaning that we might receive
-        // this packet x*syt_interval*ticks_per_frame
-        // later than expected (the real receive time)
-
-
+
+
+
+
+
+
-        
-        // the receive processing delay indicates how much
-        // extra time we need as slack
-        m_last_timestamp += RECEIVE_PROCESSING_DELAY;
-
+
+
+
+
+
+
-        //=> now estimate the device frame rate
-        if (m_last_timestamp2 && m_last_timestamp) {
@@ -1271 +1196 @@
-                m_is_disabled=false;
-                debugOutput(DEBUG_LEVEL_VERBOSE,"enabling StreamProcessor %p at %d\n", this, cycle);
+                // the previous timestamp is the one we need to start with
+                // because we're going to update the buffer again this loop
+                m_data_buffer->setBufferTailTimestamp(m_last_timestamp2);
+                
-            } else {
-                debugOutput(DEBUG_LEVEL_VERY_VERBOSE,"will enable StreamProcessor %p at %u, now is %d\n", this, m_cycle_to_enable_at, cycle);
@@ -1296 +1225 @@
-            }
-            // set the timestamps
-StreamProcessor::setBufferTimestamps(ts,
+m_data_buffer->setBufferTailTimestamp(
-            
-            return RAW1394_ISO_DEFER;
@@ -1302 +1231 @@
-        
-        //=> process the packet
-        unsigned int write_size=nevents*sizeof(quadlet_t)*m_dimension;
-        
-        // add the data payload to the ringbuffer
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
-            retval=RAW1394_ISO_OK;
+            
-            // process all ports that should be handled on a per-packet base
-            // this is MIDI for AMDTP (due to the need of DBC)
@@ -1326 +1241 @@
-                retval=RAW1394_ISO_DEFER;
-            }
+            
+        } else {
+        
+            debugWarning("Receive buffer overrun (cycle %d, FC=%d, PC=%d)\n", 
+                 cycle, m_data_buffer->getFrameCounter(), m_handler->getPacketCount());
+            
+            m_xruns++;
+            
+            // disable the processing, will be re-enabled when
+            // the xrun is handled
+            m_disabled=true;
+            m_is_disabled=true;
+
+            retval=RAW1394_ISO_DEFER;
+            
-        }
-
@@ -1434 +1364 @@
-    // currently this is in ticks
-    
-
+m_data_buffer->
-    
-    int64_t next_period_boundary =  m_last_timestamp;
@@ -1483 +1413 @@
-}
-
-
-bool AmdtpReceiveStreamProcessor::reset() {
-
-    debugOutput( DEBUG_LEVEL_VERBOSE, "Resetting...\n");
-
-    // reset the event buffer, discard all content
-    freebob_ringbuffer_reset(m_event_buffer);
-    
-    m_PeriodStat.reset();
-    m_PacketStat.reset();
@@ -1562 +1488 @@
-    ringbuffer_size_frames+=(uint)(RECEIVE_PROCESSING_DELAY/m_ticks_per_frame);
-    
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
-
-        // set the parameters of ports we can:
@@ -1671 +1595 @@
-
-bool AmdtpReceiveStreamProcessor::canClientTransferFrames(unsigned int nbframes) {
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
-        
-    // update the frame counter such that it reflects the new value,
-    // and also update the buffer head timestamp as we pull frames
-    // done in the SP base class
-
-
-
-
-
-
-
-
-
-
+
+
+
-        return false;
-    }
@@ -1780 +1621 @@
- * \brief write received events to the stream ringbuffers.
- */
-int AmdtpReceiveStreamProcessor::receive
+bool AmdtpReceiveStreamProcessor::processRead
-                                           unsigned int nevents, unsigned int offset)
-{
-
+
+
+
-
-        for ( PortVectorIterator it = m_PeriodPorts.begin();
@@ -1801 +1644 @@
-                        if(decodeMBLAEventsToPort(static_cast<AmdtpAudioPort *>(*it), (quadlet_t *)data, offset, nevents)) {
-                                debugWarning("Could not decode packet MBLA to port %s",(*it)->getName().c_str());
-problem=1
+no_problem=false
-                        }
-                        break;
@@ -1813 +1656 @@
-                }
-    }
-       return 
+    return no_
-
-}

branches/streaming-rework/src/libstreaming/AmdtpStreamProcessor.h

@@ -35 +35 @@
-#include "../debugmodule/debugmodule.h"
-#include "StreamProcessor.h"
+
-#include "cip.h"
-#include <libiec61883/iec61883.h>
-#include "libutil/ringbuffer.h"
-#include <pthread.h>
-
@@ -117 +117 @@
-    
-protected:
+    bool processWriteBlock(char *data, unsigned int nevents, unsigned int offset);
-
-    struct iec61883_cip m_cip_status;
-
-
-
+
-    int m_dimension;
-    unsigned int m_syt_interval;
@@ -184 +183 @@
-        
-    bool canClientTransferFrames(unsigned int nbframes);
-, int64_t ts
+
-
-    // We have 1 period of samples = m_period
@@ -208 +207 @@
-protected:
-
-
+
+
-    bool decodePacketPorts(quadlet_t *data, unsigned int nevents, unsigned int dbc);
-    
-    int decodeMBLAEventsToPort(AmdtpAudioPort *, quadlet_t *data, unsigned int offset, unsigned int nevents);
-
-    freebob_ringbuffer_t * m_event_buffer;
-    char* m_cluster_buffer;
-    int m_dimension;
-    unsigned int m_syt_interval;

branches/streaming-rework/src/libstreaming/IsoHandler.cpp

@@ -134 +134 @@
-    
-    if (m_TimeSource) delete m_TimeSource;
+}
+
+bool IsoHandler::iterate() {
+    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "IsoHandler (%p) iterate...\n",this);
+
+    if(m_handle) {
+        if(raw1394_loop_iterate(m_handle)) {
+            debugOutput( DEBUG_LEVEL_VERBOSE, 
+                 "IsoHandler (%p): Failed to iterate handler: %s\n",
+                 this,strerror(errno));
+            return false;
+        } else {
+            return true;
+        }
+    } else {
+        return false; 
+    }
-}
-

branches/streaming-rework/src/libstreaming/IsoHandler.h

@@ -72 +72 @@
-        virtual bool stop();
-        
-int iterate() { if(m_handle) return raw1394_loop_iterate(m_handle); else return -1; }
+bool iterate()
-        
-        void setVerboseLevel(int l);

branches/streaming-rework/src/libstreaming/IsoHandlerManager.cpp

@@ -45 +45 @@
-IsoHandlerManager::IsoHandlerManager() :
-   m_State(E_Created),
-
+00
-   m_realtime(false), m_priority(0)
-{
@@ -323 +323 @@
-                unsigned int packets_per_period=stream->getPacketsPerPeriod();
-                
-1
+0
-                // hardware interrupts occur when one DMA block is full, and the size of one DMA
-                // block = PAGE_SIZE. Setting the max_packet_size makes sure that the HW irq is 

branches/streaming-rework/src/libstreaming/StreamProcessor.cpp

@@ -52 +52 @@
-        , m_is_disabled(true)
-        , m_cycle_to_enable_at(0)
-        , m_framecounter(0)
-        , m_SyncSource(NULL)
-        , m_ticks_per_frame(0)
-{
+    // create the timestamped buffer and register ourselves as its client
+    m_data_buffer=new FreebobUtil::TimestampedBuffer(this);
-
-}
-
-StreamProcessor::~StreamProcessor() {
-
+
-}
-
-void StreamProcessor::dumpInfo()
-{
-    int64_t diff=(int64_t)m_buffer_head_timestamp - (int64_t)m_buffer_tail_timestamp;
-
-    debugOutputShort( DEBUG_LEVEL_NORMAL, " StreamProcessor information\n");
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Iso stream info:\n");
@@ -72 +71 @@
-    IsoStream::dumpInfo();
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  StreamProcessor info:\n");
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Frame counter         : %d\n", m_framecounter);
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Buffer head timestamp : %011llu\n",m_buffer_head_timestamp);
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Buffer tail timestamp : %011llu\n",m_buffer_tail_timestamp);
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Head - Tail           : %011lld\n",diff);
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Now                   : %011u\n",m_handler->getCycleTimerTicks());
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Xruns                 : %d\n", m_xruns);
@@ -82 +77 @@
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "   enable status        : %s\n", m_is_disabled ? "No" : "Yes");
-    
+    m_data_buffer->dumpInfo();
+    
-//     m_PeriodStat.dumpInfo();
-//     m_PacketStat.dumpInfo();
@@ -92 +89 @@
-    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "enter...\n");
-    
-pthread_mutex_init(&m_framecounter_lock, NULL
-
+m_data_buffer->init(
+
-    return IsoStream::init();
-}
@@ -105 +102 @@
-    debugOutput( DEBUG_LEVEL_VERBOSE, "Resetting...\n");
-
-
-
+
+
+
+
+
+
-    resetXrunCounter();
-
@@ -125 +126 @@
-    
-bool StreamProcessor::prepareForEnable() {
-    int64_t diff=(int64_t)m_buffer_head_timestamp - (int64_t)m_buffer_tail_timestamp;
-    
-    debugOutput(DEBUG_LEVEL_VERBOSE," StreamProcessor::prepareForEnable for (%p)\n",this);
-    debugOutput(DEBUG_LEVEL_VERBOSE," Frame Counter         : %05d\n",m_framecounter);
-    debugOutput(DEBUG_LEVEL_VERBOSE," Buffer head timestamp : %011llu\n",m_buffer_head_timestamp);
-    debugOutput(DEBUG_LEVEL_VERBOSE," Buffer tail timestamp : %011llu\n",m_buffer_tail_timestamp);
-    debugOutput(DEBUG_LEVEL_VERBOSE," Head - Tail           : %011lld\n",diff);
-    debugOutput(DEBUG_LEVEL_VERBOSE," Now                   : %011u\n",m_handler->getCycleTimerTicks());
+    m_data_buffer->dumpInfo();
-    return true;
-}
-
-bool StreamProcessor::prepareForDisable() {
-    int64_t diff=(int64_t)m_buffer_head_timestamp - (int64_t)m_buffer_tail_timestamp;
-    
-    debugOutput(DEBUG_LEVEL_VERBOSE," StreamProcessor::prepareForDisable for (%p)\n",this);
-    debugOutput(DEBUG_LEVEL_VERBOSE," Frame Counter         : %05d\n",m_framecounter);
-    debugOutput(DEBUG_LEVEL_VERBOSE," Buffer head timestamp : %011llu\n",m_buffer_head_timestamp);
-    debugOutput(DEBUG_LEVEL_VERBOSE," Buffer tail timestamp : %011llu\n",m_buffer_tail_timestamp);
-    debugOutput(DEBUG_LEVEL_VERBOSE," Head - Tail           : %011lld\n",diff);
-    debugOutput(DEBUG_LEVEL_VERBOSE," Now                   : %011u\n",m_handler->getCycleTimerTicks());
-return true
-
+m_data_buffer->dumpInfo()
+
-}
-
@@ -152 +142 @@
-
-        debugOutput( DEBUG_LEVEL_VERBOSE, "Preparing...\n");
-
-
+
-        // init the ports
-        
@@ -191 +180 @@
-
-        debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "Putting %d frames for %llu into frame buffer...\n", nbframes,ts);
-
+m_data_buffer->
-        return true;
-}
@@ -202 +191 @@
- *
- * @param nbframes the number of frames that are read from the internal buffers
- * @param ts the new timestamp of the 'head' of the buffer, i.e. the first sample
- *           present in the buffer.
- * @return true if successful
- */
-, int64_t ts
-
- at (%011lld)...\n", nbframes, t
-decrementFrameCounter(nbframes, t
+
+
+...\n", nbframe
+m_data_buffer->decrementFrameCounter(nbframe
-        return true;
-}
@@ -266 +253 @@
-
-/**
- * Decrements the frame counter, in a atomic way. This
- * also sets the buffer head timestamp
- * is thread safe.
- */
-void StreamProcessor::decrementFrameCounter(int nbframes, uint64_t new_timestamp) {
-    debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "Setting buffer head timestamp for (%p) to %11llu\n",
-                this, new_timestamp);
-                
-    pthread_mutex_lock(&m_framecounter_lock);
-    m_framecounter -= nbframes;
-    m_buffer_head_timestamp = new_timestamp;
-    pthread_mutex_unlock(&m_framecounter_lock);
-}
-
-/**
- * Increments the frame counter, in a atomic way.
- * also sets the buffer tail timestamp
- * This is thread safe.
- */
-void StreamProcessor::incrementFrameCounter(int nbframes, uint64_t new_timestamp) {
-    debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "Setting buffer tail timestamp for (%p) to %11llu\n",
-                this, new_timestamp);
-    
-    pthread_mutex_lock(&m_framecounter_lock);
-    m_framecounter += nbframes;
-    m_buffer_tail_timestamp = new_timestamp;
-    pthread_mutex_unlock(&m_framecounter_lock);
-    
-}
-
-/**
- * Sets the buffer tail timestamp (in usecs)
- * This is thread safe.
- */
-void StreamProcessor::setBufferTailTimestamp(uint64_t new_timestamp) {
-    debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "Setting buffer tail timestamp for (%p) to %11llu\n",
-                this, new_timestamp);
-    
-    pthread_mutex_lock(&m_framecounter_lock);
-    m_buffer_tail_timestamp = new_timestamp;
-    pthread_mutex_unlock(&m_framecounter_lock);
-}
-
-/**
- * Sets the buffer head timestamp (in usecs)
- * This is thread safe.
- */
-void StreamProcessor::setBufferHeadTimestamp(uint64_t new_timestamp) {
-    debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "Setting buffer head timestamp for (%p) to %11llu\n",
-                this, new_timestamp);
-
-    pthread_mutex_lock(&m_framecounter_lock);
-    m_buffer_head_timestamp = new_timestamp;
-    pthread_mutex_unlock(&m_framecounter_lock);
-}
-
-/**
- * Sets both the buffer head and tail timestamps (in usecs)
- * (avoids multiple mutex lock/unlock's)
- * This is thread safe.
- */
-void StreamProcessor::setBufferTimestamps(uint64_t new_head, uint64_t new_tail) {
-    debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "Setting buffer head timestamp for (%p) to %11llu\n",
-                this, new_head);
-    debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "    and buffer tail timestamp for (%p) to %11llu\n",
-                this, new_tail);
-   
-    pthread_mutex_lock(&m_framecounter_lock);
-    m_buffer_head_timestamp = new_head;
-    m_buffer_tail_timestamp = new_tail;
-    pthread_mutex_unlock(&m_framecounter_lock);
-}
-/**
- * \brief return the timestamp of the first frame in the buffer
- * 
- * This function returns the timestamp of the very first sample in
- * the StreamProcessor's buffer. This is useful for slave StreamProcessors 
- * to find out what the base for their timestamp generation should
- * be. It also returns the framecounter value for which this timestamp
- * is valid.
- *
- * The system is built in such a way that we assume that the processing
- * of the buffers doesn't take any time. Assume we have a buffer transfer at 
- * time T1, meaning that the last sample of this buffer occurs at T1. As 
- * processing does not take time, we don't have to add anything to T1. When
- * transferring the processed buffer to the xmit processor, the timestamp
- * of the last sample is still T1.
- *
- * When starting the streams, we don't have any information on this last
- * timestamp. We prefill the buffer at the xmit side, and we should find
- * out what the timestamp for the last sample in the buffer is. If we sync
- * on a receive SP, we know that the last prefilled sample corresponds with
- * the first sample received - 1 sample duration. This is the same as if the last
- * transfer from iso to client would have emptied the receive buffer.
- *
- *
- * @param ts address to store the timestamp in
- * @param fc address to store the associated framecounter in
- */
-void StreamProcessor::getBufferHeadTimestamp(uint64_t *ts, uint64_t *fc) {
-    pthread_mutex_lock(&m_framecounter_lock);
-    *fc = m_framecounter;
-    *ts = m_buffer_head_timestamp;
-    pthread_mutex_unlock(&m_framecounter_lock);
-}
-        
-/**
- * \brief return the timestamp of the last frame in the buffer
- * 
- * This function returns the timestamp of the last frame in
- * the StreamProcessor's buffer. It also returns the framecounter 
- * value for which this timestamp is valid.
- *
- * @param ts address to store the timestamp in
- * @param fc address to store the associated framecounter in
- */
-void StreamProcessor::getBufferTailTimestamp(uint64_t *ts, uint64_t *fc) {
-    pthread_mutex_lock(&m_framecounter_lock);
-    *fc = m_framecounter;
-    *ts = m_buffer_tail_timestamp;
-    pthread_mutex_unlock(&m_framecounter_lock);
-}
-
-/**
- * Resets the frame counter, in a atomic way. This
- * is thread safe.
- */
-void StreamProcessor::resetFrameCounter() {
-    pthread_mutex_lock(&m_framecounter_lock);
-    m_framecounter = 0;
-    pthread_mutex_unlock(&m_framecounter_lock);
-}
-
-/**
- * Resets the xrun counter, in a atomic way. This
- * is thread safe.

branches/streaming-rework/src/libstreaming/StreamProcessor.h

@@ -37 +37 @@
-
-#include "libutil/StreamStatistics.h"
+
+#include "libutil/TimestampedBuffer.h"
-
-namespace FreebobStreaming {
@@ -51 +53 @@
-*/
-class StreamProcessor : public IsoStream, 
-
+
+
-
-    friend class StreamProcessorManager;
@@ -86 +89 @@
-
-    virtual bool putFrames(unsigned int nbframes, int64_t ts); ///< transfer the buffer contents from client
-, int64_t ts
+
-
-    virtual bool reset(); ///< reset the streams & buffers (e.g. after xrun)
@@ -104 +107 @@
-    virtual bool prepareForDisable();
-
+public:
+    FreebobUtil::TimestampedBuffer *m_data_buffer;
+
-protected:
-        
-
-    void setManager(StreamProcessorManager *manager) {m_manager=manager;};
@@ -145 +150 @@
-         */
-        virtual bool canClientTransferFrames(unsigned int nframes) {return true;};
-        
-        int getFrameCounter() {return m_framecounter;};
-    
-        void decrementFrameCounter(int nbframes, uint64_t new_timestamp);
-        void incrementFrameCounter(int nbframes, uint64_t new_timestamp);
-        void resetFrameCounter();
-        
-        /**
@@ -188 +187 @@
-        uint64_t getTimeNow();
-        
-        void getBufferHeadTimestamp(uint64_t *ts, uint64_t *fc);
-        void getBufferTailTimestamp(uint64_t *ts, uint64_t *fc);
-        
-        void setBufferTailTimestamp(uint64_t new_timestamp);
-        void setBufferHeadTimestamp(uint64_t new_timestamp);
-        void setBufferTimestamps(uint64_t new_head, uint64_t new_tail);
-        
-        bool setSyncSource(StreamProcessor *s);
-        float getTicksPerFrame() {return m_ticks_per_frame;};
-        
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
-        
-    protected:
@@ -217 +198 @@
-        float m_ticks_per_frame;
-        
-
-
-
-
-
-
+
-
-};
@@ -252 +228 @@
-
-protected:
-
-
+
+
+
-
-};
@@ -282 +259 @@
-
-protected:
-
-
+
+
+
-
-

branches/streaming-rework/src/libstreaming/StreamProcessorManager.cpp

@@ -392 +392 @@
-    
-    // FIXME: this should not be in cycles, but in 'time'
-
-
+
+
+
-    debugOutput( DEBUG_LEVEL_VERBOSE, " Sync Source StreamProcessor...\n");
-    if (!m_SyncSource->prepareForEnable()) {
@@ -835 +836 @@
-        if ((*it)->xrunOccurred()) {
-            debugWarning("Xrun on RECV SP %p due to ISO xrun\n",*it);
+            (*it)->dumpInfo();
-        }
-        if (!((*it)->canClientTransferFrames(m_period))) {
-            debugWarning("Xrun on RECV SP %p due to buffer xrun\n",*it);
+            (*it)->dumpInfo();
-        }
-#endif
@@ -927 +930 @@
-            #endif
-    
-, (int64_t)m_time_of_transfer
+
-                    debugOutput(DEBUG_LEVEL_VERBOSE,"could not getFrames(%u, %11llu) from stream processor (%p)",
-                            m_period, m_time_of_transfer,*it);

branches/streaming-rework/src/Makefile.am

@@ -46 +46 @@
-        libutil/ringbuffer.h libutil/PacketBuffer.h libutil/StreamStatistics.h \
-        libutil/serialize.h libutil/SystemTimeSource.h libutil/Thread.h libutil/Time.h \
-
+ libutil/TimestampedBuffer.h
-
-libfreebob_la_SOURCES = \
@@ -105 +105 @@
-        libutil/SystemTimeSource.cpp \
-        libutil/Time.c \
-
+
+
-
-libfreebob_la_LDFLAGS =                                 \