env_test.cc 46.6 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
//
J
jorlow@chromium.org 已提交
6 7 8 9
// 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.

D
Dmitri Smirnov 已提交
10
#ifndef OS_WIN
S
sdong 已提交
11
#include <sys/ioctl.h>
D
Dmitri Smirnov 已提交
12
#endif
K
krad 已提交
13 14

#ifdef ROCKSDB_MALLOC_USABLE_SIZE
15 16 17
#ifdef OS_FREEBSD
#include <malloc_np.h>
#else
K
krad 已提交
18 19
#include <malloc.h>
#endif
20
#endif
S
sdong 已提交
21
#include <sys/types.h>
J
jorlow@chromium.org 已提交
22

23
#include <iostream>
24
#include <unordered_set>
I
Igor Canadi 已提交
25
#include <atomic>
26
#include <list>
27

28
#ifdef OS_LINUX
A
Andrew Kryczka 已提交
29
#include <fcntl.h>
30 31
#include <linux/fs.h>
#include <stdlib.h>
32 33 34 35
#include <sys/stat.h>
#include <unistd.h>
#endif

36 37 38 39
#ifdef ROCKSDB_FALLOCATE_PRESENT
#include <errno.h>
#endif

40
#include "env/env_chroot.h"
J
jorlow@chromium.org 已提交
41
#include "port/port.h"
A
Andrew Kryczka 已提交
42
#include "rocksdb/env.h"
43
#include "util/coding.h"
S
sdong 已提交
44
#include "util/log_buffer.h"
45
#include "util/mutexlock.h"
46
#include "util/string_util.h"
K
krad 已提交
47
#include "util/sync_point.h"
J
jorlow@chromium.org 已提交
48
#include "util/testharness.h"
49
#include "util/testutil.h"
J
jorlow@chromium.org 已提交
50

51 52 53 54 55 56
#ifdef OS_LINUX
static const size_t kPageSize = sysconf(_SC_PAGESIZE);
#else
static const size_t kPageSize = 4 * 1024;
#endif

57
namespace rocksdb {
J
jorlow@chromium.org 已提交
58 59 60

static const int kDelayMicros = 100000;

K
krad 已提交
61 62 63 64 65 66 67 68 69 70 71 72 73
struct Deleter {
  explicit Deleter(void (*fn)(void*)) : fn_(fn) {}

  void operator()(void* ptr) {
    assert(fn_);
    assert(ptr);
    (*fn_)(ptr);
  }

  void (*fn_)(void*);
};

std::unique_ptr<char, Deleter> NewAligned(const size_t size, const char ch) {
K
krad 已提交
74
  char* ptr = nullptr;
K
krad 已提交
75
#ifdef OS_WIN
76
  if (nullptr == (ptr = reinterpret_cast<char*>(_aligned_malloc(size, kPageSize)))) {
K
krad 已提交
77 78 79 80
    return std::unique_ptr<char, Deleter>(nullptr, Deleter(_aligned_free));
  }
  std::unique_ptr<char, Deleter> uptr(ptr, Deleter(_aligned_free));
#else
81
  if (posix_memalign(reinterpret_cast<void**>(&ptr), kPageSize, size) != 0) {
K
krad 已提交
82
    return std::unique_ptr<char, Deleter>(nullptr, Deleter(free));
K
krad 已提交
83
  }
K
krad 已提交
84 85
  std::unique_ptr<char, Deleter> uptr(ptr, Deleter(free));
#endif
K
krad 已提交
86 87 88 89
  memset(uptr.get(), ch, size);
  return uptr;
}

I
Igor Sugak 已提交
90
class EnvPosixTest : public testing::Test {
J
jorlow@chromium.org 已提交
91 92 93 94 95 96
 private:
  port::Mutex mu_;
  std::string events_;

 public:
  Env* env_;
A
Aaron Gao 已提交
97 98
  bool direct_io_;
  EnvPosixTest() : env_(Env::Default()), direct_io_(false) {}
J
jorlow@chromium.org 已提交
99 100
};

A
Aaron Gao 已提交
101 102 103
class EnvPosixTestWithParam
    : public EnvPosixTest,
      public ::testing::WithParamInterface<std::pair<Env*, bool>> {
A
Andrew Kryczka 已提交
104
 public:
A
Aaron Gao 已提交
105 106 107 108 109
  EnvPosixTestWithParam() {
    std::pair<Env*, bool> param_pair = GetParam();
    env_ = param_pair.first;
    direct_io_ = param_pair.second;
  }
110 111 112 113 114 115 116 117 118 119 120 121

  void WaitThreadPoolsEmpty() {
    // Wait until the thread pools are empty.
    while (env_->GetThreadPoolQueueLen(Env::Priority::LOW) != 0) {
      Env::Default()->SleepForMicroseconds(kDelayMicros);
    }
    while (env_->GetThreadPoolQueueLen(Env::Priority::HIGH) != 0) {
      Env::Default()->SleepForMicroseconds(kDelayMicros);
    }
  }

  ~EnvPosixTestWithParam() { WaitThreadPoolsEmpty(); }
A
Andrew Kryczka 已提交
122 123
};

J
jorlow@chromium.org 已提交
124
static void SetBool(void* ptr) {
125
  reinterpret_cast<std::atomic<bool>*>(ptr)->store(true);
J
jorlow@chromium.org 已提交
126 127
}

128 129 130 131 132 133 134 135
TEST_F(EnvPosixTest, RunImmediately) {
  for (int pri = Env::BOTTOM; pri < Env::TOTAL; ++pri) {
    std::atomic<bool> called(false);
    env_->SetBackgroundThreads(1, static_cast<Env::Priority>(pri));
    env_->Schedule(&SetBool, &called, static_cast<Env::Priority>(pri));
    Env::Default()->SleepForMicroseconds(kDelayMicros);
    ASSERT_TRUE(called.load());
  }
J
jorlow@chromium.org 已提交
136 137
}

A
Andrew Kryczka 已提交
138
TEST_P(EnvPosixTestWithParam, UnSchedule) {
139 140 141 142
  std::atomic<bool> called(false);
  env_->SetBackgroundThreads(1, Env::LOW);

  /* Block the low priority queue */
143 144
  test::SleepingBackgroundTask sleeping_task, sleeping_task1;
  env_->Schedule(&test::SleepingBackgroundTask::DoSleepTask, &sleeping_task,
145 146 147
                 Env::Priority::LOW);

  /* Schedule another task */
148
  env_->Schedule(&test::SleepingBackgroundTask::DoSleepTask, &sleeping_task1,
149 150 151 152 153 154 155 156 157 158 159 160 161
                 Env::Priority::LOW, &sleeping_task1);

  /* Remove it with a different tag  */
  ASSERT_EQ(0, env_->UnSchedule(&called, Env::Priority::LOW));

  /* Remove it from the queue with the right tag */
  ASSERT_EQ(1, env_->UnSchedule(&sleeping_task1, Env::Priority::LOW));

  // Unblock background thread
  sleeping_task.WakeUp();

  /* Schedule another task */
  env_->Schedule(&SetBool, &called);
162
  for (int i = 0; i < kDelayMicros; i++) {
163
    if (called.load()) {
164 165 166 167
      break;
    }
    Env::Default()->SleepForMicroseconds(1);
  }
168
  ASSERT_TRUE(called.load());
169 170

  ASSERT_TRUE(!sleeping_task.IsSleeping() && !sleeping_task1.IsSleeping());
171
  WaitThreadPoolsEmpty();
172 173
}

A
Andrew Kryczka 已提交
174
TEST_P(EnvPosixTestWithParam, RunMany) {
I
Igor Canadi 已提交
175
  std::atomic<int> last_id(0);
J
jorlow@chromium.org 已提交
176 177

  struct CB {
I
Igor Canadi 已提交
178 179
    std::atomic<int>* last_id_ptr;  // Pointer to shared slot
    int id;                         // Order# for the execution of this callback
J
jorlow@chromium.org 已提交
180

I
Igor Canadi 已提交
181
    CB(std::atomic<int>* p, int i) : last_id_ptr(p), id(i) {}
J
jorlow@chromium.org 已提交
182 183 184

    static void Run(void* v) {
      CB* cb = reinterpret_cast<CB*>(v);
185
      int cur = cb->last_id_ptr->load();
I
Igor Canadi 已提交
186
      ASSERT_EQ(cb->id - 1, cur);
187
      cb->last_id_ptr->store(cb->id);
J
jorlow@chromium.org 已提交
188 189 190 191 192 193 194 195 196 197 198 199 200 201
    }
  };

  // Schedule in different order than start time
  CB cb1(&last_id, 1);
  CB cb2(&last_id, 2);
  CB cb3(&last_id, 3);
  CB cb4(&last_id, 4);
  env_->Schedule(&CB::Run, &cb1);
  env_->Schedule(&CB::Run, &cb2);
  env_->Schedule(&CB::Run, &cb3);
  env_->Schedule(&CB::Run, &cb4);

  Env::Default()->SleepForMicroseconds(kDelayMicros);
I
Igor Canadi 已提交
202 203
  int cur = last_id.load(std::memory_order_acquire);
  ASSERT_EQ(4, cur);
204
  WaitThreadPoolsEmpty();
J
jorlow@chromium.org 已提交
205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220
}

struct State {
  port::Mutex mu;
  int val;
  int num_running;
};

static void ThreadBody(void* arg) {
  State* s = reinterpret_cast<State*>(arg);
  s->mu.Lock();
  s->val += 1;
  s->num_running -= 1;
  s->mu.Unlock();
}

A
Andrew Kryczka 已提交
221
TEST_P(EnvPosixTestWithParam, StartThread) {
J
jorlow@chromium.org 已提交
222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237
  State state;
  state.val = 0;
  state.num_running = 3;
  for (int i = 0; i < 3; i++) {
    env_->StartThread(&ThreadBody, &state);
  }
  while (true) {
    state.mu.Lock();
    int num = state.num_running;
    state.mu.Unlock();
    if (num == 0) {
      break;
    }
    Env::Default()->SleepForMicroseconds(kDelayMicros);
  }
  ASSERT_EQ(state.val, 3);
238
  WaitThreadPoolsEmpty();
J
jorlow@chromium.org 已提交
239 240
}

A
Andrew Kryczka 已提交
241
TEST_P(EnvPosixTestWithParam, TwoPools) {
242 243 244 245 246
  // Data structures to signal tasks to run.
  port::Mutex mutex;
  port::CondVar cv(&mutex);
  bool should_start = false;

H
Haobo Xu 已提交
247 248
  class CB {
   public:
249 250
    CB(const std::string& pool_name, int pool_size, port::Mutex* trigger_mu,
       port::CondVar* trigger_cv, bool* _should_start)
H
Haobo Xu 已提交
251 252 253 254
        : mu_(),
          num_running_(0),
          num_finished_(0),
          pool_size_(pool_size),
255 256 257 258
          pool_name_(pool_name),
          trigger_mu_(trigger_mu),
          trigger_cv_(trigger_cv),
          should_start_(_should_start) {}
H
Haobo Xu 已提交
259 260 261 262 263 264 265 266 267 268 269

    static void Run(void* v) {
      CB* cb = reinterpret_cast<CB*>(v);
      cb->Run();
    }

    void Run() {
      {
        MutexLock l(&mu_);
        num_running_++;
        // make sure we don't have more than pool_size_ jobs running.
270
        ASSERT_LE(num_running_, pool_size_.load());
H
Haobo Xu 已提交
271 272
      }

273 274 275 276 277 278
      {
        MutexLock l(trigger_mu_);
        while (!(*should_start_)) {
          trigger_cv_->Wait();
        }
      }
H
Haobo Xu 已提交
279 280 281 282 283 284 285 286 287 288 289 290 291

      {
        MutexLock l(&mu_);
        num_running_--;
        num_finished_++;
      }
    }

    int NumFinished() {
      MutexLock l(&mu_);
      return num_finished_;
    }

292 293 294 295 296
    void Reset(int pool_size) {
      pool_size_.store(pool_size);
      num_finished_ = 0;
    }

H
Haobo Xu 已提交
297 298 299 300
   private:
    port::Mutex mu_;
    int num_running_;
    int num_finished_;
301
    std::atomic<int> pool_size_;
H
Haobo Xu 已提交
302
    std::string pool_name_;
303 304 305
    port::Mutex* trigger_mu_;
    port::CondVar* trigger_cv_;
    bool* should_start_;
H
Haobo Xu 已提交
306 307 308 309 310 311
  };

  const int kLowPoolSize = 2;
  const int kHighPoolSize = 4;
  const int kJobs = 8;

312 313
  CB low_pool_job("low", kLowPoolSize, &mutex, &cv, &should_start);
  CB high_pool_job("high", kHighPoolSize, &mutex, &cv, &should_start);
H
Haobo Xu 已提交
314 315 316 317

  env_->SetBackgroundThreads(kLowPoolSize);
  env_->SetBackgroundThreads(kHighPoolSize, Env::Priority::HIGH);

C
Caio SBA 已提交
318 319
  ASSERT_EQ(0U, env_->GetThreadPoolQueueLen(Env::Priority::LOW));
  ASSERT_EQ(0U, env_->GetThreadPoolQueueLen(Env::Priority::HIGH));
320

H
Haobo Xu 已提交
321 322 323 324 325
  // schedule same number of jobs in each pool
  for (int i = 0; i < kJobs; i++) {
    env_->Schedule(&CB::Run, &low_pool_job);
    env_->Schedule(&CB::Run, &high_pool_job, Env::Priority::HIGH);
  }
326
  // Wait a short while for the jobs to be dispatched.
327 328 329 330 331 332 333 334 335 336 337
  int sleep_count = 0;
  while ((unsigned int)(kJobs - kLowPoolSize) !=
             env_->GetThreadPoolQueueLen(Env::Priority::LOW) ||
         (unsigned int)(kJobs - kHighPoolSize) !=
             env_->GetThreadPoolQueueLen(Env::Priority::HIGH)) {
    env_->SleepForMicroseconds(kDelayMicros);
    if (++sleep_count > 100) {
      break;
    }
  }

C
Caio SBA 已提交
338 339
  ASSERT_EQ((unsigned int)(kJobs - kLowPoolSize),
            env_->GetThreadPoolQueueLen());
C
Caio SBA 已提交
340
  ASSERT_EQ((unsigned int)(kJobs - kLowPoolSize),
341
            env_->GetThreadPoolQueueLen(Env::Priority::LOW));
C
Caio SBA 已提交
342
  ASSERT_EQ((unsigned int)(kJobs - kHighPoolSize),
343
            env_->GetThreadPoolQueueLen(Env::Priority::HIGH));
H
Haobo Xu 已提交
344

345 346 347 348 349 350 351
  // Trigger jobs to run.
  {
    MutexLock l(&mutex);
    should_start = true;
    cv.SignalAll();
  }

H
Haobo Xu 已提交
352 353 354 355 356
  // wait for all jobs to finish
  while (low_pool_job.NumFinished() < kJobs ||
         high_pool_job.NumFinished() < kJobs) {
    env_->SleepForMicroseconds(kDelayMicros);
  }
357

C
Caio SBA 已提交
358 359
  ASSERT_EQ(0U, env_->GetThreadPoolQueueLen(Env::Priority::LOW));
  ASSERT_EQ(0U, env_->GetThreadPoolQueueLen(Env::Priority::HIGH));
360

361 362 363
  // Hold jobs to schedule;
  should_start = false;

364 365 366 367 368 369 370 371 372 373 374 375 376
  // call IncBackgroundThreadsIfNeeded to two pools. One increasing and
  // the other decreasing
  env_->IncBackgroundThreadsIfNeeded(kLowPoolSize - 1, Env::Priority::LOW);
  env_->IncBackgroundThreadsIfNeeded(kHighPoolSize + 1, Env::Priority::HIGH);
  high_pool_job.Reset(kHighPoolSize + 1);
  low_pool_job.Reset(kLowPoolSize);

  // schedule same number of jobs in each pool
  for (int i = 0; i < kJobs; i++) {
    env_->Schedule(&CB::Run, &low_pool_job);
    env_->Schedule(&CB::Run, &high_pool_job, Env::Priority::HIGH);
  }
  // Wait a short while for the jobs to be dispatched.
377 378 379 380 381 382 383 384 385 386
  sleep_count = 0;
  while ((unsigned int)(kJobs - kLowPoolSize) !=
             env_->GetThreadPoolQueueLen(Env::Priority::LOW) ||
         (unsigned int)(kJobs - (kHighPoolSize + 1)) !=
             env_->GetThreadPoolQueueLen(Env::Priority::HIGH)) {
    env_->SleepForMicroseconds(kDelayMicros);
    if (++sleep_count > 100) {
      break;
    }
  }
387 388 389 390 391 392 393
  ASSERT_EQ((unsigned int)(kJobs - kLowPoolSize),
            env_->GetThreadPoolQueueLen());
  ASSERT_EQ((unsigned int)(kJobs - kLowPoolSize),
            env_->GetThreadPoolQueueLen(Env::Priority::LOW));
  ASSERT_EQ((unsigned int)(kJobs - (kHighPoolSize + 1)),
            env_->GetThreadPoolQueueLen(Env::Priority::HIGH));

394 395 396 397 398 399 400
  // Trigger jobs to run.
  {
    MutexLock l(&mutex);
    should_start = true;
    cv.SignalAll();
  }

401 402 403 404 405 406 407
  // wait for all jobs to finish
  while (low_pool_job.NumFinished() < kJobs ||
         high_pool_job.NumFinished() < kJobs) {
    env_->SleepForMicroseconds(kDelayMicros);
  }

  env_->SetBackgroundThreads(kHighPoolSize, Env::Priority::HIGH);
408
  WaitThreadPoolsEmpty();
H
Haobo Xu 已提交
409 410
}

A
Andrew Kryczka 已提交
411
TEST_P(EnvPosixTestWithParam, DecreaseNumBgThreads) {
412
  std::vector<test::SleepingBackgroundTask> tasks(10);
413 414 415 416 417 418 419

  // Set number of thread to 1 first.
  env_->SetBackgroundThreads(1, Env::Priority::HIGH);
  Env::Default()->SleepForMicroseconds(kDelayMicros);

  // Schedule 3 tasks. 0 running; Task 1, 2 waiting.
  for (size_t i = 0; i < 3; i++) {
420
    env_->Schedule(&test::SleepingBackgroundTask::DoSleepTask, &tasks[i],
421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455
                   Env::Priority::HIGH);
    Env::Default()->SleepForMicroseconds(kDelayMicros);
  }
  ASSERT_EQ(2U, env_->GetThreadPoolQueueLen(Env::Priority::HIGH));
  ASSERT_TRUE(tasks[0].IsSleeping());
  ASSERT_TRUE(!tasks[1].IsSleeping());
  ASSERT_TRUE(!tasks[2].IsSleeping());

  // Increase to 2 threads. Task 0, 1 running; 2 waiting
  env_->SetBackgroundThreads(2, Env::Priority::HIGH);
  Env::Default()->SleepForMicroseconds(kDelayMicros);
  ASSERT_EQ(1U, env_->GetThreadPoolQueueLen(Env::Priority::HIGH));
  ASSERT_TRUE(tasks[0].IsSleeping());
  ASSERT_TRUE(tasks[1].IsSleeping());
  ASSERT_TRUE(!tasks[2].IsSleeping());

  // Shrink back to 1 thread. Still task 0, 1 running, 2 waiting
  env_->SetBackgroundThreads(1, Env::Priority::HIGH);
  Env::Default()->SleepForMicroseconds(kDelayMicros);
  ASSERT_EQ(1U, env_->GetThreadPoolQueueLen(Env::Priority::HIGH));
  ASSERT_TRUE(tasks[0].IsSleeping());
  ASSERT_TRUE(tasks[1].IsSleeping());
  ASSERT_TRUE(!tasks[2].IsSleeping());

  // The last task finishes. Task 0 running, 2 waiting.
  tasks[1].WakeUp();
  Env::Default()->SleepForMicroseconds(kDelayMicros);
  ASSERT_EQ(1U, env_->GetThreadPoolQueueLen(Env::Priority::HIGH));
  ASSERT_TRUE(tasks[0].IsSleeping());
  ASSERT_TRUE(!tasks[1].IsSleeping());
  ASSERT_TRUE(!tasks[2].IsSleeping());

  // Increase to 5 threads. Task 0 and 2 running.
  env_->SetBackgroundThreads(5, Env::Priority::HIGH);
  Env::Default()->SleepForMicroseconds(kDelayMicros);
456
  ASSERT_EQ((unsigned int)0, env_->GetThreadPoolQueueLen(Env::Priority::HIGH));
457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483
  ASSERT_TRUE(tasks[0].IsSleeping());
  ASSERT_TRUE(tasks[2].IsSleeping());

  // Change number of threads a couple of times while there is no sufficient
  // tasks.
  env_->SetBackgroundThreads(7, Env::Priority::HIGH);
  Env::Default()->SleepForMicroseconds(kDelayMicros);
  tasks[2].WakeUp();
  ASSERT_EQ(0U, env_->GetThreadPoolQueueLen(Env::Priority::HIGH));
  env_->SetBackgroundThreads(3, Env::Priority::HIGH);
  Env::Default()->SleepForMicroseconds(kDelayMicros);
  ASSERT_EQ(0U, env_->GetThreadPoolQueueLen(Env::Priority::HIGH));
  env_->SetBackgroundThreads(4, Env::Priority::HIGH);
  Env::Default()->SleepForMicroseconds(kDelayMicros);
  ASSERT_EQ(0U, env_->GetThreadPoolQueueLen(Env::Priority::HIGH));
  env_->SetBackgroundThreads(5, Env::Priority::HIGH);
  Env::Default()->SleepForMicroseconds(kDelayMicros);
  ASSERT_EQ(0U, env_->GetThreadPoolQueueLen(Env::Priority::HIGH));
  env_->SetBackgroundThreads(4, Env::Priority::HIGH);
  Env::Default()->SleepForMicroseconds(kDelayMicros);
  ASSERT_EQ(0U, env_->GetThreadPoolQueueLen(Env::Priority::HIGH));

  Env::Default()->SleepForMicroseconds(kDelayMicros * 50);

  // Enqueue 5 more tasks. Thread pool size now is 4.
  // Task 0, 3, 4, 5 running;6, 7 waiting.
  for (size_t i = 3; i < 8; i++) {
484
    env_->Schedule(&test::SleepingBackgroundTask::DoSleepTask, &tasks[i],
485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500
                   Env::Priority::HIGH);
  }
  Env::Default()->SleepForMicroseconds(kDelayMicros);
  ASSERT_EQ(2U, env_->GetThreadPoolQueueLen(Env::Priority::HIGH));
  ASSERT_TRUE(tasks[3].IsSleeping());
  ASSERT_TRUE(tasks[4].IsSleeping());
  ASSERT_TRUE(tasks[5].IsSleeping());
  ASSERT_TRUE(!tasks[6].IsSleeping());
  ASSERT_TRUE(!tasks[7].IsSleeping());

  // Wake up task 0, 3 and 4. Task 5, 6, 7 running.
  tasks[0].WakeUp();
  tasks[3].WakeUp();
  tasks[4].WakeUp();

  Env::Default()->SleepForMicroseconds(kDelayMicros);
501
  ASSERT_EQ((unsigned int)0, env_->GetThreadPoolQueueLen(Env::Priority::HIGH));
502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525
  for (size_t i = 5; i < 8; i++) {
    ASSERT_TRUE(tasks[i].IsSleeping());
  }

  // Shrink back to 1 thread. Still task 5, 6, 7 running
  env_->SetBackgroundThreads(1, Env::Priority::HIGH);
  Env::Default()->SleepForMicroseconds(kDelayMicros);
  ASSERT_TRUE(tasks[5].IsSleeping());
  ASSERT_TRUE(tasks[6].IsSleeping());
  ASSERT_TRUE(tasks[7].IsSleeping());

  // Wake up task  6. Task 5, 7 running
  tasks[6].WakeUp();
  Env::Default()->SleepForMicroseconds(kDelayMicros);
  ASSERT_TRUE(tasks[5].IsSleeping());
  ASSERT_TRUE(!tasks[6].IsSleeping());
  ASSERT_TRUE(tasks[7].IsSleeping());

  // Wake up threads 7. Task 5 running
  tasks[7].WakeUp();
  Env::Default()->SleepForMicroseconds(kDelayMicros);
  ASSERT_TRUE(!tasks[7].IsSleeping());

  // Enqueue thread 8 and 9. Task 5 running; one of 8, 9 might be running.
526
  env_->Schedule(&test::SleepingBackgroundTask::DoSleepTask, &tasks[8],
527
                 Env::Priority::HIGH);
528
  env_->Schedule(&test::SleepingBackgroundTask::DoSleepTask, &tasks[9],
529 530
                 Env::Priority::HIGH);
  Env::Default()->SleepForMicroseconds(kDelayMicros);
531
  ASSERT_GT(env_->GetThreadPoolQueueLen(Env::Priority::HIGH), (unsigned int)0);
532 533 534 535 536
  ASSERT_TRUE(!tasks[8].IsSleeping() || !tasks[9].IsSleeping());

  // Increase to 4 threads. Task 5, 8, 9 running.
  env_->SetBackgroundThreads(4, Env::Priority::HIGH);
  Env::Default()->SleepForMicroseconds(kDelayMicros);
537
  ASSERT_EQ((unsigned int)0, env_->GetThreadPoolQueueLen(Env::Priority::HIGH));
538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559
  ASSERT_TRUE(tasks[8].IsSleeping());
  ASSERT_TRUE(tasks[9].IsSleeping());

  // Shrink to 1 thread
  env_->SetBackgroundThreads(1, Env::Priority::HIGH);

  // Wake up thread 9.
  tasks[9].WakeUp();
  Env::Default()->SleepForMicroseconds(kDelayMicros);
  ASSERT_TRUE(!tasks[9].IsSleeping());
  ASSERT_TRUE(tasks[8].IsSleeping());

  // Wake up thread 8
  tasks[8].WakeUp();
  Env::Default()->SleepForMicroseconds(kDelayMicros);
  ASSERT_TRUE(!tasks[8].IsSleeping());

  // Wake up the last thread
  tasks[5].WakeUp();

  Env::Default()->SleepForMicroseconds(kDelayMicros);
  ASSERT_TRUE(!tasks[5].IsSleeping());
560
  WaitThreadPoolsEmpty();
561 562
}

D
Dmitri Smirnov 已提交
563
#if (defined OS_LINUX || defined OS_WIN)
I
Igor Canadi 已提交
564 565 566
// Travis doesn't support fallocate or getting unique ID from files for whatever
// reason.
#ifndef TRAVIS
567

I
Igor Canadi 已提交
568
namespace {
569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586
bool IsSingleVarint(const std::string& s) {
  Slice slice(s);

  uint64_t v;
  if (!GetVarint64(&slice, &v)) {
    return false;
  }

  return slice.size() == 0;
}

bool IsUniqueIDValid(const std::string& s) {
  return !s.empty() && !IsSingleVarint(s);
}

const size_t MAX_ID_SIZE = 100;
char temp_id[MAX_ID_SIZE];

587

588
}  // namespace
589 590

// Determine whether we can use the FS_IOC_GETVERSION ioctl
591
// on a file in directory DIR.  Create a temporary file therein,
592 593 594
// try to apply the ioctl (save that result), cleanup and
// return the result.  Return true if it is supported, and
// false if anything fails.
595 596 597
// Note that this function "knows" that dir has just been created
// and is empty, so we create a simply-named test file: "f".
bool ioctl_support__FS_IOC_GETVERSION(const std::string& dir) {
D
Dmitri Smirnov 已提交
598 599 600
#ifdef OS_WIN
  return true;
#else
601 602 603 604 605
  const std::string file = dir + "/f";
  int fd;
  do {
    fd = open(file.c_str(), O_CREAT | O_RDWR | O_TRUNC, 0644);
  } while (fd < 0 && errno == EINTR);
606 607 608 609
  long int version;
  bool ok = (fd >= 0 && ioctl(fd, FS_IOC_GETVERSION, &version) >= 0);

  close(fd);
610
  unlink(file.c_str());
611 612

  return ok;
D
Dmitri Smirnov 已提交
613
#endif
614 615
}

616 617 618 619 620 621 622 623 624 625 626 627 628 629 630
// To ensure that Env::GetUniqueId-related tests work correctly, the files
// should be stored in regular storage like "hard disk" or "flash device",
// and not on a tmpfs file system (like /dev/shm and /tmp on some systems).
// Otherwise we cannot get the correct id.
//
// This function serves as the replacement for test::TmpDir(), which may be
// customized to be on a file system that doesn't work with GetUniqueId().

class IoctlFriendlyTmpdir {
 public:
  explicit IoctlFriendlyTmpdir() {
    char dir_buf[100];

    const char *fmt = "%s/rocksdb.XXXXXX";
    const char *tmp = getenv("TEST_IOCTL_FRIENDLY_TMPDIR");
D
Dmitri Smirnov 已提交
631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646

#ifdef OS_WIN
#define rmdir _rmdir
    if(tmp == nullptr) {
      tmp = getenv("TMP");
    }

    snprintf(dir_buf, sizeof dir_buf, fmt, tmp);
    auto result = _mktemp(dir_buf);
    assert(result != nullptr);
    BOOL ret = CreateDirectory(dir_buf, NULL);
    assert(ret == TRUE);
    dir_ = dir_buf;
#else
    std::list<std::string> candidate_dir_list = {"/var/tmp", "/tmp"};

647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663
    // If $TEST_IOCTL_FRIENDLY_TMPDIR/rocksdb.XXXXXX fits, use
    // $TEST_IOCTL_FRIENDLY_TMPDIR; subtract 2 for the "%s", and
    // add 1 for the trailing NUL byte.
    if (tmp && strlen(tmp) + strlen(fmt) - 2 + 1 <= sizeof dir_buf) {
      // use $TEST_IOCTL_FRIENDLY_TMPDIR value
      candidate_dir_list.push_front(tmp);
    }

    for (const std::string& d : candidate_dir_list) {
      snprintf(dir_buf, sizeof dir_buf, fmt, d.c_str());
      if (mkdtemp(dir_buf)) {
        if (ioctl_support__FS_IOC_GETVERSION(dir_buf)) {
          dir_ = dir_buf;
          return;
        } else {
          // Diagnose ioctl-related failure only if this is the
          // directory specified via that envvar.
D
dx9 已提交
664
          if (tmp && tmp == d) {
665 666 667 668 669 670 671 672 673 674 675 676 677 678 679
            fprintf(stderr, "TEST_IOCTL_FRIENDLY_TMPDIR-specified directory is "
                    "not suitable: %s\n", d.c_str());
          }
          rmdir(dir_buf);  // ignore failure
        }
      } else {
        // mkdtemp failed: diagnose it, but don't give up.
        fprintf(stderr, "mkdtemp(%s/...) failed: %s\n", d.c_str(),
                strerror(errno));
      }
    }

    fprintf(stderr, "failed to find an ioctl-friendly temporary directory;"
            " specify one via the TEST_IOCTL_FRIENDLY_TMPDIR envvar\n");
    std::abort();
D
Dmitri Smirnov 已提交
680 681
#endif
}
682 683 684 685

  ~IoctlFriendlyTmpdir() {
    rmdir(dir_.c_str());
  }
D
Dmitri Smirnov 已提交
686 687

  const std::string& name() const {
688 689 690 691 692 693
    return dir_;
  }

 private:
  std::string dir_;
};
694

695
#ifndef ROCKSDB_LITE
A
Aaron Gao 已提交
696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724
TEST_F(EnvPosixTest, PositionedAppend) {
  unique_ptr<WritableFile> writable_file;
  EnvOptions options;
  options.use_direct_writes = true;
  options.use_mmap_writes = false;
  IoctlFriendlyTmpdir ift;
  ASSERT_OK(env_->NewWritableFile(ift.name() + "/f", &writable_file, options));
  const size_t kBlockSize = 4096;
  const size_t kDataSize = kPageSize;
  // Write a page worth of 'a'
  auto data_ptr = NewAligned(kDataSize, 'a');
  Slice data_a(data_ptr.get(), kDataSize);
  ASSERT_OK(writable_file->PositionedAppend(data_a, 0U));
  // Write a page worth of 'b' right after the first sector
  data_ptr = NewAligned(kDataSize, 'b');
  Slice data_b(data_ptr.get(), kDataSize);
  ASSERT_OK(writable_file->PositionedAppend(data_b, kBlockSize));
  ASSERT_OK(writable_file->Close());
  // The file now has 1 sector worth of a followed by a page worth of b

  // Verify the above
  unique_ptr<SequentialFile> seq_file;
  ASSERT_OK(env_->NewSequentialFile(ift.name() + "/f", &seq_file, options));
  char scratch[kPageSize * 2];
  Slice result;
  ASSERT_OK(seq_file->Read(sizeof(scratch), &result, scratch));
  ASSERT_EQ(kPageSize + kBlockSize, result.size());
  ASSERT_EQ('a', result[kBlockSize - 1]);
  ASSERT_EQ('b', result[kBlockSize]);
725
}
726
#endif  // !ROCKSDB_LITE
727

K
kailiu 已提交
728
// Only works in linux platforms
A
Aaron Gao 已提交
729 730 731
TEST_P(EnvPosixTestWithParam, RandomAccessUniqueID) {
  // Create file.
  if (env_ == Env::Default()) {
K
krad 已提交
732
    EnvOptions soptions;
A
Aaron Gao 已提交
733
    soptions.use_direct_reads = soptions.use_direct_writes = direct_io_;
K
krad 已提交
734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769
    IoctlFriendlyTmpdir ift;
    std::string fname = ift.name() + "/testfile";
    unique_ptr<WritableFile> wfile;
    ASSERT_OK(env_->NewWritableFile(fname, &wfile, soptions));

    unique_ptr<RandomAccessFile> file;

    // Get Unique ID
    ASSERT_OK(env_->NewRandomAccessFile(fname, &file, soptions));
    size_t id_size = file->GetUniqueId(temp_id, MAX_ID_SIZE);
    ASSERT_TRUE(id_size > 0);
    std::string unique_id1(temp_id, id_size);
    ASSERT_TRUE(IsUniqueIDValid(unique_id1));

    // Get Unique ID again
    ASSERT_OK(env_->NewRandomAccessFile(fname, &file, soptions));
    id_size = file->GetUniqueId(temp_id, MAX_ID_SIZE);
    ASSERT_TRUE(id_size > 0);
    std::string unique_id2(temp_id, id_size);
    ASSERT_TRUE(IsUniqueIDValid(unique_id2));

    // Get Unique ID again after waiting some time.
    env_->SleepForMicroseconds(1000000);
    ASSERT_OK(env_->NewRandomAccessFile(fname, &file, soptions));
    id_size = file->GetUniqueId(temp_id, MAX_ID_SIZE);
    ASSERT_TRUE(id_size > 0);
    std::string unique_id3(temp_id, id_size);
    ASSERT_TRUE(IsUniqueIDValid(unique_id3));

    // Check IDs are the same.
    ASSERT_EQ(unique_id1, unique_id2);
    ASSERT_EQ(unique_id2, unique_id3);

    // Delete the file
    env_->DeleteFile(fname);
  }
770 771
}

772 773
// only works in linux platforms
#ifdef ROCKSDB_FALLOCATE_PRESENT
A
Aaron Gao 已提交
774 775
TEST_P(EnvPosixTestWithParam, AllocateTest) {
  if (env_ == Env::Default()) {
K
krad 已提交
776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801
    IoctlFriendlyTmpdir ift;
    std::string fname = ift.name() + "/preallocate_testfile";

    // Try fallocate in a file to see whether the target file system supports
    // it.
    // Skip the test if fallocate is not supported.
    std::string fname_test_fallocate = ift.name() + "/preallocate_testfile_2";
    int fd = -1;
    do {
      fd = open(fname_test_fallocate.c_str(), O_CREAT | O_RDWR | O_TRUNC, 0644);
    } while (fd < 0 && errno == EINTR);
    ASSERT_GT(fd, 0);

    int alloc_status = fallocate(fd, 0, 0, 1);

    int err_number = 0;
    if (alloc_status != 0) {
      err_number = errno;
      fprintf(stderr, "Warning: fallocate() fails, %s\n", strerror(err_number));
    }
    close(fd);
    ASSERT_OK(env_->DeleteFile(fname_test_fallocate));
    if (alloc_status != 0 && err_number == EOPNOTSUPP) {
      // The filesystem containing the file does not support fallocate
      return;
    }
802

K
krad 已提交
803 804
    EnvOptions soptions;
    soptions.use_mmap_writes = false;
A
Aaron Gao 已提交
805
    soptions.use_direct_reads = soptions.use_direct_writes = direct_io_;
K
krad 已提交
806 807
    unique_ptr<WritableFile> wfile;
    ASSERT_OK(env_->NewWritableFile(fname, &wfile, soptions));
808

K
krad 已提交
809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841
    // allocate 100 MB
    size_t kPreallocateSize = 100 * 1024 * 1024;
    size_t kBlockSize = 512;
    size_t kPageSize = 4096;
    size_t kDataSize = 1024 * 1024;
    auto data_ptr = NewAligned(kDataSize, 'A');
    Slice data(data_ptr.get(), kDataSize);
    wfile->SetPreallocationBlockSize(kPreallocateSize);
    wfile->PrepareWrite(wfile->GetFileSize(), kDataSize);
    ASSERT_OK(wfile->Append(data));
    ASSERT_OK(wfile->Flush());

    struct stat f_stat;
    ASSERT_EQ(stat(fname.c_str(), &f_stat), 0);
    ASSERT_EQ((unsigned int)kDataSize, f_stat.st_size);
    // verify that blocks are preallocated
    // Note here that we don't check the exact number of blocks preallocated --
    // we only require that number of allocated blocks is at least what we
    // expect.
    // It looks like some FS give us more blocks that we asked for. That's fine.
    // It might be worth investigating further.
    ASSERT_LE((unsigned int)(kPreallocateSize / kBlockSize), f_stat.st_blocks);

    // close the file, should deallocate the blocks
    wfile.reset();

    stat(fname.c_str(), &f_stat);
    ASSERT_EQ((unsigned int)kDataSize, f_stat.st_size);
    // verify that preallocated blocks were deallocated on file close
    // Because the FS might give us more blocks, we add a full page to the size
    // and expect the number of blocks to be less or equal to that.
    ASSERT_GE((f_stat.st_size + kPageSize + kBlockSize - 1) / kBlockSize,
              (unsigned int)f_stat.st_blocks);
842
  }
843
}
I
Igor Canadi 已提交
844
#endif  // ROCKSDB_FALLOCATE_PRESENT
845

846 847 848 849 850 851 852 853 854 855 856 857 858 859 860
// Returns true if any of the strings in ss are the prefix of another string.
bool HasPrefix(const std::unordered_set<std::string>& ss) {
  for (const std::string& s: ss) {
    if (s.empty()) {
      return true;
    }
    for (size_t i = 1; i < s.size(); ++i) {
      if (ss.count(s.substr(0, i)) != 0) {
        return true;
      }
    }
  }
  return false;
}

D
Dmitri Smirnov 已提交
861
// Only works in linux and WIN platforms
A
Aaron Gao 已提交
862 863
TEST_P(EnvPosixTestWithParam, RandomAccessUniqueIDConcurrent) {
  if (env_ == Env::Default()) {
K
krad 已提交
864 865
    // Check whether a bunch of concurrently existing files have unique IDs.
    EnvOptions soptions;
A
Aaron Gao 已提交
866
    soptions.use_direct_reads = soptions.use_direct_writes = direct_io_;
K
krad 已提交
867 868 869 870 871 872 873 874 875 876 877

    // Create the files
    IoctlFriendlyTmpdir ift;
    std::vector<std::string> fnames;
    for (int i = 0; i < 1000; ++i) {
      fnames.push_back(ift.name() + "/" + "testfile" + ToString(i));

      // Create file.
      unique_ptr<WritableFile> wfile;
      ASSERT_OK(env_->NewWritableFile(fnames[i], &wfile, soptions));
    }
878

K
krad 已提交
879 880 881 882 883 884 885 886 887 888
    // Collect and check whether the IDs are unique.
    std::unordered_set<std::string> ids;
    for (const std::string fname : fnames) {
      unique_ptr<RandomAccessFile> file;
      std::string unique_id;
      ASSERT_OK(env_->NewRandomAccessFile(fname, &file, soptions));
      size_t id_size = file->GetUniqueId(temp_id, MAX_ID_SIZE);
      ASSERT_TRUE(id_size > 0);
      unique_id = std::string(temp_id, id_size);
      ASSERT_TRUE(IsUniqueIDValid(unique_id));
889

K
krad 已提交
890 891 892
      ASSERT_TRUE(ids.count(unique_id) == 0);
      ids.insert(unique_id);
    }
893

K
krad 已提交
894 895 896 897
    // Delete the files
    for (const std::string fname : fnames) {
      ASSERT_OK(env_->DeleteFile(fname));
    }
898

K
krad 已提交
899
    ASSERT_TRUE(!HasPrefix(ids));
900
  }
K
krad 已提交
901
}
902

D
Dmitri Smirnov 已提交
903
// Only works in linux and WIN platforms
A
Aaron Gao 已提交
904 905
TEST_P(EnvPosixTestWithParam, RandomAccessUniqueIDDeletes) {
  if (env_ == Env::Default()) {
K
krad 已提交
906
    EnvOptions soptions;
A
Aaron Gao 已提交
907
    soptions.use_direct_reads = soptions.use_direct_writes = direct_io_;
K
krad 已提交
908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938

    IoctlFriendlyTmpdir ift;
    std::string fname = ift.name() + "/" + "testfile";

    // Check that after file is deleted we don't get same ID again in a new
    // file.
    std::unordered_set<std::string> ids;
    for (int i = 0; i < 1000; ++i) {
      // Create file.
      {
        unique_ptr<WritableFile> wfile;
        ASSERT_OK(env_->NewWritableFile(fname, &wfile, soptions));
      }

      // Get Unique ID
      std::string unique_id;
      {
        unique_ptr<RandomAccessFile> file;
        ASSERT_OK(env_->NewRandomAccessFile(fname, &file, soptions));
        size_t id_size = file->GetUniqueId(temp_id, MAX_ID_SIZE);
        ASSERT_TRUE(id_size > 0);
        unique_id = std::string(temp_id, id_size);
      }

      ASSERT_TRUE(IsUniqueIDValid(unique_id));
      ASSERT_TRUE(ids.count(unique_id) == 0);
      ids.insert(unique_id);

      // Delete the file
      ASSERT_OK(env_->DeleteFile(fname));
    }
939

K
krad 已提交
940 941
    ASSERT_TRUE(!HasPrefix(ids));
  }
942 943
}

K
kailiu 已提交
944
// Only works in linux platforms
D
Dmitri Smirnov 已提交
945 946 947
#ifdef OS_WIN
TEST_P(EnvPosixTestWithParam, DISABLED_InvalidateCache) {
#else
K
krad 已提交
948
TEST_P(EnvPosixTestWithParam, InvalidateCache) {
D
Dmitri Smirnov 已提交
949
#endif
K
krad 已提交
950 951
  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
    EnvOptions soptions;
A
Aaron Gao 已提交
952
    soptions.use_direct_reads = soptions.use_direct_writes = direct_io_;
K
krad 已提交
953
    std::string fname = test::TmpDir(env_) + "/" + "testfile";
954

K
krad 已提交
955
    const size_t kSectorSize = 512;
A
Aaron Gao 已提交
956
    auto data = NewAligned(kSectorSize, 0);
K
krad 已提交
957
    Slice slice(data.get(), kSectorSize);
958 959 960 961

    // Create file.
    {
      unique_ptr<WritableFile> wfile;
T
Tomas Kolda 已提交
962
#if !defined(OS_MACOSX) && !defined(OS_WIN) && !defined(OS_SOLARIS) && !defined(OS_AIX)
K
krad 已提交
963
      if (soptions.use_direct_writes) {
A
Aaron Gao 已提交
964
        soptions.use_direct_writes = false;
K
krad 已提交
965
      }
K
krad 已提交
966
#endif
967
      ASSERT_OK(env_->NewWritableFile(fname, &wfile, soptions));
D
Dmitri Smirnov 已提交
968 969 970
      ASSERT_OK(wfile->Append(slice));
      ASSERT_OK(wfile->InvalidateCache(0, 0));
      ASSERT_OK(wfile->Close());
971 972
    }

K
krad 已提交
973
    // Random Read
974 975
    {
      unique_ptr<RandomAccessFile> file;
A
Aaron Gao 已提交
976
      auto scratch = NewAligned(kSectorSize, 0);
K
krad 已提交
977
      Slice result;
T
Tomas Kolda 已提交
978
#if !defined(OS_MACOSX) && !defined(OS_WIN) && !defined(OS_SOLARIS) && !defined(OS_AIX)
K
krad 已提交
979
      if (soptions.use_direct_reads) {
A
Aaron Gao 已提交
980
        soptions.use_direct_reads = false;
K
krad 已提交
981
      }
K
krad 已提交
982
#endif
983
      ASSERT_OK(env_->NewRandomAccessFile(fname, &file, soptions));
A
Aaron Gao 已提交
984 985
      ASSERT_OK(file->Read(0, kSectorSize, &result, scratch.get()));
      ASSERT_EQ(memcmp(scratch.get(), data.get(), kSectorSize), 0);
D
Dmitri Smirnov 已提交
986 987
      ASSERT_OK(file->InvalidateCache(0, 11));
      ASSERT_OK(file->InvalidateCache(0, 0));
988 989
    }

K
krad 已提交
990 991 992
    // Sequential Read
    {
      unique_ptr<SequentialFile> file;
A
Aaron Gao 已提交
993
      auto scratch = NewAligned(kSectorSize, 0);
K
krad 已提交
994
      Slice result;
T
Tomas Kolda 已提交
995
#if !defined(OS_MACOSX) && !defined(OS_WIN) && !defined(OS_SOLARIS) && !defined(OS_AIX)
K
krad 已提交
996
      if (soptions.use_direct_reads) {
A
Aaron Gao 已提交
997
        soptions.use_direct_reads = false;
K
krad 已提交
998
      }
K
krad 已提交
999
#endif
K
krad 已提交
1000
      ASSERT_OK(env_->NewSequentialFile(fname, &file, soptions));
1001
      if (file->use_direct_io()) {
A
Aaron Gao 已提交
1002 1003 1004 1005 1006
        ASSERT_OK(file->PositionedRead(0, kSectorSize, &result, scratch.get()));
      } else {
        ASSERT_OK(file->Read(kSectorSize, &result, scratch.get()));
      }
      ASSERT_EQ(memcmp(scratch.get(), data.get(), kSectorSize), 0);
D
Dmitri Smirnov 已提交
1007 1008
      ASSERT_OK(file->InvalidateCache(0, 11));
      ASSERT_OK(file->InvalidateCache(0, 0));
K
krad 已提交
1009
    }
1010 1011
    // Delete the file
    ASSERT_OK(env_->DeleteFile(fname));
K
krad 已提交
1012
  rocksdb::SyncPoint::GetInstance()->ClearTrace();
1013
}
I
Igor Canadi 已提交
1014
#endif  // not TRAVIS
D
Dmitri Smirnov 已提交
1015
#endif  // OS_LINUX || OS_WIN
1016

S
sdong 已提交
1017 1018
class TestLogger : public Logger {
 public:
F
fyrz 已提交
1019
  using Logger::Logv;
S
sdong 已提交
1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030
  virtual void Logv(const char* format, va_list ap) override {
    log_count++;

    char new_format[550];
    std::fill_n(new_format, sizeof(new_format), '2');
    {
      va_list backup_ap;
      va_copy(backup_ap, ap);
      int n = vsnprintf(new_format, sizeof(new_format) - 1, format, backup_ap);
      // 48 bytes for extra information + bytes allocated

S
sdong 已提交
1031
// When we have n == -1 there is not a terminating zero expected
D
Dmitri Smirnov 已提交
1032 1033 1034 1035 1036 1037
#ifdef OS_WIN
      if (n < 0) {
        char_0_count++;
      }
#endif

S
sdong 已提交
1038 1039
      if (new_format[0] == '[') {
        // "[DEBUG] "
1040
        ASSERT_TRUE(n <= 56 + (512 - static_cast<int>(sizeof(struct timeval))));
S
sdong 已提交
1041
      } else {
1042
        ASSERT_TRUE(n <= 48 + (512 - static_cast<int>(sizeof(struct timeval))));
S
sdong 已提交
1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059
      }
      va_end(backup_ap);
    }

    for (size_t i = 0; i < sizeof(new_format); i++) {
      if (new_format[i] == 'x') {
        char_x_count++;
      } else if (new_format[i] == '\0') {
        char_0_count++;
      }
    }
  }
  int log_count;
  int char_x_count;
  int char_0_count;
};

A
Andrew Kryczka 已提交
1060
TEST_P(EnvPosixTestWithParam, LogBufferTest) {
S
sdong 已提交
1061
  TestLogger test_logger;
I
Igor Canadi 已提交
1062
  test_logger.SetInfoLogLevel(InfoLogLevel::INFO_LEVEL);
S
sdong 已提交
1063 1064 1065
  test_logger.log_count = 0;
  test_logger.char_x_count = 0;
  test_logger.char_0_count = 0;
I
Igor Canadi 已提交
1066
  LogBuffer log_buffer(InfoLogLevel::INFO_LEVEL, &test_logger);
1067
  LogBuffer log_buffer_debug(DEBUG_LEVEL, &test_logger);
S
sdong 已提交
1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078

  char bytes200[200];
  std::fill_n(bytes200, sizeof(bytes200), '1');
  bytes200[sizeof(bytes200) - 1] = '\0';
  char bytes600[600];
  std::fill_n(bytes600, sizeof(bytes600), '1');
  bytes600[sizeof(bytes600) - 1] = '\0';
  char bytes9000[9000];
  std::fill_n(bytes9000, sizeof(bytes9000), '1');
  bytes9000[sizeof(bytes9000) - 1] = '\0';

1079 1080 1081 1082 1083
  ROCKS_LOG_BUFFER(&log_buffer, "x%sx", bytes200);
  ROCKS_LOG_BUFFER(&log_buffer, "x%sx", bytes600);
  ROCKS_LOG_BUFFER(&log_buffer, "x%sx%sx%sx", bytes200, bytes200, bytes200);
  ROCKS_LOG_BUFFER(&log_buffer, "x%sx%sx", bytes200, bytes600);
  ROCKS_LOG_BUFFER(&log_buffer, "x%sx%sx", bytes600, bytes9000);
S
sdong 已提交
1084

1085
  ROCKS_LOG_BUFFER(&log_buffer_debug, "x%sx", bytes200);
1086
  test_logger.SetInfoLogLevel(DEBUG_LEVEL);
1087 1088
  ROCKS_LOG_BUFFER(&log_buffer_debug, "x%sx%sx%sx", bytes600, bytes9000,
                   bytes200);
S
sdong 已提交
1089 1090 1091 1092 1093 1094 1095 1096 1097

  ASSERT_EQ(0, test_logger.log_count);
  log_buffer.FlushBufferToLog();
  log_buffer_debug.FlushBufferToLog();
  ASSERT_EQ(6, test_logger.log_count);
  ASSERT_EQ(6, test_logger.char_0_count);
  ASSERT_EQ(10, test_logger.char_x_count);
}

1098 1099 1100
class TestLogger2 : public Logger {
 public:
  explicit TestLogger2(size_t max_log_size) : max_log_size_(max_log_size) {}
F
fyrz 已提交
1101
  using Logger::Logv;
1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118
  virtual void Logv(const char* format, va_list ap) override {
    char new_format[2000];
    std::fill_n(new_format, sizeof(new_format), '2');
    {
      va_list backup_ap;
      va_copy(backup_ap, ap);
      int n = vsnprintf(new_format, sizeof(new_format) - 1, format, backup_ap);
      // 48 bytes for extra information + bytes allocated
      ASSERT_TRUE(
          n <= 48 + static_cast<int>(max_log_size_ - sizeof(struct timeval)));
      ASSERT_TRUE(n > static_cast<int>(max_log_size_ - sizeof(struct timeval)));
      va_end(backup_ap);
    }
  }
  size_t max_log_size_;
};

A
Andrew Kryczka 已提交
1119
TEST_P(EnvPosixTestWithParam, LogBufferMaxSizeTest) {
1120 1121 1122 1123 1124 1125 1126 1127 1128
  char bytes9000[9000];
  std::fill_n(bytes9000, sizeof(bytes9000), '1');
  bytes9000[sizeof(bytes9000) - 1] = '\0';

  for (size_t max_log_size = 256; max_log_size <= 1024;
       max_log_size += 1024 - 256) {
    TestLogger2 test_logger(max_log_size);
    test_logger.SetInfoLogLevel(InfoLogLevel::INFO_LEVEL);
    LogBuffer log_buffer(InfoLogLevel::INFO_LEVEL, &test_logger);
1129
    ROCKS_LOG_BUFFER_MAX_SZ(&log_buffer, max_log_size, "%s", bytes9000);
1130 1131 1132 1133
    log_buffer.FlushBufferToLog();
  }
}

A
Andrew Kryczka 已提交
1134
TEST_P(EnvPosixTestWithParam, Preallocation) {
K
krad 已提交
1135 1136 1137 1138
  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
    const std::string src = test::TmpDir(env_) + "/" + "testfile";
    unique_ptr<WritableFile> srcfile;
    EnvOptions soptions;
A
Aaron Gao 已提交
1139
    soptions.use_direct_reads = soptions.use_direct_writes = direct_io_;
Z
zach shipko 已提交
1140
#if !defined(OS_MACOSX) && !defined(OS_WIN) && !defined(OS_SOLARIS) && !defined(OS_AIX) && !defined(OS_OPENBSD)
K
krad 已提交
1141 1142 1143 1144 1145 1146 1147
    if (soptions.use_direct_writes) {
      rocksdb::SyncPoint::GetInstance()->SetCallBack(
          "NewWritableFile:O_DIRECT", [&](void* arg) {
            int* val = static_cast<int*>(arg);
            *val &= ~O_DIRECT;
          });
    }
K
krad 已提交
1148
#endif
K
krad 已提交
1149 1150 1151 1152 1153 1154 1155 1156 1157
    ASSERT_OK(env_->NewWritableFile(src, &srcfile, soptions));
    srcfile->SetPreallocationBlockSize(1024 * 1024);

    // No writes should mean no preallocation
    size_t block_size, last_allocated_block;
    srcfile->GetPreallocationStatus(&block_size, &last_allocated_block);
    ASSERT_EQ(last_allocated_block, 0UL);

    // Small write should preallocate one block
A
Aaron Gao 已提交
1158
    size_t kStrSize = 4096;
K
krad 已提交
1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185
    auto data = NewAligned(kStrSize, 'A');
    Slice str(data.get(), kStrSize);
    srcfile->PrepareWrite(srcfile->GetFileSize(), kStrSize);
    srcfile->Append(str);
    srcfile->GetPreallocationStatus(&block_size, &last_allocated_block);
    ASSERT_EQ(last_allocated_block, 1UL);

    // Write an entire preallocation block, make sure we increased by two.
    {
      auto buf_ptr = NewAligned(block_size, ' ');
      Slice buf(buf_ptr.get(), block_size);
      srcfile->PrepareWrite(srcfile->GetFileSize(), block_size);
      srcfile->Append(buf);
      srcfile->GetPreallocationStatus(&block_size, &last_allocated_block);
      ASSERT_EQ(last_allocated_block, 2UL);
    }

    // Write five more blocks at once, ensure we're where we need to be.
    {
      auto buf_ptr = NewAligned(block_size * 5, ' ');
      Slice buf = Slice(buf_ptr.get(), block_size * 5);
      srcfile->PrepareWrite(srcfile->GetFileSize(), buf.size());
      srcfile->Append(buf);
      srcfile->GetPreallocationStatus(&block_size, &last_allocated_block);
      ASSERT_EQ(last_allocated_block, 7UL);
    }
  rocksdb::SyncPoint::GetInstance()->ClearTrace();
1186 1187
}

1188 1189
// Test that the two ways to get children file attributes (in bulk or
// individually) behave consistently.
A
Andrew Kryczka 已提交
1190
TEST_P(EnvPosixTestWithParam, ConsistentChildrenAttributes) {
K
krad 已提交
1191 1192
  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
    EnvOptions soptions;
A
Aaron Gao 已提交
1193
    soptions.use_direct_reads = soptions.use_direct_writes = direct_io_;
K
krad 已提交
1194 1195 1196 1197 1198 1199 1200 1201
    const int kNumChildren = 10;

    std::string data;
    for (int i = 0; i < kNumChildren; ++i) {
      std::ostringstream oss;
      oss << test::TmpDir(env_) << "/testfile_" << i;
      const std::string path = oss.str();
      unique_ptr<WritableFile> file;
Z
zach shipko 已提交
1202
#if !defined(OS_MACOSX) && !defined(OS_WIN) && !defined(OS_SOLARIS) && !defined(OS_AIX) && !defined(OS_OPENBSD)
K
krad 已提交
1203 1204 1205 1206 1207 1208 1209
      if (soptions.use_direct_writes) {
        rocksdb::SyncPoint::GetInstance()->SetCallBack(
            "NewWritableFile:O_DIRECT", [&](void* arg) {
              int* val = static_cast<int*>(arg);
              *val &= ~O_DIRECT;
            });
      }
K
krad 已提交
1210
#endif
K
krad 已提交
1211 1212 1213 1214
      ASSERT_OK(env_->NewWritableFile(path, &file, soptions));
      auto buf_ptr = NewAligned(data.size(), 'T');
      Slice buf(buf_ptr.get(), data.size());
      file->Append(buf);
A
Aaron Gao 已提交
1215
      data.append(std::string(4096, 'T'));
K
krad 已提交
1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231
    }

    std::vector<Env::FileAttributes> file_attrs;
    ASSERT_OK(env_->GetChildrenFileAttributes(test::TmpDir(env_), &file_attrs));
    for (int i = 0; i < kNumChildren; ++i) {
      std::ostringstream oss;
      oss << "testfile_" << i;
      const std::string name = oss.str();
      const std::string path = test::TmpDir(env_) + "/" + name;

      auto file_attrs_iter = std::find_if(
          file_attrs.begin(), file_attrs.end(),
          [&name](const Env::FileAttributes& fm) { return fm.name == name; });
      ASSERT_TRUE(file_attrs_iter != file_attrs.end());
      uint64_t size;
      ASSERT_OK(env_->GetFileSize(path, &size));
A
Aaron Gao 已提交
1232
      ASSERT_EQ(size, 4096 * i);
K
krad 已提交
1233 1234
      ASSERT_EQ(size, file_attrs_iter->size_bytes);
    }
1235
    rocksdb::SyncPoint::GetInstance()->ClearTrace();
1236 1237
}

1238
// Test that all WritableFileWrapper forwards all calls to WritableFile.
A
Andrew Kryczka 已提交
1239
TEST_P(EnvPosixTestWithParam, WritableFileWrapper) {
1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251
  class Base : public WritableFile {
   public:
    mutable int *step_;

    void inc(int x) const {
      EXPECT_EQ(x, (*step_)++);
    }

    explicit Base(int* step) : step_(step) {
      inc(0);
    }

1252 1253
    Status Append(const Slice& data) override { inc(1); return Status::OK(); }
    Status Truncate(uint64_t size) override { return Status::OK(); }
1254 1255 1256 1257
    Status Close() override { inc(2); return Status::OK(); }
    Status Flush() override { inc(3); return Status::OK(); }
    Status Sync() override { inc(4); return Status::OK(); }
    Status Fsync() override { inc(5); return Status::OK(); }
1258
    void SetIOPriority(Env::IOPriority pri) override { inc(6); }
1259
    uint64_t GetFileSize() override { inc(7); return 0; }
1260 1261
    void GetPreallocationStatus(size_t* block_size,
                                size_t* last_allocated_block) override {
1262 1263
      inc(8);
    }
1264
    size_t GetUniqueId(char* id, size_t max_size) const override {
1265 1266 1267
      inc(9);
      return 0;
    }
1268
    Status InvalidateCache(size_t offset, size_t length) override {
1269 1270 1271 1272 1273
      inc(10);
      return Status::OK();
    }

   protected:
1274
    Status Allocate(uint64_t offset, uint64_t len) override {
1275 1276 1277
      inc(11);
      return Status::OK();
    }
1278
    Status RangeSync(uint64_t offset, uint64_t nbytes) override {
1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319
      inc(12);
      return Status::OK();
    }

   public:
    ~Base() {
      inc(13);
    }
  };

  class Wrapper : public WritableFileWrapper {
   public:
    explicit Wrapper(WritableFile* target) : WritableFileWrapper(target) {}

    void CallProtectedMethods() {
      Allocate(0, 0);
      RangeSync(0, 0);
    }
  };

  int step = 0;

  {
    Base b(&step);
    Wrapper w(&b);
    w.Append(Slice());
    w.Close();
    w.Flush();
    w.Sync();
    w.Fsync();
    w.SetIOPriority(Env::IOPriority::IO_HIGH);
    w.GetFileSize();
    w.GetPreallocationStatus(nullptr, nullptr);
    w.GetUniqueId(nullptr, 0);
    w.InvalidateCache(0, 0);
    w.CallProtectedMethods();
  }

  EXPECT_EQ(14, step);
}

I
Islam AbdelRahman 已提交
1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378 1379 1380 1381 1382 1383 1384 1385 1386 1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471
TEST_P(EnvPosixTestWithParam, PosixRandomRWFile) {
  const std::string path = test::TmpDir(env_) + "/random_rw_file";

  env_->DeleteFile(path);

  std::unique_ptr<RandomRWFile> file;
  ASSERT_OK(env_->NewRandomRWFile(path, &file, EnvOptions()));

  char buf[10000];
  Slice read_res;

  ASSERT_OK(file->Write(0, "ABCD"));
  ASSERT_OK(file->Read(0, 10, &read_res, buf));
  ASSERT_EQ(read_res.ToString(), "ABCD");

  ASSERT_OK(file->Write(2, "XXXX"));
  ASSERT_OK(file->Read(0, 10, &read_res, buf));
  ASSERT_EQ(read_res.ToString(), "ABXXXX");

  ASSERT_OK(file->Write(10, "ZZZ"));
  ASSERT_OK(file->Read(10, 10, &read_res, buf));
  ASSERT_EQ(read_res.ToString(), "ZZZ");

  ASSERT_OK(file->Write(11, "Y"));
  ASSERT_OK(file->Read(10, 10, &read_res, buf));
  ASSERT_EQ(read_res.ToString(), "ZYZ");

  ASSERT_OK(file->Write(200, "FFFFF"));
  ASSERT_OK(file->Read(200, 10, &read_res, buf));
  ASSERT_EQ(read_res.ToString(), "FFFFF");

  ASSERT_OK(file->Write(205, "XXXX"));
  ASSERT_OK(file->Read(200, 10, &read_res, buf));
  ASSERT_EQ(read_res.ToString(), "FFFFFXXXX");

  ASSERT_OK(file->Write(5, "QQQQ"));
  ASSERT_OK(file->Read(0, 9, &read_res, buf));
  ASSERT_EQ(read_res.ToString(), "ABXXXQQQQ");

  ASSERT_OK(file->Read(2, 4, &read_res, buf));
  ASSERT_EQ(read_res.ToString(), "XXXQ");

  // Close file and reopen it
  file->Close();
  ASSERT_OK(env_->NewRandomRWFile(path, &file, EnvOptions()));

  ASSERT_OK(file->Read(0, 9, &read_res, buf));
  ASSERT_EQ(read_res.ToString(), "ABXXXQQQQ");

  ASSERT_OK(file->Read(10, 3, &read_res, buf));
  ASSERT_EQ(read_res.ToString(), "ZYZ");

  ASSERT_OK(file->Read(200, 9, &read_res, buf));
  ASSERT_EQ(read_res.ToString(), "FFFFFXXXX");

  ASSERT_OK(file->Write(4, "TTTTTTTTTTTTTTTT"));
  ASSERT_OK(file->Read(0, 10, &read_res, buf));
  ASSERT_EQ(read_res.ToString(), "ABXXTTTTTT");

  // Clean up
  env_->DeleteFile(path);
}

class RandomRWFileWithMirrorString {
 public:
  explicit RandomRWFileWithMirrorString(RandomRWFile* _file) : file_(_file) {}

  void Write(size_t offset, const std::string& data) {
    // Write to mirror string
    StringWrite(offset, data);

    // Write to file
    Status s = file_->Write(offset, data);
    ASSERT_OK(s) << s.ToString();
  }

  void Read(size_t offset = 0, size_t n = 1000000) {
    Slice str_res(nullptr, 0);
    if (offset < file_mirror_.size()) {
      size_t str_res_sz = std::min(file_mirror_.size() - offset, n);
      str_res = Slice(file_mirror_.data() + offset, str_res_sz);
      StopSliceAtNull(&str_res);
    }

    Slice file_res;
    Status s = file_->Read(offset, n, &file_res, buf_);
    ASSERT_OK(s) << s.ToString();
    StopSliceAtNull(&file_res);

    ASSERT_EQ(str_res.ToString(), file_res.ToString()) << offset << " " << n;
  }

  void SetFile(RandomRWFile* _file) { file_ = _file; }

 private:
  void StringWrite(size_t offset, const std::string& src) {
    if (offset + src.size() > file_mirror_.size()) {
      file_mirror_.resize(offset + src.size(), '\0');
    }

    char* pos = const_cast<char*>(file_mirror_.data() + offset);
    memcpy(pos, src.data(), src.size());
  }

  void StopSliceAtNull(Slice* slc) {
    for (size_t i = 0; i < slc->size(); i++) {
      if ((*slc)[i] == '\0') {
        *slc = Slice(slc->data(), i);
        break;
      }
    }
  }

  char buf_[10000];
  RandomRWFile* file_;
  std::string file_mirror_;
};

TEST_P(EnvPosixTestWithParam, PosixRandomRWFileRandomized) {
  const std::string path = test::TmpDir(env_) + "/random_rw_file_rand";
  env_->DeleteFile(path);

  unique_ptr<RandomRWFile> file;
  ASSERT_OK(env_->NewRandomRWFile(path, &file, EnvOptions()));
  RandomRWFileWithMirrorString file_with_mirror(file.get());

  Random rnd(301);
  std::string buf;
  for (int i = 0; i < 10000; i++) {
    // Genrate random data
    test::RandomString(&rnd, 10, &buf);

    // Pick random offset for write
    size_t write_off = rnd.Next() % 1000;
    file_with_mirror.Write(write_off, buf);

    // Pick random offset for read
    size_t read_off = rnd.Next() % 1000;
    size_t read_sz = rnd.Next() % 20;
    file_with_mirror.Read(read_off, read_sz);

    if (i % 500 == 0) {
      // Reopen the file every 500 iters
      ASSERT_OK(env_->NewRandomRWFile(path, &file, EnvOptions()));
      file_with_mirror.SetFile(file.get());
    }
  }

  // clean up
  env_->DeleteFile(path);
}

A
Aaron Gao 已提交
1472 1473 1474 1475 1476 1477 1478 1479 1480
INSTANTIATE_TEST_CASE_P(DefaultEnvWithoutDirectIO, EnvPosixTestWithParam,
                        ::testing::Values(std::pair<Env*, bool>(Env::Default(),
                                                                false)));
#if !defined(ROCKSDB_LITE)
INSTANTIATE_TEST_CASE_P(DefaultEnvWithDirectIO, EnvPosixTestWithParam,
                        ::testing::Values(std::pair<Env*, bool>(Env::Default(),
                                                                true)));
#endif  // !defined(ROCKSDB_LITE)

A
Andrew Kryczka 已提交
1481
#if !defined(ROCKSDB_LITE) && !defined(OS_WIN)
1482 1483
static unique_ptr<Env> chroot_env(NewChrootEnv(Env::Default(),
                                               test::TmpDir(Env::Default())));
A
Aaron Gao 已提交
1484 1485 1486 1487 1488 1489
INSTANTIATE_TEST_CASE_P(
    ChrootEnvWithoutDirectIO, EnvPosixTestWithParam,
    ::testing::Values(std::pair<Env*, bool>(chroot_env.get(), false)));
INSTANTIATE_TEST_CASE_P(
    ChrootEnvWithDirectIO, EnvPosixTestWithParam,
    ::testing::Values(std::pair<Env*, bool>(chroot_env.get(), true)));
A
Andrew Kryczka 已提交
1490 1491
#endif  // !defined(ROCKSDB_LITE) && !defined(OS_WIN)

1492
}  // namespace rocksdb
J
jorlow@chromium.org 已提交
1493 1494

int main(int argc, char** argv) {
I
Igor Sugak 已提交
1495 1496
  ::testing::InitGoogleTest(&argc, argv);
  return RUN_ALL_TESTS();
J
jorlow@chromium.org 已提交
1497
}