Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: duplicate read proto #1741

Merged
merged 24 commits into from
Apr 18, 2024
Merged
Show file tree
Hide file tree
Changes from 17 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 11 additions & 7 deletions src/config/reader.rs
Original file line number Diff line number Diff line change
Expand Up @@ -10,24 +10,25 @@ use super::{ConfigModule, Content, Link, LinkType};
use crate::config::{Config, ConfigReaderContext, Source};
use crate::merge_right::MergeRight;
use crate::proto_reader::ProtoReader;
use crate::resource_reader::ResourceReader;
use crate::resource_reader::{Cached, ResourceReader, ResourceReaderHandler};
use crate::rest::EndpointSet;
use crate::runtime::TargetRuntime;

/// Reads the configuration from a file or from an HTTP URL and resolves all
/// linked extensions to create a ConfigModule.
pub struct ConfigReader {
runtime: TargetRuntime,
resource_reader: ResourceReader,
proto_reader: ProtoReader,
resource_reader: ResourceReader<Cached>,
proto_reader: ProtoReader<Cached>,
}

impl ConfigReader {
pub fn init(runtime: TargetRuntime) -> Self {
let resource_reader = ResourceReader::<Cached>::cached(runtime.clone());
Self {
runtime: runtime.clone(),
resource_reader: ResourceReader::init(runtime.clone()),
proto_reader: ProtoReader::init(runtime),
resource_reader: resource_reader.clone(),
proto_reader: ProtoReader::init(resource_reader),
}
}

Expand Down Expand Up @@ -157,12 +158,15 @@ impl ConfigReader {
}

/// Reads a single file and returns the config
pub async fn read<T: ToString>(&self, file: T) -> anyhow::Result<ConfigModule> {
pub async fn read<T: ToString + Send + Sync>(&self, file: T) -> anyhow::Result<ConfigModule> {
self.read_all(&[file]).await
}

/// Reads all the files and returns a merged config
pub async fn read_all<T: ToString>(&self, files: &[T]) -> anyhow::Result<ConfigModule> {
pub async fn read_all<T: ToString + Send + Sync>(
&self,
files: &[T],
) -> anyhow::Result<ConfigModule> {
let files = self.resource_reader.read_files(files).await?;
let mut config_module = ConfigModule::default();

Expand Down
9 changes: 7 additions & 2 deletions src/generator/generator.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,14 +5,19 @@
use crate::generator::source::Source;
use crate::merge_right::MergeRight;
use crate::proto_reader::ProtoReader;
use crate::resource_reader::{Direct, ResourceReader};
use crate::runtime::TargetRuntime;

pub struct Generator {
proto_reader: ProtoReader,
proto_reader: ProtoReader<Direct>,
}
impl Generator {

Check warning on line 14 in src/generator/generator.rs

View workflow job for this annotation

GitHub Actions / Run Formatter and Lint Check

Diff in /home/runner/work/tailcall/tailcall/src/generator/generator.rs
pub fn init(runtime: TargetRuntime) -> Self {
Self { proto_reader: ProtoReader::init(runtime) }
Self {
proto_reader: ProtoReader::init(ResourceReader::<Direct>::direct(
runtime,
)),
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why are we changing the API here? I don't think it's required.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ProtoReader need to share ResourceReader to avoid duplicate reading.

}

pub async fn read_all<T: AsRef<str>>(
Expand Down
22 changes: 12 additions & 10 deletions src/proto_reader.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,21 +5,20 @@ use futures_util::future::join_all;
use prost_reflect::prost_types::{FileDescriptorProto, FileDescriptorSet};
use protox::file::{FileResolver, GoogleFileResolver};

use crate::resource_reader::ResourceReader;
use crate::runtime::TargetRuntime;
use crate::resource_reader::{ResourceReader, ResourceReaderHandler};

pub struct ProtoReader {
resource_reader: ResourceReader,
pub struct ProtoReader<A: ResourceReaderHandler> {
resource_reader: ResourceReader<A>,
}

pub struct ProtoMetadata {
pub descriptor_set: FileDescriptorSet,
pub path: String,
}

impl ProtoReader {
pub fn init(runtime: TargetRuntime) -> Self {
Self { resource_reader: ResourceReader::init(runtime) }
impl<A: ResourceReaderHandler> ProtoReader<A> {
pub fn init(resource_reader: ResourceReader<A>) -> Self {
Self { resource_reader }
}

pub async fn read_all<T: AsRef<str>>(&self, paths: &[T]) -> anyhow::Result<Vec<ProtoMetadata>> {
Expand Down Expand Up @@ -100,11 +99,14 @@ mod test_proto_config {
use pretty_assertions::assert_eq;

use crate::proto_reader::ProtoReader;
use crate::resource_reader::{Cached, ResourceReader};

#[tokio::test]
async fn test_resolve() {
// Skipping IO tests as they are covered in reader.rs
let reader = ProtoReader::init(crate::runtime::test::init(None));
let reader = ProtoReader::init(ResourceReader::<Cached>::cached(
crate::runtime::test::init(None),
));
reader
.read_proto("google/protobuf/empty.proto")
.await
Expand Down Expand Up @@ -133,7 +135,7 @@ mod test_proto_config {
let runtime = crate::runtime::test::init(None);
let file_rt = runtime.file.clone();

let reader = ProtoReader::init(runtime);
let reader = ProtoReader::init(ResourceReader::<Cached>::cached(runtime));
let helper_map = reader
.resolve_descriptors(reader.read_proto(&test_file).await?)
.await?;
Expand Down Expand Up @@ -177,7 +179,7 @@ mod test_proto_config {
#[tokio::test]
async fn test_proto_no_pkg() -> Result<()> {
let runtime = crate::runtime::test::init(None);
let reader = ProtoReader::init(runtime);
let reader = ProtoReader::init(ResourceReader::<Cached>::cached(runtime));
let mut proto_no_pkg = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
proto_no_pkg.push("src/grpc/tests/proto_no_pkg.graphql");
let config_module = reader.read(proto_no_pkg.to_str().unwrap()).await;
Expand Down
113 changes: 108 additions & 5 deletions src/resource_reader.rs
Original file line number Diff line number Diff line change
@@ -1,3 +1,7 @@
use std::collections::HashMap;
use std::ops::Deref;
use std::sync::{Arc, Mutex};

use futures_util::future::join_all;
use futures_util::TryFutureExt;
use url::Url;
Expand All @@ -11,16 +15,52 @@
pub path: String,
}

pub struct ResourceReader {
#[derive(Clone)]
pub struct ResourceReader<A>(A);

impl<A> ResourceReader<A> {
pub fn direct(runtime: TargetRuntime) -> ResourceReader<Direct> {
ResourceReader::<Direct>(Direct::init(runtime))
}

pub fn cached(runtime: TargetRuntime) -> ResourceReader<Cached> {
ResourceReader::<Cached>(Cached::init(runtime))
}
}

impl<A> Deref for ResourceReader<A> {
type Target = A;

fn deref(&self) -> &Self::Target {
&self.0
}
}

#[async_trait::async_trait]
pub trait ResourceReaderHandler {
async fn read_file<T: ToString + Send>(&self, file: T) -> anyhow::Result<FileRead>;

async fn read_files<T: ToString + Send + Sync>(
&self,
files: &[T],
) -> anyhow::Result<Vec<FileRead>>;
}

#[derive(Clone)]
pub struct Direct {
runtime: TargetRuntime,
}

impl ResourceReader {
impl Direct {
pub fn init(runtime: TargetRuntime) -> Self {
Self { runtime }
}
}

#[async_trait::async_trait]
impl ResourceReaderHandler for Direct {
/// Reads a file from the filesystem or from an HTTP URL
pub async fn read_file<T: ToString>(&self, file: T) -> anyhow::Result<FileRead> {
async fn read_file<T: ToString + Send>(&self, file: T) -> anyhow::Result<FileRead> {
// Is an HTTP URL
let content = if let Ok(url) = Url::parse(&file.to_string()) {
if url.scheme().starts_with("http") {
Expand All @@ -41,12 +81,75 @@

self.runtime.file.read(&file.to_string()).await?
};

Ok(FileRead { content, path: file.to_string() })
}

/// Reads all the files in parallel
pub async fn read_files<T: ToString>(&self, files: &[T]) -> anyhow::Result<Vec<FileRead>> {
async fn read_files<T: ToString + Send + Sync>(
&self,
files: &[T],
) -> anyhow::Result<Vec<FileRead>> {
let files = files.iter().map(|x| {
self.read_file(x.to_string())
.map_err(|e| e.context(x.to_string()))
});
let content = join_all(files)
.await
.into_iter()
.collect::<anyhow::Result<Vec<_>>>()?;
Ok(content)
}

Check warning on line 101 in src/resource_reader.rs

View check run for this annotation

Codecov / codecov/patch

src/resource_reader.rs#L91-L101

Added lines #L91 - L101 were not covered by tests
Shylock-Hg marked this conversation as resolved.
Show resolved Hide resolved
}

#[derive(Clone)]
pub struct Cached {
direct: Direct,
// Cache file content, path->content
cache: Arc<Mutex<HashMap<String, String>>>,
}

Check warning on line 110 in src/resource_reader.rs

View workflow job for this annotation

GitHub Actions / Run Formatter and Lint Check

Diff in /home/runner/work/tailcall/tailcall/src/resource_reader.rs
impl Cached {
pub fn init(runtime: TargetRuntime) -> Self {
Self {
direct: Direct::init(runtime),
cache: Default::default(),
tusharmath marked this conversation as resolved.
Show resolved Hide resolved
}
}
}

#[async_trait::async_trait]
impl ResourceReaderHandler for Cached {
/// Reads a file from the filesystem or from an HTTP URL with cache
async fn read_file<T: ToString + Send>(&self, file: T) -> anyhow::Result<FileRead> {
// check cache
let file_path = file.to_string();
let content = self
.cache
.as_ref()
.lock()
.unwrap()
tusharmath marked this conversation as resolved.
Show resolved Hide resolved
.get(&file_path)
.map(|v| v.to_owned());
let content = if let Some(content) = content {
content.to_owned()
} else {
let file_read = self.direct.read_file(file.to_string()).await?;
self.cache
.as_ref()
.lock()
.unwrap()
.insert(file_path.to_owned(), file_read.content.clone());
tusharmath marked this conversation as resolved.
Show resolved Hide resolved
file_read.content
};

Ok(FileRead { content, path: file_path })
}

/// Reads all the files in parallel with cache
async fn read_files<T: ToString + Send + Sync>(
&self,
files: &[T],
) -> anyhow::Result<Vec<FileRead>> {
let files = files.iter().map(|x| {
self.read_file(x.to_string())
.map_err(|e| e.context(x.to_string()))
Expand Down
Loading