diff --git a/radicle-node/src/runtime.rs b/radicle-node/src/runtime.rs index c290f00c..41719c52 100644 --- a/radicle-node/src/runtime.rs +++ b/radicle-node/src/runtime.rs @@ -23,7 +23,6 @@ use crate::service::{routing, tracking}; use crate::wire; use crate::wire::Wire; use crate::worker; -use crate::worker::{WorkerPool, WorkerReq}; use crate::{service, LocalTime}; pub use handle::Error as HandleError; @@ -76,7 +75,7 @@ pub struct Runtime { pub storage: Storage, pub reactor: Reactor>, pub daemon: net::SocketAddr, - pub pool: WorkerPool, + pub pool: worker::Pool, pub local_addrs: Vec, } @@ -131,7 +130,7 @@ impl Runtime { sig: EcSign::sign(&signer, id.as_slice()), }; - let (worker_send, worker_recv) = chan::unbounded::>(); + let (worker_send, worker_recv) = chan::unbounded::>(); let mut wire = Wire::new(service, worker_send, cert, signer, proxy, clock); let mut local_addrs = Vec::new(); @@ -155,7 +154,7 @@ impl Runtime { ); } - let pool = WorkerPool::with( + let pool = worker::Pool::with( worker_recv, handle.clone(), worker::Config { diff --git a/radicle-node/src/runtime/handle.rs b/radicle-node/src/runtime/handle.rs index 77ab7384..83483a6b 100644 --- a/radicle-node/src/runtime/handle.rs +++ b/radicle-node/src/runtime/handle.rs @@ -16,7 +16,7 @@ use crate::service; use crate::service::{CommandError, QueryState}; use crate::service::{NodeId, Sessions}; use crate::wire; -use crate::worker::WorkerResp; +use crate::worker::TaskResult; /// An error resulting from a handle method. #[derive(Error, Debug)] @@ -89,7 +89,7 @@ impl Handle { } } - pub fn worker_result(&mut self, resp: WorkerResp) -> Result<(), Error> { + pub fn worker_result(&mut self, resp: TaskResult) -> Result<(), Error> { match self.controller.cmd(wire::Control::Worker(resp)) { Ok(()) => {} Err(err) if err.kind() == io::ErrorKind::BrokenPipe => return Err(Error::NotConnected), diff --git a/radicle-node/src/wire/protocol.rs b/radicle-node/src/wire/protocol.rs index 3b955dc0..ace54ffe 100644 --- a/radicle-node/src/wire/protocol.rs +++ b/radicle-node/src/wire/protocol.rs @@ -27,7 +27,7 @@ use crate::crypto::Signer; use crate::service::reactor::{Fetch, Io}; use crate::service::{routing, session, DisconnectReason, Message, Service}; use crate::wire::{Decode, Encode}; -use crate::worker::{WorkerReq, WorkerResp}; +use crate::worker::{Task, TaskResult}; use crate::Link; use crate::{address, service}; @@ -37,7 +37,7 @@ pub enum Control { /// Message from the user to the service. User(service::Command), /// Message from a worker to the service. - Worker(WorkerResp), + Worker(TaskResult), } impl fmt::Debug for Control { @@ -186,7 +186,7 @@ pub struct Wire { /// Backing service instance. service: Service, /// Worker pool interface. - worker: chan::Sender>, + worker: chan::Sender>, /// Used for authentication; keeps local identity. cert: Cert, /// Used for authentication. @@ -210,7 +210,7 @@ where { pub fn new( mut service: Service, - worker: chan::Sender>, + worker: chan::Sender>, cert: Cert, signer: G, proxy: net::SocketAddr, @@ -307,7 +307,7 @@ where if self .worker - .send(WorkerReq { + .send(Task { fetch, session, drain: self.read_queue.drain(..).collect(), @@ -318,10 +318,10 @@ where } } - fn worker_result(&mut self, resp: WorkerResp) { - log::debug!(target: "wire", "Fetch completed: {:?}", resp.result); + fn worker_result(&mut self, task: TaskResult) { + log::debug!(target: "wire", "Fetch completed: {:?}", task.result); - let session = resp.session; + let session = task.session; let fd = session.as_connection().as_raw_fd(); let peer = self.peer_mut_by_fd(fd); @@ -342,7 +342,7 @@ where peer.downgrade(); self.actions.push_back(Action::RegisterTransport(session)); - self.service.fetched(resp.result); + self.service.fetched(task.result); } } @@ -488,7 +488,7 @@ where fn handle_command(&mut self, cmd: Self::Command) { match cmd { Control::User(cmd) => self.service.command(cmd), - Control::Worker(resp) => self.worker_result(resp), + Control::Worker(result) => self.worker_result(result), } } diff --git a/radicle-node/src/worker.rs b/radicle-node/src/worker.rs index 563693da..5800e95d 100644 --- a/radicle-node/src/worker.rs +++ b/radicle-node/src/worker.rs @@ -34,15 +34,16 @@ pub struct Config { pub storage: Storage, } -/// Worker request. -pub struct WorkerReq { +/// Task to be accomplished on a worker thread. +/// This is either going to be an outgoing or incoming fetch. +pub struct Task { pub fetch: Fetch, pub session: WireSession, pub drain: Vec, } /// Worker response. -pub struct WorkerResp { +pub struct TaskResult { pub result: FetchResult, pub session: WireSession, } @@ -50,7 +51,7 @@ pub struct WorkerResp { /// A worker that replicates git objects. struct Worker { storage: Storage, - tasks: chan::Receiver>, + tasks: chan::Receiver>, daemon: net::SocketAddr, timeout: time::Duration, handle: Handle, @@ -68,8 +69,8 @@ impl Worker { } } - fn process(&mut self, task: WorkerReq) { - let WorkerReq { + fn process(&mut self, task: Task) { + let Task { fetch, session, drain, @@ -86,7 +87,7 @@ impl Worker { if self .handle - .worker_result(WorkerResp { result, session }) + .worker_result(TaskResult { result, session }) .is_err() { log::error!(target: "worker", "Unable to report fetch result: worker channel disconnected"); @@ -256,14 +257,14 @@ impl Worker { } /// A pool of workers. One thread is allocated for each worker. -pub struct WorkerPool { +pub struct Pool { pool: Vec>>, } -impl WorkerPool { +impl Pool { /// Create a new worker pool with the given parameters. pub fn with( - tasks: chan::Receiver>, + tasks: chan::Receiver>, handle: Handle, config: Config, ) -> Self {