Basic mixer for OSX -- multi demod streams working

RtAudio can’t open multiple streams, so now opening a new device will
start a static audio thread and all other threads will attach/detach
their input queues there.

src/CubicSDR.cpp

@@ -72,6 +72,10 @@ int CubicSDR::OnExit() {
 
     demodMgr.terminateAll();
 
+#ifdef __APPLE__
+    AudioThread::deviceCleanup();
+#endif
+
     delete threadCmdQueueSDR;
 
     delete iqVisualQueue;

src/CubicSDRDefs.h

@@ -1,7 +1,7 @@
 #pragma once
 
 #ifdef __APPLE__
-#define BUF_SIZE (16384*3)
+#define BUF_SIZE (16384*2)
 #define SRATE 2000000
 #else
 #define BUF_SIZE (16384*4)

src/audio/AudioThread.cpp

@@ -1,52 +1,163 @@
 #include "AudioThread.h"
 #include "CubicSDRDefs.h"
 #include <vector>
+#include <algorithm>
 #include "DemodulatorThread.h"
 
-AudioThread::AudioThread(AudioThreadInputQueue *inputQueue, DemodulatorThreadCommandQueue* threadQueueNotify) :
-        inputQueue(inputQueue), terminated(false), audio_queue_ptr(0), underflow_count(0), threadQueueNotify(threadQueueNotify) {
+#ifdef __APPLE__
+std::map<int, AudioThread *> AudioThread::deviceController;
+std::map<int, std::thread *> AudioThread::deviceThread;
+#endif
 
+AudioThread::AudioThread(AudioThreadInputQueue *inputQueue, DemodulatorThreadCommandQueue* threadQueueNotify) :
+        inputQueue(inputQueue), terminated(false), audio_queue_ptr(0), underflow_count(0), threadQueueNotify(threadQueueNotify), gain(1.0), active(
+                false) {
+#ifdef __APPLE__
+    boundThreads = new std::vector<AudioThread *>;
+#endif
 }
 
 AudioThread::~AudioThread() {
+#ifdef __APPLE__
+    delete boundThreads.load();
+#endif
+}
 
+#ifdef __APPLE__
+void AudioThread::bindThread(AudioThread *other) {
+    boundThreads.load()->push_back(other);
+}
+
+void AudioThread::removeThread(AudioThread *other) {
+    std::vector<AudioThread *>::iterator i;
+    i = std::find(boundThreads.load()->begin(), boundThreads.load()->end(), other);
+    if (i != boundThreads.load()->end()) {
+        boundThreads.load()->erase(i);
+    }
+}
+
+void AudioThread::deviceCleanup() {
+    std::map<int,AudioThread *>::iterator i;
+
+    for (i = deviceController.begin(); i != deviceController.end(); i++) {
+        i->second->terminate();
+    }
 }
 
 static int audioCallback(void *outputBuffer, void *inputBuffer, unsigned int nBufferFrames, double streamTime, RtAudioStreamStatus status,
         void *userData) {
     AudioThread *src = (AudioThread *) userData;
     float *out = (float*) outputBuffer;
+    memset(out, 0, nBufferFrames * 2 * sizeof(float));
     if (status) {
         std::cout << "Audio buffer underflow.." << (src->underflow_count++) << std::endl;
     }
 
-    if (src->audio_queue_ptr == src->currentInput.data.size()) {
-        if (src->terminated) {
-            return 1;
+    if (!src->boundThreads.load()->empty()) {
+        src->gain = 1.0 / src->boundThreads.load()->size();
+    } else {
+        return 0;
     }
-        src->inputQueue->pop(src->currentInput);
-        src->audio_queue_ptr = 0;
+
+    for (int j = 0; j < src->boundThreads.load()->size(); j++) {
+        AudioThread *srcmix = (*(src->boundThreads.load()))[j];
+        if (srcmix->terminated || !srcmix->inputQueue || srcmix->inputQueue->empty() || !srcmix->isActive()) {
+            continue;
         }
 
         for (int i = 0; i < nBufferFrames * 2; i++) {
-        out[i] = src->currentInput.data[src->audio_queue_ptr];
-        src->audio_queue_ptr++;
-        if (src->audio_queue_ptr == src->currentInput.data.size()) {
-            if (src->terminated) {
-                return 1;
+            if (srcmix->audio_queue_ptr >= srcmix->currentInput.data.size()) {
+                srcmix->inputQueue->pop(srcmix->currentInput);
+                srcmix->audio_queue_ptr = 0;
             }
-            src->inputQueue->pop(src->currentInput);
-            src->audio_queue_ptr = 0;
+            out[i] = out[i] + srcmix->currentInput.data[srcmix->audio_queue_ptr] * src->gain;
+            srcmix->audio_queue_ptr++;
         }
     }
 
     return 0;
 }
 
+#else
+
+static int audioCallback(void *outputBuffer, void *inputBuffer, unsigned int nBufferFrames, double streamTime, RtAudioStreamStatus status,
+        void *userData) {
+    AudioThread *src = (AudioThread *) userData;
+    float *out = (float*) outputBuffer;
+    memset(out, 0, nBufferFrames * 2 * sizeof(float));
+    if (status) {
+        std::cout << "Audio buffer underflow.." << (src->underflow_count++) << std::endl;
+    }
+
+    for (int i = 0; i < nBufferFrames * 2; i++) {
+        if (src->audio_queue_ptr >= src->currentInput.data.size()) {
+            if (src->terminated) {
+                break;
+            }
+            src->inputQueue->pop(src->currentInput);
+            src->audio_queue_ptr = 0;
+        }
+        out[i] = src->currentInput.data[src->audio_queue_ptr] * src->gain;
+        src->audio_queue_ptr++;
+    }
+
+    return 0;
+}
+#endif
+
+
+void AudioThread::enumerateDevices() {
+    int numDevices = dac.getDeviceCount();
+
+    for (int i = 0; i < numDevices; i++) {
+        RtAudio::DeviceInfo info = dac.getDeviceInfo(i);
+
+        std::cout << std::endl;
+
+        std::cout << "Audio Device #" << i << " " << info.name << std::endl;
+        std::cout << "\tDefault Output? " << (info.isDefaultOutput ? "Yes" : "No") << std::endl;
+        std::cout << "\tDefault Input? " << (info.isDefaultOutput ? "Yes" : "No") << std::endl;
+        std::cout << "\tInput channels: " << info.inputChannels << std::endl;
+        std::cout << "\tOutput channels: " << info.outputChannels << std::endl;
+        std::cout << "\tDuplex channels: " << info.duplexChannels << std::endl;
+
+        std::cout << "\t" << "Native formats:" << std::endl;
+        RtAudioFormat nFormats = info.nativeFormats;
+        if (nFormats & RTAUDIO_SINT8) {
+            std::cout << "\t\t8-bit signed integer." << std::endl;
+        }
+        if (nFormats & RTAUDIO_SINT16) {
+            std::cout << "\t\t16-bit signed integer." << std::endl;
+        }
+        if (nFormats & RTAUDIO_SINT24) {
+            std::cout << "\t\t24-bit signed integer." << std::endl;
+        }
+        if (nFormats & RTAUDIO_SINT32) {
+            std::cout << "\t\t32-bit signed integer." << std::endl;
+        }
+        if (nFormats & RTAUDIO_FLOAT32) {
+            std::cout << "\t\t32-bit float normalized between plus/minus 1.0." << std::endl;
+        }
+        if (nFormats & RTAUDIO_FLOAT64) {
+            std::cout << "\t\t32-bit float normalized between plus/minus 1.0." << std::endl;
+        }
+
+        std::vector<unsigned int>::iterator srate;
+
+        std::cout << "\t" << "Supported sample rates:" << std::endl;
+
+        for (srate = info.sampleRates.begin(); srate != info.sampleRates.end(); srate++) {
+            std::cout << "\t\t" << (*srate) << "hz" << std::endl;
+        }
+
+        std::cout << std::endl;
+    }
+}
+
 void AudioThread::threadMain() {
 #ifdef __APPLE__
     pthread_t tID = pthread_self();	 // ID of this thread
-    int priority = sched_get_priority_min( SCHED_RR );
+    int priority = sched_get_priority_max( SCHED_RR) - 1;
     sched_param prio = { priority }; // scheduling priority of thread
     pthread_setschedparam(tID, SCHED_RR, &prio);
 #endif
@@ -58,7 +169,6 @@ void AudioThread::threadMain() {
         return;
     }
 
-    RtAudio::StreamParameters parameters;
     parameters.deviceId = dac.getDefaultOutputDevice();
     parameters.nChannels = 2;
     parameters.firstChannel = 0;
@@ -70,11 +180,27 @@ void AudioThread::threadMain() {
 //    | RTAUDIO_MINIMIZE_LATENCY;
 //    opts.flags = RTAUDIO_MINIMIZE_LATENCY;
     opts.streamName = "CubicSDR Audio Output";
-//    opts.priority = sched_get_priority_max(SCHED_FIFO);
+    opts.priority = sched_get_priority_max(SCHED_FIFO);
 
     try {
+
+#ifdef __APPLE__
+        if (deviceController.find(parameters.deviceId) == deviceController.end()) {
+            deviceController[parameters.deviceId] = new AudioThread(NULL, NULL);
+            deviceController[parameters.deviceId]->bindThread(this);
+            deviceThread[parameters.deviceId] = new std::thread(&AudioThread::threadMain, deviceController[parameters.deviceId]);
+        } else if (deviceController[parameters.deviceId] == this) {
             dac.openStream(&parameters, NULL, RTAUDIO_FLOAT32, sampleRate, &bufferFrames, &audioCallback, (void *) this, &opts);
             dac.startStream();
+        } else {
+            deviceController[parameters.deviceId]->bindThread(this);
+        }
+        active = true;
+#else
+        dac.openStream(&parameters, NULL, RTAUDIO_FLOAT32, sampleRate, &bufferFrames, &audioCallback, (void *) this, &opts);
+        dac.startStream();
+
+#endif
     } catch (RtAudioError& e) {
         e.printMessage();
         return;
@@ -85,9 +211,22 @@ void AudioThread::threadMain() {
         cmdQueue.pop(command);
     }
 
+#ifdef __APPLE__
+    if (deviceController[parameters.deviceId] != this) {
+        deviceController[parameters.deviceId]->removeThread(this);
+    } else {
+        try {
+            dac.stopStream();
+            dac.closeStream();
+        } catch (RtAudioError& e) {
+            e.printMessage();
+        }
+    }
+#else
     try {
         // Stop the stream
         dac.stopStream();
+        dac.closeStream();
     } catch (RtAudioError& e) {
         e.printMessage();
     }
@@ -95,16 +234,41 @@ void AudioThread::threadMain() {
     if (dac.isStreamOpen()) {
         dac.closeStream();
     }
+#endif
 
     std::cout << "Audio thread done." << std::endl;
 
+    if (threadQueueNotify != NULL) {
         DemodulatorThreadCommand tCmd(DemodulatorThreadCommand::DEMOD_THREAD_CMD_AUDIO_TERMINATED);
         tCmd.context = this;
         threadQueueNotify->push(tCmd);
     }
+}
 
 void AudioThread::terminate() {
     terminated = true;
     AudioThreadCommand endCond;   // push an empty input to bump the queue
     cmdQueue.push(endCond);
 }
+
+bool AudioThread::isActive() {
+    return active;
+}
+
+void AudioThread::setActive(bool state) {
+#ifdef __APPLE__
+    AudioThreadInput dummy;
+    if (state && !active) {
+        deviceController[parameters.deviceId]->bindThread(this);
+        while (!inputQueue->empty()) {  // flush queue
+            inputQueue->pop(dummy);
+        }
+    } else if (!state && active) {
+        deviceController[parameters.deviceId]->removeThread(this);
+        while (!inputQueue->empty()) {  // flush queue
+            inputQueue->pop(dummy);
+        }
+    }
+#endif
+    active = state;
+}

src/audio/AudioThread.h

@@ -2,6 +2,7 @@
 
 #include <queue>
 #include <vector>
+#include <map>
 #include <string>
 #include <atomic>
 #include "wx/wxprec.h"
@@ -28,11 +29,11 @@ public:
 class AudioThreadCommand {
 public:
     enum AudioThreadCommandEnum {
-        AUTIO_THREAD_CMD_NULL, AUTIO_THREAD_CMD_SET_DEVICE,
+        AUDIO_THREAD_CMD_NULL, AUDIO_THREAD_CMD_SET_DEVICE
     };
 
     AudioThreadCommand() :
-            cmd(AUTIO_THREAD_CMD_NULL), int_value(0) {
+            cmd(AUDIO_THREAD_CMD_NULL), int_value(0) {
     }
 
     AudioThreadCommandEnum cmd;
@@ -54,12 +55,33 @@ public:
     AudioThread(AudioThreadInputQueue *inputQueue, DemodulatorThreadCommandQueue* threadQueueNotify);
     ~AudioThread();
 
+    void enumerateDevices();
+
     void threadMain();
     void terminate();
 
+    bool isActive();
+    void setActive(bool state);
+
+#ifdef __APPLE__
+    void bindThread(AudioThread *other);
+    void removeThread(AudioThread *other);
+#endif
+
 private:
     RtAudio dac;
+    RtAudio::StreamParameters parameters;
     AudioThreadCommandQueue cmdQueue;
     DemodulatorThreadCommandQueue* threadQueueNotify;
+
+#ifdef __APPLE__
+public:
+    static std::map<int,AudioThread *> deviceController;
+    static std::map<int,std::thread *> deviceThread;
+    static void deviceCleanup();
+    std::atomic<std::vector<AudioThread *> *> boundThreads;
+    float gain;
+    std::atomic<bool> active;
+#endif
 };
 

src/demod/DemodulatorInstance.cpp

@@ -2,7 +2,7 @@
 
 DemodulatorInstance::DemodulatorInstance() :
         t_Demod(NULL), t_PreDemod(NULL), t_Audio(NULL), threadQueueDemod(NULL), demodulatorThread(NULL), terminated(false), audioTerminated(false), demodTerminated(
-        false), preDemodTerminated(false) {
+        false), preDemodTerminated(false), active(false) {
 
     label = new std::string("Unnamed");
     threadQueueDemod = new DemodulatorThreadInputQueue;
@@ -42,14 +42,12 @@ void DemodulatorInstance::run() {
     pthread_attr_init(&attr);
     pthread_attr_setstacksize(&attr, 2048000);
     pthread_attr_getstacksize(&attr, &size);
-    pthread_attr_setschedpolicy(&attr, SCHED_RR);
     pthread_create(&t_PreDemod, &attr, &DemodulatorPreThread::pthread_helper, demodulatorPreThread);
     pthread_attr_destroy(&attr);
 
     pthread_attr_init(&attr);
     pthread_attr_setstacksize(&attr, 2048000);
     pthread_attr_getstacksize(&attr, &size);
-    pthread_attr_setschedpolicy(&attr, SCHED_RR);
     pthread_create(&t_Demod, &attr, &DemodulatorThread::pthread_helper, demodulatorThread);
     pthread_attr_destroy(&attr);
 
@@ -59,6 +57,7 @@ void DemodulatorInstance::run() {
     t_PreDemod = new std::thread(&DemodulatorPreThread::threadMain, demodulatorPreThread);
     t_Demod = new std::thread(&DemodulatorThread::threadMain, demodulatorThread);
 #endif
+    active = true;
 }
 
 void DemodulatorInstance::updateLabel(int freq) {
@@ -137,3 +136,11 @@ bool DemodulatorInstance::isTerminated() {
     return terminated;
 }
 
+bool DemodulatorInstance::isActive() {
+    return active;
+}
+
+void DemodulatorInstance::setActive(bool state) {
+    active = state;
+    audioThread->setActive(state);
+}

src/demod/DemodulatorInstance.h

@@ -44,7 +44,15 @@ public:
     bool isTerminated();
     void updateLabel(int freq);
 
+    bool isActive();
+    void setActive(bool state);
+
 private:
-    std::atomic<std::string *> label;bool terminated;bool demodTerminated;bool audioTerminated;bool preDemodTerminated;
+    std::atomic<std::string *> label;
+    bool terminated;
+    bool demodTerminated;
+    bool audioTerminated;
+    bool preDemodTerminated;
+    std::atomic<bool> active;
 };
 

src/demod/DemodulatorPreThread.cpp

@@ -89,9 +89,9 @@ void DemodulatorPreThread::threadMain() {
 #endif
 #ifdef __APPLE__
     pthread_t tID = pthread_self();  // ID of this thread
-    int priority = sched_get_priority_min( SCHED_RR );
+    int priority = sched_get_priority_max( SCHED_FIFO )-1;
     sched_param prio = { priority }; // scheduling priority of thread
-    pthread_setschedparam( tID, SCHED_RR, &prio );
+    pthread_setschedparam(tID, SCHED_FIFO, &prio);
 #endif
 
     if (!initialized) {

src/demod/DemodulatorThread.cpp

@@ -23,9 +23,9 @@ void DemodulatorThread::threadMain() {
 #endif
 #ifdef __APPLE__
     pthread_t tID = pthread_self();  // ID of this thread
-    int priority = sched_get_priority_min( SCHED_RR );
+    int priority = sched_get_priority_max( SCHED_FIFO )-1;
     sched_param prio = { priority }; // scheduling priority of thread
-    pthread_setschedparam( tID, SCHED_RR, &prio );
+    pthread_setschedparam(tID, SCHED_FIFO, &prio);
 #endif
 
     msresamp_crcf audio_resampler = NULL;
@@ -93,8 +93,11 @@ void DemodulatorThread::threadMain() {
             audioInputQueue->push(ati);
         }
 
-        if (visOutQueue != NULL) {
-            visOutQueue->push(ati);
+        if (visOutQueue != NULL && visOutQueue->empty()) {
+            AudioThreadInput ati_vis;
+            ati_vis.data.assign(demod_output,demod_output+num_written);
+            visOutQueue->push(ati_vis);
+//            visOutQueue->push(ati);
         }
     }
 

src/sdr/SDRPostThread.cpp

@@ -42,12 +42,12 @@ void SDRPostThread::threadMain() {
     int n_read;
     double seconds = 0.0;
 
-//#ifdef __APPLE__
-//	    pthread_t tID = pthread_self();	 // ID of this thread
-//	    int priority = sched_get_priority_min( SCHED_RR );
-//	    sched_param prio = { priority }; // scheduling priority of thread
-//	    pthread_setschedparam( tID, SCHED_RR, &prio );
-//#endif
+#ifdef __APPLE__
+    pthread_t tID = pthread_self();  // ID of this thread
+    int priority = sched_get_priority_max( SCHED_FIFO) - 1;
+    sched_param prio = { priority }; // scheduling priority of thread
+    pthread_setschedparam(tID, SCHED_FIFO, &prio);
+#endif
 
     dcFilter = iirfilt_crcf_create_dc_blocker(0.0005);
 
@@ -96,15 +96,20 @@ void SDRPostThread::threadMain() {
                 demodDataOut.bandwidth = data_in.bandwidth;
                 demodDataOut.data = data_in.data;
 
-                for (int i = 0, iMax = demodulators.size(); i < iMax; i++) {
-                    DemodulatorInstance *demod = demodulators[i];
+                std::vector<DemodulatorInstance *>::iterator i;
+                for (i = demodulators.begin(); i != demodulators.end(); i++) {
+                    DemodulatorInstance *demod = *i;
                     DemodulatorThreadInputQueue *demodQueue = demod->threadQueueDemod;
 
-                    if (demod->getParams().frequency != data_in.frequency) {
-                        if (abs(data_in.frequency - demod->getParams().frequency) > (int) ((float) ((float) SRATE / 2.0) * 1.15)) {
+                    if (demod->getParams().frequency != data_in.frequency
+                            && abs(data_in.frequency - demod->getParams().frequency) > (int) ((float) ((float) SRATE / 2.0) * 1.15)) {
+                        if (demod->isActive()) {
+                            demod->setActive(false);
                             demodQueue->push(dummyDataOut);
                             continue;
                         }
+                    } else if (!demod->isActive()) {
+                        demod->setActive(true);
                     }
 
                     demodQueue->push(demodDataOut);

src/sdr/SDRThread.cpp

@@ -89,12 +89,12 @@ int SDRThread::enumerate_rtl() {
 }
 
 void SDRThread::threadMain() {
-//#ifdef __APPLE__
-//	    pthread_t tID = pthread_self();	 // ID of this thread
-//	    int priority = sched_get_priority_min( SCHED_RR );
-//	    sched_param prio = { priority }; // scheduling priority of thread
-//	    pthread_setschedparam( tID, SCHED_RR, &prio );
-//#endif
+#ifdef __APPLE__
+    pthread_t tID = pthread_self();  // ID of this thread
+    int priority = sched_get_priority_max( SCHED_FIFO )-1;
+    sched_param prio = { priority }; // scheduling priority of thread
+    pthread_setschedparam(tID, SCHED_FIFO, &prio);
+#endif
 
     std::cout << "SDR thread initializing.." << std::endl;
 

src/visual/WaterfallCanvas.cpp

@@ -176,8 +176,8 @@ void WaterfallCanvas::OnKeyDown(wxKeyEvent& event) {
         if (!activeDemod) {
             break;
         }
-        wxGetApp().getDemodMgr().deleteThread(activeDemod);
         wxGetApp().removeDemodulator(activeDemod);
+        wxGetApp().getDemodMgr().deleteThread(activeDemod);
         break;
 
     default: