Add server support for halting the simulation

nexosim/src/server/api/simulation.proto

@@ -49,6 +49,14 @@ message InitReply {
   }
 }
 
+message HaltRequest {}
+message HaltReply {
+  oneof result { // Always returns exactly 1 variant.
+    google.protobuf.Empty empty = 1;
+    Error error = 100;
+  }
+}
+
 message TimeRequest {}
 message TimeReply {
   oneof result { // Always returns exactly 1 variant.
@@ -160,21 +168,23 @@ message CloseSinkReply {
 message AnyRequest {
   oneof request { // Expects exactly 1 variant.
     InitRequest init_request = 1;
-    TimeRequest time_request = 2;
+    HaltRequest halt_request = 2;
-    StepRequest step_request = 3;
+    TimeRequest time_request = 3;
-    StepUntilRequest step_until_request = 4;
+    StepRequest step_request = 4;
-    ScheduleEventRequest schedule_event_request = 5;
+    StepUntilRequest step_until_request = 5;
-    CancelEventRequest cancel_event_request = 6;
+    ScheduleEventRequest schedule_event_request = 6;
-    ProcessEventRequest process_event_request = 7;
+    CancelEventRequest cancel_event_request = 7;
-    ProcessQueryRequest process_query_request = 8;
+    ProcessEventRequest process_event_request = 8;
-    ReadEventsRequest read_events_request = 9;
+    ProcessQueryRequest process_query_request = 9;
-    OpenSinkRequest open_sink_request = 10;
+    ReadEventsRequest read_events_request = 10;
-    CloseSinkRequest close_sink_request = 11;
+    OpenSinkRequest open_sink_request = 11;
+    CloseSinkRequest close_sink_request = 12;
   }
 }
 
 service Simulation {
   rpc Init(InitRequest) returns (InitReply);
+  rpc Halt(HaltRequest) returns (HaltReply);
   rpc Time(TimeRequest) returns (TimeReply);
   rpc Step(StepRequest) returns (StepReply);
   rpc StepUntil(StepUntilRequest) returns (StepUntilReply);

nexosim/src/server/codegen/simulation.v1.rs

@@ -36,6 +36,25 @@ pub mod init_reply {
     }
 }
 #[derive(Clone, Copy, PartialEq, ::prost::Message)]
+pub struct HaltRequest {}
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct HaltReply {
+    /// Always returns exactly 1 variant.
+    #[prost(oneof = "halt_reply::Result", tags = "1, 100")]
+    pub result: ::core::option::Option<halt_reply::Result>,
+}
+/// Nested message and enum types in `HaltReply`.
+pub mod halt_reply {
+    /// Always returns exactly 1 variant.
+    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    pub enum Result {
+        #[prost(message, tag = "1")]
+        Empty(()),
+        #[prost(message, tag = "100")]
+        Error(super::Error),
+    }
+}
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct TimeRequest {}
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct TimeReply {
@@ -299,7 +318,10 @@ pub mod close_sink_reply {
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct AnyRequest {
     /// Expects exactly 1 variant.
-    #[prost(oneof = "any_request::Request", tags = "1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11")]
+    #[prost(
+        oneof = "any_request::Request",
+        tags = "1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12"
+    )]
     pub request: ::core::option::Option<any_request::Request>,
 }
 /// Nested message and enum types in `AnyRequest`.
@@ -310,24 +332,26 @@ pub mod any_request {
         #[prost(message, tag = "1")]
         InitRequest(super::InitRequest),
         #[prost(message, tag = "2")]
-        TimeRequest(super::TimeRequest),
+        HaltRequest(super::HaltRequest),
         #[prost(message, tag = "3")]
-        StepRequest(super::StepRequest),
+        TimeRequest(super::TimeRequest),
         #[prost(message, tag = "4")]
-        StepUntilRequest(super::StepUntilRequest),
+        StepRequest(super::StepRequest),
         #[prost(message, tag = "5")]
-        ScheduleEventRequest(super::ScheduleEventRequest),
+        StepUntilRequest(super::StepUntilRequest),
         #[prost(message, tag = "6")]
-        CancelEventRequest(super::CancelEventRequest),
+        ScheduleEventRequest(super::ScheduleEventRequest),
         #[prost(message, tag = "7")]
-        ProcessEventRequest(super::ProcessEventRequest),
+        CancelEventRequest(super::CancelEventRequest),
         #[prost(message, tag = "8")]
-        ProcessQueryRequest(super::ProcessQueryRequest),
+        ProcessEventRequest(super::ProcessEventRequest),
         #[prost(message, tag = "9")]
-        ReadEventsRequest(super::ReadEventsRequest),
+        ProcessQueryRequest(super::ProcessQueryRequest),
         #[prost(message, tag = "10")]
-        OpenSinkRequest(super::OpenSinkRequest),
+        ReadEventsRequest(super::ReadEventsRequest),
         #[prost(message, tag = "11")]
+        OpenSinkRequest(super::OpenSinkRequest),
+        #[prost(message, tag = "12")]
         CloseSinkRequest(super::CloseSinkRequest),
     }
 }
@@ -431,6 +455,10 @@ pub mod simulation_server {
             &self,
             request: tonic::Request<super::InitRequest>,
         ) -> std::result::Result<tonic::Response<super::InitReply>, tonic::Status>;
+        async fn halt(
+            &self,
+            request: tonic::Request<super::HaltRequest>,
+        ) -> std::result::Result<tonic::Response<super::HaltReply>, tonic::Status>;
         async fn time(
             &self,
             request: tonic::Request<super::TimeRequest>,
@@ -603,6 +631,49 @@ pub mod simulation_server {
                     };
                     Box::pin(fut)
                 }
+                "/simulation.v1.Simulation/Halt" => {
+                    #[allow(non_camel_case_types)]
+                    struct HaltSvc<T: Simulation>(pub Arc<T>);
+                    impl<T: Simulation> tonic::server::UnaryService<super::HaltRequest>
+                    for HaltSvc<T> {
+                        type Response = super::HaltReply;
+                        type Future = BoxFuture<
+                            tonic::Response<Self::Response>,
+                            tonic::Status,
+                        >;
+                        fn call(
+                            &mut self,
+                            request: tonic::Request<super::HaltRequest>,
+                        ) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move {
+                                <T as Simulation>::halt(&inner, request).await
+                            };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = HaltSvc(inner);
+                        let codec = tonic::codec::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(
+                                accept_compression_encodings,
+                                send_compression_encodings,
+                            )
+                            .apply_max_message_size_config(
+                                max_decoding_message_size,
+                                max_encoding_message_size,
+                            );
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
                 "/simulation.v1.Simulation/Time" => {
                     #[allow(non_camel_case_types)]
                     struct TimeSvc<T: Simulation>(pub Arc<T>);

nexosim/src/server/run.rs

@@ -1,6 +1,7 @@
 //! Simulation server.
 
 use std::net::SocketAddr;
+#[cfg(unix)]
 use std::path::Path;
 use std::sync::Arc;
 use std::sync::Mutex;
@@ -210,6 +211,11 @@ impl simulation_server::Simulation for GrpcSimulationService {
 
         Ok(Response::new(reply))
     }
+    async fn halt(&self, request: Request<HaltRequest>) -> Result<Response<HaltReply>, Status> {
+        let request = request.into_inner();
+
+        Ok(Response::new(self.scheduler().halt(request)))
+    }
     async fn time(&self, request: Request<TimeRequest>) -> Result<Response<TimeReply>, Status> {
         let request = request.into_inner();
 

nexosim/src/server/services/scheduler_service.rs

@@ -191,6 +191,22 @@ impl SchedulerService {
             }),
         }
     }
+
+    /// Requests the simulation to stop when advancing to the next step.
+    pub(crate) fn halt(&mut self, _request: HaltRequest) -> HaltReply {
+        let reply = match self {
+            Self::Started { scheduler, .. } => {
+                scheduler.halt();
+
+                halt_reply::Result::Empty(())
+            }
+            Self::NotStarted => halt_reply::Result::Error(simulation_not_started_error()),
+        };
+
+        HaltReply {
+            result: Some(reply),
+        }
+    }
 }
 
 impl fmt::Debug for SchedulerService {

nexosim/src/simulation/scheduler.rs

@@ -183,7 +183,7 @@ impl Scheduler {
         )
     }
 
-    /// Stops the simulation on the next step.
+    /// Requests the simulation to stop when advancing to the next step.
     pub fn halt(&mut self) {
         self.0.halt()
     }
@@ -557,7 +557,7 @@ impl GlobalScheduler {
         Ok(event_key)
     }
 
-    /// Stops the simulation on the next step.
+    /// Requests the simulation to stop when advancing to the next step.
     pub(crate) fn halt(&mut self) {
         self.is_halted.store(true, Ordering::Relaxed);
     }