HDDS-475. Block Allocation returns same BlockID on different keys creation.

Contributed by Nanda Kumar.

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/BlockManagerImpl.java

@@ -48,8 +48,6 @@
 import java.util.Map;
 import java.util.Random;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.ReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 import static org.apache.hadoop.hdds.scm.exceptions.SCMException.ResultCodes
     .CHILL_MODE_EXCEPTION;
@@ -76,7 +74,6 @@ public class BlockManagerImpl implements EventHandler<Boolean>,
   private final NodeManager nodeManager;
   private final Mapping containerManager;
 
-  private final ReadWriteLock lock;
   private final long containerSize;
 
   private final DeletedBlockLog deletedBlockLog;
@@ -113,7 +110,6 @@ public BlockManagerImpl(final Configuration conf,
             ScmConfigKeys.OZONE_SCM_CONTAINER_PROVISION_BATCH_SIZE,
             ScmConfigKeys.OZONE_SCM_CONTAINER_PROVISION_BATCH_SIZE_DEFAULT);
     rand = new Random();
-    this.lock = new ReentrantReadWriteLock();
 
     mxBean = MBeans.register("BlockManager", "BlockManagerImpl", this);
 
@@ -223,8 +219,6 @@ public AllocatedBlock allocateBlock(final long size,
 
     ContainerWithPipeline containerWithPipeline;
 
-    lock.readLock().lock();
-    try {
     // This is to optimize performance, if the below condition is evaluated
     // to false, then we can be sure that there are no containers in
     // ALLOCATED state.
@@ -302,16 +296,9 @@ public AllocatedBlock allocateBlock(final long size,
     // we have tried all strategies we know and but somehow we are not able
     // to get a container for this block. Log that info and return a null.
     LOG.error(
-          "Unable to allocate a block for the size: {}, type: {}, " +
+        "Unable to allocate a block for the size: {}, type: {}, factor: {}",
-              "factor: {}",
+        size, type, factor);
-          size,
-          type,
-          factor);
     return null;
-
-    } finally {
-      lock.readLock().unlock();
-    }
   }
 
   /**
@@ -330,9 +317,7 @@ private AllocatedBlock newBlock(ContainerWithPipeline containerWithPipeline,
     }
 
     // TODO : Revisit this local ID allocation when HA is added.
-    // TODO: this does not work well if multiple allocation kicks in a tight
+    long localID = UniqueId.next();
-    // loop.
-    long localID = Time.getUtcTime();
     long containerID = containerInfo.getContainerID();
 
     boolean createContainer = (state == HddsProtos.LifeCycleState.ALLOCATED);
@@ -463,4 +448,51 @@ public boolean isScmInChillMode() {
   public static Logger getLogger() {
     return LOG;
   }
+
+  /**
+   * This class uses system current time milliseconds to generate unique id.
+   */
+  public static final class UniqueId {
+    /*
+     * When we represent time in milliseconds using 'long' data type,
+     * the LSB bits are used. Currently we are only using 44 bits (LSB),
+     * 20 bits (MSB) are not used.
+     * We will exhaust this 44 bits only when we are in year 2525,
+     * until then we can safely use this 20 bits (MSB) for offset to generate
+     * unique id within millisecond.
+     *
+     * Year        : Mon Dec 31 18:49:04 IST 2525
+     * TimeInMillis: 17545641544247
+     * Binary Representation:
+     *   MSB (20 bits): 0000 0000 0000 0000 0000
+     *   LSB (44 bits): 1111 1111 0101 0010 1001 1011 1011 0100 1010 0011 0111
+     *
+     * We have 20 bits to run counter, we should exclude the first bit (MSB)
+     * as we don't want to deal with negative values.
+     * To be on safer side we will use 'short' data type which is of length
+     * 16 bits and will give us 65,536 values for offset.
+     *
+     */
+
+    private static volatile short offset = 0;
+
+    /**
+     * Private constructor so that no one can instantiate this class.
+     */
+    private UniqueId() {}
+
+    /**
+     * Calculate and returns next unique id based on System#currentTimeMillis.
+     *
+     * @return unique long value
+     */
+    public static synchronized long next() {
+      long utcTime = Time.getUtcTime();
+      if ((utcTime & 0xFFFF000000000000L) == 0) {
+        return utcTime << Short.SIZE | (offset++ & 0x0000FFFF);
+      }
+      throw new RuntimeException("Got invalid UTC time," +
+          " cannot generate unique Id. UTC Time: " + utcTime);
+    }
+  }
 }