devmadhuu commented on code in PR #9243: URL: https://github.com/apache/ozone/pull/9243#discussion_r2503137625
########## hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/util/ParallelTableIteratorOperation.java: ########## @@ -0,0 +1,218 @@ +/* + * 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.recon.tasks.util; + +import java.io.Closeable; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashSet; +import java.util.LinkedList; +import java.util.List; +import java.util.Objects; +import java.util.Queue; +import java.util.Set; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; +import java.util.stream.Collectors; +import org.apache.hadoop.hdds.utils.db.Codec; +import org.apache.hadoop.hdds.utils.db.RDBStore; +import org.apache.hadoop.hdds.utils.db.Table; +import org.apache.hadoop.hdds.utils.db.TableIterator; +import org.apache.hadoop.ozone.om.OMMetadataManager; +import org.rocksdb.LiveFileMetaData; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Class to iterate through a table in parallel by breaking table into multiple iterators. + */ +public class ParallelTableIteratorOperation<K extends Comparable<K>, V> implements Closeable { + private final Table<K, V> table; + private final Codec<K> keyCodec; + private final ExecutorService iteratorExecutor; + private final ExecutorService valueExecutors; + private final int maxNumberOfVals; + private final OMMetadataManager metadataManager; + private final int maxIteratorTasks; + private final int maxWorkerTasks; + private final long logCountThreshold; + + private static final Logger LOG = LoggerFactory.getLogger(ParallelTableIteratorOperation.class); + public ParallelTableIteratorOperation(OMMetadataManager metadataManager, Table<K, V> table, Codec<K> keyCodec, + int iteratorCount, int workerCount, int maxNumberOfValsInMemory, + long logThreshold) { + this.table = table; + this.keyCodec = keyCodec; + this.metadataManager = metadataManager; + this.maxIteratorTasks = 2 * iteratorCount; + this.maxWorkerTasks = workerCount * 2; + this.iteratorExecutor = new ThreadPoolExecutor(iteratorCount, iteratorCount, 1, TimeUnit.MINUTES, + new ArrayBlockingQueue<>(iteratorCount * 2)); + this.valueExecutors = new ThreadPoolExecutor(workerCount, workerCount, 1, TimeUnit.MINUTES, + new ArrayBlockingQueue<>(workerCount * 2)); + this.maxNumberOfVals = Math.max(10, maxNumberOfValsInMemory / (workerCount)); + this.logCountThreshold = logThreshold; + } + + + private List<K> getBounds(K startKey, K endKey) throws IOException { + RDBStore store = (RDBStore) this.metadataManager.getStore(); + List<LiveFileMetaData> sstFiles = store.getDb().getSstFileList(); + Set<K> keys = new HashSet<>(); + String tableName = table.getName(); + byte[] tableNameBytes = tableName.getBytes(StandardCharsets.UTF_8); + for (LiveFileMetaData sstFile : sstFiles) { + // Filter SST files by column family to get bounds only for this specific table + if (Arrays.equals(sstFile.columnFamilyName(), tableNameBytes)) { + keys.add(this.keyCodec.fromPersistedFormat(sstFile.smallestKey())); + keys.add(this.keyCodec.fromPersistedFormat(sstFile.largestKey())); + } + } + if (startKey != null) { + keys.add(startKey); + } + if (endKey != null) { + keys.add(endKey); + } + + return keys.stream().sorted().filter(Objects::nonNull) + .filter(key -> startKey == null || key.compareTo(startKey) >= 0) + .filter(key -> endKey == null || endKey.compareTo(key) >= 0) + .collect(Collectors.toList()); + } + + private void waitForQueueSize(Queue<Future<?>> futures, int expectedSize) + throws ExecutionException, InterruptedException { + while (!futures.isEmpty() && futures.size() > expectedSize) { + Future<?> f = futures.poll(); + f.get(); + } + } + + public void performTaskOnTableVals(String taskName, K startKey, K endKey, + Function<Table.KeyValue<K, V>, Void> keyOperation) throws IOException, ExecutionException, InterruptedException { + List<K> bounds = getBounds(startKey, endKey); + + // Fallback for small tables (no SST files yet - data only in memtable) + if (bounds.size() < 2) { + try (TableIterator<K, ? extends Table.KeyValue<K, V>> iter = table.iterator()) { + if (startKey != null) { + iter.seek(startKey); + } + while (iter.hasNext()) { + Table.KeyValue<K, V> kv = iter.next(); + if (endKey != null && kv.getKey().compareTo(endKey) > 0) { + break; + } + keyOperation.apply(kv); + } + } + return; + } + + Queue<Future<?>> iterFutures = new LinkedList<>(); + Queue<Future<?>> workerFutures = new ConcurrentLinkedQueue<>(); + AtomicLong keyCounter = new AtomicLong(); + AtomicLong prevLogCounter = new AtomicLong(); + for (int idx = 1; idx < bounds.size(); idx++) { + K beg = bounds.get(idx - 1); + K end = bounds.get(idx); + boolean inclusive = idx == bounds.size() - 1; + waitForQueueSize(iterFutures, maxIteratorTasks - 1); + iterFutures.add(iteratorExecutor.submit(() -> { + try (TableIterator<K, ? extends Table.KeyValue<K, V>> iter = table.iterator()) { + iter.seek(beg); + while (iter.hasNext()) { + List<Table.KeyValue<K, V>> keyValues = new ArrayList<>(); + boolean reachedEnd = false; + while (iter.hasNext()) { + Table.KeyValue<K, V> kv = iter.next(); + K key = kv.getKey(); + + // Check if key is within this segment's range + boolean withinBounds; + if (inclusive) { + // Last segment: include everything from beg onwards (or until endKey if specified) + withinBounds = (endKey == null || key.compareTo(endKey) <= 0); + } else { + // Middle segment: include keys in range [beg, end) + withinBounds = key.compareTo(end) < 0; + } + + if (withinBounds) { + keyValues.add(kv); + } else { + reachedEnd = true; + break; + } + if (keyValues.size() >= maxNumberOfVals) { + break; + } + } + if (!keyValues.isEmpty()) { + waitForQueueSize(workerFutures, maxWorkerTasks - 10); + workerFutures.add(valueExecutors.submit(() -> { + for (Table.KeyValue<K, V> kv : keyValues) { + keyOperation.apply(kv); + } + keyCounter.addAndGet(keyValues.size()); + if (keyCounter.get() - prevLogCounter.get() > logCountThreshold) { + synchronized (keyCounter) { + if (keyCounter.get() - prevLogCounter.get() > logCountThreshold) { + long cnt = keyCounter.get(); + LOG.info("Iterated through {} keys while performing task: {}", keyCounter.get(), taskName); + prevLogCounter.set(cnt); + } + } + } + })); + } + if (reachedEnd) { + break; + } + } + } catch (IOException | ExecutionException | InterruptedException e) { + throw new RuntimeException(e); + } + })); + } + waitForQueueSize(iterFutures, 0); + waitForQueueSize(workerFutures, 0); + + // Log final stats + LOG.info("{}: Parallel iteration completed - Total keys processed: {}", taskName, keyCounter.get()); + } + + @Override + public void close() throws IOException { + iteratorExecutor.shutdown(); Review Comment: This can introduce resource leaks. Better wait for tasks to finish and use timeout mechanism is recommended: ``` public void close() throws IOException { iteratorExecutor.shutdown(); valueExecutors.shutdown(); try { if (!iteratorExecutor.awaitTermination(60, TimeUnit.SECONDS)) { iteratorExecutor.shutdownNow(); } if (!valueExecutors.awaitTermination(60, TimeUnit.SECONDS)) { valueExecutors.shutdownNow(); } } catch (InterruptedException e) { iteratorExecutor.shutdownNow(); valueExecutors.shutdownNow(); Thread.currentThread().interrupt(); } } ``` ########## hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/util/ParallelTableIteratorOperation.java: ########## @@ -0,0 +1,218 @@ +/* + * 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.recon.tasks.util; + +import java.io.Closeable; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashSet; +import java.util.LinkedList; +import java.util.List; +import java.util.Objects; +import java.util.Queue; +import java.util.Set; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; +import java.util.stream.Collectors; +import org.apache.hadoop.hdds.utils.db.Codec; +import org.apache.hadoop.hdds.utils.db.RDBStore; +import org.apache.hadoop.hdds.utils.db.Table; +import org.apache.hadoop.hdds.utils.db.TableIterator; +import org.apache.hadoop.ozone.om.OMMetadataManager; +import org.rocksdb.LiveFileMetaData; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Class to iterate through a table in parallel by breaking table into multiple iterators. + */ +public class ParallelTableIteratorOperation<K extends Comparable<K>, V> implements Closeable { + private final Table<K, V> table; + private final Codec<K> keyCodec; + private final ExecutorService iteratorExecutor; + private final ExecutorService valueExecutors; + private final int maxNumberOfVals; + private final OMMetadataManager metadataManager; + private final int maxIteratorTasks; + private final int maxWorkerTasks; + private final long logCountThreshold; + + private static final Logger LOG = LoggerFactory.getLogger(ParallelTableIteratorOperation.class); + public ParallelTableIteratorOperation(OMMetadataManager metadataManager, Table<K, V> table, Codec<K> keyCodec, + int iteratorCount, int workerCount, int maxNumberOfValsInMemory, + long logThreshold) { + this.table = table; + this.keyCodec = keyCodec; + this.metadataManager = metadataManager; + this.maxIteratorTasks = 2 * iteratorCount; + this.maxWorkerTasks = workerCount * 2; + this.iteratorExecutor = new ThreadPoolExecutor(iteratorCount, iteratorCount, 1, TimeUnit.MINUTES, + new ArrayBlockingQueue<>(iteratorCount * 2)); Review Comment: This has a potential issue where if queue fills up, there is no RejectionPolicy defined and `RejectedExecutionException` lead Recon to crash ########## hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ContainerKeyMapperHelper.java: ########## @@ -99,21 +110,38 @@ public static boolean reprocess(OMMetadataManager omMetadataManager, // Get the appropriate table based on BucketLayout Table<String, OmKeyInfo> omKeyInfoTable = omMetadataManager.getKeyTable(bucketLayout); - // Iterate through the table and process keys - try (TableIterator<String, ? extends Table.KeyValue<String, OmKeyInfo>> keyIter = omKeyInfoTable.iterator()) { - while (keyIter.hasNext()) { - Table.KeyValue<String, OmKeyInfo> kv = keyIter.next(); - handleKeyReprocess(kv.getKey(), kv.getValue(), containerKeyMap, containerKeyCountMap, - reconContainerMetadataManager); - omKeyCount++; - - // Check and flush data if it reaches the batch threshold - if (!checkAndCallFlushToDB(containerKeyMap, containerKeyFlushToDBMaxThreshold, - reconContainerMetadataManager)) { - LOG.error("Failed to flush container key data for {}", taskName); - return false; + ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); + // Use parallel table iteration + Function<Table.KeyValue<String, OmKeyInfo>, Void> kvOperation = kv -> { + try { + try { + lock.readLock().lock(); + handleKeyReprocess(kv.getKey(), kv.getValue(), containerKeyMap, containerKeyCountMap, + reconContainerMetadataManager); + } finally { + lock.readLock().unlock(); + } + omKeyCount.incrementAndGet(); + if (containerKeyMap.size() >= containerKeyFlushToDBMaxThreshold) { Review Comment: This can create race condition. ########## hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ContainerKeyMapperHelper.java: ########## @@ -99,21 +110,38 @@ public static boolean reprocess(OMMetadataManager omMetadataManager, // Get the appropriate table based on BucketLayout Table<String, OmKeyInfo> omKeyInfoTable = omMetadataManager.getKeyTable(bucketLayout); - // Iterate through the table and process keys - try (TableIterator<String, ? extends Table.KeyValue<String, OmKeyInfo>> keyIter = omKeyInfoTable.iterator()) { - while (keyIter.hasNext()) { - Table.KeyValue<String, OmKeyInfo> kv = keyIter.next(); - handleKeyReprocess(kv.getKey(), kv.getValue(), containerKeyMap, containerKeyCountMap, - reconContainerMetadataManager); - omKeyCount++; - - // Check and flush data if it reaches the batch threshold - if (!checkAndCallFlushToDB(containerKeyMap, containerKeyFlushToDBMaxThreshold, - reconContainerMetadataManager)) { - LOG.error("Failed to flush container key data for {}", taskName); - return false; + ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); + // Use parallel table iteration + Function<Table.KeyValue<String, OmKeyInfo>, Void> kvOperation = kv -> { + try { + try { + lock.readLock().lock(); + handleKeyReprocess(kv.getKey(), kv.getValue(), containerKeyMap, containerKeyCountMap, + reconContainerMetadataManager); + } finally { + lock.readLock().unlock(); + } + omKeyCount.incrementAndGet(); + if (containerKeyMap.size() >= containerKeyFlushToDBMaxThreshold) { + try { + lock.writeLock().lock(); Review Comment: ALL 20 threads can reach here. **Problem:** 1. All 20 threads check `containerKeyMap.size()` without holding any lock (line 125) 2. If size >= threshold, ALL 20 threads try to acquire write lock 3. 19 threads block waiting 4. First thread flushes and clears the map 5. Second thread wakes up, acquires lock, checks size again (inside checkAndCallFlushToDB line 173) - map is now empty! 6. Repeat for all 19 threads - massive waste! **Impact:** - 95% of threads blocked (19 out of 20) - Throughput drops from ~100 keys/sec to ~5 keys/sec - Worker threads starve waiting for write lock ########## hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/FileSizeCountTaskHelper.java: ########## @@ -134,31 +152,40 @@ public static boolean reprocessBucketLayout(BucketLayout bucketLayout, OMMetadataManager omMetadataManager, Map<FileSizeCountKey, Long> fileSizeCountMap, ReconFileMetadataManager reconFileMetadataManager, - String taskName) { + String taskName, + int maxIterators, + int maxWorkers, + int maxKeysInMemory) { + LOG.info("{}: Starting parallel iteration with {} iterators, {} workers for bucket layout {}", + taskName, maxIterators, maxWorkers, bucketLayout); Table<String, OmKeyInfo> omKeyInfoTable = omMetadataManager.getKeyTable(bucketLayout); - int totalKeysProcessed = 0; + long startTime = Time.monotonicNow(); - try (TableIterator<String, ? extends Table.KeyValue<String, OmKeyInfo>> keyIter = - omKeyInfoTable.iterator()) { - while (keyIter.hasNext()) { - Table.KeyValue<String, OmKeyInfo> kv = keyIter.next(); - handlePutKeyEvent(kv.getValue(), fileSizeCountMap); - totalKeysProcessed++; - - // Flush to RocksDB periodically. - if (fileSizeCountMap.size() >= 100000) { - // For reprocess, we don't need to check existing values since table was truncated - LOG.debug("Flushing {} accumulated counts to RocksDB for {}", fileSizeCountMap.size(), taskName); - writeCountsToDB(fileSizeCountMap, reconFileMetadataManager); - fileSizeCountMap.clear(); - } - } - } catch (IOException ioEx) { - LOG.error("Unable to populate File Size Count for {} in RocksDB.", taskName, ioEx); + // Use parallel table iteration + Function<Table.KeyValue<String, OmKeyInfo>, Void> kvOperation = kv -> { + handlePutKeyEvent(kv.getValue(), fileSizeCountMap); + return null; + }; + + try (ParallelTableIteratorOperation<String, OmKeyInfo> keyIter = + new ParallelTableIteratorOperation<>(omMetadataManager, omKeyInfoTable, + StringCodec.get(), maxIterators, maxWorkers, maxKeysInMemory, 100000)) { Review Comment: Where the flush logic gone ? Why removed ? ########## hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ContainerKeyMapperHelper.java: ########## @@ -99,21 +110,38 @@ public static boolean reprocess(OMMetadataManager omMetadataManager, // Get the appropriate table based on BucketLayout Table<String, OmKeyInfo> omKeyInfoTable = omMetadataManager.getKeyTable(bucketLayout); - // Iterate through the table and process keys - try (TableIterator<String, ? extends Table.KeyValue<String, OmKeyInfo>> keyIter = omKeyInfoTable.iterator()) { - while (keyIter.hasNext()) { - Table.KeyValue<String, OmKeyInfo> kv = keyIter.next(); - handleKeyReprocess(kv.getKey(), kv.getValue(), containerKeyMap, containerKeyCountMap, - reconContainerMetadataManager); - omKeyCount++; - - // Check and flush data if it reaches the batch threshold - if (!checkAndCallFlushToDB(containerKeyMap, containerKeyFlushToDBMaxThreshold, - reconContainerMetadataManager)) { - LOG.error("Failed to flush container key data for {}", taskName); - return false; + ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); Review Comment: Use fair locking, else thread starvation issue can happen. - ReentrantReadWriteLock is non-fair by default - With 20 worker threads continuously acquiring read locks, write lock requests can be starved indefinitely - When containerKeyMap reaches threshold, threads need write lock to flush, but may wait forever Scenario: Thread 1: [Read Lock] -> processing key 1 Thread 2: [Read Lock] -> processing key 2 Thread 3: [Read Lock] -> processing key 3 ... Thread 20: [Read Lock] -> processing key 20 Thread 21: [Waiting Write Lock] <- BLOCKED! (containerKeyMap full, needs to flush) Thread 1: [Read Lock] -> processing key 21 (lock released and reacquired) Thread 2: [Read Lock] -> processing key 22 ... Thread 21 STILL WAITING -- 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]
