You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by "belliottsmith (via GitHub)" <gi...@apache.org> on 2024/05/31 20:55:24 UTC
[PR] Periodic commit log [cassandra]
belliottsmith opened a new pull request, #3343:
URL: https://github.com/apache/cassandra/pull/3343
Thanks for sending a pull request! Here are some tips if you're new here:
* Ensure you have added or run the [appropriate tests](https://cassandra.apache.org/_/development/testing.html) for your PR.
* Be sure to keep the PR description updated to reflect all changes.
* Write your PR title to summarize what this PR proposes.
* If possible, provide a concise example to reproduce the issue for a faster review.
* Read our [contributor guidelines](https://cassandra.apache.org/_/development/index.html)
* If you're making a documentation change, see our [guide to documentation contribution](https://cassandra.apache.org/_/development/documentation.html)
Commit messages should follow the following format:
```
<One sentence description, usually Jira title or CHANGES.txt summary>
<Optional lengthier description (context on patch)>
patch by <Authors>; reviewed by <Reviewers> for CASSANDRA-#####
Co-authored-by: Name1 <email1>
Co-authored-by: Name2 <email2>
```
The [Cassandra Jira](https://issues.apache.org/jira/projects/CASSANDRA/issues/)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Periodic commit log [cassandra]
Posted by "iamaleksey (via GitHub)" <gi...@apache.org>.
iamaleksey commented on code in PR #3343:
URL: https://github.com/apache/cassandra/pull/3343#discussion_r1627437894
##########
src/java/org/apache/cassandra/service/accord/AccordKeyspace.java:
##########
@@ -99,6 +99,7 @@
import org.apache.cassandra.db.SinglePartitionReadCommand;
import org.apache.cassandra.db.Slices;
import org.apache.cassandra.db.filter.ClusteringIndexFilter;
+import org.apache.cassandra.db.filter.ClusteringIndexNamesFilter;
import org.apache.cassandra.db.filter.ClusteringIndexSliceFilter;
Review Comment:
Nit: unused import breaking checkstyle/build
##########
src/java/org/apache/cassandra/service/accord/AccordKeyspace.java:
##########
@@ -99,6 +99,7 @@
import org.apache.cassandra.db.SinglePartitionReadCommand;
import org.apache.cassandra.db.Slices;
Review Comment:
Nit: unused import breaking checkstyle/build
##########
src/java/org/apache/cassandra/journal/ActiveSegment.java:
##########
@@ -50,6 +51,8 @@ final class ActiveSegment<K, V> extends Segment<K, V>
* Everything before this offset has been written and flushed.
*/
private volatile int lastFlushedOffset = 0;
+ private volatile int lastFsyncOffset = 0;
+ private static final AtomicIntegerFieldUpdater lastFsyncOffsetUpdater = AtomicIntegerFieldUpdater.newUpdater(ActiveSegment.class, "lastFsyncOffset");
Review Comment:
Nit: specify at least `<ActiveSegment>` with `@SuppressWarnings("rawtypes")`?
##########
src/java/org/apache/cassandra/journal/Segments.java:
##########
@@ -98,6 +98,21 @@ void selectActive(long maxTimestamp, Collection<ActiveSegment<K, V>> into)
into.add(segment.asActive());
}
+ ActiveSegment<K, V> oldest()
Review Comment:
Nit: rename to `oldestActive()`?
##########
src/java/org/apache/cassandra/journal/ActiveSegment.java:
##########
@@ -261,21 +264,35 @@ boolean shouldFlush()
* TODO FIXME: calls from outside Flusher + callbacks
* @return last synced offset
*/
- synchronized int flush()
+ synchronized int flush(boolean fsync)
{
int allocatePosition = this.allocatePosition.get();
if (lastFlushedOffset >= allocatePosition)
return lastFlushedOffset;
waitForModifications();
- flushInternal();
+ if (fsync) flushInternal();
lastFlushedOffset = allocatePosition;
+ if (fsync)
+ lastFsyncOffsetUpdater.accumulateAndGet(this, allocatePosition, Math::max);
int syncedOffset = Math.min(allocatePosition, endOfBuffer);
- syncedOffsets.mark(syncedOffset);
+ syncedOffsets.mark(syncedOffset, fsync);
flushComplete.signalAll();
return syncedOffset;
}
+ // provides no ordering guarantees
+ void fsync()
+ {
+ int lastFlushed = lastFlushedOffset;
+ if (lastFsyncOffset >= lastFlushed)
+ return;
+
+ flushInternal();
Review Comment:
Rename to `fsyncInternal()`? flush vs fsync terminology becomes a little confusing here overall, I wonder if there is a better way to deconfuse these.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Periodic commit log [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith commented on code in PR #3343:
URL: https://github.com/apache/cassandra/pull/3343#discussion_r1622970312
##########
src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java:
##########
@@ -99,6 +100,13 @@ public void serialize(T ts, DataOutputPlus out, int version) throws IOException
TopologySerializers.nodeId.serialize(ts.node, out, version);
}
+ public void serialize(T ts, DataOutputPlus out) throws IOException
Review Comment:
because we have some other callers that don't currently have a version, so easier to invoke ones without a version
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Periodic commit log [cassandra]
Posted by "ifesdjeen (via GitHub)" <gi...@apache.org>.
ifesdjeen commented on code in PR #3343:
URL: https://github.com/apache/cassandra/pull/3343#discussion_r1624216910
##########
src/java/org/apache/cassandra/journal/Flusher.java:
##########
@@ -131,8 +260,8 @@ public void run(Interruptible.State state) throws InterruptedException
public void doRun(Interruptible.State state) throws InterruptedException
{
- long startedRunAt = clock.now();
- boolean flushToDisk = lastFlushedAt + flushPeriodNanos() <= startedRunAt || state != NORMAL || flushRequested;
+ long startedAt = clock.now();
+ boolean flushToDisk = flushStartedAt + flushPeriodNanos() <= startedAt || state != NORMAL || flushRequested;
Review Comment:
nit: can move `flushPeriodNanos` that dereferences the configurable above this line, and reuse it
##########
src/java/org/apache/cassandra/journal/Flusher.java:
##########
@@ -97,20 +102,144 @@ void start()
void shutdown()
{
flushExecutor.shutdown();
+ if (fsyncExecutor != null)
+ fsyncExecutor.shutdown();
}
@Simulate(with={MONITORS,GLOBAL_CLOCK,LOCK_SUPPORT})
private class FlushRunnable implements Interruptible.Task
{
- private final MonotonicClock clock;
+ @Simulate(with={MONITORS,GLOBAL_CLOCK,LOCK_SUPPORT})
+ private class FSyncRunnable implements Interruptible.Task
+ {
+ // this is written only by the Flusher thread, and read only by the Fsync thread
+ ActiveSegment<K, V> fsyncTo;
Review Comment:
nit: how about `fsyncUpTo`? Not a native speaker, but when I see the variable first association was that this is a target of fsync.
##########
src/java/org/apache/cassandra/journal/Flusher.java:
##########
@@ -162,90 +286,100 @@ public void doRun(Interruptible.State state) throws InterruptedException
}
else
{
- long wakeUpAt = startedRunAt + flushPeriodNanos;
- if (wakeUpAt > now)
- haveWork.tryAcquireUntil(1, wakeUpAt);
+ long wakeUpAt = startedAt + flushPeriodNanos;
+ haveWork.tryAcquireUntil(1, wakeUpAt);
}
}
- private void doFlush()
+ private void doFlush(long startedAt) throws InterruptedException
{
- journal.selectSegmentToFlush(segmentsToFlush);
- segmentsToFlush.sort(comparing(s -> s.descriptor));
+ boolean synchronousFsync = fSyncRunnable == null;
- try
- {
- long syncedSegment = -1;
- int syncedOffset = -1;
+ if (current == null)
+ current = journal.oldestActiveSegment();
+ ActiveSegment<K, V> newCurrent = journal.currentActiveSegment();
- for (ActiveSegment<K, V> segment : segmentsToFlush)
- {
- if (!segment.shouldFlush())
- break;
+ if (newCurrent == current && (newCurrent == null || !newCurrent.shouldFlush()))
+ {
+ if (synchronousFsync) fsyncFinishedFor = startedAt;
+ else fSyncRunnable.doNoOpFlush(startedAt);
+ return;
+ }
- syncedSegment = segment.descriptor.timestamp;
- syncedOffset = segment.flush();
+ Invariants.checkState(newCurrent != null);
- // if an older segment isn't fully complete + flushed yet, don't attempt to flush any younger ones
- if (!segment.isCompletedAndFullyFlushed(syncedOffset))
- break;
+ try
+ {
+ while (current != newCurrent)
+ {
+ current.discardUnusedTail();
+ current.flush(synchronousFsync);
+ if (synchronousFsync)
+ journal.closeActiveSegmentAndOpenAsStatic(current);
+ current = journal.getActiveSegment(current.descriptor.timestamp + 1);
}
+ int syncedOffset = current.flush(synchronousFsync);
- // invoke the onFlush() callback once, covering entire flushed range across all flushed segments
- if (syncedSegment != -1 && syncedOffset != -1)
- callbacks.onFlush(syncedSegment, syncedOffset);
+ if (synchronousFsync) postFlushSync(startedAt, current.descriptor.timestamp, syncedOffset);
Review Comment:
nit: maybe rename this to `afterFsync` or `afterFlushSync`?
##########
src/java/org/apache/cassandra/journal/Flusher.java:
##########
@@ -60,16 +62,19 @@ final class Flusher<K, V>
private final AsyncCallbacks<K, V> callbacks;
private volatile Interruptible flushExecutor;
+ private volatile Interruptible fsyncExecutor;
// counts of total pending write and written entries
private final AtomicLong pending = new AtomicLong(0);
private final AtomicLong written = new AtomicLong(0);
- // all Allocations written before this time will be flushed
- volatile long lastFlushedAt = currentTimeMillis();
+ // the time of the last initiated flush
+ volatile long flushStartedAt = nanoTime();
+ // the time of the earliest flush that has completed an fsync; all Allocations written before this time are durable
+ volatile long fsyncFinishedFor = flushStartedAt;
// a signal that writers can wait on to be notified of a completed flush in PERIODIC FlushMode
- private final WaitQueue flushComplete = newWaitQueue();
+ private final WaitQueue flushComplete = newWaitQueue(); // TODO (expected): this is only used for testing, can we remove this?
Review Comment:
Since there is now distinction between flush and fsync, should we rename this to `fsyncComplete` ?
##########
src/java/org/apache/cassandra/journal/Flusher.java:
##########
@@ -97,20 +102,144 @@ void start()
void shutdown()
{
flushExecutor.shutdown();
+ if (fsyncExecutor != null)
+ fsyncExecutor.shutdown();
}
@Simulate(with={MONITORS,GLOBAL_CLOCK,LOCK_SUPPORT})
private class FlushRunnable implements Interruptible.Task
{
- private final MonotonicClock clock;
+ @Simulate(with={MONITORS,GLOBAL_CLOCK,LOCK_SUPPORT})
+ private class FSyncRunnable implements Interruptible.Task
+ {
+ // this is written only by the Flusher thread, and read only by the Fsync thread
+ ActiveSegment<K, V> fsyncTo;
+ ActiveSegment<K, V> fsyncing;
+
+ private volatile Thread awaitingWork;
+
+ // all Allocations written before this time will be written to at least the OS page cache;
+ volatile long fsyncWaitingSince = 0;
+ // the time of the earliest flush that has begun participating in an fsync
+ volatile long fsyncStartedFor = 0;
+
+ @Override
+ public void run(Interruptible.State state) throws InterruptedException
+ {
+ try
+ {
+ doRun(state);
+ }
+ catch (Throwable t)
+ {
+ if (!journal.handleError("Failed to flush segments to disk", t))
+ throw new TerminateException();
+ }
+ }
+
+ private void awaitWork() throws InterruptedException
+ {
+ long lastStartedAt = fsyncStartedFor;
+ if (fsyncWaitingSince != lastStartedAt)
+ return;
+
+ awaitingWork = Thread.currentThread();
+ do
+ {
+ if (Thread.interrupted())
+ {
+ awaitingWork = null;
+ throw new InterruptedException();
+ }
+
+ LockSupport.park();
+ }
+ while (fsyncWaitingSince == lastStartedAt);
+
+ awaitingWork = null;
+ }
+
+ void notify(Thread notify)
+ {
+ if (notify != null)
+ LockSupport.unpark(notify);
+ }
+
+ public void doRun(Interruptible.State state) throws InterruptedException
+ {
+ awaitWork();
+ if (fsyncing == null)
+ fsyncing = journal.oldestActiveSegment();
+
+ // invert order of access; we might see a future fsyncTo, but at worst this means redundantly invoking fsync before updating fsyncStartedFor
+ long startedAt = fsyncWaitingSince;
+ ActiveSegment<K, V> fsyncTo = this.fsyncTo;
+ fsyncStartedFor = startedAt;
+ // synchronized to prevent thread interrupts while performing IO operations and also
+ // clear interrupted status to prevent ClosedByInterruptException in ActiveSegment::flush
+ synchronized (this)
+ {
+ boolean ignore = Thread.interrupted();
+ while (fsyncing != fsyncTo)
+ {
+ fsyncing.fsync();
+ journal.closeActiveSegmentAndOpenAsStatic(fsyncing);
+ fsyncing = journal.getActiveSegment(fsyncing.descriptor.timestamp + 1);
+ }
+ fsyncing.fsync();
+ }
+ fsyncFinishedFor = startedAt;
+ flushComplete.signalAll();
+ long finishedAt = clock.now();
+ processDuration(startedAt, finishedAt);
+ }
+
+ void postFlush(long startedAt, ActiveSegment<K, V> segment, int syncedOffset) throws InterruptedException
+ {
+ long requireFsyncTo = startedAt - periodicFlushLagBlockNanos();
+
+ fsyncTo = segment;
Review Comment:
nit: I understand that this is an additional allocation; but maybe we could make these two fields a single variable?
An alternative could be a sentinel boolean that would make another thread spin.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Periodic commit log [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith commented on code in PR #3343:
URL: https://github.com/apache/cassandra/pull/3343#discussion_r1622970585
##########
test/distributed/org/apache/cassandra/distributed/test/TestBaseImpl.java:
##########
@@ -81,6 +84,7 @@ public void afterEach() {
@BeforeClass
public static void beforeClass() throws Throwable
{
+ CassandraRelevantProperties.SIMULATOR_STARTED.setString(Long.toString(MILLISECONDS.toSeconds(currentTimeMillis())));
Review Comment:
Yes, it's for log.xml compatibility
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Periodic commit log [cassandra]
Posted by "iamaleksey (via GitHub)" <gi...@apache.org>.
iamaleksey commented on code in PR #3343:
URL: https://github.com/apache/cassandra/pull/3343#discussion_r1629505033
##########
src/java/org/apache/cassandra/journal/ActiveSegment.java:
##########
@@ -261,21 +264,35 @@ boolean shouldFlush()
* TODO FIXME: calls from outside Flusher + callbacks
* @return last synced offset
*/
- synchronized int flush()
+ synchronized int flush(boolean fsync)
{
int allocatePosition = this.allocatePosition.get();
if (lastFlushedOffset >= allocatePosition)
return lastFlushedOffset;
waitForModifications();
- flushInternal();
+ if (fsync) flushInternal();
lastFlushedOffset = allocatePosition;
Review Comment:
Nit: move this one line up and put the `flushIntearnal()` call (`fsyncInternal()`?) and updating the last fsync offset in one `if (fsync)` block?
##########
src/java/org/apache/cassandra/journal/Journal.java:
##########
@@ -659,6 +657,47 @@ void selectSegmentToFlush(Collection<ActiveSegment<K, V>> into)
segments().selectActive(currentSegment.descriptor.timestamp, into);
}
+ ActiveSegment<K, V> oldestActiveSegment()
+ {
+ ActiveSegment<K, V> current = currentSegment;
+ if (current == null)
Review Comment:
I don't think this can be `null`, at least as currently invoked by `Flusher`, started after we've advanced for the first time already. I don't mind the check, but if we have it here then we should probably put one in `getActiveSegment()` below as well for consistency at least?
##########
src/java/org/apache/cassandra/journal/Journal.java:
##########
@@ -659,6 +657,47 @@ void selectSegmentToFlush(Collection<ActiveSegment<K, V>> into)
segments().selectActive(currentSegment.descriptor.timestamp, into);
}
+ ActiveSegment<K, V> oldestActiveSegment()
+ {
+ ActiveSegment<K, V> current = currentSegment;
+ if (current == null)
+ return null;
+
+ ActiveSegment<K, V> oldest = segments().oldest();
+ if (oldest == null || oldest.descriptor.timestamp > current.descriptor.timestamp)
+ return current;
+
+ return oldest;
+ }
+
+ ActiveSegment<K, V> currentActiveSegment()
+ {
+ return currentSegment;
+ }
+
+ ActiveSegment<K, V> getActiveSegment(long timestamp)
+ {
+ // we can race with segment addition to the segments() collection, with a new segment appearing in currentSegment first
+ // since we are most likely to be requesting the currentSegment anyway, we resolve this case by checking currentSegment first
+ // and resort to the segments() collection only if we do not match
+ ActiveSegment<K, V> currentSegment = this.currentSegment;
+ long currentSegmentTimestamp = currentSegment.descriptor.timestamp;
+ if (timestamp == currentSegmentTimestamp)
+ {
+ return currentSegment;
+ }
+ else if (timestamp > currentSegmentTimestamp)
+ {
+ throw new IllegalArgumentException("Requested a newer timestamp " + timestamp + " than the current active segment " + currentSegmentTimestamp);
+ }
+ else
+ {
+ Segment<K, V> segment = segments().get(timestamp);
+ Invariants.checkState(segment != null, "Segment %d expected to be found, but neither current segment %d nor in active segments", timestamp, currentSegmentTimestamp);
+ return segment.asActive();
Review Comment:
Given the way it's currently used, I imagine it should never be a static segment, but maybe still ensure that we are only querying for active segments? Maybe rename `get` to `getActive()` and add an `isActive()` check there, and the more specific return type so no casting is needed here?
##########
src/java/org/apache/cassandra/journal/Journal.java:
##########
@@ -659,6 +657,47 @@ void selectSegmentToFlush(Collection<ActiveSegment<K, V>> into)
segments().selectActive(currentSegment.descriptor.timestamp, into);
}
+ ActiveSegment<K, V> oldestActiveSegment()
+ {
+ ActiveSegment<K, V> current = currentSegment;
+ if (current == null)
+ return null;
+
+ ActiveSegment<K, V> oldest = segments().oldest();
+ if (oldest == null || oldest.descriptor.timestamp > current.descriptor.timestamp)
+ return current;
+
+ return oldest;
+ }
+
+ ActiveSegment<K, V> currentActiveSegment()
+ {
+ return currentSegment;
+ }
+
+ ActiveSegment<K, V> getActiveSegment(long timestamp)
+ {
+ // we can race with segment addition to the segments() collection, with a new segment appearing in currentSegment first
+ // since we are most likely to be requesting the currentSegment anyway, we resolve this case by checking currentSegment first
+ // and resort to the segments() collection only if we do not match
+ ActiveSegment<K, V> currentSegment = this.currentSegment;
+ long currentSegmentTimestamp = currentSegment.descriptor.timestamp;
Review Comment:
See the comment re: `null` check above ^
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Periodic commit log [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith commented on code in PR #3343:
URL: https://github.com/apache/cassandra/pull/3343#discussion_r1622970964
##########
test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java:
##########
@@ -43,20 +55,35 @@ public class AccordLoadTest extends AccordTestBase
@BeforeClass
public static void setUp() throws IOException
{
- AccordTestBase.setupCluster(builder -> builder.withConfig(config -> config.set("lwt_strategy", "accord").set("non_serial_write_strategy", "accord")), 2);
+ CassandraRelevantProperties.SIMULATOR_STARTED.setString(Long.toString(MILLISECONDS.toSeconds(currentTimeMillis())));
Review Comment:
the simulator log file is the most suitable for running this, and this exists only to setup logging for the instances for that file
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Periodic commit log [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith commented on code in PR #3343:
URL: https://github.com/apache/cassandra/pull/3343#discussion_r1622972817
##########
test/simulator/main/org/apache/cassandra/simulator/paxos/PairOfSequencesAccordSimulation.java:
##########
@@ -149,7 +149,7 @@ public PairOfSequencesAccordSimulation(SimulatedSystems simulated,
@Override
protected String createTableStmt()
{
- return "CREATE TABLE " + KEYSPACE + ".tbl (pk int, count int, seq text, PRIMARY KEY (pk))";
+ return "CREATE TABLE " + KEYSPACE + ".tbl (pk int, count int, seq text, PRIMARY KEY (pk)) WITH transactional_mode = 'full'";
Review Comment:
fair, I should pass this through yes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Periodic commit log [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith commented on PR #3343:
URL: https://github.com/apache/cassandra/pull/3343#issuecomment-2173718842
I think I have addressed feedback - could those waiting for changes confirm? Thx
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Periodic commit log [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith commented on code in PR #3343:
URL: https://github.com/apache/cassandra/pull/3343#discussion_r1622972664
##########
test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java:
##########
@@ -198,7 +198,7 @@ public static abstract class Builder<S extends Simulation>
protected HeapPool.Logged.Listener memoryListener;
protected SimulatedTime.Listener timeListener = (i1, i2) -> {};
protected LongConsumer onThreadLocalRandomCheck;
- protected String lwtStrategy = "migration";
+ protected String transactionMode = "migration";
Review Comment:
I think lwtStrategy doesn't exist anymore? I was probably doing the bare minimum to get it working
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Periodic commit log [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith commented on code in PR #3343:
URL: https://github.com/apache/cassandra/pull/3343#discussion_r1622970097
##########
src/java/org/apache/cassandra/journal/ActiveSegment.java:
##########
@@ -261,21 +264,34 @@ boolean shouldFlush()
* TODO FIXME: calls from outside Flusher + callbacks
* @return last synced offset
*/
- synchronized int flush()
+ synchronized int flush(boolean fsync)
{
int allocatePosition = this.allocatePosition.get();
if (lastFlushedOffset >= allocatePosition)
return lastFlushedOffset;
waitForModifications();
- flushInternal();
+ if (fsync) flushInternal();
lastFlushedOffset = allocatePosition;
+ lastFsyncOffsetUpdater.accumulateAndGet(this, allocatePosition, Math::max);
Review Comment:
good catch, so much for late night updates (when will I learn)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Periodic commit log [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith commented on code in PR #3343:
URL: https://github.com/apache/cassandra/pull/3343#discussion_r1639889148
##########
src/java/org/apache/cassandra/journal/Journal.java:
##########
@@ -659,6 +657,47 @@ void selectSegmentToFlush(Collection<ActiveSegment<K, V>> into)
segments().selectActive(currentSegment.descriptor.timestamp, into);
}
+ ActiveSegment<K, V> oldestActiveSegment()
+ {
+ ActiveSegment<K, V> current = currentSegment;
+ if (current == null)
Review Comment:
My feeling was that `getActiveSegment` expresses the belief that segment exists, so there must be an active segment. Whereas `oldestActiveSegment` doesn't necessarily express a belief that such a segment exists (without a `get` or `try` I suppose it is a bit ambiguous, but in particular without providing a key it is less implied).
That said, I don't have a strong feeling at all and happy to go with your preference here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Periodic commit log [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3343:
URL: https://github.com/apache/cassandra/pull/3343#discussion_r1622956753
##########
src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java:
##########
@@ -99,6 +100,13 @@ public void serialize(T ts, DataOutputPlus out, int version) throws IOException
TopologySerializers.nodeId.serialize(ts.node, out, version);
}
+ public void serialize(T ts, DataOutputPlus out) throws IOException
Review Comment:
why include ones without a version? what caused this change?
##########
test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java:
##########
@@ -43,20 +55,35 @@ public class AccordLoadTest extends AccordTestBase
@BeforeClass
public static void setUp() throws IOException
{
- AccordTestBase.setupCluster(builder -> builder.withConfig(config -> config.set("lwt_strategy", "accord").set("non_serial_write_strategy", "accord")), 2);
+ CassandraRelevantProperties.SIMULATOR_STARTED.setString(Long.toString(MILLISECONDS.toSeconds(currentTimeMillis())));
Review Comment:
why are you setting this? we don't setup simulator for jvm-dtest
##########
src/java/org/apache/cassandra/journal/ActiveSegment.java:
##########
@@ -261,21 +264,34 @@ boolean shouldFlush()
* TODO FIXME: calls from outside Flusher + callbacks
* @return last synced offset
*/
- synchronized int flush()
+ synchronized int flush(boolean fsync)
{
int allocatePosition = this.allocatePosition.get();
if (lastFlushedOffset >= allocatePosition)
return lastFlushedOffset;
waitForModifications();
- flushInternal();
+ if (fsync) flushInternal();
lastFlushedOffset = allocatePosition;
+ lastFsyncOffsetUpdater.accumulateAndGet(this, allocatePosition, Math::max);
Review Comment:
is this correct? if `fsync == false` then we didn't fsync... so why update the pointer?
##########
test/distributed/org/apache/cassandra/distributed/test/TestBaseImpl.java:
##########
@@ -81,6 +84,7 @@ public void afterEach() {
@BeforeClass
public static void beforeClass() throws Throwable
{
+ CassandraRelevantProperties.SIMULATOR_STARTED.setString(Long.toString(MILLISECONDS.toSeconds(currentTimeMillis())));
Review Comment:
did you mean to add this?
##########
test/simulator/main/org/apache/cassandra/simulator/paxos/PairOfSequencesAccordSimulation.java:
##########
@@ -149,7 +149,7 @@ public PairOfSequencesAccordSimulation(SimulatedSystems simulated,
@Override
protected String createTableStmt()
{
- return "CREATE TABLE " + KEYSPACE + ".tbl (pk int, count int, seq text, PRIMARY KEY (pk))";
+ return "CREATE TABLE " + KEYSPACE + ".tbl (pk int, count int, seq text, PRIMARY KEY (pk)) WITH transactional_mode = 'full'";
Review Comment:
"should" this use `transactionMode` since you are trying to pass that around now?
##########
test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java:
##########
@@ -75,14 +102,66 @@ public void testLoad() throws Exception
inFlight.acquire();
rateLimiter.acquire();
long commandStart = System.nanoTime();
- coordinator.executeWithResult((success, fail) -> {
- inFlight.release();
- if (fail == null) histogram.add(NANOSECONDS.toMicros(System.nanoTime() - commandStart));
-// else exceptions.add(fail);
- }, "UPDATE " + qualifiedTableName + " SET v += 1 WHERE k = ? IF EXISTS;", ConsistencyLevel.SERIAL, ConsistencyLevel.QUORUM, random.nextInt(keyCount));
+ if (random.nextFloat() < readChance)
+ {
+ coordinator.executeWithResult((success, fail) -> {
+ inFlight.release();
+ if (fail == null) histogram.add(NANOSECONDS.toMicros(System.nanoTime() - commandStart));
+ // else exceptions.add(fail);
+ }, "SELECT * FROM " + qualifiedTableName + " WHERE k = ?;", ConsistencyLevel.SERIAL, random.nextInt(keyCount));
+ }
+ else
+ {
+ coordinator.executeWithResult((success, fail) -> {
+ inFlight.release();
+ if (fail == null) histogram.add(NANOSECONDS.toMicros(System.nanoTime() - commandStart));
+ // else exceptions.add(fail);
+ }, "UPDATE " + qualifiedTableName + " SET v += 1 WHERE k = ? IF EXISTS;", ConsistencyLevel.SERIAL, ConsistencyLevel.QUORUM, random.nextInt(keyCount));
+ }
+ }
+
+ if ((nextRepairAt -= batchSize) <= 0)
+ {
+ nextRepairAt += repairInterval;
+ System.out.println("repairing...");
+ cluster.coordinator(1).instance().nodetool("repair", qualifiedTableName);
Review Comment:
nit:
```suggestion
cluster.coordinator(1).instance().nodetoolResult("repair", qualifiedTableName).asserts().success();
```
if repairs fails you won't know
##########
test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java:
##########
@@ -75,14 +102,66 @@ public void testLoad() throws Exception
inFlight.acquire();
rateLimiter.acquire();
long commandStart = System.nanoTime();
- coordinator.executeWithResult((success, fail) -> {
- inFlight.release();
- if (fail == null) histogram.add(NANOSECONDS.toMicros(System.nanoTime() - commandStart));
-// else exceptions.add(fail);
- }, "UPDATE " + qualifiedTableName + " SET v += 1 WHERE k = ? IF EXISTS;", ConsistencyLevel.SERIAL, ConsistencyLevel.QUORUM, random.nextInt(keyCount));
+ if (random.nextFloat() < readChance)
+ {
+ coordinator.executeWithResult((success, fail) -> {
Review Comment:
TIL; `Future<?> executeWithResult(BiConsumer<SimpleQueryResult, Throwable> callback, String query, ConsistencyLevel serialConsistencyLevel, ConsistencyLevel commitConsistencyLevel, Object... boundValues);`
heh...
##########
test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java:
##########
@@ -198,7 +198,7 @@ public static abstract class Builder<S extends Simulation>
protected HeapPool.Logged.Listener memoryListener;
protected SimulatedTime.Listener timeListener = (i1, i2) -> {};
protected LongConsumer onThreadLocalRandomCheck;
- protected String lwtStrategy = "migration";
+ protected String transactionMode = "migration";
Review Comment:
did you only keep this so paxos running accord doesn't change topologies? I deleted in my branch but I am working on trying to get topology changes working,
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Periodic commit log [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith commented on code in PR #3343:
URL: https://github.com/apache/cassandra/pull/3343#discussion_r1622971228
##########
test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java:
##########
@@ -75,14 +102,66 @@ public void testLoad() throws Exception
inFlight.acquire();
rateLimiter.acquire();
long commandStart = System.nanoTime();
- coordinator.executeWithResult((success, fail) -> {
- inFlight.release();
- if (fail == null) histogram.add(NANOSECONDS.toMicros(System.nanoTime() - commandStart));
-// else exceptions.add(fail);
- }, "UPDATE " + qualifiedTableName + " SET v += 1 WHERE k = ? IF EXISTS;", ConsistencyLevel.SERIAL, ConsistencyLevel.QUORUM, random.nextInt(keyCount));
+ if (random.nextFloat() < readChance)
+ {
+ coordinator.executeWithResult((success, fail) -> {
+ inFlight.release();
+ if (fail == null) histogram.add(NANOSECONDS.toMicros(System.nanoTime() - commandStart));
+ // else exceptions.add(fail);
+ }, "SELECT * FROM " + qualifiedTableName + " WHERE k = ?;", ConsistencyLevel.SERIAL, random.nextInt(keyCount));
+ }
+ else
+ {
+ coordinator.executeWithResult((success, fail) -> {
+ inFlight.release();
+ if (fail == null) histogram.add(NANOSECONDS.toMicros(System.nanoTime() - commandStart));
+ // else exceptions.add(fail);
+ }, "UPDATE " + qualifiedTableName + " SET v += 1 WHERE k = ? IF EXISTS;", ConsistencyLevel.SERIAL, ConsistencyLevel.QUORUM, random.nextInt(keyCount));
+ }
+ }
+
+ if ((nextRepairAt -= batchSize) <= 0)
+ {
+ nextRepairAt += repairInterval;
+ System.out.println("repairing...");
+ cluster.coordinator(1).instance().nodetool("repair", qualifiedTableName);
Review Comment:
yeah, this is just for chaos
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org