*/
package org.opendaylight.controller.akka.segjournal;
-import static com.google.common.base.Verify.verify;
-
import akka.actor.ActorSystem;
-import akka.persistence.AtomicWrite;
import akka.persistence.PersistentRepr;
import com.codahale.metrics.Histogram;
-import io.atomix.storage.StorageLevel;
-import io.atomix.storage.journal.Indexed;
+import com.google.common.base.VerifyException;
+import io.atomix.storage.journal.JournalReader;
+import io.atomix.storage.journal.JournalSerdes;
+import io.atomix.storage.journal.JournalWriter;
+import io.atomix.storage.journal.SegmentedByteBufJournal;
import io.atomix.storage.journal.SegmentedJournal;
-import io.atomix.storage.journal.SegmentedJournalReader;
-import io.atomix.storage.journal.SegmentedJournalWriter;
-import io.atomix.utils.serializer.Namespace;
+import io.atomix.storage.journal.StorageLevel;
import java.io.File;
import java.io.Serializable;
+import java.util.ArrayList;
import java.util.List;
import org.opendaylight.controller.akka.segjournal.DataJournalEntry.FromPersistence;
import org.opendaylight.controller.akka.segjournal.DataJournalEntry.ToPersistence;
import org.opendaylight.controller.akka.segjournal.SegmentedJournalActor.ReplayMessages;
import org.opendaylight.controller.akka.segjournal.SegmentedJournalActor.WriteMessages;
+import org.opendaylight.controller.akka.segjournal.SegmentedJournalActor.WrittenMessages;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.jdk.javaapi.CollectionConverters;
/**
* Version 0 data journal, where every journal entry maps to exactly one segmented file entry.
- *
- * @author Robert Varga
*/
final class DataJournalV0 extends DataJournal {
private static final Logger LOG = LoggerFactory.getLogger(DataJournalV0.class);
DataJournalV0(final String persistenceId, final Histogram messageSize, final ActorSystem system,
final StorageLevel storage, final File directory, final int maxEntrySize, final int maxSegmentSize) {
super(persistenceId, messageSize);
- entries = SegmentedJournal.<DataJournalEntry>builder()
- .withStorageLevel(storage).withDirectory(directory).withName("data")
- .withNamespace(Namespace.builder()
- .register(new DataJournalEntrySerializer(system), FromPersistence.class, ToPersistence.class)
- .build())
- .withMaxEntrySize(maxEntrySize).withMaxSegmentSize(maxSegmentSize)
- .build();
+
+ final var serdes = JournalSerdes.builder()
+ .register(new DataJournalEntrySerdes(system), FromPersistence.class, ToPersistence.class)
+ .build();
+
+ entries = new SegmentedJournal<>(SegmentedByteBufJournal.builder()
+ .withDirectory(directory)
+ .withName("data")
+ .withStorageLevel(storage)
+ .withMaxEntrySize(maxEntrySize)
+ .withMaxSegmentSize(maxSegmentSize)
+ .build(), serdes.toReadMapper(), serdes.toWriteMapper());
}
@Override
long lastWrittenSequenceNr() {
- return entries.writer().getLastIndex();
+ return entries.lastIndex();
}
@Override
@Override
void close() {
+ flush();
entries.close();
}
+ @Override
+ void flush() {
+ entries.writer().flush();
+ }
+
@Override
@SuppressWarnings("checkstyle:illegalCatch")
void handleReplayMessages(final ReplayMessages message, final long fromSequenceNr) {
- try (SegmentedJournalReader<DataJournalEntry> reader = entries.openReader(fromSequenceNr)) {
- int count = 0;
- while (reader.hasNext() && count < message.max) {
- final Indexed<DataJournalEntry> next = reader.next();
- if (next.index() > message.toSequenceNr) {
- break;
- }
-
- LOG.trace("{}: replay {}", persistenceId, next);
- updateLargestSize(next.size());
- final DataJournalEntry entry = next.entry();
- verify(entry instanceof FromPersistence, "Unexpected entry %s", entry);
-
- final PersistentRepr repr = ((FromPersistence) entry).toRepr(persistenceId, next.index());
- LOG.debug("{}: replaying {}", persistenceId, repr);
- message.replayCallback.accept(repr);
- count++;
- }
- LOG.debug("{}: successfully replayed {} entries", persistenceId, count);
+ try (var reader = entries.openReader(fromSequenceNr)) {
+ handleReplayMessages(reader, message);
} catch (Exception e) {
LOG.warn("{}: failed to replay messages for {}", persistenceId, message, e);
message.promise.failure(e);
}
}
+ private void handleReplayMessages(final JournalReader<DataJournalEntry> reader, final ReplayMessages message) {
+ int count = 0;
+ while (count < message.max && reader.getNextIndex() <= message.toSequenceNr) {
+ final var repr = reader.tryNext((index, entry, size) -> {
+ LOG.trace("{}: replay index={} entry={}", persistenceId, index, entry);
+ updateLargestSize(size);
+ if (entry instanceof FromPersistence fromPersistence) {
+ return fromPersistence.toRepr(persistenceId, index);
+ }
+ throw new VerifyException("Unexpected entry " + entry);
+ });
+
+ if (repr == null) {
+ break;
+ }
+
+ LOG.debug("{}: replaying {}", persistenceId, repr);
+ message.replayCallback.accept(repr);
+ count++;
+ }
+ LOG.debug("{}: successfully replayed {} entries", persistenceId, count);
+ }
+
@Override
@SuppressWarnings("checkstyle:illegalCatch")
- void handleWriteMessages(final WriteMessages message) {
+ WrittenMessages handleWriteMessages(final WriteMessages message) {
final int count = message.size();
- final SegmentedJournalWriter<DataJournalEntry> writer = entries.writer();
+ final var responses = new ArrayList<>();
+ final var writer = entries.writer();
+ long writtenBytes = 0;
for (int i = 0; i < count; ++i) {
- final long mark = writer.getLastIndex();
- final AtomicWrite request = message.getRequest(i);
+ final long prevNextIndex = writer.getNextIndex();
+ final var request = message.getRequest(i);
- final List<PersistentRepr> reprs = CollectionConverters.asJava(request.payload());
- LOG.trace("{}: append {}/{}: {} items at mark {}", persistenceId, i, count, mark, reprs.size());
+ final var reprs = CollectionConverters.asJava(request.payload());
+ LOG.trace("{}: append {}/{}: {} items at mark {}", persistenceId, i, count, reprs.size(), prevNextIndex);
try {
- for (PersistentRepr repr : reprs) {
- final Object payload = repr.payload();
- if (!(payload instanceof Serializable)) {
- throw new UnsupportedOperationException("Non-serializable payload encountered "
- + payload.getClass());
- }
-
- LOG.trace("{}: starting append of {}", persistenceId, payload);
- final Indexed<ToPersistence> entry = writer.append(new ToPersistence(repr));
- final int size = entry.size();
- LOG.trace("{}: finished append of {} with {} bytes at {}", persistenceId, payload, size,
- entry.index());
- recordMessageSize(size);
- }
+ writtenBytes += writePayload(writer, reprs);
} catch (Exception e) {
- LOG.warn("{}: failed to write out request", persistenceId, e);
- message.setFailure(i, e);
- writer.truncate(mark);
+ LOG.warn("{}: failed to write out request {}/{} reverting to {}", persistenceId, i, count,
+ prevNextIndex, e);
+ responses.add(e);
+ writer.reset(prevNextIndex);
continue;
}
+ responses.add(null);
+ }
+
+ return new WrittenMessages(message, responses, writtenBytes);
+ }
+
+ private long writePayload(final JournalWriter<DataJournalEntry> writer, final List<PersistentRepr> reprs) {
+ long bytes = 0;
+ for (var repr : reprs) {
+ final Object payload = repr.payload();
+ if (!(payload instanceof Serializable)) {
+ throw new UnsupportedOperationException("Non-serializable payload encountered "
+ + payload.getClass());
+ }
- message.setSuccess(i);
+ LOG.trace("{}: starting append of {}", persistenceId, payload);
+ final var entry = writer.append(new ToPersistence(repr));
+ final int size = entry.size();
+ LOG.trace("{}: finished append of {} with {} bytes at {}", persistenceId, payload, size, entry.index());
+ recordMessageSize(size);
+ bytes += size;
}
- writer.flush();
+ return bytes;
}
}
Code Review / controller.git / blobdiff