| 333 | TaskNode* const TaskNode::UNCONNECTED = (TaskNode*)-1L; |
| 334 | |
| 335 | ExecutionQueueBase::scoped_ptr_t ExecutionQueueBase::address(uint64_t id) { |
| 336 | scoped_ptr_t ret; |
| 337 | const slot_id_t slot = slot_of_id(id); |
| 338 | ExecutionQueueBase* const m = butil::address_resource(slot); |
| 339 | if (BAIDU_LIKELY(m != NULL)) { |
| 340 | // acquire fence makes sure this thread sees latest changes before |
| 341 | // _dereference() |
| 342 | const uint64_t vref1 = m->_versioned_ref.fetch_add( |
| 343 | 1, butil::memory_order_acquire); |
| 344 | const uint32_t ver1 = _version_of_vref(vref1); |
| 345 | if (ver1 == _version_of_id(id)) { |
| 346 | ret.reset(m); |
| 347 | return ret.Pass(); |
| 348 | } |
| 349 | |
| 350 | const uint64_t vref2 = m->_versioned_ref.fetch_sub( |
| 351 | 1, butil::memory_order_release); |
| 352 | const int32_t nref = _ref_of_vref(vref2); |
| 353 | if (nref > 1) { |
| 354 | return ret.Pass(); |
| 355 | } else if (__builtin_expect(nref == 1, 1)) { |
| 356 | const uint32_t ver2 = _version_of_vref(vref2); |
| 357 | if ((ver2 & 1)) { |
| 358 | if (ver1 == ver2 || ver1 + 1 == ver2) { |
| 359 | uint64_t expected_vref = vref2 - 1; |
| 360 | if (m->_versioned_ref.compare_exchange_strong( |
| 361 | expected_vref, _make_vref(ver2 + 1, 0), |
| 362 | butil::memory_order_acquire, |
| 363 | butil::memory_order_relaxed)) { |
| 364 | m->_on_recycle(); |
| 365 | // We don't return m immediatly when the reference count |
| 366 | // reaches 0 as there might be in processing tasks. Instead |
| 367 | // _on_recycle would push a `stop_task', after which |
| 368 | // is executed m would be finally reset and returned |
| 369 | } |
| 370 | } else { |
| 371 | CHECK(false) << "ref-version=" << ver1 |
| 372 | << " unref-version=" << ver2; |
| 373 | } |
| 374 | } else { |
| 375 | CHECK_EQ(ver1, ver2); |
| 376 | // Addressed a free slot. |
| 377 | } |
| 378 | } else { |
| 379 | CHECK(false) << "Over dereferenced id=" << id; |
| 380 | } |
| 381 | } |
| 382 | return ret.Pass(); |
| 383 | } |
| 384 | |
| 385 | int ExecutionQueueBase::create(uint64_t* id, const ExecutionQueueOptions* options, |
| 386 | execute_func_t execute_func, |
nothing calls this directly
no test coverage detected