class Oplog implements CompactableOplog, Flushable {

  @Override
  public int compact(OplogCompactor compactor) {
    if (!needsCompaction()) {
      return 0; // @todo check new logic that deals with not compacting oplogs
                // which have unrecovered regions
    }
    isCompactorThread.set(Boolean.TRUE);
    assert calledByCompactorThread();
    getParent().acquireCompactorReadLock();
    try {
      if (!compactor.keepCompactorRunning()) {
        return 0;
      }
      lockCompactor();
      try {
        if (hasNoLiveValues()) {
          handleNoLiveValues();
          return 0; // do this while holding compactorLock
        }

        // Start with a fresh wrapper on every compaction so that
        // if previous run used some high memory byte array which was
        // exceptional, it gets garbage collected.
        long opStart = getStats().getStatTime();
        BytesAndBitsForCompactor wrapper = new BytesAndBitsForCompactor();

        DiskEntry de;
        DiskEntry lastDe = null;
        boolean compactFailed = /*
                                 * getParent().getOwner().isDestroyed ||
                                 */!compactor.keepCompactorRunning();
        int totalCount = 0;
        for (DiskRegionInfo dri : this.regionMap.values()) {
          final DiskRegionView dr = dri.getDiskRegion();
          if (dr == null)
            continue;
          boolean didCompact = false;
          while ((de = dri.getNextLiveEntry()) != null) {
            if (/*
                 * getParent().getOwner().isDestroyed ||
                 */!compactor.keepCompactorRunning()) {
              compactFailed = true;
              break;
            }
            if (lastDe != null) {
              if (lastDe == de) {
                throw new IllegalStateException("compactor would have gone into infinite loop");
              }
              assert lastDe != de;
            }
            lastDe = de;
            didCompact = false;
            synchronized (de) { // fix for bug 41797
              DiskId did = de.getDiskId();
              assert did != null;
              synchronized (did) {
                long oplogId = did.getOplogId();
                if (oplogId != getOplogId()) {
                  continue;
                }
                boolean toCompact = getBytesAndBitsForCompaction(dr, de, wrapper);
                if (toCompact) {
                  if (oplogId != did.getOplogId()) {
                    // @todo: Is this even possible? Perhaps I should just assert here
                    // skip this one, its oplogId changed
                    if (!wrapper.isReusable()) {
                      wrapper = new BytesAndBitsForCompactor();
                    } else if (wrapper.getOffHeapData() != null) {
                      wrapper.setOffHeapData(null, (byte) 0);
                    }
                    continue;
                  }
                  // write it to the current oplog
                  getOplogSet().getChild().copyForwardModifyForCompact(dr, de, wrapper);
                  // the did's oplogId will now be set to the current active oplog
                  didCompact = true;
                }
              } // did
            } // de
            if (didCompact) {
              totalCount++;
              getStats().endCompactionUpdate(opStart);
              opStart = getStats().getStatTime();
              // Check if the value byte array happens to be any of the
              // constant
              // static byte arrays or references the value byte array of
              // underlying RegionEntry.
              // If so for preventing data corruption across regions
              // ( in case of static byte arrays) & for RegionEntry,
              // recreate the wrapper
              if (!wrapper.isReusable()) {
                wrapper = new BytesAndBitsForCompactor();
              }
            }
          }
        }

        cleanupAfterCompaction(compactFailed);
        return totalCount;
      } finally {
        unlockCompactor();
      }
    } finally {
      getParent().releaseCompactorReadLock();
      assert calledByCompactorThread();
      isCompactorThread.remove();
    }
  }

    Oplog(long oplogId, PersistentOplogSet parent, DirectoryHolder dirHolder);
  private  Oplog(long oplogId, DirectoryHolder dirHolder, Oplog prevOplog);
    Oplog(long oplogId, PersistentOplogSet parent);

}