| 33 | |
| 34 | |
| 35 | bool DoExecute(Lock *lock, LockDb::Type db_type, const int paxos_group_id, |
| 36 | const uint64_t instance_id, const proto::LockPaxosArgs &args, |
| 37 | const bool sync = false, phxpaxos::SMCtx *sm_ctx = nullptr) { |
| 38 | LockContext *lc{nullptr}; |
| 39 | if (sm_ctx && sm_ctx->m_pCtx) { |
| 40 | // only master set ctx |
| 41 | lc = static_cast<LockContext *>(sm_ctx->m_pCtx); |
| 42 | } |
| 43 | |
| 44 | if (!lock || !lock->GetLockMgr()) { |
| 45 | NLErr("topic_id %d paxos_group_id %d instance_id %llu lock_mgr null", |
| 46 | lock->GetTopicID(), paxos_group_id, instance_id); |
| 47 | if (lc) |
| 48 | lc->result = comm::RetCode::RET_ERR_LOGIC; |
| 49 | |
| 50 | return false; |
| 51 | } |
| 52 | |
| 53 | NLVerb("topic_id %d paxos_group_id %d instance_id %llu", |
| 54 | lock->GetTopicID(), paxos_group_id, instance_id); |
| 55 | |
| 56 | if (args.has_acquire_lock_req()) { |
| 57 | NLInfo("topic_id %d paxos_group_id %d instance_id %llu acquire lock sync %d", |
| 58 | lock->GetTopicID(), paxos_group_id, instance_id, sync); |
| 59 | auto &&lock_info(args.acquire_lock_req().lock_info()); |
| 60 | |
| 61 | proto::LockKeyInfo lock_key_info; |
| 62 | lock_key_info.set_version(lock_info.version()); |
| 63 | lock_key_info.set_lock_key(lock_info.lock_key()); |
| 64 | |
| 65 | proto::LocalLockInfo local_lock_info; |
| 66 | local_lock_info.set_version(instance_id); |
| 67 | local_lock_info.set_client_id(lock_info.client_id()); |
| 68 | local_lock_info.set_lease_time_ms(lock_info.lease_time_ms()); |
| 69 | local_lock_info.set_expire_time_ms(comm::utils::Time::GetSteadyClockMS() + lock_info.lease_time_ms()); |
| 70 | |
| 71 | comm::RetCode ret{comm::RetCode::RET_ERR_UNINIT}; |
| 72 | if (LockDb::Type::MAP == db_type) { |
| 73 | comm::utils::MutexGuard guard(lock->GetLockMgr()->map(paxos_group_id).mutex()); |
| 74 | ret = lock->GetLockMgr()->map(paxos_group_id). |
| 75 | AcquireLock(lock_key_info, local_lock_info); |
| 76 | |
| 77 | // update last instance id |
| 78 | lock->GetLockMgr()->set_last_instance_id(paxos_group_id, instance_id); |
| 79 | NLInfo("topic_id %d paxos_group_id %d last_instance_id %llu", |
| 80 | lock->GetTopicID(), paxos_group_id, instance_id); |
| 81 | } else if (LockDb::Type::LEVELDB == db_type) { |
| 82 | ret = lock->GetLockMgr()->leveldb(paxos_group_id). |
| 83 | AcquireLock(lock_key_info, local_lock_info, sync); |
| 84 | |
| 85 | // update checkpoint |
| 86 | if (comm::RetCode::RET_OK == ret && sync) { |
| 87 | ret = lock->GetLockMgr()->WriteCheckpoint(paxos_group_id, instance_id); |
| 88 | if (comm::RetCode::RET_OK == ret) { |
| 89 | NLInfo("topic_id %d paxos_group_id %d WriteCheckpoint cp %llu ok", |
| 90 | lock->GetTopicID(), paxos_group_id, instance_id); |
| 91 | } else { |
| 92 | NLErr("topic_id %d paxos_group_id %d WriteCheckpoint cp %llu err %d", |
no test coverage detected