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
10 #ifdef ROCKSDB_LIB_IO_POSIX
11 #include "env/io_posix.h"
12 #include <errno.h>
13 #include <fcntl.h>
14 #include <algorithm>
15 #if defined(OS_LINUX)
16 #include <linux/fs.h>
17 #ifndef FALLOC_FL_KEEP_SIZE
18 #include <linux/falloc.h>
19 #endif
20 #endif
21 #include <stdio.h>
22 #include <stdlib.h>
23 #include <string.h>
24 #include <sys/ioctl.h>
25 #include <sys/mman.h>
26 #include <sys/stat.h>
27 #include <sys/types.h>
28 #ifdef OS_LINUX
29 #include <sys/statfs.h>
30 #include <sys/syscall.h>
31 #include <sys/sysmacros.h>
32 #endif
33 #include "monitoring/iostats_context_imp.h"
34 #include "port/port.h"
35 #include "rocksdb/slice.h"
36 #include "test_util/sync_point.h"
37 #include "util/autovector.h"
38 #include "util/coding.h"
39 #include "util/string_util.h"
40
41 #if defined(OS_LINUX) && !defined(F_SET_RW_HINT)
42 #define F_LINUX_SPECIFIC_BASE 1024
43 #define F_SET_RW_HINT (F_LINUX_SPECIFIC_BASE + 12)
44 #endif
45
46 namespace ROCKSDB_NAMESPACE {
47
IOErrorMsg(const std::string & context,const std::string & file_name)48 std::string IOErrorMsg(const std::string& context,
49 const std::string& file_name) {
50 if (file_name.empty()) {
51 return context;
52 }
53 return context + ": " + file_name;
54 }
55
56 // file_name can be left empty if it is not unkown.
IOError(const std::string & context,const std::string & file_name,int err_number)57 IOStatus IOError(const std::string& context, const std::string& file_name,
58 int err_number) {
59 switch (err_number) {
60 case ENOSPC: {
61 IOStatus s = IOStatus::NoSpace(IOErrorMsg(context, file_name),
62 strerror(err_number));
63 s.SetRetryable(true);
64 return s;
65 }
66 case ESTALE:
67 return IOStatus::IOError(IOStatus::kStaleFile);
68 case ENOENT:
69 return IOStatus::PathNotFound(IOErrorMsg(context, file_name),
70 strerror(err_number));
71 default:
72 return IOStatus::IOError(IOErrorMsg(context, file_name),
73 strerror(err_number));
74 }
75 }
76
77 // A wrapper for fadvise, if the platform doesn't support fadvise,
78 // it will simply return 0.
Fadvise(int fd,off_t offset,size_t len,int advice)79 int Fadvise(int fd, off_t offset, size_t len, int advice) {
80 #ifdef OS_LINUX
81 return posix_fadvise(fd, offset, len, advice);
82 #else
83 (void)fd;
84 (void)offset;
85 (void)len;
86 (void)advice;
87 return 0; // simply do nothing.
88 #endif
89 }
90
91 namespace {
92
93 // On MacOS (and probably *BSD), the posix write and pwrite calls do not support
94 // buffers larger than 2^31-1 bytes. These two wrappers fix this issue by
95 // cutting the buffer in 1GB chunks. We use this chunk size to be sure to keep
96 // the writes aligned.
97
PosixWrite(int fd,const char * buf,size_t nbyte)98 bool PosixWrite(int fd, const char* buf, size_t nbyte) {
99 const size_t kLimit1Gb = 1UL << 30;
100
101 const char* src = buf;
102 size_t left = nbyte;
103
104 while (left != 0) {
105 size_t bytes_to_write = std::min(left, kLimit1Gb);
106
107 ssize_t done = write(fd, src, bytes_to_write);
108 if (done < 0) {
109 if (errno == EINTR) {
110 continue;
111 }
112 return false;
113 }
114 left -= done;
115 src += done;
116 }
117 return true;
118 }
119
PosixPositionedWrite(int fd,const char * buf,size_t nbyte,off_t offset)120 bool PosixPositionedWrite(int fd, const char* buf, size_t nbyte, off_t offset) {
121 const size_t kLimit1Gb = 1UL << 30;
122
123 const char* src = buf;
124 size_t left = nbyte;
125
126 while (left != 0) {
127 size_t bytes_to_write = std::min(left, kLimit1Gb);
128
129 ssize_t done = pwrite(fd, src, bytes_to_write, offset);
130 if (done < 0) {
131 if (errno == EINTR) {
132 continue;
133 }
134 return false;
135 }
136 left -= done;
137 offset += done;
138 src += done;
139 }
140
141 return true;
142 }
143
144 #ifdef ROCKSDB_RANGESYNC_PRESENT
145
146 #if !defined(ZFS_SUPER_MAGIC)
147 // The magic number for ZFS was not exposed until recently. It should be fixed
148 // forever so we can just copy the magic number here.
149 #define ZFS_SUPER_MAGIC 0x2fc12fc1
150 #endif
151
IsSyncFileRangeSupported(int fd)152 bool IsSyncFileRangeSupported(int fd) {
153 // The approach taken in this function is to build a blacklist of cases where
154 // we know `sync_file_range` definitely will not work properly despite passing
155 // the compile-time check (`ROCKSDB_RANGESYNC_PRESENT`). If we are unsure, or
156 // if any of the checks fail in unexpected ways, we allow `sync_file_range` to
157 // be used. This way should minimize risk of impacting existing use cases.
158 struct statfs buf;
159 int ret = fstatfs(fd, &buf);
160 assert(ret == 0);
161 if (ret == 0 && buf.f_type == ZFS_SUPER_MAGIC) {
162 // Testing on ZFS showed the writeback did not happen asynchronously when
163 // `sync_file_range` was called, even though it returned success. Avoid it
164 // and use `fdatasync` instead to preserve the contract of `bytes_per_sync`,
165 // even though this'll incur extra I/O for metadata.
166 return false;
167 }
168
169 ret = sync_file_range(fd, 0 /* offset */, 0 /* nbytes */, 0 /* flags */);
170 assert(!(ret == -1 && errno != ENOSYS));
171 if (ret == -1 && errno == ENOSYS) {
172 // `sync_file_range` is not implemented on all platforms even if
173 // compile-time checks pass and a supported filesystem is in-use. For
174 // example, using ext4 on WSL (Windows Subsystem for Linux),
175 // `sync_file_range()` returns `ENOSYS`
176 // ("Function not implemented").
177 return false;
178 }
179 // None of the cases on the blacklist matched, so allow `sync_file_range` use.
180 return true;
181 }
182
183 #undef ZFS_SUPER_MAGIC
184
185 #endif // ROCKSDB_RANGESYNC_PRESENT
186
187 } // anonymous namespace
188
189 /*
190 * DirectIOHelper
191 */
192 #ifndef NDEBUG
193 namespace {
194
IsSectorAligned(const size_t off,size_t sector_size)195 bool IsSectorAligned(const size_t off, size_t sector_size) {
196 return off % sector_size == 0;
197 }
198
IsSectorAligned(const void * ptr,size_t sector_size)199 bool IsSectorAligned(const void* ptr, size_t sector_size) {
200 return uintptr_t(ptr) % sector_size == 0;
201 }
202
203 } // namespace
204 #endif
205
206 /*
207 * PosixSequentialFile
208 */
PosixSequentialFile(const std::string & fname,FILE * file,int fd,size_t logical_block_size,const EnvOptions & options)209 PosixSequentialFile::PosixSequentialFile(const std::string& fname, FILE* file,
210 int fd, size_t logical_block_size,
211 const EnvOptions& options)
212 : filename_(fname),
213 file_(file),
214 fd_(fd),
215 use_direct_io_(options.use_direct_reads),
216 logical_sector_size_(logical_block_size) {
217 assert(!options.use_direct_reads || !options.use_mmap_reads);
218 }
219
~PosixSequentialFile()220 PosixSequentialFile::~PosixSequentialFile() {
221 if (!use_direct_io()) {
222 assert(file_);
223 fclose(file_);
224 } else {
225 assert(fd_);
226 close(fd_);
227 }
228 }
229
Read(size_t n,const IOOptions &,Slice * result,char * scratch,IODebugContext *)230 IOStatus PosixSequentialFile::Read(size_t n, const IOOptions& /*opts*/,
231 Slice* result, char* scratch,
232 IODebugContext* /*dbg*/) {
233 assert(result != nullptr && !use_direct_io());
234 IOStatus s;
235 size_t r = 0;
236 do {
237 clearerr(file_);
238 r = fread_unlocked(scratch, 1, n, file_);
239 } while (r == 0 && ferror(file_) && errno == EINTR);
240 *result = Slice(scratch, r);
241 if (r < n) {
242 if (feof(file_)) {
243 // We leave status as ok if we hit the end of the file
244 // We also clear the error so that the reads can continue
245 // if a new data is written to the file
246 clearerr(file_);
247 } else {
248 // A partial read with an error: return a non-ok status
249 s = IOError("While reading file sequentially", filename_, errno);
250 }
251 }
252 return s;
253 }
254
PositionedRead(uint64_t offset,size_t n,const IOOptions &,Slice * result,char * scratch,IODebugContext *)255 IOStatus PosixSequentialFile::PositionedRead(uint64_t offset, size_t n,
256 const IOOptions& /*opts*/,
257 Slice* result, char* scratch,
258 IODebugContext* /*dbg*/) {
259 assert(use_direct_io());
260 assert(IsSectorAligned(offset, GetRequiredBufferAlignment()));
261 assert(IsSectorAligned(n, GetRequiredBufferAlignment()));
262 assert(IsSectorAligned(scratch, GetRequiredBufferAlignment()));
263
264 IOStatus s;
265 ssize_t r = -1;
266 size_t left = n;
267 char* ptr = scratch;
268 while (left > 0) {
269 r = pread(fd_, ptr, left, static_cast<off_t>(offset));
270 if (r <= 0) {
271 if (r == -1 && errno == EINTR) {
272 continue;
273 }
274 break;
275 }
276 ptr += r;
277 offset += r;
278 left -= r;
279 if (r % static_cast<ssize_t>(GetRequiredBufferAlignment()) != 0) {
280 // Bytes reads don't fill sectors. Should only happen at the end
281 // of the file.
282 break;
283 }
284 }
285 if (r < 0) {
286 // An error: return a non-ok status
287 s = IOError(
288 "While pread " + ToString(n) + " bytes from offset " + ToString(offset),
289 filename_, errno);
290 }
291 *result = Slice(scratch, (r < 0) ? 0 : n - left);
292 return s;
293 }
294
Skip(uint64_t n)295 IOStatus PosixSequentialFile::Skip(uint64_t n) {
296 if (fseek(file_, static_cast<long int>(n), SEEK_CUR)) {
297 return IOError("While fseek to skip " + ToString(n) + " bytes", filename_,
298 errno);
299 }
300 return IOStatus::OK();
301 }
302
InvalidateCache(size_t offset,size_t length)303 IOStatus PosixSequentialFile::InvalidateCache(size_t offset, size_t length) {
304 #ifndef OS_LINUX
305 (void)offset;
306 (void)length;
307 return IOStatus::OK();
308 #else
309 if (!use_direct_io()) {
310 // free OS pages
311 int ret = Fadvise(fd_, offset, length, POSIX_FADV_DONTNEED);
312 if (ret != 0) {
313 return IOError("While fadvise NotNeeded offset " + ToString(offset) +
314 " len " + ToString(length),
315 filename_, errno);
316 }
317 }
318 return IOStatus::OK();
319 #endif
320 }
321
322 /*
323 * PosixRandomAccessFile
324 */
325 #if defined(OS_LINUX)
GetUniqueIdFromFile(int fd,char * id,size_t max_size)326 size_t PosixHelper::GetUniqueIdFromFile(int fd, char* id, size_t max_size) {
327 if (max_size < kMaxVarint64Length * 3) {
328 return 0;
329 }
330
331 struct stat buf;
332 int result = fstat(fd, &buf);
333 if (result == -1) {
334 return 0;
335 }
336
337 long version = 0;
338 result = ioctl(fd, FS_IOC_GETVERSION, &version);
339 TEST_SYNC_POINT_CALLBACK("GetUniqueIdFromFile:FS_IOC_GETVERSION", &result);
340 if (result == -1) {
341 return 0;
342 }
343 uint64_t uversion = (uint64_t)version;
344
345 char* rid = id;
346 rid = EncodeVarint64(rid, buf.st_dev);
347 rid = EncodeVarint64(rid, buf.st_ino);
348 rid = EncodeVarint64(rid, uversion);
349 assert(rid >= id);
350 return static_cast<size_t>(rid - id);
351 }
352 #endif
353
354 #if defined(OS_MACOSX) || defined(OS_AIX)
GetUniqueIdFromFile(int fd,char * id,size_t max_size)355 size_t PosixHelper::GetUniqueIdFromFile(int fd, char* id, size_t max_size) {
356 if (max_size < kMaxVarint64Length * 3) {
357 return 0;
358 }
359
360 struct stat buf;
361 int result = fstat(fd, &buf);
362 if (result == -1) {
363 return 0;
364 }
365
366 char* rid = id;
367 rid = EncodeVarint64(rid, buf.st_dev);
368 rid = EncodeVarint64(rid, buf.st_ino);
369 rid = EncodeVarint64(rid, buf.st_gen);
370 assert(rid >= id);
371 return static_cast<size_t>(rid - id);
372 }
373 #endif
374
375 #ifdef OS_LINUX
RemoveTrailingSlash(const std::string & path)376 std::string RemoveTrailingSlash(const std::string& path) {
377 std::string p = path;
378 if (p.size() > 1 && p.back() == '/') {
379 p.pop_back();
380 }
381 return p;
382 }
383
RefAndCacheLogicalBlockSize(const std::vector<std::string> & directories)384 Status LogicalBlockSizeCache::RefAndCacheLogicalBlockSize(
385 const std::vector<std::string>& directories) {
386 std::vector<std::string> dirs;
387 dirs.reserve(directories.size());
388 for (auto& d : directories) {
389 dirs.emplace_back(RemoveTrailingSlash(d));
390 }
391
392 std::map<std::string, size_t> dir_sizes;
393 {
394 ReadLock lock(&cache_mutex_);
395 for (const auto& dir : dirs) {
396 if (cache_.find(dir) == cache_.end()) {
397 dir_sizes.emplace(dir, 0);
398 }
399 }
400 }
401
402 Status s;
403 for (auto& dir_size : dir_sizes) {
404 s = get_logical_block_size_of_directory_(dir_size.first, &dir_size.second);
405 if (!s.ok()) {
406 return s;
407 }
408 }
409
410 WriteLock lock(&cache_mutex_);
411 for (const auto& dir : dirs) {
412 auto& v = cache_[dir];
413 v.ref++;
414 auto dir_size = dir_sizes.find(dir);
415 if (dir_size != dir_sizes.end()) {
416 v.size = dir_size->second;
417 }
418 }
419 return Status::OK();
420 }
421
UnrefAndTryRemoveCachedLogicalBlockSize(const std::vector<std::string> & directories)422 void LogicalBlockSizeCache::UnrefAndTryRemoveCachedLogicalBlockSize(
423 const std::vector<std::string>& directories) {
424 std::vector<std::string> dirs;
425 dirs.reserve(directories.size());
426 for (auto& dir : directories) {
427 dirs.emplace_back(RemoveTrailingSlash(dir));
428 }
429
430 WriteLock lock(&cache_mutex_);
431 for (const auto& dir : dirs) {
432 auto it = cache_.find(dir);
433 if (it != cache_.end() && !(--(it->second.ref))) {
434 cache_.erase(it);
435 }
436 }
437 }
438
GetLogicalBlockSize(const std::string & fname,int fd)439 size_t LogicalBlockSizeCache::GetLogicalBlockSize(const std::string& fname,
440 int fd) {
441 std::string dir = fname.substr(0, fname.find_last_of("/"));
442 if (dir.empty()) {
443 dir = "/";
444 }
445 {
446 ReadLock lock(&cache_mutex_);
447 auto it = cache_.find(dir);
448 if (it != cache_.end()) {
449 return it->second.size;
450 }
451 }
452 return get_logical_block_size_of_fd_(fd);
453 }
454 #endif
455
GetLogicalBlockSizeOfDirectory(const std::string & directory,size_t * size)456 Status PosixHelper::GetLogicalBlockSizeOfDirectory(const std::string& directory,
457 size_t* size) {
458 int fd = open(directory.c_str(), O_DIRECTORY | O_RDONLY);
459 if (fd == -1) {
460 close(fd);
461 return Status::IOError("Cannot open directory " + directory);
462 }
463 *size = PosixHelper::GetLogicalBlockSizeOfFd(fd);
464 close(fd);
465 return Status::OK();
466 }
467
GetLogicalBlockSizeOfFd(int fd)468 size_t PosixHelper::GetLogicalBlockSizeOfFd(int fd) {
469 #ifdef OS_LINUX
470 struct stat buf;
471 int result = fstat(fd, &buf);
472 if (result == -1) {
473 return kDefaultPageSize;
474 }
475 if (major(buf.st_dev) == 0) {
476 // Unnamed devices (e.g. non-device mounts), reserved as null device number.
477 // These don't have an entry in /sys/dev/block/. Return a sensible default.
478 return kDefaultPageSize;
479 }
480
481 // Reading queue/logical_block_size does not require special permissions.
482 const int kBufferSize = 100;
483 char path[kBufferSize];
484 char real_path[PATH_MAX + 1];
485 snprintf(path, kBufferSize, "/sys/dev/block/%u:%u", major(buf.st_dev),
486 minor(buf.st_dev));
487 if (realpath(path, real_path) == nullptr) {
488 return kDefaultPageSize;
489 }
490 std::string device_dir(real_path);
491 if (!device_dir.empty() && device_dir.back() == '/') {
492 device_dir.pop_back();
493 }
494 // NOTE: sda3 and nvme0n1p1 do not have a `queue/` subdir, only the parent sda
495 // and nvme0n1 have it.
496 // $ ls -al '/sys/dev/block/8:3'
497 // lrwxrwxrwx. 1 root root 0 Jun 26 01:38 /sys/dev/block/8:3 ->
498 // ../../block/sda/sda3
499 // $ ls -al '/sys/dev/block/259:4'
500 // lrwxrwxrwx 1 root root 0 Jan 31 16:04 /sys/dev/block/259:4 ->
501 // ../../devices/pci0000:17/0000:17:00.0/0000:18:00.0/nvme/nvme0/nvme0n1/nvme0n1p1
502 size_t parent_end = device_dir.rfind('/', device_dir.length() - 1);
503 if (parent_end == std::string::npos) {
504 return kDefaultPageSize;
505 }
506 size_t parent_begin = device_dir.rfind('/', parent_end - 1);
507 if (parent_begin == std::string::npos) {
508 return kDefaultPageSize;
509 }
510 std::string parent =
511 device_dir.substr(parent_begin + 1, parent_end - parent_begin - 1);
512 std::string child = device_dir.substr(parent_end + 1, std::string::npos);
513 if (parent != "block" &&
514 (child.compare(0, 4, "nvme") || child.find('p') != std::string::npos)) {
515 device_dir = device_dir.substr(0, parent_end);
516 }
517 std::string fname = device_dir + "/queue/logical_block_size";
518 FILE* fp;
519 size_t size = 0;
520 fp = fopen(fname.c_str(), "r");
521 if (fp != nullptr) {
522 char* line = nullptr;
523 size_t len = 0;
524 if (getline(&line, &len, fp) != -1) {
525 sscanf(line, "%zu", &size);
526 }
527 free(line);
528 fclose(fp);
529 }
530 if (size != 0 && (size & (size - 1)) == 0) {
531 return size;
532 }
533 #endif
534 (void)fd;
535 return kDefaultPageSize;
536 }
537
538 /*
539 * PosixRandomAccessFile
540 *
541 * pread() based random-access
542 */
PosixRandomAccessFile(const std::string & fname,int fd,size_t logical_block_size,const EnvOptions & options,ThreadLocalPtr * thread_local_io_urings)543 PosixRandomAccessFile::PosixRandomAccessFile(
544 const std::string& fname, int fd, size_t logical_block_size,
545 const EnvOptions& options
546 #if defined(ROCKSDB_IOURING_PRESENT)
547 ,
548 ThreadLocalPtr* thread_local_io_urings
549 #endif
550 )
551 : filename_(fname),
552 fd_(fd),
553 use_direct_io_(options.use_direct_reads),
554 logical_sector_size_(logical_block_size)
555 #if defined(ROCKSDB_IOURING_PRESENT)
556 ,
557 thread_local_io_urings_(thread_local_io_urings)
558 #endif
559 {
560 assert(!options.use_direct_reads || !options.use_mmap_reads);
561 assert(!options.use_mmap_reads || sizeof(void*) < 8);
562 }
563
~PosixRandomAccessFile()564 PosixRandomAccessFile::~PosixRandomAccessFile() { close(fd_); }
565
Read(uint64_t offset,size_t n,const IOOptions &,Slice * result,char * scratch,IODebugContext *) const566 IOStatus PosixRandomAccessFile::Read(uint64_t offset, size_t n,
567 const IOOptions& /*opts*/, Slice* result,
568 char* scratch,
569 IODebugContext* /*dbg*/) const {
570 if (use_direct_io()) {
571 assert(IsSectorAligned(offset, GetRequiredBufferAlignment()));
572 assert(IsSectorAligned(n, GetRequiredBufferAlignment()));
573 assert(IsSectorAligned(scratch, GetRequiredBufferAlignment()));
574 }
575 IOStatus s;
576 ssize_t r = -1;
577 size_t left = n;
578 char* ptr = scratch;
579 while (left > 0) {
580 r = pread(fd_, ptr, left, static_cast<off_t>(offset));
581 if (r <= 0) {
582 if (r == -1 && errno == EINTR) {
583 continue;
584 }
585 break;
586 }
587 ptr += r;
588 offset += r;
589 left -= r;
590 if (use_direct_io() &&
591 r % static_cast<ssize_t>(GetRequiredBufferAlignment()) != 0) {
592 // Bytes reads don't fill sectors. Should only happen at the end
593 // of the file.
594 break;
595 }
596 }
597 if (r < 0) {
598 // An error: return a non-ok status
599 s = IOError(
600 "While pread offset " + ToString(offset) + " len " + ToString(n),
601 filename_, errno);
602 }
603 *result = Slice(scratch, (r < 0) ? 0 : n - left);
604 return s;
605 }
606
MultiRead(FSReadRequest * reqs,size_t num_reqs,const IOOptions & options,IODebugContext * dbg)607 IOStatus PosixRandomAccessFile::MultiRead(FSReadRequest* reqs,
608 size_t num_reqs,
609 const IOOptions& options,
610 IODebugContext* dbg) {
611 #if defined(ROCKSDB_IOURING_PRESENT)
612 struct io_uring* iu = nullptr;
613 if (thread_local_io_urings_) {
614 iu = static_cast<struct io_uring*>(thread_local_io_urings_->Get());
615 if (iu == nullptr) {
616 iu = CreateIOUring();
617 if (iu != nullptr) {
618 thread_local_io_urings_->Reset(iu);
619 }
620 }
621 }
622
623 // Init failed, platform doesn't support io_uring. Fall back to
624 // serialized reads
625 if (iu == nullptr) {
626 return FSRandomAccessFile::MultiRead(reqs, num_reqs, options, dbg);
627 }
628
629 struct WrappedReadRequest {
630 FSReadRequest* req;
631 struct iovec iov;
632 size_t finished_len;
633 explicit WrappedReadRequest(FSReadRequest* r) : req(r), finished_len(0) {}
634 };
635
636 autovector<WrappedReadRequest, 32> req_wraps;
637 autovector<WrappedReadRequest*, 4> incomplete_rq_list;
638
639 for (size_t i = 0; i < num_reqs; i++) {
640 req_wraps.emplace_back(&reqs[i]);
641 }
642
643 size_t reqs_off = 0;
644 while (num_reqs > reqs_off || !incomplete_rq_list.empty()) {
645 size_t this_reqs = (num_reqs - reqs_off) + incomplete_rq_list.size();
646
647 // If requests exceed depth, split it into batches
648 if (this_reqs > kIoUringDepth) this_reqs = kIoUringDepth;
649
650 assert(incomplete_rq_list.size() <= this_reqs);
651 for (size_t i = 0; i < this_reqs; i++) {
652 WrappedReadRequest* rep_to_submit;
653 if (i < incomplete_rq_list.size()) {
654 rep_to_submit = incomplete_rq_list[i];
655 } else {
656 rep_to_submit = &req_wraps[reqs_off++];
657 }
658 assert(rep_to_submit->req->len > rep_to_submit->finished_len);
659 rep_to_submit->iov.iov_base =
660 rep_to_submit->req->scratch + rep_to_submit->finished_len;
661 rep_to_submit->iov.iov_len =
662 rep_to_submit->req->len - rep_to_submit->finished_len;
663
664 struct io_uring_sqe* sqe;
665 sqe = io_uring_get_sqe(iu);
666 io_uring_prep_readv(
667 sqe, fd_, &rep_to_submit->iov, 1,
668 rep_to_submit->req->offset + rep_to_submit->finished_len);
669 io_uring_sqe_set_data(sqe, rep_to_submit);
670 }
671 incomplete_rq_list.clear();
672
673 ssize_t ret =
674 io_uring_submit_and_wait(iu, static_cast<unsigned int>(this_reqs));
675 if (static_cast<size_t>(ret) != this_reqs) {
676 fprintf(stderr, "ret = %ld this_reqs: %ld\n", (long)ret, (long)this_reqs);
677 }
678 assert(static_cast<size_t>(ret) == this_reqs);
679
680 for (size_t i = 0; i < this_reqs; i++) {
681 struct io_uring_cqe* cqe;
682 WrappedReadRequest* req_wrap;
683
684 // We could use the peek variant here, but this seems safer in terms
685 // of our initial wait not reaping all completions
686 ret = io_uring_wait_cqe(iu, &cqe);
687 assert(!ret);
688
689 req_wrap = static_cast<WrappedReadRequest*>(io_uring_cqe_get_data(cqe));
690 FSReadRequest* req = req_wrap->req;
691 if (cqe->res < 0) {
692 req->result = Slice(req->scratch, 0);
693 req->status = IOError("Req failed", filename_, cqe->res);
694 } else {
695 size_t bytes_read = static_cast<size_t>(cqe->res);
696 TEST_SYNC_POINT_CALLBACK(
697 "PosixRandomAccessFile::MultiRead:io_uring_result", &bytes_read);
698 if (bytes_read == req_wrap->iov.iov_len) {
699 req->result = Slice(req->scratch, req->len);
700 req->status = IOStatus::OK();
701 } else if (bytes_read == 0) {
702 // cqe->res == 0 can means EOF, or can mean partial results. See
703 // comment
704 // https://github.com/facebook/rocksdb/pull/6441#issuecomment-589843435
705 // Fall back to pread in this case.
706 Slice tmp_slice;
707 req->status =
708 Read(req->offset + req_wrap->finished_len,
709 req->len - req_wrap->finished_len, options, &tmp_slice,
710 req->scratch + req_wrap->finished_len, dbg);
711 req->result =
712 Slice(req->scratch, req_wrap->finished_len + tmp_slice.size());
713 } else if (bytes_read < req_wrap->iov.iov_len) {
714 assert(bytes_read > 0);
715 assert(bytes_read + req_wrap->finished_len < req->len);
716 req_wrap->finished_len += bytes_read;
717 incomplete_rq_list.push_back(req_wrap);
718 } else {
719 req->result = Slice(req->scratch, 0);
720 req->status = IOError("Req returned more bytes than requested",
721 filename_, cqe->res);
722 }
723 }
724 io_uring_cqe_seen(iu, cqe);
725 }
726 }
727 return IOStatus::OK();
728 #else
729 return FSRandomAccessFile::MultiRead(reqs, num_reqs, options, dbg);
730 #endif
731 }
732
Prefetch(uint64_t offset,size_t n,const IOOptions &,IODebugContext *)733 IOStatus PosixRandomAccessFile::Prefetch(uint64_t offset, size_t n,
734 const IOOptions& /*opts*/,
735 IODebugContext* /*dbg*/) {
736 IOStatus s;
737 if (!use_direct_io()) {
738 ssize_t r = 0;
739 #ifdef OS_LINUX
740 r = readahead(fd_, offset, n);
741 #endif
742 #ifdef OS_MACOSX
743 radvisory advice;
744 advice.ra_offset = static_cast<off_t>(offset);
745 advice.ra_count = static_cast<int>(n);
746 r = fcntl(fd_, F_RDADVISE, &advice);
747 #endif
748 if (r == -1) {
749 s = IOError("While prefetching offset " + ToString(offset) + " len " +
750 ToString(n),
751 filename_, errno);
752 }
753 }
754 return s;
755 }
756
757 #if defined(OS_LINUX) || defined(OS_MACOSX) || defined(OS_AIX)
GetUniqueId(char * id,size_t max_size) const758 size_t PosixRandomAccessFile::GetUniqueId(char* id, size_t max_size) const {
759 return PosixHelper::GetUniqueIdFromFile(fd_, id, max_size);
760 }
761 #endif
762
Hint(AccessPattern pattern)763 void PosixRandomAccessFile::Hint(AccessPattern pattern) {
764 if (use_direct_io()) {
765 return;
766 }
767 switch (pattern) {
768 case kNormal:
769 Fadvise(fd_, 0, 0, POSIX_FADV_NORMAL);
770 break;
771 case kRandom:
772 Fadvise(fd_, 0, 0, POSIX_FADV_RANDOM);
773 break;
774 case kSequential:
775 Fadvise(fd_, 0, 0, POSIX_FADV_SEQUENTIAL);
776 break;
777 case kWillNeed:
778 Fadvise(fd_, 0, 0, POSIX_FADV_WILLNEED);
779 break;
780 case kWontNeed:
781 Fadvise(fd_, 0, 0, POSIX_FADV_DONTNEED);
782 break;
783 default:
784 assert(false);
785 break;
786 }
787 }
788
InvalidateCache(size_t offset,size_t length)789 IOStatus PosixRandomAccessFile::InvalidateCache(size_t offset, size_t length) {
790 if (use_direct_io()) {
791 return IOStatus::OK();
792 }
793 #ifndef OS_LINUX
794 (void)offset;
795 (void)length;
796 return IOStatus::OK();
797 #else
798 // free OS pages
799 int ret = Fadvise(fd_, offset, length, POSIX_FADV_DONTNEED);
800 if (ret == 0) {
801 return IOStatus::OK();
802 }
803 return IOError("While fadvise NotNeeded offset " + ToString(offset) +
804 " len " + ToString(length),
805 filename_, errno);
806 #endif
807 }
808
809 /*
810 * PosixMmapReadableFile
811 *
812 * mmap() based random-access
813 */
814 // base[0,length-1] contains the mmapped contents of the file.
PosixMmapReadableFile(const int fd,const std::string & fname,void * base,size_t length,const EnvOptions & options)815 PosixMmapReadableFile::PosixMmapReadableFile(const int fd,
816 const std::string& fname,
817 void* base, size_t length,
818 const EnvOptions& options)
819 : fd_(fd), filename_(fname), mmapped_region_(base), length_(length) {
820 #ifdef NDEBUG
821 (void)options;
822 #endif
823 fd_ = fd_ + 0; // suppress the warning for used variables
824 assert(options.use_mmap_reads);
825 assert(!options.use_direct_reads);
826 }
827
~PosixMmapReadableFile()828 PosixMmapReadableFile::~PosixMmapReadableFile() {
829 int ret = munmap(mmapped_region_, length_);
830 if (ret != 0) {
831 fprintf(stdout, "failed to munmap %p length %" ROCKSDB_PRIszt " \n",
832 mmapped_region_, length_);
833 }
834 close(fd_);
835 }
836
Read(uint64_t offset,size_t n,const IOOptions &,Slice * result,char *,IODebugContext *) const837 IOStatus PosixMmapReadableFile::Read(uint64_t offset, size_t n,
838 const IOOptions& /*opts*/, Slice* result,
839 char* /*scratch*/,
840 IODebugContext* /*dbg*/) const {
841 IOStatus s;
842 if (offset > length_) {
843 *result = Slice();
844 return IOError("While mmap read offset " + ToString(offset) +
845 " larger than file length " + ToString(length_),
846 filename_, EINVAL);
847 } else if (offset + n > length_) {
848 n = static_cast<size_t>(length_ - offset);
849 }
850 *result = Slice(reinterpret_cast<char*>(mmapped_region_) + offset, n);
851 return s;
852 }
853
InvalidateCache(size_t offset,size_t length)854 IOStatus PosixMmapReadableFile::InvalidateCache(size_t offset, size_t length) {
855 #ifndef OS_LINUX
856 (void)offset;
857 (void)length;
858 return IOStatus::OK();
859 #else
860 // free OS pages
861 int ret = Fadvise(fd_, offset, length, POSIX_FADV_DONTNEED);
862 if (ret == 0) {
863 return IOStatus::OK();
864 }
865 return IOError("While fadvise not needed. Offset " + ToString(offset) +
866 " len" + ToString(length),
867 filename_, errno);
868 #endif
869 }
870
871 /*
872 * PosixMmapFile
873 *
874 * We preallocate up to an extra megabyte and use memcpy to append new
875 * data to the file. This is safe since we either properly close the
876 * file before reading from it, or for log files, the reading code
877 * knows enough to skip zero suffixes.
878 */
UnmapCurrentRegion()879 IOStatus PosixMmapFile::UnmapCurrentRegion() {
880 TEST_KILL_RANDOM("PosixMmapFile::UnmapCurrentRegion:0", rocksdb_kill_odds);
881 if (base_ != nullptr) {
882 int munmap_status = munmap(base_, limit_ - base_);
883 if (munmap_status != 0) {
884 return IOError("While munmap", filename_, munmap_status);
885 }
886 file_offset_ += limit_ - base_;
887 base_ = nullptr;
888 limit_ = nullptr;
889 last_sync_ = nullptr;
890 dst_ = nullptr;
891
892 // Increase the amount we map the next time, but capped at 1MB
893 if (map_size_ < (1 << 20)) {
894 map_size_ *= 2;
895 }
896 }
897 return IOStatus::OK();
898 }
899
MapNewRegion()900 IOStatus PosixMmapFile::MapNewRegion() {
901 #ifdef ROCKSDB_FALLOCATE_PRESENT
902 assert(base_ == nullptr);
903 TEST_KILL_RANDOM("PosixMmapFile::UnmapCurrentRegion:0", rocksdb_kill_odds);
904 // we can't fallocate with FALLOC_FL_KEEP_SIZE here
905 if (allow_fallocate_) {
906 IOSTATS_TIMER_GUARD(allocate_nanos);
907 int alloc_status = fallocate(fd_, 0, file_offset_, map_size_);
908 if (alloc_status != 0) {
909 // fallback to posix_fallocate
910 alloc_status = posix_fallocate(fd_, file_offset_, map_size_);
911 }
912 if (alloc_status != 0) {
913 return IOStatus::IOError("Error allocating space to file : " + filename_ +
914 "Error : " + strerror(alloc_status));
915 }
916 }
917
918 TEST_KILL_RANDOM("PosixMmapFile::Append:1", rocksdb_kill_odds);
919 void* ptr = mmap(nullptr, map_size_, PROT_READ | PROT_WRITE, MAP_SHARED, fd_,
920 file_offset_);
921 if (ptr == MAP_FAILED) {
922 return IOStatus::IOError("MMap failed on " + filename_);
923 }
924 TEST_KILL_RANDOM("PosixMmapFile::Append:2", rocksdb_kill_odds);
925
926 base_ = reinterpret_cast<char*>(ptr);
927 limit_ = base_ + map_size_;
928 dst_ = base_;
929 last_sync_ = base_;
930 return IOStatus::OK();
931 #else
932 return IOStatus::NotSupported("This platform doesn't support fallocate()");
933 #endif
934 }
935
Msync()936 IOStatus PosixMmapFile::Msync() {
937 if (dst_ == last_sync_) {
938 return IOStatus::OK();
939 }
940 // Find the beginnings of the pages that contain the first and last
941 // bytes to be synced.
942 size_t p1 = TruncateToPageBoundary(last_sync_ - base_);
943 size_t p2 = TruncateToPageBoundary(dst_ - base_ - 1);
944 last_sync_ = dst_;
945 TEST_KILL_RANDOM("PosixMmapFile::Msync:0", rocksdb_kill_odds);
946 if (msync(base_ + p1, p2 - p1 + page_size_, MS_SYNC) < 0) {
947 return IOError("While msync", filename_, errno);
948 }
949 return IOStatus::OK();
950 }
951
PosixMmapFile(const std::string & fname,int fd,size_t page_size,const EnvOptions & options)952 PosixMmapFile::PosixMmapFile(const std::string& fname, int fd, size_t page_size,
953 const EnvOptions& options)
954 : filename_(fname),
955 fd_(fd),
956 page_size_(page_size),
957 map_size_(Roundup(65536, page_size)),
958 base_(nullptr),
959 limit_(nullptr),
960 dst_(nullptr),
961 last_sync_(nullptr),
962 file_offset_(0) {
963 #ifdef ROCKSDB_FALLOCATE_PRESENT
964 allow_fallocate_ = options.allow_fallocate;
965 fallocate_with_keep_size_ = options.fallocate_with_keep_size;
966 #else
967 (void)options;
968 #endif
969 assert((page_size & (page_size - 1)) == 0);
970 assert(options.use_mmap_writes);
971 assert(!options.use_direct_writes);
972 }
973
~PosixMmapFile()974 PosixMmapFile::~PosixMmapFile() {
975 if (fd_ >= 0) {
976 PosixMmapFile::Close(IOOptions(), nullptr);
977 }
978 }
979
Append(const Slice & data,const IOOptions &,IODebugContext *)980 IOStatus PosixMmapFile::Append(const Slice& data, const IOOptions& /*opts*/,
981 IODebugContext* /*dbg*/) {
982 const char* src = data.data();
983 size_t left = data.size();
984 while (left > 0) {
985 assert(base_ <= dst_);
986 assert(dst_ <= limit_);
987 size_t avail = limit_ - dst_;
988 if (avail == 0) {
989 IOStatus s = UnmapCurrentRegion();
990 if (!s.ok()) {
991 return s;
992 }
993 s = MapNewRegion();
994 if (!s.ok()) {
995 return s;
996 }
997 TEST_KILL_RANDOM("PosixMmapFile::Append:0", rocksdb_kill_odds);
998 }
999
1000 size_t n = (left <= avail) ? left : avail;
1001 assert(dst_);
1002 memcpy(dst_, src, n);
1003 dst_ += n;
1004 src += n;
1005 left -= n;
1006 }
1007 return IOStatus::OK();
1008 }
1009
Close(const IOOptions &,IODebugContext *)1010 IOStatus PosixMmapFile::Close(const IOOptions& /*opts*/,
1011 IODebugContext* /*dbg*/) {
1012 IOStatus s;
1013 size_t unused = limit_ - dst_;
1014
1015 s = UnmapCurrentRegion();
1016 if (!s.ok()) {
1017 s = IOError("While closing mmapped file", filename_, errno);
1018 } else if (unused > 0) {
1019 // Trim the extra space at the end of the file
1020 if (ftruncate(fd_, file_offset_ - unused) < 0) {
1021 s = IOError("While ftruncating mmaped file", filename_, errno);
1022 }
1023 }
1024
1025 if (close(fd_) < 0) {
1026 if (s.ok()) {
1027 s = IOError("While closing mmapped file", filename_, errno);
1028 }
1029 }
1030
1031 fd_ = -1;
1032 base_ = nullptr;
1033 limit_ = nullptr;
1034 return s;
1035 }
1036
Flush(const IOOptions &,IODebugContext *)1037 IOStatus PosixMmapFile::Flush(const IOOptions& /*opts*/,
1038 IODebugContext* /*dbg*/) {
1039 return IOStatus::OK();
1040 }
1041
Sync(const IOOptions &,IODebugContext *)1042 IOStatus PosixMmapFile::Sync(const IOOptions& /*opts*/,
1043 IODebugContext* /*dbg*/) {
1044 if (fdatasync(fd_) < 0) {
1045 return IOError("While fdatasync mmapped file", filename_, errno);
1046 }
1047
1048 return Msync();
1049 }
1050
1051 /**
1052 * Flush data as well as metadata to stable storage.
1053 */
Fsync(const IOOptions &,IODebugContext *)1054 IOStatus PosixMmapFile::Fsync(const IOOptions& /*opts*/,
1055 IODebugContext* /*dbg*/) {
1056 if (fsync(fd_) < 0) {
1057 return IOError("While fsync mmaped file", filename_, errno);
1058 }
1059
1060 return Msync();
1061 }
1062
1063 /**
1064 * Get the size of valid data in the file. This will not match the
1065 * size that is returned from the filesystem because we use mmap
1066 * to extend file by map_size every time.
1067 */
GetFileSize(const IOOptions &,IODebugContext *)1068 uint64_t PosixMmapFile::GetFileSize(const IOOptions& /*opts*/,
1069 IODebugContext* /*dbg*/) {
1070 size_t used = dst_ - base_;
1071 return file_offset_ + used;
1072 }
1073
InvalidateCache(size_t offset,size_t length)1074 IOStatus PosixMmapFile::InvalidateCache(size_t offset, size_t length) {
1075 #ifndef OS_LINUX
1076 (void)offset;
1077 (void)length;
1078 return IOStatus::OK();
1079 #else
1080 // free OS pages
1081 int ret = Fadvise(fd_, offset, length, POSIX_FADV_DONTNEED);
1082 if (ret == 0) {
1083 return IOStatus::OK();
1084 }
1085 return IOError("While fadvise NotNeeded mmapped file", filename_, errno);
1086 #endif
1087 }
1088
1089 #ifdef ROCKSDB_FALLOCATE_PRESENT
Allocate(uint64_t offset,uint64_t len,const IOOptions &,IODebugContext *)1090 IOStatus PosixMmapFile::Allocate(uint64_t offset, uint64_t len,
1091 const IOOptions& /*opts*/,
1092 IODebugContext* /*dbg*/) {
1093 assert(offset <= static_cast<uint64_t>(std::numeric_limits<off_t>::max()));
1094 assert(len <= static_cast<uint64_t>(std::numeric_limits<off_t>::max()));
1095 TEST_KILL_RANDOM("PosixMmapFile::Allocate:0", rocksdb_kill_odds);
1096 int alloc_status = 0;
1097 if (allow_fallocate_) {
1098 alloc_status =
1099 fallocate(fd_, fallocate_with_keep_size_ ? FALLOC_FL_KEEP_SIZE : 0,
1100 static_cast<off_t>(offset), static_cast<off_t>(len));
1101 }
1102 if (alloc_status == 0) {
1103 return IOStatus::OK();
1104 } else {
1105 return IOError(
1106 "While fallocate offset " + ToString(offset) + " len " + ToString(len),
1107 filename_, errno);
1108 }
1109 }
1110 #endif
1111
1112 /*
1113 * PosixWritableFile
1114 *
1115 * Use posix write to write data to a file.
1116 */
PosixWritableFile(const std::string & fname,int fd,size_t logical_block_size,const EnvOptions & options)1117 PosixWritableFile::PosixWritableFile(const std::string& fname, int fd,
1118 size_t logical_block_size,
1119 const EnvOptions& options)
1120 : FSWritableFile(options),
1121 filename_(fname),
1122 use_direct_io_(options.use_direct_writes),
1123 fd_(fd),
1124 filesize_(0),
1125 logical_sector_size_(logical_block_size) {
1126 #ifdef ROCKSDB_FALLOCATE_PRESENT
1127 allow_fallocate_ = options.allow_fallocate;
1128 fallocate_with_keep_size_ = options.fallocate_with_keep_size;
1129 #endif
1130 #ifdef ROCKSDB_RANGESYNC_PRESENT
1131 sync_file_range_supported_ = IsSyncFileRangeSupported(fd_);
1132 #endif // ROCKSDB_RANGESYNC_PRESENT
1133 assert(!options.use_mmap_writes);
1134 }
1135
~PosixWritableFile()1136 PosixWritableFile::~PosixWritableFile() {
1137 if (fd_ >= 0) {
1138 PosixWritableFile::Close(IOOptions(), nullptr);
1139 }
1140 }
1141
Append(const Slice & data,const IOOptions &,IODebugContext *)1142 IOStatus PosixWritableFile::Append(const Slice& data, const IOOptions& /*opts*/,
1143 IODebugContext* /*dbg*/) {
1144 if (use_direct_io()) {
1145 assert(IsSectorAligned(data.size(), GetRequiredBufferAlignment()));
1146 assert(IsSectorAligned(data.data(), GetRequiredBufferAlignment()));
1147 }
1148 const char* src = data.data();
1149 size_t nbytes = data.size();
1150
1151 if (!PosixWrite(fd_, src, nbytes)) {
1152 return IOError("While appending to file", filename_, errno);
1153 }
1154
1155 filesize_ += nbytes;
1156 return IOStatus::OK();
1157 }
1158
PositionedAppend(const Slice & data,uint64_t offset,const IOOptions &,IODebugContext *)1159 IOStatus PosixWritableFile::PositionedAppend(const Slice& data, uint64_t offset,
1160 const IOOptions& /*opts*/,
1161 IODebugContext* /*dbg*/) {
1162 if (use_direct_io()) {
1163 assert(IsSectorAligned(offset, GetRequiredBufferAlignment()));
1164 assert(IsSectorAligned(data.size(), GetRequiredBufferAlignment()));
1165 assert(IsSectorAligned(data.data(), GetRequiredBufferAlignment()));
1166 }
1167 assert(offset <= static_cast<uint64_t>(std::numeric_limits<off_t>::max()));
1168 const char* src = data.data();
1169 size_t nbytes = data.size();
1170 if (!PosixPositionedWrite(fd_, src, nbytes, static_cast<off_t>(offset))) {
1171 return IOError("While pwrite to file at offset " + ToString(offset),
1172 filename_, errno);
1173 }
1174 filesize_ = offset + nbytes;
1175 return IOStatus::OK();
1176 }
1177
Truncate(uint64_t size,const IOOptions &,IODebugContext *)1178 IOStatus PosixWritableFile::Truncate(uint64_t size, const IOOptions& /*opts*/,
1179 IODebugContext* /*dbg*/) {
1180 IOStatus s;
1181 int r = ftruncate(fd_, size);
1182 if (r < 0) {
1183 s = IOError("While ftruncate file to size " + ToString(size), filename_,
1184 errno);
1185 } else {
1186 filesize_ = size;
1187 }
1188 return s;
1189 }
1190
Close(const IOOptions &,IODebugContext *)1191 IOStatus PosixWritableFile::Close(const IOOptions& /*opts*/,
1192 IODebugContext* /*dbg*/) {
1193 IOStatus s;
1194
1195 size_t block_size;
1196 size_t last_allocated_block;
1197 GetPreallocationStatus(&block_size, &last_allocated_block);
1198 if (last_allocated_block > 0) {
1199 // trim the extra space preallocated at the end of the file
1200 // NOTE(ljin): we probably don't want to surface failure as an IOError,
1201 // but it will be nice to log these errors.
1202 int dummy __attribute__((__unused__));
1203 dummy = ftruncate(fd_, filesize_);
1204 #if defined(ROCKSDB_FALLOCATE_PRESENT) && defined(FALLOC_FL_PUNCH_HOLE) && \
1205 !defined(TRAVIS)
1206 // in some file systems, ftruncate only trims trailing space if the
1207 // new file size is smaller than the current size. Calling fallocate
1208 // with FALLOC_FL_PUNCH_HOLE flag to explicitly release these unused
1209 // blocks. FALLOC_FL_PUNCH_HOLE is supported on at least the following
1210 // filesystems:
1211 // XFS (since Linux 2.6.38)
1212 // ext4 (since Linux 3.0)
1213 // Btrfs (since Linux 3.7)
1214 // tmpfs (since Linux 3.5)
1215 // We ignore error since failure of this operation does not affect
1216 // correctness.
1217 // TRAVIS - this code does not work on TRAVIS filesystems.
1218 // the FALLOC_FL_KEEP_SIZE option is expected to not change the size
1219 // of the file, but it does. Simple strace report will show that.
1220 // While we work with Travis-CI team to figure out if this is a
1221 // quirk of Docker/AUFS, we will comment this out.
1222 struct stat file_stats;
1223 int result = fstat(fd_, &file_stats);
1224 // After ftruncate, we check whether ftruncate has the correct behavior.
1225 // If not, we should hack it with FALLOC_FL_PUNCH_HOLE
1226 if (result == 0 &&
1227 (file_stats.st_size + file_stats.st_blksize - 1) /
1228 file_stats.st_blksize !=
1229 file_stats.st_blocks / (file_stats.st_blksize / 512)) {
1230 IOSTATS_TIMER_GUARD(allocate_nanos);
1231 if (allow_fallocate_) {
1232 fallocate(fd_, FALLOC_FL_KEEP_SIZE | FALLOC_FL_PUNCH_HOLE, filesize_,
1233 block_size * last_allocated_block - filesize_);
1234 }
1235 }
1236 #endif
1237 }
1238
1239 if (close(fd_) < 0) {
1240 s = IOError("While closing file after writing", filename_, errno);
1241 }
1242 fd_ = -1;
1243 return s;
1244 }
1245
1246 // write out the cached data to the OS cache
Flush(const IOOptions &,IODebugContext *)1247 IOStatus PosixWritableFile::Flush(const IOOptions& /*opts*/,
1248 IODebugContext* /*dbg*/) {
1249 return IOStatus::OK();
1250 }
1251
Sync(const IOOptions &,IODebugContext *)1252 IOStatus PosixWritableFile::Sync(const IOOptions& /*opts*/,
1253 IODebugContext* /*dbg*/) {
1254 if (fdatasync(fd_) < 0) {
1255 return IOError("While fdatasync", filename_, errno);
1256 }
1257 return IOStatus::OK();
1258 }
1259
Fsync(const IOOptions &,IODebugContext *)1260 IOStatus PosixWritableFile::Fsync(const IOOptions& /*opts*/,
1261 IODebugContext* /*dbg*/) {
1262 if (fsync(fd_) < 0) {
1263 return IOError("While fsync", filename_, errno);
1264 }
1265 return IOStatus::OK();
1266 }
1267
IsSyncThreadSafe() const1268 bool PosixWritableFile::IsSyncThreadSafe() const { return true; }
1269
GetFileSize(const IOOptions &,IODebugContext *)1270 uint64_t PosixWritableFile::GetFileSize(const IOOptions& /*opts*/,
1271 IODebugContext* /*dbg*/) {
1272 return filesize_;
1273 }
1274
SetWriteLifeTimeHint(Env::WriteLifeTimeHint hint)1275 void PosixWritableFile::SetWriteLifeTimeHint(Env::WriteLifeTimeHint hint) {
1276 #ifdef OS_LINUX
1277 // Suppress Valgrind "Unimplemented functionality" error.
1278 #ifndef ROCKSDB_VALGRIND_RUN
1279 if (hint == write_hint_) {
1280 return;
1281 }
1282 if (fcntl(fd_, F_SET_RW_HINT, &hint) == 0) {
1283 write_hint_ = hint;
1284 }
1285 #else
1286 (void)hint;
1287 #endif // ROCKSDB_VALGRIND_RUN
1288 #else
1289 (void)hint;
1290 #endif // OS_LINUX
1291 }
1292
InvalidateCache(size_t offset,size_t length)1293 IOStatus PosixWritableFile::InvalidateCache(size_t offset, size_t length) {
1294 if (use_direct_io()) {
1295 return IOStatus::OK();
1296 }
1297 #ifndef OS_LINUX
1298 (void)offset;
1299 (void)length;
1300 return IOStatus::OK();
1301 #else
1302 // free OS pages
1303 int ret = Fadvise(fd_, offset, length, POSIX_FADV_DONTNEED);
1304 if (ret == 0) {
1305 return IOStatus::OK();
1306 }
1307 return IOError("While fadvise NotNeeded", filename_, errno);
1308 #endif
1309 }
1310
1311 #ifdef ROCKSDB_FALLOCATE_PRESENT
Allocate(uint64_t offset,uint64_t len,const IOOptions &,IODebugContext *)1312 IOStatus PosixWritableFile::Allocate(uint64_t offset, uint64_t len,
1313 const IOOptions& /*opts*/,
1314 IODebugContext* /*dbg*/) {
1315 assert(offset <= static_cast<uint64_t>(std::numeric_limits<off_t>::max()));
1316 assert(len <= static_cast<uint64_t>(std::numeric_limits<off_t>::max()));
1317 TEST_KILL_RANDOM("PosixWritableFile::Allocate:0", rocksdb_kill_odds);
1318 IOSTATS_TIMER_GUARD(allocate_nanos);
1319 int alloc_status = 0;
1320 if (allow_fallocate_) {
1321 alloc_status =
1322 fallocate(fd_, fallocate_with_keep_size_ ? FALLOC_FL_KEEP_SIZE : 0,
1323 static_cast<off_t>(offset), static_cast<off_t>(len));
1324 }
1325 if (alloc_status == 0) {
1326 return IOStatus::OK();
1327 } else {
1328 return IOError(
1329 "While fallocate offset " + ToString(offset) + " len " + ToString(len),
1330 filename_, errno);
1331 }
1332 }
1333 #endif
1334
RangeSync(uint64_t offset,uint64_t nbytes,const IOOptions & opts,IODebugContext * dbg)1335 IOStatus PosixWritableFile::RangeSync(uint64_t offset, uint64_t nbytes,
1336 const IOOptions& opts,
1337 IODebugContext* dbg) {
1338 #ifdef ROCKSDB_RANGESYNC_PRESENT
1339 assert(offset <= static_cast<uint64_t>(std::numeric_limits<off_t>::max()));
1340 assert(nbytes <= static_cast<uint64_t>(std::numeric_limits<off_t>::max()));
1341 if (sync_file_range_supported_) {
1342 int ret;
1343 if (strict_bytes_per_sync_) {
1344 // Specifying `SYNC_FILE_RANGE_WAIT_BEFORE` together with an offset/length
1345 // that spans all bytes written so far tells `sync_file_range` to wait for
1346 // any outstanding writeback requests to finish before issuing a new one.
1347 ret =
1348 sync_file_range(fd_, 0, static_cast<off_t>(offset + nbytes),
1349 SYNC_FILE_RANGE_WAIT_BEFORE | SYNC_FILE_RANGE_WRITE);
1350 } else {
1351 ret = sync_file_range(fd_, static_cast<off_t>(offset),
1352 static_cast<off_t>(nbytes), SYNC_FILE_RANGE_WRITE);
1353 }
1354 if (ret != 0) {
1355 return IOError("While sync_file_range returned " + ToString(ret),
1356 filename_, errno);
1357 }
1358 return IOStatus::OK();
1359 }
1360 #endif // ROCKSDB_RANGESYNC_PRESENT
1361 return FSWritableFile::RangeSync(offset, nbytes, opts, dbg);
1362 }
1363
1364 #ifdef OS_LINUX
GetUniqueId(char * id,size_t max_size) const1365 size_t PosixWritableFile::GetUniqueId(char* id, size_t max_size) const {
1366 return PosixHelper::GetUniqueIdFromFile(fd_, id, max_size);
1367 }
1368 #endif
1369
1370 /*
1371 * PosixRandomRWFile
1372 */
1373
PosixRandomRWFile(const std::string & fname,int fd,const EnvOptions &)1374 PosixRandomRWFile::PosixRandomRWFile(const std::string& fname, int fd,
1375 const EnvOptions& /*options*/)
1376 : filename_(fname), fd_(fd) {}
1377
~PosixRandomRWFile()1378 PosixRandomRWFile::~PosixRandomRWFile() {
1379 if (fd_ >= 0) {
1380 Close(IOOptions(), nullptr);
1381 }
1382 }
1383
Write(uint64_t offset,const Slice & data,const IOOptions &,IODebugContext *)1384 IOStatus PosixRandomRWFile::Write(uint64_t offset, const Slice& data,
1385 const IOOptions& /*opts*/,
1386 IODebugContext* /*dbg*/) {
1387 const char* src = data.data();
1388 size_t nbytes = data.size();
1389 if (!PosixPositionedWrite(fd_, src, nbytes, static_cast<off_t>(offset))) {
1390 return IOError(
1391 "While write random read/write file at offset " + ToString(offset),
1392 filename_, errno);
1393 }
1394
1395 return IOStatus::OK();
1396 }
1397
Read(uint64_t offset,size_t n,const IOOptions &,Slice * result,char * scratch,IODebugContext *) const1398 IOStatus PosixRandomRWFile::Read(uint64_t offset, size_t n,
1399 const IOOptions& /*opts*/, Slice* result,
1400 char* scratch, IODebugContext* /*dbg*/) const {
1401 size_t left = n;
1402 char* ptr = scratch;
1403 while (left > 0) {
1404 ssize_t done = pread(fd_, ptr, left, offset);
1405 if (done < 0) {
1406 // error while reading from file
1407 if (errno == EINTR) {
1408 // read was interrupted, try again.
1409 continue;
1410 }
1411 return IOError("While reading random read/write file offset " +
1412 ToString(offset) + " len " + ToString(n),
1413 filename_, errno);
1414 } else if (done == 0) {
1415 // Nothing more to read
1416 break;
1417 }
1418
1419 // Read `done` bytes
1420 ptr += done;
1421 offset += done;
1422 left -= done;
1423 }
1424
1425 *result = Slice(scratch, n - left);
1426 return IOStatus::OK();
1427 }
1428
Flush(const IOOptions &,IODebugContext *)1429 IOStatus PosixRandomRWFile::Flush(const IOOptions& /*opts*/,
1430 IODebugContext* /*dbg*/) {
1431 return IOStatus::OK();
1432 }
1433
Sync(const IOOptions &,IODebugContext *)1434 IOStatus PosixRandomRWFile::Sync(const IOOptions& /*opts*/,
1435 IODebugContext* /*dbg*/) {
1436 if (fdatasync(fd_) < 0) {
1437 return IOError("While fdatasync random read/write file", filename_, errno);
1438 }
1439 return IOStatus::OK();
1440 }
1441
Fsync(const IOOptions &,IODebugContext *)1442 IOStatus PosixRandomRWFile::Fsync(const IOOptions& /*opts*/,
1443 IODebugContext* /*dbg*/) {
1444 if (fsync(fd_) < 0) {
1445 return IOError("While fsync random read/write file", filename_, errno);
1446 }
1447 return IOStatus::OK();
1448 }
1449
Close(const IOOptions &,IODebugContext *)1450 IOStatus PosixRandomRWFile::Close(const IOOptions& /*opts*/,
1451 IODebugContext* /*dbg*/) {
1452 if (close(fd_) < 0) {
1453 return IOError("While close random read/write file", filename_, errno);
1454 }
1455 fd_ = -1;
1456 return IOStatus::OK();
1457 }
1458
~PosixMemoryMappedFileBuffer()1459 PosixMemoryMappedFileBuffer::~PosixMemoryMappedFileBuffer() {
1460 // TODO should have error handling though not much we can do...
1461 munmap(this->base_, length_);
1462 }
1463
1464 /*
1465 * PosixDirectory
1466 */
1467
~PosixDirectory()1468 PosixDirectory::~PosixDirectory() { close(fd_); }
1469
Fsync(const IOOptions &,IODebugContext *)1470 IOStatus PosixDirectory::Fsync(const IOOptions& /*opts*/,
1471 IODebugContext* /*dbg*/) {
1472 #ifndef OS_AIX
1473 if (fsync(fd_) == -1) {
1474 return IOError("While fsync", "a directory", errno);
1475 }
1476 #endif
1477 return IOStatus::OK();
1478 }
1479 } // namespace ROCKSDB_NAMESPACE
1480 #endif
1481