From 4924a19e353c3fc574af8e996941d30a5df11cdf Mon Sep 17 00:00:00 2001 From: Ethan Rose Date: Fri, 28 Mar 2025 16:29:16 -0400 Subject: [PATCH 1/4] Initial gatekeeper prototype --- .../apache/hadoop/hdds/utils/LockInfo.java | 72 ++++++++++ .../hadoop/ozone/om/lock/OmLockInfo.java | 98 ++++++++++++++ .../ozone/om/lock/OmRequestGatekeeper.java | 127 ++++++++++++++++++ 3 files changed, 297 insertions(+) create mode 100644 hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/LockInfo.java create mode 100644 hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmLockInfo.java create mode 100644 hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmRequestGatekeeper.java diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/LockInfo.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/LockInfo.java new file mode 100644 index 000000000000..42674f6998a1 --- /dev/null +++ b/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 { + 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()); + } +} diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmLockInfo.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmLockInfo.java new file mode 100644 index 000000000000..7ff70225c4ed --- /dev/null +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmLockInfo.java @@ -0,0 +1,98 @@ +/* + * 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. + */ +public final class OmLockInfo { + private final LockInfo volumeLock; + private final LockInfo bucketLock; + private final Set keyLocks; + + private OmLockInfo(Builder builder) { + volumeLock = builder.volumeLock; + bucketLock = builder.bucketLock; + keyLocks = builder.keyLocks; + } + + public Optional getVolumeLock() { + return Optional.ofNullable(volumeLock); + } + + public Optional getBucketLock() { + return Optional.ofNullable(bucketLock); + } + + public Optional> 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 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); + } + } +} diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmRequestGatekeeper.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmRequestGatekeeper.java new file mode 100644 index 000000000000..51a6852f2387 --- /dev/null +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmRequestGatekeeper.java @@ -0,0 +1,127 @@ +/* + * 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. + */ +public class OmRequestGatekeeper { + private final Striped volumeLocks; + private final Striped bucketLocks; + // Currently only key level write locks are required. This lets us use the bulkGet API from the Striped class. + private final Striped 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 locks = new ArrayList<>(4); + + Optional optionalVolumeLock = lockInfo.getVolumeLock(); + Optional optionalBucketLock = lockInfo.getBucketLock(); + Optional> 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); + } + + private void acquireLocks(List locks) throws TimeoutException, InterruptedException { + List 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 locks) { + ListIterator reverseIterator = locks.listIterator(); + while (reverseIterator.hasPrevious()) { + Lock lock = reverseIterator.previous(); + lock.unlock(); + } + } +} From 4303a059c6e0be8a552fac0750e5fe24071dff0d Mon Sep 17 00:00:00 2001 From: Ethan Rose Date: Fri, 28 Mar 2025 18:13:59 -0400 Subject: [PATCH 2/4] Add comment on error handling --- .../apache/hadoop/ozone/om/lock/OmRequestGatekeeper.java | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmRequestGatekeeper.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmRequestGatekeeper.java index 51a6852f2387..3e7f448afccb 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmRequestGatekeeper.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmRequestGatekeeper.java @@ -98,6 +98,15 @@ public AutoCloseable lock(OmLockInfo lockInfo) throws InterruptedException, Time 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 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 locks) throws TimeoutException, InterruptedException { List acquiredLocks = new ArrayList<>(locks.size()); for (Lock lock: locks) { From 6246ca642b29548d23308cc6e15b6521e3e227e6 Mon Sep 17 00:00:00 2001 From: Ethan Rose Date: Fri, 28 Mar 2025 18:19:17 -0400 Subject: [PATCH 3/4] Add sample usage --- .../java/org/apache/hadoop/ozone/om/lock/OmLockInfo.java | 5 +++++ .../org/apache/hadoop/ozone/om/lock/OmRequestGatekeeper.java | 4 ++++ 2 files changed, 9 insertions(+) diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmLockInfo.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmLockInfo.java index 7ff70225c4ed..f42ccd11406f 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmLockInfo.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmLockInfo.java @@ -27,6 +27,11 @@ * 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. */ public final class OmLockInfo { private final LockInfo volumeLock; diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmRequestGatekeeper.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmRequestGatekeeper.java index 3e7f448afccb..c83a6c42f84e 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmRequestGatekeeper.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmRequestGatekeeper.java @@ -32,6 +32,10 @@ /** * 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. */ public class OmRequestGatekeeper { private final Striped volumeLocks; From f2e903cd39c96bb2e4879c648044c867188ad177 Mon Sep 17 00:00:00 2001 From: Ethan Rose Date: Fri, 28 Mar 2025 18:29:35 -0400 Subject: [PATCH 4/4] Update comments --- .../java/org/apache/hadoop/ozone/om/lock/OmLockInfo.java | 3 +++ .../apache/hadoop/ozone/om/lock/OmRequestGatekeeper.java | 6 +++++- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmLockInfo.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmLockInfo.java index f42ccd11406f..d21786c84df1 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmLockInfo.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmLockInfo.java @@ -32,6 +32,9 @@ * {@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; diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmRequestGatekeeper.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmRequestGatekeeper.java index c83a6c42f84e..d56c36f87b93 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmRequestGatekeeper.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OmRequestGatekeeper.java @@ -36,6 +36,10 @@ * 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 volumeLocks; @@ -107,7 +111,7 @@ public AutoCloseable lock(OmLockInfo lockInfo) throws InterruptedException, Time 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 we are blocked on a request for a completely + 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. */