cansocket.hpp

/*
 * Copyright 2025 TierOne Software
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#pragma once

#include <array>
#include <atomic>
#include <cerrno>
#include <chrono>
#include <cstring> // for strerror
#include <fcntl.h>
#include <iomanip>
#include <iostream>
#include <memory>
#include <memory_resource>
#include <mutex>
#include <net/if.h>
#include <optional>
#include <queue>
#include <shared_mutex>
#include <stdexcept>
#include <string>
#include <sys/ioctl.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <sys/uio.h>
#include <sys/select.h>
#include <thread>
#include <unistd.h>
#include <unordered_map>
#include <variant>
#include <net/if.h> // for IFNAMSIZ

#include <linux/can.h>
#include <linux/can/raw.h>

#include "fmt/format.h"

namespace CAN
{

// Memory pool for efficient frame allocation
template<typename T, size_t POOL_SIZE = 1024>
class FramePool {
private:
    alignas(T) std::array<uint8_t, sizeof(T) * POOL_SIZE> pool_storage;
    std::array<bool, POOL_SIZE> used;
    size_t next_free = 0;

public:
    FramePool() {
        used.fill(false);
    }

    [[nodiscard]] T* allocate() {
        // Find next free slot
        for (size_t i = 0; i < POOL_SIZE; ++i) {
            size_t idx = (next_free + i) % POOL_SIZE;
            if (!used[idx]) {
                used[idx] = true;
                next_free = (idx + 1) % POOL_SIZE;
                return reinterpret_cast<T*>(pool_storage.data() + idx * sizeof(T));
            }
        }
        return nullptr; // Pool exhausted
    }

    void deallocate(T* ptr) {
        auto pool_start = reinterpret_cast<T*>(pool_storage.data());
        auto pool_end = reinterpret_cast<T*>(pool_storage.data() + pool_storage.size());
        if (ptr >= pool_start && ptr < pool_end) {
            size_t idx = ptr - pool_start;
            if (idx < POOL_SIZE) {
                used[idx] = false;
            }
        }
    }

    [[nodiscard]] size_t available() const {
        size_t count = 0;
        for (bool is_used : used) {
            if (!is_used) ++count;
        }
        return count;
    }

    [[nodiscard]] size_t capacity() const {
        return POOL_SIZE;
    }
};

// Batch operations for high-performance scenarios
struct BatchContext {
    static constexpr size_t DEFAULT_BATCH_SIZE = 100;
    size_t batch_size = DEFAULT_BATCH_SIZE;
    std::chrono::microseconds timeout{1000};
    bool use_zero_copy = true;
};

// Custom exception types for better error handling
class SocketException final : public std::runtime_error {
public:
    explicit SocketException(const std::string& message) : std::runtime_error(message) {}
};

class FrameException final : public std::runtime_error {
public:
    explicit FrameException(const std::string& message) : std::runtime_error(message) {}
};

class FilterException final : public std::runtime_error {
public:
    explicit FilterException(const std::string& message) : std::runtime_error(message) {}
};

class IDException final : public std::runtime_error {
public:
    explicit IDException(const std::string& message) : std::runtime_error(message) {}
};

// Socket protocol mode enum
enum class ProtocolMode {
    ClassicCAN,  // Classic CAN mode (CAN 2.0)
    CanFD        // CAN FD mode
};

// Classic CAN constants
constexpr uint32_t EFF_FLAG = CAN_EFF_FLAG; // EFF/SFF is set in the MSB
constexpr uint32_t RTR_FLAG = CAN_RTR_FLAG; // remote transmission request
constexpr uint32_t ERR_FLAG = CAN_ERR_FLAG; // error frame

constexpr uint32_t SFF_MASK = CAN_SFF_MASK; // standard frame
constexpr uint32_t EFF_MASK = CAN_EFF_MASK; // extended frame
constexpr uint32_t ERR_MASK = CAN_ERR_MASK; // omit EFF, RTR, ERR flags

// CAN FD Constants
namespace FD {
    constexpr uint8_t MAX_DLC = 15;        // Maximum DLC value for CAN FD
    constexpr uint8_t MAX_DLEN = 64;       // Maximum data length for CAN FD
    constexpr uint8_t BRS_FLAG = CANFD_BRS; // Bit rate switch
    constexpr uint8_t ESI_FLAG = CANFD_ESI; // Error state indicator
    constexpr uint8_t FDF_FLAG = CANFD_FDF; // FD frame format
}

// CAN frame ID
// bit 0-28: CAN ID (11/29 bit)
// bit 29: error message frame flag (0=Data, 1=Error)
// bit 30: remote transmission request flag (RTR) (0=Data, 1=RTR)
// bit 31: frame format flag (0=Standard, 1=Extended)
class ID {
	uint32_t id;

    protected:
	explicit ID(const uint32_t id) : id(id)
	{
	}

	virtual ~ID() = default;

    public:
	[[nodiscard]] uint32_t get() const
	{
		return this->id;
	}
};

// StandardID is a wrapper around ID
class StandardID final : public ID {
public:
    // Constructor that validates and sets the value to fit in 11 bits
    explicit StandardID(const uint32_t id) : ID(id & SFF_MASK)
    {
        // Throw exception if the original ID exceeded 11 bits
        if ((id & ~SFF_MASK) != 0) {
            throw IDException(fmt::format("ID value 0x{:x} exceeds 11-bit limit for Standard CAN ID (max: 0x{:x})", id, SFF_MASK));
        }
    }

    // Template version for compile-time checking with literals
    template <uint32_t id>
    static StandardID create()
    {
        static_assert((id & ~SFF_MASK) == 0,
                      "Standard CAN ID exceeds 11-bit limit (0x7FF)");
        return StandardID(id);
    }

    ~StandardID() override = default;

    [[nodiscard]] uint32_t justSFF() const
    {
        return get() & SFF_MASK;
    }
};

// ExtendedID is a wrapper around ID
class ExtendedID final : public ID {
public:
    // Constructor that validates and sets the value to fit in 29 bits
    explicit ExtendedID(const uint32_t id) : ID((id & EFF_MASK) | EFF_FLAG)
    {
        // Throw exception if the original ID exceeded 29 bits
        if ((id & ~EFF_MASK) != 0) {
           throw IDException(fmt::format("ID value 0x{:x} exceeds 29-bit limit for Extended CAN ID (max: 0x{:x})", id, EFF_MASK));
        }
    }

    // Template version for compile-time checking with literals
    template <uint32_t id>
    static ExtendedID create()
    {
        static_assert((id & ~EFF_MASK) == 0,
                      "Extended CAN ID exceeds 29-bit limit (0x1FFFFFFF)");
        return ExtendedID(id);
    }

    ~ExtendedID() override = default;

    [[nodiscard]] uint32_t justEFF() const
    {
        return get() & EFF_MASK;
    }
};

// CAN frame is always 16 bytes on linux,
// regardless of the actual data payload size.
// Bytes 0-3  : CAN ID
// Byte  4    : Frame payload length in bytes
// Bytes 5-7  : Reserved
// Bytes 8-15 : Frame payload
class Frame final {
    public:
	uint32_t id;
	uint8_t dlc;
	std::array<uint8_t, 8> data;
	std::optional<timespec> hw_timestamp;

	// Default constructor (needed for containers)
	Frame() : id(0), dlc(0), data{}, hw_timestamp(std::nullopt) {}

	Frame(const uint32_t id, const uint8_t dlc, const std::array<uint8_t, 8>& data)
		: id(id), dlc(dlc), data(data), hw_timestamp(std::nullopt)
	{
		validateFrame();
	}

	Frame(const StandardID& id, const uint8_t dlc, const std::array<uint8_t, 8>& data)
		: id(id.get()), dlc(dlc), data(data), hw_timestamp(std::nullopt)
	{
		validateFrame();
	}

	Frame(const ExtendedID& id, const uint8_t dlc, const std::array<uint8_t, 8>& data)
		: id(id.get()), dlc(dlc), data(data), hw_timestamp(std::nullopt)
	{
		validateFrame();
	}

	Frame(const uint32_t id, const uint8_t dlc, const std::array<uint8_t, 8>& data, const timespec& timestamp)
		: id(id), dlc(dlc), data(data), hw_timestamp(timestamp)
	{
		validateFrame();
	}

	// Copy constructor
	Frame(const Frame& other) = default;

	// Move constructor
	Frame(Frame&& other) noexcept = default;

	// Copy assignment
	Frame& operator=(const Frame& other) = default;

	// Move assignment
	Frame& operator=(Frame&& other) noexcept = default;

	~Frame() = default;

private:
	void validateFrame() const {
		if (dlc > 8) {
			throw FrameException("DLC cannot exceed 8 bytes for standard CAN frames");
		}
	}

public:

	[[nodiscard]] uint32_t getCANID() const noexcept
	{
		// check if EFF flag is set
		if (id & EFF_FLAG) {
			return id & EFF_MASK;
		}
		// otherwise, it's a standard frame
		return id & SFF_MASK;
	}

	[[nodiscard]] bool isExtended() const noexcept {
		return (id & EFF_FLAG) != 0;
	}

	[[nodiscard]] bool isRTR() const noexcept {
		return (id & RTR_FLAG) != 0;
	}

	[[nodiscard]] bool isError() const noexcept {
		return (id & ERR_FLAG) != 0;
	}

	// Stream output operator for debugging
	friend std::ostream& operator<<(std::ostream& os, const Frame& frame) {
		os << "CAN Frame [ID: 0x" << std::hex << std::uppercase;
		if (frame.isExtended()) {
			os << std::setfill('0') << std::setw(8);
		}
		os << frame.getCANID() << std::nouppercase << std::dec
		   << ", DLC: " << static_cast<int>(frame.dlc)
		   << ", Data: [";
		for (size_t i = 0; i < frame.dlc && i < 8; ++i) {
			if (i > 0) os << ", ";
			os << "0x" << std::hex << std::setfill('0') << std::setw(2) 
			   << static_cast<int>(frame.data[i]) << std::dec;
		}
		os << "]";
		if (frame.isExtended()) os << " (Extended)";
		if (frame.isRTR()) os << " (RTR)";
		if (frame.isError()) os << " (Error)";
		return os;
	}
};

// Frame Builder Pattern for convenient frame construction
class FrameBuilder {
private:
    uint32_t id = 0;
    uint8_t dlc = 0;
    std::array<uint8_t, 8> data{};
    std::optional<timespec> timestamp;

public:
    FrameBuilder& setID(uint32_t id) {
        this->id = id;
        return *this;
    }

    FrameBuilder& setID(const StandardID& id) {
        this->id = id.get();
        return *this;
    }

    FrameBuilder& setID(const ExtendedID& id) {
        this->id = id.get();
        return *this;
    }

    FrameBuilder& setDLC(uint8_t dlc) {
        this->dlc = dlc;
        return *this;
    }

    FrameBuilder& setData(const std::array<uint8_t, 8>& data) {
        this->data = data;
        if (dlc == 0) {
            dlc = 8;  // When setting a full array, assume DLC of 8
        }
        return *this;
    }

    FrameBuilder& setData(const std::vector<uint8_t>& data) {
        std::fill(this->data.begin(), this->data.end(), 0);
        size_t copy_len = std::min(data.size(), size_t(8));
        std::copy(data.begin(), data.begin() + copy_len, this->data.begin());
        if (dlc == 0) {
            dlc = static_cast<uint8_t>(copy_len);
        }
        return *this;
    }

    FrameBuilder& setData(const std::initializer_list<uint8_t>& data) {
        std::fill(this->data.begin(), this->data.end(), 0);
        size_t copy_len = std::min(data.size(), size_t(8));
        std::copy(data.begin(), data.begin() + copy_len, this->data.begin());
        if (dlc == 0) {
            dlc = static_cast<uint8_t>(copy_len);
        }
        return *this;
    }

    FrameBuilder& setTimestamp(const timespec& timestamp) {
        this->timestamp = timestamp;
        return *this;
    }

    [[nodiscard]] Frame build() const {
        if (timestamp) {
            return Frame(id, dlc, data, *timestamp);
        } else {
            return Frame(id, dlc, data);
        }
    }

    // Convenience method for creating a frame with automatic DLC
    [[nodiscard]] Frame buildAuto() const {
        uint8_t actual_dlc = dlc;
        if (actual_dlc == 0) {
            // Calculate DLC based on non-zero data
            actual_dlc = 8;
            for (int i = 7; i >= 0; --i) {
                if (data[i] != 0) {
                    actual_dlc = i + 1;
                    break;
                }
            }
        }
        
        if (timestamp) {
            return Frame(id, actual_dlc, data, *timestamp);
        } else {
            return Frame(id, actual_dlc, data);
        }
    }
};

// Forward declaration of the Socket class
class Socket;

namespace FD {
    // CAN FD frame with up to 64 bytes of data
    class Frame final {
    public:
        uint32_t id;
        uint8_t len;         // Frame payload length in bytes (0-64)
        uint8_t flags;       // Additional flags for CAN FD (BRS, ESI, FDF)
        std::array<uint8_t, FD::MAX_DLEN> data;
        std::optional<timespec> hw_timestamp;
        
        // Static helper method to convert a standard CAN frame to CAN FD frame
        static Frame fromCANFrame(const CAN::Frame& canFrame, const bool withBitRateSwitch = false) {
            // Create a new CAN FD frame with the same ID and data
            std::array<uint8_t, FD::MAX_DLEN> fdData{};
            
            // Copy data from CAN frame (maximum 8 bytes)
            for (int i = 0; i < canFrame.dlc && i < 8; i++) {
                fdData[i] = canFrame.data[i];
            }
            
            // Create FD frame with appropriate flags
            const uint8_t fdFlags = withBitRateSwitch ? 
                FD::FDF_FLAG | FD::BRS_FLAG : 
                FD::FDF_FLAG;
            
            // Preserve any hardware timestamp
            if (canFrame.hw_timestamp) {
                return Frame(canFrame.id, canFrame.dlc, fdData, *canFrame.hw_timestamp, fdFlags);
            } else {
                return Frame(canFrame.id, canFrame.dlc, fdData, fdFlags);
            }
        }

        // Constructor with array data
        Frame(const uint32_t id, const uint8_t len, const std::array<uint8_t, FD::MAX_DLEN>& data, const uint8_t flags = FD::FDF_FLAG)
            : id(id), len(len), flags(flags), data(data), hw_timestamp(std::nullopt)
        {
            // Always set the FDF flag to indicate a CAN FD frame
            this->flags |= FD::FDF_FLAG;
            validateFrame();
        }

        // Constructors with ID objects
        Frame(const StandardID& id, const uint8_t len, const std::array<uint8_t, FD::MAX_DLEN>& data, const uint8_t flags = FD::FDF_FLAG)
            : id(id.get()), len(len), flags(flags), data(data), hw_timestamp(std::nullopt)
        {
            this->flags |= FD::FDF_FLAG;
            validateFrame();
        }

        Frame(const ExtendedID& id, const uint8_t len, const std::array<uint8_t, FD::MAX_DLEN>& data, const uint8_t flags = FD::FDF_FLAG)
            : id(id.get()), len(len), flags(flags), data(data), hw_timestamp(std::nullopt)
        {
            this->flags |= FD::FDF_FLAG;
            validateFrame();
        }

        // Constructor with timestamp
        Frame(const uint32_t id, const uint8_t len, const std::array<uint8_t, FD::MAX_DLEN>& data, 
              const timespec& timestamp, const uint8_t flags = FD::FDF_FLAG)
            : id(id), len(len), flags(flags), data(data), hw_timestamp(timestamp)
        {
            this->flags |= FD::FDF_FLAG;
            validateFrame();
        }

        // Copy constructor
        Frame(const Frame& other) = default;

        // Move constructor
        Frame(Frame&& other) noexcept = default;

        // Copy assignment
        Frame& operator=(const Frame& other) = default;

        // Move assignment
        Frame& operator=(Frame&& other) noexcept = default;

        ~Frame() = default;

    private:
        void validateFrame() const {
            if (len > FD::MAX_DLEN) {
                throw FrameException("Frame length cannot exceed 64 bytes for CAN FD frames");
            }
        }

    public:

        // Vector data constructor for flexible length input
        Frame(const uint32_t id, const std::vector<uint8_t>& vec_data, const uint8_t flags = FD::FDF_FLAG)
            : id(id), len(static_cast<uint8_t>(vec_data.size())), flags(flags), hw_timestamp(std::nullopt)
        {
            this->flags |= FD::FDF_FLAG;
            
            // Copy data from vector to array
            std::fill(data.begin(), data.end(), 0); // Initialize with zeros
            std::copy(vec_data.begin(), vec_data.end(), data.begin());
            validateFrame();
        }

        [[nodiscard]] uint32_t getCANID() const noexcept
        {
            // Check if EFF flag is set
            if (id & EFF_FLAG) {
                return id & EFF_MASK;
            }
            // Otherwise, it's a standard frame
            return id & SFF_MASK;
        }

        [[nodiscard]] bool isExtended() const noexcept {
            return (id & EFF_FLAG) != 0;
        }

        [[nodiscard]] bool isRTR() const noexcept {
            return (id & RTR_FLAG) != 0;
        }

        [[nodiscard]] bool isError() const noexcept {
            return (id & ERR_FLAG) != 0;
        }

        // Set/get flags related methods
        void setBRS(bool enable)
        {
            if (enable) {
                flags |= FD::BRS_FLAG;
            } else {
                flags &= ~FD::BRS_FLAG;
            }
        }

        [[nodiscard]] bool hasBRS() const noexcept
        {
            return (flags & FD::BRS_FLAG) != 0;
        }

        void setESI(bool enable)
        {
            if (enable) {
                flags |= FD::ESI_FLAG;
            } else {
                flags &= ~FD::ESI_FLAG;
            }
        }

        [[nodiscard]] bool hasESI() const noexcept
        {
            return (flags & FD::ESI_FLAG) != 0;
        }

        // Stream output operator for debugging
        friend std::ostream& operator<<(std::ostream& os, const Frame& frame) {
            os << "CAN FD Frame [ID: 0x" << std::hex << frame.getCANID() << std::dec
               << ", Length: " << static_cast<int>(frame.len)
               << ", Data: [";
            for (size_t i = 0; i < frame.len && i < FD::MAX_DLEN; ++i) {
                if (i > 0) os << ", ";
                os << "0x" << std::hex << std::setfill('0') << std::setw(2) 
                   << static_cast<int>(frame.data[i]) << std::dec;
            }
            os << "], Flags: [";
            if (frame.hasBRS()) os << "BRS ";
            if (frame.hasESI()) os << "ESI ";
            os << "FDF]";
            if (frame.isExtended()) os << " (Extended)";
            if (frame.isRTR()) os << " (RTR)";
            if (frame.isError()) os << " (Error)";
            return os;
        }
    };

    // CAN FD Frame Builder Pattern
    class FrameBuilder {
    private:
        uint32_t id = 0;
        uint8_t len = 0;
        uint8_t flags = FD::FDF_FLAG;
        std::array<uint8_t, FD::MAX_DLEN> data{};
        std::optional<timespec> timestamp;

    public:
        FrameBuilder& setID(uint32_t id) {
            this->id = id;
            return *this;
        }

        FrameBuilder& setID(const StandardID& id) {
            this->id = id.get();
            return *this;
        }

        FrameBuilder& setID(const ExtendedID& id) {
            this->id = id.get();
            return *this;
        }

        FrameBuilder& setLength(uint8_t len) {
            this->len = len;
            return *this;
        }

        FrameBuilder& setData(const std::array<uint8_t, FD::MAX_DLEN>& data) {
            this->data = data;
            return *this;
        }

        FrameBuilder& setData(const std::vector<uint8_t>& data) {
            std::fill(this->data.begin(), this->data.end(), 0);
            size_t copy_len = std::min(data.size(), size_t(FD::MAX_DLEN));
            std::copy(data.begin(), data.begin() + copy_len, this->data.begin());
            if (len == 0) {
                len = static_cast<uint8_t>(copy_len);
            }
            return *this;
        }

        FrameBuilder& setData(const std::initializer_list<uint8_t>& data) {
            std::fill(this->data.begin(), this->data.end(), 0);
            size_t copy_len = std::min(data.size(), size_t(FD::MAX_DLEN));
            std::copy(data.begin(), data.begin() + copy_len, this->data.begin());
            if (len == 0) {
                len = static_cast<uint8_t>(copy_len);
            }
            return *this;
        }

        FrameBuilder& enableBRS(bool enable = true) {
            if (enable) {
                flags |= FD::BRS_FLAG;
            } else {
                flags &= ~FD::BRS_FLAG;
            }
            return *this;
        }

        FrameBuilder& enableESI(bool enable = true) {
            if (enable) {
                flags |= FD::ESI_FLAG;
            } else {
                flags &= ~FD::ESI_FLAG;
            }
            return *this;
        }

        FrameBuilder& setTimestamp(const timespec& timestamp) {
            this->timestamp = timestamp;
            return *this;
        }

        [[nodiscard]] Frame build() const {
            if (timestamp) {
                return Frame(id, len, data, *timestamp, flags);
            } else {
                return Frame(id, len, data, flags);
            }
        }

        // Convenience method for creating a frame with automatic length
        [[nodiscard]] Frame buildAuto() const {
            uint8_t actual_len = len;
            if (actual_len == 0) {
                // Calculate length based on non-zero data
                actual_len = FD::MAX_DLEN;
                for (int i = FD::MAX_DLEN - 1; i >= 0; --i) {
                    if (data[i] != 0) {
                        actual_len = i + 1;
                        break;
                    }
                }
            }
            
            if (timestamp) {
                return Frame(id, actual_len, data, *timestamp, flags);
            } else {
                return Frame(id, actual_len, data, flags);
            }
        }
    };
} // namespace FD

// Unified Socket class that can handle both CAN and CAN FD frames
class Socket {
    int sock;
    ProtocolMode mode;

private:
    // Helper function to wait for socket readability with timeout
    [[nodiscard]] bool waitForRead(std::optional<std::chrono::milliseconds> timeout) const {
        if (!timeout.has_value()) {
            return true; // No timeout specified, rely on socket timeout
        }
        
        fd_set readfds;
        FD_ZERO(&readfds);
        FD_SET(sock, &readfds);
        
        timeval tv;
        tv.tv_sec = timeout->count() / 1000;
        tv.tv_usec = (timeout->count() % 1000) * 1000;
        
        int result = select(sock + 1, &readfds, nullptr, nullptr, &tv);
        if (result == -1) {
            throw SocketException(fmt::format("Select failed during read timeout: {}", strerror(errno)));
        }
        
        return result > 0; // Returns true if data is available
    }
    
    // Helper function to wait for socket writability with timeout
    [[nodiscard]] bool waitForWrite(std::optional<std::chrono::milliseconds> timeout) const {
        if (!timeout.has_value()) {
            return true; // No timeout specified
        }
        
        fd_set writefds;
        FD_ZERO(&writefds);
        FD_SET(sock, &writefds);
        
        timeval tv;
        tv.tv_sec = timeout->count() / 1000;
        tv.tv_usec = (timeout->count() % 1000) * 1000;
        
        int result = select(sock + 1, nullptr, &writefds, nullptr, &tv);
        if (result == -1) {
            throw SocketException(fmt::format("Select failed during write timeout: {}", strerror(errno)));
        }
        
        return result > 0; // Returns true if socket is writable
    }
    
    // Helper to temporarily set socket to non-blocking mode
    class NonBlockingGuard {
        int sock;
        int original_flags;
        bool was_changed;
        
    public:
        explicit NonBlockingGuard(int socket_fd) : sock(socket_fd), was_changed(false) {
            original_flags = fcntl(sock, F_GETFL, 0);
            if (original_flags == -1) {
                throw SocketException(fmt::format("Failed to get socket flags: {}", strerror(errno)));
            }
            
            if (!(original_flags & O_NONBLOCK)) {
                if (fcntl(sock, F_SETFL, original_flags | O_NONBLOCK) == -1) {
                    throw SocketException(fmt::format("Failed to set non-blocking mode: {}", strerror(errno)));
                }
                was_changed = true;
            }
        }
        
        ~NonBlockingGuard() {
            if (was_changed) {
                fcntl(sock, F_SETFL, original_flags); // Restore original flags
            }
        }
        
        // Delete copy constructor and assignment
        NonBlockingGuard(const NonBlockingGuard&) = delete;
        NonBlockingGuard& operator=(const NonBlockingGuard&) = delete;
    };

public:
    explicit Socket(const std::string& interface,
                    const ProtocolMode mode = ProtocolMode::ClassicCAN,
                    const std::chrono::milliseconds& timeout = std::chrono::milliseconds(100),
                    const int rcvbuf_size = 327680)
        : mode(mode)
    {
        this->sock = socket(PF_CAN, SOCK_RAW, CAN_RAW);
        if (this->sock == -1) {
            throw SocketException(fmt::format("Failed to create CAN socket: {}", strerror(errno)));
        }

        // Enable CAN FD frames if requested
        if (mode == ProtocolMode::CanFD) {
            try {
                setSocketOption(SOL_CAN_RAW, CAN_RAW_FD_FRAMES, 1, "Failed to enable CAN FD frames");
            } catch (const std::exception& e) {
                ::close(this->sock);
                throw SocketException(fmt::format("Failed to enable CAN FD frames: {}", e.what()));
            }
        }

        // Get interface
        ifreq ifr{};
        strncpy(ifr.ifr_name, interface.c_str(), IFNAMSIZ - 1);
        // Ensure null termination
        ifr.ifr_name[IFNAMSIZ - 1] = '\0';
        if (ioctl(this->sock, SIOCGIFINDEX, &ifr) == -1) {
            ::close(this->sock);
            throw SocketException(fmt::format("Failed to get CAN interface '{}': {}", interface, strerror(errno)));
        }

        // Set timeout
        timeval tv{};
        tv.tv_sec = timeout.count() / 1000;
        tv.tv_usec = (timeout.count() % 1000) * 1000;
        if (setsockopt(this->sock, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)) == -1) {
            ::close(this->sock);
            throw SocketException(fmt::format("Failed to set CAN socket timeout: {}", strerror(errno)));
        }

        // Set address
        sockaddr_can addr{};
        memset(&addr, 0, sizeof(addr));
        addr.can_family = AF_CAN;
        addr.can_ifindex = ifr.ifr_ifindex;
        // Bind
        if (bind(this->sock, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) == -1) {
            ::close(this->sock);
            throw SocketException(fmt::format("Failed to bind CAN socket to interface '{}': {}", interface, strerror(errno)));
        }

        // Set receive buffer size if specified
        if (rcvbuf_size > 0) {
            try {
                setSocketOption(SOL_SOCKET, SO_RCVBUF, rcvbuf_size, "Failed to set socket receive buffer size");
            } catch (const std::exception& e) {
                ::close(this->sock);
                throw SocketException(fmt::format("Failed to set socket receive buffer size: {}", e.what()));
            }
        }
    }

    // Disable copy construction and assignment
    Socket(const Socket&) = delete;
    Socket& operator=(const Socket&) = delete;

    // Enable move construction and assignment
    Socket(Socket&& other) noexcept : sock(other.sock), mode(other.mode) {
        other.sock = -1;
    }

    Socket& operator=(Socket&& other) noexcept {
        if (this != &other) {
            close();
            sock = other.sock;
            mode = other.mode;
            other.sock = -1;
        }
        return *this;
    }

    ~Socket()
    {
        close();
    }

    // Get the current mode (CAN or CAN FD)
    [[nodiscard]] ProtocolMode getMode() const
    {
        return mode;
    }

    // Generic helper for setting socket options
    template<typename T>
    void setSocketOption(int level, int option_name, const T& value, const std::string& error_message) const {
        if (!isOpen()) {
            throw SocketException("Socket is not open");
        }
        
        if (setsockopt(this->sock, level, option_name, &value, sizeof(value)) == -1) {
            throw SocketException(fmt::format("{}: {}", error_message, strerror(errno)));
        }
    }
    
    // Generic helper for getting socket options
    template<typename T>
    T getSocketOption(int level, int option_name, const std::string& error_message) const {
        if (!isOpen()) {
            throw SocketException("Socket is not open");
        }
        
        T value;
        socklen_t len = sizeof(value);
        if (getsockopt(this->sock, level, option_name, &value, &len) == -1) {
            throw SocketException(fmt::format("{}: {}", error_message, strerror(errno)));
        }
        return value;
    }

    // Set the socket receive buffer size
    int setReceiveBufferSize(const int size) const
    {
        if (size <= 0) {
            throw std::invalid_argument("Buffer size must be positive");
        }

        setSocketOption(SOL_SOCKET, SO_RCVBUF, size, "Failed to set socket receive buffer size");

        // Verify the buffer size was set
        int actual_size = getReceiveBufferSize();
        // Linux doubles the value we set, so divide by 2 for comparison
        if (actual_size / 2 < size) {
            throw std::runtime_error(fmt::format(
                "Failed to set receive buffer size to {}, actual size is {}", size, actual_size / 2));
        }

        return actual_size / 2; // Return the actual size (accounting for Linux doubling)
    }

    // Get the current socket receive buffer size
    [[nodiscard]] int getReceiveBufferSize() const
    {
        return getSocketOption<int>(SOL_SOCKET, SO_RCVBUF, "Failed to get socket receive buffer size");
        // Note: This is doubled by Linux
    }

    [[nodiscard]] int getFd() const noexcept
    {
        return this->sock;
    }

    void close() noexcept
    {
        if (isOpen() == false) {
            return;
        }
        ::close(this->sock);
        this->sock = -1;
    }

    [[nodiscard]] bool isOpen() const noexcept
    {
        return this->sock != -1;
    }

    [[nodiscard]] int setFilter(const uint32_t id, const uint32_t mask) const
    {
        can_filter filter[1];
        filter[0].can_id = id;
        filter[0].can_mask = mask;
        
        // Using plain setsockopt because the templated version expects a reference to a single value
        if (!isOpen()) {
            throw SocketException("Socket is not open");
        }
        
        if (setsockopt(this->sock, SOL_CAN_RAW, CAN_RAW_FILTER, &filter, sizeof(filter)) == -1) {
            throw FilterException(fmt::format("Failed to set CAN filter (ID: 0x{:x}, Mask: 0x{:x}): {}", id, mask, strerror(errno)));
        }
        
        return 0;
    }

    [[nodiscard]] int setFilter(const StandardID& id, const uint32_t mask) const
    {
        return setFilter(id.get(), mask);
    }

    [[nodiscard]] int setFilter(const ExtendedID& id, const uint32_t mask) const
    {
        return setFilter(id.get(), mask);
    }

    // Base implementation for setting CAN filters
    [[nodiscard]] int setFilters(const std::vector<std::pair<uint32_t, uint32_t>>& filters) const
    {
        // Create the filter array
        std::vector<can_filter> can_filters;
        can_filters.reserve(filters.size());
        for (const auto& [fst, snd] : filters) {
            can_filter filter{};
            filter.can_id = fst;
            filter.can_mask = snd;
            can_filters.push_back(filter);
        }
        
        // Set the filters
        if (!isOpen()) {
            throw SocketException("Socket is not open");
        }
        
        if (setsockopt(this->sock, SOL_CAN_RAW, CAN_RAW_FILTER, can_filters.data(),
                    can_filters.size() * sizeof(can_filter)) == -1) {
            throw FilterException(fmt::format("Failed to set {} CAN filters: {}", filters.size(), strerror(errno)));
        }
        
        return 0;
    }

    // Template function to handle any ID type that has a get() method
    template<typename T>
    [[nodiscard]] int setFilters(const std::vector<std::pair<T, uint32_t>>& filters) const
    {
        std::vector<std::pair<uint32_t, uint32_t>> can_filters;
        can_filters.reserve(filters.size());
        for (const auto& [fst, snd] : filters) {
            can_filters.emplace_back(fst.get(), snd);
        }
        return setFilters(can_filters);
    }

    // Write a single standard CAN frame
    [[nodiscard]] ssize_t write(const Frame& frame, std::optional<std::chrono::milliseconds> timeout = std::nullopt) const
    {
        // Check if socket is writable within timeout
        if (!waitForWrite(timeout)) {
            throw SocketException("Write operation timed out");
        }
        // For CAN FD sockets, automatically convert CAN frame to CAN FD frame
        if (mode == ProtocolMode::CanFD) {
            // Convert the standard CAN frame to CAN FD format and send it
            const auto fdFrame = FD::Frame::fromCANFrame(frame);
            return write(fdFrame);
        }

        // Standard CAN socket
        can_frame cf{};
        cf.can_id = frame.id;
        cf.can_dlc = frame.dlc;
        for (int i = 0; i < frame.dlc; i++) {
            cf.data[i] = frame.data[i];
        }
        return ::write(this->sock, &cf, sizeof(can_frame));
    }

    // Write a single CAN FD frame
    [[nodiscard]] ssize_t write(const FD::Frame& frame, std::optional<std::chrono::milliseconds> timeout = std::nullopt) const
    {
        // Check if socket is writable within timeout
        if (!waitForWrite(timeout)) {
            throw SocketException("Write operation timed out");
        }
        if (mode != ProtocolMode::CanFD) {
            throw SocketException("Cannot send a CAN FD frame on a standard CAN socket");
        }

        canfd_frame cf{};
        cf.can_id = frame.id;
        cf.len = frame.len;
        cf.flags = frame.flags;
        
        // Copy data from Frame to canfd_frame
        for (int i = 0; i < frame.len && i < FD::MAX_DLEN; i++) {
            cf.data[i] = frame.data[i];
        }
        
        return ::write(this->sock, &cf, sizeof(canfd_frame));
    }

    // Write multiple standard CAN frames
    [[nodiscard]] ssize_t write(const std::vector<Frame>& frames, std::optional<std::chrono::milliseconds> timeout = std::nullopt) const
    {
        ssize_t nbytes = 0;
        for (const auto& frame : frames) {
            nbytes += write(frame, timeout);
        }
        return nbytes;
    }
    
    // Write multiple standard CAN frames from a unique_ptr
    [[nodiscard]] ssize_t write(const std::unique_ptr<std::vector<Frame>>& frames, std::optional<std::chrono::milliseconds> timeout = std::nullopt) const
    {
        ssize_t nbytes = 0;
        for (const auto& frame : *frames) {
            nbytes += write(frame, timeout);
        }
        return nbytes;
    }
    
    // Write multiple standard CAN frames from a queue
    [[nodiscard]] ssize_t write(std::queue<Frame>& frames, std::optional<std::chrono::milliseconds> timeout = std::nullopt) const
    {
        ssize_t nbytes = 0;
        while (!frames.empty()) {
            nbytes += write(frames.front(), timeout);
            frames.pop();
        }
        return nbytes;
    }

    // Write multiple CAN FD frames
    [[nodiscard]] ssize_t write(const std::vector<FD::Frame>& frames, std::optional<std::chrono::milliseconds> timeout = std::nullopt) const
    {
        ssize_t nbytes = 0;
        for (const auto& frame : frames) {
            nbytes += write(frame, timeout);
        }
        return nbytes;
    }

    // Read a CAN frame (either standard or FD based on socket mode)
    [[nodiscard]] auto read(std::optional<std::chrono::milliseconds> timeout = std::nullopt) const -> std::variant<std::optional<std::unique_ptr<Frame>>, std::optional<std::unique_ptr<FD::Frame>>>
    {
        // Check if data is available within timeout
        if (!waitForRead(timeout)) {
            // Timeout occurred
            if (mode == ProtocolMode::ClassicCAN) {
                return std::optional<std::unique_ptr<Frame>>(std::nullopt);
            } else {
                return std::optional<std::unique_ptr<FD::Frame>>(std::nullopt);
            }
        }
        if (mode == ProtocolMode::ClassicCAN) {
            can_frame frame{};
            ssize_t nbytes = ::read(this->sock, &frame, sizeof(can_frame));
            
            if (nbytes == -1) {
                if (errno == EAGAIN || errno == EWOULDBLOCK) {
                    return std::optional<std::unique_ptr<Frame>>(std::nullopt);
                }
                throw SocketException(fmt::format("Failed to read CAN frame: {}", strerror(errno)));
            }
            
            if (nbytes == 0) {
                return std::optional<std::unique_ptr<Frame>>(std::nullopt);
            }
            
            std::array<uint8_t, 8> data{};
            for (int i = 0; i < frame.can_dlc; i++) {
                data[i] = frame.data[i];
            }
            
            auto f = std::make_unique<Frame>(frame.can_id, frame.can_dlc, data);
            return std::optional<std::unique_ptr<Frame>>(std::move(f));
        } else {
            // CAN FD mode
            canfd_frame frame{};
            ssize_t nbytes = ::read(this->sock, &frame, sizeof(canfd_frame));
            
            if (nbytes == -1) {
                if (errno == EAGAIN || errno == EWOULDBLOCK) {
                    return std::optional<std::unique_ptr<FD::Frame>>(std::nullopt);
                }
                throw SocketException(fmt::format("Failed to read CAN FD frame: {}", strerror(errno)));
            }
            
            if (nbytes == 0) {
                return std::optional<std::unique_ptr<FD::Frame>>(std::nullopt);
            }
            
            std::array<uint8_t, FD::MAX_DLEN> data{};
            for (int i = 0; i < frame.len && i < FD::MAX_DLEN; i++) {
                data[i] = frame.data[i];
            }
            
            auto f = std::make_unique<FD::Frame>(frame.can_id, frame.len, data, frame.flags);
            return std::optional<std::unique_ptr<FD::Frame>>(std::move(f));
        }
    }
    
    // Read multiple frames over a duration
    [[nodiscard]] auto read(const std::chrono::milliseconds& duration) const 
        -> std::variant<std::vector<Frame>, std::vector<FD::Frame>>
    {
        if (mode == ProtocolMode::ClassicCAN) {
            std::vector<Frame> frames;
            const auto start = std::chrono::steady_clock::now();
            
            while (std::chrono::steady_clock::now() - start < duration) {
                try {
                    const auto frame_variant = read();
                    if (const auto* frame_opt = std::get_if<std::optional<std::unique_ptr<Frame>>>(&frame_variant)) {
                        if (frame_opt->has_value()) {
                            frames.push_back(*std::move(frame_opt->value()));
                        }
                    }
                } catch (const std::runtime_error& err) {
                    // Ignore errors during the read duration
                }
            }
            
            return frames;
        } else {
            // CAN FD mode
            std::vector<FD::Frame> frames;
            const auto start = std::chrono::steady_clock::now();
            
            while (std::chrono::steady_clock::now() - start < duration) {
                try {
                    const auto frame_variant = read();
                    if (const auto* frame_opt = std::get_if<std::optional<std::unique_ptr<FD::Frame>>>(&frame_variant)) {
                        if (frame_opt->has_value()) {
                            frames.push_back(*std::move(frame_opt->value()));
                        }
                    }
                } catch (const std::runtime_error& err) {
                    // Ignore errors during the read duration
                }
            }
            
            return frames;
        }
    }
    
    // Read a CAN frame with hardware timestamp using recvmsg
    [[nodiscard]] auto readMsg(std::optional<std::chrono::milliseconds> timeout = std::nullopt) const 
        -> std::variant<std::optional<std::unique_ptr<Frame>>, std::optional<std::unique_ptr<FD::Frame>>>
    {
        // Check if data is available within timeout
        if (!waitForRead(timeout)) {
            // Timeout occurred
            if (mode == ProtocolMode::ClassicCAN) {
                return std::optional<std::unique_ptr<Frame>>(std::nullopt);
            } else {
                return std::optional<std::unique_ptr<FD::Frame>>(std::nullopt);
            }
        }
        if (mode == ProtocolMode::ClassicCAN) {
            can_frame frame{};
            char control[CMSG_SPACE(sizeof(timespec))];
            iovec iov = { .iov_base = &frame, .iov_len = sizeof(frame) };
            msghdr msg = {};
            msg.msg_iov = &iov;
            msg.msg_iovlen = 1;
            msg.msg_control = control;
            msg.msg_controllen = sizeof(control);

            ssize_t nbytes = recvmsg(this->sock, &msg, 0);
            if (nbytes == -1) {
                if (errno == EAGAIN || errno == EWOULDBLOCK) {
                    return std::optional<std::unique_ptr<Frame>>(std::nullopt);
                }
                throw SocketException(fmt::format("Failed to read CAN frame with timestamp: {}", strerror(errno)));
            }

            timespec hw_time {};
            cmsghdr* cmsg = CMSG_FIRSTHDR(&msg);
            while (cmsg != nullptr) {
                if (cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SO_TIMESTAMPING) {
                    hw_time = *reinterpret_cast<timespec*>(CMSG_DATA(cmsg));
                    break;
                }
                cmsg = CMSG_NXTHDR(&msg, cmsg);
            }

            std::array<uint8_t, 8> data{};
            for (int i = 0; i < frame.can_dlc; i++) {
                data[i] = frame.data[i];
            }
            
            auto f = std::make_unique<Frame>(frame.can_id, frame.can_dlc, data, hw_time);
            return std::optional<std::unique_ptr<Frame>>(std::move(f));
        } else {
            // CAN FD mode
            canfd_frame frame{};
            char control[CMSG_SPACE(sizeof(timespec))];
            iovec iov = { .iov_base = &frame, .iov_len = sizeof(frame) };
            msghdr msg = {};
            msg.msg_iov = &iov;
            msg.msg_iovlen = 1;
            msg.msg_control = control;
            msg.msg_controllen = sizeof(control);

            ssize_t nbytes = recvmsg(this->sock, &msg, 0);
            if (nbytes == -1) {
                if (errno == EAGAIN || errno == EWOULDBLOCK) {
                    return std::optional<std::unique_ptr<FD::Frame>>(std::nullopt);
                }
                throw SocketException(fmt::format("Failed to read CAN FD frame with timestamp: {}", strerror(errno)));
            }

            timespec hw_time {};
            cmsghdr* cmsg = CMSG_FIRSTHDR(&msg);
            while (cmsg != nullptr) {
                if (cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SO_TIMESTAMPING) {
                    hw_time = *reinterpret_cast<timespec*>(CMSG_DATA(cmsg));
                    break;
                }
                cmsg = CMSG_NXTHDR(&msg, cmsg);
            }

            std::array<uint8_t, FD::MAX_DLEN> data{};
            for (int i = 0; i < frame.len && i < FD::MAX_DLEN; i++) {
                data[i] = frame.data[i];
            }
            
            auto f = std::make_unique<FD::Frame>(frame.can_id, frame.len, data, hw_time, frame.flags);
            return std::optional<std::unique_ptr<FD::Frame>>(std::move(f));
        }
    }

    // Enable non-blocking mode
    void setNonBlocking() const
    {
        const int flags = fcntl(this->sock, F_GETFL, 0);
        if (flags == -1) {
            throw SocketException(fmt::format("Failed to get socket flags: {}", strerror(errno)));
        }
        if (fcntl(this->sock, F_SETFL, flags | O_NONBLOCK) == -1) {
            throw SocketException(fmt::format("Failed to set socket to non-blocking mode: {}", strerror(errno)));
        }
    }

    // Disable non-blocking mode
    void setBlocking() const
    {
        const int flags = fcntl(this->sock, F_GETFL, 0);
        if (flags == -1) {
            throw SocketException(fmt::format("Failed to get socket flags: {}", strerror(errno)));
        }
        if (fcntl(this->sock, F_SETFL, flags & ~O_NONBLOCK) == -1) {
            throw SocketException(fmt::format("Failed to set socket to blocking mode: {}", strerror(errno)));
        }
    }

    // Convenience methods for creating and sending frames
    [[nodiscard]] ssize_t sendFrame(uint32_t id, const std::initializer_list<uint8_t>& data, std::optional<std::chrono::milliseconds> timeout = std::nullopt) const {
        std::array<uint8_t, 8> frame_data{};
        std::copy(data.begin(), data.end(), frame_data.begin());
        Frame frame(id, static_cast<uint8_t>(data.size()), frame_data);
        return write(frame, timeout);
    }

    [[nodiscard]] ssize_t sendFrame(const StandardID& id, const std::initializer_list<uint8_t>& data, std::optional<std::chrono::milliseconds> timeout = std::nullopt) const {
        std::array<uint8_t, 8> frame_data{};
        std::copy(data.begin(), data.end(), frame_data.begin());
        Frame frame(id, static_cast<uint8_t>(data.size()), frame_data);
        return write(frame, timeout);
    }

    [[nodiscard]] ssize_t sendFrame(const ExtendedID& id, const std::initializer_list<uint8_t>& data, std::optional<std::chrono::milliseconds> timeout = std::nullopt) const {
        std::array<uint8_t, 8> frame_data{};
        std::copy(data.begin(), data.end(), frame_data.begin());
        Frame frame(id, static_cast<uint8_t>(data.size()), frame_data);
        return write(frame, timeout);
    }

    // Convenience methods for CAN FD frames
    [[nodiscard]] ssize_t sendFDFrame(uint32_t id, const std::initializer_list<uint8_t>& data, uint8_t flags = FD::FDF_FLAG, std::optional<std::chrono::milliseconds> timeout = std::nullopt) const {
        if (mode != ProtocolMode::CanFD) {
            throw SocketException("Cannot send CAN FD frame on standard CAN socket");
        }
        FD::Frame frame(id, std::vector<uint8_t>(data), flags);
        return write(frame, timeout);
    }

    [[nodiscard]] ssize_t sendFDFrame(const StandardID& id, const std::initializer_list<uint8_t>& data, uint8_t flags = FD::FDF_FLAG, std::optional<std::chrono::milliseconds> timeout = std::nullopt) const {
        if (mode != ProtocolMode::CanFD) {
            throw SocketException("Cannot send CAN FD frame on standard CAN socket");
        }
        FD::Frame frame(id.get(), std::vector<uint8_t>(data), flags);
        return write(frame, timeout);
    }

    [[nodiscard]] ssize_t sendFDFrame(const ExtendedID& id, const std::initializer_list<uint8_t>& data, uint8_t flags = FD::FDF_FLAG, std::optional<std::chrono::milliseconds> timeout = std::nullopt) const {
        if (mode != ProtocolMode::CanFD) {
            throw SocketException("Cannot send CAN FD frame on standard CAN socket");
        }
        FD::Frame frame(id.get(), std::vector<uint8_t>(data), flags);
        return write(frame, timeout);
    }

    // Convenience method for setting common filter presets
    [[nodiscard]] int setStandardFilter(uint32_t id) const {
        return setFilter(id, SFF_MASK);
    }

    [[nodiscard]] int setExtendedFilter(uint32_t id) const {
        return setFilter(id | EFF_FLAG, EFF_MASK | EFF_FLAG);
    }

    [[nodiscard]] int setIDRangeFilter(uint32_t min_id, uint32_t max_id) const {
        // Create a mask that covers the range
        uint32_t mask = 0;
        uint32_t diff = max_id ^ min_id;
        
        // Find the highest bit that differs
        for (int i = 31; i >= 0; --i) {
            if ((diff >> i) & 1) {
                mask = ~((1u << (i + 1)) - 1);
                break;
            }
        }
        
        return setFilter(min_id, mask);
    }

    // Convenience method for accepting all frames
    [[nodiscard]] int acceptAllFrames() const {
        return setFilter(0, 0);
    }

    // Utility method for getting socket statistics
    struct SocketStats {
        int receive_buffer_size;
        bool is_non_blocking;
        ProtocolMode mode;
        bool is_open;
    };

    [[nodiscard]] SocketStats getStats() const {
        SocketStats stats{};
        stats.receive_buffer_size = getReceiveBufferSize();
        stats.mode = getMode();
        stats.is_open = isOpen();
        
        // Check if socket is non-blocking
        if (isOpen()) {
            const int flags = fcntl(this->sock, F_GETFL, 0);
            stats.is_non_blocking = (flags & O_NONBLOCK) != 0;
        } else {
            stats.is_non_blocking = false;
        }
        
        return stats;
    }

    // High-performance batch operations
    [[nodiscard]] ssize_t writeBatch(const std::vector<Frame>& frames, const BatchContext& ctx = {}) const {
        if (frames.empty()) return 0;
        
        ssize_t total_bytes = 0;
        size_t processed = 0;
        
        // Process frames in batches to reduce system call overhead
        while (processed < frames.size()) {
            // Limit batch size to avoid EINVAL from writev (IOV_MAX is typically 1024)
            const size_t iov_max = 1024;  // Conservative limit for writev
            size_t max_batch = std::min(ctx.batch_size, iov_max);
            size_t batch_end = std::min(processed + max_batch, frames.size());
            
            // Use writev for batch writing when possible
            std::vector<iovec> iov_frames;
            iov_frames.reserve(batch_end - processed);
            
            for (size_t i = processed; i < batch_end; ++i) {
                // Convert CAN frame to raw format for zero-copy operation
                static thread_local std::array<can_frame, BatchContext::DEFAULT_BATCH_SIZE> raw_frames;
                
                if (i - processed < raw_frames.size()) {
                    can_frame& cf = raw_frames[i - processed];
                    memset(&cf, 0, sizeof(can_frame));
                    cf.can_id = frames[i].id;
                    cf.can_dlc = frames[i].dlc;
                    std::copy(frames[i].data.begin(), frames[i].data.begin() + frames[i].dlc, cf.data);
                    
                    iovec iov{};
                    iov.iov_base = &cf;
                    iov.iov_len = sizeof(can_frame);
                    iov_frames.push_back(iov);
                }
            }
            
            if (!iov_frames.empty()) {
                ssize_t bytes = writev(this->sock, iov_frames.data(), iov_frames.size());
                if (bytes > 0) {
                    total_bytes += bytes;
                } else if (bytes == -1) {
                    // Check for specific error conditions
                    if (errno == EAGAIN || errno == EWOULDBLOCK) {
                        // Non-blocking socket, no data written - continue
                    } else if (errno == EINVAL || errno == EMSGSIZE) {
                        // Try writing frames one by one if batch fails
                        for (const auto& iov : iov_frames) {
                            ssize_t single_bytes = ::write(this->sock, iov.iov_base, iov.iov_len);
                            if (single_bytes > 0) {
                                total_bytes += single_bytes;
                            }
                        }
                    } else {
                        // Other errors are fatal
                        throw SocketException(fmt::format("Failed to write CAN frames: {}", strerror(errno)));
                    }
                }
            }
            
            processed = batch_end;
        }
        
        return total_bytes;
    }

    [[nodiscard]] ssize_t writeBatch(const std::vector<FD::Frame>& frames, const BatchContext& ctx = {}) const {
        if (mode != ProtocolMode::CanFD) {
            throw SocketException("Cannot send CAN FD frames on standard CAN socket");
        }
        
        if (frames.empty()) return 0;
        
        ssize_t total_bytes = 0;
        size_t processed = 0;
        
        while (processed < frames.size()) {
            // Limit batch size to avoid EINVAL from writev (IOV_MAX is typically 1024)
            const size_t iov_max = 1024;  // Conservative limit for writev
            size_t max_batch = std::min(ctx.batch_size, iov_max);
            size_t batch_end = std::min(processed + max_batch, frames.size());
            
            std::vector<iovec> iov_frames;
            iov_frames.reserve(batch_end - processed);
            
            for (size_t i = processed; i < batch_end; ++i) {
                static thread_local std::array<canfd_frame, BatchContext::DEFAULT_BATCH_SIZE> raw_frames;
                
                if (i - processed < raw_frames.size()) {
                    canfd_frame& cf = raw_frames[i - processed];
                    memset(&cf, 0, sizeof(canfd_frame));
                    cf.can_id = frames[i].id;
                    cf.len = frames[i].len;
                    cf.flags = frames[i].flags;
                    std::copy(frames[i].data.begin(), frames[i].data.begin() + frames[i].len, cf.data);
                    
                    iovec iov{};
                    iov.iov_base = &cf;
                    iov.iov_len = sizeof(canfd_frame);
                    iov_frames.push_back(iov);
                }
            }
            
            if (!iov_frames.empty()) {
                ssize_t bytes = writev(this->sock, iov_frames.data(), iov_frames.size());
                if (bytes > 0) {
                    total_bytes += bytes;
                } else if (bytes == -1) {
                    // Check for specific error conditions
                    if (errno == EAGAIN || errno == EWOULDBLOCK) {
                        // Non-blocking socket, no data written - continue
                    } else if (errno == EINVAL || errno == EMSGSIZE) {
                        // Try writing frames one by one if batch fails
                        for (const auto& iov : iov_frames) {
                            ssize_t single_bytes = ::write(this->sock, iov.iov_base, iov.iov_len);
                            if (single_bytes > 0) {
                                total_bytes += single_bytes;
                            }
                        }
                    } else {
                        // Other errors are fatal
                        throw SocketException(fmt::format("Failed to write CAN frames: {}", strerror(errno)));
                    }
                }
            }
            
            processed = batch_end;
        }
        
        return total_bytes;
    }

    // High-performance batch read with pre-allocated buffer
    [[nodiscard]] size_t readBatch(std::vector<Frame>& frames, const BatchContext& ctx = {}) const {
        if (mode != ProtocolMode::ClassicCAN) {
            throw SocketException("readBatch for standard frames requires ClassicCAN mode");
        }
        
        frames.clear();
        frames.reserve(ctx.batch_size);
        
        auto start_time = std::chrono::steady_clock::now();
        size_t frames_read = 0;
        
        while (frames_read < ctx.batch_size) {
            auto elapsed = std::chrono::steady_clock::now() - start_time;
            if (elapsed > ctx.timeout) break;
            
            can_frame raw_frame{};
            ssize_t bytes = ::read(this->sock, &raw_frame, sizeof(can_frame));
            
            if (bytes == sizeof(can_frame)) {
                std::array<uint8_t, 8> data{};
                std::copy(raw_frame.data, raw_frame.data + raw_frame.can_dlc, data.begin());
                frames.emplace_back(raw_frame.can_id, raw_frame.can_dlc, data);
                ++frames_read;
            } else if (bytes == -1) {
                if (errno == EAGAIN || errno == EWOULDBLOCK) {
                    // No more data available
                    break;
                } else {
                    throw SocketException(fmt::format("Error reading batch: {}", strerror(errno)));
                }
            }
        }
        
        return frames_read;
    }

    [[nodiscard]] size_t readBatch(std::vector<FD::Frame>& frames, const BatchContext& ctx = {}) const {
        if (mode != ProtocolMode::CanFD) {
            throw SocketException("readBatch for FD frames requires CanFD mode");
        }
        
        frames.clear();
        frames.reserve(ctx.batch_size);
        
        auto start_time = std::chrono::steady_clock::now();
        size_t frames_read = 0;
        
        while (frames_read < ctx.batch_size) {
            auto elapsed = std::chrono::steady_clock::now() - start_time;
            if (elapsed > ctx.timeout) break;
            
            canfd_frame raw_frame{};
            ssize_t bytes = ::read(this->sock, &raw_frame, sizeof(canfd_frame));
            
            if (bytes == sizeof(canfd_frame)) {
                std::array<uint8_t, FD::MAX_DLEN> data{};
                std::copy(raw_frame.data, raw_frame.data + raw_frame.len, data.begin());
                frames.emplace_back(raw_frame.can_id, raw_frame.len, data, raw_frame.flags);
                ++frames_read;
            } else if (bytes == -1) {
                if (errno == EAGAIN || errno == EWOULDBLOCK) {
                    break;
                } else {
                    throw SocketException(fmt::format("Error reading FD batch: {}", strerror(errno)));
                }
            }
        }
        
        return frames_read;
    }

    // Memory pool management for high-frequency applications
    struct PoolManager {
        FramePool<Frame> frame_pool;
        FramePool<FD::Frame> fd_frame_pool;
        
        [[nodiscard]] Frame* allocateFrame() {
            return frame_pool.allocate();
        }
        
        [[nodiscard]] FD::Frame* allocateFDFrame() {
            return fd_frame_pool.allocate();
        }
        
        void deallocateFrame(Frame* frame) {
            frame_pool.deallocate(frame);
        }
        
        void deallocateFDFrame(FD::Frame* frame) {
            fd_frame_pool.deallocate(frame);
        }
        
        [[nodiscard]] std::pair<size_t, size_t> getPoolStats() const {
            return {frame_pool.available(), fd_frame_pool.available()};
        }
    };
    
    // Get access to memory pools for high-performance applications
    [[nodiscard]] static PoolManager& getPoolManager() {
        static thread_local PoolManager manager;
        return manager;
    }
};

// ============================================================================
// Thread Safety Policies and Wrappers
// ============================================================================

// Forward declarations for thread-safe components
template<typename T, size_t Size> class SPSCRingBuffer;
template<typename T, size_t Size> class MPSCQueue;
template<typename ThreadingPolicy> class ThreadSafeSocket;

// Threading policy tags
struct NoLocking {};
struct SharedMutexPolicy {};
struct PerThreadPolicy {};
struct LockFreePolicy {};

// ============================================================================
// NoLocking Policy - Zero overhead, no thread safety
// ============================================================================
template<>
class ThreadSafeSocket<NoLocking> {
private:
    Socket socket;

public:
    // Constructor forwarding
    template<typename... Args>
    explicit ThreadSafeSocket(Args&&... args) : socket(std::forward<Args>(args)...) {}

    // Direct delegation to socket methods - no locking overhead
    [[nodiscard]] ssize_t write(const Frame& frame, std::optional<std::chrono::milliseconds> timeout = std::nullopt) { return socket.write(frame, timeout); }
    [[nodiscard]] ssize_t write(const FD::Frame& frame, std::optional<std::chrono::milliseconds> timeout = std::nullopt) { return socket.write(frame, timeout); }
    [[nodiscard]] ssize_t write(const std::vector<Frame>& frames, std::optional<std::chrono::milliseconds> timeout = std::nullopt) { return socket.write(frames, timeout); }
    [[nodiscard]] ssize_t write(const std::vector<FD::Frame>& frames, std::optional<std::chrono::milliseconds> timeout = std::nullopt) { return socket.write(frames, timeout); }
    
    [[nodiscard]] auto read(std::optional<std::chrono::milliseconds> timeout = std::nullopt) { return socket.read(timeout); }
    [[nodiscard]] auto readMsg(std::optional<std::chrono::milliseconds> timeout = std::nullopt) { return socket.readMsg(timeout); }
    
    [[nodiscard]] int setFilter(uint32_t id, uint32_t mask) { return socket.setFilter(id, mask); }
    [[nodiscard]] int setFilters(const std::vector<std::pair<uint32_t, uint32_t>>& filters) { 
        return socket.setFilters(filters); 
    }
    
    void close() noexcept { socket.close(); }
    [[nodiscard]] bool isOpen() const noexcept { return socket.isOpen(); }
    [[nodiscard]] int getFd() const noexcept { return socket.getFd(); }
    [[nodiscard]] ProtocolMode getMode() const { return socket.getMode(); }
    
    void setNonBlocking() { socket.setNonBlocking(); }
    void setBlocking() { socket.setBlocking(); }
    
    [[nodiscard]] int setReceiveBufferSize(int size) { return socket.setReceiveBufferSize(size); }
    [[nodiscard]] int getReceiveBufferSize() const { return socket.getReceiveBufferSize(); }
    
    // Batch operations
    [[nodiscard]] ssize_t writeBatch(const std::vector<Frame>& frames, const BatchContext& ctx = {}) {
        return socket.writeBatch(frames, ctx);
    }
    [[nodiscard]] ssize_t writeBatch(const std::vector<FD::Frame>& frames, const BatchContext& ctx = {}) {
        return socket.writeBatch(frames, ctx);
    }
    
    [[nodiscard]] size_t readBatch(std::vector<Frame>& frames, const BatchContext& ctx = {}) {
        return socket.readBatch(frames, ctx);
    }
    [[nodiscard]] size_t readBatch(std::vector<FD::Frame>& frames, const BatchContext& ctx = {}) {
        return socket.readBatch(frames, ctx);
    }
    
    // Convenience methods
    [[nodiscard]] ssize_t sendFrame(uint32_t id, const std::initializer_list<uint8_t>& data, std::optional<std::chrono::milliseconds> timeout = std::nullopt) {
        return socket.sendFrame(id, data, timeout);
    }
    [[nodiscard]] ssize_t sendFDFrame(uint32_t id, const std::initializer_list<uint8_t>& data, uint8_t flags = FD::FDF_FLAG, std::optional<std::chrono::milliseconds> timeout = std::nullopt) {
        return socket.sendFDFrame(id, data, flags, timeout);
    }
    
    // Direct access to underlying socket for advanced use cases
    Socket& getSocket() { return socket; }
    const Socket& getSocket() const { return socket; }
};

// ============================================================================
// SharedMutex Policy - Multiple readers, single writer
// ============================================================================
template<>
class ThreadSafeSocket<SharedMutexPolicy> {
private:
    Socket socket;
    mutable std::shared_mutex mutex;

public:
    // Constructor forwarding
    template<typename... Args>
    explicit ThreadSafeSocket(Args&&... args) : socket(std::forward<Args>(args)...) {}

    // Write operations need exclusive lock
    [[nodiscard]] ssize_t write(const Frame& frame, std::optional<std::chrono::milliseconds> timeout = std::nullopt) {
        std::unique_lock lock(mutex);
        return socket.write(frame, timeout);
    }
    
    [[nodiscard]] ssize_t write(const FD::Frame& frame, std::optional<std::chrono::milliseconds> timeout = std::nullopt) {
        std::unique_lock lock(mutex);
        return socket.write(frame, timeout);
    }
    
    [[nodiscard]] ssize_t write(const std::vector<Frame>& frames, std::optional<std::chrono::milliseconds> timeout = std::nullopt) {
        std::unique_lock lock(mutex);
        return socket.write(frames, timeout);
    }
    
    [[nodiscard]] ssize_t write(const std::vector<FD::Frame>& frames, std::optional<std::chrono::milliseconds> timeout = std::nullopt) {
        std::unique_lock lock(mutex);
        return socket.write(frames, timeout);
    }
    
    // Read operations can use shared lock
    [[nodiscard]] auto read(std::optional<std::chrono::milliseconds> timeout = std::nullopt) {
        std::shared_lock lock(mutex);
        return socket.read(timeout);
    }
    
    [[nodiscard]] auto readMsg(std::optional<std::chrono::milliseconds> timeout = std::nullopt) {
        std::shared_lock lock(mutex);
        return socket.readMsg(timeout);
    }
    
    // Configuration operations need exclusive lock
    [[nodiscard]] int setFilter(uint32_t id, uint32_t mask) {
        std::unique_lock lock(mutex);
        return socket.setFilter(id, mask);
    }
    
    [[nodiscard]] int setFilters(const std::vector<std::pair<uint32_t, uint32_t>>& filters) {
        std::unique_lock lock(mutex);
        return socket.setFilters(filters);
    }
    
    void close() noexcept {
        std::unique_lock lock(mutex);
        socket.close();
    }
    
    [[nodiscard]] bool isOpen() const noexcept {
        std::shared_lock lock(mutex);
        return socket.isOpen();
    }
    
    [[nodiscard]] int getFd() const noexcept {
        std::shared_lock lock(mutex);
        return socket.getFd();
    }
    
    [[nodiscard]] ProtocolMode getMode() const {
        std::shared_lock lock(mutex);
        return socket.getMode();
    }
    
    void setNonBlocking() {
        std::unique_lock lock(mutex);
        socket.setNonBlocking();
    }
    
    void setBlocking() {
        std::unique_lock lock(mutex);
        socket.setBlocking();
    }
    
    [[nodiscard]] int setReceiveBufferSize(int size) {
        std::unique_lock lock(mutex);
        return socket.setReceiveBufferSize(size);
    }
    
    [[nodiscard]] int getReceiveBufferSize() const {
        std::shared_lock lock(mutex);
        return socket.getReceiveBufferSize();
    }
    
    // Batch operations
    [[nodiscard]] ssize_t writeBatch(const std::vector<Frame>& frames, const BatchContext& ctx = {}) {
        std::unique_lock lock(mutex);
        return socket.writeBatch(frames, ctx);
    }
    
    [[nodiscard]] ssize_t writeBatch(const std::vector<FD::Frame>& frames, const BatchContext& ctx = {}) {
        std::unique_lock lock(mutex);
        return socket.writeBatch(frames, ctx);
    }
    
    [[nodiscard]] size_t readBatch(std::vector<Frame>& frames, const BatchContext& ctx = {}) {
        std::shared_lock lock(mutex);
        return socket.readBatch(frames, ctx);
    }
    
    [[nodiscard]] size_t readBatch(std::vector<FD::Frame>& frames, const BatchContext& ctx = {}) {
        std::shared_lock lock(mutex);
        return socket.readBatch(frames, ctx);
    }
    
    // Convenience methods
    [[nodiscard]] ssize_t sendFrame(uint32_t id, const std::initializer_list<uint8_t>& data, std::optional<std::chrono::milliseconds> timeout = std::nullopt) {
        std::unique_lock lock(mutex);
        return socket.sendFrame(id, data, timeout);
    }
    
    [[nodiscard]] ssize_t sendFDFrame(uint32_t id, const std::initializer_list<uint8_t>& data, uint8_t flags = FD::FDF_FLAG, std::optional<std::chrono::milliseconds> timeout = std::nullopt) {
        std::unique_lock lock(mutex);
        return socket.sendFDFrame(id, data, flags, timeout);
    }
};

// ============================================================================
// PerThread Policy - Each thread gets its own socket instance
// ============================================================================
template<>
class ThreadSafeSocket<PerThreadPolicy> {
private:
    struct SocketInfo {
        std::string interface;
        ProtocolMode mode;
        std::chrono::milliseconds timeout;
        int rcvbuf_size;
    };
    
    SocketInfo info;
    mutable std::unordered_map<std::thread::id, std::unique_ptr<Socket>> sockets;
    mutable std::mutex map_mutex;  // Only for map access, not socket operations

    Socket& getThreadSocket() const {
        auto tid = std::this_thread::get_id();
        
        // Fast path - check if socket exists
        {
            std::lock_guard lock(map_mutex);
            auto it = sockets.find(tid);
            if (it != sockets.end()) {
                return *it->second;
            }
        }
        
        // Slow path - create new socket
        auto new_socket = std::make_unique<Socket>(info.interface, info.mode, info.timeout, info.rcvbuf_size);
        
        std::lock_guard lock(map_mutex);
        auto [it, inserted] = sockets.emplace(tid, std::move(new_socket));
        return *it->second;
    }

public:
    explicit ThreadSafeSocket(const std::string& interface,
                             ProtocolMode mode = ProtocolMode::ClassicCAN,
                             const std::chrono::milliseconds& timeout = std::chrono::milliseconds(100),
                             int rcvbuf_size = 327680)
        : info{interface, mode, timeout, rcvbuf_size} {}

    // All operations use thread-local socket - no locking needed!
    [[nodiscard]] ssize_t write(const Frame& frame, std::optional<std::chrono::milliseconds> timeout = std::nullopt) { 
        return getThreadSocket().write(frame, timeout); 
    }
    
    [[nodiscard]] ssize_t write(const FD::Frame& frame, std::optional<std::chrono::milliseconds> timeout = std::nullopt) { 
        return getThreadSocket().write(frame, timeout); 
    }
    
    [[nodiscard]] ssize_t write(const std::vector<Frame>& frames, std::optional<std::chrono::milliseconds> timeout = std::nullopt) { 
        return getThreadSocket().write(frames, timeout); 
    }
    
    [[nodiscard]] ssize_t write(const std::vector<FD::Frame>& frames, std::optional<std::chrono::milliseconds> timeout = std::nullopt) { 
        return getThreadSocket().write(frames, timeout); 
    }
    
    [[nodiscard]] auto read(std::optional<std::chrono::milliseconds> timeout = std::nullopt) { 
        return getThreadSocket().read(timeout); 
    }
    
    [[nodiscard]] auto readMsg(std::optional<std::chrono::milliseconds> timeout = std::nullopt) { 
        return getThreadSocket().readMsg(timeout); 
    }
    
    [[nodiscard]] int setFilter(uint32_t id, uint32_t mask) { 
        return getThreadSocket().setFilter(id, mask); 
    }
    
    [[nodiscard]] int setFilters(const std::vector<std::pair<uint32_t, uint32_t>>& filters) { 
        return getThreadSocket().setFilters(filters); 
    }
    
    void close() noexcept {
        auto tid = std::this_thread::get_id();
        std::lock_guard lock(map_mutex);
        auto it = sockets.find(tid);
        if (it != sockets.end()) {
            it->second->close();
        }
    }
    
    void closeAll() noexcept {
        std::lock_guard lock(map_mutex);
        for (auto& [tid, socket] : sockets) {
            socket->close();
        }
        sockets.clear();
    }
    
    [[nodiscard]] bool isOpen() const noexcept { 
        return getThreadSocket().isOpen(); 
    }
    
    [[nodiscard]] int getFd() const noexcept { 
        return getThreadSocket().getFd(); 
    }
    
    [[nodiscard]] ProtocolMode getMode() const { 
        return info.mode; 
    }
    
    void setNonBlocking() { 
        getThreadSocket().setNonBlocking(); 
    }
    
    void setBlocking() { 
        getThreadSocket().setBlocking(); 
    }
    
    [[nodiscard]] int setReceiveBufferSize(int size) { 
        return getThreadSocket().setReceiveBufferSize(size); 
    }
    
    [[nodiscard]] int getReceiveBufferSize() const { 
        return getThreadSocket().getReceiveBufferSize(); 
    }
    
    // Batch operations
    [[nodiscard]] ssize_t writeBatch(const std::vector<Frame>& frames, const BatchContext& ctx = {}) {
        return getThreadSocket().writeBatch(frames, ctx);
    }
    
    [[nodiscard]] ssize_t writeBatch(const std::vector<FD::Frame>& frames, const BatchContext& ctx = {}) {
        return getThreadSocket().writeBatch(frames, ctx);
    }
    
    [[nodiscard]] size_t readBatch(std::vector<Frame>& frames, const BatchContext& ctx = {}) {
        return getThreadSocket().readBatch(frames, ctx);
    }
    
    [[nodiscard]] size_t readBatch(std::vector<FD::Frame>& frames, const BatchContext& ctx = {}) {
        return getThreadSocket().readBatch(frames, ctx);
    }
    
    // Convenience methods
    [[nodiscard]] ssize_t sendFrame(uint32_t id, const std::initializer_list<uint8_t>& data, std::optional<std::chrono::milliseconds> timeout = std::nullopt) {
        return getThreadSocket().sendFrame(id, data, timeout);
    }
    
    [[nodiscard]] ssize_t sendFDFrame(uint32_t id, const std::initializer_list<uint8_t>& data, uint8_t flags = FD::FDF_FLAG, std::optional<std::chrono::milliseconds> timeout = std::nullopt) {
        return getThreadSocket().sendFDFrame(id, data, flags, timeout);
    }
    
    // Get number of active thread sockets
    [[nodiscard]] size_t getThreadCount() const {
        std::lock_guard lock(map_mutex);
        return sockets.size();
    }
};

// ============================================================================
// Lock-Free SPSC Ring Buffer for high-performance scenarios
// ============================================================================
template<typename T, size_t Size>
class SPSCRingBuffer {
    static_assert((Size & (Size - 1)) == 0, "Size must be a power of 2");

private:
    alignas(64) std::atomic<size_t> write_pos{0};
    alignas(64) std::atomic<size_t> read_pos{0};
    alignas(64) std::array<T, Size> buffer;

public:
    [[nodiscard]] bool push(const T& item) {
        const size_t current_write = write_pos.load(std::memory_order_relaxed);
        const size_t next_write = (current_write + 1) & (Size - 1);
        
        if (next_write == read_pos.load(std::memory_order_acquire)) {
            return false;  // Buffer full
        }
        
        buffer[current_write] = item;
        write_pos.store(next_write, std::memory_order_release);
        return true;
    }
    
    [[nodiscard]] bool pop(T& item) {
        const size_t current_read = read_pos.load(std::memory_order_relaxed);
        
        if (current_read == write_pos.load(std::memory_order_acquire)) {
            return false;  // Buffer empty
        }
        
        item = buffer[current_read];
        read_pos.store((current_read + 1) & (Size - 1), std::memory_order_release);
        return true;
    }
    
    [[nodiscard]] bool empty() const {
        return read_pos.load(std::memory_order_acquire) == 
               write_pos.load(std::memory_order_acquire);
    }
    
    [[nodiscard]] size_t size() const {
        const size_t write = write_pos.load(std::memory_order_acquire);
        const size_t read = read_pos.load(std::memory_order_acquire);
        return (write - read) & (Size - 1);
    }
};

// ============================================================================
// Lock-Free MPSC Queue for multiple producers, single consumer
// ============================================================================
template<typename T, size_t Size>
class MPSCQueue {
private:
    struct Node {
        std::atomic<Node*> next{nullptr};
        T data;
    };
    
    alignas(64) std::atomic<Node*> head;
    alignas(64) Node* tail;
    std::array<Node, Size> node_pool;
    SPSCRingBuffer<Node*, Size> free_list;

public:
    MPSCQueue() {
        // Initialize free list with all nodes
        for (auto& node : node_pool) {
            (void)free_list.push(&node);  // Suppress nodiscard warning
        }
        
        // Create sentinel node
        Node* sentinel;
        (void)free_list.pop(sentinel);  // Suppress nodiscard warning
        head.store(sentinel);
        tail = sentinel;
    }
    
    [[nodiscard]] bool push(const T& item) {
        Node* new_node;
        if (!free_list.pop(new_node)) {
            return false;  // No free nodes
        }
        
        new_node->data = item;
        new_node->next.store(nullptr, std::memory_order_relaxed);
        
        Node* prev_head = head.exchange(new_node, std::memory_order_acq_rel);
        prev_head->next.store(new_node, std::memory_order_release);
        return true;
    }
    
    [[nodiscard]] bool pop(T& item) {
        Node* tail_next = tail->next.load(std::memory_order_acquire);
        if (tail_next == nullptr) {
            return false;  // Queue empty
        }
        
        item = tail_next->data;
        (void)free_list.push(tail);  // Suppress nodiscard warning
        tail = tail_next;
        return true;
    }
};

// ============================================================================
// LockFree Policy - Uses lock-free queue with dedicated I/O thread
// ============================================================================
template<>
class ThreadSafeSocket<LockFreePolicy> {
private:
    static constexpr size_t QUEUE_SIZE = 4096;
    
    Socket socket;
    MPSCQueue<Frame, QUEUE_SIZE> write_queue;
    SPSCRingBuffer<Frame, QUEUE_SIZE> read_buffer;
    std::atomic<bool> running{true};
    std::thread io_thread;
    
    void ioThreadFunc() {
        while (running.load(std::memory_order_acquire)) {
            // Process writes
            Frame frame(0, 0, {});  // Temporary frame for popping
            while (write_queue.pop(frame)) {
                (void)socket.write(frame);  // Suppress nodiscard warning
            }
            
            // Process reads (non-blocking)
            socket.setNonBlocking();
            auto result = socket.read();
            if (auto* opt = std::get_if<std::optional<std::unique_ptr<Frame>>>(&result)) {
                if (opt->has_value()) {
                    (void)read_buffer.push(*opt->value());  // Suppress nodiscard warning
                }
            }
            
            // Small sleep to prevent busy waiting
            std::this_thread::sleep_for(std::chrono::microseconds(10));
        }
    }

public:
    template<typename... Args>
    explicit ThreadSafeSocket(Args&&... args) 
        : socket(std::forward<Args>(args)...), 
          io_thread(&ThreadSafeSocket::ioThreadFunc, this) {}
    
    ~ThreadSafeSocket() {
        running.store(false, std::memory_order_release);
        if (io_thread.joinable()) {
            io_thread.join();
        }
    }
    
    // Non-blocking push to queue
    [[nodiscard]] bool write(const Frame& frame) {
        return write_queue.push(frame);
    }
    
    // Non-blocking read from buffer
    [[nodiscard]] std::optional<Frame> read() {
        Frame frame(0, 0, {});  // Temporary frame for popping
        if (read_buffer.pop(frame)) {
            return frame;
        }
        return std::nullopt;
    }
    
    // Batch write - pushes all frames to queue
    [[nodiscard]] size_t writeBatch(const std::vector<Frame>& frames) {
        size_t written = 0;
        for (const auto& frame : frames) {
            if (write_queue.push(frame)) {
                ++written;
            } else {
                break;  // Queue full
            }
        }
        return written;
    }
    
    // Note: Many methods not applicable for lock-free version
    // Users should use this for high-frequency write-only or read-only scenarios
};

// ============================================================================
// Factory functions for creating thread-safe sockets
// ============================================================================
enum class ThreadingModel {
    None,           // No thread safety (NoLocking)
    SharedMutex,    // Read-heavy workloads
    PerThread,      // Zero contention
    LockFree        // High-frequency producer-consumer
};

// Base class for polymorphic use if needed
class ThreadSafeSocketBase {
public:
    virtual ~ThreadSafeSocketBase() = default;
    virtual bool isOpen() const = 0;
    virtual void close() = 0;
};

// Helper functions for creating specific socket types
template<typename... Args>
[[nodiscard]] std::unique_ptr<ThreadSafeSocket<NoLocking>> createNoLockingSocket(Args&&... args) {
    return std::make_unique<ThreadSafeSocket<NoLocking>>(std::forward<Args>(args)...);
}

template<typename... Args>
[[nodiscard]] std::unique_ptr<ThreadSafeSocket<SharedMutexPolicy>> createSharedMutexSocket(Args&&... args) {
    return std::make_unique<ThreadSafeSocket<SharedMutexPolicy>>(std::forward<Args>(args)...);
}

template<typename... Args>
[[nodiscard]] std::unique_ptr<ThreadSafeSocket<PerThreadPolicy>> createPerThreadSocket(Args&&... args) {
    return std::make_unique<ThreadSafeSocket<PerThreadPolicy>>(std::forward<Args>(args)...);
}

template<typename... Args>
[[nodiscard]] std::unique_ptr<ThreadSafeSocket<LockFreePolicy>> createLockFreeSocket(Args&&... args) {
    return std::make_unique<ThreadSafeSocket<LockFreePolicy>>(std::forward<Args>(args)...);
}

} // namespace CAN