Refactor worker thread capture procedures as internal functions. (#1971)

* Refactor worker thread capture procedures as internal functions.

* Temp removed exception handling code.

* Lint

Author: Dimi1010

Pcap++/header/PcapLiveDevice.h

@@ -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;
 
@@ -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:

Pcap++/src/PcapLiveDevice.cpp

@@ -272,6 +272,126 @@ 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;
+			}
+
+			// TODO: Add exception handling to tunnel the exceptions from here to the capture thread
+			// through the C code boundary
+
+			RawPacket rawPacket(packet, pkthdr->caplen, pkthdr->ts, false, context->device->getLinkType());
+			context->callback(&rawPacket, context->device, context->userCookie);
+		}
+
+		/// @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;
+			}
+
+			// TODO: Add exception handling to tunnel the exceptions from here to the capture thread
+			// through the C code boundary
+
+			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));
+		}
+
+		/// @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,
@@ -309,50 +429,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)
 	{
@@ -371,37 +456,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' };
@@ -654,11 +708,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);
@@ -710,11 +766,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)
@@ -752,8 +812,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;