Skip to content

Refactor worker thread capture procedures as internal functions. #1971

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: dev
Choose a base branch
from
Open

Refactor worker thread capture procedures as internal functions. #1971

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 0 additions & 9 deletions Pcap++/header/PcapLiveDevice.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,12 +103,8 @@ namespace pcpp
// Should be set to true by the Callee for the Caller
std::atomic<bool> m_CaptureThreadStarted;

OnPacketArrivesCallback m_cbOnPacketArrives;
void* m_cbOnPacketArrivesUserCookie;
OnPacketArrivesStopBlocking m_cbOnPacketArrivesBlockingMode;
void* m_cbOnPacketArrivesBlockingModeUserCookie;
RawPacketVector* m_CapturedPackets;
bool m_CaptureCallbackMode;
LinkLayerType m_LinkType;
bool m_UsePoll;

Expand All @@ -124,11 +120,6 @@ namespace pcpp
void setDeviceMacAddress();
void setDefaultGateway();

// threads
void captureThreadMain();

static void onPacketArrives(uint8_t* user, const struct pcap_pkthdr* pkthdr, const uint8_t* packet);
static void onPacketArrivesNoCallback(uint8_t* user, const struct pcap_pkthdr* pkthdr, const uint8_t* packet);
static void onPacketArrivesBlockingMode(uint8_t* user, const struct pcap_pkthdr* pkthdr, const uint8_t* packet);

public:
Expand Down
226 changes: 150 additions & 76 deletions Pcap++/src/PcapLiveDevice.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -272,6 +272,142 @@ namespace pcpp
}
PCPP_LOG_DEBUG("Ended periodic statistics update procedure");
}

struct CaptureContext
{
PcapLiveDevice* device = nullptr;
OnPacketArrivesCallback callback;
void* userCookie = nullptr;
};

struct AccumulatorCaptureContext
{
PcapLiveDevice* device = nullptr;
RawPacketVector* capturedPackets = nullptr;
};

// A noop function to be used when no callback is set
void onPacketArrivesNoop(uint8_t* user, const pcap_pkthdr* pkthdr, const uint8_t* packet)
{}

// @brief Wraps the raw packet data into a RawPacket instance and calls the user callback
// @param user A pointer to a CaptureContext instance
// @param pkthdr A pointer to the pcap_pkthdr struct
// @param packet A pointer to the raw packet data
void onPacketArrivesCallback(uint8_t* user, const pcap_pkthdr* pkthdr, const uint8_t* packet)
{
auto* context = reinterpret_cast<CaptureContext*>(user);
if (context == nullptr || context->device == nullptr || context->callback == nullptr)
{
PCPP_LOG_ERROR("Unable to extract PcapLiveDevice instance or callback");
return;
}

try
{
RawPacket rawPacket(packet, pkthdr->caplen, pkthdr->ts, false, context->device->getLinkType());
context->callback(&rawPacket, context->device, context->userCookie);
}
catch (const std::exception& ex)
{
PCPP_LOG_ERROR("Exception occurred while invoking packet arrival callback: " << ex.what());
}
catch (...)
{
PCPP_LOG_ERROR("Unknown exception occurred while invoking packet arrival callback");
}
}

/// @brief Wraps the raw packet data into a RawPacket instance and adds it to the captured packets vector
/// @param user A pointer to an AccumulatorCaptureContext instance
/// @param pkthdr A pointer to the pcap_pkthdr struct
/// @param packet A pointer to the raw packet data
void onPacketArrivesAccumulator(uint8_t* user, const pcap_pkthdr* pkthdr, const uint8_t* packet)
{
auto* context = reinterpret_cast<AccumulatorCaptureContext*>(user);
if (context == nullptr || context->device == nullptr || context->capturedPackets == nullptr)
{
PCPP_LOG_ERROR("Unable to extract PcapLiveDevice instance or captured packets vector");
return;
}

try
{
uint8_t* packetData = new uint8_t[pkthdr->caplen];
std::memcpy(packetData, packet, pkthdr->caplen);
auto rawPacket = std::make_unique<RawPacket>(packetData, pkthdr->caplen, pkthdr->ts, true,
context->device->getLinkType());
context->capturedPackets->pushBack(std::move(rawPacket));
}
catch (const std::exception& ex)
{
PCPP_LOG_ERROR("Exception occurred while invoking packet arrival callback: " << ex.what());
}
catch (...)
{
PCPP_LOG_ERROR("Unknown exception occurred while invoking packet arrival callback");
}
}

/// @brief A procedure that dispatches packets to a user-defined callback function whenever a packet arrives.
///
/// The procedure runs in a loop until the `stopFlag` is set to true.
/// The procedure will set the `hasStarted` flag to true at the start of the procedure.
///
/// @param stopFlag A reference to an atomic boolean that indicates when to stop capturing packets
/// @param hasStarted A reference to an atomic boolean that indicates when the capture thread has started
/// @param pcapDescriptor A reference to the pcap handle on which to call pcap_dispatch
/// @param context A CaptureContext instance that holds the device, callback and user cookie.
void captureThreadMain(std::atomic_bool& stopFlag, std::atomic_bool& hasStarted,
internal::PcapHandle const& pcapDescriptor, CaptureContext context)
{
PCPP_LOG_DEBUG("Started capture thread for device '" << context.device->getName() << "'");
hasStarted.store(true);

// If the callback is null, we use a no-op handler to avoid unnecessary overhead
// Statistics only capture still requires pcap_dispatch to be called, but we don't need to process
// packets.
pcap_handler callbackHandler = context.callback ? onPacketArrivesCallback : onPacketArrivesNoop;

while (!stopFlag.load())
{
if (pcap_dispatch(pcapDescriptor.get(), -1, callbackHandler, reinterpret_cast<uint8_t*>(&context)) ==
-1)
{
PCPP_LOG_ERROR("pcap_dispatch returned an error: " << pcapDescriptor.getLastError());
stopFlag.store(true);
}
}

PCPP_LOG_DEBUG("Ended capture thread for device '" << context.device->getName() << "'");
}

/// @brief A procedure that accumulates captured packets into a vector.
///
/// The procedure runs in a loop until the `stopFlag` is set to true.
/// The procedure will set the `hasStarted` flag to true at the start of the procedure.
///
/// @param stopFlag A reference to an atomic boolean that indicates when to stop capturing packets
/// @param hasStarted A reference to an atomic boolean that indicates when the capture thread has started
/// @param pcapDescriptor A reference to the pcap handle on which to call pcap_dispatch
/// @param context An AccumulatorCaptureContext instance that holds the device and the captured packets vector.
void captureThreadMainAccumulator(std::atomic_bool& stopFlag, std::atomic_bool& hasStarted,
internal::PcapHandle const& pcapDescriptor, AccumulatorCaptureContext context)
{
PCPP_LOG_DEBUG("Started capture thread for device '" << context.device->getName() << "'");
hasStarted.store(true);
while (!stopFlag.load())
{
if (pcap_dispatch(pcapDescriptor.get(), 100, onPacketArrivesAccumulator,
reinterpret_cast<uint8_t*>(&context)) == -1)
{
PCPP_LOG_ERROR("pcap_dispatch returned an error: " << pcapDescriptor.getLastError());
stopFlag.store(true);
}
}

PCPP_LOG_DEBUG("Ended capture thread for device '" << context.device->getName() << "'");
}
} // namespace

PcapLiveDevice::PcapLiveDevice(DeviceInterfaceDetails interfaceDetails, bool calculateMTU, bool calculateMacAddress,
Expand Down Expand Up @@ -309,50 +445,15 @@ namespace pcpp
m_CaptureThreadStarted = false;
m_StopThread = false;
m_CaptureThread = {};
m_cbOnPacketArrives = nullptr;
m_cbOnPacketArrivesBlockingMode = nullptr;
m_cbOnPacketArrivesBlockingModeUserCookie = nullptr;
m_cbOnPacketArrivesUserCookie = nullptr;
m_CaptureCallbackMode = true;
m_CapturedPackets = nullptr;
if (calculateMacAddress)
{
setDeviceMacAddress();
PCPP_LOG_DEBUG(" MAC addr: " << m_MacAddress);
}
}

void PcapLiveDevice::onPacketArrives(uint8_t* user, const struct pcap_pkthdr* pkthdr, const uint8_t* packet)
{
PcapLiveDevice* pThis = reinterpret_cast<PcapLiveDevice*>(user);
if (pThis == nullptr)
{
PCPP_LOG_ERROR("Unable to extract PcapLiveDevice instance");
return;
}

RawPacket rawPacket(packet, pkthdr->caplen, pkthdr->ts, false, pThis->getLinkType());

if (pThis->m_cbOnPacketArrives != nullptr)
pThis->m_cbOnPacketArrives(&rawPacket, pThis, pThis->m_cbOnPacketArrivesUserCookie);
}

void PcapLiveDevice::onPacketArrivesNoCallback(uint8_t* user, const struct pcap_pkthdr* pkthdr,
const uint8_t* packet)
{
PcapLiveDevice* pThis = reinterpret_cast<PcapLiveDevice*>(user);
if (pThis == nullptr)
{
PCPP_LOG_ERROR("Unable to extract PcapLiveDevice instance");
return;
}

uint8_t* packetData = new uint8_t[pkthdr->caplen];
memcpy(packetData, packet, pkthdr->caplen);
RawPacket* rawPacketPtr = new RawPacket(packetData, pkthdr->caplen, pkthdr->ts, true, pThis->getLinkType());
pThis->m_CapturedPackets->pushBack(rawPacketPtr);
}

void PcapLiveDevice::onPacketArrivesBlockingMode(uint8_t* user, const struct pcap_pkthdr* pkthdr,
const uint8_t* packet)
{
Expand All @@ -371,37 +472,6 @@ namespace pcpp
pThis->m_StopThread = true;
}

void PcapLiveDevice::captureThreadMain()
{
PCPP_LOG_DEBUG("Started capture thread for device '" << m_InterfaceDetails.name << "'");
m_CaptureThreadStarted = true;

if (m_CaptureCallbackMode)
{
while (!m_StopThread)
{
if (pcap_dispatch(m_PcapDescriptor.get(), -1, onPacketArrives, reinterpret_cast<uint8_t*>(this)) == -1)
{
PCPP_LOG_ERROR("pcap_dispatch returned an error: " << m_PcapDescriptor.getLastError());
m_StopThread = true;
}
}
}
else
{
while (!m_StopThread)
{
if (pcap_dispatch(m_PcapDescriptor.get(), 100, onPacketArrivesNoCallback,
reinterpret_cast<uint8_t*>(this)) == -1)
{
PCPP_LOG_ERROR("pcap_dispatch returned an error: " << m_PcapDescriptor.getLastError());
m_StopThread = true;
}
}
}
PCPP_LOG_DEBUG("Ended capture thread for device '" << m_InterfaceDetails.name << "'");
}

internal::PcapHandle PcapLiveDevice::doOpen(const DeviceConfiguration& config)
{
char errbuf[PCAP_ERRBUF_SIZE] = { '\0' };
Expand Down Expand Up @@ -651,11 +721,13 @@ namespace pcpp
return false;
}

m_CaptureCallbackMode = true;
m_cbOnPacketArrives = std::move(onPacketArrives);
m_cbOnPacketArrivesUserCookie = onPacketArrivesUserCookie;
CaptureContext context;
context.device = this;
context.callback = std::move(onPacketArrives);
context.userCookie = onPacketArrivesUserCookie;

m_CaptureThread = std::thread(&pcpp::PcapLiveDevice::captureThreadMain, this);
m_CaptureThread = std::thread(&captureThreadMain, std::ref(m_StopThread), std::ref(m_CaptureThreadStarted),
std::ref(m_PcapDescriptor), std::move(context));

// Wait thread to be start
// C++20 = m_CaptureThreadStarted.wait(true);
Expand Down Expand Up @@ -707,11 +779,15 @@ namespace pcpp
return false;
}

m_CapturedPackets = &capturedPacketsVector;
m_CapturedPackets->clear();
capturedPacketsVector.clear();

AccumulatorCaptureContext context;
context.device = this;
context.capturedPackets = &capturedPacketsVector;

m_CaptureThread = std::thread(&captureThreadMainAccumulator, std::ref(m_StopThread),
std::ref(m_CaptureThreadStarted), std::ref(m_PcapDescriptor), std::move(context));

m_CaptureCallbackMode = false;
m_CaptureThread = std::thread(&pcpp::PcapLiveDevice::captureThreadMain, this);
// Wait thread to be start
// C++20 = m_CaptureThreadStarted.wait(true);
while (m_CaptureThreadStarted != true)
Expand Down Expand Up @@ -749,8 +825,6 @@ namespace pcpp
PCPP_LOG_ERROR("Failed to prepare capture: " << ex.what());
return 0;
}
m_cbOnPacketArrives = nullptr;
m_cbOnPacketArrivesUserCookie = nullptr;

m_cbOnPacketArrivesBlockingMode = std::move(onPacketArrives);
m_cbOnPacketArrivesBlockingModeUserCookie = userCookie;
Expand Down
Loading