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

©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.