From 46fa95d7461d38b1c6cafe3ab00fa2798f5517e0 Mon Sep 17 00:00:00 2001
From: Liang Zhang <zhangliang5@huawei.com>
Date: Sat, 20 Feb 2021 16:27:36 +0800
Subject: [PATCH] qos: add support for iops

1. add support for Timer used by qos
2. add leaky bucket algorithm
3. add cmdline configure interface and use guide

Signed-off-by: Liang Zhang <zhangliang5@huawei.com>
---
 device_model/src/micro_vm/mod.rs  |   1 +
 device_model/src/virtio/block.rs  | 124 +++++++++++++++++++++---------
 device_model/src/virtio/queue.rs  |   7 ++
 docs/config_guidebook.md          |   8 +-
 machine_manager/src/cmdline.rs    |   2 +-
 machine_manager/src/config/fs.rs  |  18 ++++-
 machine_manager/src/event_loop.rs |  40 +++++-----
 util/src/leak_bucket.rs           | 116 ++++++++++++++++++++++++++++
 util/src/lib.rs                   |   1 +
 util/src/loop_context.rs          |  94 +++++++++++++++++++++-
 10 files changed, 346 insertions(+), 65 deletions(-)
 create mode 100644 util/src/leak_bucket.rs

diff --git a/device_model/src/micro_vm/mod.rs b/device_model/src/micro_vm/mod.rs
index 23243c2..2e9c58a 100644
--- a/device_model/src/micro_vm/mod.rs
+++ b/device_model/src/micro_vm/mod.rs
@@ -1006,6 +1006,7 @@ impl DeviceInterface for LightMachine {
             direct,
             serial_num: None,
             iothread: None,
+            iops: None,
         };
 
         match self.bus.add_replaceable_config(node_name, Arc::new(config)) {
diff --git a/device_model/src/virtio/block.rs b/device_model/src/virtio/block.rs
index 913afd3..ae31cf6 100644
--- a/device_model/src/virtio/block.rs
+++ b/device_model/src/virtio/block.rs
@@ -28,6 +28,7 @@ use machine_manager::{
 };
 use util::aio::{Aio, AioCb, AioCompleteFunc, IoCmd, Iovec};
 use util::byte_code::ByteCode;
+use util::leak_bucket::LeakBucket;
 use util::loop_context::{
     read_fd, EventNotifier, EventNotifierHelper, NotifierCallback, NotifierOperation,
 };
@@ -381,31 +382,35 @@ impl Request {
 }
 
 /// Control block of Block IO.
-pub struct BlockIoHandler {
+struct BlockIoHandler {
     /// The virtqueue.
-    pub queue: Arc<Mutex<Queue>>,
+    queue: Arc<Mutex<Queue>>,
     /// Eventfd of the virtqueue for IO event.
-    pub queue_evt: EventFd,
+    queue_evt: EventFd,
     /// The address space to which the block device belongs.
-    pub mem_space: Arc<AddressSpace>,
+    mem_space: Arc<AddressSpace>,
     /// The image file opened by the block device.
-    pub disk_image: Option<File>,
+    disk_image: Option<File>,
     /// The number of sectors of the disk image.
-    pub disk_sectors: u64,
+    disk_sectors: u64,
     /// Serial number of the block device.
-    pub serial_num: Option<String>,
+    serial_num: Option<String>,
     /// if use direct access io.
-    pub direct: bool,
+    direct: bool,
     /// Aio context.
-    pub aio: Option<Box<Aio<AioCompleteCb>>>,
+    aio: Option<Box<Aio<AioCompleteCb>>>,
     /// Bit mask of features negotiated by the backend and the frontend.
-    pub driver_features: u64,
+    driver_features: u64,
     /// The receiving half of Rust's channel to receive the image file.
     receiver: Receiver<SenderConfig>,
     /// Eventfd for config space update.
     update_evt: RawFd,
     /// Callback to trigger an interrupt.
-    pub interrupt_cb: Arc<VirtioBlockInterrupt>,
+    interrupt_cb: Arc<VirtioBlockInterrupt>,
+    /// thread name of io handler
+    iothread: Option<String>,
+    /// Using the leak bucket to implement IO limits
+    leak_bucket: Option<LeakBucket>,
 }
 
 impl BlockIoHandler {
@@ -417,13 +422,24 @@ impl BlockIoHandler {
         let mut last_aio_req_index = 0;
         let mut need_interrupt = false;
 
-        while let Ok(elem) = self
-            .queue
-            .lock()
-            .unwrap()
-            .vring
-            .pop_avail(&self.mem_space, self.driver_features)
-        {
+        let mut queue = self.queue.lock().unwrap();
+
+        while let Ok(elem) = queue.vring.pop_avail(&self.mem_space, self.driver_features) {
+            // limit io operations if iops is configured
+            if let Some(lb) = self.leak_bucket.as_mut() {
+                if let Some(ctx) = EventLoop::get_ctx(self.iothread.as_ref()) {
+                    if lb.throttled(ctx) {
+                        queue.vring.push_back();
+                        break;
+                    }
+                } else {
+                    bail!(
+                        "IOThread {:?} of Block is not found in cmdline.",
+                        self.iothread,
+                    );
+                };
+            }
+
             match Request::new(&self.mem_space, &elem) {
                 Ok(req) => {
                     match req.out_header.request_type {
@@ -445,6 +461,9 @@ impl BlockIoHandler {
             };
         }
 
+        // unlock queue, because it will be hold below.
+        drop(queue);
+
         if let Some(disk_img) = self.disk_image.as_mut() {
             req_index = 0;
             for req in req_queue.iter() {
@@ -530,7 +549,7 @@ impl BlockIoHandler {
     }
 
     /// Build an aio context.
-    pub fn build_aio(&self) -> Result<Box<Aio<AioCompleteCb>>> {
+    fn build_aio(&self) -> Result<Box<Aio<AioCompleteCb>>> {
         let complete_func = Arc::new(Box::new(move |aiocb: &AioCb<AioCompleteCb>, ret: i64| {
             let status = if ret < 0 {
                 ret
@@ -614,26 +633,25 @@ fn build_event_notifier(fd: RawFd, handler: Box<NotifierCallback>) -> EventNotif
 }
 
 impl EventNotifierHelper for BlockIoHandler {
-    fn internal_notifiers(block_io: Arc<Mutex<Self>>) -> Vec<EventNotifier> {
+    fn internal_notifiers(handler: Arc<Mutex<Self>>) -> Vec<EventNotifier> {
+        let handler_raw = handler.lock().unwrap();
         let mut notifiers = Vec::new();
-        let locked_block_io = block_io.lock().unwrap();
 
         // Register event notifier for update_evt.
-        let cloned_block_io = block_io.clone();
-        let handler: Box<NotifierCallback> = Box::new(move |_, fd: RawFd| {
+        let h_clone = handler.clone();
+        let h: Box<NotifierCallback> = Box::new(move |_, fd: RawFd| {
             read_fd(fd);
-            cloned_block_io.lock().unwrap().update_evt_handler();
+            h_clone.lock().unwrap().update_evt_handler();
             None
         });
-        notifiers.push(build_event_notifier(locked_block_io.update_evt, handler));
+        notifiers.push(build_event_notifier(handler_raw.update_evt, h));
 
         // Register event notifier for queue_evt.
-        let cloned_block_io = block_io.clone();
-        let handler: Box<NotifierCallback> = Box::new(move |_, fd: RawFd| {
+        let h_clone = handler.clone();
+        let h: Box<NotifierCallback> = Box::new(move |_, fd: RawFd| {
             read_fd(fd);
 
-            let mut locked_block_io = cloned_block_io.lock().unwrap();
-            if let Err(ref e) = locked_block_io.process_queue() {
+            if let Err(ref e) = h_clone.lock().unwrap().process_queue() {
                 error!(
                     "Failed to handle block IO {}",
                     error_chain::ChainedError::display_chain(e)
@@ -641,18 +659,36 @@ impl EventNotifierHelper for BlockIoHandler {
             }
             None
         });
-        notifiers.push(build_event_notifier(
-            locked_block_io.queue_evt.as_raw_fd(),
-            handler,
-        ));
+        notifiers.push(build_event_notifier(handler_raw.queue_evt.as_raw_fd(), h));
+
+        // Register timer event notifier for IO limits
+        if let Some(lb) = handler_raw.leak_bucket.as_ref() {
+            let h_clone = handler.clone();
+            let h: Box<NotifierCallback> = Box::new(move |_, fd: RawFd| {
+                read_fd(fd);
+
+                if let Some(lb) = h_clone.lock().unwrap().leak_bucket.as_mut() {
+                    lb.clear_timer();
+                }
+
+                if let Err(ref e) = h_clone.lock().unwrap().process_queue() {
+                    error!(
+                        "Failed to handle block IO {}",
+                        error_chain::ChainedError::display_chain(e)
+                    );
+                }
+                None
+            });
+            notifiers.push(build_event_notifier(lb.as_raw_fd(), h));
+        }
 
         // Register event notifier for aio.
-        let cloned_block_io = block_io.clone();
-        if let Some(ref aio) = locked_block_io.aio {
-            let handler: Box<NotifierCallback> = Box::new(move |_, fd: RawFd| {
+        if let Some(ref aio) = handler_raw.aio {
+            let h_clone = handler.clone();
+            let h: Box<NotifierCallback> = Box::new(move |_, fd: RawFd| {
                 read_fd(fd);
 
-                if let Some(aio) = &mut cloned_block_io.lock().unwrap().aio {
+                if let Some(aio) = &mut h_clone.lock().unwrap().aio {
                     if let Err(ref e) = aio.handle() {
                         error!(
                             "Failed to handle aio, {}",
@@ -662,7 +698,7 @@ impl EventNotifierHelper for BlockIoHandler {
                 }
                 None
             });
-            notifiers.push(build_event_notifier(aio.fd.as_raw_fd(), handler));
+            notifiers.push(build_event_notifier(aio.fd.as_raw_fd(), h));
         }
 
         notifiers
@@ -728,6 +764,16 @@ impl Block {
 impl VirtioDevice for Block {
     /// Realize vhost virtio network device.
     fn realize(&mut self) -> Result<()> {
+        // if iothread not found, return err
+        if self.blk_cfg.iothread.is_some()
+            && EventLoop::get_ctx(self.blk_cfg.iothread.as_ref()).is_none()
+        {
+            bail!(
+                "IOThread {:?} of Block is not configured in params.",
+                self.blk_cfg.iothread,
+            );
+        }
+
         self.device_features = (1_u64 << VIRTIO_F_VERSION_1) | (1_u64 << VIRTIO_BLK_F_FLUSH);
         if self.blk_cfg.read_only {
             self.device_features |= 1_u64 << VIRTIO_BLK_F_RO;
@@ -879,6 +925,8 @@ impl VirtioDevice for Block {
             receiver,
             update_evt: self.update_evt.as_raw_fd(),
             interrupt_cb: cb,
+            iothread: self.blk_cfg.iothread.clone(),
+            leak_bucket: self.blk_cfg.iops.map(LeakBucket::new),
         };
 
         handler.aio = Some(handler.build_aio()?);
diff --git a/device_model/src/virtio/queue.rs b/device_model/src/virtio/queue.rs
index 2b10d91..396cb15 100644
--- a/device_model/src/virtio/queue.rs
+++ b/device_model/src/virtio/queue.rs
@@ -141,6 +141,9 @@ pub trait VringOps {
     /// * `features` - Bit mask of features negotiated by the backend and the frontend.
     fn pop_avail(&mut self, sys_mem: &Arc<AddressSpace>, features: u64) -> Result<Element>;
 
+    /// Rollback the entry which is pop from available queue by `pop_avail`.
+    fn push_back(&mut self);
+
     /// Fill the used vring after processing the IO request.
     ///
     /// # Arguments
@@ -763,6 +766,10 @@ impl VringOps for SplitVring {
         }
     }
 
+    fn push_back(&mut self) {
+        self.next_avail -= Wrapping(1);
+    }
+
     fn add_used(&mut self, sys_mem: &Arc<AddressSpace>, index: u16, len: u32) -> Result<()> {
         if index >= self.size {
             return Err(ErrorKind::QueueIndex(index, self.size).into());
diff --git a/docs/config_guidebook.md b/docs/config_guidebook.md
index 989dde5..bc0f5d5 100644
--- a/docs/config_guidebook.md
+++ b/docs/config_guidebook.md
@@ -191,7 +191,7 @@ There is only one argument for iothread:
 
 Virtio block device is a virtual block device, which process read and write requests in virtio queue from guest.
 
-Six properties are supported for virtio block device.
+Seven properties are supported for virtio block device.
 
 * drive_id: unique device-id in StratoVirt
 * path_on_host: the path of block device in host
@@ -199,13 +199,14 @@ Six properties are supported for virtio block device.
 * read_only: whether virtio block device is read-only or not
 * direct: open block device with `O_DIRECT` mode or not
 * iothread: indicate which iothread will be used, if not specified the main thread will be used
+* iops: used to limit IO operations for block device
 
 If you want to boot VM with a virtio block device as rootfs, you should add `root=DEVICE_NAME_IN_GUESTOS`
  in Kernel Parameters. `DEVICE_NAME_IN_GUESTOS` will from `vda` to `vdz` in order.
 
 ```shell
 # cmdline
--drive id=drive_id,file=path_on_host,serial=serial_num,readonly=off,direct=off[,iothread=iothread1]
+-drive id=drive_id,file=path_on_host,serial=serial_num,readonly=off,direct=off[,iothread=iothread1,iops=200]
 
 # json
 {
@@ -217,7 +218,8 @@ If you want to boot VM with a virtio block device as rootfs, you should add `roo
             "serial_num": "11111111",
             "direct": false,
             "read_only": false,
-            "iothread": "iothread1"
+            "iothread": "iothread1",
+            "iops": 200
         }
     ],
     ...
diff --git a/machine_manager/src/cmdline.rs b/machine_manager/src/cmdline.rs
index b0d53e7..a80bfc7 100644
--- a/machine_manager/src/cmdline.rs
+++ b/machine_manager/src/cmdline.rs
@@ -153,7 +153,7 @@ pub fn create_args_parser<'a>() -> ArgParser<'a> {
             Arg::with_name("drive")
                 .multiple(true)
                 .long("drive")
-                .value_name("file=path,id=str[,readonly=][,direct=][,serial=][,iothread=]")
+                .value_name("file=path,id=str[,readonly=][,direct=][,serial=][,iothread=][iops=]")
                 .help("use 'file' as a drive image")
                 .takes_values(true),
         )
diff --git a/machine_manager/src/config/fs.rs b/machine_manager/src/config/fs.rs
index c0e40b3..5ca843e 100644
--- a/machine_manager/src/config/fs.rs
+++ b/machine_manager/src/config/fs.rs
@@ -21,6 +21,7 @@ use crate::config::{CmdParser, ConfigCheck, ExBool, VmConfig};
 const MAX_STRING_LENGTH: usize = 255;
 const MAX_PATH_LENGTH: usize = 4096;
 const MAX_SERIAL_NUM: usize = 20;
+const MAX_IOPS: u64 = 1000000;
 
 /// Config struct for `drive`.
 /// Contains block device's attr.
@@ -33,6 +34,7 @@ pub struct DriveConfig {
     pub direct: bool,
     pub serial_num: Option<String>,
     pub iothread: Option<String>,
+    pub iops: Option<u64>,
 }
 
 impl DriveConfig {
@@ -56,6 +58,7 @@ impl Default for DriveConfig {
             direct: true,
             serial_num: None,
             iothread: None,
+            iops: None,
         }
     }
 }
@@ -94,6 +97,17 @@ impl ConfigCheck for DriveConfig {
             .into());
         }
 
+        if self.iops.is_some() && self.iops.unwrap() > MAX_IOPS {
+            return Err(ErrorKind::IllegalValue(
+                "iops of block device".to_string(),
+                0,
+                true,
+                MAX_IOPS,
+                true,
+            )
+            .into());
+        }
+
         Ok(())
     }
 }
@@ -126,7 +140,8 @@ impl VmConfig {
             .push("readonly")
             .push("direct")
             .push("serial")
-            .push("iothread");
+            .push("iothread")
+            .push("iops");
 
         cmd_parser.parse(drive_config)?;
 
@@ -149,6 +164,7 @@ impl VmConfig {
         }
         drive.serial_num = cmd_parser.get_value::<String>("serial")?;
         drive.iothread = cmd_parser.get_value::<String>("iothread")?;
+        drive.iops = cmd_parser.get_value::<u64>("iops")?;
 
         self.add_drive(drive)
     }
diff --git a/machine_manager/src/event_loop.rs b/machine_manager/src/event_loop.rs
index 03f55f7..9715422 100644
--- a/machine_manager/src/event_loop.rs
+++ b/machine_manager/src/event_loop.rs
@@ -54,20 +54,20 @@ impl EventLoop {
                     main_loop: EventLoopContext::new(),
                     io_threads,
                 });
-            }
 
-            if let Some(event_loop) = GLOBAL_EVENT_LOOP.as_mut() {
-                for (id, ctx) in &mut event_loop.io_threads {
-                    thread::Builder::new().name(id.to_string()).spawn(move || {
-                        while let Ok(ret) = ctx.run() {
-                            if !ret {
-                                break;
+                if let Some(event_loop) = GLOBAL_EVENT_LOOP.as_mut() {
+                    for (id, ctx) in &mut event_loop.io_threads {
+                        thread::Builder::new().name(id.to_string()).spawn(move || {
+                            while let Ok(ret) = ctx.run() {
+                                if !ret {
+                                    break;
+                                }
                             }
-                        }
-                    })?;
+                        })?;
+                    }
+                } else {
+                    bail!("Global Event Loop have not been initialized.")
                 }
-            } else {
-                bail!("Global Event Loop have not been initialized.")
             }
         }
 
@@ -79,16 +79,14 @@ impl EventLoop {
     /// # Arguments
     ///
     /// * `name` - if None, return main loop, OR return io-thread-loop which is related to `name`.
-    pub fn get_ctx(name: Option<&String>) -> &mut EventLoopContext {
+    pub fn get_ctx(name: Option<&String>) -> Option<&mut EventLoopContext> {
         unsafe {
             if let Some(event_loop) = GLOBAL_EVENT_LOOP.as_mut() {
                 if let Some(name) = name {
-                    if let Some(ctx) = event_loop.io_threads.get_mut(name) {
-                        return ctx;
-                    }
+                    return event_loop.io_threads.get_mut(name);
                 }
 
-                return &mut event_loop.main_loop;
+                return Some(&mut event_loop.main_loop);
             }
         }
 
@@ -102,7 +100,9 @@ impl EventLoop {
     /// * `manager` - The main part to manager the event loop specified by name.
     /// * `name` - specify which event loop to manage
     pub fn set_manager(manager: Arc<dyn EventLoopManager>, name: Option<&String>) {
-        Self::get_ctx(name).set_manager(manager)
+        if let Some(ctx) = Self::get_ctx(name) {
+            ctx.set_manager(manager)
+        }
     }
 
     /// Update event notifiers to  event loop
@@ -115,7 +115,11 @@ impl EventLoop {
         notifiers: Vec<EventNotifier>,
         name: Option<&String>,
     ) -> util::errors::Result<()> {
-        Self::get_ctx(name).update_events(notifiers)
+        if let Some(ctx) = Self::get_ctx(name) {
+            ctx.update_events(notifiers)
+        } else {
+            bail!("Loop Context not found in EventLoop.")
+        }
     }
 
     /// Start to run main loop
diff --git a/util/src/leak_bucket.rs b/util/src/leak_bucket.rs
new file mode 100644
index 0000000..5792a38
--- /dev/null
+++ b/util/src/leak_bucket.rs
@@ -0,0 +1,116 @@
+// Copyright (c) 2020 Huawei Technologies Co.,Ltd. All rights reserved.
+//
+// StratoVirt is licensed under Mulan PSL v2.
+// You can use this software according to the terms and conditions of the Mulan
+// PSL v2.
+// You may obtain a copy of Mulan PSL v2 at:
+//         http://license.coscl.org.cn/MulanPSL2
+// THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY
+// KIND, EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO
+// NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+// See the Mulan PSL v2 for more details.
+
+/// We use Leaky Bucket Algorithm to limit iops of block device and qmp.
+use std::os::unix::io::{AsRawFd, RawFd};
+use std::time::Instant;
+
+use vmm_sys_util::eventfd::EventFd;
+
+use crate::loop_context::EventLoopContext;
+
+/// Used to improve the accuracy of bucket level.
+const ACCURACY_SCALE: u64 = 1000;
+/// Nanoseconds per second.
+const NANOS_PER_SEC: u64 = 1000000000;
+
+/// Structure used to describe a Leaky Bucket.
+pub struct LeakBucket {
+    /// Indicate the capacity of bucket, which is config by user.
+    capacity: u64,
+    /// Current water level.
+    level: u64,
+    /// Internal used to calculate the delay of timer.
+    prev_time: Instant,
+    /// Indicate whether the timer started.
+    timer_started: bool,
+    /// When bucket is ready for allowing more IO operation, the internal callback will write this FD.
+    /// This FD should be listened by IO thread.
+    timer_wakeup: EventFd,
+}
+
+impl LeakBucket {
+    /// Construct function
+    ///
+    /// # Arguments
+    ///
+    /// * `units_ps` - iops value, eg. 100
+    pub fn new(units_ps: u64) -> Self {
+        LeakBucket {
+            capacity: units_ps * ACCURACY_SCALE,
+            level: 0,
+            prev_time: Instant::now(),
+            timer_started: false,
+            timer_wakeup: EventFd::new(libc::EFD_NONBLOCK).unwrap(),
+        }
+    }
+
+    /// Return true if the bucket is full, and caller must return directly instead of launching IO.
+    /// Otherwise, caller should not be affected.
+    ///
+    /// # Arguments
+    ///
+    /// * `loop_context` - used for delay function call.
+    pub fn throttled(&mut self, loop_context: &mut EventLoopContext) -> bool {
+        // capacity value is zero, indicating that there is no need to limit
+        if self.capacity == 0 {
+            return false;
+        }
+        if self.timer_started {
+            return true;
+        }
+
+        // update the water level
+        let now = Instant::now();
+        let nanos = (now - self.prev_time).as_nanos();
+        if nanos > (self.level * NANOS_PER_SEC / self.capacity) as u128 {
+            self.level = 0;
+        } else {
+            self.level -= nanos as u64 * self.capacity / NANOS_PER_SEC;
+        }
+
+        self.prev_time = now;
+
+        // need to be throttled
+        if self.level > self.capacity {
+            let wakeup_clone = self.timer_wakeup.try_clone().unwrap();
+            let func = Box::new(move || {
+                wakeup_clone
+                    .write(1)
+                    .unwrap_or_else(|e| error!("LeakBucket send event to device failed {}", e));
+            });
+
+            loop_context.delay_call(
+                func,
+                (self.level - self.capacity) * NANOS_PER_SEC / self.capacity,
+            );
+
+            self.timer_started = true;
+
+            return true;
+        }
+
+        self.level += ACCURACY_SCALE;
+
+        false
+    }
+
+    /// Clear the timer state.
+    pub fn clear_timer(&mut self) {
+        self.timer_started = false;
+    }
+
+    /// Get raw fd of wakeup event.
+    pub fn as_raw_fd(&self) -> RawFd {
+        self.timer_wakeup.as_raw_fd()
+    }
+}
diff --git a/util/src/lib.rs b/util/src/lib.rs
index ad46cc0..17899e6 100644
--- a/util/src/lib.rs
+++ b/util/src/lib.rs
@@ -27,6 +27,7 @@ pub mod checksum;
 pub mod daemonize;
 #[cfg(target_arch = "aarch64")]
 pub mod device_tree;
+pub mod leak_bucket;
 mod link_list;
 pub mod loop_context;
 pub mod num_ops;
diff --git a/util/src/loop_context.rs b/util/src/loop_context.rs
index 0f637aa..ea410e2 100644
--- a/util/src/loop_context.rs
+++ b/util/src/loop_context.rs
@@ -15,6 +15,7 @@ extern crate vmm_sys_util;
 use std::collections::BTreeMap;
 use std::os::unix::io::RawFd;
 use std::sync::{Arc, Mutex, RwLock};
+use std::time::{Duration, Instant};
 
 use libc::{c_void, read};
 use vmm_sys_util::epoll::{ControlOperation, Epoll, EpollEvent, EventSet};
@@ -101,6 +102,29 @@ pub trait EventLoopManager: Send + Sync {
     fn loop_cleanup(&self) -> Result<()>;
 }
 
+/// Timer structure is used for delay function execution.
+struct Timer {
+    /// Given the function that will be called.
+    func: Box<dyn Fn()>,
+    /// Given the real time when the `func` will be called.
+    expire_time: Instant,
+}
+
+impl Timer {
+    /// Construct function
+    ///
+    /// # Arguments
+    ///
+    /// * `func` - the function will be called later.
+    /// * `nsec` - delay time in nanosecond.
+    pub fn new(func: Box<dyn Fn()>, nsec: u64) -> Self {
+        Timer {
+            func,
+            expire_time: Instant::now() + Duration::new(0, nsec as u32),
+        }
+    }
+}
+
 /// Epoll Loop Context
 #[allow(clippy::vec_box)]
 pub struct EventLoopContext {
@@ -114,6 +138,8 @@ pub struct EventLoopContext {
     gc: Arc<RwLock<Vec<Box<EventNotifier>>>>,
     /// Temp events vector, store wait returned events.
     ready_events: Vec<EpollEvent>,
+    /// Timer list
+    timers: Vec<Timer>,
 }
 
 unsafe impl Sync for EventLoopContext {}
@@ -128,6 +154,7 @@ impl EventLoopContext {
             events: Arc::new(RwLock::new(BTreeMap::new())),
             gc: Arc::new(RwLock::new(Vec::new())),
             ready_events: vec![EpollEvent::default(); READY_EVENT_MAX],
+            timers: Vec::new(),
         }
     }
 
@@ -259,10 +286,11 @@ impl EventLoopContext {
             }
         }
 
-        let ev_count = match self
-            .epoll
-            .wait(READY_EVENT_MAX, -1, &mut self.ready_events[..])
-        {
+        let ev_count = match self.epoll.wait(
+            READY_EVENT_MAX,
+            self.timers_min_timeout(),
+            &mut self.ready_events[..],
+        ) {
             Ok(ev_count) => ev_count,
             Err(e) if e.raw_os_error() == Some(libc::EINTR) => 0,
             Err(e) => return Err(ErrorKind::EpollWait(e).into()),
@@ -289,10 +317,68 @@ impl EventLoopContext {
             }
         }
 
+        self.run_timers();
+
         self.clear_gc();
 
         Ok(true)
     }
+
+    /// Call the function given by `func` after `nsec` nanoseconds.
+    ///
+    /// # Arguments
+    ///
+    /// * `func` - the function will be called later.
+    /// * `nsec` - delay time in nanoseconds.
+    pub fn delay_call(&mut self, func: Box<dyn Fn()>, nsec: u64) {
+        let timer = Timer::new(func, nsec);
+
+        // insert in order of expire_time
+        let mut index = self.timers.len();
+        for (i, t) in self.timers.iter().enumerate() {
+            if timer.expire_time < t.expire_time {
+                index = i;
+                break;
+            }
+        }
+        self.timers.insert(index, timer);
+    }
+
+    /// Get the expire_time of the soonest Timer, and then translate it to timeout.
+    fn timers_min_timeout(&self) -> i32 {
+        if self.timers.is_empty() {
+            return -1;
+        }
+
+        let now = Instant::now();
+        if self.timers[0].expire_time <= now {
+            return 0;
+        }
+
+        let timeout = (self.timers[0].expire_time - now).as_millis();
+        if timeout >= i32::MAX as u128 {
+            i32::MAX - 1
+        } else {
+            timeout as i32
+        }
+    }
+
+    /// Call function of the timers which have already expired.
+    fn run_timers(&mut self) {
+        let now = Instant::now();
+        let mut expired_nr = 0;
+
+        for timer in &self.timers {
+            if timer.expire_time > now {
+                break;
+            }
+
+            expired_nr += 1;
+            (timer.func)();
+        }
+
+        self.timers.drain(0..expired_nr);
+    }
 }
 
 impl Default for EventLoopContext {
-- 
2.27.0

