Source code

001/**
-002 * Licensed to the Apache Software Foundation (ASF) under one
-003 * or more contributor license agreements.  See the NOTICE file
-004 * distributed with this work for additional information
-005 * regarding copyright ownership.  The ASF licenses this file
-006 * to you under the Apache License, Version 2.0 (the
-007 * "License"); you may not use this file except in compliance
-008 * with the License.  You may obtain a copy of the License at
-009 *
-010 *     http://www.apache.org/licenses/LICENSE-2.0
-011 *
-012 * Unless required by applicable law or agreed to in writing, software
-013 * distributed under the License is distributed on an "AS IS" BASIS,
-014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-015 * See the License for the specific language governing permissions and
-016 * limitations under the License.
-017 */
-018
-019package org.apache.hadoop.hbase.master.procedure;
-020
-021import java.io.IOException;
-022import java.util.ArrayDeque;
-023import java.util.Deque;
-024import java.util.concurrent.locks.Condition;
-025import java.util.concurrent.locks.ReentrantLock;
-026
-027import org.apache.commons.logging.Log;
-028import org.apache.commons.logging.LogFactory;
-029import org.apache.hadoop.conf.Configuration;
-030import org.apache.hadoop.hbase.ServerName;
-031import org.apache.hadoop.hbase.TableName;
-032import org.apache.hadoop.hbase.TableExistsException;
-033import org.apache.hadoop.hbase.TableNotFoundException;
-034import org.apache.hadoop.hbase.classification.InterfaceAudience;
-035import org.apache.hadoop.hbase.classification.InterfaceStability;
-036import org.apache.hadoop.hbase.procedure2.Procedure;
-037import org.apache.hadoop.hbase.procedure2.ProcedureFairRunQueues;
-038import org.apache.hadoop.hbase.procedure2.ProcedureRunnableSet;
-039import org.apache.hadoop.hbase.master.TableLockManager;
-040import org.apache.hadoop.hbase.master.TableLockManager.TableLock;
-041import org.apache.hadoop.hbase.master.procedure.TableProcedureInterface.TableOperationType;
-042
-043/**
-044 * ProcedureRunnableSet for the Master Procedures.
-045 * This RunnableSet tries to provide to the ProcedureExecutor procedures
-046 * that can be executed without having to wait on a lock.
-047 * Most of the master operations can be executed concurrently, if they
-048 * are operating on different tables (e.g. two create table can be performed
-049 * at the same, time assuming table A and table B) or against two different servers; say
-050 * two servers that crashed at about the same time.
-051 *
-052 * <p>Each procedure should implement an interface providing information for this queue.
-053 * for example table related procedures should implement TableProcedureInterface.
-054 * each procedure will be pushed in its own queue, and based on the operation type
-055 * we may take smarter decision. e.g. we can abort all the operations preceding
-056 * a delete table, or similar.
-057 */
-058@InterfaceAudience.Private
-059@InterfaceStability.Evolving
-060public class MasterProcedureQueue implements ProcedureRunnableSet {
-061  private static final Log LOG = LogFactory.getLog(MasterProcedureQueue.class);
-062
-063  // Two queues to ensure that server procedures run ahead of table precedures always.
-064  private final ProcedureFairRunQueues<TableName, RunQueue> tableFairQ;
-065  /**
-066   * Rely on basic fair q. ServerCrashProcedure will yield if meta is not assigned. This way, the
-067   * server that was carrying meta should rise to the top of the queue (this is how it used to
-068   * work when we had handlers and ServerShutdownHandler ran). TODO: special handling of servers
-069   * that were carrying system tables on crash; do I need to have these servers have priority?
-070   *
-071   * <p>Apart from the special-casing of meta and system tables, fairq is what we want
-072   */
-073  private final ProcedureFairRunQueues<ServerName, RunQueue> serverFairQ;
-074
-075  private final ReentrantLock lock = new ReentrantLock();
-076  private final Condition waitCond = lock.newCondition();
-077  private final TableLockManager lockManager;
-078
-079  private final int metaTablePriority;
-080  private final int userTablePriority;
-081  private final int sysTablePriority;
-082  private static final int DEFAULT_SERVER_PRIORITY = 1;
-083
-084  /**
-085   * Keeps count across server and table queues.
-086   */
-087  private int queueSize;
-088
-089  public MasterProcedureQueue(final Configuration conf, final TableLockManager lockManager) {
-090    this.tableFairQ = new ProcedureFairRunQueues<TableName, RunQueue>(1);
-091    this.serverFairQ = new ProcedureFairRunQueues<ServerName, RunQueue>(1);
-092    this.lockManager = lockManager;
-093
-094    // TODO: should this be part of the HTD?
-095    metaTablePriority = conf.getInt("hbase.master.procedure.queue.meta.table.priority", 3);
-096    sysTablePriority = conf.getInt("hbase.master.procedure.queue.system.table.priority", 2);
-097    userTablePriority = conf.getInt("hbase.master.procedure.queue.user.table.priority", 1);
-098  }
-099
-100  @Override
-101  public void addFront(final Procedure proc) {
-102    lock.lock();
-103    try {
-104      getRunQueueOrCreate(proc).addFront(proc);
-105      queueSize++;
-106      waitCond.signal();
-107    } finally {
-108      lock.unlock();
-109    }
-110  }
-111
-112  @Override
-113  public void addBack(final Procedure proc) {
-114    lock.lock();
-115    try {
-116      getRunQueueOrCreate(proc).addBack(proc);
-117      queueSize++;
-118      waitCond.signal();
-119    } finally {
-120      lock.unlock();
-121    }
-122  }
-123
-124  @Override
-125  public void yield(final Procedure proc) {
-126    addBack(proc);
-127  }
-128
-129  @Override
-130  @edu.umd.cs.findbugs.annotations.SuppressWarnings("WA_AWAIT_NOT_IN_LOOP")
-131  public Long poll() {
-132    Long pollResult = null;
-133    lock.lock();
-134    try {
-135      if (queueSize == 0) {
-136        waitCond.await();
-137        if (queueSize == 0) {
-138          return null;
-139        }
-140      }
-141      // For now, let server handling have precedence over table handling; presumption is that it
-142      // is more important handling crashed servers than it is running the
-143      // enabling/disabling tables, etc.
-144      pollResult = doPoll(serverFairQ.poll());
-145      if (pollResult == null) {
-146        pollResult = doPoll(tableFairQ.poll());
-147      }
-148    } catch (InterruptedException e) {
-149      Thread.currentThread().interrupt();
-150    } finally {
-151      lock.unlock();
-152    }
-153    return pollResult;
-154  }
-155
-156  private Long doPoll(final RunQueue rq) {
-157    if (rq == null || !rq.isAvailable()) return null;
-158    this.queueSize--;
-159    return rq.poll();
-160  }
-161
-162  @Override
-163  public void signalAll() {
-164    lock.lock();
-165    try {
-166      waitCond.signalAll();
-167    } finally {
-168      lock.unlock();
-169    }
-170  }
-171
-172  @Override
-173  public void clear() {
-174    lock.lock();
-175    try {
-176      serverFairQ.clear();
-177      tableFairQ.clear();
-178      queueSize = 0;
-179    } finally {
-180      lock.unlock();
-181    }
-182  }
-183
-184  @Override
-185  public int size() {
-186    lock.lock();
-187    try {
-188      return queueSize;
-189    } finally {
-190      lock.unlock();
-191    }
-192  }
-193
-194  @Override
-195  public String toString() {
-196    lock.lock();
-197    try {
-198      return "MasterProcedureQueue size=" + queueSize + ": tableFairQ: " + tableFairQ +
-199        ", serverFairQ: " + serverFairQ;
-200    } finally {
-201      lock.unlock();
-202    }
-203  }
-204
-205  @Override
-206  public void completionCleanup(Procedure proc) {
-207    if (proc instanceof TableProcedureInterface) {
-208      TableProcedureInterface iProcTable = (TableProcedureInterface)proc;
-209      boolean tableDeleted;
-210      if (proc.hasException()) {
-211        IOException procEx =  proc.getException().unwrapRemoteException();
-212        if (iProcTable.getTableOperationType() == TableOperationType.CREATE) {
-213          // create failed because the table already exist
-214          tableDeleted = !(procEx instanceof TableExistsException);
-215        } else {
-216          // the operation failed because the table does not exist
-217          tableDeleted = (procEx instanceof TableNotFoundException);
-218        }
-219      } else {
-220        // the table was deleted
-221        tableDeleted = (iProcTable.getTableOperationType() == TableOperationType.DELETE);
-222      }
-223      if (tableDeleted) {
-224        markTableAsDeleted(iProcTable.getTableName());
-225      }
-226    }
-227    // No cleanup for ServerProcedureInterface types, yet.
-228  }
-229
-230  private RunQueue getRunQueueOrCreate(final Procedure proc) {
-231    if (proc instanceof TableProcedureInterface) {
-232      final TableName table = ((TableProcedureInterface)proc).getTableName();
-233      return getRunQueueOrCreate(table);
-234    }
-235    if (proc instanceof ServerProcedureInterface) {
-236      return getRunQueueOrCreate((ServerProcedureInterface)proc);
-237    }
-238    // TODO: at the moment we only have Table and Server procedures
-239    // if you are implementing a non-table/non-server procedure, you have two options: create
-240    // a group for all the non-table/non-server procedures or try to find a key for your
-241    // non-table/non-server procedures and implement something similar to the TableRunQueue.
-242    throw new UnsupportedOperationException("RQs for non-table procedures are not implemented yet");
-243  }
-244
-245  private TableRunQueue getRunQueueOrCreate(final TableName table) {
-246    final TableRunQueue queue = getRunQueue(table);
-247    if (queue != null) return queue;
-248    return (TableRunQueue)tableFairQ.add(table, createTableRunQueue(table));
-249  }
-250
-251  private ServerRunQueue getRunQueueOrCreate(final ServerProcedureInterface spi) {
-252    final ServerRunQueue queue = getRunQueue(spi.getServerName());
-253    if (queue != null) return queue;
-254    return (ServerRunQueue)serverFairQ.add(spi.getServerName(), createServerRunQueue(spi));
-255  }
-256
-257  private TableRunQueue createTableRunQueue(final TableName table) {
-258    int priority = userTablePriority;
-259    if (table.equals(TableName.META_TABLE_NAME)) {
-260      priority = metaTablePriority;
-261    } else if (table.isSystemTable()) {
-262      priority = sysTablePriority;
-263    }
-264    return new TableRunQueue(priority);
-265  }
-266
-267  private ServerRunQueue createServerRunQueue(final ServerProcedureInterface spi) {
-268    return new ServerRunQueue(DEFAULT_SERVER_PRIORITY);
-269  }
-270
-271  private TableRunQueue getRunQueue(final TableName table) {
-272    return (TableRunQueue)tableFairQ.get(table);
-273  }
-274
-275  private ServerRunQueue getRunQueue(final ServerName sn) {
-276    return (ServerRunQueue)serverFairQ.get(sn);
-277  }
-278
-279  /**
-280   * Try to acquire the write lock on the specified table.
-281   * other operations in the table-queue will be executed after the lock is released.
-282   * @param table Table to lock
-283   * @param purpose Human readable reason for locking the table
-284   * @return true if we were able to acquire the lock on the table, otherwise false.
-285   */
-286  public boolean tryAcquireTableExclusiveLock(final TableName table, final String purpose) {
-287    return getRunQueueOrCreate(table).tryExclusiveLock(lockManager, table, purpose);
-288  }
-289
-290  /**
-291   * Release the write lock taken with tryAcquireTableWrite()
-292   * @param table the name of the table that has the write lock
-293   */
-294  public void releaseTableExclusiveLock(final TableName table) {
-295    getRunQueue(table).releaseExclusiveLock(lockManager, table);
-296  }
-297
-298  /**
-299   * Try to acquire the read lock on the specified table.
-300   * other read operations in the table-queue may be executed concurrently,
-301   * otherwise they have to wait until all the read-locks are released.
-302   * @param table Table to lock
-303   * @param purpose Human readable reason for locking the table
-304   * @return true if we were able to acquire the lock on the table, otherwise false.
-305   */
-306  public boolean tryAcquireTableSharedLock(final TableName table, final String purpose) {
-307    return getRunQueueOrCreate(table).trySharedLock(lockManager, table, purpose);
-308  }
-309
-310  /**
-311   * Release the read lock taken with tryAcquireTableRead()
-312   * @param table the name of the table that has the read lock
-313   */
-314  public void releaseTableSharedLock(final TableName table) {
-315    getRunQueue(table).releaseSharedLock(lockManager, table);
-316  }
-317
-318  /**
-319   * Try to acquire the write lock on the specified server.
-320   * @see #releaseServerExclusiveLock(ServerProcedureInterface)
-321   * @param spi Server to lock
-322   * @return true if we were able to acquire the lock on the server, otherwise false.
-323   */
-324  public boolean tryAcquireServerExclusiveLock(final ServerProcedureInterface spi) {
-325    return getRunQueueOrCreate(spi).tryExclusiveLock();
-326  }
-327
-328  /**
-329   * Release the write lock
-330   * @see #tryAcquireServerExclusiveLock(ServerProcedureInterface)
-331   * @param spi the server that has the write lock
-332   */
-333  public void releaseServerExclusiveLock(final ServerProcedureInterface spi) {
-334    getRunQueue(spi.getServerName()).releaseExclusiveLock();
-335  }
-336
-337  /**
-338   * Try to acquire the read lock on the specified server.
-339   * @see #releaseServerSharedLock(ServerProcedureInterface)
-340   * @param spi Server to lock
-341   * @return true if we were able to acquire the lock on the server, otherwise false.
-342   */
-343  public boolean tryAcquireServerSharedLock(final ServerProcedureInterface spi) {
-344    return getRunQueueOrCreate(spi).trySharedLock();
-345  }
-346
-347  /**
-348   * Release the read lock taken
-349   * @see #tryAcquireServerSharedLock(ServerProcedureInterface)
-350   * @param spi the server that has the read lock
-351   */
-352  public void releaseServerSharedLock(final ServerProcedureInterface spi) {
-353    getRunQueue(spi.getServerName()).releaseSharedLock();
-354  }
-355
-356  /**
-357   * Tries to remove the queue and the table-lock of the specified table.
-358   * If there are new operations pending (e.g. a new create),
-359   * the remove will not be performed.
-360   * @param table the name of the table that should be marked as deleted
-361   * @return true if deletion succeeded, false otherwise meaning that there are
-362   *    other new operations pending for that table (e.g. a new create).
-363   */
-364  protected boolean markTableAsDeleted(final TableName table) {
-365    TableRunQueue queue = getRunQueue(table);
-366    if (queue != null) {
-367      lock.lock();
-368      try {
-369        if (queue.isEmpty() && queue.acquireDeleteLock()) {
-370          tableFairQ.remove(table);
-371
-372          // Remove the table lock
-373          try {
-374            lockManager.tableDeleted(table);
-375          } catch (IOException e) {
-376            LOG.warn("Received exception from TableLockManager.tableDeleted:", e); //not critical
-377          }
-378        } else {
-379          // TODO: If there are no create, we can drop all the other ops
-380          return false;
-381        }
-382      } finally {
-383        lock.unlock();
-384      }
-385    }
-386    return true;
-387  }
-388
-389  private interface RunQueue extends ProcedureFairRunQueues.FairObject {
-390    void addFront(Procedure proc);
-391    void addBack(Procedure proc);
-392    Long poll();
-393    boolean acquireDeleteLock();
-394  }
-395
-396  /**
-397   * Base abstract class for RunQueue implementations.
-398   * Be careful honoring synchronizations in subclasses. In here we protect access but if you are
-399   * acting on a state found in here, be sure dependent code keeps synchronization.
-400   * Implements basic in-memory read/write locking mechanism to prevent procedure steps being run
-401   * in parallel.
-402   */
-403  private static abstract class AbstractRunQueue implements RunQueue {
-404    // All modification of runnables happens with #lock held.
-405    private final Deque<Long> runnables = new ArrayDeque<Long>();
-406    private final int priority;
-407    private boolean exclusiveLock = false;
-408    private int sharedLock = 0;
-409
-410    public AbstractRunQueue(int priority) {
-411      this.priority = priority;
-412    }
-413
-414    boolean isEmpty() {
-415      return this.runnables.isEmpty();
-416    }
-417
-418    @Override
-419    public boolean isAvailable() {
-420      synchronized (this) {
-421        return !exclusiveLock && !runnables.isEmpty();
-422      }
-423    }
-424
-425    @Override
-426    public int getPriority() {
-427      return this.priority;
-428    }
-429
-430    @Override
-431    public void addFront(Procedure proc) {
-432      this.runnables.addFirst(proc.getProcId());
-433    }
-434
-435    @Override
-436    public void addBack(Procedure proc) {
-437      this.runnables.addLast(proc.getProcId());
-438    }
-439
-440    @Override
-441    public Long poll() {
-442      return this.runnables.poll();
-443    }
-444
-445    @Override
-446    public synchronized boolean acquireDeleteLock() {
-447      return tryExclusiveLock();
-448    }
-449
-450    public synchronized boolean isLocked() {
-451      return isExclusiveLock() || sharedLock > 0;
-452    }
-453
-454    public synchronized boolean isExclusiveLock() {
-455      return this.exclusiveLock;
-456    }
-457
-458    public synchronized boolean trySharedLock() {
-459      if (isExclusiveLock()) return false;
-460      sharedLock++;
-461      return true;
-462    }
-463
-464    public synchronized void releaseSharedLock() {
-465      sharedLock--;
-466    }
-467
-468    /**
-469     * @return True if only one instance of a shared lock outstanding.
-470     */
-471    synchronized boolean isSingleSharedLock() {
-472      return sharedLock == 1;
-473    }
-474
-475    public synchronized boolean tryExclusiveLock() {
-476      if (isLocked()) return false;
-477      exclusiveLock = true;
-478      return true;
-479    }
-480
-481    public synchronized void releaseExclusiveLock() {
-482      exclusiveLock = false;
-483    }
-484
-485    @Override
-486    public String toString() {
-487      return this.runnables.toString();
-488    }
-489  }
-490
-491  /**
-492   * Run Queue for Server procedures.
-493   */
-494  private static class ServerRunQueue extends AbstractRunQueue {
-495    public ServerRunQueue(int priority) {
-496      super(priority);
-497    }
-498  }
-499
-500  /**
-501   * Run Queue for a Table. It contains a read-write lock that is used by the
-502   * MasterProcedureQueue to decide if we should fetch an item from this queue
-503   * or skip to another one which will be able to run without waiting for locks.
-504   */
-505  private static class TableRunQueue extends AbstractRunQueue {
-506    private TableLock tableLock = null;
-507
-508    public TableRunQueue(int priority) {
-509      super(priority);
-510    }
-511
-512    // TODO: Improve run-queue push with TableProcedureInterface.getType()
-513    //       we can take smart decisions based on the type of the operation (e.g. create/delete)
-514    @Override
-515    public void addBack(final Procedure proc) {
-516      super.addBack(proc);
-517    }
-518
-519    public synchronized boolean trySharedLock(final TableLockManager lockManager,
-520        final TableName tableName, final String purpose) {
-521      if (isExclusiveLock()) return false;
-522
-523      // Take zk-read-lock
-524      tableLock = lockManager.readLock(tableName, purpose);
-525      try {
-526        tableLock.acquire();
-527      } catch (IOException e) {
-528        LOG.error("failed acquire read lock on " + tableName, e);
-529        tableLock = null;
-530        return false;
-531      }
-532      trySharedLock();
-533      return true;
-534    }
-535
-536    public synchronized void releaseSharedLock(final TableLockManager lockManager,
-537        final TableName tableName) {
-538      releaseTableLock(lockManager, isSingleSharedLock());
-539      releaseSharedLock();
-540    }
-541
-542    public synchronized boolean tryExclusiveLock(final TableLockManager lockManager,
-543        final TableName tableName, final String purpose) {
-544      if (isLocked()) return false;
-545      // Take zk-write-lock
-546      tableLock = lockManager.writeLock(tableName, purpose);
-547      try {
-548        tableLock.acquire();
-549      } catch (IOException e) {
-550        LOG.error("failed acquire write lock on " + tableName, e);
-551        tableLock = null;
-552        return false;
-553      }
-554      tryExclusiveLock();
-555      return true;
-556    }
-557
-558    public synchronized void releaseExclusiveLock(final TableLockManager lockManager,
-559        final TableName tableName) {
-560      releaseTableLock(lockManager, true);
-561      releaseExclusiveLock();
-562    }
-563
-564    private void releaseTableLock(final TableLockManager lockManager, boolean reset) {
-565      for (int i = 0; i < 3; ++i) {
-566        try {
-567          tableLock.release();
-568          if (reset) {
-569            tableLock = null;
-570          }
-571          break;
-572        } catch (IOException e) {
-573          LOG.warn("Could not release the table write-lock", e);
-574        }
-575      }
-576    }
-577  }
-578}
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-