接收音频块的异步 TCP 服务器收到访问冲突错误答案

【问题标题】：Async TCP server to receive audio chunks get Access Violation errors接收音频块的异步 TCP 服务器收到访问冲突错误
【发布时间】：2018-08-07 14:03:54
【问题描述】：

我使用 boost::asio 和 SMFL 音频库启动了一个语音聊天服务器项目。我认为我的互斥逻辑不正确，因为我收到此错误：

Unhandled exception at 0x00B7871D in TCPAudioReceiver.exe: 0xC0000005: Access violation reading location 0x00000000.

调试器在 main 中的 if(mServer.samplesAvailable()) 行停止，我还注意到服务器中的 if(mSamplesQueue.front().size() > 256) return true; 行读取的大小为 {???}。我认为锁定服务器的互斥体足以防止此类问题，但我认为我离好的设计还有一两个概念。

我使用 recursive_mutex 没有特别的原因，这更多是为了在我的故障排除中走得更远。

你会建议我如何解决这个问题？

主要

#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();
}

服务器头

#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;
};

服务器类

#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);
}

会话标题

#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;
};

会话类

#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();
      }
    }
  );
}

【问题讨论】：

投票结束，因为代码太多了。无论如何，一个建议：记录哪些数据是共享的，哪些互斥锁必须被锁定才能访问这些数据。然后，审核您的代码，确保此规则在任何地方都没有被破坏。
@UlrichEckhardt 不同的标准:)

标签： c++ boost boost-asio mutex sfml

【解决方案1】：

哇，好长啊。

你最明显的错误在这里：

if (mSamplesQueue.front().size() > 256)
    return true; // <-- mSamplesQueue.front() has undefined size

您应该检查mSamplesQueue.empty() 不正确，否则使用front() 是UB。

代替

bool Server::samplesAvailable() {
    std::lock_guard<std::recursive_mutex> lock(mServerMutex);

    if (mSamplesQueue.front().size() > 256)
        return true;
    return false;
}

你可以简单地写

bool Server::samplesAvailable() {
    std::lock_guard<std::recursive_mutex> lock(mServerMutex);
    return mSamplesQueue.size() && mSamplesQueue.front().size()>256;
}

审核：

里面有 UB
```
wSamples.push_back(static_cast<Int16>(mData[i] << 8 | mData[i]));
```
您需要括号，并且在许多编译器上char 已签名，这导致UB 左移。
永远不要在头文件中使用using-directives。 特别是不是using namespace std; (Why is "using namespace std" considered bad practice?)
有很多不必要的缓冲区复制，为什么不只是在具有适当字节顺序的平台上重新解释转换 char[] 缓冲区，和/或为什么不在 pushSamples() 中使用 std::move：
```
mSamplesQueue.push_back(std::move(iSamples));
```

直接使用类成员是一种代码味道 (Law of Demeter)。尤其是，这肯定表明您在编写时混合了抽象级别：

{
    std::lock_guard<std::recursive_mutex> lock(mServerPtr->mServerMutex);
    mServerPtr->pushSamples(std::move(wSamples));
}

尤其是你已经有了

void Server::pushSamples(std::vector<sf::Int16> iSamples) {
    std::lock_guard<std::recursive_mutex> lock(mServerMutex);
    mSamplesQueue.push_back(std::move(iSamples));
}

这意味着此时您需要一个递归互斥锁，否则您会遇到死锁。

Session::mSampleBufferQueue 未使用
getNextSamples 的签名错误。您将永远看不到任何效果，因为参数是按值传递的。要么声明它：
```
void getNextSamples(std::vector<sf::Int16>& oSamples);
```
或
```
std::vector<sf::Int16> getNextSamples();
```
考虑创建类似samplesAvailable() const 的方法。为此，您需要标记互斥锁mutable (Always declare std::mutex as mutable in C++11?)

概念问题

从概念上讲，接受并发客户端存在问题。如果确实发生这种情况，您最终将随机交错播放流。

修正版

main.cpp

//#include "AudioStream.h"
#include "TcpAsyncServer.h"
#include "TcpAsyncSession.h"
#include <thread>

struct AudioStream {
    enum Status { Playing, Buffering };
    Status getStatus() const { return Buffering; }
    void load(sf::SoundBuffer const& buf) {
        // 
        std::cout << __PRETTY_FUNCTION__ 
            << " rate=" << buf.getSampleRate() 
            << " channels=" << buf.getChannelCount() 
            << " duration=" << buf.getDuration().asSeconds() << "s "
            << " samples=" << buf.getSampleCount()
            << "\n";
    }
    void play() {}
};

int main() {
    boost::asio::io_service ioService;
    std::deque<std::vector<sf::Int16> > mSamplesQueue;
    sf::SoundBuffer mSoundBuffer;
    AudioStream mAudioStream;

    std::cout << "Starting server..." << std::endl;
    Server mServer(ioService, PORT); // start async accept loop
    std::thread serviceThread([&]() { ioService.run(); });

    std::cout << "Starting reception loop..." << std::endl;
    for (;;) {
        {
            std::lock_guard<std::recursive_mutex> lock(mServer.mServerMutex);

            // Look for new samples
            if (mServer.samplesAvailable()) {
                std::cout << "Samples available..." << std::endl;
                std::vector<sf::Int16> wSamples;
                mServer.getNextSamples(wSamples);
                mSamplesQueue.push_back(wSamples);
            }
        }

        // Loading and playing samples
        if ((mAudioStream.getStatus() != AudioStream::Playing) && !mSamplesQueue.empty()) {
            std::cout << "Loading and playing audio stream..." << std::endl;
            if (mSoundBuffer.loadFromSamples(reinterpret_cast<sf::Int16 *>(mSamplesQueue.front().data()),
                                             mSamplesQueue.front().size(), 2, 48000)) {
                std::cout << "SoundBuffer loaded successfully..." << std::endl;
                mAudioStream.load(mSoundBuffer);
                mAudioStream.play();
                mSamplesQueue.pop_front();
            } else
                std::cout << "SoundBuffer failed to load..." << std::endl;
        }

        // Give it some room to play the sound
        while (mAudioStream.getStatus() == AudioStream::Playing) {
            sleep(sf::milliseconds(50));
        }
    }
    serviceThread.join();
}

TcpAsyncServer.h

#include <SFML/Audio.hpp>
#include <boost/asio.hpp>
#include <deque>
#include <iostream>
#include <mutex>
#include <vector>

#define PORT 2113
#define SAMPLE_BUFFER 5120000

class Server {
    using tcp = boost::asio::ip::tcp;

  public:
    // Ctor
    Server(boost::asio::io_service &iService, short iPort);

    // Methods
    bool samplesAvailable() const;
    void getNextSamples(std::vector<sf::Int16>& oSamples);
    void pushSamples(std::vector<sf::Int16> iSamples);

    // Variables
    mutable std::recursive_mutex mServerMutex;

  private:
    // Methods
    void doAccept();

    // Variables
    tcp::acceptor mAcceptor;
    tcp::socket mSocket;
    std::deque<std::vector<sf::Int16> > mSamplesQueue;
};

TcpAsyncSession.h

#include <boost/asio.hpp>

#define BUFFER_SIZE 10240000

class Server;

class Session : public std::enable_shared_from_this<Session> {
    using tcp = boost::asio::ip::tcp;

  public:
    // Ctor
    Session(tcp::socket &&iSocket, Server *iServerPtr);

    // Methods
    void start();

  private:
    // Methods
    void doRead();

    // Variables
    tcp::socket mSocket;
    uint8_t mData[BUFFER_SIZE];
    Server *mServerPtr;
};

TcpAsyncServer.cpp

#include "TcpAsyncServer.h"
#include "TcpAsyncSession.h"

Server::Server(boost::asio::io_service &service, short port) : mAcceptor(service, tcp::endpoint(tcp::v4(), port)), mSocket(service) {
    mAcceptor.set_option(tcp::acceptor::reuse_address());
    doAccept();
}

void Server::doAccept() {
    mAcceptor.async_accept(mSocket, [this](boost::system::error_code error) {
        if (!error)
            std::make_shared<Session>(std::move(mSocket), this)->start();
        doAccept();
    });
}

bool Server::samplesAvailable() const {
    std::lock_guard<std::recursive_mutex> lock(mServerMutex);
    return mSamplesQueue.size() && mSamplesQueue.front().size()>256;
}

void Server::getNextSamples(std::vector<sf::Int16>& oSamples) {
    std::lock_guard<std::recursive_mutex> lock(mServerMutex);
    oSamples = std::move(mSamplesQueue.front());
    mSamplesQueue.pop_front();
}

void Server::pushSamples(std::vector<sf::Int16> iSamples) {
    std::lock_guard<std::recursive_mutex> lock(mServerMutex);
    mSamplesQueue.push_back(std::move(iSamples));
}

TcpAsyncSession.cpp

#include "TcpAsyncServer.h"
#include "TcpAsyncSession.h"

//#include <SFML/Audio.hpp>
#include <iostream>

Session::Session(tcp::socket &&iSocket, Server *iServerPtr) : mSocket(std::move(iSocket)), mServerPtr(iServerPtr) {}

void Session::start() { doRead(); }

void Session::doRead() {

    std::shared_ptr<Session> self(shared_from_this());

    mSocket.async_read_some(
            boost::asio::buffer(mData, BUFFER_SIZE),
            [this, self](boost::system::error_code error, size_t iBytesReceived) {
                if (error)
                    return;
                std::cout << "Receiving " << iBytesReceived << " bytes..." << std::endl;
                std::vector<sf::Int16> wSamples;

                for (unsigned int i = 0; i < iBytesReceived; i += 2) {
                    wSamples.push_back(static_cast<sf::Int16>((mData[i] << 8) | mData[i]));
                }

                mServerPtr->pushSamples(std::move(wSamples));

                doRead();
        });
}

演示运行

给它一个 138M 的 mp3：

Starting server...
Starting reception loop...
Receiving 8192 bytes...
Samples available...
Loading and playing audio stream...
SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=0.042666s  samples=4096
Receiving 562829 bytes...
Samples available...
Receiving 745525 bytes...
Loading and playing audio stream...
An internal OpenAL call failed in SoundBuffer.cpp(265).
Expression:
   alBufferData(m_buffer, format, &m_samples[0], size, sampleRate)
Error description:
   AL_INVALID_VALUE
   A numeric argument is out of range.

SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=2.93141s  samples=281415
Samples available...
Loading and playing audio stream...
Receiving 2815769 bytes...
An internal OpenAL call failed in SoundBuffer.cpp(265).
Expression:
   alBufferData(m_buffer, format, &m_samples[0], size, sampleRate)
Error description:
   AL_INVALID_VALUE
   A numeric argument is out of range.

SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=3.88295s  samples=372763
Samples available...
Receiving 4978211 bytes...
Loading and playing audio stream...
An internal OpenAL call failed in SoundBuffer.cpp(265).
Expression:
   alBufferData(m_buffer, format, &m_samples[0], size, sampleRate)
Error description:
   AL_INVALID_VALUE
   A numeric argument is out of range.

SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=14.6655s  samples=1407885
Samples available...
Receiving 5632954 bytes...
Loading and playing audio stream...
SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=25.9282s  samples=2489106
Samples available...
Receiving 5893470 bytes...
Loading and playing audio stream...
An internal OpenAL call failed in SoundBuffer.cpp(265).
Expression:
   alBufferData(m_buffer, format, &m_samples[0], size, sampleRate)
Error description:
   AL_INVALID_VALUE
   A numeric argument is out of range.

SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=29.3383s  samples=2816477
Samples available...
Receiving 5895401 bytes...
Loading and playing audio stream...
An internal OpenAL call failed in SoundBuffer.cpp(265).
Expression:
   alBufferData(m_buffer, format, &m_samples[0], size, sampleRate)
Error description:
   AL_INVALID_VALUE
   A numeric argument is out of range.

SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=30.6952s  samples=2946735
Samples available...
Receiving 5894091 bytes...
Loading and playing audio stream...
An internal OpenAL call failed in SoundBuffer.cpp(265).
Expression:
   alBufferData(m_buffer, format, &m_samples[0], size, sampleRate)
Error description:
   AL_INVALID_VALUE
   A numeric argument is out of range.

SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=30.7052s  samples=2947701
Samples available...
Receiving 5894197 bytes...
Loading and playing audio stream...
SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=30.6984s  samples=2947046
Samples available...
Receiving 5894303 bytes...
Loading and playing audio stream...
An internal OpenAL call failed in SoundBuffer.cpp(265).
Expression:
   alBufferData(m_buffer, format, &m_samples[0], size, sampleRate)
Error description:
   AL_INVALID_VALUE
   A numeric argument is out of range.

SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=30.6989s  samples=2947099
Samples available...
Receiving 5894144 bytes...
Loading and playing audio stream...
SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=30.6995s  samples=2947152
Samples available...
Receiving 5896192 bytes...
Loading and playing audio stream...
SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=30.6987s  samples=2947072
Samples available...
Receiving 5961675 bytes...
Loading and playing audio stream...
SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=30.7093s  samples=2948096
Samples available...
Receiving 5961728 bytes...
Loading and playing audio stream...
SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=31.0504s  samples=2980838
Samples available...
Receiving 5960615 bytes...
Loading and playing audio stream...
SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=31.0507s  samples=2980864
Samples available...
Receiving 5960793 bytes...
Loading and playing audio stream...
SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=31.0449s  samples=2980308
Samples available...
Receiving 5960668 bytes...
Loading and playing audio stream...
An internal OpenAL call failed in SoundBuffer.cpp(265).
Expression:
   alBufferData(m_buffer, format, &m_samples[0], size, sampleRate)
Error description:
   AL_INVALID_VALUE
   A numeric argument is out of range.

SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=31.0458s  samples=2980397
Samples available...
Receiving 5960740 bytes...
Loading and playing audio stream...
SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=31.0451s  samples=2980334
Samples available...
Receiving 5960668 bytes...
Loading and playing audio stream...
SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=31.0455s  samples=2980370
Samples available...
Receiving 5960740 bytes...
Loading and playing audio stream...
SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=31.0451s  samples=2980334
Samples available...
Receiving 5960668 bytes...
Loading and playing audio stream...
SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=31.0455s  samples=2980370
Samples available...
Receiving 5960740 bytes...
Loading and playing audio stream...
SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=31.0451s  samples=2980334
Samples available...
Receiving 5960668 bytes...
Loading and playing audio stream...
SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=31.0455s  samples=2980370
Samples available...
Receiving 5960740 bytes...
Loading and playing audio stream...
SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=31.0451s  samples=2980334
Samples available...
Receiving 5960668 bytes...
Loading and playing audio stream...
SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=31.0455s  samples=2980370
Samples available...
Receiving 5960740 bytes...
Loading and playing audio stream...
SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=31.0451s  samples=2980334
Samples available...
Receiving 4770135 bytes...
Loading and playing audio stream...
SoundBuffer loaded successfully...
void AudioStream::load(const sf::SoundBuffer &) rate=48000 channels=2 duration=31.0455s  samples=2980370
Samples available...
Loading and playing audio stream...
SoundBuffer loaded successfully...

除了我的模拟 AudioStream 没有正确初始化音频库之外，我觉得这很好。

请注意，我大幅增加了缓冲区大小，因此日志不会太大。

【讨论】：

Added 一个关于 const 正确性的评论提示。