Repository: tikv/tikv
File: src/storage/config_manager.rs (Rust)

Code before:
                    let s: ReadableSize = size.into();
                    self.configurable_db
                        .set_shared_block_cache_capacity(s.0 as usize)?;
                    // Write config to metric
                    CONFIG_ROCKSDB_GAUGE
                        .with_label_values(&[CF_DEFAULT, "block_cache_size"])
                        .set(s.0 as f64);
                }
            }
        } else if let Some(v) = change.remove("ttl_check_poll_interval") {
            let interval: ReadableDuration = v.into();
            self.ttl_checker_scheduler
                .schedule(TtlCheckerTask::UpdatePollInterval(interval.into()))
                .unwrap();
        } else if let Some(ConfigValue::Module(flow_control)) = change.remove("flow_control") {
            if let Some(v) = flow_control.get("enable") {
                let enable: bool = v.into();
                let enable_str = if enable { "true" } else { "false" };
                for cf in ALL_CFS {
                    self.configurable_db
                        .set_cf_config(cf, &[("disable_write_stall", enable_str)])
                        .unwrap();
                }
                self.flow_controller.enable(enable);
            }
            self.flow_controller.update_config(flow_control)?;
        } else if let Some(v) = change.get("scheduler_worker_pool_size") {
            let pool_size: usize = v.into();
            self.scheduler.scale_pool_size(pool_size);
        } else if let Some(v) = change.remove("memory_quota") {
            let cap: ReadableSize = v.into();
            self.scheduler.set_memory_quota_capacity(cap.0 as usize);
        }
        if let Some(ConfigValue::Module(mut io_rate_limit)) = change.remove("io_rate_limit") {
            let limiter = match get_io_rate_limiter() {
                None => return Err("IO rate limiter is not present".into()),
                Some(limiter) => limiter,
            };
            if let Some(limit) = io_rate_limit.remove("max_bytes_per_sec") {
                let limit: ReadableSize = limit.into();
                limiter.set_io_rate_limit(limit.0 as usize);
            }

            for t in IoType::iter() {
                if let Some(priority) = io_rate_limit.remove(&(t.as_str().to_owned() + "_priority"))
                {
                    let priority: IoPriority = priority.try_into()?;
                    limiter.set_io_priority(t, priority);
                }
            }
        }

Review comment (bug):
Should we call `flow_controller.update_config(flow_control)` before L79 (`flow_controller.enable(enable)`) ? As it is, I feel like there could be a small race condition and metric issue in the following scenario:

When we disable the flow controller, `checker.reset_statistics()` will be called first, but `current_cfg.enable` may still be `true` when `on_flow_info_msg` is called. So metrics like `SCHED_PENDING_COMPACTION_BYTES_GAUGE` may be non-zero after we disable the flow controller, which might be unexpected. Not sure if I understand all this correctly though. 

Code after:
                    let s: ReadableSize = size.into();
                    self.configurable_db
                        .set_shared_block_cache_capacity(s.0 as usize)?;
                    // Write config to metric
                    CONFIG_ROCKSDB_GAUGE
                        .with_label_values(&[CF_DEFAULT, "block_cache_size"])
                        .set(s.0 as f64);
                }
            }
        } else if let Some(v) = change.remove("ttl_check_poll_interval") {
            let interval: ReadableDuration = v.into();
            self.ttl_checker_scheduler
                .schedule(TtlCheckerTask::UpdatePollInterval(interval.into()))
                .unwrap();
        } else if let Some(ConfigValue::Module(flow_control)) = change.remove("flow_control") {
            // we first update the config here then trigger the side-effect of
            // `flow-control.enable`.
            self.flow_controller.update_config(flow_control.clone())?;
            if let Some(v) = flow_control.get("enable") {
                let enable: bool = v.into();
                let enable_str = if enable { "true" } else { "false" };
                for cf in ALL_CFS {
                    self.configurable_db
                        .set_cf_config(cf, &[("disable_write_stall", enable_str)])
                        .unwrap();
                }
                self.flow_controller.enable(enable);
            }
        } else if let Some(v) = change.get("scheduler_worker_pool_size") {
            let pool_size: usize = v.into();
            self.scheduler.scale_pool_size(pool_size);
        } else if let Some(v) = change.remove("memory_quota") {
            let cap: ReadableSize = v.into();
            self.scheduler.set_memory_quota_capacity(cap.0 as usize);
        }
        if let Some(ConfigValue::Module(mut io_rate_limit)) = change.remove("io_rate_limit") {
            let limiter = match get_io_rate_limiter() {
                None => return Err("IO rate limiter is not present".into()),
                Some(limiter) => limiter,
            };
            if let Some(limit) = io_rate_limit.remove("max_bytes_per_sec") {
                let limit: ReadableSize = limit.into();
                limiter.set_io_rate_limit(limit.0 as usize);
            }

            for t in IoType::iter() {
                if let Some(priority) = io_rate_limit.remove(&(t.as_str().to_owned() + "_priority"))
                {
                    let priority: IoPriority = priority.try_into()?;
                    limiter.set_io_priority(t, priority);
                }