async fn run_execute_request(
    &self,
    execute_request: ExecuteRequest,
    process: Process,
    context: &Context,
  ) -> Result<FallibleProcessResultWithPlatform, String> {
    const MAX_RETRIES: u32 = 5;
    const MAX_BACKOFF_DURATION: Duration = Duration::from_secs(10);

    let start_time = Instant::now();
    let mut current_operation_name: Option<String> = None;
    let mut num_retries = 0;

    loop {
      // If we are currently retrying a request, then delay using an exponential backoff.
      if num_retries > 0 {
        let multiplier = thread_rng().gen_range(0, 2_u32.pow(num_retries) + 1);
        let sleep_time = self.retry_interval_duration * multiplier;
        let sleep_time = sleep_time.min(MAX_BACKOFF_DURATION);
        debug!("delaying {:?} before retry", sleep_time);
        tokio::time::delay_for(sleep_time).await;
      }

      let call_opt = call_option(&self.headers, Some(context.build_id.clone()))?;
      let rpc_result = match current_operation_name {
        None => {
          // The request has not been submitted yet. Submit the request using the REv2
          // Execute method.
          debug!(
            "no current operation: submitting execute request: build_id={}; execute_request={:?}",
            context.build_id, execute_request
          );
          self
            .execution_client
            .execute_opt(&execute_request, call_opt)
        }

        Some(ref operation_name) => {
          // The request has been submitted already. Reconnect to the status stream
          // using the REv2 WaitExecution method.
          debug!(
            "existing operation: reconnecting to operation stream: build_id={}; operation_name={}",
            context.build_id, operation_name
          );
          let mut wait_execution_request = WaitExecutionRequest::new();
          wait_execution_request.set_name(operation_name.to_owned());
          self
            .execution_client
            .wait_execution_opt(&wait_execution_request, call_opt)
        }
      };

      // Take action based on whether we received an output stream or whether there is an
      // error to resolve.
      let actionable_result = match rpc_result {
        Ok(operation_stream) => {
          // Monitor the operation stream until there is an actionable operation
          // or status to interpret.
          let compat_stream = operation_stream.compat();
          let stream_outcome = self
            .wait_on_operation_stream(compat_stream, &context.build_id)
            .await;

          match stream_outcome {
            StreamOutcome::Complete(status) => {
              trace!(
                "wait_on_operation_stream (build_id={}) returned completion={:?}",
                context.build_id,
                status
              );
              status
            }
            StreamOutcome::StreamClosed(operation_name_opt) => {
              trace!("wait_on_operation_stream (build_id={}) returned stream close, will retry operation_name={:?}", context.build_id, operation_name_opt);

              // Check if the number of request attempts sent thus far have exceeded the number
              // of retries allowed since the last successful connection. (There is no point in
              // continually submitting a request if ultimately futile.)
              if num_retries >= MAX_RETRIES {
                return Err(
                  "Too many failures from server. The last event was the server disconnecting with no error given.".to_owned(),
                );
              } else {
                // Increment the retry counter and allow loop to retry.
                num_retries += 1;
              }

              // Iterate the loop to reconnect to the operation.
              current_operation_name = operation_name_opt;
              continue;
            }
          }
        }
        Err(err) => match err {
          grpcio::Error::RpcFailure(rpc_status) => {
            let mut status = Status::new();
            status.code = rpc_status.status.into();
            OperationOrStatus::Status(status)
          }
          _ => {
            return Err(format!("gRPC error: {}", err));
          }
        },
      };

      match self.extract_execute_response(actionable_result).await {
        Ok(result) => return Ok(result),
        Err(err) => match err {
          ExecutionError::Fatal(e) => return Err(e),
          ExecutionError::Retryable(e) => {
            // Check if the number of request attempts sent thus far have exceeded the number
            // of retries allowed since the last successful connection. (There is no point in
            // continually submitting a request if ultimately futile.)
            trace!("retryable error: {}", e);
            if num_retries >= MAX_RETRIES {
              return Err(format!(
                "Too many failures from server. The last error was: {}",
                e
              ));
            } else {
              // Increment the retry counter and allow loop to retry.
              num_retries += 1;
            }
          }
          ExecutionError::MissingDigests(missing_digests) => {
            trace!(
              "Server reported missing digests; trying to upload: {:?}",
              missing_digests,
            );

            let _ = self
              .store
              .ensure_remote_has_recursive(missing_digests)
              .compat()
              .await?;
          }
          ExecutionError::Timeout => {
            return populate_fallible_execution_result_for_timeout(
              &self.store,
              &process.description,
              process.timeout,
              start_time.elapsed(),
              Vec::new(),
              self.platform,
            )
            .await
          }
        },
      }
    }
  }
// Main loop: This function connects to the RE server and submits the given remote execution
// request via the REv2 Execute method. It then monitors the operation stream until the
// request completes. It will reconnect using the REv2 WaitExecution method if the connection
// is dropped.
//
// The `run` method on CommandRunner uses this function to implement the bulk of the
// processing for remote execution requests. The `run` method wraps the call with the method
// with an overall deadline timeout.
Main loop: This function connects to the RE server and submits the given remote execution
request via the REv2 Execute method. It then monitors the operation stream until the
request completes. It will reconnect using the REv2 WaitExecution method if the connection
is dropped.

The `run` method on CommandRunner uses this function to implement the bulk of the
processing for remote execution requests. The `run` method wraps the call with the method
with an overall deadline timeout.
Main loop : This function connects to the RE server and submits the given remote execution request via the REv2 Execute method . It then monitors the operation stream until the request completes . It will reconnect using the REv2 WaitExecution method if the connection is dropped . The ` run ` method on CommandRunner uses this function to implement the bulk of the processing for remote execution requests . The ` run ` method wraps the call with the method with an overall deadline timeout .
async fn run_execute_request(
    &self,
    execute_request: ExecuteRequest,
    process: Process,
    context: &Context,
  ) -> Result<FallibleProcessResultWithPlatform, String> {
    const MAX_RETRIES: u32 = 5;
    const MAX_BACKOFF_DURATION: Duration = Duration::from_secs(10);
    let start_time = Instant::now();
    let mut current_operation_name: Option<String> = None;
    let mut num_retries = 0;
    loop {
      // If we are currently retrying a request, then delay using an exponential backoff.
      if num_retries > 0 {
        let multiplier = thread_rng().gen_range(0, 2_u32.pow(num_retries) + 1);
        let sleep_time = self.retry_interval_duration * multiplier;
        let sleep_time = sleep_time.min(MAX_BACKOFF_DURATION);
        debug!("delaying {:?} before retry", sleep_time);
        tokio::time::delay_for(sleep_time).await;
      }
      let call_opt = call_option(&self.headers, Some(context.build_id.clone()))?;
      let rpc_result = match current_operation_name {
        None => {
          // The request has not been submitted yet. Submit the request using the REv2
          // Execute method.
          debug!(
            "no current operation: submitting execute request: build_id={}; execute_request={:?}",
            context.build_id, execute_request
          );
          self
            .execution_client
            .execute_opt(&execute_request, call_opt)
        }
        Some(ref operation_name) => {
          // The request has been submitted already. Reconnect to the status stream
          // using the REv2 WaitExecution method.
          debug!(
            "existing operation: reconnecting to operation stream: build_id={}; operation_name={}",
            context.build_id, operation_name
          );
          let mut wait_execution_request = WaitExecutionRequest::new();
          wait_execution_request.set_name(operation_name.to_owned());
          self
            .execution_client
            .wait_execution_opt(&wait_execution_request, call_opt)
        }
      };
      // Take action based on whether we received an output stream or whether there is an
      // error to resolve.
      let actionable_result = match rpc_result {
        Ok(operation_stream) => {
          // Monitor the operation stream until there is an actionable operation
          // or status to interpret.
          let compat_stream = operation_stream.compat();
          let stream_outcome = self
            .wait_on_operation_stream(compat_stream, &context.build_id)
            .await;
          match stream_outcome {
            StreamOutcome::Complete(status) => {
              trace!(
                "wait_on_operation_stream (build_id={}) returned completion={:?}",
                context.build_id,
                status
              );
              status
            }
            StreamOutcome::StreamClosed(operation_name_opt) => {
              trace!("wait_on_operation_stream (build_id={}) returned stream close, will retry operation_name={:?}", context.build_id, operation_name_opt);
              // Check if the number of request attempts sent thus far have exceeded the number
              // of retries allowed since the last successful connection. (There is no point in
              // continually submitting a request if ultimately futile.)
              if num_retries >= MAX_RETRIES {
                return Err(
                  "Too many failures from server. The last event was the server disconnecting with no error given.".to_owned(),
                );
              } else {
                // Increment the retry counter and allow loop to retry.
                num_retries += 1;
              }
              // Iterate the loop to reconnect to the operation.
              current_operation_name = operation_name_opt;
              continue;
            }
          }
        }
        Err(err) => match err {
          grpcio::Error::RpcFailure(rpc_status) => {
            let mut status = Status::new();
            status.code = rpc_status.status.into();
            OperationOrStatus::Status(status)
          }
          _ => {
            return Err(format!("gRPC error: {}", err));
          }
        },
      };
      match self.extract_execute_response(actionable_result).await {
        Ok(result) => return Ok(result),
        Err(err) => match err {
          ExecutionError::Fatal(e) => return Err(e),
          ExecutionError::Retryable(e) => {
            // Check if the number of request attempts sent thus far have exceeded the number
            // of retries allowed since the last successful connection. (There is no point in
            // continually submitting a request if ultimately futile.)
            trace!("retryable error: {}", e);
            if num_retries >= MAX_RETRIES {
              return Err(format!(
                "Too many failures from server. The last error was: {}",
                e
              ));
            } else {
              // Increment the retry counter and allow loop to retry.
              num_retries += 1;
            }
          }
          ExecutionError::MissingDigests(missing_digests) => {
            trace!(
              "Server reported missing digests; trying to upload: {:?}",
              missing_digests,
            );
            let _ = self
              .store
              .ensure_remote_has_recursive(missing_digests)
              .compat()
              .await?;
          }
          ExecutionError::Timeout => {
            return populate_fallible_execution_result_for_timeout(
              &self.store,
              &process.description,
              process.timeout,
              start_time.elapsed(),
              Vec::new(),
              self.platform,
            )
            .await
          }
        },
      }
    }
  }
async fn run_execute_request ( & self , execute_request : ExecuteRequest , process : Process , context : & Context , ) -> Result < FallibleProcessResultWithPlatform , String > { const MAX_RETRIES : u32 = 5 ; const MAX_BACKOFF_DURATION : Duration = Duration :: from_secs ( 10 ) ; let start_time = Instant :: now ( ) ; let mut current_operation_name : Option < String > = None ; let mut num_retries = 0 ; loop { // If we are currently retrying a request, then delay using an exponential backoff. if num_retries > 0 { let multiplier = thread_rng ( ) . gen_range ( 0 , 2_u32 . pow ( num_retries ) + 1 ) ; let sleep_time = self . retry_interval_duration * multiplier ; let sleep_time = sleep_time . min ( MAX_BACKOFF_DURATION ) ; debug ! ( " " sleep_time ) ; tokio :: time :: delay_for ( sleep_time ) . await ; } let call_opt = call_option ( & self . headers , Some ( context . build_id . clone ( ) ) ) ? ; let rpc_result = match current_operation_name { None => { // The request has not been submitted yet. Submit the request using the REv2 // Execute method. debug ! ( " " context build_id execute_request ) ; self . execution_client . execute_opt ( & execute_request , call_opt ) } Some ( ref operation_name ) => { // The request has been submitted already. Reconnect to the status stream // using the REv2 WaitExecution method. debug ! ( " " context build_id operation_name ) ; let mut wait_execution_request = WaitExecutionRequest :: new ( ) ; wait_execution_request . set_name ( operation_name . to_owned ( ) ) ; self . execution_client . wait_execution_opt ( & wait_execution_request , call_opt ) } } ; // Take action based on whether we received an output stream or whether there is an // error to resolve. let actionable_result = match rpc_result { Ok ( operation_stream ) => { // Monitor the operation stream until there is an actionable operation // or status to interpret. let compat_stream = operation_stream . compat ( ) ; let stream_outcome = self . wait_on_operation_stream ( compat_stream , & context . build_id ) . await ; match stream_outcome { StreamOutcome :: Complete ( status ) => { trace ! ( " " context build_id status ) ; status } StreamOutcome :: StreamClosed ( operation_name_opt ) => { trace ! ( " " context build_id operation_name_opt ) ; // Check if the number of request attempts sent thus far have exceeded the number // of retries allowed since the last successful connection. (There is no point in // continually submitting a request if ultimately futile.) if num_retries >= MAX_RETRIES { return Err ( " " . to_owned ( ) , ) ; } else { // Increment the retry counter and allow loop to retry. num_retries += 1 ; } // Iterate the loop to reconnect to the operation. current_operation_name = operation_name_opt ; continue ; } } } Err ( err ) => match err { grpcio :: Error :: RpcFailure ( rpc_status ) => { let mut status = Status :: new ( ) ; status . code = rpc_status . status . into ( ) ; OperationOrStatus :: Status ( status ) } _ => { return Err ( format ! ( " " err ) ) ; } } , } ; match self . extract_execute_response ( actionable_result ) . await { Ok ( result ) => return Ok ( result ) , Err ( err ) => match err { ExecutionError :: Fatal ( e ) => return Err ( e ) , ExecutionError :: Retryable ( e ) => { // Check if the number of request attempts sent thus far have exceeded the number // of retries allowed since the last successful connection. (There is no point in // continually submitting a request if ultimately futile.) trace ! ( " " e ) ; if num_retries >= MAX_RETRIES { return Err ( format ! ( " " e ) ) ; } else { // Increment the retry counter and allow loop to retry. num_retries += 1 ; } } ExecutionError :: MissingDigests ( missing_digests ) => { trace ! ( " " missing_digests ) ; let _ = self . store . ensure_remote_has_recursive ( missing_digests ) . compat ( ) . await ? ; } ExecutionError :: Timeout => { return populate_fallible_execution_result_for_timeout ( & self . store , & process . description , process . timeout , start_time . elapsed ( ) , Vec :: new ( ) , self . platform , ) . await } } , } } }
Main loop: This function connects to the RE server and submits the given remote execution
request via the REv2 Execute method.
Main loop : This function connects to the RE server and submits the given remote execution request via the REv2 Execute method .
&self
execute_request
ExecuteRequest
process
Process
context
&Context
&self
execute_request
ExecuteRequest
process
Process
context
&Context