Changeset 719 for branches

Timestamp:

11/22/07 06:43:39 (15 years ago)

Author:

ppalmers

Message:

backup commit

Files:

• branches/ppalmers-streaming/src/ffado_streaming.cpp (modified) (1 diff)
• branches/ppalmers-streaming/src/genericavc/avc_avdevice.cpp (modified) (1 diff)
• branches/ppalmers-streaming/src/libstreaming/amdtp/AmdtpReceiveStreamProcessor.cpp (modified) (7 diffs)
• branches/ppalmers-streaming/src/libstreaming/amdtp/AmdtpReceiveStreamProcessor.h (modified) (3 diffs)
• branches/ppalmers-streaming/src/libstreaming/amdtp/AmdtpTransmitStreamProcessor.cpp (modified) (11 diffs)
• branches/ppalmers-streaming/src/libstreaming/amdtp/AmdtpTransmitStreamProcessor.h (modified) (1 diff)
• branches/ppalmers-streaming/src/libstreaming/generic/IsoStream.cpp (modified) (2 diffs)
• branches/ppalmers-streaming/src/libstreaming/generic/IsoStream.h (modified) (2 diffs)
• branches/ppalmers-streaming/src/libstreaming/generic/Port.cpp (modified) (2 diffs)
• branches/ppalmers-streaming/src/libstreaming/generic/PortManager.cpp (modified) (1 diff)
• branches/ppalmers-streaming/src/libstreaming/generic/StreamProcessor.cpp (modified) (6 diffs)
• branches/ppalmers-streaming/src/libstreaming/generic/StreamProcessor.h (modified) (10 diffs)
• branches/ppalmers-streaming/src/libstreaming/StreamProcessorManager.cpp (modified) (29 diffs)
• branches/ppalmers-streaming/src/libstreaming/StreamProcessorManager.h (modified) (3 diffs)
• branches/ppalmers-streaming/src/libstreaming/util/IsoHandler.cpp (modified) (1 diff)
• branches/ppalmers-streaming/src/libstreaming/util/IsoHandlerManager.cpp (modified) (1 diff)
• branches/ppalmers-streaming/src/libutil/TimestampedBuffer.cpp (modified) (4 diffs)
• branches/ppalmers-streaming/src/libutil/TimestampedBuffer.h (modified) (1 diff)
• branches/ppalmers-streaming/tests/test-cycletimer.cpp (modified) (1 diff)
• branches/ppalmers-streaming/tests/test-sytmonitor.cpp (modified) (1 diff)

branches/ppalmers-streaming/src/ffado_streaming.cpp

@@ -340 +340 @@
-
-int ffado_streaming_transfer_playback_buffers(ffado_device_t *dev) {
-Receive
+Transmit
-}
-

branches/ppalmers-streaming/src/genericavc/avc_avdevice.cpp

@@ -592 +592 @@
-int
-AvDevice::getStreamCount() {
-
+
+
-}
-

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

@@ -45 +45 @@
-    : StreamProcessor(ePT_Receive , port)
-    , m_dimension(dimension)
-    , m_last_timestamp(0)
-    , m_last_timestamp2(0)
-    , m_dropped(0) 
-{}
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
-
-    m_PeriodStat.setName("RCV PERIOD");
@@ -273 +62 @@
-    debugOutput( DEBUG_LEVEL_VERBOSE, "Preparing (%p)...\n", this);
-
-    // prepare all non-device specific stuff
-    // i.e. the iso stream and the associated ports
-    if(!StreamProcessor::prepare()) {
-        debugFatal("Could not prepare base class\n");
-        return false;
-    }
-
-    switch (m_manager->getNominalRate()) {
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-            return false;
-        }
-
-        switch ((*it)->getPortType()) {
-            case Port::E_Audio:
-                if(!(*it)->setSignalType(Port::E_PeriodSignalled)) {
-                    debugFatal("Could not set signal type to PeriodSignalling");
-                    return false;
-                }
-                // buffertype and datatype are dependant on the API
-                debugWarning("---------------- ! Doing hardcoded dummy setup ! --------------\n");
-                // buffertype and datatype are dependant on the API
-                if(!(*it)->setBufferType(Port::E_PointerBuffer)) {
-                    debugFatal("Could not set buffer type");
-                    return false;
-                }
-                if(!(*it)->useExternalBuffer(true)) {
-                    debugFatal("Could not set external buffer usage");
-                    return false;
-                }
-                if(!(*it)->setDataType(Port::E_Float)) {
-                    debugFatal("Could not set data type");
-                    return false;
-                }
-                break;
-            case Port::E_Midi:
-                if(!(*it)->setSignalType(Port::E_PacketSignalled)) {
-                    debugFatal("Could not set signal type to PacketSignalling");
-                    return false;
-                }
-                // buffertype and datatype are dependant on the API
-                // buffertype and datatype are dependant on the API
-                debugWarning("---------------- ! Doing hardcoded test setup ! --------------\n");
-                // buffertype and datatype are dependant on the API
-                if(!(*it)->setBufferType(Port::E_RingBuffer)) {
-                    debugFatal("Could not set buffer type");
-                    return false;
-                }
-                if(!(*it)->setDataType(Port::E_MidiEvent)) {
-                    debugFatal("Could not set data type");
-                    return false;
-                }
-                break;
-            default:
-                debugWarning("Unsupported port type specified\n");
-                break;
-        }
-    }
-
-    // the API specific settings of the ports should already be set,
-    // as this is called from the processorManager->prepare()
-    // so we can init the ports
-    if(!initPorts()) {
-        debugFatal("Could not initialize ports!\n");
-        return false;
-    }
-
-    if(!preparePorts()) {
-        debugFatal("Could not initialize ports!\n");
-        return false;
-    }
-
-    debugOutput( DEBUG_LEVEL_VERBOSE, "Prepared for:\n");
-    debugOutput( DEBUG_LEVEL_VERBOSE, " Samplerate: %d, DBS: %d, SYT: %d\n",
-m_dimension,
+ m_dimension, 
-    debugOutput( DEBUG_LEVEL_VERBOSE, " PeriodSize: %d, NbBuffers: %d\n",
-             m_manager->getPeriodSize(), m_manager->getNbBuffers());
@@ -408 +90 @@
-
-    return true;
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
- */
-bool AmdtpReceiveStreamProcessor::processReadBlock(char *data,
@@ -498 +185 @@
-          ++it )
-    {
-
-        if((*it)->isDisabled()) {continue;};
-
@@ -523 +209 @@
-    }
-    return no_problem;
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-}
-
@@ -583 +305 @@
-}
-
-
+
+
+
-                       unsigned int offset, unsigned int nevents)
-{
-    unsigned int j=0;
-
-//     printf("****************\n");
-//     hexDumpQuadlets(data,m_dimension*4);
-//     printf("****************\n");
-
-    quadlet_t *target_event;
-
@@ -640 +359 @@
-}
-
+int
+AmdtpReceiveStreamProcessor::provideSilenceToPort(
+                       AmdtpAudioPort *p, unsigned int offset, unsigned int nevents)
+{
+    unsigned int j=0;
+    switch(p->getDataType()) {
+        default:
+        case Port::E_Int24:
+            {
+                quadlet_t *buffer=(quadlet_t *)(p->getBufferAddress());
+                assert(nevents + offset <= p->getBufferSize());
+                buffer+=offset;
+
+                for(j = 0; j < nevents; j += 1) { // decode max nsamples
+                    *(buffer)=0;
+                    buffer++;
+                }
+            }
+            break;
+        case Port::E_Float:
+            {
+                float *buffer=(float *)(p->getBufferAddress());
+                assert(nevents + offset <= p->getBufferSize());
+                buffer+=offset;
+
+                for(j = 0; j < nevents; j += 1) { // decode max nsamples
+                    *buffer = 0.0;
+                    buffer++;
+                }
+            }
+            break;
+    }
+    return 0;
+}
+
-} // end of namespace Streaming

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

@@ -72 +72 @@
-     * Create a AMDTP receive StreamProcessor
-     * @param port 1394 port
-     * @param framerate frame rate
-     * @param dimension number of substreams in the ISO stream
-     *                  (midi-muxed is only one stream)
@@ -79 +78 @@
-    virtual ~AmdtpReceiveStreamProcessor() {};
-
-
+
+
+
+
-                  unsigned char channel, unsigned char tag, unsigned char sy,
-                  unsigned int cycle, unsigned int dropped);
-
+    virtual bool prepareChild();
-
-
-bool reset();
-bool prepare()
-
-bool prepareForStop()
-bool prepareForStart();
-
-bool getFrames(unsigned int nbframes, int64_t ts); ///< transfer the buffer contents to the client
-bool getFramesDry(unsigned int nbframes, int64_t ts)
+
+virtual unsigned int getEventSize() 
+                {return 4;}
+
+                {return 4 * (2 + m_syt_interval * m_dimension);}
+virtual unsigned int getEventsPerFrame() 
+
+virtual unsigned int getUpdatePeriod() 
+                {return m_syt_interval;}
-
-    // We have 1 period of samples = m_period
@@ -101 +104 @@
-    // however, if we only count the number of used packets
-    // it is m_period / m_syt_interval
-
+virtual 
-
-    unsigned int getMaxPacketSize() {return 4 * (2 + m_syt_interval * m_dimension);};
-
-    void dumpInfo();
-protected:
-
-    bool processReadBlock(char *data, unsigned int nevents, unsigned int offset);
+    bool provideSilenceBlock(unsigned int nevents, unsigned int offset);
-
-    bool decodePacketPorts(quadlet_t *data, unsigned int nevents, unsigned int dbc);
-
-    int decodeMBLAEventsToPort(AmdtpAudioPort *, quadlet_t *data, unsigned int offset, unsigned int nevents);
-void updatePreparedState(
+int provideSilenceToPort(AmdtpAudioPort *p, unsigned int offset, unsigned int nevents
-
-    int m_dimension;
-    unsigned int m_syt_interval;
-
-    uint64_t m_dropped; /// FIXME:debug
-    uint64_t m_last_dropped; /// FIXME:debug
-    uint64_t m_last_syt; /// FIXME:debug
-    uint64_t m_last_now; /// FIXME:debug
-    int m_last_good_cycle; /// FIXME:debug
-    uint64_t m_last_timestamp; /// last timestamp (in ticks)
-    uint64_t m_last_timestamp2; /// last timestamp (in ticks)
-    uint64_t m_last_timestamp_at_period_ticks;
-};
-

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

@@ -49 +49 @@
-{}
-
-/**
- * @return
- */
-bool AmdtpTransmitStreamProcessor::init() {
-
-    debugOutput( DEBUG_LEVEL_VERBOSE, "Initializing (%p)...\n", this);
-    // call the parent init
-    // this has to be done before allocating the buffers,
-    // because this sets the buffersizes from the processormanager
-    if(!StreamProcessor::init()) {
-        debugFatal("Could not do base class init (%p)\n",this);
-        return false;
-    }
-    return true;
-}
-
-enum raw1394_iso_disposition
-AmdtpTransmitStreamProcessor::getPacket(unsigned char *data, unsigned int *length,
@@ -73 +57 @@
-    if (cycle<0) {
-        debugOutput(DEBUG_LEVEL_ULTRA_VERBOSE,"Xmit handler for cycle %d, (running=%d)\n",
-m_running
+isRunning()
-        *tag = 0;
-        *sy = 0;
@@ -81 +65 @@
-
-    debugOutput(DEBUG_LEVEL_ULTRA_VERBOSE,"Xmit handler for cycle %d, (running=%d)\n",
-m_running
+isRunning()
-
-    if (addCycles(m_last_cycle, 1) != cycle) {
@@ -99 +83 @@
-    /* Our node ID can change after a bus reset, so it is best to fetch
-     * our node ID for each packet. */
-get
+m_handler->getLocal
-
-    packet->dbs = m_dimension;
@@ -126 +110 @@
-
-#ifdef DEBUG
-m_running
+isRunning()
-        debugWarning("Requesting packet for cycle %04d which is in the past (now=%04dcy)\n",
-            cycle, now_cycles);
@@ -139 +123 @@
-    // to be 'running'
-    // NOTE: this works only at startup
-m_running
+isRunning()
-            debugOutput(DEBUG_LEVEL_VERBOSE, "Xmit StreamProcessor %p started running at cycle %d\n",this, cycle);
-            m_running=true;
-    }
-
@@ -169 +152 @@
-    const int max_cycles_to_transmit_early = 5;
-
-m_running
+isRunning()
-        debugOutput(DEBUG_LEVEL_ULTRA_VERBOSE,
-                    "Not running (%d) or buffer not enabled (enabled=%d)\n",
-m_running
+isRunning()
-
-        // not running or not enabled
@@ -357 +340 @@
-}
-
+unsigned int
+AmdtpTransmitStreamProcessor::getEventsPerFrame()
+{
+    return m_dimension;
+}
+
+unsigned int
+AmdtpTransmitStreamProcessor::getUpdatePeriod()
+{
+    return m_syt_interval;
+}
+
+
-unsigned int AmdtpTransmitStreamProcessor::fillDataPacketHeader(
-        struct iec61883_packet *packet, unsigned int* length,
@@ -417 +413 @@
-    m_data_buffer->setTickOffset(0);
-
-
-
-
-
-
-
+// 
+// 
+// 
+// 
+// 
+// 
-
-    // we should prefill the event buffer
@@ -433 +429 @@
-}
-
-
+Child
-    m_PeriodStat.setName("XMT PERIOD");
-    m_PacketStat.setName("XMT PACKET");
@@ -620 +616 @@
-bool AmdtpTransmitStreamProcessor::prepareForEnable(uint64_t time_to_enable_at) {
-
-
-
-
-
+// 
+// 
+// 
+// 
-
-    return true;

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

@@ -85 +85 @@
-                    int cycle, unsigned int dropped, unsigned int max_length);
-
-
+
+
+
+
-    bool reset();
-
+Child
-
-    bool prepareForStop();

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

@@ -55 +55 @@
-    *sy = 0;
-
-
-    return RAW1394_ISO_OK;
-}
-
-int IsoStream::getNodeId() {
-    if (m_handler) {
-        return m_handler->getLocalNodeId();
-    }
-    return -1;
-}
-
-
-void IsoStream::dumpInfo()
-{
-
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Address        : %p\n",this);
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Stream type    : %s\n",
-            (this->getStreamType()==eST_Receive ? "Receive" : "Transmit"));
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Port, Channel  : %d, %d\n",
-            m_port, m_channel);
-
-}
-
-bool IsoStream::setChannel(int c) {
-    debugOutput( DEBUG_LEVEL_VERBOSE, "setting channel for (%p) to %d\n",this, c);
-
-    m_channel=c;
-    return true;
-}
-
-
-bool IsoStream::reset() {
-    debugOutput( DEBUG_LEVEL_VERBOSE, "enter...\n");
-    return true;
-}
-
-bool IsoStream::prepare() {
-    debugOutput( DEBUG_LEVEL_VERBOSE, "enter...\n");
-    return true;
-}
-
-bool IsoStream::init() {
-    debugOutput( DEBUG_LEVEL_VERBOSE, "enter...\n");
-    return true;
-
-}
-
@@ -109 +71 @@
-void IsoStream::clearHandler() {
-    debugOutput( DEBUG_LEVEL_VERBOSE, "clearing handler of isostream %p\n", this);
+    m_handler=0;
+}
-
-
+
+
+
+
+
+
+
+
-
-}
-
-
-}

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

@@ -74 +74 @@
-        virtual unsigned int getMaxPacketSize() {return 1024;}; //FIXME: arbitrary
-
-        virtual bool init();
-
-        virtual enum raw1394_iso_disposition
-                putPacket(unsigned char *data, unsigned int length,
@@ -86 +84 @@
-
-        void dumpInfo();
-
-        int getNodeId();
-
-        virtual bool reset();
-        virtual bool prepare();
-
-    protected:

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

@@ -111 +111 @@
-};
-
+bool Port::setName(std::string name) {
+    debugOutput( DEBUG_LEVEL_VERBOSE, "Setting name to %s for port %s\n",name.c_str(),m_Name.c_str());
+
+    if (m_State != E_Created) {
+        debugFatal("Port (%s) not in E_Created state: %d\n",m_Name.c_str(),m_State);
+        return false;
+    }
+    m_Name=name;
+    return true;
+}
+
+bool Port::setBufferSize(unsigned int newsize) {
+    debugOutput( DEBUG_LEVEL_VERBOSE, "Setting buffersize to %d for port %s\n",newsize,m_Name.c_str());
+    if (m_State != E_Created) {
+        debugFatal("Port (%s) not in E_Created state: %d\n",m_Name.c_str(),m_State);
+        return false;
+    }
+    m_buffersize=newsize;
+    return true;
+}
+
+unsigned int Port::getEventSize() {
+    switch (m_DataType) {
+        case E_Float:
+            return sizeof(float);
+        case E_Int24: // 24 bit 2's complement, packed in a 32bit integer (LSB's)
+            return sizeof(uint32_t);
+        case E_MidiEvent:
+            return sizeof(uint32_t);
+        default:
+            return 0;
+    }
+}
+
+bool Port::setDataType(enum E_DataType d) {
+    debugOutput( DEBUG_LEVEL_VERBOSE, "Setting datatype to %d for port %s\n",(int) d,m_Name.c_str());
+    if (m_State != E_Created) {
+        debugFatal("Port (%s) not in E_Created state: %d\n",m_Name.c_str(),m_State);
+        return false;
+    }
+
+    // do some sanity checks
+    bool type_is_ok=false;
+    switch (m_PortType) {
+        case E_Audio:
+            if(d == E_Int24) type_is_ok=true;
+            if(d == E_Float) type_is_ok=true;
+            break;
+        case E_Midi:
+            if(d == E_MidiEvent) type_is_ok=true;
+            break;
+        case E_Control:
+            if(d == E_Default) type_is_ok=true;
+            break;
+        default:
+            break;
+    }
+
+    if(!type_is_ok) {
+        debugFatal("Datatype not supported by this type of port!\n");
+        return false;
+    }
+
+    m_DataType=d;
+    return true;
+}
+
+bool Port::setSignalType(enum E_SignalType s) {
+    debugOutput( DEBUG_LEVEL_VERBOSE, "Setting signaltype to %d for port %s\n",(int)s,m_Name.c_str());
+    if (m_State != E_Created) {
+        debugFatal("Port (%s) not in E_Created state: %d\n",m_Name.c_str(),m_State);
+        return false;
+    }
+
+    // do some sanity checks
+    bool type_is_ok=false;
+    switch (m_PortType) {
+        case E_Audio:
+            if(s == E_PeriodSignalled) type_is_ok=true;
+            break;
+        case E_Midi:
+            if(s == E_PacketSignalled) type_is_ok=true;
+            break;
+        case E_Control:
+            if(s == E_PeriodSignalled) type_is_ok=true;
+            break;
+        default:
+            break;
+    }
+    if(!type_is_ok) {
+        debugFatal("Signalling type not supported by this type of port!\n");
+        return false;
+    }
+    m_SignalType=s;
+    return true;
+}
+
+bool Port::setBufferType(enum E_BufferType b) {
+    debugOutput( DEBUG_LEVEL_VERBOSE, "Setting buffer type to %d for port %s\n",(int)b,m_Name.c_str());
+    if (m_State != E_Created) {
+        debugFatal("Port (%s) not in E_Created state: %d\n",m_Name.c_str(),m_State);
+        return false;
+    }
+    // do some sanity checks
+    bool type_is_ok=false;
+    switch (m_PortType) {
+        case E_Audio:
+            if(b == E_PointerBuffer) type_is_ok=true;
+            break;
+        case E_Midi:
+            if(b == E_RingBuffer) type_is_ok=true;
+            break;
+        case E_Control:
+            break;
+        default:
+            break;
+    }
+    if(!type_is_ok) {
+        debugFatal("Buffer type not supported by this type of port!\n");
+        return false;
+    }
+    m_BufferType=b;
+    return true;
+}
+
+bool Port::useExternalBuffer(bool b) {
+    // If called on an initialised stream but the request isn't for a change silently
+    // allow it (relied on by C API as used by jack backend driver)
+    if (m_State==E_Initialized && m_use_external_buffer==b)
+        return true;
+
+    debugOutput( DEBUG_LEVEL_VERBOSE, "Setting external buffer use to %d for port %s\n",(int)b,m_Name.c_str());
+
+    if (m_State != E_Created) {
+        debugFatal("Port (%s) not in E_Created state: %d\n",m_Name.c_str(),m_State);
+        return false;
+    }
+    m_use_external_buffer=b;
+    return true;
+}
+
+// buffer handling api's for pointer buffers
+/**
+ * Get the buffer address (being the external or the internal one).
+ *
+ * @param buff
+ */
+void *Port::getBufferAddress() {
+    assert(m_BufferType==E_PointerBuffer);
+    return m_buffer;
+};
+
+/**
+ * Set the external buffer address.
+ * only call this when you have specified that you will use
+ * an external buffer before doing the init()
+ *
+ * @param buff
+ */
+void Port::setExternalBufferAddress(void *buff) {
+    assert(m_BufferType==E_PointerBuffer);
+    assert(m_use_external_buffer); // don't call this with an internal buffer!
+    m_buffer=buff;
+};
+
+// buffer handling api's for ringbuffers
+bool Port::writeEvent(void *event) {
+
+#ifdef DEBUG
+    if (m_State != E_Initialized) {
+        debugFatal("Port (%s) not in E_Initialized state: %d\n",m_Name.c_str(),m_State);
+        return false;
+    }
+    
+    if(m_BufferType!=E_RingBuffer) {
+        debugError("operation not allowed on non E_RingBuffer ports\n");
+        show();
+        return false;
+    }
+    assert(m_ringbuffer);
+#endif
+
+    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "Writing event %08X with size %d to port %s\n",*((quadlet_t *)event),m_eventsize, m_Name.c_str());
+
+    return (ffado_ringbuffer_write(m_ringbuffer, (char *)event, m_eventsize)==m_eventsize);
+}
+
+bool Port::readEvent(void *event) {
+
+#ifdef DEBUG
+    if (m_State != E_Initialized) {
+        debugFatal("Port (%s) not in E_Initialized state: %d\n",m_Name.c_str(),m_State);
+        return false;
+    }
+    
+    if(m_BufferType!=E_RingBuffer) {
+        debugError("operation not allowed on non E_RingBuffer ports\n");
+        show();
+        return false;
+    }
+    assert(m_ringbuffer);
+#endif
+
+    
+    unsigned int read=ffado_ringbuffer_read(m_ringbuffer, (char *)event, m_eventsize);
+    
+    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "Reading event %X with size %d from port %s\n",*((quadlet_t *)event),m_eventsize,m_Name.c_str());
+
+
+    return (read==m_eventsize);
+}
+
+int Port::writeEvents(void *event, unsigned int nevents) {
+
+#ifdef DEBUG
+    if (m_State != E_Initialized) {
+        debugFatal("Port (%s) not in E_Initialized state: %d\n",m_Name.c_str(),m_State);
+        return false;
+    }
+    
+    if(m_BufferType!=E_RingBuffer) {
+        debugError("operation not allowed on non E_RingBuffer ports\n");
+        show();
+        return false;
+    }
+    assert(m_ringbuffer);
+#endif
+
+
+    unsigned int bytes2write=m_eventsize*nevents;
+
+    unsigned int written=ffado_ringbuffer_write(m_ringbuffer, (char *)event,bytes2write)/m_eventsize;
+
+#ifdef DEBUG
+    if(written) {
+        unsigned int i=0;
+        quadlet_t * tmp=(quadlet_t *)event;
+        debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "Written %d events (",written);
+        for (i=0;i<written;i++) {
+            debugOutputShort(DEBUG_LEVEL_VERY_VERBOSE, "%X ", *(tmp+i));
+        }
+        debugOutputShort(DEBUG_LEVEL_VERY_VERBOSE, ") to port %s\n",m_Name.c_str());
+    }
+#endif
+
+    return written;
+
+}
+
+int Port::readEvents(void *event, unsigned int nevents) {
+
+#ifdef DEBUG
+    if (m_State != E_Initialized) {
+        debugFatal("Port (%s) not in E_Initialized state: %d\n",m_Name.c_str(),m_State);
+        return false;
+    }
+    if(m_BufferType!=E_RingBuffer) {
+        debugError("operation not allowed on non E_RingBuffer ports\n");
+        show();
+        return false;
+    }
+    assert(m_ringbuffer);
+#endif
+
+
+    unsigned int bytes2read=m_eventsize*nevents;
+
+    unsigned int read=ffado_ringbuffer_read(m_ringbuffer, (char *)event, bytes2read)/m_eventsize;
+
+#ifdef DEBUG
+    if(read) {
+        unsigned int i=0;
+        quadlet_t * tmp=(quadlet_t *)event;
+        debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "Read %d events (",read);
+        for (i=0;i<read;i++) {
+            debugOutputShort(DEBUG_LEVEL_VERY_VERBOSE, "%X ", *(tmp+i));
+        }
+        debugOutputShort(DEBUG_LEVEL_VERY_VERBOSE, ") from port %s\n",m_Name.c_str());
+    }
+#endif
+
+    return read;
+}
+
+/* rate control */
+bool Port::canRead() {
+    bool byte_present_in_buffer;
+
+    bool retval=false;
+
+    assert(m_ringbuffer);
+
+    byte_present_in_buffer=(ffado_ringbuffer_read_space(m_ringbuffer) >= m_eventsize);
+
+    if(byte_present_in_buffer) {
+
+        if(!m_do_ratecontrol) {
+            return true;
+        }
+
+        if(m_rate_counter <= 0) {
+            // update the counter
+            if(m_average_ratecontrol) {
+                m_rate_counter += m_event_interval;
+                assert(m_rate_counter<m_event_interval);
+            } else {
+                m_rate_counter = m_event_interval;
+            }
+
+            retval=true;
+        } else {
+            debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "Rate limit (%s)! rate_counter=%d \n",m_Name.c_str(),m_rate_counter);
+
+        }
+    }
+
+
+    m_rate_counter -= m_slot_interval;
+
+    // we have to limit the decrement of the ratecounter somehow.
+    // m_rate_counter_minimum is initialized when enabling ratecontrol
+    if(m_rate_counter < m_rate_counter_minimum) {
+        m_rate_counter = m_rate_counter_minimum;
+    }
+
+    return retval;
+}
+
+bool Port::useRateControl(bool use, unsigned int slot_interval,
+                                unsigned int event_interval, bool average) {
+
+    if (use) {
+        debugOutput(DEBUG_LEVEL_VERBOSE, "Enabling rate control for port %s...\n",m_Name.c_str());
+        if(slot_interval>event_interval) {
+            debugWarning("Rate control not needed!\n",m_Name.c_str());
+            m_do_ratecontrol=false;
+            return false;
+        }
+        if(slot_interval==0) {
+            debugFatal("Cannot have slot interval == 0!\n");
+            m_do_ratecontrol=false;
+            return false;
+        }
+        if(event_interval==0) {
+            debugFatal("Cannot have event interval == 0!\n");
+            m_do_ratecontrol=false;
+            return false;
+        }
+        m_do_ratecontrol=use;
+        m_event_interval=event_interval;
+        m_slot_interval=slot_interval;
+        m_rate_counter=0;
+
+        // NOTE: pretty arbitrary, but in average mode this limits the peak stream rate
+        m_rate_counter_minimum=-(2*event_interval);
+
+        m_average_ratecontrol=average;
+
+    } else {
+        debugOutput(DEBUG_LEVEL_VERBOSE, "Disabling rate control for port %s...\n",m_Name.c_str());
+        m_do_ratecontrol=use;
+    }
+    return true;
+}
+
+/// Enable the port. (this can be called anytime)
+void
+Port::enable()  {
+    debugOutput(DEBUG_LEVEL_VERBOSE, "Enabling port %s...\n",m_Name.c_str());
+    m_disabled=false;
+};
+
+/// Disable the port. (this can be called anytime)
+void
+Port::disable() {
+    debugOutput(DEBUG_LEVEL_VERBOSE, "Disabling port %s...\n",m_Name.c_str());
+    m_disabled=false;
+};
+
+
+/* Private functions */
+
+bool Port::allocateInternalBuffer() {
+    int event_size=getEventSize();
+
+    debugOutput(DEBUG_LEVEL_VERBOSE,
+                "Allocating internal buffer of %d events with size %d (%s)\n",
+                m_buffersize, event_size, m_Name.c_str());
+
+    if(m_buffer) {
+        debugWarning("already has an internal buffer attached, re-allocating\n");
+        freeInternalBuffer();
+    }
+
+    m_buffer=calloc(m_buffersize,event_size);
+    if (!m_buffer) {
+        debugFatal("could not allocate internal buffer\n");
+        m_buffersize=0;
+        return false;
+    }
+
+    return true;
+}
+
+void Port::freeInternalBuffer() {
+    debugOutput(DEBUG_LEVEL_VERBOSE,
+                "Freeing internal buffer (%s)\n",m_Name.c_str());
+
+    if(m_buffer) {
+        free(m_buffer);
+        m_buffer=0;
+    }
+}
+
+bool Port::allocateInternalRingBuffer() {
+    int event_size=getEventSize();
+
+    debugOutput(DEBUG_LEVEL_VERBOSE,
+                "Allocating internal buffer of %d events with size %d (%s)\n",
+                m_buffersize, event_size, m_Name.c_str());
+
+    if(m_ringbuffer) {
+        debugWarning("already has an internal ringbuffer attached, re-allocating\n");
+        freeInternalRingBuffer();
+    }
+
+    m_ringbuffer=ffado_ringbuffer_create(m_buffersize * event_size);
+    if (!m_ringbuffer) {
+        debugFatal("could not allocate internal ringbuffer\n");
+        m_buffersize=0;
+        return false;
+    }
+
+    return true;
+}
+
+void Port::freeInternalRingBuffer() {
+    debugOutput(DEBUG_LEVEL_VERBOSE,
+                "Freeing internal ringbuffer (%s)\n",m_Name.c_str());
+
+    if(m_ringbuffer) {
+        ffado_ringbuffer_free(m_ringbuffer);
+        m_ringbuffer=0;
+    }
+}
+
-void Port::show() {
-    debugOutput(DEBUG_LEVEL_VERBOSE,"Name          : %s\n", m_Name.c_str());
@@ -129 +575 @@
-}
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+

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

@@ -251 +251 @@
-        }
-    }
-
-
-
-    return true;
-}

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

@@ -39 +39 @@
-    , m_processor_type ( type )
-    , m_state( ePS_Created )
+    , m_next_state( ePS_Invalid )
+    , m_cycle_to_switch_state( 0 )
-    , m_xruns( 0 )
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
-{
-    // create the timestamped buffer and register ourselves as its client
-    m_data_buffer=new Util::TimestampedBuffer(this);
-
-}
-
@@ -58 +58 @@
-}
-
-void
-StreamProcessor::setState(enum eProcessorState s) {
-    #ifdef DEBUG
-        // check the state transistion
-        debugOutput( DEBUG_LEVEL_VERBOSE, "State transition from %s to %s",
-            ePSToString(m_state), ePSToString(s) );
-    #endif
-    m_state = s;
-}
-
-void StreamProcessor::dumpInfo()
-{
-    debugOutputShort( DEBUG_LEVEL_NORMAL, " StreamProcessor information\n");
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Iso stream info:\n");
-
-    IsoStream::dumpInfo();
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  StreamProcessor info:\n");
-    if (m_handler)
-        debugOutputShort( DEBUG_LEVEL_NORMAL, "  Now                   : %011u\n",m_handler->getCycleTimerTicks());
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Xruns                 : %d\n", m_xruns);
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Running               : %d\n", m_running);
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Enabled               : %s\n", m_disabled ? "No" : "Yes");
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "   enable status        : %s\n", m_is_disabled ? "No" : "Yes");
-
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Nominal framerate     : %u\n", m_manager->getNominalRate());
-    debugOutputShort( DEBUG_LEVEL_NORMAL, "  Device framerate      : Sync: %f, Buffer %f\n",
-        24576000.0/getSyncSource().m_data_buffer->getRate(),
-        24576000.0/m_data_buffer->getRate()
-        );
-
-    m_data_buffer->dumpInfo();
-
-    m_PeriodStat.dumpInfo();
-    m_PacketStat.dumpInfo();
-//     m_WakeupStat.dumpInfo();
-}
-
-bool StreamProcessor::init()
-{
-    debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "enter...\n");
-    m_data_buffer->init();
-    return IsoStream::init();
-}
-
-/**
- * Resets the frame counter, the xrun counter, the ports and the iso stream.
- * @return true if reset succeeded
- */
-bool StreamProcessor::reset() {
-
-    debugOutput( DEBUG_LEVEL_VERBOSE, "Resetting...\n");
-
-    // reset the event buffer, discard all content
-    if (!m_data_buffer->reset()) {
-        debugFatal("Could not reset data buffer\n");
-        return false;
-    }
-
-    resetXrunCounter();
-
-    // loop over the ports to reset them
-    if (!PortManager::resetPorts()) {
-        debugFatal("Could not reset ports\n");
-        return false;
-    }
-
-    // reset the iso stream
-    if (!IsoStream::reset()) {
-        debugFatal("Could not reset isostream\n");
-        return false;
-    }
-    return true;
-
-}
-
-bool StreamProcessor::prepareForEnable(uint64_t time_to_enable_at) {
-    debugOutput(DEBUG_LEVEL_VERBOSE," StreamProcessor::prepareForEnable for (%p)\n",this);
-    debugOutput(DEBUG_LEVEL_VERBOSE,"  Now                   : %011u\n",m_handler->getCycleTimerTicks());
-    debugOutput(DEBUG_LEVEL_VERBOSE,"  Enable at             : %011u\n",time_to_enable_at);
-    m_data_buffer->dumpInfo();
-    return true;
-}
-
-bool StreamProcessor::prepareForDisable() {
-    debugOutput(DEBUG_LEVEL_VERBOSE," StreamProcessor::prepareForDisable for (%p)\n",this);
-    debugOutput(DEBUG_LEVEL_VERBOSE,"  Now                   : %011u\n",m_handler->getCycleTimerTicks());
-    m_data_buffer->dumpInfo();
-    return true;
-}
-
-bool StreamProcessor::prepare() {
-
-    debugOutput( DEBUG_LEVEL_VERBOSE, "Preparing...\n");
-    if(!m_manager) {
-        debugFatal("Not attached to a manager!\n");
-        return -1;
-    }
-
-    // init the ports
-    // loop over the ports to reset them
-    PortManager::preparePorts();
-
-    // reset the iso stream
-    IsoStream::prepare();
-
-    return true;
-
-}
-
-StreamProcessor&
-StreamProcessor::getSyncSource()
-{
-    return m_manager->getSyncSource();
-};
-
-int StreamProcessor::getBufferFill() {
-//     return m_data_buffer->getFrameCounter();
-    return m_data_buffer->getBufferFill();
-}
-
-uint64_t StreamProcessor::getTimeNow() {
-    return m_handler->getCycleTimerTicks();
-}
-
-
-int StreamProcessor::getMaxFrameLatency() {
@@ -191 +70 @@
-}
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-    uint64_t time_at_period=getTimeAtPeriod();
-
@@ -260 +97 @@
-    // pass before these packets are processed. Adding this extra term makes that
-    // the period boundary is signalled later
-
+m_manager->
-
-    uint64_t cycle_timer=m_handler->getCycleTimerTicks();
@@ -278 +115 @@
-}
-
-
+
+
+
-    return (uint64_t)((float)getTimeAtPeriod() * TICKS_PER_USEC);
-}
-
-bool StreamProcessor::dropFrames(unsigned int nbframes) {
-    debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "StreamProcessor::dropFrames(%d)\n", nbframes);
-    return m_data_buffer->dropFrames(nbframes);
-}
-
-/**
- * Resets the xrun counter, in a atomic way. This
- * is thread safe.
- */
-void StreamProcessor::resetXrunCounter() {
-    ZERO_ATOMIC((SInt32 *)&m_xruns);
-}
-
-void StreamProcessor::setVerboseLevel(int l) {
-    setDebugLevel(l);
-    IsoStream::setVerboseLevel(l);
-    PortManager::setVerboseLevel(l);
-    m_data_buffer->setVerboseLevel(l);
-}
-
-uint64_t
-
+
+
-    if (getType() == ePT_Receive) {
-        ffado_timestamp_t next_period_boundary=m_data_buffer->getTimestampFromHead(m_manager->getPeriodSize());
@@ -333 +153 @@
-}
-
+float
+StreamProcessor::getTicksPerFrame()
+{
+    assert(m_data_buffer != NULL);
+    return m_data_buffer->getRate();
+}
+
-bool
-
+
+
+
+
-    if (getType() == ePT_Receive) {
-return m_data_buffer->getFrameCounter()
+can_transfer = fc
-    } else {
-        bool can_transfer;
-        // there has to be enough space to put the frames in
-m_data_buffer->getFrameCounter()
+fc
-        // or the buffer is transparent
-        can_transfer |= m_data_buffer->isTransparent();
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-const char *
-StreamProcessor::ePSToString(enum eProcessorState s) {
-    switch (s) {
+        case ePS_Invalid: return "ePS_Invalid";
-        case ePS_Created: return "ePS_Created";
-Initialized: return "ePS_Initializ
-RunningStream: return "ePS_WaitingForRunning
+Stopped: return "ePS_Stopp
+Stream: return "ePS_WaitingFor
-        case ePS_DryRunning: return "ePS_DryRunning";
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-}
-

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

@@ -50 +50 @@
-                        public PortManager,
-                        public Util::TimestampedBufferClient,
-
+
+
-
-    friend class StreamProcessorManager; // FIXME: get rid of this
@@ -65 +66 @@
-    // this can only be set by the constructor
-    enum eProcessorType m_processor_type;
-
+
+
-protected:
-    ///> the state the streamprocessor is in
-    enum eProcessorState {
+        ePS_Invalid,
-        ePS_Created,
-
-
+
+
+
-        ePS_DryRunning,
-EnabledStream
-StreamEnabled
-DisabledStream
+StreamEnable
+Running
+StreamDisable
-    };
-    
@@ -84 +88 @@
-private:
-    enum eProcessorState m_state;
+    // state switching
+    enum eProcessorState m_next_state;
+    unsigned int m_cycle_to_switch_state;
+    bool updateState();
+    // pretty printing
-    const char *ePSToString(enum eProcessorState);
-
+    bool doStop();
+    bool doWaitForRunningStream();
+    bool doDryRunning();
+    bool doWaitForStreamEnable();
+    bool doRunning();
+    bool doWaitForStreamDisable();
+
+    bool scheduleStateTransition(enum eProcessorState state, uint64_t time_instant);
+    bool scheduleAndWaitForStateTransition(enum eProcessorState state, 
+                                           uint64_t time_instant, 
+                                           enum eProcessorState wait_state);
+public:
+    bool isRunning()
+            {return m_state == ePS_Running;};
+    bool isDryRunning()
+            {return m_state == ePS_DryRunning;};
+
+//--- state stuff (TODO: cleanup)
+    bool startDryRunning(int64_t time_to_start_at);
+    bool startRunning(int64_t time_to_start_at);
+    bool stopDryRunning(int64_t time_to_stop_at);
+    bool stopRunning(int64_t time_to_stop_at);
+
+    // the main difference between init and prepare is that when prepare is called,
+    // the SP is registered to a manager (FIXME: can't it be called by the manager?)
+    bool init();
+    bool prepare();
+    ///> stop the SP from running or dryrunning
+    bool stop();
-// constructor/destructor
-public:
@@ -109 +147 @@
-
-    // the receive interface accepts packets and provides frames
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
-        putPacket(unsigned char *data, unsigned int length,
-                  unsigned char channel, unsigned char tag, unsigned char sy,
-
-
-
-
-
-
+
+
+
+
+
-    virtual bool processReadBlock(char *data, unsigned int nevents, unsigned int offset)
-        {debugWarning("call not allowed\n"); return false;};
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
-
-    // move to private?
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
-
-//--- data buffering and accounting
@@ -170 +208 @@
-         *         false if it can't
-         */
-virtual 
+
-
-        /**
@@ -181 +219 @@
-         * @return true if the operation was successful
-         */
-virtual 
+
-
-        /**
@@ -215 +253 @@
-         * @return the time in internal units
-         */
-virtual 
+
-
-        uint64_t getTimeNow();
@@ -230 +268 @@
-         * @param d sync delay
-         */
-=
+ = 
-
-        /**
@@ -247 +285 @@
-         * @return maximal frame latency
-         */
-
-
-
+
-
-        float getTicksPerFrame();
@@ -256 +292 @@
-
-        int getBufferFill();
+
+        // Child implementation interface
+        /**
+        * @brief prepare the child SP
+        * @return true if successful, false otherwise
+        * @pre the m_manager pointer points to a valid manager
+        * @post getEventsPerFrame() returns the correct value
+        * @post getEventSize() returns the correct value
+        * @post getUpdatePeriod() returns the correct value
+        * @post processPacketHeader(...) can be called
+        * @post processPacketData(...) can be called
+        */
+        virtual bool prepareChild() = 0;
+        /**
+         * @brief get the number of events contained in one frame
+         * @return the number of events contained in one frame
+         */
+        virtual unsigned int getEventsPerFrame() = 0;
+
+        /**
+         * @brief get the size of one frame in bytes
+         * @return the size of one frame in bytes
+         */
+        virtual unsigned int getEventSize() = 0;
+
+        /**
+         * @brief get the nominal number of frames between buffer updates
+         * @return the nominal number of frames between buffer updates
+         */
+        virtual unsigned int getUpdatePeriod() = 0;
-
-    protected:

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

@@ -33 +33 @@
-#define PREPARE_TIMEOUT_MSEC 4000
-#define ENABLE_TIMEOUT_MSEC 4000
+
+// allows to add some processing margin. This shifts the time
+// at which the buffer is transfer()'ed, making things somewhat
+// more robust. It should be noted though that shifting the transfer
+// time to a later time instant also causes the xmit buffer fill to be
+// lower on average.
+#define FFADO_SIGNAL_DELAY_TICKS 3072
-
-namespace Streaming {
@@ -53 +60 @@
-StreamProcessorManager::~StreamProcessorManager() {
-    if (m_isoManager) delete m_isoManager;
-
-}
-
@@ -90 +96 @@
-
-    debugFatal("Unsupported processor type!\n");
-
-    return false;
-}
@@ -102 +107 @@
-
-        for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
-
-++it ) {
-
+  
+  ++it )
+
-            if ( *it == processor ) {
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
-                }
+                return true;
+            }
-        }
-    }
@@ -124 +124 @@
-    if (processor->getType()==StreamProcessor::ePT_Transmit) {
-        for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
-
-++it ) {
-
+  
+  ++it )
+
-            if ( *it == processor ) {
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
-                }
+                return true;
+            }
-        }
-    }
-
-    debugFatal("Processor (%p) not found!\n",processor);
-
-    return false; //not found
-
-}
-
-bool StreamProcessorManager::setSyncSource(StreamProcessor *s) {
-    debugOutput( DEBUG_LEVEL_VERBOSE, "Setting sync source to (%p)\n", s);
-
-    m_SyncSource=s;
-    return true;
@@ -160 +152 @@
-{
-    debugOutput( DEBUG_LEVEL_VERBOSE, "enter...\n");
-
-    m_isoManager = new IsoHandlerManager(m_thread_realtime, m_thread_priority + 1);
-
-    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");
@@ -177 +164 @@
-
-    m_xrun_happened=false;
-
-    return true;
-}
@@ -195 +181 @@
-    }
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-    debugOutput( DEBUG_LEVEL_VERBOSE, "Prepare Receive processors...\n");
-    for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
@@ -228 +216 @@
-        }
-    }
-
-    debugOutput( DEBUG_LEVEL_VERBOSE, "Prepare Transmit processors...\n");
-    for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
@@ -248 +235 @@
-        return false;
-    }
-
-    return true;
-}
-
+bool StreamProcessorManager::startDryRunning() {
+    debugOutput( DEBUG_LEVEL_VERBOSE, "Waiting for StreamProcessor streams to start dry-running...\n");
+    for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
+            it != m_ReceiveProcessors.end();
+            ++it ) {
+        if(!(*it)->startDryRunning(-1)) {
+            debugError("Could not put SP %p into the dry-running state\n", *it);
+            return false;
+        }
+    }
+    for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
+            it != m_TransmitProcessors.end();
+            ++it ) {
+        if(!(*it)->startDryRunning(-1)) {
+            debugError("Could not put SP %p into the dry-running state\n", *it);
+            return false;
+        }
+    }
+    debugOutput( DEBUG_LEVEL_VERBOSE, " StreamProcessor streams dry-running...\n");
+    return true;
+}
+
-bool StreamProcessorManager::syncStartAll() {
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
-
-    // now find out how long we have to delay the wait operation such that
@@ -318 +277 @@
-    }
-
+    // add some processing margin. This only shifts the time
+    // at which the buffer is transfer()'ed. This makes things somewhat
+    // more robust. It should be noted though that shifting the transfer
+    // time to a later time instant also causes the xmit buffer fill to be
+    // lower on average.
+    max_of_min_delay += FFADO_SIGNAL_DELAY_TICKS;
-    debugOutput( DEBUG_LEVEL_VERBOSE, "  %d ticks (%03us %04uc %04ut)...\n", 
-        max_of_min_delay,
@@ -325 +290 @@
-    m_SyncSource->setSyncDelay(max_of_min_delay);
-
-
+
+
+
-    //sleep(2); // FIXME: be smarter here
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
-    debugOutput( DEBUG_LEVEL_VERBOSE, "Waiting for sync...\n");
-
-
+
-    int nb_sync_runs=20;
+    int64_t time_till_next_period;
-    while(nb_sync_runs--) { // or while not sync-ed?
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
-    debugOutput( DEBUG_LEVEL_VERBOSE, " sync at TS=%011llu (%03us %04uc %04ut)...\n", 
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-    // we now should have decent sync info
-    // the buffers of the receive streams should be (approx) empty
-    // the buffers of the xmit streams should be full
-    
-    // note what should the timestamp of the first sample be?
-    
-    // at this point the buffer head timestamp of the transmit buffers can be
@@ -405 +343 @@
-//     int64_t transmit_timestamp = addTicks(m_time_of_transfer, one_ringbuffer_in_ticks);
-
-
-
-
-
-
-
-
+// 
+// 
+// 
+// 
+// 
+// 
+// 
-    
-    dumpInfo();
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-    return true;
-}
-
-void StreamProcessorManager::resetXrunCounters(){
-    debugOutput( DEBUG_LEVEL_VERBOSE, "Resetting xrun flags...\n");
-    for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
-        it != m_ReceiveProcessors.end();
-        ++it )
-    {
-        (*it)->resetXrunCounter();
-    }
-
-    for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
-        it != m_TransmitProcessors.end();
-        ++it ) 
-    {
-        (*it)->resetXrunCounter();
-    }
-}
-
@@ -546 +383 @@
-    debugOutput( DEBUG_LEVEL_VERBOSE, " Receive processors...\n");
-    for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
-    debugOutput( DEBUG_LEVEL_VERBOSE, " Transmit processors...\n");
-    for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
-
-    debugOutput( DEBUG_LEVEL_VERBOSE, "Preparing IsoHandlerManager...\n");
@@ -578 +408 @@
-    }
-
-    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(-1)) {
@@ -590 +414 @@
-    }
-
+    // put all SP's into dry-running state
+    if (!startDryRunning()) {
+        debugFatal("Could not put SP's in dry-running state\n");
+        return false;
+    }
+
-    // start all SP's synchonized
-    if (!syncStartAll()) {
@@ -602 +432 @@
-
-    return true;
-
-}
-
@@ -613 +442 @@
-    // (like the MOTU) need to do a few things before it's safe to turn off the iso
-    // handling.
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
-    }
-
@@ -649 +467 @@
-    debugOutput( DEBUG_LEVEL_VERBOSE, " Receive processors...\n");
-    for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
-    debugOutput( DEBUG_LEVEL_VERBOSE, " Transmit processors...\n");
-    for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
-
-    return true;
@@ -893 +509 @@
-     * 3) Re-enable the SP's
-     */
-
-
-
+
+
+
+
-        return false;
-    }
@@ -960 +577 @@
-    // this is to notify the client of the delay
-    // that we introduced
-=
+ = 
-
-    // we save the 'ideal' time of the transfer at this point,
@@ -1001 +618 @@
-
-        // if this is true, a xrun will occur
- && (*it)->isEnabled()
+
-
-#ifdef DEBUG
@@ -1008 +625 @@
-            (*it)->dumpInfo();
-        }
- && (*it)->isEnabled()
+
-            debugWarning("Xrun on RECV SP %p due to buffer xrun\n",*it);
-            (*it)->dumpInfo();
@@ -1022 +639 @@
-
-        // if this is true, a xrun will occur
- && (*it)->isEnabled()
+
-
-#ifdef DEBUG
@@ -1028 +645 @@
-            debugWarning("Xrun on XMIT SP %p due to ISO xrun\n",*it);
-        }
- && (*it)->isEnabled()
+
-            debugWarning("Xrun on XMIT SP %p due to buffer xrun\n",*it);
-        }
@@ -1049 +666 @@
-bool StreamProcessorManager::transfer() {
-
-
+Y_VER
-    bool retval=true;
-dryRun
-dryRun
+transfer
+transfer
-    return retval;
-}
@@ -1066 +683 @@
-
-bool StreamProcessorManager::transfer(enum StreamProcessor::eProcessorType t) {
-...\n"
+ for type (%d)...\n", t
-    bool retval = true;
-    // a static cast could make sure that there is no performance
@@ -1089 +706 @@
-                it != m_ReceiveProcessors.end();
-                ++it ) {
-
-            if(!(*it)->getFrames(m_period, receive_timestamp)) {
-Output(DEBUG_LEVEL_VERBOSE,
+Warning(
-                            m_period, m_time_of_transfer,*it);
-                retval &= false; // buffer underrun
@@ -1110 +726 @@
-
-            if(!(*it)->putFrames(m_period, transmit_timestamp)) {
-Output(DEBUG_LEVEL_VERBOSE, 
+Warning(
-                        m_period, transmit_timestamp, *it);
-                retval &= false; // buffer underrun

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

@@ -62 +62 @@
-    bool stop();
-
+    bool startDryRunning();
-    bool syncStartAll();
-
@@ -67 +68 @@
-    bool registerProcessor(StreamProcessor *processor); ///< start managing a streamprocessor
-    bool unregisterProcessor(StreamProcessor *processor); ///< stop managing a streamprocessor
-
-    bool enableStreamProcessors(uint64_t time_to_enable_at); /// enable registered StreamProcessors
-    bool disableStreamProcessors(); /// disable registered StreamProcessors
-
-    void setPeriodSize(unsigned int period);
@@ -99 +97 @@
-
-private:
-    void resetXrunCounters();
-
-    int m_delayed_usecs;

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

@@ -437 +437 @@
-
-    debugOutput( DEBUG_LEVEL_VERBOSE, "Preparing iso receive handler (%p)\n",this);
-
-
-
-
+
+
+
+
+
+
-
-    if(m_irq_interval > 1) {

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

@@ -327 +327 @@
-        unsigned int irq_interval=packets_per_period / MINIMUM_INTERRUPTS_PER_PERIOD;
-        if(irq_interval <= 0) irq_interval=1;
+        // FIXME: test
+        irq_interval=1;
+        #warning Using fixed irq_interval
+
-#else
-        // hardware interrupts occur when one DMA block is full, and the size of one DMA

branches/ppalmers-streaming/src/libutil/TimestampedBuffer.cpp

@@ -147 +147 @@
-        diff += m_wrap_at;
-    }
-
+
-    float rate=((float)diff)/((float) m_update_period);
+    if (rate<0.0) debugError("rate < 0! (%f)\n",rate);
-    if (fabsf(m_nominal_rate - rate)>(m_nominal_rate*0.1)) {
-        debugWarning("(%p) rate (%10.5f) more that 10%% off nominal (rate=%10.5f, diff="TIMESTAMP_FORMAT_SPEC", update_period=%d)\n",
-                     this, rate,m_nominal_rate,diff, m_update_period);
-
+
-        return m_nominal_rate;
-    } else {
@@ -379 +380 @@
-
-    if (m_transparent) {
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
-        setBufferTailTimestamp(ts);
-    } else {
@@ -446 +428 @@
-    unsigned int read_size=nframes*m_event_size*m_events_per_frame;
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
-}
-
@@ -712 +695 @@
-    EXIT_CRITICAL_SECTION;
-
-
+Y_VER
-                                          TIMESTAMP_FORMAT_SPEC" => "TIMESTAMP_FORMAT_SPEC", NTS="
-                                          TIMESTAMP_FORMAT_SPEC", DLL2=%f, RATE=%f\n",

branches/ppalmers-streaming/src/libutil/TimestampedBuffer.h

@@ -150 +150 @@
-    // dll stuff
-    bool setNominalRate(float r);
+    float getNominalRate() {return m_nominal_rate;};
-    float getRate();
-

branches/ppalmers-streaming/tests/test-cycletimer.cpp

@@ -296 +296 @@
-
-    s->setVerboseLevel(DEBUG_LEVEL_VERBOSE);
-
-    if (!s->init()) {
-        debugOutput(DEBUG_LEVEL_NORMAL, "Could not init IsoStream\n");
-        goto finish;
-    }
-
-    s->setChannel(0);
-

branches/ppalmers-streaming/tests/test-sytmonitor.cpp

@@ -233 +233 @@
-
-            monitors[i]->setVerboseLevel(DEBUG_LEVEL_VERBOSE);
-
-            if (!monitors[i]->init()) {
-                debugOutput(DEBUG_LEVEL_NORMAL, "Could not init SytMonitor %d\n", i);
-                goto finish;
-            }
-
-            monitors[i]->setChannel(arguments.args[i].channel);
-