Page History

...

Create a separate file for compaction but share the same allocation logic

The idea is that we reuse the current allocation logic but use a different file suffix for compaction. In this way, we can make minimum changes but achieve the goal of using a separate file for compaction.

Code Block

EntryLogWriteChannel createNewCompactionLog() throws IOException {
    synchronized (createCompactionLogLock) {
        List<File> writableDirs;
        try {
            writableDirs = new ArrayList<>(ledgerDirsManager.getWritableLedgerDirs());
        } catch (LedgerDirsManager.NoWritableLedgerDirException nlde) {
            // if all ledgers dirs are full, try to pick a ledger dir that has enough space for compaction log.
            writableDirs = ledgerDirsManager.getAllLedgerDirs();
            for (File dir : writableDirs) {
                if (dir.getUsableSpace() <= logSizeLimit) {
                    writableDirs.remove(dir);
                }
            }
        }
        EntryLogWriteChannel bc = allocateNewLogFile(COMPACTING_SUFFIX, writableDirs);
        LOG.info("Created new compaction logger {}.", bc.getLogId());
        return bc;
    }
}

...

Introduce Compaction Transactional Phases

We can add a separate class CompactionWorker to handle all the compaction logic in the same place. The main compact function would be like this:

Code Block

/**
 * Compaction is composed of 3 transactional phases:
 * 1. Scan data from entry log file into compaction log file and keep track of new entry locations
 * 2. Flush compaction log to make sure it's persisted in disk and roll a new compaction log
 * 3. Update entry locations to cache and flush index, finally delete the old entry log file
 * <p>
 * If compaction failed in any phase, abort the current compaction gracefully.
 */
synchronized boolean compact(EntryLogMetadata metadata) {
    if (metadata != null) {
        LOG.info("Compacting entry log {} : {}.", metadata.entryLogId, metadata);
        CompactionPhase scanEntryLog = new ScanEntryLogPhase(metadata);
        if (!scanEntryLog.run()) {
            LOG.info("Compaction for entry log {} end in ScanEntryLogPhase.", metadata.entryLogId);
            return false;
        }
        File compactionLogFile = entryLogger.getCurCompactionLogFile();
        CompactionPhase flushCompactionLog = new FlushCompactionLogPhase(metadata.entryLogId);
        if (!flushCompactionLog.run()) {
            LOG.info("Compaction for entry log {} end in FlushCompactionLogPhase.", metadata.entryLogId);
            return false;
        }
        File compactedLogFile = getCompactedLogFile(compactionLogFile, metadata.entryLogId);
        CompactionPhase updateIndex = new UpdateIndexPhase(compactedLogFile);
        if (!updateIndex.run()) {
            LOG.info("Compaction for entry log {} end in UpdateIndexPhase.", metadata.entryLogId);
            return false;
        }
        LOG.info("Compacted entry log : {}.", metadata.entryLogId);
        return true;
    }
    return false;
}

And this is the abstract class for all compaction phases.

Code Block

/**
 *

Code Block

/**
 * An abstract class that would be extended to be the actual transactional phases for compaction
 */
abstract static class CompactionPhase {
    private String phaseName = "";

    CompactionPhase() {
    }

    CompactionPhase(String phaseName) {
        this.phaseName = phaseName;
    }

    boolean run() {
        try {
            start();
            return complete();
        } catch (IOException e) {
            LOG.error("Encounter exception in compaction phase {}. Abort current compaction.", phaseName, e); current compaction.", phaseName, e);
            abort();
        }
        return false;
    }

    abstract void start() throws IOException;

    abstract boolean abortcomplete() throws IOException;

    abstract    }void abort();

}

Code Block

/**
 * Assume   we're compacting entry returnlog false;
1 to entry log }3.

 * The first abstractphase voidis start()to throws IOException;

    abstract boolean complete() throws IOException;

    abstract void abort();

}

Code Block

/**
 * This phase is to scan entry log and copy to compaction log file "x.log.compacting"scan entries in 1.log and copy them to compaction log file "3.log.compacting".
 * We'll try to allocate a new compaction log before scanning to make sure we have a log file to write.
 * If after scanning, there's no data written, it means there's no valid entries to be compacted,
 * so we can remove 1.log directly, clear the logentry directlylocations and end the compaction.
 * Otherwise.
 * Otherwise, we should move on to the next phase.
 *
 * If anything failed in this phase, we should move todelete the compaction log and clean the next phaseoffsets.
 */
class ScanEntryLogPhase extends CompactionPhase {
	...
}

Code Block

/**
 * ThisAssume phasewe're iscompacting tolog flush1 theto compactionlog log3.
 and* createThis aphase newis logto forflush nextthe compaction log.
 * When this phase starts, there should be a compaction log file like "x3.log.compacting"
 * When compaction log is flushed, in order to indicate this phase is completed,
 * a hardlink file "x3.log.y1.compacted" should be created,
 * and "x3.log.compacting" should be deleted.
 */
class FlushCompactionLogPhase extends CompactionPhase {
	...
}

Code Block

/**/**
 * Assume we're compacting log 1 to log 3.
 * This phase is to update the entry locations and flush the index.
 * When the phase start, there should be a compacted file like "x3.log.y1.compacted",
 * where x3 is the new entrycompaction logId being compacted to and y1 is the old entry logId.
 * WhenAfter the index the flushed successfully, a hardlink "x3.log" file should be created,
 * and x3.log.y1.compacted file should be deleted to indicate the phase is succeed.
 * 
 * This phase iscan also used to recover partially flushed index when we pass isInRecovery=true
 */
class UpdateIndexPhase extends CompactionPhase {
	...
}

...

Child pages

Versions Compared

Old Version 8

New Version Current

Key

Create a separate file for compaction but share the same allocation logic

Introduce Compaction Transactional Phases