This is an automated email from the ASF dual-hosted git repository.
chenBright pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new ad5b904f2 Fix RemoteTaskQueue empty check race (#3641)
ad5b904f2 is described below
commit ad5b904f23206abe7335d514f941801076733c45
Author: nkysg <[email protected]>
AuthorDate: Mon Oct 26 10:38:19 2026 +0800
Fix RemoteTaskQueue empty check race (#3641)
---
src/bthread/remote_task_queue.h | 18 +++++++++++++++---
1 file changed, 15 insertions(+), 3 deletions(-)
diff --git a/src/bthread/remote_task_queue.h b/src/bthread/remote_task_queue.h
index 98fc522b6..e87129abb 100644
--- a/src/bthread/remote_task_queue.h
+++ b/src/bthread/remote_task_queue.h
@@ -23,6 +23,7 @@
#define BTHREAD_REMOTE_TASK_QUEUE_H
#include "butil/containers/bounded_queue.h"
+#include "butil/atomicops.h"
#include "butil/macros.h"
namespace bthread {
@@ -35,7 +36,7 @@ class TaskGroup;
// The function names should be self-explanatory.
class RemoteTaskQueue {
public:
- RemoteTaskQueue() {}
+ RemoteTaskQueue() : _ntask(0) {}
int init(size_t cap) {
const size_t memsize = sizeof(bthread_t) * cap;
@@ -49,11 +50,16 @@ public:
}
bool pop(bthread_t* task) {
- if (_tasks.empty()) {
+ if (_ntask.load(butil::memory_order_relaxed) == 0) {
return false;
}
_mutex.lock();
const bool result = _tasks.pop(task);
+ if (result) {
+ const size_t ntask = _ntask.load(butil::memory_order_relaxed);
+ DCHECK_GT(ntask, 0u);
+ _ntask.store(ntask - 1, butil::memory_order_relaxed);
+ }
_mutex.unlock();
return result;
}
@@ -66,7 +72,12 @@ public:
}
bool push_locked(bthread_t task) {
- return _tasks.push(task);
+ if (!_tasks.push(task)) {
+ return false;
+ }
+ const size_t ntask = _ntask.load(butil::memory_order_relaxed);
+ _ntask.store(ntask + 1, butil::memory_order_relaxed);
+ return true;
}
size_t capacity() const { return _tasks.capacity(); }
@@ -76,6 +87,7 @@ friend class TaskGroup;
DISALLOW_COPY_AND_ASSIGN(RemoteTaskQueue);
butil::BoundedQueue<bthread_t> _tasks;
butil::Mutex _mutex;
+ butil::atomic<size_t> _ntask;
};
} // namespace bthread
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
