Using mutex to synchronize threads

Using mutex to synchronize threads

transmitter.cpp

@@ -35,6 +35,7 @@
 #include "mailbox.h"
 #include <bcm_host.h>
 #include <thread>
+#include <chrono>
 #include <cmath>
 #include <fcntl.h>
 #include <sys/mman.h>
@@ -120,7 +121,8 @@ struct AllocatedMemory {
 bool Transmitter::transmitting = false;
 volatile ClockRegisters *Transmitter::output = nullptr;
 uint32_t Transmitter::sampleOffset, Transmitter::clockDivisor, Transmitter::divisorRange, Transmitter::sampleRate;
-std::vector<Sample> *Transmitter::loadedSamples;
+std::vector<Sample> Transmitter::samples;
+std::mutex Transmitter::samplesMutex;
 void *Transmitter::peripherals;
 
 Transmitter::Transmitter()
@@ -213,16 +215,16 @@ volatile PWMRegisters *Transmitter::initPwmController()
     volatile ClockRegisters *pwmClk = reinterpret_cast<ClockRegisters *>(getPeripheralVirtAddress(PWMCLK_BASE_OFFSET));
     float pwmClkFreq = PWM_WRITES_PER_SAMPLE * PWM_CHANNEL_RANGE * sampleRate / 1000000;
     pwmClk->ctl = (0x5A << 24) | 0x06;
-    usleep(1000);
+    std::this_thread::sleep_for(std::chrono::microseconds(1000));
     pwmClk->div = (0x5A << 24) | static_cast<uint32_t>(getSourceFreq() * (0x01 << 12) / pwmClkFreq);
     pwmClk->ctl = (0x5A << 24) | (0x01 << 4) | 0x06;
 
     volatile PWMRegisters *pwm = reinterpret_cast<PWMRegisters *>(getPeripheralVirtAddress(PWM_BASE_OFFSET));
     pwm->ctl = 0x00000000;
-    usleep(1000);
+    std::this_thread::sleep_for(std::chrono::microseconds(1000));
     pwm->status = 0x01FC;
     pwm->ctl = (0x01 << 6);
-    usleep(1000);
+    std::this_thread::sleep_for(std::chrono::microseconds(1000));
     pwm->chn1Range = PWM_CHANNEL_RANGE;
     pwm->dmaConf = (0x01 << 31) | 0x0707;
     pwm->ctl = (0x01 << 5) | (0x01 << 2) | 0x01;
@@ -238,7 +240,7 @@ volatile DMARegisters *Transmitter::startDma(AllocatedMemory &memory, volatile D
 {
     volatile DMARegisters *dma = reinterpret_cast<DMARegisters *>(getPeripheralVirtAddress((dmaChannel < 15) ? DMA0_BASE_OFFSET + dmaChannel * 0x100 : DMA15_BASE_OFFSET));
     dma->ctlStatus = (0x01 << 31);
-    usleep(1000);
+    std::this_thread::sleep_for(std::chrono::microseconds(1000));
     dma->ctlStatus = (0x01 << 2) | (0x01 << 1);
     dma->cbAddress = getMemoryPhysAddress(memory, dmaCb);
     dma->ctlStatus = (0xFF << 16) | 0x01;
@@ -255,7 +257,7 @@ volatile ClockRegisters *Transmitter::initClockOutput()
     volatile ClockRegisters *clock = reinterpret_cast<ClockRegisters *>(getPeripheralVirtAddress(CLK0_BASE_OFFSET));
     volatile uint32_t *gpio = reinterpret_cast<uint32_t *>(getPeripheralVirtAddress(GPIO_BASE_OFFSET));
     clock->ctl = (0x5A << 24) | 0x06;
-    usleep(1000);
+    std::this_thread::sleep_for(std::chrono::microseconds(1000));
     clock->div = (0x5A << 24) | clockDivisor;
     clock->ctl = (0x5A << 24) | (0x01 << 9) | (0x01 << 4) | 0x06;
     *gpio = (*gpio & 0xFFFF8FFF) | (0x01 << 14);
@@ -293,7 +295,6 @@ void Transmitter::transmit(WaveReader &reader, float frequency, float bandwidth,
             output = nullptr;
         }
     };
-
     try {
         if (dmaChannel != 0xFF) {
             transmitViaDma(reader, bufferSize, dmaChannel);
@@ -305,33 +306,35 @@ void Transmitter::transmit(WaveReader &reader, float frequency, float bandwidth,
         finally();
         throw catched;
     }
-
     finally();
 }
 
 void Transmitter::transmitViaCpu(WaveReader &reader, uint32_t bufferSize)
 {
-    std::vector<Sample> samples = reader.getSamples(bufferSize, transmitting);
+    samples = reader.getSamples(bufferSize, transmitting);
     if (!samples.size()) {
         return;
     }
-
     sampleOffset = 0;
-    loadedSamples = &samples;
 
     bool eof = samples.size() < bufferSize;
     std::thread txThread(Transmitter::transmitThread);
 
-    usleep(BUFFER_TIME / 2);
+    std::this_thread::sleep_for(std::chrono::microseconds(BUFFER_TIME / 2));
+    bool wait = false;
 
     auto finally = [&]() {
         transmitting = false;
         txThread.join();
+        samples.clear();
     };
-
     try {
         while (!eof && transmitting) {
-            if (loadedSamples == nullptr) {
+            if (wait) {
+                std::this_thread::sleep_for(std::chrono::microseconds(BUFFER_TIME / 2));
+            }
+            std::lock_guard<std::mutex> locked(samplesMutex);
+            if (!samples.size()) {
                 if (!reader.setSampleOffset(sampleOffset + bufferSize)) {
                     break;
                 }
@@ -340,15 +343,13 @@ void Transmitter::transmitViaCpu(WaveReader &reader, uint32_t bufferSize)
                     break;
                 }
                 eof = samples.size() < bufferSize;
-                loadedSamples = &samples;
             }
-            usleep(BUFFER_TIME / 2);
+            wait = true;
         }
     } catch (std::runtime_error &catched) {
         finally();
         throw catched;
     }
-
     finally();
 }
 
@@ -358,7 +359,7 @@ void Transmitter::transmitViaDma(WaveReader &reader, uint32_t bufferSize, uint8_
         throw std::runtime_error("DMA channel number out of range (0 - 15)");
     }
 
-    std::vector<Sample> samples = reader.getSamples(bufferSize, transmitting);
+    samples = reader.getSamples(bufferSize, transmitting);
     if (!samples.size()) {
         return;
     }
@@ -409,12 +410,12 @@ void Transmitter::transmitViaDma(WaveReader &reader, uint32_t bufferSize, uint8_
 
     volatile DMARegisters *dma = startDma(dmaMemory, dmaCb, dmaChannel);
 
-    usleep(BUFFER_TIME / 4);
+    std::this_thread::sleep_for(std::chrono::microseconds(BUFFER_TIME / 4));
 
     auto finally = [&]() {
         dmaCb[(cbOffset < 2 * bufferSize) ? cbOffset : 0].nextCbAddress = 0x00000000;
         while (dma->cbAddress != 0x00000000) {
-            usleep(1000);
+            std::this_thread::sleep_for(std::chrono::microseconds(1000));
         }
 
         closeDma(dma);
@@ -422,8 +423,8 @@ void Transmitter::transmitViaDma(WaveReader &reader, uint32_t bufferSize, uint8_
 
         freeMemory(dmaMemory);
         transmitting = false;
+        samples.clear();
     };
-
     try {
         while (!eof && transmitting) {
             samples = reader.getSamples(bufferSize, transmitting);
@@ -438,7 +439,7 @@ void Transmitter::transmitViaDma(WaveReader &reader, uint32_t bufferSize, uint8_
                 value = preEmphasis.filter(value);
 #endif
                 while (i == ((dma->cbAddress - getMemoryPhysAddress(dmaMemory, dmaCb)) / (2 * sizeof(DMAControllBlock)))) {
-                    usleep(1000);
+                    std::this_thread::sleep_for(std::chrono::microseconds(1000));
                 }
                 clkDiv[i] = (0x5A << 24) | (clockDivisor - static_cast<int32_t>(round(value * divisorRange)));
                 cbOffset += 2;
@@ -448,7 +449,6 @@ void Transmitter::transmitViaDma(WaveReader &reader, uint32_t bufferSize, uint8_
         finally();
         throw catched;
     }
-
     finally();
 }
 
@@ -465,32 +465,36 @@ void Transmitter::transmitThread()
     while (transmitting) {
         uint64_t start = current;
 
-        while ((loadedSamples == nullptr) && transmitting) {
+        bool locked = samplesMutex.try_lock();
-            usleep(1);
+        while (!locked && transmitting) {
+            std::this_thread::sleep_for(std::chrono::microseconds(1));
             current = *(reinterpret_cast<volatile uint64_t *>(&timer->low));
+            locked = samplesMutex.try_lock();
         }
         if (!transmitting) {
+            if (locked) {
+                samplesMutex.unlock();
+            }
             break;
         }
-
+        std::vector<Sample> loaded(std::move(samples));
-        std::vector<Sample> samples(*loadedSamples);
+        samplesMutex.unlock();
-        loadedSamples = nullptr;
 
         sampleOffset = (current - playbackStart) * sampleRate / 1000000;
         uint32_t offset = (current - start) * sampleRate / 1000000;
 
         while (true) {
-            if (offset >= samples.size()) {
+            if (offset >= loaded.size()) {
                 break;
             }
             uint32_t prevOffset = offset;
-            float value = samples[offset].getMonoValue();
+            float value = loaded[offset].getMonoValue();
 #ifndef NO_PREEMP
             value = preEmphasis.filter(value);
 #endif
             output->div = (0x5A << 24) | (clockDivisor - static_cast<int32_t>(round(value * divisorRange)));
             while (offset == prevOffset) {
-                usleep(1); // asm("nop");
+                std::this_thread::sleep_for(std::chrono::microseconds(1)); // asm("nop");
                 current = *(reinterpret_cast<volatile uint64_t *>(&timer->low));;
                 offset = (current - start) * sampleRate / 1000000;
             }

transmitter.hpp

@@ -35,6 +35,7 @@
 #define TRANSMITTER_HPP
 
 #include "wave_reader.hpp"
+#include <mutex>
 
 struct AllocatedMemory;
 struct PWMRegisters;
@@ -77,7 +78,8 @@ class Transmitter
         static bool transmitting;
         static uint32_t sampleOffset, clockDivisor, divisorRange, sampleRate;
         static volatile ClockRegisters *output;
-        static std::vector<Sample> *loadedSamples;
+        static std::vector<Sample> samples;
+        static std::mutex samplesMutex;
 };
 
 #endif // TRANSMITTER_HPP

wave_reader.cpp

@@ -34,6 +34,8 @@
 #include "wave_reader.hpp"
 #include <stdexcept>
 #include <cstring>
+#include <thread>
+#include <chrono>
 #include <unistd.h>
 #include <fcntl.h>
 
@@ -115,7 +117,7 @@ std::vector<uint8_t> WaveReader::readData(uint32_t bytesToRead, bool headerBytes
                 data.resize(bytes);
                 break;
             } else {
-                usleep(1);
+                std::this_thread::sleep_for(std::chrono::microseconds(1));
             }
         }
     }