szetszwo commented on code in PR #10852:
URL: https://github.com/apache/ozone/pull/10852#discussion_r3641374745
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OzoneManagerLock.java:
##########
@@ -97,41 +93,172 @@ public class OzoneManagerLock implements IOzoneManagerLock
{
private static final Logger LOG =
LoggerFactory.getLogger(OzoneManagerLock.class);
- private final Map<Class<? extends Resource>,
- Pair<Map<Resource, Striped<ReadWriteLock>>, ResourceLockTracker>>
resourcelockMap;
+ private final ResourceLocks<LeveledResource> leveledResourceLocks;
+ private final ResourceLocks<DAGLeveledResource> dagLeveledResourceLocks;
- private OMLockMetrics omLockMetrics;
+ private final OMLockMetrics omLockMetrics = OMLockMetrics.create();
+
+ class ResourceLocks<R extends Resource> {
+ private final Map<Resource, Striped<ReentrantReadWriteLock>> lockMap;
+ private final ResourceLockTracker<R> tracker;
+
+ class Context {
+ private final R resource;
+ private final long startWaitingTimeNanos = Time.monotonicNowNanos();
+
+ Context(R r) {
+ this.resource = r;
+ }
+
+ R getResource() {
+ return resource;
+ }
+
+ long getWaitingTimeNanos() {
+ return Time.monotonicNowNanos() - startWaitingTimeNanos;
+ }
+
+ ResourceLockTracker<R> getTracker() {
+ return tracker;
+ }
+ }
+
+ ResourceLocks(Map<Resource, Striped<ReentrantReadWriteLock>> lockMap,
ResourceLockTracker<R> tracker) {
+ this.lockMap = lockMap;
+ this.tracker = tracker;
+ }
+
+ ResourceLockTracker<R> getTracker() {
+ return tracker;
+ }
+
+ private R cast(Resource resource) {
+ if (resource.getClass() != tracker.getResourceClass()) {
+ throw new IllegalArgumentException("Unexpected resource " +
resource.getClass()
+ + " for tracker " + tracker.getResourceClass());
+ }
+ @SuppressWarnings("unchecked")
+ final R r = (R) resource;
+ return r;
+ }
+
+ Context newContext(Resource resource) {
+ final R r = cast(resource);
+ tracker.clearLockDetails();
+ if (!tracker.canLockResource(r)) {
+ final String errorMessage = "Thread '" +
Thread.currentThread().getName() + "' cannot acquire "
+ + resource.getName() + " lock while holding " + getCurrentLocks()
+ " lock(s).";
+ LOG.error(errorMessage);
+ // TODO: change it to IllegalStateException
+ throw new RuntimeException(errorMessage);
+ }
+ return new Context(r);
+ }
+
+ private ReentrantReadWriteLock getLock(Resource resource, String... keys) {
+ return lockMap.get(resource).get(CompositeKey.combineKeys(keys));
+ }
+
+ private void acquireLock(ResourceLocks<?>.Context context, boolean isRead,
ReentrantReadWriteLock lock) {
+ if (isRead) {
+ lock.readLock().lock();
+ updateReadLockMetrics(context, lock);
+ } else {
+ lock.writeLock().lock();
+ updateWriteLockMetrics(context, lock);
+ }
+ }
+
+ OMLockDetails acquire(Resource resource, boolean isRead,
+ Function<Striped<ReentrantReadWriteLock>,
Iterable<ReentrantReadWriteLock>> getLocks) {
+ final Context context = newContext(resource);
+ for (ReentrantReadWriteLock lock :
getLocks.apply(lockMap.get(resource))) {
+ acquireLock(context, isRead, lock);
+ }
+ return tracker.lockResource(context.getResource());
+ }
+
+ OMLockDetails acquire(Resource resource, boolean isRead, String... keys) {
+ final Context context = newContext(resource);
+ acquireLock(context, isRead, getLock(resource, keys));
+ return tracker.lockResource(context.getResource());
+ }
+
+ void releaseLock(Resource resource, boolean isRead, ReentrantReadWriteLock
lock) {
+ if (isRead) {
+ lock.readLock().unlock();
+ updateReadUnlockMetrics(resource, lock);
+ } else {
+ boolean isWriteLocked = lock.isWriteLockedByCurrentThread();
+ lock.writeLock().unlock();
+ updateWriteUnlockMetrics(resource, lock, isWriteLocked);
+ }
+ }
+
+ OMLockDetails release(Resource resource, boolean isRead, String... keys) {
+ tracker.clearLockDetails();
+ ReentrantReadWriteLock lock = getLock(resource, keys);
+ releaseLock(resource, isRead, lock);
+ return tracker.unlockResource(cast(resource));
+ }
+
+ private OMLockDetails release(Resource resource, boolean isRead,
+ Function<Striped<ReentrantReadWriteLock>,
Iterable<ReentrantReadWriteLock>> getLock) {
+ tracker.clearLockDetails();
+ final Iterable<ReentrantReadWriteLock> i =
getLock.apply(lockMap.get(resource));
+ final List<ReentrantReadWriteLock> locks =
StreamSupport.stream(i.spliterator(), false)
+ .collect(Collectors.toList());
+ // Release locks in reverse order.
+ Collections.reverse(locks);
+ for (ReentrantReadWriteLock lock : locks) {
+ releaseLock(resource, isRead, lock);
+ }
+ return tracker.unlockResource(cast(resource));
+ }
+
+ List<String> getCurrentLocks() {
+ return tracker.getCurrentLockedResources()
+ .map(Resource::getName)
+ .collect(Collectors.toList());
+ }
+ }
/**
* Creates new OzoneManagerLock instance.
* @param conf Configuration object
*/
public OzoneManagerLock(ConfigurationSource conf) {
- omLockMetrics = OMLockMetrics.create();
- this.resourcelockMap = ImmutableMap.of(LeveledResource.class,
getLeveledLocks(conf), DAGLeveledResource.class,
- getFlatLocks(conf));
+ this.leveledResourceLocks =
newResourceLocks(LeveledResourceLockTracker.get(), conf);
+ this.dagLeveledResourceLocks =
newResourceLocks(DAGResourceLockTracker.get(), conf);
}
- private Pair<Map<Resource, Striped<ReadWriteLock>>, ResourceLockTracker>
getLeveledLocks(
- ConfigurationSource conf) {
- Map<LeveledResource, Striped<ReadWriteLock>> stripedLockMap = new
EnumMap<>(LeveledResource.class);
- for (LeveledResource r : LeveledResource.values()) {
- stripedLockMap.put(r, createStripeLock(r, conf));
+ private <T extends Resource> ResourceLocks<T> cast(ResourceLocks<?>
resourceLocks) {
+ @SuppressWarnings("unchecked")
+ final ResourceLocks<T> locks = (ResourceLocks<T>) resourceLocks;
+ return locks;
+ }
+
+ private <T extends Resource> ResourceLocks<T> getResourceLocks(Resource
instance) {
Review Comment:
That's a good point! ResourceLocks<?> seems better
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
