Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-3587 Eliminate use of read-write lock to guard loaded coprocessor collection 

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1089685 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Gary Helmling 2011-04-06 23:59:34 +00:00
parent 1596f98835
commit 9e6f97fd61
7 changed files with 722 additions and 773 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGES.txt
View File

@ -139,6 +139,8 @@ Release 0.91.0 - Unreleased
(Doug Meil via Stack) (Doug Meil via Stack)
HBASE-3738 Book.xml - expanding Architecture Client section HBASE-3738 Book.xml - expanding Architecture Client section
(Doug Meil via Stack) (Doug Meil via Stack)
HBASE-3587 Eliminate use of read-write lock to guard loaded
coprocessor collection
TASK TASK
HBASE-3559 Move report of split to master OFF the heartbeat channel HBASE-3559 Move report of split to master OFF the heartbeat channel

47
src/main/java/org/apache/hadoop/hbase/coprocessor/CoprocessorHost.java
View File

@ -31,6 +31,7 @@ import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.coprocessor.Batch; import org.apache.hadoop.hbase.client.coprocessor.Batch;
import org.apache.hadoop.hbase.ipc.CoprocessorProtocol; import org.apache.hadoop.hbase.ipc.CoprocessorProtocol;
import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.SortedCopyOnWriteSet;
import org.apache.hadoop.hbase.util.VersionInfo; import org.apache.hadoop.hbase.util.VersionInfo;
import java.io.File; import java.io.File;
@ -38,7 +39,6 @@ import java.io.IOException;
import java.net.URL; import java.net.URL;
import java.net.URLClassLoader; import java.net.URLClassLoader;
import java.util.*; import java.util.*;
import java.util.concurrent.locks.ReentrantReadWriteLock;
/** /**
* Provides the common setup framework and runtime services for coprocessor * Provides the common setup framework and runtime services for coprocessor
@ -56,9 +56,9 @@ public abstract class CoprocessorHost<E extends CoprocessorEnvironment> {
private static final Log LOG = LogFactory.getLog(CoprocessorHost.class); private static final Log LOG = LogFactory.getLog(CoprocessorHost.class);
/** Ordered set of loaded coprocessors with lock */ /** Ordered set of loaded coprocessors with lock */
protected final ReentrantReadWriteLock coprocessorLock = new ReentrantReadWriteLock(); protected SortedSet<E> coprocessors =
protected Set<E> coprocessors = new SortedCopyOnWriteSet<E>(new EnvironmentPriorityComparator());
new TreeSet<E>(new EnvironmentPriorityComparator()); protected Configuration conf;
// unique file prefix to use for local copies of jars when classloading // unique file prefix to use for local copies of jars when classloading
protected String pathPrefix; protected String pathPrefix;
@ -79,6 +79,7 @@ public abstract class CoprocessorHost<E extends CoprocessorEnvironment> {
return; return;
StringTokenizer st = new StringTokenizer(defaultCPClasses, ","); StringTokenizer st = new StringTokenizer(defaultCPClasses, ",");
int priority = Coprocessor.Priority.SYSTEM.intValue(); int priority = Coprocessor.Priority.SYSTEM.intValue();
List<E> configured = new ArrayList<E>();
while (st.hasMoreTokens()) { while (st.hasMoreTokens()) {
String className = st.nextToken(); String className = st.nextToken();
if (findCoprocessor(className) != null) { if (findCoprocessor(className) != null) {
@ -88,7 +89,7 @@ public abstract class CoprocessorHost<E extends CoprocessorEnvironment> {
Thread.currentThread().setContextClassLoader(cl); Thread.currentThread().setContextClassLoader(cl);
try { try {
implClass = cl.loadClass(className); implClass = cl.loadClass(className);
load(implClass, Coprocessor.Priority.SYSTEM); configured.add(loadInstance(implClass, Coprocessor.Priority.SYSTEM));
LOG.info("System coprocessor " + className + " was loaded " + LOG.info("System coprocessor " + className + " was loaded " +
"successfully with priority (" + priority++ + ")."); "successfully with priority (" + priority++ + ").");
} catch (ClassNotFoundException e) { } catch (ClassNotFoundException e) {
@ -99,6 +100,9 @@ public abstract class CoprocessorHost<E extends CoprocessorEnvironment> {
e.getMessage()); e.getMessage());
} }
} }
// add entire set to the collection for COW efficiency
coprocessors.addAll(configured);
} }
/** /**
@ -109,7 +113,7 @@ public abstract class CoprocessorHost<E extends CoprocessorEnvironment> {
* @throws java.io.IOException Exception * @throws java.io.IOException Exception
*/ */
@SuppressWarnings("deprecation") @SuppressWarnings("deprecation")
public void load(Path path, String className, Coprocessor.Priority priority) public E load(Path path, String className, Coprocessor.Priority priority)
throws IOException { throws IOException {
Class<?> implClass = null; Class<?> implClass = null;
@ -163,7 +167,7 @@ public abstract class CoprocessorHost<E extends CoprocessorEnvironment> {
} }
} }
load(implClass, priority); return loadInstance(implClass, priority);
} }
/** /**
@ -173,6 +177,12 @@ public abstract class CoprocessorHost<E extends CoprocessorEnvironment> {
*/ */
public void load(Class<?> implClass, Coprocessor.Priority priority) public void load(Class<?> implClass, Coprocessor.Priority priority)
throws IOException { throws IOException {
E env = loadInstance(implClass, priority);
coprocessors.add(env);
}
public E loadInstance(Class<?> implClass, Coprocessor.Priority priority)
throws IOException {
// create the instance // create the instance
Coprocessor impl; Coprocessor impl;
Object o = null; Object o = null;
@ -189,13 +199,7 @@ public abstract class CoprocessorHost<E extends CoprocessorEnvironment> {
if (env instanceof Environment) { if (env instanceof Environment) {
((Environment)env).startup(); ((Environment)env).startup();
} }
return env;
try {
coprocessorLock.writeLock().lock();
coprocessors.add(env);
} finally {
coprocessorLock.writeLock().unlock();
}
} }
/** /**
@ -220,18 +224,13 @@ public abstract class CoprocessorHost<E extends CoprocessorEnvironment> {
*/ */
public Coprocessor findCoprocessor(String className) { public Coprocessor findCoprocessor(String className) {
// initialize the coprocessors // initialize the coprocessors
try { for (E env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance().getClass().getName().equals(className) ||
for (E env: coprocessors) { env.getInstance().getClass().getSimpleName().equals(className)) {
if (env.getInstance().getClass().getName().equals(className) || return env.getInstance();
env.getInstance().getClass().getSimpleName().equals(className)) {
return env.getInstance();
}
} }
return null;
} finally {
coprocessorLock.readLock().unlock();
} }
return null;
} }
/** /**

446
src/main/java/org/apache/hadoop/hbase/master/MasterCoprocessorHost.java
View File

@ -70,483 +70,343 @@ public class MasterCoprocessorHost
/* Implementation of hooks for invoking MasterObservers */ /* Implementation of hooks for invoking MasterObservers */
void preCreateTable(HTableDescriptor desc, byte[][] splitKeys) void preCreateTable(HTableDescriptor desc, byte[][] splitKeys)
throws IOException { throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).preCreateTable(env, desc, splitKeys);
if (env.getInstance() instanceof MasterObserver) { if (env.shouldComplete()) {
((MasterObserver)env.getInstance()).preCreateTable(env, desc, splitKeys); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
void postCreateTable(HRegionInfo[] regions, boolean sync) throws IOException { void postCreateTable(HRegionInfo[] regions, boolean sync) throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).postCreateTable(env, regions, sync);
if (env.getInstance() instanceof MasterObserver) { if (env.shouldComplete()) {
((MasterObserver)env.getInstance()).postCreateTable(env, regions, sync); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
void preDeleteTable(byte[] tableName) throws IOException { void preDeleteTable(byte[] tableName) throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).preDeleteTable(env, tableName);
if (env.getInstance() instanceof MasterObserver) { if (env.shouldComplete()) {
((MasterObserver)env.getInstance()).preDeleteTable(env, tableName); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
void postDeleteTable(byte[] tableName) throws IOException { void postDeleteTable(byte[] tableName) throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).postDeleteTable(env, tableName);
if (env.getInstance() instanceof MasterObserver) { if (env.shouldComplete()) {
((MasterObserver)env.getInstance()).postDeleteTable(env, tableName); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
void preModifyTable(final byte[] tableName, HTableDescriptor htd) void preModifyTable(final byte[] tableName, HTableDescriptor htd)
throws IOException { throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).preModifyTable(env, tableName, htd);
if (env.getInstance() instanceof MasterObserver) { if (env.shouldComplete()) {
((MasterObserver)env.getInstance()).preModifyTable(env, tableName, htd); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
void postModifyTable(final byte[] tableName, HTableDescriptor htd) void postModifyTable(final byte[] tableName, HTableDescriptor htd)
throws IOException { throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).postModifyTable(env, tableName, htd);
if (env.getInstance() instanceof MasterObserver) { if (env.shouldComplete()) {
((MasterObserver)env.getInstance()).postModifyTable(env, tableName, htd); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
void preAddColumn(byte [] tableName, HColumnDescriptor column) void preAddColumn(byte [] tableName, HColumnDescriptor column)
throws IOException { throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).preAddColumn(env, tableName, column);
if (env.getInstance() instanceof MasterObserver) { if (env.shouldComplete()) {
((MasterObserver)env.getInstance()).preAddColumn(env, tableName, column); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
void postAddColumn(byte [] tableName, HColumnDescriptor column) void postAddColumn(byte [] tableName, HColumnDescriptor column)
throws IOException { throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).postAddColumn(env, tableName, column);
if (env.getInstance() instanceof MasterObserver) { if (env.shouldComplete()) {
((MasterObserver)env.getInstance()).postAddColumn(env, tableName, column); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
void preModifyColumn(byte [] tableName, HColumnDescriptor descriptor) void preModifyColumn(byte [] tableName, HColumnDescriptor descriptor)
throws IOException { throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).preModifyColumn(
if (env.getInstance() instanceof MasterObserver) { env, tableName, descriptor);
((MasterObserver)env.getInstance()).preModifyColumn( if (env.shouldComplete()) {
env, tableName, descriptor); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
void postModifyColumn(byte [] tableName, HColumnDescriptor descriptor) void postModifyColumn(byte [] tableName, HColumnDescriptor descriptor)
throws IOException { throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).postModifyColumn(
if (env.getInstance() instanceof MasterObserver) { env, tableName, descriptor);
((MasterObserver)env.getInstance()).postModifyColumn( if (env.shouldComplete()) {
env, tableName, descriptor); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
void preDeleteColumn(final byte [] tableName, final byte [] c) void preDeleteColumn(final byte [] tableName, final byte [] c)
throws IOException { throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).preDeleteColumn(env, tableName, c);
if (env.getInstance() instanceof MasterObserver) { if (env.shouldComplete()) {
((MasterObserver)env.getInstance()).preDeleteColumn(env, tableName, c); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
void postDeleteColumn(final byte [] tableName, final byte [] c) void postDeleteColumn(final byte [] tableName, final byte [] c)
throws IOException { throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).postDeleteColumn(env, tableName, c);
if (env.getInstance() instanceof MasterObserver) { if (env.shouldComplete()) {
((MasterObserver)env.getInstance()).postDeleteColumn(env, tableName, c); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
void preEnableTable(final byte [] tableName) throws IOException { void preEnableTable(final byte [] tableName) throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).preEnableTable(env, tableName);
if (env.getInstance() instanceof MasterObserver) { if (env.shouldComplete()) {
((MasterObserver)env.getInstance()).preEnableTable(env, tableName); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
void postEnableTable(final byte [] tableName) throws IOException { void postEnableTable(final byte [] tableName) throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).postEnableTable(env, tableName);
if (env.getInstance() instanceof MasterObserver) { if (env.shouldComplete()) {
((MasterObserver)env.getInstance()).postEnableTable(env, tableName); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
void preDisableTable(final byte [] tableName) throws IOException { void preDisableTable(final byte [] tableName) throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).preDisableTable(env, tableName);
if (env.getInstance() instanceof MasterObserver) { if (env.shouldComplete()) {
((MasterObserver)env.getInstance()).preDisableTable(env, tableName); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
void postDisableTable(final byte [] tableName) throws IOException { void postDisableTable(final byte [] tableName) throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).postDisableTable(env, tableName);
if (env.getInstance() instanceof MasterObserver) { if (env.shouldComplete()) {
((MasterObserver)env.getInstance()).postDisableTable(env, tableName); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
void preMove(final HRegionInfo region, final HServerInfo srcServer, final HServerInfo destServer) void preMove(final HRegionInfo region, final HServerInfo srcServer, final HServerInfo destServer)
throws UnknownRegionException { throws UnknownRegionException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).preMove(
if (env.getInstance() instanceof MasterObserver) { env, region, srcServer, destServer);
((MasterObserver)env.getInstance()).preMove( if (env.shouldComplete()) {
env, region, srcServer, destServer); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
void postMove(final HRegionInfo region, final HServerInfo srcServer, final HServerInfo destServer) void postMove(final HRegionInfo region, final HServerInfo srcServer, final HServerInfo destServer)
throws UnknownRegionException { throws UnknownRegionException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).postMove(
if (env.getInstance() instanceof MasterObserver) { env, region, srcServer, destServer);
((MasterObserver)env.getInstance()).postMove( if (env.shouldComplete()) {
env, region, srcServer, destServer); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
boolean preAssign(final byte [] regionName, final boolean force) boolean preAssign(final byte [] regionName, final boolean force)
throws IOException { throws IOException {
boolean bypass = false; boolean bypass = false;
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).preAssign(env, regionName, force);
if (env.getInstance() instanceof MasterObserver) { bypass |= env.shouldBypass();
((MasterObserver)env.getInstance()).preAssign(env, regionName, force); if (env.shouldComplete()) {
bypass |= env.shouldBypass(); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
return bypass; return bypass;
} }
void postAssign(final HRegionInfo regionInfo) throws IOException { void postAssign(final HRegionInfo regionInfo) throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).postAssign(env, regionInfo);
if (env.getInstance() instanceof MasterObserver) { if (env.shouldComplete()) {
((MasterObserver)env.getInstance()).postAssign(env, regionInfo); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
boolean preUnassign(final byte [] regionName, final boolean force) boolean preUnassign(final byte [] regionName, final boolean force)
throws IOException { throws IOException {
boolean bypass = false; boolean bypass = false;
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).preUnassign(
if (env.getInstance() instanceof MasterObserver) { env, regionName, force);
((MasterObserver)env.getInstance()).preUnassign( bypass |= env.shouldBypass();
env, regionName, force); if (env.shouldComplete()) {
bypass |= env.shouldBypass(); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
return bypass; return bypass;
} }
void postUnassign(final HRegionInfo regionInfo, final boolean force) void postUnassign(final HRegionInfo regionInfo, final boolean force)
throws IOException { throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).postUnassign(
if (env.getInstance() instanceof MasterObserver) { env, regionInfo, force);
((MasterObserver)env.getInstance()).postUnassign( if (env.shouldComplete()) {
env, regionInfo, force); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
boolean preBalance() throws IOException { boolean preBalance() throws IOException {
try { boolean bypass = false;
boolean bypass = false; for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).preBalance(env);
if (env.getInstance() instanceof MasterObserver) { bypass |= env.shouldBypass();
((MasterObserver)env.getInstance()).preBalance(env); if (env.shouldComplete()) {
bypass |= env.shouldBypass(); break;
if (env.shouldComplete()) {
break;
}
} }
} }
return bypass;
} finally {
coprocessorLock.readLock().unlock();
} }
return bypass;
} }
void postBalance() throws IOException { void postBalance() throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).postBalance(env);
if (env.getInstance() instanceof MasterObserver) { if (env.shouldComplete()) {
((MasterObserver)env.getInstance()).postBalance(env); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
boolean preBalanceSwitch(final boolean b) throws IOException { boolean preBalanceSwitch(final boolean b) throws IOException {
boolean balance = b; boolean balance = b;
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { balance = ((MasterObserver)env.getInstance()).preBalanceSwitch(
if (env.getInstance() instanceof MasterObserver) { env, balance);
balance = ((MasterObserver)env.getInstance()).preBalanceSwitch( if (env.shouldComplete()) {
env, balance); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
return balance; return balance;
} }
void postBalanceSwitch(final boolean oldValue, final boolean newValue) void postBalanceSwitch(final boolean oldValue, final boolean newValue)
throws IOException { throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).postBalanceSwitch(
if (env.getInstance() instanceof MasterObserver) { env, oldValue, newValue);
((MasterObserver)env.getInstance()).postBalanceSwitch( if (env.shouldComplete()) {
env, oldValue, newValue); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
void preShutdown() throws IOException { void preShutdown() throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).preShutdown(env);
if (env.getInstance() instanceof MasterObserver) { if (env.shouldComplete()) {
((MasterObserver)env.getInstance()).preShutdown(env); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
void preStopMaster() throws IOException { void preStopMaster() throws IOException {
try { for (MasterEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof MasterObserver) {
for (MasterEnvironment env: coprocessors) { ((MasterObserver)env.getInstance()).preStopMaster(env);
if (env.getInstance() instanceof MasterObserver) { if (env.shouldComplete()) {
((MasterObserver)env.getInstance()).preStopMaster(env); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }

678
src/main/java/org/apache/hadoop/hbase/regionserver/RegionCoprocessorHost.java
View File

@ -120,6 +120,7 @@ public class RegionCoprocessorHost
void loadTableCoprocessors () { void loadTableCoprocessors () {
// scan the table attributes for coprocessor load specifications // scan the table attributes for coprocessor load specifications
// initialize the coprocessors // initialize the coprocessors
List<RegionEnvironment> configured = new ArrayList<RegionEnvironment>();
for (Map.Entry<ImmutableBytesWritable,ImmutableBytesWritable> e: for (Map.Entry<ImmutableBytesWritable,ImmutableBytesWritable> e:
region.getTableDesc().getValues().entrySet()) { region.getTableDesc().getValues().entrySet()) {
String key = Bytes.toString(e.getKey().get()); String key = Bytes.toString(e.getKey().get());
@ -133,7 +134,7 @@ public class RegionCoprocessorHost
String className = matcher.group(2); String className = matcher.group(2);
Coprocessor.Priority priority = Coprocessor.Priority priority =
Coprocessor.Priority.valueOf(matcher.group(3)); Coprocessor.Priority.valueOf(matcher.group(3));
load(path, className, priority); configured.add(load(path, className, priority));
LOG.info("Load coprocessor " + className + " from HTD of " + LOG.info("Load coprocessor " + className + " from HTD of " +
Bytes.toString(region.getTableDesc().getName()) + Bytes.toString(region.getTableDesc().getName()) +
" successfully."); " successfully.");
@ -145,6 +146,8 @@ public class RegionCoprocessorHost
} }
} }
} }
// add together to coprocessor set for COW efficiency
coprocessors.addAll(configured);
} }
@Override @Override
@ -170,18 +173,13 @@ public class RegionCoprocessorHost
*/ */
public void preOpen() { public void preOpen() {
loadTableCoprocessors(); loadTableCoprocessors();
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).preOpen(env);
if (env.getInstance() instanceof RegionObserver) { if (env.shouldComplete()) {
((RegionObserver)env.getInstance()).preOpen(env); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
@ -189,18 +187,13 @@ public class RegionCoprocessorHost
* Invoked after a region open * Invoked after a region open
*/ */
public void postOpen() { public void postOpen() {
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).postOpen(env);
if (env.getInstance() instanceof RegionObserver) { if (env.shouldComplete()) {
((RegionObserver)env.getInstance()).postOpen(env); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
@ -209,15 +202,10 @@ public class RegionCoprocessorHost
* @param abortRequested true if the server is aborting * @param abortRequested true if the server is aborting
*/ */
public void preClose(boolean abortRequested) { public void preClose(boolean abortRequested) {
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.writeLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).preClose(env, abortRequested);
if (env.getInstance() instanceof RegionObserver) {
((RegionObserver)env.getInstance()).preClose(env, abortRequested);
}
} }
} finally {
coprocessorLock.writeLock().unlock();
} }
} }
@ -226,16 +214,11 @@ public class RegionCoprocessorHost
* @param abortRequested true if the server is aborting * @param abortRequested true if the server is aborting
*/ */
public void postClose(boolean abortRequested) { public void postClose(boolean abortRequested) {
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.writeLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).postClose(env, abortRequested);
if (env.getInstance() instanceof RegionObserver) {
((RegionObserver)env.getInstance()).postClose(env, abortRequested);
}
shutdown(env);
} }
} finally { shutdown(env);
coprocessorLock.writeLock().unlock();
} }
} }
@ -244,18 +227,13 @@ public class RegionCoprocessorHost
* @param willSplit true if the compaction is about to trigger a split * @param willSplit true if the compaction is about to trigger a split
*/ */
public void preCompact(boolean willSplit) { public void preCompact(boolean willSplit) {
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).preCompact(env, willSplit);
if (env.getInstance() instanceof RegionObserver) { if (env.shouldComplete()) {
((RegionObserver)env.getInstance()).preCompact(env, willSplit); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
@ -264,18 +242,13 @@ public class RegionCoprocessorHost
* @param willSplit true if the compaction is about to trigger a split * @param willSplit true if the compaction is about to trigger a split
*/ */
public void postCompact(boolean willSplit) { public void postCompact(boolean willSplit) {
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).postCompact(env, willSplit);
if (env.getInstance() instanceof RegionObserver) { if (env.shouldComplete()) {
((RegionObserver)env.getInstance()).postCompact(env, willSplit); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
@ -283,18 +256,13 @@ public class RegionCoprocessorHost
* Invoked before a memstore flush * Invoked before a memstore flush
*/ */
public void preFlush() { public void preFlush() {
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).preFlush(env);
if (env.getInstance() instanceof RegionObserver) { if (env.shouldComplete()) {
((RegionObserver)env.getInstance()).preFlush(env); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
@ -302,18 +270,13 @@ public class RegionCoprocessorHost
* Invoked after a memstore flush * Invoked after a memstore flush
*/ */
public void postFlush() { public void postFlush() {
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).postFlush(env);
if (env.getInstance() instanceof RegionObserver) { if (env.shouldComplete()) {
((RegionObserver)env.getInstance()).postFlush(env); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
@ -321,18 +284,13 @@ public class RegionCoprocessorHost
* Invoked just before a split * Invoked just before a split
*/ */
public void preSplit() { public void preSplit() {
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).preSplit(env);
if (env.getInstance() instanceof RegionObserver) { if (env.shouldComplete()) {
((RegionObserver)env.getInstance()).preSplit(env); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
@ -342,18 +300,13 @@ public class RegionCoprocessorHost
* @param r the new right-hand daughter region * @param r the new right-hand daughter region
*/ */
public void postSplit(HRegion l, HRegion r) { public void postSplit(HRegion l, HRegion r) {
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).postSplit(env, l, r);
if (env.getInstance() instanceof RegionObserver) { if (env.shouldComplete()) {
((RegionObserver)env.getInstance()).postSplit(env, l, r); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
@ -368,23 +321,18 @@ public class RegionCoprocessorHost
*/ */
public boolean preGetClosestRowBefore(final byte[] row, final byte[] family, public boolean preGetClosestRowBefore(final byte[] row, final byte[] family,
final Result result) throws IOException { final Result result) throws IOException {
try { boolean bypass = false;
boolean bypass = false; for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).preGetClosestRowBefore(env, row, family,
if (env.getInstance() instanceof RegionObserver) { result);
((RegionObserver)env.getInstance()).preGetClosestRowBefore(env, row, family, bypass |= env.shouldBypass();
result); if (env.shouldComplete()) {
bypass |= env.shouldBypass(); break;
if (env.shouldComplete()) {
break;
}
} }
} }
return bypass;
} finally {
coprocessorLock.readLock().unlock();
} }
return bypass;
} }
/** /**
@ -395,19 +343,14 @@ public class RegionCoprocessorHost
*/ */
public void postGetClosestRowBefore(final byte[] row, final byte[] family, public void postGetClosestRowBefore(final byte[] row, final byte[] family,
final Result result) throws IOException { final Result result) throws IOException {
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).postGetClosestRowBefore(env, row, family,
if (env.getInstance() instanceof RegionObserver) { result);
((RegionObserver)env.getInstance()).postGetClosestRowBefore(env, row, family, if (env.shouldComplete()) {
result); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
@ -418,22 +361,17 @@ public class RegionCoprocessorHost
*/ */
public boolean preGet(final Get get, final List<KeyValue> results) public boolean preGet(final Get get, final List<KeyValue> results)
throws IOException { throws IOException {
try { boolean bypass = false;
boolean bypass = false; for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).preGet(env, get, results);
if (env.getInstance() instanceof RegionObserver) { bypass |= env.shouldBypass();
((RegionObserver)env.getInstance()).preGet(env, get, results); if (env.shouldComplete()) {
bypass |= env.shouldBypass(); break;
if (env.shouldComplete()) {
break;
}
} }
} }
return bypass;
} finally {
coprocessorLock.readLock().unlock();
} }
return bypass;
} }
/** /**
@ -444,18 +382,13 @@ public class RegionCoprocessorHost
*/ */
public void postGet(final Get get, final List<KeyValue> results) public void postGet(final Get get, final List<KeyValue> results)
throws IOException { throws IOException {
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).postGet(env, get, results);
if (env.getInstance() instanceof RegionObserver) { if (env.shouldComplete()) {
((RegionObserver)env.getInstance()).postGet(env, get, results); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
@ -466,23 +399,18 @@ public class RegionCoprocessorHost
* @exception IOException Exception * @exception IOException Exception
*/ */
public Boolean preExists(final Get get) throws IOException { public Boolean preExists(final Get get) throws IOException {
try { boolean bypass = false;
boolean bypass = false; boolean exists = false;
boolean exists = false; for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { exists = ((RegionObserver)env.getInstance()).preExists(env, get, exists);
if (env.getInstance() instanceof RegionObserver) { bypass |= env.shouldBypass();
exists = ((RegionObserver)env.getInstance()).preExists(env, get, exists); if (env.shouldComplete()) {
bypass |= env.shouldBypass(); break;
if (env.shouldComplete()) {
break;
} }
} }
}
return bypass ? exists : null;
} finally {
coprocessorLock.readLock().unlock();
} }
return bypass ? exists : null;
} }
/** /**
@ -493,20 +421,15 @@ public class RegionCoprocessorHost
*/ */
public boolean postExists(final Get get, boolean exists) public boolean postExists(final Get get, boolean exists)
throws IOException { throws IOException {
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { exists = ((RegionObserver)env.getInstance()).postExists(env, get, exists);
if (env.getInstance() instanceof RegionObserver) { if (env.shouldComplete()) {
exists = ((RegionObserver)env.getInstance()).postExists(env, get, exists); break;
if (env.shouldComplete()) {
break;
}
} }
} }
return exists;
} finally {
coprocessorLock.readLock().unlock();
} }
return exists;
} }
/** /**
@ -517,22 +440,17 @@ public class RegionCoprocessorHost
*/ */
public boolean prePut(final Map<byte[], List<KeyValue>> familyMap, public boolean prePut(final Map<byte[], List<KeyValue>> familyMap,
final boolean writeToWAL) throws IOException { final boolean writeToWAL) throws IOException {
try { boolean bypass = false;
boolean bypass = false; for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).prePut(env, familyMap, writeToWAL);
if (env.getInstance() instanceof RegionObserver) { bypass |= env.shouldBypass();
((RegionObserver)env.getInstance()).prePut(env, familyMap, writeToWAL); if (env.shouldComplete()) {
bypass |= env.shouldBypass(); break;
if (env.shouldComplete()) {
break;
}
} }
} }
return bypass;
} finally {
coprocessorLock.readLock().unlock();
} }
return bypass;
} }
/** /**
@ -542,18 +460,13 @@ public class RegionCoprocessorHost
*/ */
public void postPut(final Map<byte[], List<KeyValue>> familyMap, public void postPut(final Map<byte[], List<KeyValue>> familyMap,
final boolean writeToWAL) throws IOException { final boolean writeToWAL) throws IOException {
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).postPut(env, familyMap, writeToWAL);
if (env.getInstance() instanceof RegionObserver) { if (env.shouldComplete()) {
((RegionObserver)env.getInstance()).postPut(env, familyMap, writeToWAL); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
@ -565,22 +478,17 @@ public class RegionCoprocessorHost
*/ */
public boolean preDelete(final Map<byte[], List<KeyValue>> familyMap, public boolean preDelete(final Map<byte[], List<KeyValue>> familyMap,
final boolean writeToWAL) throws IOException { final boolean writeToWAL) throws IOException {
try { boolean bypass = false;
boolean bypass = false; for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).preDelete(env, familyMap, writeToWAL);
if (env.getInstance() instanceof RegionObserver) { bypass |= env.shouldBypass();
((RegionObserver)env.getInstance()).preDelete(env, familyMap, writeToWAL); if (env.shouldComplete()) {
bypass |= env.shouldBypass(); break;
if (env.shouldComplete()) {
break;
}
} }
} }
return bypass;
} finally {
coprocessorLock.readLock().unlock();
} }
return bypass;
} }
/** /**
@ -590,18 +498,13 @@ public class RegionCoprocessorHost
*/ */
public void postDelete(final Map<byte[], List<KeyValue>> familyMap, public void postDelete(final Map<byte[], List<KeyValue>> familyMap,
final boolean writeToWAL) throws IOException { final boolean writeToWAL) throws IOException {
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).postDelete(env, familyMap, writeToWAL);
if (env.getInstance() instanceof RegionObserver) { if (env.shouldComplete()) {
((RegionObserver)env.getInstance()).postDelete(env, familyMap, writeToWAL); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
@ -619,26 +522,20 @@ public class RegionCoprocessorHost
public Boolean preCheckAndPut(final byte [] row, final byte [] family, public Boolean preCheckAndPut(final byte [] row, final byte [] family,
final byte [] qualifier, final CompareOp compareOp, final byte [] qualifier, final CompareOp compareOp,
final WritableByteArrayComparable comparator, Put put) final WritableByteArrayComparable comparator, Put put)
throws IOException throws IOException {
{ boolean bypass = false;
try { boolean result = false;
boolean bypass = false; for (RegionEnvironment env: coprocessors) {
boolean result = false; if (env.getInstance() instanceof RegionObserver) {
coprocessorLock.readLock().lock(); result = ((RegionObserver)env.getInstance()).preCheckAndPut(env, row, family,
for (RegionEnvironment env: coprocessors) { qualifier, compareOp, comparator, put, result);
if (env.getInstance() instanceof RegionObserver) { bypass |= env.shouldBypass();
result = ((RegionObserver)env.getInstance()).preCheckAndPut(env, row, family, if (env.shouldComplete()) {
qualifier, compareOp, comparator, put, result); break;
bypass |= env.shouldBypass();
if (env.shouldComplete()) {
break;
}
} }
} }
return bypass ? result : null;
} finally {
coprocessorLock.readLock().unlock();
} }
return bypass ? result : null;
} }
/** /**
@ -654,23 +551,17 @@ public class RegionCoprocessorHost
final byte [] qualifier, final CompareOp compareOp, final byte [] qualifier, final CompareOp compareOp,
final WritableByteArrayComparable comparator, final Put put, final WritableByteArrayComparable comparator, final Put put,
boolean result) boolean result)
throws IOException throws IOException {
{ for (RegionEnvironment env: coprocessors) {
try { if (env.getInstance() instanceof RegionObserver) {
coprocessorLock.readLock().lock(); result = ((RegionObserver)env.getInstance()).postCheckAndPut(env, row,
for (RegionEnvironment env: coprocessors) { family, qualifier, compareOp, comparator, put, result);
if (env.getInstance() instanceof RegionObserver) { if (env.shouldComplete()) {
result = ((RegionObserver)env.getInstance()).postCheckAndPut(env, row, break;
family, qualifier, compareOp, comparator, put, result);
if (env.shouldComplete()) {
break;
}
} }
} }
return result;
} finally {
coprocessorLock.readLock().unlock();
} }
return result;
} }
/** /**
@ -688,24 +579,19 @@ public class RegionCoprocessorHost
final byte [] qualifier, final CompareOp compareOp, final byte [] qualifier, final CompareOp compareOp,
final WritableByteArrayComparable comparator, Delete delete) final WritableByteArrayComparable comparator, Delete delete)
throws IOException { throws IOException {
try { boolean bypass = false;
boolean bypass = false; boolean result = false;
boolean result = false; for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { result = ((RegionObserver)env.getInstance()).preCheckAndDelete(env, row,
if (env.getInstance() instanceof RegionObserver) { family, qualifier, compareOp, comparator, delete, result);
result = ((RegionObserver)env.getInstance()).preCheckAndDelete(env, row, bypass |= env.shouldBypass();
family, qualifier, compareOp, comparator, delete, result); if (env.shouldComplete()) {
bypass |= env.shouldBypass(); break;
if (env.shouldComplete()) {
break;
}
} }
} }
return bypass ? result : null;
} finally {
coprocessorLock.readLock().unlock();
} }
return bypass ? result : null;
} }
/** /**
@ -721,24 +607,18 @@ public class RegionCoprocessorHost
final byte [] qualifier, final CompareOp compareOp, final byte [] qualifier, final CompareOp compareOp,
final WritableByteArrayComparable comparator, final Delete delete, final WritableByteArrayComparable comparator, final Delete delete,
boolean result) boolean result)
throws IOException throws IOException {
{ for (RegionEnvironment env: coprocessors) {
try { if (env.getInstance() instanceof RegionObserver) {
coprocessorLock.readLock().lock(); result = ((RegionObserver)env.getInstance())
for (RegionEnvironment env: coprocessors) { .postCheckAndDelete(env, row, family, qualifier, compareOp,
if (env.getInstance() instanceof RegionObserver) { comparator, delete, result);
result = ((RegionObserver)env.getInstance()) if (env.shouldComplete()) {
.postCheckAndDelete(env, row, family, qualifier, compareOp, break;
comparator, delete, result);
if (env.shouldComplete()) {
break;
} }
} }
}
return result;
} finally {
coprocessorLock.readLock().unlock();
} }
return result;
} }
/** /**
@ -754,23 +634,18 @@ public class RegionCoprocessorHost
public Long preIncrementColumnValue(final byte [] row, final byte [] family, public Long preIncrementColumnValue(final byte [] row, final byte [] family,
final byte [] qualifier, long amount, final boolean writeToWAL) final byte [] qualifier, long amount, final boolean writeToWAL)
throws IOException { throws IOException {
try { boolean bypass = false;
boolean bypass = false; for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { amount = ((RegionObserver)env.getInstance()).preIncrementColumnValue(env,
if (env.getInstance() instanceof RegionObserver) { row, family, qualifier, amount, writeToWAL);
amount = ((RegionObserver)env.getInstance()).preIncrementColumnValue(env, bypass |= env.shouldBypass();
row, family, qualifier, amount, writeToWAL); if (env.shouldComplete()) {
bypass |= env.shouldBypass(); break;
if (env.shouldComplete()) {
break;
}
} }
} }
return bypass ? amount : null;
} finally {
coprocessorLock.readLock().unlock();
} }
return bypass ? amount : null;
} }
/** /**
@ -786,19 +661,14 @@ public class RegionCoprocessorHost
public long postIncrementColumnValue(final byte [] row, final byte [] family, public long postIncrementColumnValue(final byte [] row, final byte [] family,
final byte [] qualifier, final long amount, final boolean writeToWAL, final byte [] qualifier, final long amount, final boolean writeToWAL,
long result) throws IOException { long result) throws IOException {
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { result = ((RegionObserver)env.getInstance()).postIncrementColumnValue(env,
if (env.getInstance() instanceof RegionObserver) { row, family, qualifier, amount, writeToWAL, result);
result = ((RegionObserver)env.getInstance()).postIncrementColumnValue(env, if (env.shouldComplete()) {
row, family, qualifier, amount, writeToWAL, result); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
return result; return result;
} }
@ -811,23 +681,18 @@ public class RegionCoprocessorHost
*/ */
public Result preIncrement(Increment increment) public Result preIncrement(Increment increment)
throws IOException { throws IOException {
try { boolean bypass = false;
boolean bypass = false; Result result = new Result();
Result result = new Result(); for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).preIncrement(env, increment, result);
if (env.getInstance() instanceof RegionObserver) { bypass |= env.shouldBypass();
((RegionObserver)env.getInstance()).preIncrement(env, increment, result); if (env.shouldComplete()) {
bypass |= env.shouldBypass(); break;
if (env.shouldComplete()) {
break;
}
} }
} }
return bypass ? result : null;
} finally {
coprocessorLock.readLock().unlock();
} }
return bypass ? result : null;
} }
/** /**
@ -837,18 +702,13 @@ public class RegionCoprocessorHost
*/ */
public void postIncrement(final Increment increment, Result result) public void postIncrement(final Increment increment, Result result)
throws IOException { throws IOException {
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).postIncrement(env, increment, result);
if (env.getInstance() instanceof RegionObserver) { if (env.shouldComplete()) {
((RegionObserver)env.getInstance()).postIncrement(env, increment, result); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
@ -859,23 +719,18 @@ public class RegionCoprocessorHost
* @exception IOException Exception * @exception IOException Exception
*/ */
public InternalScanner preScannerOpen(Scan scan) throws IOException { public InternalScanner preScannerOpen(Scan scan) throws IOException {
try { boolean bypass = false;
boolean bypass = false; InternalScanner s = null;
InternalScanner s = null; for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { s = ((RegionObserver)env.getInstance()).preScannerOpen(env, scan, s);
if (env.getInstance() instanceof RegionObserver) { bypass |= env.shouldBypass();
s = ((RegionObserver)env.getInstance()).preScannerOpen(env, scan, s); if (env.shouldComplete()) {
bypass |= env.shouldBypass(); break;
if (env.shouldComplete()) {
break;
}
} }
} }
return bypass ? s : null;
} finally {
coprocessorLock.readLock().unlock();
} }
return bypass ? s : null;
} }
/** /**
@ -886,20 +741,15 @@ public class RegionCoprocessorHost
*/ */
public InternalScanner postScannerOpen(final Scan scan, InternalScanner s) public InternalScanner postScannerOpen(final Scan scan, InternalScanner s)
throws IOException { throws IOException {
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { s = ((RegionObserver)env.getInstance()).postScannerOpen(env, scan, s);
if (env.getInstance() instanceof RegionObserver) { if (env.shouldComplete()) {
s = ((RegionObserver)env.getInstance()).postScannerOpen(env, scan, s); break;
if (env.shouldComplete()) {
break;
}
} }
} }
return s;
} finally {
coprocessorLock.readLock().unlock();
} }
return s;
} }
/** /**
@ -912,24 +762,19 @@ public class RegionCoprocessorHost
*/ */
public Boolean preScannerNext(final InternalScanner s, public Boolean preScannerNext(final InternalScanner s,
final List<Result> results, int limit) throws IOException { final List<Result> results, int limit) throws IOException {
try { boolean bypass = false;
boolean bypass = false; boolean hasNext = false;
boolean hasNext = false; for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { hasNext = ((RegionObserver)env.getInstance()).preScannerNext(env, s, results,
if (env.getInstance() instanceof RegionObserver) { limit, hasNext);
hasNext = ((RegionObserver)env.getInstance()).preScannerNext(env, s, results, bypass |= env.shouldBypass();
limit, hasNext); if (env.shouldComplete()) {
bypass |= env.shouldBypass(); break;
if (env.shouldComplete()) {
break;
}
} }
} }
return bypass ? hasNext : null;
} finally {
coprocessorLock.readLock().unlock();
} }
return bypass ? hasNext : null;
} }
/** /**
@ -943,21 +788,16 @@ public class RegionCoprocessorHost
public boolean postScannerNext(final InternalScanner s, public boolean postScannerNext(final InternalScanner s,
final List<Result> results, final int limit, boolean hasMore) final List<Result> results, final int limit, boolean hasMore)
throws IOException { throws IOException {
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { hasMore = ((RegionObserver)env.getInstance()).postScannerNext(env, s,
if (env.getInstance() instanceof RegionObserver) { results, limit, hasMore);
hasMore = ((RegionObserver)env.getInstance()).postScannerNext(env, s, if (env.shouldComplete()) {
results, limit, hasMore); break;
if (env.shouldComplete()) {
break;
}
} }
} }
return hasMore;
} finally {
coprocessorLock.readLock().unlock();
} }
return hasMore;
} }
/** /**
@ -967,22 +807,17 @@ public class RegionCoprocessorHost
*/ */
public boolean preScannerClose(final InternalScanner s) public boolean preScannerClose(final InternalScanner s)
throws IOException { throws IOException {
try { boolean bypass = false;
boolean bypass = false; for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).preScannerClose(env, s);
if (env.getInstance() instanceof RegionObserver) { bypass |= env.shouldBypass();
((RegionObserver)env.getInstance()).preScannerClose(env, s); if (env.shouldComplete()) {
bypass |= env.shouldBypass(); break;
if (env.shouldComplete()) {
break;
}
} }
} }
return bypass;
} finally {
coprocessorLock.readLock().unlock();
} }
return bypass;
} }
/** /**
@ -991,18 +826,13 @@ public class RegionCoprocessorHost
*/ */
public void postScannerClose(final InternalScanner s) public void postScannerClose(final InternalScanner s)
throws IOException { throws IOException {
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).postScannerClose(env, s);
if (env.getInstance() instanceof RegionObserver) { if (env.shouldComplete()) {
((RegionObserver)env.getInstance()).postScannerClose(env, s); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
@ -1015,23 +845,18 @@ public class RegionCoprocessorHost
*/ */
public boolean preWALRestore(HRegionInfo info, HLogKey logKey, public boolean preWALRestore(HRegionInfo info, HLogKey logKey,
WALEdit logEdit) throws IOException { WALEdit logEdit) throws IOException {
try { boolean bypass = false;
boolean bypass = false; for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).preWALRestore(env, info, logKey,
if (env.getInstance() instanceof RegionObserver) { logEdit);
((RegionObserver)env.getInstance()).preWALRestore(env, info, logKey, }
logEdit); bypass |= env.shouldBypass();
} if (env.shouldComplete()) {
bypass |= env.shouldBypass(); break;
if (env.shouldComplete()) {
break;
}
} }
return bypass;
} finally {
coprocessorLock.readLock().unlock();
} }
return bypass;
} }
/** /**
@ -1042,19 +867,14 @@ public class RegionCoprocessorHost
*/ */
public void postWALRestore(HRegionInfo info, HLogKey logKey, public void postWALRestore(HRegionInfo info, HLogKey logKey,
WALEdit logEdit) throws IOException { WALEdit logEdit) throws IOException {
try { for (RegionEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof RegionObserver) {
for (RegionEnvironment env: coprocessors) { ((RegionObserver)env.getInstance()).postWALRestore(env, info,
if (env.getInstance() instanceof RegionObserver) { logKey, logEdit);
((RegionObserver)env.getInstance()).postWALRestore(env, info, }
logKey, logEdit); if (env.shouldComplete()) {
} break;
if (env.shouldComplete()) {
break;
}
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
} }

44
src/main/java/org/apache/hadoop/hbase/regionserver/wal/WALCoprocessorHost.java
View File

@ -95,24 +95,19 @@ public class WALCoprocessorHost
*/ */
public boolean preWALWrite(HRegionInfo info, HLogKey logKey, WALEdit logEdit) public boolean preWALWrite(HRegionInfo info, HLogKey logKey, WALEdit logEdit)
throws IOException { throws IOException {
try { boolean bypass = false;
boolean bypass = false; for (WALEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof
for (WALEnvironment env: coprocessors) { org.apache.hadoop.hbase.coprocessor.WALObserver) {
if (env.getInstance() instanceof ((org.apache.hadoop.hbase.coprocessor.WALObserver)env.getInstance()).
org.apache.hadoop.hbase.coprocessor.WALObserver) { preWALWrite(env, info, logKey, logEdit);
((org.apache.hadoop.hbase.coprocessor.WALObserver)env.getInstance()). bypass |= env.shouldBypass();
preWALWrite(env, info, logKey, logEdit); if (env.shouldComplete()) {
bypass |= env.shouldBypass(); break;
if (env.shouldComplete()) {
break;
}
} }
} }
return bypass;
} finally {
coprocessorLock.readLock().unlock();
} }
return bypass;
} }
/** /**
@ -123,20 +118,15 @@ public class WALCoprocessorHost
*/ */
public void postWALWrite(HRegionInfo info, HLogKey logKey, WALEdit logEdit) public void postWALWrite(HRegionInfo info, HLogKey logKey, WALEdit logEdit)
throws IOException { throws IOException {
try { for (WALEnvironment env: coprocessors) {
coprocessorLock.readLock().lock(); if (env.getInstance() instanceof
for (WALEnvironment env: coprocessors) { org.apache.hadoop.hbase.coprocessor.WALObserver) {
if (env.getInstance() instanceof ((org.apache.hadoop.hbase.coprocessor.WALObserver)env.getInstance()).
org.apache.hadoop.hbase.coprocessor.WALObserver) { postWALWrite(env, info, logKey, logEdit);
((org.apache.hadoop.hbase.coprocessor.WALObserver)env.getInstance()). if (env.shouldComplete()) {
postWALWrite(env, info, logKey, logEdit); break;
if (env.shouldComplete()) {
break;
}
} }
} }
} finally {
coprocessorLock.readLock().unlock();
} }
} }
} }

175
src/main/java/org/apache/hadoop/hbase/util/SortedCopyOnWriteSet.java Normal file
View File

@ -0,0 +1,175 @@
/*
* Copyright 2011 The Apache Software Foundation
*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.hadoop.hbase.util;
import java.util.Collection;
import java.util.Comparator;
import java.util.Iterator;
import java.util.SortedSet;
import java.util.TreeSet;
/**
* Simple {@link java.util.SortedSet} implementation that uses an internal
* {@link java.util.TreeSet} to provide ordering. All mutation operations
* create a new copy of the <code>TreeSet</code> instance, so are very
* expensive. This class is only intended for use on small, very rarely
* written collections that expect highly concurrent reads. Read operations
* are performed on a reference to the internal <code>TreeSet</code> at the
* time of invocation, so will not see any mutations to the collection during
* their operation.
*
* <p>Note that due to the use of a {@link java.util.TreeSet} internally,
* a {@link java.util.Comparator} instance must be provided, or collection
* elements must implement {@link java.lang.Comparable}.
* </p>
* @param <E> A class implementing {@link java.lang.Comparable} or able to be
* compared by a provided comparator.
*/
public class SortedCopyOnWriteSet<E> implements SortedSet<E> {
private SortedSet<E> internalSet;
public SortedCopyOnWriteSet() {
this.internalSet = new TreeSet<E>();
}
public SortedCopyOnWriteSet(Collection<? extends E> c) {
this.internalSet = new TreeSet<E>(c);
}
public SortedCopyOnWriteSet(Comparator<? super E> comparator) {
this.internalSet = new TreeSet<E>(comparator);
}
@Override
public int size() {
return internalSet.size();
}
@Override
public boolean isEmpty() {
return internalSet.isEmpty();
}
@Override
public boolean contains(Object o) {
return internalSet.contains(o);
}
@Override
public Iterator<E> iterator() {
return internalSet.iterator();
}
@Override
public Object[] toArray() {
return internalSet.toArray();
}
@Override
public <T> T[] toArray(T[] a) {
return internalSet.toArray(a);
}
@Override
public synchronized boolean add(E e) {
SortedSet<E> newSet = new TreeSet<E>(internalSet);
boolean added = newSet.add(e);
internalSet = newSet;
return added;
}
@Override
public synchronized boolean remove(Object o) {
SortedSet<E> newSet = new TreeSet<E>(internalSet);
boolean removed = newSet.remove(o);
internalSet = newSet;
return removed;
}
@Override
public boolean containsAll(Collection<?> c) {
return internalSet.containsAll(c);
}
@Override
public synchronized boolean addAll(Collection<? extends E> c) {
SortedSet<E> newSet = new TreeSet<E>(internalSet);
boolean changed = newSet.addAll(c);
internalSet = newSet;
return changed;
}
@Override
public synchronized boolean retainAll(Collection<?> c) {
SortedSet<E> newSet = new TreeSet<E>(internalSet);
boolean changed = newSet.retainAll(c);
internalSet = newSet;
return changed;
}
@Override
public synchronized boolean removeAll(Collection<?> c) {
SortedSet<E> newSet = new TreeSet<E>(internalSet);
boolean changed = newSet.removeAll(c);
internalSet = newSet;
return changed;
}
@Override
public synchronized void clear() {
Comparator<? super E> comparator = internalSet.comparator();
if (comparator != null) {
internalSet = new TreeSet<E>(comparator);
} else {
internalSet = new TreeSet<E>();
}
}
@Override
public Comparator<? super E> comparator() {
return internalSet.comparator();
}
@Override
public SortedSet<E> subSet(E fromElement, E toElement) {
return internalSet.subSet(fromElement, toElement);
}
@Override
public SortedSet<E> headSet(E toElement) {
return internalSet.headSet(toElement);
}
@Override
public SortedSet<E> tailSet(E fromElement) {
return internalSet.tailSet(fromElement);
}
@Override
public E first() {
return internalSet.first();
}
@Override
public E last() {
return internalSet.last();
}
}

103
src/test/java/org/apache/hadoop/hbase/util/TestSortedCopyOnWriteSet.java Normal file
View File

@ -0,0 +1,103 @@
/*
* Copyright 2011 The Apache Software Foundation
*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.hadoop.hbase.util;
import static org.junit.Assert.*;
import java.util.Arrays;
import java.util.Iterator;
import com.google.common.collect.Lists;
import org.junit.Test;
public class TestSortedCopyOnWriteSet {
@Test
public void testSorting() throws Exception {
SortedCopyOnWriteSet<String> set = new SortedCopyOnWriteSet<String>();
set.add("c");
set.add("d");
set.add("a");
set.add("b");
String[] expected = new String[]{"a", "b", "c", "d"};
String[] stored = set.toArray(new String[4]);
assertArrayEquals(expected, stored);
set.add("c");
assertEquals(4, set.size());
stored = set.toArray(new String[4]);
assertArrayEquals(expected, stored);
}
@Test
public void testIteratorIsolation() throws Exception {
SortedCopyOnWriteSet<String> set = new SortedCopyOnWriteSet<String>(
Lists.newArrayList("a", "b", "c", "d", "e"));
// isolation of remove()
Iterator<String> iter = set.iterator();
set.remove("c");
boolean found = false;
while (iter.hasNext() && !found) {
found = "c".equals(iter.next());
}
assertTrue(found);
iter = set.iterator();
found = false;
while (iter.hasNext() && !found) {
found = "c".equals(iter.next());
}
assertFalse(found);
// isolation of add()
iter = set.iterator();
set.add("f");
found = false;
while (iter.hasNext() && !found) {
String next = iter.next();
found = "f".equals(next);
}
assertFalse(found);
// isolation of addAll()
iter = set.iterator();
set.addAll(Lists.newArrayList("g", "h", "i"));
found = false;
while (iter.hasNext() && !found) {
String next = iter.next();
found = "g".equals(next) || "h".equals(next) || "i".equals(next);
}
assertFalse(found);
// isolation of clear()
iter = set.iterator();
set.clear();
assertEquals(0, set.size());
int size = 0;
while (iter.hasNext()) {
iter.next();
size++;
}
assertTrue(size > 0);
}
}