当前位置：首页 > article >正文

windows C++ 并行编程-使用消息块筛选器

article 2024/11/15 11:17:08

本文档演示了如何使用筛选器函数，使异步消息块能够根据消息的有效负载接受或拒绝消息。

创建消息块对象(例如 concurrency::unbounded_buffer、concurrency::call 或 concurrency::transformer)时，可以提供筛选器函数，用于确定消息块是接受还是拒绝消息。筛选器函数是保证消息块仅接收特定值的有效方式。

筛选器函数很重要，因为它们使你能够连接消息块以形成数据流网络。在数据流网络中，消息块通过仅处理满足特定标准的消息来控制数据流。将数据流网络与控制流模型进行比较，在后者中，数据流是通过使用条件语句、循环等控制结构来调节的。

本文档提供了有关如何使用消息筛选器的基本示例。

示例：count_primes 函数

考虑以下函数 count_primes，该函数说明了不筛选传入消息的消息块的基本用法。此消息块将质数追加到 std::vector 对象。 count_primes 函数将几个数字发送到消息块，从消息块接收输出值，并将这些数字打印到控制台。

// Illustrates usage of a message buffer that does not use filtering.
void count_primes(unsigned long random_seed)
{
    // Holds prime numbers.
    vector<unsigned long> primes;

    // Adds numbers that are prime to the vector object.
    transformer<unsigned long, unsigned long> t([&primes](unsigned long n) -> unsigned long
    {
        if (is_prime(n))
        {
            primes.push_back(n);
        }
        return n;
    });

    // Send random values to the message buffer.
    mt19937 generator(random_seed);
    for (int i = 0; i < 20; ++i)
    {
        send(t, static_cast<unsigned long>(generator()%10000));
    }

    // Receive from the message buffer the same number of times
    // to ensure that the message buffer has processed each message.
    for (int i = 0; i < 20; ++i)
    {
        receive(t);
    }

    // Print the prime numbers to the console.
    wcout << L"The following numbers are prime: " << endl;
    for(unsigned long prime : primes)
    {
        wcout << prime << endl;
    }
}

transformer 对象处理所有输入值；但是，它只需要那些为质数的值。尽管可以编写应用程序以使消息发送方仅发送质数，但并不能始终得知消息接收方的要求。

示例：count_primes_filter 函数

以下函数 count_primes_filter 执行与 count_primes 函数相同的任务。但是，此版本中的 transformer 对象使用筛选器函数以便仅接受那些为质数的值。执行该操作的函数仅接收质数；因此，它不必调用 is_prime 函数。

由于 transformer 对象仅接收质数，所以 transformer 对象本身可以保存质数。换言之，此示例中的 transformer 对象不需要将质数添加到 vector 对象。

// Illustrates usage of a message buffer that uses filtering.
void count_primes_filter(unsigned long random_seed)
{
    // Accepts numbers that are prime.
    transformer<unsigned long, unsigned long> t([](unsigned long n) -> unsigned long
    {
        // The filter function guarantees that the input value is prime.
        // Return the input value.
        return n;
    },
    nullptr,
    [](unsigned long n) -> bool
    {
        // Filter only values that are prime.
        return is_prime(n);
    });

    // Send random values to the message buffer.
    mt19937 generator(random_seed);
    size_t prime_count = 0;
    for (int i = 0; i < 20; ++i)
    {
        if (send(t, static_cast<unsigned long>(generator()%10000)))
        {
            ++prime_count;
        }
    }

    // Print the prime numbers to the console. 
    wcout << L"The following numbers are prime: " << endl;
    while (prime_count-- > 0)
    {
        wcout << receive(t) << endl;
    }
}

transformer 对象现在仅处理那些为质数的值。在前面的示例中，transformer 对象处理所有消息。因此，前面的示例必须接收与其发送的相同数量的消息。此示例使用 concurrency::send 函数的结果来确定要从 transformer 对象接收的消息数。 send 函数在消息缓冲区接受消息时返回 true，在消息缓冲区拒绝消息时返回 false。因此，消息缓冲区接受消息的次数与质数的计数相匹配。

示例：已完成的消息块筛选器代码示例

以下代码显示完整示例。该示例同时调用 count_primes 函数和 count_primes_filter 函数。

// primes-filter.cpp
// compile with: /EHsc
#include <agents.h>
#include <algorithm>
#include <iostream>
#include <random>

using namespace concurrency;
using namespace std;

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

// Illustrates usage of a message buffer that does not use filtering.
void count_primes(unsigned long random_seed)
{
    // Holds prime numbers.
    vector<unsigned long> primes;

    // Adds numbers that are prime to the vector object.
    transformer<unsigned long, unsigned long> t([&primes](unsigned long n) -> unsigned long
    {
        if (is_prime(n))
        {
            primes.push_back(n);
        }
        return n;
    });

    // Send random values to the message buffer.
    mt19937 generator(random_seed);
    for (int i = 0; i < 20; ++i)
    {
        send(t, static_cast<unsigned long>(generator()%10000));
    }

    // Receive from the message buffer the same number of times
    // to ensure that the message buffer has processed each message.
    for (int i = 0; i < 20; ++i)
    {
        receive(t);
    }

    // Print the prime numbers to the console.
    wcout << L"The following numbers are prime: " << endl;
    for(unsigned long prime : primes)
    {
        wcout << prime << endl;
    }
}

// Illustrates usage of a message buffer that uses filtering.
void count_primes_filter(unsigned long random_seed)
{
    // Accepts numbers that are prime.
    transformer<unsigned long, unsigned long> t([](unsigned long n) -> unsigned long
    {
        // The filter function guarantees that the input value is prime.
        // Return the input value.
        return n;
    },
    nullptr,
    [](unsigned long n) -> bool
    {
        // Filter only values that are prime.
        return is_prime(n);
    });

    // Send random values to the message buffer.
    mt19937 generator(random_seed);
    size_t prime_count = 0;
    for (int i = 0; i < 20; ++i)
    {
        if (send(t, static_cast<unsigned long>(generator()%10000)))
        {
            ++prime_count;
        }
    }

    // Print the prime numbers to the console. 
    wcout << L"The following numbers are prime: " << endl;
    while (prime_count-- > 0)
    {
        wcout << receive(t) << endl;
    }
}

int wmain()
{
    const unsigned long random_seed = 99714;

    wcout << L"Without filtering:" << endl;
    count_primes(random_seed);

    wcout << L"With filtering:" << endl;
    count_primes_filter(random_seed);

    /* Output:
    9973
    9349
    9241
    8893
    1297
    7127
    8647
    3229
    With filtering:
    The following numbers are prime:
    9973
    9349
    9241
    8893
    1297
    7127
    8647
    3229
    */
}