目录

Function Flow Runtime Concurrent Queue (C)

Overview

The FFRT concurrent queue provides the capability of setting the priority and queue concurrency. Tasks in the queue can be executed on multiple threads at the same time, achieving better effects.

Example: Bank Service System

For example, each customer (common customer or VIP customer) submits a service request to the bank service system. The service request of the VIP customer can be processed first. The bank system has two windows for handling service requests submitted by customers.

You can use the FFRT paradigm to perform the following modeling:

The implementation code is as follows:

#include <cstdio>
#include <unistd.h>
#include "hilog/log.h"
#include "ffrt/ffrt.h" // From the HarmonyOS third-party library "@ppd/ffrt"

#undef LOG_TAG
#define LOG_TAG "ConcurrentTag"
const int SLEEP_TIME = 100 * 1000; // 100ms
const int BANK_CONCURRENCY = 2;

ffrt_queue_t CreateBankSystem(const char *name, int concurrency)
{
    ffrt_queue_attr_t queue_attr;
    (void)ffrt_queue_attr_init(&queue_attr);
    ffrt_queue_attr_set_max_concurrency(&queue_attr, concurrency);

    // Create a concurrent queue.
    ffrt_queue_t queue = ffrt_queue_create(ffrt_queue_concurrent, name, &queue_attr);

    // Destroy the queue attributes after the queue is created.
    ffrt_queue_attr_destroy(&queue_attr);
    if (!queue) {
        OH_LOG_ERROR(LOG_APP, "create queue failed");
        return NULL;
    }

    OH_LOG_INFO(LOG_APP, "create bank system successfully");
    return queue;
}

void DestroyBankSystem(ffrt_queue_t queue_handle)
{
    ffrt_queue_destroy(queue_handle);
    OH_LOG_INFO(LOG_APP, "destroy bank system successfully");
}

void BankBusiness(void *arg)
{
    usleep(SLEEP_TIME);
    const char *data = (const char *)arg;
    OH_LOG_INFO(LOG_APP, "saving or withdraw for %{public}s", data);
}

// Encapsulate the queue task submission function.
ffrt_task_handle_t CommitRequest(ffrt_queue_t bank, void (*func)(void *), const char *name,
    ffrt_queue_priority_t level, int delay)
{
    ffrt_task_attr_t task_attr;
    (void)ffrt_task_attr_init(&task_attr);
    ffrt_task_attr_set_name(&task_attr, name);
    ffrt_task_attr_set_queue_priority(&task_attr, level);
    ffrt_task_attr_set_delay(&task_attr, delay);

    ffrt_task_handle_t handle = ffrt_queue_submit_h_f(bank, func, (void*)name, &task_attr);
    ffrt_task_attr_destroy(&task_attr);
    return handle;
}

// Encapsulate the queue task cancellation function.
int CancelRequest(ffrt_task_handle_t request)
{
    return ffrt_queue_cancel(request);
}

// Encapsulate the queue task waiting function.
void WaitForRequest(ffrt_task_handle_t task)
{
    ffrt_queue_wait(task);
}

int ConcurrentQueueCExec()
{
    ffrt_queue_t bank = CreateBankSystem("Bank", BANK_CONCURRENCY);
    if (!bank) {
        printf("create bank system failed\n");
        OH_LOG_ERROR(LOG_APP, "create bank system failed");
        return -1;
    }

    ffrt_task_handle_t task1 = CommitRequest(bank, BankBusiness, "customer1", ffrt_queue_priority_low, 0);
    ffrt_task_handle_t task2 = CommitRequest(bank, BankBusiness, "customer2", ffrt_queue_priority_low, 0);
    // VIPs enjoy higher-priority service.
    ffrt_task_handle_t task3 = CommitRequest(bank, BankBusiness, "customer3 VIP", ffrt_queue_priority_high, 0);
    ffrt_task_handle_t task4 = CommitRequest(bank, BankBusiness, "customer4", ffrt_queue_priority_low, 0);
    ffrt_task_handle_t task5 = CommitRequest(bank, BankBusiness, "customer5", ffrt_queue_priority_low, 0);

    // Cancel the service for customer 4.
    CancelRequest(task4);

    // Wait for all customer services to complete.
    WaitForRequest(task5);
    DestroyBankSystem(bank);

    ffrt_task_handle_destroy(task1);
    ffrt_task_handle_destroy(task2);
    ffrt_task_handle_destroy(task3);
    ffrt_task_handle_destroy(task4);
    ffrt_task_handle_destroy(task5);
    return 0;
}

The ffrt_queue_submit_h_f API can receive a raw function pointer task as a parameter. If there are pre-processing and post-processing operations on the task, refer to the ffrt_alloc_auto_managed_function_storage_base function to construct the task structure.

Available APIs

The main FFRT APIs involved in the preceding sample are as follows. For details, see the methods in Function Flow Runtime C API:

Name Description
ffrt_queue_create Creates a queue.
ffrt_queue_destroy Destroys a queue.
ffrt_task_attr_set_queue_priority Sets the task priority of a queue.
ffrt_queue_attr_set_max_concurrency Sets the concurrency of a concurrent queue.
ffrt_queue_submit_h_f

Submits a task to a queue.

Note: This API is supported since API version 20.
  • For details about how to use FFRT C++ APIs, see Using FFRT C++ APIs.
  • When using FFRT C or C++ APIs, you can use the FFRT C++ API third-party library to simplify header file inclusion, that is, use the #include "ffrt/ffrt.h" header file inclusion statement.

Constraints

  1. ffrt_queue_attr_t must be initialized by calling ffrt_queue_attr_init before the attribute is set or obtained. If ffrt_queue_attr_t is no longer used, ffrt_queue_attr_destroy needs to be explicitly called to release resources.

  2. ffrt_queue_t must explicitly call ffrt_queue_destroy to release resources before the process exits.

  3. It is recommended that the maximum concurrency of a concurrent queue be within a proper range. If the value is too large, it is meaningless to exceed the number of Worker threads. If the value is too small, the system resource utilization may be low.