liburing

CHANGELOG.md

@@ -1,6 +1,11 @@
 # Changelog
 
 
+### NEXT
+
+* Enable liburing usage for linux (kernel versions >= 6) ([PR #1218](https://github.com/versatica/mediasoup/pull/1218)).
+
+
 ### 3.13.1
 
 * Node: Extract version from `package.json` using `require()` ([PR #1217](https://github.com/versatica/mediasoup/pull/1217) by @arcinston).

worker/include/DepLibUring.hpp

@@ -0,0 +1,80 @@
+#ifndef MS_DEP_LIBURING_HPP
+#define MS_DEP_LIBURING_HPP
+
+#ifdef MS_LIBURING_ENABLED
+
+#include "DepLibUV.hpp"
+#include <functional>
+#include <liburing.h>
+#include <queue>
+
+class DepLibUring
+{
+public:
+	using onSendCallback = const std::function<void(bool sent)>;
+
+	/* Struct for the user data field of SQE and CQE. */
+	struct UserData
+	{
+		uint8_t store[1500]{};
+		onSendCallback* cb{ nullptr };
+		size_t idx{ 0 };
+	};
+
+	/* Number of submission queue entries (SQE). */
+	static constexpr size_t QueueDepth{ 1024 * 4 };
+
+	static void ClassInit();
+	static void ClassDestroy();
+
+	thread_local static DepLibUring* liburing;
+
+public:
+	DepLibUring();
+	~DepLibUring();
+	bool PrepareSend(
+	  int sockfd, const void* data, size_t len, const struct sockaddr* addr, onSendCallback* cb);
+	bool PrepareWrite(
+	  int sockfd, const void* data1, size_t len1, const void* data2, size_t len2, onSendCallback* cb);
+	void Submit();
+	void SetActive()
+	{
+		this->active = true;
+	}
+	bool IsActive() const
+	{
+		return this->active;
+	}
+	io_uring* GetRing()
+	{
+		return &this->ring;
+	}
+	int GetEventFd() const
+	{
+		return this->efd;
+	}
+	void ReleaseUserDataEntry(size_t idx)
+	{
+		this->availableUserDataEntries.push(idx);
+	}
+
+private:
+	UserData* GetUserData();
+
+private:
+	// io_uring instance.
+	io_uring ring;
+	// Event file descriptor to watch for completions.
+	int efd;
+	// libuv handle used to poll uring completions.
+	uv_poll_t* uvHandle{ nullptr };
+	// Whether we are currently sending RTP over io_uring.
+	bool active{ false };
+	// Pre-allocated UserData entries.
+	UserData userDataBuffer[QueueDepth]{};
+	// Indexes of available UserData entries.
+	std::queue<size_t> availableUserDataEntries;
+};
+
+#endif
+#endif

worker/include/handles/TcpConnectionHandle.hpp

@@ -143,6 +143,7 @@ class TcpConnectionHandle
 	uv_tcp_t* uvHandle{ nullptr };
 	// Others.
 	struct sockaddr_storage* localAddr{ nullptr };
+	uv_os_fd_t fd{ 0u };
 	bool closed{ false };
 	size_t recvBytes{ 0u };
 	size_t sentBytes{ 0u };

worker/include/handles/UdpSocketHandle.hpp

@@ -103,6 +103,7 @@ class UdpSocketHandle
 	// Allocated by this (may be passed by argument).
 	uv_udp_t* uvHandle{ nullptr };
 	// Others.
+	uv_os_fd_t fd{ 0u };
 	bool closed{ false };
 	size_t recvBytes{ 0u };
 	size_t sentBytes{ 0u };

worker/meson.build

@@ -257,6 +257,26 @@ if host_machine.system() == 'windows'
   ]
 endif
 
+if host_machine.system() == 'linux'
+  kernel_version = run_command('uname', '-r').stdout().strip()
+
+  # Enable liburing for kernel versions greather than or equal to 6.
+  if kernel_version[0].to_int() >= 6
+    liburing_proj = subproject('liburing', default_options: ['default_library=static'], required: true)
+    liburing = liburing_proj.get_variable('uring')
+
+    dependencies += [liburing]
+
+    common_sources += [
+      'src/DepLibUring.cpp',
+      ]
+
+    cpp_args += [
+      '-DMS_LIBURING_ENABLED',
+      ]
+  endif
+endif
+
 libmediasoup_worker = library(
   'libmediasoup-worker',
   name_prefix: '',