Move shared functionality to common, remove worker -> host dep

The worker binaries must be built first so that the host could embed them into `polkadot`. Therefore `pvf/worker` could not depend on `pvf`, so to remove the dependency, common functionality was extracted into `pvf/common`.

(NOTE: We already needed to do this host/worker/common separation as part of https://github.com/paritytech/polkadot/issues/7116, it's just unfortunate that it had to be done here and complicate this PR.)

Integration tests were moved from `pvf/worker/tests` to `pvf/tests` because they need the PVF host.

Author: mrcnski

.editorconfig

@@ -4,7 +4,7 @@ root = true
 indent_style=tab
 indent_size=tab
 tab_width=4
-max_line_length=120
+max_line_length=100
 end_of_line=lf
 charset=utf-8
 trim_trailing_whitespace=true

Cargo.lock

@@ -80,6 +80,7 @@ members = [
 	"node/core/parachains-inherent",
 	"node/core/provisioner",
 	"node/core/pvf",
+	"node/core/pvf/common",
     "node/core/pvf/worker",
 	"node/core/pvf-checker",
 	"node/core/runtime-api",

Cargo.toml

@@ -19,6 +19,9 @@ parity-scale-codec = { version = "3.4.0", default-features = false, features = [
 
 polkadot-parachain = { path = "../../../parachain" }
 polkadot-core-primitives = { path = "../../../core-primitives" }
+polkadot-node-core-pvf-common = { path = "common" }
+# Must depend on the worker because the binaries must have been already built.
+polkadot-node-core-pvf-worker = { path = "worker" }
 polkadot-node-metrics = { path = "../../metrics" }
 polkadot-node-primitives = { path = "../../primitives" }
 polkadot-primitives = { path = "../../../primitives" }
@@ -35,3 +38,6 @@ substrate-build-script-utils = { git = "https://github.com/paritytech/substrate"
 assert_matches = "1.4.0"
 hex-literal = "0.3.4"
 tempfile = "3.3.0"
+
+adder = { package = "test-parachain-adder", path = "../../../parachain/test-parachains/adder" }
+halt = { package = "test-parachain-halt", path = "../../../parachain/test-parachains/halt" }

node/core/pvf/Cargo.toml

@@ -16,24 +16,4 @@
 
 //! Puppet worker used for integration tests.
 
-use sp_tracing;
-
-fn main() {
-	sp_tracing::try_init_simple();
-
-	let args = std::env::args().collect::<Vec<_>>();
-	if args.len() < 3 {
-		panic!("wrong number of arguments");
-	}
-
-	let subcommand = &args[1];
-	match subcommand.as_ref() {
-		"exit" => {
-			std::process::exit(1);
-		},
-		"sleep" => {
-			std::thread::sleep(std::time::Duration::from_secs(5));
-		},
-		other => panic!("unknown subcommand: {}", other),
-	}
-}
+polkadot_node_core_pvf::decl_puppet_worker_main!();

node/core/pvf/worker/bin/puppet_worker.rs renamed to node/core/pvf/bin/puppet_worker.rs

@@ -0,0 +1,15 @@
+[package]
+name = "polkadot-node-core-pvf-common"
+version.workspace = true
+authors.workspace = true
+edition.workspace = true
+
+[dependencies]
+tokio = { version = "1.24.2", features = ["fs", "process", "io-util"] }
+
+parity-scale-codec = { version = "3.4.0", default-features = false, features = ["derive"] }
+
+polkadot-parachain = { path = "../../../../parachain" }
+polkadot-primitives = { path = "../../../../primitives" }
+
+sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" }

node/core/pvf/common/Cargo.toml

@@ -0,0 +1,75 @@
+// Copyright (C) Parity Technologies (UK) Ltd.
+// This file is part of Polkadot.
+
+// Polkadot is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Polkadot is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// 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::prepare::PrepareStats;
+use parity_scale_codec::{Decode, Encode};
+use std::fmt;
+
+/// Result of PVF preparation performed by the validation host. Contains stats about the preparation if
+/// successful
+pub type PrepareResult = Result<PrepareStats, PrepareError>;
+
+/// An error that occurred during the prepare part of the PVF pipeline.
+#[derive(Debug, Clone, Encode, Decode)]
+pub enum PrepareError {
+	/// During the prevalidation stage of preparation an issue was found with the PVF.
+	Prevalidation(String),
+	/// Compilation failed for the given PVF.
+	Preparation(String),
+	/// An unexpected panic has occurred in the preparation worker.
+	Panic(String),
+	/// Failed to prepare the PVF due to the time limit.
+	TimedOut,
+	/// An IO error occurred. This state is reported by either the validation host or by the worker.
+	IoErr(String),
+	/// The temporary file for the artifact could not be created at the given cache path. This state is reported by the
+	/// validation host (not by the worker).
+	CreateTmpFileErr(String),
+	/// The response from the worker is received, but the file cannot be renamed (moved) to the final destination
+	/// location. This state is reported by the validation host (not by the worker).
+	RenameTmpFileErr(String),
+}
+
+impl PrepareError {
+	/// Returns whether this is a deterministic error, i.e. one that should trigger reliably. Those
+	/// errors depend on the PVF itself and the sc-executor/wasmtime logic.
+	///
+	/// Non-deterministic errors can happen spuriously. Typically, they occur due to resource
+	/// starvation, e.g. under heavy load or memory pressure. Those errors are typically transient
+	/// but may persist e.g. if the node is run by overwhelmingly underpowered machine.
+	pub fn is_deterministic(&self) -> bool {
+		use PrepareError::*;
+		match self {
+			Prevalidation(_) | Preparation(_) | Panic(_) => true,
+			TimedOut | IoErr(_) | CreateTmpFileErr(_) | RenameTmpFileErr(_) => false,
+		}
+	}
+}
+
+impl fmt::Display for PrepareError {
+	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+		use PrepareError::*;
+		match self {
+			Prevalidation(err) => write!(f, "prevalidation: {}", err),
+			Preparation(err) => write!(f, "preparation: {}", err),
+			Panic(err) => write!(f, "panic: {}", err),
+			TimedOut => write!(f, "prepare: timeout"),
+			IoErr(err) => write!(f, "prepare: io error while receiving response: {}", err),
+			CreateTmpFileErr(err) => write!(f, "prepare: error creating tmp file: {}", err),
+			RenameTmpFileErr(err) => write!(f, "prepare: error renaming tmp file: {}", err),
+		}
+	}
+}

node/core/pvf/common/src/error.rs

@@ -0,0 +1,65 @@
+// Copyright (C) Parity Technologies (UK) Ltd.
+// This file is part of Polkadot.
+
+// Polkadot is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Polkadot is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
+
+use parity_scale_codec::{Decode, Encode};
+use polkadot_parachain::primitives::ValidationResult;
+use polkadot_primitives::ExecutorParams;
+use std::time::Duration;
+
+/// The payload of the one-time handshake that is done when a worker process is created. Carries
+/// data from the host to the worker.
+#[derive(Encode, Decode)]
+pub struct Handshake {
+	/// The executor parameters.
+	pub executor_params: ExecutorParams,
+}
+
+/// The response from an execution job on the worker.
+#[derive(Encode, Decode)]
+pub enum Response {
+	/// The job completed successfully.
+	Ok {
+		/// The result of parachain validation.
+		result_descriptor: ValidationResult,
+		/// The amount of CPU time taken by the job.
+		duration: Duration,
+	},
+	/// The candidate is invalid.
+	InvalidCandidate(String),
+	/// The job timed out.
+	TimedOut,
+	/// Some internal error occurred. Should only be used for errors independent of the candidate.
+	InternalError(String),
+}
+
+impl Response {
+	/// Creates an invalid response from a context `ctx` and a message `msg` (which can be empty).
+	pub fn format_invalid(ctx: &'static str, msg: &str) -> Self {
+		if msg.is_empty() {
+			Self::InvalidCandidate(ctx.to_string())
+		} else {
+			Self::InvalidCandidate(format!("{}: {}", ctx, msg))
+		}
+	}
+	/// Creates an internal response from a context `ctx` and a message `msg` (which can be empty).
+	pub fn format_internal(ctx: &'static str, msg: &str) -> Self {
+		if msg.is_empty() {
+			Self::InternalError(ctx.to_string())
+		} else {
+			Self::InternalError(format!("{}: {}", ctx, msg))
+		}
+	}
+}

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

@@ -0,0 +1,48 @@
+// Copyright (C) Parity Technologies (UK) Ltd.
+// This file is part of Polkadot.
+
+// Polkadot is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Polkadot is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
+
+pub mod error;
+pub mod execute;
+pub mod prepare;
+pub mod pvf;
+
+use std::mem;
+use tokio::io::{self, AsyncRead, AsyncReadExt as _, AsyncWrite, AsyncWriteExt as _};
+
+pub mod tests {
+	use std::time::Duration;
+
+	pub const TEST_EXECUTION_TIMEOUT: Duration = Duration::from_secs(3);
+	pub const TEST_PREPARATION_TIMEOUT: Duration = Duration::from_secs(30);
+}
+
+/// Write some data prefixed by its length into `w`.
+pub async fn framed_send(w: &mut (impl AsyncWrite + Unpin), buf: &[u8]) -> io::Result<()> {
+	let len_buf = buf.len().to_le_bytes();
+	w.write_all(&len_buf).await?;
+	w.write_all(buf).await?;
+	Ok(())
+}
+
+/// Read some data prefixed by its length from `r`.
+pub async fn framed_recv(r: &mut (impl AsyncRead + Unpin)) -> io::Result<Vec<u8>> {
+	let mut len_buf = [0u8; mem::size_of::<usize>()];
+	r.read_exact(&mut len_buf).await?;
+	let len = usize::from_le_bytes(len_buf);
+	let mut buf = vec![0; len];
+	r.read_exact(&mut buf).await?;
+	Ok(buf)
+}

node/core/pvf/common/src/lib.rs

@@ -0,0 +1,48 @@
+// Copyright (C) Parity Technologies (UK) Ltd.
+// This file is part of Polkadot.
+
+// Polkadot is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Polkadot is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
+
+use parity_scale_codec::{Decode, Encode};
+
+/// Preparation statistics, including the CPU time and memory taken.
+#[derive(Debug, Clone, Default, Encode, Decode)]
+pub struct PrepareStats {
+	/// The CPU time that elapsed for the preparation job.
+	pub cpu_time_elapsed: std::time::Duration,
+	/// The observed memory statistics for the preparation job.
+	pub memory_stats: MemoryStats,
+}
+
+/// Helper struct to contain all the memory stats, including `MemoryAllocationStats` and, if
+/// supported by the OS, `ru_maxrss`.
+#[derive(Clone, Debug, Default, Encode, Decode)]
+pub struct MemoryStats {
+	/// Memory stats from `tikv_jemalloc_ctl`.
+	#[cfg(any(target_os = "linux", feature = "jemalloc-allocator"))]
+	pub memory_tracker_stats: Option<MemoryAllocationStats>,
+	/// `ru_maxrss` from `getrusage`. `None` if an error occurred.
+	#[cfg(target_os = "linux")]
+	pub max_rss: Option<i64>,
+}
+
+/// Statistics of collected memory metrics.
+#[cfg(any(target_os = "linux", feature = "jemalloc-allocator"))]
+#[derive(Clone, Debug, Default, Encode, Decode)]
+pub struct MemoryAllocationStats {
+	/// Total resident memory, in bytes.
+	pub resident: u64,
+	/// Total allocated memory, in bytes.
+	pub allocated: u64,
+}