LimeSDR Windows build: upgraded to LimeSuite 17.06.0

liblimesuite/srcmw/ConnectionSTREAM/ConnectionSTREAMing.cpp

@@ -1,4 +1,4 @@
-    /**
+/**
     @file ConnectionSTREAMing.cpp
     @author Lime Microsystems
     @brief Implementation of STREAM board connection (streaming API)
@@ -20,93 +20,40 @@
 using namespace lime;
 using namespace std;
 
-int ConnectionSTREAM::UploadWFM(const void* const* samples, uint8_t chCount, size_t sample_count, StreamConfig::StreamDataFormat format)
-{
-    WriteRegister(0x000C, 0x3); //channels 0,1
-    WriteRegister(0x000E, 0x2); //12bit samples
-    WriteRegister(0x000D, 0x0004); //WFM_LOAD
-
-    lime::FPGA_DataPacket pkt;
-    size_t samplesUsed = 0;
-
-    const complex16_t* const* src = (const complex16_t* const*)samples;
-    int cnt = sample_count;
-
-    const lime::complex16_t** batch = new const lime::complex16_t*[chCount];
-    while(cnt > 0)
-    {
-        pkt.counter = 0;
-        pkt.reserved[0] = 0;
-        int samplesToSend = cnt > 1360/chCount ? 1360/chCount : cnt;
-        cnt -= samplesToSend;
-
-        for(uint8_t i=0; i<chCount; ++i)
-            batch[i] = &src[i][samplesUsed];
-        samplesUsed += samplesToSend;
-
-        size_t bufPos = 0;
-        lime::fpga::Samples2FPGAPacketPayload(batch, samplesToSend, chCount, format, pkt.data, &bufPos);
-        int payloadSize = (bufPos / 4) * 4;
-        if(bufPos % 4 != 0)
-            lime::error("Packet samples count not multiple of 4");
-        pkt.reserved[2] = (payloadSize >> 8) & 0xFF; //WFM loading
-        pkt.reserved[1] = payloadSize & 0xFF; //WFM loading
-        pkt.reserved[0] = 0x1 << 5; //WFM loading
-
-        long bToSend = 16+payloadSize;
-        int context = BeginDataSending((char*)&pkt, bToSend );
-        if(WaitForSending(context, 250) == false)
-        {
-            FinishDataSending((char*)&pkt, bToSend , context);
-            break;
-        }
-        FinishDataSending((char*)&pkt, bToSend , context);
-    }
-    delete[] batch;
-    /*Give FX3 some time to load samples to FPGA*/
-    std::this_thread::sleep_for(std::chrono::milliseconds(30));
-    if(cnt == 0)
-        return 0;
-    else
-        return ReportError(-1, "Failed to upload waveform");
-}
-
 /** @brief Configures FPGA PLLs to LimeLight interface frequency
 */
 int ConnectionSTREAM::UpdateExternalDataRate(const size_t channel, const double txRate_Hz, const double rxRate_Hz, const double txPhase, const double rxPhase)
 {
-#ifndef NDEBUG
-    lime::debug("ConnectionSTREAM::ConfigureFPGA_PLL(tx=%gMHz, rx=%gMHz)", txRate_Hz/1e6, rxRate_Hz/1e6);
-#endif
+    lime::fpga::FPGA_PLL_clock clocks[2];
+
+    if (channel == 2)
+    {
+        clocks[0].index = 0;
+        clocks[0].outFrequency = rxRate_Hz;
+        clocks[1].index = 1;
+        clocks[1].outFrequency = txRate_Hz;
+        return lime::fpga::SetPllFrequency(this, 4, 30.72e6, clocks, 2);
+    }
+
     const float txInterfaceClk = 2 * txRate_Hz;
     const float rxInterfaceClk = 2 * rxRate_Hz;
     mExpectedSampleRate = rxRate_Hz;
+    const int pll_ind = (channel == 1) ? 2 : 0;
 
-    lime::fpga::FPGA_PLL_clock clocks[2];
-    clocks[0].bypass = false;
     clocks[0].index = 0;
     clocks[0].outFrequency = rxInterfaceClk;
-    clocks[0].phaseShift_deg = 0;
-    clocks[0].findPhase = false;
-    clocks[1].bypass = false;
     clocks[1].index = 1;
     clocks[1].outFrequency = rxInterfaceClk;
     clocks[1].phaseShift_deg = rxPhase;
-    clocks[1].findPhase = false;
-    if (lime::fpga::SetPllFrequency(this, 1, rxInterfaceClk, clocks, 2)!=0)
+    if (lime::fpga::SetPllFrequency(this, pll_ind+1, rxInterfaceClk, clocks, 2)!=0)
         return -1;
 
-    clocks[0].bypass = false;
     clocks[0].index = 0;
     clocks[0].outFrequency = txInterfaceClk;
-    clocks[0].phaseShift_deg = 0;
-    clocks[0].findPhase = false;
-    clocks[1].bypass = false;
     clocks[1].index = 1;
     clocks[1].outFrequency = txInterfaceClk;
     clocks[1].phaseShift_deg = txPhase;
-    clocks[1].findPhase = false;
-    if (lime::fpga::SetPllFrequency(this, 0, txInterfaceClk, clocks, 2)!=0)
+    if (lime::fpga::SetPllFrequency(this, pll_ind, txInterfaceClk, clocks, 2)!=0)
         return -1;
 
     return 0;
@@ -116,11 +63,9 @@ int ConnectionSTREAM::UpdateExternalDataRate(const size_t channel, const double
 */
 int ConnectionSTREAM::UpdateExternalDataRate(const size_t channel, const double txRate_Hz, const double rxRate_Hz)
 {
-#ifndef NDEBUG
-    lime::debug("ConnectionSTREAM::ConfigureFPGA_PLL(tx=%gMHz, rx=%gMHz)", txRate_Hz/1e6, rxRate_Hz/1e6);
-#endif
     const float txInterfaceClk = 2 * txRate_Hz;
     const float rxInterfaceClk = 2 * rxRate_Hz;
+    const int pll_ind = (channel == 1) ? 2 : 0;
     int status = 0;
     uint32_t reg20;
     const double rxPhC1[] = { 91.08, 89.46 };
@@ -128,18 +73,17 @@ int ConnectionSTREAM::UpdateExternalDataRate(const size_t channel, const double
     const double txPhC1[] = { 89.75, 89.61 };
     const double txPhC2[] = { -3.0e-7, 2.71e-7 };
 
-    const std::vector<uint32_t> spiAddr = {0x0021, 0x0022, 0x0023, 0x0024,
-                                           0x0027, 0x002A, 0x0400, 0x040C,
-                                           0x040B, 0x0400, 0x040B, 0x0400};
+    const std::vector<uint32_t> spiAddr = { 0x021, 0x022, 0x023, 0x024, 0x027, 0x02A,
+                                            0x400, 0x40C, 0x40B, 0x400, 0x40B, 0x400};
     const int bakRegCnt = spiAddr.size() - 4;
     auto info = GetDeviceInfo();
-    const int addrLMS7002M = info.addrsLMS7002M.at(0);
     bool phaseSearch = false;
-    if (this->chipVersion == 0x3841 && stoi(info.gatewareRevision) >= 7 && stoi(info.gatewareVersion) >= 2) //0x3840 LMS7002Mr2, 0x3841 LMS7002Mr3
-        if(rxInterfaceClk >= 5e6 || txInterfaceClk >= 5e6)
-            phaseSearch = true;
-    mExpectedSampleRate = rxRate_Hz;
+    if (!(mStreamers.size() > channel && (mStreamers[channel]->rxRunning || mStreamers[channel]->txRunning)))
+        if (this->chipVersion == 0x3841 && stoi(info.gatewareRevision) >= 7 && stoi(info.gatewareVersion) >= 2) //0x3840 LMS7002Mr2, 0x3841 LMS7002Mr3
+            if(rxInterfaceClk >= 5e6 || txInterfaceClk >= 5e6)
+                phaseSearch = true;
 
+    mExpectedSampleRate = rxRate_Hz;
     std::vector<uint32_t> dataWr;
     std::vector<uint32_t> dataRd;
 
@@ -149,78 +93,61 @@ int ConnectionSTREAM::UpdateExternalDataRate(const size_t channel, const double
         dataRd.resize(spiAddr.size());
         //backup registers
         dataWr[0] = (uint32_t(0x0020) << 16);
-        TransactSPI(addrLMS7002M, dataWr.data(), &reg20, 1);
+        ReadLMS7002MSPI(dataWr.data(), &reg20, 1, channel);
 
         dataWr[0] = (1 << 31) | (uint32_t(0x0020) << 16) | 0xFFFD; //msbit 1=SPI write
-        TransactSPI(addrLMS7002M, dataWr.data(), nullptr, 1);
+        WriteLMS7002MSPI(dataWr.data(), 1, channel);
 
         for (int i = 0; i < bakRegCnt; ++i)
             dataWr[i] = (spiAddr[i] << 16);
-        TransactSPI(addrLMS7002M, dataWr.data(), dataRd.data(), bakRegCnt);
-        UpdateThreads(true);
+        ReadLMS7002MSPI(dataWr.data(),dataRd.data(), bakRegCnt, channel);
     }
 
     if(rxInterfaceClk >= 5e6)
     {
         if (phaseSearch)
         {
-            const std::vector<uint32_t> spiData = { 0x0E9F, 0x07FF, 0x5550, 0xE4E4,
-                                                    0xE4E4, 0x0086, 0x028D, 0x00FF,
-                                                    0x5555, 0x02CD, 0xAAAA, 0x02ED};
+            const std::vector<uint32_t> spiData = { 0x0E9F, 0x07FF, 0x5550, 0xE4E4, 0xE4E4, 0x0086,
+                                                    0x028D, 0x00FF, 0x5555, 0x02CD, 0xAAAA, 0x02ED};
             //Load test config
             const int setRegCnt = spiData.size();
             for (int i = 0; i < setRegCnt; ++i)
                 dataWr[i] = (1 << 31) | (uint32_t(spiAddr[i]) << 16) | spiData[i]; //msbit 1=SPI write
-            TransactSPI(addrLMS7002M, dataWr.data(), nullptr, setRegCnt);
+            WriteLMS7002MSPI(dataWr.data(), setRegCnt, channel);
         }
         lime::fpga::FPGA_PLL_clock clocks[2];
-        clocks[0].bypass = false;
         clocks[0].index = 0;
         clocks[0].outFrequency = rxInterfaceClk;
-        clocks[0].phaseShift_deg = 0;
-        clocks[0].findPhase = false;
-        clocks[1].bypass = false;
         clocks[1].index = 1;
         clocks[1].outFrequency = rxInterfaceClk;
         if (this->chipVersion == 0x3841)
             clocks[1].phaseShift_deg = rxPhC1[1] + rxPhC2[1] * rxInterfaceClk;
         else
             clocks[1].phaseShift_deg = rxPhC1[0] + rxPhC2[0] * rxInterfaceClk;
-
         if (phaseSearch)
-        {
             clocks[1].findPhase = true;
-        }
-        else
-        {
-            clocks[1].findPhase = false;
-        }
-        status = lime::fpga::SetPllFrequency(this, 1, rxInterfaceClk, clocks, 2);
+        status = lime::fpga::SetPllFrequency(this, pll_ind+1, rxInterfaceClk, clocks, 2);
     }
     else
-        status = lime::fpga::SetDirectClocking(this, 1, rxInterfaceClk, 90);
+        status = lime::fpga::SetDirectClocking(this, pll_ind+1, rxInterfaceClk, 90);
 
     if(txInterfaceClk >= 5e6)
     {
         if (phaseSearch)
         {
-            const std::vector<uint32_t> spiData = {0x0E9F, 0x07FF, 0x5550, 0xE4E4,
-                                                   0xE4E4, 0x0484};
+            const std::vector<uint32_t> spiData = {0x0E9F, 0x07FF, 0x5550, 0xE4E4, 0xE4E4, 0x0484};
             WriteRegister(0x000A, 0x0000);
             //Load test config
             const int setRegCnt = spiData.size();
             for (int i = 0; i < setRegCnt; ++i)
                 dataWr[i] = (1 << 31) | (uint32_t(spiAddr[i]) << 16) | spiData[i]; //msbit 1=SPI write
-            TransactSPI(addrLMS7002M, dataWr.data(), nullptr, setRegCnt);
+            WriteLMS7002MSPI(dataWr.data(), setRegCnt, channel);
         }
 
         lime::fpga::FPGA_PLL_clock clocks[2];
-        clocks[0].bypass = false;
         clocks[0].index = 0;
         clocks[0].outFrequency = txInterfaceClk;
         clocks[0].phaseShift_deg = 0;
-        clocks[0].findPhase = false;
-        clocks[1].bypass = false;
         clocks[1].index = 1;
         clocks[1].outFrequency = txInterfaceClk;
         if (this->chipVersion == 0x3841)
@@ -233,27 +160,21 @@ int ConnectionSTREAM::UpdateExternalDataRate(const size_t channel, const double
             clocks[1].findPhase = true;
             WriteRegister(0x000A, 0x0200);
         }
-        else
-        {
-            clocks[1].findPhase = false;
-        }
-        status = lime::fpga::SetPllFrequency(this, 0, txInterfaceClk, clocks, 2);
+        status = lime::fpga::SetPllFrequency(this, pll_ind, txInterfaceClk, clocks, 2);
     }
     else
-        status = lime::fpga::SetDirectClocking(this, 0, txInterfaceClk, 90);
+        status = lime::fpga::SetDirectClocking(this, pll_ind, txInterfaceClk, 90);
 
     if (phaseSearch)
     {
         //Restore registers
         for (int i = 0; i < bakRegCnt; ++i)
             dataWr[i] = (1 << 31) | (uint32_t(spiAddr[i]) << 16) | dataRd[i]; //msbit 1=SPI write
-        TransactSPI(addrLMS7002M, dataWr.data(), nullptr, bakRegCnt);
+        WriteLMS7002MSPI(dataWr.data(), bakRegCnt, channel);
         dataWr[0] = (1 << 31) | (uint32_t(0x0020) << 16) | reg20; //msbit 1=SPI write
-        TransactSPI(addrLMS7002M, dataWr.data(), nullptr, 1);
+        WriteLMS7002MSPI(dataWr.data(), 1, channel);
         WriteRegister(0x000A, 0);
-        UpdateThreads();
     }
-
     return status;
 }
 
@@ -266,57 +187,44 @@ int ConnectionSTREAM::ResetStreamBuffers()
     return TransferPacket(ctrPkt);
 }
 
-int ConnectionSTREAM::ReadRawStreamData(char* buffer, unsigned length, int timeout_ms)
+int ConnectionSTREAM::ReadRawStreamData(char* buffer, unsigned length, int epIndex, int timeout_ms)
 {
-        fpga::StopStreaming(this);
+    const unsigned char ep = 0x81;
+    fpga::StopStreaming(this, epIndex);
 
-        ResetStreamBuffers();
-        WriteRegister(0x0008, 0x0100 | 0x2);
-        WriteRegister(0x0007, 1);
+    ResetStreamBuffers();
+    WriteRegister(0x0008, 0x0100 | 0x2);
+    WriteRegister(0x0007, 1);
+    fpga::StartStreaming(this, epIndex);
 
-        fpga::StartStreaming(this);
+    int totalBytesReceived = ReceiveData(buffer,length, epIndex, timeout_ms);
+    fpga::StopStreaming(this, epIndex);
+    AbortReading(ep);
 
-        int handle = BeginDataReading(buffer, length);
-        if (WaitForReading(handle, timeout_ms) == false)
-        {
-            AbortReading();
-        }
-
-        fpga::StopStreaming(this);
-
-        int totalBytesReceived = FinishDataReading(buffer, length, handle);
-
-	return totalBytesReceived;
+    return totalBytesReceived;
 }
 
 /** @brief Function dedicated for receiving data samples from board
-    @param rxFIFO FIFO to store received data
-    @param terminate periodically pooled flag to terminate thread
-    @param dataRate_Bps (optional) if not NULL periodically returns data rate in bytes per second
+    @param stream a pointer to an active receiver stream
 */
-void ConnectionSTREAM::ReceivePacketsLoop(const ThreadData args)
+void ConnectionSTREAM::ReceivePacketsLoop(Streamer* stream)
 {
-    //auto dataPort = args.dataPort;
-    auto terminate = args.terminate;
-    auto dataRate_Bps = args.dataRate_Bps;
-    auto generateData = args.generateData;
-    auto safeToConfigInterface = args.safeToConfigInterface;
-
     //at this point FPGA has to be already configured to output samples
-    const uint8_t chCount = args.channels.size();
-    const auto link = args.channels[0]->config.linkFormat;
+    const uint8_t chCount = stream->mRxStreams.size();
+    const auto link = stream->mRxStreams[0]->config.linkFormat;
     const uint32_t samplesInPacket = (link == StreamConfig::STREAM_12_BIT_COMPRESSED ? 1360 : 1020)/chCount;
+    const unsigned char ep = 0x81;
+    const int chipID = stream->mChipID;
 
-    double latency=0;
+    float latency=0;
     for (int i = 0; i < chCount; i++)
-    {
-        latency += args.channels[i]->config.performanceLatency/chCount;
-    }
+        latency += stream->mRxStreams[i]->config.performanceLatency/chCount;
+
     const unsigned tmp_cnt = (latency * 6)+0.5;
 
     const uint8_t packetsToBatch = (1<<tmp_cnt);
     const uint32_t bufferSize = packetsToBatch*sizeof(FPGA_DataPacket);
-    const uint8_t buffersCount = (tmp_cnt < 3) ? 32 : 16; // must be power of 2
+    const uint8_t buffersCount = 16;
     vector<int> handles(buffersCount, 0);
     vector<char>buffers(buffersCount*bufferSize, 0);
     vector<StreamChannel::Frame> chFrames;
@@ -330,24 +238,18 @@ void ConnectionSTREAM::ReceivePacketsLoop(const ThreadData args)
         return;
     }
 
-    uint8_t activeTransfers = 0;
+    int activeTransfers = 0;
     for (int i = 0; i<buffersCount; ++i)
     {
-        handles[i] = this->BeginDataReading(&buffers[i*bufferSize], bufferSize);
+        handles[i] = this->BeginDataReading(&buffers[i*bufferSize], bufferSize, ep);
         ++activeTransfers;
     }
 
     int bi = 0;
     unsigned long totalBytesReceived = 0; //for data rate calculation
-    int m_bufferFailures = 0;
-    int32_t droppedSamples = 0;
-    int32_t packetLoss = 0;
-
-    vector<uint32_t> samplesCollected(chCount, 0);
-    vector<uint32_t> samplesReceived(chCount, 0);
 
     auto t1 = chrono::high_resolution_clock::now();
-    auto t2 = chrono::high_resolution_clock::now();
+    auto t2 = t1;
 
     std::mutex txFlagsLock;
     condition_variable resetTxFlags;
@@ -360,24 +262,24 @@ void ConnectionSTREAM::ReceivePacketsLoop(const ThreadData args)
         uint32_t reg9;
         port->ReadRegister(0x0009, reg9);
         const uint32_t addr[] = {0x0009, 0x0009};
-        const uint32_t data[] = {reg9 | (1 << 1), reg9 & ~(1 << 1)};
+        const uint32_t data[] = {reg9 | (5 << 1), reg9 & ~(5 << 1)};
         while (not terminate->load())
         {
             std::unique_lock<std::mutex> lck(*spiLock);
             doWork->wait(lck);
             port->WriteRegisters(addr, data, 2);
         }
-    }, this, terminate, &txFlagsLock, &resetTxFlags);
+    }, this, &stream->terminateRx, &txFlagsLock, &resetTxFlags);
 
     int resetFlagsDelay = 128;
     uint64_t prevTs = 0;
-    while (terminate->load() == false)
+    while (stream->terminateRx.load() == false)
     {
-        if(generateData->load())
+        if(stream->generateData.load())
         {
             if(activeTransfers == 0) //stop FPGA when last transfer completes
-                fpga::StopStreaming(this);
-            safeToConfigInterface->notify_all(); //notify that it's safe to change chip config
+                fpga::StopStreaming(this, chipID);
+            stream->safeToConfigInterface.notify_all(); //notify that it's safe to change chip config
             const int batchSize = (this->mExpectedSampleRate/chFrames[0].samplesCount)/10;
             IStreamChannel::Metadata meta;
             for(int i=0; i<batchSize; ++i)
@@ -390,8 +292,7 @@ void ConnectionSTREAM::ReceivePacketsLoop(const ThreadData args)
                         chFrames[ch].samples[j].i = 0;
                         chFrames[ch].samples[j].q = 0;
                     }
-                    uint32_t samplesPushed = args.channels[ch]->Write((const void*)chFrames[ch].samples, chFrames[ch].samplesCount, &meta);
-                    samplesReceived[ch] += chFrames[ch].samplesCount;
+                    uint32_t samplesPushed = stream->mRxStreams[ch]->Write((const void*)chFrames[ch].samples, chFrames[ch].samplesCount, &meta);
                     if(samplesPushed != chFrames[ch].samplesCount)
                         lime::warning("Rx samples pushed %i/%i", samplesPushed, chFrames[ch].samplesCount);
                 }
@@ -401,13 +302,13 @@ void ConnectionSTREAM::ReceivePacketsLoop(const ThreadData args)
         int32_t bytesReceived = 0;
         if(handles[bi] >= 0)
         {
-            if (this->WaitForReading(handles[bi], 1000) == false)
-                ++m_bufferFailures;
-            bytesReceived = this->FinishDataReading(&buffers[bi*bufferSize], bufferSize, handles[bi]);
+            if (this->WaitForReading(handles[bi], 1000) == true)
+                bytesReceived = this->FinishDataReading(&buffers[bi*bufferSize], bufferSize, handles[bi]);
             --activeTransfers;
             totalBytesReceived += bytesReceived;
             if (bytesReceived != int32_t(bufferSize)) //data should come in full sized packets
-                ++m_bufferFailures;
+                for(auto value: stream->mRxStreams)
+                    value->underflow++;
         }
         bool txLate=false;
         for (uint8_t pktIndex = 0; pktIndex < bytesReceived / sizeof(FPGA_DataPacket); ++pktIndex)
@@ -424,20 +325,23 @@ void ConnectionSTREAM::ReceivePacketsLoop(const ThreadData args)
                     lime::info("L %llu", (unsigned long long)pkt[pktIndex].counter);
                     resetTxFlags.notify_one();
                     resetFlagsDelay = packetsToBatch*buffersCount;
-                    if (args.reportLateTx) args.reportLateTx(pkt[pktIndex].counter);
+                    stream->txLastLateTime.store(pkt[pktIndex].counter);
+                    for(auto value: stream->mTxStreams)
+                        value->pktLost++;
                 }
             }
             uint8_t* pktStart = (uint8_t*)pkt[pktIndex].data;
             if(pkt[pktIndex].counter - prevTs != samplesInPacket && pkt[pktIndex].counter != prevTs)
             {
+                int packetLoss = ((pkt[pktIndex].counter - prevTs)/samplesInPacket)-1;
 #ifndef NDEBUG
-                lime::debug("\tRx pktLoss@%i - ts diff: %li  pktLoss: %.1f", pktIndex, pkt[pktIndex].counter - prevTs, float(pkt[pktIndex].counter - prevTs)/samplesInPacket);
+                printf("\tRx pktLoss: ts diff: %li  pktLoss: %i\n", pkt[pktIndex].counter - prevTs, packetLoss);
 #endif
-                packetLoss += (pkt[pktIndex].counter - prevTs)/samplesInPacket;
+                for(auto value: stream->mRxStreams)
+                    value->pktLost += packetLoss;
             }
             prevTs = pkt[pktIndex].counter;
-            if(args.lastTimestamp)
-                args.lastTimestamp->store(pkt[pktIndex].counter);
+            stream->rxLastTimestamp.store(prevTs);
             //parse samples
             vector<complex16_t*> dest(chCount);
             for(uint8_t c=0; c<chCount; ++c)
@@ -450,19 +354,19 @@ void ConnectionSTREAM::ReceivePacketsLoop(const ThreadData args)
                 IStreamChannel::Metadata meta;
                 meta.timestamp = pkt[pktIndex].counter;
                 meta.flags = RingFIFO::OVERWRITE_OLD;
-                uint32_t samplesPushed = args.channels[ch]->Write((const void*)chFrames[ch].samples, samplesCount, &meta, 100);
+                uint32_t samplesPushed = stream->mRxStreams[ch]->Write((const void*)chFrames[ch].samples, samplesCount, &meta, 100);
                 if(samplesPushed != samplesCount)
-                    droppedSamples += samplesCount-samplesPushed;
+                    stream->mRxStreams[ch]->overflow++;
             }
         }
         // Re-submit this request to keep the queue full
-        if(not generateData->load())
+        if(not stream->generateData.load())
         {
             if(activeTransfers == 0) //reactivate FPGA and USB transfers
-                fpga::StartStreaming(this);
+                fpga::StartStreaming(this, chipID);
             for(int i=0; i<buffersCount-activeTransfers; ++i)
             {
-                handles[bi] = this->BeginDataReading(&buffers[bi*bufferSize], bufferSize);
+                handles[bi] = this->BeginDataReading(&buffers[bi*bufferSize], bufferSize, ep);
                 bi = (bi + 1) & (buffersCount-1);
                 ++activeTransfers;
             }
@@ -480,21 +384,13 @@ void ConnectionSTREAM::ReceivePacketsLoop(const ThreadData args)
             //total number of bytes sent per second
             double dataRate = 1000.0*totalBytesReceived / timePeriod;
 #ifndef NDEBUG
-            //each channel sample rate
-            float samplingRate = 1000.0*samplesReceived[0] / timePeriod;
-            lime::debug("Rx: %.3f MB/s, Fs: %.3f MHz, overrun: %i, loss: %i", dataRate / 1000000.0, samplingRate / 1000000.0, droppedSamples, packetLoss);
+            printf("Rx: %.3f MB/s\n", dataRate / 1000000.0);
 #endif
-            samplesReceived[0] = 0;
             totalBytesReceived = 0;
-            m_bufferFailures = 0;
-            droppedSamples = 0;
-            packetLoss = 0;
-
-            if (dataRate_Bps)
-                dataRate_Bps->store((uint32_t)dataRate);
+            stream->rxDataRate_Bps.store((uint32_t)dataRate);
         }
     }
-    this->AbortReading();
+    AbortReading(ep);
     for (int j = 0; j<buffersCount; j++)
     {
         if(handles[bi] >= 0)
@@ -506,35 +402,27 @@ void ConnectionSTREAM::ReceivePacketsLoop(const ThreadData args)
     }
     resetTxFlags.notify_one();
     txReset.join();
-    if (dataRate_Bps)
-        dataRate_Bps->store(0);
+    stream->rxDataRate_Bps.store(0);
 }
 
 /** @brief Functions dedicated for transmitting packets to board
-    @param txFIFO data source FIFO
-    @param terminate periodically pooled flag to terminate thread
-    @param dataRate_Bps (optional) if not NULL periodically returns data rate in bytes per second
+    @param stream an active transmit stream
 */
-void ConnectionSTREAM::TransmitPacketsLoop(const ThreadData args)
+void ConnectionSTREAM::TransmitPacketsLoop(Streamer* stream)
 {
-    //auto dataPort = args.dataPort;
-    auto terminate = args.terminate;
-    auto dataRate_Bps = args.dataRate_Bps;
-
     //at this point FPGA has to be already configured to output samples
     const uint8_t maxChannelCount = 2;
-    const uint8_t chCount = args.channels.size();
-    const auto link = args.channels[0]->config.linkFormat;
+    const uint8_t chCount = stream->mTxStreams.size();
+    const auto link = stream->mTxStreams[0]->config.linkFormat;
+    const unsigned char ep  = 0x01;
 
     double latency=0;
     for (int i = 0; i < chCount; i++)
-    {
-        latency += args.channels[i]->config.performanceLatency/chCount;
-    }
+        latency += stream->mTxStreams[i]->config.performanceLatency/chCount;
+
     const unsigned tmp_cnt = (latency * 6)+0.5;
 
     const uint8_t buffersCount = 16; // must be power of 2
-    assert(buffersCount % 2 == 0);
     const uint8_t packetsToBatch = (1<<tmp_cnt); //packets in single USB transfer
     const uint32_t bufferSize = packetsToBatch*4096;
     const uint32_t popTimeout_ms = 100;
@@ -542,7 +430,6 @@ void ConnectionSTREAM::TransmitPacketsLoop(const ThreadData args)
     const int maxSamplesBatch = (link==StreamConfig::STREAM_12_BIT_COMPRESSED?1360:1020)/chCount;
     vector<int> handles(buffersCount, 0);
     vector<bool> bufferUsed(buffersCount, 0);
-    vector<uint32_t> bytesToSend(buffersCount, 0);
     vector<complex16_t> samples[maxChannelCount];
     vector<char> buffers;
     try
@@ -553,49 +440,57 @@ void ConnectionSTREAM::TransmitPacketsLoop(const ThreadData args)
     }
     catch (const std::bad_alloc& ex) //not enough memory for buffers
     {
-        lime::error("Error allocating Tx buffers, not enough memory");
-        return;
+        return lime::error("Error allocating Tx buffers, not enough memory");
     }
 
-    int m_bufferFailures = 0;
     long totalBytesSent = 0;
-
-    uint32_t samplesSent = 0;
-
     auto t1 = chrono::high_resolution_clock::now();
-    auto t2 = chrono::high_resolution_clock::now();
+    auto t2 = t1;
 
     uint8_t bi = 0; //buffer index
-    while (terminate->load() != true)
+    while (stream->terminateTx.load() != true)
     {
         if (bufferUsed[bi])
         {
-            if (this->WaitForSending(handles[bi], 1000) == false)
-                ++m_bufferFailures;
-            uint32_t bytesSent = this->FinishDataSending(&buffers[bi*bufferSize], bytesToSend[bi], handles[bi]);
-            totalBytesSent += bytesSent;
-            if (bytesSent != bytesToSend[bi])
-                ++m_bufferFailures;
+    	    unsigned bytesSent = 0; 
+            if (this->WaitForSending(handles[bi], 1000) == true) {
+                bytesSent = this->FinishDataSending(&buffers[bi*bufferSize], bufferSize, handles[bi]);
+	    }
+
+            if (bytesSent != bufferSize) {
+	      for (auto value : stream->mTxStreams) {
+		value->overflow++;
+	      }
+            }
+            else {
+                totalBytesSent += bytesSent;
+	    }
             bufferUsed[bi] = false;
         }
         int i=0;
 
-        while(i<packetsToBatch && terminate->load() != true)
+        while(i<packetsToBatch && stream->terminateTx.load() != true)
         {
             IStreamChannel::Metadata meta;
             FPGA_DataPacket* pkt = reinterpret_cast<FPGA_DataPacket*>(&buffers[bi*bufferSize]);
+            bool badSamples = false;
             for(int ch=0; ch<chCount; ++ch)
             {
-                int samplesPopped = args.channels[ch]->Read(samples[ch].data(), maxSamplesBatch, &meta, popTimeout_ms);
+                int samplesPopped = stream->mTxStreams[ch]->Read(samples[ch].data(), maxSamplesBatch, &meta, popTimeout_ms);
                 if (samplesPopped != maxSamplesBatch)
                 {
-                #ifndef NDEBUG
-                    lime::warning("popping from TX, samples popped %i/%i", samplesPopped, maxSamplesBatch);
-                #endif
+                    badSamples = true;
+                    stream->mTxStreams[ch]->underflow++;
+                    stream->txDataRate_Bps.store(0);
+#ifndef NDEBUG
+                    printf("popping from TX, samples popped %i/%i\n", samplesPopped, maxSamplesBatch);
+#endif
+                    break;
                 }
-
             }
-            if(terminate->load() == true) //early termination
+            if (badSamples)
+                continue;
+            if(stream->terminateTx.load() == true) //early termination
                 break;
             pkt[i].counter = meta.timestamp;
             pkt[i].reserved[0] = 0;
@@ -608,12 +503,10 @@ void ConnectionSTREAM::TransmitPacketsLoop(const ThreadData args)
                 src[c] = (samples[c].data());
             uint8_t* const dataStart = (uint8_t*)pkt[i].data;
             fpga::Samples2FPGAPacketPayload(src.data(), maxSamplesBatch, chCount, link, dataStart, nullptr);
-            samplesSent += maxSamplesBatch;
             ++i;
         }
 
-        bytesToSend[bi] = bufferSize;
-        handles[bi] = this->BeginDataSending(&buffers[bi*bufferSize], bytesToSend[bi]);
+        handles[bi] = this->BeginDataSending(&buffers[bi*bufferSize], bufferSize, ep);
         bufferUsed[bi] = true;
 
         t2 = chrono::high_resolution_clock::now();
@@ -622,23 +515,18 @@ void ConnectionSTREAM::TransmitPacketsLoop(const ThreadData args)
         {
             //total number of bytes sent per second
             float dataRate = 1000.0*totalBytesSent / timePeriod;
-            if(dataRate_Bps)
-                dataRate_Bps->store(dataRate);
-            m_bufferFailures = 0;
-            samplesSent = 0;
+            stream->txDataRate_Bps.store(dataRate);
             totalBytesSent = 0;
             t1 = t2;
 #ifndef NDEBUG
-            //total number of samples from all channels per second
-            float sampleRate = 1000.0*samplesSent / timePeriod;
-            lime::debug("Tx: %.3f MB/s, Fs: %.3f MHz, failures: %i", dataRate / 1000000.0, sampleRate / 1000000.0, m_bufferFailures);
+            printf("Tx: %.3f MB/s\n", dataRate / 1000000.0);
 #endif
         }
         bi = (bi + 1) & (buffersCount-1);
     }
 
     // Wait for all the queued requests to be cancelled
-    this->AbortSending();
+    AbortSending(ep);
     for (int j = 0; j<buffersCount; j++)
     {
         if (bufferUsed[bi])
@@ -648,7 +536,6 @@ void ConnectionSTREAM::TransmitPacketsLoop(const ThreadData args)
         }
         bi = (bi + 1) & (buffersCount-1);
     }
-    if (dataRate_Bps)
-        dataRate_Bps->store(0);
+    stream->txDataRate_Bps.store(0);
 }