Feat[MQB, MWC]: report the number of tcp connections #384

waldgange · 2024-07-31T13:59:21Z

No description provided.

Signed-off-by: Anton Pryakhin <[email protected]>

678098 · 2024-08-07T13:23:07Z

src/groups/mqb/mqbstat/mqbstat_statcontroller.cpp

+                                                 &d_portsMutex,
+                                                 bdlf::PlaceHolders::_1)));
+            channelStatContext = portStatContext->addSubcontext(statConfig);
+            d_portsMap.emplace(portStatContext->name(), portStatContext);


Unfortunately, it doesn't compile on Solaris

[2024-08-07T12:57:33.062Z] "/SunOS/opt/bb/include/bslstl_pair.h", line 2528: Error: Could not find a match for BloombergLP::bslma::ManagedPtr<BloombergLP::mwcst::StatContext>::ManagedPtr<MANAGED_TYPE>(const BloombergLP::bslma::ManagedPtr<BloombergLP::mwcst::StatContext>) needed in void bsl::Pair_Second<BloombergLP::bslma::ManagedPtr<BloombergLP::mwcst::StatContext>>::Pair_Second<BloombergLP::bslma::ManagedPtr<BloombergLP::mwcst::StatContext>>(const BloombergLP::bslma::ManagedPtr<BloombergLP::mwcst::StatContext>&, BloombergLP::bslma::Allocator*, bsl::integral_constant<int, 0>). ... [2024-08-07T12:57:33.062Z] "/opt/bb/include/bslstl_unorderedmap_cpp03.h", line 3617: Where: Instantiated from bsl::pair<BloombergLP::bslstl::HashTableIterator<bsl::pair<const bsl::basic_string<char, std::char_traits<char>, bsl::allocator<char>>, BloombergLP::bslma::ManagedPtr<BloombergLP::mwcst::StatContext>>, long>, bool> bsl::unordered_map<bsl::basic_string<char, std::char_traits<char>, bsl::allocator<char>>, BloombergLP::bslma::ManagedPtr<BloombergLP::mwcst::StatContext>, bsl::hash<bsl::basic_string<char, std::char_traits<char>, bsl::allocator<char>>>, bsl::equal_to<bsl::basic_string<char, std::char_traits<char>, bsl::allocator<char>>>, bsl::allocator<bsl::pair<const bsl::basic_string<char, std::char_traits<char>, bsl::allocator<char>>, BloombergLP::bslma::ManagedPtr<BloombergLP::mwcst::StatContext>>>>::emplace<bsl::basic_string<char, std::char_traits<char>, bsl::allocator<char>>, BloombergLP::bslma::ManagedPtr<BloombergLP::mwcst::StatContext>>(const bsl::basic_string<char, std::char_traits<char>, bsl::allocator<char>>&, const BloombergLP::bslma::ManagedPtr<BloombergLP::mwcst::StatContext>&).

You might introduce parallel structures bsl::list<ManagedPtr<...> > and bsl::unordered_map<bsl::string, StatContext*> like it is done in this PR:
#389

Or you might find another solution

678098 · 2024-08-07T13:23:51Z

src/groups/mqb/mqbnet/mqbnet_tcpsessionfactory.cpp

@@ -100,6 +101,21 @@ const int k_SESSION_DESTROY_WAIT = 20;
 const int k_CLIENT_CLOSE_WAIT = 20;
 // Time to wait incrementally (in seconds) for all clients and
 // proxies to be destroyed during stop sequence.
+const char k_PORT_PATTERN[] = ":(\\d{1,5})";
+
+bsl::string portFromUri(const bsl::string& endpoint)


Suggested change

bsl::string portFromUri(const bsl::string& endpoint)

inline bsl::string portFromUri(const bsl::string& endpoint)

678098 · 2024-08-07T13:25:00Z

src/groups/mwc/mwcst/mwcst_statcontext.h

@@ -799,6 +799,10 @@ class StatContext {
    /// all sibling subcontexts.
    int uniqueId() const;

+    /// Return te number of times this 'StatContext' had 'snapshot' called on


Suggested change

/// Return te number of times this 'StatContext' had 'snapshot' called on

/// Return the number of times this 'StatContext' had 'snapshot' called on

678098 · 2024-08-07T13:35:59Z

src/groups/mqb/mqbnet/mqbnet_tcpsessionfactory.cpp

+    std::vector<bsl::string_view> matchVector;
+
+    BSLS_ASSERT_SAFE(
+        0 == regEx.prepare(&errorMessage, &errorOffset, k_PORT_PATTERN));


It's not necessary to prepare the same regEx on any opened channel.
We can initialize it once and reuse it from different threads since it's const thread-safe:
https://bloomberg.github.io/bde-resources/doxygen/bde_api_prod/group__bdlpcre__regex.html#3.5

678098 · 2024-08-07T13:39:27Z

src/groups/mqb/mqbnet/mqbnet_tcpsessionfactory.cpp

+
+    BSLS_ASSERT_SAFE(
+        0 == regEx.prepare(&errorMessage, &errorOffset, k_PORT_PATTERN));
+    BSLS_ASSERT_SAFE(0 == regEx.match(&matchVector, endpoint));


I think it's better to use match overload with one string_view* instead, not needed to construct an entire vector for this
https://bloomberg.github.io/bde-resources/doxygen/bde_api_prod/classbdlpcre_1_1RegEx.html#a6c7d41da3bddf9caa0aba7fd2677a68b

678098 · 2024-08-07T13:41:41Z

src/groups/mqb/mqbnet/mqbnet_tcpsessionfactory.cpp

+        0 == regEx.prepare(&errorMessage, &errorOffset, k_PORT_PATTERN));
+    BSLS_ASSERT_SAFE(0 == regEx.match(&matchVector, endpoint));
+
+    return bsl::string(matchVector[1]);


We can return a string_view as well, since it relies on the string passed in arg, so we don't construct a string with a default allocator

678098 · 2024-08-07T13:43:39Z

src/groups/mqb/mqbnet/mqbnet_tcpsessionfactory.cpp

-    BSLS_ASSERT_SAFE(parent);
-
-    bsl::string name;
+    bsl::string name, localPort;


Would rather use the provided allocator. If we do so, we can see in the stats table any small allocations within this component, and debug it if needed

Suggested change

bsl::string name, localPort;

bsl::string name(d_allocator_p), localPort(d_allocator_p);

678098 · 2024-08-07T13:54:43Z

src/groups/mqb/mqbstat/mqbstat_statcontroller.cpp

+    const mwcst::StatContext& context)
+{
+    // Lookup the port's StatContext and remove it from the 'map'
+    bslmt::LockGuard<bslmt::Mutex> guard(mutex);  // LOCK


With this change, you introduce a synchronization between different threads. We might assume it's okay, but if it's not, it's difficult to diagnose.

I think it's possible to avoid this, by storing ManagedPtr<StatContext> within a corresponding channel object. So if we close this channel and call stop or destructor (not sure which one), we free this ManagedPtr. StatContext class already has the mechanism to keep track of deleted subcontexts. So, if we call a destructor for ManagedPtr to subcontext, the parent context will know about this and remove raw pointer stored to it.

Unfortunately, the threadsafety of StatContext class is only partial.

blazingmq/src/groups/mwc/mwcst/mwcst_statcontext.h

Lines 96 to 98 in 7d4b3d7

// In simpler terms, only 'addSubContext', 'addSubTable', 'adjustValue',

// 'reportValue', and 'setValue' are thread-safe. All other functions should

// be considered not thread safe.

For example, I can't iterate over subcontexts that can be added/deleted without additional synchronisation anywhere except the snapshot() method or its synchronious pre- or post- steps, as handling the deleted subcontexts happens inside this method. Please find below some arguments in favor of the current imlementation:
Each channel stores its own StatContext, and this approach hasn't changed. What I have done is another 'layer' of StatContexts between channel contexts and their parent. This layer serves for breaking down the channels stats by ports. I really need to have only one StatContext for each port as I report aggregated stats. The inter-thread synchronisation is required here, because:

portsDeleter() is always called from StatController's thread, but TCPSessionFactory::channelStatContextCreator and, hence, constructors and destructors of channel-related StatContexts are called from TCPSessionFactory's threads.

Although StatContext::addSubcontext() is a thread-safe method, it will create a new StatContext instance even if there are sub-contexts with the same name.

So, in order to add a channel StatContext, I need to check if there is a port StatContext for that port and use it, otherways I have to create a new port Statcontext. After that I add a channel sub context to it. So I must have a ManagedPtr's to the StatContexts of the ports and to lock a container (currently, the unordered_map), where they are stored. Otherways race conditions are possible, either with two insertions of the same name port StatContexts, or parent deletion during the adding a new channel subcontext.

If you have an idea how to reach it without the synchronization, please let me know

If you have an idea how to reach it without the synchronization, please let me know

@waldgange let me think about this for some time

Hi @waldgange
I think the correct way to do it is to pass ManagedPtr to the StatChannel here, instead of shared_ptr:

blazingmq/src/groups/mwc/mwcio/mwcio_statchannelfactory.cpp

Lines 126 to 133 in a6a6cef

bsl::shared_ptr<mwcst::StatContext> statContext(

d_config.d_statContextCreator(channel, handleSp));

// Create the channel and notify user

bsl::shared_ptr<StatChannel> newChannel;

newChannel.createInplace(

handleSp->d_allocator_p,

StatChannelConfig(channel, statContext, handleSp->d_allocator_p),

handleSp->d_allocator_p);

So, StatChannel keeps ownership of this StatContext. When a StatChannel goes down, the ManagedPtr to its StatContext goes down too.

Note that it requires to change MWC interfaces. However, StatChannel and StatChannelFactory are used only within BlazingMQ, so I don't see any harm in this. Also, since we plan to own ManagedPtr in StatChannel, the StatChannel class is not copy constructible anymore, so we might explicitly delete copy constructor and assignment

Signed-off-by: Anton Pryakhin <[email protected]>

src/groups/mqb/mqbnet/mqbnet_tcpsessionfactory.cpp

678098 · 2024-08-13T13:30:15Z

src/groups/mqb/mqbnet/mqbnet_tcpsessionfactory.cpp

+{
+    bsl::string_view result;
+
+    BSLS_ASSERT_SAFE(0 == d_regex.match(&result, endpoint));


Is it possible to pass an incorrect endpoint here and cause the broker to crash on assert? I am thinking if we should crash here or handle incorrect endpoint

It never happens as an endpoint is always a host:port string. However, I can remove this assert and just return an empty string in case of not matching endpoint. Then broker will proceed working and all the channels with unknown ports will report their TCP connections number metric with an empty tag port

678098 · 2024-08-13T13:31:44Z

src/groups/mqb/mqbnet/mqbnet_tcpsessionfactory.cpp

+
+    BSLS_ASSERT_SAFE(0 == d_regex.match(&result, endpoint));
+
+    return result.substr(1);


Do we make substr(1) here because we capture the leading :?

Yes, definetley. If we used the match() overload that puts results to a vector, we would just take the [1] element of the vector, that contains sub expresion of the regex, and the [0] element would contain the full match with : in the beginning. However current match() overload only returns full matched expression.

678098 · 2024-08-13T13:35:30Z

src/groups/mwc/mwcio/mwcio_statchannel.cpp

 }

 StatChannel::~StatChannel()
 {
-    // NOTHING
+    d_config.d_statContext_sp->adjustValue(Stat::e_CONNECTIONS, -1);


If we do per-channel stat contexts, then the number of connections will always be 0 or 1, right?

Correct, I utilize the aggregation functionality Of StatContextTable here in order to deal only with the Channel's StatContext here. Otherways I would have to deal with its parent (port's StatContext), and, hence, store its shared_ptr here in StatChannel class

678098 · 2024-08-13T13:37:14Z

src/groups/mwc/mwcst/mwcst_statcontext.h

@@ -799,6 +799,10 @@ class StatContext {
    /// all sibling subcontexts.
    int uniqueId() const;

+    /// Return the number of times this 'StatContext' had 'snapshot' called on
+    /// it.
+    bsls::Types::Int64 numSnapshots() const;


This is useful, but also it's an interface change in MWC. Do we want to keep it?

I need to check it in portsDeleter() as before the first snapshot numSubcontexts() returns zero, and I don't want to delete new StatContexts in this case. It happens because StatContext::addSubcontext() puts the newly created contexts to d_newSubcontexts vector.

blazingmq/src/groups/mwc/mwcst/mwcst_statcontext.cpp

Line 631 in 7d4b3d7

d_newSubcontexts.push_back(newContext);

Then it moves them to d_statContext vector on next StatContext::snapshot()

blazingmq/src/groups/mwc/mwcst/mwcst_statcontext.cpp

Line 339 in 7d4b3d7

moveNewSubcontexts();

However StatContext::numSubcontexts() doesn't take into account d_newSubcontext and will return 0 if there haven't been any stapshots yet.

blazingmq/src/groups/mwc/mwcst/mwcst_statcontext.h

Lines 1136 to 1140 in 7d4b3d7

inline int StatContext::numSubcontexts() const

{

return static_cast<int>(d_subcontexts.size() +

d_deletedSubcontexts.size());

}

So if we don't check the number of snapshots in portsDeleter() we can unintentionally delete the StatContext that has just been added.

Signed-off-by: Anton Pryakhin <[email protected]>

bmq-oss-ci

Build 167 of commit 3d29a39 has completed with FAILURE

bmq-oss-ci

Build 167 of commit 3d29a39 has completed with FAILURE

Signed-off-by: Anton Pryakhin <[email protected]>

bmq-oss-ci

Build 232 of commit 3555a56 has completed with FAILURE

bmq-oss-ci

Build 236 of commit 4cf0b83 has completed with FAILURE

Signed-off-by: Anton Pryakhin <[email protected]>

bmq-oss-ci

Build 239 of commit 0565d51 has completed with FAILURE

678098

Some thoughts

src/groups/mqb/mqbnet/mqbnet_tcpsessionfactory.cpp

src/groups/mqb/mqbnet/mqbnet_tcpsessionfactory.h

678098 · 2024-09-05T10:59:45Z

src/groups/mqb/mqbnet/mqbnet_tcpsessionfactory.cpp

+        mwcst::StatContextConfiguration     portConfig(port, &localAllocator);
+        bsl::shared_ptr<mwcst::StatContext> portStatContext =
+            parent->addSubcontext(
+                portConfig.storeExpiredSubcontextValues(true));


Here, we convert bslma::ManagedPtr to bsl::shared_ptr. How does the ownership transfer works in this situiation? Is this operation safe? Are we sure that the underlying StatContext is not accidentally destroyed?

Seems like this is a common practice:

blazingmq/src/groups/mwc/mwcio/mwcio_statchannelfactory.cpp

Lines 126 to 127 in 5aaf596

bsl::shared_ptr<mwcst::StatContext> statContext(

d_config.d_statContextCreator(channel, handleSp));

blazingmq/src/groups/mqb/mqbstat/mqbstat_clusterstats.cpp

Lines 254 to 258 in 5aaf596

d_partitionsStatContexts.emplace_back(

bsl::shared_ptr<mwcst::StatContext>(

d_statContext_mp->addSubcontext(

mwcst::StatContextConfiguration(partitionName,

&localAllocator))));

blazingmq/src/groups/mqb/mqbblp/mqbblp_cluster.cpp

Lines 2582 to 2584 in 5aaf596

StatContextMp statContextMp =

d_clusterData.clusterNodesStatContext()->addSubcontext(config);

StatContextSp statContextSp(statContextMp, d_allocator_p);

and so on

src/groups/mqb/mqbnet/mqbnet_tcpsessionfactory.h

Signed-off-by: Anton Pryakhin <[email protected]>

bmq-oss-ci

Build 254 of commit 2e5e9e8 has completed with FAILURE

678098

LGTM!

Signed-off-by: Anton Pryakhin <[email protected]>

678098 · 2024-09-18T14:29:06Z

src/groups/mqb/mqbnet/mqbnet_tcpsessionfactory.cpp

+}
+
+void TCPSessionFactory::PortManager::onDeleteChannelContext(
+    const bsl::uint16_t port)


For basic numeric types we don't typically specify const in arguments list

678098 · 2024-09-18T14:29:46Z

src/groups/mwc/mwcio/mwcio_ntcchannel.cpp

+        return d_streamSocket_sp->sourceEndpoint();
+    }
+    else {
+        return ntsa::Endpoint();


Might replace with ternary operator

Signed-off-by: Anton Pryakhin <[email protected]>

bmq-oss-ci

Build 259 of commit 7f29d3a has completed with FAILURE

waldgange force-pushed the tcpconnections branch from ee0e30a to bdd9c6d Compare July 31, 2024 14:01

chrisbeard requested a review from hallfox July 31, 2024 14:05

waldgange force-pushed the tcpconnections branch 4 times, most recently from d785037 to 2282eb1 Compare August 7, 2024 12:02

waldgange marked this pull request as ready for review August 7, 2024 12:28

waldgange requested a review from a team as a code owner August 7, 2024 12:28

Report the number of tcp connections

2ceb0f5

Signed-off-by: Anton Pryakhin <[email protected]>

waldgange force-pushed the tcpconnections branch from 2282eb1 to 2ceb0f5 Compare August 7, 2024 13:10

678098 requested changes Aug 7, 2024

View reviewed changes

waldgange added 2 commits August 9, 2024 22:20

Address review

1a98e23

Signed-off-by: Anton Pryakhin <[email protected]>

Move mp

29506cc

Signed-off-by: Anton Pryakhin <[email protected]>

678098 changed the title ~~Report the number of tcp connections~~ Feat[MQB, MWC]: report the number of tcp connections Aug 13, 2024

678098 reviewed Aug 13, 2024

View reviewed changes

waldgange added 2 commits August 14, 2024 15:25

Address review pt2

d2171c8

Signed-off-by: Anton Pryakhin <[email protected]>

Remove logs

3d29a39

Signed-off-by: Anton Pryakhin <[email protected]>

bmq-oss-ci bot reviewed Aug 26, 2024

View reviewed changes

Address review pt3

3555a56

Signed-off-by: Anton Pryakhin <[email protected]>

bmq-oss-ci bot reviewed Sep 4, 2024

View reviewed changes

Merge branch 'main' into tcpconnections

4cf0b83

bmq-oss-ci bot reviewed Sep 4, 2024

View reviewed changes

cleanup

0565d51

Signed-off-by: Anton Pryakhin <[email protected]>

bmq-oss-ci bot reviewed Sep 5, 2024

View reviewed changes

678098 reviewed Sep 5, 2024

View reviewed changes

Address review pt4

2e5e9e8

Signed-off-by: Anton Pryakhin <[email protected]>

bmq-oss-ci bot reviewed Sep 16, 2024

View reviewed changes

678098 previously approved these changes Sep 16, 2024

View reviewed changes

Get rid of regex-based port extraction

eff0b97

Signed-off-by: Anton Pryakhin <[email protected]>

waldgange dismissed 678098’s stale review via eff0b97 September 18, 2024 11:41

Merge branch 'main' into tcpconnections

3c3025e

678098 previously approved these changes Sep 18, 2024

View reviewed changes

Fix plugin

7f29d3a

Signed-off-by: Anton Pryakhin <[email protected]>

waldgange dismissed 678098’s stale review via 7f29d3a September 18, 2024 15:08

678098 approved these changes Sep 18, 2024

View reviewed changes

bmq-oss-ci bot reviewed Sep 18, 2024

View reviewed changes

waldgange merged commit 6044740 into bloomberg:main Sep 19, 2024
34 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Feat[MQB, MWC]: report the number of tcp connections #384

Feat[MQB, MWC]: report the number of tcp connections #384

waldgange commented Jul 31, 2024

678098 Aug 7, 2024

678098 Aug 7, 2024

678098 Aug 7, 2024

678098 Aug 7, 2024

678098 Aug 7, 2024

678098 Aug 7, 2024

678098 Aug 7, 2024

678098 Aug 7, 2024

waldgange Aug 9, 2024 •

edited

Loading

678098 Aug 9, 2024 •

edited

Loading

678098 Aug 13, 2024

678098 Aug 13, 2024

waldgange Aug 14, 2024 •

edited

Loading

678098 Aug 13, 2024

waldgange Aug 14, 2024

678098 Aug 13, 2024

waldgange Aug 14, 2024

678098 Aug 13, 2024

waldgange Aug 14, 2024

bmq-oss-ci bot left a comment

bmq-oss-ci bot left a comment

bmq-oss-ci bot left a comment

bmq-oss-ci bot left a comment

bmq-oss-ci bot left a comment

678098 left a comment

678098 Sep 5, 2024

waldgange Sep 12, 2024

bmq-oss-ci bot left a comment

678098 left a comment

678098 Sep 18, 2024

678098 Sep 18, 2024

bmq-oss-ci bot left a comment

	bsl::string portFromUri(const bsl::string& endpoint)
	inline bsl::string portFromUri(const bsl::string& endpoint)

	/// Return te number of times this 'StatContext' had 'snapshot' called on
	/// Return the number of times this 'StatContext' had 'snapshot' called on

	bsl::string name, localPort;
	bsl::string name(d_allocator_p), localPort(d_allocator_p);

	// In simpler terms, only 'addSubContext', 'addSubTable', 'adjustValue',
	// 'reportValue', and 'setValue' are thread-safe. All other functions should
	// be considered not thread safe.

	bsl::shared_ptr<mwcst::StatContext> statContext(
	d_config.d_statContextCreator(channel, handleSp));
	// Create the channel and notify user
	bsl::shared_ptr<StatChannel> newChannel;
	newChannel.createInplace(
	handleSp->d_allocator_p,
	StatChannelConfig(channel, statContext, handleSp->d_allocator_p),
	handleSp->d_allocator_p);


		BSLS_ASSERT_SAFE(0 == d_regex.match(&result, endpoint));

		return result.substr(1);

	inline int StatContext::numSubcontexts() const
	{
	return static_cast<int>(d_subcontexts.size() +
	d_deletedSubcontexts.size());
	}

	d_partitionsStatContexts.emplace_back(
	bsl::shared_ptr<mwcst::StatContext>(
	d_statContext_mp->addSubcontext(
	mwcst::StatContextConfiguration(partitionName,
	&localAllocator))));

	StatContextMp statContextMp =
	d_clusterData.clusterNodesStatContext()->addSubcontext(config);
	StatContextSp statContextSp(statContextMp, d_allocator_p);

Feat[MQB, MWC]: report the number of tcp connections #384

Feat[MQB, MWC]: report the number of tcp connections #384

Conversation

waldgange commented Jul 31, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

waldgange Aug 9, 2024 • edited Loading

Choose a reason for hiding this comment

678098 Aug 9, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

waldgange Aug 14, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bmq-oss-ci bot left a comment

Choose a reason for hiding this comment

bmq-oss-ci bot left a comment

Choose a reason for hiding this comment

bmq-oss-ci bot left a comment

Choose a reason for hiding this comment

bmq-oss-ci bot left a comment

Choose a reason for hiding this comment

bmq-oss-ci bot left a comment

Choose a reason for hiding this comment

678098 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bmq-oss-ci bot left a comment

Choose a reason for hiding this comment

678098 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bmq-oss-ci bot left a comment

Choose a reason for hiding this comment

waldgange Aug 9, 2024 •

edited

Loading

678098 Aug 9, 2024 •

edited

Loading

waldgange Aug 14, 2024 •

edited

Loading