Changeset 1005

Timestamp:

04/21/08 01:27:47 (13 years ago)

Author:

ppalmers

Message:

Improve thread synchronisation. Switch back to separate threads for transmit and
receive since it is not possible to statically schedule things properly. One
of the threads (i.e. the client thread) is out of our control, hence it's
execution can't be controlled. Using separate threads and correct priorities
will shift this problem to the OS. Note that the priority of the packet
receive thread should be lower than the client thread (such that the client
thread is woken ASAP), and the priority of the transmit thread should be
higher than the client thread (such that packets are queued ASAP).
Extra benefit: multi-cores are used.

Some other startup improvements.

Files:

• trunk/libffado/config.h.in (modified) (2 diffs)
• trunk/libffado/src/libieee1394/IsoHandler.cpp (modified) (6 diffs)
• trunk/libffado/src/libieee1394/IsoHandler.h (modified) (5 diffs)
• trunk/libffado/src/libieee1394/IsoHandlerManager.cpp (modified) (28 diffs)
• trunk/libffado/src/libieee1394/IsoHandlerManager.h (modified) (8 diffs)
• trunk/libffado/src/libstreaming/generic/StreamProcessor.cpp (modified) (30 diffs)
• trunk/libffado/src/libstreaming/generic/StreamProcessor.h (modified) (3 diffs)
• trunk/libffado/src/libstreaming/StreamProcessorManager.cpp (modified) (11 diffs)
• trunk/libffado/src/libstreaming/StreamProcessorManager.h (modified) (2 diffs)

trunk/libffado/config.h.in

@@ -80 +80 @@
-#define ISOHANDLERMANAGER_MAX_STREAMS_PER_ISOTHREAD         16
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-// allows to add some processing margin. This shifts the time
@@ -97 +110 @@
-#define STREAMPROCESSORMANAGER_SYNCSTART_TRIES              10
-#define STREAMPROCESSORMANAGER_SYNC_WAIT_TIME_MSEC          200
-2
+4
-#define STREAMPROCESSORMANAGER_NB_ALIGN_TRIES               40
-

trunk/libffado/src/libieee1394/IsoHandler.cpp

@@ -26 +26 @@
-#include "IsoHandler.h"
-#include "ieee1394service.h" 
+#include "IsoHandlerManager.h"
-
-#include "libstreaming/generic/StreamProcessor.h"
@@ -91 +92 @@
-   , m_speed( RAW1394_ISO_SPEED_400 )
-   , m_prebuffers( 0 )
+   , m_dont_exit_iterate_loop( true )
-   , m_State( E_Created )
-#ifdef DEBUG
@@ -150 +152 @@
-
-bool
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
-    if(m_Client) {
-        bool result;
@@ -186 +169 @@
-    return false;
-}
-/*
-bool
-IsoHandler::Execute()
-{
-    debugOutputExtreme( DEBUG_LEVEL_VERY_VERBOSE, "%p: Execute thread...\n", this);
-
-    // bypass if not running
-    if (m_State != E_Running) {
-        debugOutput( DEBUG_LEVEL_VERBOSE, "%p: not polling since not running...\n", this);
-        usleep(m_poll_timeout * 1000);
-        debugOutput( DEBUG_LEVEL_VERBOSE, "%p: done sleeping...\n", this);
-        return true;
-    }
-
-    // wait for the availability of frames in the client
-    // (blocking for transmit handlers)
-    debugOutputExtreme(DEBUG_LEVEL_VERY_VERBOSE, "(%p, %s) Waiting for Client activity...\n", this, getTypeString());
-    if (waitForClient()) {
-#if ISOHANDLER_USE_POLL
-        bool result = true;
-        while(result && m_Client && tryWaitForClient()) {
-            int err = poll(&m_poll_fd, 1, m_poll_timeout);
-            if (err == -1) {
-                if (errno == EINTR) {
-                    return true;
-                }
-                debugFatal("%p, poll error: %s\n", this, strerror (errno));
-                return false;
-            }
-
-            if(m_poll_fd.revents & (POLLIN)) {
-                result=iterate();
-                if(!result) {
-                    debugOutput( DEBUG_LEVEL_VERBOSE,
-                                "IsoHandler (%p): Failed to iterate handler\n",
-                                this);
-                }
-            } else {
-                if (m_poll_fd.revents & POLLERR) {
-                    debugWarning("error on fd for %p\n", this);
-                }
-                if (m_poll_fd.revents & POLLHUP) {
-                    debugWarning("hangup on fd for %p\n",this);
-                }
-                break;
-            }
-        }
-        return result;
-#else
-        // iterate() is blocking if no 1394 data is available
-        // so poll'ing is not really necessary
-        bool result = true;
-        while(result && m_Client && tryWaitForClient()) {
-            result = iterate();
-//             if (getType() == eHT_Receive) {
-//                 debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "(%p, %s) Iterate returned: %d\n",
-//                             this, (m_type==eHT_Receive?"Receive":"Transmit"), result);
-//             }
-        }
-        return result;
-#endif
-    } else {
-        debugError("waitForClient() failed.\n");
-        return false;
-    }
-}*/
-
-bool
@@ -440 +357 @@
-    #endif
-    if(m_Client) {
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-    }
-
@@ -467 +397 @@
-        }
-        #endif
-
+
+
+
+
+
+
+
+
+
+
+
+
+
-    }
-    *tag = 0;

trunk/libffado/src/libieee1394/IsoHandler.h

@@ -26 +26 @@
-
-#include "debugmodule/debugmodule.h"
-#include "IsoHandlerManager.h"
-
-#include "libutil/Thread.h"
@@ -32 +31 @@
-enum raw1394_iso_disposition;
-
+class IsoHandlerManager;
-namespace Streaming {
-    class StreamProcessor;
@@ -121 +121 @@
-    bool unregisterStream(Streaming::StreamProcessor *);
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-private:
@@ -132 +146 @@
-    int             m_irq_interval;
-
-
+ // FIXME: implement with functors
-
-    int handleBusReset(unsigned int generation);
@@ -140 +154 @@
-    enum raw1394_iso_speed m_speed;
-    unsigned int m_prebuffers;
+    bool m_dont_exit_iterate_loop;
-
-    // the state machine

trunk/libffado/src/libieee1394/IsoHandlerManager.cpp

@@ -25 +25 @@
-#include "IsoHandlerManager.h"
-#include "ieee1394service.h" 
-#include "IsoHandler.h"
-#include "libstreaming/generic/StreamProcessor.h"
-
@@ -42 +41 @@
-// --- ISO Thread --- //
-
-
+, enum IsoHandler::EHandlerType t
-    : m_manager( manager )
-    , m_SyncIsoHandler ( NULL )
-
+
+
+
+
+
+
+
-}
-
@@ -65 +70 @@
-    #endif
-
+    sem_init(&m_activity_semaphore, 0, 0);
-    return true;
-}
@@ -89 +95 @@
-        IsoHandler *h = m_manager.m_IsoHandlers.at(i);
-        assert(h);
+
+        // skip the handlers not intended for us
+        if(h->getType() != m_handlerType) continue;
-
-        if (h->isEnabled()) {
@@ -129 +138 @@
-{
-    debugOutputExtreme(DEBUG_LEVEL_VERY_VERBOSE,
-
+
+
-    int err;
-    unsigned int i;
@@ -139 +149 @@
-    if(diff < 100) {
-        debugOutputExtreme(DEBUG_LEVEL_VERY_VERBOSE,
-
-
+
+
+
-        m_successive_short_loops++;
-
+00
-            debugError("Shutting down runaway thread\n");
-            return false;
@@ -163 +174 @@
-    if (m_poll_nfds_shadow == 0) {
-        debugOutputExtreme(DEBUG_LEVEL_VERY_VERBOSE,
-
-
+, %s
+, (m_handlerType == IsoHandler::eHT_Transmit? "Transmit": "Receive")
-        usleep(m_poll_timeout * 1000);
-        return true;
@@ -179 +190 @@
-            short events = 0;
-            IsoHandler *h = m_IsoHandler_map_shadow[i];
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
-            }
-            m_poll_fds_shadow[i].events = events;
@@ -203 +207 @@
-
-        if(no_one_to_poll) {
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-        }
-    }
-
-    // Use a shadow map of the fd's such that we don't have to update
-
-
+
-    err = poll (m_poll_fds_shadow, m_poll_nfds_shadow, m_poll_timeout);
-
-    if (err < 0) {
-        if (errno == EINTR) {
+            debugOutput(DEBUG_LEVEL_VERBOSE, "Ignoring poll return due to signal\n");
-            return true;
-        }
@@ -247 +249 @@
-        #ifdef DEBUG
-        if(m_poll_fds_shadow[i].revents) {
-
-
-
+
+
+
+
-                        i, m_poll_nfds_shadow,
-                        m_IsoHandler_map_shadow[i],
@@ -283 +286 @@
-    }
-    return true;
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-}
-
@@ -295 +360 @@
-   , m_service( service )
-   , m_realtime(false), m_priority(0)
-
-
-
+
+
+
+
+
+
-
-IsoHandlerManager::IsoHandlerManager(Ieee1394Service& service, bool run_rt, int rt_prio)
@@ -303 +371 @@
-   , m_service( service )
-   , m_realtime(run_rt), m_priority(rt_prio)
-
-
-
+
+
+
+
+
+
-
-IsoHandlerManager::~IsoHandlerManager()
@@ -314 +385 @@
-        debugError("Still some handlers in use\n");
-    }
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
-    }
-}
@@ -326 +404 @@
-IsoHandlerManager::requestShadowMapUpdate()
-{
-
+
+
-}
-
@@ -336 +415 @@
-    m_priority = priority;
-
-
+Transmit
-        if (m_realtime) {
-
+
+
+
-        } else {
-
+
+
+
+
+
+
+
+
+
+
-        }
-    }
@@ -356 +446 @@
-    }
-
-a thread
-
- = new IsoTask( *this
-
+threads
+ transmit
+Transmit = new IsoTask( *this, IsoHandler::eHT_Transmit
+Transmit
-        debugFatal("No task\n");
-        return false;
-    }
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-        debugFatal("No thread\n");
-        return false;
@@ -374 +481 @@
-    Util::Watchdog *watchdog = m_service.getWatchdog();
-    if(watchdog) {
-
-
+
+
+
+
+
-        }
-    } else {
@@ -381 +491 @@
-    }
-
-
-
+
+
+
+
+
+
-        return false;
-    }
@@ -401 +515 @@
-        if ((*it) == h) {
-            result = h->disable();
-
+
+
+
+
+
-            debugOutput(DEBUG_LEVEL_VERY_VERBOSE, " disabled\n");
-            return result;
@@ -422 +540 @@
-        if ((*it) == h) {
-            result = h->enable();
-
+
+
+
+
+
-            debugOutput(DEBUG_LEVEL_VERY_VERBOSE, " enabled\n");
-            return result;
@@ -430 +552 @@
-    debugError("Handler not found\n");
-    return false;
+}
+
+void
+IsoHandlerManager::signalActivityTransmit()
+{
+    assert(m_IsoTaskTransmit);
+    m_IsoTaskTransmit->signalActivity();
+}
+
+void
+IsoHandlerManager::signalActivityReceive()
+{
+    assert(m_IsoTaskReceive);
+    m_IsoTaskReceive->signalActivity();
-}
-
@@ -438 +574 @@
-    handler->setVerboseLevel(getDebugLevel());
-    m_IsoHandlers.push_back(handler);
-
+
+
-}
-
@@ -452 +589 @@
-        if ( *it == handler ) {
-            m_IsoHandlers.erase(it);
-
+
+
-        }
-    }
@@ -690 +828 @@
-                return false;
-            }
-
+
+
+
+
+
+
+
-                debugOutput( DEBUG_LEVEL_VERBOSE, " could not update shadow map for handler (%p)\n",*it);
-                return false;
@@ -751 +895 @@
-                return false;
-            }
-
+
+
+
+
+
+
+
-                debugOutput( DEBUG_LEVEL_VERBOSE, " could not update shadow map for handler (%p)\n",*it);
-                return false;
@@ -782 +932 @@
-            retval=false;
-        }
-
+
+
+
+
+
+
+
-            debugOutput( DEBUG_LEVEL_VERBOSE, " could not update shadow map for handler (%p)\n",*it);
-val=
+urn 
-        }
-    }
@@ -816 +972 @@
-        (*it)->setVerboseLevel(i);
-    }
-
-
+
+
+
+
-}
-

trunk/libffado/src/libieee1394/IsoHandlerManager.h

@@ -26 +26 @@
-
-#include "config.h"
-
-#include "debugmodule/debugmodule.h"
-
-#include "libutil/Thread.h"
+
+#include "IsoHandler.h"
-
-#include <sys/poll.h>
-#include <errno.h>
-
-#include <vector>
+#include <semaphore.h>
-
-class Ieee1394Service;
-
+
+
-
-namespace Streaming {
@@ -57 +59 @@
-{
-    public:
-
- {}
+, enum IsoHandler::EHandlerType
+
-
-    public:
@@ -65 +67 @@
-
-        /**
-
+@brief 
-         */
-        bool requestShadowMapUpdate();
+        enum eActivityResult {
+            eAR_Activity,
+            eAR_Timeout,
+            eAR_Interrupted,
+            eAR_Error
+        };
+
+        /**
+         * @brief signals that something happened in one of the clients of this task
+         */
+        void signalActivity();
+        /**
+         * @brief wait until something happened in one of the clients of this task
+         */
+        enum eActivityResult waitForActivity();
-
-        void setVerboseLevel(int i);
@@ -92 +109 @@
-#endif
-
+        // activity signaling
+        sem_t m_activity_semaphore;
+
+        enum IsoHandler::EHandlerType m_handlerType;
-        // debug stuff
-        DECLARE_DEBUG_MODULE;
@@ -110 +131 @@
-class IsoHandlerManager
-{
-    friend class Streaming::StreamProcessorManager;
-    friend class IsoTask;
-    friend class IsoHandler;
-
-    public:
@@ -138 +157 @@
-        bool disable(IsoHandler *); ///< disables a handler
-        bool enable(IsoHandler *); ///< enables a handler
+
+        /**
+         * @brief signals that something happened in one of the clients
+         */
+        void signalActivityTransmit();
+        void signalActivityReceive();
+
-        ///> disables the handler attached to the stream
-        bool stopHandlerForStream(Streaming::StreamProcessor *);
@@ -157 +183 @@
-        Ieee1394Service& get1394Service() {return m_service;};
-
-    protected:
-        void requestShadowMapUpdate();
-
@@ -195 +220 @@
-        bool            m_realtime;
-        int             m_priority;
-
-
+
+
+
+
-
-        // debug stuff

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

@@ -40 +40 @@
-#include <math.h>
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
-}
-
@@ -71 +80 @@
-    , m_sync_delay( 0 )
-    , m_in_xrun( false )
+    , m_min_ahead( 7999 )
-{
-    // create the timestamped buffer and register ourselves as its client
-    m_data_buffer = new Util::TimestampedBuffer(this);
-    pthread_mutex_init(&m_activity_cond_lock, NULL);
-    pthread_cond_init(&m_activity_cond, NULL);
-}
-
@@ -84 +92 @@
-    }
-
-    // lock the condition mutex to keep threads from blocking on
-    // the condition var while we destroy it
-    pthread_mutex_lock(&m_activity_cond_lock);
-    // now signal activity, releasing threads that
-    // are already blocking on the condition variable
-    pthread_cond_broadcast(&m_activity_cond);
-    // then destroy it
-    pthread_cond_destroy(&m_activity_cond);
-    pthread_mutex_unlock(&m_activity_cond_lock);
-
-    // destroy the mutexes
-    pthread_mutex_destroy(&m_activity_cond_lock);
-    if (m_data_buffer) delete m_data_buffer;
-    if (m_scratch_buffer) delete[] m_scratch_buffer;
@@ -108 +104 @@
-        debugError("Failed to stop SP\n");
-    }
-
+_ALL
-}
-
@@ -313 +309 @@
-            m_dropped += dropped_cycles;
-            m_last_cycle = cycle;
-
-
-
+
-        }
-    }
@@ -481 +475 @@
-        } else if(result2 == eCRV_OK) {
-            // no problem here
-
+
+
+
+
+
+
+
+
-            return RAW1394_ISO_OK;
-        } else {
@@ -589 +590 @@
-
-    if(cycle_diff < 0 && (m_state == ePS_Running || m_state == ePS_DryRunning)) {
-
-
+
+
+
-        if(m_state == ePS_Running) {
-            debugShowBackLogLines(200);
@@ -642 +644 @@
-            generateSilentPacketData(data, length, tag, sy, cycle, dropped_cycles, max_length);
-            return RAW1394_ISO_OK;
-be possible too
+also be a possibility
-        //} else if (result == eCRV_EmptyPacket) {
-        //    goto send_empty_packet;
@@ -687 +689 @@
-        enum eChildReturnValue result = generatePacketHeader(data, length, tag, sy, cycle, dropped_cycles, max_length);
-        if (result == eCRV_Packet || result == eCRV_Defer) {
-
-
-
+
+
+
+
+
-            // update some accounting
-            m_last_good_cycle = cycle;
@@ -804 +808 @@
-        }
-    }
-
-
-
-
+
+
+
+
+
+
+
+
+
-    generateEmptyPacketHeader(data, length, tag, sy, cycle, dropped_cycles, max_length);
-    generateEmptyPacketData(data, length, tag, sy, cycle, dropped_cycles, max_length);
-    return RAW1394_ISO_OK;
-}
-
-
-// Frame Transfer API
@@ -823 +831 @@
-bool StreamProcessor::getFrames(unsigned int nbframes, int64_t ts) {
-    bool result;
-Y_VER
-%p.getFrames(%d, %11llu)
-
+
+(%p, %s) getFrames(%d, %11llu)\n
+this, getTypeString(), 
-    assert( getType() == ePT_Receive );
-    if(isDryRunning()) result = getFramesDry(nbframes, ts);
-    else result = getFramesWet(nbframes, ts);
-
+_ISO_RECV
-    return result;
-}
@@ -890 +898 @@
-    debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "StreamProcessor::dropFrames(%d, %lld)\n", nbframes, ts);
-    result = m_data_buffer->dropFrames(nbframes);
-
+_ISO_RECV
-    return result;
-}
@@ -897 +905 @@
-{
-    bool result;
-DEBUG_LEVEL_VERY
-"%p.putFrames(%d, %11llu)
-
+ DEBUG_LEVEL
+ "(%p, %s) putFrames(%d, %11llu)\n
+ this, getTypeString(), 
-    assert( getType() == ePT_Transmit );
-    if(isDryRunning()) result = putFramesDry(nbframes, ts);
-    else result = putFramesWet(nbframes, ts);
-
+_ISO_XMIT
-    return result;
-}
@@ -933 +941 @@
-StreamProcessor::putSilenceFrames(unsigned int nbframes, int64_t ts)
-{
-Extreme(DEBUG_LEVEL_ULTRA
+(DEBUG_LEVEL_VERY
-                       "StreamProcessor::putSilenceFrames(%d, %llu)\n",
-                       nbframes, ts);
@@ -955 +963 @@
-    }
-
-
+_ISO_XMIT
-    return true;
-}
@@ -966 +974 @@
-    if(nbframes > 0) {
-        result = m_data_buffer->dropFrames(nbframes);
-
+_ALL
-        return result;
-    } else {
@@ -973 +981 @@
-            result &= m_data_buffer->writeDummyFrame();
-        }
-
+_ALL
-        return result;
-    }
@@ -1135 +1143 @@
-    // wake up any threads that might be waiting on data in the buffers
-    // since a state transition can cause data to become available
-
+_ALL
-    return true;
-}
@@ -1412 +1420 @@
-    }
-    #endif
-
+_ALL
-    return result;
-}
@@ -1446 +1454 @@
-    }
-    #endif
-
+_ALL
-    return true;
-}
@@ -1496 +1504 @@
-    }
-    #endif
-
+_ALL
-    return result;
-}
@@ -1545 +1553 @@
-    }
-    #endif
-
+_ALL
-    return true;
-}
@@ -1569 +1577 @@
-                                             this, m_last_cycle);
-            m_in_xrun = false;
+            m_min_ahead = 7999;
-            m_local_node_id = m_1394service.getLocalNodeId() & 0x3f;
-            m_data_buffer->setTransparent(false);
@@ -1583 +1592 @@
-    }
-    #endif
-
+_ALL
-    return result;
-}
@@ -1615 +1624 @@
-    }
-    #endif
-
+_ALL
-    return true;
-}
@@ -1728 +1737 @@
-    debugError("Invalid state transition: %s => %s\n",
-        ePSToString(m_state), ePSToString(next_state));
-
+_ALL
-    return false;
-updateState_exit_change_failed:
-    debugError("State transition failed: %s => %s\n",
-        ePSToString(m_state), ePSToString(next_state));
-
+_ALL
-    return false;
-}
-
-bool StreamProcessor::waitForProducePacket()
-{
-    return waitForProduce(getNominalFramesPerPacket());
-}
-bool StreamProcessor::waitForProducePeriod()
-{
-    return waitForProduce(m_StreamProcessorManager.getPeriodSize());
-}
-bool StreamProcessor::waitForProduce(unsigned int nframes)
-{
-    debugOutputExtreme(DEBUG_LEVEL_VERY_VERBOSE,
-                       "(%p, %s) wait ...\n",
-                       this, getTypeString());
-    struct timespec ts;
-    int result;
-    int max_runs = 1000;
-
-    if (clock_gettime(CLOCK_REALTIME, &ts) == -1) {
-        debugError("clock_gettime failed\n");
-        return false;
-    }
-
-    // FIXME: hardcoded timeout of 10 sec
-//     ts.tv_nsec += 1000 * 1000000LL;
-//     while (ts.tv_nsec > 1000000000LL) {
-//         ts.tv_sec += 1;
-//         ts.tv_nsec -= 1000000000LL;
-//     }
-    ts.tv_sec += 2;
-    
-    pthread_mutex_lock(&m_activity_cond_lock);
-    while(!canProduce(nframes) && max_runs) {
-        result = pthread_cond_timedwait(&m_activity_cond, &m_activity_cond_lock, &ts);
-    
-        if(result != 0) {
-            if (result == ETIMEDOUT) {
-                debugOutput(DEBUG_LEVEL_VERBOSE,
-                            "(%p, %s) pthread_cond_timedwait() timed out (result=%d)\n",
-                            this, getTypeString(), result);
-                pthread_mutex_unlock(&m_activity_cond_lock);
-                dumpInfo();
-                return false;
-            } else if (result == EINTR) {
-                debugOutput(DEBUG_LEVEL_VERBOSE,
-                            "(%p, %s) pthread_cond_timedwait() interrupted by signal (result=%d)\n",
-                            this, getTypeString(), result);
-                pthread_mutex_unlock(&m_activity_cond_lock);
-                dumpInfo();
-                return false;
-            } else {
-                debugError("(%p, %s) pthread_cond_timedwait error (result=%d)\n", 
-                            this, getTypeString(), result);
-                pthread_mutex_unlock(&m_activity_cond_lock);
-                dumpInfo();
-                return false;
-            }
-        }
-    }
-    pthread_mutex_unlock(&m_activity_cond_lock);
-    if(max_runs == 0) {
-        debugWarning("(%p) runaway loop\n");
-    }
-    return true;
-}
-
-bool StreamProcessor::waitForConsumePacket()
-{
-    return waitForConsume(getNominalFramesPerPacket());
-}
-bool StreamProcessor::waitForConsumePeriod()
-{
-    return waitForConsume(m_StreamProcessorManager.getPeriodSize());
-}
-bool StreamProcessor::waitForConsume(unsigned int nframes)
-{
-    debugOutputExtreme(DEBUG_LEVEL_VERY_VERBOSE,
-                       "(%p, %s) wait ...\n",
-                       this, getTypeString());
-    struct timespec ts;
-    int result;
-
-    int max_runs = 1000;
-
-    if (clock_gettime(CLOCK_REALTIME, &ts) == -1) {
-        debugError("clock_gettime failed\n");
-        return false;
-    }
-
-    // FIXME: hardcoded timeout of 10 sec
-//     ts.tv_nsec += 1000 * 1000000LL;
-//     while (ts.tv_nsec > 1000000000LL) {
-//         ts.tv_sec += 1;
-//         ts.tv_nsec -= 1000000000LL;
-//     }
-    ts.tv_sec += 2;
-
-    pthread_mutex_lock(&m_activity_cond_lock);
-    while(!canConsume(nframes) && max_runs) {
-        result = pthread_cond_timedwait(&m_activity_cond, &m_activity_cond_lock, &ts);
-        if(result != 0) {
-            if (result == ETIMEDOUT) {
-                debugOutput(DEBUG_LEVEL_VERBOSE,
-                            "(%p, %s) pthread_cond_timedwait() timed out (result=%d)\n",
-                            this, getTypeString(), result);
-                pthread_mutex_unlock(&m_activity_cond_lock);
-                dumpInfo();
-                return false;
-            } else if (result == EINTR) {
-                debugOutput(DEBUG_LEVEL_VERBOSE,
-                            "(%p, %s) pthread_cond_timedwait() interrupted by signal (result=%d)\n",
-                            this, getTypeString(), result);
-                pthread_mutex_unlock(&m_activity_cond_lock);
-                dumpInfo();
-                return false;
-            } else {
-                debugError("(%p, %s) pthread_cond_timedwait error (result=%d)\n", 
-                            this, getTypeString(), result);
-                pthread_mutex_unlock(&m_activity_cond_lock);
-                dumpInfo();
-                return false;
-            }
-        }
-        max_runs--;
-    }
-    pthread_mutex_unlock(&m_activity_cond_lock);
-    
-    if(max_runs == 0) {
-        debugWarning("(%p) runaway loop\n");
-    }
-    
-    debugOutputExtreme(DEBUG_LEVEL_VERY_VERBOSE,
-                       "(%p, %s) leave ...\n",
-                       this, getTypeString());
-    return true;
-}
-
@@ -1885 +1758 @@
-    if(m_in_xrun) return true;
-    if(m_state == ePS_Running && m_next_state == ePS_Running) {
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
-    } else {
-        if(getType() == ePT_Transmit) {
@@ -1947 +1809 @@
- * Helper routines                             *
- ***********************************************/
+// FIXME: I think this can be removed and replaced by putSilenceFrames
-bool
-StreamProcessor::transferSilence(unsigned int nframes)
-{
-    bool retval;
+
+    #ifdef DEBUG
-    signed int fc;
-    ffado_timestamp_t ts_tail_tmp;
+    m_data_buffer->getBufferTailTimestamp(&ts_tail_tmp, &fc);
+    if (fc != 0) {
+        debugWarning("Prefilling a buffer that already contains %d frames\n", fc);
+    }
+    #endif
-
-    // prepare a buffer of silence
-    char *dummybuffer = (char *)calloc(getEventSize(), 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
@@ -1970 +1835 @@
-        retval = false;
-    }
+
-    free(dummybuffer);
-    return retval;
@@ -2023 +1889 @@
-                        (unsigned int)TICKS_TO_CYCLES(now),
-                        (unsigned int)TICKS_TO_OFFSET(now));
+    if(getType() == ePT_Transmit) {
+        debugOutputShort( DEBUG_LEVEL_NORMAL, "  Min ISOXMT bufferfill : %04d\n", m_min_ahead);
+    }
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Xrun?                 : %s\n", (m_in_xrun ? "True":"False"));
-    if (m_state == m_next_state) {

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

@@ -163 +163 @@
-    bool putFrames(unsigned int nbframes, int64_t ts); ///< transfer the client contents to the buffer
-
-    //FIXME: document wait functions
-    bool waitForProducePacket();
-    bool waitForProducePeriod();
-    bool waitForProduce(unsigned int nframes);
-
-    bool waitForConsumePacket();
-    bool waitForConsumePeriod();
-    bool waitForConsume(unsigned int nframes);
-
-    bool canProducePacket();
-    bool canProducePeriod();
@@ -476 +467 @@
-    private:
-        bool m_in_xrun;
-        pthread_mutex_t m_activity_cond_lock;
-        pthread_cond_t  m_activity_cond;
-
-public:
@@ -488 +477 @@
-    const char *getTypeString()
-        {return ePTToString(getType());};
-
-StreamStatistics m_PeriodStat;
-
+
+int m_min_ahead; // DEBUG
+
-    DECLARE_DEBUG_MODULE;
-};

trunk/libffado/src/libstreaming/StreamProcessorManager.cpp

@@ -42 +42 @@
-    , m_SyncSource(NULL)
-    , m_xrun_happened( false )
+    , m_activity_wait_timeout_usec( 1000*1000 )
-    , m_nb_buffers( 0 )
-    , m_period( 0 )
@@ -51 +52 @@
-{
-    addOption(Util::OptionContainer::Option("slaveMode",false));
+    sem_init(&m_activity_semaphore, 0, 0);
-}
-
@@ -57 +59 @@
-    , m_SyncSource(NULL)
-    , m_xrun_happened( false )
+    , m_activity_wait_timeout_usec( 1000*1000 )
-    , m_nb_buffers(nb_buffers)
-    , m_period(period)
@@ -66 +69 @@
-{
-    addOption(Util::OptionContainer::Option("slaveMode",false));
+    sem_init(&m_activity_semaphore, 0, 0);
-}
-
-StreamProcessorManager::~StreamProcessorManager() {
+    sem_post(&m_activity_semaphore);
+    sem_destroy(&m_activity_semaphore);
-}
-
@@ -100 +106 @@
-
-    m_WaitLock.Unlock();
+}
+
+void
+StreamProcessorManager::signalActivity()
+{
+    sem_post(&m_activity_semaphore);
+    debugOutputExtreme(DEBUG_LEVEL_VERBOSE,"%p activity\n", this);
+}
+
+enum StreamProcessorManager::eActivityResult
+StreamProcessorManager::waitForActivity()
+{
+    debugOutputExtreme(DEBUG_LEVEL_VERBOSE,"%p waiting for activity\n", this);
+    struct timespec ts;
+    int result;
+
+    if (clock_gettime(CLOCK_REALTIME, &ts) == -1) {
+        debugError("clock_gettime failed\n");
+        return eAR_Error;
+    }
+    long long int timeout_nsec=0;
+    int timeout_sec = 0;
+    if (m_activity_wait_timeout_usec >= 0) {
+        timeout_nsec = m_activity_wait_timeout_usec * 1000LL;
+        timeout_sec = 0;
+        while(timeout_nsec >= 1000000000LL) {
+            timeout_sec += 1;
+            timeout_nsec -= 1000000000LL;
+        }
+        ts.tv_nsec += timeout_nsec;
+        ts.tv_sec += timeout_sec;
+    }
+
+    if (m_activity_wait_timeout_usec >= 0) {
+        result = sem_timedwait(&m_activity_semaphore, &ts);
+    } else {
+        result = sem_wait(&m_activity_semaphore);
+    }
+
+    if(result != 0) {
+        if (result == ETIMEDOUT) {
+            debugOutput(DEBUG_LEVEL_VERBOSE,
+                        "(%p) pthread_cond_timedwait() timed out (result=%d)\n",
+                        this, result);
+            return eAR_Timeout;
+        } else if (result == EINTR) {
+            debugOutput(DEBUG_LEVEL_VERBOSE,
+                        "(%p) pthread_cond_[timed]wait() interrupted by signal (result=%d)\n",
+                        this, result);
+            return eAR_Interrupted;
+        } else {
+            debugError("(%p) pthread_cond_[timed]wait error (result=%d)\n", 
+                        this, result);
+            debugError("(%p) timeout_sec=%d timeout_nsec=%lld ts.sec=%d ts.nsec=%lld\n", 
+                       this, timeout_sec, timeout_nsec, ts.tv_sec, ts.tv_nsec);
+            return eAR_Error;
+        }
+    }
+
+    debugOutputExtreme(DEBUG_LEVEL_VERBOSE,"%p got activity\n", this);
+    return eAR_Activity;
-}
-
@@ -545 +612 @@
-            };
-
+            // before we do anything else, transfer
+            if(!transferSilence()) {
+                debugError("Could not transfer silence\n");
+                return false;
+            }
+
+            // now calculate the stream offset
-            i = 0;
-            for ( i = 0; i < nb_rcv_sp; i++) {
@@ -558 +632 @@
-            }
-
-            if(!transferSilence()) {
-                debugError("Could not transfer silence\n");
-                return false;
-            }
-            nb_sync_runs--;
-        }
@@ -806 +876 @@
-
-    while(period_not_ready) {
-Y_VER
+
-                           "waiting for period (%d frames in buffer)...\n",
-                           m_SyncSource->getBufferFill());
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-        // HACK: this should be solved more elegantly
@@ -838 +917 @@
-            }
-        }
+        debugOutputExtreme(DEBUG_LEVEL_VERBOSE, " period not ready? %d...\n", period_not_ready);
+
-        // check for underruns on the ISO side,
-        // those should make us bail out of the wait loop
@@ -885 +966 @@
-    m_time_of_transfer2 = m_time_of_transfer;
-    #endif
-
-Y_VER
+
+
-                        "transfer at %llu ticks...\n",
-                        m_time_of_transfer);
@@ -1027 +1108 @@
-    debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "Transferring silent period...\n");
-    bool retval=true;
+    // NOTE: the order here is opposite from the order in
+    // normal operation (transmit is before receive), because
+    // we can do that here (data=silence=available) and 
+    // it increases reliability (esp. on startup)
+    retval &= transferSilence(StreamProcessor::ePT_Transmit);
-    retval &= transferSilence(StreamProcessor::ePT_Receive);
-    retval &= transferSilence(StreamProcessor::ePT_Transmit);
-    return retval;
-}

trunk/libffado/src/libstreaming/StreamProcessorManager.h

@@ -69 +69 @@
-    bool startDryRunning();
-    bool syncStartAll();
+    // activity signaling
+    enum eActivityResult {
+        eAR_Activity,
+        eAR_Timeout,
+        eAR_Interrupted,
+        eAR_Error
+    };
+    void signalActivity();
+    enum eActivityResult waitForActivity();
-
-    // this is the setup API
@@ -139 +148 @@
-        {return *m_SyncSource;};
-
-
+ // FIXME: private?
-
-sync primitive
+related var
-    bool m_xrun_happened;
+    int m_activity_wait_timeout_usec;
-    bool m_thread_realtime;
-    int m_thread_priority;
+
+    // activity signaling
+    sem_t m_activity_semaphore;
-
-    // processor list