kcenon::thread::backpressure_job_queue Class Reference

A job queue with comprehensive backpressure mechanisms. More...

#include <backpressure_job_queue.h>

Inheritance diagram for kcenon::thread::backpressure_job_queue:

Collaboration diagram for kcenon::thread::backpressure_job_queue:

Public Member Functions
	backpressure_job_queue (std::size_t max_size, backpressure_config config=backpressure_config{})
	Constructs a backpressure-aware job queue.
	~backpressure_job_queue () override
	Virtual destructor.
	backpressure_job_queue (const backpressure_job_queue &)=delete
backpressure_job_queue &	operator= (const backpressure_job_queue &)=delete
	backpressure_job_queue (backpressure_job_queue &&)=delete
backpressure_job_queue &	operator= (backpressure_job_queue &&)=delete
auto	enqueue (std::unique_ptr< job > &&value) -> common::VoidResult override
	Enqueues a job with backpressure handling.
auto	enqueue_batch (std::vector< std::unique_ptr< job > > &&jobs) -> common::VoidResult override
	Enqueues a batch of jobs with backpressure handling.
auto	get_pressure_level () const -> pressure_level
	Returns the current pressure level.
auto	get_pressure_ratio () const -> double
	Returns the current pressure as a ratio.
auto	set_backpressure_config (backpressure_config config) -> void
	Sets the backpressure configuration.
auto	get_backpressure_config () const -> const backpressure_config &
	Returns the current backpressure configuration.
auto	is_rate_limited () const -> bool
	Checks if rate limiting is causing delays.
auto	get_available_tokens () const -> std::size_t
	Returns available rate limit tokens.
auto	get_backpressure_stats () const -> backpressure_stats_snapshot
	Returns backpressure statistics snapshot.
auto	reset_stats () -> void
	Resets backpressure statistics.
auto	to_string () const -> std::string override
	Returns string representation including backpressure state.
Public Member Functions inherited from kcenon::thread::job_queue
	job_queue (std::optional< std::size_t > max_size=std::nullopt)
	Constructs a new, empty job_queue.
virtual	~job_queue (void)
	Virtual destructor. Cleans up resources used by the job_queue.
auto	get_ptr (void) -> std::shared_ptr< job_queue >
	Obtains a std::shared_ptr that points to this queue instance.
auto	is_stopped () const -> bool
	Checks if the queue is in a "stopped" state.
auto	set_notify (bool notify) -> void
	Sets the 'notify' flag for this queue.
auto	schedule (std::unique_ptr< job > &&value) -> common::VoidResult override
	scheduler_interface implementation: schedule a job
auto	get_next_job () -> common::Result< std::unique_ptr< job > > override
	scheduler_interface implementation: get next job
template<typename JobType , typename = std::enable_if_t<std::is_base_of_v<job, JobType>>>
auto	enqueue (std::unique_ptr< JobType > &&value) -> common::VoidResult
	Type-safe enqueue for job subclasses.
virtual auto	dequeue (void) -> common::Result< std::unique_ptr< job > >
	Dequeues a job from the queue in FIFO order (blocking operation).
virtual auto	try_dequeue (void) -> common::Result< std::unique_ptr< job > >
	Attempts to dequeue a job from the queue without blocking.
virtual auto	dequeue_batch (void) -> std::deque< std::unique_ptr< job > >
	Dequeues all remaining jobs from the queue without processing them.
virtual auto	dequeue_batch_limited (std::size_t max_count) -> std::deque< std::unique_ptr< job > >
	Dequeues up to N jobs in a single operation (micro-batching).
virtual auto	clear (void) -> void
	Removes all jobs currently in the queue without processing them.
auto	empty (void) const -> bool
	Checks if the queue is currently empty.
auto	size (void) const -> std::size_t
	Returns the current number of jobs in the queue.
auto	get_memory_stats () const -> memory_stats
	Get memory footprint statistics for debugging and monitoring.
auto	stop (void) -> void
	Signals the queue to stop waiting for new jobs (e.g., during shutdown).
auto	get_capabilities () const -> queue_capabilities override
	Returns capabilities of this job_queue implementation.
auto	is_bounded () const -> bool
	Check if queue has a size limit.
auto	get_max_size () const -> std::optional< std::size_t >
	Get the maximum queue size.
auto	set_max_size (std::optional< std::size_t > max_size) -> void
	Set maximum queue size.
auto	is_full () const -> bool
	Check if queue is at capacity.
virtual auto	inspect_pending_jobs (std::size_t limit=100) const -> std::vector< diagnostics::job_info >
	Inspects pending jobs in the queue without removing them.
Public Member Functions inherited from kcenon::thread::scheduler_interface
virtual	~scheduler_interface ()=default
Public Member Functions inherited from kcenon::thread::queue_capabilities_interface
virtual	~queue_capabilities_interface ()=default
auto	has_exact_size () const -> bool
	Check if size() returns exact values.
auto	has_atomic_empty () const -> bool
	Check if empty() check is atomic.
auto	is_lock_free () const -> bool
	Check if this is a lock-free implementation.
auto	is_wait_free () const -> bool
	Check if this is a wait-free implementation.
auto	supports_batch () const -> bool
	Check if batch operations are supported.
auto	supports_blocking_wait () const -> bool
	Check if blocking wait is supported.
auto	supports_stop () const -> bool
	Check if stop signaling is supported.

Private Member Functions
auto	apply_backpressure (std::unique_ptr< job > &&value) -> common::VoidResult
	Applies backpressure logic for a single job.
auto	apply_rate_limiting () -> bool
	Applies rate limiting check.
auto	update_pressure_state () -> void
	Updates pressure level and triggers callbacks if changed.
auto	handle_block_policy (std::unique_ptr< job > &&value) -> common::VoidResult
	Handles blocking policy with timeout.
auto	handle_drop_oldest_policy (std::unique_ptr< job > &&value) -> common::VoidResult
	Handles drop_oldest policy.
auto	handle_callback_policy (std::unique_ptr< job > &&value) -> common::VoidResult
	Handles callback policy.
auto	handle_adaptive_policy (std::unique_ptr< job > &&value) -> common::VoidResult
	Handles adaptive policy.
auto	direct_enqueue (std::unique_ptr< job > &&value) -> common::VoidResult
	Directly enqueues without backpressure (internal use).

Private Attributes
backpressure_config	config_
	Backpressure configuration.
std::mutex	config_mutex_
	Mutex for configuration access.
std::unique_ptr< token_bucket >	rate_limiter_
	Token bucket rate limiter (nullptr if disabled).
std::atomic< pressure_level >	current_pressure_ {pressure_level::none}
	Current pressure level (atomic for lock-free reads).
std::atomic< double >	current_pressure_ratio_ {0.0}
	Current pressure ratio (atomic for lock-free reads).
backpressure_stats	stats_
	Backpressure statistics.
std::condition_variable	space_available_
	Condition variable for blocking policy wait.

Additional Inherited Members
Protected Attributes inherited from kcenon::thread::job_queue
std::atomic_bool	notify_
	If true, threads waiting for new jobs are notified when a new job is enqueued. If false, enqueuing does not automatically trigger a notification.
std::atomic_bool	stop_
	Indicates whether the queue has been signaled to stop.
std::mutex	mutex_
	Mutex to protect access to the underlying queue_ container and related state.
std::condition_variable	condition_
	Condition variable used to signal worker threads.

Detailed Description

A job queue with comprehensive backpressure mechanisms.

Extends job_queue with backpressure handling to prevent resource exhaustion and provide graceful degradation under high load.

Features

• Multiple Policies: Block, drop_oldest, drop_newest, callback, adaptive
• Watermark-Based Pressure: Graduated response based on queue depth
• Rate Limiting: Token bucket algorithm for sustained throughput control
• Adaptive Control: Auto-adjusts based on latency targets
• Statistics: Comprehensive metrics for monitoring

Pressure Response

Queue Depth vs Response:
 
0%     50%      80%      100%
|------|--------|--------|
  OK    Warning   High   Critical
       (callback) (slow) (reject)

Thread Safety

All methods are thread-safe. The queue inherits mutex-based synchronization from job_queue.

Usage Example

// Create backpressure queue with 1000 max size
backpressure_config config;
config.policy = backpressure_policy::adaptive;
config.high_watermark = 0.8;
config.enable_rate_limiting = true;
config.rate_limit_tokens_per_second = 5000;
 
auto queue = std::make_shared<backpressure_job_queue>(1000, config);
 
// Enqueue with backpressure handling
auto result = queue->enqueue(std::make_unique<my_job>());
if (result.is_err()) {
    // Handle backpressure (rejected, timeout, etc.)
}
 
// Check pressure level
if (queue->get_pressure_level() == pressure_level::high) {
    // Consider reducing load
}

See also

job_queue Base queue class

backpressure_config Configuration options

token_bucket Rate limiting implementation

Definition at line 87 of file backpressure_job_queue.h.

Constructor & Destructor Documentation

backpressure_job_queue() [1/3]

kcenon::thread::backpressure_job_queue::backpressure_job_queue ( std::size_t max_size, backpressure_config config = backpressure_config{} )

explicit

Constructs a backpressure-aware job queue.

Parameters

max_size	Maximum queue capacity.
config	Backpressure configuration.

The queue is immediately ready for use after construction. If rate limiting is enabled, the token bucket is initialized with the specified parameters.

Implementation details:

• Calls base job_queue constructor with max_size
• Stores configuration
• Creates token bucket if rate limiting is enabled
• Initializes pressure state to none

Parameters

max_size	Maximum queue capacity
config	Backpressure configuration

Definition at line 34 of file backpressure_job_queue.cpp.

        : job_queue(max_size)
        , config_(std::move(config))
        , current_pressure_(pressure_level::none)
        , current_pressure_ratio_(0.0)
    {
        // Initialize rate limiter if enabled
        if (config_.enable_rate_limiting)
        {
            rate_limiter_ = std::make_unique<token_bucket>(
                config_.rate_limit_tokens_per_second,
                config_.rate_limit_burst_size);
        }
    }

References config_, and kcenon::thread::backpressure_config::enable_rate_limiting.

~backpressure_job_queue()

kcenon::thread::backpressure_job_queue::~backpressure_job_queue ( )

overridedefault

Virtual destructor.

Destructor.

backpressure_job_queue() [2/3]

kcenon::thread::backpressure_job_queue::backpressure_job_queue ( const backpressure_job_queue & )

delete

backpressure_job_queue() [3/3]

kcenon::thread::backpressure_job_queue::backpressure_job_queue ( backpressure_job_queue && )

delete

Member Function Documentation

apply_backpressure()

auto kcenon::thread::backpressure_job_queue::apply_backpressure ( std::unique_ptr< job > && value ) -> common::VoidResult

nodiscardprivate

Applies backpressure logic for a single job.

Core backpressure implementation for single job.

Parameters

value

The job to potentially enqueue.

Returns

VoidResult indicating success or error.

This is the core backpressure implementation that handles all policy logic, rate limiting, and statistics tracking.

Implementation details:

• Applies rate limiting first
• Checks capacity before attempting enqueue
• Routes to policy handler if queue is full
• Updates statistics

Parameters

value

The job to enqueue

Returns

VoidResult indicating success or error

Definition at line 252 of file backpressure_job_queue.cpp.

    {
        // Apply rate limiting
        if (!apply_rate_limiting())
        {
            stats_.jobs_rejected.fetch_add(1, std::memory_order_relaxed);
            return make_error_result(error_code::operation_timeout,
                "rate limit exceeded");
        }
 
        // Check if queue is full before deciding on policy
        bool queue_is_full = is_full();
 
        // If not full, try direct enqueue
        if (!queue_is_full)
        {
            auto result = direct_enqueue(std::move(value));
            if (result.is_ok())
            {
                stats_.jobs_accepted.fetch_add(1, std::memory_order_relaxed);
                update_pressure_state();
                return result;
            }
            // If direct_enqueue failed but queue wasn't reported as full,
            // it might be a race condition - treat as full
            queue_is_full = true;
        }
 
        // Queue is full, apply policy
        switch (config_.policy)
        {
            case backpressure_policy::block:
                return handle_block_policy(std::move(value));
 
            case backpressure_policy::drop_oldest:
                return handle_drop_oldest_policy(std::move(value));
 
            case backpressure_policy::drop_newest:
                stats_.jobs_rejected.fetch_add(1, std::memory_order_relaxed);
                stats_.pressure_events.fetch_add(1, std::memory_order_relaxed);
                update_pressure_state();
                return make_error_result(error_code::queue_full,
                    "queue full: drop_newest policy rejected job");
 
            case backpressure_policy::callback:
                return handle_callback_policy(std::move(value));
 
            case backpressure_policy::adaptive:
                return handle_adaptive_policy(std::move(value));
 
            default:
                stats_.jobs_rejected.fetch_add(1, std::memory_order_relaxed);
                return make_error_result(error_code::queue_full,
                    "queue full: unknown policy");
        }
    }

References kcenon::thread::adaptive, kcenon::thread::block, kcenon::thread::callback, kcenon::thread::drop_newest, kcenon::thread::drop_oldest, kcenon::thread::result< T >::is_ok(), kcenon::thread::make_error_result(), kcenon::thread::operation_timeout, and kcenon::thread::queue_full.

Here is the call graph for this function:

apply_rate_limiting()

auto kcenon::thread::backpressure_job_queue::apply_rate_limiting ( ) -> bool

nodiscardprivate

Applies rate limiting check.

Applies rate limiting using token bucket.

Returns

true if rate limit allows operation, false if should wait.

true if allowed to proceed, false if rate limited

Definition at line 315 of file backpressure_job_queue.cpp.

    {
        if (!config_.enable_rate_limiting || !rate_limiter_)
        {
            return true;  // No rate limiting
        }
 
        // Try immediate acquire
        if (rate_limiter_->try_acquire())
        {
            return true;
        }
 
        // Rate limited - record and try waiting
        stats_.rate_limit_waits.fetch_add(1, std::memory_order_relaxed);
 
        auto start = std::chrono::steady_clock::now();
        bool acquired = rate_limiter_->try_acquire_for(1, config_.block_timeout);
        auto elapsed = std::chrono::steady_clock::now() - start;
 
        stats_.total_block_time_ns.fetch_add(
            std::chrono::duration_cast<std::chrono::nanoseconds>(elapsed).count(),
            std::memory_order_relaxed);
 
        return acquired;
    }

direct_enqueue()

auto kcenon::thread::backpressure_job_queue::direct_enqueue ( std::unique_ptr< job > && value ) -> common::VoidResult

nodiscardprivate

Directly enqueues without backpressure (internal use).

Directly enqueues to base class without backpressure.

Parameters

value

The job to enqueue.

Returns

VoidResult from base class enqueue.

Definition at line 598 of file backpressure_job_queue.cpp.

    {
        return job_queue::enqueue(std::move(value));
    }

References kcenon::thread::job_queue::enqueue().

Here is the call graph for this function:

enqueue()

auto kcenon::thread::backpressure_job_queue::enqueue ( std::unique_ptr< job > && value ) -> common::VoidResult

nodiscardoverridevirtual

Enqueues a job with backpressure handling.

Parameters

value

The job to enqueue.

Returns

VoidResult indicating success or error.

Depending on the configured policy:

• block: Waits up to block_timeout for space
• drop_oldest: Removes oldest job if full
• drop_newest: Rejects new job if full
• callback: Calls decision_callback for custom handling
• adaptive: Adjusts behavior based on current conditions

Rate limiting (if enabled) is applied before policy-based handling.

Error Codes:

• queue_full: Queue at capacity and job rejected
• queue_stopped: Queue has been stopped
• operation_timeout: Block timeout expired
• invalid_argument: Null job provided

Implementation details:

• Early validation (stopped, null check)
• Apply rate limiting if enabled
• Route to policy-specific handler based on config
• Update pressure state and statistics

Parameters

value

The job to enqueue

Returns

VoidResult indicating success or error

Reimplemented from kcenon::thread::job_queue.

Definition at line 68 of file backpressure_job_queue.cpp.

    {
        // Early validation
        if (is_stopped())
        {
            return make_error_result(error_code::queue_stopped);
        }
 
        if (value == nullptr)
        {
            return make_error_result(error_code::invalid_argument, "cannot enqueue null job");
        }
 
        // Apply backpressure logic
        return apply_backpressure(std::move(value));
    }

References kcenon::thread::invalid_argument, kcenon::thread::make_error_result(), and kcenon::thread::queue_stopped.

Here is the call graph for this function:

enqueue_batch()

auto kcenon::thread::backpressure_job_queue::enqueue_batch ( std::vector< std::unique_ptr< job > > && jobs ) -> common::VoidResult

nodiscardoverridevirtual

Enqueues a batch of jobs with backpressure handling.

Parameters

jobs	Vector of jobs to enqueue.

Returns

VoidResult indicating success or error.

Batch enqueue respects backpressure settings. If the batch would exceed capacity, behavior depends on policy:

• block/drop_newest: Entire batch rejected
• drop_oldest: Oldest jobs dropped to make room
• callback: Called once for the batch decision
• adaptive: May partially accept (future enhancement)

Implementation details:

• Validates batch (stopped, empty, null jobs)
• Checks if batch fits within capacity constraints
• For drop_oldest: drops enough jobs to fit batch
• For other policies: rejects entire batch if won't fit

Parameters

jobs	Vector of jobs to enqueue

Returns

VoidResult indicating success or error

Reimplemented from kcenon::thread::job_queue.

Definition at line 97 of file backpressure_job_queue.cpp.

    {
        if (is_stopped())
        {
            return make_error_result(error_code::queue_stopped);
        }
 
        if (jobs.empty())
        {
            return make_error_result(error_code::invalid_argument, "cannot enqueue empty batch");
        }
 
        // Validate all jobs
        for (const auto& job_ptr : jobs)
        {
            if (job_ptr == nullptr)
            {
                return make_error_result(error_code::invalid_argument,
                    "cannot enqueue null job in batch");
            }
        }
 
        // Apply rate limiting for batch
        if (config_.enable_rate_limiting && rate_limiter_)
        {
            if (!rate_limiter_->try_acquire(jobs.size()))
            {
                stats_.rate_limit_waits.fetch_add(1, std::memory_order_relaxed);
 
                // Try waiting for tokens
                auto start = std::chrono::steady_clock::now();
                if (!rate_limiter_->try_acquire_for(jobs.size(), config_.block_timeout))
                {
                    stats_.jobs_rejected.fetch_add(jobs.size(), std::memory_order_relaxed);
                    return make_error_result(error_code::operation_timeout,
                        "rate limit timeout for batch");
                }
                auto elapsed = std::chrono::steady_clock::now() - start;
                stats_.total_block_time_ns.fetch_add(
                    std::chrono::duration_cast<std::chrono::nanoseconds>(elapsed).count(),
                    std::memory_order_relaxed);
            }
        }
 
        // Check capacity
        auto max_sz = get_max_size();
        if (!max_sz.has_value())
        {
            // Unbounded queue, just enqueue
            auto result = job_queue::enqueue_batch(std::move(jobs));
            if (result.is_ok())
            {
                stats_.jobs_accepted.fetch_add(jobs.size(), std::memory_order_relaxed);
            }
            update_pressure_state();
            return result;
        }
 
        std::size_t current_size = size();
        std::size_t available_space = max_sz.value() > current_size
            ? max_sz.value() - current_size
            : 0;
 
        if (jobs.size() <= available_space)
        {
            // Batch fits
            auto result = job_queue::enqueue_batch(std::move(jobs));
            if (result.is_ok())
            {
                stats_.jobs_accepted.fetch_add(jobs.size(), std::memory_order_relaxed);
            }
            update_pressure_state();
            return result;
        }
 
        // Batch doesn't fit - handle based on policy
        switch (config_.policy)
        {
            case backpressure_policy::drop_oldest:
            {
                // Drop enough jobs to make room
                std::size_t to_drop = jobs.size() - available_space;
                {
                    std::scoped_lock<std::mutex> lock(mutex_);
                    for (std::size_t i = 0; i < to_drop && !empty(); ++i)
                    {
                        auto batch = dequeue_batch_limited(1);
                        stats_.jobs_dropped.fetch_add(batch.size(), std::memory_order_relaxed);
                    }
                }
                auto result = job_queue::enqueue_batch(std::move(jobs));
                if (result.is_ok())
                {
                    stats_.jobs_accepted.fetch_add(jobs.size(), std::memory_order_relaxed);
                }
                update_pressure_state();
                return result;
            }
 
            case backpressure_policy::block:
            {
                // Wait for space (simplified: just check once with timeout)
                auto start = std::chrono::steady_clock::now();
                std::unique_lock<std::mutex> lock(mutex_);
                bool got_space = space_available_.wait_for(
                    lock,
                    config_.block_timeout,
                    [this, needed = jobs.size(), max = max_sz.value()]() {
                        return size() + needed <= max || is_stopped();
                    });
 
                if (!got_space || is_stopped())
                {
                    stats_.jobs_rejected.fetch_add(jobs.size(), std::memory_order_relaxed);
                    auto elapsed = std::chrono::steady_clock::now() - start;
                    stats_.total_block_time_ns.fetch_add(
                        std::chrono::duration_cast<std::chrono::nanoseconds>(elapsed).count(),
                        std::memory_order_relaxed);
                    return make_error_result(error_code::operation_timeout,
                        "timeout waiting for space for batch");
                }
 
                lock.unlock();
                auto result = job_queue::enqueue_batch(std::move(jobs));
                if (result.is_ok())
                {
                    stats_.jobs_accepted.fetch_add(jobs.size(), std::memory_order_relaxed);
                }
                update_pressure_state();
                return result;
            }
 
            default:
                // drop_newest, callback, adaptive - reject batch
                stats_.jobs_rejected.fetch_add(jobs.size(), std::memory_order_relaxed);
                stats_.pressure_events.fetch_add(1, std::memory_order_relaxed);
                update_pressure_state();
                return make_error_result(error_code::queue_full,
                    "queue cannot fit entire batch");
        }
    }

References kcenon::thread::block, kcenon::thread::drop_oldest, kcenon::thread::job_queue::enqueue_batch(), kcenon::thread::invalid_argument, kcenon::thread::result< T >::is_ok(), kcenon::thread::make_error_result(), kcenon::thread::operation_timeout, kcenon::thread::queue_full, and kcenon::thread::queue_stopped.

Here is the call graph for this function:

get_available_tokens()

auto kcenon::thread::backpressure_job_queue::get_available_tokens ( ) const -> std::size_t

nodiscard

Returns available rate limit tokens.

Returns

Available tokens, or max if rate limiting disabled.

Definition at line 684 of file backpressure_job_queue.cpp.

    {
        if (!config_.enable_rate_limiting || !rate_limiter_)
        {
            return std::numeric_limits<std::size_t>::max();
        }
        return rate_limiter_->available_tokens();
    }

References config_, kcenon::thread::backpressure_config::enable_rate_limiting, and rate_limiter_.

get_backpressure_config()

auto kcenon::thread::backpressure_job_queue::get_backpressure_config ( ) const -> const backpressure_config&

nodiscard

Returns the current backpressure configuration.

Returns current configuration.

Returns

Reference to current configuration.

Definition at line 662 of file backpressure_job_queue.cpp.

    {
        return config_;
    }

References config_.

get_backpressure_stats()

auto kcenon::thread::backpressure_job_queue::get_backpressure_stats ( ) const -> backpressure_stats_snapshot

nodiscard

Returns backpressure statistics snapshot.

Returns

Snapshot of current statistics.

Returns a snapshot of current stats. For ongoing monitoring, call periodically.

Definition at line 696 of file backpressure_job_queue.cpp.

    {
        return stats_.snapshot();
    }

References kcenon::thread::backpressure_stats::snapshot(), and stats_.

Here is the call graph for this function:

get_pressure_level()

auto kcenon::thread::backpressure_job_queue::get_pressure_level ( ) const -> pressure_level

nodiscard

Returns the current pressure level.

Returns current pressure level.

Returns

Current pressure_level enum value.

Calculated based on current queue depth relative to watermarks:

• none: depth < low_watermark * max_size
• low: depth < high_watermark * max_size
• high: depth < max_size
• critical: depth >= max_size

Definition at line 607 of file backpressure_job_queue.cpp.

    {
        return current_pressure_.load(std::memory_order_acquire);
    }

References current_pressure_.

Referenced by to_string().

Here is the caller graph for this function:

get_pressure_ratio()

auto kcenon::thread::backpressure_job_queue::get_pressure_ratio ( ) const -> double

nodiscard

Returns the current pressure as a ratio.

Returns current pressure ratio.

Returns

Ratio of current depth to max_size (0.0 to 1.0+).

Can exceed 1.0 if queue somehow exceeds max_size (shouldn't happen with proper backpressure, but included for robustness).

Definition at line 615 of file backpressure_job_queue.cpp.

    {
        // Recalculate for accuracy
        auto max_sz = get_max_size();
        if (!max_sz.has_value() || max_sz.value() == 0)
        {
            return 0.0;
        }
        return static_cast<double>(size()) / static_cast<double>(max_sz.value());
    }

References kcenon::thread::job_queue::get_max_size(), and kcenon::thread::job_queue::size().

Referenced by to_string().

Here is the call graph for this function:

Here is the caller graph for this function:

handle_adaptive_policy()

auto kcenon::thread::backpressure_job_queue::handle_adaptive_policy ( std::unique_ptr< job > && value ) -> common::VoidResult

nodiscardprivate

Handles adaptive policy.

Parameters

value

The job to potentially enqueue.

Returns

VoidResult based on adaptive decision.

Implementation details:

• Below high_watermark: accept normally
• Above high_watermark: probabilistic acceptance based on pressure
• At capacity: block briefly then reject

Definition at line 525 of file backpressure_job_queue.cpp.

    {
        double ratio = get_pressure_ratio();
 
        // Below high watermark: accept
        if (ratio < config_.high_watermark)
        {
            auto result = direct_enqueue(std::move(value));
            if (result.is_ok())
            {
                stats_.jobs_accepted.fetch_add(1, std::memory_order_relaxed);
            }
            update_pressure_state();
            return result;
        }
 
        // Above high watermark but not full: probabilistic acceptance
        // Acceptance probability decreases as we approach capacity
        if (ratio < 1.0)
        {
            // Simple linear decrease from 1.0 at high_watermark to 0 at 1.0
            double accept_prob = (1.0 - ratio) / (1.0 - config_.high_watermark);
 
            // Use simple threshold (not truly random, but deterministic)
            // This avoids random number generation overhead
            static thread_local std::size_t counter = 0;
            bool should_accept = (counter++ % 100) < (accept_prob * 100);
 
            if (should_accept)
            {
                auto result = direct_enqueue(std::move(value));
                if (result.is_ok())
                {
                    stats_.jobs_accepted.fetch_add(1, std::memory_order_relaxed);
                }
                update_pressure_state();
                return result;
            }
        }
 
        // At or above capacity: brief wait then reject
        {
            std::unique_lock<std::mutex> lock(mutex_);
            auto brief_wait = std::chrono::milliseconds{10};
            bool got_space = space_available_.wait_for(
                lock,
                brief_wait,
                [this]() { return !is_full() || is_stopped(); });
 
            if (got_space && !is_stopped())
            {
                lock.unlock();
                auto result = direct_enqueue(std::move(value));
                if (result.is_ok())
                {
                    stats_.jobs_accepted.fetch_add(1, std::memory_order_relaxed);
                }
                update_pressure_state();
                return result;
            }
        }
 
        stats_.jobs_rejected.fetch_add(1, std::memory_order_relaxed);
        stats_.pressure_events.fetch_add(1, std::memory_order_relaxed);
        update_pressure_state();
        return make_error_result(error_code::queue_full,
            "adaptive policy rejected job due to high pressure");
    }

References kcenon::thread::result< T >::is_ok(), kcenon::thread::make_error_result(), and kcenon::thread::queue_full.

Here is the call graph for this function:

handle_block_policy()

auto kcenon::thread::backpressure_job_queue::handle_block_policy ( std::unique_ptr< job > && value ) -> common::VoidResult

nodiscardprivate

Handles blocking policy with timeout.

Parameters

value

The job to enqueue.

Returns

VoidResult indicating success or timeout.

Definition at line 405 of file backpressure_job_queue.cpp.

    {
        auto start = std::chrono::steady_clock::now();
        auto deadline = start + config_.block_timeout;
 
        while (std::chrono::steady_clock::now() < deadline)
        {
            // Try to enqueue
            auto result = direct_enqueue(std::move(value));
            if (result.is_ok())
            {
                stats_.jobs_accepted.fetch_add(1, std::memory_order_relaxed);
                update_pressure_state();
                return result;
            }
 
            // Wait for space
            std::unique_lock<std::mutex> lock(mutex_);
            auto remaining = deadline - std::chrono::steady_clock::now();
            if (remaining <= std::chrono::milliseconds{0})
            {
                break;
            }
 
            space_available_.wait_for(lock, remaining, [this]() {
                return !is_full() || is_stopped();
            });
 
            if (is_stopped())
            {
                break;
            }
 
            // Job was moved, need to return error
            // This is a design issue - we can't retry after move
            break;
        }
 
        auto elapsed = std::chrono::steady_clock::now() - start;
        stats_.total_block_time_ns.fetch_add(
            std::chrono::duration_cast<std::chrono::nanoseconds>(elapsed).count(),
            std::memory_order_relaxed);
        stats_.jobs_rejected.fetch_add(1, std::memory_order_relaxed);
 
        return make_error_result(error_code::operation_timeout,
            "timeout waiting for queue space");
    }

References kcenon::thread::result< T >::is_ok(), kcenon::thread::make_error_result(), and kcenon::thread::operation_timeout.

Here is the call graph for this function:

handle_callback_policy()

auto kcenon::thread::backpressure_job_queue::handle_callback_policy ( std::unique_ptr< job > && value ) -> common::VoidResult

nodiscardprivate

Handles callback policy.

Parameters

value

The job to potentially enqueue.

Returns

VoidResult based on callback decision.

Definition at line 477 of file backpressure_job_queue.cpp.

    {
        if (!config_.decision_callback)
        {
            // No callback, fall back to reject
            stats_.jobs_rejected.fetch_add(1, std::memory_order_relaxed);
            return make_error_result(error_code::queue_full,
                "queue full and no decision callback");
        }
 
        backpressure_decision decision = config_.decision_callback(value);
 
        switch (decision)
        {
            case backpressure_decision::accept:
                // Force accept (may exceed max_size temporarily)
                return job_queue::enqueue(std::move(value));
 
            case backpressure_decision::reject:
                stats_.jobs_rejected.fetch_add(1, std::memory_order_relaxed);
                return make_error_result(error_code::queue_full,
                    "callback rejected job");
 
            case backpressure_decision::drop_and_accept:
                return handle_drop_oldest_policy(std::move(value));
 
            case backpressure_decision::delay:
                // For now, treat as reject with retry hint
                stats_.jobs_rejected.fetch_add(1, std::memory_order_relaxed);
                return make_error_result(error_code::queue_busy,
                    "callback requested delay, retry later");
 
            default:
                stats_.jobs_rejected.fetch_add(1, std::memory_order_relaxed);
                return make_error_result(error_code::queue_full,
                    "unknown callback decision");
        }
    }

References kcenon::thread::accept, kcenon::thread::delay, kcenon::thread::drop_and_accept, kcenon::thread::job_queue::enqueue(), kcenon::thread::make_error_result(), kcenon::thread::queue_busy, kcenon::thread::queue_full, and kcenon::thread::reject.

Here is the call graph for this function:

handle_drop_oldest_policy()

auto kcenon::thread::backpressure_job_queue::handle_drop_oldest_policy ( std::unique_ptr< job > && value ) -> common::VoidResult

nodiscardprivate

Handles drop_oldest policy.

Parameters

value

The job to enqueue after dropping oldest.

Returns

VoidResult (always succeeds unless stopped).

Definition at line 457 of file backpressure_job_queue.cpp.

    {
        // Drop oldest job - dequeue_batch_limited handles its own locking
        auto dropped = dequeue_batch_limited(1);
        stats_.jobs_dropped.fetch_add(dropped.size(), std::memory_order_relaxed);
 
        // Now enqueue new job
        auto result = direct_enqueue(std::move(value));
        if (result.is_ok())
        {
            stats_.jobs_accepted.fetch_add(1, std::memory_order_relaxed);
        }
        update_pressure_state();
        return result;
    }

References kcenon::thread::result< T >::is_ok().

Here is the call graph for this function:

is_rate_limited()

auto kcenon::thread::backpressure_job_queue::is_rate_limited ( ) const -> bool

nodiscard

Checks if rate limiting is causing delays.

Checks if rate limiting is active.

Returns

true if rate limiter is constraining throughput.

Returns false if rate limiting is disabled.

Definition at line 670 of file backpressure_job_queue.cpp.

    {
        if (!config_.enable_rate_limiting || !rate_limiter_)
        {
            return false;
        }
        // Rate limited if tokens are low (less than burst size * 10%)
        return rate_limiter_->available_tokens() <
            (config_.rate_limit_burst_size / 10);
    }

References config_, kcenon::thread::backpressure_config::enable_rate_limiting, kcenon::thread::backpressure_config::rate_limit_burst_size, and rate_limiter_.

operator=() [1/2]

backpressure_job_queue & kcenon::thread::backpressure_job_queue::operator= ( backpressure_job_queue && )

delete

operator=() [2/2]

backpressure_job_queue & kcenon::thread::backpressure_job_queue::operator= ( const backpressure_job_queue & )

delete

reset_stats()

auto kcenon::thread::backpressure_job_queue::reset_stats

(

)

-> void

Resets backpressure statistics.

Resets statistics.

Definition at line 704 of file backpressure_job_queue.cpp.

    {
        stats_.reset();
    }

set_backpressure_config()

auto kcenon::thread::backpressure_job_queue::set_backpressure_config

(

backpressure_config

config

)

-> void

Sets the backpressure configuration.

Sets backpressure configuration.

Parameters

config

New configuration to apply.

Updates take effect immediately. If rate limiting is being enabled or its parameters change, the token bucket is recreated or updated.

Definition at line 629 of file backpressure_job_queue.cpp.

    {
        std::scoped_lock<std::mutex> lock(config_mutex_);
 
        bool rate_limiting_changed =
            config.enable_rate_limiting != config_.enable_rate_limiting ||
            config.rate_limit_tokens_per_second != config_.rate_limit_tokens_per_second ||
            config.rate_limit_burst_size != config_.rate_limit_burst_size;
 
        config_ = std::move(config);
 
        // Update rate limiter if needed
        if (rate_limiting_changed)
        {
            if (config_.enable_rate_limiting)
            {
                rate_limiter_ = std::make_unique<token_bucket>(
                    config_.rate_limit_tokens_per_second,
                    config_.rate_limit_burst_size);
            }
            else
            {
                rate_limiter_.reset();
            }
        }
 
        // Update pressure state with new config
        update_pressure_state();
    }

to_string()

auto kcenon::thread::backpressure_job_queue::to_string ( void ) const -> std::string

nodiscardoverridevirtual

Returns string representation including backpressure state.

Returns string representation.

Returns

Formatted string with queue and backpressure information.

Reimplemented from kcenon::thread::job_queue.

Definition at line 712 of file backpressure_job_queue.cpp.

    {
        return utility_module::formatter::format(
            "backpressure_job_queue[size={}, pressure={}, policy={}, ratio={:.1f}%]",
            size(),
            pressure_level_to_string(get_pressure_level()),
            backpressure_policy_to_string(config_.policy),
            get_pressure_ratio() * 100.0);
    }

References kcenon::thread::backpressure_policy_to_string(), config_, utility_module::formatter::format(), get_pressure_level(), get_pressure_ratio(), kcenon::thread::backpressure_config::policy, kcenon::thread::pressure_level_to_string(), and kcenon::thread::job_queue::size().

Here is the call graph for this function:

update_pressure_state()

auto kcenon::thread::backpressure_job_queue::update_pressure_state ( ) -> void

private

Updates pressure level and triggers callbacks if changed.

Updates pressure state and triggers callbacks.

Definition at line 345 of file backpressure_job_queue.cpp.

    {
        auto max_sz = get_max_size();
        if (!max_sz.has_value() || max_sz.value() == 0)
        {
            current_pressure_.store(pressure_level::none, std::memory_order_relaxed);
            current_pressure_ratio_.store(0.0, std::memory_order_relaxed);
            return;
        }
 
        std::size_t current = size();
        double ratio = static_cast<double>(current) / static_cast<double>(max_sz.value());
        current_pressure_ratio_.store(ratio, std::memory_order_relaxed);
 
        pressure_level new_level;
        if (current >= max_sz.value())
        {
            new_level = pressure_level::critical;
        }
        else if (ratio >= config_.high_watermark)
        {
            new_level = pressure_level::high;
        }
        else if (ratio >= config_.low_watermark)
        {
            new_level = pressure_level::low;
        }
        else
        {
            new_level = pressure_level::none;
        }
 
        pressure_level old_level = current_pressure_.exchange(
            new_level, std::memory_order_acq_rel);
 
        // Trigger callback on level change or when entering high/critical
        if (config_.pressure_callback)
        {
            if (new_level != old_level ||
                new_level == pressure_level::high ||
                new_level == pressure_level::critical)
            {
                if (new_level >= pressure_level::high)
                {
                    stats_.pressure_events.fetch_add(1, std::memory_order_relaxed);
                }
                config_.pressure_callback(current, ratio);
            }
        }
 
        // Notify waiting threads if space became available
        if (new_level < old_level)
        {
            space_available_.notify_all();
        }
    }

References kcenon::thread::critical, kcenon::thread::high, kcenon::thread::low, and kcenon::thread::none.

Member Data Documentation

config_

backpressure_config kcenon::thread::backpressure_job_queue::config_

private

Backpressure configuration.

Definition at line 302 of file backpressure_job_queue.h.

Referenced by backpressure_job_queue(), get_available_tokens(), get_backpressure_config(), is_rate_limited(), and to_string().

config_mutex_

std::mutex kcenon::thread::backpressure_job_queue::config_mutex_

mutableprivate

Mutex for configuration access.

Definition at line 307 of file backpressure_job_queue.h.

current_pressure_

std::atomic<pressure_level> kcenon::thread::backpressure_job_queue::current_pressure_ {pressure_level::none}

private

Current pressure level (atomic for lock-free reads).

Definition at line 317 of file backpressure_job_queue.h.

{pressure_level::none};

Referenced by get_pressure_level().

current_pressure_ratio_

std::atomic<double> kcenon::thread::backpressure_job_queue::current_pressure_ratio_ {0.0}

private

Current pressure ratio (atomic for lock-free reads).

Definition at line 322 of file backpressure_job_queue.h.

{0.0};

rate_limiter_

std::unique_ptr<token_bucket> kcenon::thread::backpressure_job_queue::rate_limiter_

private

Token bucket rate limiter (nullptr if disabled).

Definition at line 312 of file backpressure_job_queue.h.

Referenced by get_available_tokens(), and is_rate_limited().

space_available_

std::condition_variable kcenon::thread::backpressure_job_queue::space_available_

private

Condition variable for blocking policy wait.

Definition at line 332 of file backpressure_job_queue.h.

stats_

backpressure_stats kcenon::thread::backpressure_job_queue::stats_

private

Backpressure statistics.

Definition at line 327 of file backpressure_job_queue.h.

Referenced by get_backpressure_stats().

---

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

• include/kcenon/thread/core/backpressure_job_queue.h
• src/core/backpressure_job_queue.cpp