intermediate changes

ref:cde9a383711a11544ce7e107a78147fb96cc4029

1 files changed, 67 insertions, 0 deletions

diff --git a/library/cpp/messagebus/remote_client_session_semaphore.cpp b/library/cpp/messagebus/remote_client_session_semaphore.cpp
new file mode 100644
index 0000000000..f877ed4257
--- /dev/null
+++ b/library/cpp/messagebus/remote_client_session_semaphore.cpp
@@ -0,0 +1,67 @@
+#include "remote_client_session_semaphore.h"
+
+#include <util/stream/output.h>
+#include <util/system/yassert.h>
+
+using namespace NBus;
+using namespace NBus::NPrivate;
+
+TRemoteClientSessionSemaphore::TRemoteClientSessionSemaphore(TAtomicBase limit, const char* name)
+    : Name(name)
+    , Limit(limit)
+    , Current(0)
+    , StopSignal(0)
+{
+    Y_VERIFY(limit > 0, "limit must be > 0");
+    Y_UNUSED(Name);
+}
+
+TRemoteClientSessionSemaphore::~TRemoteClientSessionSemaphore() {
+    Y_VERIFY(AtomicGet(Current) == 0);
+}
+
+bool TRemoteClientSessionSemaphore::TryAcquire() {
+    if (!TryWait()) {
+        return false;
+    }
+
+    AtomicIncrement(Current);
+    return true;
+}
+
+bool TRemoteClientSessionSemaphore::TryWait() {
+    if (AtomicGet(Current) < Limit)
+        return true;
+    if (Y_UNLIKELY(AtomicGet(StopSignal)))
+        return true;
+    return false;
+}
+
+void TRemoteClientSessionSemaphore::Acquire() {
+    Wait();
+
+    Increment();
+}
+
+void TRemoteClientSessionSemaphore::Increment() {
+    IncrementMultiple(1);
+}
+
+void TRemoteClientSessionSemaphore::IncrementMultiple(TAtomicBase count) {
+    AtomicAdd(Current, count);
+    Updated();
+}
+
+void TRemoteClientSessionSemaphore::Release() {
+    ReleaseMultiple(1);
+}
+
+void TRemoteClientSessionSemaphore::ReleaseMultiple(TAtomicBase count) {
+    AtomicSub(Current, count);
+    Updated();
+}
+
+void TRemoteClientSessionSemaphore::Stop() {
+    AtomicSet(StopSignal, 1);
+    Updated();
+}