OPENDJ-1902 (CR-6667) Remove Storage.getWriteableTransaction()

Jean-Noel Rouvignac

23.48.2015 25eb96def3b1fb3aea5d69a631f848bcab95f1e9

OPENDJ-1902 (CR-6667) Remove Storage.getWriteableTransaction()

Thanks to Yannick who showed me the way with r12096, I finally found how to do it.

*Storage.java:
Removed getWriteableStorage() and close() methods.

Importer.java:
In IndexDBWriteTask:
- In ctor, accept a Storage rather than a WriteableTransaction + renamed txn field to storage and changed its type from WriteableTransaction to Storage.
- In call(), extracted all code to call0(WriteableTransaction) + now only open a WriteableTransaction and call call0(WriteableTransaction) + propagate WriteableTransaction down the method calls
- renamed flushMapToDB() to flushToDB() + renamed map parameter to idSets and its type from Map to Collection.
- (Unrelated to this change) renamed doRebuildIndexes() to rebuildIndexes0().

ReadableTransaction.java:
No longer extends Closeable becasue it is not needed anymore.

 opendj-server-legacy/src/main/java/org/opends/server/backends/persistit/PersistItStorage.java

@@ -27,6 +27,7 @@

import static com.persistit.Transaction.CommitPolicy.*;
import static java.util.Arrays.*;

import static org.opends.messages.BackendMessages.*;
import static org.opends.messages.ConfigMessages.*;
import static org.opends.messages.JebMessages.*;
@@ -459,8 +460,7 @@
      return exchange;
    }

    @Override
    public void close()
    private void release()
    {
      for (final Exchange ex : exchanges.values())
      {
@@ -628,7 +628,7 @@
        }
        finally
        {
          storageImpl.close();
          storageImpl.release();
        }
      }
      catch (final RollbackException e)
@@ -700,7 +700,7 @@
        }
        finally
        {
          storageImpl.close();
          storageImpl.release();
        }
      }
      catch (final RollbackException e)
@@ -720,12 +720,6 @@
  }

  @Override
  public WriteableTransaction getWriteableTransaction()
  {
    return new StorageImpl();
  }

  @Override
  public boolean supportsBackupAndRestore()
  {
    return true;

 opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/Importer.java

@@ -835,7 +835,7 @@
      }
      else
      {
        doRebuildIndexes();
        rebuildIndexes0();
      }
    }
    catch (Exception e)
@@ -861,7 +861,7 @@
    });
  }

  private void doRebuildIndexes() throws Exception
  private void rebuildIndexes0() throws Exception
  {
    final long startTime = System.currentTimeMillis();
    final Storage storage = rootContainer.getStorage();
@@ -1224,10 +1224,8 @@
  {
    for (IndexManager indexMgr : indexMgrs)
    {
      // avoid threading issues by allocating one writeable storage per thread
      // DB transactions are generally tied to a single thread
      WriteableTransaction txn = this.rootContainer.getStorage().getWriteableTransaction();
      futures.add(dbService.submit(new IndexDBWriteTask(indexMgr, txn, permits, buffers, readAheadSize)));
      futures.add(dbService.submit(
          new IndexDBWriteTask(rootContainer.getStorage(), indexMgr, permits, buffers, readAheadSize)));
    }
  }

@@ -1441,10 +1439,6 @@
        isCanceled = true;
        throw e;
      }
      finally
      {
        txn.close();
      }
    }

    void processEntry(WriteableTransaction txn, Entry entry, Suffix suffix)
@@ -1577,10 +1571,6 @@
        isCanceled = true;
        throw e;
      }
      finally
      {
        txn.close();
      }
    }

    void processEntry(WriteableTransaction txn, Entry entry, EntryID entryID, Suffix suffix)
@@ -1779,6 +1769,7 @@
   */
  private final class IndexDBWriteTask implements Callable<Void>
  {
    private final Storage storage;
    private final IndexManager indexMgr;
    private final int cacheSize;
    /** indexID => DNState map */
@@ -1796,15 +1787,14 @@
    private int nextBufferID;
    private int ownedPermits;
    private volatile boolean isRunning;
    private final WriteableTransaction txn;

    /**
     * Creates a new index DB writer.
     *
     * @param indexMgr
     *          The index manager.
     * @param txn
     *          The database transaction
     * @param storage
     *          Where to store data
     * @param permits
     *          The semaphore used for restricting the number of buffer allocations.
     * @param maxPermits
@@ -1812,11 +1802,10 @@
     * @param cacheSize
     *          The buffer cache size.
     */
    public IndexDBWriteTask(IndexManager indexMgr, WriteableTransaction txn, Semaphore permits, int maxPermits,
        int cacheSize)
    public IndexDBWriteTask(Storage storage, IndexManager indexMgr, Semaphore permits, int maxPermits, int cacheSize)
    {
      this.storage = storage;
      this.indexMgr = indexMgr;
      this.txn = txn;
      this.permits = permits;
      this.maxPermits = maxPermits;
      this.cacheSize = cacheSize;
@@ -1895,10 +1884,8 @@
      return buffers;
    }

    /**
     * Finishes this task.
     */
    public void endWriteTask()
    /** Finishes this task. */
    private void endWriteTask(WriteableTransaction txn)
    {
      isRunning = false;

@@ -1915,7 +1902,7 @@
        {
          for (DNState dnState : dnStateMap.values())
          {
            dnState.flush();
            dnState.flush(txn);
          }
          if (!isCanceled)
          {
@@ -1932,7 +1919,7 @@
      }
      finally
      {
        close(bufferFile, bufferIndexFile, txn);
        close(bufferFile, bufferIndexFile);

        indexMgr.getBufferFile().delete();
        indexMgr.getBufferIndexFile().delete();
@@ -1972,9 +1959,22 @@
    @Override
    public Void call() throws Exception
    {
      storage.write(new WriteOperation()
      {
        @Override
        public void run(WriteableTransaction txn) throws Exception
        {
          call0(txn);
        }
      });
      return null;
    }

    private void call0(WriteableTransaction txn) throws Exception
    {
      if (isCanceled)
      {
        return null;
        return;
      }

      ImportIDSet insertIDSet = null;
@@ -1989,7 +1989,7 @@
        {
          if (isCanceled)
          {
            return null;
            return;
          }

          while (!bufferSet.isEmpty())
@@ -1999,7 +1999,7 @@
            {
              if (previousRecord != null)
              {
                addToDB(previousRecord.getIndexID(), insertIDSet, deleteIDSet);
                addToDB(txn, previousRecord.getIndexID(), insertIDSet, deleteIDSet);
              }

              // this is a new record
@@ -2023,10 +2023,9 @@

          if (previousRecord != null)
          {
            addToDB(previousRecord.getIndexID(), insertIDSet, deleteIDSet);
            addToDB(txn, previousRecord.getIndexID(), insertIDSet, deleteIDSet);
          }
        }
        return null;
      }
      catch (Exception e)
      {
@@ -2035,7 +2034,7 @@
      }
      finally
      {
        endWriteTask();
        endWriteTask(txn);
      }
    }

@@ -2050,12 +2049,13 @@
      return new ImportIDSet(record.getKey(), newDefinedSet(), index.getIndexEntryLimit(), index.getMaintainCount());
    }

    private void addToDB(int indexID, ImportIDSet insertSet, ImportIDSet deleteSet) throws DirectoryException
    private void addToDB(WriteableTransaction txn, int indexID, ImportIDSet insertSet, ImportIDSet deleteSet)
        throws DirectoryException
    {
      keyCount.incrementAndGet();
      if (indexMgr.isDN2ID())
      {
        addDN2ID(indexID, insertSet);
        addDN2ID(txn, indexID, insertSet);
      }
      else
      {
@@ -2072,7 +2072,7 @@
      }
    }

    private void addDN2ID(int indexID, ImportIDSet idSet) throws DirectoryException
    private void addDN2ID(WriteableTransaction txn, int indexID, ImportIDSet idSet) throws DirectoryException
    {
      DNState dnState = dnStateMap.get(indexID);
      if (dnState == null)
@@ -2082,7 +2082,7 @@
      }
      if (dnState.checkParent(txn, idSet))
      {
        dnState.writeToDN2ID(idSet);
        dnState.writeToDN2ID(txn, idSet);
      }
    }

@@ -2197,7 +2197,7 @@
        return true;
      }

      private void id2child(EntryID childID) throws DirectoryException
      private void id2child(WriteableTransaction txn, EntryID childID) throws DirectoryException
      {
        if (parentID == null)
        {
@@ -2207,7 +2207,7 @@
        getId2childtreeImportIDSet().addEntryID(childID);
        if (id2childTree.size() > DN_STATE_CACHE_SIZE)
        {
          flushMapToDB(id2childTree, entryContainer.getID2Children(), true);
          flushToDB(txn, id2childTree.values(), entryContainer.getID2Children(), true);
        }
      }

@@ -2223,7 +2223,7 @@
        return idSet;
      }

      private void id2SubTree(ReadableTransaction txn, EntryID childID) throws DirectoryException
      private void id2SubTree(WriteableTransaction txn, EntryID childID) throws DirectoryException
      {
        if (parentID == null)
        {
@@ -2246,7 +2246,7 @@
        }
        if (id2subtreeTree.size() > DN_STATE_CACHE_SIZE)
        {
          flushMapToDB(id2subtreeTree, entryContainer.getID2Subtree(), true);
          flushToDB(txn, id2subtreeTree.values(), entryContainer.getID2Subtree(), true);
        }
      }

@@ -2282,32 +2282,32 @@
        return idSet;
      }

      public void writeToDN2ID(ImportIDSet idSet) throws DirectoryException
      public void writeToDN2ID(WriteableTransaction txn, ImportIDSet idSet) throws DirectoryException
      {
        txn.put(dn2id, idSet.getKey(), entryID.toByteString());
        indexMgr.addTotDNCount(1);
        if (parentDN != null)
        {
          id2child(entryID);
          id2child(txn, entryID);
          id2SubTree(txn, entryID);
        }
      }

      public void flush()
      public void flush(WriteableTransaction txn)
      {
        flushMapToDB(id2childTree, entryContainer.getID2Children(), false);
        flushMapToDB(id2subtreeTree, entryContainer.getID2Subtree(), false);
        flushToDB(txn, id2childTree.values(), entryContainer.getID2Children(), false);
        flushToDB(txn, id2subtreeTree.values(), entryContainer.getID2Subtree(), false);
      }

      private void flushMapToDB(Map<ByteString, ImportIDSet> map, Index index, boolean clearMap)
      private void flushToDB(WriteableTransaction txn, Collection<ImportIDSet> idSets, Index index, boolean clearIDSets)
      {
        for (ImportIDSet idSet : map.values())
        for (ImportIDSet idSet : idSets)
        {
          index.importPut(txn, idSet);
        }
        if (clearMap)
        if (clearIDSets)
        {
          map.clear();
          idSets.clear();
        }
      }
    }

 opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/TracedStorage.java

@@ -91,9 +91,7 @@
    }
  }

  /**
   * Decorates an {@link ReadableTransaction} with additional trace logging.
   */
  /** Decorates an {@link ReadableTransaction} with additional trace logging. */
  private final class TracedReadableStorage implements ReadableTransaction
  {
    private final ReadableTransaction txn;
@@ -130,21 +128,13 @@
      return value;
    }

    @Override
    public void close()
    {
      logger.trace("Storage@%s.ReadableStorage@%s.close()", storageId(), id());
    }

    private int id()
    {
      return System.identityHashCode(this);
    }
  }

  /**
   * Decorates an {@link WriteableTransaction} with additional trace logging.
   */
  /** Decorates an {@link WriteableTransaction} with additional trace logging. */
  private final class TracedWriteableStorage implements WriteableTransaction
  {
    private final WriteableTransaction txn;
@@ -231,12 +221,6 @@
      return isUpdated;
    }

    @Override
    public void close()
    {
      logger.trace("Storage@%s.WriteableStorage@%s.close()", storageId(), id());
    }

    private int id()
    {
      return System.identityHashCode(this);
@@ -356,17 +340,6 @@
    storage.write(op);
  }

  @Override
  public WriteableTransaction getWriteableTransaction()
  {
    final WriteableTransaction writeableStorage = storage.getWriteableTransaction();
    if (logger.isTraceEnabled())
    {
      return new TracedWriteableStorage(writeableStorage);
    }
    return writeableStorage;
  }

  private String hex(final ByteSequence bytes)
  {
    return bytes != null ? bytes.toByteString().toHexString() : null;

 opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/spi/ReadableTransaction.java

@@ -25,15 +25,13 @@
 */
package org.opends.server.backends.pluggable.spi;

import java.io.Closeable;

import org.forgerock.opendj.ldap.ByteSequence;
import org.forgerock.opendj.ldap.ByteString;

/**
 * Represents a readable transaction on a storage engine.
 */
public interface ReadableTransaction extends Closeable
public interface ReadableTransaction
{
  /**
   * Reads the record's value associated to the provided key, in the tree whose name is provided.
@@ -63,7 +61,4 @@
   * @return the number of key/value pairs in the provided tree.
   */
  long getRecordCount(TreeName treeName);

  @Override
  public void close();
}

 opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/spi/Storage.java

@@ -80,13 +80,6 @@
  void write(WriteOperation writeOperation) throws Exception;

  /**
   * Returns a new writeable transaction.
   *
   * @return a new writeable transaction
   */
  WriteableTransaction getWriteableTransaction();

  /**
   * Remove all files for a backend of this storage.
   *
   * @throws StorageRuntimeException if removal fails