Integration tests

CMakeLists.txt

@@ -47,7 +47,8 @@ endif()
 
 if (YDB_SDK_TESTS)
   enable_testing()
-  add_subdirectory(tests)
+  add_subdirectory(tests/ut)
+  add_subdirectory(tests/integration)
 endif()
 
 if (YDB_SDK_INSTALL)

README.md

@@ -122,6 +122,20 @@ cmake --build --preset release
 
 Specify a level of parallelism by passing the `-j<level>` option into the command below (e.g. `-j$(nproc)`)
 
+Running all tests:
+
 ```bash
 ctest -j$(nproc) --preset release
 ```
+
+Running unit tests only:
+
+```bash
+ctest -j$(nproc) --preset release -R .*_ut
+```
+
+Running integration tests only:
+
+```bash
+ctest -j$(nproc) --preset release -R .*_it
+```

include/ydb-cpp-sdk/library/monlib/counters/counters.h

@@ -32,7 +32,7 @@ namespace NMonitoring {
     // https://wiki.yandex-team.ru/solomon/libs/monlib_cpp/
     class TDeprecatedCounter {
     public:
-        using TValue = TAtomic;
+        using TValue = std::atomic<int>;
         using TValueBase = TAtomicBase;
 
         TDeprecatedCounter()
@@ -52,28 +52,28 @@ namespace NMonitoring {
         }
 
         operator TValueBase() const {
-            return AtomicGet(Value);
+            return Value.load();
         }
         TValueBase Val() const {
-            return AtomicGet(Value);
+            return Value.load();
         }
 
         void Set(TValueBase val) {
-            AtomicSet(Value, val);
+            Value.store(val);
         }
 
         TValueBase Inc() {
-            return AtomicIncrement(Value);
+            return ++Value;
         }
         TValueBase Dec() {
-            return AtomicDecrement(Value);
+            return --Value;
         }
 
         TValueBase Add(const TValueBase val) {
-            return AtomicAdd(Value, val);
+            return Value.fetch_add(val);
         }
         TValueBase Sub(const TValueBase val) {
-            return AtomicAdd(Value, -val);
+            return Value.fetch_sub(val);
         }
 
         // operator overloads convinient
@@ -99,20 +99,19 @@ namespace NMonitoring {
         }
 
         TValueBase operator=(TValueBase rhs) {
-            AtomicSwap(&Value, rhs);
-            return rhs;
+            return Value.exchange(rhs);
         }
 
         bool operator!() const {
-            return AtomicGet(Value) == 0;
+            return Value.load() == 0;
         }
 
-        TAtomic& GetAtomic() {
+        std::atomic<int>& GetAtomic() {
             return Value;
         }
 
     private:
-        TAtomic Value;
+        std::atomic<int> Value;
         bool Derivative;
     };
 

include/ydb-cpp-sdk/library/monlib/dynamic_counters/counters.h

@@ -228,7 +228,7 @@ namespace NMonitoring {
 
         /// XXX: hack for deferred removal of expired counters. Remove once Output* functions are not used for serialization
         mutable TCounters Counters;
-        mutable TAtomic ExpiringCount = 0;
+        mutable std::atomic<int> ExpiringCount = 0;
 
     public:
         TDynamicCounters(TCountableBase::EVisibility visibility = TCountableBase::EVisibility::Public);

src/client/federated_topic/ut/basic_usage_ut.cpp

@@ -461,7 +461,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
         auto totalReceived = 0u;
 
         auto f = checkedPromise.GetFuture();
-        TAtomic check = 1;
+        std::atomic<int> check = 1;
 
         // Create read session.
         NYdb::NFederatedTopic::TFederatedReadSessionSettings readSettings;
@@ -472,7 +472,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
 
         readSettings.FederatedEventHandlers_.SimpleDataHandlers([&](TReadSessionEvent::TDataReceivedEvent& ev) mutable {
             std::cerr << ">>> event from dataHandler: " << DebugString(ev) << std::endl;
-            Y_VERIFY_S(AtomicGet(check) != 0, "check is false");
+            Y_VERIFY_S(check.load() != 0, "check is false");
             auto& messages = ev.GetMessages();
             for (size_t i = 0u; i < messages.size(); ++i) {
                 auto& message = messages[i];
@@ -517,7 +517,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
 
         f.GetValueSync();
         ReadSession->Close();
-        AtomicSet(check, 0);
+        check.store(0);
     }
 
     Y_UNIT_TEST(BasicWriteSession) {

src/client/persqueue_public/impl/write_session_impl.cpp

@@ -87,7 +87,7 @@ TWriteSessionImpl::THandleResult TWriteSessionImpl::RestartImpl(const TPlainStat
     Y_ABORT_UNLESS(Lock.IsLocked());
 
     THandleResult result;
-    if (AtomicGet(Aborting)) {
+    if (Aborting.load()) {
         LOG_LAZY(DbDriverState->Log, TLOG_DEBUG, LogPrefix() << "Write session is aborting and will not restart");
         return result;
     }
@@ -128,7 +128,7 @@ void TWriteSessionImpl::DoCdsRequest(TDuration delay) {
     bool cdsRequestIsUnnecessary;
     {
         std::lock_guard guard(Lock);
-        if (AtomicGet(Aborting)) {
+        if (Aborting.load()) {
             return;
         }
         LOG_LAZY(DbDriverState->Log, TLOG_INFO, LogPrefix() << "Write session: Do CDS request");
@@ -1231,7 +1231,7 @@ void TWriteSessionImpl::SendImpl() {
 
 // Client method, no Lock
 bool TWriteSessionImpl::Close(TDuration closeTimeout) {
-    if (AtomicGet(Aborting))
+    if (Aborting.load())
         return false;
     LOG_LAZY(DbDriverState->Log,
         TLOG_INFO,
@@ -1247,7 +1247,7 @@ bool TWriteSessionImpl::Close(TDuration closeTimeout) {
             if (OriginalMessagesToSend.empty() && SentOriginalMessages.empty()) {
                 ready = true;
             }
-            if (AtomicGet(Aborting))
+            if (Aborting.load())
                 break;
         }
         if (ready) {
@@ -1258,7 +1258,7 @@ bool TWriteSessionImpl::Close(TDuration closeTimeout) {
     }
     {
         std::lock_guard guard(Lock);
-        ready = (OriginalMessagesToSend.empty() && SentOriginalMessages.empty()) && !AtomicGet(Aborting);
+        ready = (OriginalMessagesToSend.empty() && SentOriginalMessages.empty()) && !Aborting.load();
     }
     {
         std::lock_guard guard(Lock);
@@ -1284,7 +1284,7 @@ void TWriteSessionImpl::HandleWakeUpImpl() {
     Y_ABORT_UNLESS(Lock.IsLocked());
 
     FlushWriteIfRequiredImpl();
-    if (AtomicGet(Aborting)) {
+    if (Aborting.load()) {
         return;
     }
     auto callback = [cbContext = SelfContext] (bool ok)
@@ -1352,9 +1352,9 @@ void TWriteSessionImpl::UpdateTimedCountersImpl() {
 void TWriteSessionImpl::AbortImpl() {
     Y_ABORT_UNLESS(Lock.IsLocked());
 
-    if (!AtomicGet(Aborting)) {
+    if (!Aborting.load()) {
         LOG_LAZY(DbDriverState->Log, TLOG_DEBUG, LogPrefix() << "Write session: aborting");
-        AtomicSet(Aborting, 1);
+        Aborting.store(1);
         Cancel(ConnectContext);
         Cancel(ConnectTimeoutContext);
         Cancel(ConnectDelayContext);
@@ -1395,7 +1395,7 @@ TWriteSessionImpl::~TWriteSessionImpl() {
     bool needClose = false;
     {
         std::lock_guard guard(Lock);
-        if (!AtomicGet(Aborting)) {
+        if (!Aborting.load()) {
             CloseImpl(EStatus::SUCCESS, NYql::TIssues{});
 
             needClose = !InitSeqNoSetDone && (InitSeqNoSetDone = true);

src/client/persqueue_public/impl/write_session_impl.h

@@ -419,7 +419,7 @@ class TWriteSessionImpl : public TContinuationTokenIssuer,
     const size_t MaxBlockMessageCount = 1; //!< Max message count that can be packed into a single block. In block version 0 is equal to 1 for compatibility
     bool Connected = false;
     bool Started = false;
-    TAtomic Aborting = 0;
+    std::atomic<int> Aborting = 0;
     bool SessionEstablished = false;
     ui32 PartitionId = 0;
     ui64 NextId = 0;

src/client/topic/impl/write_session_impl.cpp

@@ -167,7 +167,7 @@ TWriteSessionImpl::THandleResult TWriteSessionImpl::RestartImpl(const TPlainStat
         TRACE_KV("status", status.Status));
 
     THandleResult result;
-    if (AtomicGet(Aborting)) {
+    if (Aborting.load()) {
         LOG_LAZY(DbDriverState->Log, TLOG_DEBUG, LogPrefix() << "Write session is aborting and will not restart");
         return result;
     }
@@ -226,7 +226,7 @@ void TWriteSessionImpl::ConnectToPreferredPartitionLocation(const TDuration& del
     Y_ABORT_UNLESS(Lock.IsLocked());
     Y_ABORT_UNLESS(Settings.DirectWriteToPartition_ && (Settings.PartitionId_.has_value() || DirectWriteToPartitionId.has_value()));
 
-    if (AtomicGet(Aborting)) {
+    if (Aborting.load()) {
         return;
     }
 
@@ -1425,7 +1425,7 @@ void TWriteSessionImpl::SendImpl() {
 
 // Client method, no Lock
 bool TWriteSessionImpl::Close(TDuration closeTimeout) {
-    if (AtomicGet(Aborting)) {
+    if (Aborting.load()) {
         return false;
     }
     LOG_LAZY(DbDriverState->Log, TLOG_INFO, LogPrefix() << "Write session: close. Timeout " << closeTimeout);
@@ -1439,7 +1439,7 @@ bool TWriteSessionImpl::Close(TDuration closeTimeout) {
             if (OriginalMessagesToSend.empty() && SentOriginalMessages.empty()) {
                 ready = true;
             }
-            if (AtomicGet(Aborting))
+            if (Aborting.load())
                 break;
         }
         if (ready) {
@@ -1450,7 +1450,7 @@ bool TWriteSessionImpl::Close(TDuration closeTimeout) {
     }
     {
         std::lock_guard guard(Lock);
-        ready = (OriginalMessagesToSend.empty() && SentOriginalMessages.empty()) && !AtomicGet(Aborting);
+        ready = (OriginalMessagesToSend.empty() && SentOriginalMessages.empty()) && !Aborting.load();
     }
     {
         std::lock_guard guard(Lock);
@@ -1472,7 +1472,7 @@ void TWriteSessionImpl::HandleWakeUpImpl() {
     Y_ABORT_UNLESS(Lock.IsLocked());
 
     FlushWriteIfRequiredImpl();
-    if (AtomicGet(Aborting)) {
+    if (Aborting.load()) {
         return;
     }
     auto callback = [cbContext = SelfContext] (bool ok)
@@ -1538,9 +1538,9 @@ void TWriteSessionImpl::UpdateTimedCountersImpl() {
 void TWriteSessionImpl::AbortImpl() {
     Y_ABORT_UNLESS(Lock.IsLocked());
 
-    if (!AtomicGet(Aborting)) {
+    if (!Aborting.load()) {
         LOG_LAZY(DbDriverState->Log, TLOG_DEBUG, LogPrefix() << "Write session: aborting");
-        AtomicSet(Aborting, 1);
+        Aborting.store(1);
         Cancel(DescribePartitionContext);
         Cancel(ConnectContext);
         Cancel(ConnectTimeoutContext);
@@ -1582,7 +1582,7 @@ TWriteSessionImpl::~TWriteSessionImpl() {
     bool needClose = false;
     {
         std::lock_guard guard(Lock);
-        if (!AtomicGet(Aborting)) {
+        if (!Aborting.load()) {
             CloseImpl(EStatus::SUCCESS, NYql::TIssues{});
 
             needClose = !InitSeqNoSetDone && (InitSeqNoSetDone = true);

src/client/topic/impl/write_session_impl.h

@@ -452,7 +452,7 @@ class TWriteSessionImpl : public TContinuationTokenIssuer,
     const size_t MaxBlockMessageCount = 1; //!< Max message count that can be packed into a single block. In block version 0 is equal to 1 for compatibility
     bool Connected = false;
     bool Started = false;
-    TAtomic Aborting = 0;
+    std::atomic<int> Aborting = 0;
     bool SessionEstablished = false;
     ui32 PartitionId = 0;
     TPartitionLocation PreferredPartitionLocation = {};

src/client/topic/ut/basic_usage_ut.cpp

@@ -71,20 +71,20 @@ void WriteAndReadToEndWithRestarts(TReadSessionSettings readSettings, TWriteSess
 
 
     NThreading::TPromise<void> checkedPromise = NThreading::NewPromise<void>();
-    TAtomic lastOffset = 0u;
+    std::atomic<int> lastOffset = 0u;
 
     auto f = checkedPromise.GetFuture();
     readSettings.EventHandlers_.SimpleDataHandlers(
         [&]
         (TReadSessionEvent::TDataReceivedEvent& ev) mutable {
-        AtomicSet(lastOffset, ev.GetMessages().back().GetOffset());
+        lastOffset.store(ev.GetMessages().back().GetOffset());
         std::cerr << ">>> TEST: last offset = " << lastOffset << std::endl;
     });
 
     ReadSession = topicClient.CreateReadSession(readSettings);
 
     ui32 i = 0;
-    while (AtomicGet(lastOffset) + 1 < count) {
+    while (lastOffset.load() + 1 < count) {
         RunTasks(decompressor, {i++});
     }
 
@@ -302,12 +302,12 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
         auto totalReceived = 0u;
 
         auto f = checkedPromise.GetFuture();
-        TAtomic check = 1;
+        std::atomic<int> check = 1;
         readSettings.EventHandlers_.SimpleDataHandlers(
             // [checkedPromise = std::move(checkedPromise), &check, &sentMessages, &totalReceived]
             [&]
             (TReadSessionEvent::TDataReceivedEvent& ev) mutable {
-            Y_VERIFY_S(AtomicGet(check) != 0, "check is false");
+            Y_VERIFY_S(check.load() != 0, "check is false");
             auto& messages = ev.GetMessages();
             for (size_t i = 0u; i < messages.size(); ++i) {
                 auto& message = messages[i];
@@ -322,7 +322,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
 
         f.GetValueSync();
         ReadSession->Close(TDuration::MilliSeconds(10));
-        AtomicSet(check, 0);
+        check.store(0);
 
         auto status = topicClient.CommitOffset(setup->GetTestTopic(), 0, setup->GetTestConsumer(), 50);
         UNIT_ASSERT(status.GetValueSync().IsSuccess());

src/library/http/server/http.cpp

@@ -469,15 +469,15 @@ class THttpServer::TImpl {
     }
 
     inline void DecreaseConnections() noexcept {
-        AtomicDecrement(ConnectionCount);
+        ConnectionCount--;
     }
 
     inline void IncreaseConnections() noexcept {
-        AtomicIncrement(ConnectionCount);
+        ConnectionCount++;
     }
 
     inline i64 GetClientCount() const {
-        return AtomicGet(ConnectionCount);
+        return ConnectionCount.load();
     }
 
     inline bool MaxRequestsReached() const {
@@ -491,7 +491,7 @@ class THttpServer::TImpl {
     TPipeHandle ListenWakeupWriteFd;
     TMtpQueueRef Requests;
     TMtpQueueRef FailRequests;
-    TAtomic ConnectionCount = 0;
+    std::atomic<int> ConnectionCount = 0;
     THolder<TSocketPoller> Poller;
     THolder<TConnections> Connections;
     int ErrorCode = 0;