[#1096] Copy a value into Persistit once rather than twice, and write the l...

Valery Kharseko

3 hours ago d30ff782c1c046a28939278b2e50d012d945f362

[#1096] Copy a value into Persistit once rather than twice, and write the large values of PDBStorageTest through a small buffer pool (#1097)

 opendj-server-legacy/src/main/java/org/opends/server/backends/pdb/PDBStorage.java

@@ -139,6 +139,8 @@
  private static final String JOURNAL_NAME = VOLUME_NAME + "_journal";
  /** The buffer / page size used by the PersistIt storage. */
  private static final int BUFFER_SIZE = 16 * 1024;
  /** Encoded by {@link #bytesToValue(Value, ByteSequence)} for the header Persistit puts before a byte array. */
  private static final byte[] EMPTY_BYTES = new byte[0];

  /** PersistIt implementation of the {@link Cursor} interface. */
  private final class CursorImpl implements Cursor<ByteString, ByteString>
@@ -578,7 +580,7 @@
          }
          else
          {
            ex.getValue().clear().putByteArray(newValue.toByteArray());
            bytesToValue(ex.getValue(), newValue);
            ex.store();
          }
          return true;
@@ -1542,9 +1544,21 @@
    return key.clear().appendByteArray(tmp, 0, tmp.length);
  }

  /**
   * Encodes the bytes as a byte array value, copying them once, straight into the encoded bytes of the value.
   * {@code putByteArray(bytes.toByteArray())} would copy them twice, and for a value of 63 MB the extra copy is
   * one more humongous array live next to the source and the value buffer. Persistit encodes a byte array as a
   * header followed by the bytes as they are, so the header is taken from Persistit itself - by encoding an empty
   * array - and the bytes are appended behind it.
   */
  private static Value bytesToValue(final Value value, final ByteSequence bytes)
  {
    value.clear().putByteArray(bytes.toByteArray());
    value.clear().putByteArray(EMPTY_BYTES);
    final int headerSize = value.getEncodedSize();
    value.ensureFit(bytes.length());
    // ensureFit() may have replaced the encoded bytes, so they are read only after it
    bytes.copyTo(value.getEncodedBytes(), headerSize);
    value.setEncodedSize(headerSize + bytes.length());
    return value;
  }


 opendj-server-legacy/src/test/java/org/opends/server/backends/pdb/PDBStorageTest.java

@@ -17,6 +17,7 @@
package org.opends.server.backends.pdb;

import static org.assertj.core.api.Assertions.*;
import static org.mockito.AdditionalAnswers.delegatesTo;
import static org.mockito.Mockito.*;
import static org.forgerock.opendj.config.ConfigurationMock.*;
import static org.opends.server.util.StaticUtils.*;
@@ -36,7 +37,9 @@
import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.opendj.config.server.ConfigChangeResult;
import org.forgerock.opendj.config.server.ConfigException;
import org.forgerock.opendj.ldap.ByteSequence;
import org.forgerock.opendj.ldap.ByteString;
import org.forgerock.opendj.ldap.ByteStringBuilder;
import org.forgerock.opendj.ldap.ResultCode;
import org.mockito.ArgumentCaptor;
import org.opends.server.DirectoryServerTestCase;
@@ -49,6 +52,7 @@
import org.opends.server.backends.pluggable.spi.StorageInUseException;
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.WriteOperation;
import org.opends.server.backends.pluggable.spi.WriteableTransaction;
import org.opends.server.core.DirectoryServer;
@@ -73,6 +77,8 @@
  private static final long SHORT_RETRY_WINDOW_NANOS = 200L * 1000L * 1000L; //200 ms
  /** An attempt long enough to outlast {@link #SHORT_RETRY_WINDOW_NANOS} on its own, in milliseconds. */
  private static final long ATTEMPT_LONGER_THAN_SHORT_WINDOW_MS = 300;
  /** The buffer pool {@link #testCanAddLargeValues()} writes through, well under the 20% of the other methods. */
  private static final long LARGE_VALUES_DB_CACHE_SIZE = 16L * MB;

  private final TreeName treeName = new TreeName("dc=test", "test");
  private ServerContext serverContext;
@@ -159,16 +165,20 @@
  }

  /**
   * The sources are wrapped rather than copied: a value on its way into Persistit is copied twice more -
   * {@code toByteArray()} and the value buffer of the exchange, which doubles up to 64 MB - so with a copy
   * here as well the 63 MB value had four copies of itself live at once, on top of the buffer pool and the
   * server: about 310 MB left after a collection, in a JVM of 512 MB, which one CI leg ran out of. Wrapped,
   * about 165 MB. The three values stay in one transaction on purpose: the value buffer the 32 MB one
   * grew fits the 63 MB one without growing again.
   * The sources are wrapped rather than copied, and the storage is reopened with a buffer pool of
   * {@link #LARGE_VALUES_DB_CACHE_SIZE}: in a JVM of 512 MB each 63 MB array needs a free run of 64 regions,
   * and two CI legs ran out of one. A value on its way into Persistit is copied once more, into the value buffer
   * of the exchange, which doubles up to 64 MB; the 20% cache of the other methods would allocate 76 MB of
   * buffers up front, which this test does not need. The three values stay in one transaction on purpose:
   * the value buffer the 32 MB one grew fits the 63 MB one without growing again.
   */
  @Test
  public void testCanAddLargeValues() throws Exception
  {
    closeAndRemove(storage);
    storage = new PDBStorage(createBackendCfg(LARGE_VALUES_DB_CACHE_SIZE), serverContext);
    storage.open(AccessMode.READ_WRITE);

    storage.write(new WriteOperation()
    {
      private final TreeName treeName = new TreeName("dc=test", "test");
@@ -185,6 +195,93 @@
    });
  }

  /**
   * A value is copied straight into the encoded bytes of the Persistit value, behind the header Persistit
   * writes for a byte array: each of these reads back as it was written - an empty one, one which starts past
   * the offset of the array behind it, one which is not a {@link ByteString} at all, and one which outgrows the
   * encoded bytes the value had, so that it is copied into the ones {@code ensureFit()} put in their place.
   */
  @Test
  public void testValuesReadBackAsWritten() throws Exception
  {
    final ByteString empty = ByteString.empty();
    final ByteString inTheMiddle = wrap(new byte[] { 9, 1, 2, 3, 9 }, 1, 3);
    final ByteStringBuilder builder = new ByteStringBuilder().appendUtf8("built");
    final byte[] patterned = new byte[64 * KB];
    for (int i = 0; i < patterned.length; i++)
    {
      patterned[i] = (byte) i;
    }
    final ByteString large = wrap(patterned);
    createTree();
    storage.write(new WriteOperation()
    {
      @Override
      public void run(WriteableTransaction txn) throws Exception
      {
        txn.put(treeName, valueOfUtf8("empty"), empty);
        txn.put(treeName, valueOfUtf8("inTheMiddle"), inTheMiddle);
        txn.put(treeName, valueOfUtf8("builder"), builder);
        txn.put(treeName, valueOfUtf8("large"), large);
      }
    });

    assertThat(read("empty")).isEqualTo(empty);
    assertThat(read("inTheMiddle")).isEqualTo(valueOfBytes(new byte[] { 1, 2, 3 }));
    assertThat(read("builder")).isEqualTo(valueOfUtf8("built"));
    assertThat(read("large")).isEqualTo(large);
  }

  /**
   * A put copies the value once, straight into the encoded bytes of the Persistit value: the source is never
   * asked for a copy of its own, which {@code putByteArray(bytes.toByteArray())} would make.
   */
  @Test
  public void testPutValueIsCopiedOnlyOnce() throws Exception
  {
    final ByteString large = wrap(new byte[64 * KB]);
    final ByteSequence value = mock(ByteSequence.class, delegatesTo(large));
    createTree();
    storage.write(new WriteOperation()
    {
      @Override
      public void run(WriteableTransaction txn) throws Exception
      {
        txn.put(treeName, valueOfUtf8("large"), value);
      }
    });

    verify(value, never()).toByteArray();
    assertThat(read("large")).isEqualTo(large);
  }

  /** The new value an update computes is copied once as well, the same way as the value of a put. */
  @Test
  public void testUpdatedValueIsCopiedOnlyOnce() throws Exception
  {
    final ByteString large = wrap(new byte[64 * KB]);
    final ByteSequence value = mock(ByteSequence.class, delegatesTo(large));
    createTree();
    storage.write(new WriteOperation()
    {
      @Override
      public void run(WriteableTransaction txn) throws Exception
      {
        txn.update(treeName, valueOfUtf8("large"), new UpdateFunction()
        {
          @Override
          public ByteSequence computeNewValue(ByteSequence oldValue)
          {
            return value;
          }
        });
      }
    });

    verify(value, never()).toByteArray();
    assertThat(read("large")).isEqualTo(large);
  }

  @Test
  public void testExchangeWithSmallValuesAreReleasedToPool() throws Exception
  {