Skip to content

Commit 647807b

Browse files
vyazelenkovyazelenko
committed
[C++ Wrapper] Fix pending AsyncDestination memory leak when not polled to completion.
1 parent 4478c72 commit 647807b

4 files changed

Lines changed: 250 additions & 4 deletions

File tree

‎aeron-client/src/main/cpp_wrapper/ExclusivePublication.h‎

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,11 @@ class ExclusivePublication
7575
~ExclusivePublication()
7676
{
7777
aeron_exclusive_publication_close(m_publication, nullptr, nullptr);
78+
79+
for (const std::pair<const std::int64_t, AsyncDestination *>& e : m_pendingDestinations)
80+
{
81+
aeron_async_cmd_free(e.second);
82+
}
7883
}
7984

8085
inline void close()
@@ -765,7 +770,21 @@ class ExclusivePublication
765770
throw IllegalArgumentException("Unknown correlation id", SOURCEINFO);
766771
}
767772

768-
return findDestinationResponse(search->second);
773+
auto async = search->second;
774+
try
775+
{
776+
bool result = findDestinationResponse(async);
777+
if (result)
778+
{
779+
m_pendingDestinations.erase(correlationId);
780+
}
781+
return result;
782+
}
783+
catch (...)
784+
{
785+
m_pendingDestinations.erase(correlationId);
786+
throw;
787+
}
769788
}
770789

771790
/// @cond HIDDEN_SYMBOLS

‎aeron-client/src/main/cpp_wrapper/Publication.h‎

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -107,6 +107,11 @@ class Publication
107107
~Publication()
108108
{
109109
aeron_publication_close(m_publication, nullptr, nullptr);
110+
111+
for (const std::pair<const std::int64_t, AsyncDestination *>& e : m_pendingDestinations)
112+
{
113+
aeron_async_cmd_free(e.second);
114+
}
110115
}
111116

112117
inline void close()
@@ -776,7 +781,21 @@ class Publication
776781
throw IllegalArgumentException("Unknown correlation id", SOURCEINFO);
777782
}
778783

779-
return findDestinationResponse(search->second);
784+
auto async = search->second;
785+
try
786+
{
787+
bool result = findDestinationResponse(async);
788+
if (result)
789+
{
790+
m_pendingDestinations.erase(correlationId);
791+
}
792+
return result;
793+
}
794+
catch (...)
795+
{
796+
m_pendingDestinations.erase(correlationId);
797+
throw;
798+
}
780799
}
781800

782801
/// @cond HIDDEN_SYMBOLS

‎aeron-client/src/main/cpp_wrapper/Subscription.h‎

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -125,6 +125,11 @@ class Subscription
125125
delete m_addSubscription;
126126
}
127127
}
128+
129+
for (const std::pair<const std::int64_t, AsyncDestination *>& e : m_pendingDestinations)
130+
{
131+
aeron_async_cmd_free(e.second);
132+
}
128133
}
129134

130135
/**
@@ -392,7 +397,21 @@ class Subscription
392397
throw IllegalArgumentException("Unknown correlation id", SOURCEINFO);
393398
}
394399

395-
return findDestinationResponse(search->second);
400+
auto async = search->second;
401+
try
402+
{
403+
bool result = findDestinationResponse(async);
404+
if (result)
405+
{
406+
m_pendingDestinations.erase(correlationId);
407+
}
408+
return result;
409+
}
410+
catch (...)
411+
{
412+
m_pendingDestinations.erase(correlationId);
413+
throw;
414+
}
396415
}
397416

398417
/**

‎aeron-client/src/test/cpp_wrapper/WrapperSystemTest.cpp‎

Lines changed: 190 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -134,7 +134,7 @@ TEST_F(WrapperSystemTest, shouldRemovePendingAsyncPublicationUponError)
134134
AgentInvoker<ClientConductor> &invoker = aeron->conductorAgentInvoker();
135135
invoker.start();
136136

137-
auto channel = "aeron:udp?control=localhost:99999";
137+
auto channel = "aeron:udp?endpoint=localhost:99999";
138138
int stream_id = 1000;
139139
int64_t registration_id = aeron->addPublication(channel, stream_id);
140140

@@ -161,6 +161,65 @@ TEST_F(WrapperSystemTest, shouldRemovePendingAsyncPublicationUponError)
161161
}
162162
}
163163

164+
TEST_F(WrapperSystemTest, shouldRemovePendingAsyncPublicationUponSuccess)
165+
{
166+
Context ctx;
167+
ctx.useConductorAgentInvoker(true);
168+
169+
std::shared_ptr<Aeron> aeron = Aeron::connect(ctx);
170+
AgentInvoker<ClientConductor> &invoker = aeron->conductorAgentInvoker();
171+
invoker.start();
172+
auto channel = "aeron:udp?endpoint=localhost:5555";
173+
int stream_id = 1000;
174+
int64_t registration_id = aeron->addPublication(channel, stream_id);
175+
POLL_FOR_NON_NULL(publication, aeron->findPublication(registration_id), invoker);
176+
177+
try
178+
{
179+
POLL_FOR_NON_NULL(publication2, aeron->findPublication(registration_id), invoker);
180+
FAIL();
181+
}
182+
catch( const IllegalArgumentException& e )
183+
{
184+
auto errorMsg = std::string(e.what());
185+
EXPECT_NE(std::string::npos, errorMsg.find(std::string("Unknown registration id: ").append(std::to_string(registration_id)), 0));
186+
}
187+
}
188+
189+
TEST_F(WrapperSystemTest, shouldManuallyFreeAsyncPublicationIfNotPolled)
190+
{
191+
Context ctx;
192+
ctx.useConductorAgentInvoker(true);
193+
194+
std::shared_ptr<Aeron> aeron = Aeron::connect(ctx);
195+
AgentInvoker<ClientConductor> &invoker = aeron->conductorAgentInvoker();
196+
invoker.start();
197+
198+
auto channel = "aeron:udp?endpoint=localhost:5555";
199+
int stream_id = 1000;
200+
auto async = aeron->addPublicationAsync(channel, stream_id);
201+
aeron_async_cmd_free(async); // safe since client conductor is not running concurrently
202+
}
203+
204+
TEST_F(WrapperSystemTest, shouldManuallyFreeAsyncPublicationIfNotPolledConductor)
205+
{
206+
Context ctx;
207+
ctx.useConductorAgentInvoker(false);
208+
209+
std::shared_ptr<Aeron> aeron = Aeron::connect(ctx);
210+
211+
auto channel = "aeron:udp?endpoint=localhost:5555";
212+
int stream_id = 1000;
213+
auto async = aeron->addPublicationAsync(channel, stream_id);
214+
EXPECT_NE(nullptr, async);
215+
216+
// wait for addPublicationAsync to complete
217+
int64_t counterId = aeron->addCounter(1000, nullptr, 0, "test");
218+
WAIT_FOR_NON_NULL(counter, aeron->findCounter(counterId));
219+
220+
aeron_async_cmd_free(async); // it is safe to delete now since client conductor processed this command
221+
}
222+
164223
TEST_F(WrapperSystemTest, shouldRemovePendingAsyncExclusivePublicationUponError)
165224
{
166225
Context ctx;
@@ -299,6 +358,31 @@ TEST_F(WrapperSystemTest, asyncSubscriptionMustBeManuallyFreedAfterUsage)
299358
delete async;
300359
}
301360

361+
TEST_F(WrapperSystemTest, asyncSubscriptionMustBeManuallyFreedAfterUsageConductor)
362+
{
363+
Context ctx;
364+
ctx.useConductorAgentInvoker(false);
365+
366+
std::shared_ptr<Aeron> aeron = Aeron::connect(ctx);
367+
368+
auto channel = "aeron:udp?endpoint=localhost:99999";
369+
int stream_id = 1000;
370+
auto async = aeron->addSubscriptionAsync(channel, stream_id);
371+
372+
try
373+
{
374+
WAIT_FOR_NON_NULL(subscription, aeron->findSubscription(async));
375+
FAIL();
376+
}
377+
catch( const AeronException& e )
378+
{
379+
auto errorMsg = std::string(e.what());
380+
EXPECT_NE(std::string::npos, errorMsg.find("port out of range: 99999", 0));
381+
}
382+
383+
delete async;
384+
}
385+
302386
TEST_F(WrapperSystemTest, nonPolledAsyncSubscriptionMustBeManuallyFreedAfterUsage)
303387
{
304388
Context ctx;
@@ -437,3 +521,108 @@ TEST_F(WrapperSystemTest, polledSubscriptionShouldCloseAllAllocatedResourcesAfte
437521

438522
EXPECT_EQ(registration_id, subscription->registrationId());
439523
}
524+
525+
TEST_F(WrapperSystemTest, nonPolledPendingAsyncDestinationsAreAutomaticallyFreedSubscription)
526+
{
527+
Context ctx;
528+
ctx.useConductorAgentInvoker(false);
529+
530+
std::shared_ptr<Aeron> aeron = Aeron::connect(ctx);
531+
532+
auto channel = "aeron:udp?control-mode=manual";
533+
auto dest1Uri = "aeron:udp?endpoint=localhost:4554";
534+
auto dest2Uri = "aeron:udp?endpoint=localhost:7777";
535+
auto dest3Uri = "aeron:udp?endpoint=localhost:10000";
536+
int stream_id = 1000;
537+
538+
int64_t registration_id = aeron->addSubscription(channel, stream_id);
539+
WAIT_FOR_NON_NULL(subscription, aeron->findSubscription(registration_id));
540+
541+
int64_t addDest1RegistrationId = subscription->addDestination(dest1Uri);
542+
auto addDest3Async = subscription->addDestinationAsync(dest3Uri);
543+
int64_t addDest2RegistrationId = subscription->addDestination(dest2Uri);
544+
545+
auto removeDest1Async = subscription->removeDestinationAsync(dest1Uri);
546+
int64_t removeDest2RegistrationId = subscription->removeDestination(dest2Uri);
547+
int64_t removeDest3RegistrationId = subscription->removeDestination(dest3Uri);
548+
549+
EXPECT_GT(removeDest2RegistrationId, addDest1RegistrationId);
550+
EXPECT_GT(removeDest3RegistrationId, removeDest2RegistrationId);
551+
552+
WAIT_FOR(subscription->findDestinationResponse(addDest2RegistrationId));
553+
WAIT_FOR(subscription->findDestinationResponse(removeDest3RegistrationId));
554+
555+
// safe to delete after commands were processed by the client conductor thread
556+
aeron_async_cmd_free(addDest3Async);
557+
aeron_async_cmd_free(removeDest1Async);
558+
}
559+
560+
TEST_F(WrapperSystemTest, nonPolledPendingAsyncDestinationsAreAutomaticallyFreedPublication)
561+
{
562+
Context ctx;
563+
ctx.useConductorAgentInvoker(false);
564+
565+
std::shared_ptr<Aeron> aeron = Aeron::connect(ctx);
566+
567+
auto channel = "aeron:udp?control-mode=manual";
568+
auto dest1Uri = "aeron:udp?endpoint=localhost:4554";
569+
auto dest2Uri = "aeron:udp?endpoint=localhost:7777";
570+
auto dest3Uri = "aeron:udp?endpoint=localhost:10000";
571+
int stream_id = 1000;
572+
573+
int64_t registration_id = aeron->addPublication(channel, stream_id);
574+
WAIT_FOR_NON_NULL(publication, aeron->findPublication(registration_id));
575+
576+
int64_t addDest1RegistrationId = publication->addDestination(dest1Uri);
577+
auto addDest3Async = publication->addDestinationAsync(dest3Uri);
578+
int64_t addDest2RegistrationId = publication->addDestination(dest2Uri);
579+
580+
auto removeDest1Async = publication->removeDestinationAsync(dest1Uri);
581+
int64_t removeDest2RegistrationId = publication->removeDestination(dest2Uri);
582+
int64_t removeDest3RegistrationId = publication->removeDestination(dest3Uri);
583+
584+
EXPECT_GT(removeDest2RegistrationId, addDest1RegistrationId);
585+
EXPECT_GT(removeDest3RegistrationId, removeDest2RegistrationId);
586+
587+
WAIT_FOR(publication->findDestinationResponse(addDest2RegistrationId));
588+
WAIT_FOR(publication->findDestinationResponse(removeDest3RegistrationId));
589+
590+
// safe to delete after commands were processed by the client conductor thread
591+
aeron_async_cmd_free(addDest3Async);
592+
aeron_async_cmd_free(removeDest1Async);
593+
}
594+
595+
TEST_F(WrapperSystemTest, nonPolledPendingAsyncDestinationsAreAutomaticallyFreedExclusivePublication)
596+
{
597+
Context ctx;
598+
ctx.useConductorAgentInvoker(false);
599+
600+
std::shared_ptr<Aeron> aeron = Aeron::connect(ctx);
601+
602+
auto channel = "aeron:udp?control-mode=manual";
603+
auto dest1Uri = "aeron:udp?endpoint=localhost:4554";
604+
auto dest2Uri = "aeron:udp?endpoint=localhost:7777";
605+
auto dest3Uri = "aeron:udp?endpoint=localhost:10000";
606+
int stream_id = 1000;
607+
608+
int64_t registration_id = aeron->addExclusivePublication(channel, stream_id);
609+
WAIT_FOR_NON_NULL(publication, aeron->findExclusivePublication(registration_id));
610+
611+
int64_t addDest1RegistrationId = publication->addDestination(dest1Uri);
612+
auto addDest3Async = publication->addDestinationAsync(dest3Uri);
613+
int64_t addDest2RegistrationId = publication->addDestination(dest2Uri);
614+
615+
auto removeDest1Async = publication->removeDestinationAsync(dest1Uri);
616+
int64_t removeDest2RegistrationId = publication->removeDestination(dest2Uri);
617+
int64_t removeDest3RegistrationId = publication->removeDestination(dest3Uri);
618+
619+
EXPECT_GT(removeDest2RegistrationId, addDest1RegistrationId);
620+
EXPECT_GT(removeDest3RegistrationId, removeDest2RegistrationId);
621+
622+
WAIT_FOR(publication->findDestinationResponse(addDest2RegistrationId));
623+
WAIT_FOR(publication->findDestinationResponse(removeDest3RegistrationId));
624+
625+
// safe to delete after commands were processed by the client conductor thread
626+
aeron_async_cmd_free(addDest3Async);
627+
aeron_async_cmd_free(removeDest1Async);
628+
}

0 commit comments

Comments
 (0)