41117c8e32754f47a7b7a258cd803d31d83eedab
[controller.git] / opendaylight / md-sal / sal-akka-segmented-journal / src / main / java / org / opendaylight / controller / akka / segjournal / SegmentedJournalActor.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 com.google.common.base.Verify.verifyNotNull;
11 import static java.util.Objects.requireNonNull;
12
13 import akka.actor.AbstractActor;
14 import akka.actor.Props;
15 import akka.persistence.AtomicWrite;
16 import akka.persistence.PersistentRepr;
17 import com.codahale.metrics.Histogram;
18 import com.codahale.metrics.Meter;
19 import com.codahale.metrics.MetricRegistry;
20 import com.codahale.metrics.Timer;
21 import com.google.common.base.MoreObjects;
22 import io.atomix.storage.StorageLevel;
23 import io.atomix.storage.journal.Indexed;
24 import io.atomix.storage.journal.SegmentedJournal;
25 import io.atomix.storage.journal.SegmentedJournalWriter;
26 import io.atomix.utils.serializer.Namespace;
27 import java.io.File;
28 import java.util.ArrayList;
29 import java.util.List;
30 import java.util.Optional;
31 import java.util.concurrent.TimeUnit;
32 import java.util.function.Consumer;
33 import org.opendaylight.controller.cluster.common.actor.MeteringBehavior;
34 import org.opendaylight.controller.cluster.reporting.MetricsReporter;
35 import org.slf4j.Logger;
36 import org.slf4j.LoggerFactory;
37 import scala.concurrent.Future;
38 import scala.concurrent.Promise;
39
40 /**
41  * This actor handles a single PersistentActor's journal. The journal is split into two {@link SegmentedJournal}s:
42  * <ul>
43  *     <li>A memory-mapped data journal, containing actual data entries</li>
44  *     <li>A simple file journal, containing sequence numbers of last deleted entry</li>
45  * </ul>
46  *
47  * <p>
48  * This is a conscious design decision to minimize the amount of data that is being stored in the data journal while
49  * speeding up normal operations. Since the SegmentedJournal is an append-only linear log and Akka requires the ability
50  * to delete persistence entries, we need ability to mark a subset of a SegmentedJournal as deleted. While we could
51  * treat such delete requests as normal events, this leads to a mismatch between SegmentedJournal indices (as exposed by
52  * {@link Indexed}) and Akka sequence numbers -- requiring us to potentially perform costly deserialization to find the
53  * index corresponding to a particular sequence number, or maintain moderately-complex logic and data structures to
54  * perform that mapping in sub-linear time complexity.
55  *
56  * <p>
57  * Split-file approach allows us to treat sequence numbers and indices as equivalent, without maintaining any explicit
58  * mapping information. The only additional information we need to maintain is the last deleted sequence number.
59  *
60  * @author Robert Varga
61  */
62 final class SegmentedJournalActor extends AbstractActor {
63     abstract static class AsyncMessage<T> {
64         final Promise<T> promise = Promise.apply();
65     }
66
67     private static final class ReadHighestSequenceNr extends AsyncMessage<Long> {
68         private final long fromSequenceNr;
69
70         ReadHighestSequenceNr(final long fromSequenceNr) {
71             this.fromSequenceNr = fromSequenceNr;
72         }
73
74         @Override
75         public String toString() {
76             return MoreObjects.toStringHelper(this).add("fromSequenceNr", fromSequenceNr).toString();
77         }
78     }
79
80     static final class ReplayMessages extends AsyncMessage<Void> {
81         private final long fromSequenceNr;
82         final long toSequenceNr;
83         final long max;
84         final Consumer<PersistentRepr> replayCallback;
85
86         ReplayMessages(final long fromSequenceNr,
87                 final long toSequenceNr, final long max, final Consumer<PersistentRepr> replayCallback) {
88             this.fromSequenceNr = fromSequenceNr;
89             this.toSequenceNr = toSequenceNr;
90             this.max = max;
91             this.replayCallback = requireNonNull(replayCallback);
92         }
93
94         @Override
95         public String toString() {
96             return MoreObjects.toStringHelper(this).add("fromSequenceNr", fromSequenceNr)
97                     .add("toSequenceNr", toSequenceNr).add("max", max).toString();
98         }
99     }
100
101     static final class WriteMessages {
102         private final List<AtomicWrite> requests = new ArrayList<>();
103         private final List<Promise<Optional<Exception>>> results = new ArrayList<>();
104
105         Future<Optional<Exception>> add(final AtomicWrite write) {
106             final Promise<Optional<Exception>> promise = Promise.apply();
107             requests.add(write);
108             results.add(promise);
109             return promise.future();
110         }
111
112         int size() {
113             return requests.size();
114         }
115
116         AtomicWrite getRequest(final int index) {
117             return requests.get(index);
118         }
119
120         void setFailure(final int index, final Exception cause) {
121             results.get(index).success(Optional.of(cause));
122
123         }
124
125         void setSuccess(final int index) {
126             results.get(index).success(Optional.empty());
127         }
128
129         @Override
130         public String toString() {
131             return MoreObjects.toStringHelper(this).add("requests", requests).toString();
132         }
133     }
134
135     private static final class DeleteMessagesTo extends AsyncMessage<Void> {
136         final long toSequenceNr;
137
138         DeleteMessagesTo(final long toSequenceNr) {
139             this.toSequenceNr = toSequenceNr;
140         }
141
142         @Override
143         public String toString() {
144             return MoreObjects.toStringHelper(this).add("toSequenceNr", toSequenceNr).toString();
145         }
146     }
147
148     private static final Logger LOG = LoggerFactory.getLogger(SegmentedJournalActor.class);
149     private static final Namespace DELETE_NAMESPACE = Namespace.builder().register(Long.class).build();
150     private static final int DELETE_SEGMENT_SIZE = 64 * 1024;
151
152     private final String persistenceId;
153     private final StorageLevel storage;
154     private final int maxSegmentSize;
155     private final int maxEntrySize;
156     private final File directory;
157
158     // Tracks the time it took us to write a batch of messages
159     private Timer batchWriteTime;
160     // Tracks the number of individual messages written
161     private Meter messageWriteCount;
162     // Tracks the size distribution of messages
163     private Histogram messageSize;
164
165     private DataJournal dataJournal;
166     private SegmentedJournal<Long> deleteJournal;
167     private long lastDelete;
168
169     SegmentedJournalActor(final String persistenceId, final File directory, final StorageLevel storage,
170             final int maxEntrySize, final int maxSegmentSize) {
171         this.persistenceId = requireNonNull(persistenceId);
172         this.directory = requireNonNull(directory);
173         this.storage = requireNonNull(storage);
174         this.maxEntrySize = maxEntrySize;
175         this.maxSegmentSize = maxSegmentSize;
176     }
177
178     static Props props(final String persistenceId, final File directory, final StorageLevel storage,
179             final int maxEntrySize, final int maxSegmentSize) {
180         return Props.create(SegmentedJournalActor.class, requireNonNull(persistenceId), directory, storage,
181             maxEntrySize, maxSegmentSize);
182     }
183
184     @Override
185     public Receive createReceive() {
186         return receiveBuilder()
187                 .match(DeleteMessagesTo.class, this::handleDeleteMessagesTo)
188                 .match(ReadHighestSequenceNr.class, this::handleReadHighestSequenceNr)
189                 .match(ReplayMessages.class, this::handleReplayMessages)
190                 .match(WriteMessages.class, this::handleWriteMessages)
191                 .matchAny(this::handleUnknown)
192                 .build();
193     }
194
195     @Override
196     public void preStart() throws Exception {
197         LOG.debug("{}: actor starting", persistenceId);
198         super.preStart();
199
200         final MetricRegistry registry = MetricsReporter.getInstance(MeteringBehavior.DOMAIN).getMetricsRegistry();
201         final String actorName = self().path().parent().toStringWithoutAddress() + '/' + directory.getName();
202
203         batchWriteTime = registry.timer(MetricRegistry.name(actorName, "batchWriteTime"));
204         messageWriteCount = registry.meter(MetricRegistry.name(actorName, "messageWriteCount"));
205         messageSize = registry.histogram(MetricRegistry.name(actorName, "messageSize"));
206     }
207
208     @Override
209     public void postStop() throws Exception {
210         LOG.debug("{}: actor stopping", persistenceId);
211         if (dataJournal != null) {
212             dataJournal.close();
213             LOG.debug("{}: data journal closed", persistenceId);
214             dataJournal = null;
215         }
216         if (deleteJournal != null) {
217             deleteJournal.close();
218             LOG.debug("{}: delete journal closed", persistenceId);
219             deleteJournal = null;
220         }
221         LOG.debug("{}: actor stopped", persistenceId);
222         super.postStop();
223     }
224
225     static AsyncMessage<Void> deleteMessagesTo(final long toSequenceNr) {
226         return new DeleteMessagesTo(toSequenceNr);
227     }
228
229     static AsyncMessage<Long> readHighestSequenceNr(final long fromSequenceNr) {
230         return new ReadHighestSequenceNr(fromSequenceNr);
231     }
232
233     static AsyncMessage<Void> replayMessages(final long fromSequenceNr, final long toSequenceNr, final long max,
234             final Consumer<PersistentRepr> replayCallback) {
235         return new ReplayMessages(fromSequenceNr, toSequenceNr, max, replayCallback);
236     }
237
238     private void handleDeleteMessagesTo(final DeleteMessagesTo message) {
239         ensureOpen();
240
241         LOG.debug("{}: delete messages {}", persistenceId, message);
242         final long to = Long.min(dataJournal.lastWrittenIndex(), message.toSequenceNr);
243         LOG.debug("{}: adjusted delete to {}", persistenceId, to);
244
245         if (lastDelete < to) {
246             LOG.debug("{}: deleting entries up to {}", persistenceId, to);
247
248             lastDelete = to;
249             final SegmentedJournalWriter<Long> deleteWriter = deleteJournal.writer();
250             final Indexed<Long> entry = deleteWriter.append(lastDelete);
251             deleteWriter.commit(entry.index());
252             dataJournal.commitTo(lastDelete);
253
254             LOG.debug("{}: compaction started", persistenceId);
255             dataJournal.compactTo(lastDelete + 1);
256             deleteJournal.compact(entry.index());
257             LOG.debug("{}: compaction finished", persistenceId);
258         } else {
259             LOG.debug("{}: entries up to {} already deleted", persistenceId, lastDelete);
260         }
261
262         message.promise.success(null);
263     }
264
265     private void handleReadHighestSequenceNr(final ReadHighestSequenceNr message) {
266         LOG.debug("{}: looking for highest sequence on {}", persistenceId, message);
267         final Long sequence;
268         if (directory.isDirectory()) {
269             ensureOpen();
270             sequence = dataJournal.lastWrittenIndex();
271         } else {
272             sequence = 0L;
273         }
274
275         LOG.debug("{}: highest sequence is {}", message, sequence);
276         message.promise.success(sequence);
277     }
278
279     @SuppressWarnings("checkstyle:illegalCatch")
280     private void handleReplayMessages(final ReplayMessages message) {
281         LOG.debug("{}: replaying messages {}", persistenceId, message);
282         ensureOpen();
283
284         final long from = Long.max(lastDelete + 1, message.fromSequenceNr);
285         LOG.debug("{}: adjusted fromSequenceNr to {}", persistenceId, from);
286
287         dataJournal.handleReplayMessages(message, from);
288     }
289
290     private void handleWriteMessages(final WriteMessages message) {
291         ensureOpen();
292
293         final long startTicks = System.nanoTime();
294         final long start = dataJournal.lastWrittenIndex();
295
296         dataJournal.handleWriteMessages(message);
297
298         batchWriteTime.update(System.nanoTime() - startTicks, TimeUnit.NANOSECONDS);
299         messageWriteCount.mark(dataJournal.lastWrittenIndex() - start);
300     }
301
302     private void handleUnknown(final Object message) {
303         LOG.error("{}: Received unknown message {}", persistenceId, message);
304     }
305
306     private void ensureOpen() {
307         if (dataJournal != null) {
308             verifyNotNull(deleteJournal);
309             return;
310         }
311
312         deleteJournal = SegmentedJournal.<Long>builder().withDirectory(directory).withName("delete")
313                 .withNamespace(DELETE_NAMESPACE).withMaxSegmentSize(DELETE_SEGMENT_SIZE).build();
314         final Indexed<Long> lastEntry = deleteJournal.writer().getLastEntry();
315         lastDelete = lastEntry == null ? 0 : lastEntry.entry();
316
317         dataJournal = new DataJournalV0(persistenceId, messageSize, context().system(), storage, directory,
318             maxEntrySize, maxSegmentSize);
319         dataJournal.commitTo(lastDelete);
320         LOG.debug("{}: journal open with last index {}, deleted to {}", persistenceId, dataJournal.lastWrittenIndex(),
321             lastDelete);
322     }
323 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.