Changeset 399

Timestamp:

02/17/07 06:22:15 (17 years ago)

Author:

pieterpalmers

Message:

- code cleanup
- introduce sync delay concept to fix latency issues due to intermediate ISO buffering
- made SytMonitor? use cycletimer.h functions

Files:

• branches/streaming-rework/src/libstreaming/AmdtpStreamProcessor.cpp (modified) (9 diffs)
• branches/streaming-rework/src/libstreaming/AmdtpStreamProcessor.h (modified) (2 diffs)
• branches/streaming-rework/src/libstreaming/StreamProcessor.cpp (modified) (2 diffs)
• branches/streaming-rework/src/libstreaming/StreamProcessor.h (modified) (3 diffs)
• branches/streaming-rework/src/libstreaming/StreamProcessorManager.cpp (modified) (4 diffs)
• branches/streaming-rework/src/libutil/TimestampedBuffer.cpp (modified) (2 diffs)
• branches/streaming-rework/tests/SytMonitor.cpp (modified) (5 diffs)
• branches/streaming-rework/tests/SytMonitor.h (modified) (1 diff)

branches/streaming-rework/src/libstreaming/AmdtpStreamProcessor.cpp

@@ -39 +39 @@
-
-// in ticks
-9
+6
-// the number of cycles to send a packet in advance of it's timestamp
-#define TRANSMIT_ADVANCE_CYCLES 1U
@@ -132 +132 @@
-    
-#ifdef DEBUG
-cycle_diff < 0
+m_running && (cycle_diff < 0)
-        debugWarning("Requesting packet for cycle %04d which is in the past (now=%04dcy)\n",
-            cycle, now_cycles);
@@ -318 +318 @@
-        return RAW1394_ISO_OK;
-        
+    } else if (now_cycles<cycle) {
+        // we can still postpone the queueing of the packets
+        return RAW1394_ISO_AGAIN;
-    } else { // there is no more data in the ringbuffer
-
@@ -399 +402 @@
-}
-
+int AmdtpTransmitStreamProcessor::getMinimalSyncDelay() {
+    return 0;
+}
+
-bool AmdtpTransmitStreamProcessor::prefill() {
-
@@ -419 +426 @@
-    m_PacketStat.reset();
-    m_WakeupStat.reset();
+    
+    // we have to make sure that the buffer HEAD timestamp
+    // lies in the future for every possible buffer fill case.
+    int offset=(int)(m_ringbuffer_size_frames*m_ticks_per_frame);
+    
+    // we can substract the delay as it introduces
+    // unnescessary delay
+    offset -= m_SyncSource->getSyncDelay();
+    
+    m_data_buffer->setTickOffset(offset);
-    
-    // reset all non-device specific stuff
@@ -505 +522 @@
-    
-    m_data_buffer->setWrapValue(128L*TICKS_PER_SECOND);
-    
-    // we have to make sure that the buffer HEAD timestamp
-    // lies in the future for every possible buffer fill case.
-    m_data_buffer->setTickOffset((int)(m_ringbuffer_size_frames*m_ticks_per_frame));
-    
-    m_data_buffer->prepare();
@@ -1127 +1140 @@
-}
-
+// returns the delay between the actual (real) time of a timestamp as received,
+// and the timestamp that is passed on for the same event. This is to cope with
+// ISO buffering
+int AmdtpReceiveStreamProcessor::getMinimalSyncDelay() {
+    return (int)(m_handler->getWakeupInterval() * m_syt_interval * m_ticks_per_frame);
+//     return RECEIVE_PROCESSING_DELAY;
+}
+
-void AmdtpReceiveStreamProcessor::dumpInfo()
-{
@@ -1160 +1181 @@
-    // pass before these packets are processed. Adding this extra term makes that
-    // the period boundary is signalled later
-
-
+
-
-    // reset all non-device specific stuff
@@ -1236 +1256 @@
-    m_data_buffer->setWrapValue(128L*TICKS_PER_SECOND);
-    
-
-
-
-
-
-
-
-
-
-
-
+
+
-    m_data_buffer->prepare();
-

branches/streaming-rework/src/libstreaming/AmdtpStreamProcessor.h

@@ -113 +113 @@
-    uint64_t getTimeAtPeriodUsecs();
-    uint64_t getTimeAtPeriod();
+    int getMinimalSyncDelay();
-    
-    void setVerboseLevel(int l);
@@ -202 +203 @@
-    uint64_t getTimeAtPeriodUsecs();
-    uint64_t getTimeAtPeriod();
+    int getMinimalSyncDelay();
-        
-    void setVerboseLevel(int l);

branches/streaming-rework/src/libstreaming/StreamProcessor.cpp

@@ -54 +54 @@
-        , m_SyncSource(NULL)
-        , m_ticks_per_frame(0)
+        , m_last_cycle(0)
+        , m_sync_delay(0)
-{
-    // create the timestamped buffer and register ourselves as its client
@@ -172 +174 @@
-}
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
-
-uint64_t StreamProcessor::getTimeNow() {

branches/streaming-rework/src/libstreaming/StreamProcessor.h

@@ -188 +188 @@
-
-        uint64_t getTimeNow();
+        
+        
+        /**
+         * Returns the sync delay. This is the time a syncsource
+         * delays a period signal, e.g. to cope with buffering.
+         * @return the sync delay
+         */
+        int getSyncDelay() {return m_sync_delay;};
+        /**
+         * sets the sync delay
+         * @param d sync delay
+         */
+        void setSyncDelay(int d) {m_sync_delay=d;};
+        
+        /**
+         * Returns the minimal sync delay a SP needs
+         * @return minimal sync delay
+         */
+        virtual int getMinimalSyncDelay() = 0;
-
-        bool setSyncSource(StreamProcessor *s);
@@ -194 +213 @@
-        int getLastCycle() {return m_last_cycle;};
-
+        int getBufferFill();
+        
-    protected:
-        StreamProcessor *m_SyncSource;
@@ -200 +221 @@
-
-        int m_last_cycle;
+        int m_sync_delay;
-
-};

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

@@ -312 +312 @@
-
-    debugOutput( DEBUG_LEVEL_VERBOSE, " StreamProcessor streams running...\n");
+    
+    debugOutput( DEBUG_LEVEL_VERBOSE, "Finding minimal sync delay...\n");
+
+    int max_of_min_delay=0;
+    int min_delay=0;
+    for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
+            it != m_ReceiveProcessors.end();
+            ++it ) {
+        min_delay=(*it)->getMinimalSyncDelay();
+        if(min_delay>max_of_min_delay) max_of_min_delay=min_delay;
+    }
+    
+    for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
+            it != m_TransmitProcessors.end();
+            ++it ) {
+        min_delay=(*it)->getMinimalSyncDelay();
+        if(min_delay>max_of_min_delay) max_of_min_delay=min_delay;
+    }
+    
+    debugOutput( DEBUG_LEVEL_VERBOSE, "  %d ticks\n", max_of_min_delay);
+    m_SyncSource->setSyncDelay(max_of_min_delay);
+    
+    
-    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();
-        }
-
-    }
-    
@@ -338 +346 @@
-            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();
-        }
-    }
-        
@@ -776 +771 @@
-    }
-    
--
+
-    
-    // this is to notify the client of the delay 
@@ -792 +787 @@
-        m_time_of_transfer);
-    
+#ifdef DEBUG
+    int rcv_bf=0, xmt_bf=0;
+    for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
+        it != m_ReceiveProcessors.end();
+        ++it ) {
+        rcv_bf = (*it)->getBufferFill();
+    }
+    for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
+        it != m_TransmitProcessors.end();
+        ++it ) {
+        xmt_bf = (*it)->getBufferFill();
+    }
+    debugOutput( DEBUG_LEVEL_VERBOSE, "XF at %011llu ticks, RBF=%d, XBF=%d, SUM=%d...\n", 
+        m_time_of_transfer,rcv_bf,xmt_bf,rcv_bf+xmt_bf);
+    
+#endif
+
-    xrun_occurred=false;
-    

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

@@ -285 +285 @@
-    if (freebob_ringbuffer_write(m_event_buffer,data,write_size) < write_size) 
-    {
-
+// 
-        return false;
-    }
@@ -311 +311 @@
-    if ((freebob_ringbuffer_read(m_event_buffer,data,read_size)) < read_size) 
-    {
-
+// 
-        return false;
-    }

branches/streaming-rework/tests/SytMonitor.cpp

@@ -34 +34 @@
-
-
-NORMAL
+VERBOSE
-
-
@@ -58 +58 @@
-    struct iec61883_packet *packet = (struct iec61883_packet *) data;
-    unsigned int syt_timestamp=ntohs(packet->syt);
-    unsigned int m_full_timestamp;
-    
-
-
+
+
+
+
+
+
+
+
-        m_port,channel,
-        cycle,syt_timestamp,
@@ -72 +77 @@
-    // reconstruct the full cycle
-    unsigned int cc=m_handler->getCycleTimer();
-    unsigned int cc_ticks=CYCLE_TIMER_TO_TICKS(cc);
-    
-    unsigned int cc_cycles=CYCLE_TIMER_GET_CYCLES(cc);
@@ -83 +87 @@
-    // was received
-    if (cycle>cc_cycles) cc_seconds--;
-
-     // reconstruct the top part of the timestamp using the current cycle number
-    unsigned int now_cycle_masked=cycle & 0xF;
-    unsigned int syt_cycle=CYCLE_TIMER_GET_CYCLES(syt_timestamp);
-    
-// if this is true, wraparound has occurred, undo this wraparound
-if(syt_cycle<now_cycle_masked) syt_cycle += 0x10
+m_full_timestamp_ticks=sytRecvToFullTicks((uint32_t)syt_timestamp, cycle, cc);
+m_full_timestamp=TICKS_TO_CYCLE_TIMER(m_full_timestamp_ticks)
-    
-    // this is the difference in cycles wrt the cycle the
-    // timestamp was received
-    unsigned int delta_cycles=syt_cycle-now_cycle_masked;
-    
-    // reconstruct the cycle part of the timestamp
-    unsigned int new_cycles=cycle + delta_cycles;
-    
-    // if the cycles cause a wraparound of the cycle timer,
-    // perform this wraparound
-    if(new_cycles>7999) {
-        debugOutput(DEBUG_LEVEL_VERY_VERBOSE,
-            "Detected wraparound: %d + %d = %d\n",
-            cycle,delta_cycles,new_cycles);
-        
-        new_cycles-=8000; // wrap around
-        wraparound_occurred=true;
-        
-    }
-    
-    m_full_timestamp = (new_cycles) << 12;
-    
-    // now add the offset part on top of that
-    m_full_timestamp |= (syt_timestamp & 0xFFF);
-    
-    // and the seconds part
-    // if the timestamp causes a wraparound of the cycle timer,
-    // the timestamp lies in the next second
-    // if it didn't, the timestamp lies in this second
-    if (wraparound_occurred) {
-        m_full_timestamp |= ((cc_seconds+1 << 25) & 0xFE000000);
-    } else {
-        m_full_timestamp |= ((cc_seconds << 25) & 0xFE000000);
-    }
-
-    struct cycle_info cif;
-    cif.cycle=cycle;
@@ -136 +102 @@
-    cif.pres_cycle   = CYCLE_TIMER_GET_CYCLES(m_full_timestamp);
-    cif.pres_offset  = CYCLE_TIMER_GET_OFFSET(m_full_timestamp);
-
-
-
-
-
-
+
-    
-    if (cif.pres_offset != (syt_timestamp & 0xFFF)) {

branches/streaming-rework/tests/SytMonitor.h

@@ -60 +60 @@
-                                  
-                void dumpInfo();
+                unsigned int getMaxPacketSize() {return 4096;};
+                unsigned int getPacketsPerPeriod() {return 1;};
-                
-        bool readNextCycleInfo(struct cycle_info *cif);