Socket接收数据回调支持无拷贝数据转移

ZLMediaKit · Jun 24, 2024 · 91f8919 · 91f8919
1 parent b45dac7
commit 91f8919
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 6 deletions.
diff --git a/src/Network/Socket.cpp b/src/Network/Socket.cpp
@@ -282,7 +282,8 @@ class MMsgBuffer {
 #endif
 
     MMsgBuffer(size_t count, size_t size)
-        : _iovec(count)
+        : _size(size)
+        , _iovec(count)
         , _mmsgs(count)
         , _buffers(count)
         , _address(count) {
@@ -307,8 +308,16 @@ class MMsgBuffer {
     }
 
     ssize_t recvFromSocket(int fd, ssize_t &count) {
-        for (auto &mmsg : _mmsgs) {
+        for (auto i = 0u; i < _mmsgs.size(); ++i) {
+            auto &mmsg = _mmsgs[i];
             mmsg.msg_hdr.msg_namelen = sizeof(struct sockaddr_storage);
+            auto &buf = _buffers[i];
+            if (!buf) {
+                auto raw = BufferRaw::create();
+                raw->setCapacity(_size);
+                buf = raw;
+                mmsg.msg_hdr.msg_iov->iov_base = buf->data();
+            }
         }
         do {
             count = recvmmsg(fd, &_mmsgs[0], _mmsgs.size(), 0, nullptr);
@@ -330,9 +339,9 @@ class MMsgBuffer {
         return nread;
     }
 
-    const Buffer::Ptr &getBuffer(size_t index) const { return _buffers[index]; }
+    Buffer::Ptr &getBuffer(size_t index) { return _buffers[index]; }
 
-    const struct sockaddr_storage &getAddress(size_t index) const { return _address[index]; }
+    struct sockaddr_storage &getAddress(size_t index) { return _address[index]; }
 
 private:
 #if !defined(__linux)
@@ -347,6 +356,7 @@ class MMsgBuffer {
 #endif
 
 private:
+    size_t _size;
     std::vector<struct iovec> _iovec;
     std::vector<struct mmsghdr> _mmsgs;
     std::vector<Buffer::Ptr> _buffers;

diff --git a/src/Network/Socket.h b/src/Network/Socket.h
@@ -282,8 +282,8 @@ class Socket : public std::enable_shared_from_this<Socket>, public noncopyable,
 public:
     using Ptr = std::shared_ptr<Socket>;
     //接收数据回调
-    using onReadCB = std::function<void(const Buffer::Ptr &buf, struct sockaddr *addr, int addr_len)>;
-    using onMultiReadCB = std::function<void(const Buffer::Ptr *buf, const struct sockaddr_storage *addr, size_t count)>;
+    using onReadCB = std::function<void(Buffer::Ptr &buf, struct sockaddr *addr, int addr_len)>;
+    using onMultiReadCB = std::function<void(Buffer::Ptr *buf, struct sockaddr_storage *addr, size_t count)>;
 
     //发生错误回调
     using onErrCB = std::function<void(const SockException &err)>;