I started a voice chat server project, using boost::asio and SMFL audio library. I think my mutex logic is not correct, because I get this error:
Unhandled exception at 0x00B7871D in TCPAudioReceiver.exe: 0xC0000005: Access violation reading location 0x00000000.
The debugger stops at the if(mServer.samplesAvailable()) line in the main and I also notice that the if(mSamplesQueue.front().size() > 256) return true; line in the server reads a size of {???}. I thought that locking the server's mutex would be enough to protect against this kind of problem, but I think I'm still one or two concepts away from the good design.
I used recursive_mutex for no particular reason a this point, it was more an attempt of getting farther in my troubleshooting.
How would you suggest I solve this problem?
main
#include "TcpAsyncServer.h"
#include "TcpAsyncSession.h"
#include "AudioStream.h"
#include <thread>
int main()
{
io_service ioService;
deque<vector<Int16>> mSamplesQueue;
SoundBuffer mSoundBuffer;
AudioStream mAudioStream;
cout << "Starting server..." << endl;
Server mServer(ioService, PORT);
thread serviceThread([&](){ioService.run();});
cout << "Starting reception loop..." << endl;
for(;;) {
{
lock_guard<recursive_mutex> lock(mServer.mServerMutex);
// Look for new samples
if(mServer.samplesAvailable()) { // <-- debugger stops here
cout << "Samples available..." << endl;
vector<Int16> wSamples;
mServer.getNextSamples(wSamples);
mSamplesQueue.push_back(wSamples);
}
}
// Loading and playing samples
if((mAudioStream.getStatus() != AudioStream::Playing) && !mSamplesQueue.empty()) {
cout << "Loading and playing audio stream..." << endl;
if(mSoundBuffer.loadFromSamples(
reinterpret_cast<Int16*>(mSamplesQueue.front().data()),
mSamplesQueue.front().size(), 2, 48000)
)
{
cout << "SoundBuffer loaded successfully..." << endl;
mAudioStream.load(mSoundBuffer);
mAudioStream.play();
mSamplesQueue.pop_front();
}
else cout << "SoundBuffer failed to load..." << endl;
}
// Give it some room to play the sound
while (mAudioStream.getStatus() == AudioStream::Playing) {
sleep(milliseconds(50));
}
}
serviceThread.join();
}
Server header
#include <mutex>
#include <deque>
#include <vector>
#include <iostream>
#include <boost/asio.hpp>
#include <SFML/Audio.hpp>
using namespace sf;
using namespace boost::asio;
using namespace boost::asio::ip;
using namespace std;
#define PORT 2112
#define SAMPLE_BUFFER 512
class Server
{
public:
// Ctor
Server(io_service & iService, short iPort);
// Methods
bool samplesAvailable();
void getNextSamples(vector<Int16> oSamples);
void pushSamples(vector<Int16> iSamples);
// Variables
recursive_mutex mServerMutex;
private:
// Methods
void doAccept();
// Variables
tcp::acceptor mAcceptor;
tcp::socket mSocket;
deque<vector<Int16>> mSamplesQueue;
};
Server class
#include "TcpAsyncServer.h"
#include "TcpAsyncSession.h"
Server::Server(io_service & service, short port)
: mAcceptor(service, tcp::endpoint(tcp::v4(), port)),
mSocket(service) {
doAccept();
}
void Server::doAccept()
{
mAcceptor.async_accept(mSocket,
[this]
(boost::system::error_code error) {
if(!error) make_shared<Session>(move(mSocket),this)->start();
doAccept();
}
);
}
bool Server::samplesAvailable() {
lock_guard<recursive_mutex> lock(mServerMutex);
if(mSamplesQueue.front().size() > 256) return true; // <-- mSamplesQueue.front() has undefined size
return false;
}
void Server::getNextSamples(vector<Int16> oSamples) {
lock_guard<recursive_mutex> lock(mServerMutex);
oSamples = mSamplesQueue.front();
mSamplesQueue.pop_front();
}
void Server::pushSamples(vector<Int16> iSamples) {
lock_guard<recursive_mutex> lock(mServerMutex);
mSamplesQueue.push_back(iSamples);
}
Session header
#include <iostream>
#include <mutex>
#include <deque>
#include <vector>
#include <boost/asio.hpp>
#include <SFML/Audio.hpp>
using namespace std;
using namespace sf;
using namespace boost::asio;
using namespace boost::asio::ip;
#define BUFFER_SIZE 1024
class Server;
class Session : public enable_shared_from_this<Session>
{
public:
// Ctor
Session(tcp::socket & iSocket, Server* iServerPtr);
// Methods
void start();
private:
// Methods
void doRead();
// Variables
tcp::socket mSocket;
char mData[BUFFER_SIZE];
Server* mServerPtr;
deque<vector<Int16>> mSampleBufferQueue;
};
Session class
#include "TcpAsyncSession.h"
#include "TcpAsyncServer.h"
Session::Session(tcp::socket & iSocket, Server* iServerPtr)
: mSocket(move(iSocket)),
mServerPtr(iServerPtr)
{}
void Session::start() {
doRead();
}
void Session::doRead() {
shared_ptr<Session> self(shared_from_this());
mSocket.async_read_some(buffer(mData,BUFFER_SIZE),
[this,self]
(boost::system::error_code error, size_t iBytesReceived) {
if(!error) {
cout << "Receiving " << iBytesReceived << " bytes..." << endl;
vector<Int16> wSamples;
for(unsigned int i = 0; i < iBytesReceived; i+=2) {
wSamples.push_back(static_cast<Int16>( mData[i]<<8 | mData[i] ));
}
{
lock_guard<recursive_mutex> lock(mServerPtr->mServerMutex);
mServerPtr->pushSamples(move(wSamples));
}
doRead();
}
}
);
}
