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 static com.google.common.base.Verify.verify;
  11
  12 import akka.actor.ActorSystem;
  13 import akka.persistence.AtomicWrite;
  14 import akka.persistence.PersistentRepr;
  15 import com.codahale.metrics.Histogram;
  16 import io.atomix.storage.StorageLevel;
  17 import io.atomix.storage.journal.Indexed;
  18 import io.atomix.storage.journal.SegmentedJournal;
  19 import io.atomix.storage.journal.SegmentedJournalReader;
  20 import io.atomix.storage.journal.SegmentedJournalWriter;
  21 import io.atomix.utils.serializer.Namespace;
  22 import java.io.File;
  23 import java.io.Serializable;
  24 import java.util.List;
  25 import org.opendaylight.controller.akka.segjournal.DataJournalEntry.FromPersistence;
  26 import org.opendaylight.controller.akka.segjournal.DataJournalEntry.ToPersistence;
  27 import org.opendaylight.controller.akka.segjournal.SegmentedJournalActor.ReplayMessages;
  28 import org.opendaylight.controller.akka.segjournal.SegmentedJournalActor.WriteMessages;
  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  * @author Robert Varga
  37  */
  38 final class DataJournalV0 extends DataJournal {
  39     private static final Logger LOG = LoggerFactory.getLogger(DataJournalV0.class);
  40
  41     private final SegmentedJournal<DataJournalEntry> entries;
  42
  43     DataJournalV0(final String persistenceId, final Histogram messageSize, final ActorSystem system,
  44             final StorageLevel storage, final File directory, final int maxEntrySize, final int maxSegmentSize) {
  45         super(persistenceId, messageSize);
  46         entries = SegmentedJournal.<DataJournalEntry>builder()
  47                 .withStorageLevel(storage).withDirectory(directory).withName("data")
  48                 .withNamespace(Namespace.builder()
  49                     .register(new DataJournalEntrySerializer(system), FromPersistence.class, ToPersistence.class)
  50                     .build())
  51                 .withMaxEntrySize(maxEntrySize).withMaxSegmentSize(maxSegmentSize)
  52                 .build();
  53     }
  54
  55     @Override
  56     long lastWrittenSequenceNr() {
  57         return entries.writer().getLastIndex();
  58     }
  59
  60     @Override
  61     void deleteTo(final long sequenceNr) {
  62         entries.writer().commit(sequenceNr);
  63     }
  64
  65     @Override
  66     void compactTo(final long sequenceNr) {
  67         entries.compact(sequenceNr + 1);
  68     }
  69
  70     @Override
  71     void close() {
  72         entries.close();
  73     }
  74
  75     @Override
  76     @SuppressWarnings("checkstyle:illegalCatch")
  77     void handleReplayMessages(final ReplayMessages message, final long fromSequenceNr) {
  78         try (SegmentedJournalReader<DataJournalEntry> reader = entries.openReader(fromSequenceNr)) {
  79             int count = 0;
  80             while (reader.hasNext() && count < message.max) {
  81                 final Indexed<DataJournalEntry> next = reader.next();
  82                 if (next.index() > message.toSequenceNr) {
  83                     break;
  84                 }
  85
  86                 LOG.trace("{}: replay {}", persistenceId, next);
  87                 updateLargestSize(next.size());
  88                 final DataJournalEntry entry = next.entry();
  89                 verify(entry instanceof FromPersistence, "Unexpected entry %s", entry);
  90
  91                 final PersistentRepr repr = ((FromPersistence) entry).toRepr(persistenceId, next.index());
  92                 LOG.debug("{}: replaying {}", persistenceId, repr);
  93                 message.replayCallback.accept(repr);
  94                 count++;
  95             }
  96             LOG.debug("{}: successfully replayed {} entries", persistenceId, count);
  97         } catch (Exception e) {
  98             LOG.warn("{}: failed to replay messages for {}", persistenceId, message, e);
  99             message.promise.failure(e);
 100         } finally {
 101             message.promise.success(null);
 102         }
 103     }
 104
 105     @Override
 106     @SuppressWarnings("checkstyle:illegalCatch")
 107     void handleWriteMessages(final WriteMessages message) {
 108         final int count = message.size();
 109         final SegmentedJournalWriter<DataJournalEntry> writer = entries.writer();
 110
 111         for (int i = 0; i < count; ++i) {
 112             final long mark = writer.getLastIndex();
 113             final AtomicWrite request = message.getRequest(i);
 114
 115             final List<PersistentRepr> reprs = CollectionConverters.asJava(request.payload());
 116             LOG.trace("{}: append {}/{}: {} items at mark {}", persistenceId, i, count, mark, reprs.size());
 117             try {
 118                 for (PersistentRepr repr : reprs) {
 119                     final Object payload = repr.payload();
 120                     if (!(payload instanceof Serializable)) {
 121                         throw new UnsupportedOperationException("Non-serializable payload encountered "
 122                                 + payload.getClass());
 123                     }
 124
 125                     LOG.trace("{}: starting append of {}", persistenceId, payload);
 126                     final Indexed<ToPersistence> entry = writer.append(new ToPersistence(repr));
 127                     final int size = entry.size();
 128                     LOG.trace("{}: finished append of {} with {} bytes at {}", persistenceId, payload, size,
 129                         entry.index());
 130                     recordMessageSize(size);
 131                 }
 132             } catch (Exception e) {
 133                 LOG.warn("{}: failed to write out request", persistenceId, e);
 134                 message.setFailure(i, e);
 135                 writer.truncate(mark);
 136                 continue;
 137             }
 138
 139             message.setSuccess(i);
 140         }
 141         writer.flush();
 142     }
 143 }