openharmony/resourceschedule_ffrt
1
0
Fork 0
mirror of https://github.com/openharmony/resourceschedule_ffrt.git synced 2026-08-24 18:26:04 -04:00
Code Issues Packages Projects Releases Wiki Activity

Fix concurrency-pattern correctness and multi-DSO support

Browse Source 
Port fixes from the ffrt_api repo:
- job_ring: fix memory order and wakeup defect — release the drained
  count with memory_order_release and notify waiters after draining;
  drop an unused variable.
- job_utils: strengthen mpmc_queue slot reads to memory_order_acquire;
  make ref_obj copy/move assignment inc-before-dec to be self-assignment
  safe; use plain variadic signatures for FFRT_API_LOG/TRACE placeholder
  macros.
- fiber: save/restore thread-local `cur` across start()/suspend() so
  nested or cross-thread fiber resume keeps the outer fiber's identity.
- job_partner/fiber: add multi-DSO extern convergence for singleton
  accessors (visibility("default"), optional extern definitions), and
  add submit_to_master AllowRunOnCallerStack inline-execution switch.
- ffrt.h: include job_partner.h and job_ring.h.

Signed-off-by: chuchihtung <zhuzhidong2@huawei.com>

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
cct
2026-07-17 18:08:24 +08:00
parent ec491b59ba
commit dabb2898b2
4 changed files with 58 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
interfaces/kits/cpp/pattern/job_partner.h
+18 -4
View File
@@ -252,11 +252,15 @@ struct job_partner : ref_obj<job_partner<UsageId>>, detail::non_copyable {
* @return Reference to the main thread's job_partner smart pointer.
* @since 20
*/
static __attribute__((noinline)) auto& get_main_partner()
#ifdef FFRT_MULTI_DSO_EXTERN_MAIN_PARTNER
static __attribute__((visibility("default"), noinline)) auto& get_main_partner();
#else
static __attribute__((visibility("default"), noinline)) auto& get_main_partner()
{
static auto s = ref_obj<job_partner<UsageId>>::make(getpid());
return s;
}
#endif
/**
* @brief Gets the main thread's job_partner instance with updated attributes.
@@ -278,11 +282,15 @@ struct job_partner : ref_obj<job_partner<UsageId>>, detail::non_copyable {
* @return Reference to the current thread's job_partner smart pointer.
* @since 20
*/
static __attribute__((noinline)) auto& get_partner_of_this_thread()
#ifdef FFRT_MULTI_DSO_EXTERN_THREAD_PARTNER
static __attribute__((visibility("default"), noinline)) auto& get_partner_of_this_thread();
#else
static __attribute__((visibility("default"), noinline)) auto& get_partner_of_this_thread()
{
static thread_local auto s = ref_obj<job_partner<UsageId>>::make(tid());
return s;
}
#endif
/**
* @brief Gets the current thread's job_partner instance with updated attributes.
@@ -338,14 +346,20 @@ struct job_partner : ref_obj<job_partner<UsageId>>, detail::non_copyable {
* - If called outside a job context (no current task), the job executes directly.
* - If the master thread queue is full, it retries until successful submission.
*
* @param job The task function to be executed by the master thread.
* @tparam AllowRunOnCallerStack If true, @p job may be executed inline on the caller's current
* stack when the caller is the master fiber (skipping suspend/queue/resume); enable only
* if @p job is safe on a fiber (coroutine) stack, which the caller's stack may be. If
* false (default), @p job never runs on a fiber stack: it runs inline only when there is
* no current task (native stack), else on the master thread's native stack.
* @param job Indicates the task function to be executed by the master thread.
* @since 20
*/
template <bool AllowRunOnCallerStack = false>
static inline void submit_to_master(const std::function<void()>& job)
{
auto& e = job_t::env();
auto j = e.cur;
if (j == nullptr || j->local().partner->token == e.tl.thread_id) {
if (j == nullptr || (AllowRunOnCallerStack && j->local().partner->token == e.tl.thread_id)) {
return job();
}
j->local().partner->submit_to_master(e, j, job);
interfaces/kits/cpp/pattern/job_ring.h
+4 -2
View File
@@ -224,7 +224,6 @@ struct job_ring : ref_obj<job_ring<MultiProducer>> {
{
FFRT_API_LOGD("ring_try_submit %" X_PUBLIC "p", this);
FFRT_API_TRACE_SCOPE("ring_try_submit");
uint64_t us = 1;
if (!q.try_push(std::forward<std::function<void()>>(job))) {
return false;
}
@@ -307,12 +306,15 @@ struct job_ring : ref_obj<job_ring<MultiProducer>> {
if (HelpWorker) {
int32_t exp_in = 0;
while (token.compare_exchange_strong(exp_in, 1, std::memory_order_acquire)) {
num.fetch_sub(drain(), std::memory_order_relaxed);
num.fetch_sub(drain(), std::memory_order_release);
int32_t exp_out = 1;
if (token.compare_exchange_strong(exp_out, 0, std::memory_order_release)) {
if (!all_done()) { // have new task enqueue
continue;
}
if (waiter.exchange(0) == 1) {
waiter.notify_all();
}
return;
}
}
interfaces/kits/cpp/pattern/job_utils.h
+34 -16
View File
@@ -56,16 +56,16 @@
#endif
#ifndef FFRT_API_LOGE
#define FFRT_API_LOGE(fmt, ...)
#define FFRT_API_LOGE(...)
#endif
#ifndef FFRT_API_LOGD
#define FFRT_API_LOGD(fmt, ...)
#define FFRT_API_LOGD(...)
#endif
#ifndef FFRT_API_TRACE_INT64
#define FFRT_API_TRACE_INT64(name, value)
#endif
#ifndef FFRT_API_TRACE_SCOPE
#define FFRT_API_TRACE_SCOPE(fmt, ...)
#define FFRT_API_TRACE_SCOPE(...)
#endif
namespace ffrt {
@@ -284,13 +284,14 @@ struct ref_obj {
inline ptr& operator=(ptr const& h)
{
if (this != &h) {
if (p) {
p->dec_ref();
}
T* op = p;
p = h.p;
if (p) {
p->inc_ref();
}
if (op) {
op->dec_ref();
}
}
return *this;
}
@@ -314,11 +315,12 @@ struct ref_obj {
inline ptr& operator=(ptr&& h)
{
if (this != &h) {
if (p) {
p->dec_ref();
}
T* op = p;
p = h.p;
h.p = nullptr;
if (op) {
op->dec_ref();
}
}
return *this;
}
@@ -504,7 +506,7 @@ struct mpmc_queue : detail::non_copyable {
auto iwrite = iwrite_.load(std::memory_order_relaxed);
for (;;) {
i = &q[iwrite & mask];
if (i->iwrite_exp.load(std::memory_order_relaxed) != iwrite) {
if (i->iwrite_exp.load(std::memory_order_acquire) != iwrite) {
return false;
}
if ((iwrite_.compare_exchange_weak(iwrite, iwrite + 1, std::memory_order_relaxed))) {
@@ -530,7 +532,7 @@ struct mpmc_queue : detail::non_copyable {
auto iread = iread_.load(std::memory_order_relaxed);
for (;;) {
i = &q[iread & mask];
if (i->iread_exp.load(std::memory_order_relaxed) != iread) {
if (i->iread_exp.load(std::memory_order_acquire) != iread) {
return false;
}
if (iread_.compare_exchange_weak(iread, iread + 1, std::memory_order_relaxed)) {
@@ -826,19 +828,23 @@ struct fiber : detail::non_copyable {
/**
* @brief Returns the thread-local environment.
*/
static __attribute__((noinline)) thread_env& env()
#ifdef FFRT_MULTI_DSO_EXTERN_FIBER_ENV
static __attribute__((visibility("default"), noinline)) thread_env& env();
#else
static __attribute__((visibility("default"), noinline)) thread_env& env()
{
static thread_local thread_env ctx;
return ctx;
}
#endif
/**
* @brief Initializes a fiber with a function and stack.
*
* @param f Function to run.
* @param stack Stack memory.
* @param stack_size Stack size.
* @return Pointer to the created fiber.
* @param stack_size Stack size. Must be large enough to hold both the fiber header and the fiber context.
* @return Pointer to the created fiber; `nullptr` if `stack_size` is too small.
*/
static fiber* init(std::function<void()>&& f, void* stack, size_t stack_size)
{
@@ -873,7 +879,16 @@ struct fiber : detail::non_copyable {
{
bool done;
auto& e = fiber::env();
// Save the fiber that was running on this thread before switching in.
// `e.cur` is thread-local state read by suspend() and submit_to_master() to
// identify the running fiber. start() may be (re)entered while another fiber is
// already in flight on this thread — via a nested start() (a fiber's body starts
// a child fiber) or a cross-thread resume (a partner worker restarts a suspended
// fiber through suspendable_job_func). We must restore, not reset, `cur`;
// otherwise the outer fiber resumes with a stale/null `cur`, making a later
// suspend() dereference null or submit_to_master() run its job inline on a
// fiber stack.
auto saved_cur = e.cur;
do {
e.cond = nullptr;
e.cur = this;
@@ -883,6 +898,7 @@ struct fiber : detail::non_copyable {
done = this->id_ == 0;
} while (e.cond && !(e.cond)(this));
e.cond = nullptr;
e.cur = saved_cur; // resume the outer fiber, or nullptr if entered from a native stack
return done;
}
@@ -902,7 +918,9 @@ struct fiber : detail::non_copyable {
} else {
e.cond = cond;
}
e.cur = nullptr;
// `e.cur` is intentionally left unchanged: its lifecycle is owned by start(),
// which saves/restores it around each switch-in. Clearing it here would drop
// the outer fiber's identity on a nested or cross-thread resume (see start()).
ffrt_fiber_switch(&j->fb, &j->link);
}
interfaces/kits/ffrt.h
+2
View File
@@ -50,6 +50,8 @@
#include "cpp/condition_variable.h"
#include "cpp/sleep.h"
#include "cpp/queue.h"
#include "cpp/pattern/job_partner.h"
#include "cpp/pattern/job_ring.h"
#include "c/timer.h"
#include "c/loop.h"
#include "c/fiber.h"