Sockets dimensioned by Num queues

Pcap++/header/XdpDevice.h

@@ -6,20 +6,31 @@
 #include <memory>
 #include <utility>
 #include <functional>
+#include <vector>
 
 /// @namespace pcpp
 /// @
 namespace pcpp
 {
+
+	#define PCPP_MAX_XDP_NUMBER_QUEUES 64
+
+	/// Forward Declaration of XdpSocket
+	class XdpSocket;
+
 	/// @class XdpDevice
 	/// A class wrapping the main functionality of using AF_XDP (XSK) sockets
 	/// which are optimized for high performance packet processing.
 	///
-	/// It provides methods for configuring and initializing an AF_XDP socket, and then send and receive packets through
-	/// it. It also provides a method for gathering statistics from the socket.
+	/// It provides methods for configuring and initializing multipler AF_XDP socket, and then send and receive packets through
+	/// these. It also provides a method for gathering statistics from the socket.
 	class XdpDevice : public IDevice
 	{
+
+	friend class XdpSocket;
+
 	public:
+
 		/// @typedef OnPacketsArrive
 		/// The callback that is called whenever packets are received on the socket
 		/// @param[in] packets An array of the raw packets received
@@ -48,6 +59,10 @@ namespace pcpp
 			/// AF_XDP operation mode
 			AttachMode attachMode;
 
+			/// number of queues. Should be less than or equal to the number of hardware queues supported by the device
+			// the queue ids are inferred as consecutive starting at zero
+			uint32_t numQueues;
+
 			/// UMEM is a region of virtual contiguous memory, divided into equal-sized frames.
 			/// This parameter determines the number of frames that will be allocated as pert of the UMEM.
 			uint16_t umemNumFrames;
@@ -87,10 +102,11 @@ namespace pcpp
 			/// @param[in] txSize The size of the TX ring used by the AF_XDP socket. The default value is 2048
 			/// @param[in] rxTxBatchSize The max number of packets to be received or sent in one batch. The default
 			/// value is 64
+			/// @param[in] numQueues The number of queues, and therefore sockets, to configure with device. The default is 1
 			explicit XdpDeviceConfiguration(AttachMode attachMode = AutoMode, uint16_t umemNumFrames = 0,
 			                                uint16_t umemFrameSize = 0, uint32_t fillRingSize = 0,
 			                                uint32_t completionRingSize = 0, uint32_t rxSize = 0, uint32_t txSize = 0,
-			                                uint16_t rxTxBatchSize = 0)
+			                                uint16_t rxTxBatchSize = 0, uint32_t numQueues = 0)
 			{
 				this->attachMode = attachMode;
 				this->umemNumFrames = umemNumFrames;
@@ -100,6 +116,7 @@ namespace pcpp
 				this->rxSize = rxSize;
 				this->txSize = txSize;
 				this->rxTxBatchSize = rxTxBatchSize;
+				this->numQueues = numQueues;
 			}
 		};
 
@@ -155,37 +172,37 @@ namespace pcpp
 			uint64_t umemFreeFrames;
 		};
 
-		/// A c'tor for this class. Please note that calling this c'tor doesn't initialize the AF_XDP socket. In order
-		/// to set up the socket call open().
-		/// @param[in] interfaceName The interface name to open the AF_XDP socket on
+		/// A c'tor for this class. Please note that calling this c'tor doesn't initialize the AF_XDP sockets. In order
+		/// to set up the sockets call open().
+		/// @param[in] interfaceName The interface name to open the AF_XDP sockets on
 		explicit XdpDevice(std::string interfaceName);
 
-		/// A d'tor for this class. It closes the device if it's open.
+		/// A d'tor for this class. It closes the device if it's open and frees up the sockets
 		~XdpDevice() override;
 
 		/// Open the device with default configuration. Call getConfig() after opening the device to get the
 		/// current configuration.
-		/// This method initializes the UMEM, and then creates and configures the AF_XDP socket. If it succeeds the
-		/// socket is ready to receive and send packets.
+		/// This method creates and configures the AF_XDP sockets per queue. If it succeeds the
+		/// sockets are ready to receive and send packets.
 		/// @return True if device was opened successfully, false otherwise
 		bool open() override;
 
 		/// Open the device with custom configuration set by the user.
-		/// This method initializes the UMEM, and then creates and configures the AF_XDP socket. If it succeeds the
-		/// socket is ready to receive and send packets.
+		/// This method initializes the UMEM, and then creates and configures the AF_XDP sockets. If it succeeds the
+		/// sockets are ready to receive and send packets.
 		/// @param[in] config The configuration to use for opening the device
 		/// @return True if device was opened successfully, false otherwise
 		bool open(const XdpDeviceConfiguration& config);
 
-		/// Close the device. This method closes the AF_XDP socket and frees the UMEM that was allocated for it.
+		/// Close the device. This method closes the AF_XDP sockets and frees the associated socket resources.
 		void close() override;
 
 		bool isOpened() const override
 		{
 			return m_DeviceOpened;
 		}
 
-		/// Start receiving packets. In order to use this method the device should be open. Note that this method is
+		/// Start receiving packets on queue id = 0. In order to use this method the device should be open. Note that this method is
 		/// blocking and will return if:
 		/// - stopReceivePackets() was called from within the user callback
 		/// - timeoutMS passed without receiving any packets
@@ -203,7 +220,7 @@ namespace pcpp
 		/// want to stop receiving packets.
 		void stopReceivePackets();
 
-		/// Send a vector of packet pointers.
+		/// Send a vector of packet pointers on queue id zero
 		/// @param[in] packets A vector of packet pointers to send
 		/// @param[in] waitForTxCompletion Wait for confirmation from the kernel that packets were sent. If set to true
 		/// this method will wait until the number of packets in the completion ring is equal or greater to the number
@@ -215,7 +232,7 @@ namespace pcpp
 		bool sendPackets(const RawPacketVector& packets, bool waitForTxCompletion = false,
 		                 int waitForTxCompletionTimeoutMS = 5000);
 
-		/// Send an array of packets.
+		/// Send an array of packets on queue id zero
 		/// @param[in] packets An array of raw packets to send
 		/// @param[in] packetCount The length of the packet array
 		/// @param[in] waitForTxCompletion Wait for confirmation from the kernel that packets were sent. If set to true
@@ -232,13 +249,144 @@ namespace pcpp
 		XdpDeviceConfiguration* getConfig() const
 		{
 			// TODO: Return a copy or const ref to avoid user modifying config?
-			return m_Config.get();
+			return m_Config;
 		}
 
-		/// @return Current device statistics
+		/// @return Current device statistics for queue id zero
 		XdpDeviceStats getStatistics();
 
+		/// Get the XdpSocket object for the specified queue id
+		/// @param[in] queueid the queueid of the related socket
+		/// @return the pointer to the XdpSocket object if queueid valid otherwise nullptr
+		XdpSocket *getSocket(uint32_t queueid = 0)
+		{
+			if(queueid < m_Socket.size())
+			{
+				return m_Socket[queueid].get();
+			}
+
+			return nullptr;
+		}
+
+	private:
+
+		struct XdpPrevDeviceStats
+		{
+			timespec timestamp;
+			uint64_t rxPackets;
+			uint64_t rxBytes;
+			uint64_t txSentPackets;
+			uint64_t txSentBytes;
+			uint64_t txCompletedPackets;
+		};
+
+		bool m_DeviceOpened = false;
+
+		std::string m_InterfaceName;
+		XdpDeviceConfiguration* m_Config;
+		std::vector<std::unique_ptr<XdpSocket>> m_Socket;
+
+		bool sendPackets(const std::function<RawPacket(uint32_t)>& getPacketAt,
+			const std::function<uint32_t()>& getPacketCount, bool waitForTxCompletion = false,
+			int waitForTxCompletionTimeoutMS = 5000);
+		bool initConfig();
+
+		// for backward compatibility
+		OnPacketsArrive OnPacketsArriveCB_;	
+		static void onPacketsArriveSocketZero(RawPacket packets[], uint32_t packetCount, XdpSocket* socket, void* userCookie);
+
+	};
+
+	/// @class XdpSocket
+	/// A class wrapping the main functionality of a single AF_XDP (XSK) socket
+	/// which are optimized for high performance packet processing.
+	///
+	/// It provides methods for configuring and initializing AF_XDP socket, and then send and receive packets through
+	/// it. It also provides a method for gathering statistics from the socket.
+	class XdpSocket
+	{
+
+	friend class XdpDevice;
+
+	public:
+
+		/// A c'tor for this class. Please note that calling this c'tor doesn't initialize the AF_XDP socket. In order
+		/// to set up the socket call configureSocket().
+		/// @param[in] device The custodial XdpDevice
+		/// @param[in] queueid The queue id for the socket
+		explicit XdpSocket(XdpDevice *device, uint32_t queueid);
+
+		/// A d'tor for this class. It deletes the underlying socket resource and frees allocated UMEM
+		~XdpSocket();
+
+		/// Configure the socket with device configuration.
+		/// This method initializes the UMEM, and then creates and configures the AF_XDP socket for a queue. If it succeeds the
+		/// sockets are ready to receive and send packets.
+		/// @return True if socket was configured successfully, false otherwise
+		bool configureSocket();
+
+		/// A socket is associated with a device
+		XdpDevice *getDevice() { return m_Device; }
+
+		/// A socket is associated with one of the device queues. 
+		uint32_t getQueueId() { return m_Queueid; }
+
+		/// @typedef OnPacketsArriveSocket
+		/// The callback that is called whenever packets are received on the socket
+		/// @param[in] packets An array of the raw packets received
+		/// @param[in] packetCount The number of packets received
+		/// @param[in] socket The XdpSocket packets are received from (represents the AF_XDP socket)
+		/// @param[in] userCookie A pointer to an object set by the user when receivePackets() started
+		typedef void (*OnPacketsArriveSocket)(RawPacket packets[], uint32_t packetCount, XdpSocket* socket, void* userCookie);
+
+		/// Start receiving packets on socket. In order to use this method the device should be open. Note that this method is
+		/// blocking and will return if:
+		/// - stopReceivePackets() was called from within the user callback
+		/// - timeoutMS passed without receiving any packets
+		/// - Some error occurred (an error log will be printed)
+		/// @param[in] onPacketsArriveSocket A callback to be called when packets are received on particular socket
+		/// @param[in] onPacketsArriveSocketUserCookie The callback is invoked with this cookie as a parameter. It can be used
+		/// to pass information from the user application to the callback
+		/// @param[in] timeoutMS Timeout in milliseconds to stop if no packets are received. The default value is 5000
+		/// ms
+		/// @return True if stopped receiving packets because stopReceivePackets() was called or because timeoutMS
+		/// passed, or false if an error occurred.
+		bool receivePackets(OnPacketsArriveSocket onPacketsArriveSocket, void* onPacketsArriveSocketUserCookie, int timeoutMS = 5000);
+
+		/// Stop receiving packets. Call this method from within the callback passed to receivePackets() whenever you
+		/// want to stop receiving packets on the socket.
+		void stopReceivePackets();
+
+		/// Send a vector of packet pointers.
+		/// @param[in] packets A vector of packet pointers to send
+		/// @param[in] waitForTxCompletion Wait for confirmation from the kernel that packets were sent. If set to true
+		/// this method will wait until the number of packets in the completion ring is equal or greater to the number
+		/// of packets that were sent. The default value is false
+		/// @param[in] waitForTxCompletionTimeoutMS If waitForTxCompletion is set to true, poll the completion ring with
+		/// this timeout. The default value is 5000 ms
+		/// @return True if all packets were sent, or if waitForTxCompletion is true - all sent packets were confirmed.
+		/// Returns false if an error occurred or if poll timed out.
+		bool sendPackets(const RawPacketVector& packets, bool waitForTxCompletion = false,
+						int waitForTxCompletionTimeoutMS = 5000);
+
+		/// Send an array of packets.
+		/// @param[in] packets An array of raw packets to send
+		/// @param[in] packetCount The length of the packet array
+		/// @param[in] waitForTxCompletion Wait for confirmation from the kernel that packets were sent. If set to true
+		/// this method will wait until the number of packets in the completion ring is equal or greater to the number
+		/// of packets sent. The default value is false
+		/// @param[in] waitForTxCompletionTimeoutMS If waitForTxCompletion is set to true, poll the completion ring with
+		/// this timeout. The default value is 5000 ms
+		/// @return True if all packets were sent, or if waitForTxCompletion is true - all sent packets were confirmed.
+		/// Returns false if an error occurred or if poll timed out.
+		bool sendPackets(RawPacket packets[], size_t packetCount, bool waitForTxCompletion = false,
+						int waitForTxCompletionTimeoutMS = 5000);
+
+		/// @return Current device statistics
+		XdpDevice::XdpDeviceStats getStatistics();
+
 	private:
+
 		class XdpUmem
 		{
 		public:
@@ -282,35 +430,23 @@ namespace pcpp
 			std::vector<uint64_t> m_FreeFrames;
 		};
 
-		struct XdpPrevDeviceStats
-		{
-			timespec timestamp;
-			uint64_t rxPackets;
-			uint64_t rxBytes;
-			uint64_t txSentPackets;
-			uint64_t txSentBytes;
-			uint64_t txCompletedPackets;
-		};
+		XdpDevice *m_Device;
+		uint32_t m_Queueid;
 
-		bool m_DeviceOpened = false;
-
-		std::string m_InterfaceName;
-		std::unique_ptr<XdpDeviceConfiguration> m_Config;
-		bool m_ReceivingPackets;
-		XdpUmem* m_Umem;
-		void* m_SocketInfo;
-		XdpDeviceStats m_Stats;
-		XdpPrevDeviceStats m_PrevStats;
+		bool m_ReceivingPackets = false;
+		XdpUmem* m_Umem = nullptr;
+		void* m_SocketInfo = nullptr;
+		XdpDevice::XdpDeviceStats m_Stats;
+		XdpDevice::XdpPrevDeviceStats m_PrevStats;
 
 		bool sendPackets(const std::function<RawPacket(uint32_t)>& getPacketAt,
-		                 const std::function<uint32_t()>& getPacketCount, bool waitForTxCompletion = false,
-		                 int waitForTxCompletionTimeoutMS = 5000);
+						const std::function<uint32_t()>& getPacketCount, bool waitForTxCompletion = false,
+						int waitForTxCompletionTimeoutMS = 5000);
+
 		bool populateFillRing(uint32_t count, uint32_t rxId = 0);
 		bool populateFillRing(const std::vector<uint64_t>& addresses, uint32_t rxId);
 		uint32_t checkCompletionRing();
-		bool configureSocket();
 		bool initUmem();
-		bool populateConfigDefaults(XdpDeviceConfiguration& config) const;
 		bool getSocketStats();
 	};
 }  // namespace pcpp