OPENDJ-1848 CR-6383: Remove Storage getRMW and putIfAbsent methods

Matthew Swift

24.03.2015 b506a6cd184bd8cf477f7d9a7f968c990f153528

OPENDJ-1848 CR-6383: Remove Storage getRMW and putIfAbsent methods

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

@@ -346,12 +346,6 @@
    }

    @Override
    public ByteString getRMW(final TreeName treeName, final ByteSequence key)
    {
      return read(treeName, key);
    }

    @Override
    public Cursor openCursor(final TreeName treeName)
    {
      try
@@ -388,34 +382,6 @@
    }

    @Override
    public boolean putIfAbsent(final TreeName treeName, final ByteSequence key,
        final ByteSequence value)
    {
      try
      {
        // There is no CAS (Compare And Swap) operation to do here :)
        // Following code is fine because Persistit provides snapshot isolation.
        // If another thread tries to update the same key, we'll get a RollbackException
        // And the write operation will be retried (see write() method in this class)
        final Exchange ex = getExchangeFromCache(treeName);
        bytesToKey(ex.getKey(), key);
        ex.fetch();
        final Value exValue = ex.getValue();
        if (exValue.isDefined())
        {
          return false;
        }
        bytesToValue(exValue, value);
        ex.store();
        return true;
      }
      catch (final Exception e)
      {
        throw new StorageRuntimeException(e);
      }
    }

    @Override
    public ByteString read(final TreeName treeName, final ByteSequence key)
    {
      try
@@ -449,8 +415,15 @@
        final ByteSequence newValue = f.computeNewValue(oldValue);
        if (!equals(newValue, oldValue))
        {
          ex.getValue().clear().putByteArray(newValue.toByteArray());
          ex.store();
          if (newValue == null)
          {
            ex.remove();
          }
          else
          {
            ex.getValue().clear().putByteArray(newValue.toByteArray());
            ex.store();
          }
          return true;
        }
        return false;

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

@@ -58,21 +58,19 @@
  }

  /**
   * Insert a new record into the DN database.
   * Adds a new record into the DN database replacing any existing record having the same DN.
   * @param txn A JE database transaction to be used for the database operation,
   * or null if none.
   * @param dn The entry DN, which is the key to the record.
   * @param id The entry ID, which is the value of the record.
   * @return true if the record was inserted, false if a record with that key
   * already exists.
   * @throws StorageRuntimeException If an error occurred while attempting to insert
   * the new record.
   */
  boolean insert(WriteableStorage txn, DN dn, EntryID id) throws StorageRuntimeException
  void put(WriteableStorage txn, DN dn, EntryID id) throws StorageRuntimeException
  {
    ByteString key = dnToDNKey(dn, prefixRDNComponents);
    ByteString value = id.toByteString();
    return txn.putIfAbsent(getName(), key, value);
    txn.create(getName(), key, value);
  }

  /**
@@ -109,15 +107,4 @@
    }
    return null;
  }

  EntryID getRMW(ReadableStorage txn, DN dn) throws StorageRuntimeException
  {
    ByteString key = dnToDNKey(dn, prefixRDNComponents);
    ByteString value = txn.getRMW(getName(), key);
    if (value != null)
    {
      return new EntryID(value);
    }
    return null;
  }
}

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

@@ -48,6 +48,7 @@
import org.opends.server.backends.pluggable.spi.ReadableStorage;
import org.opends.server.backends.pluggable.spi.StorageRuntimeException;
import org.opends.server.backends.pluggable.spi.TreeName;
import org.opends.server.backends.pluggable.spi.UpdateFunction;
import org.opends.server.backends.pluggable.spi.WriteableStorage;
import org.opends.server.core.DirectoryServer;
import org.opends.server.core.SearchOperation;
@@ -114,13 +115,12 @@

  private ByteSequence encode(DN dn, Collection<String> col)
  {
    if (col != null)
    if (col != null && !col.isEmpty())
    {
      ByteStringBuilder b = new ByteStringBuilder();
      byte[] dnBytes = StaticUtils.getBytes(dn.toString());
      b.append(dnBytes.length);
      b.append(dnBytes);

      b.append(col.size());
      for (String s : col)
      {
@@ -130,7 +130,7 @@
      }
      return b;
    }
    return ByteString.empty();
    return null;
  }

  private Pair<DN, List<String>> decode(ByteSequence bs) throws StorageRuntimeException
@@ -161,7 +161,7 @@
  }

  /**
   * Insert a URI value in the referral database.
   * Puts a URI value in the referral database.
   *
   * @param txn A database transaction used for the update, or null if none is
   * required.
@@ -169,24 +169,31 @@
   * @param labeledURIs The labeled URI value of the ref attribute.
   * @throws StorageRuntimeException If an error occurs in the JE database.
   */
  private void insert(WriteableStorage txn, DN dn, Collection<String> labeledURIs) throws StorageRuntimeException
  private void put(final WriteableStorage txn, final DN dn, final Collection<String> labeledURIs)
      throws StorageRuntimeException
  {
    ByteString key = toKey(dn);

    ByteString oldValue = txn.getRMW(getName(), key);
    if (oldValue != null)
    final ByteString key = toKey(dn);
    txn.update(getName(), key, new UpdateFunction()
    {
      final Pair<DN, List<String>> dnAndUris = decode(oldValue);
      final Collection<String> newUris = dnAndUris.getSecond();
      if (newUris.addAll(labeledURIs))
      @Override
      public ByteSequence computeNewValue(ByteSequence oldValue)
      {
        txn.create(getName(), key, encode(dn, newUris));
        if (oldValue != null)
        {
          final Pair<DN, List<String>> dnAndUris = decode(oldValue);
          final Collection<String> newUris = dnAndUris.getSecond();
          if (newUris.addAll(labeledURIs))
          {
            return encode(dn, newUris);
          }
          return oldValue;
        }
        else
        {
          return encode(dn, labeledURIs);
        }
      }
    }
    else
    {
      txn.putIfAbsent(getName(), key, encode(dn, labeledURIs));
    }
    });
    containsReferrals = ConditionResult.TRUE;
  }

@@ -218,27 +225,30 @@
   * required.
   * @param dn The DN of the referral entry.
   * @param labeledURIs The URI value to be deleted.
   * @return true if the value was deleted, false if not.
   * @throws StorageRuntimeException If an error occurs in the JE database.
   */
  private boolean delete(WriteableStorage txn, DN dn, Collection<String> labeledURIs)
  private void delete(final WriteableStorage txn, final DN dn, final Collection<String> labeledURIs)
      throws StorageRuntimeException
  {
    ByteString key = toKey(dn);

    ByteString oldValue = txn.getRMW(getName(), key);
    if (oldValue != null)
    txn.update(getName(), key, new UpdateFunction()
    {
      final Pair<DN, List<String>> dnAndUris = decode(oldValue);
      final Collection<String> oldUris = dnAndUris.getSecond();
      if (oldUris.removeAll(labeledURIs))
      @Override
      public ByteSequence computeNewValue(ByteSequence oldValue)
      {
        txn.create(getName(), key, encode(dn, oldUris));
        containsReferrals = containsReferrals(txn);
        return true;
        if (oldValue != null)
        {
          final Pair<DN, List<String>> dnAndUris = decode(oldValue);
          final Collection<String> oldUris = dnAndUris.getSecond();
          if (oldUris.removeAll(labeledURIs))
          {
            return encode(dn, oldUris);
          }
        }
        return oldValue;
      }
    }
    return false;
    });
    containsReferrals = containsReferrals(txn);
  }

  /**
@@ -296,7 +306,7 @@
          case ADD:
            if (a != null)
            {
              insert(txn, entryDN, toStrings(a));
              put(txn, entryDN, toStrings(a));
            }
            break;

@@ -319,7 +329,7 @@
            delete(txn, entryDN);
            if (a != null)
            {
              insert(txn, entryDN, toStrings(a));
              put(txn, entryDN, toStrings(a));
            }
            break;
        }
@@ -365,18 +375,16 @@
   * @param txn A database transaction used for the update, or null if none is
   * required.
   * @param entry The entry to be added.
   * @return True if the entry was added successfully or False otherwise.
   * @throws StorageRuntimeException If an error occurs in the JE database.
   */
  boolean addEntry(WriteableStorage txn, Entry entry)
  void addEntry(WriteableStorage txn, Entry entry)
       throws StorageRuntimeException
  {
    Set<String> labeledURIs = entry.getReferralURLs();
    if (labeledURIs != null)
    {
      insert(txn, entry.getName(), labeledURIs);
      put(txn, entry.getName(), labeledURIs);
    }
    return true;
  }

  /**

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

@@ -1528,30 +1528,9 @@
            }

            EntryID entryID = rootContainer.getNextEntryID();

            // Insert into dn2id.
            if (!dn2id.insert(txn, entry.getName(), entryID))
            {
              // Do not ever expect to come through here.
              throw new DirectoryException(ResultCode.ENTRY_ALREADY_EXISTS, ERR_JEB_ADD_ENTRY_ALREADY_EXISTS.get(entry
                  .getName()));
            }

            // Update the referral database for referral entries.
            if (!dn2uri.addEntry(txn, entry))
            {
              // Do not ever expect to come through here.
              throw new DirectoryException(ResultCode.ENTRY_ALREADY_EXISTS, ERR_JEB_ADD_ENTRY_ALREADY_EXISTS.get(entry
                  .getName()));
            }

            // Insert into id2entry.
            if (!id2entry.insert(txn, entryID, entry))
            {
              // Do not ever expect to come through here.
              throw new DirectoryException(ResultCode.ENTRY_ALREADY_EXISTS, ERR_JEB_ADD_ENTRY_ALREADY_EXISTS.get(entry
                  .getName()));
            }
            dn2id.put(txn, entry.getName(), entryID);
            dn2uri.addEntry(txn, entry);
            id2entry.put(txn, entryID, entry);

            // Insert into the indexes, in index configuration order.
            final IndexBuffer indexBuffer = new IndexBuffer(EntryContainer.this);
@@ -1811,7 +1790,8 @@
      {
        leafDNKey = dnToDNKey(targetDN, baseDN.size());
      }
      ByteString value = txn.getRMW(dn2id.getName(), leafDNKey);
      // FIXME: previously this used a RMW lock - see OPENDJ-1878.
      ByteString value = txn.read(dn2id.getName(), leafDNKey);
      if (value == null)
      {
        LocalizableMessage message = ERR_JEB_DELETE_NO_SUCH_OBJECT.get(targetDN);
@@ -1831,7 +1811,8 @@
    }

    // Check that the entry exists in id2entry and read its contents.
    Entry entry = id2entry.getRMW(txn, leafID);
    // FIXME: previously this used a RMW lock - see OPENDJ-1878.
    Entry entry = id2entry.get(txn, leafID);
    if (entry == null)
    {
      throw new DirectoryException(DirectoryServer.getServerErrorResultCode(),
@@ -2023,7 +2004,7 @@
        {
          try
          {
            EntryID entryID = dn2id.getRMW(txn, newEntry.getName());
            EntryID entryID = dn2id.get(txn, newEntry.getName());
            if (entryID == null)
            {
              LocalizableMessage message = ERR_JEB_MODIFY_NO_SUCH_OBJECT.get(newEntry.getName());
@@ -2138,6 +2119,7 @@
  void renameEntry(final DN currentDN, final Entry entry, final ModifyDNOperation modifyDNOperation)
      throws StorageRuntimeException, DirectoryException, CanceledOperationException
  {
    // FIXME: consistency + isolation cannot be maintained lock free - see OPENDJ-1878.
    try
    {
      storage.write(new WriteOperation()
@@ -2374,11 +2356,8 @@
                           ModifyDNOperation modifyDNOperation)
      throws DirectoryException, StorageRuntimeException
  {
    if (!dn2id.insert(txn, newEntry.getName(), newID))
    {
      LocalizableMessage message = ERR_JEB_MODIFYDN_ALREADY_EXISTS.get(newEntry.getName());
      throw new DirectoryException(ResultCode.ENTRY_ALREADY_EXISTS, message);
    }
    // FIXME: the core server should validate that the new subtree location is empty.
    dn2id.put(txn, newEntry.getName(), newID);
    id2entry.put(txn, newID, newEntry);
    dn2uri.addEntry(txn, newEntry);


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

@@ -309,35 +309,6 @@
    }
  }



  /**
   * Insert a record into the entry database.
   *
   * @param txn The database transaction or null if none.
   * @param id The entry ID which forms the key.
   * @param entry The LDAP entry.
   * @return true if the entry was inserted, false if a record with that
   *         ID already existed.
   * @throws StorageRuntimeException If an error occurs in the JE database.
   * @throws  DirectoryException  If a problem occurs while attempting to encode
   *                              the entry.
   */
  boolean insert(WriteableStorage txn, EntryID id, Entry entry) throws StorageRuntimeException, DirectoryException
  {
    ByteString key = id.toByteString();
    EntryCodec codec = acquireEntryCodec();
    try
    {
      ByteString value = codec.encodeInternal(entry, dataConfig);
      return txn.putIfAbsent(getName(), key, value);
    }
    finally
    {
      codec.release();
    }
  }

  /**
   * Write a record in the entry database.
   *
@@ -392,11 +363,6 @@
    return get0(id, txn.read(getName(), id.toByteString()));
  }

  public Entry getRMW(ReadableStorage txn, EntryID id) throws DirectoryException, StorageRuntimeException
  {
    return get0(id, txn.getRMW(getName(), id.toByteString()));
  }

  private Entry get0(EntryID id, ByteString value) throws DirectoryException
  {
    if (value == null)

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

@@ -46,6 +46,7 @@
import org.opends.server.backends.pluggable.spi.ReadableStorage;
import org.opends.server.backends.pluggable.spi.StorageRuntimeException;
import org.opends.server.backends.pluggable.spi.TreeName;
import org.opends.server.backends.pluggable.spi.UpdateFunction;
import org.opends.server.backends.pluggable.spi.WriteableStorage;
import org.opends.server.types.Entry;
import org.opends.server.types.Modification;
@@ -243,16 +244,11 @@
      }
      else if (trusted)
      {
        if (deletedIDs != null)
        {
          logIndexCorruptError(txn, key);
        }

        if (isNotNullOrEmpty(addedIDs)
            && !txn.putIfAbsent(getName(), key, addedIDs.toByteString()))
        {
          updateKeyWithRMW(txn, key, deletedIDs, addedIDs);
        }
        /*
         * The key was not present, but we cannot simply add it because another thread may have
         * added since.
         */
        updateKeyWithRMW(txn, key, deletedIDs, addedIDs);
      }
    }
  }
@@ -262,43 +258,44 @@
    return entryIDSet == null || entryIDSet.size() == 0;
  }

  private boolean isNotNullOrEmpty(EntryIDSet entryIDSet)
  private boolean isNotEmpty(EntryIDSet entryIDSet)
  {
    return entryIDSet != null && entryIDSet.size() > 0;
  }

  private void updateKeyWithRMW(WriteableStorage txn, ByteString key, EntryIDSet deletedIDs, EntryIDSet addedIDs)
      throws StorageRuntimeException
  private void updateKeyWithRMW(final WriteableStorage txn, final ByteString key, final EntryIDSet deletedIDs,
      final EntryIDSet addedIDs) throws StorageRuntimeException
  {
    final ByteString value = txn.getRMW(getName(), key);
    if (value != null)
    txn.update(getName(), key, new UpdateFunction()
    {
      EntryIDSet entryIDSet = computeEntryIDSet(key, value, deletedIDs, addedIDs);
      ByteString after = entryIDSet.toByteString();
      if (!after.isEmpty())
      @Override
      public ByteSequence computeNewValue(final ByteSequence oldValue)
      {
        txn.create(getName(), key, after);
        if (oldValue != null)
        {
          EntryIDSet entryIDSet = computeEntryIDSet(key, oldValue.toByteString(), deletedIDs, addedIDs);
          ByteString after = entryIDSet.toByteString();
          /*
           * If there are no more IDs then return null indicating that the record should be removed.
           * If index is not trusted then this will cause all subsequent reads for this key to
           * return undefined set.
           */
          return after.isEmpty() ? null : after;
        }
        else if (trusted)
        {
          if (deletedIDs != null)
          {
            logIndexCorruptError(txn, key);
          }
          if (isNotEmpty(addedIDs))
          {
            return addedIDs.toByteString();
          }
        }
        return null; // no change.
      }
      else
      {
        // No more IDs, so remove the key. If index is not
        // trusted then this will cause all subsequent reads
        // for this key to return undefined set.
        txn.delete(getName(), key);
      }
    }
    else if (trusted)
    {
      if (deletedIDs != null)
      {
        logIndexCorruptError(txn, key);
      }

      if (isNotNullOrEmpty(addedIDs))
      {
        txn.putIfAbsent(getName(), key, addedIDs.toByteString());
      }
    }
    });
  }

  private EntryIDSet computeEntryIDSet(ByteString key, ByteString value, EntryIDSet deletedIDs, EntryIDSet addedIDs)

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

@@ -227,21 +227,10 @@
    }
  }

  private void store(final TreeName treeName, final byte[] key, final ByteStringBuilder value)
      throws DirectoryException
  {
    if (!putNoOverwrite(treeName, key, value))
    {
      final LocalizableMessage m = ERR_JEB_COMPSCHEMA_CANNOT_STORE_MULTIPLE_FAILURES.get();
      throw new DirectoryException(DirectoryServer.getServerErrorResultCode(), m);
    }
  }

  private boolean putNoOverwrite(final TreeName treeName, final byte[] key, final ByteStringBuilder value)
  private boolean store(final TreeName treeName, final byte[] key, final ByteStringBuilder value)
      throws DirectoryException
  {
    final ByteString keyEntry = ByteString.wrap(key);
    final ByteString valueEntry = ByteString.wrap(value.getBackingArray(), 0, value.length());
    try
    {
      storage.write(new WriteOperation()
@@ -249,11 +238,7 @@
        @Override
        public void run(WriteableStorage txn) throws Exception
        {
          if (!txn.putIfAbsent(treeName, keyEntry, valueEntry))
          {
            final LocalizableMessage m = ERR_JEB_COMPSCHEMA_CANNOT_STORE_STATUS.get(false);
            throw new DirectoryException(DirectoryServer.getServerErrorResultCode(), m);
          }
          txn.create(treeName, keyEntry, value);
        }
      });
      return true;

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

@@ -112,15 +112,6 @@
    }

    @Override
    public ByteString getRMW(final TreeName name, final ByteSequence key)
    {
      final ByteString value = txn.getRMW(name, key);
      logger.trace("Storage@%s.ReadableStorage@%s.getRMW(%s, %s, %s) = %s",
          storageId(), id(), backendId, name, hex(key), hex(value));
      return value;
    }

    @Override
    public Cursor openCursor(final TreeName name)
    {
      final Cursor cursor = txn.openCursor(name);
@@ -197,15 +188,6 @@
    }

    @Override
    public ByteString getRMW(final TreeName name, final ByteSequence key)
    {
      final ByteString value = txn.getRMW(name, key);
      logger.trace("Storage@%s.WriteableStorage@%s.getRMW(%s, %s, %s) = %s",
          storageId(), id(), backendId, name, hex(key), hex(value));
      return value;
    }

    @Override
    public Cursor openCursor(final TreeName name)
    {
      final Cursor cursor = txn.openCursor(name);
@@ -223,15 +205,6 @@
    }

    @Override
    public boolean putIfAbsent(final TreeName name, final ByteSequence key, final ByteSequence value)
    {
      final boolean isCreated = txn.putIfAbsent(name, key, value);
      logger.trace("Storage@%s.WriteableStorage@%s.putIfAbsent(%s, %s, %s, %s) = %s",
          storageId(), id(), backendId, name, hex(key), hex(value), isCreated);
      return isCreated;
    }

    @Override
    public ByteString read(final TreeName name, final ByteSequence key)
    {
      final ByteString value = txn.read(name, key);

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

@@ -227,7 +227,7 @@

  /** {@inheritDoc} */
  @Override
  public void open(WriteableStorage txn) throws StorageRuntimeException
  void open(WriteableStorage txn) throws StorageRuntimeException
  {
    super.open(txn);

@@ -402,12 +402,12 @@
      DirectoryException
  {
    ByteString key = encodeKey(entryID, values, types);
    return getSortValuesSet(txn, key, false);
    ByteString value = txn.read(getName(), key);
    return decodeSortValuesSet(key, value);
  }

  private SortValuesSet getSortValuesSet(ReadableStorage txn, ByteString key, boolean isRMW)
  private SortValuesSet decodeSortValuesSet(ByteString key, ByteString value)
  {
    ByteString value = isRMW ? txn.getRMW(getName(), key) : txn.read(getName(), key);
    if (value == null)
    {
      // There are no records in the database
@@ -544,7 +544,12 @@
        break;
      }

      final SortValuesSet sortValuesSet = getSortValuesSet(txn, key, true);
      /*
       * FIXME: replace getRMW+updates with single call to update()
       */
      ByteString value = txn.read(getName(), key);
      final SortValuesSet sortValuesSet = decodeSortValuesSet(key, value);

      int oldSize = sortValuesSet.size();
      if(key.length() == 0)
      {
@@ -581,6 +586,10 @@
      int newSize = sortValuesSet.size();
      if(newSize >= sortedSetCapacity)
      {
        /*
         * FIXME: is one record becoming two or three? The call to split() looks like it is changing
         * the key.
         */
        SortValuesSet splitSortValuesSet = sortValuesSet.split(newSize / 2);
        txn.create(getName(), splitSortValuesSet.getKeyBytes(), splitSortValuesSet.toByteString()); // splitAfter
        txn.create(getName(), sortValuesSet.getKeyBytes(), sortValuesSet.toByteString()); // after

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

@@ -47,20 +47,6 @@
  ByteString read(TreeName treeName, ByteSequence key);

  /**
   * Reads the record's value associated to the provided key in a Read-Modify-Write fashion, in the
   * tree whose name is provided.
   *
   * @param treeName
   *          the tree name
   * @param key
   *          the record's key
   * @return the record's value, or {@code null} if none exists
   * @deprecated use {@link WriteableStorage#update(TreeName, ByteSequence, UpdateFunction)} instead
   */
  @Deprecated
  ByteString getRMW(TreeName treeName, ByteSequence key);

  /**
   * Opens a cursor on the tree whose name is provided.
   *
   * @param treeName

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

@@ -72,22 +72,6 @@
  void create(TreeName treeName, ByteSequence key, ByteSequence value);

  /**
   * Creates a new record with the provided key and value, in the tree whose name is provided, if
   * the key was not previously associated to any record.
   *
   * @param treeName
   *          the tree name
   * @param key
   *          the key of the new record
   * @param value
   *          the value of the new record
   * @return {@code true} if the new record could be created, {@code false} otherwise
   * @deprecated use {@link #update(TreeName, ByteSequence, UpdateFunction)} instead
   */
  @Deprecated
  boolean putIfAbsent(TreeName treeName, ByteSequence key, ByteSequence value);

  /**
   * Updates a record with the provided key according to the new value computed by the update function.
   *
   * @param treeName

 opendj-server-legacy/src/main/java/org/opends/server/workflowelement/localbackend/LocalBackendModifyDNOperation.java

@@ -308,6 +308,12 @@
    // Check for a request to cancel this operation.
    checkIfCanceled(false);

    /*
     * FIXME: we lock the target DN and the renamed target DN, but not the parent of the target DN,
     * which seems inconsistent with the add operation implementation. Specifically, this
     * implementation does not defend against concurrent deletes of the parent of the renamed entry.
     */

    // Acquire write locks for the current and new DN.
    final Lock currentLock = LockManager.lockWrite(entryDN);
    Lock newLock = null;