HDDS-12356. OM Gatekeeper Prototype

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/LockInfo.java

@@ -0,0 +1,72 @@
+/*
+ * 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.hdds.utils;
+
+import java.util.Objects;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * This class provides specifications about a lock's requirements.
+ */
+public final class LockInfo implements Comparable<LockInfo> {
+  private final String key;
+  private final boolean isWriteLock;
+
+  private LockInfo(String key, boolean isWriteLock) {
+    this.key = key;
+    this.isWriteLock = isWriteLock;
+  }
+
+  public static LockInfo writeLockInfo(String key) {
+    return new LockInfo(key, true);
+  }
+
+  public static LockInfo readLockInfo(String key) {
+    return new LockInfo(key, false);
+  }
+
+  public String getKey() {
+    return key;
+  }
+
+  public boolean isWriteLock() {
+    return isWriteLock;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof LockInfo)) {
+      return false;
+    }
+    LockInfo lockInfo = (LockInfo) o;
+    return isWriteLock == lockInfo.isWriteLock && Objects.equals(key, lockInfo.key);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(key, isWriteLock);
+  }
+
+  @Override
+  public int compareTo(@NotNull LockInfo other) {
+    return Integer.compare(hashCode(), other.hashCode());
+  }
+}

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmLockInfo.java

@@ -0,0 +1,106 @@
+/*
+ * 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.ozone.om.lock;
+
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+import org.apache.hadoop.hdds.utils.LockInfo;
+import org.apache.hadoop.ozone.OzoneConsts;
+
+/**
+ * This class represents the locking information that is required by an OM request.
+ * Requests can define whether they need volume, bucket, or key locks of read or write types.
+ * A request gatekeeper can handle these requirements to identify conflicting requests.
+ *
+ * Sample usage:
+ * {@link org.apache.hadoop.ozone.om.request.OMClientRequest} adds a required method {@code getLockInfo} which returns
+ * an instance of this class. All OM requests then implement this method and return an instance of this class to define
+ * their locking requirements.
+ *
+ * Note that this class defines the interface for each of the ~100 OM requests to communicate their requirements to the
+ * gatekeeper. Any breaking changes to this interface will have a large ripple effect.
+ */
+public final class OmLockInfo {
+  private final LockInfo volumeLock;
+  private final LockInfo bucketLock;
+  private final Set<LockInfo> keyLocks;
+
+  private OmLockInfo(Builder builder) {
+    volumeLock = builder.volumeLock;
+    bucketLock = builder.bucketLock;
+    keyLocks = builder.keyLocks;
+  }
+
+  public Optional<LockInfo> getVolumeLock() {
+    return Optional.ofNullable(volumeLock);
+  }
+
+  public Optional<LockInfo> getBucketLock() {
+    return Optional.ofNullable(bucketLock);
+  }
+
+  public Optional<Set<LockInfo>> getKeyLocks() {
+    return Optional.ofNullable(keyLocks);
+  }
+
+  /**
+   * Builds an {@link OmLockInfo} object with optional volume, bucket or key locks.
+   */
+  public static final class Builder {
+    private LockInfo volumeLock;
+    private LockInfo bucketLock;
+    private Set<LockInfo> keyLocks;
+
+    public Builder() {
+    }
+
+    public void addVolumeReadLock(String volume) {
+      volumeLock = LockInfo.writeLockInfo(volume);
+    }
+
+    public void addVolumeWriteLock(String volume) {
+      volumeLock = LockInfo.readLockInfo(volume);
+    }
+
+    public void addBucketReadLock(String volume, String bucket) {
+      bucketLock = LockInfo.readLockInfo(joinStrings(volume, bucket));
+    }
+
+    public void addBucketWriteLock(String volume, String bucket) {
+      bucketLock = LockInfo.writeLockInfo(joinStrings(volume, bucket));
+    }
+
+    // Currently there is no use case for key level read locks.
+    public void addKeyWriteLock(String volume, String bucket, String key) {
+      // Lazy init keys.
+      if (keyLocks == null) {
+        keyLocks = new HashSet<>();
+      }
+      keyLocks.add(LockInfo.writeLockInfo(joinStrings(volume, bucket, key)));
+    }
+
+    private String joinStrings(String... parts) {
+      return String.join(OzoneConsts.OZONE_URI_DELIMITER, parts);
+    }
+
+    public OmLockInfo build() {
+      return new OmLockInfo(this);
+    }
+  }
+}

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmRequestGatekeeper.java

@@ -0,0 +1,144 @@
+/*
+ * 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.ozone.om.lock;
+
+import com.google.common.util.concurrent.Striped;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.ListIterator;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+import org.apache.hadoop.hdds.utils.LockInfo;
+
+/**
+ * This class allows acquiring and releasing locks defined in an {@link OmLockInfo} instance.
+ * This can be used to allow non-conflicting OM requests to be executed in parallel.
+ *
+ * Sample Usage:
+ * {@link org.apache.hadoop.ozone.om.execution.OMExecutionFlow} instantiates this class and calls {@code lock} with the
+ * {@link OmLockInfo} of each request it is processing.
+ *
+ * Note that this class only provides one public method and will only be consumed by the OMExecutionFlow. Although this
+ * implementation uses 3 bins of striped locks, We can change the lock management implementation later as needed with
+ * little to no effect on any other classes.
+ */
+public class OmRequestGatekeeper {
+  private final Striped<ReadWriteLock> volumeLocks;
+  private final Striped<ReadWriteLock> bucketLocks;
+  // Currently only key level write locks are required. This lets us use the bulkGet API from the Striped class.
+  private final Striped<Lock> keyLocks;
+  private final long timeoutMillis;
+
+  // TODO Arbitrary values for now.
+  private static final int NUM_VOLUME_STRIPES = 128;
+  private static final int NUM_BUCKET_STRIPES = 1024;
+  private static final int NUM_KEY_STRIPES = 4096;
+
+  // If stats about the locks need to be added, they should be tracked here in the gatekeeper.
+
+  public OmRequestGatekeeper(long timeoutMillis) {
+    this.timeoutMillis = timeoutMillis;
+    volumeLocks = Striped.readWriteLock(NUM_VOLUME_STRIPES);
+    bucketLocks = Striped.readWriteLock(NUM_BUCKET_STRIPES);
+    keyLocks = Striped.lock(NUM_KEY_STRIPES);
+  }
+
+  /**
+   * Acquires all the locks specified by the {@code lockInfo} parameter. Caller is responsible for releasing the locks
+   * by closing the returned object. If an exception is thrown, no locks will be acquired when the method exits.
+   * @param lockInfo Defines the locks to acquire.
+   * @return A handle that can be closed to release the acquired locks.
+   * @throws InterruptedException If the thread is interrupted while waiting to acquire some of the locks.
+   * @throws TimeoutException If the timeout elapses while waiting to acquire any of the locks individually.
+   */
+  public AutoCloseable lock(OmLockInfo lockInfo) throws InterruptedException, TimeoutException {
+    // Common case is 1 volume, 1 bucket, and at most 2 key locks.
+    List<Lock> locks = new ArrayList<>(4);
+
+    Optional<LockInfo> optionalVolumeLock = lockInfo.getVolumeLock();
+    Optional<LockInfo> optionalBucketLock = lockInfo.getBucketLock();
+    Optional<Set<LockInfo>> optionalKeyLocks = lockInfo.getKeyLocks();
+
+    if (optionalVolumeLock.isPresent()) {
+      LockInfo volumeLockInfo = optionalVolumeLock.get();
+      if (volumeLockInfo.isWriteLock()) {
+        locks.add(volumeLocks.get(volumeLockInfo.getKey()).writeLock());
+      } else {
+        locks.add(volumeLocks.get(volumeLockInfo.getKey()).readLock());
+      }
+    }
+
+    if (optionalBucketLock.isPresent()) {
+      LockInfo bucketLockInfo = optionalBucketLock.get();
+      if (bucketLockInfo.isWriteLock()) {
+        locks.add(bucketLocks.get(bucketLockInfo.getKey()).writeLock());
+      } else {
+        locks.add(bucketLocks.get(bucketLockInfo.getKey()).readLock());
+      }
+    }
+
+    if (optionalKeyLocks.isPresent()) {
+      for (Lock keyLock: keyLocks.bulkGet(optionalKeyLocks.get())) {
+        locks.add(keyLock);
+      }
+    }
+
+    acquireLocks(locks);
+    return () -> releaseLocks(locks);
+  }
+
+  /*
+  Optional: If we want more diagnostic info on the type of lock that failed to be acquired (volume, bucket, or key),
+  We can make the parameter a list of objects that wrap the Lock with information about its type.
+
+  Note that logging the specific volume, bucket or keys this lock was trying to acquire is not helpful and
+  misleading because collisions within the stripe lock might mean we are blocked on a request for a completely
+  different part of the namespace.
+  Obtaining the thread ID that we were waiting on would be more useful, but there is no easy way to do that.
+   */
+  private void acquireLocks(List<Lock> locks) throws TimeoutException, InterruptedException {
+    List<Lock> acquiredLocks = new ArrayList<>(locks.size());
+    for (Lock lock: locks) {
+      if (lock.tryLock(timeoutMillis, TimeUnit.MILLISECONDS)) {
+        try {
+          acquiredLocks.add(lock);
+        } catch (Throwable e) {
+          // We acquired this lock but were unable to add it to our acquired locks list.
+          lock.unlock();
+          releaseLocks(acquiredLocks);
+          throw e;
+        }
+      } else {
+        releaseLocks(acquiredLocks);
+        throw new TimeoutException("Failed to acquire lock after the given timeout.");
+      }
+    }
+  }
+
+  private void releaseLocks(List<Lock> locks) {
+    ListIterator<Lock> reverseIterator = locks.listIterator();
+    while (reverseIterator.hasPrevious()) {
+      Lock lock = reverseIterator.previous();
+      lock.unlock();
+    }
+  }
+}