1 //  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
2 //  This source code is licensed under both the GPLv2 (found in the
3 //  COPYING file in the root directory) and Apache 2.0 License
4 //  (found in the LICENSE.Apache file in the root directory).
5 //
6 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
7 // Use of this source code is governed by a BSD-style license that can be
8 // found in the LICENSE file. See the AUTHORS file for names of contributors.
9 #pragma once
10 #include <errno.h>
11 #if defined(ROCKSDB_IOURING_PRESENT)
12 #include <liburing.h>
13 #include <sys/uio.h>
14 #endif
15 #include <unistd.h>
16 #include <atomic>
17 #include <functional>
18 #include <map>
19 #include <string>
20 #include "port/port.h"
21 #include "rocksdb/env.h"
22 #include "rocksdb/file_system.h"
23 #include "rocksdb/io_status.h"
24 #include "util/mutexlock.h"
25 #include "util/thread_local.h"
26 
27 // For non linux platform, the following macros are used only as place
28 // holder.
29 #if !(defined OS_LINUX) && !(defined CYGWIN) && !(defined OS_AIX)
30 #define POSIX_FADV_NORMAL 0     /* [MC1] no further special treatment */
31 #define POSIX_FADV_RANDOM 1     /* [MC1] expect random page refs */
32 #define POSIX_FADV_SEQUENTIAL 2 /* [MC1] expect sequential page refs */
33 #define POSIX_FADV_WILLNEED 3   /* [MC1] will need these pages */
34 #define POSIX_FADV_DONTNEED 4   /* [MC1] don't need these pages */
35 #endif
36 
37 namespace ROCKSDB_NAMESPACE {
38 std::string IOErrorMsg(const std::string& context,
39                        const std::string& file_name);
40 // file_name can be left empty if it is not unkown.
41 IOStatus IOError(const std::string& context, const std::string& file_name,
42                  int err_number);
43 
44 class PosixHelper {
45  public:
46   static size_t GetUniqueIdFromFile(int fd, char* id, size_t max_size);
47   static size_t GetLogicalBlockSizeOfFd(int fd);
48   static Status GetLogicalBlockSizeOfDirectory(const std::string& directory,
49                                                size_t* size);
50 };
51 
52 #ifdef OS_LINUX
53 // Files under a specific directory have the same logical block size.
54 // This class caches the logical block size for the specified directories to
55 // save the CPU cost of computing the size.
56 // Safe for concurrent access from multiple threads without any external
57 // synchronization.
58 class LogicalBlockSizeCache {
59  public:
60   LogicalBlockSizeCache(
61       std::function<size_t(int)> get_logical_block_size_of_fd =
62           PosixHelper::GetLogicalBlockSizeOfFd,
63       std::function<Status(const std::string&, size_t*)>
64           get_logical_block_size_of_directory =
65               PosixHelper::GetLogicalBlockSizeOfDirectory)
get_logical_block_size_of_fd_(get_logical_block_size_of_fd)66       : get_logical_block_size_of_fd_(get_logical_block_size_of_fd),
67         get_logical_block_size_of_directory_(
68             get_logical_block_size_of_directory) {}
69 
70   // Takes the following actions:
71   // 1. Increases reference count of the directories;
72   // 2. If the directory's logical block size is not cached,
73   //    compute the buffer size and cache the result.
74   Status RefAndCacheLogicalBlockSize(
75       const std::vector<std::string>& directories);
76 
77   // Takes the following actions:
78   // 1. Decreases reference count of the directories;
79   // 2. If the reference count of a directory reaches 0, remove the directory
80   //    from the cache.
81   void UnrefAndTryRemoveCachedLogicalBlockSize(
82       const std::vector<std::string>& directories);
83 
84   // Returns the logical block size for the file.
85   //
86   // If the file is under a cached directory, return the cached size.
87   // Otherwise, the size is computed.
88   size_t GetLogicalBlockSize(const std::string& fname, int fd);
89 
GetRefCount(const std::string & dir)90   int GetRefCount(const std::string& dir) {
91     ReadLock lock(&cache_mutex_);
92     auto it = cache_.find(dir);
93     if (it == cache_.end()) {
94       return 0;
95     }
96     return it->second.ref;
97   }
98 
Size()99   size_t Size() const { return cache_.size(); }
100 
Contains(const std::string & dir)101   bool Contains(const std::string& dir) {
102     ReadLock lock(&cache_mutex_);
103     return cache_.find(dir) != cache_.end();
104   }
105 
106  private:
107   struct CacheValue {
CacheValueCacheValue108     CacheValue() : size(0), ref(0) {}
109 
110     // Logical block size of the directory.
111     size_t size;
112     // Reference count of the directory.
113     int ref;
114   };
115 
116   std::function<size_t(int)> get_logical_block_size_of_fd_;
117   std::function<Status(const std::string&, size_t*)>
118       get_logical_block_size_of_directory_;
119 
120   std::map<std::string, CacheValue> cache_;
121   port::RWMutex cache_mutex_;
122 };
123 #endif
124 
125 class PosixSequentialFile : public FSSequentialFile {
126  private:
127   std::string filename_;
128   FILE* file_;
129   int fd_;
130   bool use_direct_io_;
131   size_t logical_sector_size_;
132 
133  public:
134   PosixSequentialFile(const std::string& fname, FILE* file, int fd,
135                       size_t logical_block_size,
136                       const EnvOptions& options);
137   virtual ~PosixSequentialFile();
138 
139   virtual IOStatus Read(size_t n, const IOOptions& opts, Slice* result,
140                         char* scratch, IODebugContext* dbg) override;
141   virtual IOStatus PositionedRead(uint64_t offset, size_t n,
142                                   const IOOptions& opts, Slice* result,
143                                   char* scratch, IODebugContext* dbg) override;
144   virtual IOStatus Skip(uint64_t n) override;
145   virtual IOStatus InvalidateCache(size_t offset, size_t length) override;
use_direct_io()146   virtual bool use_direct_io() const override { return use_direct_io_; }
GetRequiredBufferAlignment()147   virtual size_t GetRequiredBufferAlignment() const override {
148     return logical_sector_size_;
149   }
150 };
151 
152 #if defined(ROCKSDB_IOURING_PRESENT)
153 // io_uring instance queue depth
154 const unsigned int kIoUringDepth = 256;
155 
DeleteIOUring(void * p)156 inline void DeleteIOUring(void* p) {
157   struct io_uring* iu = static_cast<struct io_uring*>(p);
158   delete iu;
159 }
160 
CreateIOUring()161 inline struct io_uring* CreateIOUring() {
162   struct io_uring* new_io_uring = new struct io_uring;
163   int ret = io_uring_queue_init(kIoUringDepth, new_io_uring, 0);
164   if (ret) {
165     delete new_io_uring;
166     new_io_uring = nullptr;
167   }
168   return new_io_uring;
169 }
170 #endif  // defined(ROCKSDB_IOURING_PRESENT)
171 
172 class PosixRandomAccessFile : public FSRandomAccessFile {
173  protected:
174   std::string filename_;
175   int fd_;
176   bool use_direct_io_;
177   size_t logical_sector_size_;
178 #if defined(ROCKSDB_IOURING_PRESENT)
179   ThreadLocalPtr* thread_local_io_urings_;
180 #endif
181 
182  public:
183   PosixRandomAccessFile(const std::string& fname, int fd,
184                         size_t logical_block_size,
185                         const EnvOptions& options
186 #if defined(ROCKSDB_IOURING_PRESENT)
187                         ,
188                         ThreadLocalPtr* thread_local_io_urings
189 #endif
190   );
191   virtual ~PosixRandomAccessFile();
192 
193   virtual IOStatus Read(uint64_t offset, size_t n, const IOOptions& opts,
194                         Slice* result, char* scratch,
195                         IODebugContext* dbg) const override;
196 
197   virtual IOStatus MultiRead(FSReadRequest* reqs, size_t num_reqs,
198                              const IOOptions& options,
199                              IODebugContext* dbg) override;
200 
201   virtual IOStatus Prefetch(uint64_t offset, size_t n, const IOOptions& opts,
202                             IODebugContext* dbg) override;
203 
204 #if defined(OS_LINUX) || defined(OS_MACOSX) || defined(OS_AIX)
205   virtual size_t GetUniqueId(char* id, size_t max_size) const override;
206 #endif
207   virtual void Hint(AccessPattern pattern) override;
208   virtual IOStatus InvalidateCache(size_t offset, size_t length) override;
use_direct_io()209   virtual bool use_direct_io() const override { return use_direct_io_; }
GetRequiredBufferAlignment()210   virtual size_t GetRequiredBufferAlignment() const override {
211     return logical_sector_size_;
212   }
213 };
214 
215 class PosixWritableFile : public FSWritableFile {
216  protected:
217   const std::string filename_;
218   const bool use_direct_io_;
219   int fd_;
220   uint64_t filesize_;
221   size_t logical_sector_size_;
222 #ifdef ROCKSDB_FALLOCATE_PRESENT
223   bool allow_fallocate_;
224   bool fallocate_with_keep_size_;
225 #endif
226 #ifdef ROCKSDB_RANGESYNC_PRESENT
227   // Even if the syscall is present, the filesystem may still not properly
228   // support it, so we need to do a dynamic check too.
229   bool sync_file_range_supported_;
230 #endif  // ROCKSDB_RANGESYNC_PRESENT
231 
232  public:
233   explicit PosixWritableFile(const std::string& fname, int fd,
234                              size_t logical_block_size,
235                              const EnvOptions& options);
236   virtual ~PosixWritableFile();
237 
238   // Need to implement this so the file is truncated correctly
239   // with direct I/O
240   virtual IOStatus Truncate(uint64_t size, const IOOptions& opts,
241                             IODebugContext* dbg) override;
242   virtual IOStatus Close(const IOOptions& opts, IODebugContext* dbg) override;
243   virtual IOStatus Append(const Slice& data, const IOOptions& opts,
244                           IODebugContext* dbg) override;
245   virtual IOStatus PositionedAppend(const Slice& data, uint64_t offset,
246                                     const IOOptions& opts,
247                                     IODebugContext* dbg) override;
248   virtual IOStatus Flush(const IOOptions& opts, IODebugContext* dbg) override;
249   virtual IOStatus Sync(const IOOptions& opts, IODebugContext* dbg) override;
250   virtual IOStatus Fsync(const IOOptions& opts, IODebugContext* dbg) override;
251   virtual bool IsSyncThreadSafe() const override;
use_direct_io()252   virtual bool use_direct_io() const override { return use_direct_io_; }
253   virtual void SetWriteLifeTimeHint(Env::WriteLifeTimeHint hint) override;
254   virtual uint64_t GetFileSize(const IOOptions& opts,
255                                IODebugContext* dbg) override;
256   virtual IOStatus InvalidateCache(size_t offset, size_t length) override;
GetRequiredBufferAlignment()257   virtual size_t GetRequiredBufferAlignment() const override {
258     return logical_sector_size_;
259   }
260 #ifdef ROCKSDB_FALLOCATE_PRESENT
261   virtual IOStatus Allocate(uint64_t offset, uint64_t len,
262                             const IOOptions& opts,
263                             IODebugContext* dbg) override;
264 #endif
265   virtual IOStatus RangeSync(uint64_t offset, uint64_t nbytes,
266                              const IOOptions& opts,
267                              IODebugContext* dbg) override;
268 #ifdef OS_LINUX
269   virtual size_t GetUniqueId(char* id, size_t max_size) const override;
270 #endif
271 };
272 
273 // mmap() based random-access
274 class PosixMmapReadableFile : public FSRandomAccessFile {
275  private:
276   int fd_;
277   std::string filename_;
278   void* mmapped_region_;
279   size_t length_;
280 
281  public:
282   PosixMmapReadableFile(const int fd, const std::string& fname, void* base,
283                         size_t length, const EnvOptions& options);
284   virtual ~PosixMmapReadableFile();
285   virtual IOStatus Read(uint64_t offset, size_t n, const IOOptions& opts,
286                         Slice* result, char* scratch,
287                         IODebugContext* dbg) const override;
288   virtual IOStatus InvalidateCache(size_t offset, size_t length) override;
289 };
290 
291 class PosixMmapFile : public FSWritableFile {
292  private:
293   std::string filename_;
294   int fd_;
295   size_t page_size_;
296   size_t map_size_;       // How much extra memory to map at a time
297   char* base_;            // The mapped region
298   char* limit_;           // Limit of the mapped region
299   char* dst_;             // Where to write next  (in range [base_,limit_])
300   char* last_sync_;       // Where have we synced up to
301   uint64_t file_offset_;  // Offset of base_ in file
302 #ifdef ROCKSDB_FALLOCATE_PRESENT
303   bool allow_fallocate_;  // If false, fallocate calls are bypassed
304   bool fallocate_with_keep_size_;
305 #endif
306 
307   // Roundup x to a multiple of y
Roundup(size_t x,size_t y)308   static size_t Roundup(size_t x, size_t y) { return ((x + y - 1) / y) * y; }
309 
TruncateToPageBoundary(size_t s)310   size_t TruncateToPageBoundary(size_t s) {
311     s -= (s & (page_size_ - 1));
312     assert((s % page_size_) == 0);
313     return s;
314   }
315 
316   IOStatus MapNewRegion();
317   IOStatus UnmapCurrentRegion();
318   IOStatus Msync();
319 
320  public:
321   PosixMmapFile(const std::string& fname, int fd, size_t page_size,
322                 const EnvOptions& options);
323   ~PosixMmapFile();
324 
325   // Means Close() will properly take care of truncate
326   // and it does not need any additional information
Truncate(uint64_t,const IOOptions &,IODebugContext *)327   virtual IOStatus Truncate(uint64_t /*size*/, const IOOptions& /*opts*/,
328                             IODebugContext* /*dbg*/) override {
329     return IOStatus::OK();
330   }
331   virtual IOStatus Close(const IOOptions& opts, IODebugContext* dbg) override;
332   virtual IOStatus Append(const Slice& data, const IOOptions& opts,
333                           IODebugContext* dbg) override;
334   virtual IOStatus Flush(const IOOptions& opts, IODebugContext* dbg) override;
335   virtual IOStatus Sync(const IOOptions& opts, IODebugContext* dbg) override;
336   virtual IOStatus Fsync(const IOOptions& opts, IODebugContext* dbg) override;
337   virtual uint64_t GetFileSize(const IOOptions& opts,
338                                IODebugContext* dbg) override;
339   virtual IOStatus InvalidateCache(size_t offset, size_t length) override;
340 #ifdef ROCKSDB_FALLOCATE_PRESENT
341   virtual IOStatus Allocate(uint64_t offset, uint64_t len,
342                             const IOOptions& opts,
343                             IODebugContext* dbg) override;
344 #endif
345 };
346 
347 class PosixRandomRWFile : public FSRandomRWFile {
348  public:
349   explicit PosixRandomRWFile(const std::string& fname, int fd,
350                              const EnvOptions& options);
351   virtual ~PosixRandomRWFile();
352 
353   virtual IOStatus Write(uint64_t offset, const Slice& data,
354                          const IOOptions& opts, IODebugContext* dbg) override;
355 
356   virtual IOStatus Read(uint64_t offset, size_t n, const IOOptions& opts,
357                         Slice* result, char* scratch,
358                         IODebugContext* dbg) const override;
359 
360   virtual IOStatus Flush(const IOOptions& opts, IODebugContext* dbg) override;
361   virtual IOStatus Sync(const IOOptions& opts, IODebugContext* dbg) override;
362   virtual IOStatus Fsync(const IOOptions& opts, IODebugContext* dbg) override;
363   virtual IOStatus Close(const IOOptions& opts, IODebugContext* dbg) override;
364 
365  private:
366   const std::string filename_;
367   int fd_;
368 };
369 
370 struct PosixMemoryMappedFileBuffer : public MemoryMappedFileBuffer {
PosixMemoryMappedFileBufferPosixMemoryMappedFileBuffer371   PosixMemoryMappedFileBuffer(void* _base, size_t _length)
372       : MemoryMappedFileBuffer(_base, _length) {}
373   virtual ~PosixMemoryMappedFileBuffer();
374 };
375 
376 class PosixDirectory : public FSDirectory {
377  public:
PosixDirectory(int fd)378   explicit PosixDirectory(int fd) : fd_(fd) {}
379   ~PosixDirectory();
380   virtual IOStatus Fsync(const IOOptions& opts, IODebugContext* dbg) override;
381 
382  private:
383   int fd_;
384 };
385 
386 }  // namespace ROCKSDB_NAMESPACE
387