HBASE-21468 separate workers for meta table is not working

This commit is contained in:
Allan Yang 2018-11-14 11:43:41 +08:00
parent c95832159f
commit 0f295de156
18 changed files with 29 additions and 15 deletions

View File

@ -124,7 +124,7 @@ public abstract class AbstractProcedureScheduler implements ProcedureScheduler {
try { try {
enqueue(procedure, addFront); enqueue(procedure, addFront);
if (notify) { if (notify) {
schedWaitCond.signal(); schedWaitCond.signalAll();
} }
} finally { } finally {
schedUnlock(); schedUnlock();
@ -311,10 +311,6 @@ public abstract class AbstractProcedureScheduler implements ProcedureScheduler {
if (waitingCount <= 0) { if (waitingCount <= 0) {
return; return;
} }
if (waitingCount == 1) { schedWaitCond.signalAll();
schedWaitCond.signal();
} else {
schedWaitCond.signalAll();
}
} }
} }

View File

@ -211,7 +211,7 @@ public class ProcedureExecutor<TEnvironment> {
/** /**
* Worker thread only for urgent tasks. * Worker thread only for urgent tasks.
*/ */
private List<WorkerThread> urgentWorkerThreads; private CopyOnWriteArrayList<WorkerThread> urgentWorkerThreads;
/** /**
* Created in the {@link #init(int, boolean)} method. Terminated in {@link #join()} (FIX! Doing * Created in the {@link #init(int, boolean)} method. Terminated in {@link #join()} (FIX! Doing
@ -564,7 +564,7 @@ public class ProcedureExecutor<TEnvironment> {
* is found on replay. otherwise false. * is found on replay. otherwise false.
*/ */
public void init(int numThreads, boolean abortOnCorruption) throws IOException { public void init(int numThreads, boolean abortOnCorruption) throws IOException {
init(numThreads, 1, abortOnCorruption); init(numThreads, 0, abortOnCorruption);
} }
/** /**
@ -595,7 +595,7 @@ public class ProcedureExecutor<TEnvironment> {
// Create the workers // Create the workers
workerId.set(0); workerId.set(0);
workerThreads = new CopyOnWriteArrayList<>(); workerThreads = new CopyOnWriteArrayList<>();
urgentWorkerThreads = new ArrayList<>(); urgentWorkerThreads = new CopyOnWriteArrayList<>();
for (int i = 0; i < corePoolSize; ++i) { for (int i = 0; i < corePoolSize; ++i) {
workerThreads.add(new WorkerThread(threadGroup)); workerThreads.add(new WorkerThread(threadGroup));
} }
@ -637,7 +637,7 @@ public class ProcedureExecutor<TEnvironment> {
return; return;
} }
// Start the executors. Here we must have the lastProcId set. // Start the executors. Here we must have the lastProcId set.
LOG.debug("Start workers {}, urgent workers", workerThreads.size(), LOG.debug("Start workers {}, urgent workers {}", workerThreads.size(),
urgentWorkerThreads.size()); urgentWorkerThreads.size());
timeoutExecutor.start(); timeoutExecutor.start();
for (WorkerThread worker: workerThreads) { for (WorkerThread worker: workerThreads) {
@ -2023,7 +2023,8 @@ public class ProcedureExecutor<TEnvironment> {
long lastUpdate = EnvironmentEdgeManager.currentTime(); long lastUpdate = EnvironmentEdgeManager.currentTime();
try { try {
while (isRunning() && keepAlive(lastUpdate)) { while (isRunning() && keepAlive(lastUpdate)) {
Procedure<TEnvironment> proc = scheduler.poll(keepAliveTime, TimeUnit.MILLISECONDS); Procedure<TEnvironment> proc = scheduler
.poll(onlyPollUrgent, keepAliveTime, TimeUnit.MILLISECONDS);
if (proc == null) { if (proc == null) {
continue; continue;
} }

View File

@ -87,7 +87,7 @@ public class ProcedureTestingUtility {
public static void initAndStartWorkers(ProcedureExecutor<?> procExecutor, int numThreads, public static void initAndStartWorkers(ProcedureExecutor<?> procExecutor, int numThreads,
boolean abortOnCorruption, boolean startWorkers) throws IOException { boolean abortOnCorruption, boolean startWorkers) throws IOException {
initAndStartWorkers(procExecutor, numThreads, 1, abortOnCorruption, startWorkers); initAndStartWorkers(procExecutor, numThreads, 0, abortOnCorruption, startWorkers);
} }
public static void initAndStartWorkers(ProcedureExecutor<?> procExecutor, int numThreads, public static void initAndStartWorkers(ProcedureExecutor<?> procExecutor, int numThreads,

View File

@ -380,7 +380,6 @@ public class TestYieldProcedures {
@Override @Override
public Procedure poll() { public Procedure poll() {
LOG.error("polled()");
pollCalls++; pollCalls++;
return super.poll(); return super.poll();
} }
@ -388,10 +387,15 @@ public class TestYieldProcedures {
@Override @Override
public Procedure poll(long timeout, TimeUnit unit) { public Procedure poll(long timeout, TimeUnit unit) {
pollCalls++; pollCalls++;
LOG.error("polled(long timeout, TimeUnit unit)");
return super.poll(timeout, unit); return super.poll(timeout, unit);
} }
@Override
public Procedure poll(boolean onlyUrgent, long timeout, TimeUnit unit) {
pollCalls++;
return super.poll(onlyUrgent, timeout, unit);
}
@Override @Override
public void completionCleanup(Procedure proc) { public void completionCleanup(Procedure proc) {
completionCalls++; completionCalls++;

View File

@ -62,6 +62,7 @@ public class TestAssignmentOnRSCrash {
private static void setupConf(Configuration conf) { private static void setupConf(Configuration conf) {
conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1); conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1);
conf.setInt(MasterProcedureConstants.MASTER_URGENT_PROCEDURE_THREADS, 0);
conf.set("hbase.balancer.tablesOnMaster", "none"); conf.set("hbase.balancer.tablesOnMaster", "none");
} }

View File

@ -86,6 +86,7 @@ public class TestMergeTableRegionsProcedure {
conf.setInt("hbase.master.maximum.ping.server.attempts", 3); conf.setInt("hbase.master.maximum.ping.server.attempts", 3);
conf.setInt("hbase.master.ping.server.retry.sleep.interval", 1); conf.setInt("hbase.master.ping.server.retry.sleep.interval", 1);
conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1); conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1);
conf.setInt(MasterProcedureConstants.MASTER_URGENT_PROCEDURE_THREADS, 0);
} }
@BeforeClass @BeforeClass

View File

@ -94,6 +94,7 @@ public class TestRogueRSAssignment {
conf.setInt("hbase.master.maximum.ping.server.attempts", 3); conf.setInt("hbase.master.maximum.ping.server.attempts", 3);
conf.setInt("hbase.master.ping.server.retry.sleep.interval", 1); conf.setInt("hbase.master.ping.server.retry.sleep.interval", 1);
conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1); conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1);
conf.setInt(MasterProcedureConstants.MASTER_URGENT_PROCEDURE_THREADS, 0);
} }
@BeforeClass @BeforeClass

View File

@ -70,6 +70,7 @@ public class TestLockManager {
private static void setupConf(Configuration conf) { private static void setupConf(Configuration conf) {
conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1); conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1);
conf.setInt(MasterProcedureConstants.MASTER_URGENT_PROCEDURE_THREADS, 0);
conf.setBoolean("hbase.procedure.check.owner.set", false); // since rpc user will be null conf.setBoolean("hbase.procedure.check.owner.set", false); // since rpc user will be null
conf.setInt(LockProcedure.LOCAL_MASTER_LOCKS_TIMEOUT_MS_CONF, LOCAL_LOCKS_TIMEOUT); conf.setInt(LockProcedure.LOCAL_MASTER_LOCKS_TIMEOUT_MS_CONF, LOCAL_LOCKS_TIMEOUT);
} }

View File

@ -98,6 +98,7 @@ public class TestLockProcedure {
private static void setupConf(Configuration conf) { private static void setupConf(Configuration conf) {
conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1); conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1);
conf.setInt(MasterProcedureConstants.MASTER_URGENT_PROCEDURE_THREADS, 0);
conf.setBoolean("hbase.procedure.check.owner.set", false); // since rpc user will be null conf.setBoolean("hbase.procedure.check.owner.set", false); // since rpc user will be null
conf.setInt(LockProcedure.REMOTE_LOCKS_TIMEOUT_MS_CONF, HEARTBEAT_TIMEOUT); conf.setInt(LockProcedure.REMOTE_LOCKS_TIMEOUT_MS_CONF, HEARTBEAT_TIMEOUT);
conf.setInt(LockProcedure.LOCAL_MASTER_LOCKS_TIMEOUT_MS_CONF, LOCAL_LOCKS_TIMEOUT); conf.setInt(LockProcedure.LOCAL_MASTER_LOCKS_TIMEOUT_MS_CONF, LOCAL_LOCKS_TIMEOUT);

View File

@ -57,6 +57,7 @@ public class TestCreateNamespaceProcedure {
private static void setupConf(Configuration conf) { private static void setupConf(Configuration conf) {
conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1); conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1);
conf.setInt(MasterProcedureConstants.MASTER_URGENT_PROCEDURE_THREADS, 0);
} }
@BeforeClass @BeforeClass

View File

@ -63,6 +63,7 @@ public class TestDeleteNamespaceProcedure {
private static void setupConf(Configuration conf) { private static void setupConf(Configuration conf) {
conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1); conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1);
conf.setInt(MasterProcedureConstants.MASTER_URGENT_PROCEDURE_THREADS, 0);
} }
@BeforeClass @BeforeClass

View File

@ -67,6 +67,7 @@ public class TestMasterFailoverWithProcedures {
conf.setInt(WALProcedureStore.ROLL_RETRIES_CONF_KEY, 1); conf.setInt(WALProcedureStore.ROLL_RETRIES_CONF_KEY, 1);
conf.setInt(WALProcedureStore.MAX_SYNC_FAILURE_ROLL_CONF_KEY, 1); conf.setInt(WALProcedureStore.MAX_SYNC_FAILURE_ROLL_CONF_KEY, 1);
conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1); conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1);
conf.setInt(MasterProcedureConstants.MASTER_URGENT_PROCEDURE_THREADS, 0);
} }
@Before @Before

View File

@ -72,6 +72,7 @@ public class TestMasterObserverPostCalls {
private static void setupConf(Configuration conf) { private static void setupConf(Configuration conf) {
conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1); conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1);
conf.setInt(MasterProcedureConstants.MASTER_URGENT_PROCEDURE_THREADS, 0);
conf.set(MasterCoprocessorHost.MASTER_COPROCESSOR_CONF_KEY, conf.set(MasterCoprocessorHost.MASTER_COPROCESSOR_CONF_KEY,
MasterObserverForTest.class.getName()); MasterObserverForTest.class.getName());
} }

View File

@ -63,6 +63,7 @@ public class TestMasterProcedureEvents {
private static void setupConf(Configuration conf) { private static void setupConf(Configuration conf) {
conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1); conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1);
conf.setInt(MasterProcedureConstants.MASTER_URGENT_PROCEDURE_THREADS, 0);
conf.setBoolean(WALProcedureStore.USE_HSYNC_CONF_KEY, false); conf.setBoolean(WALProcedureStore.USE_HSYNC_CONF_KEY, false);
} }

View File

@ -56,6 +56,7 @@ public class TestModifyNamespaceProcedure {
private static void setupConf(Configuration conf) { private static void setupConf(Configuration conf) {
conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1); conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1);
conf.setInt(MasterProcedureConstants.MASTER_URGENT_PROCEDURE_THREADS, 0);
} }
@BeforeClass @BeforeClass

View File

@ -60,6 +60,7 @@ public class TestProcedureAdmin {
private static void setupConf(Configuration conf) { private static void setupConf(Configuration conf) {
conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1); conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1);
conf.setInt(MasterProcedureConstants.MASTER_URGENT_PROCEDURE_THREADS, 0);
} }
@BeforeClass @BeforeClass

View File

@ -58,6 +58,7 @@ public class TestSafemodeBringsDownMaster {
private static void setupConf(Configuration conf) { private static void setupConf(Configuration conf) {
conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1); conf.setInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, 1);
conf.setInt(MasterProcedureConstants.MASTER_URGENT_PROCEDURE_THREADS, 0);
conf.set(BaseLoadBalancer.TABLES_ON_MASTER, "none"); conf.set(BaseLoadBalancer.TABLES_ON_MASTER, "none");
} }

View File

@ -169,7 +169,7 @@ public class TestUrgentProcedureWorker {
MasterProcedureScheduler scheduler = new MasterProcedureScheduler(pid -> null); MasterProcedureScheduler scheduler = new MasterProcedureScheduler(pid -> null);
procExec = new ProcedureExecutor<>(UTIL.getConfiguration(), new TestEnv(scheduler), procStore, procExec = new ProcedureExecutor<>(UTIL.getConfiguration(), new TestEnv(scheduler), procStore,
scheduler); scheduler);
procExec.init(1, false); procExec.init(1, 1, false);
procExec.startWorkers(); procExec.startWorkers();
} }