HDDS-2026. Overlapping chunk region cannot be read concurrently

Signed-off-by: Anu Engineer <aengineer@apache.org>

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/ChunkUtils.java

@@ -18,6 +18,7 @@
 
 package org.apache.hadoop.ozone.container.keyvalue.helpers;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos
     .ContainerCommandRequestProto;
@@ -36,17 +37,20 @@
 import org.apache.hadoop.ozone.container.keyvalue.impl.ChunkManagerImpl;
 import org.apache.hadoop.ozone.container.common.volume.VolumeIOStats;
 import org.apache.hadoop.util.Time;
+import org.apache.ratis.util.function.CheckedSupplier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.File;
 import java.io.IOException;
 import java.nio.ByteBuffer;
-import java.nio.channels.AsynchronousFileChannel;
 import java.nio.channels.FileChannel;
 import java.nio.channels.FileLock;
+import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.security.NoSuchAlgorithmException;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
 
 import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.*;
@@ -56,6 +60,8 @@
  */
 public final class ChunkUtils {
 
+  private static final Set<Path> LOCKS = ConcurrentHashMap.newKeySet();
+
   /** Never constructed. **/
   private ChunkUtils() {
 
@@ -67,9 +73,8 @@ private ChunkUtils() {
    * @param chunkFile - File to write data to.
    * @param chunkInfo - Data stream to write.
    * @param data - The data buffer.
-   * @param volumeIOStats
+   * @param volumeIOStats statistics collector
    * @param sync whether to do fsync or not
-   * @throws StorageContainerException
    */
   public static void writeData(File chunkFile, ChunkInfo chunkInfo,
       ByteBuffer data, VolumeIOStats volumeIOStats, boolean sync)
@@ -85,58 +90,43 @@ public static void writeData(File chunkFile, ChunkInfo chunkInfo,
       throw new StorageContainerException(err, INVALID_WRITE_SIZE);
     }
 
-    FileChannel file = null;
-    FileLock lock = null;
-
-    try {
-      long writeTimeStart = Time.monotonicNow();
-
-      // skip SYNC and DSYNC to reduce contention on file.lock
-      file = FileChannel.open(chunkFile.toPath(),
-              StandardOpenOption.CREATE,
-              StandardOpenOption.WRITE,
-              StandardOpenOption.SPARSE);
-
-      lock = file.lock();
-      int size = file.write(data, chunkInfo.getOffset());
-      // Increment volumeIO stats here.
-      volumeIOStats.incWriteTime(Time.monotonicNow() - writeTimeStart);
-      volumeIOStats.incWriteOpCount();
-      volumeIOStats.incWriteBytes(size);
-      if (size != bufferSize) {
-        log.error("Invalid write size found. Size:{}  Expected: {} ", size,
-            bufferSize);
-        throw new StorageContainerException("Invalid write size found. " +
-            "Size: " + size + " Expected: " + bufferSize, INVALID_WRITE_SIZE);
-      }
-    } catch (StorageContainerException ex) {
-      throw ex;
-    } catch(IOException e) {
-      throw new StorageContainerException(e, IO_EXCEPTION);
-
-    } finally {
-      if (lock != null) {
-        try {
-          lock.release();
-        } catch (IOException e) {
-          log.error("Unable to release lock ??, Fatal Error.");
-          throw new StorageContainerException(e, CONTAINER_INTERNAL_ERROR);
+    Path path = chunkFile.toPath();
+    long startTime = Time.monotonicNow();
+    processFileExclusively(path, () -> {
+      FileChannel file = null;
+      try {
+        // skip SYNC and DSYNC to reduce contention on file.lock
+        file = FileChannel.open(path,
+            StandardOpenOption.CREATE,
+            StandardOpenOption.WRITE,
+            StandardOpenOption.SPARSE);
 
+        int size;
+        try (FileLock ignored = file.lock()) {
+          size = file.write(data, chunkInfo.getOffset());
         }
-      }
-      if (file != null) {
-        try {
-          if (sync) {
-            // ensure data and metadata is persisted. Outside the lock
-            file.force(true);
-          }
-          file.close();
-        } catch (IOException e) {
-          throw new StorageContainerException("Error closing chunk file",
-              e, CONTAINER_INTERNAL_ERROR);
+
+        // Increment volumeIO stats here.
+        volumeIOStats.incWriteTime(Time.monotonicNow() - startTime);
+        volumeIOStats.incWriteOpCount();
+        volumeIOStats.incWriteBytes(size);
+        if (size != bufferSize) {
+          log.error("Invalid write size found. Size:{}  Expected: {} ", size,
+              bufferSize);
+          throw new StorageContainerException("Invalid write size found. " +
+              "Size: " + size + " Expected: " + bufferSize, INVALID_WRITE_SIZE);
         }
+      } catch (StorageContainerException ex) {
+        throw ex;
+      } catch (IOException e) {
+        throw new StorageContainerException(e, IO_EXCEPTION);
+      } finally {
+        closeFile(file, sync);
       }
-    }
+
+      return null;
+    });
+
     log.debug("Write Chunk completed for chunkFile: {}, size {}", chunkFile,
         bufferSize);
   }
@@ -146,11 +136,8 @@ public static void writeData(File chunkFile, ChunkInfo chunkInfo,
    *
    * @param chunkFile - file where data lives.
    * @param data - chunk definition.
-   * @param volumeIOStats
+   * @param volumeIOStats statistics collector
    * @return ByteBuffer
-   * @throws StorageContainerException
-   * @throws ExecutionException
-   * @throws InterruptedException
    */
   public static ByteBuffer readData(File chunkFile, ChunkInfo data,
       VolumeIOStats volumeIOStats) throws StorageContainerException,
@@ -165,38 +152,36 @@ public static ByteBuffer readData(File chunkFile, ChunkInfo data,
           data.toString(), UNABLE_TO_FIND_CHUNK);
     }
 
-    AsynchronousFileChannel file = null;
-    FileLock lock = null;
-    try {
-      long readStartTime = Time.monotonicNow();
-      file =
-          AsynchronousFileChannel.open(chunkFile.toPath(),
-              StandardOpenOption.READ);
-      lock = file.lock(data.getOffset(), data.getLen(), true).get();
+    long offset = data.getOffset();
+    long len = data.getLen();
+    ByteBuffer buf = ByteBuffer.allocate((int) len);
 
-      ByteBuffer buf = ByteBuffer.allocate((int) data.getLen());
-      file.read(buf, data.getOffset()).get();
+    Path path = chunkFile.toPath();
+    long startTime = Time.monotonicNow();
+    return processFileExclusively(path, () -> {
+      FileChannel file = null;
 
-      // Increment volumeIO stats here.
-      volumeIOStats.incReadTime(Time.monotonicNow() - readStartTime);
-      volumeIOStats.incReadOpCount();
-      volumeIOStats.incReadBytes(data.getLen());
+      try {
+        file = FileChannel.open(path, StandardOpenOption.READ);
 
-      return buf;
-    } catch (IOException e) {
-      throw new StorageContainerException(e, IO_EXCEPTION);
-    } finally {
-      if (lock != null) {
-        try {
-          lock.release();
-        } catch (IOException e) {
-          log.error("I/O error is lock release.");
+        try (FileLock ignored = file.lock(offset, len, true)) {
+          file.read(buf, offset);
+        }
+
+        // Increment volumeIO stats here.
+        volumeIOStats.incReadTime(Time.monotonicNow() - startTime);
+        volumeIOStats.incReadOpCount();
+        volumeIOStats.incReadBytes(len);
+
+        return buf;
+      } catch (IOException e) {
+        throw new StorageContainerException(e, IO_EXCEPTION);
+      } finally {
+        if (file != null) {
+          IOUtils.closeStream(file);
         }
       }
-      if (file != null) {
-        IOUtils.closeStream(file);
-      }
-    }
+    });
   }
 
   /**
@@ -326,4 +311,37 @@ public static ContainerCommandResponseProto getReadChunkResponse(
     builder.setReadChunk(response);
     return builder.build();
   }
+
+  @VisibleForTesting
+  static <T, E extends Exception> T processFileExclusively(
+      Path path, CheckedSupplier<T, E> op
+  ) throws E {
+    for (;;) {
+      if (LOCKS.add(path)) {
+        break;
+      }
+    }
+
+    try {
+      return op.get();
+    } finally {
+      LOCKS.remove(path);
+    }
+  }
+
+  private static void closeFile(FileChannel file, boolean sync)
+      throws StorageContainerException {
+    if (file != null) {
+      try {
+        if (sync) {
+          // ensure data and metadata is persisted
+          file.force(true);
+        }
+        file.close();
+      } catch (IOException e) {
+        throw new StorageContainerException("Error closing chunk file",
+            e, CONTAINER_INTERNAL_ERROR);
+      }
+    }
+  }
 }

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/helpers/TestChunkUtils.java

@@ -0,0 +1,164 @@
+/*
+ * 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.container.keyvalue.helpers;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.ozone.container.common.helpers.ChunkInfo;
+import org.apache.hadoop.ozone.container.common.volume.VolumeIOStats;
+import org.apache.hadoop.test.GenericTestUtils;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+/**
+ * Tests for {@link ChunkUtils}.
+ */
+public class TestChunkUtils {
+
+  private static final Logger LOG =
+      LoggerFactory.getLogger(TestChunkUtils.class);
+
+  private static final String PREFIX = TestChunkUtils.class.getSimpleName();
+
+  @Test
+  public void concurrentReadOfSameFile() throws Exception {
+    String s = "Hello World";
+    byte[] array = s.getBytes();
+    ByteBuffer data = ByteBuffer.wrap(array);
+    Path tempFile = Files.createTempFile(PREFIX, "concurrent");
+    try {
+      ChunkInfo chunkInfo = new ChunkInfo(tempFile.toString(),
+          0, data.capacity());
+      File file = tempFile.toFile();
+      VolumeIOStats stats = new VolumeIOStats();
+      ChunkUtils.writeData(file, chunkInfo, data, stats, true);
+      int threads = 10;
+      ExecutorService executor = new ThreadPoolExecutor(threads, threads,
+          0, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
+      AtomicInteger processed = new AtomicInteger();
+      AtomicBoolean failed = new AtomicBoolean();
+      for (int i = 0; i < threads; i++) {
+        final int threadNumber = i;
+        executor.submit(() -> {
+          try {
+            ByteBuffer readBuffer = ChunkUtils.readData(file, chunkInfo, stats);
+            LOG.info("Read data ({}): {}", threadNumber,
+                new String(readBuffer.array()));
+            if (!Arrays.equals(array, readBuffer.array())) {
+              failed.set(true);
+            }
+          } catch (Exception e) {
+            LOG.error("Failed to read data ({})", threadNumber, e);
+            failed.set(true);
+          }
+          processed.incrementAndGet();
+        });
+      }
+      try {
+        GenericTestUtils.waitFor(() -> processed.get() == threads,
+            100, (int) TimeUnit.SECONDS.toMillis(5));
+      } finally {
+        executor.shutdownNow();
+      }
+      assertEquals(threads * stats.getWriteBytes(), stats.getReadBytes());
+      assertFalse(failed.get());
+    } finally {
+      Files.deleteIfExists(tempFile);
+    }
+  }
+
+  @Test
+  public void concurrentProcessing() throws Exception {
+    final int perThreadWait = 1000;
+    final int maxTotalWait = 5000;
+    int threads = 20;
+    List<Path> paths = new LinkedList<>();
+
+    try {
+      ExecutorService executor = new ThreadPoolExecutor(threads, threads,
+          0, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
+      AtomicInteger processed = new AtomicInteger();
+      for (int i = 0; i < threads; i++) {
+        Path path = Files.createTempFile(PREFIX, String.valueOf(i));
+        paths.add(path);
+        executor.submit(() -> {
+          ChunkUtils.processFileExclusively(path, () -> {
+            try {
+              Thread.sleep(perThreadWait);
+            } catch (InterruptedException e) {
+              e.printStackTrace();
+            }
+            processed.incrementAndGet();
+            return null;
+          });
+        });
+      }
+      try {
+        GenericTestUtils.waitFor(() -> processed.get() == threads,
+            100, maxTotalWait);
+      } finally {
+        executor.shutdownNow();
+      }
+    } finally {
+      for (Path path : paths) {
+        FileUtils.deleteQuietly(path.toFile());
+      }
+    }
+  }
+
+  @Test
+  public void serialRead() throws Exception {
+    String s = "Hello World";
+    byte[] array = s.getBytes();
+    ByteBuffer data = ByteBuffer.wrap(array);
+    Path tempFile = Files.createTempFile(PREFIX, "serial");
+    try {
+      ChunkInfo chunkInfo = new ChunkInfo(tempFile.toString(),
+          0, data.capacity());
+      File file = tempFile.toFile();
+      VolumeIOStats stats = new VolumeIOStats();
+      ChunkUtils.writeData(file, chunkInfo, data, stats, true);
+      ByteBuffer readBuffer = ChunkUtils.readData(file, chunkInfo, stats);
+      assertArrayEquals(array, readBuffer.array());
+      assertEquals(stats.getWriteBytes(), stats.getReadBytes());
+    } catch (Exception e) {
+      LOG.error("Failed to read data", e);
+    } finally {
+      Files.deleteIfExists(tempFile);
+    }
+  }
+
+}