Bug:BcastListener: Add SpinLock for races around stop()
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
#include <algorithm>
|
||||
#include <iostream>
|
||||
#include <functional>
|
||||
#include <opts.h>
|
||||
#include <componentThread.h>
|
||||
#include "broadcastListener.h"
|
||||
@@ -54,7 +55,6 @@ void BroadcastListener::broadcastMsgInd(
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
// Use placement new to construct BroadcastMessage in the buffer
|
||||
BroadcastMessage* msg = new (bcastMsgRecvBuffer) BroadcastMessage;
|
||||
|
||||
@@ -94,6 +94,8 @@ void BroadcastListener::broadcastMsgInd(
|
||||
reinterpret_cast<const char*>(msg->broadcast_code));
|
||||
|
||||
// Early return if device already exists
|
||||
smo::SpinLock::Guard lock(isListeningLock);
|
||||
|
||||
if (deviceExists(broadcastCode))
|
||||
{
|
||||
// Device already exists, just log the update
|
||||
@@ -103,21 +105,23 @@ void BroadcastListener::broadcastMsgInd(
|
||||
<< ": Received broadcast from known device: "
|
||||
<< broadcastCode << " at " << senderIP << "\n";
|
||||
}
|
||||
return;
|
||||
}
|
||||
else
|
||||
{
|
||||
// Create new DiscoveredDevice using conversion constructor
|
||||
auto device = std::make_shared<DiscoveredDevice>(*msg, senderIP);
|
||||
discoveredDevices.push_back(device);
|
||||
// Output device information using stringify
|
||||
std::cout << __func__ << ": Discovered new Livox device: "
|
||||
<< device->stringify() << "\n";
|
||||
}
|
||||
|
||||
// Create new DiscoveredDevice using conversion constructor
|
||||
auto device = std::make_shared<DiscoveredDevice>(*msg, senderIP);
|
||||
discoveredDevices.push_back(device);
|
||||
|
||||
// Output device information using stringify
|
||||
std::cout << __func__ << ": Discovered new Livox device: "
|
||||
<< device->stringify() << "\n";
|
||||
startReceive();
|
||||
}
|
||||
|
||||
void BroadcastListener::start(void)
|
||||
{
|
||||
if (isListening.load()) { return; }
|
||||
if (isListening) { return; }
|
||||
|
||||
try
|
||||
{
|
||||
@@ -132,7 +136,7 @@ void BroadcastListener::start(void)
|
||||
socket.open(boost::asio::ip::udp::v4());
|
||||
socket.bind(listeningEndpoint);
|
||||
|
||||
isListening.store(true);
|
||||
isListening = true;
|
||||
// Start the first async receive operation
|
||||
startReceive();
|
||||
std::cout << __func__ << ": BroadcastListener started on port "
|
||||
@@ -140,7 +144,7 @@ void BroadcastListener::start(void)
|
||||
}
|
||||
catch (const boost::system::system_error& e)
|
||||
{
|
||||
isListening.store(false);
|
||||
isListening = false;
|
||||
std::cerr << __func__ << ": Failed to start BroadcastListener: "
|
||||
<< e.what() << std::endl;
|
||||
throw;
|
||||
@@ -149,27 +153,25 @@ void BroadcastListener::start(void)
|
||||
|
||||
void BroadcastListener::startReceive(void)
|
||||
{
|
||||
if (!isListening.load()) { return; }
|
||||
if (!isListening) { return; }
|
||||
|
||||
socket.async_receive_from(
|
||||
boost::asio::buffer(bcastMsgRecvBuffer, sizeof(bcastMsgRecvBuffer)),
|
||||
senderEndpoint,
|
||||
[this](const boost::system::error_code& ec, std::size_t bytes_received)
|
||||
{
|
||||
broadcastMsgInd(ec, bytes_received);
|
||||
|
||||
// Continue listening for the next packet
|
||||
if (isListening.load())
|
||||
{ startReceive(); }
|
||||
}
|
||||
std::bind(
|
||||
&BroadcastListener::broadcastMsgInd, this,
|
||||
std::placeholders::_1, std::placeholders::_2)
|
||||
);
|
||||
}
|
||||
|
||||
void BroadcastListener::stop(void)
|
||||
{
|
||||
if (!isListening.load()) { return; }
|
||||
{
|
||||
smo::SpinLock::Guard lock(isListeningLock);
|
||||
if (!isListening) { return; }
|
||||
|
||||
isListening.store(false);
|
||||
isListening = false;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
|
||||
@@ -8,6 +8,7 @@
|
||||
#include <atomic>
|
||||
#include <boost/asio/ip/udp.hpp>
|
||||
#include <user/senseApiDesc.h>
|
||||
#include <spinLock.h>
|
||||
#include "device.h"
|
||||
|
||||
namespace livoxProto1 {
|
||||
@@ -67,7 +68,8 @@ private:
|
||||
|
||||
boost::asio::ip::udp::socket socket;
|
||||
boost::asio::ip::udp::endpoint listeningEndpoint, senderEndpoint;
|
||||
std::atomic<bool> isListening;
|
||||
smo::SpinLock isListeningLock;
|
||||
bool isListening;
|
||||
|
||||
uint8_t bcastMsgRecvBuffer[UDP_BCAST_MSG_BUFFER_NBYTES];
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user