You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "RobertIndie (via GitHub)" <gi...@apache.org> on 2023/07/18 08:49:54 UTC
[GitHub] [pulsar-dotpulsar] RobertIndie opened a new pull request, #167: Fix PingPongHandler doesn't kill the connection
RobertIndie opened a new pull request, #167:
URL: https://github.com/apache/pulsar-dotpulsar/pull/167
Fixes #166
# Description
The root cause of #166 is that the PingPongHandler doesn't work expectedly.
The expected behavior for PingPongHandler is as follows:
- Send a ping every keepAliveIntervarl(default is 30s)
- Waiting for the Pong responses or any other command.
- Trigger the Watch task at the next keepAliveIntervarl. If there are no incoming commands in this period. Then the connection is dead, and we should dispose the connection.
But the current behavior is Triigger the Watch every keepAliveIntervarl. If there are o incoming commands, then send the Ping. And there are any actions to the connection.
# Regression
It's not a regression.
# Testing
It seems hard to add a test for it. But I have tested it that it could fix #166 .
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-dotpulsar] RobertIndie commented on a diff in pull request #167: Fix PingPongHandler doesn't kill the connection
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on code in PR #167:
URL: https://github.com/apache/pulsar-dotpulsar/pull/167#discussion_r1295954536
##########
src/DotPulsar/Internal/Abstractions/IConnection.cs:
##########
@@ -42,4 +42,6 @@ public interface IConnection : IAsyncDisposable
Task Send(SendPackage command, TaskCompletionSource<BaseCommand> responseTcs, CancellationToken cancellationToken);
Task<BaseCommand> Send(CommandGetOrCreateSchema command, CancellationToken cancellationToken);
Task<BaseCommand> Send(CommandPartitionedTopicMetadata command, CancellationToken cancellationToken);
+
+ void MarkInactive();
Review Comment:
But this way, the PingPongHandler will be subjected to closeInactiveConnectionsInterval. This will make users confused.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-dotpulsar] blankensteiner merged pull request #167: Fix PingPongHandler doesn't kill the connection
Posted by "blankensteiner (via GitHub)" <gi...@apache.org>.
blankensteiner merged PR #167:
URL: https://github.com/apache/pulsar-dotpulsar/pull/167
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-dotpulsar] RobertIndie commented on a diff in pull request #167: Fix PingPongHandler doesn't kill the connection
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on code in PR #167:
URL: https://github.com/apache/pulsar-dotpulsar/pull/167#discussion_r1302739462
##########
src/DotPulsar/Internal/Connection.cs:
##########
@@ -336,6 +338,14 @@ public async ValueTask DisposeAsync()
await _stream.DisposeAsync().ConfigureAwait(false);
}
+ public async void MarkInactive()
Review Comment:
Make sense to me. I am trying to implement it.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-dotpulsar] blankensteiner commented on a diff in pull request #167: Fix PingPongHandler doesn't kill the connection
Posted by "blankensteiner (via GitHub)" <gi...@apache.org>.
blankensteiner commented on code in PR #167:
URL: https://github.com/apache/pulsar-dotpulsar/pull/167#discussion_r1280401766
##########
src/DotPulsar/Internal/PingPongHandler.cs:
##########
@@ -58,16 +55,13 @@ private void Watch(object? state)
{
try
{
- var lastCommand = Interlocked.Read(ref _lastCommand);
- var now = Stopwatch.GetTimestamp();
- var elapsed = TimeSpan.FromSeconds((now - lastCommand) / Stopwatch.Frequency);
- if (elapsed >= _keepAliveInterval)
+ if (_waitForPongResponse)
{
- Task.Factory.StartNew(() => SendPing());
- _timer.Change(_keepAliveInterval, TimeSpan.Zero);
+ _connection.DisposeAsync();
Review Comment:
It is the responsibility of the ConnectionPool to dispose connections. We can create a way for the PingPongHandler to tell the connection that it is inactive so that the ConnectionPool will detect it.
##########
src/DotPulsar/Internal/PingPongHandler.cs:
##########
@@ -24,26 +24,23 @@ namespace DotPulsar.Internal;
public sealed class PingPongHandler : IAsyncDisposable
{
private readonly IConnection _connection;
- private readonly TimeSpan _keepAliveInterval;
private readonly Timer _timer;
private readonly CommandPing _ping;
private readonly CommandPong _pong;
- private long _lastCommand;
+ private volatile bool _waitForPongResponse;
Review Comment:
Changing this from a timestamp to a bool will lengthen the reaction time.
Let's say we have a keepalive interval of 60 seconds and that we receive a message 1 second after the last check, but then the connection is lost.
In the old implementation, we wake up 59 seconds later and see that we have a timeout in 1 second, so we reschedule to wake up in 1 second and start acting.
With the new implementation, we wake up and see we got a message since the last check, so we wait 60 seconds again before we realize we need to act.
##########
src/DotPulsar/Internal/PingPongHandler.cs:
##########
@@ -24,26 +24,23 @@ namespace DotPulsar.Internal;
public sealed class PingPongHandler : IAsyncDisposable
{
private readonly IConnection _connection;
- private readonly TimeSpan _keepAliveInterval;
private readonly Timer _timer;
private readonly CommandPing _ping;
private readonly CommandPong _pong;
- private long _lastCommand;
+ private volatile bool _waitForPongResponse;
public PingPongHandler(IConnection connection, TimeSpan keepAliveInterval)
{
_connection = connection;
- _keepAliveInterval = keepAliveInterval;
_timer = new Timer(Watch);
- _timer.Change(_keepAliveInterval, TimeSpan.Zero);
+ _timer.Change(keepAliveInterval, keepAliveInterval);
Review Comment:
This can cause multiple simultaneous Watch invocations, which is why I set it explicitly and thereby also set it for the actual time we want/need to wait.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-dotpulsar] blankensteiner commented on a diff in pull request #167: Fix PingPongHandler doesn't kill the connection
Posted by "blankensteiner (via GitHub)" <gi...@apache.org>.
blankensteiner commented on code in PR #167:
URL: https://github.com/apache/pulsar-dotpulsar/pull/167#discussion_r1301505115
##########
src/DotPulsar/Internal/Connection.cs:
##########
@@ -336,6 +338,14 @@ public async ValueTask DisposeAsync()
await _stream.DisposeAsync().ConfigureAwait(false);
}
+ public async void MarkInactive()
Review Comment:
Point taken, but it is the ConnectionPool that creates the connections and should be the one disposing them. Splitting lifecycle management is an issue and especially if classes owned by the connection kill the connection. An alternative is for the connection to expose a Task that completes if a timeout occurs. The ConnectionPool can have those tasks in the wait loop when checking inactive connections. Make sense and up for that? Otherwise, I currently have time to implement this, if you are busy.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-dotpulsar] RobertIndie commented on a diff in pull request #167: Fix PingPongHandler doesn't kill the connection
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on code in PR #167:
URL: https://github.com/apache/pulsar-dotpulsar/pull/167#discussion_r1302851690
##########
src/DotPulsar/Internal/Connection.cs:
##########
@@ -336,6 +338,14 @@ public async ValueTask DisposeAsync()
await _stream.DisposeAsync().ConfigureAwait(false);
}
+ public async void MarkInactive()
Review Comment:
I'm not sure if I get your point correctly. I moved the connection disposing to the lifecycle loop in the ConnectionPool.
But this way, it's hard to add a test for the ConnnectionPool. I have tested it manually and it seems work fine.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-dotpulsar] RobertIndie commented on a diff in pull request #167: Fix PingPongHandler doesn't kill the connection
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on code in PR #167:
URL: https://github.com/apache/pulsar-dotpulsar/pull/167#discussion_r1288265087
##########
src/DotPulsar/Internal/PingPongHandler.cs:
##########
@@ -58,16 +55,13 @@ private void Watch(object? state)
{
try
{
- var lastCommand = Interlocked.Read(ref _lastCommand);
- var now = Stopwatch.GetTimestamp();
- var elapsed = TimeSpan.FromSeconds((now - lastCommand) / Stopwatch.Frequency);
- if (elapsed >= _keepAliveInterval)
+ if (_waitForPongResponse)
{
- Task.Factory.StartNew(() => SendPing());
- _timer.Change(_keepAliveInterval, TimeSpan.Zero);
+ _connection.DisposeAsync();
Review Comment:
I try to add a callback hook to the connection to tell the ConnectionPool to close it.
##########
src/DotPulsar/Internal/PingPongHandler.cs:
##########
@@ -58,16 +55,13 @@ private void Watch(object? state)
{
try
{
- var lastCommand = Interlocked.Read(ref _lastCommand);
- var now = Stopwatch.GetTimestamp();
- var elapsed = TimeSpan.FromSeconds((now - lastCommand) / Stopwatch.Frequency);
- if (elapsed >= _keepAliveInterval)
+ if (_waitForPongResponse)
{
- Task.Factory.StartNew(() => SendPing());
- _timer.Change(_keepAliveInterval, TimeSpan.Zero);
+ _connection.DisposeAsync();
Review Comment:
I try to add a callback hook to the connection to tell the ConnectionPool to close it. PTAL. Thanks.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-dotpulsar] blankensteiner commented on a diff in pull request #167: Fix PingPongHandler doesn't kill the connection
Posted by "blankensteiner (via GitHub)" <gi...@apache.org>.
blankensteiner commented on code in PR #167:
URL: https://github.com/apache/pulsar-dotpulsar/pull/167#discussion_r1293305237
##########
src/DotPulsar/Internal/Abstractions/IConnection.cs:
##########
@@ -42,4 +42,6 @@ public interface IConnection : IAsyncDisposable
Task Send(SendPackage command, TaskCompletionSource<BaseCommand> responseTcs, CancellationToken cancellationToken);
Task<BaseCommand> Send(CommandGetOrCreateSchema command, CancellationToken cancellationToken);
Task<BaseCommand> Send(CommandPartitionedTopicMetadata command, CancellationToken cancellationToken);
+
+ void MarkInactive();
Review Comment:
Currently, IConnection has a "HasChannels" method, we should just rename this to "IsActive". Then it is still the ConnectionPool's "CloseInactiveConnections" method that does the checking and disposing.
##########
src/DotPulsar/Internal/Connection.cs:
##########
@@ -336,6 +338,14 @@ public async ValueTask DisposeAsync()
await _stream.DisposeAsync().ConfigureAwait(false);
}
+ public async void MarkInactive()
Review Comment:
Instead of a callback, the connection already has a reference to the PingPongHandler and should use that. The "IsActive" method should just check if the PingPongHandler thinks the connection is still good (and if the connection has channels). The PingPongHandler could have a "HasTimedOut" boolean property or something like that.
##########
tests/DotPulsar.Tests/Internal/PingPongHandlerTest.cs:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed 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.
+ */
+
+namespace DotPulsar.Tests.Internal;
+
+using DotPulsar.Internal;
+using DotPulsar.Internal.Abstractions;
+using DotPulsar.Internal.PulsarApi;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+[Trait("Category", "Unit")]
+public class PingPongHandlerTest
+{
+ [Fact]
+ public async Task Watch_GivenConnectionNotAlive_ShouldDisposeConnection()
+ {
+ var countDown = new CountdownEvent(1);
+ var connection = new MockConnection();
+ var keepAliveInterval = TimeSpan.FromSeconds(1);
+ connection.PingCallback = () => countDown.Signal();
+ var pingPongHandler = new PingPongHandler(connection, keepAliveInterval);
+
+ // Wait for the first Ping
+ // The ping arrive time should be at (keepAliveInterval, 2*keepAliveInterval)
+ countDown.Wait(2 * keepAliveInterval);
+ pingPongHandler.Incoming(BaseCommand.Type.Pong);
+ Assert.False(connection.IsDispose);
+
+ // Wait for the second Ping, but we don't reply with Pong.
+ // The connection disposed time should be at (2*keepAliveInterval, 3*keepAliveInterval)
+ await Task.Delay(3 * keepAliveInterval);
+ Assert.True(connection.IsDispose);
Review Comment:
Should use FluentAssertions
##########
tests/DotPulsar.Tests/Internal/PingPongHandlerTest.cs:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed 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.
+ */
+
+namespace DotPulsar.Tests.Internal;
+
+using DotPulsar.Internal;
+using DotPulsar.Internal.Abstractions;
+using DotPulsar.Internal.PulsarApi;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+[Trait("Category", "Unit")]
+public class PingPongHandlerTest
+{
+ [Fact]
+ public async Task Watch_GivenConnectionNotAlive_ShouldDisposeConnection()
+ {
+ var countDown = new CountdownEvent(1);
+ var connection = new MockConnection();
+ var keepAliveInterval = TimeSpan.FromSeconds(1);
+ connection.PingCallback = () => countDown.Signal();
+ var pingPongHandler = new PingPongHandler(connection, keepAliveInterval);
+
+ // Wait for the first Ping
+ // The ping arrive time should be at (keepAliveInterval, 2*keepAliveInterval)
+ countDown.Wait(2 * keepAliveInterval);
+ pingPongHandler.Incoming(BaseCommand.Type.Pong);
+ Assert.False(connection.IsDispose);
+
+ // Wait for the second Ping, but we don't reply with Pong.
+ // The connection disposed time should be at (2*keepAliveInterval, 3*keepAliveInterval)
+ await Task.Delay(3 * keepAliveInterval);
+ Assert.True(connection.IsDispose);
+ }
+
+ private class MockConnection : IConnection
Review Comment:
Can't NSubstitute mock this?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-dotpulsar] RobertIndie commented on a diff in pull request #167: Fix PingPongHandler doesn't kill the connection
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on code in PR #167:
URL: https://github.com/apache/pulsar-dotpulsar/pull/167#discussion_r1295959711
##########
src/DotPulsar/Internal/Connection.cs:
##########
@@ -336,6 +338,14 @@ public async ValueTask DisposeAsync()
await _stream.DisposeAsync().ConfigureAwait(false);
}
+ public async void MarkInactive()
Review Comment:
Same as the above comment. The PingPongHandler will still be subjected to closeInactiveConnectionsInterval.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org