Cancelling tasks

On this page

A task that is no longer needed - a worker when the application closes, a handler whose window is gone, a request whose client has left - should stop soon and clean up after itself. CancelTask asks a task to do that. This guide shows how a task checks for the request, which waits it ends, and how to shut down a group of tasks. The examples are small pieces of guide-cancellation.cpp, a complete program that uses them and checks the results.

Like Your first program, the program lets main convert its thread to a task and start a scheduler, scheduler, before the parts shown here. It uses these modules and names:

#include <ThinkMeta.ConcurrentTasks/Core.h>
import ThinkMeta.ConcurrentTasks.Tasks.Task;
import ThinkMeta.ConcurrentTasks.Tasks.WaitableTask;
import ThinkMeta.ConcurrentTasks.Synchronization.AutoResetEvent;
import ThinkMeta.ConcurrentTasks.Synchronization.ManualResetEvent;
import ThinkMeta.ConcurrentTasks.Synchronization.TaskGroup;
import ThinkMeta.ConcurrentTasks.AsyncIo.AsyncTcpSocket;

namespace TC = ThinkMeta::ConcurrentTasks;
using TC::AsyncIo::AsyncTcpSocket4;
using TC::Windows::WinsockError;
using TC::Windows::WinsockErrorCode;

Cancellation is cooperative

A running task is never stopped from outside: its fiber holds locks, buffers and half-done work that only the task itself can put in order. CancelTask sets a request on the task, and the task ends itself:

while it runs, by asking IsCancellationRequested at points where it can stop;
while it waits, because a cancellable wait returns WaitStatus::Cancelled instead of waiting on;
before it has started, without its help: it exits without running.

A task that never asks and never waits cancellably runs to its end. Any Task can be cancelled; Task::Cancel is the same as CancelTask(this). A task class of your own needs control storage (GetControlData), as for a start by time.

What a request does depends on where the task is:

The task …	Effect of `CancelTask`
runs	Nothing at once; `IsCancellationRequested` returns `true` from now on.
was added, but has not started	It exits without running: only `OnTaskExited` is called.
waits to start by time	The start is taken back; it exits without running.
is not added at all	The request stands; the next run is skipped.
sleeps in SleepTask	The sleep ends; `SleepTask` returns `false`.
waits with `Wait(Duration)` on a single-waiter primitive	The wait ends with `WaitStatus::Cancelled`; nothing was taken.
waits with `Wait()` or on a multi-waiter primitive	It waits on; it sees the request when the wait is over.
waits for asynchronous I/O	The operation is cancelled; the call returns `OperationAborted`.

A later cancellable wait of a task with a standing request returns Cancelled at once, without suspending. The request stays until the task has exited.

Checking while running

A task with long work checks between two steps whether it is to stop:

// A worker with far more work than it gets time for. It asks between two steps
// whether it is to stop, and ends itself.
class Worker : public TC::Tasks::Task
{
public:
    std::uint64_t steps    = 0;
    bool          finished = false;

    void Run() noexcept override
    {
        std::uint64_t value = 0;
        for (std::uint64_t i = 0; i < 1'000'000'000'000; ++i) {
            if (TC::IsCancellationRequested())
                return; // cancelled: leave the rest of the work undone
            value = Step(value);
            ++steps;
        }
        finished = value != 1; // practically never reached
    }
};

IsCancellationRequested asks for the calling task and is cheap enough for every pass of a loop. Check where stopping is safe: not halfway through an update that others will read.

Shutting down a group of tasks

There is no "cancel everything" call: whoever starts tasks knows them and cancels them. A pool that keeps its workers in an array cancels each one, then waits for the group:

// The pool knows its workers: on shutdown it cancels every one of them, then waits
// until all have exited. Only then may the workers go away.
class Pool
{
public:
    static constexpr int Size = 4;
    Worker               workers[Size];

    void Start(TC::ITaskScheduler& scheduler) noexcept
    {
        for (Worker& worker : workers) {
            running_.AddTask(&worker);
            scheduler.AddTask(&worker);
        }
    }

    void Shutdown() noexcept
    {
        for (Worker& worker : workers)
            worker.Cancel(); // the same as TC::CancelTask(&worker)
        running_.Wait();
    }

private:
    TC::Synchronization::TaskGroup_s running_;
};

CancelTask never blocks; the TaskGroup_s wait is what makes the shutdown complete. Workers that had not started yet exit without running and leave the group all the same.

Important

Cancelling does not end a task's lifetime early. The task object must stay valid during CancelTask and, as for any task, until its OnTaskExited has been called. Wait for the tasks, then destroy them.

Ending a wait

The Wait(Duration timeout) of the single-waiter primitives - AutoResetEvent_s, ManualResetEvent_s, Semaphore_s, TaskGroup_s, WaitableTask_s and the waitable I/O requests - returns a WaitStatus: Signaled, TimedOut or Cancelled. Duration::Infinite waits without a time limit, but a cancel still ends the wait:

// Waits for the next request. Wait(Duration::Infinite()) has no time limit, but a
// cancel ends it. Wait() without an argument would wait on: it returns only with
// the event set.
class RequestHandler : public TC::Tasks::WaitableTask_s
{
public:
    TC::Synchronization::AutoResetEvent_s   request;
    TC::Synchronization::ManualResetEvent_s started; // for the test driver
    TC::WaitStatus                          status = TC::WaitStatus::Signaled;

    void Run() noexcept override
    {
        started.Set();
        status = request.Wait(TC::Duration::Infinite());
        if (status == TC::WaitStatus::Cancelled)
            return; // no request came; nothing was taken from the event
        // ... handle the request ...
    }
};

int main()
{
    // ...
    RequestHandler handler;
    scheduler->AddTask(&handler);
    handler.started.Wait(); // test driver: the handler runs
    handler.Cancel();       // ends the wait, or the next one at once
    handler.Wait();         // the handler has exited
    // ...
}

A cancelled wait has taken nothing: no signal, no permit. A Set that comes later stays for the next wait. Such a wait needs no EnableTimers: only a finite timeout needs the timer thread.

Wait() without an argument cannot be cancelled. It returns only when the condition holds - the event set, the permit taken, the group empty - and costs nothing extra; the task sees the request afterwards. Use it where the wait must not end without its result, and Wait(Duration::Infinite()) where a cancel should end it. The multi-waiter variants (AutoResetEvent, ManualResetEvent, Semaphore, TaskGroup, WaitableTask) and the critical sections cannot be cancelled either.

Ending I/O

The operations of AsyncFile and the asynchronous sockets end on a cancel, also with their default timeout, Duration::Infinite(): the operation is cancelled with CancelIoEx, the call waits until the cancellation has completed, and it returns WinsockError::OperationAborted (for files the Win32 code ERROR_OPERATION_ABORTED). A client that waits for news from its server, without a time limit:

// A client that waits for news from its server. Its Recv has no time limit; a
// cancel ends it with WinsockError::OperationAborted.
class Subscriber : public TC::Tasks::WaitableTask_s
{
public:
    sockaddr_in                             server{};
    TC::Synchronization::ManualResetEvent_s connected; // for the test driver
    WinsockErrorCode                        error = 0;

    void Run() noexcept override
    {
        AsyncTcpSocket4 socket;
        sockaddr_in     local{};
        local.sin_family      = AF_INET;
        local.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
        if (WinsockError::Failed(socket.Create())
            || WinsockError::Failed(socket.Bind(local))
            || WinsockError::Failed(socket.Connect(server)))
            return;
        connected.Set();

        char  news[256];
        DWORD received = 0;
        error          = socket.Recv(news, sizeof(news), &received);
        // After an aborted operation a stream socket is fit for closing only.
        socket.Close();
    }
};

On shutdown it is cancelled like any task:

int main()
{
    // ...
    // Shutdown: the subscriber's Recv ends, the subscriber closes its socket.
    subscriber.Cancel();
    subscriber.Wait();
    // ...
}

Important

After an aborted operation a stream socket is fit for closing only: data may have been lost. If the operation completed before the cancellation took effect, the call returns its own result.

Tasks that have not started

A cancelled task that has not started exits without running: its OnTaskExited is called on its scheduler, so a waiter of a WaitableTask_s or a task group is released as usual. This holds for a task that is added and waiting for a thread, for a task waiting to start by time, and for a task cancelled before it was added at all:

// A report that is not wanted any more before it has run.
class Report : public TC::Tasks::WaitableTask_s
{
public:
    bool ran = false;

    void Run() noexcept override { ran = true; }
};

int main()
{
    // ...
    // Cancelled before it was added: the request stands, and the next run is
    // skipped.
    Report early;
    early.Cancel();
    scheduler->AddTask(&early);
    early.Wait(); // passes: the task exits without running

    // Planned for later, then taken back: it exits without running as well.
    Report later;
    if (scheduler->AddTimedTask(&later, TC::TaskPriority::Normal,
                                TC::Duration::Seconds(10))) {
        later.Cancel();
        later.Wait();
    }
    // ...
}

CancelTask returns true if the request is new and false if one is standing already. Called again, it still acts: a start by time added since is taken back as well.

Ending a sleep

A cancel cuts a SleepTask or Task::SleepFor short; it then returns false, as it does when timers are not enabled:

// Polls something every ten seconds; a cancel cuts the sleep short.
class Poller : public TC::Tasks::WaitableTask_s
{
public:
    TC::Synchronization::ManualResetEvent_s started; // for the test driver
    int                                     polls = 0;

    void Run() noexcept override
    {
        started.Set();
        for (;;) {
            ++polls; // ... poll ...
            if (!TC::Tasks::Task::SleepFor(TC::Duration::Seconds(10)))
                return; // cancelled (or timers not enabled)
        }
    }
};

Your own primitives

A wait of your own is cancellable if it suspends with SuspendTaskFor, also with Duration::Infinite(), and its callback can take the task back in OnWithdraw. For a cancel, OnWithdraw runs on the thread that called CancelTask. See Custom primitives.

The program prints:

workers stopped:        yes (in 0 ms)
wait without limit:     Cancelled
Wait() ends on cancel:  no
receive on shutdown:    OperationAborted
reports run:            none
sleep cut short:        yes
OK

Show the whole program

// Guide "Cancelling tasks": workers that check for a cancel request, a cancellable
// wait, a socket receive ended on shutdown, tasks that are taken back before they
// start, and a sleep cut short. The first three cases run without timers. Uses
// 127.0.0.1 and an ephemeral port; checks its own results outside the regions that
// the guide shows.
#include <WinSock2.h>
#include <Windows.h>

#include <atomic>
#include <chrono>
#include <cstdint>
#include <cstdio>

#include <ThinkMeta.ConcurrentTasks/Core.h>
import ThinkMeta.ConcurrentTasks.Tasks.Task;
import ThinkMeta.ConcurrentTasks.Tasks.WaitableTask;
import ThinkMeta.ConcurrentTasks.Synchronization.AutoResetEvent;
import ThinkMeta.ConcurrentTasks.Synchronization.ManualResetEvent;
import ThinkMeta.ConcurrentTasks.Synchronization.TaskGroup;
import ThinkMeta.ConcurrentTasks.AsyncIo.AsyncTcpSocket;

namespace TC = ThinkMeta::ConcurrentTasks;
using TC::AsyncIo::AsyncTcpSocket4;
using TC::Windows::WinsockError;
using TC::Windows::WinsockErrorCode;
using Clock = std::chrono::steady_clock;

static long long Milliseconds(Clock::duration duration) noexcept
{
    return std::chrono::duration_cast<std::chrono::milliseconds>(duration).count();
}

static const char* Name(TC::WaitStatus status) noexcept
{
    switch (status) {
        case TC::WaitStatus::Signaled: return "Signaled";
        case TC::WaitStatus::TimedOut: return "TimedOut";
        case TC::WaitStatus::Cancelled: return "Cancelled";
    }
    return "?";
}

// One step of made-up work.
static std::uint64_t Step(std::uint64_t value) noexcept
{
    for (int i = 0; i < 1000; ++i)
        value = value * 6364136223846793005ULL + 1442695040888963407ULL;
    return value;
}

// A worker with far more work than it gets time for. It asks between two steps
// whether it is to stop, and ends itself.
class Worker : public TC::Tasks::Task
{
public:
    std::uint64_t steps    = 0;
    bool          finished = false;

    void Run() noexcept override
    {
        std::uint64_t value = 0;
        for (std::uint64_t i = 0; i < 1'000'000'000'000; ++i) {
            if (TC::IsCancellationRequested())
                return; // cancelled: leave the rest of the work undone
            value = Step(value);
            ++steps;
        }
        finished = value != 1; // practically never reached
    }
};

// The pool knows its workers: on shutdown it cancels every one of them, then waits
// until all have exited. Only then may the workers go away.
class Pool
{
public:
    static constexpr int Size = 4;
    Worker               workers[Size];

    void Start(TC::ITaskScheduler& scheduler) noexcept
    {
        for (Worker& worker : workers) {
            running_.AddTask(&worker);
            scheduler.AddTask(&worker);
        }
    }

    void Shutdown() noexcept
    {
        for (Worker& worker : workers)
            worker.Cancel(); // the same as TC::CancelTask(&worker)
        running_.Wait();
    }

private:
    TC::Synchronization::TaskGroup_s running_;
};

// Waits for the next request. Wait(Duration::Infinite()) has no time limit, but a
// cancel ends it. Wait() without an argument would wait on: it returns only with
// the event set.
class RequestHandler : public TC::Tasks::WaitableTask_s
{
public:
    TC::Synchronization::AutoResetEvent_s   request;
    TC::Synchronization::ManualResetEvent_s started; // for the test driver
    TC::WaitStatus                          status = TC::WaitStatus::Signaled;

    void Run() noexcept override
    {
        started.Set();
        status = request.Wait(TC::Duration::Infinite());
        if (status == TC::WaitStatus::Cancelled)
            return; // no request came; nothing was taken from the event
        // ... handle the request ...
    }
};

// Test driver: the same wait with Wait(), which a cancel does not end.
class PlainWaiter : public TC::Tasks::WaitableTask_s
{
public:
    TC::Synchronization::AutoResetEvent_s   request;
    TC::Synchronization::ManualResetEvent_s started;
    std::atomic<bool>                       returned{false};
    bool                                    askedToCancel = false;

    void Run() noexcept override
    {
        started.Set();
        request.Wait();
        returned      = true;
        askedToCancel = TC::IsCancellationRequested();
    }
};

// A client that waits for news from its server. Its Recv has no time limit; a
// cancel ends it with WinsockError::OperationAborted.
class Subscriber : public TC::Tasks::WaitableTask_s
{
public:
    sockaddr_in                             server{};
    TC::Synchronization::ManualResetEvent_s connected; // for the test driver
    WinsockErrorCode                        error = 0;

    void Run() noexcept override
    {
        AsyncTcpSocket4 socket;
        sockaddr_in     local{};
        local.sin_family      = AF_INET;
        local.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
        if (WinsockError::Failed(socket.Create())
            || WinsockError::Failed(socket.Bind(local))
            || WinsockError::Failed(socket.Connect(server)))
            return;
        connected.Set();

        char  news[256];
        DWORD received = 0;
        error          = socket.Recv(news, sizeof(news), &received);
        // After an aborted operation a stream socket is fit for closing only.
        socket.Close();
    }
};

// A report that is not wanted any more before it has run.
class Report : public TC::Tasks::WaitableTask_s
{
public:
    bool ran = false;

    void Run() noexcept override { ran = true; }
};

// Polls something every ten seconds; a cancel cuts the sleep short.
class Poller : public TC::Tasks::WaitableTask_s
{
public:
    TC::Synchronization::ManualResetEvent_s started; // for the test driver
    int                                     polls = 0;

    void Run() noexcept override
    {
        started.Set();
        for (;;) {
            ++polls; // ... poll ...
            if (!TC::Tasks::Task::SleepFor(TC::Duration::Seconds(10)))
                return; // cancelled (or timers not enabled)
        }
    }
};

int main()
{
    TC::ConvertThreadToTask();
    if (WinsockError::Failed(TC::Windows::Socket::Startup())
        || !TC::AsyncIo::EnableAsyncIo())
        return 1;

    TC::SchedulerInfo   info{};
    TC::ITaskScheduler* scheduler
        = TC::CreateTaskScheduler(TC::SchedulerKind::Default, &info);
    if (scheduler == nullptr)
        return 1;
    scheduler->Start();
    bool ok = true;

    {
        // The workers run for a while, then the pool shuts down. None finishes its
        // work.
        static Pool pool;
        pool.Start(*scheduler);
        ::Sleep(50);
        const auto start = Clock::now();
        pool.Shutdown();
        const auto took = Milliseconds(Clock::now() - start);
        bool       none = true;
        for (const Worker& worker : pool.workers)
            none = none && !worker.finished;
        std::printf("workers stopped:        %s (in %lld ms)\n",
                    none ? "yes" : "no", took);
        ok = ok && none && took < 1000;
    }

    {
        RequestHandler handler;
        scheduler->AddTask(&handler);
        handler.started.Wait(); // test driver: the handler runs
        handler.Cancel();       // ends the wait, or the next one at once
        handler.Wait();         // the handler has exited
        std::printf("wait without limit:     %s\n", Name(handler.status));
        ok = ok && handler.status == TC::WaitStatus::Cancelled;
    }

    {
        // Wait() keeps waiting after the cancel; the task sees the request when the
        // event is set.
        PlainWaiter waiter;
        scheduler->AddTask(&waiter);
        waiter.started.Wait();
        waiter.Cancel();
        ::Sleep(50);
        const bool stillWaiting = !waiter.returned;
        waiter.request.Set();
        waiter.Wait();
        std::printf("Wait() ends on cancel:  %s\n", stillWaiting ? "no" : "yes");
        ok = ok && stillWaiting && waiter.askedToCancel;
    }

    {
        AsyncTcpSocket4 listener;
        sockaddr_in     address{};
        address.sin_family      = AF_INET;
        address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
        address.sin_port        = 0; // any free port
        if (WinsockError::Failed(listener.Create())
            || WinsockError::Failed(listener.Bind(address))
            || WinsockError::Failed(listener.GetSockName(address))
            || WinsockError::Failed(listener.Listen()))
            return 1;

        Subscriber subscriber;
        subscriber.server = address;
        scheduler->AddTask(&subscriber);
        AsyncTcpSocket4 peer; // the server side; it never sends
        if (WinsockError::Failed(listener.Accept(peer)))
            return 1;
        subscriber.connected.Wait();
        // Shutdown: the subscriber's Recv ends, the subscriber closes its socket.
        subscriber.Cancel();
        subscriber.Wait();
        peer.Close();
        listener.Close();
        const bool aborted = subscriber.error == WinsockError::OperationAborted;
        std::printf("receive on shutdown:    %s\n",
                    aborted ? "OperationAborted" : "other result");
        ok = ok && aborted;
    }

    // From here on with timers: a start by time and a sleep.
    if (!TC::EnableTimers())
        return 1;

    {
        // Cancelled before it was added: the request stands, and the next run is
        // skipped.
        Report early;
        early.Cancel();
        scheduler->AddTask(&early);
        early.Wait(); // passes: the task exits without running

        // Planned for later, then taken back: it exits without running as well.
        Report later;
        if (scheduler->AddTimedTask(&later, TC::TaskPriority::Normal,
                                    TC::Duration::Seconds(10))) {
            later.Cancel();
            later.Wait();
        }
        std::printf("reports run:            %s\n",
                    early.ran || later.ran ? "yes" : "none");
        ok = ok && !early.ran && !later.ran;
    }

    {
        Poller poller;
        scheduler->AddTask(&poller);
        poller.started.Wait();
        const auto start = Clock::now();
        poller.Cancel();
        poller.Wait();
        const auto took = Milliseconds(Clock::now() - start);
        std::printf("sleep cut short:        %s\n",
                    took < 1000 && poller.polls == 1 ? "yes" : "no");
        ok = ok && took < 1000 && poller.polls == 1;
    }

    scheduler->Stop();
    scheduler->Destroy();
    TC::DisableTimers();
    TC::AsyncIo::DisableAsyncIo(); // all sockets are closed, nothing is pending
    TC::Windows::Socket::Cleanup();
    TC::ConvertTaskToThread();
    std::printf(ok ? "OK\n" : "FAILED\n");
    return ok ? 0 : 1;
}

Source and project on GitHub