[pvf-worker] Refactor execute request handling (#8908)

# Description

Fixes #8886

PVF execution worker communication was organized into a single
ExecuteRequest struct. This should improve performance: one
encode/decode operation instead of four. Also, no more chance of
ordering mistakes.



## Integration

This is an internal refactoring of the PVF execution worker. Downstream
projects will not need any code changes.

---------

Co-authored-by: cmd[bot] <41898282+github-actions[bot]@users.noreply.github.com>

Author: AndreiEres

Cargo.lock

@@ -21,6 +21,7 @@ thiserror = { workspace = true }
 
 codec = { features = ["derive"], workspace = true }
 
+polkadot-node-primitives = { workspace = true, default-features = true }
 polkadot-parachain-primitives = { workspace = true, default-features = true }
 polkadot-primitives = { workspace = true, default-features = true }
 

polkadot/node/core/pvf/common/Cargo.toml

@@ -14,10 +14,11 @@
 // You should have received a copy of the GNU General Public License
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
 
-use crate::error::InternalValidationError;
+use crate::{error::InternalValidationError, ArtifactChecksum};
 use codec::{Decode, Encode};
+use polkadot_node_primitives::PoV;
 use polkadot_parachain_primitives::primitives::ValidationResult;
-use polkadot_primitives::ExecutorParams;
+use polkadot_primitives::{ExecutorParams, PersistedValidationData};
 use std::time::Duration;
 
 /// The payload of the one-time handshake that is done when a worker process is created. Carries
@@ -28,6 +29,19 @@ pub struct Handshake {
 	pub executor_params: ExecutorParams,
 }
 
+/// A request to execute a PVF
+#[derive(Encode, Decode)]
+pub struct ExecuteRequest {
+	/// Persisted validation data.
+	pub pvd: PersistedValidationData,
+	/// Proof-of-validity.
+	pub pov: PoV,
+	/// Execution timeout.
+	pub execution_timeout: Duration,
+	/// Checksum of the artifact to execute.
+	pub artifact_checksum: ArtifactChecksum,
+}
+
 /// The response from the execution worker.
 #[derive(Debug, Encode, Decode)]
 pub struct WorkerResponse {

polkadot/node/core/pvf/common/src/execute.rs

@@ -41,7 +41,9 @@ use nix::{
 use polkadot_node_core_pvf_common::{
 	compute_checksum,
 	error::InternalValidationError,
-	execute::{Handshake, JobError, JobResponse, JobResult, WorkerError, WorkerResponse},
+	execute::{
+		ExecuteRequest, Handshake, JobError, JobResponse, JobResult, WorkerError, WorkerResponse,
+	},
 	executor_interface::params_to_wasmtime_semantics,
 	framed_recv_blocking, framed_send_blocking,
 	worker::{
@@ -91,40 +93,15 @@ fn recv_execute_handshake(stream: &mut UnixStream) -> io::Result<Handshake> {
 fn recv_request(
 	stream: &mut UnixStream,
 ) -> io::Result<(PersistedValidationData, PoV, Duration, ArtifactChecksum)> {
-	let pvd = framed_recv_blocking(stream)?;
-	let pvd = PersistedValidationData::decode(&mut &pvd[..]).map_err(|_| {
-		io::Error::new(
-			io::ErrorKind::Other,
-			"execute pvf recv_request: failed to decode persisted validation data".to_string(),
-		)
-	})?;
-
-	let pov = framed_recv_blocking(stream)?;
-	let pov = PoV::decode(&mut &pov[..]).map_err(|_| {
+	let request_bytes = framed_recv_blocking(stream)?;
+	let request = ExecuteRequest::decode(&mut &request_bytes[..]).map_err(|_| {
 		io::Error::new(
 			io::ErrorKind::Other,
-			"execute pvf recv_request: failed to decode PoV".to_string(),
+			"execute pvf recv_request: failed to decode ExecuteRequest".to_string(),
 		)
 	})?;
 
-	let execution_timeout = framed_recv_blocking(stream)?;
-	let execution_timeout = Duration::decode(&mut &execution_timeout[..]).map_err(|_| {
-		io::Error::new(
-			io::ErrorKind::Other,
-			"execute pvf recv_request: failed to decode duration".to_string(),
-		)
-	})?;
-
-	let artifact_checksum = framed_recv_blocking(stream)?;
-	let artifact_checksum =
-		ArtifactChecksum::decode(&mut &artifact_checksum[..]).map_err(|_| {
-			io::Error::new(
-				io::ErrorKind::Other,
-				"execute pvf recv_request: failed to decode artifact checksum".to_string(),
-			)
-		})?;
-
-	Ok((pvd, pov, execution_timeout, artifact_checksum))
+	Ok((request.pvd, request.pov, request.execution_timeout, request.artifact_checksum))
 }
 
 /// Sends an error to the host and returns the original error wrapped in `io::Error`.

polkadot/node/core/pvf/execute-worker/src/lib.rs

@@ -295,10 +295,13 @@ async fn send_request(
 	execution_timeout: Duration,
 	artifact_checksum: ArtifactChecksum,
 ) -> io::Result<()> {
-	framed_send(stream, &pvd.encode()).await?;
-	framed_send(stream, &pov.encode()).await?;
-	framed_send(stream, &execution_timeout.encode()).await?;
-	framed_send(stream, &artifact_checksum.encode()).await
+	let request = polkadot_node_core_pvf_common::execute::ExecuteRequest {
+		pvd: (*pvd).clone(),
+		pov: (*pov).clone(),
+		execution_timeout,
+		artifact_checksum,
+	};
+	framed_send(stream, &request.encode()).await
 }
 
 async fn recv_result(stream: &mut UnixStream) -> io::Result<Result<WorkerResponse, WorkerError>> {

polkadot/node/core/pvf/src/execute/worker_interface.rs

@@ -0,0 +1,13 @@
+title: '[pvf-worker] Refactor execute request handling'
+doc:
+- audience: Node Dev
+  description: |-
+    PVF execution worker communication was organized into a single ExecuteRequest struct. This should improve performance: one encode/decode operation instead of four. Also, no more chance of ordering mistakes.
+
+crates:
+- name: polkadot-node-core-pvf-common
+  bump: minor
+- name: polkadot-node-core-pvf-execute-worker
+  bump: patch
+- name: polkadot-node-core-pvf
+  bump: patch