1 --- a/lock/lock_deadlock.c
2 +++ b/lock/lock_deadlock.c
3 @@ -121,7 +121,7 @@ __lock_detect(env, atype, rejectp)
7 - u_int32_t *bitmap, *copymap, **deadp, **free_me, *tmpmap;
8 + u_int32_t *bitmap, *copymap, **deadp, **deadlist, *tmpmap;
9 u_int32_t i, cid, keeper, killid, limit, nalloc, nlockers;
10 u_int32_t lock_max, txn_max;
12 @@ -133,7 +133,8 @@ __lock_detect(env, atype, rejectp)
13 if (IS_REP_CLIENT(env))
14 atype = DB_LOCK_MINWRITE;
17 + copymap = tmpmap = NULL;
22 @@ -179,11 +180,11 @@ __lock_detect(env, atype, rejectp)
23 memcpy(copymap, bitmap, nlockers * sizeof(u_int32_t) * nalloc);
25 if ((ret = __os_calloc(env, sizeof(u_int32_t), nalloc, &tmpmap)) != 0)
29 /* Find a deadlock. */
31 - __dd_find(env, bitmap, idmap, nlockers, nalloc, &deadp)) != 0)
32 + __dd_find(env, bitmap, idmap, nlockers, nalloc, &deadlist)) != 0)
36 @@ -204,8 +205,7 @@ __lock_detect(env, atype, rejectp)
37 txn_max = TXN_MAXIMUM;
41 - for (; *deadp != NULL; deadp++) {
42 + for (deadp = deadlist; *deadp != NULL; deadp++) {
45 killid = (u_int32_t)(*deadp - bitmap) / nalloc;
46 @@ -342,11 +342,12 @@ dokill: if (killid == BAD_KILLID) {
48 "Aborting locker %lx", (u_long)idmap[killid].id);
50 - __os_free(env, tmpmap);
51 -err1: __os_free(env, copymap);
53 -err: if (free_me != NULL)
54 - __os_free(env, free_me);
55 +err: if(copymap != NULL)
56 + __os_free(env, copymap);
57 + if (deadlist != NULL)
58 + __os_free(env, deadlist);
60 + __os_free(env, tmpmap);
61 __os_free(env, bitmap);
62 __os_free(env, idmap);
64 @@ -360,6 +361,17 @@ err: if (free_me != NULL)
66 #define DD_INVALID_ID ((u_int32_t) -1)
70 + * Build the lock dependency bit maps.
71 + * Notes on synchronization:
72 + * LOCK_SYSTEM_LOCK is used to hold objects locked when we have
73 + * a single partition.
74 + * LOCK_LOCKERS is held while we are walking the lockers list and
75 + * to single thread the use of lockerp->dd_id.
76 + * LOCK_DD protects the DD list of objects.
80 __dd_build(env, atype, bmp, nlockers, allocp, idmap, rejectp)
82 @@ -393,6 +405,7 @@ __dd_build(env, atype, bmp, nlockers, al
83 * In particular we do not build the conflict array and our caller
84 * needs to expect this.
86 + LOCK_SYSTEM_LOCK(lt, region);
87 if (atype == DB_LOCK_EXPIRE) {
88 skip: LOCK_DD(env, region);
89 op = SH_TAILQ_FIRST(®ion->dd_objs, __db_lockobj);
90 @@ -430,17 +443,18 @@ skip: LOCK_DD(env, region);
91 OBJECT_UNLOCK(lt, region, indx);
93 UNLOCK_DD(env, region);
94 + LOCK_SYSTEM_UNLOCK(lt, region);
99 - * We'll check how many lockers there are, add a few more in for
100 - * good measure and then allocate all the structures. Then we'll
101 - * verify that we have enough room when we go back in and get the
102 - * mutex the second time.
103 + * Allocate after locking the region
104 + * to make sure the structures are large enough.
106 -retry: count = region->stat.st_nlockers;
107 + LOCK_LOCKERS(env, region);
108 + count = region->stat.st_nlockers;
110 + UNLOCK_LOCKERS(env, region);
114 @@ -448,50 +462,37 @@ retry: count = region->stat.st_nlockers;
115 if (FLD_ISSET(env->dbenv->verbose, DB_VERB_DEADLOCK))
116 __db_msg(env, "%lu lockers", (u_long)count);
119 nentries = (u_int32_t)DB_ALIGN(count, 32) / 32;
122 - * Allocate enough space for a count by count bitmap matrix.
125 - * We can probably save the malloc's between iterations just
126 - * reallocing if necessary because count grew by too much.
128 + /* Allocate enough space for a count by count bitmap matrix. */
129 if ((ret = __os_calloc(env, (size_t)count,
130 - sizeof(u_int32_t) * nentries, &bitmap)) != 0)
131 + sizeof(u_int32_t) * nentries, &bitmap)) != 0) {
132 + UNLOCK_LOCKERS(env, region);
136 if ((ret = __os_calloc(env,
137 sizeof(u_int32_t), nentries, &tmpmap)) != 0) {
138 + UNLOCK_LOCKERS(env, region);
139 __os_free(env, bitmap);
143 if ((ret = __os_calloc(env,
144 (size_t)count, sizeof(locker_info), &id_array)) != 0) {
145 + UNLOCK_LOCKERS(env, region);
146 __os_free(env, bitmap);
147 __os_free(env, tmpmap);
152 - * Now go back in and actually fill in the matrix.
154 - if (region->stat.st_nlockers > count) {
155 - __os_free(env, bitmap);
156 - __os_free(env, tmpmap);
157 - __os_free(env, id_array);
162 * First we go through and assign each locker a deadlock detector id.
165 - LOCK_LOCKERS(env, region);
166 SH_TAILQ_FOREACH(lip, ®ion->lockers, ulinks, __db_locker) {
167 if (lip->master_locker == INVALID_ROFF) {
168 + DB_ASSERT(env, id < count);
170 id_array[lip->dd_id].id = lip->id;
172 @@ -510,7 +511,6 @@ retry: count = region->stat.st_nlockers;
173 lip->dd_id = DD_INVALID_ID;
176 - UNLOCK_LOCKERS(env, region);
179 * We only need consider objects that have waiters, so we use
180 @@ -669,7 +669,6 @@ again: memset(bitmap, 0, count * sizeof
181 * status after building the bit maps so that we will not detect
182 * a blocked transaction without noting that it is already aborting.
184 - LOCK_LOCKERS(env, region);
185 for (id = 0; id < count; id++) {
186 if (!id_array[id].valid)
188 @@ -738,6 +737,7 @@ get_lock: id_array[id].last_lock = R_OF
189 id_array[id].in_abort = 1;
191 UNLOCK_LOCKERS(env, region);
192 + LOCK_SYSTEM_UNLOCK(lt, region);
195 * Now we can release everything except the bitmap matrix that we
196 @@ -839,6 +839,7 @@ __dd_abort(env, info, statusp)
199 /* We must lock so this locker cannot go away while we abort it. */
200 + LOCK_SYSTEM_LOCK(lt, region);
201 LOCK_LOCKERS(env, region);
204 @@ -895,6 +896,7 @@ __dd_abort(env, info, statusp)
205 done: OBJECT_UNLOCK(lt, region, info->last_ndx);
207 out: UNLOCK_LOCKERS(env, region);
208 + LOCK_SYSTEM_UNLOCK(lt, region);