一、锁的冲突检测

在PG中如果申请的锁是强锁，或者已经有其他进程对表加了强锁，如要将本地锁表中的所有锁注册到共享锁表（主锁表）中进行冲突检测，来决定当前申请的锁是否需要等待。

PG对每一级表锁都设置一个冲突检测位图：

static const LOCKMASK LockConflicts[] = {0,/* AccessShareLock */LOCKBIT_ON(AccessExclusiveLock),/* RowShareLock */LOCKBIT_ON(ExclusiveLock) | LOCKBIT_ON(AccessExclusiveLock),/* RowExclusiveLock */LOCKBIT_ON(ShareLock) | LOCKBIT_ON(ShareRowExclusiveLock) |LOCKBIT_ON(ExclusiveLock) | LOCKBIT_ON(AccessExclusiveLock),/* ShareUpdateExclusiveLock */LOCKBIT_ON(ShareUpdateExclusiveLock) |LOCKBIT_ON(ShareLock) | LOCKBIT_ON(ShareRowExclusiveLock) |LOCKBIT_ON(ExclusiveLock) | LOCKBIT_ON(AccessExclusiveLock),/* ShareLock */LOCKBIT_ON(RowExclusiveLock) | LOCKBIT_ON(ShareUpdateExclusiveLock) |LOCKBIT_ON(ShareRowExclusiveLock) |LOCKBIT_ON(ExclusiveLock) | LOCKBIT_ON(AccessExclusiveLock),/* ShareRowExclusiveLock */LOCKBIT_ON(RowExclusiveLock) | LOCKBIT_ON(ShareUpdateExclusiveLock) |LOCKBIT_ON(ShareLock) | LOCKBIT_ON(ShareRowExclusiveLock) |LOCKBIT_ON(ExclusiveLock) | LOCKBIT_ON(AccessExclusiveLock),/* ExclusiveLock */LOCKBIT_ON(RowShareLock) |LOCKBIT_ON(RowExclusiveLock) | LOCKBIT_ON(ShareUpdateExclusiveLock) |LOCKBIT_ON(ShareLock) | LOCKBIT_ON(ShareRowExclusiveLock) |LOCKBIT_ON(ExclusiveLock) | LOCKBIT_ON(AccessExclusiveLock),/* AccessExclusiveLock */LOCKBIT_ON(AccessShareLock) | LOCKBIT_ON(RowShareLock) |LOCKBIT_ON(RowExclusiveLock) | LOCKBIT_ON(ShareUpdateExclusiveLock) |LOCKBIT_ON(ShareLock) | LOCKBIT_ON(ShareRowExclusiveLock) |LOCKBIT_ON(ExclusiveLock) | LOCKBIT_ON(AccessExclusiveLock)};

锁的冲突检测过程：

（1）如果锁位图与主锁表中的已授予锁位图的锁没有冲突，返回STATUS_OK。

 if (!(conflictMask & lock->grantMask)){PROCLOCK_PRINT("LockCheckConflicts: no conflict", proclock);return STATUS_OK;}

（2）如果有冲突，减去当前进程中的对应的锁个数。因为一个进程中的锁互不冲突。如果最后得到的冲突锁个数是零，返回STATUS_OK。

 myLocks = proclock->holdMask;               /*当前进程中的锁位图*/for (i = 1; i <= numLockModes; i++){if ((conflictMask & LOCKBIT_ON(i)) == 0){conflictsRemaining[i] = 0;continue;}/*主锁表的授予锁位图 减去 当前进程的锁位图*/conflictsRemaining[i] = lock->granted[i];if (myLocks & LOCKBIT_ON(i))--conflictsRemaining[i];totalConflictsRemaining += conflictsRemaining[i];}/* If no conflicts remain, we get the lock. */if (totalConflictsRemaining == 0){PROCLOCK_PRINT("LockCheckConflicts: resolved (simple)", proclock);return STATUS_OK;}/* If no group locking, it's definitely a conflict. */if (proclock->groupLeader == MyProc && MyProc->lockGroupLeader == NULL){Assert(proclock->tag.myProc == MyProc);PROCLOCK_PRINT("LockCheckConflicts: conflicting (simple)",proclock);return STATUS_FOUND;}

（3）如果还有冲突，在（2）中保留的冲突位图基础上减去当前进程组中的其他进程的锁，检测冲突。

 procLocks = &(lock->procLocks);/*得到进程组中的其他进程的锁*/otherproclock = (PROCLOCK *)SHMQueueNext(procLocks, procLocks, offsetof(PROCLOCK, lockLink));while (otherproclock != NULL){if (proclock != otherproclock &&proclock->groupLeader == otherproclock->groupLeader &&(otherproclock->holdMask & conflictMask) != 0){int          intersectMask = otherproclock->holdMask & conflictMask;     /*得到当前进程组的其他进程在冲突锁位图中对应的锁的个数*/for (i = 1; i <= numLockModes; i++){if ((intersectMask & LOCKBIT_ON(i)) != 0)         {if (conflictsRemaining[i] <= 0)elog(PANIC, "proclocks held do not match lock");conflictsRemaining[i]--;totalConflictsRemaining--;}}if (totalConflictsRemaining == 0){PROCLOCK_PRINT("LockCheckConflicts: resolved (group)",proclock);return STATUS_OK;}}otherproclock = (PROCLOCK *)SHMQueueNext(procLocks, &otherproclock->lockLink,offsetof(PROCLOCK, lockLink));}/* Nope, it's a real conflict. */PROCLOCK_PRINT("LockCheckConflicts: conflicting (group)", proclock);return STATUS_FOUND;

代码不好理解，引用一位大佬画的图更容易理解：

（1）conflictMak:当前申请的而所得冲突位图表。

Lock->granted：所有的锁授予的数量，该位图表存储在主锁表中。

proclock->holdMask ：当前进程持有的锁信息

conflictsRemaining: Lock->granted-proclock->holdMask 减去当前进程锁的保留锁表位图，此时如果该位图中的总数是0，代表主锁表中的锁与当前申请的锁没有冲突，如果大于0，需要做下一步判断。

（2）otherprocLock->holdMask:同进程中其他进程持锁信息。

intersectMask:减去本进程冲突锁后剩余的冲突数量。

conflictsRemaining: (1)中的 conflictsRemaining - intersectMask 减去当前进程组其他进程的锁的保留锁位图。这里已经在主锁表中排除当前进程组中的所有锁，如果conflictsRemaining在conflictMask位图的对应位数量仍然大于0，代表当前申请的锁与主锁表中的锁有冲突。

二、死锁检测

在postgresql数据库中，对申请事务锁的顺序没有严格的限制，在申请锁时也没有严格的检查，因此不可避免的会发生死锁。PG使用“乐观锁”来检测死锁，即锁等待的时间超过了时间deadlock_timeout（postgresql.conf中配置，deadlock_timeout = 1s），那么开始进行死锁检测，如果没有死锁继续等待。

WFG（有向图）：

如果一个事务存在冲突模式，需要另一个持有冲突锁的事务释放锁才能持有该锁，这里就存在等待过程。假设T2在等待T1，可以用一个有向图来表示。

如果等待是一条单线，则不存在死锁，只需要按照顺序执行事务即可。但是如果有向图中产生环，就会产生死锁：

实边（hard edge）：

在常规锁申请过程中，假设事务A持有表的共享锁或者排他锁，此时事务B申请排他锁时，就需要进入等待过程，PG称这种等待的边是实边。

虚边（soft edge）：

如果事务A持有共享锁，B申请排他锁，此时又有事务C申请共享锁。虽然事务C和事务A持有的共享锁不冲突，但是和等待队列中的B申请的排他锁冲突，所以事务C也需要进入等待队列，此时

C<-B 就是一个虚边。

pg死锁即WFG存在环，从起点进程开始按照有向图能够回到起点。如果环中存在soft edge,可以通过调整soft edge两个顶点进程的执行顺序来解决死锁。如果没有soft edge，只能通过杀掉其中的一个进程来解决死锁。

死锁检测也是检测WFG中环的状态。

相关结构体：
（1）死锁检测状态

typedef enum
{DS_NOT_YET_CHECKED,            /* 没有死锁检测 */DS_NO_DEADLOCK,             /* 没有死锁 */DS_SOFT_DEADLOCK,         /* 死锁，有虚边，可以通过调整虚边进程的顺序来解决死锁 */DS_HARD_DEADLOCK,          /* 死锁，没有虚边，只能报错 */DS_BLOCKED_BY_AUTOVACUUM    /* no deadlock; queue blocked by autovacuum* worker */
} DeadLockState;

（2）WFG边的结构体

typedef struct
{PGPROC    *waiter;         /* 等待的进程组的leader，EDGE边的等待进程，需要等待 blocker释放锁*/PGPROC      *blocker;        /* 被等待的进程组的leader*/LOCK    *lock;           /* 等待的锁 */int           pred;           /* workspace for TopoSort */int         link;           /* workspace for TopoSort */
} EDGE;

（3）已经检测到的死锁WFG图中每条边的信息

typedef struct
{LOCKTAG        locktag;        /* ID of awaited lock object */LOCKMODE lockmode;       /* type of lock we're waiting for */int            pid;            /* PID of blocked backend */
} DEADLOCK_INFO;

（4）新的等待队列，每个数组元素都是一个等待队列

typedef struct
{LOCK      *lock;           /* 调整的是这个锁的等待队列 */PGPROC      **procs;          /*等待队列数组 */int          nProcs;
} WAIT_ORDER;

死锁检测函数

（1）死锁检测的入口函数是CheckDeadLock函数。需要注意的是，死锁检测的过程是互斥的，就是说在死锁检测开始的时候，锁表就需要被保护起来，防止被其他人修改。

static void
CheckDeadLock(void)
{int            i;/*死锁检测时需要将所有锁表保护起来，防止其他进程修改锁的持有状态与锁的等待状态*/for (i = 0; i < NUM_LOCK_PARTITIONS; i++)LWLockAcquire(LockHashPartitionLockByIndex(i), LW_EXCLUSIVE);/** Check to see if we've been awoken by anyone in the interim.** If we have, we can return and resume our transaction -- happy day.* Before we are awoken the process releasing the lock grants it to us so* we know that we don't have to wait anymore.** We check by looking to see if we've been unlinked from the wait queue.* This is safe because we hold the lock partition lock.*//*只有当前的会话，不会构成死锁*/if (MyProc->links.prev == NULL ||MyProc->links.next == NULL)goto check_done;#ifdef LOCK_DEBUGif (Debug_deadlocks)DumpAllLocks();
#endif/* Run the deadlock check, and set deadlock_state for use by ProcSleep *//* 检测锁链中是否有环，死锁检测，返回死锁状态 */deadlock_state = DeadLockCheck(MyProc);if (deadlock_state == DS_HARD_DEADLOCK){Assert(MyProc->waitLock != NULL);RemoveFromWaitQueue(MyProc, LockTagHashCode(&(MyProc->waitLock->tag)));}check_done:for (i = NUM_LOCK_PARTITIONS; --i >= 0;)LWLockRelease(LockHashPartitionLockByIndex(i));
}

（2）死锁检测的入口函数，一旦发现了死锁，该函数会重新排序锁的等待队列以消除这个死锁，然后将调整后的等待队列替换现有的等待队列。如果这个死锁无法消除，则返回DS_ HEAD_ DEAD_LOCK。该返回值提醒调用者中断正在处理的事务。

DeadLockState
DeadLockCheck(PGPROC *proc)
{int            i,j;/* Initialize to "no constraints" */nCurConstraints = 0;         /* 被探索的虚边的个数， "约束"代表可以通过调整虚边的顺序来解决队列中死锁的问题，* 每次死锁检测时都会从PossibleConstraints中找到个虚边进程探索，* 被探索的虚边保存在CurConstraints中*/nPossibleConstraints = 0;      //可以被调整的虚边的个数nWaitOrders = 0;/* Initialize to not blocked by an autovacuum worker */blocking_autovacuum_proc = NULL;/* Search for deadlocks and possible fixes */if (DeadLockCheckRecurse(proc))                      /* 递归的检测等待队列中有没有死锁环（包括实环与虚环） */{/** Call FindLockCycle one more time, to record the correct* deadlockDetails[] for the basic state with no rearrangements.*/int           nSoftEdges;TRACE_POSTGRESQL_DEADLOCK_FOUND();nWaitOrders = 0;if (!FindLockCycle(proc, possibleConstraints, &nSoftEdges))       elog(FATAL, "deadlock seems to have disappeared");return DS_HARD_DEADLOCK;    /* cannot find a non-deadlocked state */}/* Apply any needed rearrangements of wait queues *//*如果调整过后没有死锁，用调整过得等待队列代替现有的等待队列*/for (i = 0; i < nWaitOrders; i++){LOCK      *lock = waitOrders[i].lock;PGPROC     **procs = waitOrders[i].procs;int            nProcs = waitOrders[i].nProcs;PROC_QUEUE *waitQueue = &(lock->waitProcs);Assert(nProcs == waitQueue->size);#ifdef DEBUG_DEADLOCKPrintLockQueue(lock, "DeadLockCheck:");
#endif/* Reset the queue and re-add procs in the desired order */ProcQueueInit(waitQueue);for (j = 0; j < nProcs; j++){SHMQueueInsertBefore(&(waitQueue->links), &(procs[j]->links));waitQueue->size++;}#ifdef DEBUG_DEADLOCKPrintLockQueue(lock, "rearranged to:");
#endif/* See if any waiters for the lock can be woken up now */ProcLockWakeup(GetLocksMethodTable(lock), lock);}/* Return code tells caller if we had to escape a deadlock or not */if (nWaitOrders > 0)return DS_SOFT_DEADLOCK;else if (blocking_autovacuum_proc != NULL)return DS_BLOCKED_BY_AUTOVACUUM;elsereturn DS_NO_DEADLOCK;
}

（3）上面的函数中通过DeadLockCheckRecurse来递归的的检测死锁，具体方法是将当前"约束"（curConstraints：里面存放的是所有的调整过顺序的虚边）的等待列表做拓扑排序，通过拓扑排序得到一个新的等待队列，检测死锁环。如果有死锁环并且是实环就报死锁，如果是虚边就将该环放到curConstraints（里面保存的是等待队列的虚边），然后从后对虚环中的两个进程调整，返回PROC，然后再调用DeadLockCheckRecurse来检测，依次递归直到没有虚环或者报死锁。

static bool
DeadLockCheckRecurse(PGPROC *proc)              /* 逐一检测 PGPROC 结构体链表，返回死锁状态 */
{int            nEdges;int          oldPossibleConstraints;bool     savedList;int           i;/*  */nEdges = TestConfiguration(proc);if (nEdges < 0)return true;            /* hard deadlock --- no solution */if (nEdges == 0)return false;          /* good configuration found */if (nCurConstraints >= maxCurConstraints)                             /* curConstraints:当前的配置 */return true;          /* out of room for active constraints? */oldPossibleConstraints = nPossibleConstraints;if (nPossibleConstraints + nEdges + MaxBackends <= maxPossibleConstraints){/* We can save the edge list in possibleConstraints[] */nPossibleConstraints += nEdges;                                      /*  */savedList = true;}else{/* Not room; will need to regenerate the edges on-the-fly */savedList = false;}/** Try each available soft edge as an addition to the configuration.  尝试将每个可用的软边缘添加到配置中。*/for (i = 0; i < nEdges; i++){if (!savedList && i > 0){/* Regenerate the list of possible added constraints */if (nEdges != TestConfiguration(proc))elog(FATAL, "inconsistent results during deadlock check");}/* 将检测到的虚边分别尝试加入新的Configuration，然后进行死锁检测 */curConstraints[nCurConstraints] =possibleConstraints[oldPossibleConstraints + i];nCurConstraints++;if (!DeadLockCheckRecurse(proc))    //递归检测return false;     /* found a valid solution! *//* give up on that added constraint, try again */nCurConstraints--;}nPossibleConstraints = oldPossibleConstraints;return true;                /* no solution found */
}

（4）TestConfiguration函数：对“约束”里面的虚边的等待队列进行拓扑排序，然后检测有没有环。

static int
TestConfiguration(PGPROC *startProc)
{int            softFound = 0;EDGE    *softEdges = possibleConstraints + nPossibleConstraints;               /*possibleConstraints:可以被调整的虚边* nPossibleConstraints:可以更改的虚边的个数*/int            nSoftEdges;                         //遍历得到的虚边的个数int         i;/** Make sure we have room for FindLockCycle's output.*/return -1;if (nPossibleConstraints + MaxBackends > maxPossibleConstraints)/** Expand current constraint set into wait orderings.  Fail if the* constraint set is not self-consistent.*//* 对约束中的边的等待队列拓扑排序，按照后向扫描的方法，因为新的虚环是从后面加入约束的，拓扑排序的话更容易检测到环 */if (!ExpandConstraints(curConstraints, nCurConstraints))return -1;/** Check for cycles involving startProc or any of the procs mentioned in* constraints.  We check startProc last because if it has a soft cycle* still to be dealt with, we want to deal with that first.*//* 从"约束的边" 的上下顶点找环 */for (i = 0; i < nCurConstraints; i++){if (FindLockCycle(curConstraints[i].waiter, softEdges, &nSoftEdges)){if (nSoftEdges == 0)return -1;       /* hard deadlock detected */softFound = nSoftEdges;}if (FindLockCycle(curConstraints[i].blocker, softEdges, &nSoftEdges)){if (nSoftEdges == 0)return -1;     /* hard deadlock detected */softFound = nSoftEdges;}}/* 从起始定点找环 */if (FindLockCycle(startProc, softEdges, &nSoftEdges)){if (nSoftEdges == 0)return -1;           /* hard deadlock detected */softFound = nSoftEdges;}return softFound;
}

postgresql锁的冲突检测死锁检测相关推荐

linux 内核死锁检查,一种linux内核自旋锁死锁检测报告系统和方法与流程
本发明涉及内核死锁检测领域,具体的说是一种linux内核自旋锁死锁检测报告系统和方法. 背景技术: linux内核死锁是长期困扰内核开发人员的问题之一,但自内核引入lockdep调试模块之后,内核死锁 ...
10、MySQL锁等待，死锁，死锁检测
使用数据库时,有时会出现死锁.对于实际应用来说,就是出现系统卡顿. 死锁是指两个或两个以上的事务在执行过程中,因争夺资源而造成的一种互相等待的现象.就是所谓的锁资源请求产生了回路现象,即死循环,此时称 ...
mysql二级封锁协议_MySQL 行锁、两阶段锁协议、死锁以及死锁检测
行锁 MySQL的行锁都是在引擎层实现的,但是 MyISAM 不支持行锁,意味着并发控制只能使用表锁,同一张表任何时刻只能被一个更新在执行,影响到业务并发度.InnoDB 是支持行锁的,这也是 MyI ...
postgresql 锁_PostgreSQL中的锁：3.其他锁
postgresql 锁 object-level locks (specifically, relation-level locks), as well as 对象级别的锁 (特别是关系级别的锁), ...
linux死锁检测的一种思路【转】
转自:http://www.cnblogs.com/mumuxinfei/p/4365697.html 前言: 上一篇博文讲述了pstack的使用和原理. 和jstack一样, pstack能获取进 ...
MySQL - 锁等待及死锁初探
文章目录生猛干货版本信息 MySQL 行锁分析 MySQL死锁演示排查过程查看近期死锁日志信息查询锁等待命令及kill 锁优化建议搞定MySQL 生猛干货带你搞定MySQL实战,轻松对 ...
java简单的死锁检测(转载线下代码)
一个简单的死锁检测功能. 1.探测出死锁主要是:java.lang.management类的关于线程的系列方法一个线程监控类 ThreadMXBean x=ManagementFactory.ge ...
MySQL 锁与MVCC ：数据库的锁、MVCC、当前读、快照读、锁算法、死锁
文章目录 lock与latch 锁的类型 MVCC 一致性非锁定读(快照读) 一致性锁定读(当前读) 锁算法死锁锁升级 lock与latch 在了解数据库锁之前,首先就要区分开lock和latch ...
《MySQL——幻读与next-key lock与间隙锁带来的死锁》
create table 't' ('id' int(11) not null,'c' int(11) default null,'d' int(11) default null,primary ke ...

postgresql锁的冲突检测死锁检测

一、锁的冲突检测

二、死锁检测

postgresql锁的冲突检测死锁检测相关推荐

最新文章

热门文章

postgresql锁的冲突检测 死锁检测

一、锁的冲突检测

二、死锁检测

postgresql锁的冲突检测 死锁检测相关推荐

最新文章

热门文章

postgresql锁的冲突检测死锁检测

postgresql锁的冲突检测死锁检测相关推荐