in hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCache.java [238:434]
private ICachedPage findPageInner(long dpid) {
CachedPage cPage;
/*
* Hash dpid to get a bucket and then check if the page exists in
* the bucket.
*/
int hash = hash(dpid);
CacheBucket bucket = pageMap[hash];
bucket.bucketLock.lock();
try {
cPage = bucket.cachedPage;
while (cPage != null) {
if (DEBUG) {
assert bucket.cachedPage != bucket.cachedPage.next;
}
if (cPage.dpid == dpid) {
if (DEBUG) {
assert !cPage.confiscated.get();
}
cPage.incrementAndGetPinCount();
return cPage;
}
cPage = cPage.next;
}
} finally {
bucket.bucketLock.unlock();
}
/*
* If we got here, the page was not in the hash table. Now we ask
* the page replacement strategy to find us a victim.
*/
CachedPage victim = (CachedPage) pageReplacementStrategy.findVictim();
if (victim == null) {
return null;
}
/*
* We have a victim with the following invariants. 1. The dpid
* on the CachedPage may or may not be valid. 2. We have a pin
* on the CachedPage. We have to deal with three cases here.
* Case 1: The dpid on the CachedPage is invalid (-1). This
* indicates that this buffer has never been used or is a
* confiscated page. So we are the only ones holding it. Get a lock
* on the required dpid's hash bucket, check if someone inserted
* the page we want into the table. If so, decrement the
* pincount on the victim and return the winner page in the
* table. If such a winner does not exist, insert the victim and
* return it. Case 2: The dpid on the CachedPage is valid. Case
* 2a: The current dpid and required dpid hash to the same
* bucket. Get the bucket lock, check that the victim is still
* at pinCount == 1 If so check if there is a winning CachedPage
* with the required dpid. If so, decrement the pinCount on the
* victim and return the winner. If not, update the contents of
* the CachedPage to hold the required dpid and return it. If
* the picCount on the victim was != 1 or CachedPage was dirty
* someone used the victim for its old contents -- Decrement the
* pinCount and retry. Case 2b: The current dpid and required
* dpid hash to different buckets. Get the two bucket locks in
* the order of the bucket indexes (Ordering prevents
* deadlocks). Check for the existence of a winner in the new
* bucket and for potential use of the victim (pinCount != 1).
* If everything looks good, remove the CachedPage from the old
* bucket, and add it to the new bucket and update its header
* with the new dpid.
*/
if (victim.dpid < 0) {
/*
* Case 1.
*/
bucket.bucketLock.lock();
try {
if (!victim.pinCount.compareAndSet(0, 1)) {
return null;
}
// now that we have the pin, ensure the victim's dpid still is < 0, if it's not, decrement
// pin count and try again
if (victim.dpid >= 0) {
victim.decrementAndGetPinCount();
return null;
}
if (DEBUG) {
confiscateLock.lock();
try {
if (confiscatedPages.contains(victim)) {
throw new IllegalStateException();
}
} finally {
confiscateLock.unlock();
}
}
cPage = findTargetInBucket(dpid, bucket.cachedPage, victim);
if (cPage != null) {
return cPage;
}
victim.reset(dpid);
victim.next = bucket.cachedPage;
bucket.cachedPage = victim;
} finally {
bucket.bucketLock.unlock();
}
if (DEBUG) {
assert !victim.confiscated.get();
}
return victim;
}
int victimHash = hash(victim.dpid);
if (victimHash == hash) {
/*
* Case 2a.
*/
bucket.bucketLock.lock();
try {
if (!victim.pinCount.compareAndSet(0, 1)) {
return null;
}
// now that we have the pin, ensure the victim's bucket hasn't changed, if it has, decrement
// pin count and try again
if (victimHash != hash(victim.dpid)) {
victim.decrementAndGetPinCount();
return null;
}
if (DEBUG) {
confiscateLock.lock();
try {
if (confiscatedPages.contains(victim)) {
throw new IllegalStateException();
}
} finally {
confiscateLock.unlock();
}
}
cPage = findTargetInBucket(dpid, bucket.cachedPage, victim);
if (cPage != null) {
return cPage;
}
victim.reset(dpid);
} finally {
bucket.bucketLock.unlock();
}
if (DEBUG) {
assert !victim.confiscated.get();
}
return victim;
} else {
/*
* Case 2b.
*/
CacheBucket victimBucket = pageMap[victimHash];
if (victimHash < hash) {
victimBucket.bucketLock.lock();
bucket.bucketLock.lock();
} else {
bucket.bucketLock.lock();
victimBucket.bucketLock.lock();
}
try {
if (!victim.pinCount.compareAndSet(0, 1)) {
return null;
}
// now that we have the pin, ensure the victim's bucket hasn't changed, if it has, decrement
// pin count and try again
if (victimHash != hash(victim.dpid)) {
victim.decrementAndGetPinCount();
return null;
}
if (DEBUG && confiscatedPages.contains(victim)) {
throw new IllegalStateException();
}
cPage = findTargetInBucket(dpid, bucket.cachedPage, victim);
if (cPage != null) {
return cPage;
}
if (victimBucket.cachedPage == victim) {
victimBucket.cachedPage = victim.next;
} else {
CachedPage victimPrev = victimBucket.cachedPage;
while (victimPrev.next != victim) {
victimPrev = victimPrev.next;
if (victimPrev == null) {
throw new IllegalStateException();
}
}
victimPrev.next = victim.next;
}
victim.reset(dpid);
victim.next = bucket.cachedPage;
bucket.cachedPage = victim;
} finally {
victimBucket.bucketLock.unlock();
bucket.bucketLock.unlock();
}
if (DEBUG) {
assert !victim.confiscated.get();
}
return victim;
}
}