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

   1 /*
   2  * Copyright (c) 2019 Pantheon Technologies, 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 akka.actor.ActorRef.noSender;
  11 import static com.google.common.base.Preconditions.checkArgument;
  12 import static com.google.common.base.Preconditions.checkState;
  13
  14 import akka.actor.ActorRef;
  15 import akka.dispatch.Futures;
  16 import akka.persistence.AtomicWrite;
  17 import akka.persistence.PersistentRepr;
  18 import akka.persistence.journal.japi.AsyncWriteJournal;
  19 import com.typesafe.config.Config;
  20 import com.typesafe.config.ConfigMemorySize;
  21 import io.atomix.storage.StorageLevel;
  22 import io.atomix.storage.journal.SegmentedJournal;
  23 import java.io.File;
  24 import java.nio.charset.StandardCharsets;
  25 import java.util.ArrayList;
  26 import java.util.Base64;
  27 import java.util.HashMap;
  28 import java.util.List;
  29 import java.util.Map;
  30 import java.util.Optional;
  31 import java.util.function.Consumer;
  32 import org.opendaylight.controller.akka.segjournal.SegmentedJournalActor.AsyncMessage;
  33 import org.opendaylight.controller.akka.segjournal.SegmentedJournalActor.WriteMessages;
  34 import org.slf4j.Logger;
  35 import org.slf4j.LoggerFactory;
  36 import scala.concurrent.Future;
  37
  38 /**
  39  * An Akka persistence journal implementation on top of {@link SegmentedJournal}. This actor represents aggregation
  40  * of multiple journals and performs a receptionist job between Akka and invidual per-persistenceId actors. See
  41  * {@link SegmentedJournalActor} for details on how the persistence works.
  42  *
  43  * @author Robert Varga
  44  */
  45 public class SegmentedFileJournal extends AsyncWriteJournal {
  46     public static final String STORAGE_ROOT_DIRECTORY = "root-directory";
  47     public static final String STORAGE_MAX_ENTRY_SIZE = "max-entry-size";
  48     public static final int STORAGE_MAX_ENTRY_SIZE_DEFAULT = 16 * 1024 * 1024;
  49     public static final String STORAGE_MAX_SEGMENT_SIZE = "max-segment-size";
  50     public static final int STORAGE_MAX_SEGMENT_SIZE_DEFAULT = STORAGE_MAX_ENTRY_SIZE_DEFAULT * 8;
  51     public static final String STORAGE_MEMORY_MAPPED = "memory-mapped";
  52
  53     private static final Logger LOG = LoggerFactory.getLogger(SegmentedFileJournal.class);
  54
  55     private final Map<String, ActorRef> handlers = new HashMap<>();
  56     private final File rootDir;
  57     private final StorageLevel storage;
  58     private final int maxEntrySize;
  59     private final int maxSegmentSize;
  60
  61     public SegmentedFileJournal(final Config config) {
  62         rootDir = new File(config.getString(STORAGE_ROOT_DIRECTORY));
  63         if (!rootDir.exists()) {
  64             LOG.debug("Creating directory {}", rootDir);
  65             checkState(rootDir.mkdirs(), "Failed to create root directory %s", rootDir);
  66         }
  67         checkArgument(rootDir.isDirectory(), "%s is not a directory", rootDir);
  68
  69         maxEntrySize = getBytes(config, STORAGE_MAX_ENTRY_SIZE, STORAGE_MAX_ENTRY_SIZE_DEFAULT);
  70         maxSegmentSize = getBytes(config, STORAGE_MAX_SEGMENT_SIZE, STORAGE_MAX_SEGMENT_SIZE_DEFAULT);
  71
  72         if (config.hasPath(STORAGE_MEMORY_MAPPED)) {
  73             storage = config.getBoolean(STORAGE_MEMORY_MAPPED) ? StorageLevel.MAPPED : StorageLevel.DISK;
  74         } else {
  75             storage = StorageLevel.DISK;
  76         }
  77
  78         LOG.info("Initialized with root directory {} with storage {}", rootDir, storage);
  79     }
  80
  81     @Override
  82     public Future<Iterable<Optional<Exception>>> doAsyncWriteMessages(final Iterable<AtomicWrite> messages) {
  83         final Map<ActorRef, WriteMessages> map = new HashMap<>();
  84         final List<Future<Optional<Exception>>> result = new ArrayList<>();
  85
  86         for (AtomicWrite message : messages) {
  87             final String persistenceId = message.persistenceId();
  88             final ActorRef handler = handlers.computeIfAbsent(persistenceId, this::createHandler);
  89             result.add(map.computeIfAbsent(handler, key -> new WriteMessages()).add(message));
  90         }
  91
  92         // Send requests to actors and zip the futures back
  93         map.forEach((handler, message) -> {
  94             LOG.trace("Sending {} to {}", message, handler);
  95             handler.tell(message, noSender());
  96         });
  97         return Futures.sequence(result, context().dispatcher());
  98     }
  99
 100     @Override
 101     public Future<Void> doAsyncDeleteMessagesTo(final String persistenceId, final long toSequenceNr) {
 102         return delegateMessage(persistenceId, SegmentedJournalActor.deleteMessagesTo(toSequenceNr));
 103     }
 104
 105     @Override
 106     public Future<Void> doAsyncReplayMessages(final String persistenceId, final long fromSequenceNr,
 107             final long toSequenceNr, final long max, final Consumer<PersistentRepr> replayCallback) {
 108         return delegateMessage(persistenceId,
 109             SegmentedJournalActor.replayMessages(fromSequenceNr, toSequenceNr, max, replayCallback));
 110     }
 111
 112     @Override
 113     public Future<Long> doAsyncReadHighestSequenceNr(final String persistenceId, final long fromSequenceNr) {
 114         return delegateMessage(handlers.computeIfAbsent(persistenceId, this::createHandler),
 115             SegmentedJournalActor.readHighestSequenceNr(fromSequenceNr));
 116     }
 117
 118     private ActorRef createHandler(final String persistenceId) {
 119         final String directoryName = Base64.getUrlEncoder().encodeToString(persistenceId.getBytes(
 120             StandardCharsets.UTF_8));
 121         final File directory = new File(rootDir, directoryName);
 122         LOG.debug("Creating handler for {} in directory {}", persistenceId, directory);
 123
 124         final ActorRef handler = context().actorOf(SegmentedJournalActor.props(persistenceId, directory, storage,
 125             maxEntrySize, maxSegmentSize));
 126         LOG.debug("Directory {} handled by {}", directory, handler);
 127         return handler;
 128     }
 129
 130     private <T> Future<T> delegateMessage(final String persistenceId, final AsyncMessage<T> message) {
 131         final ActorRef handler = handlers.get(persistenceId);
 132         if (handler == null) {
 133             return Futures.failed(new IllegalStateException("Cannot find handler for " + persistenceId));
 134         }
 135
 136         return delegateMessage(handler, message);
 137     }
 138
 139     private static <T> Future<T> delegateMessage(final ActorRef handler, final AsyncMessage<T> message) {
 140         LOG.trace("Delegating {} to {}", message, handler);
 141         handler.tell(message, noSender());
 142         return message.promise.future();
 143     }
 144
 145     private static int getBytes(final Config config, final String path, final int defaultValue) {
 146         if (!config.hasPath(path)) {
 147             return defaultValue;
 148         }
 149         final ConfigMemorySize value = config.getMemorySize(path);
 150         final long result = value.toBytes();
 151         checkArgument(result <= Integer.MAX_VALUE, "Size %s exceeds maximum allowed %s", Integer.MAX_VALUE);
 152         return (int) result;
 153     }
 154 }