audio_capture_manager.cpp

/*
 * If not stated otherwise in this file or this component's LICENSE file the
 * following copyright and licenses apply:
 *
 * Copyright 2016 RDK Management
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
*/
#include <stdlib.h>
#include <string.h>
#include <sstream>
#include "audio_capture_manager.h"
#include <algorithm>
#include <chrono>
#include <unistd.h>
#include "rmfAudioCapture.h"

using namespace audiocapturemgr;

const unsigned int QUEUE_CHECK_LOOP_TIME_US = 1000 * 50; //50ms
static const size_t DEFAULT_FIFO_SIZE = 64 * 1024;
static const size_t	DEFAULT_THRESHOLD = 8 * 1024;
static const unsigned int DEFAULT_DELAY_COMPENSATION = 0;
static const unsigned int MAX_QMGR_BUFFER_DURATION_S = 30; //safe maximum, beyond which the queue will be flushed without waiting for buffers to be consumed.

static void * q_mgr_thread_launcher(void * data)
{
	q_mgr * ptr = (q_mgr *) data;
	ptr->data_processor_thread();
	return NULL;
}

rmf_Error q_mgr::data_callback(void *context, void *buf, unsigned int size)
{
	((q_mgr *)context)->add_data((unsigned char*)buf, size);
	return RMF_SUCCESS;
}


inline void q_mgr::lock(pthread_mutex_t &mutex)
{
	REPORT_IF_UNEQUAL(0, pthread_mutex_lock(&mutex));
}
inline void q_mgr::unlock(pthread_mutex_t &mutex)
{
	REPORT_IF_UNEQUAL(0, pthread_mutex_unlock(&mutex));
}
inline void q_mgr::notify_data_ready()
{
	DEBUG("Enter.\n");
	if(m_notify_new_data)
	{
		DEBUG("Posting semaphore.\n");
		REPORT_IF_UNEQUAL(0, sem_post(&m_sem));
		m_notify_new_data = false;
	}
}
void q_mgr::swap_queues() //caller must lock before invoking this.
{
	if(!m_current_outgoing_q->empty())
	{
		WARN("Outgoing queue wasn't empty. Flushing it now.\n");
		flush_queue(m_current_outgoing_q);
	}
	std::vector <audio_buffer *> * temp = m_current_incoming_q;
	m_current_incoming_q = m_current_outgoing_q;
	m_current_outgoing_q = temp;
}

void q_mgr::flush_queue(std::vector <audio_buffer *> *q)
{
	DEBUG("Flushing queue.\n");
	std::vector <audio_buffer *>::iterator iter;
	for(iter = q->begin(); iter != q->end(); iter++)
	{
		free_audio_buffer(*iter);	
	}
	q->clear();
}

namespace audiocapturemgr
{
	void get_individual_audio_parameters(const audio_properties_t &audio_props, unsigned int &sampling_rate, unsigned int &bits_per_sample, unsigned int &num_channels)
	{
		bits_per_sample = 0;
		sampling_rate= 0;
		num_channels = 0;

		switch(audio_props.sampling_frequency)
		{
			case racFreq_e16000:
				sampling_rate = 16000;
				break;
			case racFreq_e24000:
				sampling_rate = 24000;
				break;
			case racFreq_e32000:
				sampling_rate = 32000;
				break;
			case racFreq_e44100:
				sampling_rate = 44100;
				break;
			case racFreq_e48000:
				sampling_rate = 48000;
				break;
			default:
				ERROR("Bad sampling rate: 0x%x\n", audio_props.sampling_frequency);
				break;
		}

		switch(audio_props.format)
		{
			case racFormat_e16BitStereo:
				bits_per_sample = 16; num_channels = 2;
				break;
			case racFormat_e24BitStereo:
				bits_per_sample = 24; num_channels = 2;
				break;
			case racFormat_e16BitMonoLeft:
				bits_per_sample = 16; num_channels = 1;
				break;
			case racFormat_e16BitMonoRight:
				bits_per_sample = 16; num_channels = 1;
				break;
			case racFormat_e16BitMono:
				bits_per_sample = 16; num_channels = 1;
				break;
			case racFormat_e24Bit5_1:
				bits_per_sample = 24; num_channels = 6;
				break;
			default:
				ERROR("Unknown format 0x%x\n", audio_props.format);
		}
	}

	unsigned int calculate_data_rate(const audio_properties_t &audio_props)
	{
		unsigned int bits_per_sample = 0;
		unsigned int sampling_rate= 0;
		unsigned int num_channels = 0;
		get_individual_audio_parameters(audio_props, sampling_rate, bits_per_sample, num_channels); 	
		unsigned int data_rate = bits_per_sample * sampling_rate * num_channels / 8;
		INFO("Audio properties: %dkHz, %d bit, %d channel, byte rate: %d\n", sampling_rate, bits_per_sample, num_channels, data_rate);
		return data_rate;
	}
	
	std::string get_suffix(unsigned int ticker)
	{
		std::ostringstream stream;
		stream << ticker;
		std::string outstring = stream.str();
		return outstring;
	}
}

q_mgr::q_mgr() : m_inflow_byte_counter(0), m_num_clients(0), m_notify_new_data(false), m_started(false), m_device_handle(NULL), m_stop_data_monitor(true)
{
	INFO("Creating instance 0x%p.\n", static_cast <void *>(this));
	pthread_mutexattr_t mutex_attribute;
	REPORT_IF_UNEQUAL(0, pthread_mutexattr_init(&mutex_attribute));
	REPORT_IF_UNEQUAL(0, pthread_mutexattr_settype(&mutex_attribute, PTHREAD_MUTEX_ERRORCHECK));
	REPORT_IF_UNEQUAL(0, pthread_mutex_init(&m_q_mutex, &mutex_attribute));
	REPORT_IF_UNEQUAL(0, pthread_mutex_init(&m_client_mutex, &mutex_attribute));
	REPORT_IF_UNEQUAL(0, sem_init(&m_sem, 0, 0));
	m_current_incoming_q = new std::vector <audio_buffer *>;
	m_current_outgoing_q = new std::vector <audio_buffer *>; 
	m_processing_thread_alive = false; //CID:80565 :  Initialize bool variable

	REPORT_IF_UNEQUAL(0, pthread_create(&m_thread, NULL, q_mgr_thread_launcher, (void *) this));
	
	int ret = RMF_AudioCapture_Open(&m_device_handle);
	INFO("open() result is 0x%x\n", ret);
	
	RMF_AudioCapture_Settings settings;
	RMF_AudioCapture_GetDefaultSettings(&settings);
	m_audio_properties.format = settings.format;
	m_audio_properties.sampling_frequency = settings.samplingFreq;
	m_audio_properties.fifo_size = DEFAULT_FIFO_SIZE; 
	m_audio_properties.threshold = DEFAULT_THRESHOLD;
	m_audio_properties.delay_compensation_ms = DEFAULT_DELAY_COMPENSATION; 
	m_bytes_per_second = calculate_data_rate(m_audio_properties);
	m_max_queue_size = (MAX_QMGR_BUFFER_DURATION_S * m_bytes_per_second) / m_audio_properties.threshold;
	INFO("Max incoming queue size is now %d\n", m_max_queue_size);
}
q_mgr::~q_mgr()
{
	INFO("Deleting instance 0x%p.\n", static_cast <void *>(this));
	if(true == m_started)
	{
		stop();
	}
    if(NULL != m_device_handle)
    {
		int ret = RMF_AudioCapture_Close(m_device_handle);
		INFO("close() result is 0x%x\n", ret);
		m_device_handle = NULL;
    }
	

	m_processing_thread_alive = false;
	REPORT_IF_UNEQUAL(0, sem_post(&m_sem));
	REPORT_IF_UNEQUAL(0, pthread_join(m_thread, NULL));
	REPORT_IF_UNEQUAL(0, sem_destroy(&m_sem));
	REPORT_IF_UNEQUAL(0, pthread_mutex_destroy(&m_q_mutex));
	REPORT_IF_UNEQUAL(0, pthread_mutex_destroy(&m_client_mutex));
	flush_queue(m_current_incoming_q);
	flush_queue(m_current_outgoing_q);
	delete m_current_incoming_q;
	delete m_current_outgoing_q;
}


int q_mgr::set_audio_properties(audio_properties_t &in_properties)
{
	DEBUG("Enter.\n");
	lock(m_q_mutex);
	if(0 == memcmp(&m_audio_properties, &in_properties, sizeof(m_audio_properties)))
	{
		unlock(m_q_mutex);
		return 0; //No change to properties
	}

	/* Validate incoming parameters.*/
	if((racFormat_eMax <= in_properties.format) || (racFreq_eMax <= in_properties.sampling_frequency))
	{
		ERROR("Bad parameters. Format: 0x%x, sampling freq: 0x%x\n", in_properties.format, in_properties.sampling_frequency);
		unlock(m_q_mutex);
		return -1;
	}

	m_audio_properties = in_properties;

	/*Update data rate.*/
	m_bytes_per_second = calculate_data_rate(m_audio_properties);
	m_max_queue_size = (MAX_QMGR_BUFFER_DURATION_S * m_bytes_per_second) / m_audio_properties.threshold;
	INFO("Max incoming queue size is now %d\n", m_max_queue_size);
	unlock(m_q_mutex);

	lock(m_client_mutex);
	bool needs_restart = m_started;
	if(m_started)
	{
		stop();
	}
	std::vector<audio_capture_client *>::iterator iter;
	for(iter = m_clients.begin(); iter != m_clients.end(); iter++)
	{
		(*iter)->notify_event(AUDIO_SETTINGS_CHANGE_EVENT);
	}
	if(needs_restart)
	{
		INFO("Restarting audio after settings change.\n");
		start();
	}

	unlock(m_client_mutex);
	return 0;
}

void q_mgr::get_audio_properties(audio_properties_t &out_properties)
{
	out_properties = m_audio_properties;
}

void q_mgr::get_default_audio_properties(audio_properties_t &out_properties)
{
	RMF_AudioCapture_Settings settings;
	RMF_AudioCapture_GetDefaultSettings(&settings);
	out_properties.format = settings.format;
	out_properties.sampling_frequency = settings.samplingFreq;
	out_properties.fifo_size = settings.fifoSize;
	out_properties.threshold = settings.threshold;
	out_properties.delay_compensation_ms = settings.delayCompensation_ms;
}
unsigned int q_mgr::get_data_rate()
{
	return m_bytes_per_second;
}

void q_mgr::add_data(unsigned char *buf, unsigned int size)
{
	DEBUG("Adding data.\n");
	lock(m_q_mutex);
	audio_buffer * temp = create_new_audio_buffer(buf, size, 0, m_num_clients);
	if(m_max_queue_size < m_current_incoming_q->size())
	{
		WARN("Incoming queue size over limit. Flushing.\n");
		flush_queue(m_current_incoming_q);
	}
	m_current_incoming_q->push_back(temp);
	//TODO: guard against accumulating audio_buffers if there is no consumption.
	notify_data_ready();
	unlock(m_q_mutex);
	m_inflow_byte_counter += size;
}
void q_mgr::data_processor_thread()
{
	DEBUG("Launching.\n");
	m_processing_thread_alive = true;
	while(m_processing_thread_alive)
	{
		/*
		 * 1. If there are audio_buffers remaining in outbound queue, process them.
		 * 2. If no audio_buffers remain in outbound queue, wait for semaphore.
		 * 3. When semaphore is posted, if data is available in the other queue, swap queues.
		 * */
		DEBUG("Enter processing loop.\n");
		while(!m_current_outgoing_q->empty())
		{
			process_data();	
		}
		DEBUG("No data in outgoing queue.\n");
		/* No more data to be consumed. Gotta check the other queue */
		lock(m_q_mutex);
		if(!m_current_incoming_q->empty())
		{	
			DEBUG("Incoming queue has buffers. Swapping them.\n");
			swap_queues();
			unlock(m_q_mutex);
		}
		else
		{
			/*Block until data is available.*/
			m_notify_new_data = true;
			unlock(m_q_mutex);
			DEBUG("Incoming queue is empty as well. Waiting until a buffer arrives.\n");
			if(0 != sem_wait(&m_sem))
			{
				ERROR("Critical semaphore error. Exiting\n");
				return;
			}
			/* Just in case the semaphore was posted to shut down this thred... */
			if(!m_processing_thread_alive)
			{
				break;
			}

			lock(m_q_mutex);
			DEBUG("New data available in incoming queue. Swapping them.\n");
			swap_queues();
			unlock(m_q_mutex);
		}

	}
	DEBUG("Exiting.\n");
}

void q_mgr::update_buffer_references()
{
	std::vector <audio_buffer *>::iterator buffer_iter;
	audio_buffer_get_global_lock();
	for(buffer_iter = m_current_outgoing_q->begin(); buffer_iter != m_current_outgoing_q->end(); buffer_iter++)
	{
		set_ref_audio_buffer(*buffer_iter, m_num_clients);	
	}
	audio_buffer_release_global_lock();
}

void q_mgr::process_data()
{
	DEBUG("Processing %d buffers of data.\n", m_current_outgoing_q->size());

	lock(m_client_mutex);
	
	update_buffer_references();
	
	std::vector <audio_buffer *>::iterator buffer_iter;
	for(buffer_iter = m_current_outgoing_q->begin(); buffer_iter != m_current_outgoing_q->end(); buffer_iter++)
	{
		std::vector <audio_capture_client *>::iterator client_iter;
		for(client_iter = m_clients.begin(); client_iter != m_clients.end(); client_iter++)
		{
			(*client_iter)->data_callback(*buffer_iter);	
		}
	}
	unlock(m_client_mutex);
	m_current_outgoing_q->clear();
}

int q_mgr::register_client(audio_capture_client * client)
{
	DEBUG("Enter.\n");
	lock(m_client_mutex);
	if(std::find(m_clients.begin(), m_clients.end(), client) == m_clients.end()) //Add only if it is not already present.
	{
		m_clients.push_back(client);
		m_num_clients = m_clients.size();
		if(1 == m_num_clients)
		{
			start();
		}
	}
	else
	{
		INFO("Will not register client as it is already present.\n");
	}
	unlock(m_client_mutex);
	INFO("Total clients: %d.\n", m_num_clients);
	return 0;
}
int q_mgr::unregister_client(audio_capture_client * client)
{
	DEBUG("Enter.\n");
	lock(m_client_mutex);
	m_clients.erase(std::remove(m_clients.begin(), m_clients.end(), client), m_clients.end());
	m_num_clients = m_clients.size();
	if(0 == m_num_clients)
	{
		stop();
	}
	unlock(m_client_mutex);
	INFO("Total clients: %d.\n", m_num_clients);
	return 0;
}

void q_mgr::flush_system()
{
	/*
	 * Flush everything in incoming queue.
	 * Wait until outbound queue is exhausted.
	 * Return.
	 * */
	flush_queue(m_current_incoming_q);
	while(!m_current_outgoing_q->empty())
	{
		usleep(QUEUE_CHECK_LOOP_TIME_US);
	}
	INFO("Exit.\n");
}

static void log_settings(RMF_AudioCapture_Settings &settings)
{
	INFO("Format 0x%x, sampling freq: 0x%x, FIFO size: %d, threshold: %d, delay compensation: %d\n",
		settings.format, settings.samplingFreq, settings.fifoSize, settings.threshold, settings.delayCompensation_ms);
}

void q_mgr::data_monitor()
{
	INFO("data_monitor thread has launched.\n");
	unsigned int saved_byte_counter = 0;
	bool is_stalled = false;
	std::unique_lock<std::mutex> wlock(m_data_monitor_mutex);
	while(false == m_stop_data_monitor)
	{
		auto ret = m_data_monitor_cv.wait_for(wlock, std::chrono::seconds(5), [this](){return m_stop_data_monitor;});
		if(false == ret)
		{ //This indicates a timeout or spurious wake.
			if(saved_byte_counter == m_inflow_byte_counter)
			{
				if(false == is_stalled)
				{
					WARN("Data inflow has stalled at %u bytes for instance 0x%p.\n", saved_byte_counter, static_cast <void *>(this));
					is_stalled = true;
				}
				else
					continue; //already logged the stall once. 
			}
			else
			{
				saved_byte_counter = m_inflow_byte_counter;
				if(true == is_stalled)
				{
					INFO("Data inflow has resumed for instance 0x%p.\n", static_cast <void *>(this));
					is_stalled = false;
				}
			}
		}
	}
	INFO("data_monitor thread exiting.\n");
}

int q_mgr::start()
{
	if(true == m_started)
	{
		WARN("Looks like device was already started.\n");
		return 0;
	}
	
	m_inflow_byte_counter = 0;
	RMF_AudioCapture_Settings settings;
	memset (&settings, 0, sizeof(RMF_AudioCapture_Settings));
	RMF_AudioCapture_GetDefaultSettings(&settings);
	
	settings.cbBufferReady = &q_mgr::data_callback;
	settings.cbBufferReadyParm = (void *)this;
	settings.fifoSize = m_audio_properties.fifo_size; 
	settings.threshold = m_audio_properties.threshold;
	settings.delayCompensation_ms = m_audio_properties.delay_compensation_ms;
	settings.format = m_audio_properties.format;
	settings.samplingFreq = m_audio_properties.sampling_frequency;

	log_settings(settings);
	
	int ret = RMF_AudioCapture_Start(m_device_handle, &settings);
	INFO("start() result is 0x%x\n", ret);
	m_started = true;

	std::unique_lock<std::mutex> wlock(m_data_monitor_mutex);
	m_stop_data_monitor = false;
	m_data_monitor_thread = std::thread(&q_mgr::data_monitor, this);
	return ret;
}

int q_mgr::stop()
{
	if(false == m_started)
	{
		WARN("Looks like device is already stopped.\n");
		return 0;
	}
	
	{
		std::unique_lock<std::mutex> wlock(m_data_monitor_mutex);
		m_stop_data_monitor = true;
	}
	m_data_monitor_cv.notify_all();
	m_data_monitor_thread.join();

	int ret = RMF_AudioCapture_Stop(m_device_handle);
	INFO("stop() result is 0x%x\n", ret);
	m_started = false;
	return ret;
}

void audio_capture_client::release_buffer(audio_buffer *ptr)
{ 
	unref_audio_buffer(ptr);
} 

int audio_capture_client::data_callback(audio_buffer *buf)
{ 
	release_buffer(buf);
	return 0;
} 
audio_capture_client::audio_capture_client(q_mgr * manager): m_priority(0), m_manager(manager)
{ 
	pthread_mutexattr_t mutex_attribute;
	REPORT_IF_UNEQUAL(0, pthread_mutexattr_init(&mutex_attribute));
	REPORT_IF_UNEQUAL(0, pthread_mutexattr_settype(&mutex_attribute, PTHREAD_MUTEX_ERRORCHECK));
	REPORT_IF_UNEQUAL(0, pthread_mutex_init(&m_mutex, &mutex_attribute));
} 
audio_capture_client::~audio_capture_client()
{	
	REPORT_IF_UNEQUAL(0, pthread_mutex_destroy(&m_mutex));
} 

void audio_capture_client::set_manager(q_mgr *mgr)
{
	m_manager = mgr;
}

int audio_capture_client::set_audio_properties(audio_properties_t &properties)
{
	return m_manager->set_audio_properties(properties);
}

void audio_capture_client::get_audio_properties(audio_properties_t &properties)
{
	m_manager->get_audio_properties(properties);
}

void audio_capture_client::get_default_audio_properties(audio_properties_t &properties)
{
	m_manager->get_default_audio_properties(properties);
}

int audio_capture_client::start()
{
	return m_manager->register_client(this);
}

int audio_capture_client::stop()
{
	return m_manager->unregister_client(this);
}

void audio_capture_client::lock()
{
    REPORT_IF_UNEQUAL(0, pthread_mutex_lock(&m_mutex));
}

void audio_capture_client::unlock()
{
    REPORT_IF_UNEQUAL(0, pthread_mutex_unlock(&m_mutex));
}