Changeset 803

Timestamp:

01/02/08 14:11:58 (16 years ago)

Author:

ppalmers

Message:

more reliable streaming. hackish, but a start for a better implementation

Files:

• trunk/libffado/config.h.in (modified) (2 diffs)
• trunk/libffado/src/libieee1394/IsoHandler.cpp (modified) (3 diffs)
• trunk/libffado/src/libieee1394/IsoHandler.h (modified) (1 diff)
• trunk/libffado/src/libieee1394/IsoHandlerManager.cpp (modified) (5 diffs)
• trunk/libffado/src/libstreaming/generic/StreamProcessor.cpp (modified) (4 diffs)
• trunk/libffado/src/libstreaming/generic/StreamProcessor.h (modified) (3 diffs)
• trunk/libffado/src/libstreaming/StreamProcessorManager.cpp (modified) (1 diff)
• trunk/libffado/src/libutil/TimestampedBuffer.cpp (modified) (8 diffs)
• trunk/libffado/src/libutil/TimestampedBuffer.h (modified) (4 diffs)

trunk/libffado/config.h.in

@@ -41 +41 @@
-#define MAX_ISO_XMIT_BUFFER_FILL_PCT    50
-
-0
+1
-#define ISOHANDLER_USE_POLL             0
-
@@ -51 +51 @@
-// time to a later time instant also causes the xmit buffer fill to be
-// lower on average.
-4
+6
-
-#define STREAMPROCESSORMANAGER_CYCLES_FOR_DRYRUN            40000

trunk/libffado/src/libieee1394/IsoHandler.cpp

@@ -158 +158 @@
-
-bool
-
+
+
-    debugOutput( DEBUG_LEVEL_VERBOSE, "%p: Init thread...\n", this);
-    m_poll_fd.fd = getFileDescriptor();
@@ -171 +172 @@
-
-bool
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "%p: Execute thread...\n", this);
+
-    // bypass if not running
-    if (m_State != E_Running) {
@@ -183 +208 @@
-    }
-
+    // wait for the availability of frames in the client
+    // (blocking)
+    if (getType()==eHT_Receive || waitForClient()) {
+
-#if ISOHANDLER_USE_POLL
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-    } else {
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
-}
-

trunk/libffado/src/libieee1394/IsoHandler.h

@@ -126 +126 @@
-    bool unregisterStream(Streaming::StreamProcessor *);
-
+    bool waitForClient();
+    bool tryWaitForClient();
+
-private:
-    IsoHandlerManager& m_manager;

trunk/libffado/src/libieee1394/IsoHandlerManager.cpp

@@ -140 +140 @@
-    unsigned int m_poll_timeout = 100;
-
-
-
-
-
-
-
+
-    // bypass if no handlers are registered
-    if (m_poll_nfds_shadow == 0) {
+        debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "bypass iterate since no handlers registered\n");
-        usleep(m_poll_timeout * 1000);
-        return true;
@@ -182 +178 @@
-
-        if(m_poll_fds_shadow[i].revents & (POLLIN)) {
-            m_IsoHandler_map_shadow[i]->iterate();
-            if (m_IsoHandler_map_shadow[i]->getType() == IsoHandler::eHT_Receive) {
+                m_IsoHandler_map_shadow[i]->iterate();
-                nb_rcv++;
-            } else {
-
+
+
+
+
+
-            }
-        }
-    }
-    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,
@@ -209 +209 @@
-
-#if ISOHANDLER_PER_HANDLER_THREAD
+    // the IsoHandlers will create their own thread.
-#else
-    // create a thread to iterate our handlers
@@ -274 +275 @@
-    handler->setVerboseLevel(getDebugLevel());
-    m_IsoHandlers.push_back(handler);
+    updateShadowVars();
-    return true;
-}
@@ -288 +290 @@
-        if ( *it == handler ) {
-            m_IsoHandlers.erase(it);
+            updateShadowVars();
-            return true;
-        }

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

@@ -110 +110 @@
-StreamProcessor::getNbPacketsIsoXmitBuffer()
-{
+#if ISOHANDLER_PER_HANDLER_THREAD
+    // if we use one thread per packet, we can put every frame into the ISO buffer
+    // the waitForClient in IsoHandler will take care of the fact that the frames are
+    // not present in time
+    unsigned int packets_to_prebuffer = (getPacketsPerPeriod() * (m_StreamProcessorManager.getNbBuffers()-1));
+    debugOutput(DEBUG_LEVEL_VERBOSE, "Nominal prebuffer: %u\n", packets_to_prebuffer);
+    return packets_to_prebuffer;
+#else
-    // the target is to have all of the transmit buffer (at period transfer) as ISO packets
-    // when one period is received, there will be approx (NbBuffers - 1) * period_size frames
@@ -133 +141 @@
-    
-    return packets_to_prebuffer;
+#endif
-}
-
@@ -762 +771 @@
-    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);
@@ -807 +817 @@
-    }
-    return true;
+}
+
+bool
+StreamProcessor::waitForFrames()
+{
+    if(m_state == ePS_Running) {
+        assert(m_data_buffer);
+        if(getType() == ePT_Receive) {
+            return m_data_buffer->waitForFrames(m_StreamProcessorManager.getPeriodSize());
+        } else {
+            return m_data_buffer->waitForFrames(getNominalFramesPerPacket());
+        }
+    } else {
+        // when we're not running, we can always provide frames
+        debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "Not running...\n");
+        return true;
+    }
+}
+
+bool
+StreamProcessor::tryWaitForFrames()
+{
+    if(m_state == ePS_Running) {
+        assert(m_data_buffer);
+        if(getType() == ePT_Receive) {
+            return m_data_buffer->tryWaitForFrames(m_StreamProcessorManager.getPeriodSize());
+        } else {
+            return m_data_buffer->tryWaitForFrames(getNominalFramesPerPacket());
+        }
+    } else {
+        // when we're not running, we can always provide frames
+        debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "Not running...\n");
+        return true;
+    }
-}
-

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

@@ -34 +34 @@
-
-#include "debugmodule/debugmodule.h"
-
-#include <pthread.h>
-
-class Ieee1394Service;
@@ -160 +158 @@
-
-    /**
+     * @brief waits for the availability of frames (blocking)
+     * @param nframes number of frames
+     *
+     * @return true if frames are available, false if not (e.g. signal occurred)
+     */
+    bool waitForFrames();
+
+    /**
+     * @brief waits for the availability of frames (non-blocking)
+     * @param nframes number of frames
+     *
+     * @return true if frames are available, false if not
+     */
+    bool tryWaitForFrames();
+
+    /**
-     * @brief drop nframes from the internal buffer as if they were transferred to the client side
-     *
@@ -446 +460 @@
-    private:
-        bool m_in_xrun;
-
-public:
-    // debug stuff

trunk/libffado/src/libstreaming/StreamProcessorManager.cpp

@@ -666 +666 @@
-    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "enter...\n");
-
-=
+ = 
-
-    while(time_till_next_period > 0) {

trunk/libffado/src/libutil/TimestampedBuffer.cpp

@@ -52 +52 @@
-    pthread_mutex_unlock(&m_framecounter_lock); \
-    }
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-namespace Util {
-
@@ -77 +91 @@
-    ffado_ringbuffer_free(m_event_buffer);
-    free(m_cluster_buffer);
+    sem_destroy(&m_frame_semaphore);
-}
-
@@ -108 +123 @@
-    m_update_period=n;
-    return true;
+}
+
+/**
+ * \brief Get the nominal update period (in frames)
+ *
+ * Gets the nominal update period. This period is the number of frames
+ * between two timestamp updates (hence buffer writes)
+ *
+ * @return period in frames
+ */
+unsigned int TimestampedBuffer::getUpdatePeriod() {
+    return m_update_period;
-}
-
@@ -262 +289 @@
- */
-bool TimestampedBuffer::init() {
+    if (sem_init(&m_frame_semaphore, 0, 0) == -1) {
+        debugError("Could not init frame semaphore");
+        return false;
+    }
-    return true;
-}
@@ -340 +371 @@
-}
-
+bool
+TimestampedBuffer::waitForFrames(unsigned int nframes)
+{
+    int result;
+    do {
+        unsigned int bufferfill = getBufferFill();
+        if(bufferfill >= nframes) {
+            // first make the semaphore 0
+            while((result=sem_trywait(&m_frame_semaphore)) == 0) {};
+            return true;
+        } else {
+            debugOutput(DEBUG_LEVEL_VERY_VERBOSE,
+                        "only %d frames in buffer, waiting for more (%d)\n", 
+                        bufferfill, nframes);
+        }
+    } while((result=sem_wait(&m_frame_semaphore)) == 0);
+    debugOutput(DEBUG_LEVEL_VERBOSE,
+                "sem_wait returns: %d\n",
+                result);
+    return false;
+}
+
+bool
+TimestampedBuffer::tryWaitForFrames(unsigned int nframes)
+{
+    int result;
+    do {
+        unsigned int bufferfill = getBufferFill();
+        if(bufferfill >= nframes) {
+            // first make the semaphore 0
+            while((result=sem_trywait(&m_frame_semaphore)) == 0) {};
+            return true;
+        } else {
+            debugOutput(DEBUG_LEVEL_VERY_VERBOSE,
+                        "only %d frames in buffer, waiting for more (%d)\n", 
+                        bufferfill, nframes);
+        }
+    } while((result=sem_trywait(&m_frame_semaphore)) == 0);
+    debugOutput(DEBUG_LEVEL_VERY_VERBOSE,
+                "sem_trywait returns: %d\n", 
+                result);
+    return false;
+}
+
+bool
+TimestampedBuffer::waitForFrames()
+{
+    return waitForFrames(m_update_period);
+}
+
+bool
+TimestampedBuffer::tryWaitForFrames()
+{
+    return tryWaitForFrames(m_update_period);
+}
+
+
-/**
- * @brief Insert a dummy frame to the head buffer
@@ -371 +459 @@
-    m_framecounter++;
-    EXIT_CRITICAL_SECTION;
-
+
+
-    return true;
-}
@@ -449 +538 @@
-    }
-
+    POST_SEMAPHORE;
-    return true;
-}
@@ -1220 +1310 @@
-    // ts(x) = m_buffer_tail_timestamp +
-    //         (m_buffer_next_tail_timestamp - m_buffer_tail_timestamp)/(samples_between_updates)*x
-
+
-}
-

trunk/libffado/src/libutil/TimestampedBuffer.h

@@ -27 +27 @@
-#include "../debugmodule/debugmodule.h"
-#include "libutil/ringbuffer.h"
+#include <semaphore.h>
-
-//typedef float ffado_timestamp_t;
@@ -89 +90 @@
-        virtual ~TimestampedBuffer();
-
+        /**
+         * @brief waits for the availability of frames (blocking)
+         * @param nframes number of frames
+         *
+         * @return true if frames are available, false if not (e.g. signal occurred)
+         */
+        bool waitForFrames(unsigned int nframes);
+
+        /**
+         * @brief waits for the availability of frames (blocking)
+         *
+         * waits for one update period of frames
+         *
+         * @return true if frames are available, false if not (e.g. signal occurred)
+         */
+        bool waitForFrames();
+
+        /**
+         * @brief waits for the availability of frames (non-blocking)
+         * @param nframes number of frames
+         *
+         * @return true if frames are available, false if not
+         */
+        bool tryWaitForFrames(unsigned int nframes);
+
+        /**
+         * @brief waits for the availability of frames (non-blocking)
+         *
+         * waits for one update period of frames
+         *
+         * @return true if frames are available, false if not
+         */
+        bool tryWaitForFrames();
+
-        bool writeDummyFrame();
-        bool dropFrames ( unsigned int nbframes );
@@ -152 +187 @@
-
-        bool setUpdatePeriod ( unsigned int t );
+        unsigned int getUpdatePeriod();
-
-        // misc stuff
@@ -208 +244 @@
-        float m_current_rate;
-        unsigned int m_update_period;
+
+        sem_t m_frame_semaphore;
-};
-