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