opendaylight/md-sal/sal-akka-segmented-journal/src/main/java/org/opendaylight/controller/akka/segjournal/DataJournalV0.java

   1 /*
   2  * Copyright (c) 2019, 2020 PANTHEON.tech, s.r.o. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.akka.segjournal;
   9
  10 import akka.actor.ActorSystem;
  11 import akka.persistence.PersistentRepr;
  12 import com.codahale.metrics.Histogram;
  13 import com.google.common.base.VerifyException;
  14 import io.atomix.storage.journal.JournalReader;
  15 import io.atomix.storage.journal.JournalSerdes;
  16 import io.atomix.storage.journal.JournalWriter;
  17 import io.atomix.storage.journal.SegmentedByteBufJournal;
  18 import io.atomix.storage.journal.SegmentedJournal;
  19 import io.atomix.storage.journal.StorageLevel;
  20 import java.io.File;
  21 import java.io.Serializable;
  22 import java.util.ArrayList;
  23 import java.util.List;
  24 import org.opendaylight.controller.akka.segjournal.DataJournalEntry.FromPersistence;
  25 import org.opendaylight.controller.akka.segjournal.DataJournalEntry.ToPersistence;
  26 import org.opendaylight.controller.akka.segjournal.SegmentedJournalActor.ReplayMessages;
  27 import org.opendaylight.controller.akka.segjournal.SegmentedJournalActor.WriteMessages;
  28 import org.opendaylight.controller.akka.segjournal.SegmentedJournalActor.WrittenMessages;
  29 import org.slf4j.Logger;
  30 import org.slf4j.LoggerFactory;
  31 import scala.jdk.javaapi.CollectionConverters;
  32
  33 /**
  34  * Version 0 data journal, where every journal entry maps to exactly one segmented file entry.
  35  */
  36 final class DataJournalV0 extends DataJournal {
  37     private static final Logger LOG = LoggerFactory.getLogger(DataJournalV0.class);
  38
  39     private final SegmentedJournal<DataJournalEntry> entries;
  40
  41     DataJournalV0(final String persistenceId, final Histogram messageSize, final ActorSystem system,
  42             final StorageLevel storage, final File directory, final int maxEntrySize, final int maxSegmentSize) {
  43         super(persistenceId, messageSize);
  44         entries = new SegmentedJournal<>(SegmentedByteBufJournal.builder()
  45             .withDirectory(directory)
  46             .withName("data")
  47             .withStorageLevel(storage)
  48             .withMaxEntrySize(maxEntrySize)
  49             .withMaxSegmentSize(maxSegmentSize)
  50             .build(),
  51             JournalSerdes.builder()
  52                 .register(new DataJournalEntrySerdes(system), FromPersistence.class, ToPersistence.class)
  53                 .build().toMapper());
  54     }
  55
  56     @Override
  57     long lastWrittenSequenceNr() {
  58         return entries.lastIndex();
  59     }
  60
  61     @Override
  62     void deleteTo(final long sequenceNr) {
  63         entries.writer().commit(sequenceNr);
  64     }
  65
  66     @Override
  67     void compactTo(final long sequenceNr) {
  68         entries.compact(sequenceNr + 1);
  69     }
  70
  71     @Override
  72     void close() {
  73         flush();
  74         entries.close();
  75     }
  76
  77     @Override
  78     void flush() {
  79         entries.writer().flush();
  80     }
  81
  82     @Override
  83     @SuppressWarnings("checkstyle:illegalCatch")
  84     void handleReplayMessages(final ReplayMessages message, final long fromSequenceNr) {
  85         try (var reader = entries.openReader(fromSequenceNr)) {
  86             handleReplayMessages(reader, message);
  87         } catch (Exception e) {
  88             LOG.warn("{}: failed to replay messages for {}", persistenceId, message, e);
  89             message.promise.failure(e);
  90         } finally {
  91             message.promise.success(null);
  92         }
  93     }
  94
  95     private void handleReplayMessages(final JournalReader<DataJournalEntry> reader, final ReplayMessages message) {
  96         int count = 0;
  97         while (count < message.max && reader.getNextIndex() <= message.toSequenceNr) {
  98             final var repr = reader.tryNext((index, entry, size) -> {
  99                 LOG.trace("{}: replay index={} entry={}", persistenceId, index, entry);
 100                 updateLargestSize(size);
 101                 if (entry instanceof FromPersistence fromPersistence) {
 102                     return fromPersistence.toRepr(persistenceId, index);
 103                 }
 104                 throw new VerifyException("Unexpected entry " + entry);
 105             });
 106
 107             if (repr == null) {
 108                 break;
 109             }
 110
 111             LOG.debug("{}: replaying {}", persistenceId, repr);
 112             message.replayCallback.accept(repr);
 113             count++;
 114         }
 115         LOG.debug("{}: successfully replayed {} entries", persistenceId, count);
 116     }
 117
 118     @Override
 119     @SuppressWarnings("checkstyle:illegalCatch")
 120     WrittenMessages handleWriteMessages(final WriteMessages message) {
 121         final int count = message.size();
 122         final var responses = new ArrayList<>();
 123         final var writer = entries.writer();
 124         long writtenBytes = 0;
 125
 126         for (int i = 0; i < count; ++i) {
 127             final long prevNextIndex = writer.getNextIndex();
 128             final var request = message.getRequest(i);
 129
 130             final var reprs = CollectionConverters.asJava(request.payload());
 131             LOG.trace("{}: append {}/{}: {} items at mark {}", persistenceId, i, count, reprs.size(), prevNextIndex);
 132             try {
 133                 writtenBytes += writePayload(writer, reprs);
 134             } catch (Exception e) {
 135                 LOG.warn("{}: failed to write out request {}/{} reverting to {}", persistenceId, i, count,
 136                     prevNextIndex, e);
 137                 responses.add(e);
 138                 writer.reset(prevNextIndex);
 139                 continue;
 140             }
 141             responses.add(null);
 142         }
 143
 144         return new WrittenMessages(message, responses, writtenBytes);
 145     }
 146
 147     private long writePayload(final JournalWriter<DataJournalEntry> writer, final List<PersistentRepr> reprs) {
 148         long bytes = 0;
 149         for (var repr : reprs) {
 150             final Object payload = repr.payload();
 151             if (!(payload instanceof Serializable)) {
 152                 throw new UnsupportedOperationException("Non-serializable payload encountered "
 153                         + payload.getClass());
 154             }
 155
 156             LOG.trace("{}: starting append of {}", persistenceId, payload);
 157             final var entry = writer.append(new ToPersistence(repr));
 158             final int size = entry.size();
 159             LOG.trace("{}: finished append of {} with {} bytes at {}", persistenceId, payload, size, entry.index());
 160             recordMessageSize(size);
 161             bytes += size;
 162         }
 163         return bytes;
 164     }
 165 }