kcenon::network::internal::tcp_socket Class Reference

A lightweight wrapper around asio::ip::tcp::socket, enabling asynchronous read and write operations. More...

#include <tcp_socket.h>

Inheritance diagram for kcenon::network::internal::tcp_socket:

Collaboration diagram for kcenon::network::internal::tcp_socket:

Public Types
using	backpressure_callback = std::function<void(bool apply_backpressure)>
	Callback type for backpressure notifications.

Public Member Functions
	tcp_socket (asio::ip::tcp::socket socket)
	Constructs a tcp_socket by taking ownership of a moved socket.
	tcp_socket (asio::ip::tcp::socket socket, const socket_config &config)
	Constructs a tcp_socket with custom configuration.
	~tcp_socket ()=default
	Default destructor (no special cleanup needed).
auto	attach_observer (std::shared_ptr< network_core::interfaces::socket_observer > observer) -> void
	Attaches an observer to receive socket events.
auto	detach_observer (std::shared_ptr< network_core::interfaces::socket_observer > observer) -> void
	Detaches a previously attached observer.
auto	set_receive_callback (std::function< void(const std::vector< uint8_t > &)> callback) -> void
	Sets a callback to receive inbound data chunks.
auto	set_receive_callback_view (std::function< void(std::span< const uint8_t >)> callback) -> void
	Sets a zero-copy callback to receive inbound data as a view.
auto	set_error_callback (std::function< void(std::error_code)> callback) -> void
	Sets a callback to handle socket errors (e.g., read/write failures).
auto	start_read () -> void
	Begins the continuous asynchronous read loop.
auto	async_send (std::vector< uint8_t > &&data, std::function< void(std::error_code, std::size_t)> handler) -> void
	Initiates an asynchronous write of the given data buffer.
auto	set_backpressure_callback (backpressure_callback callback) -> void
	Sets a callback for backpressure notifications.
auto	try_send (std::vector< uint8_t > &&data, std::function< void(std::error_code, std::size_t)> handler) -> bool
	Attempts to send data without blocking.
auto	pending_bytes () const -> std::size_t
	Returns current pending bytes count.
auto	is_backpressure_active () const -> bool
	Checks if backpressure is currently active.
auto	metrics () const -> const socket_metrics &
	Returns socket metrics for monitoring.
auto	reset_metrics () -> void
	Resets socket metrics to zero.
auto	config () const -> const socket_config &
	Returns the current socket configuration.
auto	socket () -> asio::ip::tcp::socket &
	Provides direct access to the underlying asio::ip::tcp::socket in case advanced operations are needed.
auto	stop_read () -> void
	Stops the read loop to prevent further async operations.
auto	close () -> void
	Safely closes the socket and stops all async operations.
auto	is_closed () const -> bool
	Checks if the socket has been closed.

Private Types
using	receive_callback_t = std::function<void(const std::vector<uint8_t>&)>
	Callback type aliases for lock-free storage.
using	receive_callback_view_t = std::function<void(std::span<const uint8_t>)>
using	error_callback_t = std::function<void(std::error_code)>

Private Member Functions
auto	do_read () -> void
	Internal function to handle the read logic with async_read_some().
auto	notify_observers_receive (std::span< const uint8_t > data) -> void
	Helper to notify all observers of receive events.
auto	notify_observers_error (std::error_code ec) -> void
	Helper to notify all observers of error events.
auto	notify_observers_backpressure (bool apply) -> void
	Helper to notify all observers of backpressure events.

Private Attributes
asio::ip::tcp::socket	socket_
std::array< uint8_t, 4096 >	read_buffer_
std::shared_ptr< receive_callback_t >	receive_callback_
std::shared_ptr< receive_callback_view_t >	receive_callback_view_
std::shared_ptr< error_callback_t >	error_callback_
std::mutex	callback_mutex_
socket_config	config_
	Backpressure configuration.
socket_metrics	metrics_
	Socket runtime metrics.
std::vector< std::weak_ptr< network_core::interfaces::socket_observer > >	observers_ {}
	List of socket observers (using weak_ptr for automatic cleanup). Protected by callback_mutex_.
std::atomic< bool >	is_reading_ {false}
std::atomic< bool >	is_closed_ {false}
std::atomic< std::size_t >	pending_bytes_ {0}
	Current pending bytes in send buffer.
std::atomic< bool >	backpressure_active_ {false}
	Backpressure state flag.
std::shared_ptr< backpressure_callback >	backpressure_callback_
	Backpressure notification callback.

Detailed Description

A lightweight wrapper around asio::ip::tcp::socket, enabling asynchronous read and write operations.

Key Features

• Maintains a socket_ (from ASIO) for TCP communication.
• Exposes set_receive_callback() to handle inbound data and set_error_callback() for error handling.
• start_read() begins an ongoing loop of async_read_some().
• async_send() performs an async_write of a given data buffer.

Thread Safety

• All public methods are thread-safe. Callback registration is protected by callback_mutex_.
• ASIO operations are serialized through the io_context, ensuring read_buffer_ is only accessed by one async operation at a time.
• The provided callbacks will be invoked on an ASIO worker thread; ensure that your callback logic is thread-safe if it shares data.

Definition at line 43 of file tcp_socket.h.

Member Typedef Documentation

backpressure_callback

using kcenon::network::internal::tcp_socket::backpressure_callback = std::function<void(bool apply_backpressure)>

Callback type for backpressure notifications.

Parameters

apply_backpressure

True when backpressure should be applied (high water mark reached), false when it can be released (low water mark reached).

Definition at line 52 of file tcp_socket.h.

error_callback_t

using kcenon::network::internal::tcp_socket::error_callback_t = std::function<void(std::error_code)>

private

Definition at line 305 of file tcp_socket.h.

receive_callback_t

using kcenon::network::internal::tcp_socket::receive_callback_t = std::function<void(const std::vector<uint8_t>&)>

private

Callback type aliases for lock-free storage.

Definition at line 303 of file tcp_socket.h.

receive_callback_view_t

using kcenon::network::internal::tcp_socket::receive_callback_view_t = std::function<void(std::span<const uint8_t>)>

private

Definition at line 304 of file tcp_socket.h.

Constructor & Destructor Documentation

tcp_socket() [1/2]

kcenon::network::internal::tcp_socket::tcp_socket

(

asio::ip::tcp::socket

socket

)

Constructs a tcp_socket by taking ownership of a moved socket.

Parameters

socket

An asio::ip::tcp::socket that must be open/connected or at least valid.

After construction, you can immediately call start_read() to begin receiving data. For sending, call async_send().

Definition at line 15 of file tcp_socket.cpp.

        : socket_(std::move(socket)), config_{}
    {
        // constructor body empty
    }

tcp_socket() [2/2]

kcenon::network::internal::tcp_socket::tcp_socket	(	asio::ip::tcp::socket	socket,
		const socket_config &	config )

Constructs a tcp_socket with custom configuration.

Parameters

socket	An asio::ip::tcp::socket that must be open/connected.
config	Socket configuration including backpressure settings.

Definition at line 21 of file tcp_socket.cpp.

        : socket_(std::move(socket)), config_(config)
    {
        // constructor body empty
    }

References config.

~tcp_socket()

kcenon::network::internal::tcp_socket::~tcp_socket ( )

default

Default destructor (no special cleanup needed).

Member Function Documentation

async_send()

auto kcenon::network::internal::tcp_socket::async_send	(	std::vector< uint8_t > &&	data,
		std::function< void(std::error_code, std::size_t)>	handler ) -> void

Initiates an asynchronous write of the given data buffer.

Parameters

data	The buffer to send over TCP (moved for efficiency).
handler	A completion handler with signature void(std::error_code, std::size_t) that is invoked upon success or failure.

The handler receives:

• ec : the std::error_code from the write operation,
• bytes_transferred : how many bytes were actually written.

Example

auto sock = std::make_shared<network::tcp_socket>(...);
std::vector<uint8_t> buf = {0x01, 0x02, 0x03};
sock->async_send(std::move(buf), [](std::error_code ec, std::size_t len) {
    if(ec) {
        // handle error
    }
    else {
        // handle success
    }
});

Note

Data is moved (not copied) to avoid memory allocation.

The original vector will be empty after this call.

Definition at line 253 of file tcp_socket.cpp.

{
    // Check if socket has been closed or is no longer open before starting async operation
    // Both checks are needed: is_closed_ for explicit close() calls, is_open() for ASIO state
    // This prevents UBSAN errors from accessing null descriptor_state in epoll_reactor
    if (is_closed_.load() || !socket_.is_open())
    {
        if (handler)
        {
            handler(asio::error::not_connected, 0);
        }
        return;
    }
 
    // Additional check: verify native handle is valid before async operation
    // This prevents SEGV when socket is in transitional state after move construction
#ifdef _WIN32
    if (socket_.native_handle() == INVALID_SOCKET)
#else
    if (socket_.native_handle() < 0)
#endif
    {
        if (handler)
        {
            handler(asio::error::not_connected, 0);
        }
        return;
    }
 
    // Verify ASIO internal state (descriptor_state) is valid before async operations
    // by accessing the socket's non-blocking property. This forces ASIO to access
    // its internal implementation state, which will fail gracefully if invalid.
    std::error_code state_check_ec;
    socket_.non_blocking(socket_.non_blocking(), state_check_ec);
    if (state_check_ec)
    {
        if (handler)
        {
            handler(state_check_ec, 0);
        }
        return;
    }
 
    auto self = shared_from_this();
    std::size_t data_size = data.size();
 
    // Track pending bytes
    std::size_t new_pending = pending_bytes_.fetch_add(data_size) + data_size;
    metrics_.current_pending_bytes.store(new_pending);
 
    // Update peak pending bytes
    std::size_t peak = metrics_.peak_pending_bytes.load();
    while (new_pending > peak &&
           !metrics_.peak_pending_bytes.compare_exchange_weak(peak, new_pending))
    {
        // Loop until we successfully update or find a higher value
    }
 
    // Check high water mark for backpressure
    if (config_.high_water_mark > 0 && !backpressure_active_.load() &&
        new_pending >= config_.high_water_mark)
    {
        backpressure_active_.store(true);
        metrics_.backpressure_events.fetch_add(1);
 
        // Invoke backpressure callback
        auto bp_cb = std::atomic_load(&backpressure_callback_);
        if (bp_cb && *bp_cb)
        {
            (*bp_cb)(true);
        }
    }
 
    // Move data into shared_ptr for lifetime management
    auto buffer = std::make_shared<std::vector<uint8_t>>(std::move(data));
    try
    {
        asio::async_write(
            socket_, asio::buffer(*buffer),
            [handler = std::move(handler), self, buffer, data_size](std::error_code ec, std::size_t bytes_transferred)
            {
                // Update pending bytes on completion
                std::size_t remaining = self->pending_bytes_.fetch_sub(data_size) - data_size;
                self->metrics_.current_pending_bytes.store(remaining);
 
                if (!ec)
                {
                    self->metrics_.total_bytes_sent.fetch_add(bytes_transferred);
                    self->metrics_.send_count.fetch_add(1);
                }
 
                // Check low water mark to release backpressure
                if (self->config_.low_water_mark > 0 && self->backpressure_active_.load() &&
                    remaining <= self->config_.low_water_mark)
                {
                    self->backpressure_active_.store(false);
 
                    // Invoke backpressure callback
                    auto bp_cb = std::atomic_load(&self->backpressure_callback_);
                    if (bp_cb && *bp_cb)
                    {
                        (*bp_cb)(false);
                    }
                }
 
                if constexpr (std::is_invocable_v<decltype(handler), std::error_code, std::size_t>)
                {
                    if (handler)
                    {
                        handler(ec, bytes_transferred);
                    }
                }
            });
    }
    catch (const std::exception&)
    {
        // Socket may have been closed between our checks and async operation
        // Rollback pending bytes and notify handler
        pending_bytes_.fetch_sub(data_size);
        metrics_.current_pending_bytes.store(pending_bytes_.load());
        if (handler)
        {
            handler(asio::error::not_connected, 0);
        }
    }
}

attach_observer()

auto kcenon::network::internal::tcp_socket::attach_observer

(

std::shared_ptr< network_core::interfaces::socket_observer >

observer

)

-> void

Attaches an observer to receive socket events.

Parameters

observer

Shared pointer to socket_observer implementation.

Multiple observers can be attached to the same socket. All attached observers will be notified of socket events.

See also

socket_observer

detach_observer()

close()

auto kcenon::network::internal::tcp_socket::close

(

)

-> void

Safely closes the socket and stops all async operations.

This method atomically sets the closed flag before closing the socket, preventing data races between the close operation and async read operations. Thread-safe with respect to concurrent async operations.

Definition at line 89 of file tcp_socket.cpp.

    {
        // Atomically mark socket as closed before actual close
        // This prevents data races with concurrent async operations
        is_closed_.store(true);
        is_reading_.store(false);
 
        // Post the actual socket close to the socket's executor to ensure
        // thread-safety. ASIO's epoll_reactor has internal data structures that
        // must be accessed from the same thread as async operations to avoid
        // data races and SEGV errors detected by ThreadSanitizer/AddressSanitizer.
        try
        {
            auto self = shared_from_this();
            asio::post(socket_.get_executor(), [self]() {
                std::error_code ec;
                if (self->socket_.is_open())
                {
                    self->socket_.cancel(ec);
                    self->socket_.close(ec);
                }
                // Ignore close errors - socket may already be closed
            });
        }
        catch (...)
        {
            // If post fails (e.g., executor not running or shared_from_this fails),
            // fall back to direct close. This may race but is better than leaking
            // the socket.
            std::error_code ec;
            if (socket_.is_open())
            {
                socket_.cancel(ec);
                socket_.close(ec);
            }
        }
    }

config()

auto kcenon::network::internal::tcp_socket::config ( ) const -> const socket_config&

nodiscard

Returns the current socket configuration.

Returns

Const reference to socket_config struct.

Definition at line 429 of file tcp_socket.cpp.

{
    return config_;
}

detach_observer()

auto kcenon::network::internal::tcp_socket::detach_observer

(

std::shared_ptr< network_core::interfaces::socket_observer >

observer

)

-> void

Detaches a previously attached observer.

Parameters

observer

The observer to detach.

If the observer was not attached, this method has no effect.

do_read()

auto kcenon::network::internal::tcp_socket::do_read ( ) -> void

private

Internal function to handle the read logic with async_read_some().

Upon success, it calls receive_callback_ if set, then schedules another read. On error, it calls error_callback_ if available.

Definition at line 132 of file tcp_socket.cpp.

    {
        // Check if reading has been stopped before initiating new async operation
        if (!is_reading_.load())
        {
            return;
        }
 
        // Check if socket has been closed or is no longer open before starting async operation
        // This prevents data races and UBSAN errors from accessing null descriptor_state
        // Both checks are needed: is_closed_ for explicit close() calls, is_open() for ASIO state
        if (is_closed_.load() || !socket_.is_open())
        {
            is_reading_.store(false);
            return;
        }
 
        // Additional check: verify native handle is valid before async operation
        // This prevents SEGV when socket is in transitional state after move construction
        // On POSIX systems, invalid handles are < 0; on Windows, INVALID_SOCKET
#ifdef _WIN32
        if (socket_.native_handle() == INVALID_SOCKET)
#else
        if (socket_.native_handle() < 0)
#endif
        {
            is_reading_.store(false);
            return;
        }
 
        // Verify ASIO internal state (descriptor_state) is valid before async operations
        // by accessing the socket's non-blocking property. This forces ASIO to access
        // its internal implementation state, which will fail gracefully if invalid
        // instead of causing UBSAN errors from null pointer dereferencing.
        std::error_code state_check_ec;
        socket_.non_blocking(socket_.non_blocking(), state_check_ec);
        if (state_check_ec)
        {
            is_reading_.store(false);
            // Notify via error callback if available
            auto error_cb = std::atomic_load(&error_callback_);
            if (error_cb && *error_cb)
            {
                (*error_cb)(state_check_ec);
            }
            return;
        }
 
        auto self = shared_from_this();
        try
        {
            socket_.async_read_some(
                asio::buffer(read_buffer_),
                [this, self](std::error_code ec, std::size_t length)
                {
                    // Check if reading has been stopped or socket closed at callback time
                    // This prevents accessing invalid socket state after close()
                    if (!is_reading_.load() || is_closed_.load())
                    {
                        return;
                    }
 
                    if (ec)
                    {
                        // On error, invoke the error callback
                        // Lock-free callback access via atomic_load
                        auto error_cb = std::atomic_load(&error_callback_);
                        if (error_cb && *error_cb)
                        {
                            (*error_cb)(ec);
                        }
                        return;
                    }
 
                    // On success, if length > 0, dispatch to the appropriate callback
                    if (length > 0)
                    {
                        // Lock-free callback access via atomic_load
                        // Prefer view callback (zero-copy) over vector callback
                        auto view_cb = std::atomic_load(&receive_callback_view_);
                        if (view_cb && *view_cb)
                        {
                            // Zero-copy path: create span view directly into read_buffer_
                            // No std::vector allocation or copy required
                            std::span<const uint8_t> data_view(read_buffer_.data(), length);
                            (*view_cb)(data_view);
                        }
                        else
                        {
                            // Legacy path: allocate and copy into vector for compatibility
                            auto recv_cb = std::atomic_load(&receive_callback_);
                            if (recv_cb && *recv_cb)
                            {
                                std::vector<uint8_t> chunk(read_buffer_.begin(),
                                                           read_buffer_.begin() + length);
                                (*recv_cb)(chunk);
                            }
                        }
                    }
 
                    // Continue reading only if still active and socket is not closed
                    // Use atomic is_closed_ flag to prevent data race with close()
                    if (is_reading_.load() && !is_closed_.load())
                    {
                        do_read();
                    }
                });
        }
        catch (const std::exception&)
        {
            // Socket may have been invalidated between checks and async operation
            // This catch prevents crashes from race conditions during socket teardown
            is_reading_.store(false);
            auto error_cb = std::atomic_load(&error_callback_);
            if (error_cb && *error_cb)
            {
                (*error_cb)(asio::error::not_connected);
            }
        }
    }

is_backpressure_active()

auto kcenon::network::internal::tcp_socket::is_backpressure_active ( ) const -> bool

nodiscard

Checks if backpressure is currently active.

Returns

true if pending bytes exceed high_water_mark.

Definition at line 414 of file tcp_socket.cpp.

{
    return backpressure_active_.load();
}

is_closed()

auto kcenon::network::internal::tcp_socket::is_closed ( ) const -> bool

nodiscard

Checks if the socket has been closed.

Returns

true if close() has been called on this socket.

Definition at line 127 of file tcp_socket.cpp.

    {
        return is_closed_.load();
    }

metrics()

auto kcenon::network::internal::tcp_socket::metrics ( ) const -> const socket_metrics&

nodiscard

Returns socket metrics for monitoring.

Returns

Const reference to socket_metrics struct.

Definition at line 419 of file tcp_socket.cpp.

{
    return metrics_;
}

notify_observers_backpressure()

auto kcenon::network::internal::tcp_socket::notify_observers_backpressure ( bool apply ) -> void

private

Helper to notify all observers of backpressure events.

notify_observers_error()

auto kcenon::network::internal::tcp_socket::notify_observers_error ( std::error_code ec ) -> void

private

Helper to notify all observers of error events.

notify_observers_receive()

auto kcenon::network::internal::tcp_socket::notify_observers_receive ( std::span< const uint8_t > data ) -> void

private

Helper to notify all observers of receive events.

pending_bytes()

auto kcenon::network::internal::tcp_socket::pending_bytes ( ) const -> std::size_t

nodiscard

Returns current pending bytes count.

Returns

Number of bytes currently pending in send buffer.

Definition at line 409 of file tcp_socket.cpp.

{
    return pending_bytes_.load();
}

reset_metrics()

auto kcenon::network::internal::tcp_socket::reset_metrics

(

)

-> void

Resets socket metrics to zero.

Definition at line 424 of file tcp_socket.cpp.

{
    metrics_.reset();
}

set_backpressure_callback()

auto kcenon::network::internal::tcp_socket::set_backpressure_callback

(

backpressure_callback

callback

)

-> void

Sets a callback for backpressure notifications.

Parameters

callback

Function invoked when backpressure state changes.

The callback receives true when pending bytes exceed high_water_mark (apply backpressure), and false when they drop below low_water_mark (release backpressure).

Definition at line 382 of file tcp_socket.cpp.

{
    auto new_cb = std::make_shared<backpressure_callback>(std::move(callback));
    std::lock_guard<std::mutex> lock(callback_mutex_);
    std::atomic_store(&backpressure_callback_, new_cb);
}

set_error_callback()

auto kcenon::network::internal::tcp_socket::set_error_callback

(

std::function< void(std::error_code)>

callback

)

-> void

Sets a callback to handle socket errors (e.g., read/write failures).

Parameters

callback

A function with signature void(std::error_code), invoked when any asynchronous operation fails.

If no callback is set, errors are not explicitly handled here (beyond stopping reads).

Definition at line 43 of file tcp_socket.cpp.

    {
        auto new_cb = std::make_shared<error_callback_t>(std::move(callback));
        std::lock_guard<std::mutex> lock(callback_mutex_);
        std::atomic_store(&error_callback_, new_cb);
    }

set_receive_callback()

auto kcenon::network::internal::tcp_socket::set_receive_callback

(

std::function< void(const std::vector< uint8_t > &)>

callback

)

-> void

Sets a callback to receive inbound data chunks.

Parameters

callback

A function with signature void(const std::vector<uint8_t>&), called whenever a chunk of data is successfully read.

If no callback is set, received data is effectively discarded.

Note

For better performance, consider using set_receive_callback_view() instead.

Definition at line 27 of file tcp_socket.cpp.

    {
        auto new_cb = std::make_shared<receive_callback_t>(std::move(callback));
        std::lock_guard<std::mutex> lock(callback_mutex_);
        std::atomic_store(&receive_callback_, new_cb);
    }

set_receive_callback_view()

auto kcenon::network::internal::tcp_socket::set_receive_callback_view

(

std::function< void(std::span< const uint8_t >)>

callback

)

-> void

Sets a zero-copy callback to receive inbound data as a view.

Parameters

callback

A function with signature void(std::span<const uint8_t>), called whenever a chunk of data is successfully read.

Zero-Copy Performance

Unlike set_receive_callback(), this callback receives data as a non-owning view directly into the internal read buffer, avoiding per-read std::vector allocations and copies.

Lifetime Contract

• The span is valid only until the callback returns.
• Callers must not store, capture, or use the span after returning from the callback.
• If data must be retained, copy it into your own container within the callback.

Dispatch Priority

• If both view and vector callbacks are set, the view callback takes priority and the vector callback is not invoked.

Example

sock->set_receive_callback_view([](std::span<const uint8_t> data) {
    // Process data directly (zero-copy)
    process_bytes(data.data(), data.size());
    // If you need to keep the data:
    // my_buffer.insert(my_buffer.end(), data.begin(), data.end());
});

Definition at line 35 of file tcp_socket.cpp.

    {
        auto new_cb = std::make_shared<receive_callback_view_t>(std::move(callback));
        std::lock_guard<std::mutex> lock(callback_mutex_);
        std::atomic_store(&receive_callback_view_, new_cb);
    }

socket()

auto kcenon::network::internal::tcp_socket::socket ( ) -> asio::ip::tcp::socket&

inline

Provides direct access to the underlying asio::ip::tcp::socket in case advanced operations are needed.

Returns

A reference to the wrapped asio::ip::tcp::socket.

Definition at line 269 of file tcp_socket.h.

{ return socket_; }

References socket_.

start_read()

auto kcenon::network::internal::tcp_socket::start_read

(

)

-> void

Begins the continuous asynchronous read loop.

Once called, the class repeatedly calls async_read_some(). If an error occurs, on_error() is triggered, stopping further reads.

Definition at line 51 of file tcp_socket.cpp.

    {
        // Prevent duplicate read operations - only start if not already reading
        bool expected = false;
        if (!is_reading_.compare_exchange_strong(expected, true))
        {
            // Already reading, don't start another async operation
            return;
        }
 
        // Post the first read operation to the socket's executor to ensure
        // the executor's internal state (descriptor_state) is fully initialized.
        // This prevents SEGV errors when start_read() is called immediately
        // after socket construction from a moved asio::ip::tcp::socket.
        try
        {
            auto self = shared_from_this();
            asio::post(socket_.get_executor(), [self]() {
                if (self->is_reading_.load() && !self->is_closed_.load())
                {
                    self->do_read();
                }
            });
        }
        catch (...)
        {
            // If post fails, fall back to direct call
            // This may happen if executor is stopped
            is_reading_.store(false);
        }
    }

stop_read()

auto kcenon::network::internal::tcp_socket::stop_read

(

)

-> void

Stops the read loop to prevent further async operations.

Definition at line 83 of file tcp_socket.cpp.

    {
        // Stop further read operations
        is_reading_.store(false);
    }

try_send()

auto kcenon::network::internal::tcp_socket::try_send ( std::vector< uint8_t > && data, std::function< void(std::error_code, std::size_t)> handler ) -> bool

nodiscard

Attempts to send data without blocking.

Parameters

data	The buffer to send (moved for efficiency).
handler	Completion handler for async operation.

Returns

true if send was initiated, false if backpressure is active.

Unlike async_send(), this method checks backpressure limits before initiating the send. Returns false immediately if:

• max_pending_bytes is set and would be exceeded
• Backpressure is currently active

Example

if (!sock->try_send(std::move(data), handler)) {
    // Queue data for later or drop it
}

Definition at line 389 of file tcp_socket.cpp.

{
    std::size_t data_size = data.size();
    std::size_t current = pending_bytes_.load();
 
    // Check if max_pending_bytes limit would be exceeded
    if (config_.max_pending_bytes > 0 &&
        current + data_size > config_.max_pending_bytes)
    {
        metrics_.rejected_sends.fetch_add(1);
        return false;
    }
 
    // Proceed with async_send
    async_send(std::move(data), std::move(handler));
    return true;
}

Member Data Documentation

backpressure_active_

std::atomic<bool> kcenon::network::internal::tcp_socket::backpressure_active_ {false}

private

Backpressure state flag.

Definition at line 341 of file tcp_socket.h.

{false};

backpressure_callback_

std::shared_ptr<backpressure_callback> kcenon::network::internal::tcp_socket::backpressure_callback_

private

Backpressure notification callback.

Definition at line 344 of file tcp_socket.h.

callback_mutex_

std::mutex kcenon::network::internal::tcp_socket::callback_mutex_

private

Protects callback registration only.

Definition at line 320 of file tcp_socket.h.

config_

socket_config kcenon::network::internal::tcp_socket::config_

private

Backpressure configuration.

Definition at line 323 of file tcp_socket.h.

error_callback_

std::shared_ptr<error_callback_t> kcenon::network::internal::tcp_socket::error_callback_

private

Definition at line 318 of file tcp_socket.h.

is_closed_

std::atomic<bool> kcenon::network::internal::tcp_socket::is_closed_ {false}

private

Flag to indicate socket is closed.

Definition at line 335 of file tcp_socket.h.

{false};  

is_reading_

std::atomic<bool> kcenon::network::internal::tcp_socket::is_reading_ {false}

private

Flag to prevent read after stop.

Definition at line 334 of file tcp_socket.h.

{false}; 

metrics_

socket_metrics kcenon::network::internal::tcp_socket::metrics_

mutableprivate

Socket runtime metrics.

Definition at line 326 of file tcp_socket.h.

observers_

std::vector<std::weak_ptr<network_core::interfaces::socket_observer> > kcenon::network::internal::tcp_socket::observers_ {}

private

List of socket observers (using weak_ptr for automatic cleanup). Protected by callback_mutex_.

Definition at line 332 of file tcp_socket.h.

{};

pending_bytes_

std::atomic<std::size_t> kcenon::network::internal::tcp_socket::pending_bytes_ {0}

private

Current pending bytes in send buffer.

Definition at line 338 of file tcp_socket.h.

{0};

read_buffer_

std::array<uint8_t, 4096> kcenon::network::internal::tcp_socket::read_buffer_

private

Buffer for receiving data in do_read().

Definition at line 310 of file tcp_socket.h.

receive_callback_

std::shared_ptr<receive_callback_t> kcenon::network::internal::tcp_socket::receive_callback_

private

Lock-free callback storage using shared_ptr + atomic operations. This eliminates mutex contention on the receive hot path.

Definition at line 316 of file tcp_socket.h.

receive_callback_view_

std::shared_ptr<receive_callback_view_t> kcenon::network::internal::tcp_socket::receive_callback_view_

private

Definition at line 317 of file tcp_socket.h.

socket_

asio::ip::tcp::socket kcenon::network::internal::tcp_socket::socket_

private

The underlying ASIO TCP socket.

Definition at line 307 of file tcp_socket.h.

Referenced by socket().

---

The documentation for this class was generated from the following files:

• src/internal/tcp/tcp_socket.h
• src/internal/tcp_socket.cpp