HBASE-13082 Coarsen StoreScanner locks to RegionScanner.
This commit is contained in:
parent
6d4a4a48fe
commit
ec1eff9b69
|
@ -5233,6 +5233,8 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver { //
|
||||||
scan.getFamilyMap().entrySet()) {
|
scan.getFamilyMap().entrySet()) {
|
||||||
Store store = stores.get(entry.getKey());
|
Store store = stores.get(entry.getKey());
|
||||||
KeyValueScanner scanner = store.getScanner(scan, entry.getValue(), this.readPt);
|
KeyValueScanner scanner = store.getScanner(scan, entry.getValue(), this.readPt);
|
||||||
|
// pass the RegionScanner object to lock out concurrent changes to set of readers
|
||||||
|
scanner.setReaderLock(this);
|
||||||
if (this.filter == null || !scan.doLoadColumnFamiliesOnDemand()
|
if (this.filter == null || !scan.doLoadColumnFamiliesOnDemand()
|
||||||
|| this.filter.isFamilyEssential(entry.getKey())) {
|
|| this.filter.isFamilyEssential(entry.getKey())) {
|
||||||
scanners.add(scanner);
|
scanners.add(scanner);
|
||||||
|
|
|
@ -162,4 +162,11 @@ public interface KeyValueScanner {
|
||||||
* if known, or null otherwise
|
* if known, or null otherwise
|
||||||
*/
|
*/
|
||||||
public Cell getNextIndexedKey();
|
public Cell getNextIndexedKey();
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Set the object to lock when the scanner's readers (if any) are updated (this is here so that the
|
||||||
|
* coprocessors creating {@link StoreScanner}'s do not have to change)
|
||||||
|
* @param obj lock object to use
|
||||||
|
*/
|
||||||
|
public void setReaderLock(Object obj);
|
||||||
}
|
}
|
||||||
|
|
|
@ -71,4 +71,7 @@ public abstract class NonLazyKeyValueScanner implements KeyValueScanner {
|
||||||
public Cell getNextIndexedKey() {
|
public Cell getNextIndexedKey() {
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void setReaderLock(Object obj) {/* NO OP */}
|
||||||
}
|
}
|
||||||
|
|
|
@ -124,24 +124,13 @@ class ReversedStoreScanner extends StoreScanner implements KeyValueScanner {
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public boolean seekToPreviousRow(Cell key) throws IOException {
|
public boolean seekToPreviousRow(Cell key) throws IOException {
|
||||||
lock.lock();
|
checkReseek();
|
||||||
try {
|
return this.heap.seekToPreviousRow(key);
|
||||||
checkReseek();
|
|
||||||
return this.heap.seekToPreviousRow(key);
|
|
||||||
} finally {
|
|
||||||
lock.unlock();
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public boolean backwardSeek(Cell key) throws IOException {
|
public boolean backwardSeek(Cell key) throws IOException {
|
||||||
lock.lock();
|
checkReseek();
|
||||||
try {
|
return this.heap.backwardSeek(key);
|
||||||
checkReseek();
|
|
||||||
return this.heap.backwardSeek(key);
|
|
||||||
} finally {
|
|
||||||
lock.unlock();
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -489,4 +489,7 @@ public class StoreFileScanner implements KeyValueScanner {
|
||||||
public Cell getNextIndexedKey() {
|
public Cell getNextIndexedKey() {
|
||||||
return hfs.getNextIndexedKey();
|
return hfs.getNextIndexedKey();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void setReaderLock(Object obj) {/* NO OP */}
|
||||||
}
|
}
|
||||||
|
|
|
@ -25,7 +25,6 @@ import java.util.ArrayList;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.NavigableSet;
|
import java.util.NavigableSet;
|
||||||
import java.util.concurrent.CountDownLatch;
|
import java.util.concurrent.CountDownLatch;
|
||||||
import java.util.concurrent.locks.ReentrantLock;
|
|
||||||
|
|
||||||
import org.apache.commons.logging.Log;
|
import org.apache.commons.logging.Log;
|
||||||
import org.apache.commons.logging.LogFactory;
|
import org.apache.commons.logging.LogFactory;
|
||||||
|
@ -103,10 +102,17 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
|
||||||
|
|
||||||
// A flag whether use pread for scan
|
// A flag whether use pread for scan
|
||||||
private boolean scanUsePread = false;
|
private boolean scanUsePread = false;
|
||||||
protected ReentrantLock lock = new ReentrantLock();
|
|
||||||
|
|
||||||
private final long readPt;
|
private final long readPt;
|
||||||
|
|
||||||
|
// lock to use for updateReaders
|
||||||
|
// creator needs to ensure that:
|
||||||
|
// 1. *all* calls to public methods (except updateReaders and close) are locked with this lock
|
||||||
|
// (this can be done by passing the RegionScannerImpl object down)
|
||||||
|
// OR
|
||||||
|
// 2. updateReader is *never* called (such as in flushes or compactions)
|
||||||
|
private Object readerLock;
|
||||||
|
|
||||||
// used by the injection framework to test race between StoreScanner construction and compaction
|
// used by the injection framework to test race between StoreScanner construction and compaction
|
||||||
enum StoreScannerCompactionRace {
|
enum StoreScannerCompactionRace {
|
||||||
BEFORE_SEEK,
|
BEFORE_SEEK,
|
||||||
|
@ -395,15 +401,10 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public Cell peek() {
|
public Cell peek() {
|
||||||
lock.lock();
|
|
||||||
try {
|
|
||||||
if (this.heap == null) {
|
if (this.heap == null) {
|
||||||
return this.lastTop;
|
return this.lastTop;
|
||||||
}
|
}
|
||||||
return this.heap.peek();
|
return this.heap.peek();
|
||||||
} finally {
|
|
||||||
lock.unlock();
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -414,8 +415,6 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void close() {
|
public void close() {
|
||||||
lock.lock();
|
|
||||||
try {
|
|
||||||
if (this.closing) return;
|
if (this.closing) return;
|
||||||
this.closing = true;
|
this.closing = true;
|
||||||
// under test, we dont have a this.store
|
// under test, we dont have a this.store
|
||||||
|
@ -425,21 +424,13 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
|
||||||
this.heap.close();
|
this.heap.close();
|
||||||
this.heap = null; // CLOSED!
|
this.heap = null; // CLOSED!
|
||||||
this.lastTop = null; // If both are null, we are closed.
|
this.lastTop = null; // If both are null, we are closed.
|
||||||
} finally {
|
|
||||||
lock.unlock();
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public boolean seek(Cell key) throws IOException {
|
public boolean seek(Cell key) throws IOException {
|
||||||
lock.lock();
|
|
||||||
try {
|
|
||||||
// reset matcher state, in case that underlying store changed
|
// reset matcher state, in case that underlying store changed
|
||||||
checkReseek();
|
checkReseek();
|
||||||
return this.heap.seek(key);
|
return this.heap.seek(key);
|
||||||
} finally {
|
|
||||||
lock.unlock();
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
@ -464,8 +455,6 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
|
||||||
@Override
|
@Override
|
||||||
public NextState next(List<Cell> outResult, int limit, long remainingResultSize)
|
public NextState next(List<Cell> outResult, int limit, long remainingResultSize)
|
||||||
throws IOException {
|
throws IOException {
|
||||||
lock.lock();
|
|
||||||
try {
|
|
||||||
if (checkReseek()) {
|
if (checkReseek()) {
|
||||||
return NextState.makeState(NextState.State.MORE_VALUES, 0);
|
return NextState.makeState(NextState.State.MORE_VALUES, 0);
|
||||||
}
|
}
|
||||||
|
@ -617,9 +606,6 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
|
||||||
// No more keys
|
// No more keys
|
||||||
close();
|
close();
|
||||||
return NextState.makeState(NextState.State.NO_MORE_VALUES, totalHeapSize);
|
return NextState.makeState(NextState.State.NO_MORE_VALUES, totalHeapSize);
|
||||||
} finally {
|
|
||||||
lock.unlock();
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/*
|
/*
|
||||||
|
@ -663,8 +649,7 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
|
||||||
// Implementation of ChangedReadersObserver
|
// Implementation of ChangedReadersObserver
|
||||||
@Override
|
@Override
|
||||||
public void updateReaders() throws IOException {
|
public void updateReaders() throws IOException {
|
||||||
lock.lock();
|
synchronized(readerLock) {
|
||||||
try {
|
|
||||||
if (this.closing) return;
|
if (this.closing) return;
|
||||||
|
|
||||||
// All public synchronized API calls will call 'checkReseek' which will cause
|
// All public synchronized API calls will call 'checkReseek' which will cause
|
||||||
|
@ -684,8 +669,6 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
|
||||||
this.heap = null; // the re-seeks could be slow (access HDFS) free up memory ASAP
|
this.heap = null; // the re-seeks could be slow (access HDFS) free up memory ASAP
|
||||||
|
|
||||||
// Let the next() call handle re-creating and seeking
|
// Let the next() call handle re-creating and seeking
|
||||||
} finally {
|
|
||||||
lock.unlock();
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -776,8 +759,6 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public boolean reseek(Cell kv) throws IOException {
|
public boolean reseek(Cell kv) throws IOException {
|
||||||
lock.lock();
|
|
||||||
try {
|
|
||||||
//Heap will not be null, if this is called from next() which.
|
//Heap will not be null, if this is called from next() which.
|
||||||
//If called from RegionScanner.reseek(...) make sure the scanner
|
//If called from RegionScanner.reseek(...) make sure the scanner
|
||||||
//stack is reset if needed.
|
//stack is reset if needed.
|
||||||
|
@ -786,9 +767,6 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
|
||||||
return heap.requestSeek(kv, true, useRowColBloom);
|
return heap.requestSeek(kv, true, useRowColBloom);
|
||||||
}
|
}
|
||||||
return heap.reseek(kv);
|
return heap.reseek(kv);
|
||||||
} finally {
|
|
||||||
lock.unlock();
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -863,5 +841,9 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
|
||||||
public Cell getNextIndexedKey() {
|
public Cell getNextIndexedKey() {
|
||||||
return this.heap.getNextIndexedKey();
|
return this.heap.getNextIndexedKey();
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void setReaderLock(Object obj) {
|
||||||
|
this.readerLock = obj;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
Loading…
Reference in New Issue