How to lock the queue_node from enqueuing and keep dequeuing until we unlock the node?

[wujiazheng2020]

there is a situation that the predecessor node keep "try_put()" to my queue_node at 1000HZ, and the successor of my queue_node keep "try_get()" at 10HZ. However, it is hard to find a API that prevent the predecessor from "try_put()", so when the successor node call the "try_get()", the number of products we give varies from 20 to 1000! So, does anyone know if there are lock-related API desigened for queue_node? for instance: enqueue_lock(), enqueue_unlock() and so on. i'm very eager to solve this problem. Thanks!

[wujiazheng2020]

or how to get the size of queue_node?

[vlserov]

If I understand correctly, you would like to limit number of messages what can walk through your queue_node. Try to use limiter_node before your queue_node (userguide), predecessor node communicates with limiter_node and successor node should allows limiter_node to accept new messages by directly calling limiter_node.decrementer().try_put() or it can be made automatically. Example below shows how to make graph with automatic . After execution function_node sends message to limiter_node for decrement passed message and accept one more.

tbb::flow::graph g;

tbb::flow::limiter_node<size_t> limiter(g, limit_threshold);
tbb::flow::queue_node<size_t> queue(g);
tbb::flow::function_node<size_t> func(g, oneapi::tbb::flow::unlimited, [&](const size_t value)
    { printf("%zd\n", value); }
);

tbb::flow::make_edge(limiter, queue);
tbb::flow::make_edge(queue, func);
tbb::flow::make_edge(func, limiter.decrementer());

g.wait_for_all();

You didn't specify which node are you using as successor of queue_node, but if it's one of functional nodes (function_node, multifunction_node, async_node and continue_node), try to use functional node with Rejecting Policy.

[isaevil]

@wujiazheng2020 is this issue still relevant for you? Could you please respond?

[isaevil]

Closing it since no response. Feel free to reopen if any questions are left.