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.

Tests/encryption_thread_pool_lifecycle_test.cpp

@@ -0,0 +1,322 @@
+/* Copyright (c) 2026 AM Crypto. Licensed under the Apache License 2.0. */
+#include <algorithm>
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <cstdlib>
+#include <cstring>
+#include <future>
+#include <iostream>
+#include <mutex>
+#include <thread>
+#include <unistd.h>
+#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 <class Predicate> 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<std::mutex> 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<std::mutex> lock (Mutex);
+        Check (Changed.wait_for (lock, std::chrono::seconds (10), [=] { return Arrivals >= count; }), "arrival timed out");
+    }
+    void Release () { std::lock_guard<std::mutex> 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<SyncEvent *> 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<EncryptionMode> GetNew () const override { return shared_ptr<EncryptionMode>(); }
+    void SetKey (const ConstBufferPtr &) override { }
+    mutable std::atomic<int> 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<VeraCrypt::Hash> GetHash () const override { return shared_ptr<VeraCrypt::Hash>(); }
+    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<size_t> Outstanding;
+    std::vector<shared_ptr<EncryptionThreadPool::KeyDerivationWorkItem>> Items;
+    KdfRequest () : Outstanding (0) { }
+    void Submit (Gate *gate = nullptr)
+    {
+        shared_ptr<Pkcs5Kdf> kdf (new TestKdf (gate));
+        shared_ptr<EncryptionThreadPool::KeyDerivationWorkItem> 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<void> &task)
+{
+    Check (task.wait_for (std::chrono::seconds (10)) == std::future_status::ready, "task hung");
+    task.get();
+}
+static void StillWaiting (std::future<void> &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<uint8> 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<size_t> (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;
+}

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)

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()

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 <Pkcs5Kdf> 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 <size_t> &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 <ThreadFunctor> 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 <bool> EncryptionThreadPool::ThreadPoolRunning (false);
+	std::atomic <bool> EncryptionThreadPool::StopPending (false);
+	std::atomic <uint32> EncryptionThreadPool::PoolReferences (PoolClosing);
+	SyncEvent EncryptionThreadPool::PoolDrainedEvent;
 
 	size_t EncryptionThreadPool::ThreadCount;
 

src/Volume/EncryptionThreadPool.h

@@ -13,6 +13,7 @@
 #ifndef TC_HEADER_Volume_EncryptionThreadPool
 #define TC_HEADER_Volume_EncryptionThreadPool
 
+#include <atomic>
 #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 <size_t> &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 <uint32> 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 <Thread> > RunningThreads;
-		static volatile bool StopPending;
+		static std::atomic <bool> StopPending;
 		static size_t ThreadCount;
-		static volatile bool ThreadPoolRunning;
+		static std::atomic <bool> ThreadPoolRunning;
 		static SyncEvent WorkItemCompletedEvent;
 		static WorkItem WorkItemQueue[QueueSize];
 		static SyncEvent WorkItemReadyEvent;