Merging from FasterLog branch for epoch and allocator goodness.

microsoft · Oct 5, 2019 · 72cdfdb · 72cdfdb
2 parents b9404b6 + 6dc7af6
commit 72cdfdb
Show file tree

Hide file tree

Showing 32 changed files with 2,112 additions and 382 deletions.
diff --git a/cc/src/core/faster.h b/cc/src/core/faster.h
@@ -92,10 +92,10 @@ class FasterKv {
   typedef AsyncPendingRmwContext<key_t> async_pending_rmw_context_t;
 
   FasterKv(uint64_t table_size, uint64_t log_size, const std::string& filename,
-           double log_mutable_fraction = 0.9)
+           double log_mutable_fraction = 0.9, bool pre_allocate_log = false)
     : min_table_size_{ table_size }
     , disk{ filename, epoch_ }
-    , hlog{ log_size, epoch_, disk, disk.log(), log_mutable_fraction }
+    , hlog{ log_size, epoch_, disk, disk.log(), log_mutable_fraction, pre_allocate_log }
     , system_state_{ Action::None, Phase::REST, 1 }
     , num_pending_ios{ 0 } {
     if(!Utility::IsPowerOfTwo(table_size)) {
@@ -1405,22 +1405,33 @@ OperationStatus FasterKv<K, V, D>::InternalContinuePendingRmw(ExecutionContext&
   assert(address < hlog.begin_address.load() || address == pending_context->entry.address());
 
   // We have to do copy-on-write/RCU and write the updated value to the tail of the log.
-  uint32_t record_size = record_t::size(key, pending_context->value_size());
-  Address new_address = BlockAllocate(record_size);
-  record_t* new_record = reinterpret_cast<record_t*>(hlog.Get(new_address));
-
-  new(new_record) record_t{
-    RecordInfo{
-      static_cast<uint16_t>(context.version), true, false, false,
-      expected_entry.address() },
-    key };
+  Address new_address;
+  record_t* new_record;
   if(io_context.address < hlog.begin_address.load()) {
     // The on-disk trace back failed to find a key match.
+    uint32_t record_size = record_t::size(key, pending_context->value_size());
+    new_address = BlockAllocate(record_size);
+    new_record = reinterpret_cast<record_t*>(hlog.Get(new_address));
+
+    new(new_record) record_t{
+      RecordInfo{
+        static_cast<uint16_t>(context.version), true, false, false,
+        expected_entry.address() },
+      key };
     pending_context->RmwInitial(new_record);
   } else {
     // The record we read from disk.
     const record_t* disk_record = reinterpret_cast<const record_t*>(
                                     io_context.record.GetValidPointer());
+    uint32_t record_size = record_t::size(key, pending_context->value_size(disk_record));
+    new_address = BlockAllocate(record_size);
+    new_record = reinterpret_cast<record_t*>(hlog.Get(new_address));
+
+    new(new_record) record_t{
+      RecordInfo{
+        static_cast<uint16_t>(context.version), true, false, false,
+        expected_entry.address() },
+      key };
     pending_context->RmwCopy(disk_record, new_record);
   }
 

diff --git a/cc/src/core/persistent_memory_malloc.h b/cc/src/core/persistent_memory_malloc.h
@@ -244,7 +244,7 @@ class PersistentMemoryMalloc {
   static constexpr uint32_t kNumHeadPages = 4;
 
   PersistentMemoryMalloc(uint64_t log_size, LightEpoch& epoch, disk_t& disk_, log_file_t& file_,
-                         Address start_address, double log_mutable_fraction)
+                         Address start_address, double log_mutable_fraction, bool pre_allocate_log)
     : sector_size{ static_cast<uint32_t>(file_.alignment()) }
     , epoch_{ &epoch }
     , disk{ &disk_ }
@@ -260,7 +260,8 @@ class PersistentMemoryMalloc {
     , tail_page_offset_{ start_address }
     , buffer_size_{ 0 }
     , pages_{ nullptr }
-    , page_status_{ nullptr } {
+    , page_status_{ nullptr }
+    , pre_allocate_log_{ pre_allocate_log } {
     assert(start_address.page() <= Address::kMaxPage);
 
     if(log_size % kPageSize != 0) {
@@ -282,21 +283,28 @@ class PersistentMemoryMalloc {
       throw std::invalid_argument{ "Must have at least 2 mutable pages" };
     }
 
+    page_status_ = new FullPageStatus[buffer_size_];
+
     pages_ = new uint8_t* [buffer_size_];
     for(uint32_t idx = 0; idx < buffer_size_; ++idx) {
-      pages_[idx] = nullptr;
+      if (pre_allocate_log_) {
+        pages_[idx] = reinterpret_cast<uint8_t*>(aligned_alloc(sector_size, kPageSize));
+        std::memset(pages_[idx], 0, kPageSize);
+        // Mark the page as accessible.
+        page_status_[idx].status.store(FlushStatus::Flushed, CloseStatus::Open);
+      } else {
+        pages_[idx] = nullptr;
+      }
     }
 
-    page_status_ = new FullPageStatus[buffer_size_];
-
     PageOffset tail_page_offset = tail_page_offset_.load();
     AllocatePage(tail_page_offset.page());
     AllocatePage(tail_page_offset.page() + 1);
   }
 
   PersistentMemoryMalloc(uint64_t log_size, LightEpoch& epoch, disk_t& disk_, log_file_t& file_,
-                         double log_mutable_fraction)
-    : PersistentMemoryMalloc(log_size, epoch, disk_, file_, Address{ 0 }, log_mutable_fraction) {
+                         double log_mutable_fraction, bool pre_allocate_log)
+    : PersistentMemoryMalloc(log_size, epoch, disk_, file_, Address{ 0 }, log_mutable_fraction, pre_allocate_log) {
     /// Allocate the invalid page. Supports allocations aligned up to kCacheLineBytes.
     uint32_t discard;
     Allocate(Constants::kCacheLineBytes, discard);
@@ -555,6 +563,7 @@ class PersistentMemoryMalloc {
 
  private:
   uint32_t buffer_size_;
+  bool pre_allocate_log_;
 
   /// -- the latest N pages should be mutable.
   uint32_t num_mutable_pages_;
@@ -567,18 +576,20 @@ class PersistentMemoryMalloc {
 
   // Global address of the current tail (next element to be allocated from the circular buffer)
   AtomicPageOffset tail_page_offset_;
+
 };
 
 /// Implementations.
 template <class D>
 inline void PersistentMemoryMalloc<D>::AllocatePage(uint32_t index) {
   index = index % buffer_size_;
-  assert(pages_[index] == nullptr);
-  pages_[index] = reinterpret_cast<uint8_t*>(aligned_alloc(sector_size, kPageSize));;
-  std::memset(pages_[index], 0, kPageSize);
-
-  // Mark the page as accessible.
-  page_status_[index].status.store(FlushStatus::Flushed, CloseStatus::Open);
+  if (!pre_allocate_log_) {
+    assert(pages_[index] == nullptr);
+    pages_[index] = reinterpret_cast<uint8_t*>(aligned_alloc(sector_size, kPageSize));
+    std::memset(pages_[index], 0, kPageSize);
+    // Mark the page as accessible.
+    page_status_[index].status.store(FlushStatus::Flushed, CloseStatus::Open);
+  }
 }
 
 template <class D>

diff --git a/cs/FASTER.sln b/cs/FASTER.sln
@@ -44,6 +44,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "devices", "devices", "{A6B1
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "FASTER.devices.AzureStorageDevice", "src\devices\AzureStorageDevice\FASTER.devices.AzureStorageDevice.csproj", "{E571E686-01A0-44D5-BFF5-B7678284258B}"
 EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "FasterLogSample", "playground\FasterLogSample\FasterLogSample.csproj", "{25C5C6B6-4A8A-46DD-88C1-EB247033FE58}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -156,6 +158,12 @@ Global
 		{E571E686-01A0-44D5-BFF5-B7678284258B}.Release|Any CPU.Build.0 = Release|Any CPU
 		{E571E686-01A0-44D5-BFF5-B7678284258B}.Release|x64.ActiveCfg = Release|x64
 		{E571E686-01A0-44D5-BFF5-B7678284258B}.Release|x64.Build.0 = Release|x64
+		{25C5C6B6-4A8A-46DD-88C1-EB247033FE58}.Debug|Any CPU.ActiveCfg = Debug|x64
+		{25C5C6B6-4A8A-46DD-88C1-EB247033FE58}.Debug|x64.ActiveCfg = Debug|x64
+		{25C5C6B6-4A8A-46DD-88C1-EB247033FE58}.Debug|x64.Build.0 = Debug|x64
+		{25C5C6B6-4A8A-46DD-88C1-EB247033FE58}.Release|Any CPU.ActiveCfg = Release|x64
+		{25C5C6B6-4A8A-46DD-88C1-EB247033FE58}.Release|x64.ActiveCfg = Release|x64
+		{25C5C6B6-4A8A-46DD-88C1-EB247033FE58}.Release|x64.Build.0 = Release|x64
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -176,6 +184,7 @@ Global
 		{7EBB5ADF-D9EA-4B8B-AAE7-C48A98EBF780} = {E6026D6A-01C5-4582-B2C1-64751490DABE}
 		{A6B14415-D316-4955-BE5F-725BB2DEBEBE} = {28800357-C8CE-4CD0-A2AD-D4A910ABB496}
 		{E571E686-01A0-44D5-BFF5-B7678284258B} = {A6B14415-D316-4955-BE5F-725BB2DEBEBE}
+		{25C5C6B6-4A8A-46DD-88C1-EB247033FE58} = {E6026D6A-01C5-4582-B2C1-64751490DABE}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {A0750637-2CCB-4139-B25E-F2CE740DCFAC}

diff --git a/cs/playground/FasterLogSample/App.config b/cs/playground/FasterLogSample/App.config
@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="utf-8"?>
+<configuration>
+    <startup> 
+        <supportedRuntime version="v4.0" sku=".NETFramework,Version=v4.6" />
+    </startup>
+</configuration>
diff --git a/cs/playground/FasterLogSample/FasterLogSample.csproj b/cs/playground/FasterLogSample/FasterLogSample.csproj
@@ -0,0 +1,38 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>netcoreapp2.2</TargetFramework>
+    <Platforms>x64</Platforms>
+    <RuntimeIdentifier>win7-x64</RuntimeIdentifier>
+  </PropertyGroup>
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
+    <RootNamespace>StructSample</RootNamespace>
+    <ErrorReport>prompt</ErrorReport>
+    <RestoreProjectStyle>PackageReference</RestoreProjectStyle>
+    <Prefer32Bit>true</Prefer32Bit>
+  </PropertyGroup>
+
+  <PropertyGroup Condition="'$(Configuration)' == 'Debug'">
+    <DefineConstants>TRACE;DEBUG</DefineConstants>
+    <DebugType>full</DebugType>
+    <DebugSymbols>true</DebugSymbols>
+    <OutputPath>bin\x64\Debug\</OutputPath>
+  </PropertyGroup>
+  <PropertyGroup Condition="'$(Configuration)' == 'Release'">
+    <DefineConstants>TRACE</DefineConstants>
+    <DebugType>pdbonly</DebugType>
+    <Optimize>true</Optimize>
+    <OutputPath>bin\x64\Release\</OutputPath>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <None Include="App.config" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\src\core\FASTER.core.csproj" />
+  </ItemGroup>
+</Project>