TCP input TLS - Optimize TlsDecoder.

This also brings new api for `DecodeBuffer` and simplification of IpfixDecoder.

Author: BonnyAD9

src/plugins/input/tcp/CMakeLists.txt

@@ -13,6 +13,8 @@ add_library(tcp-input MODULE
     src/Acceptor.cpp
     src/Plugin.cpp
     src/IpxPlugin.cpp
+    src/TcpReader.cpp
+    src/RingBufferReader.cpp
     src/tls/Ssl.cpp
     src/tls/TlsDecoder.cpp
     src/tls/DecoderFactory.cpp

src/plugins/input/tcp/src/DecodeBuffer.cpp

@@ -19,58 +19,34 @@
 
 namespace tcp_in {
 
-void DecodeBuffer::read_from(const uint8_t *data, size_t size) {
-    // read until there is something to read
-    while (size) {
-        // get the message size
-        if (!read_header(&data, &size)) {
-            // There is not enough data to read the whole header.
+void DecodeBuffer::read_from(Reader &reader, std::size_t consume) {
+    consume += m_total_bytes_decoded;
+    while (!enough_data() || consume < m_total_bytes_decoded) {
+        if (!read_header(reader)) {
             break;
         }
 
-        // Read the body of the message.
-        if (!read_body(&data, &size)) {
-            // There is not enough data to read the whole body of the message.
+        if (!read_body(reader)) {
             break;
         }
     }
 }
 
-void DecodeBuffer::read_from(
-    const uint8_t *data,
-    size_t buffer_size,
-    size_t data_size,
-    size_t position
-) {
-    // read until wrap
-    auto block = data + position;
-    auto block_size = std::min(data_size, buffer_size - position);
-    read_from(block, block_size);
-
-    if (block_size == data_size) {
-        // there was no need to wrap
-        return;
-    }
-
-    // read the second block (after wrap)
-    read_from(data, data_size - block_size);
-}
-
 void DecodeBuffer::signal_eof() {
     if (m_part_decoded.size() != 0) {
         throw std::runtime_error("Received incomplete message.");
     }
     m_eof_reached = true;
 }
 
-bool DecodeBuffer::read_header(const uint8_t **data, size_t *size) {
+bool DecodeBuffer::read_header(Reader &reader) {
     if (m_decoded_size != 0) {
         // The header is already read, but the message body is incomplete.
         return true;
     }
 
     // Read the header.
-    if (!read_until_n(sizeof(fds_ipfix_msg_hdr), data, size)) {
+    if (!read_until_n(sizeof(fds_ipfix_msg_hdr), reader)) {
         // There is not enough data to read the whole header.
         return false;
     }
@@ -86,9 +62,9 @@ bool DecodeBuffer::read_header(const uint8_t **data, size_t *size) {
     return true;
 }
 
-bool DecodeBuffer::read_body(const uint8_t **data, size_t *size) {
+bool DecodeBuffer::read_body(Reader &reader) {
     // Read the body
-    if (!read_until_n(m_decoded_size, data, size)) {
+    if (!read_until_n(m_decoded_size, reader)) {
         // There is not enough data to read the whole body.
         return false;
     }
@@ -99,22 +75,10 @@ bool DecodeBuffer::read_body(const uint8_t **data, size_t *size) {
     return true;
 }
 
-bool DecodeBuffer::read_until_n(size_t n, const uint8_t **data, size_t *data_len) {
-    m_part_decoded.reserve(n);
-    auto cnt = read_min(*data, n, *data_len);
-    *data_len -= cnt;
-    *data += cnt;
+bool DecodeBuffer::read_until_n(size_t n, Reader &reader) {
+    m_total_bytes_decoded += reader.read_until_n(n, m_part_decoded, *this);
     return m_part_decoded.size() == n;
 }
 
-size_t DecodeBuffer::read_min(const uint8_t *data, size_t size1, size_t size2) {
-    auto size = std::min(size1, size2);
-    auto filled = m_part_decoded.size();
-    m_part_decoded.resize(filled + size);
-    std::copy_n(data, size, m_part_decoded.data() + filled);
-    m_total_bytes_decoded += size;
-    return size;
-}
-
 } // namespace tcp_in
 

src/plugins/input/tcp/src/DecodeBuffer.hpp

@@ -10,11 +10,12 @@
 
 #pragma once
 
-#include <vector>  // std::vector
 #include <cstdint> // uint8_t, UINT16_MAX
 #include <cstddef> // size_t
+#include <vector>  // std::vector
 
 #include "ByteVector.hpp" // ByteVector
+#include "Reader.hpp"
 
 namespace tcp_in {
 
@@ -50,28 +51,11 @@ class DecodeBuffer {
     }
 
     /**
-     * @brief Copies IPFIX data from buffer.
-     *
-     * The data may be any part of message (possibly incomplete or even multiple messages) but
-     * multiple calls to this method must be with the message data in correct order so that it can
-     * be reconstructed.
-     * @param data data with the message
-     * @param size size of the data in `data`
+     * @brief Read data using generic reader function.
+     * @param reader Generic reader function.
+     * @param consume Minimum number of bytes that should be read from reader if possible.
      */
-    void read_from(const uint8_t *data, size_t size);
-
-    /**
-     * @brief Copies IPFIX data from circular buffer.
-     *
-     * The data may be any part of message (possibly incomplete or even multiple messages) but
-     * multiple calls to this metod must be with the message data in correct order so that it can be
-     * reconstructed.
-     * @param data data of the circullar buffer
-     * @param buffer_size size of the circullar buffer (allocated space)
-     * @param data_size size of data to copy from the buffer
-     * @param position start position of the data in the buffer.
-     */
-    void read_from(const uint8_t *data, size_t buffer_size, size_t data_size, size_t position);
+    void read_from(Reader &reader, std::size_t consume = 0);
 
     /**
      * @brief Checks whether enough data has been read since last time. When this returns true,
@@ -104,32 +88,23 @@ class DecodeBuffer {
 private:
     /**
      * @brief Reads the length from ipfix header to `m_decoded_size`.
-     * @param[in,out] data Pointer to data. It is modified by the amount of read data.
-     * @param[in,out] size Size of the data. It is modified by the amount of read data.
+     * @param reader Reader from which to read.
      * @return `true` if the whole header could be read.
      */
-    bool read_header(const uint8_t **data, size_t *size);
+    bool read_header(Reader &reader);
     /**
      * @brief Reads the body of ipfix message.
-     * @param data Pointer to data. It is modified by the amount of read data.
-     * @param size Size of the data. It is modified by the amount of read data.
+     * @param reader Reader from which to read.
      * @return `true` if the whole header could be read.
      */
-    bool read_body(const uint8_t **data, size_t *size);
+    bool read_body(Reader &reader);
     /**
      * @brief Reads to `m_part_decoded` until it reaches length of `n`.
      * @param n Target length for `m_part_decoded`.
-     * @param[in,out] data Data to read from. It is modified by the read amount.
-     * @param[in,out] data_len Length of data to read. It is modified by the read amount.
+     * @param reader Generic reader from which to read.
      * @return `true` if `m_part_decoded` has reached length of `n`.
      */
-    bool read_until_n(size_t n, const uint8_t **data, size_t *data_len);
-    /**
-     * @brief Adds N elements from `src` to `m_part_decoded` where N is the smaller of the two
-     * sizes.
-     * @return The smaller of the two sizes.
-     */
-    size_t read_min(const uint8_t *data, size_t size1, size_t size2);
+    bool read_until_n(size_t n, Reader &reader);
 
     /** number of bytes readed since last call to `get_decoded` */
     size_t m_total_bytes_decoded;

src/plugins/input/tcp/src/IpfixDecoder.cpp

@@ -10,74 +10,13 @@
 
 #include "IpfixDecoder.hpp"
 
-#include <stdexcept> // runtime_error
-#include <string>    // string
-#include <errno.h>   // errno, EWOULDBLOCK, EAGAIN
-#include <stddef.h>  // size_t
-
-#include <unistd.h> // read
-
-#include <ipfixcol2.h> // fds_ipfix_msg_hdr, ipx_strerror
-
 #include "DecodeBuffer.hpp" // DecodeBuffer
-#include "read_until_n.hpp" // read_until_n
+#include "Reader.hpp"
 
 namespace tcp_in {
 
 void IpfixDecoder::progress() {
-    while (!m_decoded.enough_data()) {
-        // Read the header of the message.
-        if (!read_header()) {
-            // There is not enough data to read the whole header.
-            break;
-        }
-
-        // Read the body of the message.
-        if (!read_body()) {
-            // There is not enough data to read the whole body of the message.
-            break;
-        }
-    }
-
-    if (m_decoded.is_eof_reached() && m_part_readed.size() != 0) {
-        throw std::runtime_error("Received incomplete message.");
-    }
-}
-
-bool IpfixDecoder::read_header() {
-    if (m_msg_size != 0) {
-        // The header is already read, but the message body is incomplete.
-        return true;
-    }
-
-    // Read the header
-    if (!read_until_n(sizeof(fds_ipfix_msg_hdr))) {
-        // There is not enough data to read the whole header.
-        return false;
-    }
-
-    // The header has been read, load the size of the message.
-    auto hdr = reinterpret_cast<fds_ipfix_msg_hdr *>(m_part_readed.data());
-    m_msg_size = ntohs(hdr->length);
-    return true;
-}
-
-bool IpfixDecoder::read_body() {
-    // Read the body.
-    if (!read_until_n(m_msg_size)) {
-        // There is not enough data to read the whole body
-        return false;
-    }
-
-    // Whole message has been read. Add it to the decode buffer.
-    m_decoded.add(std::move(m_part_readed));
-    m_part_readed = ByteVector();
-    m_msg_size = 0;
-    return true;
-}
-
-bool IpfixDecoder::read_until_n(size_t n) {
-    return ::read_until_n(n, m_fd, m_part_readed, m_decoded);
+    m_decoded.read_from(m_reader);
 }
 
 } // namespace tcp_in

src/plugins/input/tcp/src/IpfixDecoder.hpp

@@ -15,7 +15,7 @@
 
 #include "Decoder.hpp"      // Decoder
 #include "DecodeBuffer.hpp" // DecodeBuffer
-#include "ByteVector.hpp"   // ByteVector
+#include "TcpReader.hpp"
 
 namespace tcp_in {
 
@@ -29,7 +29,7 @@ class IpfixDecoder : public Decoder {
      * @brief Creates ipfix decoder.
      * @param fd TCP connection file descriptor.
      */
-    IpfixDecoder(int fd) : m_fd(fd), m_decoded(), m_part_readed(), m_msg_size(0) {}
+    IpfixDecoder(int fd) : m_reader(fd), m_decoded() {}
 
     virtual void progress() override;
 
@@ -40,19 +40,8 @@ class IpfixDecoder : public Decoder {
     }
 
 private:
-    /** returns true if there was enough data to read the header or if the header is already read */
-    bool read_header();
-    /** returns true if there was enough data to read the body */
-    bool read_body();
-    /** returns true if there was enough data to read to the given amount */
-    bool read_until_n(size_t n);
-
-    int m_fd;
+    TcpReader m_reader;
     DecodeBuffer m_decoded;
-
-    ByteVector m_part_readed;
-    /** Final size of the whole IPFIX message. When 0 IPIFX header is not fully read. */
-    size_t m_msg_size;
 };
 
 } // namespace tcp_in

src/plugins/input/tcp/src/Lz4Decoder.cpp

@@ -14,6 +14,7 @@
 #include <stdexcept> // runtime_error
 #include <cstddef>   // size_t
 #include <cerrno>    // errno, EWOULDBLOCK, EAGAIN
+#include <cstring>
 #include <string>    // string
 
 #include <netinet/in.h> // ntohl, ntohs
@@ -23,7 +24,7 @@
 #include <ipfixcol2.h> // ipx_strerror
 
 #include "DecodeBuffer.hpp" // DecodeBuffer
-#include "read_until_n.hpp" // read_until_n
+#include "RingBufferReader.hpp"
 
 namespace tcp_in {
 
@@ -38,7 +39,7 @@ struct __attribute__((__packed__)) ipfix_start_compress_header {
 };
 
 Lz4Decoder::Lz4Decoder(int fd) :
-    m_fd(fd),
+    m_reader(fd),
     m_decoded(),
     m_decoder(LZ4_createStreamDecode()),
     m_decompressed(),
@@ -158,12 +159,13 @@ void Lz4Decoder::decompress() {
     }
 
     // Copy the decompressed data into the decode buffer
-    m_decoded.read_from(
+    auto reader = RingBufferReader(
         m_decompressed.data(),
         m_decompressed.size(),
         m_decompressed_size,
         m_decompressed_pos
     );
+    m_decoded.read_from(reader, m_decompressed_size);
 
     m_decompressed_pos += m_decompressed_size;
     if (m_decompressed_pos >= m_decompressed.size()) {
@@ -175,7 +177,8 @@ void Lz4Decoder::decompress() {
 }
 
 bool Lz4Decoder::read_until_n(size_t n) {
-    return ::read_until_n(n, m_fd, m_compressed, m_decoded);
+    m_reader.read_until_n(n, m_compressed, m_decoded);
+    return m_compressed.size() == n;
 }
 
 void Lz4Decoder::reset_stream(size_t buffer_size) {

src/plugins/input/tcp/src/Lz4Decoder.hpp

@@ -19,6 +19,7 @@
 
 #include "Decoder.hpp"      // Decoder
 #include "DecodeBuffer.hpp" // DecodeBuffer
+#include "TcpReader.hpp"
 
 namespace tcp_in {
 
@@ -92,7 +93,7 @@ class Lz4Decoder : public Decoder {
      */
     void reset_stream(size_t buffer_size);
 
-    int m_fd;
+    TcpReader m_reader;
     DecodeBuffer m_decoded;
 
     std::unique_ptr<LZ4_streamDecode_t, Lz4DecodeDestructor> m_decoder;