From 996ec04851b7f286f0cc2abe1d71702a07b9de81 Mon Sep 17 00:00:00 2001 From: Mounir IDRASSI Date: Fri, 16 Oct 2026 21:17:31 +0900 Subject: [PATCH] Unix: drain encryption pool users before shutdown Retain accepted callers and KDF workers until completion, and run later requests synchronously. Clear joined threads and unwind partial startup so serialized restarts do not retain stale workers. Cover shutdown boundaries and startup failures with gated tests. --- .../encryption_thread_pool_lifecycle_test.cpp | 322 ++++++++++++++++++ .../test_encryption_thread_pool_exceptions.py | 49 +-- .../test_encryption_thread_pool_lifecycle.py | 35 ++ src/Volume/EncryptionThreadPool.cpp | 81 ++++- src/Volume/EncryptionThreadPool.h | 26 +- 5 files changed, 479 insertions(+), 34 deletions(-) create mode 100644 Tests/encryption_thread_pool_lifecycle_test.cpp create mode 100644 Tests/test_encryption_thread_pool_lifecycle.py diff --git a/Tests/encryption_thread_pool_lifecycle_test.cpp b/Tests/encryption_thread_pool_lifecycle_test.cpp new file mode 100644 index 00000000..147c9e3a --- /dev/null +++ b/Tests/encryption_thread_pool_lifecycle_test.cpp @@ -0,0 +1,322 @@ +/* Copyright (c) 2026 AM Crypto. Licensed under the Apache License 2.0. */ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "Volume/EncryptionThreadPool.h" +#include "Volume/Pkcs5Kdf.h" + +using namespace VeraCrypt; + +static thread_local int AllocationCalls, FailAllocationAt; +void *operator new (size_t size) +{ + if (++AllocationCalls == FailAllocationAt) + { + FailAllocationAt = 0; + throw std::bad_alloc(); + } + void *memory = std::malloc (size ? size : 1); + if (!memory) throw std::bad_alloc(); + return memory; +} +void operator delete (void *memory) noexcept { std::free (memory); } +void *operator new[] (size_t size) { return ::operator new (size); } +void operator delete[] (void *memory) noexcept { ::operator delete (memory); } + +static void Check (bool condition, const char *message) +{ + if (!condition) { std::cerr << "FAIL: " << message << std::endl; std::exit (1); } +} + +template static void Until (Predicate predicate) +{ + auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds (10); + while (!predicate()) + { + Check (std::chrono::steady_clock::now() < deadline, "barrier timed out"); + std::this_thread::yield(); + } +} + +class Gate +{ +public: + Gate () : Arrivals (0), Open (false) { } + void Enter () + { + std::unique_lock lock (Mutex); + ++Arrivals; + Changed.notify_all(); + Check (Changed.wait_for (lock, std::chrono::seconds (10), [this] { return Open; }), "gate timed out"); + } + void Wait (size_t count = 1) + { + std::unique_lock lock (Mutex); + Check (Changed.wait_for (lock, std::chrono::seconds (10), [=] { return Arrivals >= count; }), "arrival timed out"); + } + void Release () { std::lock_guard lock (Mutex); Open = true; Changed.notify_all(); } +private: + std::mutex Mutex; + std::condition_variable Changed; + size_t Arrivals; + bool Open; +}; + +class PoolView : public EncryptionThreadPool +{ +public: + static void LockQueue () { DequeueMutex.Lock(); } + static void UnlockQueue () { DequeueMutex.Unlock(); } + static void LockSubmit () { EnqueueMutex.Lock(); } + static void UnlockSubmit () { EnqueueMutex.Unlock(); } + static bool Closing () { return (PoolReferences.load() & PoolClosing) != 0; } + static size_t References () { return PoolReferences.load() & ~PoolClosing; } + static bool Empty () { return RunningThreads.empty() && ThreadCount == 0 && References() == 0; } + static SyncEvent *NextCompletion () { return &WorkItemQueue[EnqueuePosition].ItemCompletedEvent; } + static SyncEvent *SlotAvailable () { return &WorkItemCompletedEvent; } + static size_t Ready () + { + size_t count = 0; + for (size_t i = 0; i < QueueSize; ++i) if (WorkItemQueue[i].State == WorkItem::State::Ready) ++count; + return count; + } +}; + +static int CpuCount = 2, CreateCalls, JoinCalls, FailCreateAt; +static std::atomic PausedWait (nullptr), PausedSignal (nullptr); +static Gate *WaitGate, *SignalGate; +extern "C" long __real_sysconf (int); +extern "C" long __wrap_sysconf (int name) { return name == _SC_NPROCESSORS_ONLN ? CpuCount : __real_sysconf (name); } +extern "C" int __real_pthread_create (pthread_t *, const pthread_attr_t *, void *(*)(void *), void *); +extern "C" int __wrap_pthread_create (pthread_t *thread, const pthread_attr_t *attr, void *(*proc)(void *), void *arg) +{ + if (++CreateCalls == FailCreateAt) return EAGAIN; + return __real_pthread_create (thread, attr, proc, arg); +} +extern "C" int __real_pthread_join (pthread_t, void **); +extern "C" int __wrap_pthread_join (pthread_t thread, void **result) +{ + ++JoinCalls; + int status = __real_pthread_join (thread, result); + Check (status == 0, "invalid/repeated worker join"); + return status; +} +extern "C" void __real__ZN9VeraCrypt9SyncEvent4WaitEv (SyncEvent *); +extern "C" void __wrap__ZN9VeraCrypt9SyncEvent4WaitEv (SyncEvent *event) +{ + __real__ZN9VeraCrypt9SyncEvent4WaitEv (event); + if (event == PausedWait.load()) WaitGate->Enter(); +} +extern "C" void __real__ZN9VeraCrypt9SyncEvent6SignalEv (SyncEvent *); +extern "C" void __wrap__ZN9VeraCrypt9SyncEvent6SignalEv (SyncEvent *event) +{ + __real__ZN9VeraCrypt9SyncEvent6SignalEv (event); + SyncEvent *expected = event; + if (PausedSignal.compare_exchange_strong (expected, nullptr)) SignalGate->Enter(); +} + +class TestMode : public EncryptionMode +{ +public: + explicit TestMode (Gate *gate = nullptr) : Calls (0), Barrier (gate) { } + void Encrypt (uint8 *, uint64) const override { throw NotApplicable(); } + void Decrypt (uint8 *, uint64) const override { throw NotApplicable(); } + void EncryptSectorsCurrentThread (uint8 *data, uint64, uint64 count, size_t size) const override { Process (data, count * size); } + void DecryptSectorsCurrentThread (uint8 *data, uint64, uint64 count, size_t size) const override { Process (data, count * size); } + size_t GetKeySize () const override { return 0; } + std::wstring GetName () const override { return L"lifecycle-test"; } + shared_ptr GetNew () const override { return shared_ptr(); } + void SetKey (const ConstBufferPtr &) override { } + mutable std::atomic Calls; +private: + void Process (uint8 *data, size_t size) const + { + ++Calls; + if (Barrier) Barrier->Enter(); + std::memset (data, 0xa5, size); + } + Gate *Barrier; +}; + +class TestKdf : public Pkcs5Kdf +{ +public: + explicit TestKdf (Gate *gate = nullptr) : Barrier (gate) { } + int DeriveKey (const BufferPtr &key, const VolumePassword &, const ConstBufferPtr &, int) const override + { if (Barrier) Barrier->Enter(); std::memset (key.Get(), 0x5a, key.Size()); return 0; } + int DeriveKey (const BufferPtr &key, const VolumePassword &password, const ConstBufferPtr &salt, int count, long volatile *) const override + { return DeriveKey (key, password, salt, count); } + shared_ptr GetHash () const override { return shared_ptr(); } + int GetIterationCount (int) const override { return 1; } + std::wstring GetName () const override { return L"lifecycle-test"; } + Pkcs5Kdf *Clone () const override { return new TestKdf (Barrier); } +private: + Gate *Barrier; +}; + +struct KdfRequest +{ + VolumePassword Password; + uint8 Salt[64] = {}; + SyncEvent Completed, Drained; + SharedVal Outstanding; + std::vector> Items; + KdfRequest () : Outstanding (0) { } + void Submit (Gate *gate = nullptr) + { + shared_ptr kdf (new TestKdf (gate)); + shared_ptr item (new EncryptionThreadPool::KeyDerivationWorkItem (kdf, 64)); + Items.push_back (item); + EncryptionThreadPool::BeginKeyDerivation (*item, Password, 0, ConstBufferPtr (Salt, sizeof Salt), Completed, Drained, Outstanding, nullptr); + } + void Verify () + { + Drained.Wait(); + Check (Outstanding.Get() == 0, "KDF drain returned early"); + for (const auto &item : Items) + { + Check (item->Completed.Get() && item->Result == 0 && !item->ItemException.get(), "KDF completion lost"); + for (size_t i = 0; i < item->DerivedKey.Size(); ++i) Check (item->DerivedKey[i] == 0x5a, "KDF output incomplete"); + } + } +}; + +static void Join (std::future &task) +{ + Check (task.wait_for (std::chrono::seconds (10)) == std::future_status::ready, "task hung"); + task.get(); +} +static void StillWaiting (std::future &stop) +{ Check (stop.wait_for (std::chrono::milliseconds (50)) == std::future_status::timeout, "shutdown returned before its users"); } + +static void CipherShutdown (bool queued, EncryptionThreadPool::WorkType::Enum type) +{ + if (queued) PoolView::LockQueue(); + EncryptionThreadPool::Start(); + Gate workers, requester; + TestMode mode (queued ? nullptr : &workers); + std::vector data (CpuCount * 2 * ENCRYPTION_DATA_UNIT_SIZE); + WaitGate = &requester; + PausedWait = PoolView::NextCompletion(); + auto work = std::async (std::launch::async, [&] { EncryptionThreadPool::DoWork (type, &mode, data.data(), 0, CpuCount * 2, ENCRYPTION_DATA_UNIT_SIZE); }); + if (queued) Until ([] { return PoolView::Ready() == static_cast (CpuCount); }); + else workers.Wait (CpuCount); + auto stop = std::async (std::launch::async, [] { EncryptionThreadPool::Stop(); }); + Until ([] { return PoolView::Closing(); }); + StillWaiting (stop); + TestMode inlineMode; + uint8 inlineData[2 * ENCRYPTION_DATA_UNIT_SIZE] = {}; + EncryptionThreadPool::DoWork (type, &inlineMode, inlineData, 0, 2, ENCRYPTION_DATA_UNIT_SIZE); + Check (inlineMode.Calls == 1 && inlineData[sizeof inlineData - 1] == 0xa5, "late cipher did not run inline"); + if (queued) PoolView::UnlockQueue(); else workers.Release(); + requester.Wait(); + StillWaiting (stop); + requester.Release(); + Join (work); + Join (stop); + PausedWait = nullptr; + for (uint8 byte : data) Check (byte == 0xa5, "accepted cipher work was abandoned"); + Check (PoolView::Empty(), "cipher shutdown retained pool users"); +} + +static void KdfShutdown () +{ + EncryptionThreadPool::Start(); + Gate workers, finishing; + KdfRequest request; + for (int i = 0; i < CpuCount + 2; ++i) request.Submit (&workers); + workers.Wait (CpuCount); + SignalGate = &finishing; + PausedSignal = PoolView::SlotAvailable(); + auto stop = std::async (std::launch::async, [] { EncryptionThreadPool::Stop(); }); + Until ([] { return PoolView::Closing(); }); + StillWaiting (stop); + KdfRequest inlineRequest; + inlineRequest.Submit(); + inlineRequest.Verify(); + request.Submit(); + auto drain = std::async (std::launch::async, [&] { request.Verify(); }); + StillWaiting (drain); + workers.Release(); + finishing.Wait(); + Join (drain); + StillWaiting (stop); + finishing.Release(); + Join (stop); + Check (PoolView::Empty(), "KDF shutdown retained pool users"); +} + +static void BlockedSubmitter () +{ + EncryptionThreadPool::Start(); + PoolView::LockSubmit(); + KdfRequest request; + auto submit = std::async (std::launch::async, [&] { request.Submit(); }); + Until ([] { return PoolView::References() == 1; }); + auto stop = std::async (std::launch::async, [] { EncryptionThreadPool::Stop(); }); + Until ([] { return PoolView::Closing(); }); + StillWaiting (stop); + PoolView::UnlockSubmit(); + Join (submit); + request.Verify(); + Join (stop); + Check (PoolView::Empty(), "submitter shutdown retained pool users"); +} + +static void StartupFailures () +{ + for (int failure = 1; failure <= CpuCount; ++failure) + { + CreateCalls = JoinCalls = 0; + FailCreateAt = failure; + bool failed = false; + try { EncryptionThreadPool::Start(); } catch (const SystemException &) { failed = true; } + Check (failed && !EncryptionThreadPool::IsRunning(), "thread failure did not fail startup"); + Check (JoinCalls == failure - 1 && PoolView::Empty(), "partial startup did not join its workers"); + FailCreateAt = 0; + EncryptionThreadPool::Start(); + EncryptionThreadPool::Stop(); + Check (PoolView::Empty(), "restart retained joined threads"); + } + AllocationCalls = 0; + EncryptionThreadPool::Start(); + int allocations = AllocationCalls; + EncryptionThreadPool::Stop(); + for (int failure = 1; failure <= allocations; ++failure) + { + AllocationCalls = CreateCalls = JoinCalls = 0; + FailAllocationAt = failure; + bool failed = false; + try { EncryptionThreadPool::Start(); } catch (const std::bad_alloc &) { failed = true; } + Check (failed && !EncryptionThreadPool::IsRunning(), "allocation failure did not fail startup"); + Check (CreateCalls == JoinCalls && PoolView::Empty(), "allocation failure left an unjoined worker"); + EncryptionThreadPool::Start(); + EncryptionThreadPool::Stop(); + Check (PoolView::Empty(), "allocation-failure restart retained joined threads"); + } +} + +int main (int argc, char **argv) +{ + if (argc > 1) CpuCount = std::atoi (argv[1]); + StartupFailures(); + for (auto type : { EncryptionThreadPool::WorkType::EncryptDataUnits, EncryptionThreadPool::WorkType::DecryptDataUnits }) + { + CipherShutdown (true, type); + CipherShutdown (false, type); + } + KdfShutdown(); + BlockedSubmitter(); + KdfRequest stopped; + stopped.Submit(); + stopped.Verify(); + std::cout << "PASS: " << CpuCount << " workers: startup failures/restarts, queued/active cipher work, requester drain, KDF drain, blocked submitter and inline fallback" << std::endl; +} diff --git a/Tests/test_encryption_thread_pool_exceptions.py b/Tests/test_encryption_thread_pool_exceptions.py index 1bbc535f..31df1d6a 100644 --- a/Tests/test_encryption_thread_pool_exceptions.py +++ b/Tests/test_encryption_thread_pool_exceptions.py @@ -16,6 +16,33 @@ import tempfile ROOT = Path(__file__).resolve().parents[1] +def build_test(build, sanitize, test_source, extra_sources=(), extra_flags=()): + build.mkdir(parents=True, exist_ok=True) + sources = ["Volume/EncryptionThreadPool.cpp", "Volume/EncryptionMode.cpp"] + sources += ["Platform/" + name + ".cpp" for name in ( + "Buffer", "Exception", "Memory", "Serializable", "Serializer", "SerializerFactory", + "StringConverter", "Unix/Mutex", "Unix/SyncEvent", "Unix/SystemException", + "Unix/SystemLog", "Unix/Thread")] + sources += list(extra_sources) + exe = build / Path(test_source).stem + command = shlex.split(os.environ.get("CXX", "c++")) + [ + "-std=c++11", "-O2", "-g", "-pthread", "-ffunction-sections", "-fdata-sections", + "-Wl,--gc-sections", "-DTC_UNIX", "-DTC_LINUX", "-DARGON2_NO_THREADS", + "-D_FILE_OFFSET_BITS=64", "-I" + str(ROOT / "src"), + "-I" + str(ROOT / "src/Crypto"), "-I" + str(ROOT / "src/Crypto/Argon2/include")] + if sanitize: + command += ["-fsanitize=address", "-fno-omit-frame-pointer"] + command += list(extra_flags) + command += [str(ROOT / "Tests" / test_source)] + command += [str(ROOT / "src" / source) for source in sources] + command += ["-o", str(exe)] + result = subprocess.run(command, capture_output=True, text=True, timeout=180) + (build / "build.log").write_text(result.stdout + result.stderr) + if result.returncode: + raise RuntimeError(result.stdout + result.stderr) + return exe + + def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--build-dir", type=Path) @@ -24,27 +51,7 @@ def main(): if not sys.platform.startswith("linux"): parser.error("This build runner requires Linux") build = (args.build_dir or Path(tempfile.mkdtemp(prefix="vc-pool-exceptions-"))).resolve() - build.mkdir(parents=True, exist_ok=True) - sources = ["Volume/EncryptionThreadPool.cpp", "Volume/EncryptionMode.cpp"] - sources += ["Platform/" + name + ".cpp" for name in ( - "Buffer", "Exception", "Memory", "Serializable", "Serializer", "SerializerFactory", - "StringConverter", "Unix/Mutex", "Unix/SyncEvent", "Unix/SystemException", - "Unix/SystemLog", "Unix/Thread")] - exe = build / "encryption_thread_pool_exception_test" - command = shlex.split(os.environ.get("CXX", "c++")) + [ - "-std=c++11", "-O2", "-g", "-pthread", "-ffunction-sections", "-fdata-sections", - "-Wl,--gc-sections", "-DTC_UNIX", "-DTC_LINUX", "-DARGON2_NO_THREADS", - "-D_FILE_OFFSET_BITS=64", "-I" + str(ROOT / "src"), - "-I" + str(ROOT / "src/Crypto"), "-I" + str(ROOT / "src/Crypto/Argon2/include")] - if args.sanitize: - command += ["-fsanitize=address", "-fno-omit-frame-pointer"] - command += [str(ROOT / "Tests/encryption_thread_pool_exception_test.cpp")] - command += [str(ROOT / "src" / source) for source in sources] - command += ["-o", str(exe)] - result = subprocess.run(command, capture_output=True, text=True, timeout=180) - (build / "build.log").write_text(result.stdout + result.stderr) - if result.returncode: - raise RuntimeError(result.stdout + result.stderr) + exe = build_test(build, args.sanitize, "encryption_thread_pool_exception_test.cpp") result = subprocess.run([str(exe)], timeout=60) if result.returncode not in (0, 77): raise SystemExit(result.returncode) diff --git a/Tests/test_encryption_thread_pool_lifecycle.py b/Tests/test_encryption_thread_pool_lifecycle.py new file mode 100644 index 00000000..779e48d2 --- /dev/null +++ b/Tests/test_encryption_thread_pool_lifecycle.py @@ -0,0 +1,35 @@ +#!/usr/bin/env python3 +"""Exercise Unix pool shutdown and restart with controlled worker/requester gates. + +Uses the production pool, platform primitives, and KDF dispatch. Only the test +cipher/KDF, CPU count, thread-creation failures and event boundary gates are fake. +Run on Linux: python3 Tests/test_encryption_thread_pool_lifecycle.py [--sanitize] +""" +import argparse +from pathlib import Path +import subprocess +import sys +import tempfile + +from test_encryption_thread_pool_exceptions import build_test + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--build-dir", type=Path) + parser.add_argument("--sanitize", action="store_true") + args = parser.parse_args() + if not sys.platform.startswith("linux"): + parser.error("This build runner requires Linux") + build = (args.build_dir or Path(tempfile.mkdtemp(prefix="vc-pool-lifecycle-"))).resolve() + symbols = ("sysconf", "pthread_create", "pthread_join", + "_ZN9VeraCrypt9SyncEvent4WaitEv", "_ZN9VeraCrypt9SyncEvent6SignalEv") + exe = build_test(build, args.sanitize, "encryption_thread_pool_lifecycle_test.cpp", + ("Volume/Pkcs5Kdf.cpp", "Volume/VolumePassword.cpp"), + ["-Wl,--wrap=" + symbol for symbol in symbols]) + for workers in (2, 8): + subprocess.run([str(exe), str(workers)], check=True, timeout=90) + + +if __name__ == "__main__": + main() diff --git a/src/Volume/EncryptionThreadPool.cpp b/src/Volume/EncryptionThreadPool.cpp index 5493f01b..337cef59 100644 --- a/src/Volume/EncryptionThreadPool.cpp +++ b/src/Volume/EncryptionThreadPool.cpp @@ -27,6 +27,23 @@ namespace VeraCrypt { + bool EncryptionThreadPool::AcquirePoolReference () + { + uint32 references = PoolReferences.load(); + while (!(references & PoolClosing)) + { + if (PoolReferences.compare_exchange_weak (references, references + 1)) + return true; + } + return false; + } + + void EncryptionThreadPool::ReleasePoolReference () + { + if (PoolReferences.fetch_sub (1) == PoolClosing + 1) + PoolDrainedEvent.Signal(); + } + EncryptionThreadPool::KeyDerivationWorkItem::KeyDerivationWorkItem (shared_ptr kdf, size_t derivedKeySize) : Completed (false), DerivedKey (derivedKeySize), Kdf (kdf), Processed (false), Result (0) { @@ -38,8 +55,22 @@ namespace VeraCrypt void EncryptionThreadPool::BeginKeyDerivation (KeyDerivationWorkItem &keyDerivationWorkItem, const VolumePassword &password, int pim, const ConstBufferPtr &salt, SyncEvent &completionEvent, SyncEvent &noOutstandingWorkItemEvent, SharedVal &outstandingWorkItemCount, long volatile *abortFlag) { - if (!ThreadPoolRunning) - throw NotInitialized (SRC_POS); + PoolReference reference; + if (!reference.IsAcquired()) + { + keyDerivationWorkItem.Completed.Set (false); + keyDerivationWorkItem.ItemException.reset(); + keyDerivationWorkItem.Processed = false; + keyDerivationWorkItem.Result = abortFlag && *abortFlag ? ERR_USER_ABORT + : keyDerivationWorkItem.Kdf->DeriveKey (keyDerivationWorkItem.DerivedKey, password, pim, salt, abortFlag); + keyDerivationWorkItem.Completed.Set (true); + completionEvent.Signal(); + // No worker was submitted, but the caller still waits on the drain event. + ScopeLock outstandingWorkItemLock (KeyDerivationCompletionMutex); + if (outstandingWorkItemCount.Get() == 0) + noOutstandingWorkItemEvent.Signal(); + return; + } ScopeLock lock (EnqueueMutex); @@ -75,6 +106,8 @@ namespace VeraCrypt noOutstandingWorkItemEvent.Reset(); } + // The submitter and worker may finish in either order. + PoolReferences.fetch_add (1); workItem->State.Set (WorkItem::State::Ready); WorkItemReadyEvent.Signal(); } @@ -94,7 +127,8 @@ namespace VeraCrypt if (unitCount == 0) return; - if (!ThreadPoolRunning || unitCount == 1) + PoolReference reference (unitCount > 1); + if (!reference.IsAcquired()) { switch (type) { @@ -230,6 +264,9 @@ namespace VeraCrypt cpuCount = MaxThreadCount; StopPending = false; + PoolDrainedEvent.Reset(); + WorkItemReadyEvent.Reset(); + WorkItemCompletedEvent.Reset(); DequeuePosition = 0; EnqueuePosition = 0; @@ -251,21 +288,32 @@ namespace VeraCrypt }; make_shared_auto (Thread, thread); - thread->Start (new ThreadFunctor ()); + unique_ptr functor (new ThreadFunctor ()); + // Allocate the list entry before starting a thread that cleanup must join. RunningThreads.push_back (thread); + try + { + thread->Start (functor.get()); + functor.release(); + } + catch (...) + { + RunningThreads.pop_back(); + throw; + } } } catch (...) { try { - ThreadPoolRunning = true; - Stop(); + StopWorkers(); } catch (...) { } throw; } + PoolReferences.store (0); ThreadPoolRunning = true; } @@ -274,6 +322,14 @@ namespace VeraCrypt if (!ThreadPoolRunning) return; + ThreadPoolRunning = false; + if (PoolReferences.fetch_or (PoolClosing) != 0) + PoolDrainedEvent.Wait(); + StopWorkers(); + } + + void EncryptionThreadPool::StopWorkers () + { StopPending = true; WorkItemReadyEvent.Signal(); @@ -281,6 +337,7 @@ namespace VeraCrypt { thread.Join(); } + RunningThreads.clear(); ThreadCount = 0; ThreadPoolRunning = false; @@ -307,12 +364,11 @@ namespace VeraCrypt WorkItemReadyEvent.Wait(); } + if (StopPending) + break; workItem->State.Set (WorkItem::State::Busy); } - if (StopPending) - break; - try { switch (workItem->Type) @@ -384,6 +440,7 @@ namespace VeraCrypt } workItem->State.Set (WorkItem::State::Free); WorkItemCompletedEvent.Signal(); + ReleasePoolReference(); continue; } @@ -409,8 +466,10 @@ namespace VeraCrypt } } - volatile bool EncryptionThreadPool::ThreadPoolRunning = false; - volatile bool EncryptionThreadPool::StopPending = false; + std::atomic EncryptionThreadPool::ThreadPoolRunning (false); + std::atomic EncryptionThreadPool::StopPending (false); + std::atomic EncryptionThreadPool::PoolReferences (PoolClosing); + SyncEvent EncryptionThreadPool::PoolDrainedEvent; size_t EncryptionThreadPool::ThreadCount; diff --git a/src/Volume/EncryptionThreadPool.h b/src/Volume/EncryptionThreadPool.h index ddf6c6e4..87780387 100644 --- a/src/Volume/EncryptionThreadPool.h +++ b/src/Volume/EncryptionThreadPool.h @@ -13,6 +13,7 @@ #ifndef TC_HEADER_Volume_EncryptionThreadPool #define TC_HEADER_Volume_EncryptionThreadPool +#include #include "Platform/Platform.h" #include "EncryptionMode.h" @@ -95,17 +96,38 @@ namespace VeraCrypt }; // Caller-owned references and pointers must remain valid until noOutstandingWorkItemEvent is signaled. + // Derivation runs synchronously when the pool is unavailable or closing. static void BeginKeyDerivation (KeyDerivationWorkItem &keyDerivationWorkItem, const VolumePassword &password, int pim, const ConstBufferPtr &salt, SyncEvent &completionEvent, SyncEvent &noOutstandingWorkItemEvent, SharedVal &outstandingWorkItemCount, long volatile *abortFlag); static void DoWork (WorkType::Enum type, const EncryptionMode *mode, uint8 *data, uint64 startUnitNo, uint64 unitCount, size_t sectorSize); static bool IsRunning () { return ThreadPoolRunning; } + // Start and Stop must be serialized by the caller. Stop drains accepted pool users; + // callers must still keep their objects alive for work running synchronously. static void Start (); static void Stop (); protected: + static bool AcquirePoolReference (); + static void ReleasePoolReference (); + static void StopWorkers (); static void WorkThreadProc (); + class PoolReference + { + public: + explicit PoolReference (bool acquire = true) : Acquired (acquire && AcquirePoolReference()) { } + ~PoolReference () { if (Acquired) ReleasePoolReference(); } + bool IsAcquired () const { return Acquired; } + private: + bool Acquired; + PoolReference (const PoolReference &); + PoolReference &operator= (const PoolReference &); + }; + static const size_t MaxThreadCount = 32; static const size_t QueueSize = MaxThreadCount * 2; + static const uint32 PoolClosing = 0x80000000U; + static std::atomic PoolReferences; + static SyncEvent PoolDrainedEvent; static Mutex DequeueMutex; static volatile size_t DequeuePosition; @@ -116,9 +138,9 @@ namespace VeraCrypt // Orders KDF outstanding-count transitions against no-outstanding event updates. static Mutex KeyDerivationCompletionMutex; static list < shared_ptr > RunningThreads; - static volatile bool StopPending; + static std::atomic StopPending; static size_t ThreadCount; - static volatile bool ThreadPoolRunning; + static std::atomic ThreadPoolRunning; static SyncEvent WorkItemCompletedEvent; static WorkItem WorkItemQueue[QueueSize]; static SyncEvent WorkItemReadyEvent;