Struct Worker 

pub struct Worker {
    pub(crate) grpc_client: ControllerClient<Channel>,
    pub(crate) hostname: String,
    pub(crate) is_busy: Arc<AtomicBool>,
    pub(crate) outbound_txs: Vec<(u32, Sender<InstructionType>)>,
    pub(crate) outbound_handles: Vec<JoinHandle<()>>,
    pub(crate) abort_inbound: Arc<AtomicBool>,
}

The worker that is run at the anycast PoPs and performs measurements as instructed by the orchestrator. The worker is responsible for establishing a connection with the orchestrator, receiving tasks, and performing measurements.

Fields

grpc_client: ControllerClient<Channel>

gRPC client to communicate with the orchestrator

hostname: String

Hostname of the worker

is_busy: Arc<AtomicBool>

Whether a measurement is currently active on this worker

outbound_txs: Vec<(u32, Sender<InstructionType>)>

Instructions senders to the outbound probing threads, paired with their origin ID

outbound_handles: Vec<JoinHandle<()>>

Join handles of the outbound probing threads, awaited on graceful end before closing inbound

abort_inbound: Arc<AtomicBool>

Atomic boolean to signal the inbound thread to immediately stop listening for packets

Implementations

impl Worker

pub(crate) async fn connect( address: String, tls: &TlsOptions<'_>, ) -> Result<ControllerClient<Channel>, Box<dyn Error>>

Connect to the orchestrator.

Arguments

• address - the address of the orchestrator in string format, containing both the address (IPv4, IPv6, or hostname) and port number
• tls - the TLS settings given on the command line

Returns

A gRPC client that is connected to the orchestrator

Remarks

When TLS is enabled, the connection is secured and the orchestrator authenticated.

pub(crate) async fn connect_to_server(&mut self) -> Result<(), Box<dyn Error>>

Establish a formal connection with the orchestrator. Obtains a unique worker ID from the orchestrator, establishes a stream for receiving tasks, and handles tasks as they come in.

fn handle_start_instruction( &mut self, start: Start, worker_id: u16, abort_outbound: Arc<AtomicBool>, ) -> Result<(), Box<dyn Error>>

Start a new measurement. Marks the worker as busy, Initializes the abort signals to False (for outbound and inbound threads) Calls the function to initialize the measurement

Arguments

start - The definition of the new measurement worker_id - ID of this worker abort_outbound - Abort signal to forcefully close the outbound thread

async fn handle_end_instruction( &mut self, end_instruction: End, abort_outbound: Arc<AtomicBool>, ) -> Result<(), Box<dyn Error>>

End an ongoing measurement.

Graceful end (code 0): the outbound threads first drain any tasks still queued in their channels, then the inbound listener stays open for a grace period to capture in-flight replies before it is closed. Forceful end (code != 0): outbound and inbound threads are closed immediately, discarding any queued tasks.

Arguments

end_instruction - End instruction sent by the Orchestrator with an ending code abort_outbound - Shared boolean to forcefully close the outbound/sending thread

impl Worker

pub(crate) fn init( &mut self, start: Start, worker_id: u16, abort_outbound: Arc<AtomicBool>, ) -> Result<(), Box<dyn Error>>

Initialize a new measurement by creating outbound and inbound threads, and ensures task results are sent back to the orchestrator.

Extracts the protocol type from the measurement definition, and determines which source address to use. Creates a socket to send out probes and receive replies with, calls the appropriate inbound & outbound functions. Creates an additional thread that forwards task results to the orchestrator.

Arguments

• start - Definition of the new measurement
• worker_id - the unique ID of this worker
• abort_outbound - Forcefully signal the outbound thread to stop sending probes

fn log_probe_details(&self, origin: &Origin)

Print the Origins (i.e., source address and port values) used for this measurement

Arguments

• p_type - Protocol used
• origins - Sending origins used by this Worker

fn get_socket( is_ipv6: bool, p_type: ProtocolType, origin: Origin, is_traceroute: bool, m_id: u32, ) -> Arc<Socket>

Obtain a raw socket for the given IP version and protocol.

Arguments

• is_ipv6 - IP version used (true: IPv6)
• p_type - Protocol type used (ICMP, UDP, or TCP)
• origin - Origin used in this measurement (anycast or local unicast address)
• is_traceroute - Whether this is a traceroute measurement

Returns

Arc<Socket> containing the raw socket to send/receive from

fn try_raw_socket( domain: Domain, protocol: Protocol, is_ipv6: bool, ) -> Option<Socket>

Try to create a RAW socket with IP_HDRINCL and appropriate options. Returns None if creation or setup fails (e.g. missing CAP_NET_RAW).

impl Worker

pub async fn new(args: &ArgMatches) -> Result<Worker, Box<dyn Error>>

Create a worker instance, which includes establishing a connection with the orchestrator.

Arguments

• args - contains the parsed command-line arguments