vmm: api: make REST API multi threaded

On-behalf-of: SAP sebastian.eydam@sap.com
Signed-off-by: Sebastian Eydam <sebastian.eydam@cyberus-technology.de>

Author: amphi

vmm/src/api/http/mod.rs

@@ -6,6 +6,7 @@
 use std::collections::BTreeMap;
 use std::error::Error;
 use std::fs::File;
+use std::os::fd::AsRawFd;
 use std::os::unix::io::{IntoRawFd, RawFd};
 use std::os::unix::net::UnixListener;
 use std::panic::AssertUnwindSafe;
@@ -16,11 +17,13 @@ use std::thread;
 
 use hypervisor::HypervisorType;
 use micro_http::{
-    Body, HttpServer, MediaType, Method, Request, Response, ServerError, StatusCode, Version,
+    Body, HttpServer, MediaType, Method, Request, Response, ServerError, ServerRequest,
+    ServerResponse, StatusCode, Version,
 };
 use seccompiler::{SeccompAction, apply_filter};
 use serde_json::Error as SerdeError;
 use thiserror::Error;
+use vmm_sys_util::epoll::{ControlOperation, Epoll, EpollEvent, EventSet};
 use vmm_sys_util::eventfd::EventFd;
 
 use self::http_endpoint::{VmActionHandler, VmCreate, VmInfo, VmmPing, VmmShutdown};
@@ -314,6 +317,188 @@ fn handle_http_request(
     response
 }
 
+// This answer is returned by the worker threads to the HTTP Server thread. The
+// ID is used to mark the worker that just sent the response as ready.
+struct WorkerAnswer {
+    id: usize,
+    response: ServerResponse,
+}
+struct HttpWorkerThreads {
+    // The worker threads themselves.
+    threads: Vec<thread::JoinHandle<Result<()>>>,
+    // Workers that are ready for more work are marked with true.
+    threads_ready: Vec<bool>,
+    // An MPSC channel for every worker. We use this to send server requests
+    // to the workers.
+    request_txs: Vec<std::sync::mpsc::SyncSender<ServerRequest>>,
+    // If all workers are busy, we queue server requests here.
+    request_queue: Vec<ServerRequest>,
+    // A counter to select the workers round-robin.
+    next_request: usize,
+    // An MPSC channel that the workers use to send responses to the HTTP
+    // server thread.
+    response_rx: std::sync::mpsc::Receiver<WorkerAnswer>,
+    // Workers signal this eventfd when they have a response for the HTTP
+    // server thread.
+    response_event: EventFd,
+}
+
+impl HttpWorkerThreads {
+    fn new(
+        thread_count: usize,
+        api_notifier: EventFd,
+        api_sender: Sender<ApiRequest>,
+        seccomp_action: &SeccompAction,
+        hypervisor_type: HypervisorType,
+        landlock_enable: bool,
+        exit_evt: EventFd,
+    ) -> Result<Self> {
+        let response_event = EventFd::new(libc::EFD_NONBLOCK).map_err(VmmError::EventFdCreate)?;
+        let (response_tx, response_rx) =
+            std::sync::mpsc::sync_channel::<WorkerAnswer>(thread_count);
+
+        let mut threads = Vec::new();
+        let mut threads_ready = Vec::new();
+        let mut request_txs = Vec::new();
+
+        // We use the same seccomp filter that we already use for the HTTP server thread.
+        let api_seccomp_filter =
+            get_seccomp_filter(seccomp_action, Thread::HttpApi, hypervisor_type)
+                .map_err(VmmError::CreateSeccompFilter)?;
+
+        for n in 0..thread_count {
+            let response_event = response_event.try_clone().map_err(VmmError::EventFdClone)?;
+
+            let response_tx = response_tx.clone();
+            let (request_tx, request_rx) = std::sync::mpsc::sync_channel::<ServerRequest>(1);
+
+            let api_notifier = api_notifier.try_clone().map_err(VmmError::EventFdClone)?;
+            let api_sender = api_sender.clone();
+
+            let api_seccomp_filter = api_seccomp_filter.clone();
+            let exit_evt = exit_evt.try_clone().map_err(VmmError::EventFdClone)?;
+
+            let thread = thread::Builder::new()
+                .name(format!("http-worker-{n}").to_string())
+                .spawn(move || {
+                    info!("Spawned HTTP worker thread with id {n}",);
+                    if !api_seccomp_filter.is_empty() {
+                        apply_filter(&api_seccomp_filter)
+                            .map_err(VmmError::ApplySeccompFilter)
+                            .map_err(|e| {
+                                error!("Error applying seccomp filter: {:?}", e);
+                                exit_evt.write(1).ok();
+                                e
+                            })?;
+                    }
+
+                    if landlock_enable {
+                        Landlock::new()
+                            .map_err(VmmError::CreateLandlock)?
+                            .restrict_self()
+                            .map_err(VmmError::ApplyLandlock)
+                            .map_err(|e| {
+                                error!("Error applying landlock to http-worker thread: {:?}", e);
+                                exit_evt.write(1).ok();
+                                e
+                            })?;
+                    }
+
+                    std::panic::catch_unwind(AssertUnwindSafe(move || {
+                        loop {
+                            let id = n;
+                            match request_rx.recv() {
+                                Ok(msg) => {
+                                    // Process the server request
+                                    let response = msg.process(|request| {
+                                        handle_http_request(request, &api_notifier, &api_sender)
+                                    });
+
+                                    // Send the response to the HTTP server thread together with this
+                                    // threads id.
+                                    let response = WorkerAnswer { id, response };
+                                    if let Err(e) = response_tx.send(response) {
+                                        error!(
+                                            "HTTP worker thread {id}: error sending response {}",
+                                            e
+                                        );
+                                        break;
+                                    }
+
+                                    // Notify the HTTP server thread.
+                                    response_event.write(1).ok();
+                                }
+                                Err(e) => {
+                                    error!(
+                                        "HTTP worker thread {id}: error receiving request {}",
+                                        e
+                                    );
+                                    break;
+                                }
+                            }
+                        }
+                    }))
+                    .map_err(|_| {
+                        error!("http-worker thread {n} panicked");
+                        exit_evt.write(1).ok()
+                    })
+                    .ok();
+
+                    Ok(())
+                })
+                .map_err(VmmError::HttpThreadSpawn)?;
+
+            threads.push(thread);
+            threads_ready.push(true);
+            request_txs.push(request_tx);
+        }
+
+        Ok(Self {
+            threads,
+            threads_ready,
+            request_txs,
+            request_queue: Vec::new(),
+            next_request: 0,
+            response_rx,
+            response_event,
+        })
+    }
+
+    /// Returns a channel to a worker thread that is ready for more work, if
+    /// any workers are ready.
+    fn next_worker(&mut self) -> std::option::Option<&std::sync::mpsc::SyncSender<ServerRequest>> {
+        let next: usize = self.next_request;
+        self.next_request = next.wrapping_add(1);
+
+        // Check if any worker is ready.
+        if !self.threads_ready.iter().any(|v| *v == true) {
+            return None;
+        }
+
+        let worker = next % self.threads.len();
+        if self.threads_ready[worker] {
+            // Happy path: the next worker is ready for more work.
+            self.threads_ready[worker] = false;
+            return Some(&self.request_txs[worker]);
+        }
+
+        // If the selected worker is not ready, check the next worker.
+        self.next_worker()
+    }
+}
+
+impl Drop for HttpWorkerThreads {
+    fn drop(&mut self) {
+        // Dropping the Sender side of the request channels to throw the worker
+        // threads out of their loops.
+        self.request_txs.clear();
+        // Now we can join each thread.
+        self.threads
+            .drain(..)
+            .for_each(|thread| thread.join().unwrap().unwrap());
+    }
+}
+
 fn start_http_thread(
     mut server: HttpServer,
     api_notifier: EventFd,
@@ -334,6 +519,42 @@ fn start_http_thread(
         .add_kill_switch(api_shutdown_fd_clone)
         .map_err(VmmError::CreateApiServer)?;
 
+    // We use the epoll mechanism to parallelize this. The epoll tokens are
+    // attached when registering the FDs with epoll. That way we can later
+    // check why we were notified.
+    const HTTP_EPOLL_TOKEN: u64 = 1;
+    const WORKER_EPOLL_TOKEN: u64 = 2;
+
+    // The epoll instance our HTTP server thread will wait on.
+    let outer_epoll = Epoll::new().unwrap();
+    let mut worker_threads = HttpWorkerThreads::new(
+        2,
+        api_notifier,
+        api_sender,
+        seccomp_action,
+        hypervisor_type,
+        landlock_enable,
+        exit_evt.try_clone().unwrap(),
+    )?;
+
+    // Register the fd that the worker threads will signal.
+    outer_epoll
+        .ctl(
+            ControlOperation::Add,
+            worker_threads.response_event.as_raw_fd(),
+            EpollEvent::new(EventSet::IN, WORKER_EPOLL_TOKEN),
+        )
+        .unwrap();
+
+    // Register the HttpServer's fd.
+    outer_epoll
+        .ctl(
+            ControlOperation::Add,
+            server.epoll().as_raw_fd(),
+            EpollEvent::new(EventSet::IN, HTTP_EPOLL_TOKEN),
+        )
+        .unwrap();
+
     let thread = thread::Builder::new()
         .name("http-server".to_string())
         .spawn(move || {
@@ -361,27 +582,60 @@ fn start_http_thread(
             }
 
             std::panic::catch_unwind(AssertUnwindSafe(move || {
+                let mut events = vec![EpollEvent::default(); 32];
                 server.start_server().unwrap();
+
                 loop {
-                    match server.requests() {
-                        Ok(request_vec) => {
-                            for server_request in request_vec {
-                                if let Err(e) = server.respond(server_request.process(|request| {
-                                    handle_http_request(request, &api_notifier, &api_sender)
-                                })) {
+                    let n = outer_epoll.wait(-1, &mut events).unwrap();
+                    for ev in events.iter().take(n) {
+                        match ev.data() {
+                            HTTP_EPOLL_TOKEN => {
+                                // The HttpServer got a request, handle that.
+                                match server.requests() {
+                                    Ok(request_vec) => {
+                                        for server_request in request_vec {
+                                            match worker_threads.next_worker() {
+                                                Some(worker_tx) => worker_tx.send(server_request).unwrap(),
+                                                None => worker_threads.request_queue.push(server_request),
+                                            }
+                                        }
+                                    }
+                                    Err(ServerError::ShutdownEvent) => {
+                                        server.flush_outgoing_writes();
+                                        return;
+                                    }
+                                    Err(e) => {
+                                        error!(
+                                            "HTTP server error on retrieving incoming request. Error: {}",
+                                            e
+                                        );
+                                    }
+                                }
+                            }
+                            WORKER_EPOLL_TOKEN => {
+                                // One of the worker threads has a response.
+                                // We clear the eventfd first.
+                                let _ = worker_threads.response_event.read().unwrap();
+                                let response = worker_threads.response_rx.recv().unwrap();
+                                if let Err(e) = server.respond(response.response){
                                     error!("HTTP server error on response: {}", e);
                                 }
+
+                                // We mark that worker thread as ready for more work.
+                                worker_threads.threads_ready[response.id] = true;
+                                // We check whether a request arrived while we no workers were ready.
+                                if !worker_threads.request_queue.is_empty() {
+                                    let request = worker_threads.request_queue.pop().unwrap();
+                                    match worker_threads.next_worker() {
+                                        Some(worker_tx) => worker_tx.send(request).unwrap(),
+                                        None => {
+                                            // This should not happen, because we just marked a worker as ready.
+                                            warn!("HTTP Server: no worker ready, but one worker was just marked as ready.");
+                                            worker_threads.request_queue.push(request)},
+                                    }
+                                }
                             }
-                        }
-                        Err(ServerError::ShutdownEvent) => {
-                            server.flush_outgoing_writes();
-                            return;
-                        }
-                        Err(e) => {
-                            error!(
-                                "HTTP server error on retrieving incoming request. Error: {}",
-                                e
-                            );
+                            _ => { }
                         }
                     }
                 }

vmm/src/seccomp_filters.rs

@@ -868,6 +868,7 @@ fn http_api_thread_rules() -> Result<Vec<(i64, Vec<SeccompRule>)>, BackendError>
         (libc::SYS_rt_sigprocmask, vec![]),
         (libc::SYS_getcwd, vec![]),
         (libc::SYS_clock_nanosleep, vec![]),
+        (libc::SYS_read, vec![]),
     ])
 }