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

HBASE-15106 Procedure v2 - Procedure Queue pass Procedure for better debuggability

This commit is contained in:
Matteo Bertozzi 2016-01-19 11:54:24 -08:00
parent 7a096ac6cb
commit 7d7a2d8712
15 changed files with 179 additions and 147 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/AddColumnFamilyProcedure.java
View File

@ -185,12 +185,12 @@ public class AddColumnFamilyProcedure
@Override
protected boolean acquireLock(final MasterProcedureEnv env) {
if (env.waitInitialized(this)) return false;
return env.getProcedureQueue().tryAcquireTableExclusiveLock(tableName, "add family");
return env.getProcedureQueue().tryAcquireTableExclusiveLock(this, tableName);
}
@Override
protected void releaseLock(final MasterProcedureEnv env) {
env.getProcedureQueue().releaseTableExclusiveLock(tableName);
env.getProcedureQueue().releaseTableExclusiveLock(this, tableName);
}
@Override

4
hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/CreateNamespaceProcedure.java
View File

@ -207,12 +207,12 @@ public class CreateNamespaceProcedure
return false;
}
}
return env.getProcedureQueue().tryAcquireNamespaceExclusiveLock(getNamespaceName());
return env.getProcedureQueue().tryAcquireNamespaceExclusiveLock(this, getNamespaceName());
}
@Override
protected void releaseLock(final MasterProcedureEnv env) {
env.getProcedureQueue().releaseNamespaceExclusiveLock(getNamespaceName());
env.getProcedureQueue().releaseNamespaceExclusiveLock(this, getNamespaceName());
}
@Override

4
hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/CreateTableProcedure.java
View File

@ -273,12 +273,12 @@ public class CreateTableProcedure
if (!getTableName().isSystemTable() && env.waitInitialized(this)) {
return false;
}
return env.getProcedureQueue().tryAcquireTableExclusiveLock(getTableName(), "create table");
return env.getProcedureQueue().tryAcquireTableExclusiveLock(this, getTableName());
}
@Override
protected void releaseLock(final MasterProcedureEnv env) {
env.getProcedureQueue().releaseTableExclusiveLock(getTableName());
env.getProcedureQueue().releaseTableExclusiveLock(this, getTableName());
}
private boolean prepareCreate(final MasterProcedureEnv env) throws IOException {

4
hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/DeleteColumnFamilyProcedure.java
View File

@ -201,12 +201,12 @@ public class DeleteColumnFamilyProcedure
@Override
protected boolean acquireLock(final MasterProcedureEnv env) {
if (env.waitInitialized(this)) return false;
return env.getProcedureQueue().tryAcquireTableExclusiveLock(tableName, "delete family");
return env.getProcedureQueue().tryAcquireTableExclusiveLock(this, tableName);
}
@Override
protected void releaseLock(final MasterProcedureEnv env) {
env.getProcedureQueue().releaseTableExclusiveLock(tableName);
env.getProcedureQueue().releaseTableExclusiveLock(this, tableName);
}
@Override

4
hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/DeleteNamespaceProcedure.java
View File

@ -213,12 +213,12 @@ public class DeleteNamespaceProcedure
@Override
protected boolean acquireLock(final MasterProcedureEnv env) {
if (env.waitInitialized(this)) return false;
return env.getProcedureQueue().tryAcquireNamespaceExclusiveLock(getNamespaceName());
return env.getProcedureQueue().tryAcquireNamespaceExclusiveLock(this, getNamespaceName());
}
@Override
protected void releaseLock(final MasterProcedureEnv env) {
env.getProcedureQueue().releaseNamespaceExclusiveLock(getNamespaceName());
env.getProcedureQueue().releaseNamespaceExclusiveLock(this, getNamespaceName());
}
@Override

4
hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/DeleteTableProcedure.java
View File

@ -199,12 +199,12 @@ public class DeleteTableProcedure
@Override
protected boolean acquireLock(final MasterProcedureEnv env) {
if (env.waitInitialized(this)) return false;
return env.getProcedureQueue().tryAcquireTableExclusiveLock(getTableName(), "delete table");
return env.getProcedureQueue().tryAcquireTableExclusiveLock(this, getTableName());
}
@Override
protected void releaseLock(final MasterProcedureEnv env) {
env.getProcedureQueue().releaseTableExclusiveLock(getTableName());
env.getProcedureQueue().releaseTableExclusiveLock(this, getTableName());
}
@Override

4
hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/DisableTableProcedure.java
View File

@ -215,12 +215,12 @@ public class DisableTableProcedure
@Override
protected boolean acquireLock(final MasterProcedureEnv env) {
if (env.waitInitialized(this)) return false;
return env.getProcedureQueue().tryAcquireTableExclusiveLock(tableName, "disable table");
return env.getProcedureQueue().tryAcquireTableExclusiveLock(this, tableName);
}
@Override
protected void releaseLock(final MasterProcedureEnv env) {
env.getProcedureQueue().releaseTableExclusiveLock(tableName);
env.getProcedureQueue().releaseTableExclusiveLock(this, tableName);
}
@Override

4
hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/EnableTableProcedure.java
View File

@ -240,12 +240,12 @@ public class EnableTableProcedure
@Override
protected boolean acquireLock(final MasterProcedureEnv env) {
if (env.waitInitialized(this)) return false;
return env.getProcedureQueue().tryAcquireTableExclusiveLock(tableName, "enable table");
return env.getProcedureQueue().tryAcquireTableExclusiveLock(this, tableName);
}
@Override
protected void releaseLock(final MasterProcedureEnv env) {
env.getProcedureQueue().releaseTableExclusiveLock(tableName);
env.getProcedureQueue().releaseTableExclusiveLock(this, tableName);
}
@Override

105
hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/MasterProcedureScheduler.java
View File

@ -709,6 +709,11 @@ public class MasterProcedureScheduler implements ProcedureRunnableSet {
throw new UnsupportedOperationException();
}
@Override
public Procedure peek() {
throw new UnsupportedOperationException();
}
@Override
public Procedure poll() {
throw new UnsupportedOperationException();
@ -731,18 +736,18 @@ public class MasterProcedureScheduler implements ProcedureRunnableSet {
/**
* Try to acquire the exclusive lock on the specified table.
* other operations in the table-queue will be executed after the lock is released.
* @param procedure the procedure trying to acquire the lock
* @param table Table to lock
* @param purpose Human readable reason for locking the table
* @return true if we were able to acquire the lock on the table, otherwise false.
*/
public boolean tryAcquireTableExclusiveLock(final TableName table, final String purpose) {
public boolean tryAcquireTableExclusiveLock(final Procedure procedure, final TableName table) {
schedLock.lock();
TableQueue queue = getTableQueue(table);
if (!queue.getNamespaceQueue().trySharedLock()) {
return false;
}
if (!queue.tryExclusiveLock()) {
if (!queue.tryExclusiveLock(procedure.getProcId())) {
queue.getNamespaceQueue().releaseSharedLock();
schedLock.unlock();
return false;
@ -752,7 +757,7 @@ public class MasterProcedureScheduler implements ProcedureRunnableSet {
schedLock.unlock();
// Zk lock is expensive...
boolean hasXLock = queue.tryZkExclusiveLock(lockManager, purpose);
boolean hasXLock = queue.tryZkExclusiveLock(lockManager, procedure.toString());
if (!hasXLock) {
schedLock.lock();
queue.releaseExclusiveLock();
@ -765,9 +770,10 @@ public class MasterProcedureScheduler implements ProcedureRunnableSet {
/**
* Release the exclusive lock taken with tryAcquireTableWrite()
* @param procedure the procedure releasing the lock
* @param table the name of the table that has the exclusive lock
*/
public void releaseTableExclusiveLock(final TableName table) {
public void releaseTableExclusiveLock(final Procedure procedure, final TableName table) {
schedLock.lock();
TableQueue queue = getTableQueue(table);
schedLock.unlock();
@ -785,44 +791,48 @@ public class MasterProcedureScheduler implements ProcedureRunnableSet {
/**
* Try to acquire the shared lock on the specified table.
* other "read" operations in the table-queue may be executed concurrently,
* @param procedure the procedure trying to acquire the lock
* @param table Table to lock
* @param purpose Human readable reason for locking the table
* @return true if we were able to acquire the lock on the table, otherwise false.
*/
public boolean tryAcquireTableSharedLock(final TableName table, final String purpose) {
public boolean tryAcquireTableSharedLock(final Procedure procedure, final TableName table) {
return tryAcquireTableQueueSharedLock(procedure, table) != null;
}
private TableQueue tryAcquireTableQueueSharedLock(final Procedure procedure,
final TableName table) {
schedLock.lock();
TableQueue queue = getTableQueue(table);
if (!queue.getNamespaceQueue().trySharedLock()) {
return false;
return null;
}
if (!queue.trySharedLock()) {
queue.getNamespaceQueue().releaseSharedLock();
schedLock.unlock();
return false;
return null;
}
schedLock.unlock();
// Zk lock is expensive...
boolean hasXLock = queue.tryZkSharedLock(lockManager, purpose);
if (!hasXLock) {
if (!queue.tryZkSharedLock(lockManager, procedure.toString())) {
schedLock.lock();
queue.releaseSharedLock();
queue.getNamespaceQueue().releaseSharedLock();
schedLock.unlock();
return null;
}
return hasXLock;
return queue;
}
/**
* Release the shared lock taken with tryAcquireTableRead()
* @param procedure the procedure releasing the lock
* @param table the name of the table that has the shared lock
*/
public void releaseTableSharedLock(final TableName table) {
schedLock.lock();
TableQueue queue = getTableQueue(table);
schedLock.unlock();
public void releaseTableSharedLock(final Procedure procedure, final TableName table) {
final TableQueue queue = getTableQueueWithLock(table);
// Zk lock is expensive...
queue.releaseZkSharedLock(lockManager);
@ -848,7 +858,7 @@ public class MasterProcedureScheduler implements ProcedureRunnableSet {
TableQueue queue = getTableQueue(table);
if (queue == null) return true;
if (queue.isEmpty() && queue.acquireDeleteLock()) {
if (queue.isEmpty() && queue.tryExclusiveLock(0)) {
// remove the table from the run-queue and the map
if (IterableList.isLinked(queue)) {
tableRunQueue.remove(queue);
@ -877,18 +887,19 @@ public class MasterProcedureScheduler implements ProcedureRunnableSet {
// ============================================================================
/**
* Try to acquire the exclusive lock on the specified namespace.
* @see #releaseNamespaceExclusiveLock(String)
* @see #releaseNamespaceExclusiveLock(Procedure,String)
* @param procedure the procedure trying to acquire the lock
* @param nsName Namespace to lock
* @return true if we were able to acquire the lock on the namespace, otherwise false.
*/
public boolean tryAcquireNamespaceExclusiveLock(final String nsName) {
public boolean tryAcquireNamespaceExclusiveLock(final Procedure procedure, final String nsName) {
schedLock.lock();
try {
TableQueue tableQueue = getTableQueue(TableName.NAMESPACE_TABLE_NAME);
if (!tableQueue.trySharedLock()) return false;
NamespaceQueue nsQueue = getNamespaceQueue(nsName);
boolean hasLock = nsQueue.tryExclusiveLock();
boolean hasLock = nsQueue.tryExclusiveLock(procedure.getProcId());
if (!hasLock) {
tableQueue.releaseSharedLock();
}
@ -900,10 +911,11 @@ public class MasterProcedureScheduler implements ProcedureRunnableSet {
/**
* Release the exclusive lock
* @see #tryAcquireNamespaceExclusiveLock(String)
* @see #tryAcquireNamespaceExclusiveLock(Procedure,String)
* @param procedure the procedure releasing the lock
* @param nsName the namespace that has the exclusive lock
*/
public void releaseNamespaceExclusiveLock(final String nsName) {
public void releaseNamespaceExclusiveLock(final Procedure procedure, final String nsName) {
schedLock.lock();
try {
TableQueue tableQueue = getTableQueue(TableName.NAMESPACE_TABLE_NAME);
@ -921,15 +933,17 @@ public class MasterProcedureScheduler implements ProcedureRunnableSet {
// ============================================================================
/**
* Try to acquire the exclusive lock on the specified server.
* @see #releaseServerExclusiveLock(ServerName)
* @see #releaseServerExclusiveLock(Procedure,ServerName)
* @param procedure the procedure trying to acquire the lock
* @param serverName Server to lock
* @return true if we were able to acquire the lock on the server, otherwise false.
*/
public boolean tryAcquireServerExclusiveLock(final ServerName serverName) {
public boolean tryAcquireServerExclusiveLock(final Procedure procedure,
final ServerName serverName) {
schedLock.lock();
try {
ServerQueue queue = getServerQueue(serverName);
if (queue.tryExclusiveLock()) {
if (queue.tryExclusiveLock(procedure.getProcId())) {
removeFromRunQueue(serverRunQueue, queue);
return true;
}
@ -941,10 +955,12 @@ public class MasterProcedureScheduler implements ProcedureRunnableSet {
/**
* Release the exclusive lock
* @see #tryAcquireServerExclusiveLock(ServerName)
* @see #tryAcquireServerExclusiveLock(Procedure,ServerName)
* @param procedure the procedure releasing the lock
* @param serverName the server that has the exclusive lock
*/
public void releaseServerExclusiveLock(final ServerName serverName) {
public void releaseServerExclusiveLock(final Procedure procedure,
final ServerName serverName) {
schedLock.lock();
try {
ServerQueue queue = getServerQueue(serverName);
@ -957,20 +973,24 @@ public class MasterProcedureScheduler implements ProcedureRunnableSet {
/**
* Try to acquire the shared lock on the specified server.
* @see #releaseServerSharedLock(ServerName)
* @see #releaseServerSharedLock(Procedure,ServerName)
* @param procedure the procedure releasing the lock
* @param serverName Server to lock
* @return true if we were able to acquire the lock on the server, otherwise false.
*/
public boolean tryAcquireServerSharedLock(final ServerName serverName) {
public boolean tryAcquireServerSharedLock(final Procedure procedure,
final ServerName serverName) {
return getServerQueueWithLock(serverName).trySharedLock();
}
/**
* Release the shared lock taken
* @see #tryAcquireServerSharedLock(ServerName)
* @see #tryAcquireServerSharedLock(Procedure,ServerName)
* @param procedure the procedure releasing the lock
* @param serverName the server that has the shared lock
*/
public void releaseServerSharedLock(final ServerName serverName) {
public void releaseServerSharedLock(final Procedure procedure,
final ServerName serverName) {
getServerQueueWithLock(serverName).releaseSharedLock();
}
@ -981,8 +1001,10 @@ public class MasterProcedureScheduler implements ProcedureRunnableSet {
boolean isAvailable();
boolean isEmpty();
int size();
void add(Procedure proc, boolean addFront);
boolean requireExclusiveLock(Procedure proc);
Procedure peek();
Procedure poll();
boolean isSuspended();
@ -997,7 +1019,7 @@ public class MasterProcedureScheduler implements ProcedureRunnableSet {
private Queue<TKey> iterPrev = null;
private boolean suspended = false;
private boolean exclusiveLock = false;
private long exclusiveLockProcIdOwner = Long.MIN_VALUE;
private int sharedLock = 0;
private final TKey key;
@ -1041,7 +1063,7 @@ public class MasterProcedureScheduler implements ProcedureRunnableSet {
}
public synchronized boolean hasExclusiveLock() {
return this.exclusiveLock;
return this.exclusiveLockProcIdOwner != Long.MIN_VALUE;
}
public synchronized boolean trySharedLock() {
@ -1058,24 +1080,21 @@ public class MasterProcedureScheduler implements ProcedureRunnableSet {
return sharedLock == 1;
}
public synchronized boolean tryExclusiveLock() {
public synchronized boolean tryExclusiveLock(long procIdOwner) {
assert procIdOwner != Long.MIN_VALUE;
if (isLocked()) return false;
exclusiveLock = true;
exclusiveLockProcIdOwner = procIdOwner;
return true;
}
public synchronized void releaseExclusiveLock() {
exclusiveLock = false;
}
public synchronized boolean acquireDeleteLock() {
return tryExclusiveLock();
exclusiveLockProcIdOwner = Long.MIN_VALUE;
}
// This should go away when we have the new AM and its events
// and we move xlock to the lock-event-queue.
public synchronized boolean isAvailable() {
return !exclusiveLock && !isEmpty();
return !hasExclusiveLock() && !isEmpty();
}
// ======================================================================
@ -1125,6 +1144,10 @@ public class MasterProcedureScheduler implements ProcedureRunnableSet {
runnables.addLast(proc);
}
public Procedure peek() {
return runnables.peek();
}
@Override
public Procedure poll() {
return runnables.poll();

4
hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/ModifyColumnFamilyProcedure.java
View File

@ -182,12 +182,12 @@ public class ModifyColumnFamilyProcedure
@Override
protected boolean acquireLock(final MasterProcedureEnv env) {
if (env.waitInitialized(this)) return false;
return env.getProcedureQueue().tryAcquireTableExclusiveLock(tableName, "modify family");
return env.getProcedureQueue().tryAcquireTableExclusiveLock(this, tableName);
}
@Override
protected void releaseLock(final MasterProcedureEnv env) {
env.getProcedureQueue().releaseTableExclusiveLock(tableName);
env.getProcedureQueue().releaseTableExclusiveLock(this, tableName);
}
@Override

4
hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/ModifyNamespaceProcedure.java
View File

@ -193,12 +193,12 @@ public class ModifyNamespaceProcedure
@Override
protected boolean acquireLock(final MasterProcedureEnv env) {
if (env.waitInitialized(this)) return false;
return env.getProcedureQueue().tryAcquireNamespaceExclusiveLock(getNamespaceName());
return env.getProcedureQueue().tryAcquireNamespaceExclusiveLock(this, getNamespaceName());
}
@Override
protected void releaseLock(final MasterProcedureEnv env) {
env.getProcedureQueue().releaseNamespaceExclusiveLock(getNamespaceName());
env.getProcedureQueue().releaseNamespaceExclusiveLock(this, getNamespaceName());
}
@Override

4
hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/ModifyTableProcedure.java
View File

@ -216,12 +216,12 @@ public class ModifyTableProcedure
@Override
protected boolean acquireLock(final MasterProcedureEnv env) {
if (env.waitInitialized(this)) return false;
return env.getProcedureQueue().tryAcquireTableExclusiveLock(getTableName(), "modify table");
return env.getProcedureQueue().tryAcquireTableExclusiveLock(this, getTableName());
}
@Override
protected void releaseLock(final MasterProcedureEnv env) {
env.getProcedureQueue().releaseTableExclusiveLock(getTableName());
env.getProcedureQueue().releaseTableExclusiveLock(this, getTableName());
}
@Override

4
hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/ServerCrashProcedure.java
View File

@ -589,12 +589,12 @@ implements ServerProcedureInterface {
@Override
protected boolean acquireLock(final MasterProcedureEnv env) {
if (env.waitServerCrashProcessingEnabled(this)) return false;
return env.getProcedureQueue().tryAcquireServerExclusiveLock(getServerName());
return env.getProcedureQueue().tryAcquireServerExclusiveLock(this, getServerName());
}
@Override
protected void releaseLock(final MasterProcedureEnv env) {
env.getProcedureQueue().releaseServerExclusiveLock(getServerName());
env.getProcedureQueue().releaseServerExclusiveLock(this, getServerName());
}
@Override

4
hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/TruncateTableProcedure.java
View File

@ -182,12 +182,12 @@ public class TruncateTableProcedure
@Override
protected boolean acquireLock(final MasterProcedureEnv env) {
if (env.waitInitialized(this)) return false;
return env.getProcedureQueue().tryAcquireTableExclusiveLock(getTableName(), "truncate table");
return env.getProcedureQueue().tryAcquireTableExclusiveLock(this, getTableName());
}
@Override
protected void releaseLock(final MasterProcedureEnv env) {
env.getProcedureQueue().releaseTableExclusiveLock(getTableName());
env.getProcedureQueue().releaseTableExclusiveLock(this, getTableName());
}
@Override

169
hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestMasterProcedureScheduler.java
View File

@ -21,6 +21,7 @@ package org.apache.hadoop.hbase.master.procedure;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Arrays;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Map;
@ -76,9 +77,11 @@ public class TestMasterProcedureScheduler {
@Override
public void run() {
try {
TestTableProcedure proc = new TestTableProcedure(1, table,
TableProcedureInterface.TableOperationType.CREATE);
while (running.get() && !failure.get()) {
if (procQueue.tryAcquireTableExclusiveLock(table, "create")) {
procQueue.releaseTableExclusiveLock(table);
if (procQueue.tryAcquireTableExclusiveLock(proc, table)) {
procQueue.releaseTableExclusiveLock(proc, table);
}
}
} catch (Throwable e) {
@ -92,9 +95,11 @@ public class TestMasterProcedureScheduler {
@Override
public void run() {
try {
TestTableProcedure proc = new TestTableProcedure(2, table,
TableProcedureInterface.TableOperationType.DELETE);
while (running.get() && !failure.get()) {
if (procQueue.tryAcquireTableExclusiveLock(table, "delete")) {
procQueue.releaseTableExclusiveLock(table);
if (procQueue.tryAcquireTableExclusiveLock(proc, table)) {
procQueue.releaseTableExclusiveLock(proc, table);
}
procQueue.markTableAsDeleted(table);
}
@ -141,8 +146,8 @@ public class TestMasterProcedureScheduler {
Procedure proc = queue.poll();
assertTrue(proc != null);
TableName tableName = ((TestTableProcedure)proc).getTableName();
queue.tryAcquireTableExclusiveLock(tableName, "test");
queue.releaseTableExclusiveLock(tableName);
queue.tryAcquireTableExclusiveLock(proc, tableName);
queue.releaseTableExclusiveLock(proc, tableName);
queue.completionCleanup(proc);
assertEquals(--count, queue.size());
assertEquals(i * 1000 + j, proc.getProcId());
@ -172,14 +177,15 @@ public class TestMasterProcedureScheduler {
assertFalse(queue.markTableAsDeleted(tableName));
// fetch item and take a lock
assertEquals(1, queue.poll().getProcId());
Procedure proc = queue.poll();
assertEquals(1, proc.getProcId());
// take the xlock
assertTrue(queue.tryAcquireTableExclusiveLock(tableName, "write"));
assertTrue(queue.tryAcquireTableExclusiveLock(proc, tableName));
// table can't be deleted because we have the lock
assertEquals(0, queue.size());
assertFalse(queue.markTableAsDeleted(tableName));
// release the xlock
queue.releaseTableExclusiveLock(tableName);
queue.releaseTableExclusiveLock(proc, tableName);
// complete the table deletion
assertTrue(queue.markTableAsDeleted(tableName));
}
@ -201,20 +207,22 @@ public class TestMasterProcedureScheduler {
// table can't be deleted because one item is in the queue
assertFalse(queue.markTableAsDeleted(tableName));
for (int i = 1; i <= nitems; ++i) {
Procedure[] procs = new Procedure[nitems];
for (int i = 0; i < nitems; ++i) {
// fetch item and take a lock
assertEquals(i, queue.poll().getProcId());
Procedure proc = procs[i] = queue.poll();
assertEquals(i + 1, proc.getProcId());
// take the rlock
assertTrue(queue.tryAcquireTableSharedLock(tableName, "read " + i));
assertTrue(queue.tryAcquireTableSharedLock(proc, tableName));
// table can't be deleted because we have locks and/or items in the queue
assertFalse(queue.markTableAsDeleted(tableName));
}
for (int i = 1; i <= nitems; ++i) {
for (int i = 0; i < nitems; ++i) {
// table can't be deleted because we have locks
assertFalse(queue.markTableAsDeleted(tableName));
// release the rlock
queue.releaseTableSharedLock(tableName);
queue.releaseTableSharedLock(procs[i], tableName);
}
// there are no items and no lock in the queeu
@ -241,49 +249,49 @@ public class TestMasterProcedureScheduler {
TableProcedureInterface.TableOperationType.READ));
// Fetch the 1st item and take the write lock
long procId = queue.poll().getProcId();
assertEquals(1, procId);
assertEquals(true, queue.tryAcquireTableExclusiveLock(tableName, "write " + procId));
Procedure proc = queue.poll();
assertEquals(1, proc.getProcId());
assertEquals(true, queue.tryAcquireTableExclusiveLock(proc, tableName));
// Fetch the 2nd item and verify that the lock can't be acquired
assertEquals(null, queue.poll(0));
// Release the write lock and acquire the read lock
queue.releaseTableExclusiveLock(tableName);
queue.releaseTableExclusiveLock(proc, tableName);
// Fetch the 2nd item and take the read lock
procId = queue.poll().getProcId();
assertEquals(2, procId);
assertEquals(true, queue.tryAcquireTableSharedLock(tableName, "read " + procId));
Procedure rdProc = queue.poll();
assertEquals(2, rdProc.getProcId());
assertEquals(true, queue.tryAcquireTableSharedLock(rdProc, tableName));
// Fetch the 3rd item and verify that the lock can't be acquired
procId = queue.poll().getProcId();
assertEquals(3, procId);
assertEquals(false, queue.tryAcquireTableExclusiveLock(tableName, "write " + procId));
Procedure wrProc = queue.poll();
assertEquals(3, wrProc.getProcId());
assertEquals(false, queue.tryAcquireTableExclusiveLock(wrProc, tableName));
// release the rdlock of item 2 and take the wrlock for the 3d item
queue.releaseTableSharedLock(tableName);
assertEquals(true, queue.tryAcquireTableExclusiveLock(tableName, "write " + procId));
queue.releaseTableSharedLock(rdProc, tableName);
assertEquals(true, queue.tryAcquireTableExclusiveLock(wrProc, tableName));
// Fetch 4th item and verify that the lock can't be acquired
assertEquals(null, queue.poll(0));
// Release the write lock and acquire the read lock
queue.releaseTableExclusiveLock(tableName);
queue.releaseTableExclusiveLock(wrProc, tableName);
// Fetch the 4th item and take the read lock
procId = queue.poll().getProcId();
assertEquals(4, procId);
assertEquals(true, queue.tryAcquireTableSharedLock(tableName, "read " + procId));
rdProc = queue.poll();
assertEquals(4, rdProc.getProcId());
assertEquals(true, queue.tryAcquireTableSharedLock(rdProc, tableName));
// Fetch the 4th item and take the read lock
procId = queue.poll().getProcId();
assertEquals(5, procId);
assertEquals(true, queue.tryAcquireTableSharedLock(tableName, "read " + procId));
Procedure rdProc2 = queue.poll();
assertEquals(5, rdProc2.getProcId());
assertEquals(true, queue.tryAcquireTableSharedLock(rdProc2, tableName));
// Release 4th and 5th read-lock
queue.releaseTableSharedLock(tableName);
queue.releaseTableSharedLock(tableName);
queue.releaseTableSharedLock(rdProc, tableName);
queue.releaseTableSharedLock(rdProc2, tableName);
// remove table queue
assertEquals(0, queue.size());
@ -306,36 +314,36 @@ public class TestMasterProcedureScheduler {
TableProcedureInterface.TableOperationType.EDIT));
// Fetch the 1st item and take the write lock
long procId = queue.poll().getProcId();
assertEquals(1, procId);
assertEquals(true, queue.tryAcquireNamespaceExclusiveLock(nsName1));
Procedure procNs1 = queue.poll();
assertEquals(1, procNs1.getProcId());
assertEquals(true, queue.tryAcquireNamespaceExclusiveLock(procNs1, nsName1));
// System tables have 2 as default priority
Procedure proc = queue.poll();
assertEquals(4, proc.getProcId());
assertEquals(true, queue.tryAcquireNamespaceExclusiveLock(nsName2));
queue.releaseNamespaceExclusiveLock(nsName2);
queue.yield(proc);
Procedure procNs2 = queue.poll();
assertEquals(4, procNs2.getProcId());
assertEquals(true, queue.tryAcquireNamespaceExclusiveLock(procNs2, nsName2));
queue.releaseNamespaceExclusiveLock(procNs2, nsName2);
queue.yield(procNs2);
// table on ns1 is locked, so we get table on ns2
procId = queue.poll().getProcId();
assertEquals(3, procId);
assertEquals(true, queue.tryAcquireTableExclusiveLock(tableName2, "lock " + procId));
procNs2 = queue.poll();
assertEquals(3, procNs2.getProcId());
assertEquals(true, queue.tryAcquireTableExclusiveLock(procNs2, tableName2));
// ns2 is not available (TODO we may avoid this one)
proc = queue.poll();
assertEquals(4, proc.getProcId());
assertEquals(false, queue.tryAcquireNamespaceExclusiveLock(nsName2));
queue.yield(proc);
Procedure procNs2b = queue.poll();
assertEquals(4, procNs2b.getProcId());
assertEquals(false, queue.tryAcquireNamespaceExclusiveLock(procNs2b, nsName2));
queue.yield(procNs2b);
// release the ns1 lock
queue.releaseNamespaceExclusiveLock(nsName1);
queue.releaseNamespaceExclusiveLock(procNs1, nsName1);
// we are now able to execute table of ns1
procId = queue.poll().getProcId();
long procId = queue.poll().getProcId();
assertEquals(2, procId);
queue.releaseTableExclusiveLock(tableName2);
queue.releaseTableExclusiveLock(procNs2, tableName2);
// we are now able to execute ns2
procId = queue.poll().getProcId();
@ -373,7 +381,7 @@ public class TestMasterProcedureScheduler {
public void run() {
while (opsCount.get() > 0) {
try {
TableProcedureInterface proc = procSet.acquire();
Procedure proc = procSet.acquire();
if (proc == null) {
queue.signalAll();
if (opsCount.get() > 0) {
@ -381,14 +389,14 @@ public class TestMasterProcedureScheduler {
}
break;
}
TableName tableId = procSet.getTableName(proc);
synchronized (concurrentTables) {
assertTrue("unexpected concurrency on " + proc.getTableName(),
concurrentTables.add(proc.getTableName()));
assertTrue("unexpected concurrency on " + tableId, concurrentTables.add(tableId));
}
assertTrue(opsCount.decrementAndGet() >= 0);
try {
long procId = ((Procedure)proc).getProcId();
TableName tableId = proc.getTableName();
long procId = proc.getProcId();
int concurrent = concurrentCount.incrementAndGet();
assertTrue("inc-concurrent="+ concurrent +" 1 <= concurrent <= "+ NUM_TABLES,
concurrent >= 1 && concurrent <= NUM_TABLES);
@ -399,7 +407,7 @@ public class TestMasterProcedureScheduler {
assertTrue("dec-concurrent=" + concurrent, concurrent < NUM_TABLES);
} finally {
synchronized (concurrentTables) {
assertTrue(concurrentTables.remove(proc.getTableName()));
assertTrue(concurrentTables.remove(tableId));
}
procSet.release(proc);
}
@ -431,43 +439,36 @@ public class TestMasterProcedureScheduler {
public static class TestTableProcSet {
private final MasterProcedureScheduler queue;
private Map<Long, TableProcedureInterface> procsMap =
new ConcurrentHashMap<Long, TableProcedureInterface>();
public TestTableProcSet(final MasterProcedureScheduler queue) {
this.queue = queue;
}
public void addBack(TableProcedureInterface tableProc) {
Procedure proc = (Procedure)tableProc;
procsMap.put(proc.getProcId(), tableProc);
public void addBack(Procedure proc) {
queue.addBack(proc);
}
public void addFront(TableProcedureInterface tableProc) {
Procedure proc = (Procedure)tableProc;
procsMap.put(proc.getProcId(), tableProc);
public void addFront(Procedure proc) {
queue.addFront(proc);
}
public TableProcedureInterface acquire() {
TableProcedureInterface proc = null;
public Procedure acquire() {
Procedure proc = null;
boolean avail = false;
while (!avail) {
Procedure xProc = queue.poll();
proc = xProc != null ? procsMap.remove(xProc.getProcId()) : null;
proc = queue.poll();
if (proc == null) break;
switch (proc.getTableOperationType()) {
switch (getTableOperationType(proc)) {
case CREATE:
case DELETE:
case EDIT:
avail = queue.tryAcquireTableExclusiveLock(proc.getTableName(),
"op="+ proc.getTableOperationType());
avail = queue.tryAcquireTableExclusiveLock(proc, getTableName(proc));
break;
case READ:
avail = queue.tryAcquireTableSharedLock(proc.getTableName(),
"op="+ proc.getTableOperationType());
avail = queue.tryAcquireTableSharedLock(proc, getTableName(proc));
break;
default:
throw new UnsupportedOperationException();
}
if (!avail) {
addFront(proc);
@ -477,18 +478,26 @@ public class TestMasterProcedureScheduler {
return proc;
}
public void release(TableProcedureInterface proc) {
switch (proc.getTableOperationType()) {
public void release(Procedure proc) {
switch (getTableOperationType(proc)) {
case CREATE:
case DELETE:
case EDIT:
queue.releaseTableExclusiveLock(proc.getTableName());
queue.releaseTableExclusiveLock(proc, getTableName(proc));
break;
case READ:
queue.releaseTableSharedLock(proc.getTableName());
queue.releaseTableSharedLock(proc, getTableName(proc));
break;
}
}
public TableName getTableName(Procedure proc) {
return ((TableProcedureInterface)proc).getTableName();
}
public TableProcedureInterface.TableOperationType getTableOperationType(Procedure proc) {
return ((TableProcedureInterface)proc).getTableOperationType();
}
}
public static class TestTableProcedure extends TestProcedure