io_posix.cc 40.8 KB
Newer Older
1
//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
S
Siying Dong 已提交
2 3 4
//  This source code is licensed under both the GPLv2 (found in the
//  COPYING file in the root directory) and Apache 2.0 License
//  (found in the LICENSE.Apache file in the root directory).
5 6 7 8 9 10
//
// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file. See the AUTHORS file for names of contributors.

#ifdef ROCKSDB_LIB_IO_POSIX
11
#include "env/io_posix.h"
S
sdong 已提交
12 13
#include <errno.h>
#include <fcntl.h>
K
krad 已提交
14
#include <algorithm>
S
sdong 已提交
15 16
#if defined(OS_LINUX)
#include <linux/fs.h>
17
#ifndef FALLOC_FL_KEEP_SIZE
18
#include <linux/falloc.h>
S
sdong 已提交
19
#endif
20
#endif
S
sdong 已提交
21 22 23 24 25 26 27 28 29 30
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/ioctl.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <sys/types.h>
#ifdef OS_LINUX
#include <sys/statfs.h>
#include <sys/syscall.h>
31
#include <sys/sysmacros.h>
S
sdong 已提交
32
#endif
33
#include "monitoring/iostats_context_imp.h"
S
sdong 已提交
34 35
#include "port/port.h"
#include "rocksdb/slice.h"
36
#include "test_util/sync_point.h"
37
#include "util/autovector.h"
S
sdong 已提交
38 39
#include "util/coding.h"
#include "util/string_util.h"
40

S
Stream  
Shaohua Li 已提交
41 42
#if defined(OS_LINUX) && !defined(F_SET_RW_HINT)
#define F_LINUX_SPECIFIC_BASE 1024
43
#define F_SET_RW_HINT (F_LINUX_SPECIFIC_BASE + 12)
S
Stream  
Shaohua Li 已提交
44 45
#endif

46
namespace ROCKSDB_NAMESPACE {
47 48

// A wrapper for fadvise, if the platform doesn't support fadvise,
49
// it will simply return 0.
50 51 52 53
int Fadvise(int fd, off_t offset, size_t len, int advice) {
#ifdef OS_LINUX
  return posix_fadvise(fd, offset, len, advice);
#else
54 55 56 57
  (void)fd;
  (void)offset;
  (void)len;
  (void)advice;
58 59 60 61
  return 0;  // simply do nothing.
#endif
}

A
Aaron Gao 已提交
62
namespace {
63

64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114
// On MacOS (and probably *BSD), the posix write and pwrite calls do not support
// buffers larger than 2^31-1 bytes. These two wrappers fix this issue by
// cutting the buffer in 1GB chunks. We use this chunk size to be sure to keep
// the writes aligned.

bool PosixWrite(int fd, const char* buf, size_t nbyte) {
  const size_t kLimit1Gb = 1UL << 30;

  const char* src = buf;
  size_t left = nbyte;

  while (left != 0) {
    size_t bytes_to_write = std::min(left, kLimit1Gb);

    ssize_t done = write(fd, src, bytes_to_write);
    if (done < 0) {
      if (errno == EINTR) {
        continue;
      }
      return false;
    }
    left -= done;
    src += done;
  }
  return true;
}

bool PosixPositionedWrite(int fd, const char* buf, size_t nbyte, off_t offset) {
  const size_t kLimit1Gb = 1UL << 30;

  const char* src = buf;
  size_t left = nbyte;

  while (left != 0) {
    size_t bytes_to_write = std::min(left, kLimit1Gb);

    ssize_t done = pwrite(fd, src, bytes_to_write, offset);
    if (done < 0) {
      if (errno == EINTR) {
        continue;
      }
      return false;
    }
    left -= done;
    offset += done;
    src += done;
  }

  return true;
}

A
Aaron Gao 已提交
115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140
size_t GetLogicalBufferSize(int __attribute__((__unused__)) fd) {
#ifdef OS_LINUX
  struct stat buf;
  int result = fstat(fd, &buf);
  if (result == -1) {
    return kDefaultPageSize;
  }
  if (major(buf.st_dev) == 0) {
    // Unnamed devices (e.g. non-device mounts), reserved as null device number.
    // These don't have an entry in /sys/dev/block/. Return a sensible default.
    return kDefaultPageSize;
  }

  // Reading queue/logical_block_size does not require special permissions.
  const int kBufferSize = 100;
  char path[kBufferSize];
  char real_path[PATH_MAX + 1];
  snprintf(path, kBufferSize, "/sys/dev/block/%u:%u", major(buf.st_dev),
           minor(buf.st_dev));
  if (realpath(path, real_path) == nullptr) {
    return kDefaultPageSize;
  }
  std::string device_dir(real_path);
  if (!device_dir.empty() && device_dir.back() == '/') {
    device_dir.pop_back();
  }
141 142
  // NOTE: sda3 and nvme0n1p1 do not have a `queue/` subdir, only the parent sda
  // and nvme0n1 have it.
A
Aaron Gao 已提交
143 144 145
  // $ ls -al '/sys/dev/block/8:3'
  // lrwxrwxrwx. 1 root root 0 Jun 26 01:38 /sys/dev/block/8:3 ->
  // ../../block/sda/sda3
146 147 148
  // $ ls -al '/sys/dev/block/259:4'
  // lrwxrwxrwx 1 root root 0 Jan 31 16:04 /sys/dev/block/259:4 ->
  // ../../devices/pci0000:17/0000:17:00.0/0000:18:00.0/nvme/nvme0/nvme0n1/nvme0n1p1
A
Aaron Gao 已提交
149 150 151 152 153 154 155 156
  size_t parent_end = device_dir.rfind('/', device_dir.length() - 1);
  if (parent_end == std::string::npos) {
    return kDefaultPageSize;
  }
  size_t parent_begin = device_dir.rfind('/', parent_end - 1);
  if (parent_begin == std::string::npos) {
    return kDefaultPageSize;
  }
157 158 159 160 161
  std::string parent =
      device_dir.substr(parent_begin + 1, parent_end - parent_begin - 1);
  std::string child = device_dir.substr(parent_end + 1, std::string::npos);
  if (parent != "block" &&
      (child.compare(0, 4, "nvme") || child.find('p') != std::string::npos)) {
A
Aaron Gao 已提交
162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182
    device_dir = device_dir.substr(0, parent_end);
  }
  std::string fname = device_dir + "/queue/logical_block_size";
  FILE* fp;
  size_t size = 0;
  fp = fopen(fname.c_str(), "r");
  if (fp != nullptr) {
    char* line = nullptr;
    size_t len = 0;
    if (getline(&line, &len, fp) != -1) {
      sscanf(line, "%zu", &size);
    }
    free(line);
    fclose(fp);
  }
  if (size != 0 && (size & (size - 1)) == 0) {
    return size;
  }
#endif
  return kDefaultPageSize;
}
183 184 185 186 187 188 189 190 191

#ifdef ROCKSDB_RANGESYNC_PRESENT

#if !defined(ZFS_SUPER_MAGIC)
// The magic number for ZFS was not exposed until recently. It should be fixed
// forever so we can just copy the magic number here.
#define ZFS_SUPER_MAGIC 0x2fc12fc1
#endif

192 193 194 195 196 197
bool IsSyncFileRangeSupported(int fd) {
  // The approach taken in this function is to build a blacklist of cases where
  // we know `sync_file_range` definitely will not work properly despite passing
  // the compile-time check (`ROCKSDB_RANGESYNC_PRESENT`). If we are unsure, or
  // if any of the checks fail in unexpected ways, we allow `sync_file_range` to
  // be used. This way should minimize risk of impacting existing use cases.
198 199 200
  struct statfs buf;
  int ret = fstatfs(fd, &buf);
  assert(ret == 0);
201
  if (ret == 0 && buf.f_type == ZFS_SUPER_MAGIC) {
202 203 204 205 206 207
    // Testing on ZFS showed the writeback did not happen asynchronously when
    // `sync_file_range` was called, even though it returned success. Avoid it
    // and use `fdatasync` instead to preserve the contract of `bytes_per_sync`,
    // even though this'll incur extra I/O for metadata.
    return false;
  }
208 209 210 211 212 213 214 215 216 217 218 219

  ret = sync_file_range(fd, 0 /* offset */, 0 /* nbytes */, 0 /* flags */);
  assert(!(ret == -1 && errno != ENOSYS));
  if (ret == -1 && errno == ENOSYS) {
    // `sync_file_range` is not implemented on all platforms even if
    // compile-time checks pass and a supported filesystem is in-use. For
    // example, using ext4 on WSL (Windows Subsystem for Linux),
    // `sync_file_range()` returns `ENOSYS`
    // ("Function not implemented").
    return false;
  }
  // None of the cases on the blacklist matched, so allow `sync_file_range` use.
220 221 222 223 224 225 226 227
  return true;
}

#undef ZFS_SUPER_MAGIC

#endif  // ROCKSDB_RANGESYNC_PRESENT

}  // anonymous namespace
A
Aaron Gao 已提交
228

K
krad 已提交
229 230 231
/*
 * DirectIOHelper
 */
232
#ifndef NDEBUG
K
krad 已提交
233 234
namespace {

A
Aaron Gao 已提交
235 236 237
bool IsSectorAligned(const size_t off, size_t sector_size) {
  return off % sector_size == 0;
}
K
krad 已提交
238

239 240
bool IsSectorAligned(const void* ptr, size_t sector_size) {
  return uintptr_t(ptr) % sector_size == 0;
K
krad 已提交
241 242
}

243
}  // namespace
244
#endif
K
krad 已提交
245

246 247 248
/*
 * PosixSequentialFile
 */
A
Aaron Gao 已提交
249 250
PosixSequentialFile::PosixSequentialFile(const std::string& fname, FILE* file,
                                         int fd, const EnvOptions& options)
251
    : filename_(fname),
A
Aaron Gao 已提交
252 253
      file_(file),
      fd_(fd),
A
Aaron Gao 已提交
254 255
      use_direct_io_(options.use_direct_reads),
      logical_sector_size_(GetLogicalBufferSize(fd_)) {
A
Aaron Gao 已提交
256 257
  assert(!options.use_direct_reads || !options.use_mmap_reads);
}
258

A
Aaron Gao 已提交
259
PosixSequentialFile::~PosixSequentialFile() {
260
  if (!use_direct_io()) {
A
Aaron Gao 已提交
261 262 263 264 265 266 267
    assert(file_);
    fclose(file_);
  } else {
    assert(fd_);
    close(fd_);
  }
}
268

269 270 271
IOStatus PosixSequentialFile::Read(size_t n, const IOOptions& /*opts*/,
                                   Slice* result, char* scratch,
                                   IODebugContext* /*dbg*/) {
272
  assert(result != nullptr && !use_direct_io());
273
  IOStatus s;
274 275 276 277 278 279 280 281 282 283 284 285 286
  size_t r = 0;
  do {
    r = fread_unlocked(scratch, 1, n, file_);
  } while (r == 0 && ferror(file_) && errno == EINTR);
  *result = Slice(scratch, r);
  if (r < n) {
    if (feof(file_)) {
      // We leave status as ok if we hit the end of the file
      // We also clear the error so that the reads can continue
      // if a new data is written to the file
      clearerr(file_);
    } else {
      // A partial read with an error: return a non-ok status
287
      s = IOError("While reading file sequentially", filename_, errno);
288 289
    }
  }
A
Aaron Gao 已提交
290 291 292
  return s;
}

293 294 295 296
IOStatus PosixSequentialFile::PositionedRead(uint64_t offset, size_t n,
                                             const IOOptions& /*opts*/,
                                             Slice* result, char* scratch,
                                             IODebugContext* /*dbg*/) {
297 298 299 300 301
  assert(use_direct_io());
  assert(IsSectorAligned(offset, GetRequiredBufferAlignment()));
  assert(IsSectorAligned(n, GetRequiredBufferAlignment()));
  assert(IsSectorAligned(scratch, GetRequiredBufferAlignment()));

302
  IOStatus s;
A
Aaron Gao 已提交
303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321
  ssize_t r = -1;
  size_t left = n;
  char* ptr = scratch;
  while (left > 0) {
    r = pread(fd_, ptr, left, static_cast<off_t>(offset));
    if (r <= 0) {
      if (r == -1 && errno == EINTR) {
        continue;
      }
      break;
    }
    ptr += r;
    offset += r;
    left -= r;
    if (r % static_cast<ssize_t>(GetRequiredBufferAlignment()) != 0) {
      // Bytes reads don't fill sectors. Should only happen at the end
      // of the file.
      break;
    }
322
  }
A
Aaron Gao 已提交
323 324
  if (r < 0) {
    // An error: return a non-ok status
325 326 327
    s = IOError(
        "While pread " + ToString(n) + " bytes from offset " + ToString(offset),
        filename_, errno);
A
Aaron Gao 已提交
328 329
  }
  *result = Slice(scratch, (r < 0) ? 0 : n - left);
330 331 332
  return s;
}

333
IOStatus PosixSequentialFile::Skip(uint64_t n) {
334
  if (fseek(file_, static_cast<long int>(n), SEEK_CUR)) {
335 336
    return IOError("While fseek to skip " + ToString(n) + " bytes", filename_,
                   errno);
337
  }
338
  return IOStatus::OK();
339 340
}

341
IOStatus PosixSequentialFile::InvalidateCache(size_t offset, size_t length) {
342
#ifndef OS_LINUX
343 344
  (void)offset;
  (void)length;
345
  return IOStatus::OK();
346
#else
347
  if (!use_direct_io()) {
A
Aaron Gao 已提交
348 349 350
    // free OS pages
    int ret = Fadvise(fd_, offset, length, POSIX_FADV_DONTNEED);
    if (ret != 0) {
351 352 353
      return IOError("While fadvise NotNeeded offset " + ToString(offset) +
                         " len " + ToString(length),
                     filename_, errno);
A
Aaron Gao 已提交
354
    }
355
  }
356
  return IOStatus::OK();
A
Aaron Gao 已提交
357
#endif
K
krad 已提交
358 359 360 361 362
}

/*
 * PosixRandomAccessFile
 */
363
#if defined(OS_LINUX)
K
krad 已提交
364
size_t PosixHelper::GetUniqueIdFromFile(int fd, char* id, size_t max_size) {
365 366 367 368 369 370 371 372 373 374 375 376
  if (max_size < kMaxVarint64Length * 3) {
    return 0;
  }

  struct stat buf;
  int result = fstat(fd, &buf);
  if (result == -1) {
    return 0;
  }

  long version = 0;
  result = ioctl(fd, FS_IOC_GETVERSION, &version);
K
krad 已提交
377
  TEST_SYNC_POINT_CALLBACK("GetUniqueIdFromFile:FS_IOC_GETVERSION", &result);
378 379 380 381 382 383 384 385 386 387 388 389
  if (result == -1) {
    return 0;
  }
  uint64_t uversion = (uint64_t)version;

  char* rid = id;
  rid = EncodeVarint64(rid, buf.st_dev);
  rid = EncodeVarint64(rid, buf.st_ino);
  rid = EncodeVarint64(rid, uversion);
  assert(rid >= id);
  return static_cast<size_t>(rid - id);
}
K
krad 已提交
390 391
#endif

T
Tomas Kolda 已提交
392
#if defined(OS_MACOSX) || defined(OS_AIX)
K
krad 已提交
393
size_t PosixHelper::GetUniqueIdFromFile(int fd, char* id, size_t max_size) {
K
krad 已提交
394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409
  if (max_size < kMaxVarint64Length * 3) {
    return 0;
  }

  struct stat buf;
  int result = fstat(fd, &buf);
  if (result == -1) {
    return 0;
  }

  char* rid = id;
  rid = EncodeVarint64(rid, buf.st_dev);
  rid = EncodeVarint64(rid, buf.st_ino);
  rid = EncodeVarint64(rid, buf.st_gen);
  assert(rid >= id);
  return static_cast<size_t>(rid - id);
410 411 412 413 414 415 416
}
#endif
/*
 * PosixRandomAccessFile
 *
 * pread() based random-access
 */
417 418 419 420 421 422 423
PosixRandomAccessFile::PosixRandomAccessFile(
    const std::string& fname, int fd, const EnvOptions& options
#if defined(ROCKSDB_IOURING_PRESENT)
    ,
    ThreadLocalPtr* thread_local_io_urings
#endif
    )
A
Aaron Gao 已提交
424 425 426
    : filename_(fname),
      fd_(fd),
      use_direct_io_(options.use_direct_reads),
427 428 429 430 431 432
      logical_sector_size_(GetLogicalBufferSize(fd_))
#if defined(ROCKSDB_IOURING_PRESENT)
      ,
      thread_local_io_urings_(thread_local_io_urings)
#endif
{
A
Aaron Gao 已提交
433
  assert(!options.use_direct_reads || !options.use_mmap_reads);
434 435 436 437 438
  assert(!options.use_mmap_reads || sizeof(void*) < 8);
}

PosixRandomAccessFile::~PosixRandomAccessFile() { close(fd_); }

439 440 441 442
IOStatus PosixRandomAccessFile::Read(uint64_t offset, size_t n,
                                     const IOOptions& /*opts*/, Slice* result,
                                     char* scratch,
                                     IODebugContext* /*dbg*/) const {
443 444 445 446 447
  if (use_direct_io()) {
    assert(IsSectorAligned(offset, GetRequiredBufferAlignment()));
    assert(IsSectorAligned(n, GetRequiredBufferAlignment()));
    assert(IsSectorAligned(scratch, GetRequiredBufferAlignment()));
  }
448
  IOStatus s;
449 450 451 452 453 454
  ssize_t r = -1;
  size_t left = n;
  char* ptr = scratch;
  while (left > 0) {
    r = pread(fd_, ptr, left, static_cast<off_t>(offset));
    if (r <= 0) {
A
Aaron Gao 已提交
455
      if (r == -1 && errno == EINTR) {
456 457 458 459 460 461 462
        continue;
      }
      break;
    }
    ptr += r;
    offset += r;
    left -= r;
463
    if (use_direct_io() &&
A
Aaron Gao 已提交
464 465 466 467 468
        r % static_cast<ssize_t>(GetRequiredBufferAlignment()) != 0) {
      // Bytes reads don't fill sectors. Should only happen at the end
      // of the file.
      break;
    }
469 470 471
  }
  if (r < 0) {
    // An error: return a non-ok status
472 473 474
    s = IOError(
        "While pread offset " + ToString(offset) + " len " + ToString(n),
        filename_, errno);
475
  }
A
Aaron Gao 已提交
476
  *result = Slice(scratch, (r < 0) ? 0 : n - left);
477 478 479
  return s;
}

480 481 482 483
IOStatus PosixRandomAccessFile::MultiRead(FSReadRequest* reqs,
                                          size_t num_reqs,
                                          const IOOptions& options,
                                          IODebugContext* dbg) {
484 485 486 487 488 489 490 491 492 493 494 495 496 497 498
#if defined(ROCKSDB_IOURING_PRESENT)
  struct io_uring* iu = nullptr;
  if (thread_local_io_urings_) {
    iu = static_cast<struct io_uring*>(thread_local_io_urings_->Get());
    if (iu == nullptr) {
      iu = CreateIOUring();
      if (iu != nullptr) {
        thread_local_io_urings_->Reset(iu);
      }
    }
  }

  // Init failed, platform doesn't support io_uring. Fall back to
  // serialized reads
  if (iu == nullptr) {
499
    return FSRandomAccessFile::MultiRead(reqs, num_reqs, options, dbg);
500 501 502
  }

  struct WrappedReadRequest {
503
    FSReadRequest* req;
504
    struct iovec iov;
S
sdong 已提交
505 506
    size_t finished_len;
    explicit WrappedReadRequest(FSReadRequest* r) : req(r), finished_len(0) {}
507 508 509
  };

  autovector<WrappedReadRequest, 32> req_wraps;
S
sdong 已提交
510
  autovector<WrappedReadRequest*, 4> incomplete_rq_list;
511 512 513 514 515

  for (size_t i = 0; i < num_reqs; i++) {
    req_wraps.emplace_back(&reqs[i]);
  }

S
sdong 已提交
516 517 518
  size_t reqs_off = 0;
  while (num_reqs > reqs_off || !incomplete_rq_list.empty()) {
    size_t this_reqs = (num_reqs - reqs_off) + incomplete_rq_list.size();
519 520 521 522

    // If requests exceed depth, split it into batches
    if (this_reqs > kIoUringDepth) this_reqs = kIoUringDepth;

S
sdong 已提交
523
    assert(incomplete_rq_list.size() <= this_reqs);
524
    for (size_t i = 0; i < this_reqs; i++) {
S
sdong 已提交
525 526 527 528 529 530 531 532 533 534 535
      WrappedReadRequest* rep_to_submit;
      if (i < incomplete_rq_list.size()) {
        rep_to_submit = incomplete_rq_list[i];
      } else {
        rep_to_submit = &req_wraps[reqs_off++];
      }
      assert(rep_to_submit->req->len > rep_to_submit->finished_len);
      rep_to_submit->iov.iov_base =
          rep_to_submit->req->scratch + rep_to_submit->finished_len;
      rep_to_submit->iov.iov_len =
          rep_to_submit->req->len - rep_to_submit->finished_len;
536

S
sdong 已提交
537
      struct io_uring_sqe* sqe;
538
      sqe = io_uring_get_sqe(iu);
S
sdong 已提交
539 540 541 542
      io_uring_prep_readv(
          sqe, fd_, &rep_to_submit->iov, 1,
          rep_to_submit->req->offset + rep_to_submit->finished_len);
      io_uring_sqe_set_data(sqe, rep_to_submit);
543
    }
S
sdong 已提交
544
    incomplete_rq_list.clear();
545

S
sdong 已提交
546 547
    ssize_t ret =
        io_uring_submit_and_wait(iu, static_cast<unsigned int>(this_reqs));
548 549 550 551 552 553 554 555 556 557 558 559 560
    if (static_cast<size_t>(ret) != this_reqs) {
      fprintf(stderr, "ret = %ld this_reqs: %ld\n", (long)ret, (long)this_reqs);
    }
    assert(static_cast<size_t>(ret) == this_reqs);

    for (size_t i = 0; i < this_reqs; i++) {
      struct io_uring_cqe* cqe;
      WrappedReadRequest* req_wrap;

      // We could use the peek variant here, but this seems safer in terms
      // of our initial wait not reaping all completions
      ret = io_uring_wait_cqe(iu, &cqe);
      assert(!ret);
S
sdong 已提交
561

562
      req_wrap = static_cast<WrappedReadRequest*>(io_uring_cqe_get_data(cqe));
563
      FSReadRequest* req = req_wrap->req;
S
sdong 已提交
564
      if (cqe->res < 0) {
565 566
        req->result = Slice(req->scratch, 0);
        req->status = IOError("Req failed", filename_, cqe->res);
S
sdong 已提交
567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595
      } else {
        size_t bytes_read = static_cast<size_t>(cqe->res);
        TEST_SYNC_POINT_CALLBACK(
            "PosixRandomAccessFile::MultiRead:io_uring_result", &bytes_read);
        if (bytes_read == req_wrap->iov.iov_len) {
          req->result = Slice(req->scratch, req->len);
          req->status = IOStatus::OK();
        } else if (bytes_read == 0) {
          // cqe->res == 0 can means EOF, or can mean partial results. See
          // comment
          // https://github.com/facebook/rocksdb/pull/6441#issuecomment-589843435
          // Fall back to pread in this case.
          Slice tmp_slice;
          req->status =
              Read(req->offset + req_wrap->finished_len,
                   req->len - req_wrap->finished_len, options, &tmp_slice,
                   req->scratch + req_wrap->finished_len, dbg);
          req->result =
              Slice(req->scratch, req_wrap->finished_len + tmp_slice.size());
        } else if (bytes_read < req_wrap->iov.iov_len) {
          assert(bytes_read > 0);
          assert(bytes_read + req_wrap->finished_len < req->len);
          req_wrap->finished_len += bytes_read;
          incomplete_rq_list.push_back(req_wrap);
        } else {
          req->result = Slice(req->scratch, 0);
          req->status = IOError("Req returned more bytes than requested",
                                filename_, cqe->res);
        }
596 597 598 599
      }
      io_uring_cqe_seen(iu, cqe);
    }
  }
600
  return IOStatus::OK();
601
#else
602
  return FSRandomAccessFile::MultiRead(reqs, num_reqs, options, dbg);
603 604 605
#endif
}

606 607 608 609
IOStatus PosixRandomAccessFile::Prefetch(uint64_t offset, size_t n,
                                         const IOOptions& /*opts*/,
                                         IODebugContext* /*dbg*/) {
  IOStatus s;
610 611 612 613 614 615 616 617 618 619 620 621
  if (!use_direct_io()) {
    ssize_t r = 0;
#ifdef OS_LINUX
    r = readahead(fd_, offset, n);
#endif
#ifdef OS_MACOSX
    radvisory advice;
    advice.ra_offset = static_cast<off_t>(offset);
    advice.ra_count = static_cast<int>(n);
    r = fcntl(fd_, F_RDADVISE, &advice);
#endif
    if (r == -1) {
622 623 624
      s = IOError("While prefetching offset " + ToString(offset) + " len " +
                      ToString(n),
                  filename_, errno);
625 626 627 628 629
    }
  }
  return s;
}

T
Tomas Kolda 已提交
630
#if defined(OS_LINUX) || defined(OS_MACOSX) || defined(OS_AIX)
631
size_t PosixRandomAccessFile::GetUniqueId(char* id, size_t max_size) const {
K
krad 已提交
632
  return PosixHelper::GetUniqueIdFromFile(fd_, id, max_size);
633 634 635 636
}
#endif

void PosixRandomAccessFile::Hint(AccessPattern pattern) {
637
  if (use_direct_io()) {
A
Aaron Gao 已提交
638 639
    return;
  }
640
  switch (pattern) {
641
    case kNormal:
642 643
      Fadvise(fd_, 0, 0, POSIX_FADV_NORMAL);
      break;
644
    case kRandom:
645 646
      Fadvise(fd_, 0, 0, POSIX_FADV_RANDOM);
      break;
647
    case kSequential:
648 649
      Fadvise(fd_, 0, 0, POSIX_FADV_SEQUENTIAL);
      break;
650
    case kWillNeed:
651 652
      Fadvise(fd_, 0, 0, POSIX_FADV_WILLNEED);
      break;
653
    case kWontNeed:
654 655 656 657 658 659 660 661
      Fadvise(fd_, 0, 0, POSIX_FADV_DONTNEED);
      break;
    default:
      assert(false);
      break;
  }
}

662
IOStatus PosixRandomAccessFile::InvalidateCache(size_t offset, size_t length) {
663
  if (use_direct_io()) {
664
    return IOStatus::OK();
A
Aaron Gao 已提交
665
  }
666
#ifndef OS_LINUX
667 668
  (void)offset;
  (void)length;
669
  return IOStatus::OK();
670 671 672 673
#else
  // free OS pages
  int ret = Fadvise(fd_, offset, length, POSIX_FADV_DONTNEED);
  if (ret == 0) {
674
    return IOStatus::OK();
675
  }
676 677 678
  return IOError("While fadvise NotNeeded offset " + ToString(offset) +
                     " len " + ToString(length),
                 filename_, errno);
679 680 681 682 683 684 685 686 687 688 689 690 691 692
#endif
}

/*
 * PosixMmapReadableFile
 *
 * mmap() based random-access
 */
// base[0,length-1] contains the mmapped contents of the file.
PosixMmapReadableFile::PosixMmapReadableFile(const int fd,
                                             const std::string& fname,
                                             void* base, size_t length,
                                             const EnvOptions& options)
    : fd_(fd), filename_(fname), mmapped_region_(base), length_(length) {
693 694 695
#ifdef NDEBUG
  (void)options;
#endif
696 697
  fd_ = fd_ + 0;  // suppress the warning for used variables
  assert(options.use_mmap_reads);
A
Aaron Gao 已提交
698
  assert(!options.use_direct_reads);
699 700 701 702 703 704 705 706
}

PosixMmapReadableFile::~PosixMmapReadableFile() {
  int ret = munmap(mmapped_region_, length_);
  if (ret != 0) {
    fprintf(stdout, "failed to munmap %p length %" ROCKSDB_PRIszt " \n",
            mmapped_region_, length_);
  }
707
  close(fd_);
708 709
}

710 711 712 713 714
IOStatus PosixMmapReadableFile::Read(uint64_t offset, size_t n,
                                     const IOOptions& /*opts*/, Slice* result,
                                     char* /*scratch*/,
                                     IODebugContext* /*dbg*/) const {
  IOStatus s;
715 716
  if (offset > length_) {
    *result = Slice();
717 718 719
    return IOError("While mmap read offset " + ToString(offset) +
                       " larger than file length " + ToString(length_),
                   filename_, EINVAL);
720 721 722 723 724 725 726
  } else if (offset + n > length_) {
    n = static_cast<size_t>(length_ - offset);
  }
  *result = Slice(reinterpret_cast<char*>(mmapped_region_) + offset, n);
  return s;
}

727
IOStatus PosixMmapReadableFile::InvalidateCache(size_t offset, size_t length) {
728
#ifndef OS_LINUX
729 730
  (void)offset;
  (void)length;
731
  return IOStatus::OK();
732 733 734 735
#else
  // free OS pages
  int ret = Fadvise(fd_, offset, length, POSIX_FADV_DONTNEED);
  if (ret == 0) {
736
    return IOStatus::OK();
737
  }
738 739 740
  return IOError("While fadvise not needed. Offset " + ToString(offset) +
                     " len" + ToString(length),
                 filename_, errno);
741 742 743 744 745 746 747 748 749 750 751
#endif
}

/*
 * PosixMmapFile
 *
 * We preallocate up to an extra megabyte and use memcpy to append new
 * data to the file.  This is safe since we either properly close the
 * file before reading from it, or for log files, the reading code
 * knows enough to skip zero suffixes.
 */
752
IOStatus PosixMmapFile::UnmapCurrentRegion() {
753 754 755 756
  TEST_KILL_RANDOM("PosixMmapFile::UnmapCurrentRegion:0", rocksdb_kill_odds);
  if (base_ != nullptr) {
    int munmap_status = munmap(base_, limit_ - base_);
    if (munmap_status != 0) {
757
      return IOError("While munmap", filename_, munmap_status);
758 759 760 761 762 763 764 765 766 767 768 769
    }
    file_offset_ += limit_ - base_;
    base_ = nullptr;
    limit_ = nullptr;
    last_sync_ = nullptr;
    dst_ = nullptr;

    // Increase the amount we map the next time, but capped at 1MB
    if (map_size_ < (1 << 20)) {
      map_size_ *= 2;
    }
  }
770
  return IOStatus::OK();
771 772
}

773
IOStatus PosixMmapFile::MapNewRegion() {
774 775 776 777 778 779 780 781 782 783 784 785
#ifdef ROCKSDB_FALLOCATE_PRESENT
  assert(base_ == nullptr);
  TEST_KILL_RANDOM("PosixMmapFile::UnmapCurrentRegion:0", rocksdb_kill_odds);
  // we can't fallocate with FALLOC_FL_KEEP_SIZE here
  if (allow_fallocate_) {
    IOSTATS_TIMER_GUARD(allocate_nanos);
    int alloc_status = fallocate(fd_, 0, file_offset_, map_size_);
    if (alloc_status != 0) {
      // fallback to posix_fallocate
      alloc_status = posix_fallocate(fd_, file_offset_, map_size_);
    }
    if (alloc_status != 0) {
786 787
      return IOStatus::IOError("Error allocating space to file : " + filename_ +
                               "Error : " + strerror(alloc_status));
788 789 790 791 792 793 794
    }
  }

  TEST_KILL_RANDOM("PosixMmapFile::Append:1", rocksdb_kill_odds);
  void* ptr = mmap(nullptr, map_size_, PROT_READ | PROT_WRITE, MAP_SHARED, fd_,
                   file_offset_);
  if (ptr == MAP_FAILED) {
795
    return IOStatus::IOError("MMap failed on " + filename_);
796 797 798 799 800 801 802
  }
  TEST_KILL_RANDOM("PosixMmapFile::Append:2", rocksdb_kill_odds);

  base_ = reinterpret_cast<char*>(ptr);
  limit_ = base_ + map_size_;
  dst_ = base_;
  last_sync_ = base_;
803
  return IOStatus::OK();
804
#else
805
  return IOStatus::NotSupported("This platform doesn't support fallocate()");
806 807 808
#endif
}

809
IOStatus PosixMmapFile::Msync() {
810
  if (dst_ == last_sync_) {
811
    return IOStatus::OK();
812 813 814 815 816 817 818 819
  }
  // Find the beginnings of the pages that contain the first and last
  // bytes to be synced.
  size_t p1 = TruncateToPageBoundary(last_sync_ - base_);
  size_t p2 = TruncateToPageBoundary(dst_ - base_ - 1);
  last_sync_ = dst_;
  TEST_KILL_RANDOM("PosixMmapFile::Msync:0", rocksdb_kill_odds);
  if (msync(base_ + p1, p2 - p1 + page_size_, MS_SYNC) < 0) {
820
    return IOError("While msync", filename_, errno);
821
  }
822
  return IOStatus::OK();
823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838
}

PosixMmapFile::PosixMmapFile(const std::string& fname, int fd, size_t page_size,
                             const EnvOptions& options)
    : filename_(fname),
      fd_(fd),
      page_size_(page_size),
      map_size_(Roundup(65536, page_size)),
      base_(nullptr),
      limit_(nullptr),
      dst_(nullptr),
      last_sync_(nullptr),
      file_offset_(0) {
#ifdef ROCKSDB_FALLOCATE_PRESENT
  allow_fallocate_ = options.allow_fallocate;
  fallocate_with_keep_size_ = options.fallocate_with_keep_size;
839 840
#else
  (void)options;
841 842 843
#endif
  assert((page_size & (page_size - 1)) == 0);
  assert(options.use_mmap_writes);
A
Aaron Gao 已提交
844
  assert(!options.use_direct_writes);
845 846 847 848
}

PosixMmapFile::~PosixMmapFile() {
  if (fd_ >= 0) {
849
    PosixMmapFile::Close(IOOptions(), nullptr);
850 851 852
  }
}

853 854
IOStatus PosixMmapFile::Append(const Slice& data, const IOOptions& /*opts*/,
                               IODebugContext* /*dbg*/) {
855 856 857 858 859 860 861
  const char* src = data.data();
  size_t left = data.size();
  while (left > 0) {
    assert(base_ <= dst_);
    assert(dst_ <= limit_);
    size_t avail = limit_ - dst_;
    if (avail == 0) {
862
      IOStatus s = UnmapCurrentRegion();
863 864 865 866 867 868 869 870 871 872 873
      if (!s.ok()) {
        return s;
      }
      s = MapNewRegion();
      if (!s.ok()) {
        return s;
      }
      TEST_KILL_RANDOM("PosixMmapFile::Append:0", rocksdb_kill_odds);
    }

    size_t n = (left <= avail) ? left : avail;
874
    assert(dst_);
875 876 877 878 879
    memcpy(dst_, src, n);
    dst_ += n;
    src += n;
    left -= n;
  }
880
  return IOStatus::OK();
881 882
}

883 884 885
IOStatus PosixMmapFile::Close(const IOOptions& /*opts*/,
                              IODebugContext* /*dbg*/) {
  IOStatus s;
886 887 888 889
  size_t unused = limit_ - dst_;

  s = UnmapCurrentRegion();
  if (!s.ok()) {
890
    s = IOError("While closing mmapped file", filename_, errno);
891 892 893
  } else if (unused > 0) {
    // Trim the extra space at the end of the file
    if (ftruncate(fd_, file_offset_ - unused) < 0) {
894
      s = IOError("While ftruncating mmaped file", filename_, errno);
895 896 897 898 899
    }
  }

  if (close(fd_) < 0) {
    if (s.ok()) {
900
      s = IOError("While closing mmapped file", filename_, errno);
901 902 903 904 905 906 907 908 909
    }
  }

  fd_ = -1;
  base_ = nullptr;
  limit_ = nullptr;
  return s;
}

910 911 912 913
IOStatus PosixMmapFile::Flush(const IOOptions& /*opts*/,
                              IODebugContext* /*dbg*/) {
  return IOStatus::OK();
}
914

915 916
IOStatus PosixMmapFile::Sync(const IOOptions& /*opts*/,
                             IODebugContext* /*dbg*/) {
917
  if (fdatasync(fd_) < 0) {
918
    return IOError("While fdatasync mmapped file", filename_, errno);
919 920 921 922 923 924 925 926
  }

  return Msync();
}

/**
 * Flush data as well as metadata to stable storage.
 */
927 928
IOStatus PosixMmapFile::Fsync(const IOOptions& /*opts*/,
                              IODebugContext* /*dbg*/) {
929
  if (fsync(fd_) < 0) {
930
    return IOError("While fsync mmaped file", filename_, errno);
931 932 933 934 935 936 937 938 939 940
  }

  return Msync();
}

/**
 * Get the size of valid data in the file. This will not match the
 * size that is returned from the filesystem because we use mmap
 * to extend file by map_size every time.
 */
941 942
uint64_t PosixMmapFile::GetFileSize(const IOOptions& /*opts*/,
                                    IODebugContext* /*dbg*/) {
943 944 945 946
  size_t used = dst_ - base_;
  return file_offset_ + used;
}

947
IOStatus PosixMmapFile::InvalidateCache(size_t offset, size_t length) {
948
#ifndef OS_LINUX
949 950
  (void)offset;
  (void)length;
951
  return IOStatus::OK();
952 953 954 955
#else
  // free OS pages
  int ret = Fadvise(fd_, offset, length, POSIX_FADV_DONTNEED);
  if (ret == 0) {
956
    return IOStatus::OK();
957
  }
958
  return IOError("While fadvise NotNeeded mmapped file", filename_, errno);
959 960 961
#endif
}

962
#ifdef ROCKSDB_FALLOCATE_PRESENT
963 964 965
IOStatus PosixMmapFile::Allocate(uint64_t offset, uint64_t len,
                                 const IOOptions& /*opts*/,
                                 IODebugContext* /*dbg*/) {
Y
Yi Wu 已提交
966 967
  assert(offset <= static_cast<uint64_t>(std::numeric_limits<off_t>::max()));
  assert(len <= static_cast<uint64_t>(std::numeric_limits<off_t>::max()));
968 969 970
  TEST_KILL_RANDOM("PosixMmapFile::Allocate:0", rocksdb_kill_odds);
  int alloc_status = 0;
  if (allow_fallocate_) {
971 972 973
    alloc_status =
        fallocate(fd_, fallocate_with_keep_size_ ? FALLOC_FL_KEEP_SIZE : 0,
                  static_cast<off_t>(offset), static_cast<off_t>(len));
974 975
  }
  if (alloc_status == 0) {
976
    return IOStatus::OK();
977
  } else {
978 979 980
    return IOError(
        "While fallocate offset " + ToString(offset) + " len " + ToString(len),
        filename_, errno);
981
  }
982
}
983
#endif
984 985 986 987 988 989 990 991

/*
 * PosixWritableFile
 *
 * Use posix write to write data to a file.
 */
PosixWritableFile::PosixWritableFile(const std::string& fname, int fd,
                                     const EnvOptions& options)
992
    : FSWritableFile(options),
993
      filename_(fname),
994
      use_direct_io_(options.use_direct_writes),
A
Aaron Gao 已提交
995
      fd_(fd),
A
Aaron Gao 已提交
996 997
      filesize_(0),
      logical_sector_size_(GetLogicalBufferSize(fd_)) {
998 999 1000 1001
#ifdef ROCKSDB_FALLOCATE_PRESENT
  allow_fallocate_ = options.allow_fallocate;
  fallocate_with_keep_size_ = options.fallocate_with_keep_size;
#endif
1002 1003 1004
#ifdef ROCKSDB_RANGESYNC_PRESENT
  sync_file_range_supported_ = IsSyncFileRangeSupported(fd_);
#endif  // ROCKSDB_RANGESYNC_PRESENT
1005 1006 1007 1008 1009
  assert(!options.use_mmap_writes);
}

PosixWritableFile::~PosixWritableFile() {
  if (fd_ >= 0) {
1010
    PosixWritableFile::Close(IOOptions(), nullptr);
1011 1012 1013
  }
}

1014 1015
IOStatus PosixWritableFile::Append(const Slice& data, const IOOptions& /*opts*/,
                                   IODebugContext* /*dbg*/) {
1016 1017 1018 1019
  if (use_direct_io()) {
    assert(IsSectorAligned(data.size(), GetRequiredBufferAlignment()));
    assert(IsSectorAligned(data.data(), GetRequiredBufferAlignment()));
  }
1020
  const char* src = data.data();
1021 1022 1023 1024
  size_t nbytes = data.size();

  if (!PosixWrite(fd_, src, nbytes)) {
    return IOError("While appending to file", filename_, errno);
1025
  }
1026 1027

  filesize_ += nbytes;
1028
  return IOStatus::OK();
1029 1030
}

1031 1032 1033
IOStatus PosixWritableFile::PositionedAppend(const Slice& data, uint64_t offset,
                                             const IOOptions& /*opts*/,
                                             IODebugContext* /*dbg*/) {
1034 1035 1036 1037 1038
  if (use_direct_io()) {
    assert(IsSectorAligned(offset, GetRequiredBufferAlignment()));
    assert(IsSectorAligned(data.size(), GetRequiredBufferAlignment()));
    assert(IsSectorAligned(data.data(), GetRequiredBufferAlignment()));
  }
Y
Yi Wu 已提交
1039
  assert(offset <= static_cast<uint64_t>(std::numeric_limits<off_t>::max()));
1040
  const char* src = data.data();
1041 1042 1043 1044
  size_t nbytes = data.size();
  if (!PosixPositionedWrite(fd_, src, nbytes, static_cast<off_t>(offset))) {
    return IOError("While pwrite to file at offset " + ToString(offset),
                   filename_, errno);
1045
  }
1046
  filesize_ = offset + nbytes;
1047
  return IOStatus::OK();
1048 1049
}

1050 1051 1052
IOStatus PosixWritableFile::Truncate(uint64_t size, const IOOptions& /*opts*/,
                                     IODebugContext* /*dbg*/) {
  IOStatus s;
A
Aaron Gao 已提交
1053 1054
  int r = ftruncate(fd_, size);
  if (r < 0) {
1055 1056
    s = IOError("While ftruncate file to size " + ToString(size), filename_,
                errno);
A
Aaron Gao 已提交
1057 1058 1059 1060 1061 1062
  } else {
    filesize_ = size;
  }
  return s;
}

1063 1064 1065
IOStatus PosixWritableFile::Close(const IOOptions& /*opts*/,
                                  IODebugContext* /*dbg*/) {
  IOStatus s;
1066 1067 1068 1069 1070 1071 1072 1073

  size_t block_size;
  size_t last_allocated_block;
  GetPreallocationStatus(&block_size, &last_allocated_block);
  if (last_allocated_block > 0) {
    // trim the extra space preallocated at the end of the file
    // NOTE(ljin): we probably don't want to surface failure as an IOError,
    // but it will be nice to log these errors.
T
Tamir Duberstein 已提交
1074
    int dummy __attribute__((__unused__));
1075
    dummy = ftruncate(fd_, filesize_);
1076
#if defined(ROCKSDB_FALLOCATE_PRESENT) && defined(FALLOC_FL_PUNCH_HOLE) && \
1077
    !defined(TRAVIS)
1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093
    // in some file systems, ftruncate only trims trailing space if the
    // new file size is smaller than the current size. Calling fallocate
    // with FALLOC_FL_PUNCH_HOLE flag to explicitly release these unused
    // blocks. FALLOC_FL_PUNCH_HOLE is supported on at least the following
    // filesystems:
    //   XFS (since Linux 2.6.38)
    //   ext4 (since Linux 3.0)
    //   Btrfs (since Linux 3.7)
    //   tmpfs (since Linux 3.5)
    // We ignore error since failure of this operation does not affect
    // correctness.
    // TRAVIS - this code does not work on TRAVIS filesystems.
    // the FALLOC_FL_KEEP_SIZE option is expected to not change the size
    // of the file, but it does. Simple strace report will show that.
    // While we work with Travis-CI team to figure out if this is a
    // quirk of Docker/AUFS, we will comment this out.
1094
    struct stat file_stats;
1095
    int result = fstat(fd_, &file_stats);
1096 1097
    // After ftruncate, we check whether ftruncate has the correct behavior.
    // If not, we should hack it with FALLOC_FL_PUNCH_HOLE
1098 1099
    if (result == 0 &&
        (file_stats.st_size + file_stats.st_blksize - 1) /
1100 1101
                file_stats.st_blksize !=
            file_stats.st_blocks / (file_stats.st_blksize / 512)) {
1102 1103 1104 1105 1106
      IOSTATS_TIMER_GUARD(allocate_nanos);
      if (allow_fallocate_) {
        fallocate(fd_, FALLOC_FL_KEEP_SIZE | FALLOC_FL_PUNCH_HOLE, filesize_,
                  block_size * last_allocated_block - filesize_);
      }
1107 1108
    }
#endif
1109 1110 1111
  }

  if (close(fd_) < 0) {
1112
    s = IOError("While closing file after writing", filename_, errno);
1113 1114 1115 1116 1117 1118
  }
  fd_ = -1;
  return s;
}

// write out the cached data to the OS cache
1119 1120 1121 1122
IOStatus PosixWritableFile::Flush(const IOOptions& /*opts*/,
                                  IODebugContext* /*dbg*/) {
  return IOStatus::OK();
}
1123

1124 1125
IOStatus PosixWritableFile::Sync(const IOOptions& /*opts*/,
                                 IODebugContext* /*dbg*/) {
1126
  if (fdatasync(fd_) < 0) {
1127
    return IOError("While fdatasync", filename_, errno);
1128
  }
1129
  return IOStatus::OK();
1130 1131
}

1132 1133
IOStatus PosixWritableFile::Fsync(const IOOptions& /*opts*/,
                                  IODebugContext* /*dbg*/) {
1134
  if (fsync(fd_) < 0) {
1135
    return IOError("While fsync", filename_, errno);
1136
  }
1137
  return IOStatus::OK();
1138 1139 1140 1141
}

bool PosixWritableFile::IsSyncThreadSafe() const { return true; }

1142 1143 1144 1145
uint64_t PosixWritableFile::GetFileSize(const IOOptions& /*opts*/,
                                        IODebugContext* /*dbg*/) {
  return filesize_;
}
1146

S
Stream  
Shaohua Li 已提交
1147 1148
void PosixWritableFile::SetWriteLifeTimeHint(Env::WriteLifeTimeHint hint) {
#ifdef OS_LINUX
1149 1150
// Suppress Valgrind "Unimplemented functionality" error.
#ifndef ROCKSDB_VALGRIND_RUN
S
Stream  
Shaohua Li 已提交
1151 1152 1153 1154 1155 1156
  if (hint == write_hint_) {
    return;
  }
  if (fcntl(fd_, F_SET_RW_HINT, &hint) == 0) {
    write_hint_ = hint;
  }
1157 1158
#else
  (void)hint;
1159
#endif  // ROCKSDB_VALGRIND_RUN
1160 1161
#else
  (void)hint;
1162
#endif  // OS_LINUX
S
Stream  
Shaohua Li 已提交
1163 1164
}

1165
IOStatus PosixWritableFile::InvalidateCache(size_t offset, size_t length) {
1166
  if (use_direct_io()) {
1167
    return IOStatus::OK();
A
Aaron Gao 已提交
1168
  }
1169
#ifndef OS_LINUX
1170 1171
  (void)offset;
  (void)length;
1172
  return IOStatus::OK();
1173 1174 1175 1176
#else
  // free OS pages
  int ret = Fadvise(fd_, offset, length, POSIX_FADV_DONTNEED);
  if (ret == 0) {
1177
    return IOStatus::OK();
1178
  }
1179
  return IOError("While fadvise NotNeeded", filename_, errno);
1180 1181 1182
#endif
}

1183
#ifdef ROCKSDB_FALLOCATE_PRESENT
1184 1185 1186
IOStatus PosixWritableFile::Allocate(uint64_t offset, uint64_t len,
                                     const IOOptions& /*opts*/,
                                     IODebugContext* /*dbg*/) {
Y
Yi Wu 已提交
1187 1188
  assert(offset <= static_cast<uint64_t>(std::numeric_limits<off_t>::max()));
  assert(len <= static_cast<uint64_t>(std::numeric_limits<off_t>::max()));
1189 1190 1191 1192
  TEST_KILL_RANDOM("PosixWritableFile::Allocate:0", rocksdb_kill_odds);
  IOSTATS_TIMER_GUARD(allocate_nanos);
  int alloc_status = 0;
  if (allow_fallocate_) {
1193 1194 1195
    alloc_status =
        fallocate(fd_, fallocate_with_keep_size_ ? FALLOC_FL_KEEP_SIZE : 0,
                  static_cast<off_t>(offset), static_cast<off_t>(len));
1196 1197
  }
  if (alloc_status == 0) {
1198
    return IOStatus::OK();
1199
  } else {
1200 1201 1202
    return IOError(
        "While fallocate offset " + ToString(offset) + " len " + ToString(len),
        filename_, errno);
1203 1204
  }
}
1205
#endif
1206

1207 1208 1209
IOStatus PosixWritableFile::RangeSync(uint64_t offset, uint64_t nbytes,
                                      const IOOptions& opts,
                                      IODebugContext* dbg) {
1210
#ifdef ROCKSDB_RANGESYNC_PRESENT
Y
Yi Wu 已提交
1211 1212
  assert(offset <= static_cast<uint64_t>(std::numeric_limits<off_t>::max()));
  assert(nbytes <= static_cast<uint64_t>(std::numeric_limits<off_t>::max()));
1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229
  if (sync_file_range_supported_) {
    int ret;
    if (strict_bytes_per_sync_) {
      // Specifying `SYNC_FILE_RANGE_WAIT_BEFORE` together with an offset/length
      // that spans all bytes written so far tells `sync_file_range` to wait for
      // any outstanding writeback requests to finish before issuing a new one.
      ret =
          sync_file_range(fd_, 0, static_cast<off_t>(offset + nbytes),
                          SYNC_FILE_RANGE_WAIT_BEFORE | SYNC_FILE_RANGE_WRITE);
    } else {
      ret = sync_file_range(fd_, static_cast<off_t>(offset),
                            static_cast<off_t>(nbytes), SYNC_FILE_RANGE_WRITE);
    }
    if (ret != 0) {
      return IOError("While sync_file_range returned " + ToString(ret),
                     filename_, errno);
    }
1230
    return IOStatus::OK();
1231
  }
1232
#endif  // ROCKSDB_RANGESYNC_PRESENT
1233
  return FSWritableFile::RangeSync(offset, nbytes, opts, dbg);
1234 1235
}

T
Tomas Kolda 已提交
1236
#ifdef OS_LINUX
1237
size_t PosixWritableFile::GetUniqueId(char* id, size_t max_size) const {
K
krad 已提交
1238
  return PosixHelper::GetUniqueIdFromFile(fd_, id, max_size);
1239
}
1240
#endif
1241

I
Islam AbdelRahman 已提交
1242 1243 1244 1245 1246
/*
 * PosixRandomRWFile
 */

PosixRandomRWFile::PosixRandomRWFile(const std::string& fname, int fd,
A
Andrew Kryczka 已提交
1247
                                     const EnvOptions& /*options*/)
I
Islam AbdelRahman 已提交
1248 1249 1250 1251
    : filename_(fname), fd_(fd) {}

PosixRandomRWFile::~PosixRandomRWFile() {
  if (fd_ >= 0) {
1252
    Close(IOOptions(), nullptr);
I
Islam AbdelRahman 已提交
1253 1254 1255
  }
}

1256 1257 1258
IOStatus PosixRandomRWFile::Write(uint64_t offset, const Slice& data,
                                  const IOOptions& /*opts*/,
                                  IODebugContext* /*dbg*/) {
I
Islam AbdelRahman 已提交
1259
  const char* src = data.data();
1260 1261 1262 1263 1264
  size_t nbytes = data.size();
  if (!PosixPositionedWrite(fd_, src, nbytes, static_cast<off_t>(offset))) {
    return IOError(
        "While write random read/write file at offset " + ToString(offset),
        filename_, errno);
I
Islam AbdelRahman 已提交
1265 1266
  }

1267
  return IOStatus::OK();
I
Islam AbdelRahman 已提交
1268 1269
}

1270 1271 1272
IOStatus PosixRandomRWFile::Read(uint64_t offset, size_t n,
                                 const IOOptions& /*opts*/, Slice* result,
                                 char* scratch, IODebugContext* /*dbg*/) const {
I
Islam AbdelRahman 已提交
1273 1274 1275 1276 1277 1278 1279 1280 1281 1282
  size_t left = n;
  char* ptr = scratch;
  while (left > 0) {
    ssize_t done = pread(fd_, ptr, left, offset);
    if (done < 0) {
      // error while reading from file
      if (errno == EINTR) {
        // read was interrupted, try again.
        continue;
      }
1283 1284 1285
      return IOError("While reading random read/write file offset " +
                         ToString(offset) + " len " + ToString(n),
                     filename_, errno);
I
Islam AbdelRahman 已提交
1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297
    } else if (done == 0) {
      // Nothing more to read
      break;
    }

    // Read `done` bytes
    ptr += done;
    offset += done;
    left -= done;
  }

  *result = Slice(scratch, n - left);
1298
  return IOStatus::OK();
I
Islam AbdelRahman 已提交
1299 1300
}

1301 1302 1303 1304
IOStatus PosixRandomRWFile::Flush(const IOOptions& /*opts*/,
                                  IODebugContext* /*dbg*/) {
  return IOStatus::OK();
}
I
Islam AbdelRahman 已提交
1305

1306 1307
IOStatus PosixRandomRWFile::Sync(const IOOptions& /*opts*/,
                                 IODebugContext* /*dbg*/) {
I
Islam AbdelRahman 已提交
1308
  if (fdatasync(fd_) < 0) {
1309
    return IOError("While fdatasync random read/write file", filename_, errno);
I
Islam AbdelRahman 已提交
1310
  }
1311
  return IOStatus::OK();
I
Islam AbdelRahman 已提交
1312 1313
}

1314 1315
IOStatus PosixRandomRWFile::Fsync(const IOOptions& /*opts*/,
                                  IODebugContext* /*dbg*/) {
I
Islam AbdelRahman 已提交
1316
  if (fsync(fd_) < 0) {
1317
    return IOError("While fsync random read/write file", filename_, errno);
I
Islam AbdelRahman 已提交
1318
  }
1319
  return IOStatus::OK();
I
Islam AbdelRahman 已提交
1320 1321
}

1322 1323
IOStatus PosixRandomRWFile::Close(const IOOptions& /*opts*/,
                                  IODebugContext* /*dbg*/) {
I
Islam AbdelRahman 已提交
1324
  if (close(fd_) < 0) {
1325
    return IOError("While close random read/write file", filename_, errno);
I
Islam AbdelRahman 已提交
1326 1327
  }
  fd_ = -1;
1328
  return IOStatus::OK();
I
Islam AbdelRahman 已提交
1329 1330
}

1331 1332
PosixMemoryMappedFileBuffer::~PosixMemoryMappedFileBuffer() {
  // TODO should have error handling though not much we can do...
D
Dmitri Smirnov 已提交
1333
  munmap(this->base_, length_);
1334 1335
}

K
krad 已提交
1336 1337 1338 1339
/*
 * PosixDirectory
 */

1340 1341
PosixDirectory::~PosixDirectory() { close(fd_); }

1342 1343
IOStatus PosixDirectory::Fsync(const IOOptions& /*opts*/,
                               IODebugContext* /*dbg*/) {
T
Tomas Kolda 已提交
1344
#ifndef OS_AIX
1345
  if (fsync(fd_) == -1) {
1346
    return IOError("While fsync", "a directory", errno);
1347
  }
T
Tomas Kolda 已提交
1348
#endif
1349
  return IOStatus::OK();
1350
}
1351
}  // namespace ROCKSDB_NAMESPACE
1352
#endif