You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by GitBox <gi...@apache.org> on 2020/01/28 19:13:18 UTC
[GitHub] [cassandra] yifan-c opened a new pull request #433: Release
capacity at the end of the test round
yifan-c opened a new pull request #433: Release capacity at the end of the test round
URL: https://github.com/apache/cassandra/pull/433
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] yifan-c commented on a change in pull request #433:
Release capacity at the end of the test round
Posted by GitBox <gi...@apache.org>.
yifan-c commented on a change in pull request #433: Release capacity at the end of the test round
URL: https://github.com/apache/cassandra/pull/433#discussion_r384060436
##########
File path: test/unit/org/apache/cassandra/net/ConnectionTest.java
##########
@@ -766,40 +767,84 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
@Test
public void testAcquireReleaseOutbound() throws Throwable
{
+ // In each test round, K capacity is reserved upfront.
+ // Two groups of threads each release/acquire for K capacity in total accordingly,
+ // i.e. if only let the release threads to run, at the end, the reserved capacity is 0 (K - K).
+ // During the test, we expect N (N <= maxFailures) acquire attempts (for M capacity) to fail.
+ // The reserved capacity (pendingBytes) at the end of the round should equal to K - N * M,
+ // which you can find in the assertion.
test((inbound, outbound, endpoint) -> {
- ExecutorService executor = Executors.newFixedThreadPool(100);
- int acquireStep = 123;
- Assert.assertTrue(outbound.unsafeAcquireCapacity(100 * 10000, 100 * 10000 * acquireStep));
+ // max capacity equals to permit-free sendQueueCapcity + the minimun of endpoint and global reserve
+ double maxSendQueueCapacity = outbound.settings().applicationSendQueueCapacityInBytes +
+ Double.min(outbound.settings().applicationSendQueueReserveEndpointCapacityInBytes,
+ outbound.settings().applicationSendQueueReserveGlobalCapacityInBytes.limit());
+ int concurrency = 100;
+ int attempts = 10000;
+ int acquireCount = concurrency * attempts;
+ long acquireStep = Math.round(maxSendQueueCapacity * 1.2 / acquireCount / 2); // It is guranteed to acquire (~20%) more
+ // The total overly acquired amount divides the amount acquired in each step. Get the ceil value so not to miss the acquire that just exceeds.
+ long maxFailures = (long) Math.ceil((acquireCount * acquireStep * 2 - maxSendQueueCapacity) / acquireStep); // The result must be in the range of lone
AtomicLong acquisitionFailures = new AtomicLong();
- for (int i = 0; i < 100; i++)
- {
- executor.submit(() -> {
- for (int j = 0; j < 10000; j++)
- {
- if (!outbound.unsafeAcquireCapacity(acquireStep))
- acquisitionFailures.incrementAndGet();
- }
+ Runnable acquirer = () -> {
+ for (int j = 0; j < attempts; j++)
+ {
+ if (!outbound.unsafeAcquireCapacity(acquireStep))
+ acquisitionFailures.incrementAndGet();
+ }
+ };
+ Runnable releaser = () -> {
+ for (int j = 0; j < attempts; j++)
+ outbound.unsafeReleaseCapacity(acquireStep);
+ };
- });
- }
+ // Start N acquirer and releaser to contend for capcaity
+ List<Runnable> invokeOrder = new ArrayList<>();
+ for (int i = 0 ; i < concurrency ; ++i)
+ invokeOrder.add(acquirer);
+ for (int i = 0 ; i < concurrency ; ++i)
+ invokeOrder.add(releaser);
+ // randomize their start order
+ randomize(invokeOrder);
- for (int i = 0; i < 100; i++)
+ try
{
- executor.submit(() -> {
- for (int j = 0; j < 10000; j++)
- outbound.unsafeReleaseCapacity(acquireStep);
- });
- }
+ // Reserve enough capacity upfront to ensure the releaser threads cannot release all reserved capacity.
+ // i.e. the pendingBytes is always positive during the test.
+ Assert.assertTrue(outbound.unsafeAcquireCapacity(acquireCount, acquireCount * acquireStep));
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
- executor.shutdown();
- executor.awaitTermination(10, TimeUnit.SECONDS);
+ invokeOrder.forEach(executor::submit);
- // We can release more than we acquire, which certainly should not happen in
- // real life, but since it's a test just for acquisition and release, it is fine
- Assert.assertEquals(100 * 10000 * acquireStep - (acquisitionFailures.get() * acquireStep), outbound.pendingBytes());
+ executor.shutdown();
+ executor.awaitTermination(10, TimeUnit.SECONDS);
+
+ Assert.assertEquals(acquireCount * acquireStep - (acquisitionFailures.get() * acquireStep), outbound.pendingBytes());
+ Assert.assertEquals(acquireCount - acquisitionFailures.get(), outbound.pendingCount());
+ Assert.assertTrue(acquisitionFailures.get() <= maxFailures);
+ }
+ finally
+ { // release the acquired capacity from this round
+ outbound.unsafeReleaseCapacity(outbound.pendingCount(), outbound.pendingBytes());
Review comment:
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] dcapwell commented on a change in pull request #433:
Release capacity at the end of the test round
Posted by GitBox <gi...@apache.org>.
dcapwell commented on a change in pull request #433: Release capacity at the end of the test round
URL: https://github.com/apache/cassandra/pull/433#discussion_r384034992
##########
File path: test/unit/org/apache/cassandra/net/ConnectionTest.java
##########
@@ -766,40 +767,84 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
@Test
public void testAcquireReleaseOutbound() throws Throwable
{
+ // In each test round, K capacity is reserved upfront.
+ // Two groups of threads each release/acquire for K capacity in total accordingly,
+ // i.e. if only let the release threads to run, at the end, the reserved capacity is 0 (K - K).
Review comment:
should be "if only the release threads run"
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] dcapwell commented on a change in pull request #433:
Release capacity at the end of the test round
Posted by GitBox <gi...@apache.org>.
dcapwell commented on a change in pull request #433: Release capacity at the end of the test round
URL: https://github.com/apache/cassandra/pull/433#discussion_r384042628
##########
File path: test/unit/org/apache/cassandra/net/ConnectionTest.java
##########
@@ -766,40 +767,84 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
@Test
public void testAcquireReleaseOutbound() throws Throwable
{
+ // In each test round, K capacity is reserved upfront.
+ // Two groups of threads each release/acquire for K capacity in total accordingly,
+ // i.e. if only let the release threads to run, at the end, the reserved capacity is 0 (K - K).
+ // During the test, we expect N (N <= maxFailures) acquire attempts (for M capacity) to fail.
+ // The reserved capacity (pendingBytes) at the end of the round should equal to K - N * M,
+ // which you can find in the assertion.
test((inbound, outbound, endpoint) -> {
- ExecutorService executor = Executors.newFixedThreadPool(100);
- int acquireStep = 123;
- Assert.assertTrue(outbound.unsafeAcquireCapacity(100 * 10000, 100 * 10000 * acquireStep));
+ // max capacity equals to permit-free sendQueueCapcity + the minimun of endpoint and global reserve
+ double maxSendQueueCapacity = outbound.settings().applicationSendQueueCapacityInBytes +
+ Double.min(outbound.settings().applicationSendQueueReserveEndpointCapacityInBytes,
+ outbound.settings().applicationSendQueueReserveGlobalCapacityInBytes.limit());
+ int concurrency = 100;
+ int attempts = 10000;
+ int acquireCount = concurrency * attempts;
+ long acquireStep = Math.round(maxSendQueueCapacity * 1.2 / acquireCount / 2); // It is guranteed to acquire (~20%) more
+ // The total overly acquired amount divides the amount acquired in each step. Get the ceil value so not to miss the acquire that just exceeds.
+ long maxFailures = (long) Math.ceil((acquireCount * acquireStep * 2 - maxSendQueueCapacity) / acquireStep); // The result must be in the range of lone
AtomicLong acquisitionFailures = new AtomicLong();
- for (int i = 0; i < 100; i++)
- {
- executor.submit(() -> {
- for (int j = 0; j < 10000; j++)
- {
- if (!outbound.unsafeAcquireCapacity(acquireStep))
- acquisitionFailures.incrementAndGet();
- }
+ Runnable acquirer = () -> {
+ for (int j = 0; j < attempts; j++)
+ {
+ if (!outbound.unsafeAcquireCapacity(acquireStep))
+ acquisitionFailures.incrementAndGet();
+ }
+ };
+ Runnable releaser = () -> {
+ for (int j = 0; j < attempts; j++)
+ outbound.unsafeReleaseCapacity(acquireStep);
+ };
- });
- }
+ // Start N acquirer and releaser to contend for capcaity
+ List<Runnable> invokeOrder = new ArrayList<>();
+ for (int i = 0 ; i < concurrency ; ++i)
+ invokeOrder.add(acquirer);
+ for (int i = 0 ; i < concurrency ; ++i)
+ invokeOrder.add(releaser);
+ // randomize their start order
+ randomize(invokeOrder);
- for (int i = 0; i < 100; i++)
+ try
{
- executor.submit(() -> {
- for (int j = 0; j < 10000; j++)
- outbound.unsafeReleaseCapacity(acquireStep);
- });
- }
+ // Reserve enough capacity upfront to ensure the releaser threads cannot release all reserved capacity.
+ // i.e. the pendingBytes is always positive during the test.
+ Assert.assertTrue(outbound.unsafeAcquireCapacity(acquireCount, acquireCount * acquireStep));
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
- executor.shutdown();
- executor.awaitTermination(10, TimeUnit.SECONDS);
+ invokeOrder.forEach(executor::submit);
- // We can release more than we acquire, which certainly should not happen in
- // real life, but since it's a test just for acquisition and release, it is fine
- Assert.assertEquals(100 * 10000 * acquireStep - (acquisitionFailures.get() * acquireStep), outbound.pendingBytes());
+ executor.shutdown();
+ executor.awaitTermination(10, TimeUnit.SECONDS);
+
+ Assert.assertEquals(acquireCount * acquireStep - (acquisitionFailures.get() * acquireStep), outbound.pendingBytes());
+ Assert.assertEquals(acquireCount - acquisitionFailures.get(), outbound.pendingCount());
+ Assert.assertTrue(acquisitionFailures.get() <= maxFailures);
+ }
+ finally
+ { // release the acquired capacity from this round
+ outbound.unsafeReleaseCapacity(outbound.pendingCount(), outbound.pendingBytes());
Review comment:
given the diff and the JIRA comments, this looks to be the only fix needed to make this test not flaky? Not asking for a smaller diff, just clarification.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] dcapwell commented on a change in pull request #433:
Release capacity at the end of the test round
Posted by GitBox <gi...@apache.org>.
dcapwell commented on a change in pull request #433: Release capacity at the end of the test round
URL: https://github.com/apache/cassandra/pull/433#discussion_r384036074
##########
File path: test/unit/org/apache/cassandra/net/ConnectionTest.java
##########
@@ -766,40 +767,84 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
@Test
public void testAcquireReleaseOutbound() throws Throwable
{
+ // In each test round, K capacity is reserved upfront.
+ // Two groups of threads each release/acquire for K capacity in total accordingly,
+ // i.e. if only let the release threads to run, at the end, the reserved capacity is 0 (K - K).
+ // During the test, we expect N (N <= maxFailures) acquire attempts (for M capacity) to fail.
+ // The reserved capacity (pendingBytes) at the end of the round should equal to K - N * M,
+ // which you can find in the assertion.
test((inbound, outbound, endpoint) -> {
- ExecutorService executor = Executors.newFixedThreadPool(100);
- int acquireStep = 123;
- Assert.assertTrue(outbound.unsafeAcquireCapacity(100 * 10000, 100 * 10000 * acquireStep));
+ // max capacity equals to permit-free sendQueueCapcity + the minimun of endpoint and global reserve
+ double maxSendQueueCapacity = outbound.settings().applicationSendQueueCapacityInBytes +
+ Double.min(outbound.settings().applicationSendQueueReserveEndpointCapacityInBytes,
+ outbound.settings().applicationSendQueueReserveGlobalCapacityInBytes.limit());
+ int concurrency = 100;
+ int attempts = 10000;
+ int acquireCount = concurrency * attempts;
+ long acquireStep = Math.round(maxSendQueueCapacity * 1.2 / acquireCount / 2); // It is guranteed to acquire (~20%) more
+ // The total overly acquired amount divides the amount acquired in each step. Get the ceil value so not to miss the acquire that just exceeds.
+ long maxFailures = (long) Math.ceil((acquireCount * acquireStep * 2 - maxSendQueueCapacity) / acquireStep); // The result must be in the range of lone
AtomicLong acquisitionFailures = new AtomicLong();
- for (int i = 0; i < 100; i++)
- {
- executor.submit(() -> {
- for (int j = 0; j < 10000; j++)
- {
- if (!outbound.unsafeAcquireCapacity(acquireStep))
- acquisitionFailures.incrementAndGet();
- }
+ Runnable acquirer = () -> {
+ for (int j = 0; j < attempts; j++)
+ {
+ if (!outbound.unsafeAcquireCapacity(acquireStep))
+ acquisitionFailures.incrementAndGet();
+ }
+ };
+ Runnable releaser = () -> {
+ for (int j = 0; j < attempts; j++)
+ outbound.unsafeReleaseCapacity(acquireStep);
+ };
- });
- }
+ // Start N acquirer and releaser to contend for capcaity
+ List<Runnable> invokeOrder = new ArrayList<>();
+ for (int i = 0 ; i < concurrency ; ++i)
+ invokeOrder.add(acquirer);
+ for (int i = 0 ; i < concurrency ; ++i)
+ invokeOrder.add(releaser);
+ // randomize their start order
+ randomize(invokeOrder);
- for (int i = 0; i < 100; i++)
+ try
{
- executor.submit(() -> {
- for (int j = 0; j < 10000; j++)
- outbound.unsafeReleaseCapacity(acquireStep);
- });
- }
+ // Reserve enough capacity upfront to ensure the releaser threads cannot release all reserved capacity.
+ // i.e. the pendingBytes is always positive during the test.
+ Assert.assertTrue(outbound.unsafeAcquireCapacity(acquireCount, acquireCount * acquireStep));
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
- executor.shutdown();
- executor.awaitTermination(10, TimeUnit.SECONDS);
+ invokeOrder.forEach(executor::submit);
- // We can release more than we acquire, which certainly should not happen in
- // real life, but since it's a test just for acquisition and release, it is fine
- Assert.assertEquals(100 * 10000 * acquireStep - (acquisitionFailures.get() * acquireStep), outbound.pendingBytes());
+ executor.shutdown();
+ executor.awaitTermination(10, TimeUnit.SECONDS);
+
+ Assert.assertEquals(acquireCount * acquireStep - (acquisitionFailures.get() * acquireStep), outbound.pendingBytes());
+ Assert.assertEquals(acquireCount - acquisitionFailures.get(), outbound.pendingCount());
+ Assert.assertTrue(acquisitionFailures.get() <= maxFailures);
+ }
+ finally
+ { // release the acquired capacity from this round
+ outbound.unsafeReleaseCapacity(outbound.pendingCount(), outbound.pendingBytes());
+ }
});
}
+ private static <V> void randomize(List<V> list)
+ {
+ ThreadLocalRandom random = ThreadLocalRandom.current();
Review comment:
when this test is flaky we will have a hard time reproducing. If you use random, can you control the seed in the test? You should log the seed and have a way to update it while running the tests.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] dcapwell commented on a change in pull request #433:
Release capacity at the end of the test round
Posted by GitBox <gi...@apache.org>.
dcapwell commented on a change in pull request #433: Release capacity at the end of the test round
URL: https://github.com/apache/cassandra/pull/433#discussion_r384039563
##########
File path: test/unit/org/apache/cassandra/net/ConnectionTest.java
##########
@@ -766,40 +767,84 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
@Test
public void testAcquireReleaseOutbound() throws Throwable
{
+ // In each test round, K capacity is reserved upfront.
+ // Two groups of threads each release/acquire for K capacity in total accordingly,
+ // i.e. if only let the release threads to run, at the end, the reserved capacity is 0 (K - K).
+ // During the test, we expect N (N <= maxFailures) acquire attempts (for M capacity) to fail.
+ // The reserved capacity (pendingBytes) at the end of the round should equal to K - N * M,
+ // which you can find in the assertion.
test((inbound, outbound, endpoint) -> {
- ExecutorService executor = Executors.newFixedThreadPool(100);
- int acquireStep = 123;
- Assert.assertTrue(outbound.unsafeAcquireCapacity(100 * 10000, 100 * 10000 * acquireStep));
+ // max capacity equals to permit-free sendQueueCapcity + the minimun of endpoint and global reserve
+ double maxSendQueueCapacity = outbound.settings().applicationSendQueueCapacityInBytes +
+ Double.min(outbound.settings().applicationSendQueueReserveEndpointCapacityInBytes,
+ outbound.settings().applicationSendQueueReserveGlobalCapacityInBytes.limit());
+ int concurrency = 100;
+ int attempts = 10000;
+ int acquireCount = concurrency * attempts;
+ long acquireStep = Math.round(maxSendQueueCapacity * 1.2 / acquireCount / 2); // It is guranteed to acquire (~20%) more
+ // The total overly acquired amount divides the amount acquired in each step. Get the ceil value so not to miss the acquire that just exceeds.
+ long maxFailures = (long) Math.ceil((acquireCount * acquireStep * 2 - maxSendQueueCapacity) / acquireStep); // The result must be in the range of lone
AtomicLong acquisitionFailures = new AtomicLong();
- for (int i = 0; i < 100; i++)
- {
- executor.submit(() -> {
- for (int j = 0; j < 10000; j++)
- {
- if (!outbound.unsafeAcquireCapacity(acquireStep))
- acquisitionFailures.incrementAndGet();
- }
+ Runnable acquirer = () -> {
+ for (int j = 0; j < attempts; j++)
+ {
+ if (!outbound.unsafeAcquireCapacity(acquireStep))
+ acquisitionFailures.incrementAndGet();
+ }
+ };
+ Runnable releaser = () -> {
+ for (int j = 0; j < attempts; j++)
+ outbound.unsafeReleaseCapacity(acquireStep);
+ };
- });
- }
+ // Start N acquirer and releaser to contend for capcaity
+ List<Runnable> invokeOrder = new ArrayList<>();
+ for (int i = 0 ; i < concurrency ; ++i)
+ invokeOrder.add(acquirer);
+ for (int i = 0 ; i < concurrency ; ++i)
+ invokeOrder.add(releaser);
+ // randomize their start order
+ randomize(invokeOrder);
- for (int i = 0; i < 100; i++)
+ try
{
- executor.submit(() -> {
- for (int j = 0; j < 10000; j++)
- outbound.unsafeReleaseCapacity(acquireStep);
- });
- }
+ // Reserve enough capacity upfront to ensure the releaser threads cannot release all reserved capacity.
+ // i.e. the pendingBytes is always positive during the test.
+ Assert.assertTrue(outbound.unsafeAcquireCapacity(acquireCount, acquireCount * acquireStep));
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
- executor.shutdown();
- executor.awaitTermination(10, TimeUnit.SECONDS);
+ invokeOrder.forEach(executor::submit);
- // We can release more than we acquire, which certainly should not happen in
- // real life, but since it's a test just for acquisition and release, it is fine
- Assert.assertEquals(100 * 10000 * acquireStep - (acquisitionFailures.get() * acquireStep), outbound.pendingBytes());
+ executor.shutdown();
+ executor.awaitTermination(10, TimeUnit.SECONDS);
Review comment:
your currency is 100, but if you run the test with 1 core is 10seconds enough time? How stable is this on the LOWER circle ci configs?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] dcapwell commented on a change in pull request #433:
Release capacity at the end of the test round
Posted by GitBox <gi...@apache.org>.
dcapwell commented on a change in pull request #433: Release capacity at the end of the test round
URL: https://github.com/apache/cassandra/pull/433#discussion_r384035142
##########
File path: test/unit/org/apache/cassandra/net/ConnectionTest.java
##########
@@ -766,40 +767,84 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
@Test
public void testAcquireReleaseOutbound() throws Throwable
{
+ // In each test round, K capacity is reserved upfront.
+ // Two groups of threads each release/acquire for K capacity in total accordingly,
+ // i.e. if only let the release threads to run, at the end, the reserved capacity is 0 (K - K).
+ // During the test, we expect N (N <= maxFailures) acquire attempts (for M capacity) to fail.
+ // The reserved capacity (pendingBytes) at the end of the round should equal to K - N * M,
+ // which you can find in the assertion.
test((inbound, outbound, endpoint) -> {
- ExecutorService executor = Executors.newFixedThreadPool(100);
- int acquireStep = 123;
- Assert.assertTrue(outbound.unsafeAcquireCapacity(100 * 10000, 100 * 10000 * acquireStep));
+ // max capacity equals to permit-free sendQueueCapcity + the minimun of endpoint and global reserve
+ double maxSendQueueCapacity = outbound.settings().applicationSendQueueCapacityInBytes +
+ Double.min(outbound.settings().applicationSendQueueReserveEndpointCapacityInBytes,
+ outbound.settings().applicationSendQueueReserveGlobalCapacityInBytes.limit());
+ int concurrency = 100;
+ int attempts = 10000;
+ int acquireCount = concurrency * attempts;
+ long acquireStep = Math.round(maxSendQueueCapacity * 1.2 / acquireCount / 2); // It is guranteed to acquire (~20%) more
+ // The total overly acquired amount divides the amount acquired in each step. Get the ceil value so not to miss the acquire that just exceeds.
+ long maxFailures = (long) Math.ceil((acquireCount * acquireStep * 2 - maxSendQueueCapacity) / acquireStep); // The result must be in the range of lone
AtomicLong acquisitionFailures = new AtomicLong();
- for (int i = 0; i < 100; i++)
- {
- executor.submit(() -> {
- for (int j = 0; j < 10000; j++)
- {
- if (!outbound.unsafeAcquireCapacity(acquireStep))
- acquisitionFailures.incrementAndGet();
- }
+ Runnable acquirer = () -> {
+ for (int j = 0; j < attempts; j++)
+ {
+ if (!outbound.unsafeAcquireCapacity(acquireStep))
+ acquisitionFailures.incrementAndGet();
+ }
+ };
+ Runnable releaser = () -> {
+ for (int j = 0; j < attempts; j++)
+ outbound.unsafeReleaseCapacity(acquireStep);
+ };
- });
- }
+ // Start N acquirer and releaser to contend for capcaity
+ List<Runnable> invokeOrder = new ArrayList<>();
Review comment:
nit: since the size is known (concurrency * 2) should use that to fill the constructor
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] dcapwell commented on a change in pull request #433:
Release capacity at the end of the test round
Posted by GitBox <gi...@apache.org>.
dcapwell commented on a change in pull request #433: Release capacity at the end of the test round
URL: https://github.com/apache/cassandra/pull/433#discussion_r384040186
##########
File path: test/unit/org/apache/cassandra/net/ConnectionTest.java
##########
@@ -766,40 +767,84 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
@Test
public void testAcquireReleaseOutbound() throws Throwable
{
+ // In each test round, K capacity is reserved upfront.
+ // Two groups of threads each release/acquire for K capacity in total accordingly,
+ // i.e. if only let the release threads to run, at the end, the reserved capacity is 0 (K - K).
+ // During the test, we expect N (N <= maxFailures) acquire attempts (for M capacity) to fail.
+ // The reserved capacity (pendingBytes) at the end of the round should equal to K - N * M,
+ // which you can find in the assertion.
test((inbound, outbound, endpoint) -> {
- ExecutorService executor = Executors.newFixedThreadPool(100);
- int acquireStep = 123;
- Assert.assertTrue(outbound.unsafeAcquireCapacity(100 * 10000, 100 * 10000 * acquireStep));
+ // max capacity equals to permit-free sendQueueCapcity + the minimun of endpoint and global reserve
+ double maxSendQueueCapacity = outbound.settings().applicationSendQueueCapacityInBytes +
+ Double.min(outbound.settings().applicationSendQueueReserveEndpointCapacityInBytes,
+ outbound.settings().applicationSendQueueReserveGlobalCapacityInBytes.limit());
+ int concurrency = 100;
+ int attempts = 10000;
+ int acquireCount = concurrency * attempts;
+ long acquireStep = Math.round(maxSendQueueCapacity * 1.2 / acquireCount / 2); // It is guranteed to acquire (~20%) more
+ // The total overly acquired amount divides the amount acquired in each step. Get the ceil value so not to miss the acquire that just exceeds.
+ long maxFailures = (long) Math.ceil((acquireCount * acquireStep * 2 - maxSendQueueCapacity) / acquireStep); // The result must be in the range of lone
AtomicLong acquisitionFailures = new AtomicLong();
- for (int i = 0; i < 100; i++)
- {
- executor.submit(() -> {
- for (int j = 0; j < 10000; j++)
- {
- if (!outbound.unsafeAcquireCapacity(acquireStep))
- acquisitionFailures.incrementAndGet();
- }
+ Runnable acquirer = () -> {
+ for (int j = 0; j < attempts; j++)
+ {
+ if (!outbound.unsafeAcquireCapacity(acquireStep))
+ acquisitionFailures.incrementAndGet();
+ }
+ };
+ Runnable releaser = () -> {
+ for (int j = 0; j < attempts; j++)
+ outbound.unsafeReleaseCapacity(acquireStep);
+ };
- });
- }
+ // Start N acquirer and releaser to contend for capcaity
+ List<Runnable> invokeOrder = new ArrayList<>();
+ for (int i = 0 ; i < concurrency ; ++i)
+ invokeOrder.add(acquirer);
+ for (int i = 0 ; i < concurrency ; ++i)
+ invokeOrder.add(releaser);
+ // randomize their start order
+ randomize(invokeOrder);
- for (int i = 0; i < 100; i++)
+ try
{
- executor.submit(() -> {
- for (int j = 0; j < 10000; j++)
- outbound.unsafeReleaseCapacity(acquireStep);
- });
- }
+ // Reserve enough capacity upfront to ensure the releaser threads cannot release all reserved capacity.
+ // i.e. the pendingBytes is always positive during the test.
+ Assert.assertTrue(outbound.unsafeAcquireCapacity(acquireCount, acquireCount * acquireStep));
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
- executor.shutdown();
- executor.awaitTermination(10, TimeUnit.SECONDS);
+ invokeOrder.forEach(executor::submit);
- // We can release more than we acquire, which certainly should not happen in
- // real life, but since it's a test just for acquisition and release, it is fine
- Assert.assertEquals(100 * 10000 * acquireStep - (acquisitionFailures.get() * acquireStep), outbound.pendingBytes());
+ executor.shutdown();
+ executor.awaitTermination(10, TimeUnit.SECONDS);
+
+ Assert.assertEquals(acquireCount * acquireStep - (acquisitionFailures.get() * acquireStep), outbound.pendingBytes());
+ Assert.assertEquals(acquireCount - acquisitionFailures.get(), outbound.pendingCount());
+ Assert.assertTrue(acquisitionFailures.get() <= maxFailures);
Review comment:
nit: as stated above, a more meaningful message would be good.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] yifan-c commented on a change in pull request #433:
Release capacity at the end of the test round
Posted by GitBox <gi...@apache.org>.
yifan-c commented on a change in pull request #433: Release capacity at the end of the test round
URL: https://github.com/apache/cassandra/pull/433#discussion_r384065570
##########
File path: test/unit/org/apache/cassandra/net/ConnectionTest.java
##########
@@ -766,40 +767,84 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
@Test
public void testAcquireReleaseOutbound() throws Throwable
{
+ // In each test round, K capacity is reserved upfront.
+ // Two groups of threads each release/acquire for K capacity in total accordingly,
+ // i.e. if only let the release threads to run, at the end, the reserved capacity is 0 (K - K).
+ // During the test, we expect N (N <= maxFailures) acquire attempts (for M capacity) to fail.
+ // The reserved capacity (pendingBytes) at the end of the round should equal to K - N * M,
+ // which you can find in the assertion.
test((inbound, outbound, endpoint) -> {
- ExecutorService executor = Executors.newFixedThreadPool(100);
- int acquireStep = 123;
- Assert.assertTrue(outbound.unsafeAcquireCapacity(100 * 10000, 100 * 10000 * acquireStep));
+ // max capacity equals to permit-free sendQueueCapcity + the minimun of endpoint and global reserve
+ double maxSendQueueCapacity = outbound.settings().applicationSendQueueCapacityInBytes +
+ Double.min(outbound.settings().applicationSendQueueReserveEndpointCapacityInBytes,
+ outbound.settings().applicationSendQueueReserveGlobalCapacityInBytes.limit());
+ int concurrency = 100;
+ int attempts = 10000;
+ int acquireCount = concurrency * attempts;
+ long acquireStep = Math.round(maxSendQueueCapacity * 1.2 / acquireCount / 2); // It is guranteed to acquire (~20%) more
+ // The total overly acquired amount divides the amount acquired in each step. Get the ceil value so not to miss the acquire that just exceeds.
+ long maxFailures = (long) Math.ceil((acquireCount * acquireStep * 2 - maxSendQueueCapacity) / acquireStep); // The result must be in the range of lone
AtomicLong acquisitionFailures = new AtomicLong();
- for (int i = 0; i < 100; i++)
- {
- executor.submit(() -> {
- for (int j = 0; j < 10000; j++)
- {
- if (!outbound.unsafeAcquireCapacity(acquireStep))
- acquisitionFailures.incrementAndGet();
- }
+ Runnable acquirer = () -> {
+ for (int j = 0; j < attempts; j++)
+ {
+ if (!outbound.unsafeAcquireCapacity(acquireStep))
+ acquisitionFailures.incrementAndGet();
+ }
+ };
+ Runnable releaser = () -> {
+ for (int j = 0; j < attempts; j++)
+ outbound.unsafeReleaseCapacity(acquireStep);
+ };
- });
- }
+ // Start N acquirer and releaser to contend for capcaity
+ List<Runnable> invokeOrder = new ArrayList<>();
+ for (int i = 0 ; i < concurrency ; ++i)
+ invokeOrder.add(acquirer);
+ for (int i = 0 ; i < concurrency ; ++i)
+ invokeOrder.add(releaser);
+ // randomize their start order
+ randomize(invokeOrder);
- for (int i = 0; i < 100; i++)
+ try
{
- executor.submit(() -> {
- for (int j = 0; j < 10000; j++)
- outbound.unsafeReleaseCapacity(acquireStep);
- });
- }
+ // Reserve enough capacity upfront to ensure the releaser threads cannot release all reserved capacity.
+ // i.e. the pendingBytes is always positive during the test.
+ Assert.assertTrue(outbound.unsafeAcquireCapacity(acquireCount, acquireCount * acquireStep));
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
- executor.shutdown();
- executor.awaitTermination(10, TimeUnit.SECONDS);
+ invokeOrder.forEach(executor::submit);
- // We can release more than we acquire, which certainly should not happen in
- // real life, but since it's a test just for acquisition and release, it is fine
- Assert.assertEquals(100 * 10000 * acquireStep - (acquisitionFailures.get() * acquireStep), outbound.pendingBytes());
+ executor.shutdown();
+ executor.awaitTermination(10, TimeUnit.SECONDS);
Review comment:
So far so good. I have been using the LOWER config, and not seeing an error from the await time.
I do not expect 10 seconds to be not enough. It can be easily increased if this becomes a failure cause.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] dcapwell commented on a change in pull request #433:
Release capacity at the end of the test round
Posted by GitBox <gi...@apache.org>.
dcapwell commented on a change in pull request #433: Release capacity at the end of the test round
URL: https://github.com/apache/cassandra/pull/433#discussion_r384036792
##########
File path: test/unit/org/apache/cassandra/net/ConnectionTest.java
##########
@@ -766,40 +767,84 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
@Test
public void testAcquireReleaseOutbound() throws Throwable
{
+ // In each test round, K capacity is reserved upfront.
+ // Two groups of threads each release/acquire for K capacity in total accordingly,
+ // i.e. if only let the release threads to run, at the end, the reserved capacity is 0 (K - K).
+ // During the test, we expect N (N <= maxFailures) acquire attempts (for M capacity) to fail.
+ // The reserved capacity (pendingBytes) at the end of the round should equal to K - N * M,
+ // which you can find in the assertion.
test((inbound, outbound, endpoint) -> {
- ExecutorService executor = Executors.newFixedThreadPool(100);
- int acquireStep = 123;
- Assert.assertTrue(outbound.unsafeAcquireCapacity(100 * 10000, 100 * 10000 * acquireStep));
+ // max capacity equals to permit-free sendQueueCapcity + the minimun of endpoint and global reserve
+ double maxSendQueueCapacity = outbound.settings().applicationSendQueueCapacityInBytes +
+ Double.min(outbound.settings().applicationSendQueueReserveEndpointCapacityInBytes,
+ outbound.settings().applicationSendQueueReserveGlobalCapacityInBytes.limit());
+ int concurrency = 100;
+ int attempts = 10000;
+ int acquireCount = concurrency * attempts;
+ long acquireStep = Math.round(maxSendQueueCapacity * 1.2 / acquireCount / 2); // It is guranteed to acquire (~20%) more
+ // The total overly acquired amount divides the amount acquired in each step. Get the ceil value so not to miss the acquire that just exceeds.
+ long maxFailures = (long) Math.ceil((acquireCount * acquireStep * 2 - maxSendQueueCapacity) / acquireStep); // The result must be in the range of lone
AtomicLong acquisitionFailures = new AtomicLong();
- for (int i = 0; i < 100; i++)
- {
- executor.submit(() -> {
- for (int j = 0; j < 10000; j++)
- {
- if (!outbound.unsafeAcquireCapacity(acquireStep))
- acquisitionFailures.incrementAndGet();
- }
+ Runnable acquirer = () -> {
+ for (int j = 0; j < attempts; j++)
+ {
+ if (!outbound.unsafeAcquireCapacity(acquireStep))
+ acquisitionFailures.incrementAndGet();
+ }
+ };
+ Runnable releaser = () -> {
+ for (int j = 0; j < attempts; j++)
+ outbound.unsafeReleaseCapacity(acquireStep);
+ };
- });
- }
+ // Start N acquirer and releaser to contend for capcaity
+ List<Runnable> invokeOrder = new ArrayList<>();
+ for (int i = 0 ; i < concurrency ; ++i)
+ invokeOrder.add(acquirer);
+ for (int i = 0 ; i < concurrency ; ++i)
+ invokeOrder.add(releaser);
+ // randomize their start order
+ randomize(invokeOrder);
- for (int i = 0; i < 100; i++)
+ try
{
- executor.submit(() -> {
- for (int j = 0; j < 10000; j++)
- outbound.unsafeReleaseCapacity(acquireStep);
- });
- }
+ // Reserve enough capacity upfront to ensure the releaser threads cannot release all reserved capacity.
+ // i.e. the pendingBytes is always positive during the test.
+ Assert.assertTrue(outbound.unsafeAcquireCapacity(acquireCount, acquireCount * acquireStep));
Review comment:
nit: `assertTrue` produces bad error messages; it is mostly better to provide a meaningful message when using `assertTrue`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] dcapwell commented on a change in pull request #433:
Release capacity at the end of the test round
Posted by GitBox <gi...@apache.org>.
dcapwell commented on a change in pull request #433: Release capacity at the end of the test round
URL: https://github.com/apache/cassandra/pull/433#discussion_r384038216
##########
File path: test/unit/org/apache/cassandra/net/ConnectionTest.java
##########
@@ -766,40 +767,84 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
@Test
public void testAcquireReleaseOutbound() throws Throwable
{
+ // In each test round, K capacity is reserved upfront.
+ // Two groups of threads each release/acquire for K capacity in total accordingly,
+ // i.e. if only let the release threads to run, at the end, the reserved capacity is 0 (K - K).
+ // During the test, we expect N (N <= maxFailures) acquire attempts (for M capacity) to fail.
+ // The reserved capacity (pendingBytes) at the end of the round should equal to K - N * M,
+ // which you can find in the assertion.
test((inbound, outbound, endpoint) -> {
- ExecutorService executor = Executors.newFixedThreadPool(100);
- int acquireStep = 123;
- Assert.assertTrue(outbound.unsafeAcquireCapacity(100 * 10000, 100 * 10000 * acquireStep));
+ // max capacity equals to permit-free sendQueueCapcity + the minimun of endpoint and global reserve
+ double maxSendQueueCapacity = outbound.settings().applicationSendQueueCapacityInBytes +
+ Double.min(outbound.settings().applicationSendQueueReserveEndpointCapacityInBytes,
+ outbound.settings().applicationSendQueueReserveGlobalCapacityInBytes.limit());
+ int concurrency = 100;
+ int attempts = 10000;
+ int acquireCount = concurrency * attempts;
+ long acquireStep = Math.round(maxSendQueueCapacity * 1.2 / acquireCount / 2); // It is guranteed to acquire (~20%) more
+ // The total overly acquired amount divides the amount acquired in each step. Get the ceil value so not to miss the acquire that just exceeds.
+ long maxFailures = (long) Math.ceil((acquireCount * acquireStep * 2 - maxSendQueueCapacity) / acquireStep); // The result must be in the range of lone
AtomicLong acquisitionFailures = new AtomicLong();
- for (int i = 0; i < 100; i++)
- {
- executor.submit(() -> {
- for (int j = 0; j < 10000; j++)
- {
- if (!outbound.unsafeAcquireCapacity(acquireStep))
- acquisitionFailures.incrementAndGet();
- }
+ Runnable acquirer = () -> {
+ for (int j = 0; j < attempts; j++)
+ {
+ if (!outbound.unsafeAcquireCapacity(acquireStep))
+ acquisitionFailures.incrementAndGet();
+ }
+ };
+ Runnable releaser = () -> {
+ for (int j = 0; j < attempts; j++)
+ outbound.unsafeReleaseCapacity(acquireStep);
+ };
- });
- }
+ // Start N acquirer and releaser to contend for capcaity
+ List<Runnable> invokeOrder = new ArrayList<>();
+ for (int i = 0 ; i < concurrency ; ++i)
+ invokeOrder.add(acquirer);
+ for (int i = 0 ; i < concurrency ; ++i)
+ invokeOrder.add(releaser);
+ // randomize their start order
+ randomize(invokeOrder);
- for (int i = 0; i < 100; i++)
+ try
{
- executor.submit(() -> {
- for (int j = 0; j < 10000; j++)
- outbound.unsafeReleaseCapacity(acquireStep);
- });
- }
+ // Reserve enough capacity upfront to ensure the releaser threads cannot release all reserved capacity.
+ // i.e. the pendingBytes is always positive during the test.
+ Assert.assertTrue(outbound.unsafeAcquireCapacity(acquireCount, acquireCount * acquireStep));
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
- executor.shutdown();
- executor.awaitTermination(10, TimeUnit.SECONDS);
+ invokeOrder.forEach(executor::submit);
Review comment:
nit: you call it `invokeOrder` but you have `concurrency` threads. It's very possible that execution happens out of order.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org