X-Git-Url: https://git.ozlabs.org/?p=ccan;a=blobdiff_plain;f=ccan%2Ftdb2%2Flock.c;h=10d1e18a512b6d0f0964e8eda80fb33b5914dd54;hp=15c97e3b98343b0a02ee910bdac0ec93664e3527;hb=c5e3f07a30056cb7c3c380bf690f80815ca4b6b1;hpb=6804501c350181dea8f531142b28c620b70edbd9;ds=sidebyside

diff --git a/ccan/tdb2/lock.c b/ccan/tdb2/lock.c
index 15c97e3b..10d1e18a 100644
--- a/ccan/tdb2/lock.c
+++ b/ccan/tdb2/lock.c
@@ -260,7 +260,7 @@ static int tdb_nest_lock(struct tdb_context *tdb, tdb_off_t offset, int ltype,
 	if (offset >= TDB_HASH_LOCK_START + TDB_HASH_LOCK_RANGE + tdb->map_size / 8) {
 		tdb->ecode = TDB_ERR_LOCK;
 		tdb->log(tdb, TDB_DEBUG_FATAL, tdb->log_priv,
-			 "tdb_lock: invalid offset %llu ltype=%d\n",
+			 "tdb_nest_lock: invalid offset %llu ltype=%d\n",
 			 (long long)offset, ltype);
 		return -1;
 	}
@@ -278,13 +278,22 @@ static int tdb_nest_lock(struct tdb_context *tdb, tdb_off_t offset, int ltype,
 		return 0;
 	}
 
+	if (tdb->num_lockrecs
+	    && offset >= TDB_HASH_LOCK_START
+	    && offset < TDB_HASH_LOCK_START + TDB_HASH_LOCK_RANGE) {
+		tdb->ecode = TDB_ERR_LOCK;
+		tdb->log(tdb, TDB_DEBUG_FATAL, tdb->log_priv,
+			 "tdb_nest_lock: already have a hash lock?\n");
+		return -1;
+	}
+
 	new_lck = (struct tdb_lock_type *)realloc(
 		tdb->lockrecs,
 		sizeof(*tdb->lockrecs) * (tdb->num_lockrecs+1));
 	if (new_lck == NULL) {
 		tdb->ecode = TDB_ERR_OOM;
 		tdb->log(tdb, TDB_DEBUG_ERROR, tdb->log_priv,
-			 "tdb_lock: unable to allocate %llu lock structure",
+			 "tdb_nest_lock: unable to allocate %llu lock struct",
 			 (long long)(tdb->num_lockrecs + 1));
 		errno = ENOMEM;
 		return -1;
@@ -351,7 +360,7 @@ static int tdb_nest_unlock(struct tdb_context *tdb, tdb_off_t off, int ltype)
 	if ((lck == NULL) || (lck->count == 0)) {
 		tdb->ecode = TDB_ERR_LOCK;
 		tdb->log(tdb, TDB_DEBUG_ERROR, tdb->log_priv,
-			 "tdb_unlock: no lock for %llu\n", (long long)off);
+			 "tdb_nest_unlock: no lock for %llu\n", (long long)off);
 		return -1;
 	}
 
@@ -433,9 +442,7 @@ static int tdb_lock_gradual(struct tdb_context *tdb,
 }
 
 /* lock/unlock entire database.  It can only be upgradable if you have some
- * other way of guaranteeing exclusivity (ie. transaction write lock).
- * Note that we don't lock the free chains: currently noone can get those locks
- * without a hash chain lock first. */
+ * other way of guaranteeing exclusivity (ie. transaction write lock). */
 int tdb_allrecord_lock(struct tdb_context *tdb, int ltype,
 		       enum tdb_lock_flags flags, bool upgradable)
 {
@@ -479,13 +486,27 @@ int tdb_allrecord_lock(struct tdb_context *tdb, int ltype,
 	}
 
 again:
+	/* Lock hashes, gradually. */
 	if (tdb_lock_gradual(tdb, ltype, flags, TDB_HASH_LOCK_START,
 			     TDB_HASH_LOCK_RANGE)) {
 		if (!(flags & TDB_LOCK_PROBE)) {
 			tdb->log(tdb, TDB_DEBUG_ERROR, tdb->log_priv,
-				 "tdb_lockall hashes failed (%s)\n",
+				 "tdb_allrecord_lock hashes failed (%s)\n",
+				 strerror(errno));
+		}
+		return -1;
+	}
+
+	/* Lock free lists: there to end of file. */
+	if (tdb_brlock(tdb, ltype, TDB_HASH_LOCK_START + TDB_HASH_LOCK_RANGE,
+		       0, flags)) {
+		if (!(flags & TDB_LOCK_PROBE)) {
+			tdb->log(tdb, TDB_DEBUG_ERROR, tdb->log_priv,
+				 "tdb_allrecord_lock freelist failed (%s)\n",
 				 strerror(errno));
 		}
+		tdb_brunlock(tdb, ltype, TDB_HASH_LOCK_START, 
+			     TDB_HASH_LOCK_RANGE);
 		return -1;
 	}
 
@@ -564,14 +585,12 @@ int tdb_allrecord_unlock(struct tdb_context *tdb, int ltype)
 	tdb->allrecord_lock.count = 0;
 	tdb->allrecord_lock.ltype = 0;
 
-	return tdb_brunlock(tdb, ltype, TDB_HASH_LOCK_START,
-			    TDB_HASH_LOCK_RANGE);
+	return tdb_brunlock(tdb, ltype, TDB_HASH_LOCK_START, 0);
 }
 
 bool tdb_has_expansion_lock(struct tdb_context *tdb)
 {
-	return find_nestlock(tdb, TDB_EXPANSION_LOCK) != NULL
-		|| (tdb->flags & TDB_NOLOCK);
+	return find_nestlock(tdb, TDB_EXPANSION_LOCK) != NULL;
 }
 
 bool tdb_has_locks(struct tdb_context *tdb)
@@ -625,6 +644,18 @@ int tdb_unlockall_read(struct tdb_context *tdb)
 }
 #endif
 
+static bool tdb_has_free_lock(struct tdb_context *tdb)
+{
+	unsigned int i;
+
+	for (i=0; i<tdb->num_lockrecs; i++) {
+		if (tdb->lockrecs[i].off
+		    > TDB_HASH_LOCK_START + TDB_HASH_LOCK_RANGE)
+			return true;
+	}
+	return false;
+}
+
 int tdb_lock_hashes(struct tdb_context *tdb,
 		    tdb_off_t hash_lock,
 		    tdb_len_t hash_range,
@@ -643,12 +674,26 @@ int tdb_lock_hashes(struct tdb_context *tdb,
 	if (tdb->allrecord_lock.count) {
 		tdb->ecode = TDB_ERR_LOCK;
 		tdb->log(tdb, TDB_DEBUG_ERROR, tdb->log_priv,
-			 "tdb_lock_list: have %s allrecordlock\n",
+			 "tdb_lock_hashes: have %s allrecordlock\n",
 			 tdb->allrecord_lock.ltype == F_RDLCK
 			 ? "read" : "write");
 		return -1;
 	}
 
+	if (tdb_has_free_lock(tdb)) {
+		tdb->ecode = TDB_ERR_LOCK;
+		tdb->log(tdb, TDB_DEBUG_ERROR, tdb->log_priv,
+			 "tdb_lock_hashes: have free lock already\n");
+		return -1;
+	}
+
+	if (tdb_has_expansion_lock(tdb)) {
+		tdb->ecode = TDB_ERR_LOCK;
+		tdb->log(tdb, TDB_DEBUG_ERROR, tdb->log_priv,
+			 "tdb_lock_hashes: have expansion lock already\n");
+		return -1;
+	}
+
 	return tdb_nest_lock(tdb, lock, ltype, waitflag);
 }
 
@@ -665,7 +710,7 @@ int tdb_unlock_hashes(struct tdb_context *tdb,
 		    && ltype == F_WRLCK) {
 			tdb->ecode = TDB_ERR_LOCK;
 			tdb->log(tdb, TDB_DEBUG_FATAL, tdb->log_priv,
-				 "tdb_unlock_list RO allrecord!\n");
+				 "tdb_unlock_hashes RO allrecord!\n");
 			return -1;
 		}
 		return 0;
@@ -681,7 +726,8 @@ int tdb_unlock_hashes(struct tdb_context *tdb,
  */
 static tdb_off_t free_lock_off(tdb_off_t b_off)
 {
-	return TDB_HASH_LOCK_START + TDB_HASH_LOCK_RANGE + b_off / sizeof(tdb_off_t);
+	return TDB_HASH_LOCK_START + TDB_HASH_LOCK_RANGE
+		+ b_off / sizeof(tdb_off_t);
 }
 
 int tdb_lock_free_bucket(struct tdb_context *tdb, tdb_off_t b_off,
@@ -689,24 +735,25 @@ int tdb_lock_free_bucket(struct tdb_context *tdb, tdb_off_t b_off,
 {
 	assert(b_off >= sizeof(struct tdb_header));
 
-	/* You're supposed to have a hash lock first! */
-	if (!(tdb->flags & TDB_NOLOCK) && !tdb_has_locks(tdb)) {
-		tdb->ecode = TDB_ERR_LOCK;
-		tdb->log(tdb, TDB_DEBUG_FATAL, tdb->log_priv,
-			 "tdb_lock_free_list without lock!\n");
-		return -1;
-	}
-
 	/* a allrecord lock allows us to avoid per chain locks */
 	if (tdb->allrecord_lock.count) {
 		if (tdb->allrecord_lock.ltype == F_WRLCK)
 			return 0;
 		tdb->ecode = TDB_ERR_LOCK;
 		tdb->log(tdb, TDB_DEBUG_FATAL, tdb->log_priv,
-			 "tdb_lock_free_list with RO allrecordlock!\n");
+			 "tdb_lock_free_bucket with RO allrecordlock!\n");
 		return -1;
 	}
 
+#if 0 /* FIXME */
+	if (tdb_has_expansion_lock(tdb)) {
+		tdb->ecode = TDB_ERR_LOCK;
+		tdb->log(tdb, TDB_DEBUG_ERROR, tdb->log_priv,
+			 "tdb_lock_free_bucket: have expansion lock already\n");
+		return -1;
+	}
+#endif
+
 	return tdb_nest_lock(tdb, free_lock_off(b_off), F_WRLCK, waitflag);
 }
 
@@ -718,34 +765,6 @@ void tdb_unlock_free_bucket(struct tdb_context *tdb, tdb_off_t b_off)
 	tdb_nest_unlock(tdb, free_lock_off(b_off), F_WRLCK);
 }
 
-/* Even if the entry isn't in this hash bucket, you'd have to lock this
- * bucket to find it. */
-static int chainlock(struct tdb_context *tdb, const TDB_DATA *key,
-		     int ltype, enum tdb_lock_flags waitflag,
-		     const char *func)
-{
-	int ret;
-	uint64_t h = tdb_hash(tdb, key->dptr, key->dsize);
-
-	ret = tdb_lock_hashes(tdb, h, 1, ltype, waitflag);
-	tdb_trace_1rec(tdb, func, *key);
-	return ret;
-}
-
-/* lock/unlock one hash chain. This is meant to be used to reduce
-   contention - it cannot guarantee how many records will be locked */
-int tdb_chainlock(struct tdb_context *tdb, TDB_DATA key)
-{
-	return chainlock(tdb, &key, F_WRLCK, TDB_LOCK_WAIT, "tdb_chainlock");
-}
-
-int tdb_chainunlock(struct tdb_context *tdb, TDB_DATA key)
-{
-	uint64_t h = tdb_hash(tdb, key.dptr, key.dsize);
-	tdb_trace_1rec(tdb, "tdb_chainunlock", key);
-	return tdb_unlock_hashes(tdb, h, 1, F_WRLCK);
-}
-
 #if 0
 /* lock/unlock one hash chain, non-blocking. This is meant to be used
    to reduce contention - it cannot guarantee how many records will be