方法: 遅延後に完了するタスクを作成する

  • [アーティクル]

この例では、concurrency::task、concurrency::cancellation_token_source、concurrency::cancellation_token、concurrency::task_completion_event、concurrency::timer、concurrency::call の各クラスを使用して、遅延後に完了するタスクを作成する方法を示します。 このメソッドを使用して、データをときどきポーリングするループを作成できます。 また、タイムアウトを導入したり、ユーザー入力の処理をあらかじめ指定された時間遅らせたりすることもできます。

例: complete_after 関数と cancel_after_timeout 関数

complete_after 関数および cancel_after_timeout 関数の例を次に示します。 complete_after 関数では、指定された遅延後に完了する task オブジェクトを作成します。 timer オブジェクトと call オブジェクトを使用して、指定された遅延後に task_completion_event オブジェクトが設定されます。 task_completion_event クラスを使用して、スレッドまたは別のタスクから、値が使用可能であることが通知された後に完了するタスクを定義できます。 イベントが設定されると、リスナー タスクが完了し、継続タスクの実行がスケジュールされます。

ヒント

非同期エージェント ライブラリに含まれる timer クラスと call クラスの詳細については、「非同期メッセージ ブロック」を参照してください。

cancel_after_timeout 関数は complete_after 関数に基づいており、指定されたタイムアウトの前にタスクが完了しない場合にそのタスクを取り消します。 cancel_after_timeout 関数では、2 つのタスクを作成します。 1 つ目のタスクは、指定されたタスクの完了後に成功を示して完了します。 2 つ目のタスクは、指定されたタイムアウト後に失敗を示して完了します。 cancel_after_timeout 関数では、成功または失敗タスクが完了したときに実行される継続タスクを作成します。 失敗タスクが最初に完了した場合、継続タスクによってトークン ソースがキャンセルされ、タスク全体が取り消されます。

// Creates a task that completes after the specified delay.
task<void> complete_after(unsigned int timeout)
{
    // A task completion event that is set when a timer fires.
    task_completion_event<void> tce;

    // Create a non-repeating timer.
    auto fire_once = new timer<int>(timeout, 0, nullptr, false);
    // Create a call object that sets the completion event after the timer fires.
    auto callback = new call<int>([tce](int)
    {
        tce.set();
    });

    // Connect the timer to the callback and start the timer.
    fire_once->link_target(callback);
    fire_once->start();

    // Create a task that completes after the completion event is set.
    task<void> event_set(tce);

    // Create a continuation task that cleans up resources and
    // and return that continuation task.
    return event_set.then([callback, fire_once]()
    {
        delete callback;
        delete fire_once;
    });
}

// Cancels the provided task after the specifed delay, if the task
// did not complete.
template<typename T>
task<T> cancel_after_timeout(task<T> t, cancellation_token_source cts, unsigned int timeout)
{
    // Create a task that returns true after the specified task completes.
    task<bool> success_task = t.then([](T)
    {
        return true;
    });
    // Create a task that returns false after the specified timeout.
    task<bool> failure_task = complete_after(timeout).then([]
    {
        return false;
    });

    // Create a continuation task that cancels the overall task 
    // if the timeout task finishes first.
    return (failure_task || success_task).then([t, cts](bool success)
    {
        if(!success)
        {
            // Set the cancellation token. The task that is passed as the
            // t parameter should respond to the cancellation and stop
            // as soon as it can.
            cts.cancel();
        }

        // Return the original task.
        return t;
    });
}

例: 素数の数を計算する

次の例では、[0, 100000] の範囲の素数の数を複数回計算します。 この操作は、指定された制限時間内に完了しない場合、失敗となります。 count_primes 関数は、cancel_after_timeout 関数の使用方法を示しています。 指定された範囲の素数の数をカウントし、指定された時間内にタスクが完了しない場合は失敗します。 wmain 関数は、count_primes 関数を複数回呼び出します。 毎回、制限時間が半分になります。 現在の制限時間内に操作が完了しないと、プログラムは終了します。

// Determines whether the input value is prime.
bool is_prime(int n)
{
    if (n < 2)
        return false;
    for (int i = 2; i < n; ++i)
    {
        if ((n % i) == 0)
            return false;
    }
    return true;
}

// Counts the number of primes in the range [0, max_value].
// The operation fails if it exceeds the specified timeout.
bool count_primes(unsigned int max_value, unsigned int timeout)
{
    cancellation_token_source cts;

    // Create a task that computes the count of prime numbers.
    // The task is canceled after the specified timeout.
    auto t = cancel_after_timeout(task<size_t>([max_value, timeout, cts]
    {
        combinable<size_t> counts;
        parallel_for<unsigned int>(0, max_value + 1, [&counts, cts](unsigned int n) 
        {
            // Respond if the overall task is cancelled by canceling 
            // the current task.
            if (cts.get_token().is_canceled())
            {
                cancel_current_task();
            }
            // NOTE: You can replace the calls to is_canceled
            // and cancel_current_task with a call to interruption_point.
            // interruption_point();

            // Increment the local counter if the value is prime.
            if (is_prime(n))
            {
                counts.local()++;
            }
        });
        // Return the sum of counts across all threads.
        return counts.combine(plus<size_t>());
    }, cts.get_token()), cts, timeout);

    // Print the result.
    try
    {
        auto primes = t.get();
        wcout << L"Found " << primes << L" prime numbers within " 
              << timeout << L" ms." << endl;
        return true;
    }
    catch (const task_canceled&)
    {
        wcout << L"The task timed out." << endl;
        return false;
    }
}

int wmain()
{
    // Compute the count of prime numbers in the range [0, 100000] 
    // until the operation fails.
    // Each time the test succeeds, the time limit is halved.

    unsigned int max = 100000;
    unsigned int timeout = 5000;
    
    bool success = true;
    do
    {
        success = count_primes(max, timeout);
        timeout /= 2;
    } while (success);
}
/* Sample output:
    Found 9592 prime numbers within 5000 ms.
    Found 9592 prime numbers within 2500 ms.
    Found 9592 prime numbers within 1250 ms.
    Found 9592 prime numbers within 625 ms.
    The task timed out.
*/

この手法を使用して遅延後にタスクを取り消した場合、タスク全体が取り消された後、まだ開始していないタスクは開始されません。 ただし、実行時間の長いタスクは取り消しに迅速に対応することが重要です。 タスクの取り消しの詳細については、「PPL における取り消し処理」を参照してください。

完全なコード例

この例の完全なコードを次に示します。

// task-delay.cpp
// compile with: /EHsc
#include <ppl.h>
#include <ppltasks.h>
#include <agents.h>
#include <iostream>

using namespace concurrency;
using namespace std;

// Creates a task that completes after the specified delay.
task<void> complete_after(unsigned int timeout)
{
    // A task completion event that is set when a timer fires.
    task_completion_event<void> tce;

    // Create a non-repeating timer.
    auto fire_once = new timer<int>(timeout, 0, nullptr, false);
    // Create a call object that sets the completion event after the timer fires.
    auto callback = new call<int>([tce](int)
    {
        tce.set();
    });

    // Connect the timer to the callback and start the timer.
    fire_once->link_target(callback);
    fire_once->start();

    // Create a task that completes after the completion event is set.
    task<void> event_set(tce);

    // Create a continuation task that cleans up resources and
    // and return that continuation task.
    return event_set.then([callback, fire_once]()
    {
        delete callback;
        delete fire_once;
    });
}

// Cancels the provided task after the specifed delay, if the task
// did not complete.
template<typename T>
task<T> cancel_after_timeout(task<T> t, cancellation_token_source cts, unsigned int timeout)
{
    // Create a task that returns true after the specified task completes.
    task<bool> success_task = t.then([](T)
    {
        return true;
    });
    // Create a task that returns false after the specified timeout.
    task<bool> failure_task = complete_after(timeout).then([]
    {
        return false;
    });

    // Create a continuation task that cancels the overall task 
    // if the timeout task finishes first.
    return (failure_task || success_task).then([t, cts](bool success)
    {
        if(!success)
        {
            // Set the cancellation token. The task that is passed as the
            // t parameter should respond to the cancellation and stop
            // as soon as it can.
            cts.cancel();
        }

        // Return the original task.
        return t;
    });
}

// Determines whether the input value is prime.
bool is_prime(int n)
{
    if (n < 2)
        return false;
    for (int i = 2; i < n; ++i)
    {
        if ((n % i) == 0)
            return false;
    }
    return true;
}

// Counts the number of primes in the range [0, max_value].
// The operation fails if it exceeds the specified timeout.
bool count_primes(unsigned int max_value, unsigned int timeout)
{
    cancellation_token_source cts;

    // Create a task that computes the count of prime numbers.
    // The task is canceled after the specified timeout.
    auto t = cancel_after_timeout(task<size_t>([max_value, timeout, cts]
    {
        combinable<size_t> counts;
        parallel_for<unsigned int>(0, max_value + 1, [&counts, cts](unsigned int n) 
        {
            // Respond if the overall task is cancelled by canceling 
            // the current task.
            if (cts.get_token().is_canceled())
            {
                cancel_current_task();
            }
            // NOTE: You can replace the calls to is_canceled
            // and cancel_current_task with a call to interruption_point.
            // interruption_point();

            // Increment the local counter if the value is prime.
            if (is_prime(n))
            {
                counts.local()++;
            }
        });
        // Return the sum of counts across all threads.
        return counts.combine(plus<size_t>());
    }, cts.get_token()), cts, timeout);

    // Print the result.
    try
    {
        auto primes = t.get();
        wcout << L"Found " << primes << L" prime numbers within " 
              << timeout << L" ms." << endl;
        return true;
    }
    catch (const task_canceled&)
    {
        wcout << L"The task timed out." << endl;
        return false;
    }
}

int wmain()
{
    // Compute the count of prime numbers in the range [0, 100000] 
    // until the operation fails.
    // Each time the test succeeds, the time limit is halved.

    unsigned int max = 100000;
    unsigned int timeout = 5000;
    
    bool success = true;
    do
    {
        success = count_primes(max, timeout);
        timeout /= 2;
    } while (success);
}
/* Sample output:
    Found 9592 prime numbers within 5000 ms.
    Found 9592 prime numbers within 2500 ms.
    Found 9592 prime numbers within 1250 ms.
    Found 9592 prime numbers within 625 ms.
    The task timed out.
*/

コードのコンパイル

このコードをコンパイルするには、コードをコピーし、Visual Studio プロジェクトに貼り付けるか、task-delay.cpp という名前のファイルに貼り付けてから、Visual Studio のコマンド プロンプト ウィンドウで次のコマンドを実行します。

cl.exe /EHsc task-delay.cpp

関連項目

タスクの並列処理
task クラス (コンカレンシー ランタイム)
cancellation_token_source クラス
cancellation_token クラス
task_completion_event クラス
timer クラス
call クラス
非同期メッセージ ブロック
PPL における取り消し処理