rate_limiter: removed custom deserialization logic

Removed the RateLimiter partial serialization logic, removing a de-facto
dependency on the VMM / custom Firecracker logic. Also changed the VMM
code that relied on the RateLimiter serialization.

Signed-off-by: Dan Horobeanu <[email protected]>

Author: dhrgit

api_server/src/request/net.rs

@@ -55,14 +55,14 @@ impl IntoParsedRequest for NetworkInterfaceUpdateConfig {
 #[cfg(test)]
 mod tests {
     extern crate net_util;
-    extern crate rate_limiter;
+    extern crate vmm;
 
     use self::net_util::MacAddr;
     use super::*;
 
     use serde_json;
 
-    use self::rate_limiter::RateLimiter;
+    use self::vmm::vmm_config::RateLimiterConfig;
 
     fn get_dummy_netif(
         iface_id: String,
@@ -117,8 +117,8 @@ mod tests {
             iface_id: String::from("foo"),
             host_dev_name: String::from("bar"),
             guest_mac: Some(MacAddr::parse_str("12:34:56:78:9A:BC").unwrap()),
-            rx_rate_limiter: Some(RateLimiter::default()),
-            tx_rate_limiter: Some(RateLimiter::default()),
+            rx_rate_limiter: Some(RateLimiterConfig::default()),
+            tx_rate_limiter: Some(RateLimiterConfig::default()),
             allow_mmds_requests: true,
             tap: None,
         };

rate_limiter/src/lib.rs

@@ -54,7 +54,6 @@ extern crate serde_derive;
 #[macro_use]
 extern crate logger;
 
-use serde::de::{self, Deserialize, Deserializer, MapAccess, Visitor};
 use std::os::unix::io::{AsRawFd, RawFd};
 use std::time::Duration;
 use std::{fmt, io};
@@ -290,67 +289,6 @@ impl fmt::Debug for RateLimiter {
     }
 }
 
-impl<'de> Deserialize<'de> for RateLimiter {
-    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
-    where
-        D: Deserializer<'de>,
-    {
-        #[derive(Deserialize)]
-        #[allow(non_camel_case_types)]
-        enum Field {
-            bandwidth,
-            ops,
-        };
-
-        struct RateLimiterVisitor;
-
-        impl<'de> Visitor<'de> for RateLimiterVisitor {
-            type Value = RateLimiter;
-
-            fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
-                formatter.write_str("struct RateLimiter")
-            }
-
-            fn visit_map<V>(self, mut map: V) -> Result<RateLimiter, V::Error>
-            where
-                V: MapAccess<'de>,
-            {
-                let mut bandwidth: Option<TokenBucket> = None;
-                let mut ops: Option<TokenBucket> = None;
-                while let Some(key) = map.next_key()? {
-                    match key {
-                        Field::bandwidth => {
-                            if bandwidth.is_some() {
-                                return Err(de::Error::duplicate_field("bandwidth"));
-                            }
-                            bandwidth = Some(map.next_value()?);
-                        }
-                        Field::ops => {
-                            if ops.is_some() {
-                                return Err(de::Error::duplicate_field("ops"));
-                            }
-                            ops = Some(map.next_value()?);
-                        }
-                    }
-                }
-                let bandwidth = bandwidth.unwrap_or_default();
-                let ops = ops.unwrap_or_default();
-                RateLimiter::new(
-                    bandwidth.size,
-                    bandwidth.one_time_burst,
-                    bandwidth.refill_time,
-                    ops.size,
-                    ops.one_time_burst,
-                    ops.refill_time,
-                )
-                .map_err(de::Error::custom)
-            }
-        }
-        const FIELDS: &'static [&'static str] = &["bandwidth", "ops"];
-        deserializer.deserialize_struct("RateLimiter", FIELDS, RateLimiterVisitor)
-    }
-}
-
 impl RateLimiter {
     // description
     fn make_bucket(
@@ -499,9 +437,14 @@ impl RateLimiter {
     /// Updates the parameters of the token buckets associated with this RateLimiter.
     // TODO: Pls note that, right now, the buckets become full after being updated.
     pub fn update_buckets(&mut self, bytes: Option<TokenBucket>, ops: Option<TokenBucket>) {
-        // TODO: We have to call make_bucket instead of directly assigning the bytes and/or ops
-        // because the input buckets are likely build via deserialization, which currently does not
-        // properly set up their internal state.
+        // TODO: We should reconcile the create and update paths, such that they use the same data
+        // format. Currently, the TokenBucket config data is used for create, but the live
+        // TokenBucket objects are used for update.
+        // We have to call make_bucket instead of directly assigning the bytes and/or ops
+        // because the RateLimiter validates the TokenBucket config data (e.g. it nullifies
+        // an unusable bucket with size 0). This is needed, because passing a 0-sized bucket is
+        // the only method the user has to disable rate limiting. I.e. if the user passes `null`
+        // as the token bucket config, the old config is left unchanged.
 
         if let Some(b) = bytes {
             self.bandwidth = Self::make_bucket(b.size, b.one_time_burst, b.refill_time);
@@ -804,51 +747,9 @@ mod tests {
         //assert!(!l.consume(u64::max_value(), TokenType::Bytes));
     }
 
-    #[test]
-    fn test_rate_limiter_deserialization() {
-        let jstr = r#"{
-            "bandwidth": { "size": 1000, "one_time_burst": 2000,  "refill_time": 1000 },
-            "ops": { "size": 10, "one_time_burst": 20, "refill_time": 1000 }
-        }"#;
-
-        let x: RateLimiter = serde_json::from_str(jstr).expect("deserialization failed.");
-        assert_eq!(x.bandwidth.as_ref().unwrap().size, 1000);
-        assert_eq!(x.bandwidth.as_ref().unwrap().one_time_burst.unwrap(), 2000);
-        assert_eq!(x.bandwidth.as_ref().unwrap().refill_time, 1000);
-        assert_eq!(x.ops.as_ref().unwrap().size, 10);
-        assert_eq!(x.ops.as_ref().unwrap().one_time_burst.unwrap(), 20);
-        assert_eq!(x.ops.as_ref().unwrap().refill_time, 1000);
-
-        let jstr = r#"{
-            "ops": { "size": 10, "one_time_burst": 20, "refill_time": 1000 }
-        }"#;
-
-        let x: RateLimiter = serde_json::from_str(jstr).expect("deserialization failed.");
-        assert!(x.bandwidth.is_none());
-        assert_eq!(x.ops.as_ref().unwrap().size, 10);
-        assert_eq!(x.ops.as_ref().unwrap().one_time_burst.unwrap(), 20);
-        assert_eq!(x.ops.as_ref().unwrap().refill_time, 1000);
-        assert_eq!(x.timer_active, false);
-
-        let jstr = r#"{
-            "bandwidth": { "size": 1000, "one_time_burst": 2000,  "refill_time": 1000 },
-            "opz": { "size": 10, "one_time_burst": 20, "refill_time": 1000 }
-        }"#;
-        assert!(serde_json::from_str::<RateLimiter>(jstr).is_err());
-
-        let jstr = r#"{
-        }"#;
-        assert!(serde_json::from_str::<RateLimiter>(jstr).is_ok());
-    }
-
     #[test]
     fn test_update_buckets() {
-        let jstr = r#"{
-            "bandwidth": { "size": 1000, "one_time_burst": 2000,  "refill_time": 1000 },
-            "ops": { "size": 10, "one_time_burst": 20, "refill_time": 1000 }
-        }"#;
-
-        let mut x: RateLimiter = serde_json::from_str(jstr).unwrap();
+        let mut x = RateLimiter::new(1000, Some(2000), 1000, 10, Some(20), 1000).unwrap();
 
         let initial_bw = x.bandwidth.clone();
         let initial_ops = x.ops.clone();

vmm/src/lib.rs

@@ -745,13 +745,21 @@ impl Vmm {
             let (epoll_config, handler_idx) = epoll_context.allocate_virtio_block_tokens();
             self.drive_handler_id_map
                 .insert(drive_config.drive_id.clone(), handler_idx);
+            let rate_limiter = match drive_config.rate_limiter {
+                Some(rlim_cfg) => Some(
+                    rlim_cfg
+                        .into_rate_limiter()
+                        .map_err(StartMicrovmError::CreateRateLimiter)?,
+                ),
+                None => None,
+            };
 
             let block_box = Box::new(
                 devices::virtio::Block::new(
                     block_file,
                     drive_config.is_read_only,
                     epoll_config,
-                    drive_config.rate_limiter.take(),
+                    rate_limiter,
                 )
                 .map_err(StartMicrovmError::CreateBlockDevice)?,
             );
@@ -783,8 +791,20 @@ impl Vmm {
                 .insert(cfg.iface_id.clone(), handler_idx);
 
             let allow_mmds_requests = cfg.allow_mmds_requests();
-            let rx_rate_limiter = cfg.rx_rate_limiter.take();
-            let tx_rate_limiter = cfg.tx_rate_limiter.take();
+            let rx_rate_limiter = match cfg.rx_rate_limiter {
+                Some(rlim) => Some(
+                    rlim.into_rate_limiter()
+                        .map_err(StartMicrovmError::CreateRateLimiter)?,
+                ),
+                None => None,
+            };
+            let tx_rate_limiter = match cfg.tx_rate_limiter {
+                Some(rlim) => Some(
+                    rlim.into_rate_limiter()
+                        .map_err(StartMicrovmError::CreateRateLimiter)?,
+                ),
+                None => None,
+            };
 
             if let Some(tap) = cfg.take_tap() {
                 let net_box = Box::new(
@@ -1565,41 +1585,23 @@ impl Vmm {
 
             // Check if we need to update the RX rate limiter.
             if let Some(new_rlim_cfg) = new_cfg.rx_rate_limiter {
-                if let Some(ref mut old_rlim) = old_cfg.rx_rate_limiter {
+                if let Some(ref mut old_rlim_cfg) = old_cfg.rx_rate_limiter {
                     // We already have an RX rate limiter set, so we'll update it.
-                    old_rlim.update_buckets(
-                        new_rlim_cfg.bandwidth.map(|b| b.into_token_bucket()),
-                        new_rlim_cfg.ops.map(|b| b.into_token_bucket()),
-                    );
+                    old_rlim_cfg.update(&new_rlim_cfg);
                 } else {
                     // No old RX rate limiter; create one now.
-                    old_cfg.rx_rate_limiter =
-                        Some(new_rlim_cfg.into_rate_limiter().map_err(|e| {
-                            VmmActionError::NetworkConfig(
-                                ErrorKind::Internal,
-                                NetworkInterfaceError::RateLimiterError(e),
-                            )
-                        })?);
+                    old_cfg.rx_rate_limiter = Some(new_rlim_cfg);
                 }
             }
 
             // Check if we need to update the TX rate limiter.
             if let Some(new_rlim_cfg) = new_cfg.tx_rate_limiter {
-                if let Some(ref mut old_rlim) = old_cfg.tx_rate_limiter {
+                if let Some(ref mut old_rlim_cfg) = old_cfg.tx_rate_limiter {
                     // We already have a TX rate limiter set, so we'll update it.
-                    old_rlim.update_buckets(
-                        new_rlim_cfg.bandwidth.map(|b| b.into_token_bucket()),
-                        new_rlim_cfg.ops.map(|b| b.into_token_bucket()),
-                    );
+                    old_rlim_cfg.update(&new_rlim_cfg);
                 } else {
                     // No old TX rate limiter; create one now.
-                    old_cfg.tx_rate_limiter =
-                        Some(new_rlim_cfg.into_rate_limiter().map_err(|e| {
-                            VmmActionError::NetworkConfig(
-                                ErrorKind::Internal,
-                                NetworkInterfaceError::RateLimiterError(e),
-                            )
-                        })?);
+                    old_cfg.tx_rate_limiter = Some(new_rlim_cfg);
                 }
             }
 

vmm/src/vmm_config/drive.rs

@@ -7,7 +7,7 @@ use std::fmt::{Display, Formatter};
 use std::path::PathBuf;
 use std::result;
 
-use rate_limiter::RateLimiter;
+use super::RateLimiterConfig;
 
 type Result<T> = result::Result<T, DriveError>;
 
@@ -74,7 +74,7 @@ pub struct BlockDeviceConfig {
     /// drive is opened as read-write.
     pub is_read_only: bool,
     /// Rate Limiter for I/O operations.
-    pub rate_limiter: Option<RateLimiter>,
+    pub rate_limiter: Option<RateLimiterConfig>,
 }
 
 impl BlockDeviceConfig {

vmm/src/vmm_config/instance_info.rs

@@ -57,6 +57,8 @@ pub enum StartMicrovmError {
     /// Internal errors are due to resource exhaustion.
     /// Users errors are due to invalid permissions.
     CreateNetDevice(devices::virtio::Error),
+    /// Failed to create a `RateLimiter` object.
+    CreateRateLimiter(std::io::Error),
     #[cfg(feature = "vsock")]
     /// Creating a vsock device can only fail if the /dev/vhost-vsock device cannot be open.
     CreateVsockDevice(devices::virtio::vhost::Error),
@@ -125,6 +127,7 @@ impl Display for StartMicrovmError {
                  the file was deleted/corrupted. Error number: {}",
                 err.errno().to_string()
             ),
+            CreateRateLimiter(ref err) => write!(f, "Cannot create RateLimiter: {}", err),
             #[cfg(feature = "vsock")]
             CreateVsockDevice(ref err) => {
                 let mut err_msg = format!("{:?}", err);

vmm/src/vmm_config/mod.rs

@@ -31,7 +31,7 @@ pub mod vsock;
 
 /// A public-facing, stateless structure, holding all the data we need to create a TokenBucket
 /// (live) object.
-#[derive(Clone, Copy, Debug, Default, Deserialize)]
+#[derive(Clone, Copy, Debug, Default, Deserialize, PartialEq)]
 pub struct TokenBucketConfig {
     /// See TokenBucket::size.
     pub size: u64,
@@ -54,7 +54,7 @@ impl TokenBucketConfig {
 
 /// A public-facing, stateless structure, holding all the data we need to create a RateLimiter
 /// (live) object.
-#[derive(Clone, Copy, Debug, Default, Deserialize)]
+#[derive(Clone, Copy, Debug, Default, Deserialize, PartialEq)]
 pub struct RateLimiterConfig {
     /// Data used to initialize the RateLimiter::bandwidth bucket.
     pub bandwidth: Option<TokenBucketConfig>,
@@ -76,4 +76,13 @@ impl RateLimiterConfig {
             ops.refill_time,
         )
     }
+    /// Updates the configuration, merging in new options from `new_config`.
+    pub fn update(&mut self, new_config: &RateLimiterConfig) {
+        if new_config.bandwidth.is_some() {
+            self.bandwidth = new_config.bandwidth;
+        }
+        if new_config.ops.is_some() {
+            self.ops = new_config.ops;
+        }
+    }
 }

vmm/src/vmm_config/net.rs

@@ -9,7 +9,6 @@ use super::super::Error as VmmInternalError;
 use super::RateLimiterConfig;
 use devices;
 use net_util::{MacAddr, Tap, TapError};
-use rate_limiter::RateLimiter;
 
 /// This struct represents the strongly typed equivalent of the json body from net iface
 /// related requests.
@@ -23,9 +22,9 @@ pub struct NetworkInterfaceConfig {
     /// Guest MAC address.
     pub guest_mac: Option<MacAddr>,
     /// Rate Limiter for received packages.
-    pub rx_rate_limiter: Option<RateLimiter>,
+    pub rx_rate_limiter: Option<RateLimiterConfig>,
     /// Rate Limiter for transmitted packages.
-    pub tx_rate_limiter: Option<RateLimiter>,
+    pub tx_rate_limiter: Option<RateLimiterConfig>,
     #[serde(default = "default_allow_mmds_requests")]
     /// If this field is set, the device model will reply to HTTP GET
     /// requests sent to the MMDS address via this interface. In this case,
@@ -297,8 +296,8 @@ mod tests {
             iface_id: String::from(id),
             host_dev_name: String::from(name),
             guest_mac: Some(MacAddr::parse_str(mac).unwrap()),
-            rx_rate_limiter: Some(RateLimiter::default()),
-            tx_rate_limiter: Some(RateLimiter::default()),
+            rx_rate_limiter: Some(RateLimiterConfig::default()),
+            tx_rate_limiter: Some(RateLimiterConfig::default()),
             allow_mmds_requests: false,
             tap: None,
         }