leveldb学习：Env

考虑到移植以及灵活性，leveldb将系统相关的处理（文件/进程/时间）抽象成Evn，用户可以自己实现相应的接口，作为option传入，默认使用自带的实现。

解压目录/util/中放有env的声明和实现代码。env.h中声明了:

虚基类env，在env_posix.cc中，派生类PosixEnv继承自env类，是leveldb的默认实现。

虚基类WritableFile、SequentialFile、RandomAccessFile，分别是文件的写抽象类，顺序读抽象类和随机读抽象类

类Logger，log文件的写入接口，log文件是防止系统异常终止造成数据丢失，是memtable在磁盘的备份

类FileLock，为文件上锁

WriteStringToFile、ReadFileToString、Log三个全局函数，封装了上述接口

下面来看看env_posix.cc中为我们写好的默认实现

顺序读：

class PosixSequentialFile: public SequentialFile {
private:
std::string filename_;
FILE* file_;
public:
PosixSequentialFile(const std::string& fname, FILE* f)
: filename_(fname), file_(f) { }
virtual ~PosixSequentialFile() { fclose(file_); }
virtual Status Read(size_t n, Slice* result, char* scratch) {
Status s;
size_t r = fread_unlocked(scratch, 1, n, file_);
*result = Slice(scratch, r);
if (r < n) {
if (feof(file_)) {
// We leave status as ok if we hit the end of the file
} else {
// A partial read with an error: return a non-ok status
s = IOError(filename_, errno);
}
}
return s;
}
virtual Status Skip(uint64_t n) {
if (fseek(file_, n, SEEK_CUR)) {
return IOError(filename_, errno);
}
return Status::OK();
}
};

这就是leveldb从磁盘读取文件的接口了，用的是C的流文件操作和FILE结构体。

随机读：

class PosixRandomAccessFile: public RandomAccessFile {
private:
std::string filename_;
int fd_;
public:
PosixRandomAccessFile(const std::string& fname, int fd)
: filename_(fname), fd_(fd) { }
virtual ~PosixRandomAccessFile() { close(fd_); }
virtual Status Read(uint64_t offset, size_t n, Slice* result,
char* scratch) const {
Status s;
ssize_t r = pread(fd_, scratch, n, static_cast<off_t>(offset));
*result = Slice(scratch, (r < 0) ? 0 : r);
if (r < 0) {
// An error: return a non-ok status
s = IOError(filename_, errno);
}
return s;
}
};

随机读取磁盘中文件的数据，可以定位读取文件中offset偏移量的数据，核心函数pread()函数，可以带偏移量地原子的从文件中读取数据，offset:读取的

其实地址偏移量，读取地址=文件地址+offset。

写入：

class PosixWritableFile : public WritableFile {
private:
std::string filename_;
FILE* file_;
public:
PosixWritableFile(const std::string& fname, FILE* f)
: filename_(fname), file_(f) { }
~PosixWritableFile() {
if (file_ != NULL) {
// Ignoring any potential errors
fclose(file_);
}
}
virtual Status Append(const Slice& data) {
size_t r = fwrite_unlocked(data.data(), 1, data.size(), file_);
if (r != data.size()) {
return IOError(filename_, errno);
}
return Status::OK();
}
virtual Status Close() {
Status result;
if (fclose(file_) != 0) {
result = IOError(filename_, errno);
}
file_ = NULL;
return result;
}
virtual Status Flush() {
if (fflush_unlocked(file_) != 0) {
return IOError(filename_, errno);
}
return Status::OK();
}
Status SyncDirIfManifest() {
const char* f = filename_.c_str();
const char* sep = strrchr(f, '/');
Slice basename;
std::string dir;
if (sep == NULL) {
dir = ".";
basename = f;
} else {
dir = std::string(f, sep - f);
basename = sep + 1;
}
Status s;
if (basename.starts_with("MANIFEST")) {
int fd = open(dir.c_str(), O_RDONLY);
if (fd < 0) {
s = IOError(dir, errno);
} else {
if (fsync(fd) < 0) {
s = IOError(dir, errno);
}
close(fd);
}
}
return s;
}
virtual Status Sync() {
// Ensure new files referred to by the manifest are in the filesystem.
Status s = SyncDirIfManifest();
if (!s.ok()) {
return s;
}
if (fflush_unlocked(file_) != 0 ||
fdatasync(fileno(file_)) != 0) {
s = Status::IOError(filename_, strerror(errno));
}
return s;
}
};

sync( )函数是同步缓存和磁盘中的数据，核心函数fflush（清空输出缓冲区，并把缓冲区内容输出），在比较关键的写入操作时，立即同步可以防止系统掉电时数据丢失。

定义完这三个文件的读写抽象类，把他们加入到PosixEnv类中，定义三个NewSequentialFile、NewRandomAccessFile、NewWritableFile函数，产生文件读写的对象，在程序上层中调用env_->NewWritableFile即可创建一个文件，并可写入数据。

文件的上锁：

static int LockOrUnlock(int fd, bool lock) {
errno = 0;
struct flock f;
memset(&f, 0, sizeof(f));
f.l_type = (lock ? F_WRLCK : F_UNLCK);
f.l_whence = SEEK_SET;
f.l_start = 0;
f.l_len = 0;        // Lock/unlock entire file
return fcntl(fd, F_SETLK, &f);
}

fd是文件的i节点（linux每个文件和文件夹都对应一个唯一的i节点，用于管理文件）。核心调用函数fcntl，F_WRLK和F_UNLK是命令的参数，上锁和解锁。在PosixEnv的成员函数LockFile和UnlockFile都调用了此函数，此外还将上锁的文件保存在了PosixEnv::PosixLockTable locks_成员变量中，PosixLockTable是一个类，内有成员std::set locked_files_保存上锁的文件名，定义了insert和remove操作，然而并没看出用处o(╯□╰)o？

PosixEnv还有一个很重要的功能，计划任务，也就是后台的compact进程，compact目的是维护数据库的均衡性，保持数据库查找的高效率。PosixEnv中定义了一个任务队列：

struct BGItem { void* arg; void (*function)(void*); };
//用的是deque双向链表作为底层的数据结构，按时间顺序执行任务，没有设立优先级队列
typedef std::deque<BGItem> BGQueue;
BGQueue queue_;

主进程一旦判定需要进行compact操作，就把compact任务压进队列queue_中，BGItem是存有任务函数和db对象指针的结构。而后台进程从一开始就不断根据队列中的函数指针执行compact任务。BGThread()函数就是不停的在queue_中取出函数指针，执行。

void PosixEnv::Schedule(void (*function)(void*), void* arg) {
PthreadCall("lock", pthread_mutex_lock(&mu_));
// Start background thread if necessary
//开启后台进程，如果没有开启
if (!started_bgthread_) {
started_bgthread_ = true;
PthreadCall(
"create thread",
pthread_create(&bgthread_, NULL,  &PosixEnv::BGThreadWrapper, this));
}
// If the queue is currently empty, the background thread may currently be
// waiting.
if (queue_.empty()) {
PthreadCall("signal", pthread_cond_signal(&bgsignal_));
}
// Add to priority queue
//将任务压入队列
queue_.push_back(BGItem());
queue_.back().function = function;
queue_.back().arg = arg;
PthreadCall("unlock", pthread_mutex_unlock(&mu_));
}
void PosixEnv::BGThread() {
while (true) {
// Wait until there is an item that is ready to run
PthreadCall("lock", pthread_mutex_lock(&mu_));
while (queue_.empty()) {
PthreadCall("wait", pthread_cond_wait(&bgsignal_, &mu_));
}
//从队列取出函数，执行
void (*function)(void*) = queue_.front().function;
void* arg = queue_.front().arg;
queue_.pop_front();
PthreadCall("unlock", pthread_mutex_unlock(&mu_));
(*function)(arg);
}
}

后台进程一直执行queue_中的任务，由于queue_是动态的，自然需要考虑queue_空了怎么办，leveldb采用的是条件量pthread_cond_t bgsignal_，队列空了就进入等待，直至主进程有新的任务加入进来，而条件变量一般是要和pthread_mutex_t mu_搭配使用，防止某些逻辑错误。核心函数有：

pthread_mutex_init(),pthread_cond_init()条件变量、互斥锁的初始化

pthread_mutex_lock(),pthread_mutex_unlock()关锁解锁

pthread_cond_signal(),pthread_cond_wait()条件变量成立，等待条件成立

pthread_create创建子线程

此外PosixEnv中还有FileExists、GetChildren、DeleteFile、CreateDir、DeleteDir、GetFileSize、RenameFile等等函数，它们的作用就如它们的名字所说一样。

好了，从env给我的收获就是：

利用虚基类的特性提供了默认的实现，也开放了用户自定义操作的权限

面向对象编程范式的学习，把一切操作定义成类

文件的加锁解锁，线程的同步

C的文件流操作，对文件名的字符提取操作，创建、删除文件和路径，这些都可以直接用到将来自己的项目中