2 * Copyright (c) 2019 Pantheon Technologies, s.r.o. and others. All rights reserved.
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
8 package org.opendaylight.controller.akka.segjournal;
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;
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;
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;
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;
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.
43 * @author Robert Varga
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";
53 private static final Logger LOG = LoggerFactory.getLogger(SegmentedFileJournal.class);
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;
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);
67 checkArgument(rootDir.isDirectory(), "%s is not a directory", rootDir);
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);
72 if (config.hasPath(STORAGE_MEMORY_MAPPED)) {
73 storage = config.getBoolean(STORAGE_MEMORY_MAPPED) ? StorageLevel.MAPPED : StorageLevel.DISK;
75 storage = StorageLevel.DISK;
78 LOG.info("Initialized with root directory {} with storage {}", rootDir, storage);
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<>();
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));
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());
97 return Futures.sequence(result, context().dispatcher());
101 public Future<Void> doAsyncDeleteMessagesTo(final String persistenceId, final long toSequenceNr) {
102 return delegateMessage(persistenceId, SegmentedJournalActor.deleteMessagesTo(toSequenceNr));
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));
113 public Future<Long> doAsyncReadHighestSequenceNr(final String persistenceId, final long fromSequenceNr) {
114 return delegateMessage(handlers.computeIfAbsent(persistenceId, this::createHandler),
115 SegmentedJournalActor.readHighestSequenceNr(fromSequenceNr));
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);
124 final ActorRef handler = context().actorOf(SegmentedJournalActor.props(persistenceId, directory, storage,
125 maxEntrySize, maxSegmentSize));
126 LOG.debug("Directory {} handled by {}", directory, handler);
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));
136 return delegateMessage(handler, message);
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();
145 private static int getBytes(final Config config, final String path, final int defaultValue) {
146 if (!config.hasPath(path)) {
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);