diff --git a/src/java/org/apache/cassandra/journal/ActiveSegment.java b/src/java/org/apache/cassandra/journal/ActiveSegment.java index 34caaa690f..373b766267 100644 --- a/src/java/org/apache/cassandra/journal/ActiveSegment.java +++ b/src/java/org/apache/cassandra/journal/ActiveSegment.java @@ -525,6 +525,11 @@ public final class ActiveSegment extends Segment { return start; } + + RecordPointer recordPointer() + { + return new RecordPointer(descriptor.timestamp, start); + } } private int maybeCompleteInProgress() diff --git a/src/java/org/apache/cassandra/journal/Flusher.java b/src/java/org/apache/cassandra/journal/Flusher.java index 2db7b0f6d0..70ec6d5593 100644 --- a/src/java/org/apache/cassandra/journal/Flusher.java +++ b/src/java/org/apache/cassandra/journal/Flusher.java @@ -401,7 +401,7 @@ final class Flusher private interface Mode { - void flushAndAwaitDurable(ActiveSegment.Allocation alloc); + RecordPointer flushAndAwaitDurable(ActiveSegment.Allocation alloc); RecordPointer flushAsync(ActiveSegment.Allocation alloc); boolean isDurable(RecordPointer recordPointer); } @@ -409,13 +409,14 @@ final class Flusher private class BatchMode implements Mode { @Override - public void flushAndAwaitDurable(ActiveSegment.Allocation alloc) + public RecordPointer flushAndAwaitDurable(ActiveSegment.Allocation alloc) { pending.incrementAndGet(); requestExtraFlush(); alloc.awaitDurable(journal.metrics.waitingOnFlush); pending.decrementAndGet(); written.incrementAndGet(); + return alloc.recordPointer(); } @Override @@ -423,7 +424,7 @@ final class Flusher { requestExtraFlush(); written.incrementAndGet(); - return new RecordPointer(alloc.descriptor().timestamp, alloc.start()); + return alloc.recordPointer(); } @Override @@ -436,19 +437,20 @@ final class Flusher private class GroupMode implements Mode { @Override - public void flushAndAwaitDurable(ActiveSegment.Allocation alloc) + public RecordPointer flushAndAwaitDurable(ActiveSegment.Allocation alloc) { pending.incrementAndGet(); alloc.awaitDurable(journal.metrics.waitingOnFlush); pending.decrementAndGet(); written.incrementAndGet(); + return alloc.recordPointer(); } @Override public RecordPointer flushAsync(ActiveSegment.Allocation alloc) { written.incrementAndGet(); - return new RecordPointer(alloc.descriptor().timestamp, alloc.start()); + return alloc.recordPointer(); } @Override @@ -461,7 +463,7 @@ final class Flusher private class PeriodicMode implements Mode { @Override - public void flushAndAwaitDurable(ActiveSegment.Allocation alloc) + public RecordPointer flushAndAwaitDurable(ActiveSegment.Allocation alloc) { RecordPointer pointer = flushAsync(alloc); @@ -472,6 +474,7 @@ final class Flusher awaitFsyncAt(expectedFsyncTime, journal.metrics.waitingOnFlush.time()); pending.decrementAndGet(); } + return pointer; } @Override diff --git a/src/java/org/apache/cassandra/journal/Journal.java b/src/java/org/apache/cassandra/journal/Journal.java index 085848588c..4db2f7bfd8 100644 --- a/src/java/org/apache/cassandra/journal/Journal.java +++ b/src/java/org/apache/cassandra/journal/Journal.java @@ -487,7 +487,7 @@ public class Journal implements Shutdownable * @param id user-provided record id, expected to roughly correlate with time and go up * @param record the record to store */ - public void blockingWrite(K id, V record) + public RecordPointer blockingWrite(K id, V record) { try (DataOutputBuffer dob = DataOutputBuffer.scratchBuffer.get()) { @@ -495,6 +495,7 @@ public class Journal implements Shutdownable ActiveSegment.Allocation alloc = allocate(dob.getLength()); alloc.writeInternal(id, dob.unsafeGetBufferAndFlip()); flusher.flushAndAwaitDurable(alloc); + return alloc.recordPointer(); } catch (IOException e) { diff --git a/src/java/org/apache/cassandra/service/tracking/MutationId.java b/src/java/org/apache/cassandra/service/tracking/MutationId.java new file mode 100644 index 0000000000..9132676530 --- /dev/null +++ b/src/java/org/apache/cassandra/service/tracking/MutationId.java @@ -0,0 +1,84 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.service.tracking; + +public class MutationId +{ + /** + * 4 byte TCM host id + 4 byte host log id packed into a long. + * Host log ID is unique within the host, allocated + * anew on host restart - one per token range replicated by the host, + * persisted on allocation, unique within the host. + */ + public final long logId; + + /** + * 4 byte position + 4 byte timestamp packed into a long. + * Position is incremented, the timestamp is monotonically non-decreasing. + * The position is enough to identify the entry within a coordinator log, + * the timestamp is added for correlation purposes. + */ + public final long sequenceId; + + MutationId(long logId, long sequenceId) + { + this.logId = logId; + this.sequenceId = sequenceId; + } + + public int hostId() + { + return (int) (0xffffffffL & (logId >> 32)); + } + + public int hostLogId() + { + return (int) (0xffffffffL & logId); + } + + public int position() + { + return (int) (0xffffffffL & (sequenceId >> 32)); + } + + public int timestamp() + { + return (int) (0xffffffffL & sequenceId); + } + + @Override + public boolean equals(Object o) + { + if (this == o) return true; + if (!(o instanceof MutationId)) return false; + MutationId that = (MutationId) o; + return this.logId == that.logId && this.sequenceId == that.sequenceId; + } + + @Override + public int hashCode() + { + return Long.hashCode(logId) + 31 * Long.hashCode(sequenceId); + } + + @Override + public String toString() + { + return "MutationId{" + logId + ", " + sequenceId + '}'; + } +} diff --git a/src/java/org/apache/cassandra/service/tracking/MutationJournal.java b/src/java/org/apache/cassandra/service/tracking/MutationJournal.java new file mode 100644 index 0000000000..3f00e4387c --- /dev/null +++ b/src/java/org/apache/cassandra/service/tracking/MutationJournal.java @@ -0,0 +1,235 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.service.tracking; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Collection; +import java.util.concurrent.TimeUnit; +import java.util.zip.Checksum; + +import javax.annotation.Nullable; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; + +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.Mutation; +import org.apache.cassandra.db.TypeSizes; +import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.DataOutputPlus; +import org.apache.cassandra.io.util.File; +import org.apache.cassandra.journal.Journal; +import org.apache.cassandra.journal.KeySupport; +import org.apache.cassandra.journal.Params; +import org.apache.cassandra.journal.RecordConsumer; +import org.apache.cassandra.journal.RecordPointer; +import org.apache.cassandra.journal.SegmentCompactor; +import org.apache.cassandra.journal.ValueSerializer; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.utils.FBUtilities; + +public class MutationJournal +{ + private final Journal journal; + + private MutationJournal() + { + this(new File(DatabaseDescriptor.getCommitLogLocation()), new JournalParams()); + } + + @VisibleForTesting + MutationJournal(File directory, Params params) + { + journal = new Journal<>("MutationJournal", directory, params, new MutationIdSupport(), new MutationSerializer(), SegmentCompactor.noop()); + } + + public void start() + { + journal.start(); + } + + public void shutdownBlocking() + { + journal.shutdown(); + } + + public RecordPointer write(MutationId id, Mutation mutation) + { + return journal.blockingWrite(id, mutation); + } + + @Nullable + public Mutation read(MutationId id) + { + return journal.readLast(id); + } + + public boolean read(MutationId id, RecordConsumer consumer) + { + return journal.readLast(id, consumer); + } + + public void readAll(Iterable ids, Collection into) + { + for (MutationId id : ids) + { + Mutation mutation = read(id); + Preconditions.checkState(mutation != null); + into.add(mutation); + } + } + + static class JournalParams implements Params + { + @Override + public int segmentSize() + { + return DatabaseDescriptor.getCommitLogSegmentSize(); + } + + @Override + public FailurePolicy failurePolicy() + { + return FailurePolicy.STOP; + } + + @Override + public FlushMode flushMode() + { + return FlushMode.PERIODIC; + } + + @Override + public boolean enableCompaction() + { + return false; + } + + @Override + public long compactionPeriod(TimeUnit units) + { + return 0; + } + + @Override + public long flushPeriod(TimeUnit units) + { + return units.convert(DatabaseDescriptor.getCommitLogSyncPeriod(), TimeUnit.MILLISECONDS); + } + + @Override + public long periodicBlockPeriod(TimeUnit units) + { + return units.convert(DatabaseDescriptor.getPeriodicCommitLogSyncBlock(), TimeUnit.MILLISECONDS); + } + + @Override + public int userVersion() + { + return MessagingService.current_version; + } + } + + static class MutationIdSupport implements KeySupport + { + static final int LOG_ID_OFFSET = 0; + static final int SEQUENCE_ID_OFFSET = LOG_ID_OFFSET + TypeSizes.LONG_SIZE; + + @Override + public int serializedSize(int userVersion) + { + return TypeSizes.LONG_SIZE // logId + + TypeSizes.LONG_SIZE; // sequenceId + } + + @Override + public void serialize(MutationId id, DataOutputPlus out, int userVersion) throws IOException + { + out.writeLong(id.logId); + out.writeLong(id.sequenceId); + } + + @Override + public void serialize(MutationId id, ByteBuffer out, int userVersion) throws IOException + { + out.putLong(id.logId); + out.putLong(id.sequenceId); + } + + @Override + public MutationId deserialize(DataInputPlus in, int userVersion) throws IOException + { + long logId = in.readLong(); + long sequenceId = in.readLong(); + return new MutationId(logId, sequenceId); + } + + @Override + public MutationId deserialize(ByteBuffer buffer, int position, int userVersion) + { + long logId = buffer.getLong(position + LOG_ID_OFFSET); + long sequenceId = buffer.getLong(position + SEQUENCE_ID_OFFSET); + return new MutationId(logId, sequenceId); + } + + @Override + public MutationId deserialize(ByteBuffer buffer, int userVersion) + { + long logId = buffer.getLong(); + long sequenceId = buffer.getLong(); + return new MutationId(logId, sequenceId); + } + + @Override + public void updateChecksum(Checksum crc, MutationId id, int userVersion) + { + FBUtilities.updateChecksumLong(crc, id.logId); + FBUtilities.updateChecksumLong(crc, id.sequenceId); + } + + @Override + public int compareWithKeyAt(MutationId id, ByteBuffer buffer, int position, int userVersion) + { + int cmp = Long.compare(id.logId, buffer.getLong(position + LOG_ID_OFFSET)); + return cmp != 0 ? cmp : Long.compare(id.sequenceId, buffer.getLong(position + SEQUENCE_ID_OFFSET)); + } + + @Override + public int compare(MutationId id1, MutationId id2) + { + int cmp = Long.compare(id1.logId, id2.logId); + return cmp != 0 ? cmp : Long.compare(id1.sequenceId, id2.sequenceId); + } + } + + static class MutationSerializer implements ValueSerializer + { + @Override + public void serialize(MutationId id, Mutation mutation, DataOutputPlus out, int userVersion) throws IOException + { + Mutation.serializer.serialize(mutation, out, userVersion); + } + + @Override + public Mutation deserialize(MutationId id, DataInputPlus in, int userVersion) throws IOException + { + return Mutation.serializer.deserialize(in, userVersion); + } + } +} diff --git a/test/unit/org/apache/cassandra/service/tracking/MutationJournalTest.java b/test/unit/org/apache/cassandra/service/tracking/MutationJournalTest.java new file mode 100644 index 0000000000..55e97f57bd --- /dev/null +++ b/test/unit/org/apache/cassandra/service/tracking/MutationJournalTest.java @@ -0,0 +1,154 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.service.tracking; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.List; + +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + +import org.apache.cassandra.SchemaLoader; +import org.apache.cassandra.db.Mutation; +import org.apache.cassandra.db.RowUpdateBuilder; +import org.apache.cassandra.db.marshal.UTF8Type; +import org.apache.cassandra.io.util.DataOutputBuffer; +import org.apache.cassandra.io.util.File; +import org.apache.cassandra.journal.TestParams; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.schema.KeyspaceParams; +import org.apache.cassandra.schema.Schema; +import org.apache.cassandra.schema.TableMetadata; + +import static org.junit.Assert.assertEquals; + +/** + * Tests to sanity-check the integration points with Journal + * (mutation id and mutation ser/de, comparison, etc.) + */ +public class MutationJournalTest +{ + private static final String KEYSPACE = "mjtks"; + private static final String TABLE = "mjtt"; + + private static MutationJournal journal; + + @BeforeClass + public static void setUp() throws IOException + { + SchemaLoader.prepareServer(); + SchemaLoader.createKeyspace(KEYSPACE, KeyspaceParams.simple(3), + TableMetadata.builder(KEYSPACE, TABLE) + .addPartitionKeyColumn("pk", UTF8Type.instance) + .addClusteringColumn("ck", UTF8Type.instance) + .addRegularColumn("value", UTF8Type.instance) + .build()); + + File directory = new File(Files.createTempDirectory("mutation-journal-test-simple")); + directory.deleteRecursiveOnExit(); + + journal = new MutationJournal(directory, TestParams.INSTANCE); + journal.start(); + } + + @AfterClass + public static void tearDown() + { + journal.shutdownBlocking(); + } + + @Test + public void testWriteOneReadOne() + { + Mutation expected = + new RowUpdateBuilder(Schema.instance.getTableMetadata(KEYSPACE, TABLE), 0, "key") + .clustering("ck") + .add("value", "value") + .build(); + + MutationId id = new MutationId(100L, 0); + journal.write(id, expected); + + // regular read + Mutation actual = journal.read(id); + assertMutationEquals(expected, actual); + + // read via RecordConsumer + journal.read(id, ((segment, position, key, buffer, userVersion) -> + { + assertEquals(id, key); + assertEquals(serialize(expected), buffer); + })); + } + + @Test + public void testWriteManyReadMany() + { + Mutation expected1 = + new RowUpdateBuilder(Schema.instance.getTableMetadata(KEYSPACE, TABLE), 0, "key1") + .clustering("ck1") + .add("value", "value1") + .build(); + Mutation expected2 = + new RowUpdateBuilder(Schema.instance.getTableMetadata(KEYSPACE, TABLE), 0, "key2") + .clustering("ck2") + .add("value", "value2") + .build(); + List expected = List.of(expected1, expected2); + + MutationId id1 = new MutationId(100L, 1); + MutationId id2 = new MutationId(100L, 2); + List ids = List.of(id1, id2); + + journal.write(id1, expected1); + journal.write(id2, expected2); + + List actual = new ArrayList<>(); + journal.readAll(ids, actual); + assertMutationsEqual(expected, actual); + } + + private static void assertMutationEquals(Mutation expected, Mutation actual) + { + assertEquals(serialize(expected), serialize(actual)); + } + + private static void assertMutationsEqual(List expected, List actual) + { + assertEquals(expected.size(), actual.size()); + for (int i = 0; i < expected.size(); i++) + assertMutationEquals(expected.get(i), actual.get(i)); + } + + private static ByteBuffer serialize(Mutation mutation) + { + try (DataOutputBuffer out = DataOutputBuffer.scratchBuffer.get()) + { + Mutation.serializer.serialize(mutation, out, MessagingService.maximum_version); + return out.asNewBuffer(); + } + catch (IOException e) + { + throw new AssertionError(e); + } + } +}