Windows cumulative patch

Summary:
This patch addressed several issues.
  Portability including db_test std::thread -> port::Thread Cc: @
  and %z to ROCKSDB portable macro. Cc: maysamyabandeh

  Implement Env::AreFilesSame

  Make the implementation of file unique number more robust

  Get rid of C-runtime and go directly to Windows API when dealing
  with file primitives.

  Implement GetSectorSize() and aling unbuffered read on the value if
  available.

  Adjust Windows Logger for the new interface, implement CloseImpl() Cc: anand1976

  Fix test running script issue where $status var was of incorrect scope
  so the failures were swallowed and not reported.

  DestroyDB() creates a logger and opens a LOG file in the directory
  being cleaned up. This holds a lock on the folder and the cleanup is
  prevented. This fails one of the checkpoin tests. We observe the same in production.
  We close the log file in this change.

 Fix DBTest2.ReadAmpBitmapLiveInCacheAfterDBClose failure where the test
 attempts to open a directory with NewRandomAccessFile which does not
 work on Windows.
  Fix DBTest.SoftLimit as it is dependent on thread timing. CC: yiwu-arbug
Closes facebook#3552

Differential Revision: D7156304

Pulled By: siying

fbshipit-source-id: 43db0a757f1dfceffeb2b7988043156639173f5b

Author: yuslepukhin

build_tools/run_ci_db_test.ps1

@@ -336,7 +336,7 @@ $InvokeTestAsync = {
 # Test limiting factor here
 [int]$count = 0
 # Overall status
-[bool]$success = $true;
+[bool]$script:success = $true;
 
 function RunJobs($Suites, $TestCmds, [int]$ConcurrencyVal)
 {
@@ -425,7 +425,7 @@ function RunJobs($Suites, $TestCmds, [int]$ConcurrencyVal)
         $log_content = @(Get-Content $log)
 
         if($completed.State -ne "Completed") {
-            $success = $false
+            $script:success = $false
             Write-Warning $message
             $log_content | Write-Warning
         } else {
@@ -449,7 +449,7 @@ function RunJobs($Suites, $TestCmds, [int]$ConcurrencyVal)
             }
 
             if(!$pass_found) {
-                $success = $false;
+                $script:success = $false;
                 Write-Warning $message
                 $log_content | Write-Warning
             } else {
@@ -473,7 +473,7 @@ New-TimeSpan -Start $StartDate -End $EndDate |
   }
 
 
-if(!$success) {
+if(!$script:success) {
 # This does not succeed killing off jobs quick
 # So we simply exit
 #    Remove-Job -Job $jobs -Force

db/db_impl.cc

@@ -2389,10 +2389,13 @@ Snapshot::~Snapshot() {
 }
 
 Status DestroyDB(const std::string& dbname, const Options& options) {
-  const ImmutableDBOptions soptions(SanitizeOptions(dbname, options));
+  ImmutableDBOptions soptions(SanitizeOptions(dbname, options));
   Env* env = soptions.env;
   std::vector<std::string> filenames;
 
+  // Reset the logger because it holds a handle to the
+  // log file and prevents cleanup and directory removal
+  soptions.info_log.reset();
   // Ignore error in case directory does not exist
   env->GetChildren(dbname, &filenames);
 

db/db_test.cc

@@ -5153,7 +5153,7 @@ TEST_F(DBTest, AutomaticConflictsWithManualCompaction) {
     }
     ASSERT_OK(Flush());
   }
-  std::thread manual_compaction_thread([this]() {
+  port::Thread manual_compaction_thread([this]() {
     CompactRangeOptions croptions;
     croptions.exclusive_manual_compaction = true;
     ASSERT_OK(db_->CompactRange(croptions, nullptr, nullptr));
@@ -5393,18 +5393,42 @@ TEST_F(DBTest, HardLimit) {
 #ifndef ROCKSDB_LITE
 class WriteStallListener : public EventListener {
  public:
-  WriteStallListener() : condition_(WriteStallCondition::kNormal) {}
+  WriteStallListener() : cond_(&mutex_), 
+    condition_(WriteStallCondition::kNormal),
+    expected_(WriteStallCondition::kNormal),
+    expected_set_(false) 
+  {}
   void OnStallConditionsChanged(const WriteStallInfo& info) override {
     MutexLock l(&mutex_);
     condition_ = info.condition.cur;
+    if (expected_set_ && 
+      condition_ == expected_) {
+        cond_.Signal();
+        expected_set_ = false;
+    }
   }
   bool CheckCondition(WriteStallCondition expected) {
     MutexLock l(&mutex_);
-    return expected == condition_;
+    if (expected != condition_) {
+      expected_ = expected;
+      expected_set_ = true;
+      while (expected != condition_) {
+        // We bail out on timeout 500 milliseconds
+        const uint64_t timeout_us = 500000;
+        if (cond_.TimedWait(timeout_us)) {
+          expected_set_ = false;
+          return false;
+        }
+      }
+    }
+    return true;
   }
  private:
-  port::Mutex mutex_;
+  port::Mutex   mutex_;
+  port::CondVar cond_;
   WriteStallCondition condition_;
+  WriteStallCondition expected_;
+  bool                expected_set_;
 };
 
 TEST_F(DBTest, SoftLimit) {
@@ -5743,7 +5767,7 @@ TEST_F(DBTest, ThreadLocalPtrDeadlock) {
     return flushes_done.load() > 10;
   };
 
-  std::thread flushing_thread([&] {
+  port::Thread flushing_thread([&] {
     for (int i = 0; !done(); ++i) {
       ASSERT_OK(db_->Put(WriteOptions(), Slice("hi"),
                          Slice(std::to_string(i).c_str())));
@@ -5753,12 +5777,12 @@ TEST_F(DBTest, ThreadLocalPtrDeadlock) {
     }
   });
 
-  std::vector<std::thread> thread_spawning_threads(10);
+  std::vector<port::Thread> thread_spawning_threads(10);
   for (auto& t: thread_spawning_threads) {
-    t = std::thread([&] {
+    t = port::Thread([&] {
       while (!done()) {
         {
-          std::thread tmp_thread([&] {
+          port::Thread tmp_thread([&] {
             auto it = db_->NewIterator(ReadOptions());
             delete it;
           });

db/db_test2.cc

@@ -1822,14 +1822,26 @@ TEST_F(DBTest2, ReadAmpBitmapLiveInCacheAfterDBClose) {
   {
     const int kIdBufLen = 100;
     char id_buf[kIdBufLen];
+#ifndef OS_WIN
+    // You can't open a directory on windows using random access file
     std::unique_ptr<RandomAccessFile> file;
-    env_->NewRandomAccessFile(dbname_, &file, EnvOptions());
+    ASSERT_OK(env_->NewRandomAccessFile(dbname_, &file, EnvOptions()));
     if (file->GetUniqueId(id_buf, kIdBufLen) == 0) {
       // fs holding db directory doesn't support getting a unique file id,
       // this means that running this test will fail because lru_cache will load
       // the blocks again regardless of them being already in the cache
       return;
     }
+#else
+    std::unique_ptr<Directory> dir;
+    ASSERT_OK(env_->NewDirectory(dbname_, &dir));
+    if (dir->GetUniqueId(id_buf, kIdBufLen) == 0) {
+      // fs holding db directory doesn't support getting a unique file id,
+      // this means that running this test will fail because lru_cache will load
+      // the blocks again regardless of them being already in the cache
+      return;
+    }
+#endif
   }
   uint32_t bytes_per_bit[2] = {1, 16};
   for (size_t k = 0; k < 2; k++) {

env/env_test.cc

@@ -135,6 +135,37 @@ TEST_F(EnvPosixTest, RunImmediately) {
   }
 }
 
+#ifdef OS_WIN
+TEST_F(EnvPosixTest, AreFilesSame) {
+  {
+    bool tmp;
+    if (env_->AreFilesSame("", "", &tmp).IsNotSupported()) {
+      fprintf(stderr,
+              "skipping EnvBasicTestWithParam.AreFilesSame due to "
+              "unsupported Env::AreFilesSame\n");
+      return;
+    }
+  }
+
+  const EnvOptions soptions;
+  auto* env = Env::Default();
+  std::string same_file_name = test::TmpDir(env) + "/same_file";
+  std::string same_file_link_name = same_file_name + "_link";
+
+  std::unique_ptr<WritableFile> same_file;
+  ASSERT_OK(env->NewWritableFile(same_file_name,
+    &same_file, soptions));
+  same_file->Append("random_data");
+  ASSERT_OK(same_file->Flush());
+  same_file.reset();
+
+  ASSERT_OK(env->LinkFile(same_file_name, same_file_link_name));
+  bool result = false;
+  ASSERT_OK(env->AreFilesSame(same_file_name, same_file_link_name, &result));
+  ASSERT_TRUE(result);
+}
+#endif
+
 TEST_P(EnvPosixTestWithParam, UnSchedule) {
   std::atomic<bool> called(false);
   env_->SetBackgroundThreads(1, Env::LOW);

include/rocksdb/env.h

@@ -802,6 +802,10 @@ class Directory {
   virtual ~Directory() {}
   // Fsync directory. Can be called concurrently from multiple threads.
   virtual Status Fsync() = 0;
+
+  virtual size_t GetUniqueId(char* id, size_t max_size) const {
+    return 0;
+  }
 };
 
 enum InfoLogLevel : unsigned char {