giuliomoro@131: #include "NetworkSend.h"
giuliomoro@111: 
giuliomoro@129: #ifdef USE_JUCE
giuliomoro@129: #else
giuliomoro@111: //initialize the static members of NetworkSend
giuliomoro@111: bool NetworkSend::staticConstructed=false;
giuliomoro@131: int NetworkSend::sleepTimeMs;
giuliomoro@131: bool NetworkSend::threadIsExiting;
giuliomoro@131: bool NetworkSend::threadRunning;
giuliomoro@111: std::vector<NetworkSend*> NetworkSend::objAddrs(0);
giuliomoro@116: AuxiliaryTask NetworkSend::sendDataTask=NULL;
giuliomoro@111: 
giuliomoro@129: void sendData(){
giuliomoro@131: 	NetworkSend::run();
giuliomoro@111: }
giuliomoro@129: void NetworkSend::staticConstructor(){
giuliomoro@129: 	if(staticConstructed==true)
giuliomoro@129: 		return;
giuliomoro@129: 	staticConstructed=true;
giuliomoro@131: 	threadIsExiting=false;
giuliomoro@131: 	threadRunning=false;
giuliomoro@129: 	sendDataTask = BeagleRT_createAuxiliaryTask(::sendData, 95, "sendDataTask"); //TODO: allow variable priority
giuliomoro@129: }
giuliomoro@111: void NetworkSend::sendAllData(){
giuliomoro@111: 	for(unsigned int n=0; n<NetworkSend::objAddrs.size(); n++){
giuliomoro@111: 		NetworkSend::objAddrs[n]->sendData();
giuliomoro@111: 	}
giuliomoro@111: }
giuliomoro@129: int NetworkSend::getNumInstances(){
giuliomoro@129: 	return objAddrs.size();
giuliomoro@129: }
giuliomoro@131: void NetworkSend::startThread(){
giuliomoro@131: 	BeagleRT_scheduleAuxiliaryTask(sendDataTask);
giuliomoro@131: }
giuliomoro@131: void NetworkSend::stopThread(){
giuliomoro@131: 	threadIsExiting=true;
giuliomoro@131: }
giuliomoro@131: bool NetworkSend::threadShouldExit(){
giuliomoro@131: 	return(gShouldStop || threadIsExiting);
giuliomoro@131: }
giuliomoro@131: bool NetworkSend::isThreadRunning(){
giuliomoro@131: 	return threadRunning;
giuliomoro@131: }
giuliomoro@129: #endif /* USE_JUCE */
giuliomoro@111: 
giuliomoro@131: #ifdef USE_JUCE
giuliomoro@131: NetworkSend::NetworkSend(const String &threadName):
giuliomoro@131: 		Thread(threadName)
giuliomoro@131: #else
giuliomoro@111: NetworkSend::NetworkSend()
giuliomoro@131: #endif /* USE_JUCE */
giuliomoro@111: {
giuliomoro@131: 	channel.buffers=NULL;
giuliomoro@131: 	channel.doneOnTime=NULL;
giuliomoro@131: 	channel.readyToBeSent=NULL;
giuliomoro@131: 	channel.enabled=false;
giuliomoro@131: 	sleepTimeMs=2; //This should actually be initialized in the staticConstructor for non-Juce code, but doing it here makes it more portable
giuliomoro@131: 	channel.sampleCount=0;
giuliomoro@111: }
giuliomoro@116: 
giuliomoro@111: NetworkSend::~NetworkSend(){
giuliomoro@131: #ifdef USE_JUCE
giuliomoro@131: 	stopThread(1000);
giuliomoro@131: #else
giuliomoro@131: 	stopThread();
giuliomoro@111: 	for(unsigned int n=0; n<objAddrs.size(); n++){ //keep track of deleted instances;
giuliomoro@111: 		if(objAddrs[n]==this){
giuliomoro@111: 			objAddrs.erase(objAddrs.begin()+n);
giuliomoro@111: 			break;
giuliomoro@111: 		}
giuliomoro@111: 	}
giuliomoro@131: #endif
giuliomoro@131: 	dealloc();
giuliomoro@131: }
giuliomoro@131: void NetworkSend::dealloc(){
giuliomoro@131: 	channel.enabled=false;
giuliomoro@131: 	if(channel.buffers!=NULL){
giuliomoro@131: 		for(int n=0; n<channel.numBuffers; n++){
giuliomoro@131: 			free(channel.buffers[n]);
giuliomoro@131: 			channel.buffers[n]=NULL;
giuliomoro@131: 		}
giuliomoro@131: 		free(channel.buffers);
giuliomoro@131: 		channel.buffers=NULL;
giuliomoro@131: 	}
giuliomoro@131: 	free(channel.readyToBeSent);
giuliomoro@131: 	channel.readyToBeSent=NULL;
giuliomoro@131: 	free(channel.doneOnTime);
giuliomoro@131: 	channel.doneOnTime=NULL;
giuliomoro@131: }
giuliomoro@131: void NetworkSend::cleanup(){
giuliomoro@131: 	dealloc();
giuliomoro@111: }
giuliomoro@111: 
giuliomoro@131: void NetworkSend::setup(float aSampleRate, int blockSize, int aChannelNumber, int aPort, const char *aServer){
giuliomoro@131: #ifdef USE_JUCE
giuliomoro@131: #else
giuliomoro@111: 	staticConstructor(); //FIXME: ideally this should be in the constructor, but this is not currently possible
giuliomoro@111: 						//because of limitations in BeagleRT_createAuxiliaryTask()
giuliomoro@111: 	//keep track of added active instances
giuliomoro@111: 	objAddrs.push_back(this);//TODO: this line should be in the constructor, but something weird happens if
giuliomoro@111: 	// an instance of NetworkSend is then declared globally: the constructor gets called,
giuliomoro@111: 	// and objAddrs.size()==1 but when you get to setup, objAddrs.size() has reverted back to 0, without
giuliomoro@111: 	// any destructor being called in between ...
giuliomoro@131: #endif /* USE_JUCE */
giuliomoro@131: 	cleanup();
giuliomoro@131: 	int numSamples=blockSize*4>4*channel.bufferLength ? blockSize*4 : 4*channel.bufferLength;
giuliomoro@131: 	channel.numBuffers= 1+numSamples/channel.bufferLength; //the +1 takes the ceil() of the division
giuliomoro@131: 	channel.buffers=(float**)malloc(channel.numBuffers*sizeof(float*));
giuliomoro@131: 	printf("NumBuffers: %d\n", channel.numBuffers);
giuliomoro@131: 	if(channel.buffers==NULL)
giuliomoro@131: 		return;
giuliomoro@131: 	for(int n=0; n<channel.numBuffers; n++){
giuliomoro@131: 		channel.buffers[n]=(float*)malloc(channel.bufferLength*sizeof(float));
giuliomoro@131: 		if(channel.buffers[n]==NULL)
giuliomoro@131: 			return;
giuliomoro@131: 	}
giuliomoro@131: 	channel.readyToBeSent=(bool*)malloc(channel.numBuffers*sizeof(bool));
giuliomoro@131: 	channel.doneOnTime=(bool*)malloc(channel.numBuffers*sizeof(bool));
giuliomoro@131: 	for(int n=0; n<channel.numBuffers; n++){
giuliomoro@131: 		channel.readyToBeSent[n]=false;
giuliomoro@131: 		channel.doneOnTime[n]=true;
giuliomoro@131: 	}
giuliomoro@131: 	if(channel.readyToBeSent==NULL || channel.doneOnTime==NULL)
giuliomoro@131: 		return;
giuliomoro@131: 	channel.writePointer=0;
giuliomoro@131: 	channel.writeBuffer=0;
giuliomoro@131: 	channel.readBuffer=0;
giuliomoro@111: 	setChannelNumber(aChannelNumber);
giuliomoro@131: 	setPort(aPort); //TODO: check for the return value
giuliomoro@131: 	setServer(aServer); //TODO: check for the return value
giuliomoro@120: 	printf("Channel %d is sending messages to: %s:%d at %fHz\n", getChannelNumber(), aServer, aPort, aSampleRate);
giuliomoro@131: 	channel.enabled=true;
giuliomoro@111: }
giuliomoro@111: 
giuliomoro@111: void NetworkSend::log(float value){ //TODO: add a vectorized version of this method
giuliomoro@131: 	if(channel.enabled==false)
giuliomoro@131: 		return;
giuliomoro@131: 	if(channel.writePointer==channel.bufferLength){ // when the buffer is filled ...
giuliomoro@131: 		channel.readyToBeSent[channel.writeBuffer]=true; // flag it as such
giuliomoro@131: //		printf("Scheduling for send %d\n",(int)channel.buffers[channel.writeBuffer][channel.headerTimestampIndex]);
giuliomoro@131: 		channel.writePointer=channel.headerLength; //reset the writePointer
giuliomoro@131: 		channel.writeBuffer=(channel.writeBuffer+1); //switch buffer
giuliomoro@131: 		if(channel.writeBuffer==channel.numBuffers) // and wrap it
giuliomoro@131: 			channel.writeBuffer=0;
giuliomoro@131: //		printf("WriteBuffer:%d\n", channel.writeBuffer);
giuliomoro@131: 		if(channel.doneOnTime[channel.writeBuffer]==false){ //check if this buffer's last sending has completed on time ...
giuliomoro@131: 			printf("Network buffer underrun. timestamp: %d :-{\n",
giuliomoro@131: 					(int)channel.buffers[channel.writeBuffer][channel.headerTimestampIndex]);
giuliomoro@111: 		}
giuliomoro@131: 		channel.doneOnTime[channel.writeBuffer]=false; // ... and then reset the flag
giuliomoro@131: #ifdef USE_JUCE
giuliomoro@131: 		if(isThreadRunning()==false){
giuliomoro@131: 			startThread(10);
giuliomoro@131: 		}
giuliomoro@131: #else
giuliomoro@131: 		if(isThreadRunning()==false){
giuliomoro@131: 			startThread();
giuliomoro@131: 		}
giuliomoro@131: #endif /* USE_JUCE */
giuliomoro@111: 	}
giuliomoro@131: 	if(channel.writePointer==channel.headerLength){ // we are about to start writing in the buffer, let's set the header
giuliomoro@131: 		//set dynamic header values here. Static values are set in setup() and setChannelNumber().
giuliomoro@131: 		channel.buffers[channel.writeBuffer][channel.headerTimestampIndex]=(float)channel.sampleCount; //timestamp
giuliomoro@131: 		channel.sampleCount++;
giuliomoro@111: 		//add here more header fields
giuliomoro@111: 	}
giuliomoro@131:     channel.buffers[channel.writeBuffer][channel.writePointer++]=value;
giuliomoro@131: //	sampleCount++;
giuliomoro@111: };
giuliomoro@111: 
giuliomoro@111: void NetworkSend::setServer(const char *aServer){
giuliomoro@131: #ifdef USE_JUCE
giuliomoro@131: 	remoteHostname=String::fromUTF8(aServer);
giuliomoro@131: #else
giuliomoro@111: 	udpClient.setServer(aServer);
giuliomoro@131: #endif /* USE_JUCE */
giuliomoro@111: }
giuliomoro@111: void NetworkSend::setPort(int aPort){
giuliomoro@131: #ifdef USE_JUCE
giuliomoro@131: 	remotePortNumber=aPort;
giuliomoro@131: #else
giuliomoro@111: 	udpClient.setPort(aPort);
giuliomoro@131: #endif /* USE_JUCE */
giuliomoro@111: }
giuliomoro@111: 
giuliomoro@111: void NetworkSend::setChannelNumber(int aChannelNumber){
giuliomoro@111: 	channel.channelNumber=aChannelNumber;
giuliomoro@131: 	for(int n=0; n<channel.numBuffers; n++){ //initialize the header
giuliomoro@131: 		channel.buffers[n][channel.headerChannelIndex]=channel.channelNumber;
giuliomoro@131: 		//add here more static header fields
giuliomoro@131: 	}
giuliomoro@111: };
giuliomoro@111: int NetworkSend::getChannelNumber(){
giuliomoro@111: 	return channel.channelNumber;
giuliomoro@111: };
giuliomoro@111: 
giuliomoro@132: int NetworkSend::getTimestamp(){
giuliomoro@132: 	return channel.buffers[channel.readBuffer][channel.headerTimestampIndex];
giuliomoro@132: }
giuliomoro@132: 
giuliomoro@111: void NetworkSend::sendData(){
giuliomoro@131: 	if(channel.enabled==false)
giuliomoro@131: 		return;
giuliomoro@131: 	while(channel.readyToBeSent[channel.readBuffer]==true){
giuliomoro@131: 		channel.readyToBeSent[channel.readBuffer]=false;
giuliomoro@131: 		void* sourceBuffer=channel.buffers[channel.readBuffer];
giuliomoro@131: //		printf("Trying to send timestamp %d\n",(int)((float*)sourceBuffer)[channel.headerTimestampIndex]);
giuliomoro@131: //		printf("ReadBuffer:%d\n", channel.readBuffer);
giuliomoro@131: 		unsigned int numBytesToSend=NETWORK_AUDIO_BUFFER_SIZE*sizeof(float);
giuliomoro@131: 		//TODO: call waitUntilReady before trying to write/send, to avoid blocks! (OR NOT?)
giuliomoro@131: #ifdef USE_JUCE
giuliomoro@131: 		if(1==udpClient.waitUntilReady(0, 5)){
giuliomoro@131: 			udpClient.write(remoteHostname, remotePortNumber, sourceBuffer, numBytesToSend);
giuliomoro@131: 			channel.doneOnTime[channel.readBuffer]=true;
giuliomoro@131: 			//  printf ("Sent timestamp: %d\n", (int)((float*)sourceBuffer)[1]);
giuliomoro@131: 		} else {
giuliomoro@131: 			//  printf ("Not ready timestamp: %d\n", (int)((float*)sourceBuffer)[1]);
giuliomoro@131: 		}
giuliomoro@131: #else
giuliomoro@131: 		udpClient.send(sourceBuffer, numBytesToSend);
giuliomoro@131: 		channel.doneOnTime[channel.readBuffer]=true;
giuliomoro@131: #endif /* USE_JUCE */
giuliomoro@131: 		channel.readBuffer++;
giuliomoro@131: 		if(channel.readBuffer==channel.numBuffers)
giuliomoro@131: 			channel.readBuffer=0;
giuliomoro@111: 	}
giuliomoro@111: }
giuliomoro@111: 
giuliomoro@131: void NetworkSend::run(){
giuliomoro@131: #ifdef USE_JUCE
giuliomoro@131: 	//  std::chrono::high_resolution_clock::time_point t1;
giuliomoro@131:     //  std::chrono::high_resolution_clock::time_point t2;
giuliomoro@131:     //  std::chrono::high_resolution_clock::time_point t3;
giuliomoro@131: 	while(threadShouldExit()==false){
giuliomoro@131: 		//  t3 = std::chrono::high_resolution_clock::now();
giuliomoro@131: 		//  t1 = std::chrono::high_resolution_clock::now();
giuliomoro@131: 		sendData();
giuliomoro@131: 		//  t2 = std::chrono::high_resolution_clock::now();
giuliomoro@131: 		//  auto duration1 = std::chrono::duration_cast<std::chrono::microseconds>( t3 - t1 ).count();
giuliomoro@131: 		//  auto duration2 = std::chrono::duration_cast<std::chrono::microseconds>( t2 - t1 ).count();
giuliomoro@131: 		//  if(duration2>0)
giuliomoro@131: 			//  std::cout << "Duration is: " << duration2 <<". Whole loop is: " << duration1 << "\n"; 
giuliomoro@131: 		sleep(1);
giuliomoro@131: 	}
giuliomoro@131: #else
giuliomoro@131: 	threadRunning=true;
giuliomoro@131: 	while(threadShouldExit()==false){
giuliomoro@131: 		sendAllData();
giuliomoro@131: 		usleep(sleepTimeMs*1000);
giuliomoro@131: 	}
giuliomoro@131: 	threadRunning=false;
giuliomoro@131: #endif
giuliomoro@131: }
giuliomoro@129: #ifdef USE_JUCE
giuliomoro@129: #else
giuliomoro@111: Scope::Scope(int aNumChannels):
giuliomoro@111: 		channels(aNumChannels)
giuliomoro@111: {};
giuliomoro@111: Scope::~Scope(){};
giuliomoro@111: 
giuliomoro@111: void Scope::log(int channel, float value){
giuliomoro@111: 	if(channel>=getNumChannels()) //TODO: assert this
giuliomoro@111: 		return;
giuliomoro@111: 	channels[channel].log(value);
giuliomoro@111: }
giuliomoro@111: 
giuliomoro@111: void Scope::setup(){
giuliomoro@111: 	setup(44100, 9999, "127.0.0.1");
giuliomoro@111: }
giuliomoro@111: 
giuliomoro@111: void Scope::setup(float sampleRate, int aPort, const char* aServer){
giuliomoro@111: 	for(int n=0; n<getNumChannels(); n++){
giuliomoro@131: 		channels[n].setup(sampleRate, 128, n, aPort, aServer); //TODO: replace 128 with the actual block size
giuliomoro@111: 	}
giuliomoro@111: }
giuliomoro@111: 
giuliomoro@119: void Scope::setPort(int port){
giuliomoro@119: 	for(int n=0; n<getNumChannels(); n++){
giuliomoro@119: 		channels[n].setPort(port);
giuliomoro@119: 	}
giuliomoro@119: }
giuliomoro@120: void Scope::setPort(int channel, int aPort){
giuliomoro@120: 	channels[channel].setPort(aPort);
giuliomoro@120: 	printf("Channel %d is now sending to port %d\n", channel, aPort);
giuliomoro@119: }
giuliomoro@119: 
giuliomoro@111: int Scope::getNumChannels(){
giuliomoro@111: 	return channels.size();
giuliomoro@111: }
giuliomoro@111: 
giuliomoro@111: void Scope::sendData(){
giuliomoro@111: 	NetworkSend::sendAllData();
giuliomoro@111: }
giuliomoro@129: #endif