Skip to content

Commit 9ed0db6

Browse files
nshestakovGazizonoki
nshestakov
authored and
Gazizonoki
committed
Moved "We guarantee the sequence of messages in the SDK with Autoscaling support" commit from ydb repo
1 parent 34db45b commit 9ed0db6

File tree

7 files changed

+222
-36
lines changed

7 files changed

+222
-36
lines changed

‎examples/topic_reader/eventloop/main.cpp

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -101,6 +101,8 @@ int main(int argc, const char* argv[]) {
101101
startPartitionSessionEvent->Confirm();
102102
} else if (auto* stopPartitionSessionEvent = std::get_if<NYdb::NTopic::TReadSessionEvent::TStopPartitionSessionEvent>(&*event)) {
103103
stopPartitionSessionEvent->Confirm();
104+
} else if (auto* endPartitionSessionEvent = std::get_if<NYdb::NTopic::TReadSessionEvent::TEndPartitionSessionEvent>(&*event)) {
105+
endPartitionSessionEvent->Confirm();
104106
} else if (auto* closeSessionEvent = std::get_if<NYdb::NTopic::TSessionClosedEvent>(&*event)) {
105107
break;
106108
}

‎include/ydb-cpp-sdk/client/topic/read_events.h

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -297,6 +297,10 @@ struct TReadSessionEvent {
297297
return ChildPartitionIds;
298298
}
299299

300+
//! Confirm partition session destruction.
301+
//! Confirm has no effect if TPartitionSessionClosedEvent for same partition session with is received.
302+
void Confirm();
303+
300304
private:
301305
std::vector<ui32> AdjacentPartitionIds;
302306
std::vector<ui32> ChildPartitionIds;

‎include/ydb-cpp-sdk/client/topic/read_session.h

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -194,7 +194,7 @@ struct TReadSessionSettings: public TRequestSettings<TReadSessionSettings> {
194194
FLUENT_SETTING_DEFAULT(TDuration, ConnectTimeout, TDuration::Seconds(30));
195195

196196
//! AutoscalingSupport.
197-
FLUENT_SETTING_OPTIONAL(bool, AutoscalingSupport);
197+
FLUENT_SETTING_DEFAULT(bool, AutoscalingSupport, false);
198198

199199
//! Log.
200200
FLUENT_SETTING_OPTIONAL(TLog, Log);

‎src/client/topic/impl/event_handlers.cpp

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -76,7 +76,8 @@ class TGracefulReleasingSimpleDataHandlers : public TThrRefBase {
7676
}
7777
}
7878

79-
void OnEndPartitionStream(TReadSessionEvent::TEndPartitionSessionEvent&) {
79+
void OnEndPartitionStream(TReadSessionEvent::TEndPartitionSessionEvent& event) {
80+
event.Confirm();
8081
}
8182

8283
void OnPartitionStreamClosed(TReadSessionEvent::TPartitionSessionClosedEvent& event) {
@@ -138,7 +139,8 @@ TReadSessionSettings::TEventHandlers& TReadSessionSettings::TEventHandlers::Simp
138139
StopPartitionSessionHandler([](TReadSessionEvent::TStopPartitionSessionEvent& event) {
139140
event.Confirm();
140141
});
141-
EndPartitionSessionHandler([](TReadSessionEvent::TEndPartitionSessionEvent&) {
142+
EndPartitionSessionHandler([](TReadSessionEvent::TEndPartitionSessionEvent& event) {
143+
event.Confirm();
142144
});
143145
CommitOffsetAcknowledgementHandler([](TReadSessionEvent::TCommitOffsetAcknowledgementEvent&){});
144146
PartitionSessionClosedHandler([](TReadSessionEvent::TPartitionSessionClosedEvent&){});

‎src/client/topic/impl/read_session_event.cpp

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -349,6 +349,12 @@ TEndPartitionSessionEvent::TEndPartitionSessionEvent(TPartitionSession::TPtr par
349349
, ChildPartitionIds(std::move(childPartitionIds)) {
350350
}
351351

352+
void TEndPartitionSessionEvent::Confirm() {
353+
if (PartitionSession) {
354+
static_cast<TPartitionStreamImpl<false>*>(PartitionSession.Get())->ConfirmEnd(GetChildPartitionIds());
355+
}
356+
}
357+
352358
void JoinIds(TStringBuilder& ret, const std::vector<ui32> ids) {
353359
ret << "[";
354360
for (size_t i = 0; i < ids.size(); ++i) {

‎src/client/topic/impl/read_session_impl.h

Lines changed: 95 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -488,7 +488,9 @@ struct TRawPartitionStreamEvent {
488488
template <bool UseMigrationProtocol>
489489
class TRawPartitionStreamEventQueue {
490490
public:
491-
TRawPartitionStreamEventQueue() = default;
491+
TRawPartitionStreamEventQueue(TCallbackContextPtr<UseMigrationProtocol> cbContext)
492+
: CbContext(cbContext) {
493+
};
492494

493495
template <class... Ts>
494496
TRawPartitionStreamEvent<UseMigrationProtocol>& emplace_back(Ts&&... event)
@@ -549,6 +551,7 @@ class TRawPartitionStreamEventQueue {
549551
std::deque<TRawPartitionStreamEvent<UseMigrationProtocol>>& queue);
550552

551553
private:
554+
TCallbackContextPtr<UseMigrationProtocol> CbContext;
552555
std::deque<TRawPartitionStreamEvent<UseMigrationProtocol>> Ready;
553556
std::deque<TRawPartitionStreamEvent<UseMigrationProtocol>> NotReady;
554557
};
@@ -582,6 +585,7 @@ class TPartitionStreamImpl : public TAPartitionStream<UseMigrationProtocol> {
582585
, AssignId(assignId)
583586
, FirstNotReadOffset(readOffset)
584587
, CbContext(std::move(cbContext))
588+
, EventsQueue(CbContext)
585589
{
586590
TAPartitionStream<true>::PartitionStreamId = partitionStreamId;
587591
TAPartitionStream<true>::TopicPath = std::move(topicPath);
@@ -602,6 +606,7 @@ class TPartitionStreamImpl : public TAPartitionStream<UseMigrationProtocol> {
602606
, AssignId(static_cast<ui64>(assignId))
603607
, FirstNotReadOffset(static_cast<ui64>(readOffset))
604608
, CbContext(std::move(cbContext))
609+
, EventsQueue(CbContext)
605610
{
606611
TAPartitionStream<false>::PartitionSessionId = partitionStreamId;
607612
TAPartitionStream<false>::TopicPath = std::move(topicPath);
@@ -624,6 +629,7 @@ class TPartitionStreamImpl : public TAPartitionStream<UseMigrationProtocol> {
624629

625630
void ConfirmCreate(std::optional<ui64> readOffset, std::optional<ui64> commitOffset);
626631
void ConfirmDestroy();
632+
void ConfirmEnd(const std::vector<ui32>& childIds);
627633

628634
void StopReading() /*override*/;
629635
void ResumeReading() /*override*/;
@@ -763,6 +769,7 @@ class TPartitionStreamImpl : public TAPartitionStream<UseMigrationProtocol> {
763769
std::mutex Lock;
764770
};
765771

772+
766773
template <bool UseMigrationProtocol>
767774
class TReadSessionEventsQueue: public TBaseSessionEventsQueue<TAReadSessionSettings<UseMigrationProtocol>,
768775
typename TAReadSessionEvent<UseMigrationProtocol>::TEvent,
@@ -822,7 +829,8 @@ class TReadSessionEventsQueue: public TBaseSessionEventsQueue<TAReadSessionSetti
822829
}
823830

824831
bool TryApplyCallbackToEventImpl(typename TParent::TEvent& event,
825-
TDeferredActions<UseMigrationProtocol>& deferred);
832+
TDeferredActions<UseMigrationProtocol>& deferred,
833+
TCallbackContextPtr<UseMigrationProtocol>& cbContext);
826834
bool HasDataEventCallback() const;
827835
void ApplyCallbackToEventImpl(TADataReceivedEvent<UseMigrationProtocol>& event,
828836
TUserRetrievedEventsInfoAccumulator<UseMigrationProtocol>&& eventsInfo,
@@ -858,13 +866,19 @@ class TReadSessionEventsQueue: public TBaseSessionEventsQueue<TAReadSessionSetti
858866

859867
void ClearAllEvents();
860868

869+
void SetCallbackContext(TCallbackContextPtr<UseMigrationProtocol>& ctx) {
870+
CbContext = ctx;
871+
}
872+
861873
private:
862874
struct THandlersVisitor : public TParent::TBaseHandlersVisitor {
863875
THandlersVisitor(const TAReadSessionSettings<UseMigrationProtocol>& settings,
864876
typename TParent::TEvent& event,
865-
TDeferredActions<UseMigrationProtocol>& deferred)
877+
TDeferredActions<UseMigrationProtocol>& deferred,
878+
TCallbackContextPtr<UseMigrationProtocol>& cbContext)
866879
: TParent::TBaseHandlersVisitor(settings, event)
867-
, Deferred(deferred) {
880+
, Deferred(deferred)
881+
, CbContext(cbContext) {
868882
}
869883

870884
#define DECLARE_HANDLER(type, handler, answer) \
@@ -879,7 +893,7 @@ class TReadSessionEventsQueue: public TBaseSessionEventsQueue<TAReadSessionSetti
879893
} \
880894
/**/
881895

882-
#define DECLARE_TEMPLATE_HANDLER(type_true, type_false, handler_true, handler_false, answer) \
896+
#define DECLARE_TEMPLATE_HANDLER(type_true, type_false, handler_true, handler_false) \
883897
bool operator()(std::conditional_t<UseMigrationProtocol, type_true, type_false>&) { \
884898
if (this->template PushHandler<std::conditional_t<UseMigrationProtocol, type_true, type_false>>( \
885899
std::move(TParent::TBaseHandlersVisitor::Event), \
@@ -891,7 +905,7 @@ class TReadSessionEventsQueue: public TBaseSessionEventsQueue<TAReadSessionSetti
891905
} \
892906
}(), \
893907
this->Settings.EventHandlers_.CommonHandler_)) { \
894-
return answer; \
908+
return true; \
895909
} \
896910
return false; \
897911
} \
@@ -900,50 +914,81 @@ class TReadSessionEventsQueue: public TBaseSessionEventsQueue<TAReadSessionSetti
900914
DECLARE_TEMPLATE_HANDLER(typename TAReadSessionEvent<true>::TDataReceivedEvent,
901915
typename TAReadSessionEvent<false>::TDataReceivedEvent,
902916
DataReceivedHandler_,
903-
DataReceivedHandler_,
904-
true);
917+
DataReceivedHandler_);
905918
DECLARE_TEMPLATE_HANDLER(typename TAReadSessionEvent<true>::TCommitAcknowledgementEvent,
906919
typename TAReadSessionEvent<false>::TCommitOffsetAcknowledgementEvent,
907920
CommitAcknowledgementHandler_,
908-
CommitOffsetAcknowledgementHandler_,
909-
true);
921+
CommitOffsetAcknowledgementHandler_);
910922
DECLARE_TEMPLATE_HANDLER(typename TAReadSessionEvent<true>::TCreatePartitionStreamEvent,
911923
typename TAReadSessionEvent<false>::TStartPartitionSessionEvent,
912924
CreatePartitionStreamHandler_,
913-
StartPartitionSessionHandler_,
914-
true);
925+
StartPartitionSessionHandler_);
915926
DECLARE_TEMPLATE_HANDLER(typename TAReadSessionEvent<true>::TDestroyPartitionStreamEvent,
916927
typename TAReadSessionEvent<false>::TStopPartitionSessionEvent,
917928
DestroyPartitionStreamHandler_,
918-
StopPartitionSessionHandler_,
919-
true);
929+
StopPartitionSessionHandler_);
920930
DECLARE_TEMPLATE_HANDLER(typename TAReadSessionEvent<true>::TPartitionStreamStatusEvent,
921931
typename TAReadSessionEvent<false>::TPartitionSessionStatusEvent,
922932
PartitionStreamStatusHandler_,
923-
PartitionSessionStatusHandler_,
924-
true);
925-
DECLARE_TEMPLATE_HANDLER(typename TAReadSessionEvent<true>::TPartitionStreamClosedEvent,
926-
typename TAReadSessionEvent<false>::TPartitionSessionClosedEvent,
927-
PartitionStreamClosedHandler_,
928-
PartitionSessionClosedHandler_,
929-
true);
933+
PartitionSessionStatusHandler_);
930934
DECLARE_HANDLER(TASessionClosedEvent<UseMigrationProtocol>, SessionClosedHandler_, false); // Not applied
931935

932936
#undef DECLARE_HANDLER
933937
#undef DECLARE_TEMPLATE_HANDLER
934938

939+
bool operator()(std::conditional_t<UseMigrationProtocol, typename TAReadSessionEvent<true>::TPartitionStreamClosedEvent, typename TAReadSessionEvent<false>::TPartitionSessionClosedEvent>&) {
940+
auto specific = [this]() {
941+
if constexpr (UseMigrationProtocol) {
942+
return this->Settings.EventHandlers_.PartitionStreamClosedHandler_;
943+
} else {
944+
return this->Settings.EventHandlers_.PartitionSessionClosedHandler_;
945+
}
946+
}();
947+
948+
if (!specific && !this->Settings.EventHandlers_.CommonHandler_) {
949+
return false;
950+
}
951+
952+
this->template PushCommonHandler<>(
953+
std::move(TParent::TBaseHandlersVisitor::Event),
954+
[specific = specific,
955+
common = this->Settings.EventHandlers_.CommonHandler_,
956+
cbContext = CbContext](auto& event) {
957+
auto& e = std::get<std::conditional_t<UseMigrationProtocol, typename TAReadSessionEvent<true>::TPartitionStreamClosedEvent, typename TAReadSessionEvent<false>::TPartitionSessionClosedEvent>>(event);
958+
if (specific) {
959+
specific(e);
960+
} else if (common) {
961+
common(event);
962+
}
963+
if constexpr (!UseMigrationProtocol) {
964+
if (auto session = cbContext->LockShared()) {
965+
session->UnregisterPartition(e.GetPartitionSession()->GetPartitionId(), e.GetPartitionSession()->GetPartitionSessionId());
966+
}
967+
}
968+
});
969+
970+
return true;
971+
}
972+
935973
template<bool E = !UseMigrationProtocol>
936974
constexpr std::enable_if_t<E, bool>
937975
operator()(typename TAReadSessionEvent<false>::TEndPartitionSessionEvent&) {
938-
if (this->template PushHandler<typename TAReadSessionEvent<false>::TEndPartitionSessionEvent>(
939-
std::move(TParent::TBaseHandlersVisitor::Event),
940-
[this](){
941-
return this->Settings.EventHandlers_.EndPartitionSessionHandler_;
942-
}(),
943-
this->Settings.EventHandlers_.CommonHandler_)) {
976+
if (!this->Settings.EventHandlers_.EndPartitionSessionHandler_ && !this->Settings.EventHandlers_.CommonHandler_) {
944977
return false;
945978
}
946-
return false;
979+
this->template PushCommonHandler<>(
980+
std::move(TParent::TBaseHandlersVisitor::Event),
981+
[specific = this->Settings.EventHandlers_.EndPartitionSessionHandler_,
982+
common = this->Settings.EventHandlers_.CommonHandler_,
983+
cbContext = CbContext](TReadSessionEvent::TEvent& event) {
984+
auto& e = std::get<TReadSessionEvent::TEndPartitionSessionEvent>(event);
985+
if (specific) {
986+
specific(e);
987+
} else if (common) {
988+
common(event);
989+
}
990+
});
991+
return true;
947992
}
948993

949994
bool Visit() {
@@ -955,6 +1000,7 @@ class TReadSessionEventsQueue: public TBaseSessionEventsQueue<TAReadSessionSetti
9551000
}
9561001

9571002
TDeferredActions<UseMigrationProtocol>& Deferred;
1003+
TCallbackContextPtr<UseMigrationProtocol> CbContext;
9581004
};
9591005

9601006
TADataReceivedEvent<UseMigrationProtocol>
@@ -963,12 +1009,13 @@ class TReadSessionEventsQueue: public TBaseSessionEventsQueue<TAReadSessionSetti
9631009
TUserRetrievedEventsInfoAccumulator<UseMigrationProtocol>& accumulator); // Assumes that we're under lock.
9641010

9651011
bool ApplyHandler(TReadSessionEventInfo<UseMigrationProtocol>& eventInfo, TDeferredActions<UseMigrationProtocol>& deferred) {
966-
THandlersVisitor visitor(this->Settings, eventInfo.GetEvent(), deferred);
1012+
THandlersVisitor visitor(this->Settings, eventInfo.GetEvent(), deferred, CbContext);
9671013
return visitor.Visit();
9681014
}
9691015

9701016
private:
9711017
bool HasEventCallbacks;
1018+
TCallbackContextPtr<UseMigrationProtocol> CbContext;
9721019
};
9731020

9741021
} // namespace NYdb::NTopic
@@ -1044,6 +1091,7 @@ class TSingleClusterReadSessionImpl : public TEnableSelfContext<TSingleClusterRe
10441091
void Start();
10451092
void ConfirmPartitionStreamCreate(const TPartitionStreamImpl<UseMigrationProtocol>* partitionStream, std::optional<ui64> readOffset, std::optional<ui64> commitOffset);
10461093
void ConfirmPartitionStreamDestroy(TPartitionStreamImpl<UseMigrationProtocol>* partitionStream);
1094+
void ConfirmPartitionStreamEnd(TPartitionStreamImpl<UseMigrationProtocol>* partitionStream, const std::vector<ui32>& childIds);
10471095
void RequestPartitionStreamStatus(const TPartitionStreamImpl<UseMigrationProtocol>* partitionStream);
10481096
void Commit(const TPartitionStreamImpl<UseMigrationProtocol>* partitionStream, ui64 startOffset, ui64 endOffset);
10491097

@@ -1091,6 +1139,16 @@ class TSingleClusterReadSessionImpl : public TEnableSelfContext<TSingleClusterRe
10911139
return Log;
10921140
}
10931141

1142+
void RegisterParentPartition(ui32 partitionId, ui32 parentPartitionId, ui64 parentPartitionSessionId);
1143+
void UnregisterPartition(ui32 partitionId, ui64 partitionSessionId);
1144+
std::vector<ui64> GetParentPartitionSessions(ui32 partitionId, ui64 partitionSessionId);
1145+
bool AllParentSessionsHasBeenRead(ui32 partitionId, ui64 partitionSessionId);
1146+
1147+
void SetSelfContext(TPtr ptr) {
1148+
TEnableSelfContext<TSingleClusterReadSessionImpl<UseMigrationProtocol>>::SetSelfContext(std::move(ptr));
1149+
EventsQueue->SetCallbackContext(TEnableSelfContext<TSingleClusterReadSessionImpl<UseMigrationProtocol>>::SelfContext);
1150+
}
1151+
10941152
private:
10951153
void BreakConnectionAndReconnectImpl(TPlainStatus&& status, TDeferredActions<UseMigrationProtocol>& deferred);
10961154

@@ -1272,6 +1330,14 @@ class TSingleClusterReadSessionImpl : public TEnableSelfContext<TSingleClusterRe
12721330
std::atomic<int> DecompressionTasksInflight = 0;
12731331
i64 ReadSizeBudget;
12741332
i64 ReadSizeServerDelta = 0;
1333+
1334+
struct TParentInfo {
1335+
ui32 PartitionId;
1336+
ui64 PartitionSessionId;
1337+
};
1338+
1339+
std::unordered_map<ui32, std::vector<TParentInfo>> HierarchyData;
1340+
std::unordered_set<ui64> ReadingFinishedData;
12751341
};
12761342

12771343
} // namespace NYdb::NTopic

0 commit comments

Comments
 (0)