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.ActorRef;
  16 import akka.actor.Props;
  17 import akka.japi.pf.ReceiveBuilder;
  18 import akka.persistence.AtomicWrite;
  19 import akka.persistence.PersistentRepr;
  20 import com.codahale.metrics.Histogram;
  21 import com.codahale.metrics.Meter;
  22 import com.codahale.metrics.MetricRegistry;
  23 import com.codahale.metrics.Timer;
  24 import com.google.common.base.MoreObjects;
  25 import com.google.common.base.Stopwatch;
  26 import io.atomix.storage.journal.Indexed;
  27 import io.atomix.storage.journal.JournalSerdes;
  28 import io.atomix.storage.journal.SegmentedJournal;
  29 import io.atomix.storage.journal.StorageLevel;
  30 import java.io.File;
  31 import java.util.ArrayDeque;
  32 import java.util.ArrayList;
  33 import java.util.List;
  34 import java.util.Optional;
  35 import java.util.concurrent.TimeUnit;
  36 import java.util.function.Consumer;
  37 import org.opendaylight.controller.cluster.common.actor.MeteringBehavior;
  38 import org.opendaylight.controller.cluster.reporting.MetricsReporter;
  39 import org.slf4j.Logger;
  40 import org.slf4j.LoggerFactory;
  41 import scala.concurrent.Future;
  42 import scala.concurrent.Promise;
  43
  44 /**
  45  * This actor handles a single PersistentActor's journal. The journal is split into two {@link SegmentedJournal}s:
  46  * <ul>
  47  *     <li>A memory-mapped data journal, containing actual data entries</li>
  48  *     <li>A simple file journal, containing sequence numbers of last deleted entry</li>
  49  * </ul>
  50  *
  51  * <p>
  52  * This is a conscious design decision to minimize the amount of data that is being stored in the data journal while
  53  * speeding up normal operations. Since the SegmentedJournal is an append-only linear log and Akka requires the ability
  54  * to delete persistence entries, we need ability to mark a subset of a SegmentedJournal as deleted. While we could
  55  * treat such delete requests as normal events, this leads to a mismatch between SegmentedJournal indices (as exposed by
  56  * {@link Indexed}) and Akka sequence numbers -- requiring us to potentially perform costly deserialization to find the
  57  * index corresponding to a particular sequence number, or maintain moderately-complex logic and data structures to
  58  * perform that mapping in sub-linear time complexity.
  59  *
  60  * <p>
  61  * Split-file approach allows us to treat sequence numbers and indices as equivalent, without maintaining any explicit
  62  * mapping information. The only additional information we need to maintain is the last deleted sequence number.
  63  */
  64 abstract sealed class SegmentedJournalActor extends AbstractActor {
  65     abstract static sealed class AsyncMessage<T> {
  66         final Promise<T> promise = Promise.apply();
  67     }
  68
  69     private static final class ReadHighestSequenceNr extends AsyncMessage<Long> {
  70         private final long fromSequenceNr;
  71
  72         ReadHighestSequenceNr(final long fromSequenceNr) {
  73             this.fromSequenceNr = fromSequenceNr;
  74         }
  75
  76         @Override
  77         public String toString() {
  78             return MoreObjects.toStringHelper(this).add("fromSequenceNr", fromSequenceNr).toString();
  79         }
  80     }
  81
  82     static final class ReplayMessages extends AsyncMessage<Void> {
  83         private final long fromSequenceNr;
  84         final long toSequenceNr;
  85         final long max;
  86         final Consumer<PersistentRepr> replayCallback;
  87
  88         ReplayMessages(final long fromSequenceNr,
  89                 final long toSequenceNr, final long max, final Consumer<PersistentRepr> replayCallback) {
  90             this.fromSequenceNr = fromSequenceNr;
  91             this.toSequenceNr = toSequenceNr;
  92             this.max = max;
  93             this.replayCallback = requireNonNull(replayCallback);
  94         }
  95
  96         @Override
  97         public String toString() {
  98             return MoreObjects.toStringHelper(this).add("fromSequenceNr", fromSequenceNr)
  99                     .add("toSequenceNr", toSequenceNr).add("max", max).toString();
 100         }
 101     }
 102
 103     static final class WriteMessages {
 104         private final List<AtomicWrite> requests = new ArrayList<>();
 105         private final List<Promise<Optional<Exception>>> results = new ArrayList<>();
 106
 107         Future<Optional<Exception>> add(final AtomicWrite write) {
 108             final var promise = Promise.<Optional<Exception>>apply();
 109             requests.add(write);
 110             results.add(promise);
 111             return promise.future();
 112         }
 113
 114         int size() {
 115             return requests.size();
 116         }
 117
 118         AtomicWrite getRequest(final int index) {
 119             return requests.get(index);
 120         }
 121
 122         void setFailure(final int index, final Exception cause) {
 123             results.get(index).success(Optional.of(cause));
 124
 125         }
 126
 127         void setSuccess(final int index) {
 128             results.get(index).success(Optional.empty());
 129         }
 130
 131         @Override
 132         public String toString() {
 133             return MoreObjects.toStringHelper(this).add("requests", requests).toString();
 134         }
 135     }
 136
 137     private static final class DeleteMessagesTo extends AsyncMessage<Void> {
 138         final long toSequenceNr;
 139
 140         DeleteMessagesTo(final long toSequenceNr) {
 141             this.toSequenceNr = toSequenceNr;
 142         }
 143
 144         @Override
 145         public String toString() {
 146             return MoreObjects.toStringHelper(this).add("toSequenceNr", toSequenceNr).toString();
 147         }
 148     }
 149
 150     // responses == null on success, Exception on failure
 151     record WrittenMessages(WriteMessages message, List<Object> responses, long writtenBytes) {
 152         WrittenMessages {
 153             verify(responses.size() == message.size(), "Mismatched %s and %s", message, responses);
 154             verify(writtenBytes >= 0, "Unexpected length %s", writtenBytes);
 155         }
 156
 157         private void complete() {
 158             for (int i = 0, size = responses.size(); i < size; ++i) {
 159                 if (responses.get(i) instanceof Exception ex) {
 160                     message.setFailure(i, ex);
 161                 } else {
 162                     message.setSuccess(i);
 163                 }
 164             }
 165         }
 166     }
 167
 168     /**
 169      * A {@link SegmentedJournalActor} which delays issuing a flush operation until a watermark is reached or when the
 170      * queue is empty.
 171      *
 172      * <p>
 173      * The problem we are addressing is that there is a queue sitting in from of the actor, which we have no direct
 174      * access to. Since a flush involves committing data to durable storage, that operation can easily end up dominating
 175      * workloads.
 176      *
 177      * <p>
 178      * We solve this by having an additional queue in which we track which messages were written and trigger a flush
 179      * only when the number of bytes we have written exceeds specified limit. The other part is that each time this
 180      * queue becomes non-empty, we send a dedicated message to self. This acts as a actor queue probe -- when we receive
 181      * it, we know we have processed all messages that were in the queue when we first delayed the write.
 182      *
 183      * <p>
 184      * The combination of these mechanisms ensure we use a minimal delay while also ensuring we take advantage of
 185      * batching opportunities.
 186      */
 187     private static final class Delayed extends SegmentedJournalActor {
 188         private static final class Flush extends AsyncMessage<Void> {
 189             final long batch;
 190
 191             Flush(final long batch) {
 192                 this.batch = batch;
 193             }
 194         }
 195
 196         private final ArrayDeque<WrittenMessages> unflushedWrites = new ArrayDeque<>();
 197         private final Stopwatch unflushedDuration = Stopwatch.createUnstarted();
 198         private final long maxUnflushedBytes;
 199
 200         private long batch = 0;
 201         private long unflushedBytes = 0;
 202
 203         Delayed(final String persistenceId, final File directory, final StorageLevel storage,
 204                 final int maxEntrySize, final int maxSegmentSize, final int maxUnflushedBytes) {
 205             super(persistenceId, directory, storage, maxEntrySize, maxSegmentSize);
 206             this.maxUnflushedBytes = maxUnflushedBytes;
 207         }
 208
 209         @Override
 210         ReceiveBuilder addMessages(final ReceiveBuilder builder) {
 211             return super.addMessages(builder).match(Flush.class, this::handleFlush);
 212         }
 213
 214         private void handleFlush(final Flush message) {
 215             if (message.batch == batch) {
 216                 flushWrites();
 217             } else {
 218                 LOG.debug("{}: batch {} not flushed by {}", persistenceId(), batch, message.batch);
 219             }
 220         }
 221
 222         @Override
 223         void onWrittenMessages(final WrittenMessages message) {
 224             boolean first = unflushedWrites.isEmpty();
 225             if (first) {
 226                 unflushedDuration.start();
 227             }
 228             unflushedWrites.addLast(message);
 229             unflushedBytes = unflushedBytes + message.writtenBytes;
 230             if (unflushedBytes >= maxUnflushedBytes) {
 231                 LOG.debug("{}: reached {} unflushed journal bytes", persistenceId(), unflushedBytes);
 232                 flushWrites();
 233             } else if (first) {
 234                 LOG.debug("{}: deferring journal flush", persistenceId());
 235                 self().tell(new Flush(++batch), ActorRef.noSender());
 236             }
 237         }
 238
 239         @Override
 240         void flushWrites() {
 241             final var unsyncedSize = unflushedWrites.size();
 242             if (unsyncedSize == 0) {
 243                 // Nothing to flush
 244                 return;
 245             }
 246
 247             LOG.debug("{}: flushing {} journal writes after {}", persistenceId(), unsyncedSize,
 248                 unflushedDuration.stop());
 249             flushJournal(unflushedBytes, unsyncedSize);
 250
 251             final var sw = Stopwatch.createStarted();
 252             unflushedWrites.forEach(WrittenMessages::complete);
 253             unflushedWrites.clear();
 254             unflushedBytes = 0;
 255             unflushedDuration.reset();
 256             LOG.debug("{}: completed {} flushed journal writes in {}", persistenceId(), unsyncedSize, sw);
 257         }
 258     }
 259
 260     private static final class Immediate extends SegmentedJournalActor {
 261         Immediate(final String persistenceId, final File directory, final StorageLevel storage,
 262                 final int maxEntrySize, final int maxSegmentSize) {
 263             super(persistenceId, directory, storage, maxEntrySize, maxSegmentSize);
 264         }
 265
 266         @Override
 267         void onWrittenMessages(final WrittenMessages message) {
 268             flushJournal(message.writtenBytes, 1);
 269             message.complete();
 270         }
 271
 272         @Override
 273         void flushWrites() {
 274             // No-op
 275         }
 276     }
 277
 278     private static final Logger LOG = LoggerFactory.getLogger(SegmentedJournalActor.class);
 279     private static final JournalSerdes DELETE_NAMESPACE = JournalSerdes.builder()
 280         .register(LongEntrySerdes.LONG_ENTRY_SERDES, Long.class)
 281         .build();
 282     private static final int DELETE_SEGMENT_SIZE = 64 * 1024;
 283
 284     private final String persistenceId;
 285     private final StorageLevel storage;
 286     private final int maxSegmentSize;
 287     private final int maxEntrySize;
 288     private final File directory;
 289
 290     // Tracks the time it took us to write a batch of messages
 291     private Timer batchWriteTime;
 292     // Tracks the number of individual messages written
 293     private Meter messageWriteCount;
 294     // Tracks the size distribution of messages
 295     private Histogram messageSize;
 296     // Tracks the number of messages completed for each flush
 297     private Histogram flushMessages;
 298     // Tracks the number of bytes completed for each flush
 299     private Histogram flushBytes;
 300     // Tracks the duration of flush operations
 301     private Timer flushTime;
 302
 303     private DataJournal dataJournal;
 304     private SegmentedJournal<Long> deleteJournal;
 305     private long lastDelete;
 306
 307     private SegmentedJournalActor(final String persistenceId, final File directory, final StorageLevel storage,
 308             final int maxEntrySize, final int maxSegmentSize) {
 309         this.persistenceId = requireNonNull(persistenceId);
 310         this.directory = requireNonNull(directory);
 311         this.storage = requireNonNull(storage);
 312         this.maxEntrySize = maxEntrySize;
 313         this.maxSegmentSize = maxSegmentSize;
 314     }
 315
 316     static Props props(final String persistenceId, final File directory, final StorageLevel storage,
 317             final int maxEntrySize, final int maxSegmentSize, final int maxUnflushedBytes) {
 318         final var pid = requireNonNull(persistenceId);
 319         return maxUnflushedBytes > 0
 320             ? Props.create(Delayed.class, pid, directory, storage, maxEntrySize, maxSegmentSize, maxUnflushedBytes)
 321             : Props.create(Immediate.class, pid, directory, storage, maxEntrySize, maxSegmentSize);
 322     }
 323
 324     final String persistenceId() {
 325         return persistenceId;
 326     }
 327
 328     final void flushJournal(final long bytes, final int messages) {
 329         final var sw = Stopwatch.createStarted();
 330         dataJournal.flush();
 331         LOG.debug("{}: journal flush completed in {}", persistenceId, sw.stop());
 332         flushBytes.update(bytes);
 333         flushMessages.update(messages);
 334         flushTime.update(sw.elapsed(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS);
 335     }
 336
 337     @Override
 338     public Receive createReceive() {
 339         return addMessages(receiveBuilder())
 340             .matchAny(this::handleUnknown)
 341             .build();
 342     }
 343
 344     ReceiveBuilder addMessages(final ReceiveBuilder builder) {
 345         return builder
 346             .match(DeleteMessagesTo.class, this::handleDeleteMessagesTo)
 347             .match(ReadHighestSequenceNr.class, this::handleReadHighestSequenceNr)
 348             .match(ReplayMessages.class, this::handleReplayMessages)
 349             .match(WriteMessages.class, this::handleWriteMessages);
 350     }
 351
 352     @Override
 353     public void preStart() throws Exception {
 354         LOG.debug("{}: actor starting", persistenceId);
 355         super.preStart();
 356
 357         final var registry = MetricsReporter.getInstance(MeteringBehavior.DOMAIN).getMetricsRegistry();
 358         final var actorName = self().path().parent().toStringWithoutAddress() + '/' + directory.getName();
 359
 360         batchWriteTime = registry.timer(MetricRegistry.name(actorName, "batchWriteTime"));
 361         messageWriteCount = registry.meter(MetricRegistry.name(actorName, "messageWriteCount"));
 362         messageSize = registry.histogram(MetricRegistry.name(actorName, "messageSize"));
 363         flushBytes = registry.histogram(MetricRegistry.name(actorName, "flushBytes"));
 364         flushMessages = registry.histogram(MetricRegistry.name(actorName, "flushMessages"));
 365         flushTime = registry.timer(MetricRegistry.name(actorName, "flushTime"));
 366     }
 367
 368     @Override
 369     public void postStop() throws Exception {
 370         LOG.debug("{}: actor stopping", persistenceId);
 371         if (dataJournal != null) {
 372             dataJournal.close();
 373             LOG.debug("{}: data journal closed", persistenceId);
 374             dataJournal = null;
 375         }
 376         if (deleteJournal != null) {
 377             deleteJournal.close();
 378             LOG.debug("{}: delete journal closed", persistenceId);
 379             deleteJournal = null;
 380         }
 381         LOG.debug("{}: actor stopped", persistenceId);
 382         super.postStop();
 383     }
 384
 385     static AsyncMessage<Void> deleteMessagesTo(final long toSequenceNr) {
 386         return new DeleteMessagesTo(toSequenceNr);
 387     }
 388
 389     static AsyncMessage<Long> readHighestSequenceNr(final long fromSequenceNr) {
 390         return new ReadHighestSequenceNr(fromSequenceNr);
 391     }
 392
 393     static AsyncMessage<Void> replayMessages(final long fromSequenceNr, final long toSequenceNr, final long max,
 394             final Consumer<PersistentRepr> replayCallback) {
 395         return new ReplayMessages(fromSequenceNr, toSequenceNr, max, replayCallback);
 396     }
 397
 398     private void handleDeleteMessagesTo(final DeleteMessagesTo message) {
 399         ensureOpen();
 400
 401         LOG.debug("{}: delete messages {}", persistenceId, message);
 402         flushWrites();
 403
 404         final long to = Long.min(dataJournal.lastWrittenSequenceNr(), message.toSequenceNr);
 405         LOG.debug("{}: adjusted delete to {}", persistenceId, to);
 406
 407         if (lastDelete < to) {
 408             LOG.debug("{}: deleting entries up to {}", persistenceId, to);
 409
 410             lastDelete = to;
 411             final var deleteWriter = deleteJournal.writer();
 412             final var entry = deleteWriter.append(lastDelete);
 413             deleteWriter.commit(entry.index());
 414             dataJournal.deleteTo(lastDelete);
 415
 416             LOG.debug("{}: compaction started", persistenceId);
 417             dataJournal.compactTo(lastDelete);
 418             deleteJournal.compact(entry.index());
 419             LOG.debug("{}: compaction finished", persistenceId);
 420         } else {
 421             LOG.debug("{}: entries up to {} already deleted", persistenceId, lastDelete);
 422         }
 423
 424         message.promise.success(null);
 425     }
 426
 427     private void handleReadHighestSequenceNr(final ReadHighestSequenceNr message) {
 428         LOG.debug("{}: looking for highest sequence on {}", persistenceId, message);
 429         final Long sequence;
 430         if (directory.isDirectory()) {
 431             ensureOpen();
 432             flushWrites();
 433             sequence = dataJournal.lastWrittenSequenceNr();
 434         } else {
 435             sequence = 0L;
 436         }
 437
 438         LOG.debug("{}: highest sequence is {}", message, sequence);
 439         message.promise.success(sequence);
 440     }
 441
 442     private void handleReplayMessages(final ReplayMessages message) {
 443         LOG.debug("{}: replaying messages {}", persistenceId, message);
 444         ensureOpen();
 445         flushWrites();
 446
 447         final long from = Long.max(lastDelete + 1, message.fromSequenceNr);
 448         LOG.debug("{}: adjusted fromSequenceNr to {}", persistenceId, from);
 449
 450         dataJournal.handleReplayMessages(message, from);
 451     }
 452
 453     private void handleWriteMessages(final WriteMessages message) {
 454         ensureOpen();
 455
 456         final var sw = Stopwatch.createStarted();
 457         final long start = dataJournal.lastWrittenSequenceNr();
 458         final var writtenMessages = dataJournal.handleWriteMessages(message);
 459         sw.stop();
 460
 461         batchWriteTime.update(sw.elapsed(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS);
 462         messageWriteCount.mark(dataJournal.lastWrittenSequenceNr() - start);
 463
 464         // log message after statistics are updated
 465         LOG.debug("{}: write of {} bytes completed in {}", persistenceId, writtenMessages.writtenBytes, sw);
 466         onWrittenMessages(writtenMessages);
 467     }
 468
 469     /**
 470      * Handle a check of written messages.
 471      *
 472      * @param message Messages which were written
 473      */
 474     abstract void onWrittenMessages(WrittenMessages message);
 475
 476     private void handleUnknown(final Object message) {
 477         LOG.error("{}: Received unknown message {}", persistenceId, message);
 478     }
 479
 480     private void ensureOpen() {
 481         if (dataJournal != null) {
 482             verifyNotNull(deleteJournal);
 483             return;
 484         }
 485
 486         final var sw = Stopwatch.createStarted();
 487         deleteJournal = SegmentedJournal.<Long>builder().withDirectory(directory).withName("delete")
 488                 .withNamespace(DELETE_NAMESPACE).withMaxSegmentSize(DELETE_SEGMENT_SIZE).build();
 489         final var lastEntry = deleteJournal.writer().getLastEntry();
 490         lastDelete = lastEntry == null ? 0 : lastEntry.entry();
 491
 492         dataJournal = new DataJournalV0(persistenceId, messageSize, context().system(), storage, directory,
 493             maxEntrySize, maxSegmentSize);
 494         dataJournal.deleteTo(lastDelete);
 495         LOG.debug("{}: journal open in {} with last index {}, deleted to {}", persistenceId, sw,
 496             dataJournal.lastWrittenSequenceNr(), lastDelete);
 497     }
 498
 499     abstract void flushWrites();
 500
 501 }