root/branches/streaming-rework/src/libstreaming/StreamProcessorManager.cpp

/* $Id$ */

/*
 *   FreeBob Streaming API
 *   FreeBob = Firewire (pro-)audio for linux
 *
 *   http://freebob.sf.net
 *
 *   Copyright (C) 2005,2006 Pieter Palmers <pieterpalmers@users.sourceforge.net>
 *
 *   This program is free software {} you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation {} either version 2 of the License, or
 *   (at your option) any later version.
 *
 *   This program is distributed in the hope that it will be useful,
 *   but WITHOUT ANY WARRANTY {} without even the implied warranty of
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *   GNU General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with this program {} if not, write to the Free Software
 *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
 *
 * 
 *
 */

#include "StreamProcessorManager.h"
#include "StreamProcessor.h"
#include "Port.h"
#include <errno.h>
#include <assert.h>

#include "../libutil/PosixThread.h"

#include "libstreaming/cycletimer.h"

#define CYCLES_TO_SLEEP_AFTER_RUN_SIGNAL 50

namespace FreebobStreaming {

IMPL_DEBUG_MODULE( StreamProcessorManager, StreamProcessorManager, DEBUG_LEVEL_NORMAL );

StreamProcessorManager::StreamProcessorManager(unsigned int period, unsigned int nb_buffers)
        : m_SyncSource(NULL), m_nb_buffers(nb_buffers), m_period(period), m_xruns(0), 
        m_isoManager(0), m_nbperiods(0) {

}

StreamProcessorManager::~StreamProcessorManager() {
        if (m_isoManager) delete m_isoManager;
        
}

/**
 * Registers \ref processor with this manager.
 *
 * also registers it with the isohandlermanager
 *
 * be sure to call isohandlermanager->init() first!
 * and be sure that the processors are also ->init()'ed
 *
 * @param processor 
 * @return true if successfull
 */
bool StreamProcessorManager::registerProcessor(StreamProcessor *processor)
{
        debugOutput( DEBUG_LEVEL_VERBOSE, "Registering processor (%p)\n",processor);
        assert(processor);
        assert(m_isoManager);

        if (processor->getType()==StreamProcessor::E_Receive) {
                processor->setVerboseLevel(getDebugLevel()); // inherit debug level
                
                m_ReceiveProcessors.push_back(processor);
                
                processor->setManager(this);
                                
                return true;
        }
        
        if (processor->getType()==StreamProcessor::E_Transmit) {
                processor->setVerboseLevel(getDebugLevel()); // inherit debug level
                
                m_TransmitProcessors.push_back(processor);
                
                processor->setManager(this);
                
                return true;
        }

        debugFatal("Unsupported processor type!\n");
        
        return false;
}

bool StreamProcessorManager::unregisterProcessor(StreamProcessor *processor)
{
        debugOutput( DEBUG_LEVEL_VERBOSE, "Unregistering processor (%p)\n",processor);
        assert(processor);

        if (processor->getType()==StreamProcessor::E_Receive) {

                for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
                        it != m_ReceiveProcessors.end();
                        ++it ) {

                        if ( *it == processor ) { 
                                        m_ReceiveProcessors.erase(it);
                                        
                                        processor->clearManager();
                                        
                                        if(!m_isoManager->unregisterStream(processor)) {
                                                debugOutput(DEBUG_LEVEL_VERBOSE,"Could not unregister receive stream processor from the Iso manager\n");
                                                
                                                return false;
                                                
                                        }
                                        
                                        return true;
                                }
                }
        }

        if (processor->getType()==StreamProcessor::E_Transmit) {
                for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
                        it != m_TransmitProcessors.end();
                        ++it ) {

                        if ( *it == processor ) { 
                                        m_TransmitProcessors.erase(it);
                                        
                                        processor->clearManager();
                                        
                                        if(!m_isoManager->unregisterStream(processor)) {
                                                debugOutput(DEBUG_LEVEL_VERBOSE,"Could not unregister transmit stream processor from the Iso manager\n");
                                                
                                                return false;
                                                
                                        }
                                        
                                        return true;
                                }
                }
        }
        
        debugFatal("Processor (%p) not found!\n",processor);

        return false; //not found

}

bool StreamProcessorManager::setSyncSource(StreamProcessor *s) {
    m_SyncSource=s;
    return true;
}

StreamProcessor *StreamProcessorManager::getSyncSource() {
    return m_SyncSource;
}

bool StreamProcessorManager::init()
{
        debugOutput( DEBUG_LEVEL_VERBOSE, "enter...\n");

        // the tread that runs the StreamProcessor
        // checking the period boundaries
        int prio=m_thread_priority+5;
        if (prio>98) prio=98;
        
        m_streamingThread=new FreebobUtil::PosixThread(this,
           m_thread_realtime, prio, 
           PTHREAD_CANCEL_DEFERRED);
           
        if(!m_streamingThread) {
                debugFatal("Could not create streaming thread\n");
                return false;
        }
        
        m_isoManager=new IsoHandlerManager(m_thread_realtime, m_thread_priority);
        
        if(!m_isoManager) {
                debugFatal("Could not create IsoHandlerManager\n");
                return false;
        }
        
        // propagate the debug level
        m_isoManager->setVerboseLevel(getDebugLevel());
        
        if(!m_isoManager->init()) {
                debugFatal("Could not initialize IsoHandlerManager\n");
                return false;
        }
        
        m_xrun_happened=false;
        
        return true;
}

bool StreamProcessorManager::Init()
{
    debugOutput( DEBUG_LEVEL_VERBOSE, "Initializing runner...\n");

    // no xrun has occurred (yet)

    return true;
}

bool StreamProcessorManager::prepare() {

        debugOutput( DEBUG_LEVEL_VERBOSE, "Preparing...\n");
        
        // if no sync source is set, select one here
        if(m_SyncSource == NULL) {
           debugWarning("Sync Source is not set. Defaulting to first StreamProcessor.\n");
        }
        
        for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
                it != m_ReceiveProcessors.end();
                ++it ) {
                        if(m_SyncSource == NULL) {
                                debugWarning(" => Sync Source is %p.\n", *it);
                                m_SyncSource = *it;
                        }
        }

        for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
                it != m_TransmitProcessors.end();
                ++it ) {
                        if(m_SyncSource == NULL) {
                                debugWarning(" => Sync Source is %p.\n", *it);
                                m_SyncSource = *it;
                        }
        }

        // now do the actual preparation
        debugOutput( DEBUG_LEVEL_VERBOSE, "Prepare Receive processors...\n");
        for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
                it != m_ReceiveProcessors.end();
                ++it ) {
                        if(!(*it)->setSyncSource(m_SyncSource)) {
                                debugFatal(  " could not set sync source (%p)...\n",(*it));
                                return false;
                        }
                        
                        if(!(*it)->prepare()) {
                                debugFatal(  " could not prepare (%p)...\n",(*it));
                                return false;
                        }
        }

        debugOutput( DEBUG_LEVEL_VERBOSE, "Prepare Transmit processors...\n");
        for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
                it != m_TransmitProcessors.end();
                ++it ) {
                        if(!(*it)->setSyncSource(m_SyncSource)) {
                                debugFatal(  " could not set sync source (%p)...\n",(*it));
                                return false;
                        }               
                        if(!(*it)->prepare()) {
                                debugFatal( " could not prepare (%p)...\n",(*it));
                                return false;
                        }
        }

    // if there are no stream processors registered, 
    // fail
    if (m_ReceiveProcessors.size() + m_TransmitProcessors.size() == 0) {
        debugFatal("No stream processors registered, can't do anything usefull\n");
        return false;
    }

        return true;
}

// FIXME: this can be removed
bool StreamProcessorManager::Execute()
{
        // temp measure, polling
        usleep(1000);

        // FIXME: move this to an IsoHandlerManager sub-thread
        // and make this private again in IHM
        m_isoManager->updateCycleTimers();
        
        return true;
}

bool StreamProcessorManager::syncStartAll() {

    debugOutput( DEBUG_LEVEL_VERBOSE, "Waiting for all StreamProcessor streams to start running...\n");
    // we have to wait until all streamprocessors indicate that they are running
    // i.e. that there is actually some data stream flowing
    int wait_cycles=2000; // two seconds
    bool notRunning=true;
    while (notRunning && wait_cycles) {
        wait_cycles--;
        notRunning=false;
        
        for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
                it != m_ReceiveProcessors.end();
                ++it ) {
            if(!(*it)->isRunning()) notRunning=true;
        }

        for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
                it != m_TransmitProcessors.end();
                ++it ) {
            if(!(*it)->isRunning()) notRunning=true;
        }
        usleep(1000);
        debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "Running check: %d\n",notRunning);
    }

    if(!wait_cycles) { // timout has occurred
        debugFatal("One or more streams are not starting up (timeout):\n");
                    
        for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
                it != m_ReceiveProcessors.end();
                ++it ) {
            if(!(*it)->isRunning()) {
                debugFatal(" receive stream %p not running\n",*it);
            } else {    
                debugFatal(" receive stream %p running\n",*it);
            }
        }

        for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
                it != m_TransmitProcessors.end();
                ++it ) {
            if(!(*it)->isRunning()) {
                debugFatal(" transmit stream %p not running\n",*it);
            } else {    
                debugFatal(" transmit stream %p running\n",*it);
            }
        }
        
        return false;
    }

    // we want to make sure that everything is running well, 
    // so wait for a while
    usleep(USECS_PER_CYCLE * CYCLES_TO_SLEEP_AFTER_RUN_SIGNAL);

    debugOutput( DEBUG_LEVEL_VERBOSE, " StreamProcessor streams running...\n");
    debugOutput( DEBUG_LEVEL_VERBOSE, "Resetting StreamProcessors...\n");

    // now we reset the frame counters
    for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
            it != m_ReceiveProcessors.end();
            ++it ) {

        debugOutput( DEBUG_LEVEL_VERBOSE, "Before:\n");

        if(getDebugLevel()>=DEBUG_LEVEL_VERBOSE) {
            (*it)->dumpInfo();
        }

        (*it)->reset();

        debugOutput( DEBUG_LEVEL_VERBOSE, "After:\n");

        if(getDebugLevel()>=DEBUG_LEVEL_VERBOSE) {
            (*it)->dumpInfo();
        }

    }
    
    for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
            it != m_TransmitProcessors.end();
            ++it ) {
            
        debugOutput( DEBUG_LEVEL_VERBOSE, "Before:\n");
        
        if(getDebugLevel()>=DEBUG_LEVEL_VERBOSE) {
            (*it)->dumpInfo();
        }
        
        (*it)->reset();
        
        debugOutput( DEBUG_LEVEL_VERBOSE, "After:\n");
        
        if(getDebugLevel()>=DEBUG_LEVEL_VERBOSE) {
            (*it)->dumpInfo();
        }
    }
        
    debugOutput( DEBUG_LEVEL_VERBOSE, "Enabling StreamProcessors...\n");
    
    uint64_t now=m_SyncSource->getTimeNow(); // fixme: should be in usecs, not ticks
    
    // FIXME: this should not be in cycles, but in 'time'
    unsigned int enable_at=TICKS_TO_CYCLES(now)+300;
        
    debugOutput( DEBUG_LEVEL_VERBOSE, " Sync Source StreamProcessor...\n");
    if (!m_SyncSource->prepareForEnable()) {
            debugFatal("Could not prepare Sync Source StreamProcessor for enable()...\n");
        return false;
    }

    m_SyncSource->enable(enable_at);

    debugOutput( DEBUG_LEVEL_VERBOSE, " All StreamProcessors...\n");
    if (!enableStreamProcessors(enable_at)) {
        debugFatal("Could not enable StreamProcessors...\n");
        return false;
    }

    return true;
}

bool StreamProcessorManager::start() {
        debugOutput( DEBUG_LEVEL_VERBOSE, "Starting Processors...\n");
        assert(m_isoManager);
        
        debugOutput( DEBUG_LEVEL_VERBOSE, "Creating handlers for the StreamProcessors...\n");
        debugOutput( DEBUG_LEVEL_VERBOSE, " Receive processors...\n");
        for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
                it != m_ReceiveProcessors.end();
                ++it ) {
                        if (!(*it)->prepareForStart()) {
                                debugOutput(DEBUG_LEVEL_VERBOSE,"Receive stream processor (%p) failed to prepare for start\n", *it);
                                return false;
                        }
                        if (!m_isoManager->registerStream(*it)) {
                                debugOutput(DEBUG_LEVEL_VERBOSE,"Could not register receive stream processor (%p) with the Iso manager\n",*it);
                                return false;
                        }
                }

        debugOutput( DEBUG_LEVEL_VERBOSE, " Transmit processors...\n");
        for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
                it != m_TransmitProcessors.end();
                ++it ) {
                        if (!(*it)->prepareForStart()) {
                                debugOutput(DEBUG_LEVEL_VERBOSE,"Transmit stream processor (%p) failed to prepare for start\n", *it);
                                return false;
                        }
                        if (!m_isoManager->registerStream(*it)) {
                                debugOutput(DEBUG_LEVEL_VERBOSE,"Could not register transmit stream processor (%p) with the Iso manager\n",*it);
                                return false;
                        }
                }

        debugOutput( DEBUG_LEVEL_VERBOSE, "Preparing IsoHandlerManager...\n");
        if (!m_isoManager->prepare()) {
                debugFatal("Could not prepare isoManager\n");
                return false;
        }

        debugOutput( DEBUG_LEVEL_VERBOSE, "Disabling StreamProcessors...\n");
        if (!disableStreamProcessors()) {
                debugFatal("Could not disable StreamProcessors...\n");
                return false;
        }
                
        debugOutput( DEBUG_LEVEL_VERBOSE, "Starting IsoHandlers...\n");
        if (!m_isoManager->startHandlers(0)) {
                debugFatal("Could not start handlers...\n");
                return false;
        }
        
        debugOutput( DEBUG_LEVEL_VERBOSE, "Starting streaming threads...\n");

        // start the runner thread
        // FIXME: not used anymore (for updatecycletimers ATM, but that's not good)
        m_streamingThread->Start();

        // start all SP's synchonized
        if (!syncStartAll()) {
                debugFatal("Could not syncStartAll...\n");
                return false;
        }
        
        // dump the iso stream information when in verbose mode
        if(getDebugLevel()>=DEBUG_LEVEL_VERBOSE) {
                m_isoManager->dumpInfo();
        }
        
        return true;
        
}

bool StreamProcessorManager::stop() {
        debugOutput( DEBUG_LEVEL_VERBOSE, "Stopping...\n");
        assert(m_isoManager);
        assert(m_streamingThread);

        debugOutput( DEBUG_LEVEL_VERBOSE, "Waiting for all StreamProcessors to prepare to stop...\n");
        // Most stream processors can just stop without special treatment.  However, some
        // (like the MOTU) need to do a few things before it's safe to turn off the iso
        // handling.
        int wait_cycles=2000; // two seconds ought to be sufficient
        bool allReady = false;
        while (!allReady && wait_cycles) {
                wait_cycles--;
                allReady = true;
                
                for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
                        it != m_ReceiveProcessors.end();
                        ++it ) {
                        if(!(*it)->prepareForStop()) allReady = false;
                }
        
                for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
                        it != m_TransmitProcessors.end();
                        ++it ) {
                        if(!(*it)->prepareForStop()) allReady = false;
                }
                usleep(1000);
        }


        debugOutput( DEBUG_LEVEL_VERBOSE, "Stopping threads...\n");
        
        m_streamingThread->Stop();
        
        debugOutput( DEBUG_LEVEL_VERBOSE, "Stopping handlers...\n");
        if(!m_isoManager->stopHandlers()) {
           debugFatal("Could not stop ISO handlers\n");
           return false;
        }
        
        debugOutput( DEBUG_LEVEL_VERBOSE, "Unregistering processors from handlers...\n");
    // now unregister all streams from iso manager
        debugOutput( DEBUG_LEVEL_VERBOSE, " Receive processors...\n");
        for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
                it != m_ReceiveProcessors.end();
                ++it ) {
                        if (!m_isoManager->unregisterStream(*it)) {
                                debugOutput(DEBUG_LEVEL_VERBOSE,"Could not unregister receive stream processor (%p) from the Iso manager\n",*it);
                                return false;
                        }
                        
                }

        debugOutput( DEBUG_LEVEL_VERBOSE, " Transmit processors...\n");
        for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
                it != m_TransmitProcessors.end();
                ++it ) {
                        if (!m_isoManager->unregisterStream(*it)) {
                                debugOutput(DEBUG_LEVEL_VERBOSE,"Could not unregister transmit stream processor (%p) from the Iso manager\n",*it);
                                return false;
                        }
                        
                }
        
        return true;
        
}

/**
 * Enables the registered StreamProcessors
 * @return true if successful, false otherwise
 */
bool StreamProcessorManager::enableStreamProcessors(unsigned int time_to_enable_at) {
    // we prepare the streamprocessors for enable
    for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
            it != m_ReceiveProcessors.end();
            ++it ) {            
        (*it)->prepareForEnable();
    }

    for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
            it != m_TransmitProcessors.end();
            ++it ) {
        (*it)->prepareForEnable();
    }

    // then we enable the streamprocessors
    for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
            it != m_ReceiveProcessors.end();
            ++it ) {            
        (*it)->enable(time_to_enable_at);
    }

    for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
            it != m_TransmitProcessors.end();
            ++it ) {
        (*it)->enable(time_to_enable_at);
    }

    // now we wait for the SP's to get enabled
    debugOutput( DEBUG_LEVEL_VERBOSE, "Waiting for all StreamProcessors to be enabled...\n");
    // we have to wait until all streamprocessors indicate that they are running
    // i.e. that there is actually some data stream flowing
    int wait_cycles=2000; // two seconds
    bool notEnabled=true;
    while (notEnabled && wait_cycles) {
        wait_cycles--;
        notEnabled=false;
        
        for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
                it != m_ReceiveProcessors.end();
                ++it ) {
            if(!(*it)->isEnabled()) notEnabled=true;
        }

        for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
                it != m_TransmitProcessors.end();
                ++it ) {
            if(!(*it)->isEnabled()) notEnabled=true;
        }
        usleep(1000); // one cycle
    }
    
    if(!wait_cycles) { // timout has occurred
        debugFatal("One or more streams couldn't be enabled (timeout):\n");
                    
        for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
                it != m_ReceiveProcessors.end();
                ++it ) {
            if(!(*it)->isEnabled()) {
                    debugFatal(" receive stream %p not enabled\n",*it);
            } else {    
                    debugFatal(" receive stream %p enabled\n",*it);
            }
        }
    
        for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
                it != m_TransmitProcessors.end();
                ++it ) {
            if(!(*it)->isEnabled()) {
                    debugFatal(" transmit stream %p not enabled\n",*it);
            } else {    
                    debugFatal(" transmit stream %p enabled\n",*it);
            }
        }
        return false;
    }
    
    debugOutput( DEBUG_LEVEL_VERBOSE, " => all StreamProcessors enabled...\n");

    return true;
}

/**
 * Disables the registered StreamProcessors
 * @return true if successful, false otherwise
 */
bool StreamProcessorManager::disableStreamProcessors() {
    // we prepare the streamprocessors for disable
    for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
            it != m_ReceiveProcessors.end();
            ++it ) {
        (*it)->prepareForDisable();
    }

    for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
            it != m_TransmitProcessors.end();
            ++it ) {
        (*it)->prepareForDisable();
    }

    // then we disable the streamprocessors
    for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
            it != m_ReceiveProcessors.end();
            ++it ) {
        (*it)->disable();
    }

    for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
            it != m_TransmitProcessors.end();
            ++it ) {
        (*it)->disable();
    }

    // now we wait for the SP's to get disabled
    debugOutput( DEBUG_LEVEL_VERBOSE, "Waiting for all StreamProcessors to be disabled...\n");
    // we have to wait until all streamprocessors indicate that they are running
    // i.e. that there is actually some data stream flowing
    int wait_cycles=2000; // two seconds
    bool enabled=true;
    while (enabled && wait_cycles) {
        wait_cycles--;
        enabled=false;
        
        for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
                it != m_ReceiveProcessors.end();
                ++it ) {
            if((*it)->isEnabled()) enabled=true;
        }

        for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
                it != m_TransmitProcessors.end();
                ++it ) {
            if((*it)->isEnabled()) enabled=true;
        }
        usleep(1000); // one cycle
    }
    
    if(!wait_cycles) { // timout has occurred
        debugFatal("One or more streams couldn't be disabled (timeout):\n");
                    
        for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
                it != m_ReceiveProcessors.end();
                ++it ) {
            if(!(*it)->isEnabled()) {
                    debugFatal(" receive stream %p not enabled\n",*it);
            } else {    
                    debugFatal(" receive stream %p enabled\n",*it);
            }
        }
    
        for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
                it != m_TransmitProcessors.end();
                ++it ) {
            if(!(*it)->isEnabled()) {
                    debugFatal(" transmit stream %p not enabled\n",*it);
            } else {    
                    debugFatal(" transmit stream %p enabled\n",*it);
            }
        }
        return false;
    }
    
    debugOutput( DEBUG_LEVEL_VERBOSE, " => all StreamProcessors disabled...\n");
        
    return true;
}

/**
 * Called upon Xrun events. This brings all StreamProcessors back
 * into their starting state, and then carries on streaming. This should
 * have the same effect as restarting the whole thing.
 *
 * @return true if successful, false otherwise
 */
bool StreamProcessorManager::handleXrun() {

        debugOutput( DEBUG_LEVEL_VERBOSE, "Handling Xrun ...\n");

        /* 
         * Reset means:
         * 1) Disabling the SP's, so that they don't process any packets
         *    note: the isomanager does keep on delivering/requesting them
         * 2) Bringing all buffers & streamprocessors into a know state
         *    - Clear all capture buffers
         *    - Put nb_periods*period_size of null frames into the playback buffers
         * 3) Re-enable the SP's
         */
        debugOutput( DEBUG_LEVEL_VERBOSE, "Disabling StreamProcessors...\n");
        if (!disableStreamProcessors()) {
                debugFatal("Could not disable StreamProcessors...\n");
                return false;
        }

        debugOutput( DEBUG_LEVEL_VERBOSE, "Restarting StreamProcessors...\n");
        // start all SP's synchonized
        if (!syncStartAll()) {
                debugFatal("Could not syncStartAll...\n");
                return false;
        }

        debugOutput( DEBUG_LEVEL_VERBOSE, "Xrun handled...\n");
        
        return true;
}

/**
 * @brief Waits until the next period of samples is ready
 *
 * This function does not return until a full period of samples is (or should be)
 * ready to be transferred. 
 *
 * @return true if the period is ready, false if an xrun occurred
 */
bool StreamProcessorManager::waitForPeriod() {
    int time_till_next_period;
    bool xrun_occurred=false;
       
    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "enter...\n");

    assert(m_SyncSource);
    
    time_till_next_period=m_SyncSource->getTimeUntilNextPeriodUsecs();
    
    while(time_till_next_period > 0) {
        debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "waiting for %d usecs...\n", time_till_next_period);
    
        // wait for the period
        usleep(time_till_next_period);
        
        // check for underruns on the ISO side,
        // those should make us bail out of the wait loop
        for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
            it != m_ReceiveProcessors.end();
            ++it ) {
            // a xrun has occurred on the Iso side
            xrun_occurred |= (*it)->xrunOccurred();
        }
        for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
            it != m_TransmitProcessors.end();
            ++it ) {
            // a xrun has occurred on the Iso side
            xrun_occurred |= (*it)->xrunOccurred();
        }

        // check if we were waked up too soon
        time_till_next_period=m_SyncSource->getTimeUntilNextPeriodUsecs();
    }
    
    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "delayed for %d usecs...\n", -time_till_next_period);
    
    // this is to notify the client of the delay 
    // that we introduced 
    m_delayed_usecs=time_till_next_period;
    
    // we save the 'ideal' time of the transfer at this point,
    // because we can have interleaved read - process - write 
    // cycles making that we modify a receiving stream's buffer
    // before we get to writing.
    // NOTE: before waitForPeriod() is called again, both the transmit
    //       and the receive processors should have done their transfer.
    m_time_of_transfer=m_SyncSource->getTimeAtPeriod();
    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "transfer at %llu ticks...\n", 
        m_time_of_transfer);
    
    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 ) {
        // 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 RECV SP %p due to ISO xrun\n",*it);
        }
        if (!((*it)->canClientTransferFrames(m_period))) {
            debugWarning("Xrun on RECV SP %p due to buffer 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;
}

/**
 * @brief Transfer one period of frames for both receive and transmit StreamProcessors
 *
 * Transfers one period of frames from the client side to the Iso side and vice versa.
 *
 * @return true if successful, false otherwise (indicates xrun).
 */
bool StreamProcessorManager::transfer() {
    
    debugOutput( DEBUG_LEVEL_VERBOSE, "Transferring period...\n");

    if (!transfer(StreamProcessor::E_Receive)) return false;
    if (!transfer(StreamProcessor::E_Transmit)) return false;

    return true;
}

/**
 * @brief Transfer one period of frames for either the receive or transmit StreamProcessors
 *
 * Transfers one period of frames from the client side to the Iso side or vice versa.
 *
 * @param t The processor type to tranfer for (receive or transmit)
 * @return true if successful, false otherwise (indicates xrun).
 */

bool StreamProcessorManager::transfer(enum StreamProcessor::EProcessorType t) {
    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "Transferring period...\n");
    
    // a static cast could make sure that there is no performance
    // penalty for the virtual functions (to be checked)
    if (t==StreamProcessor::E_Receive) {
        for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
                it != m_ReceiveProcessors.end();
                ++it ) {
            
            //#ifdef DEBUG
            #if 0
            {
                uint64_t ts_tail=0;
                uint64_t fc_tail=0;
                
                uint64_t ts_head=0;
                uint64_t fc_head=0;
                
                int cnt=0;
                
                (*it)->getBufferHeadTimestamp(&ts_head,&fc_head);
                (*it)->getBufferTailTimestamp(&ts_tail,&fc_tail);
                
                while((fc_head != fc_tail) && (cnt++ < 10)) {
                    (*it)->getBufferHeadTimestamp(&ts_head,&fc_head);
                    (*it)->getBufferTailTimestamp(&ts_tail,&fc_tail);
                }
                
                debugOutput(DEBUG_LEVEL_VERBOSE,"R => HEAD: TS=%11llu, FC=%5llu | TAIL: TS=%11llu, FC=%5llu, %d\n",
                    ts_head, fc_head, ts_tail, fc_tail, cnt);
            }
            #endif
    
            if(!(*it)->getFrames(m_period, (int64_t)m_time_of_transfer)) {
                    debugOutput(DEBUG_LEVEL_VERBOSE,"could not getFrames(%u, %11llu) from stream processor (%p)",
                            m_period, m_time_of_transfer,*it);
                    return false; // buffer underrun
            }
            
            //#ifdef DEBUG
            #if 0
            {
                uint64_t ts_tail=0;
                uint64_t fc_tail=0;
                
                uint64_t ts_head=0;
                uint64_t fc_head=0;
                
                int cnt=0;
                
                (*it)->getBufferHeadTimestamp(&ts_head,&fc_head);
                (*it)->getBufferTailTimestamp(&ts_tail,&fc_tail);
            
                while((fc_head != fc_tail) && (cnt++ < 10)) {
                    (*it)->getBufferHeadTimestamp(&ts_head,&fc_head);
                    (*it)->getBufferTailTimestamp(&ts_tail,&fc_tail);
                }
                
                debugOutput(DEBUG_LEVEL_VERBOSE,"R  > HEAD: TS=%11llu, FC=%5llu | TAIL: TS=%11llu, FC=%5llu, %d\n",
                    ts_head, fc_head, ts_tail, fc_tail, cnt);
            }
            #endif
    
        }
    } else {
        for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
                it != m_TransmitProcessors.end();
                ++it ) {
                
            //#ifdef DEBUG
            #if 0
            {
                uint64_t ts_tail=0;
                uint64_t fc_tail=0;
                
                uint64_t ts_head=0;
                uint64_t fc_head=0;
                
                int cnt=0;
                
                (*it)->getBufferHeadTimestamp(&ts_head,&fc_head);
                (*it)->getBufferTailTimestamp(&ts_tail,&fc_tail);
                
                while((fc_head != fc_tail) && (cnt++ < 10)) {
                    (*it)->getBufferHeadTimestamp(&ts_head,&fc_head);
                    (*it)->getBufferTailTimestamp(&ts_tail,&fc_tail);
                }
                
                debugOutput(DEBUG_LEVEL_VERBOSE,"T => HEAD: TS=%11llu, FC=%5llu | TAIL: TS=%11llu, FC=%5llu, %d\n",
                    ts_head, fc_head, ts_tail, fc_tail, cnt);
            }
            #endif
                
            if(!(*it)->putFrames(m_period, (int64_t)m_time_of_transfer)) {
                debugOutput(DEBUG_LEVEL_VERBOSE, "could not putFrames(%u,%llu) to stream processor (%p)",
                        m_period, m_time_of_transfer, *it);
                return false; // buffer overrun
            }
            
            //#ifdef DEBUG
            #if 0
            {
                uint64_t ts_tail=0;
                uint64_t fc_tail=0;
                
                uint64_t ts_head=0;
                uint64_t fc_head=0;
                
                int cnt=0;
                
                (*it)->getBufferHeadTimestamp(&ts_head,&fc_head);
                (*it)->getBufferTailTimestamp(&ts_tail,&fc_tail);
            
                while((fc_head != fc_tail) && (cnt++ < 10)) {
                    (*it)->getBufferHeadTimestamp(&ts_head,&fc_head);
                    (*it)->getBufferTailTimestamp(&ts_tail,&fc_tail);
                }
                
                debugOutput(DEBUG_LEVEL_VERBOSE,"T  > HEAD: TS=%11llu, FC=%5llu | TAIL: TS=%11llu, FC=%5llu, %d\n",
                    ts_head, fc_head, ts_tail, fc_tail, cnt);
            }
            #endif
        }
    }

    return true;
}

void StreamProcessorManager::dumpInfo() {
        debugOutputShort( DEBUG_LEVEL_NORMAL, "----------------------------------------------------\n");
        debugOutputShort( DEBUG_LEVEL_NORMAL, "Dumping StreamProcessorManager information...\n");
        debugOutputShort( DEBUG_LEVEL_NORMAL, "Period count: %6d\n", m_nbperiods);

        debugOutputShort( DEBUG_LEVEL_NORMAL, " Receive processors...\n");
        for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
                it != m_ReceiveProcessors.end();
                ++it ) {
                (*it)->dumpInfo();
        }

        debugOutputShort( DEBUG_LEVEL_NORMAL, " Transmit processors...\n");
        for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
                it != m_TransmitProcessors.end();
                ++it ) {
                (*it)->dumpInfo();
        }

        debugOutputShort( DEBUG_LEVEL_NORMAL, "Iso handler info:\n");
        m_isoManager->dumpInfo();
        debugOutputShort( DEBUG_LEVEL_NORMAL, "----------------------------------------------------\n");

}

void StreamProcessorManager::setVerboseLevel(int l) {
        setDebugLevel(l);

        if (m_isoManager) m_isoManager->setVerboseLevel(l);

        debugOutput( DEBUG_LEVEL_VERBOSE, " Receive processors...\n");
        for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
                it != m_ReceiveProcessors.end();
                ++it ) {
                (*it)->setVerboseLevel(l);
        }

        debugOutput( DEBUG_LEVEL_VERBOSE, " Transmit processors...\n");
        for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
                it != m_TransmitProcessors.end();
                ++it ) {
                (*it)->setVerboseLevel(l);
        }
}


int StreamProcessorManager::getPortCount(enum Port::E_PortType type, enum Port::E_Direction direction) {
        int count=0;

        if (direction == Port::E_Capture) {
                for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
                        it != m_ReceiveProcessors.end();
                        ++it ) {
                        count += (*it)->getPortCount(type);
                }
        } else {
                for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
                        it != m_TransmitProcessors.end();
                        ++it ) {
                        count += (*it)->getPortCount(type);
                }
        }
        return count;
}

int StreamProcessorManager::getPortCount(enum Port::E_Direction direction) {
        int count=0;

        if (direction == Port::E_Capture) {
                for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
                        it != m_ReceiveProcessors.end();
                        ++it ) {
                        count += (*it)->getPortCount();
                }
        } else {
                for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
                        it != m_TransmitProcessors.end();
                        ++it ) {
                        count += (*it)->getPortCount();
                }
        }
        return count;
}

// TODO: implement a port map here, instead of the loop

Port* StreamProcessorManager::getPortByIndex(int idx, enum Port::E_Direction direction) {
        int count=0;
        int prevcount=0;

        if (direction == Port::E_Capture) {
                for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
                        it != m_ReceiveProcessors.end();
                        ++it ) {
                        count += (*it)->getPortCount();
                        if (count > idx) {
                                return (*it)->getPortAtIdx(idx-prevcount);
                        }
                        prevcount=count;
                }
        } else {
                for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
                        it != m_TransmitProcessors.end();
                        ++it ) {
                        count += (*it)->getPortCount();
                        if (count > idx) {
                                return (*it)->getPortAtIdx(idx-prevcount);
                        }
                        prevcount=count;
                }
        }
        return NULL;
}

bool StreamProcessorManager::setThreadParameters(bool rt, int priority) {
    m_thread_realtime=rt;
    m_thread_priority=priority;
    return true;
}


} // end of namespace