Changeset 720

Timestamp:

11/23/07 08:11:41 (14 years ago)

Author:

ppalmers

Message:

first working version of the reworked streaming code

Files:

• branches/ppalmers-streaming/src/debugmodule/debugmodule.cpp (modified) (1 diff)
• branches/ppalmers-streaming/src/genericavc/avc_avdevice.cpp (modified) (2 diffs)
• branches/ppalmers-streaming/src/libstreaming/amdtp/AmdtpReceiveStreamProcessor.cpp (modified) (2 diffs)
• branches/ppalmers-streaming/src/libstreaming/amdtp/AmdtpReceiveStreamProcessor.h (modified) (1 diff)
• branches/ppalmers-streaming/src/libstreaming/amdtp/AmdtpTransmitStreamProcessor.cpp (modified) (14 diffs)
• branches/ppalmers-streaming/src/libstreaming/amdtp/AmdtpTransmitStreamProcessor.h (modified) (1 diff)
• branches/ppalmers-streaming/src/libstreaming/generic/StreamProcessor.cpp (modified) (30 diffs)
• branches/ppalmers-streaming/src/libstreaming/generic/StreamProcessor.h (modified) (10 diffs)
• branches/ppalmers-streaming/src/libstreaming/StreamProcessorManager.cpp (modified) (8 diffs)
• branches/ppalmers-streaming/src/libstreaming/StreamProcessorManager.h (modified) (2 diffs)
• branches/ppalmers-streaming/src/libutil/TimestampedBuffer.cpp (modified) (8 diffs)
• branches/ppalmers-streaming/src/libutil/TimestampedBuffer.h (modified) (1 diff)

branches/ppalmers-streaming/src/debugmodule/debugmodule.cpp

@@ -35 +35 @@
-
-#ifndef DO_MESSAGE_BUFFER_PRINT
-       
+    
-#endif
-

branches/ppalmers-streaming/src/genericavc/avc_avdevice.cpp

@@ -422 +422 @@
-    }
-
-    int samplerate=outputPlug->getSampleRate();
-
-    debugOutput( DEBUG_LEVEL_VERBOSE, "Initializing receive processor...\n");
-    // create & add streamprocessors
@@ -592 +590 @@
-int
-AvDevice::getStreamCount() {
-//
-
+
+//
-}
-

branches/ppalmers-streaming/src/libstreaming/amdtp/AmdtpReceiveStreamProcessor.cpp

@@ -55 +55 @@
-
-bool AmdtpReceiveStreamProcessor::prepareChild() {
-
-    m_PeriodStat.setName("RCV PERIOD");
-    m_PacketStat.setName("RCV PACKET");
-    m_WakeupStat.setName("RCV WAKEUP");
-
-    debugOutput( DEBUG_LEVEL_VERBOSE, "Preparing (%p)...\n", this);
-
@@ -153 +148 @@
-    // this packet x*syt_interval*ticks_per_frame
-    // later than expected (the real receive time)
-
-
+
+
+
+
+
+
-
-    if(m_data_buffer->writeFrames(nevents, (char *)(data+8), m_last_timestamp)) {

branches/ppalmers-streaming/src/libstreaming/amdtp/AmdtpReceiveStreamProcessor.h

@@ -94 +94 @@
-    virtual unsigned int getEventsPerFrame() 
-                    { return m_dimension; };
-UpdatePeriod
+NominalFramesPerPacket
-                    {return m_syt_interval;};
-
-    // We have 1 period of samples = m_period
-    // this period takes m_period/m_framerate seconds of time
-    // during this time, 8000 packets are sent
-//     unsigned int getPacketsPerPeriod() {return (m_period*8000)/m_framerate;};
-
-    // however, if we only count the number of used packets
-    // it is m_period / m_syt_interval
-    virtual unsigned int getPacketsPerPeriod();
-
-protected:
-
-    bool processReadBlock(char *data, unsigned int nevents, unsigned int offset);
-    bool provideSilenceBlock(unsigned int nevents, unsigned int offset);
-
+private:
-    bool decodePacketPorts(quadlet_t *data, unsigned int nevents, unsigned int dbc);
-

branches/ppalmers-streaming/src/libstreaming/amdtp/AmdtpTransmitStreamProcessor.cpp

@@ -38 +38 @@
-#define TRANSMIT_TRANSFER_DELAY DEFAULT_TRANSFER_DELAY
-
-
+
+
-
-/* transmit */
-
-
-
-
-
-
+
+
+
+
-{}
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
-    /* Our node ID can change after a bus reset, so it is best to fetch
- 
+
-    packet->sid = m_handler->getLocalNodeId() & 0x3f;
-
@@ -96 +70 @@
-    *tag = IEC61883_TAG_WITH_CIP;
-    *sy = 0;
-
-    // determine if we want to send a packet or not
-    // note that we can't use getCycleTimer directly here,
-    // because packets are queued in advance. This means that
-    // we the packet we are constructing will be sent out
-    // on 'cycle', not 'now'.
-    unsigned int ctr=m_handler->getCycleTimer();
-    int now_cycles = (int)CYCLE_TIMER_GET_CYCLES(ctr);
-
-    // the difference between the cycle this
-    // packet is intended for and 'now'
-    int cycle_diff = diffCycles(cycle, now_cycles);
-
-#ifdef DEBUG
-    if(isRunning() && (cycle_diff < 0)) {
-        debugWarning("Requesting packet for cycle %04d which is in the past (now=%04dcy)\n",
-            cycle, now_cycles);
-    }
-
-    // keep track of the lag
-    m_PacketStat.mark(cycle_diff);
-#endif
-
-    // as long as the cycle parameter is not in sync with
-    // the current time, the stream is considered not
-    // to be 'running'
-    // NOTE: this works only at startup
-    if (!isRunning() && cycle_diff >= 0 && cycle >= 0) {
-            debugOutput(DEBUG_LEVEL_VERBOSE, "Xmit StreamProcessor %p started running at cycle %d\n",this, cycle);
-    }
-
-    signed int fc;
@@ -152 +96 @@
-    const int max_cycles_to_transmit_early = 5;
-
-    if( !isRunning() || !m_data_buffer->isEnabled() ) {
-        debugOutput(DEBUG_LEVEL_ULTRA_VERBOSE,
-                    "Not running (%d) or buffer not enabled (enabled=%d)\n",
-                    isRunning(), m_data_buffer->isEnabled());
-
-        // not running or not enabled
-        goto send_empty_packet;
-    }
-
-try_block_of_frames:
-(DEBUG_LEVEL_ULTRA_VERBOSE, "Try for cycle %d\n", cycle
+ ( DEBUG_LEVEL_ULTRA_VERBOSE, "Try for cycle %d\n", cycle 
-    // check whether the packet buffer has packets for us to send.
-    // the base timestamp is the one of the next sample in the buffer
-    ffado_timestamp_t ts_head_tmp;
-(&ts_head_tmp, &fc
+ ( &ts_head_tmp, &fc 
-
-    // the timestamp gives us the time at which we want the sample block
-    // to be output by the device
-
+
+
-
-    // now we calculate the time when we have to transmit the sample block
-(presentation_time, TRANSMIT_TRANSFER_DELAY
+ ( presentation_time, TRANSMIT_TRANSFER_DELAY 
-
-    // calculate the cycle this block should be presented in
-    // (this is just a virtual calculation since at that time it should
-    //  already be in the device's buffer)
-unsigned int)(TICKS_TO_CYCLES( presentation_time )
+ unsigned int ) ( TICKS_TO_CYCLES ( presentation_time ) 
-
-    // calculate the cycle this block should be transmitted in
-unsigned int)(TICKS_TO_CYCLES( transmit_at_time )
+ unsigned int ) ( TICKS_TO_CYCLES ( transmit_at_time ) 
-
-    // we can check whether this cycle is within the 'window' we have
-    // to send this packet.
-    // first calculate the number of cycles left before presentation time
-
+ 
-
-    // we can check whether this cycle is within the 'window' we have
-    // to send this packet.
-    // first calculate the number of cycles left before presentation time
-
+
+
+
+
+
+
+
+
-
-    // two different options:
- 
+
-    //      => determine wether this is a problem, since we might still
-    //         have some time to send it
-    // 2) there are enough packets
-    //      => determine whether we have to send them in this packet
-fc < (signed int)m_syt_interval) {
-    m_PacketStat.signal(0);
+ fc < ( signed int ) m_syt_interval )
+{
-        // not enough frames in the buffer,
-(DEBUG_LEVEL_VERBOSE, 
+ ( DEBUG_LEVEL_VERBOSE,
-                    "Insufficient frames: N=%02d, CY=%04u, TC=%04u, CUT=%04d\n",
-
+ 
-        // we can still postpone the queueing of the packets
-        // if we are far enough ahead of the presentation time
-( cycles_until_presentation <= min_cycles_before_presentation ) {
-    m_PacketStat.signal(1);
+ ( cycles_until_presentation <= min_cycles_before_presentation )
+{
-            // we are too late
-            // meaning that we in some sort of xrun state
@@ -214 +157 @@
-            m_xruns++;
-            // we send an empty packet on this cycle
-
-
-
+
+
+
+
-            // there is still time left to send the packet
-            // we want the system to give this packet another go
-
+    
-            // unfortunatly the try_again doesn't work very well,
-            // so we'll have to either usleep(one cycle) and goto try_block_of_frames
-
+
-            // or just fill this with an empty packet
-            // if we have to do this too often, the presentation time will
-            // get too close and we're in trouble
-
-
-
-
+
+
+
+
+
-        // there are enough frames, so check the time they are intended for
-        // all frames have a certain 'time window' in which they can be sent
@@ -236 +181 @@
-        // in theory we can send the packet up till one cycle before the presentation time,
-        // however this is not very smart.
-
+
-        // There are 3 options:
-        // 1) the frame block is too early
@@ -245 +190 @@
-        //      => discard (and raise xrun?)
-        //         get next block of frames and repeat
-
-cycles_until_transmit <= max_cycles_to_transmit_early) {
-    m_PacketStat.signal(4);
+
+ cycles_until_transmit <= max_cycles_to_transmit_early )
+{
-            // it's time send the packet
-
-
+
+
+
+
+
-            // we are too late
-(DEBUG_LEVEL_VERBOSE, 
+ ( DEBUG_LEVEL_VERBOSE,
-                        "Too late: CY=%04u, TC=%04u, CUT=%04d, TSP=%011llu (%04u)\n",
-                        cycle,
-                        transmit_at_cycle, cycles_until_transmit,
-unsigned int)TICKS_TO_CYCLES(presentation_time)
+ unsigned int ) TICKS_TO_CYCLES ( presentation_time ) 
-
-            // however, if we can send this sufficiently before the presentation
@@ -262 +210 @@
-            // NOTE: dangerous since the device has no way of reporting that it didn't get
-            //       this packet on time.
- {
-    m_PacketStat.signal(5);
+
+{
-                // we are not that late and can still try to transmit the packet
-
-
-
+
+
+
+
+
-                // remove the samples
-(m_syt_interval
+ ( m_syt_interval 
-                // signal some xrun situation ??HERE??
-                m_xruns++;
@@ -275 +225 @@
-                goto try_block_of_frames; // UGLY but effective
-            }
-
-
-
+
+
+
+
-                        "Too early: CY=%04u, TC=%04u, CUT=%04d, TST=%011llu (%04u), TSP=%011llu (%04u)\n",
-                        cycle,
-                        transmit_at_cycle, cycles_until_transmit,
-
-
-
-
-
+
+
+
+
+
+
-                            "Way too early: CY=%04u, TC=%04u, CUT=%04d, TST=%011llu (%04u), TSP=%011llu (%04u)\n",
-                            cycle,
-                            transmit_at_cycle, cycles_until_transmit,
-unsigned int)TICKS_TO_CYCLES(transmit_at_time
-unsigned int)TICKS_TO_CYCLES(presentation_time)
-
-            
+ unsigned int ) TICKS_TO_CYCLES ( transmit_at_time 
+ unsigned int ) TICKS_TO_CYCLES ( presentation_time ) 
+
+
-            // we are too early, send only an empty packet
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-        // process all ports that should be handled on a per-packet base
-        // this is MIDI for AMDTP (due to the need of DBC)
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-    return m_syt_interval;
-}
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
-
-    // no-data packets have syt=0xFFFF
@@ -378 +343 @@
-    // FIXME: either make this a setting or choose
-    bool send_payload=true;
-
+
+
-        // this means no-data packets with payload (DICE doesn't like that)
-(quadlet_t) + m_syt_interval * m_dimension * sizeof(quadlet_t
+ ( quadlet_t ) + m_syt_interval * m_dimension * sizeof ( quadlet_t 
-        return m_syt_interval;
-
+
+
+
-        // dbc is not incremented
-        // this means no-data packets without payload
-(quadlet_t
+ ( quadlet_t 
-        return 0;
-    }
-}
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-    }
-
@@ -481 +406 @@
-        m_manager->getNominalRate(),
-        m_dimension,
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
-    for ( PortVectorIterator it = m_Ports.begin();
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-                }
-                break;
-
-
+
-                break;
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
-    for ( PortVectorIterator it = m_PeriodPorts.begin();
-
-
-
-
-
-
+
+
+
-        //FIXME: make this into a static_cast when not DEBUG?
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-        }
-    }
-    return problem;
-
-}
-
-/**
- 
- 
- 
- 
- 
- 
- 
-(quadlet_t *data, unsigned int nevents, unsigned int dbc
+
+
+
+
+
+
+
+ ( quadlet_t *data, unsigned int nevents, unsigned int dbc 
-{
-    bool ok=true;
@@ -781 +530 @@
-
-    for ( PortVectorIterator it = m_PacketPorts.begin();
-
-
+  
+  
-    {
-
-#ifdef DEBUG
-(*it
-(pinfo
+ ( *it 
+ ( pinfo 
-
-        // the only packet type of events for AMDTP is MIDI in mbla
-(pinfo->getFormat()==AmdtpPortInfo::E_Midi
+ ( pinfo->getFormat() ==AmdtpPortInfo::E_Midi 
-#endif
-
-(*it
+ ( *it 
-
-        // we encode this directly (no function call) due to the high frequency
@@ -806 +555 @@
-        // first prefill the buffer with NO_DATA's on all time muxed channels
-
-
-
+
+
+
-            quadlet_t tmpval;
-
-
-
-
-
+
+
+
+
+
+
-                byte &= 0xFF;
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
-                // can't send a byte, either because there is no byte,
-                // or because this would exceed the maximum rate
-
-IEC61883_AM824_SET_LABEL(0,IEC61883_AM824_LABEL_MIDI_NO_DATA)
-
-
+ 
+    IEC61883_AM824_SET_LABEL ( 0,IEC61883_AM824_LABEL_MIDI_NO_DATA ) 
+
+
-            *target_event=tmpval;
-        }
-
-    }
-
-    return ok;
-}
-
-
-(
-               unsigned int offset, unsigned int nevents
+ ( 
+unsigned int offset, unsigned int nevents 
-{
-    unsigned int j=0;
@@ -847 +599 @@
-    quadlet_t *target_event;
-
-
-
-
+
+
+
+
-        default:
-        case Port::E_Int24:
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-        case Port::E_Float:
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-    }
-
-    return 0;
-}
-(
-               unsigned int offset, unsigned int nevents
+ ( 
+unsigned int offset, unsigned int nevents 
-{
-    unsigned int j=0;
@@ -898 +653 @@
-    quadlet_t *target_event;
-
-
-
-
+
+
+
+
-        default:
-        case Port::E_Int24:
-        case Port::E_Float:
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
-    }
-

branches/ppalmers-streaming/src/libstreaming/amdtp/AmdtpTransmitStreamProcessor.h

@@ -80 +80 @@
-    virtual ~AmdtpTransmitStreamProcessor() {};
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
-    unsigned int getPacketsPerPeriod();
-
-    unsigned int getMaxPacketSize() {return 4 * (2 + m_syt_interval * m_dimension);};
-
-protected:
-    bool processWriteBlock(char *data, unsigned int nevents, unsigned int offset);
+    bool transmitSilenceBlock(char *data, unsigned int nevents, unsigned int offset);
-
-
-
-
-
-
-
-
-
-
+
-    unsigned int fillNoDataPacketHeader(struct iec61883_packet *packet, unsigned int* length);
-    unsigned int fillDataPacketHeader(struct iec61883_packet *packet, unsigned int* length, uint32_t ts);
-
-
-    bool transferSilence(unsigned int size);
-
-    int transmitBlock(char *data, unsigned int nevents,
-                        unsigned int offset);
-
-
+
+
+
-    int encodePortToMBLAEvents(AmdtpAudioPort *, quadlet_t *data,
-                                unsigned int offset, unsigned int nevents);
-
-    int transmitSilenceBlock(char *data, unsigned int nevents,
-                        unsigned int offset);
-    int encodeSilencePortToMBLAEvents(AmdtpAudioPort *, quadlet_t *data,
-                                unsigned int offset, unsigned int nevents);
-    void updatePreparedState();
-
-
-
+
+
+
+
-    unsigned int m_dbc;
-
-    unsigned int m_ringbuffer_size_frames;
-};
-

branches/ppalmers-streaming/src/libstreaming/generic/StreamProcessor.cpp

@@ -327 +327 @@
-}
-
+enum raw1394_iso_disposition
+StreamProcessor::getPacket(unsigned char *data, unsigned int *length,
+                           unsigned char *tag, unsigned char *sy,
+                           int cycle, unsigned int dropped, unsigned int max_length) {
+    if (cycle<0) {
+        *tag = 0;
+        *sy = 0;
+        *length = 0;
+        return RAW1394_ISO_OK;
+    }
+
+    int dropped_cycles = diffCycles(cycle, m_last_cycle) - 1;
+    if (dropped_cycles < 0) debugWarning("(%p) dropped < 1 (%d)\n", this, dropped_cycles);
+    else m_dropped += dropped_cycles;
+    if (dropped_cycles > 0) debugWarning("(%p) dropped %d packets on cycle %u\n", this, dropped_cycles, cycle);
+    m_last_cycle = cycle;
+
+    // bypass based upon state
+    if (m_state == ePS_Invalid) {
+        debugError("Should not have state %s\n", ePSToString(m_state) );
+        return RAW1394_ISO_ERROR;
+    }
+    if (m_state == ePS_Created) {
+        *tag = 0;
+        *sy = 0;
+        *length = 0;
+        return RAW1394_ISO_DEFER;
+    }
+
+    // normal processing
+    // note that we can't use getCycleTimer directly here,
+    // because packets are queued in advance. This means that
+    // we the packet we are constructing will be sent out
+    // on 'cycle', not 'now'.
+    unsigned int ctr = m_handler->getCycleTimer();
+    int now_cycles = (int)CYCLE_TIMER_GET_CYCLES(ctr);
+
+    // the difference between the cycle this
+    // packet is intended for and 'now'
+    int cycle_diff = diffCycles(cycle, now_cycles);
+
+    #ifdef DEBUG
+    if(cycle_diff < 0) {
+        debugWarning("Requesting packet for cycle %04d which is in the past (now=%04dcy)\n",
+            cycle, now_cycles);
+    }
+    #endif
+
+    // store the previous timestamp
+    m_last_timestamp2 = m_last_timestamp;
+
+    // NOTE: synchronized switching is restricted to a 0.5 sec span (4000 cycles)
+    //       it happens on the first 'good' cycle for the wait condition
+    //       or on the first received cycle that is received afterwards (might be a problem)
+
+    // check whether we are waiting for a stream to be disabled
+    if(m_state == ePS_WaitingForStreamDisable) {
+        // we then check whether we have to switch on this cycle
+        if (diffCycles(cycle, m_cycle_to_switch_state) >= 0) {
+            debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state to DryRunning\n");
+            m_next_state = ePS_DryRunning;
+            if (!updateState()) { // we are allowed to change the state directly
+                debugError("Could not update state!\n");
+                return RAW1394_ISO_ERROR;
+            }
+        } else {
+            // not time to disable yet
+        }
+    }
+    // check whether we are waiting for a stream to be enabled
+    else if(m_state == ePS_WaitingForStreamEnable) {
+        // we then check whether we have to switch on this cycle
+        if (diffCycles(cycle, m_cycle_to_switch_state) >= 0) {
+            debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state to Running\n");
+            m_next_state = ePS_Running;
+            if (!updateState()) { // we are allowed to change the state directly
+                debugError("Could not update state!\n");
+                return RAW1394_ISO_ERROR;
+            }
+        } else {
+            // not time to enable yet
+        }
+        // we are dryRunning hence data should be processed in any case
+    }
+    // check whether we are waiting for a stream to startup
+    else if(m_state == ePS_WaitingForStream) {
+        // as long as the cycle parameter is not in sync with
+        // the current time, the stream is considered not
+        // to be 'running'
+        // we then check whether we have to switch on this cycle
+        if ((cycle_diff >= 0) && (diffCycles(cycle, m_cycle_to_switch_state) >= 0)) {
+            debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state to DryRunning due to good packet\n");
+            // hence go to the dryRunning state
+            m_next_state = ePS_DryRunning;
+            if (!updateState()) { // we are allowed to change the state directly
+                debugError("Could not update state!\n");
+                return RAW1394_ISO_ERROR;
+            }
+        } else {
+            // not time (yet) to switch state
+        }
+    }
+    else if(m_state == ePS_Running) {
+        // check the packet header
+        if (generatePacketHeader(data, length, tag, sy, cycle, dropped_cycles, max_length)) {
+            debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "XMIT: CY=%04u TS=%011llu\n",
+                    cycle, m_last_timestamp);
+            // update some accounting
+            m_last_good_cycle = cycle;
+            m_last_dropped = dropped_cycles;
+
+            // check whether a state change has been requested
+            // note that only the wait state changes are synchronized with the cycles
+            if(m_state != m_next_state) {
+                debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state from %s to %s\n",
+                                                ePSToString(m_state), ePSToString(m_next_state));
+                // execute the requested change
+                if (!updateState()) { // we are allowed to change the state directly
+                    debugError("Could not update state!\n");
+                    return RAW1394_ISO_ERROR;
+                }
+            }
+
+            bool ok = generatePacketData(data, length, tag, sy, cycle, dropped_cycles, max_length);
+            // if an xrun occured, switch to the dryRunning state and
+            // allow for the xrun to be picked up
+            if (!ok) {
+                debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state to DryRunning due to xrun\n");
+                m_next_state = ePS_DryRunning;
+                // execute the requested change
+                if (!updateState()) { // we are allowed to change the state directly
+                    debugError("Could not update state!\n");
+                    return RAW1394_ISO_ERROR;
+                }
+                goto send_empty_packet;
+            }
+            return RAW1394_ISO_OK;
+        }
+    }
+    // we are not running, so send an empty packet
+    // we should generate a valid packet any time
+send_empty_packet:
+    // note that only the wait state changes are synchronized with the cycles
+    if(m_state != m_next_state) {
+        debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state from %s to %s\n",
+                                        ePSToString(m_state), ePSToString(m_next_state));
+        // execute the requested change
+        if (!updateState()) { // we are allowed to change the state directly
+            debugError("Could not update state!\n");
+            return RAW1394_ISO_ERROR;
+        }
+    }
+
+    debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "XMIT EMPTY: CY=%04u\n", cycle);
+    generateSilentPacketHeader(data, length, tag, sy, cycle, dropped_cycles, max_length);
+    generateSilentPacketData(data, length, tag, sy, cycle, dropped_cycles, max_length);
+    return RAW1394_ISO_DEFER;
+}
+
+
-// Frame Transfer API
+/**
+ * Transfer a block of frames from the event buffer to the port buffers
+ * @param nbframes number of frames to transfer
+ * @param ts the timestamp that the LAST frame in the block should have
+ * @return 
+ */
-bool StreamProcessor::getFrames(unsigned int nbframes, int64_t ts) {
-    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "%p.getFrames(%d, %11llu)", nbframes, ts);
@@ -338 +504 @@
-// FIXME: this should be done somewhere else
-#ifdef DEBUG
-hea
+expecte
-    signed int fc;
-    int32_t lag_ticks;
@@ -345 +511 @@
-    // in order to sync up multiple received streams, we should 
-    // use the ts parameter. It specifies the time of the block's 
-
+
+
+
+
+
+
-    
-    ffado_timestamp_t ts_head_tmp;
-    m_data_buffer->getBufferHeadTimestamp(&ts_head_tmp, &fc);
-
-
-
+
-    
-assert(rate!=0.0
+lag_ticks = diffTicks(ts, ts_expected
-    
-
+
+
-    
-    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "stream (%p): drifts %6d ticks = %10.5f frames (rate=%10.5f), %lld, %llu, %d\n",
-rate, ts, ts_hea
-
->=
+ srate, ts, ts_expecte
+
+ >= 
-        // the stream lags
-        debugWarning( "stream (%p): lags  with %6d ticks = %10.5f frames (rate=%10.5f), %lld, %llu, %d\n",
-rate, ts, ts_hea
-<=
+ srate, ts, ts_expecte
+ <= 
-        // the stream leads
-        debugWarning( "stream (%p): leads with %6d ticks = %10.5f frames (rate=%10.5f), %lld, %llu, %d\n",
-rate, ts, ts_hea
+ srate, ts, ts_expecte
-    }
-#endif
@@ -386 +556 @@
-}
-
+bool StreamProcessor::putFrames(unsigned int nbframes, int64_t ts) {
+    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "%p.putFrames(%d, %11llu)", nbframes, ts);
+    assert( getType() == ePT_Transmit );
+    if(isDryRunning()) return putFramesDry(nbframes, ts);
+    else return putFramesWet(nbframes, ts);
+}
+
+bool
+StreamProcessor::putFramesWet(unsigned int nbframes, int64_t ts) {
+    debugOutput(DEBUG_LEVEL_ULTRA_VERBOSE, "StreamProcessor::putFramesWet(%d, %llu)\n", nbframes, ts);
+    // transfer the data
+    m_data_buffer->blockProcessWriteFrames(nbframes, ts);
+    debugOutput(DEBUG_LEVEL_ULTRA_VERBOSE, " New timestamp: %llu\n", ts);
+    return true; // FIXME: what about failure?
+}
+
+bool
+StreamProcessor::putFramesDry(unsigned int nbframes, int64_t ts) {
+    debugOutput(DEBUG_LEVEL_ULTRA_VERBOSE, "StreamProcessor::putFramesDry(%d, %llu)\n", nbframes, ts);
+    // do nothing
+    return true;
+}
-
-/***********************************************
@@ -421 +613 @@
-bool StreamProcessor::stop()
-{
-    uint64_t time_to_stop_at = addTicks(m_handler->getCycleTimerTicks(), 200 * TICKS_PER_CYCLE);
-    int cnt;
-    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "stop...\n");
-    switch (m_state) {
@@ -437 +627 @@
-}
-
-
+
+
-{
-    // first set the time, since in the packet loop we first check m_state == m_next_state before
@@ -446 +637 @@
-}
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
-        return false;
-    }
-    while (m_state != wait_state && cnt) {
-        usleep(10000);
-        cnt++;
-    }
-    if(cnt==0) {
-        debugError("Timeout entering Stopped state\n");
-        return false;
-    }
-    debugOutput(DEBUG_LEVEL_VERBOSE, " entered state %s\n", ePSToString(wait_state));
-    return true;
-}
-
-
+cheduleS
-    uint64_t tx;
-    if (t < 0) {
@@ -474 +660 @@
-        tx = t;
-    }
- StreamProcessor::startDryRunning for (%p)\n",
+for %s SP (%p)\n", ePTToString(getType()), 
-    debugOutput(DEBUG_LEVEL_VERBOSE,"  Now                   : %011lu\n", m_handler->getCycleTimerTicks());
-    debugOutput(DEBUG_LEVEL_VERBOSE,"  Start at              : %011llu (%u)\n", tx, TICKS_TO_CYCLES(tx));
-    if (m_state == ePS_Stopped) {
-AndWaitForStateTransition(ePS_WaitingForStream, tx, ePS_DryRunning
+StateTransition(ePS_WaitingForStream, tx
-    } else if (m_state == ePS_Running) {
-AndWaitForStateTransition(ePS_WaitingForStreamDisable, tx, ePS_DryRunning
+StateTransition(ePS_WaitingForStreamDisable, tx
-    } else {
-        debugError("Cannot switch to ePS_DryRunning from %s\n", ePSToString(m_state));
@@ -487 +673 @@
-}
-
-
+cheduleS
-    uint64_t tx;
-    if (t < 0) {
@@ -494 +680 @@
-        tx = t;
-    }
- StreamProcessor::startRunning for (%p)\n",
+for %s SP (%p)\n", ePTToString(getType()), 
-    debugOutput(DEBUG_LEVEL_VERBOSE,"  Now                   : %011lu\n", m_handler->getCycleTimerTicks());
-    debugOutput(DEBUG_LEVEL_VERBOSE,"  Start at              : %011llu (%u)\n", tx, TICKS_TO_CYCLES(tx));
-AndWaitForStateTransition(ePS_WaitingForStreamEnable, tx, ePS_Running
-
-
-
+StateTransition(ePS_WaitingForStreamEnable, tx
+
+
+cheduleS
-    uint64_t tx;
-    if (t < 0) {
@@ -507 +693 @@
-        tx = t;
-    }
- StreamProcessor::stopDryRunning for (%p)\n",
+for %s SP (%p)\n", ePTToString(getType()), 
-    debugOutput(DEBUG_LEVEL_VERBOSE,"  Now                   : %011lu\n", m_handler->getCycleTimerTicks());
-    debugOutput(DEBUG_LEVEL_VERBOSE,"  Stop at               : %011llu (%u)\n", tx, TICKS_TO_CYCLES(tx));
-AndWaitForStateTransition(ePS_Stopped, tx, ePS_Stopped
-
-
-
+StateTransition(ePS_Stopped, tx
+
+
+cheduleS
-    uint64_t tx;
-    if (t < 0) {
@@ -520 +706 @@
-        tx = t;
-    }
- StreamProcessor::stopRunning for (%p)\n",
+for %s SP (%p)\n", ePTToString(getType()), 
-    debugOutput(DEBUG_LEVEL_VERBOSE,"  Now                   : %011lu\n", m_handler->getCycleTimerTicks());
-    debugOutput(DEBUG_LEVEL_VERBOSE,"  Stop at               : %011llu (%u)\n", tx, TICKS_TO_CYCLES(tx));
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-// internal state API
@@ -544 +779 @@
-{
-    float ticks_per_frame;
-
+ = m_manager->getNbBuffers() * m_manager->getPeriodSize()
-
-    debugOutput(DEBUG_LEVEL_VERBOSE, "Enter from state: %s\n", ePSToString(m_state));
@@ -554 +789 @@
-            // object just created
-            result = m_data_buffer->init();
-
+
-            // prepare the framerate estimate
-            ticks_per_frame = (TICKS_PER_SECOND*1.0) / ((float)m_manager->getNominalRate());
-            m_ticks_per_frame = ticks_per_frame;
-            debugOutput(DEBUG_LEVEL_VERBOSE,"Initializing remote ticks/frame to %f\n", ticks_per_frame);
-
+
-            // initialize internal buffer
-
-
+
-
-            result &= m_data_buffer->setEventSize( getEventSize() );
-            result &= m_data_buffer->setEventsPerFrame( getEventsPerFrame() );
-
-
+
+
+
+
+
-            result &= m_data_buffer->setNominalRate(ticks_per_frame);
-            result &= m_data_buffer->setWrapValue(128L*TICKS_PER_SECOND);
@@ -612 +849 @@
-                        }
-                        // buffertype and datatype are dependant on the API
-                        // buffertype and datatype are dependant on the API
-                        debugWarning("---------------- ! Doing hardcoded test setup ! --------------\n");
-                        // buffertype and datatype are dependant on the API
@@ -643 +879 @@
-    }
-
-
-
+
-    // make the buffer transparent
-    m_data_buffer->setTransparent(true);
@@ -652 +887 @@
-
-    m_state = ePS_Stopped;
+    #ifdef DEBUG
+    if (getDebugLevel() >= DEBUG_LEVEL_VERBOSE) {
+        debugOutput(DEBUG_LEVEL_VERBOSE, "State switch complete, dumping SP info...\n");
+        dumpInfo();
+    }
+    #endif
-    return result;
-}
@@ -679 +920 @@
-    }
-    m_state = ePS_WaitingForStream;
+    #ifdef DEBUG
+    if (getDebugLevel() >= DEBUG_LEVEL_VERBOSE) {
+        debugOutput(DEBUG_LEVEL_VERBOSE, "State switch complete, dumping SP info...\n");
+        dumpInfo();
+    }
+    #endif
-    return true;
-}
@@ -701 +948 @@
-            debugOutput(DEBUG_LEVEL_VERBOSE, "StreamProcessor %p started dry-running at cycle %d\n", this, m_last_cycle);
-            if (getType() == ePT_Receive) {
+                // this to ensure that there is no discontinuity when starting to 
+                // update the DLL based upon the received packets
-                m_data_buffer->setBufferTailTimestamp(m_last_timestamp);
-            } else {
@@ -708 +957 @@
-            break;
-        case ePS_WaitingForStreamDisable:
-reset
+clearBuffer
-            m_data_buffer->setTransparent(true);
-            break;
@@ -716 +965 @@
-    }
-    m_state = ePS_DryRunning;
+    #ifdef DEBUG
+    if (getDebugLevel() >= DEBUG_LEVEL_VERBOSE) {
+        debugOutput(DEBUG_LEVEL_VERBOSE, "State switch complete, dumping SP info...\n");
+        dumpInfo();
+    }
+    #endif
-    return result;
-}
@@ -732 +987 @@
-{
-    debugOutput(DEBUG_LEVEL_VERBOSE, "Enter from state: %s\n", ePSToString(m_state));
+    unsigned int ringbuffer_size_frames;
-    switch(m_state) {
-        case ePS_DryRunning:
@@ -737 +993 @@
-            // this basically means nothing, the state change will
-            // be picked up by the packet iterator
+
+            if(!m_data_buffer->clearBuffer()) { // FIXME: don't like the reset() name
+                debugError("Could not reset data buffer\n");
+                return false;
+            }
+            if (getType() == ePT_Transmit) {
+                ringbuffer_size_frames = m_manager->getNbBuffers() * m_manager->getPeriodSize();
+                debugOutput(DEBUG_LEVEL_VERBOSE, "Prefill transmit SP %p with %u frames\n", this, ringbuffer_size_frames);
+                // prefill the buffer
+                if(!transferSilence(ringbuffer_size_frames)) {
+                    debugFatal("Could not prefill transmit stream\n");
+                    return false;
+                }
+            }
+
-            break;
-        default:
@@ -743 +1014 @@
-    }
-    m_state = ePS_WaitingForStreamEnable;
+    #ifdef DEBUG
+    if (getDebugLevel() >= DEBUG_LEVEL_VERBOSE) {
+        debugOutput(DEBUG_LEVEL_VERBOSE, "State switch complete, dumping SP info...\n");
+        dumpInfo();
+    }
+    #endif
-    return true;
-}
@@ -765 +1042 @@
-            debugOutput(DEBUG_LEVEL_VERBOSE, "StreamProcessor %p started running at cycle %d\n", 
-                                             this, m_last_cycle);
-
-
-
-
-
-
+
+
-            break;
-        default:
@@ -777 +1050 @@
-    }
-    m_state = ePS_Running;
+    #ifdef DEBUG
+    if (getDebugLevel() >= DEBUG_LEVEL_VERBOSE) {
+        debugOutput(DEBUG_LEVEL_VERBOSE, "State switch complete, dumping SP info...\n");
+        dumpInfo();
+    }
+    #endif
-    return result;
-}
@@ -802 +1081 @@
-    }
-    m_state = ePS_WaitingForStreamDisable;
+    #ifdef DEBUG
+    if (getDebugLevel() >= DEBUG_LEVEL_VERBOSE) {
+        debugOutput(DEBUG_LEVEL_VERBOSE, "State switch complete, dumping SP info...\n");
+        dumpInfo();
+    }
+    #endif
-    return true;
-}
@@ -922 +1207 @@
-}
-
+/***********************************************
+ * Helper routines                             *
+ ***********************************************/
+bool
+StreamProcessor::transferSilence(unsigned int nframes)
+{
+    bool retval;
+    signed int fc;
+    ffado_timestamp_t ts_tail_tmp;
+
+    // prepare a buffer of silence
+    char *dummybuffer = (char *)calloc(sizeof(quadlet_t), nframes * getEventsPerFrame());
+    transmitSilenceBlock(dummybuffer, nframes, 0);
+
+    m_data_buffer->getBufferTailTimestamp(&ts_tail_tmp, &fc);
+    if (fc != 0) {
+        debugWarning("Prefilling a buffer that already contains %d frames\n", fc);
+    }
+
+    // add the silence data to the ringbuffer
+    if(m_data_buffer->preloadFrames(nframes, dummybuffer, true)) {
+        retval = true;
+    } else {
+        debugWarning("Could not write to event buffer\n");
+        retval = false;
+    }
+    free(dummybuffer);
+    return retval;
+}
-
-/**
@@ -968 +1282 @@
-    IsoStream::dumpInfo();
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  StreamProcessor info:\n");
-
-
+
+
+
+
+
+
+
+
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Xruns                 : %d\n", m_xruns);
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  State                 : %s\n", ePSToString(m_state));
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "   Next state           : %s\n", ePSToString(m_next_state));
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "    transition at       : %u\n", m_cycle_to_switch_state);
-
-
+
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Nominal framerate     : %u\n", m_manager->getNominalRate());
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Device framerate      : Sync: %f, Buffer %f\n",
@@ -983 +1302 @@
-
-    m_data_buffer->dumpInfo();
-
-    m_PeriodStat.dumpInfo();
-    m_PacketStat.dumpInfo();
-//     m_WakeupStat.dumpInfo();
-}
-

branches/ppalmers-streaming/src/libstreaming/generic/StreamProcessor.h

@@ -81 +81 @@
-        ePS_WaitingForStreamDisable,
-    };
-
+
-    ///> set the SP state to a specific value
-    void setState(enum eProcessorState);
@@ -103 +103 @@
-
-    bool scheduleStateTransition(enum eProcessorState state, uint64_t time_instant);
-
-
-
-
+
+
+
-    bool isRunning()
-            {return m_state == ePS_Running;};
@@ -112 +111 @@
-            {return m_state == ePS_DryRunning;};
-
-
+
-    bool startDryRunning(int64_t time_to_start_at);
-    bool startRunning(int64_t time_to_start_at);
@@ -118 +117 @@
-    bool stopRunning(int64_t time_to_stop_at);
-
+    // these only schedule the transition
+    bool scheduleStartDryRunning(int64_t time_to_start_at);
+    bool scheduleStartRunning(int64_t time_to_start_at);
+    bool scheduleStopDryRunning(int64_t time_to_stop_at);
+    bool scheduleStopRunning(int64_t time_to_stop_at);
+
-    // the main difference between init and prepare is that when prepare is called,
-    // the SP is registered to a manager (FIXME: can't it be called by the manager?)
-    bool init();
-    bool prepare();
+
-    ///> stop the SP from running or dryrunning
-    bool stop();
-
-
+
+ // constructor/destructor
-    StreamProcessor(enum eProcessorType type, int port);
-    virtual ~StreamProcessor();
-
-
-
+
-    // the transmit interface accepts frames and provides packets
-    // implement these for a transmit SP
-    // leave default for a receive SP
-    virtual enum raw1394_iso_disposition
-    getPacket(unsigned char *data, unsigned int *length,
-                unsigned char *tag, unsigned char *sy,
-                int cycle, unsigned int dropped, unsigned int max_length)
-        {debugWarning("call not allowed\n"); return RAW1394_ISO_STOP;};
-    virtual bool putFrames(unsigned int nbframes, int64_t ts) 
-        {debugWarning("call not allowed\n"); return false;};
-    virtual bool putFramesDry(unsigned int nbframes, int64_t ts)
-        {debugWarning("call not allowed\n"); return false;};
-    virtual bool processWriteBlock(char *data, unsigned int nevents, unsigned int offset)
-        {debugWarning("call not allowed\n"); return false;};
-
-    // the receive interface accepts packets and provides frames
-
-
-
-
-
-
-
-
-
-
-
-
+
-    enum raw1394_iso_disposition
-        putPacket(unsigned char *data, unsigned int length,
@@ -164 +147 @@
-                  unsigned int cycle, unsigned int dropped);
-
+    enum raw1394_iso_disposition
+    getPacket(unsigned char *data, unsigned int *length,
+                unsigned char *tag, unsigned char *sy,
+                int cycle, unsigned int dropped, unsigned int max_length);
+
-    bool getFrames(unsigned int nbframes, int64_t ts); ///< transfer the buffer contents to the client
-
+
+
+
-    // to be implemented by the children
+    // the following methods are to be implemented by receive SP subclasses
+    virtual bool processPacketHeader(unsigned char *data, unsigned int length,
+                                     unsigned char channel, unsigned char tag,
+                                     unsigned char sy, unsigned int cycle,
+                                     unsigned int dropped)
+        {debugWarning("call not allowed\n"); return false;};
+    virtual bool processPacketData(unsigned char *data, unsigned int length,
+                                   unsigned char channel, unsigned char tag,
+                                   unsigned char sy, unsigned int cycle,
+                                   unsigned int dropped)
+        {debugWarning("call not allowed\n"); return false;};
-    virtual bool processReadBlock(char *data, unsigned int nevents, unsigned int offset)
-        {debugWarning("call not allowed\n"); return false;};
-    virtual bool provideSilenceBlock(unsigned int nevents, unsigned int offset)
+        {debugWarning("call not allowed\n"); return false;};
+
+    // the following methods are to be implemented by transmit SP subclasses
+    virtual bool generatePacketHeader(unsigned char *data, unsigned int *length,
+                                      unsigned char *tag, unsigned char *sy,
+                                      int cycle, unsigned int dropped,
+                                      unsigned int max_length)
+        {debugWarning("call not allowed\n"); return false;};
+    virtual bool generatePacketData(unsigned char *data, unsigned int *length,
+                                    unsigned char *tag, unsigned char *sy,
+                                    int cycle, unsigned int dropped,
+                                    unsigned int max_length)
+        {debugWarning("call not allowed\n"); return false;};
+    virtual bool generateSilentPacketHeader(unsigned char *data, unsigned int *length,
+                                            unsigned char *tag, unsigned char *sy,
+                                            int cycle, unsigned int dropped,
+                                            unsigned int max_length)
+        {debugWarning("call not allowed\n"); return false;};
+    virtual bool generateSilentPacketData(unsigned char *data, unsigned int *length,
+                                          unsigned char *tag, unsigned char *sy,
+                                          int cycle, unsigned int dropped,
+                                          unsigned int max_length)
+        {debugWarning("call not allowed\n"); return false;};
+    virtual bool processWriteBlock(char *data, unsigned int nevents, unsigned int offset)
+        {debugWarning("call not allowed\n"); return false;};
+    virtual bool transmitSilenceBlock(char *data, unsigned int nevents, unsigned int offset)
-        {debugWarning("call not allowed\n"); return false;};
-
@@ -175 +202 @@
-    bool getFramesDry(unsigned int nbframes, int64_t ts);
-    bool getFramesWet(unsigned int nbframes, int64_t ts);
+    bool putFramesDry(unsigned int nbframes, int64_t ts);
+    bool putFramesWet(unsigned int nbframes, int64_t ts);
+
+    bool transferSilence(unsigned int size);
-
-    // move to private?
@@ -188 +219 @@
-
-//--- data buffering and accounting
-
+
+
+
+
+
+
+
+
+
+
+
-    Util::TimestampedBuffer *m_data_buffer;
-
@@ -255 +296 @@
-        uint64_t getTimeAtPeriod();
-
-
+ // FIXME: should disappear
-
-
@@ -318 +359 @@
-
-        /**
-between buffer updates
-between buffer updates
-
-UpdatePeriod
+in a packet
+in a packet
+
+NominalFramesPerPacket
-
-    protected:
@@ -340 +381 @@
-    StreamStatistics m_WakeupStat;
-    DECLARE_DEBUG_MODULE;
-
-};
-

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

@@ -283 +283 @@
-    // lower on average.
-    max_of_min_delay += FFADO_SIGNAL_DELAY_TICKS;
-
+sync delay =
-        max_of_min_delay,
-        (unsigned int)TICKS_TO_SECS(max_of_min_delay),
@@ -295 +295 @@
-    //sleep(2); // FIXME: be smarter here
-
-
+
+
-    debugOutput( DEBUG_LEVEL_VERBOSE, "Waiting for sync...\n");
-    // in order to obtain that, we wait for the first periods to be received.
-    int nb_sync_runs=20;
-    int64_t time_till_next_period;
-    while(nb_sync_runs--) { // or while not sync-ed?
+        // check if we were waked up too soon
-        time_till_next_period=m_SyncSource->getTimeUntilNextPeriodSignalUsecs();
-
+Y_VER
-        if(time_till_next_period > 0) {
-            // wait for the period
@@ -309 +310 @@
-    }
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
-    // FIXME: in the SPM it would be nice to have system time instead of
-    //        1394 time
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-    // STEP X: switch SP's over to the running state
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-    }
-    debugOutput( DEBUG_LEVEL_VERBOSE, " StreamProcessor streams running...\n");
@@ -577 +598 @@
-    // this is to notify the client of the delay
-    // that we introduced
-
+-
-
-    // we save the 'ideal' time of the transfer at this point,
@@ -683 +704 @@
-
-bool StreamProcessorManager::transfer(enum StreamProcessor::eProcessorType t) {
-
+
+
+
+
+
+
-    bool retval = true;
-    // a static cast could make sure that there is no performance
-    // penalty for the virtual functions (to be checked)
-    if (t==StreamProcessor::ePT_Receive) {
-        // determine the time at which we want reception to start
-        float rate=m_SyncSource->getTicksPerFrame();
-        int64_t one_frame_in_ticks=(int64_t)(((float)m_period)*rate);
-        
-        int64_t receive_timestamp = substractTicks(m_time_of_transfer, one_frame_in_ticks);
-        
-        if(receive_timestamp<0) {
-            debugWarning("receive ts < 0.0 : %lld, m_time_of_transfer= %llu, one_frame_in_ticks=%lld\n",
-             receive_timestamp, m_time_of_transfer, one_frame_in_ticks);
-        }
-        if(receive_timestamp>(128L*TICKS_PER_SECOND)) {
-            debugWarning("receive ts > 128L*TICKS_PER_SECOND : %lld, m_time_of_transfer= %llu, one_frame_in_ticks=%lld\n",
-             receive_timestamp, m_time_of_transfer, one_frame_in_ticks);
-        }
-        
-        for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
-                it != m_ReceiveProcessors.end();
-                ++it ) {
-receive_timestamp
+m_time_of_transfer
-                    debugWarning("could not getFrames(%u, %11llu) from stream processor (%p)\n",
-                            m_period, m_time_of_transfer,*it);
@@ -713 +724 @@
-        }
-    } else {
+        // FIXME: in the SPM it would be nice to have system time instead of
+        //        1394 time
+        float rate = m_SyncSource->getTicksPerFrame();
+        int64_t one_ringbuffer_in_ticks=(int64_t)(((float)(m_nb_buffers * m_period)) * rate);
+
+        // the data we are putting into the buffer is intended to be transmitted
+        // one ringbuffer size after it has been received
+        int64_t transmit_timestamp = addTicks(m_time_of_transfer, one_ringbuffer_in_ticks);
+
-        for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
-                it != m_TransmitProcessors.end();
@@ -718 +738 @@
-            // FIXME: in the SPM it would be nice to have system time instead of
-            //        1394 time
-            float rate=m_SyncSource->getTicksPerFrame();
-            int64_t one_ringbuffer_in_ticks=(int64_t)(((float)(m_nb_buffers*m_period))*rate);
-
-            // the data we are putting into the buffer is intended to be transmitted
-            // one ringbuffer size after it has been received
-            int64_t transmit_timestamp = addTicks(m_time_of_transfer, one_ringbuffer_in_ticks);
-
-            if(!(*it)->putFrames(m_period, transmit_timestamp)) {
-                debugWarning("could not putFrames(%u,%llu) to stream processor (%p)\n",
@@ -730 +743 @@
-                retval &= false; // buffer underrun
-            }
-
-        }
-    }
-    return retval;
-}
-
-/**
- * @brief Dry run one period for both receive and transmit StreamProcessors
- *
- * Process one period of frames for all streamprocessors, without touching the
- * client buffers. This only removes an incoming period from the ISO receive buffer and
- * puts one period of silence into the transmit buffers.
- *
- * @return true if successful, false otherwise (indicates xrun).
- */
-bool StreamProcessorManager::dryRun() {
-    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "Dry-running period...\n");
-    bool retval=true;
-    retval &= dryRun(StreamProcessor::ePT_Receive);
-    retval &= dryRun(StreamProcessor::ePT_Transmit);
-    return retval;
-}
-
-/**
- * @brief Dry run one period for either the receive or transmit StreamProcessors
- *
- * see dryRun()
- *
- * @param t The processor type to dryRun for (receive or transmit)
- * @return true if successful, false otherwise (indicates xrun).
- */
-
-bool StreamProcessorManager::dryRun(enum StreamProcessor::eProcessorType t) {
-    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "Dry-running period...\n");
-    bool retval = true;
-    // a static cast could make sure that there is no performance
-    // penalty for the virtual functions (to be checked)
-    if (t==StreamProcessor::ePT_Receive) {
-        // determine the time at which we want reception to start
-        float rate=m_SyncSource->getTicksPerFrame();
-        int64_t one_frame_in_ticks=(int64_t)(((float)m_period)*rate);
-        
-        int64_t receive_timestamp = substractTicks(m_time_of_transfer, one_frame_in_ticks);
-        
-        if(receive_timestamp<0) {
-            debugWarning("receive ts < 0.0 : %lld, m_time_of_transfer= %llu, one_frame_in_ticks=%lld\n",
-             receive_timestamp, m_time_of_transfer, one_frame_in_ticks);
-        }
-        if(receive_timestamp>(128L*TICKS_PER_SECOND)) {
-            debugWarning("receive ts > 128L*TICKS_PER_SECOND : %lld, m_time_of_transfer= %llu, one_frame_in_ticks=%lld\n",
-             receive_timestamp, m_time_of_transfer, one_frame_in_ticks);
-        }
-        
-        for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
-                it != m_ReceiveProcessors.end();
-                ++it ) {
-
-            if(!(*it)->getFramesDry(m_period, receive_timestamp)) {
-                    debugOutput(DEBUG_LEVEL_VERBOSE,"could not getFrames(%u, %11llu) from stream processor (%p)\n",
-                            m_period, m_time_of_transfer,*it);
-                retval &= false; // buffer underrun
-            }
-
-        }
-    } else {
-        for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
-                it != m_TransmitProcessors.end();
-                ++it ) {
-            // FIXME: in the SPM it would be nice to have system time instead of
-            //        1394 time
-            float rate=m_SyncSource->getTicksPerFrame();
-            int64_t one_ringbuffer_in_ticks=(int64_t)(((float)(m_nb_buffers*m_period))*rate);
-
-            // the data we are putting into the buffer is intended to be transmitted
-            // one ringbuffer size after it has been received
-            int64_t transmit_timestamp = addTicks(m_time_of_transfer, one_ringbuffer_in_ticks);
-
-            if(!(*it)->putFramesDry(m_period, transmit_timestamp)) {
-                debugOutput(DEBUG_LEVEL_VERBOSE, "could not putFrames(%u,%llu) to stream processor (%p)\n",
-                        m_period, transmit_timestamp, *it);
-                retval &= false; // buffer underrun
-            }
-
-        }
-    }

branches/ppalmers-streaming/src/libstreaming/StreamProcessorManager.h

@@ -83 +83 @@
-
-    bool waitForPeriod(); ///< wait for the next period
-
-    bool transfer(); ///< transfer the buffer contents from/to client
-    bool transfer(enum StreamProcessor::eProcessorType); ///< transfer the buffer contents from/to client (single processor type)
-    
-    bool dryRun();
-    bool dryRun(enum StreamProcessor::eProcessorType);
-
-    int getDelayedUsecs() {return m_delayed_usecs;};
@@ -95 +91 @@
-
-    unsigned int getNominalRate() {return m_nominal_framerate;};
+    uint64_t getTimeOfLastTransfer() { return m_time_of_transfer;};
-
-private:
-
-    int m_delayed_usecs;
-    // this stores the time at which the next transfer should occur

branches/ppalmers-streaming/src/libutil/TimestampedBuffer.cpp

@@ -266 +266 @@
- * @return true if successful
- */
-
+
+
-    ffado_ringbuffer_reset(m_event_buffer);
-
-    resetFrameCounter();
-
-    return true;
-}
@@ -397 +396 @@
-
-/**
+ * @brief Preload frames into the buffer
+ *
+ * Preload \ref nframes of frames from the buffer pointed to by \ref data to the
+ * internal ringbuffer. Does not care about transparency. Keeps the buffer head or tail
+ * timestamp constant.
+ *
+ * @note not thread safe
+ *
+ * @param nframes number of frames to copy
+ * @param data pointer to the frame buffer
+ * @param keep_head_ts if true, keep the head timestamp constant. If false, keep the
+ *                     tail timestamp constant.
+ * @return true if successful
+ */
+bool TimestampedBuffer::preloadFrames(unsigned int nframes, char *data, bool keep_head_ts) {
+    unsigned int write_size = nframes * m_event_size * m_events_per_frame;
+    // add the data payload to the ringbuffer
+    size_t written = ffado_ringbuffer_write(m_event_buffer, data, write_size);
+    if (written < write_size)
+    {
+        debugWarning("ringbuffer full, request: %u, actual: %u\n", write_size, written);
+        return false;
+    }
+    
+    // make sure the head timestamp remains identical
+    signed int fc;
+    ffado_timestamp_t ts;
+
+    if (keep_head_ts) {
+        getBufferHeadTimestamp(&ts, &fc);
+    } else {
+        getBufferTailTimestamp(&ts, &fc);
+    }
+    // update frame counter
+    m_framecounter += nframes;
+    if (keep_head_ts) {
+        setBufferHeadTimestamp(ts);
+    } else {
+        setBufferTailTimestamp(ts);
+    }
+
+    return true;
+}
+
+/**
- * @brief Drop frames from the head of the buffer
- *
@@ -410 +454 @@
-    ffado_ringbuffer_read_advance(m_event_buffer, read_size);
-    decrementFrameCounter(nframes);
-
-    return true;
-}
@@ -722 +765 @@
-#endif
-
-=
+ = 
-
-    ENTER_CRITICAL_SECTION;
@@ -1045 +1088 @@
-    
-    ENTER_CRITICAL_SECTION;
-=
+ = 
-    EXIT_CRITICAL_SECTION;
-
@@ -1054 +1097 @@
-#endif
-
-=
+ = 
-    ts += m_tick_offset;
-
@@ -1123 +1166 @@
-                this, diff);
-
-=
+ = 
-
-    debugOutputShort(DEBUG_LEVEL_VERY_VERBOSE, "diff2="TIMESTAMP_FORMAT_SPEC" err=%f\n",
@@ -1208 +1251 @@
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  TimestampedBuffer (%p) info:\n",this);
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Frame counter         : %d\n", m_framecounter);
+    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Events in buffer      : %d\n", getBufferFill());
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Buffer head timestamp : "TIMESTAMP_FORMAT_SPEC"\n",ts_head);
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Buffer tail timestamp : "TIMESTAMP_FORMAT_SPEC"\n",m_buffer_tail_timestamp);

branches/ppalmers-streaming/src/libutil/TimestampedBuffer.h

@@ -37 +37 @@
-// #define TIMESTAMP_FORMAT_SPEC "%012lld"
-
- {
-
+
+
-
-class TimestampedBufferClient;
-
-/**
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
- {
-
-
-
-
-
-virtual ~TimestampedBuffer
-
-
-bool dropFrames(unsigned int nbframes
-
-
-bool readFrames(unsigned int nbframes, char *data
-
-bool blockProcessWriteFrames(unsigned int nbframes, ffado_timestamp_t ts
-bool blockProcessReadFrames(unsigned int nbframes
-
-
-
-bool reset
-
-
-
-
-
-
-void setTransparent(bool v
-
-bool setEventSize(unsigned int s
-bool setEventsPerFrame(unsigned int s
-bool setBufferSize(unsigned int s
-
-
-
-
-bool setWrapValue(ffado_timestamp_t w
-
-
-
-
-
-
-void getBufferHeadTimestamp(ffado_timestamp_t *ts, signed int *fc
-void getBufferTailTimestamp(ffado_timestamp_t *ts, signed int *fc
-
-void setBufferTailTimestamp(ffado_timestamp_t new_timestamp
-void setBufferHeadTimestamp(ffado_timestamp_t new_timestamp
-
-
-bool syncBufferHeadToTimestamp(ffado_timestamp_t ts
-bool syncBufferTailToTimestamp(ffado_timestamp_t ts
-bool syncCorrectLag(int64_t ts
-
-ffado_timestamp_t getTimestampFromTail(int nframes
-ffado_timestamp_t getTimestampFromHead(int nframes
-
-
-
-
-
-bool setFrameOffset(int nframes
-bool setTickOffset(ffado_timestamp_t
-
-
-bool setNominalRate(float r
-
-
-
-bool setUpdatePeriod(unsigned int t
-
-
-
-void setVerboseLevel(int l) {setDebugLevel(l
-
-
-void decrementFrameCounter(int nbframes
-void incrementFrameCounter(int nbframes, ffado_timestamp_t new_timestamp
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+   
+
+
+    
+
+
+
+    bool writeDummyFrame
+
+
+    bool writeFrames ( unsigned int nbframes, char *data, ffado_timestamp_t ts 
+
+
+    bool preloadFrames ( unsigned int nbframes, char *data, bool keep_head_ts 
+
+    bool blockProcessWriteFrames ( unsigned int nbframes, ffado_timestamp_t ts 
+    bool blockProcessReadFrames ( unsigned int nbframes 
+
+    
+    
+    bool clearBuffer
+
+    
+    
+    
+
+    
+    void setTransparent ( bool v 
+
+    bool setEventSize ( unsigned int s 
+    bool setEventsPerFrame ( unsigned int s 
+    bool setBufferSize ( unsigned int s 
+    
+
+    
+
+    bool setWrapValue ( ffado_timestamp_t w 
+
+    
+
+    
+    
+
+    void getBufferHeadTimestamp ( ffado_timestamp_t *ts, signed int *fc 
+    void getBufferTailTimestamp ( ffado_timestamp_t *ts, signed int *fc 
+
+    void setBufferTailTimestamp ( ffado_timestamp_t new_timestamp 
+    void setBufferHeadTimestamp ( ffado_timestamp_t new_timestamp 
+
+    
+    bool syncBufferHeadToTimestamp ( ffado_timestamp_t ts 
+    bool syncBufferTailToTimestamp ( ffado_timestamp_t ts 
+    bool syncCorrectLag ( int64_t ts 
+
+    ffado_timestamp_t getTimestampFromTail ( int nframes 
+    ffado_timestamp_t getTimestampFromHead ( int nframes 
+
+    
+    
+    
+
+    bool setFrameOffset ( int nframes 
+    bool setTickOffset ( ffado_timestamp_t 
+
+    
+    bool setNominalRate ( float r 
+    
+    
+
+    bool setUpdatePeriod ( unsigned int t 
+
+    
+    
+    void setVerboseLevel ( int l ) {setDebugLevel ( l 
+
+    
+    void decrementFrameCounter ( int nbframes 
+    void incrementFrameCounter ( int nbframes, ffado_timestamp_t new_timestamp 
+    
+
+    
+
+    
+    
+
+    
+    
+    
+    
+    
+    
+    
+
+    
+
+    
+
+    
+
+    
+    
+    
+
+    
+    
+
+    
+    
+    
+    
+
+    
+    
+    
+
+    
-// JMW: try double for this too
-//    float m_dll_e2;
-
-
-
-
-
-
+    
+    
+    
+
+    
+    
-};
-
-/**
-
-
-
+
+
+
+
-    public:
-        TimestampedBufferClient() {};
-        virtual ~TimestampedBufferClient() {};
-
-(char *data, unsigned int nevents, unsigned int offset)
-(char *data, unsigned int nevents, unsigned int offset)
+ ( char *data, unsigned int nevents, unsigned int offset ) 
+ ( char *data, unsigned int nevents, unsigned int offset ) 
-
-};