content: /*
   * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
   * (the "License"). You may not use this work except in compliance with the License, which is
   * available at www.apache.org/licenses/LICENSE-2.0
   *
   * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
   * either express or implied, as more fully set forth in the License.
   *
   * See the NOTICE file distributed with this work for information regarding copyright ownership.
   */
  
  package alluxio.master.file.meta;
  
  import alluxio.exception.status.UnavailableException;
  import alluxio.master.block.BlockId;
  import alluxio.master.block.ContainerIdGenerable;
  import alluxio.master.file.state.DirectoryId;
  import alluxio.master.journal.JournalContext;
  import alluxio.master.journal.Journaled;
  import alluxio.master.journal.checkpoint.CheckpointName;
  import alluxio.proto.journal.File.InodeDirectoryIdGeneratorEntry;
  import alluxio.proto.journal.Journal.JournalEntry;
  import alluxio.resource.CloseableIterator;
  import alluxio.util.CommonUtils;
  
  import com.google.common.base.Preconditions;
  
  import javax.annotation.concurrent.ThreadSafe;
  
  /**
   * Inode id management for directory inodes. Keep track of a block container id, along with a block
   * sequence number. If the block sequence number reaches the limit, a new block container id is
   * retrieved.
   */
  @ThreadSafe
  public class InodeDirectoryIdGenerator implements Journaled {
    private final ContainerIdGenerable mContainerIdGenerator;
  
    private final DirectoryId mNextDirectoryId;
    private boolean mInitialized = false;
  
    /**
     * @param containerIdGenerator the container id generator to use
     */
    public InodeDirectoryIdGenerator(ContainerIdGenerable containerIdGenerator) {
      mContainerIdGenerator =
          Preconditions.checkNotNull(containerIdGenerator, "containerIdGenerator");
      mNextDirectoryId = new DirectoryId();
    }
  
    /**
     * Returns the next directory id, and journals the state.
     *
     * @return the next directory id
     */
    synchronized long getNewDirectoryId(JournalContext context) throws UnavailableException {
      initialize(context);
      long containerId = mNextDirectoryId.getContainerId();
      long sequenceNumber = mNextDirectoryId.getSequenceNumber();
      long directoryId = BlockId.createBlockId(containerId, sequenceNumber);
      if (sequenceNumber == BlockId.getMaxSequenceNumber()) {
        // No more ids in this container. Get a new container for the next id.
        containerId = mContainerIdGenerator.getNewContainerId();
        sequenceNumber = 0;
      } else {
        sequenceNumber++;
      }
      applyAndJournal(context, toEntry(containerId, sequenceNumber));
      return directoryId;
    }
  
    /**
     * @return the next directory id
     */
    public long peekDirectoryId() {
      DirectoryId directoryId = mNextDirectoryId;
      long containerId = directoryId.getContainerId();
      long sequenceNumber = directoryId.getSequenceNumber();
      return BlockId.createBlockId(containerId, sequenceNumber);
    }
  
    private void initialize(JournalContext context) throws UnavailableException {
      if (!mInitialized) {
        applyAndJournal(context, toEntry(mContainerIdGenerator.getNewContainerId(), 0));
        mInitialized = true;
      }
    }
  
    /**
     * @param containerId a container ID
     * @param sequenceNumber a sequence number
     * @return a journal entry for the given container ID and sequence number
     */
    private static JournalEntry toEntry(long containerId, long sequenceNumber) {
      return JournalEntry.newBuilder().setInodeDirectoryIdGenerator(
          InodeDirectoryIdGeneratorEntry.newBuilder()
              .setContainerId(containerId)
              .setSequenceNumber(sequenceNumber))
          .build();
    }
  
    @Override
    public boolean processJournalEntry(JournalEntry entry) {
      if (entry.hasInodeDirectoryIdGenerator()) {
        InodeDirectoryIdGeneratorEntry e = entry.getInodeDirectoryIdGenerator();
        mNextDirectoryId.setContainerId(e.getContainerId());
        mNextDirectoryId.setSequenceNumber(e.getSequenceNumber());
        return true;
      }
      return false;
    }
  
    @Override
    public void resetState() {
      mNextDirectoryId.setContainerId(0);
      mNextDirectoryId.setSequenceNumber(0);
    }
  
    @Override
    public CloseableIterator<JournalEntry> getJournalEntryIterator() {
      return CloseableIterator.noopCloseable(CommonUtils.singleElementIterator(
          JournalEntry.newBuilder()
              .setInodeDirectoryIdGenerator(InodeDirectoryIdGeneratorEntry.newBuilder()
                  .setContainerId(mNextDirectoryId.getContainerId())
                  .setSequenceNumber(mNextDirectoryId.getSequenceNumber()))
          .build()));
    }
  
    @Override
    public CheckpointName getCheckpointName() {
      return CheckpointName.INODE_DIRECTORY_ID_GENERATOR;
    }
  }
  