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