Mercurial > hg > beaglert
changeset 217:c42a6b4dc2d4 mergingClockSync
Recovered some files from ClockSync
author | Giulio Moro <giuliomoro@yahoo.it> |
---|---|
date | Sat, 13 Feb 2016 04:09:12 +0000 |
parents | 869f5e703844 |
children | a94c8e0f4ec7 |
files | core/NetworkSend.cpp core/ReceiveAudioThread.cpp core/UdpClient.cpp core/UdpServer.cpp include/NetworkSend.h include/ReceiveAudioThread.h include/Scope.h include/UdpClient.h include/UdpServer.h |
diffstat | 9 files changed, 1094 insertions(+), 160 deletions(-) [+] |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/core/NetworkSend.cpp Sat Feb 13 04:09:12 2016 +0000 @@ -0,0 +1,294 @@ +#include "NetworkSend.h" + +#ifdef USE_JUCE +#else +//initialize the static members of NetworkSend +bool NetworkSend::staticConstructed=false; +int NetworkSend::sleepTimeMs; +bool NetworkSend::threadIsExiting; +bool NetworkSend::threadRunning; +std::vector<NetworkSend*> NetworkSend::objAddrs(0); +AuxiliaryTask NetworkSend::sendDataTask=NULL; + +void sendData(){ + NetworkSend::run(); +} +void NetworkSend::staticConstructor(){ + if(staticConstructed==true) + return; + staticConstructed=true; + threadIsExiting=false; + threadRunning=false; + sendDataTask = BeagleRT_createAuxiliaryTask(::sendData, 95, "sendDataTask"); //TODO: allow variable priority +} +void NetworkSend::sendAllData(){ + for(unsigned int n=0; n<NetworkSend::objAddrs.size(); n++){ + NetworkSend::objAddrs[n]->sendData(); + } +} +int NetworkSend::getNumInstances(){ + return objAddrs.size(); +} +void NetworkSend::startThread(){ + BeagleRT_scheduleAuxiliaryTask(sendDataTask); +} +void NetworkSend::stopThread(){ + threadIsExiting=true; +} +bool NetworkSend::threadShouldExit(){ + return(gShouldStop || threadIsExiting); +} +bool NetworkSend::isThreadRunning(){ + return threadRunning; +} +#endif /* USE_JUCE */ + +#ifdef USE_JUCE +NetworkSend::NetworkSend(const String &threadName): + Thread(threadName) +#else +NetworkSend::NetworkSend() +#endif /* USE_JUCE */ +{ + channel.buffers=NULL; + channel.doneOnTime=NULL; + channel.readyToBeSent=NULL; + channel.enabled=false; + sleepTimeMs=2; //This should actually be initialized in the staticConstructor for non-Juce code, but doing it here makes it more portable + channel.sampleCount=0; +} + +NetworkSend::~NetworkSend(){ +#ifdef USE_JUCE + stopThread(1000); +#else + stopThread(); + for(unsigned int n=0; n<objAddrs.size(); n++){ //keep track of deleted instances; + if(objAddrs[n]==this){ + objAddrs.erase(objAddrs.begin()+n); + break; + } + } +#endif + dealloc(); +} +void NetworkSend::dealloc(){ + channel.enabled=false; + if(channel.buffers!=NULL){ + for(int n=0; n<channel.numBuffers; n++){ + free(channel.buffers[n]); + channel.buffers[n]=NULL; + } + free(channel.buffers); + channel.buffers=NULL; + } + free(channel.readyToBeSent); + channel.readyToBeSent=NULL; + free(channel.doneOnTime); + channel.doneOnTime=NULL; +} +void NetworkSend::cleanup(){ + dealloc(); +} + +void NetworkSend::setup(float aSampleRate, int blockSize, int aChannelNumber, int aPort, const char *aServer){ +#ifdef USE_JUCE +#else + staticConstructor(); //FIXME: ideally this should be in the constructor, but this is not currently possible + //because of limitations in BeagleRT_createAuxiliaryTask() + //keep track of added active instances + objAddrs.push_back(this);//TODO: this line should be in the constructor, but something weird happens if + // an instance of NetworkSend is then declared globally: the constructor gets called, + // and objAddrs.size()==1 but when you get to setup, objAddrs.size() has reverted back to 0, without + // any destructor being called in between ... +#endif /* USE_JUCE */ + cleanup(); + int numSamples=blockSize*4>4*channel.bufferLength ? blockSize*4 : 4*channel.bufferLength; + channel.numBuffers= 1+numSamples/channel.bufferLength; //the +1 takes the ceil() of the division + channel.buffers=(float**)malloc(channel.numBuffers*sizeof(float*)); + printf("NumBuffers: %d\n", channel.numBuffers); + if(channel.buffers==NULL) + return; + for(int n=0; n<channel.numBuffers; n++){ + channel.buffers[n]=(float*)malloc(channel.bufferLength*sizeof(float)); + if(channel.buffers[n]==NULL) + return; + } + channel.readyToBeSent=(bool*)malloc(channel.numBuffers*sizeof(bool)); + channel.doneOnTime=(bool*)malloc(channel.numBuffers*sizeof(bool)); + for(int n=0; n<channel.numBuffers; n++){ + channel.readyToBeSent[n]=false; + channel.doneOnTime[n]=true; + } + if(channel.readyToBeSent==NULL || channel.doneOnTime==NULL) + return; + channel.writePointer=0; + channel.writeBuffer=0; + channel.readBuffer=0; + setChannelNumber(aChannelNumber); + setPort(aPort); //TODO: check for the return value + setServer(aServer); //TODO: check for the return value + printf("Channel %d is sending messages to: %s:%d at %fHz\n", getChannelNumber(), aServer, aPort, aSampleRate); + channel.enabled=true; +} + +void NetworkSend::log(float value){ //TODO: add a vectorized version of this method + if(channel.enabled==false) + return; + if(channel.writePointer==channel.bufferLength){ // when the buffer is filled ... + channel.readyToBeSent[channel.writeBuffer]=true; // flag it as such +// printf("Scheduling for send %d\n",(int)channel.buffers[channel.writeBuffer][channel.headerTimestampIndex]); + channel.writePointer=channel.headerLength; //reset the writePointer + channel.writeBuffer=(channel.writeBuffer+1); //switch buffer + if(channel.writeBuffer==channel.numBuffers) // and wrap it + channel.writeBuffer=0; +// printf("WriteBuffer:%d\n", channel.writeBuffer); + if(channel.doneOnTime[channel.writeBuffer]==false){ //check if this buffer's last sending has completed on time ... + printf("Network buffer underrun. timestamp: %d :-{\n", + (int)channel.buffers[channel.writeBuffer][channel.headerTimestampIndex]); + } + channel.doneOnTime[channel.writeBuffer]=false; // ... and then reset the flag +#ifdef USE_JUCE + if(isThreadRunning()==false){ + startThread(10); + } +#else + if(isThreadRunning()==false){ + startThread(); + } +#endif /* USE_JUCE */ + } + if(channel.writePointer==channel.headerLength){ // we are about to start writing in the buffer, let's set the header + //set dynamic header values here. Static values are set in setup() and setChannelNumber(). + channel.buffers[channel.writeBuffer][channel.headerTimestampIndex]=(float)channel.sampleCount; //timestamp + channel.sampleCount++; + //add here more header fields + } + channel.buffers[channel.writeBuffer][channel.writePointer++]=value; +// sampleCount++; +}; + +void NetworkSend::setServer(const char *aServer){ +#ifdef USE_JUCE + remoteHostname=String::fromUTF8(aServer); +#else + udpClient.setServer(aServer); +#endif /* USE_JUCE */ +} +void NetworkSend::setPort(int aPort){ +#ifdef USE_JUCE + remotePortNumber=aPort; +#else + udpClient.setPort(aPort); +#endif /* USE_JUCE */ +} + +void NetworkSend::setChannelNumber(int aChannelNumber){ + channel.channelNumber=aChannelNumber; + for(int n=0; n<channel.numBuffers; n++){ //initialize the header + channel.buffers[n][channel.headerChannelIndex]=channel.channelNumber; + //add here more static header fields + } +}; +int NetworkSend::getChannelNumber(){ + return channel.channelNumber; +}; + +int NetworkSend::getTimestamp(){ + return channel.buffers[channel.readBuffer][channel.headerTimestampIndex]; +} + +void NetworkSend::sendData(){ + if(channel.enabled==false) + return; + while(channel.readyToBeSent[channel.readBuffer]==true){ + channel.readyToBeSent[channel.readBuffer]=false; + void* sourceBuffer=channel.buffers[channel.readBuffer]; +// printf("Trying to send timestamp %d\n",(int)((float*)sourceBuffer)[channel.headerTimestampIndex]); +// printf("ReadBuffer:%d\n", channel.readBuffer); + unsigned int numBytesToSend=NETWORK_AUDIO_BUFFER_SIZE*sizeof(float); + //TODO: call waitUntilReady before trying to write/send, to avoid blocks! (OR NOT?) +#ifdef USE_JUCE + if(1==udpClient.waitUntilReady(0, 5)){ + udpClient.write(remoteHostname, remotePortNumber, sourceBuffer, numBytesToSend); + channel.doneOnTime[channel.readBuffer]=true; + // printf ("Sent timestamp: %d\n", (int)((float*)sourceBuffer)[1]); + } else { + // printf ("Not ready timestamp: %d\n", (int)((float*)sourceBuffer)[1]); + } +#else + udpClient.send(sourceBuffer, numBytesToSend); + channel.doneOnTime[channel.readBuffer]=true; +#endif /* USE_JUCE */ + channel.readBuffer++; + if(channel.readBuffer==channel.numBuffers) + channel.readBuffer=0; + } +} + +void NetworkSend::run(){ +#ifdef USE_JUCE + // std::chrono::high_resolution_clock::time_point t1; + // std::chrono::high_resolution_clock::time_point t2; + // std::chrono::high_resolution_clock::time_point t3; + while(threadShouldExit()==false){ + // t3 = std::chrono::high_resolution_clock::now(); + // t1 = std::chrono::high_resolution_clock::now(); + sendData(); + // t2 = std::chrono::high_resolution_clock::now(); + // auto duration1 = std::chrono::duration_cast<std::chrono::microseconds>( t3 - t1 ).count(); + // auto duration2 = std::chrono::duration_cast<std::chrono::microseconds>( t2 - t1 ).count(); + // if(duration2>0) + // std::cout << "Duration is: " << duration2 <<". Whole loop is: " << duration1 << "\n"; + sleep(1); + } +#else + threadRunning=true; + while(threadShouldExit()==false){ + sendAllData(); + usleep(sleepTimeMs*1000); + } + threadRunning=false; +#endif +} +#ifdef USE_JUCE +#else +Scope::Scope(int aNumChannels): + channels(aNumChannels) +{}; +Scope::~Scope(){}; + +void Scope::log(int channel, float value){ + if(channel>=getNumChannels()) //TODO: assert this + return; + channels[channel].log(value); +} + +void Scope::setup(){ + setup(44100, 9999, "127.0.0.1"); +} + +void Scope::setup(float sampleRate, int aPort, const char* aServer){ + for(int n=0; n<getNumChannels(); n++){ + channels[n].setup(sampleRate, 128, n, aPort, aServer); //TODO: replace 128 with the actual block size + } +} + +void Scope::setPort(int port){ + for(int n=0; n<getNumChannels(); n++){ + channels[n].setPort(port); + } +} +void Scope::setPort(int channel, int aPort){ + channels[channel].setPort(aPort); + printf("Channel %d is now sending to port %d\n", channel, aPort); +} + +int Scope::getNumChannels(){ + return channels.size(); +} + +void Scope::sendData(){ + NetworkSend::sendAllData(); +} +#endif
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/core/ReceiveAudioThread.cpp Sat Feb 13 04:09:12 2016 +0000 @@ -0,0 +1,310 @@ +#include "ReceiveAudioThread.h" + +#ifdef USE_JUCE +#else +//initialise static members +bool ReceiveAudioThread::staticConstructed=false; +AuxiliaryTask ReceiveAudioThread::receiveDataTask=NULL; +std::vector<ReceiveAudioThread *> ReceiveAudioThread::objAddrs(0); +bool ReceiveAudioThread::threadRunning; +bool ReceiveAudioThread::threadIsExiting; +int ReceiveAudioThread::sleepTime; + +void receiveData(){ + ReceiveAudioThread::run(); +} +void ReceiveAudioThread::staticConstructor(){ + if(staticConstructed==true) + return; + staticConstructed=true; + threadIsExiting=false; + receiveDataTask=BeagleRT_createAuxiliaryTask(receiveData, 90, "receiveDataTask"); //TODO: allow different priorities +} +#endif /* USE_JUCE */ + +void ReceiveAudioThread::dealloc(){ + free(buffer); + buffer=NULL; + free(stackBuffer); + stackBuffer=NULL; +} +void ReceiveAudioThread::wrapWritePointer(){ + //this is not quite a simple wrapping as you would do 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(socket.waitUntilReady(true, waitForSocketTime)){// TODO: if waitForSocketTime here is >>5, the +#ifdef USE_JUCE +#else + lastTime=rt_timer_read(); +// rt_printf("lastTimeread= %llu\n", lastTime); +#endif /* USE_JUCE */ + // destructor (always or sometimes) never actually gets called, despite run() returns ...see issue #1381 + pushPayload(writePointer); //backup headerLength samples. This could be skipped if writePointer==0 + //read header+payload + int numBytes=socket.read(buffer+writePointer, bytesToRead, true); //read without waiting. + //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: this should not happen unless you actually receive a packet of size zero (is it at all possible?) +// printf("received 0 bytes\n"); + return 0; + } + if(numBytes != bytesToRead){ //this is equivalent to (numBytes<bytesToRead) + printf("error numBytes2: %d\n", numBytes); + return -4; //TODO: something went wrong, less bytes than expected in the payload. + } + if(channel!=(int)buffer[writePointer]){ +// printf("I am channel %d, but I received data for channel %d\n", channel, (int)buffer[writePointer]); + return -5; + } + if(buffer[writePointer+1]!=timestamp+1) + printf("missing a timestamp: %d\n",timestamp+1); + timestamp=buffer[writePointer+1]; +// rt_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! + writePointer+=payloadLength; + wrapWritePointer(); + return numBytes; + } + return 0; //timeout occurred +} +//USE_JUCE Thread(threadName), +#ifdef USE_JUCE +ReceiveAudioThread::ReceiveAudioThread(const String &threadName) : + Thread(threadName), +#else +ReceiveAudioThread::ReceiveAudioThread() : +#endif /* USE_JUCE */ + socket(0), + listening(false), + bufferReady(false), + buffer(NULL), + stackBuffer(NULL), + bufferLength(0), + lastValidPointer(0), + waitForSocketTime(5), +#ifdef USE_JUCE + threadPriority(5) +#else + threadPriority(88) +#endif /* USE_JUCE */ +{}; +ReceiveAudioThread::~ReceiveAudioThread(){ +#ifdef USE_JUCE + stopThread(1000); +#else + stopThread(); + while(threadRunning){ + usleep(sleepTime*2); //wait for thread to stop + std::cout<< "Waiting for receiveAudioTask to stop" << std::endl; + } +#endif /* USE_JUCE */ + //TODO: check if thread stopped, otherwise kill it before dealloc + dealloc(); +} +void ReceiveAudioThread::init(int aPort, int aSamplesPerBlock, int aChannel){ + dealloc(); +#ifdef USE_JUCE +#else + staticConstructor(); + objAddrs.push_back(this);//TODO: this line should be in the constructor +#endif /* USE_JUCE */ + bindToPort(aPort); + channel=aChannel; + printf("Channel %d is receiving on port %d\n",aChannel, aPort); + // fd=fopen("output.m","w"); //DEBUG + // fprintf(fd,"var=["); //DEBUG + 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; + lastValidPointer=headerLength+ ((bufferLength-headerLength)/payloadLength)*payloadLength; + memset(buffer,0,bufferLength*sizeof(float)); + stackBuffer=(float*)malloc(sizeof(float)*headerLength); + if(stackBuffer==NULL) // something wrong + return; + bufferReady=true; + bytesToRead=sizeof(float)*(payloadLength + headerLength); + writePointer=-1; + readPointer=0; + sleepTime=payloadLength/(float)44100 /4.0; //set sleepTime so that you do not check too often or too infrequently + timestamp=0; +#ifdef USE_JUCE + startThread(threadPriority); +#else + //TODO: the thread cannot be started here at the moment because init() is called in setup(), where tasks cannot be scheduled +#endif /* USE_JUCE */ +} + +void ReceiveAudioThread::bindToPort(int aPort){ + listening=socket.bindToPort(aPort); +#ifdef USE_JUCE +#else + if(listening==false) //this condition is valid also for USE_JUCE, but we do not printf in USE_JUCE + printf("Could not bind to port %d\n",aPort); +#endif /* USE_JUCE */ +} +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, int numChannelsInDestination, + int channelToWriteTo) +{ + if (!(samplingRateRatio>0 && samplingRateRatio<=2)) + return -2; + if(isListening()==false) + return -1; + static int numCalls=0; + if(writePointer<0 /*|| (numCalls&16383)==0*/){ //if writePointer has not been initalized yet ... +#ifdef USE_JUCE +#else //debug + rt_printf("reinit the writePointer, readPointer: %f;\n",readPointer); + readPointer=0; +#endif /* USE_JUCE */ + 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 + } + numCalls++; + 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*numChannelsInDestination+channelToWriteTo]=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 length; +} +int ReceiveAudioThread::getSamplesSrc(float *destination, int length, float samplingRateRatio){ + return getSamplesSrc(destination, length, samplingRateRatio, 1,0); + // TODO: rewriting this so that it does not call the override method we can save a multiply and add + // for each sample. +} +bool ReceiveAudioThread::isBufferReady(){ + return bufferReady; +} +#ifdef USE_JUCE +#else +void ReceiveAudioThread::startThread(){ + BeagleRT_scheduleAuxiliaryTask(receiveDataTask); +} +void ReceiveAudioThread::stopThread(){ + threadIsExiting=true; +} +bool ReceiveAudioThread::threadShouldExit(){ + return(gShouldStop || threadIsExiting ); +} +RTIME ReceiveAudioThread::getLastTime(){ + return lastTime; +} +#endif /* USE_JUCE */ +int ReceiveAudioThread::getTimestamp(){ + return timestamp; +} +void ReceiveAudioThread::run(){ + // fd2=fopen("buffer.m","w"); //DEBUG + // fprintf(fd2, "buf=["); //DEBUG + threadRunning=true; + int maxCount=10; + int count=0; + // Clean the socket from anything that is currently in it. +#ifdef USE_JUCE + // this is borrowed from BeagleRT's UdpServer class. + int n; + do { + float waste; + if(socket.waitUntilReady(true, 0)==0) + break; + n=socket.read((void*)&waste, sizeof(float), false); + count++; + if(n<0){ + printf("error\n"); + break; + } + printf("n: %d\n",n); + } while (n>0 && (maxCount<=0 || count<maxCount)); +#else + for(unsigned int n=0; n<objAddrs.size(); n++){ + count=objAddrs[n]->socket.empty(maxCount); + } +#endif /* USE_JUCE */ + printf("socket emptied with %d reads\n", count); + + while(!threadShouldExit()){ //TODO: check that the socket buffer is empty before starting +#ifdef USE_JUCE + readUdpToBuffer(); // read into the oldBuffer + sleep(sleepTime); +#else + for(unsigned int n=0; n<ReceiveAudioThread::objAddrs.size(); n++){ +// printf("%d\n", n); + ReceiveAudioThread::objAddrs[n]->readUdpToBuffer(); + } + usleep(sleepTime); //TODO: use rt_task_sleep instead +#endif /* USE_JUCE */ + } + threadRunning=false; + printf("Thread is not running \n"); + // 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 +}
--- a/core/UdpClient.cpp Sat Feb 13 01:51:01 2016 +0000 +++ b/core/UdpClient.cpp Sat Feb 13 04:09:12 2016 +0000 @@ -1,57 +1,77 @@ -/* - * udpClient.cpp - * - * Created on: 19 May 2015 - * Author: giulio moro - */ -#include "../include/UdpClient.h" - - UdpClient::UdpClient(){ - outSocket=socket(AF_INET, SOCK_DGRAM, 0); - isSetPort=false; - isSetServer=false; - enabled=false; - } - UdpClient::UdpClient(int aPort, const char* aServerName){ - outSocket=socket(AF_INET, SOCK_DGRAM, 0); - if(outSocket<0){ - enabled=false; - return; - } - setPort(aPort); - setServer(aServerName); - isSetPort=true; - isSetServer=true; - enabled=true; - } - UdpClient::~UdpClient(){ - close(outSocket); - } - void UdpClient::setPort(int aPort){ - port=aPort; - destinationServer.sin_port = htons(port); - destinationServer.sin_family = AF_INET; - isSetPort=true; - if(isSetServer){ - enabled=true; - } - }; - void UdpClient::setServer(const char* aServerName){ - inet_pton(AF_INET,aServerName,&destinationServer.sin_addr); - isSetServer=true; - if(isSetPort){ - enabled=true; - } - }; - int UdpClient::send(void * message, int size){ - if(!enabled) - return -1; - unsigned int length; - length=sizeof(struct sockaddr_in); - int n=sendto(outSocket,message,size,0,(const struct sockaddr *)&destinationServer,length); - if (n < 0){ - return n; - } - return 1; - }; - +/* + * udpClient.cpp + * + * Created on: 19 May 2015 + * Author: giulio moro + */ +#include "UdpClient.h" + + UdpClient::UdpClient(){ + outSocket=socket(AF_INET, SOCK_DGRAM, 0); + isSetPort=false; + isSetServer=false; + enabled=false; + } + UdpClient::UdpClient(int aPort, const char* aServerName){ + outSocket=socket(AF_INET, SOCK_DGRAM, 0); + if(outSocket<0){ + enabled=false; + return; + } + setPort(aPort); + setServer(aServerName); + isSetPort=true; + isSetServer=true; + enabled=true; + memset(&stTimeOut, 0, sizeof(struct timeval)); + } + UdpClient::~UdpClient(){ + close(outSocket); + } + void UdpClient::setPort(int aPort){ + port=aPort; + destinationServer.sin_port = htons(port); + destinationServer.sin_family = AF_INET; + isSetPort=true; + if(isSetServer){ + enabled=true; + } + }; + void UdpClient::setServer(const char* aServerName){ + inet_pton(AF_INET,aServerName,&destinationServer.sin_addr); + isSetServer=true; + if(isSetPort){ + enabled=true; + } + }; + int UdpClient::send(void * message, int size){ + if(!enabled) + return -1; + unsigned int length; + length=sizeof(struct sockaddr_in); + int n=sendto(outSocket,message,size,0,(const struct sockaddr *)&destinationServer,length); + if (n < 0){ + return n; + } + return 1; + }; + int UdpClient::write(const char* remoteHostname, int remotePortNumber, void* sourceBuffer, int numBytesToWrite){ + setServer(remoteHostname); + setPort(remotePortNumber); + send(sourceBuffer, numBytesToWrite); + } + int UdpClient::waitUntilReady(bool readyForReading, int timeoutMsecs){ +// If the socket is ready on return, this returns 1. If it times-out before the socket becomes ready, it returns 0. If an error occurs, it returns -1. + if(enabled==false) + return -1; + if(timeoutMsecs<0) + return select(outSocket+1, NULL, &stWriteFDS, NULL, NULL); //calling this with a NULL timeout will block indefinitely + FD_ZERO(&stWriteFDS); + FD_SET(outSocket, &stWriteFDS); + float timeOutSecs=timeoutMsecs*0.001; + stTimeOut.tv_sec=(int)timeOutSecs; + timeOutSecs-=(int)timeOutSecs; + stTimeOut.tv_usec=(int)(timeOutSecs*1000000); + int descriptorReady= select(outSocket+1, NULL, &stWriteFDS, NULL, &stTimeOut); + return descriptorReady>0? 1 : descriptorReady; + } \ No newline at end of file
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/core/UdpServer.cpp Sat Feb 13 04:09:12 2016 +0000 @@ -0,0 +1,119 @@ +/* + * udpServer.cpp + * + * Created on: 19 May 2015 + * Author: giulio moro + */ +#include "UdpServer.h" + +UdpServer::UdpServer(int aPort){ + init(aPort); +}; +UdpServer::UdpServer(){ + init(0); +} +UdpServer::~UdpServer(){ + close(); +}; +bool UdpServer::init(int aPort){ + enabled=true; + stZeroTimeOut.tv_sec = 0; //set timeout to 0 + stZeroTimeOut.tv_usec = 0; + inSocket=socket(AF_INET, SOCK_DGRAM, 0); + if (inSocket < 0){ + enabled=false; + } + length = sizeof(server); + server.sin_family=AF_INET; + server.sin_addr.s_addr=INADDR_ANY; + enabled=bindToPort(aPort); + wasteBufferSize=2048; + wasteBuffer=malloc(wasteBufferSize); + memset(&stTimeOut,0,sizeof(struct timeval)); + return enabled; +} + +bool UdpServer::bindToPort(int aPort){ + port=aPort; + if(port<1){ + enabled=false; + return false; + } + server.sin_port=htons(port); + if (bind(inSocket,(struct sockaddr *)&server,length)<0){ + enabled=false; + return false; + } + enabled=true; + return true; +} + +void UdpServer::close(){ + int ret=::close(inSocket); + if(ret != 0) + printf("Error while closing socket, errno: %d\n", errno);//Stop receiving data for this socket. If further data arrives, reject it. + inSocket=0; +} + +int UdpServer::waitUntilReady(bool readyForReading, int timeoutMsecs){ +// If the socket is ready on return, this returns 1. If it times-out before the socket becomes ready, it returns 0. If an error occurs, it returns -1. + if(enabled==false) + return -1; + if(timeoutMsecs<0) + return select(inSocket+1, &stReadFDS, NULL, NULL, NULL); //calling this with a NULL timeout will block indefinitely + FD_ZERO(&stReadFDS); + FD_SET(inSocket, &stReadFDS); + float timeOutSecs=timeoutMsecs*0.001; + stTimeOut.tv_sec=(long int)timeOutSecs; + timeOutSecs-=(int)timeOutSecs; + long int timeOutUsecs=timeOutSecs*1000000; + stTimeOut.tv_usec=timeOutUsecs; + int descriptorReady= select(inSocket+1, &stReadFDS, NULL, NULL, &stTimeOut); +// printf("stTimeOut.tv_sec=%ld, stTimeOut.tv_usec=%ld, descriptorReady: \n",stTimeOut.tv_sec,stTimeOut.tv_usec, descriptorReady); +// return descriptorReady>0 ? (timeOutUsecs-stTimeOut.tv_usec) : descriptorReady; + return descriptorReady>0 ? 1 : descriptorReady; +} + +int UdpServer::read(//Returns the number of bytes read, or -1 if there was an error. + void *destBuffer, + int maxBytesToRead, + bool blockUntilSpecifiedAmountHasArrived) +{ + if(enabled==false) + return -1; + FD_ZERO(&stReadFDS); + FD_SET(inSocket, &stReadFDS); + int descriptorReady= select(inSocket+1, &stReadFDS, NULL, NULL, &stZeroTimeOut); //TODO: this is not JUCE-compliant + if(descriptorReady<0){ //an error occurred + return -1; + } + int numberOfBytes=0; +// do + { + if (FD_ISSET(inSocket, &stReadFDS)) + { + // numberOfBytes=recvfrom(inSocket,destBuffer,maxBytesToRead,0,(struct sockaddr *)&from,&fromLength); + numberOfBytes+=recv(inSocket,destBuffer,maxBytesToRead-numberOfBytes,0); + if(numberOfBytes<0) + return -1; + } + } +// while (blockUntilSpecifiedAmountHasArrived && numberOfBytes==maxBytesToRead); + return numberOfBytes; +} +int UdpServer::empty(){ + return empty(0); +} +int UdpServer::empty(int maxCount){ + int count=0; + int n; + do { + if(waitUntilReady(true, 0)==0) + return 0; + float waste; + n=read(&waste, sizeof(float), false); + count++; + } while (n>0 && (maxCount<=0 || maxCount<count)); + printf("socket emptied with %d reads\n", count); + return count; +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/include/NetworkSend.h Sat Feb 13 04:09:12 2016 +0000 @@ -0,0 +1,115 @@ +//scope.cpp +#ifndef SCOPE_H_ +#define SCOPE_H_ + +#ifdef USE_JUCE +#include <JuceHeader.h> +#else +#include <BeagleRT.h> +#include <rtdk.h> +#include <cmath> +#include <UdpClient.h> +#include <vector> +#include <string> +extern bool gShouldStop; +#endif /* USE_JUCE */ + +#define NETWORK_AUDIO_BUFFER_SIZE 302 +#define UDP_BUFFER_HEADER_CHANNEL_INDEX 0 +#define UDP_BUFFER_HEADER_TIMESTAMP_INDEX 1 +#define UDP_BUFFER_HEADER_LENGTH 2 + +struct NetworkBuffer{ + int channelNumber; + int numBuffers; + int writeBuffer; + int readBuffer; + int writePointer; + float** buffers; + bool* doneOnTime; + bool* readyToBeSent; + bool enabled; + int sampleCount; + static const int bufferLength=NETWORK_AUDIO_BUFFER_SIZE; + static const int headerLength=UDP_BUFFER_HEADER_LENGTH; + static const int headerChannelIndex=UDP_BUFFER_HEADER_CHANNEL_INDEX; + static const int headerTimestampIndex=UDP_BUFFER_HEADER_TIMESTAMP_INDEX; +}; + +#ifdef USE_JUCE +class NetworkSend: public Thread { +#else +class NetworkSend { +#endif /* USE_JUCE */ + float sampleRate; +#ifdef USE_JUCE + DatagramSocket udpClient; + int sleepTimeMs; + String remoteHostname; + int remotePortNumber; +#else + UdpClient udpClient; + bool isThreadRunning(); + static int sleepTimeMs; + static bool threadShouldExit(); + static bool threadIsExiting; + static bool threadRunning; + static bool staticConstructed; + static void staticConstructor(); + static AuxiliaryTask sendDataTask; //TODO: allow different AuxiliaryTasks for different priorities (e.g.: audio vs scope) + static std::vector<NetworkSend *> objAddrs; +#endif /* USE_JUCE */ + void dealloc(); +public: + NetworkBuffer channel; +#ifdef USE_JUCE + NetworkSend(const String &threadName); +#else + NetworkSend(); +#endif + ~NetworkSend(); + void setup(float aSampleRate, int blockSize, int aChannelNumber, int aPort, const char *aServer); + void cleanup(); + void sendData(); + void log(float value); + void setPort(int aPort); + void setServer(const char* aServer); + void setChannelNumber(int aChannelNumber); + int getChannelNumber(); + int getTimestamp(); +#ifdef USE_JUCE + void run(); +#else + static int getNumInstances(); + static void sendAllData(); + static void startThread(); + static void stopThread(); + static void run(); +#endif /* USE_JUCE */ +}; + +#ifdef USE_JUCE +#else +/** + * An array of NetworkSend objects with some default parameters + * + * All sending on the same port (defaults to 9999) + * All sending to the same server (defaults to 127.0.0.1) +*/ +class Scope { + std::vector<NetworkSend> channels; + void deallocate(); +public: + Scope(int aNumChannels); + ~Scope(); + void log(int channel, float value); + void setup(); + void setup(float sampleRate, int aPort, const char* aServer); + void sendData(); + void setPort(int port); + void setPort(int channel, int port); + int getNumChannels(); +}; +#endif /* USE_JUCE */ + +#endif /* SCOPE_H */
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/include/ReceiveAudioThread.h Sat Feb 13 04:09:12 2016 +0000 @@ -0,0 +1,105 @@ +#ifndef RECEIVEAUDIOTHREAD_H_INCLUDED +#define RECEIVEAUDIOTHREAD_H_INCLUDED + +#ifdef USE_JUCE +#include <JuceHeader.h> +#else +#include <BeagleRT.h> +#include <UdpServer.h> +#include <vector> +#include <iostream> +#include <native/task.h> +#include <native/timer.h> +#include <math.h> + +#endif /*USE_JUCE*/ + +#ifdef USE_JUCE +class ReceiveAudioThread : public Thread { +#else +class ReceiveAudioThread{ +#endif /* USE_JUCE */ +private: + // FILE *fd; //DEBUG + // FILE *fd2; //DEBUG +#ifdef USE_JUCE + DatagramSocket socket; +#else + UdpServer socket; +#endif /* USE_JUCE */ + bool listening; + bool bufferReady; +#ifdef USE_JUCE + bool threadRunning; //do we really need this ? +#else + static bool threadRunning; + static bool threadIsExiting; +#endif + float *buffer; + float *stackBuffer; + int bufferLength; + float readPointer; + int writePointer; + int lastValidPointer; +#ifdef USE_JUCE + int sleepTime; +#else + static int sleepTime; +#endif + int waitForSocketTime; + int payloadLength; //size of the payload of each datagram + int headerLength; //size of the header of each datagram + int bytesToRead; + int threadPriority; + int channel; + int timestamp; + void dealloc(); + void wrapWritePointer(); + void pushPayload(int startIndex); + void popPayload(int startIndex); + int readUdpToBuffer(); +#ifdef USE_JUCE +#else + RTIME lastTime; // Used for clock synchronization + static bool threadShouldExit(); + static bool staticConstructed; + static void staticConstructor(); + static AuxiliaryTask receiveDataTask; //TODO: allow different AuxiliaryTasks for different priorities (e.g.: audio vs scope) + static std::vector<ReceiveAudioThread *> objAddrs; +#endif +public: +#ifdef USE_JUCE + ReceiveAudioThread(const String &threadName); +#else + ReceiveAudioThread(); +#endif + ~ReceiveAudioThread(); + void init(int port, int aSamplesPerBlock, int channel); + void bindToPort(int aPort); + bool isListening(); + float* getCurrentBuffer(int length); + /** + * Copies the samples to a non-interleaved buffer. + */ + int getSamplesSrc(float *destination, int length, float samplingRateRatio); + /** + * Copies the samples to an interleaved buffer. + */ + int getSamplesSrc(float *destination, int length, + float samplingRateRatio, int numChannelsInDestination, + int channelToWriteTo); + bool isBufferReady(); + int getTimestamp(); +#ifdef USE_JUCE // if we are in Juce, then we run a separate thread for each receiver + // (as each of them are typically receiving on a mono or stereo track) + void run(); +#else + RTIME getLastTime(); + void static run(); //while in BeagleRT we have a single thread that receives for all the instances. + //TODO: make run() private in BeagleRT + static void startThread(); + static void stopThread(); + static int getNumInstances(); +#endif // USE_JUCE +}; +#endif // RECEIVEAUDIOTHREAD_H_INCLUDED
--- a/include/Scope.h Sat Feb 13 01:51:01 2016 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,103 +0,0 @@ -//scope.cpp -#include <BeagleRT.h> -#include <rtdk.h> -#include <cmath> -#include <UdpClient.h> - -#define BUILD_FOR_UDPRECEIVE_PLUGIN -#define NETWORK_AUDIO_BUFFER_SIZE 302 -struct networkAudio{ - int timestamp; - int currentBuffer; - int index; - float buffers[2][NETWORK_AUDIO_BUFFER_SIZE]; - int doneOnTime; - bool toBeSent; - UdpClient udpClient; -}; - -#define NUM_SCOPE_CHANNELS 6 - -static void SendScopeData(); - -class Scope { - int sampleCount; - float sampleRate; - AuxiliaryTask scopeTask; - public: - int numChannels; - networkAudio channel[NUM_SCOPE_CHANNELS]; - Scope(){ - numChannels = NUM_SCOPE_CHANNELS; - sampleCount = 0; -#ifdef BUILD_FOR_UDPRECEIVE_PLUGIN - char server[]="192.168.7.1"; -#else - char server[]="127.0.0.1"; -#endif /* BUILD_FOR_UDPRECEIVE_PLUGIN */ - printf("Sending messages to : %s\n", server); - for(int n=0; n<numChannels; n++){ - channel[n].doneOnTime=1; - channel[n].index=2; //leave space for the heading message (channel, timestamp) - channel[n].timestamp=0; - channel[n].currentBuffer=0; - channel[n].toBeSent=false; - channel[n].udpClient.setServer(server); -#ifdef BUILD_FOR_UDPRECEIVE_PLUGIN - channel[n].udpClient.setPort(9999+n); -#else - channel[n].udpClient.setPort(9999); -#endif /* BUILD_FOR_UDPRECEIVE_PLUGIN */ - } - } - void setup(float _sampleRate); - void log(float channel1=0.0, float channel2=0.0, float channel3=0.0, float channel4=0.0, float channel5=0.0, float channel6=0.0){ - - for(int j=0; j<numChannels; j++){ - if(channel[j].index==(NETWORK_AUDIO_BUFFER_SIZE)){ // when the buffer is ready ... - channel[j].buffers[channel[j].currentBuffer][0] = (float)j; - channel[j].buffers[channel[j].currentBuffer][1] = (float)channel[j].timestamp; - channel[j].toBeSent=true; - channel[j].index=2; //reset the counter - if(channel[j].doneOnTime==0) - rt_printf("Network buffer underrun :-{\n"); - channel[j].timestamp=sampleCount; - channel[j].currentBuffer=!channel[j].currentBuffer; //switch buffer - channel[j].doneOnTime=0; - BeagleRT_scheduleAuxiliaryTask(scopeTask); //send the buffer - } - } - - channel[0].buffers[channel[0].currentBuffer][channel[0].index++]=channel1; - channel[1].buffers[channel[1].currentBuffer][channel[1].index++]=channel2; - channel[2].buffers[channel[2].currentBuffer][channel[2].index++]=channel3; - channel[3].buffers[channel[3].currentBuffer][channel[3].index++]=channel4; - channel[4].buffers[channel[4].currentBuffer][channel[4].index++]=channel5; - channel[5].buffers[channel[5].currentBuffer][channel[5].index++]=channel6; - - sampleCount++; - } -}; - -Scope* gOscilloscopeInstance; - -void Scope::setup(float _sampleRate){ - sampleRate = _sampleRate; - gOscilloscopeInstance = this; - scopeTask = BeagleRT_createAuxiliaryTask(*SendScopeData, 98, "transmit-receive-audio"); -} - -//Scope scope; - -static void SendScopeData(){ - for(int n=0; n<gOscilloscopeInstance->numChannels; n++){ - if(gOscilloscopeInstance->channel[n].toBeSent){ - gOscilloscopeInstance->channel[n].toBeSent=false; - gOscilloscopeInstance->channel[n].udpClient.send( - gOscilloscopeInstance->channel[n].buffers[!gOscilloscopeInstance->channel[n].currentBuffer], - NETWORK_AUDIO_BUFFER_SIZE*sizeof(float) - ); - gOscilloscopeInstance->channel[n].doneOnTime=1; - } - } -}
--- a/include/UdpClient.h Sat Feb 13 01:51:01 2016 +0000 +++ b/include/UdpClient.h Sat Feb 13 04:09:12 2016 +0000 @@ -23,6 +23,8 @@ int port; int enabled; int outSocket; + struct timeval stTimeOut; + fd_set stWriteFDS; bool isSetPort; bool isSetServer; struct sockaddr_in destinationServer; @@ -33,6 +35,8 @@ void setPort(int aPort); void setServer(const char* aServerName); int send(void* message, int size); + int write(const char* remoteHostname, int remotePortNumber, void* sourceBuffer, int numBytesToWrite); + int waitUntilReady(bool readyForReading, int timeoutMsecs); };
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/include/UdpServer.h Sat Feb 13 04:09:12 2016 +0000 @@ -0,0 +1,70 @@ +/* + * udpServer.h + * + * Created on: 19 May 2015 + * Author: giulio moro + */ + +#ifndef UDPSERVER_H_ +#define UDPSERVER_H_ + +#include <sys/types.h> +#include <sys/socket.h> +#include <netinet/in.h> +#include <arpa/inet.h> +#include <errno.h> +#include <netdb.h> +#include <stdio.h> +#include <stdlib.h> +#include <unistd.h> +#include <string.h> + +class UdpServer{ + private: + int port; + int enabled; + int inSocket; + struct sockaddr_in server; + struct timeval stTimeOut; + struct timeval stZeroTimeOut; + fd_set stReadFDS; + int size; + void *wasteBuffer; + int wasteBufferSize; + int length; + socklen_t fromLength; + struct sockaddr_in from; + public: + UdpServer(); + UdpServer(int aPort); + ~UdpServer(); + bool init(int aPort); + bool bindToPort(int aPort); + int getBoundPort() const; + /* + * Reads bytes from the socket. + * + * Drop-in replacement for JUCE DatagramSocket::read() + * + If blockUntilSpecifiedAmountHasArrived is true, the method will block until maxBytesToRead + bytes have been read, (or until an error occurs). If this flag is false, the method will + return as much data as is currently available without blocking. + */ + int read(void* destBuffer, int maxBytesToRead, bool blockUntilSpecifiedAmountHasArrived); + void close(); + int empty(); + int empty(int maxCount); + /* + * Waits until the socket is ready for reading or writing. + * + Drop-in replacement for JUCE DatagramSocket::waitUntilReady. + If readyForReading is true, it will wait until the socket is ready for reading; if false, it will wait until it's ready for writing. + If the timeout is < 0, it will wait forever, or else will give up after the specified time. + If the socket is ready on return, this returns 1. If it times-out before the socket becomes ready, it returns 0. If an error occurs, it returns -1. + */ + int waitUntilReady(bool readyForReading, int timeoutMsecs); +}; + + + +#endif /* UDPSERVER_H_ */