Skip to content

Commit

Permalink
Remote execution uses tower-grpc to start executions
Browse files Browse the repository at this point in the history
grpcio is still used both for GetOperation requests, and for all fs
operations. It won't be for long, though... :)
  • Loading branch information
illicitonion committed Jan 9, 2019
1 parent 0bc0178 commit 5f469b6
Show file tree
Hide file tree
Showing 8 changed files with 405 additions and 190 deletions.
10 changes: 10 additions & 0 deletions src/rust/engine/Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

13 changes: 11 additions & 2 deletions src/rust/engine/process_execution/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -13,18 +13,27 @@ bytes = "0.4.5"
digest = "0.8"
fs = { path = "../fs" }
futures = "^0.1.16"
# TODO: Switch to a release once https://github.com/alexcrichton/futures-timer/pull/11 and https://github.com/alexcrichton/futures-timer/pull/12 merge
futures-timer = { git = "https://github.com/pantsbuild/futures-timer", rev = "0b747e565309a58537807ab43c674d8951f9e5a0" }
grpcio = { git = "https://github.com/pantsbuild/grpc-rs.git", rev = "4dfafe9355dc996d7d0702e7386a6fedcd9734c0", default_features = false, features = ["protobuf-codec"] }
h2 = "0.1.13"
hashing = { path = "../hashing" }
http = "0.1"
log = "0.4"
parking_lot = "0.6"
protobuf = { version = "2.0.4", features = ["with-bytes"] }
resettable = { path = "../resettable" }
sha2 = "0.8"
tempfile = "3"
# TODO: Switch to a release once https://github.com/alexcrichton/futures-timer/pull/11 and https://github.com/alexcrichton/futures-timer/pull/12 merge
futures-timer = { git = "https://github.com/pantsbuild/futures-timer", rev = "0b747e565309a58537807ab43c674d8951f9e5a0" }
time = "0.1.40"
tokio = "0.1.14"
tokio-codec = "0.1"
tokio-connect = { git = "https://github.com/pantsbuild/tokio-connect.git", rev = "f7ad1ca437973d6e24037ac6f7d5ef1013833c0b" }
tokio-process = "0.2.1"
tower-grpc = { git = "https://github.com/pantsbuild/tower-grpc.git", rev = "ef19f2e1715f415ecb699e8f17f5845ad2b45daf" }
tower-h2 = { git = "https://github.com/pantsbuild/tower-h2.git", rev = "44b0efb4983b769283efd5b2a3bc3decbf7c33de" }
tower-http = { git = "https://github.com/pantsbuild/tower-http.git", rev = "56049ee7f31d4f6c549f5d1d5fbbfd7937df3d00" }
tower-util = { git = "https://github.com/pantsbuild/tower.git", rev = "7b61c1fc1992c1df684fd3f179644ef0ca9bfa4c" }

[dev-dependencies]
mock = { path = "../testutil/mock" }
Expand Down
58 changes: 55 additions & 3 deletions src/rust/engine/process_execution/bazel_protos/src/conversions.rs
Original file line number Diff line number Diff line change
Expand Up @@ -18,19 +18,31 @@ impl<'a> From<&'a hashing::Digest> for crate::build::bazel::remote::execution::v
}
}

impl<'a> From<&'a super::remote_execution::Digest> for Result<hashing::Digest, String> {
fn from(d: &super::remote_execution::Digest) -> Self {
impl<'a> From<&'a crate::remote_execution::Digest> for Result<hashing::Digest, String> {
fn from(d: &crate::remote_execution::Digest) -> Self {
hashing::Fingerprint::from_hex_string(d.get_hash())
.map_err(|err| format!("Bad fingerprint in Digest {:?}: {:?}", d.get_hash(), err))
.map(|fingerprint| hashing::Digest(fingerprint, d.get_size_bytes() as usize))
}
}

impl<'a> From<&'a crate::build::bazel::remote::execution::v2::Digest>
for Result<hashing::Digest, String>
{
fn from(d: &crate::build::bazel::remote::execution::v2::Digest) -> Self {
hashing::Fingerprint::from_hex_string(&d.hash)
.map_err(|err| format!("Bad fingerprint in Digest {:?}: {:?}", d.hash, err))
.map(|fingerprint| hashing::Digest(fingerprint, d.size_bytes as usize))
}
}

impl From<crate::google::longrunning::Operation> for crate::operations::Operation {
fn from(op: crate::google::longrunning::Operation) -> Self {
let mut dst = Self::new();
dst.set_name(op.name);
dst.set_metadata(prost_any_to_gcprio_any(op.metadata.unwrap()));
if let Some(metadata) = op.metadata {
dst.set_metadata(prost_any_to_gcprio_any(metadata));
}
dst.set_done(op.done);
match op.result {
Some(crate::google::longrunning::operation::Result::Response(response)) => {
Expand All @@ -45,6 +57,46 @@ impl From<crate::google::longrunning::Operation> for crate::operations::Operatio
}
}

// This should only be used in test contexts. It should be deleted when the mock systems use tower.
impl From<crate::remote_execution::ExecuteRequest>
for crate::build::bazel::remote::execution::v2::ExecuteRequest
{
fn from(req: crate::remote_execution::ExecuteRequest) -> Self {
if req.has_execution_policy() || req.has_results_cache_policy() {
panic!("Can't convert ExecuteRequest protos with execution policy or results cache policy");
}
let digest: Result<hashing::Digest, String> = req.get_action_digest().into();
Self {
action_digest: Some((&digest.expect("Bad digest converting ExecuteRequest proto")).into()),
instance_name: req.instance_name,
execution_policy: None,
results_cache_policy: None,
skip_cache_lookup: req.skip_cache_lookup,
}
}
}

// This should only be used in test contexts. It should be deleted when the mock systems use tower.
impl From<crate::build::bazel::remote::execution::v2::ExecuteRequest>
for crate::remote_execution::ExecuteRequest
{
fn from(req: crate::build::bazel::remote::execution::v2::ExecuteRequest) -> Self {
if req.execution_policy.is_some() || req.results_cache_policy.is_some() {
panic!("Can't convert ExecuteRequest protos with execution policy or results cache policy");
}
let digest: Result<hashing::Digest, String> = (&req
.action_digest
.expect("Missing digest converting ExecuteRequest proto"))
.into();

let mut ret = Self::new();
ret.set_action_digest((&digest.expect("Bad digest converting ExecuteRequest proto")).into());
ret.set_instance_name(req.instance_name);
ret.set_skip_cache_lookup(req.skip_cache_lookup);
ret
}
}

pub fn prost_any_to_gcprio_any(any: prost_types::Any) -> protobuf::well_known_types::Any {
let prost_types::Any { type_url, value } = any;
let mut dst = protobuf::well_known_types::Any::new();
Expand Down
Loading

0 comments on commit 5f469b6

Please sign in to comment.