changeset 117:ada68d50e56a scope-refactoring

ReceiveAudioThread hs been ported to BBB. The scope project now is sending audio locally and receiving it at the same time

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/core/ReceiveAudioThread.cpp	Thu Aug 20 16:37:15 2015 +0100
@@ -0,0 +1,207 @@
+#include "ReceiveAudioThread.h"
+//initialise static members
+AuxiliaryTask ReceiveAudioThread::receiveDataTask=NULL;
+
+void ReceiveAudioThread::dealloc(){
+    free(buffer);
+    buffer=NULL;
+    free(stackBuffer);
+    stackBuffer=NULL;
+}
+void ReceiveAudioThread::wrapWritePointer(){
+    //this is not quite a simple wrapping as you wold in a circular buffer,
+    //as there is no guarantee the buffer will be full at all times, given that there must alwas be enough space at the end of it
+    //to hold a full payload
+    // lastValidPointer indicates the last pointer in the buffer containing valid data
+    //
+    if(writePointer+payloadLength+headerLength>bufferLength){ //if we are going to exceed the length of the buffer with the next reading
+        //  lastValidPointer=writePointer+headerLength; //remember where the last valid data are
+        //  for(int n=headerLength;n<lastValidPointer; n++){
+            //  fprintf(fd2, "%f\n",buffer[n]); //DEBUG
+        //  }
+        writePointer=0; //and reset to beginning of the buffer
+    }
+}
+void ReceiveAudioThread::pushPayload(int startIndex){ //backup the payload samples that will be overwritten by the new header
+    for(int n=0; n<headerLength; n++){
+        stackBuffer[n]=buffer[startIndex+n];
+    }
+}
+void ReceiveAudioThread::popPayload(int startIndex){
+    for(int n=0; n<headerLength; n++){
+        buffer[startIndex+n]=stackBuffer[n];
+    }        
+}
+
+int ReceiveAudioThread::readUdpToBuffer(){
+    if(listening==false || bufferReady==false)
+        return 0;
+    if(writePointer<0)
+        return 0;
+    if(1){ //TODO: implement waitUntilReady for UdpServer
+//JUCE    if(socket.waitUntilReady(true, waitForSocketTime)){ // waitForSocketTime could have been set to -1 (wait forever),
+                                                        // but it would have made it more difficult for the thread to be killed
+        pushPayload(writePointer); //backup headerLength samples. This could be skipped if writePointer==0
+        //read header+payload
+//JUCE        int numBytes=socket.read(buffer+writePointer, bytesToRead,1);
+        int numBytes=socket.read(buffer+writePointer, bytesToRead);
+            //TODO: do something with the data in the header (e.g.: check that timestamp is sequential,
+            // check the channel number)
+            //TODO: (if using variable-length payload) validate the actual numBytes read against the size declared in the header
+        if(numBytes<0){
+            printf("error numBytes1\n");
+            return -3; //TODO: something went wrong, you have to discard the rest of the packet!
+        }
+        if(numBytes==0){//TODO: when inplementing waitUntilReady, this should not happen unless you actually receive a packate of size zero (is it at all possible?)
+//        	printf("received 0 bytes\n");
+        	return 0;
+        }
+        if(numBytes != bytesToRead){ //this is equivalent to (numBytes<numBytes)
+            printf("error numBytes2: %d\n", numBytes);
+            return -4; //TODO: something went wrong, less bytes than expected in the payload.
+        }
+        if(buffer[writePointer]!=0)
+        	return 0; //wrong channel
+        //  printf("Received a message of length %d, it was on channel %d and timestamp %d\n", numBytes, (int)buffer[writePointer], (int)buffer[writePointer+1]);
+        popPayload(writePointer); //restore headerLength payload samples. This could be skipped if writePointer==0
+
+        //even though we just wrote (payloadLength+headerLength) samples in the buffer,
+        //we only increment by payloadLength. This way, next time a socket.read is performed, we will
+        //backup the last headerLength samples that we just wrote and we will overwrite them with
+        //the header from the new read. After parsing the header we will then restore the backed up samples.
+        //This way we guarantee that, apart from the first headerLength samples, buffer is a circular buffer!
+        printf("writepointer:%d\n", writePointer);
+        writePointer+=payloadLength;
+
+        if(writePointer>lastValidPointer){
+            //  lastValidPointer=writePointer+headerLength;
+        }
+        wrapWritePointer();
+        return numBytes;
+    }
+    return 0; //timeout occurred
+}
+ReceiveAudioThread::ReceiveAudioThread() :
+//JUCE    Thread(threadName),
+    socket(0),
+    listening(false),
+    bufferReady(false),
+    threadIsExiting(false),
+    buffer(NULL),
+    stackBuffer(NULL),
+    bufferLength(0),
+    lastValidPointer(0),
+    sleepTime(2000),
+    waitForSocketTime(100),
+    threadPriority(95)
+{};
+ReceiveAudioThread::~ReceiveAudioThread(){
+//JUCE    stopThread(1000);
+	while(threadRunning){
+		sleep(sleepTime*2);	//wait for thread to stop
+		std::cout<< "Waiting for receiveAudioTask to stop" << std::endl;
+	}
+	//TODO: check if thread stopped, otherwise kill it before dealloc
+    dealloc();
+}
+ReceiveAudioThread *gRAT;
+void receiveData(){
+	gRAT->run();
+}
+void ReceiveAudioThread::init(int aSamplesPerBlock){
+  dealloc();
+  //  fd=fopen("output.m","w"); //DEBUG
+  //  fprintf(fd,"var=["); //DEBUG
+  gRAT=this;
+  headerLength=2;
+  payloadLength=300; //TODO: make sure that payloadLength and headerLength are the same as the client is sending.
+  bufferLength=std::max(headerLength+(payloadLength*4), headerLength+(aSamplesPerBlock*4)); //there are many considerations that can be done here ...
+                      //We keep a headerLength padding at the beginning of the array to allow full reads from the socket
+  buffer=(float*)malloc(sizeof(float)*bufferLength);
+  if(buffer==NULL) // something wrong
+    return;
+  bufferReady=true;
+  lastValidPointer=headerLength+ ((bufferLength-headerLength)/payloadLength)*payloadLength;
+  memset(buffer,0,bufferLength*sizeof(float));
+  stackBuffer=(float*)malloc(sizeof(float)*headerLength);
+  bytesToRead=sizeof(float)*(payloadLength + headerLength);
+  writePointer=-1;
+  readPointer=0; //TODO: this *4 is sortof a security margin
+  sleepTime=payloadLength/(float)44100 /4.0; //set sleepTime so that you do not check too often or too infrequently
+	receiveDataTask=BeagleRT_createAuxiliaryTask( receiveData, threadPriority, "receiveDataTask");
+//JUCE    startThread(threadPriority);
+}
+
+void ReceiveAudioThread::bindToPort(int aPort){
+    listening=socket.bindToPort(aPort);
+}
+bool ReceiveAudioThread::isListening(){
+    return listening;
+}
+float* ReceiveAudioThread::getCurrentBuffer(int length){ // NOTE: this cannot work all the time unless samplesPerBuffer and payloadLength are multiples
+    //TODO: make it return the number of samples actually available at the specified location
+    if(isListening()==false || length>bufferLength)
+        return NULL;
+    readPointer+=length;
+    if(readPointer>lastValidPointer){
+        readPointer=headerLength;
+    }
+    return buffer+(int)readPointer;
+};
+int ReceiveAudioThread::getSamplesSrc(float *destination, int length, float samplingRateRatio){//TODO: add interleaved version
+    if (!(samplingRateRatio>0 && samplingRateRatio<=2))
+        return -2;
+    if(isListening()==false)
+        return -1;
+    if(writePointer<0){ //if writePointer has not been initalized yet ...
+        writePointer=2*length;  // do it, so that it starts writing at a safety margin from where we write.
+                            // This will help keeping them in sync.
+                            //TODO: handle what happens when the remote stream is interrupted and then restarted
+    }
+    if(length>lastValidPointer) { 
+        //not enough samples available, we fill the buffer with what is available, but the destination buffer will not be filled completely
+        //at this very moment the other thread might be writing at most one payload into the buffer.
+        //To avoid a race condition, we need to let alone the buffer where we are currently writing
+        //as writing the payload also temporarily overwrites the previous headerLength samples, we need to account for them as well
+        //TODO: This assumes that the writePointer and readPointer do not drift. When doing clock synchronization we will find out that it is not true!
+        length=lastValidPointer-payloadLength-headerLength;
+        if(length<0) //no samples available at all! 
+            return 0;
+    }
+    for(int n=0; n<length; n++){
+        destination[n]=buffer[(int)(0.5+readPointer)];//simple ZOH non-interpolation (nearest neighbour)
+        //  fprintf(fd,"%f, %d, %f;\n",readPointer,writePointer,destination[n]); //DEBUG
+        readPointer+=samplingRateRatio;
+        if((int)(0.5+readPointer)>=lastValidPointer){
+            readPointer=readPointer-lastValidPointer+headerLength;
+        }
+    }
+    return readPointer; 
+}
+
+bool ReceiveAudioThread::isBufferReady(){
+    return bufferReady;
+}
+void ReceiveAudioThread::startThread(){
+	BeagleRT_scheduleAuxiliaryTask(receiveDataTask);
+}
+void ReceiveAudioThread::stopThread(){
+	threadIsExiting=true;
+}
+bool ReceiveAudioThread::threadShouldExit(){
+	return(gShouldStop || threadIsExiting );
+}
+void ReceiveAudioThread::run(){
+    //  fd2=fopen("buffer.m","w"); //DEBUG
+    //  fprintf(fd2, "buf=["); //DEBUG
+	threadRunning=true;
+    while(!threadShouldExit()){ //TODO: check that the socket buffer is empty before starting
+        readUdpToBuffer(); // read into the oldBuffer
+        usleep(sleepTime);
+    }
+    threadRunning=false;
+    //  fprintf(fd,"];readPointer,writePointer,lastValidPointer,destination]=deal(var(:,1), var(:,2), var(:,3), var(:,4));"); //DEBUG
+    //  fclose(fd);//DEBUG
+    //  fprintf(fd2,"];");//DEBUG
+    //  fclose(fd2); //DEBUG
+}

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/include/ReceiveAudioThread.h	Thu Aug 20 16:37:15 2015 +0100
@@ -0,0 +1,55 @@
+#ifndef RECEIVEAUDIOTHREAD_H_INCLUDED
+#define RECEIVEAUDIOTHREAD_H_INCLUDED
+#include <vector>
+#include <iostream>
+#include <UdpServer.h>
+#include <BeagleRT.h>
+#include <native/task.h>
+
+class ReceiveAudioThread{
+private:
+    //  FILE *fd; //DEBUG
+    //  FILE *fd2; //DEBUG
+    UdpServer socket;
+//JUCE    DatagramSocket socket;
+    bool listening;
+    bool bufferReady;
+    bool threadIsExiting;
+    bool threadRunning;
+    float *buffer;
+    float *stackBuffer;
+    int bufferLength;
+    float readPointer;
+    int writePointer;
+    int lastValidPointer;
+    int sleepTime;
+    int waitForSocketTime;
+    int payloadLength; //size of the payload of each datagram
+    int headerLength; //size of the header of each datagram
+    int bytesToRead;
+    int threadPriority;
+    void dealloc();
+    void wrapWritePointer();
+    void pushPayload(int startIndex);
+    void popPayload(int startIndex);
+    int readUdpToBuffer();
+    bool threadShouldExit();
+    static AuxiliaryTask receiveDataTask; //TODO: allow different AuxiliaryTasks for different priorities (e.g.: audio vs scope)
+//    static std::vector<NetworkReceive *> objAddrs;
+public:
+    ReceiveAudioThread();
+    ~ReceiveAudioThread();
+    void init(int aSamplesPerBlock);
+    void setup();
+    void bindToPort(int aPort);
+    bool isListening();
+    float* getCurrentBuffer(int length);
+    int getSamplesSrc(float *destination, int length, float samplingRateRatio);
+    bool isBufferReady();
+    void run();
+    void startThread();
+    void stopThread();
+//    static int getNumInstances();
+//    static void receiveAllData();
+};
+#endif  // RECEIVEAUDIOTHREAD_H_INCLUDED

--- a/projects/scope/render.cpp	Wed Aug 19 23:22:39 2015 +0100
+++ b/projects/scope/render.cpp	Thu Aug 20 16:37:15 2015 +0100
@@ -1,5 +1,6 @@
 #include <BeagleRT.h> 
 #include <NetworkSend.h>
+#include <ReceiveAudioThread.h>
 #include <cmath>
 
 float gPhase1, gPhase2;
@@ -17,9 +18,11 @@
 // in from the call to initAudio().
 //
 // Return true on success; returning false halts the program.
-
+ReceiveAudioThread receiveAudio;
 bool setup(BeagleRTContext *context, void *userData)
 {
+	receiveAudio.bindToPort(9999);
+	receiveAudio.init(context->audioFrames);
 	scope.setup();  //call this once in setup to initialise the scope
 //	networkSend.setup(context->audioSampleRate, 0, 9999, "192.168.7.1");
 	 
@@ -42,6 +45,9 @@
 void render(BeagleRTContext *context, void *userData)
 {
 	static int count=0;
+	if(count==0)
+		receiveAudio.startThread();
+
 	for(unsigned int n = 0; n < context->audioFrames; n++) {
 	    
 		float chn0 = sinf(gPhase1);
@@ -74,6 +80,8 @@
 			gPhase2 -= 2.0 * M_PI;
 		
 	}
+	if(count>100)
+		receiveAudio.getSamplesSrc(context->audioOut, context->audioFrames, 1);
 	count++;
 }