Changeset 759

Timestamp:

12/15/07 11:36:27 (13 years ago)

Author:

ppalmers

Message:

fix streaming problem

Files:

• trunk/libffado/src/libieee1394/IsoHandler.cpp (modified) (6 diffs)
• trunk/libffado/src/libieee1394/IsoHandler.h (modified) (2 diffs)
• trunk/libffado/src/libieee1394/IsoHandlerManager.cpp (modified) (12 diffs)
• trunk/libffado/src/libieee1394/IsoHandlerManager.h (modified) (3 diffs)
• trunk/libffado/src/libstreaming/generic/StreamProcessor.cpp (modified) (5 diffs)
• trunk/libffado/src/libstreaming/StreamProcessorManager.cpp (modified) (2 diffs)

trunk/libffado/src/libieee1394/IsoHandler.cpp

@@ -21 +21 @@
- *
- */
+
+//#define PER_HANDLER_THREAD
-
-#include "IsoHandler.h"
@@ -181 +183 @@
-    }
-
+    uint64_t poll_enter = m_manager.get1394Service().getCurrentTimeAsUsecs();
-    err = poll(&m_poll_fd, 1, m_poll_timeout);
+    uint64_t poll_exit = m_manager.get1394Service().getCurrentTimeAsUsecs();
-    if (err == -1) {
-        if (errno == EINTR) {
@@ -189 +193 @@
-        return false;
-    }
+    uint64_t iter_enter=0;
+    uint64_t iter_exit=0;
-    if(m_poll_fd.revents & (POLLIN)) {
-
+
+
-            debugOutput( DEBUG_LEVEL_VERBOSE,
-
-
-
-
+
+
+
+
+
-    } else {
-        if (m_poll_fd.revents & POLLERR) {
@@ -203 +211 @@
-            debugWarning("hangup on fd for %p\n",this);
-        }
+    }
+    debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "(%c %p) poll took %lldus, iterate took %lldus\n", 
+                (this->getType()==eHT_Receive?'R':'X'), this, 
+                poll_exit-poll_enter, iter_exit-iter_enter);
+    return true;
+}
+
+bool
+IsoHandler::iterate() {
+    if(raw1394_loop_iterate(m_handle)) {
+        debugOutput( DEBUG_LEVEL_VERBOSE,
+                    "IsoHandler (%p): Failed to iterate handler: %s\n",
+                    this,strerror(errno));
+        return false;
-    }
-    return true;
@@ -257 +279 @@
-    }
-
+#ifdef THREAD_PER_ISOHANDLER
-    // create a thread to iterate ourselves
-    debugOutput( DEBUG_LEVEL_VERBOSE, "Start thread for %p...\n", this);
@@ -269 +292 @@
-        return false;
-    }
+#endif
-
-    // update the internal state

trunk/libffado/src/libieee1394/IsoHandler.h

@@ -84 +84 @@
-    bool Init();
-    bool Execute();
+    bool iterate();
+
-    int getFileDescriptor() { return raw1394_get_fd(m_handle);};
-    bool setThreadParameters(bool rt, int priority);
@@ -90 +92 @@
-    bool prepare();
-
-    bool iterate();
-    void setVerboseLevel(int l);
-

trunk/libffado/src/libieee1394/IsoHandlerManager.cpp

@@ -29 +29 @@
-#include "libutil/Atomic.h"
-
+#include "libutil/PosixThread.h"
+
-#include <assert.h>
-
-8
+4
-
-IMPL_DEBUG_MODULE( IsoHandlerManager, IsoHandlerManager, DEBUG_LEVEL_NORMAL );
@@ -41 +43 @@
-   , m_service( service )
-   , m_realtime(false), m_priority(0)
+   , m_Thread ( NULL )
-{}
-
@@ -47 +50 @@
-   , m_service( service )
-   , m_realtime(run_rt), m_priority(rt_prio)
+   , m_Thread ( NULL )
-{}
-
@@ -52 +56 @@
-{
-    stopHandlers();
+    if (m_Thread) {
+        m_Thread->Stop();
+        delete m_Thread;
+    }
-}
-
@@ -67 +75 @@
-        result &= (*it)->setThreadParameters(m_realtime, m_priority);
-    }
+
+    if (m_Thread) {
+        if (m_realtime) {
+            m_Thread->AcquireRealTime(m_priority);
+        } else {
+            m_Thread->DropRealTime();
+        }
+    }
+
-    return result;
+}
+
+/**
+ * Update the shadow variables. Should only be called from
+ * the iso handler iteration thread
+ */
+void
+IsoHandlerManager::updateShadowVars()
+{
+    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "updating shadow vars...\n");
+    unsigned int i;
+    m_poll_nfds_shadow = m_IsoHandlers.size();
+    if(m_poll_nfds_shadow > FFADO_MAX_ISO_HANDLERS_PER_PORT) {
+        debugWarning("Too much ISO Handlers in manager...\n");
+        m_poll_nfds_shadow = FFADO_MAX_ISO_HANDLERS_PER_PORT;
+    }
+    for (i = 0; i < m_poll_nfds_shadow; i++) {
+        IsoHandler *h = m_IsoHandlers.at(i);
+        assert(h);
+        m_IsoHandler_map_shadow[i] = h;
+
+        m_poll_fds_shadow[i].fd = h->getFileDescriptor();
+        m_poll_fds_shadow[i].revents = 0;
+        if (h->isEnabled()) {
+            m_poll_fds_shadow[i].events = POLLIN;
+        } else {
+            m_poll_fds_shadow[i].events = 0;
+        }
+    }
+    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, " updated shadow vars...\n");
+}
+
+bool
+IsoHandlerManager::Init() {
+    debugOutput( DEBUG_LEVEL_VERBOSE, "%p: Init thread...\n", this);
+    bool result = true;
+    for ( IsoHandlerVectorIterator it = m_IsoHandlers.begin();
+        it != m_IsoHandlers.end();
+        ++it )
+    {
+        result &= (*it)->Init();
+    }
+    return result;
+}
+
+bool
+IsoHandlerManager::Execute() {
+    int err;
+    unsigned int i;
+
+    unsigned int m_poll_timeout = 100;
+
+    // update the shadow variables if requested
+   // if(m_request_fdmap_update) {
+        updateShadowVars();
+    //    ZERO_ATOMIC((SInt32*)&m_request_fdmap_update);
+    //}
+
+    // bypass if no handlers are registered
+    if (m_poll_nfds_shadow == 0) {
+        usleep(m_poll_timeout * 1000);
+        return true;
+    }
+
+    // Use a shadow map of the fd's such that the poll call is not in a critical section
+    uint64_t poll_enter = m_service.getCurrentTimeAsUsecs();
+    err = poll (m_poll_fds_shadow, m_poll_nfds_shadow, m_poll_timeout);
+    uint64_t poll_exit = m_service.getCurrentTimeAsUsecs();
+
+    if (err == -1) {
+        if (errno == EINTR) {
+            return true;
+        }
+        debugFatal("poll error: %s\n", strerror (errno));
+        return false;
+    }
+
+    int nb_rcv = 0;
+    int nb_xmit = 0;
+    uint64_t iter_enter = m_service.getCurrentTimeAsUsecs();
+    for (i = 0; i < m_poll_nfds_shadow; i++) {
+        if(m_poll_fds_shadow[i].revents) {
+            debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "received events: %08X for (%p)\n",
+                m_poll_fds_shadow[i].revents, m_IsoHandler_map_shadow[i]);
+        }
+        if (m_poll_fds_shadow[i].revents & POLLERR) {
+            debugWarning("error on fd for %d\n",i);
+        }
+
+        if (m_poll_fds_shadow[i].revents & POLLHUP) {
+            debugWarning("hangup on fd for %d\n",i);
+        }
+
+        if(m_poll_fds_shadow[i].revents & (POLLIN)) {
+            m_IsoHandler_map_shadow[i]->iterate();
+            if (m_IsoHandler_map_shadow[i]->getType() == IsoHandler::eHT_Receive) {
+                nb_rcv++;
+            } else {
+                nb_xmit++;
+            }
+        }
+    }
+    uint64_t iter_exit = m_service.getCurrentTimeAsUsecs();
+    
+    debugOutput(DEBUG_LEVEL_VERY_VERBOSE, " poll took %6lldus, iterate took %6lldus, iterated (R: %2d, X: %2d) handlers\n",
+                poll_exit-poll_enter, iter_exit-iter_enter,
+                nb_rcv, nb_xmit);
+
+    return true;
-}
-
@@ -78 +204 @@
-        return false;
-    }
+
+#ifndef THREAD_PER_ISOHANDLER
+    // create a thread to iterate our handlers
+    debugOutput( DEBUG_LEVEL_VERBOSE, "Start thread for %p...\n", this);
+    m_Thread = new Util::PosixThread(this, m_realtime, m_priority, 
+                                     PTHREAD_CANCEL_DEFERRED);
+    if(!m_Thread) {
+        debugFatal("No thread\n");
+        return false;
+    }
+    if (m_Thread->Start() != 0) {
+        debugFatal("Could not start update thread\n");
+        return false;
+    }
+#endif
-
-    m_State=E_Running;
@@ -197 +338 @@
-        // NOTE: try and use MINIMUM_INTERRUPTS_PER_PERIOD hardware interrupts
-        //       per period for better latency.
-
+/2
-
-        if (max_packet_size < stream->getMaxPacketSize()) {
@@ -207 +348 @@
-        // Ensure we don't request a packet size bigger than the
-        // kernel-enforced maximum which is currently 1 page.
-
-Warning("max packet size (%u) > page size (%u)\n", max_packet_size ,(unsigned int)getpagesize()
-max_packet_size = getpagesize()
+/2
+Error("max packet size (%u) > page size (%u)\n", max_packet_size, (unsigned int)getpagesize()/2
+return false
-        }
-
@@ -224 +365 @@
-
-        // configure it such that we have an irq for every PACKETS_PER_INTERRUPT packets
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
-
-        if (max_packet_size < stream->getMaxPacketSize()) {
-
-
-
-
-
-
-
+
+
+
-
-#endif
@@ -255 +384 @@
-        h = new IsoHandler(*this, IsoHandler::eHT_Receive,
-                           buffers, max_packet_size, irq_interval);
+
+        debugOutput( DEBUG_LEVEL_VERBOSE, " creating IsoRecvHandler\n");
+
-        if(!h) {
-            debugFatal("Could not create IsoRecvHandler\n");
-            return false;
-        }
-        debugOutput( DEBUG_LEVEL_VERBOSE, " registering IsoRecvHandler\n");
-
-    } else if (stream->getType()==StreamProcessor::ePT_Transmit) {
@@ -265 +396 @@
-        unsigned int packets_per_period = stream->getPacketsPerPeriod();
-
+#if 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
@@ -295 +427 @@
-        //int buffers=30;
-        //max_packet_size = getpagesize(); // HACK
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-        // the SP specifies how many packets to buffer
-        int buffers = stream->getNbPacketsIsoXmitBuffer();
+
+        debugOutput( DEBUG_LEVEL_VERBOSE, " creating IsoXmitHandler\n");
-
-        // create the actual handler
-        h = new IsoHandler(*this, IsoHandler::eHT_Transmit,
-                           buffers, max_packet_size, irq_interval);
-
-        debugOutput( DEBUG_LEVEL_VERBOSE, " registering IsoXmitHandler\n");
-
-        if(!h) {

trunk/libffado/src/libieee1394/IsoHandlerManager.h

@@ -27 +27 @@
-#include "debugmodule/debugmodule.h"
-
+#include "libutil/Thread.h"
+
-#include <sys/poll.h>
-#include <errno.h>
-
-#include <vector>
+
+#define THREAD_PER_ISOHANDLER
-
-#define FFADO_MAX_ISO_HANDLERS_PER_PORT 16
@@ -61 +65 @@
-
-*/
-
+ : public Util::RunnableInterface
-{
-    friend class Streaming::StreamProcessorManager;
+    public:
+        bool Init();
+        bool Execute();
+        void updateShadowVars();
+    private:
+        // shadow variables
+        struct pollfd m_poll_fds_shadow[FFADO_MAX_ISO_HANDLERS_PER_PORT];
+        IsoHandler *m_IsoHandler_map_shadow[FFADO_MAX_ISO_HANDLERS_PER_PORT];
+        unsigned int m_poll_nfds_shadow;
-
-    public:
@@ -143 +156 @@
-        bool m_realtime;
-        int m_priority;
+        Util::Thread *  m_Thread;
-
-        // debug stuff

trunk/libffado/src/libstreaming/generic/StreamProcessor.cpp

@@ -145 +145 @@
-void
-StreamProcessor::setSyncDelay(int d) {
-
+Y_VER
-    m_sync_delay = d;
-}
@@ -234 +234 @@
-    if (m_last_cycle != (int)cycle && m_last_cycle != -1) {
-        dropped_cycles = diffCycles(cycle, m_last_cycle) - 1;
-
+
+
+
+
-        if (dropped_cycles > 0) {
-            debugWarning("(%p) dropped %d packets on cycle %u, 'dropped'=%u, cycle=%d, m_last_cycle=%d\n",
-                this, dropped_cycles, cycle, dropped, cycle, m_last_cycle);
-            m_dropped += dropped_cycles;
+            m_in_xrun = true;
-        }
-    }
@@ -410 +414 @@
-    if (m_last_cycle != cycle && m_last_cycle != -1) {
-        dropped_cycles = diffCycles(cycle, m_last_cycle) - 1;
-
+
+
+
+
-        if (dropped_cycles > 0) {
-            debugWarning("(%p) dropped %d packets on cycle %u (last_cycle=%u, dropped=%d)\n", this, dropped_cycles, cycle, m_last_cycle, dropped);
@@ -508 +515 @@
-        // check the packet header
-        enum eChildReturnValue result = generatePacketHeader(data, length, tag, sy, cycle, dropped_cycles, max_length);
-
+ || result == eCRV_Defer
-            debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "XMIT: CY=%04u TS=%011llu\n",
-                    cycle, m_last_timestamp);
@@ -544 +551 @@
-            // skip queueing packets if we detect that there are not enough frames
-            // available
-
+ || result == eCRV_Defer
-                return RAW1394_ISO_DEFER;
-            else

trunk/libffado/src/libstreaming/StreamProcessorManager.cpp

@@ -733 +733 @@
-            usleep(125); // MAGIC: one cycle sleep...
-
-            #if 0
-            // in order to avoid this in the future, we increase the sync delay of the sync source SP
-            int d = m_SyncSource->getSyncDelay() + TICKS_PER_CYCLE;
-            m_SyncSource->setSyncDelay(d);
-            #endif
-
-            #ifdef DEBUG
@@ -744 +742 @@
-        }
-    } // we are either ready or an xrun occurred
+
+    // in order to avoid a runaway value of the sync delay, we gradually decrease
+    // it. It will be increased by a 'too early' event (cfr some lines higher)
+    // hence we'll be at a good point on average.
+    int d = m_SyncSource->getSyncDelay() - 1;
+    if (d >= 0) m_SyncSource->setSyncDelay(d);
+
-
-    #ifdef DEBUG