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;