DOCSEARCHbyAlgolia
Submit Docs

tarpc

repository·main·Indexed 25 days ago

https://github.com/google/tarpc

A high-performance RPC (Remote Procedure Call) framework for Rust that allows developers to define services directly in Rust code using the #[tarpc::service] attribute, eliminating the need for external schema languages like Protocol Buffers. It features pluggable transports, cascading cancellation, configurable deadlines, distributed tracing with OpenTelemetry, and Serde serialization.

Tokens
10.1K
Snippets
28
Records
69
Agent score
86%

What's inside tarpc

  1. Key features of tarpc

    main
    • Pluggable transport: Any type implementing Stream<Item = Request> + Sink<Response> can be used.
    • Cascading cancellation: Dropping a request sends a cancellation message to the server to cease unfinished work.
    • Configurable deadlines: Request deadlines default to 10s. Deadlines are automatically propagated to downstream requests via the Context.
    • Distributed tracing: Instrumented with tracing and OpenTelemetry support.
    • Serde serialization: Enabling the serde1 feature allows requests/responses to implement Serialize + Deserialize.
    • No separate schema language: Schema is defined in Rust code, eliminating .proto files and separate compilation steps.

  5. Implement a tarpc server

    main

    To create a server, define a struct and implement the trait generated by the #[tarpc::service] attribute. The implementation methods must accept a tarpc::context::Context as the first argument.

    #[derive(Clone)]
struct HelloServer;

impl World for HelloServer {
    async fn hello(self, _: context::Context, name: String) -> String {
        format!("Hello, {name}!")
    }
}

  7. Define an RPC service with `#[tarpc::service]`

    main

    Use the #[tarpc::service] attribute on a trait to define your RPC schema. This attribute expands into a collection of items, including a client struct (e.g., WorldClient) and a service trait that you must implement for your server.

    Unlike gRPC, tarpc defines the schema directly in Rust code, avoiding separate .proto files.

    #[tarpc::service]
trait World {
    /// Returns a greeting for name.
    async fn hello(name: String) -> String;
}

  8. Chain multiple request hooks using `before()`

    main

    To run a sequence of hooks in order, use the tarpc::server::request_hook::before() builder.

    • Use .then_fn(|ctx, req| async { ... }) to add a hook using a closure. This is often preferred for type inference.
    • Use .serving(service) to wrap your existing service with the constructed hook chain.

    Hooks are executed in the order they are chained.

    use futures::{executor::block_on, future};
use tarpc::{context, ServerError, server::{Serve, serve, request_hook::{self, 
            BeforeRequest, BeforeRequestList}}};
use std::cell::Cell;

let i = Cell::new(0);
let serve = request_hook::before()
    .then_fn(|_, _| async {
        assert!(i.get() == 0);
        i.set(1);
        Ok(())
    })
    .then_fn(|_, _| async {
        assert!(i.get() == 1);
        i.set(2);
        Ok(())
    })
    .serving(serve(|_ctx, i| async move { Ok(i + 1) }));

let response = serve.clone().serve(context::current(), 1);
assert!(block_on(response).is_ok());
assert!(i.get() == 2);

  9. Handle requests manually using `Channel::requests`

    main

    For more control over scheduling or if you are not using tokio, use Channel::requests(). This returns a Requests stream of InFlightRequest items. Each InFlightRequest can be executed using its own .execute(serve) method.

    use tarpc::server::{self, BaseChannel, Channel, serve};
use futures::prelude::*;

// ... setup transport and channel ...

let mut requests = server.requests();
tokio::spawn(async move {
    while let Some(Ok(in_flight_request)) = requests.next().await {
        in_flight_request.execute(serve(|_, i| async move { Ok(i + 1) })).await;
    }
});

  10. Run a complete tarpc client-server example

    main

    This example demonstrates an in-process communication using tarpc::transport::channel::unbounded(). In production, you would typically use network transports like serde_transport with the tcp feature enabled.

    To run this example, ensure your Cargo.toml includes:

    anyhow = "1.0"
futures = "0.3"
tarpc = { version = "0.37", features = ["tokio1"] }
tokio = { version = "1.0", features = ["rt-multi-thread", "macros"] }
    use futures::prelude::*;
use tarpc::client;
use tarpc::context;
use tarpc::server::{self, Channel};

#[tarpc::service]
trait World {
    async fn hello(name: String) -> String;
}

#[derive(Clone)]
struct HelloServer;

impl World for HelloServer {
    async fn hello(self, _: context::Context, name: String) -> String {
        format!("Hello, {name}!")
    }
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let (client_transport, server_transport) = tarpc::transport::channel::unbounded();

    let server = server::BaseChannel::with_defaults(server_transport);
    tokio::spawn(
        server.execute(HelloServer.serve())
            .for_each(|response| async move {
                tokio::spawn(response);
            })
    );

    let mut client = WorldClient::new(client::Config::default(), client_transport).spawn();

    let hello = client.hello(context::current(), "Stim".to_string()).await?;

    println!("{hello}");

    Ok(())
}

  11. Configure server channel settings

    main
    Use the Config struct to control the behavior of Channel instances. The primary setting is pending_response_buffer, which defines the number of responses that can sit in the outbound queue before request handlers begin blocking.

  12. Configure client behavior with Config

    main

    Use the Config struct to tune the performance and resource usage of the tarpc client.

    Key parameters:

    • max_in_flight_requests: Limits the number of concurrent requests. When this limit is reached, new requests are queued. Default is 1000.
    • pending_request_buffer: The size of the buffer between the client API and the network dispatch task. A higher value can improve throughput but uses more memory. Default is 100.