build on windows system.

.github/workflows/rust.yml

@@ -10,7 +10,7 @@ jobs:
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        os: [ ubuntu-latest, macos-latest ]
+        os: [ ubuntu-latest, macos-latest, windows-latest ]
     steps:
       - uses: actions/checkout@v2
         with:

Cargo.toml

@@ -68,7 +68,7 @@ kvproto = { git = "https://github.com/pingcap/kvproto.git", default-features = f
 raft = { git = "https://github.com/tikv/raft-rs", branch = "master", default-features = false, features = ["protobuf-codec"] }
 rand = "0.8"
 rand_distr = "0.4"
-tempfile = "3.1"
+tempfile = "3.6"
 toml = "0.7"
 
 [features]
@@ -87,13 +87,16 @@ swap = [
   "nightly",
   "memmap2",
 ]
+std_fs = []
 
 nightly_group = ["nightly", "swap"]
 
 [patch.crates-io]
 raft-proto = { git = "https://github.com/tikv/raft-rs", branch = "master" }
 protobuf = { git = "https://github.com/pingcap/rust-protobuf", branch = "v2.8" }
 protobuf-codegen = { git = "https://github.com/pingcap/rust-protobuf", branch = "v2.8" }
+# TODO: Use official grpc-rs once https://github.com/tikv/grpc-rs/pull/622 is merged.
+grpcio = { git = "https://github.com/tabokie/grpc-rs", branch = "v0.10.x-win" }
 
 [workspace]
 members = ["stress", "ctl"]

Makefile

@@ -67,6 +67,8 @@ else
 test_matrix: test
 	cargo ${TOOLCHAIN_ARGS} test --all ${EXTRA_CARGO_ARGS} -- --nocapture
 	cargo ${TOOLCHAIN_ARGS} test --test failpoints --features failpoints ${EXTRA_CARGO_ARGS} -- --test-threads 1 --nocapture
+	cargo ${TOOLCHAIN_ARGS} test --all --features nightly_group,std_fs ${EXTRA_CARGO_ARGS} -- --nocapture
+	cargo ${TOOLCHAIN_ARGS} test --test failpoints --features nightly_group,std_fs,failpoints ${EXTRA_CARGO_ARGS} -- --test-threads 1 --nocapture
 endif
 
 ## Build raft-engine-ctl.

src/env/default.rs

@@ -1,212 +1,14 @@
 // Copyright (c) 2017-present, PingCAP, Inc. Licensed under Apache-2.0.
 
-use std::io::{Read, Result as IoResult, Seek, SeekFrom, Write};
-use std::os::unix::io::RawFd;
+use std::io::{Error, ErrorKind, Read, Result as IoResult, Seek, SeekFrom, Write};
 use std::path::Path;
 use std::sync::Arc;
 
 use fail::fail_point;
-use log::error;
-use nix::errno::Errno;
-use nix::fcntl::{self, OFlag};
-use nix::sys::stat::Mode;
-use nix::sys::uio::{pread, pwrite};
-use nix::unistd::{close, ftruncate, lseek, Whence};
-use nix::NixPath;
 
+use crate::env::log_fd::LogFd;
 use crate::env::{FileSystem, Handle, Permission, WriteExt};
 
-fn from_nix_error(e: nix::Error, custom: &'static str) -> std::io::Error {
-    let kind = std::io::Error::from(e).kind();
-    std::io::Error::new(kind, custom)
-}
-
-impl From<Permission> for OFlag {
-    fn from(value: Permission) -> OFlag {
-        match value {
-            Permission::ReadOnly => OFlag::O_RDONLY,
-            Permission::ReadWrite => OFlag::O_RDWR,
-        }
-    }
-}
-
-/// A RAII-style low-level file. Errors occurred during automatic resource
-/// release are logged and ignored.
-///
-/// A [`LogFd`] is essentially a thin wrapper around [`RawFd`]. It's only
-/// supported on *Unix*, and primarily optimized for *Linux*.
-///
-/// All [`LogFd`] instances are opened with read and write permission.
-pub struct LogFd(RawFd);
-
-impl LogFd {
-    /// Opens a file with the given `path`.
-    pub fn open<P: ?Sized + NixPath>(path: &P, perm: Permission) -> IoResult<Self> {
-        fail_point!("log_fd::open::err", |_| {
-            Err(from_nix_error(nix::Error::EINVAL, "fp"))
-        });
-        // Permission 644
-        let mode = Mode::S_IRUSR | Mode::S_IWUSR | Mode::S_IRGRP | Mode::S_IROTH;
-        fail_point!("log_fd::open::fadvise_dontneed", |_| {
-            let fd =
-                LogFd(fcntl::open(path, perm.into(), mode).map_err(|e| from_nix_error(e, "open"))?);
-            #[cfg(target_os = "linux")]
-            unsafe {
-                extern crate libc;
-                libc::posix_fadvise64(fd.0, 0, fd.file_size()? as i64, libc::POSIX_FADV_DONTNEED);
-            }
-            Ok(fd)
-        });
-        Ok(LogFd(
-            fcntl::open(path, perm.into(), mode).map_err(|e| from_nix_error(e, "open"))?,
-        ))
-    }
-
-    /// Opens a file with the given `path`. The specified file will be created
-    /// first if not exists.
-    pub fn create<P: ?Sized + NixPath>(path: &P) -> IoResult<Self> {
-        fail_point!("log_fd::create::err", |_| {
-            Err(from_nix_error(nix::Error::EINVAL, "fp"))
-        });
-        let flags = OFlag::O_RDWR | OFlag::O_CREAT;
-        // Permission 644
-        let mode = Mode::S_IRUSR | Mode::S_IWUSR | Mode::S_IRGRP | Mode::S_IROTH;
-        let fd = fcntl::open(path, flags, mode).map_err(|e| from_nix_error(e, "open"))?;
-        Ok(LogFd(fd))
-    }
-
-    /// Closes the file.
-    pub fn close(&self) -> IoResult<()> {
-        fail_point!("log_fd::close::err", |_| {
-            Err(from_nix_error(nix::Error::EINVAL, "fp"))
-        });
-        close(self.0).map_err(|e| from_nix_error(e, "close"))
-    }
-
-    /// Reads some bytes starting at `offset` from this file into the specified
-    /// buffer. Returns how many bytes were read.
-    pub fn read(&self, mut offset: usize, buf: &mut [u8]) -> IoResult<usize> {
-        let mut readed = 0;
-        while readed < buf.len() {
-            fail_point!("log_fd::read::err", |_| {
-                Err(from_nix_error(nix::Error::EINVAL, "fp"))
-            });
-            let bytes = match pread(self.0, &mut buf[readed..], offset as i64) {
-                Ok(bytes) => bytes,
-                Err(e) if e == Errno::EINTR => continue,
-                Err(e) => return Err(from_nix_error(e, "pread")),
-            };
-            // EOF
-            if bytes == 0 {
-                break;
-            }
-            readed += bytes;
-            offset += bytes;
-        }
-        Ok(readed)
-    }
-
-    /// Writes some bytes to this file starting at `offset`. Returns how many
-    /// bytes were written.
-    pub fn write(&self, mut offset: usize, content: &[u8]) -> IoResult<usize> {
-        fail_point!("log_fd::write::zero", |_| { Ok(0) });
-        fail_point!("log_fd::write::no_space_err", |_| {
-            Err(from_nix_error(nix::Error::ENOSPC, "nospace"))
-        });
-        let mut written = 0;
-        while written < content.len() {
-            let bytes = match pwrite(self.0, &content[written..], offset as i64) {
-                Ok(bytes) => bytes,
-                Err(e) if e == Errno::EINTR => continue,
-                Err(e) if e == Errno::ENOSPC => return Err(from_nix_error(e, "nospace")),
-                Err(e) => return Err(from_nix_error(e, "pwrite")),
-            };
-            if bytes == 0 {
-                break;
-            }
-            written += bytes;
-            offset += bytes;
-        }
-        fail_point!("log_fd::write::err", |_| {
-            Err(from_nix_error(nix::Error::EINVAL, "fp"))
-        });
-        Ok(written)
-    }
-
-    /// Truncates all data after `offset`.
-    pub fn truncate(&self, offset: usize) -> IoResult<()> {
-        fail_point!("log_fd::truncate::err", |_| {
-            Err(from_nix_error(nix::Error::EINVAL, "fp"))
-        });
-        ftruncate(self.0, offset as i64).map_err(|e| from_nix_error(e, "ftruncate"))
-    }
-
-    /// Attempts to allocate space for `size` bytes starting at `offset`.
-    #[allow(unused_variables)]
-    pub fn allocate(&self, offset: usize, size: usize) -> IoResult<()> {
-        fail_point!("log_fd::allocate::err", |_| {
-            Err(from_nix_error(nix::Error::EINVAL, "fp"))
-        });
-        #[cfg(target_os = "linux")]
-        {
-            if let Err(e) = fcntl::fallocate(
-                self.0,
-                fcntl::FallocateFlags::empty(),
-                offset as i64,
-                size as i64,
-            ) {
-                if e != nix::Error::EOPNOTSUPP {
-                    return Err(from_nix_error(e, "fallocate"));
-                }
-            }
-        }
-        Ok(())
-    }
-}
-
-impl Handle for LogFd {
-    #[inline]
-    fn truncate(&self, offset: usize) -> IoResult<()> {
-        fail_point!("log_fd::truncate::err", |_| {
-            Err(from_nix_error(nix::Error::EINVAL, "fp"))
-        });
-        ftruncate(self.0, offset as i64).map_err(|e| from_nix_error(e, "ftruncate"))
-    }
-
-    #[inline]
-    fn file_size(&self) -> IoResult<usize> {
-        fail_point!("log_fd::file_size::err", |_| {
-            Err(from_nix_error(nix::Error::EINVAL, "fp"))
-        });
-        lseek(self.0, 0, Whence::SeekEnd)
-            .map(|n| n as usize)
-            .map_err(|e| from_nix_error(e, "lseek"))
-    }
-
-    #[inline]
-    fn sync(&self) -> IoResult<()> {
-        fail_point!("log_fd::sync::err", |_| {
-            Err(from_nix_error(nix::Error::EINVAL, "fp"))
-        });
-        #[cfg(target_os = "linux")]
-        {
-            nix::unistd::fdatasync(self.0).map_err(|e| from_nix_error(e, "fdatasync"))
-        }
-        #[cfg(not(target_os = "linux"))]
-        {
-            nix::unistd::fsync(self.0).map_err(|e| from_nix_error(e, "fsync"))
-        }
-    }
-}
-
-impl Drop for LogFd {
-    fn drop(&mut self) {
-        if let Err(e) = self.close() {
-            error!("error while closing file: {e}");
-        }
-    }
-}
-
 /// A low-level file adapted for standard interfaces including [`Seek`],
 /// [`Write`] and [`Read`].
 pub struct LogFile {
@@ -226,7 +28,14 @@ impl LogFile {
 
 impl Write for LogFile {
     fn write(&mut self, buf: &[u8]) -> IoResult<usize> {
+        fail_point!("log_file::write::zero", |_| { Ok(0) });
+
         let len = self.inner.write(self.offset, buf)?;
+
+        fail_point!("log_file::write::err", |_| {
+            Err(Error::new(ErrorKind::InvalidInput, "fp"))
+        });
+
         self.offset += len;
         Ok(len)
     }
@@ -238,6 +47,10 @@ impl Write for LogFile {
 
 impl Read for LogFile {
     fn read(&mut self, buf: &mut [u8]) -> IoResult<usize> {
+        fail_point!("log_file::read::err", |_| {
+            Err(Error::new(ErrorKind::InvalidInput, "fp"))
+        });
+
         let len = self.inner.read(self.offset, buf)?;
         self.offset += len;
         Ok(len)
@@ -260,12 +73,20 @@ impl Seek for LogFile {
 
 impl WriteExt for LogFile {
     fn truncate(&mut self, offset: usize) -> IoResult<()> {
+        fail_point!("log_file::truncate::err", |_| {
+            Err(Error::new(ErrorKind::InvalidInput, "fp"))
+        });
+
         self.inner.truncate(offset)?;
         self.offset = offset;
         Ok(())
     }
 
     fn allocate(&mut self, offset: usize, size: usize) -> IoResult<()> {
+        fail_point!("log_file::allocate::err", |_| {
+            Err(Error::new(ErrorKind::InvalidInput, "fp"))
+        });
+
         self.inner.allocate(offset, size)
     }
 }
@@ -278,10 +99,18 @@ impl FileSystem for DefaultFileSystem {
     type Writer = LogFile;
 
     fn create<P: AsRef<Path>>(&self, path: P) -> IoResult<Self::Handle> {
+        fail_point!("default_fs::create::err", |_| {
+            Err(Error::new(ErrorKind::InvalidInput, "fp"))
+        });
+
         LogFd::create(path.as_ref())
     }
 
     fn open<P: AsRef<Path>>(&self, path: P, perm: Permission) -> IoResult<Self::Handle> {
+        fail_point!("default_fs::open::err", |_| {
+            Err(Error::new(ErrorKind::InvalidInput, "fp"))
+        });
+
         LogFd::open(path.as_ref(), perm)
     }
 

src/env/log_fd.rs

@@ -0,0 +1,11 @@
+// Copyright (c) 2017-present, PingCAP, Inc. Licensed under Apache-2.0.
+
+#[cfg(not(any(windows, feature = "std_fs")))]
+mod unix;
+#[cfg(not(any(windows, feature = "std_fs")))]
+pub use unix::LogFd;
+
+#[cfg(any(windows, feature = "std_fs"))]
+mod plain;
+#[cfg(any(windows, feature = "std_fs"))]
+pub use plain::LogFd;