Changeset 722

Timestamp:

11/24/07 09:57:25 (16 years ago)

Author:

ppalmers

Message:

more rewrite of streaming

Files:

• branches/ppalmers-streaming/src/dice/dice_avdevice.cpp (modified) (2 diffs)
• branches/ppalmers-streaming/src/libstreaming/amdtp/AmdtpReceiveStreamProcessor.cpp (modified) (5 diffs)
• branches/ppalmers-streaming/src/libstreaming/amdtp/AmdtpReceiveStreamProcessor.h (modified) (1 diff)
• branches/ppalmers-streaming/src/libstreaming/amdtp/AmdtpTransmitStreamProcessor.cpp (modified) (11 diffs)
• branches/ppalmers-streaming/src/libstreaming/amdtp/AmdtpTransmitStreamProcessor.h (modified) (1 diff)
• branches/ppalmers-streaming/src/libstreaming/generic/StreamProcessor.cpp (modified) (14 diffs)
• branches/ppalmers-streaming/src/libstreaming/generic/StreamProcessor.h (modified) (5 diffs)
• branches/ppalmers-streaming/src/libstreaming/StreamProcessorManager.cpp (modified) (7 diffs)
• branches/ppalmers-streaming/src/libstreaming/util/IsoHandlerManager.cpp (modified) (7 diffs)
• branches/ppalmers-streaming/src/libstreaming/util/IsoHandlerManager.h (modified) (3 diffs)

branches/ppalmers-streaming/src/dice/dice_avdevice.cpp

@@ -448 +448 @@
-DiceAvDevice::prepare() {
-    // prepare receive SP's
-
+
+
-        fb_quadlet_t nb_audio;
-        fb_quadlet_t nb_midi;
@@ -534 +535 @@
-
-    // prepare transmit SP's
-
+
+
-        fb_quadlet_t nb_audio;
-        fb_quadlet_t nb_midi;

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

@@ -97 +97 @@
- * @param cycle 
- * @param dropped 
-true if this is a valid packet, false if not
-
-
+
+
+
-AmdtpReceiveStreamProcessor::processPacketHeader(unsigned char *data, unsigned int length,
-                  unsigned char channel, unsigned char tag, unsigned char sy,
@@ -106 +106 @@
-    struct iec61883_packet *packet = (struct iec61883_packet *) data;
-    assert(packet);
-retval
+ok
-                  (packet->fdf != 0xFF) &&
-                  (packet->fmt == 0x10) &&
-                  (packet->dbs > 0) &&
-                  (length >= 2*sizeof(quadlet_t));
-retval
+ok
-        uint64_t now = m_handler->getCycleTimer();
-        //=> convert the SYT to a full timestamp in ticks
@@ -117 +117 @@
-                                              cycle, now);
-    }
-retval
+(ok ? eCRV_OK : eCRV_Invalid )
-}
-
@@ -130 +130 @@
- * @param cycle 
- * @param dropped 
-true if successful, false if xrun
-
-
+
+
+
-AmdtpReceiveStreamProcessor::processPacketData(unsigned char *data, unsigned int length,
-                  unsigned char channel, unsigned char tag, unsigned char sy,
@@ -158 +158 @@
-        // process all ports that should be handled on a per-packet base
-        // this is MIDI for AMDTP (due to the need of DBC)
-
-
+
+
+
+
-        }
-true
+eCRV_OK
-    } else {
-false
+eCRV_XRun
-    }
-}

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

@@ -78 +78 @@
-    virtual ~AmdtpReceiveStreamProcessor() {};
-
-bool
+enum eChildReturnValue
-                  unsigned char channel, unsigned char tag, unsigned char sy,
-                  unsigned int cycle, unsigned int dropped);
-bool
+enum eChildReturnValue
-                  unsigned char channel, unsigned char tag, unsigned char sy,
-                  unsigned int cycle, unsigned int dropped);

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

@@ -48 +48 @@
-{}
-
-
+
-AmdtpTransmitStreamProcessor::generatePacketHeader (
-    unsigned char *data, unsigned int *length,
@@ -94 +94 @@
-    // given by TRANSMIT_TRANSFER_DELAY (in ticks), but we can send
-    // packets early if we want to. (not completely according to spec)
-5
+2
-
-try_block_of_frames:
@@ -129 +129 @@
-    cycles_until_transmit = diffCycles ( transmit_at_cycle, cycle );
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
-    // two different options:
-    // 1) there are not enough frames for one packet
@@ -154 +155 @@
-                        fc, cycle, transmit_at_cycle, cycles_until_transmit );
-            // we are too late
-
-
-
-
-
+
-        }
-        else
@@ -166 +163 @@
-                        fc, cycle, transmit_at_cycle, cycles_until_transmit );
-            // there is still time left to send the packet
-
-
-
-
-
-
-
-
-
+
+
-        }
-    }
@@ -195 +185 @@
-        //         get next block of frames and repeat
-
-
-
-
-
-
-
-
+
-        {
-            // we are too late
- ( 
+(
-                        "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
@@ -214 +198 @@
-            // NOTE: dangerous since the device has no way of reporting that it didn't get
-            //       this packet on time.
- ( cycles_until_presentation <= min_cycles_before_presentation 
+(cycles_until_presentation >= min_cycles_before_presentation
-            {
-                // we are not that late and can still try to transmit the packet
- ( packet, length, m_last_timestamp 
-true
+(packet, length, m_last_timestamp
+eCRV_Packet
-            }
-            else   // definitely too late
-            {
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
-        }
-        else
@@ -250 +235 @@
-#endif
-            // we are too early, send only an empty packet
-false
-
-
-true
-
-
-
+eCRV_EmptyPacket
+
+
+eCRV_Invalid
+
+
+
-AmdtpTransmitStreamProcessor::generatePacketData (
-    unsigned char *data, unsigned int *length,
@@ -273 +258 @@
-        debugOutput ( DEBUG_LEVEL_VERY_VERBOSE, "XMIT DATA: TSP=%011llu (%04u)\n",
-                    cycle, m_last_timestamp, ( unsigned int ) TICKS_TO_CYCLES ( m_last_timestamp ) );
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
-AmdtpTransmitStreamProcessor::generateSilentPacketHeader (
-    unsigned char *data, unsigned int *length,
@@ -308 +291 @@
-
-    m_dbc += fillNoDataPacketHeader ( packet, length );
-true
-
-
-
+eCRV_OK
+
+
+
-AmdtpTransmitStreamProcessor::generateSilentPacketData (
-    unsigned char *data, unsigned int *length,
@@ -317 +300 @@
-    int cycle, unsigned int dropped, unsigned int max_length )
-{
-true
+eCRV_OK
-}
-

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

@@ -80 +80 @@
-    virtual ~AmdtpTransmitStreamProcessor() {};
-
-bool
+enum eChildReturnValue
-                              unsigned char *tag, unsigned char *sy,
-                              int cycle, unsigned int dropped, unsigned int max_length);
-bool
+enum eChildReturnValue
-                            unsigned char *tag, unsigned char *sy,
-                            int cycle, unsigned int dropped, unsigned int max_length);
-bool
+enum eChildReturnValue
-                                    unsigned char *tag, unsigned char *sy,
-                                    int cycle, unsigned int dropped, unsigned int max_length);
-bool
+enum eChildReturnValue
-                                  unsigned char *tag, unsigned char *sy,
-                                  int cycle, unsigned int dropped, unsigned int max_length);

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

@@ -41 +41 @@
-    , m_next_state( ePS_Invalid )
-    , m_cycle_to_switch_state( 0 )
-    , m_xruns( 0 )
-    , m_manager( NULL )
-    , m_ticks_per_frame( 0 )
-0
+-1
-    , m_sync_delay( 0 )
+    , m_in_xrun( false )
-    , m_last_timestamp(0)
-    , m_last_timestamp2(0)
@@ -166 +166 @@
-    unsigned int fc = m_data_buffer->getFrameCounter();
-    if (getType() == ePT_Receive) {
-fc >= (int) nbframes
+(fc >= nbframes)
-    } else {
-        // there has to be enough space to put the frames in
@@ -193 +193 @@
-                           unsigned char channel, unsigned char tag, unsigned char sy,
-                           unsigned int cycle, unsigned int dropped) {
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
-
-    // bypass based upon state
@@ -208 +212 @@
-        return RAW1394_ISO_DEFER;
-    }
-
-    // normal processing
-    enum raw1394_iso_disposition retval = RAW1394_ISO_OK;
-
-    // store the previous timestamp
@@ -254 +255 @@
-
-    // check the packet header
-
+
+
-        debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "RECV: CY=%04u TS=%011llu\n",
-                cycle, m_last_timestamp);
@@ -301 +303 @@
-            debugWarning("(%p) Correcting timestamp for dropped cycles, discarding packet...\n", this);
-            m_data_buffer->setBufferTailTimestamp(m_last_timestamp);
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
-            // execute the requested change
-            if (!updateState()) { // we are allowed to change the state directly
@@ -321 +316 @@
-            return RAW1394_ISO_DEFER;
-        }
+
+        // for all states that reach this we are allowed to
+        // do protocol specific data reception
+        enum eChildReturnValue result2 = processPacketData(data, length, channel, tag, sy, cycle, dropped_cycles);
+
+        // if an xrun occured, switch to the dryRunning state and
+        // allow for the xrun to be picked up
+        if (result2 == eCRV_XRun) {
+            m_in_xrun = true;
+            debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state to WaitingForStreamDisable due to data xrun\n");
+            m_cycle_to_switch_state = cycle+1; // switch in the next cycle
+            m_next_state = ePS_WaitingForStreamDisable;
+            // execute the requested change
+            if (!updateState()) { // we are allowed to change the state directly
+                debugError("Could not update state!\n");
+                return RAW1394_ISO_ERROR;
+            }
+            return RAW1394_ISO_DEFER;
+        } else if(result2 == eCRV_OK) {
+            // no problem here
+            return RAW1394_ISO_OK;
+        } else {
+            debugError("Invalid response\n");
+            return RAW1394_ISO_ERROR;
+        }
+    } else if(result == eCRV_Invalid) {
+        // apparently we don't have to do anything when the packets are not valid
+        return RAW1394_ISO_OK;
-    } else {
-
-
-
+
+
+
+
+
-}
-
@@ -338 +363 @@
-    }
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
-
-    // bypass based upon state
@@ -431 +461 @@
-    else if(m_state == ePS_Running) {
-        // check the packet header
-
+
+
-            debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "XMIT: CY=%04u TS=%011llu\n",
-                    cycle, m_last_timestamp);
@@ -450 +481 @@
-            }
-
-bool ok
+enum eChildReturnValue result2
-            // if an xrun occured, switch to the dryRunning state and
-            // allow for the xrun to be picked up
-
-
-
+
+
+
+
+
-                // execute the requested change
-                if (!updateState()) { // we are allowed to change the state directly
@@ -464 +497 @@
-            }
-            return RAW1394_ISO_OK;
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-        }
-    }
@@ -968 +1037 @@
-                m_data_buffer->setBufferTailTimestamp(m_last_timestamp);
-            } else {
-
-
+
-            }
-            break;
@@ -1058 +1126 @@
-            debugOutput(DEBUG_LEVEL_VERBOSE, "StreamProcessor %p started running at cycle %d\n", 
-                                             this, m_last_cycle);
-xruns = 0
+in_xrun = false
-            m_data_buffer->setTransparent(false);
-            break;
@@ -1306 +1374 @@
-                          (unsigned int)TICKS_TO_OFFSET(now));
-    }
-d\n", m_xruns
+s\n", (m_in_xrun ? "True":"False")
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  State                 : %s\n", ePSToString(m_state));
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "   Next state           : %s\n", ePSToString(m_next_state));

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

@@ -155 +155 @@
-
-protected: // the helper receive/transmit functions
+    enum eChildReturnValue {
+        eCRV_OK,
+        eCRV_Invalid,
+        eCRV_Packet,
+        eCRV_EmptyPacket,
+        eCRV_XRun,
+        eCRV_Again,
+    };
-    // to be implemented by the children
-    // the following methods are to be implemented by receive SP subclasses
-bool
+enum eChildReturnValue
-                                     unsigned char channel, unsigned char tag,
-                                     unsigned char sy, unsigned int cycle,
-                                     unsigned int dropped)
-false
-bool
+eCRV_Invalid
+enum eChildReturnValue
-                                   unsigned char channel, unsigned char tag,
-                                   unsigned char sy, unsigned int cycle,
-                                   unsigned int dropped)
-false
+eCRV_Invalid
-    virtual bool processReadBlock(char *data, unsigned int nevents, unsigned int offset)
-        {debugWarning("call not allowed\n"); return false;};
@@ -173 +181 @@
-
-    // the following methods are to be implemented by transmit SP subclasses
-bool
+enum eChildReturnValue
-                                      unsigned char *tag, unsigned char *sy,
-                                      int cycle, unsigned int dropped,
-                                      unsigned int max_length)
-false
-bool
+eCRV_Invalid
+enum eChildReturnValue
-                                    unsigned char *tag, unsigned char *sy,
-                                    int cycle, unsigned int dropped,
-                                    unsigned int max_length)
-false
-bool
+eCRV_Invalid
+enum eChildReturnValue
-                                            unsigned char *tag, unsigned char *sy,
-                                            int cycle, unsigned int dropped,
-                                            unsigned int max_length)
-false
-bool
+eCRV_Invalid
+enum eChildReturnValue
-                                          unsigned char *tag, unsigned char *sy,
-                                          int cycle, unsigned int dropped,
-                                          unsigned int max_length)
-false
+eCRV_Invalid
-    virtual bool processWriteBlock(char *data, unsigned int nevents, unsigned int offset)
-        {debugWarning("call not allowed\n"); return false;};
@@ -207 +215 @@
-
-    // move to private?
-(m_xruns>0); }; // FIXME: m_xruns not updated
+m_in_xrun; };
-
-protected: // FIXME: move to private
@@ -232 +240 @@
-
-protected:
-    unsigned int m_xruns;
-
-    StreamProcessorManager *m_manager;
-
@@ -367 +373 @@
-        int m_last_cycle;
-        int m_sync_delay;
+    private:
+        bool m_in_xrun;
-
-protected: // SPM related

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

@@ -158 +158 @@
-    }
-    m_isoManager->setVerboseLevel(getDebugLevel());
+    m_isoManager->setTransmitBufferNbPeriods(getNbBuffers() - 1);
+
-    if(!m_isoManager->init()) {
-        debugFatal("Could not initialize IsoHandlerManager\n");
@@ -243 +245 @@
-            it != m_ReceiveProcessors.end();
-            ++it ) {
-
-
-
+
+
+
+
+
-        }
-    }
@@ -251 +255 @@
-            it != m_TransmitProcessors.end();
-            ++it ) {
-
-
-
+
+
+
+
+
-        }
-    }
@@ -621 +627 @@
-bool StreamProcessorManager::waitForPeriod() {
-    int time_till_next_period;
-=
+ = 
-
-    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "enter...\n");
@@ -649 +655 @@
-            xrun_occurred |= (*it)->xrunOccurred();
-        }
-
-        if(xrun_occurred) break;
-
-        // check if we were waked up too soon
-
-
-
-
-
-
-
-
+
+
-
-    // we save the 'ideal' time of the transfer at this point,
@@ -672 +671 @@
-        m_time_of_transfer);
-
+    // normally we can transfer frames at this time, but in some cases this is not true
+    // e.g. when there are not enough frames in the receive buffer.
+    // however this doesn't have to be a problem, since we can wait some more until we
+    // have enough frames. There is only a problem once the ISO xmit doesn't have packets
+    // to transmit, or if the receive buffer overflows. These conditions are signaled by
+    // the iso threads
+    // check if xruns occurred on the Iso side.
+    // also check if xruns will occur should we transfer() now
+    #ifdef DEBUG
+    int waited = -1;
+    #endif
+    bool ready_for_transfer = false;
+    xrun_occurred = false;
+    while (!ready_for_transfer && !xrun_occurred) {
+        ready_for_transfer = true;
+        for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
+            it != m_ReceiveProcessors.end();
+            ++it ) {
+            ready_for_transfer &= ((*it)->canClientTransferFrames(m_period));
+            xrun_occurred |= (*it)->xrunOccurred();
+        }
+        for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
+            it != m_TransmitProcessors.end();
+            ++it ) {
+            ready_for_transfer &= ((*it)->canClientTransferFrames(m_period));
+            xrun_occurred |= (*it)->xrunOccurred();
+        }
+        usleep(125); // MAGIC: one cycle sleep...
+        #ifdef DEBUG
+        waited++;
+        #endif
+    } // we are either ready or an xrun occurred
+
+    #ifdef DEBUG
+    if(waited > 0) {
+        debugOutput(DEBUG_LEVEL_VERBOSE, "Waited %d x 125us due to SP not ready for transfer\n", waited);
+    }
+    #endif
+
+    // this is to notify the client of the delay that we introduced by waiting
+    m_delayed_usecs = - m_SyncSource->getTimeUntilNextPeriodSignalUsecs();
+    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "delayed for %d usecs...\n", m_delayed_usecs);
+
-#ifdef DEBUG
-    int rcv_bf=0, xmt_bf=0;
@@ -687 +729 @@
-        m_time_of_transfer, rcv_bf, xmt_bf, rcv_bf+xmt_bf);
-
-#endif
-
-    xrun_occurred=false;
-
-    // check if xruns occurred on the Iso side.
-    // also check if xruns will occur should we transfer() now
-
-    for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
-          it != m_ReceiveProcessors.end();
-          ++it ) {
-
-
-
-
-
-
-
+
-        if ((*it)->xrunOccurred()) {
-
+side 
-            (*it)->dumpInfo();
-        }
-        if (!((*it)->canClientTransferFrames(m_period))) {
-
+side 
-            (*it)->dumpInfo();
-        }
+    }
+    for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
+          it != m_TransmitProcessors.end();
+          ++it ) {
+        if ((*it)->xrunOccurred()) {
+            debugWarning("Xrun on XMIT SP %p due to ISO side xrun\n",*it);
+        }
+        if (!((*it)->canClientTransferFrames(m_period))) {
+            debugWarning("Xrun on XMIT SP %p due to buffer side xrun\n",*it);
+        }
+    }
-#endif
-
-    }
-    for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
-          it != m_TransmitProcessors.end();
-          ++it ) {
-        // a xrun has occurred on the Iso side
-        xrun_occurred |= (*it)->xrunOccurred();
-
-        // if this is true, a xrun will occur
-        xrun_occurred |= !((*it)->canClientTransferFrames(m_period));
-
-#ifdef DEBUG
-        if ((*it)->xrunOccurred()) {
-            debugWarning("Xrun on XMIT SP %p due to ISO xrun\n",*it);
-        }
-        if (!((*it)->canClientTransferFrames(m_period))) {
-            debugWarning("Xrun on XMIT SP %p due to buffer xrun\n",*it);
-        }
-#endif
-    }
-
-    m_nbperiods++;
-
-    // now we can signal the client that we are (should be) ready
-    return !xrun_occurred;

branches/ppalmers-streaming/src/libstreaming/util/IsoHandlerManager.cpp

@@ -41 +41 @@
-   m_State(E_Created),
-   m_poll_timeout(100), m_poll_fds(0), m_poll_nfds(0),
-
+, m_xmit_nb_periods( 1 )
-{
-
@@ -49 +49 @@
-   m_State(E_Created),
-   m_poll_timeout(1), m_poll_fds(0), m_poll_nfds(0),
-
+, m_xmit_nb_periods( 1 )
-{
-
@@ -400 +400 @@
-    if (stream->getStreamType()==IsoStream::eST_Transmit) {
-        // setup the optimal parameters for the raw1394 ISO buffering
-=
+ = 
-
-#if 1
@@ -410 +410 @@
-        unsigned int max_packet_size=MINIMUM_INTERRUPTS_PER_PERIOD * getpagesize() / packets_per_period;
-        if (max_packet_size < stream->getMaxPacketSize()) {
-=
+ = 
-        }
-
@@ -418 +418 @@
-                    max_packet_size = getpagesize();
-
-=
+ = 
-         if(irq_interval <= 0) irq_interval=1;
-#else
@@ -427 +427 @@
-
-        // configure it such that we have an irq for every PACKETS_PER_INTERRUPT packets
-=
+ = 
-
-        // unless the period size doesn't allow this
-        if ((packets_per_period/MINIMUM_INTERRUPTS_PER_PERIOD) < irq_interval) {
-=
+ = 
-        }
-
-        // FIXME: test
-=
+ = 
-#warning Using fixed irq_interval
-
-=
+ = 
-
-        if (max_packet_size < stream->getMaxPacketSize()) {
-=
+ = 
-        }
-
@@ -458 +458 @@
-        // buffers get transfered, meaning that we should have at least some
-        // margin here
-
-
-
-
-
-
-
+
+
+
+
-
-        // create the actual handler

branches/ppalmers-streaming/src/libstreaming/util/IsoHandlerManager.h

@@ -43 +43 @@
-namespace Streaming
-{
-
-class IsoHandler;
-class IsoStream;
@@ -78 +77 @@
-        void setPollTimeout(int t) {m_poll_timeout=t;}; ///< set the timeout used for poll()
-        int getPollTimeout() {return m_poll_timeout;};  ///< get the timeout used for poll()
+
+        void setTransmitBufferNbPeriods(unsigned int t) {m_xmit_nb_periods = t;};
+        int getTransmitBufferNbPeriods() {return m_xmit_nb_periods;};
-
-        void setVerboseLevel(int l); ///< set the verbose level
@@ -151 +153 @@
-        Util::PosixThread *m_isoManagerThread;
-
+        // the preferred number of periods to buffer on xmit
+        unsigned int m_xmit_nb_periods;
+
-        // debug stuff
-        DECLARE_DEBUG_MODULE;