Struct tower_batch_control::worker::Worker

pub struct Worker<T, Request>
where
    T: Service<BatchControl<Request>>,
    T::Future: Send + 'static,
    T::Error: Into<Box<dyn Error + Send + Sync + 'static>>,
{
    rx: UnboundedReceiver<Message<Request, T::Future>>,
    service: T,
    pending_items: usize,
    pending_batch_timer: Option<Pin<Box<Sleep>>>,
    concurrent_batches: FuturesUnordered<BoxFuture<'static, Result<T::Response, T::Error>>>,
    failed: Option<ServiceError>,
    error_handle: ErrorHandle,
    close: PollSemaphore,
    max_items_in_batch: usize,
    max_concurrent_batches: usize,
    max_latency: Duration,
}

Task that handles processing the buffer. This type should not be used directly, instead Buffer requires an Executor that can accept this task.

The struct is pub in the private module and the type is not re-exported as part of the public API. This is the “sealed” pattern to include “private” types in public traits that are not meant for consumers of the library to implement (only call).

Fields

rx: UnboundedReceiver<Message<Request, T::Future>>

A semaphore-bounded channel for receiving requests from the batch wrapper service.

service: T

The wrapped service that processes batches.

pending_items: usize

The number of pending items sent to service, since the last batch flush.

pending_batch_timer: Option<Pin<Box<Sleep>>>

The timer for the pending batch, if it has any items.

The timer is started when the first entry of a new batch is submitted, so that the batch latency of all entries is at most self.max_latency. However, we don’t keep the timer running unless there is a pending request to prevent wakeups on idle services.

concurrent_batches: FuturesUnordered<BoxFuture<'static, Result<T::Response, T::Error>>>

The batches that the worker is concurrently executing.

failed: Option<ServiceError>

An error that’s populated on permanent service failure.

error_handle: ErrorHandle

A shared error handle that’s populated on permanent service failure.

close: PollSemaphore

A cloned copy of the wrapper service’s semaphore, used to close the semaphore.

max_items_in_batch: usize

The maximum number of items allowed in a batch.

max_concurrent_batches: usize

The maximum number of batches that are allowed to run concurrently.

max_latency: Duration

The maximum delay before processing a batch with fewer than max_items_in_batch.

Implementations

impl<T, Request> Worker<T, Request>

where T: Service<BatchControl<Request>>, T::Future: Send + 'static, T::Error: Into<Box<dyn Error + Send + Sync + 'static>>,

pub(crate) fn project<'pin>( self: Pin<&'pin mut Self>, ) -> __WorkerProjection<'pin, T, Request>

pub(crate) fn project_ref<'pin>( self: Pin<&'pin Self>, ) -> __WorkerProjectionRef<'pin, T, Request>

impl<T, Request> Worker<T, Request>

where T: Service<BatchControl<Request>>, T::Future: Send + 'static, T::Error: Into<Box<dyn Error + Send + Sync + 'static>>,

pub(crate) fn new( service: T, rx: UnboundedReceiver<Message<Request, T::Future>>, max_items_in_batch: usize, max_concurrent_batches: usize, max_latency: Duration, close: PollSemaphore, ) -> (ErrorHandle, Worker<T, Request>)

Creates a new batch worker.

See Batch::new() for details.

async fn process_req( &mut self, req: Request, tx: Sender<Result<T::Future, ServiceError>>, )

Process a single worker request.

async fn flush_service(&mut self)

Tell the inner service to flush the current batch.

Waits until the inner service is ready, then stores a future which resolves when the batch finishes.

fn can_spawn_new_batches(&self) -> bool

Is the current number of concurrent batches above the configured limit?

pub async fn run(self)

Run loop for batch requests, which implements the batch policies.

See Batch::new() for details.

fn failed(&mut self, error: Box<dyn Error + Send + Sync + 'static>)

Register an inner service failure.

The underlying service failed when we called poll_ready on it with the given error. We need to communicate this to all the Buffer handles. To do so, we wrap up the error in an Arc, send that Arc<E> to all pending requests, and store it so that subsequent requests will also fail with the same error.

Trait Implementations

impl<T, Request: Debug> Debug for Worker<T, Request>

where T: Service<BatchControl<Request>> + Debug, T::Future: Send + 'static + Debug, T::Error: Into<Box<dyn Error + Send + Sync + 'static>> + Debug, T::Response: Debug,

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<T, Request> Drop for Worker<T, Request>

where T: Service<BatchControl<Request>>, T::Future: Send + 'static, T::Error: Into<Box<dyn Error + Send + Sync + 'static>>,

fn drop(&mut self)

Executes the destructor for this type.

impl<'pin, T, Request> Unpin for Worker<T, Request>

where T: Service<BatchControl<Request>>, T::Future: Send + 'static, T::Error: Into<Box<dyn Error + Send + Sync + 'static>>, PinnedFieldsOf<__Worker<'pin, T, Request>>: Unpin,