Merge https://github.com/ned14/llfio/pull/77 into my current working tree and push the bits which are safe to push for now.

7 files changed, 210 insertions, 78 deletions

diff --git a/Build.md b/Build.md
index 6325a75e..9ccb6176 100644
--- a/Build.md
+++ b/Build.md
@@ -9,41 +9,9 @@ These compilers and OSs are regularly tested:
 
 Other compilers, architectures and OSs may work, but are not tested regularly.
 You will need a working [Filesystem TS](https://en.cppreference.com/w/cpp/experimental/fs)
-implementation in your STL, and at least C++ 14. 
-
-## Get a copy of the source
-
-Download [this archive](https://dedi5.nedprod.com/static/files/llfio-v2-source-latest.tar.xz)
-or clone from the GitHub repository:
-
-~~~
-git config --system core.longpaths true
-git clone --recursive https://github.com/ned14/llfio.git
-cd llfio
-~~~
-
-The first command is relevant so deeply nested paths on Windows will work when
-cloning the repository and submodules. It may require elevated privileges, but
-you can also use `git config --global core.longpaths true` instead.
-
-### If you already cloned before reading this
-
-If you had already cloned _this_ repository, but didn't use the `--recursive`
-switch, you can simply run the following command from inside the work tree:
-
-~~~
-git submodule update --init --recursive
-~~~
-
-## Header only usage
-
-LLFIO defaults to header only library configuration, so you don't actually need
-any of the prebuilt binaries below, or to build anything. Simply:
-
-~~~cpp
-#include "llfio/include/llfio.hpp"
-~~~
+implementation in your STL, and at least C++ 14.
 
+LLFIO has your choice of header-only, static library, and shared library build modes.
 Note that on Microsoft Windows, the default header only configuration is unsafe
 to use outside of toy projects. You will get warnings of the form:
 
@@ -67,14 +35,50 @@ which doesn't have this problem (define `LLFIO_EXPERIMENTAL_STATUS_CODE=1`)
 documentation](https://github.com/ned14/ntkernel-error-category)) (iii)
 Don't use header only LLFIO on Windows (see below).
 
+
+## Install from the vcpkg package manager
+
+This is particularly easy, and works on Mac OS, Linux and Microsoft Windows:
+
+```
+vcpkg install llfio
+```
+
+LLFIO appears at `<llfio/llfio.hpp>`.
+
 ## Prebuilt binaries
 
 It is faster to build programs using LLFIO if you don't use a header only build.
-In this situation, define `LLFIO_HEADERS_ONLY=0`, and choose one of `LLFIO_DYN_LINK` or `LLFIO_STATIC_LINK` depending on whether you are using the prebuilt shared or static libraries respectively.
+In this situation, define `LLFIO_HEADERS_ONLY=0`, and choose one of `LLFIO_DYN_LINK`
+or `LLFIO_STATIC_LINK` depending on whether you are using the prebuilt shared or
+static libraries respectively.
+
+You can find prebuilt binaries for Mac OS, Ubuntu and Microsoft Windows at
+https://github.com/ned14/llfio/releases. Choose a release, and under the Assets
+you will find the prebuilt binaries packages which include headers.
+
+## Get a copy of the source
+
+Clone from the GitHub repository:
+
+~~~
+git config --system core.longpaths true
+git clone --recursive https://github.com/ned14/llfio.git
+cd llfio
+~~~
+
+The first command is relevant so deeply nested paths on Windows will work when
+cloning the repository and submodules. It may require elevated privileges, but
+you can also use `git config --global core.longpaths true` instead.
+
+### If you already cloned before reading this
 
-- https://dedi5.nedprod.com/static/files/llfio-v2-binaries-darwin-latest.zip
-- https://dedi5.nedprod.com/static/files/llfio-v2-binaries-linux64-latest.tgz
-- https://dedi5.nedprod.com/static/files/llfio-v2-binaries-win64-latest.zip
+If you had already cloned _this_ repository, but didn't use the `--recursive`
+switch, you can simply run the following command from inside the work tree:
+
+~~~
+git submodule update --init --recursive
+~~~
 
 ## Build static libraries from source
 
diff --git a/include/llfio/llfio.hpp b/include/llfio/llfio.hpp
index 180df4ae..e5fbb0d9 100644
--- a/include/llfio/llfio.hpp
+++ b/include/llfio/llfio.hpp
@@ -1,7 +1,7 @@
 //! \file llfio/llfio.hpp The master *latest version* LLFIO include file. All LLFIO consuming libraries should include this header only.
 #include "version.hpp"
 
-#if defined(_MSC_VER) && !defined(__clang__)
+#if defined(_MSC_VER) && !defined(__clang__) && (!defined(_MSVC_TRADITIONAL) || _MSVC_TRADITIONAL)
 #define LLFIO_HEADERS_PATH2 LLFIO_VERSION_GLUE(v, LLFIO_HEADERS_VERSION, /llfio.hpp)
 #elif !__PCPP_ALWAYS_FALSE__
 #define LLFIO_HEADERS_PATH2 LLFIO_VERSION_GLUE(v, LLFIO_HEADERS_VERSION,)/llfio.hpp
diff --git a/include/llfio/v2.0/detail/impl/dynamic_thread_pool_group.ipp b/include/llfio/v2.0/detail/impl/dynamic_thread_pool_group.ipp
index 287683bf..08e6762e 100644
--- a/include/llfio/v2.0/detail/impl/dynamic_thread_pool_group.ipp
+++ b/include/llfio/v2.0/detail/impl/dynamic_thread_pool_group.ipp
@@ -1261,7 +1261,6 @@ class dynamic_thread_pool_group_impl final : public dynamic_thread_pool_group
 #else
   void *_grouph{nullptr};
   size_t _newly_added_active_work_items{0};
-  size_t _timer_work_items_remaining{0};
   size_t _active_work_items_remaining{0};
 #endif
 
diff --git a/include/llfio/v2.0/llfio.hpp b/include/llfio/v2.0/llfio.hpp
index a7c23b27..0cb651e6 100644
--- a/include/llfio/v2.0/llfio.hpp
+++ b/include/llfio/v2.0/llfio.hpp
@@ -23,7 +23,7 @@
 #define LLFIO_NAMESPACE_VERSION LLFIO_VERSION_GLUE(LLFIO_VERSION_MAJOR, _, LLFIO_VERSION_MINOR)
 
 #if defined(__cpp_modules) || defined(DOXYGEN_SHOULD_SKIP_THIS)
-#if defined(_MSC_VER) && !defined(__clang__)
+#if defined(_MSC_VER) && !defined(__clang__) && (!defined(_MSVC_TRADITIONAL) || _MSVC_TRADITIONAL)
 //! \brief The LLFIO C++ module name
 #define LLFIO_MODULE_NAME LLFIO_VERSION_GLUE(llfio_v, LLFIO_NAMESPACE_VERSION, )
 #else
diff --git a/include/llfio/version.hpp b/include/llfio/version.hpp
index f5421418..a9eb862e 100644
--- a/include/llfio/version.hpp
+++ b/include/llfio/version.hpp
@@ -25,7 +25,7 @@
 #define LLFIO_VERSION_GLUE(a, b, c) LLFIO_VERSION_GLUE2(a, b, c)
 #define LLFIO_NAMESPACE_VERSION LLFIO_VERSION_GLUE(LLFIO_VERSION_MAJOR, _, LLFIO_VERSION_MINOR)
 
-#if defined(_MSC_VER) && !defined(__clang__)
+#if defined(_MSC_VER) && !defined(__clang__) && (!defined(_MSVC_TRADITIONAL) || _MSVC_TRADITIONAL)
 #define LLFIO_HEADERS_VERSION LLFIO_VERSION_GLUE(LLFIO_VERSION_MAJOR, ., LLFIO_VERSION_MINOR)
 #else
 #define LLFIO_HEADERS_VERSION LLFIO_VERSION_MAJOR.LLFIO_VERSION_MINOR
diff --git a/programs/benchmark-io-congestion/main.cpp b/programs/benchmark-io-congestion/main.cpp
index e54cb25a..ea902938 100644
--- a/programs/benchmark-io-congestion/main.cpp
+++ b/programs/benchmark-io-congestion/main.cpp
@@ -25,11 +25,11 @@ Distributed under the Boost Software License, Version 1.0.
 //! Seconds to run the benchmark
 static constexpr unsigned BENCHMARK_DURATION = 10;
 //! Maximum work items to create
-static constexpr unsigned MAX_WORK_ITEMS = 512;
+static constexpr unsigned MAX_WORK_ITEMS = 4096;
 // Size of buffer to SHA256
-static constexpr unsigned SHA256_BUFFER_SIZE = 1024 * 1024;  // 1Mb
+static constexpr unsigned SHA256_BUFFER_SIZE = 4 * 1024;  // 64Kb
 // Size of test file
-static constexpr unsigned long long TEST_FILE_SIZE = 4ULL * SHA256_BUFFER_SIZE * MAX_WORK_ITEMS;  // 4Gb
+static constexpr unsigned long long TEST_FILE_SIZE = 4ULL * 1024 * 1024 * 1024;  // 4Gb
 
 #include "../../include/llfio/llfio.hpp"
 
@@ -58,6 +58,12 @@ static constexpr unsigned long long TEST_FILE_SIZE = 4ULL * SHA256_BUFFER_SIZE *
 
 namespace llfio = LLFIO_V2_NAMESPACE;
 
+struct benchmark_results
+{
+  std::chrono::microseconds duration;
+  llfio::utils::process_memory_usage memory_usage;
+};
+
 inline QUICKCPPLIB_NOINLINE void memcpy_s(llfio::byte *dest, const llfio::byte *s, size_t len)
 {
 #if defined(__SSE2__) || defined(_M_X64) || (defined(_M_IX86_FP) && _M_IX86_FP >= 2)
@@ -119,14 +125,14 @@ inline QUICKCPPLIB_NOINLINE void memcpy_s(llfio::byte *dest, const llfio::byte *
   }
 }
 
-#if 0
-struct llfio_runner
+struct llfio_runner_unpaced
 {
   std::atomic<bool> cancel{false};
   llfio::dynamic_thread_pool_group_ptr group = llfio::make_dynamic_thread_pool_group().value();
   std::vector<llfio::dynamic_thread_pool_group::work_item *> workitems;
 
-  ~llfio_runner()
+  llfio_runner_unpaced(llfio::io_handle * /*unused*/) {}
+  ~llfio_runner_unpaced()
   {
     for(auto *p : workitems)
     {
@@ -137,9 +143,9 @@ struct llfio_runner
   {
     struct workitem final : public llfio::dynamic_thread_pool_group::work_item
     {
-      llfio_runner *parent;
+      llfio_runner_unpaced *parent;
       F f;
-      workitem(llfio_runner *_parent, F &&_f)
+      workitem(llfio_runner_unpaced *_parent, F &&_f)
           : parent(_parent)
           , f(std::move(_f))
       {
@@ -153,18 +159,82 @@ struct llfio_runner
     };
     workitems.push_back(new workitem(this, std::move(f)));
   }
-  std::chrono::microseconds run(unsigned seconds)
+  benchmark_results run(unsigned seconds)
   {
     group->submit(workitems).value();
     auto begin = std::chrono::steady_clock::now();
     std::this_thread::sleep_for(std::chrono::seconds(seconds));
+    auto memusage = llfio::utils::current_process_memory_usage().value();
     cancel.store(true, std::memory_order_release);
     group->wait().value();
     auto end = std::chrono::steady_clock::now();
-    return std::chrono::duration_cast<std::chrono::microseconds>(end - begin);
+    return {std::chrono::duration_cast<std::chrono::microseconds>(end - begin), memusage};
   }
 };
+
+struct llfio_runner_paced
+{
+  std::atomic<bool> cancel{false};
+  llfio::dynamic_thread_pool_group_ptr group = llfio::make_dynamic_thread_pool_group().value();
+  llfio::dynamic_thread_pool_group::io_aware_work_item::io_handle_awareness awareness;
+  std::vector<llfio::dynamic_thread_pool_group::work_item *> workitems;
+  std::atomic<int64_t> last_pace{0};
+
+  llfio_runner_paced(llfio::io_handle *h)
+      : awareness{h, 1.0f /* 100% reads */}
+  {
+  }
+  ~llfio_runner_paced()
+  {
+    for(auto *p : workitems)
+    {
+      delete p;
+    }
+  }
+  template <class F> void add_workitem(F &&f)
+  {
+    struct workitem final : public llfio::dynamic_thread_pool_group::io_aware_work_item
+    {
+      llfio_runner_paced *parent;
+      F f;
+      workitem(llfio_runner_paced *_parent, F &&_f)
+          : llfio::dynamic_thread_pool_group::io_aware_work_item({&_parent->awareness, 1})
+          , parent(_parent)
+          , f(std::move(_f))
+      {
+      }
+      virtual intptr_t io_aware_next(llfio::deadline &d) noexcept override
+      {
+#if 1
+        auto last_pace = parent->last_pace.load(std::memory_order_relaxed);
+        if(last_pace != d.nsecs)
+        {
+          parent->last_pace.store(d.nsecs, std::memory_order_relaxed);
+          std::cout << "Pacing work by milliseconds " << (d.nsecs / 1000.0) << std::endl;
+        }
 #endif
+        return parent->cancel.load(std::memory_order_relaxed) ? -1 : 1;
+      }
+      virtual llfio::result<void> operator()(intptr_t /*unused*/) noexcept override
+      {
+        f();
+        return llfio::success();
+      }
+    };
+    workitems.push_back(new workitem(this, std::move(f)));
+  }
+  benchmark_results run(unsigned seconds)
+  {
+    group->submit(workitems).value();
+    auto begin = std::chrono::steady_clock::now();
+    std::this_thread::sleep_for(std::chrono::seconds(seconds));
+    auto memusage = llfio::utils::current_process_memory_usage().value();
+    cancel.store(true, std::memory_order_release);
+    group->wait().value();
+    auto end = std::chrono::steady_clock::now();
+    return {std::chrono::duration_cast<std::chrono::microseconds>(end - begin), memusage};
+  }
+};
 
 
 #if ENABLE_ASIO
@@ -173,6 +243,7 @@ struct asio_runner
   std::atomic<bool> cancel{false};
   asio::io_context ctx;
 
+  asio_runner(llfio::io_handle * /*unused*/) {}
   template <class F> struct C
   {
     asio_runner *parent;
@@ -192,7 +263,7 @@ struct asio_runner
     }
   };
   template <class F> void add_workitem(F &&f) { ctx.post(C<F>(this, std::move(f))); }
-  std::chrono::microseconds run(unsigned seconds)
+  benchmark_results run(unsigned seconds)
   {
     std::vector<std::thread> threads;
     auto do_cleanup = [&]() noexcept {
@@ -205,21 +276,37 @@ struct asio_runner
     auto cleanup = llfio::make_scope_exit(do_cleanup);
     for(size_t n = 0; n < MAX_WORK_ITEMS; n++)
     {
-      threads.emplace_back([&] { ctx.run(); });
+      try
+      {
+        threads.emplace_back([&] { ctx.run(); });
+      }
+      catch(...)
+      {
+        std::cerr << "Thread creation failed at number " << n << std::endl;
+        throw;
+      }
     }
     auto begin = std::chrono::steady_clock::now();
     std::this_thread::sleep_for(std::chrono::seconds(seconds));
+    auto memusage = llfio::utils::current_process_memory_usage().value();
     cleanup.release();
     do_cleanup();
     auto end = std::chrono::steady_clock::now();
-    return std::chrono::duration_cast<std::chrono::microseconds>(end - begin);
+    return {std::chrono::duration_cast<std::chrono::microseconds>(end - begin), memusage};
   }
 };
 #endif
 
-template <class Runner> void benchmark(llfio::span<llfio::byte> ioregion, const char *name)
+template <class Runner> void benchmark(llfio::mapped_file_handle &maph, const char *name)
 {
-  std::cout << "\nBenchmarking " << name << " ..." << std::endl;
+  if(name != nullptr)
+  {
+    std::cout << "\nBenchmarking " << name << " ..." << std::endl;
+  }
+  else
+  {
+    std::cout << "\nWarming up ..." << std::endl;
+  }
   struct shared_t
   {
     llfio::span<llfio::byte> ioregion;
@@ -240,8 +327,12 @@ template <class Runner> void benchmark(llfio::span<llfio::byte> ioregion, const
       {
         shared->max_concurrency.store(concurrency, std::memory_order_relaxed);
       }
+#if 1
+      auto offset = rand() % (TEST_FILE_SIZE - SHA256_BUFFER_SIZE - 1);
+#else
       auto offset = rand() & (TEST_FILE_SIZE - 1);
       offset &= ~(SHA256_BUFFER_SIZE - 1);
+#endif
       hash = QUICKCPPLIB_NAMESPACE::algorithm::hash::sha256_hash::hash(shared->ioregion.data() + offset, SHA256_BUFFER_SIZE);
       count++;
       shared->concurrency.fetch_sub(1, std::memory_order_relaxed);
@@ -253,38 +344,57 @@ template <class Runner> void benchmark(llfio::span<llfio::byte> ioregion, const
     }
   };
   std::vector<worker> workers;
-  std::vector<std::tuple<size_t, double, unsigned>> results;
+  struct result_t
+  {
+    size_t items;
+    double throughput;
+    size_t paged_in;
+    unsigned max_concurrency;
+  };
+  std::vector<result_t> results;
   for(size_t items = 1; items <= MAX_WORK_ITEMS; items <<= 1)
   {
-    shared_t shared{ioregion};
+    if(name == nullptr && items != 16)
+    {
+      continue;
+    }
+    shared_t shared{maph.map().as_span()};
     workers.clear();
     for(uint32_t n = 0; n < items; n++)
     {
       workers.emplace_back(&shared, n);
     }
-    Runner runner;
+    Runner runner(&maph);
     for(auto &i : workers)
     {
       runner.add_workitem([&] { i(); });
     }
-    auto duration = runner.run(BENCHMARK_DURATION);
+    auto out = runner.run(BENCHMARK_DURATION);
     uint64_t total = 0;
     for(auto &i : workers)
     {
       total += i.count;
     }
-    results.emplace_back(items, 1000000.0 * total / duration.count(), shared.max_concurrency);
-    std::cout << "   For " << std::get<0>(results.back()) << " work items got " << std::get<1>(results.back()) << " SHA256 hashes/sec with "
-              << (items * SHA256_BUFFER_SIZE / 1024.0 / 1024.0) << " Mb working set and " << std::get<2>(results.back()) << " maximum concurrency."
-              << std::endl;
+    results.push_back({items, 1000000.0 * total / out.duration.count(), out.memory_usage.total_address_space_paged_in, shared.max_concurrency});
+    std::cout << "   For " << results.back().items << " work items got " << results.back().throughput << " SHA256 hashes/sec with "
+              << (results.back().items * SHA256_BUFFER_SIZE / 1024.0 / 1024.0) << " Mb working set, " << results.back().max_concurrency
+              << " maximum concurrency, and " << (results.back().paged_in / 1024.0 / 1024.0) << " Mb paged in." << std::endl;
+    // std::cout << "      " << (out.memory_usage.total_address_space_in_use / 1024.0 / 1024.0) << ","
+    //          << (out.memory_usage.total_address_space_paged_in / 1024.0 / 1024.0) << "," << (out.memory_usage.private_committed / 1024.0 / 1024.0) << ","
+    //          << (out.memory_usage.private_paged_in / 1024.0 / 1024.0) << std::endl;
   }
-  std::ofstream out(std::string(name) + "_results.csv");
-  out << R"("Work items","SHA256 hashes/sec","Working set","Max concurrency")";
-  for(auto &i : results)
+  if(name != nullptr)
   {
-    out << "\n" << std::get<0>(i) << "," << std::get<1>(i) << "," << (std::get<0>(i) * SHA256_BUFFER_SIZE) << "," << std::get<2>(i);
+    std::ofstream out(std::string(name) + "_results.csv");
+    out << R"("Work items","SHA256 hashes/sec","Working set","Max concurrency","Paged in")";
+    for(auto &i : results)
+    {
+      out << "\n"
+          << i.items << "," << i.throughput << "," << (i.items * SHA256_BUFFER_SIZE / 1024.0 / 1024.0) << "," << i.max_concurrency << ","
+          << (i.paged_in / 1024.0 / 1024.0);
+    }
+    out << std::endl;
   }
-  out << std::endl;
 }
 
 int main(int argc, char *argv[])
@@ -308,35 +418,50 @@ int main(int argc, char *argv[])
       {
         fileh.close().value();
       }
+#if 0
       else
       {
-        std::cout << "Extending " << (TEST_FILE_SIZE / 1024.0 / 1024.0) << " Mb test file at " << fileh.current_path().value() << " ..." << std::endl;
+        std::cout << "Prefaulting " << (TEST_FILE_SIZE / 1024.0 / 1024.0) << " Mb test file at " << fileh.current_path().value() << " ..." << std::endl;
         std::vector<llfio::byte> buffer(SHA256_BUFFER_SIZE);
         for(size_t n = 0; n < TEST_FILE_SIZE; n += SHA256_BUFFER_SIZE)
         {
           memcpy_s(buffer.data(), fileh.address() + n, SHA256_BUFFER_SIZE);
         }
       }
+#endif
     }
     if(!fileh.is_valid())
     {
       fileh = llfio::mapped_file_handle::mapped_file(TEST_FILE_SIZE, where, "testfile", llfio::mapped_file_handle::mode::write,
-                                                     llfio::mapped_file_handle::creation::if_needed, llfio::mapped_file_handle::caching::reads_and_metadata)
+                                                     llfio::mapped_file_handle::creation::always_new, llfio::mapped_file_handle::caching::reads_and_metadata)
               .value();
       std::cout << "Writing " << (TEST_FILE_SIZE / 1024.0 / 1024.0) << " Mb test file at " << fileh.current_path().value() << " ..." << std::endl;
       fileh.truncate(TEST_FILE_SIZE).value();
       memset(fileh.address(), 0xff, TEST_FILE_SIZE);
     }
 
+    benchmark<llfio_runner_unpaced>(fileh, nullptr);
+
 #if 0
-    std::string llfio_name("llfio (");
-    llfio_name.append(llfio::dynamic_thread_pool_group::implementation_description());
-    llfio_name.push_back(')');
-    benchmark<llfio_runner>(llfio_name.c_str());
+    {
+      std::string llfio_name("llfio unpaced (");
+      llfio_name.append(llfio::dynamic_thread_pool_group::implementation_description());
+      llfio_name.push_back(')');
+      benchmark<llfio_runner_unpaced>(fileh, llfio_name.c_str());
+    }
 #endif
 
-#if ENABLE_ASIO
-    benchmark<asio_runner>(fileh.map().as_span(), "asio");
+#if 1
+    {
+      std::string llfio_name("llfio paced (");
+      llfio_name.append(llfio::dynamic_thread_pool_group::implementation_description());
+      llfio_name.push_back(')');
+      benchmark<llfio_runner_paced>(fileh, llfio_name.c_str());
+    }
+#endif
+
+#if 0  // ENABLE_ASIO
+    benchmark<asio_runner>(fileh, "asio");
 #endif
 
     std::cout << "\nReminder: you may wish to delete " << fileh.current_path().value() << std::endl;
diff --git a/test/tests/dynamic_thread_pool_group.cpp b/test/tests/dynamic_thread_pool_group.cpp
index dcaf292a..0c2b6dad 100644
--- a/test/tests/dynamic_thread_pool_group.cpp
+++ b/test/tests/dynamic_thread_pool_group.cpp
@@ -346,6 +346,10 @@ static inline void TestDynamicThreadPoolGroupNestingWorks()
 
 static inline void TestDynamicThreadPoolGroupIoAwareWorks()
 {
+  if(getenv("CI") != nullptr)
+  {
+    return;
+  }
   namespace llfio = LLFIO_V2_NAMESPACE;
   static constexpr size_t WORK_ITEMS = 1000;
   static constexpr size_t IO_SIZE = 1 * 65536;