Restoring authorship annotation for Alexey Borzenkov <snaury@yandex-team.ru>. Commit 1 of 2.

1 files changed, 287 insertions, 287 deletions

diff --git a/library/cpp/grpc/client/grpc_client_low.cpp b/library/cpp/grpc/client/grpc_client_low.cpp
index 73cc908ef8..bba83aee02 100644
--- a/library/cpp/grpc/client/grpc_client_low.cpp
+++ b/library/cpp/grpc/client/grpc_client_low.cpp
@@ -193,214 +193,214 @@ static void PullEvents(grpc::CompletionQueue* cq) {
     }
 }
 
-class TGRpcClientLow::TContextImpl final
-    : public std::enable_shared_from_this<TContextImpl>
-    , public IQueueClientContext
-{
-    friend class TGRpcClientLow;
-
-    using TCallback = std::function<void()>;
-    using TContextPtr = std::shared_ptr<TContextImpl>;
-
-public:
+class TGRpcClientLow::TContextImpl final 
+    : public std::enable_shared_from_this<TContextImpl> 
+    , public IQueueClientContext 
+{ 
+    friend class TGRpcClientLow; 
+ 
+    using TCallback = std::function<void()>; 
+    using TContextPtr = std::shared_ptr<TContextImpl>; 
+ 
+public: 
     ~TContextImpl() override {
-        Y_VERIFY(CountChildren() == 0,
-                "Destructor called with non-empty children");
-
-        if (Parent) {
-            Parent->ForgetContext(this);
-        } else if (Y_LIKELY(Owner)) {
-            Owner->ForgetContext(this);
-        }
-    }
-
-    /**
-     * Helper for locking child pointer from a parent container
-     */
-    static TContextPtr LockChildPtr(TContextImpl* ptr) {
-        if (ptr) {
-            // N.B. it is safe to do as long as it's done under a mutex and
-            // pointer is among valid children. When that's the case we
-            // know that TContextImpl destructor has not finished yet, so
-            // the object is valid. The lock() method may return nullptr
-            // though, if the object is being destructed right now.
-            return ptr->weak_from_this().lock();
-        } else {
-            return nullptr;
-        }
-    }
-
-    void ForgetContext(TContextImpl* child) {
-        std::unique_lock<std::mutex> guard(Mutex);
-
-        auto removed = RemoveChild(child);
-        Y_VERIFY(removed, "Unexpected ForgetContext(%p)", child);
-    }
-
-    IQueueClientContextPtr CreateContext() override {
-        auto self = shared_from_this();
-        auto child = std::make_shared<TContextImpl>();
-
-        {
-            std::unique_lock<std::mutex> guard(Mutex);
-
-            AddChild(child.get());
-
-            // It's now safe to initialize parent and owner
-            child->Parent = std::move(self);
-            child->Owner = Owner;
+        Y_VERIFY(CountChildren() == 0, 
+                "Destructor called with non-empty children"); 
+ 
+        if (Parent) { 
+            Parent->ForgetContext(this); 
+        } else if (Y_LIKELY(Owner)) { 
+            Owner->ForgetContext(this); 
+        } 
+    } 
+ 
+    /** 
+     * Helper for locking child pointer from a parent container 
+     */ 
+    static TContextPtr LockChildPtr(TContextImpl* ptr) { 
+        if (ptr) { 
+            // N.B. it is safe to do as long as it's done under a mutex and 
+            // pointer is among valid children. When that's the case we 
+            // know that TContextImpl destructor has not finished yet, so 
+            // the object is valid. The lock() method may return nullptr 
+            // though, if the object is being destructed right now. 
+            return ptr->weak_from_this().lock(); 
+        } else { 
+            return nullptr; 
+        } 
+    } 
+ 
+    void ForgetContext(TContextImpl* child) { 
+        std::unique_lock<std::mutex> guard(Mutex); 
+ 
+        auto removed = RemoveChild(child); 
+        Y_VERIFY(removed, "Unexpected ForgetContext(%p)", child); 
+    } 
+ 
+    IQueueClientContextPtr CreateContext() override { 
+        auto self = shared_from_this(); 
+        auto child = std::make_shared<TContextImpl>(); 
+ 
+        { 
+            std::unique_lock<std::mutex> guard(Mutex); 
+ 
+            AddChild(child.get()); 
+ 
+            // It's now safe to initialize parent and owner 
+            child->Parent = std::move(self); 
+            child->Owner = Owner; 
             child->CQ = CQ;
-
-            // Propagate cancellation to a child context
-            if (Cancelled.load(std::memory_order_relaxed)) {
-                child->Cancelled.store(true, std::memory_order_relaxed);
-            }
-        }
-
-        return child;
-    }
-
-    grpc::CompletionQueue* CompletionQueue() override {
-        Y_VERIFY(Owner, "Uninitialized context");
+ 
+            // Propagate cancellation to a child context 
+            if (Cancelled.load(std::memory_order_relaxed)) { 
+                child->Cancelled.store(true, std::memory_order_relaxed); 
+            } 
+        } 
+ 
+        return child; 
+    } 
+ 
+    grpc::CompletionQueue* CompletionQueue() override { 
+        Y_VERIFY(Owner, "Uninitialized context"); 
         return CQ;
-    }
-
-    bool IsCancelled() const override {
-        return Cancelled.load(std::memory_order_acquire);
-    }
-
-    bool Cancel() override {
-        TStackVec<TCallback, 1> callbacks;
-        TStackVec<TContextPtr, 2> children;
-
-        {
-            std::unique_lock<std::mutex> guard(Mutex);
-
-            if (Cancelled.load(std::memory_order_relaxed)) {
-                // Already cancelled in another thread
-                return false;
-            }
-
-            callbacks.reserve(Callbacks.size());
-            children.reserve(CountChildren());
-
-            for (auto& callback : Callbacks) {
-                callbacks.emplace_back().swap(callback);
-            }
-            Callbacks.clear();
-
-            // Collect all children we need to cancel
-            // N.B. we don't clear children links (cleared by destructors)
-            // N.B. some children may be stuck in destructors at the moment
-            for (TContextImpl* ptr : InlineChildren) {
-                if (auto child = LockChildPtr(ptr)) {
-                    children.emplace_back(std::move(child));
-                }
-            }
-            for (auto* ptr : Children) {
-                if (auto child = LockChildPtr(ptr)) {
-                    children.emplace_back(std::move(child));
-                }
-            }
-
-            Cancelled.store(true, std::memory_order_release);
-        }
-
-        // Call directly subscribed callbacks
-        if (callbacks) {
-            RunCallbacksNoExcept(callbacks);
-        }
-
-        // Cancel all children
-        for (auto& child : children) {
-            child->Cancel();
-            child.reset();
-        }
-
-        return true;
-    }
-
-    void SubscribeCancel(TCallback callback) override {
-        Y_VERIFY(callback, "SubscribeCancel called with an empty callback");
-
-        {
-            std::unique_lock<std::mutex> guard(Mutex);
-
-            if (!Cancelled.load(std::memory_order_relaxed)) {
-                Callbacks.emplace_back().swap(callback);
-                return;
-            }
-        }
-
-        // Already cancelled, run immediately
-        callback();
-    }
-
-private:
-    void AddChild(TContextImpl* child) {
-        for (TContextImpl*& slot : InlineChildren) {
-            if (!slot) {
-                slot = child;
-                return;
-            }
-        }
-
-        Children.insert(child);
-    }
-
-    bool RemoveChild(TContextImpl* child) {
-        for (TContextImpl*& slot : InlineChildren) {
-            if (slot == child) {
-                slot = nullptr;
-                return true;
-            }
-        }
-
-        return Children.erase(child);
-    }
-
-    size_t CountChildren() {
-        size_t count = 0;
-
-        for (TContextImpl* ptr : InlineChildren) {
-            if (ptr) {
-                ++count;
-            }
-        }
-
-        return count + Children.size();
-    }
-
-    template<class TCallbacks>
-    static void RunCallbacksNoExcept(TCallbacks& callbacks) noexcept {
-        for (auto& callback : callbacks) {
-            if (callback) {
-                callback();
-                callback = nullptr;
-            }
-        }
-    }
-
-private:
-    // We want a simple lock here, without extra memory allocations
-    std::mutex Mutex;
-
-    // These fields are initialized on successful registration
-    TContextPtr Parent;
-    TGRpcClientLow* Owner = nullptr;
+    } 
+ 
+    bool IsCancelled() const override { 
+        return Cancelled.load(std::memory_order_acquire); 
+    } 
+ 
+    bool Cancel() override { 
+        TStackVec<TCallback, 1> callbacks; 
+        TStackVec<TContextPtr, 2> children; 
+ 
+        { 
+            std::unique_lock<std::mutex> guard(Mutex); 
+ 
+            if (Cancelled.load(std::memory_order_relaxed)) { 
+                // Already cancelled in another thread 
+                return false; 
+            } 
+ 
+            callbacks.reserve(Callbacks.size()); 
+            children.reserve(CountChildren()); 
+ 
+            for (auto& callback : Callbacks) { 
+                callbacks.emplace_back().swap(callback); 
+            } 
+            Callbacks.clear(); 
+ 
+            // Collect all children we need to cancel 
+            // N.B. we don't clear children links (cleared by destructors) 
+            // N.B. some children may be stuck in destructors at the moment 
+            for (TContextImpl* ptr : InlineChildren) { 
+                if (auto child = LockChildPtr(ptr)) { 
+                    children.emplace_back(std::move(child)); 
+                } 
+            } 
+            for (auto* ptr : Children) { 
+                if (auto child = LockChildPtr(ptr)) { 
+                    children.emplace_back(std::move(child)); 
+                } 
+            } 
+ 
+            Cancelled.store(true, std::memory_order_release); 
+        } 
+ 
+        // Call directly subscribed callbacks 
+        if (callbacks) { 
+            RunCallbacksNoExcept(callbacks); 
+        } 
+ 
+        // Cancel all children 
+        for (auto& child : children) { 
+            child->Cancel(); 
+            child.reset(); 
+        } 
+ 
+        return true; 
+    } 
+ 
+    void SubscribeCancel(TCallback callback) override { 
+        Y_VERIFY(callback, "SubscribeCancel called with an empty callback"); 
+ 
+        { 
+            std::unique_lock<std::mutex> guard(Mutex); 
+ 
+            if (!Cancelled.load(std::memory_order_relaxed)) { 
+                Callbacks.emplace_back().swap(callback); 
+                return; 
+            } 
+        } 
+ 
+        // Already cancelled, run immediately 
+        callback(); 
+    } 
+ 
+private: 
+    void AddChild(TContextImpl* child) { 
+        for (TContextImpl*& slot : InlineChildren) { 
+            if (!slot) { 
+                slot = child; 
+                return; 
+            } 
+        } 
+ 
+        Children.insert(child); 
+    } 
+ 
+    bool RemoveChild(TContextImpl* child) { 
+        for (TContextImpl*& slot : InlineChildren) { 
+            if (slot == child) { 
+                slot = nullptr; 
+                return true; 
+            } 
+        } 
+ 
+        return Children.erase(child); 
+    } 
+ 
+    size_t CountChildren() { 
+        size_t count = 0; 
+ 
+        for (TContextImpl* ptr : InlineChildren) { 
+            if (ptr) { 
+                ++count; 
+            } 
+        } 
+ 
+        return count + Children.size(); 
+    } 
+ 
+    template<class TCallbacks> 
+    static void RunCallbacksNoExcept(TCallbacks& callbacks) noexcept { 
+        for (auto& callback : callbacks) { 
+            if (callback) { 
+                callback(); 
+                callback = nullptr; 
+            } 
+        } 
+    } 
+ 
+private: 
+    // We want a simple lock here, without extra memory allocations 
+    std::mutex Mutex; 
+ 
+    // These fields are initialized on successful registration 
+    TContextPtr Parent; 
+    TGRpcClientLow* Owner = nullptr; 
     grpc::CompletionQueue* CQ = nullptr;
-
-    // Some children are stored inline, others are in a set
-    std::array<TContextImpl*, 2> InlineChildren{ { nullptr, nullptr } };
+ 
+    // Some children are stored inline, others are in a set 
+    std::array<TContextImpl*, 2> InlineChildren{ { nullptr, nullptr } }; 
     std::unordered_set<TContextImpl*> Children;
-
-    // Single callback is stored without extra allocations
-    TStackVec<TCallback, 1> Callbacks;
-
-    // Atomic flag for a faster IsCancelled() implementation
-    std::atomic<bool> Cancelled;
-};
-
+ 
+    // Single callback is stored without extra allocations 
+    TStackVec<TCallback, 1> Callbacks; 
+ 
+    // Atomic flag for a faster IsCancelled() implementation 
+    std::atomic<bool> Cancelled; 
+}; 
+ 
 TGRpcClientLow::TGRpcClientLow(size_t numWorkerThread, bool useCompletionQueuePerThread)
     : UseCompletionQueuePerThread_(useCompletionQueuePerThread)
 {
@@ -408,7 +408,7 @@ TGRpcClientLow::TGRpcClientLow(size_t numWorkerThread, bool useCompletionQueuePe
 }
 
 void TGRpcClientLow::Init(size_t numWorkerThread) {
-    SetCqState(WORKING);
+    SetCqState(WORKING); 
     if (UseCompletionQueuePerThread_) {
         for (size_t i = 0; i < numWorkerThread; i++) {
             CQS_.push_back(std::make_unique<grpc::CompletionQueue>());
@@ -428,7 +428,7 @@ void TGRpcClientLow::Init(size_t numWorkerThread) {
     }
 }
 
-void TGRpcClientLow::AddWorkerThreadForTest() {
+void TGRpcClientLow::AddWorkerThreadForTest() { 
     if (UseCompletionQueuePerThread_) {
         CQS_.push_back(std::make_unique<grpc::CompletionQueue>());
         auto* cq = CQS_.back().get();
@@ -441,75 +441,75 @@ void TGRpcClientLow::AddWorkerThreadForTest() {
             PullEvents(cq);
         }).Release());
     }
-}
-
-TGRpcClientLow::~TGRpcClientLow() {
-    StopInternal(true);
-    WaitInternal();
-}
-
-void TGRpcClientLow::Stop(bool wait) {
-    StopInternal(false);
-
-    if (wait) {
-        WaitInternal();
+} 
+ 
+TGRpcClientLow::~TGRpcClientLow() { 
+    StopInternal(true); 
+    WaitInternal(); 
+} 
+ 
+void TGRpcClientLow::Stop(bool wait) { 
+    StopInternal(false); 
+ 
+    if (wait) { 
+        WaitInternal(); 
     }
 }
 
-void TGRpcClientLow::StopInternal(bool silent) {
-    bool shutdown;
-
-    TVector<TContextImpl::TContextPtr> cancelQueue;
-
+void TGRpcClientLow::StopInternal(bool silent) { 
+    bool shutdown; 
+ 
+    TVector<TContextImpl::TContextPtr> cancelQueue; 
+ 
     {
         std::unique_lock<std::mutex> guard(Mtx_);
 
-        auto allowStateChange = [&]() {
-            switch (GetCqState()) {
-                case WORKING:
-                    return true;
-                case STOP_SILENT:
-                    return !silent;
-                case STOP_EXPLICIT:
-                    return false;
-            }
-
-            Y_UNREACHABLE();
-        };
-
-        if (!allowStateChange()) {
-            // Completion queue is already stopping
-            return;
-        }
-
-        SetCqState(silent ? STOP_SILENT : STOP_EXPLICIT);
-
+        auto allowStateChange = [&]() { 
+            switch (GetCqState()) { 
+                case WORKING: 
+                    return true; 
+                case STOP_SILENT: 
+                    return !silent; 
+                case STOP_EXPLICIT: 
+                    return false; 
+            } 
+ 
+            Y_UNREACHABLE(); 
+        }; 
+ 
+        if (!allowStateChange()) { 
+            // Completion queue is already stopping 
+            return; 
+        } 
+ 
+        SetCqState(silent ? STOP_SILENT : STOP_EXPLICIT); 
+ 
         if (!silent && !Contexts_.empty()) {
-            cancelQueue.reserve(Contexts_.size());
-            for (auto* ptr : Contexts_) {
-                // N.B. some contexts may be stuck in destructors
-                if (auto context = TContextImpl::LockChildPtr(ptr)) {
-                    cancelQueue.emplace_back(std::move(context));
-                }
-            }
-        }
-
+            cancelQueue.reserve(Contexts_.size()); 
+            for (auto* ptr : Contexts_) { 
+                // N.B. some contexts may be stuck in destructors 
+                if (auto context = TContextImpl::LockChildPtr(ptr)) { 
+                    cancelQueue.emplace_back(std::move(context)); 
+                } 
+            } 
+        } 
+ 
         shutdown = Contexts_.empty();
-    }
-
-    for (auto& context : cancelQueue) {
-        context->Cancel();
-        context.reset();
-    }
-
-    if (shutdown) {
+    } 
+ 
+    for (auto& context : cancelQueue) { 
+        context->Cancel(); 
+        context.reset(); 
+    } 
+ 
+    if (shutdown) { 
         for (auto& cq : CQS_) {
             cq->Shutdown();
         }
     }
-}
-
-void TGRpcClientLow::WaitInternal() {
+} 
+ 
+void TGRpcClientLow::WaitInternal() { 
     std::unique_lock<std::mutex> guard(JoinMutex_);
 
     for (auto& ti : WorkerThreads_) {
@@ -517,7 +517,7 @@ void TGRpcClientLow::WaitInternal() {
     }
 }
 
-void TGRpcClientLow::WaitIdle() {
+void TGRpcClientLow::WaitIdle() { 
     std::unique_lock<std::mutex> guard(Mtx_);
 
     while (!Contexts_.empty()) {
@@ -525,7 +525,7 @@ void TGRpcClientLow::WaitIdle() {
     }
 }
 
-std::shared_ptr<IQueueClientContext> TGRpcClientLow::CreateContext() {
+std::shared_ptr<IQueueClientContext> TGRpcClientLow::CreateContext() { 
     std::unique_lock<std::mutex> guard(Mtx_);
 
     auto allowCreateContext = [&]() {
@@ -535,15 +535,15 @@ std::shared_ptr<IQueueClientContext> TGRpcClientLow::CreateContext() {
             case STOP_SILENT:
             case STOP_EXPLICIT:
                 return false;
-        }
-
+        } 
+ 
         Y_UNREACHABLE();
     };
 
     if (!allowCreateContext()) {
         // New context creation is forbidden
         return nullptr;
-    }
+    } 
 
     auto context = std::make_shared<TContextImpl>();
     Contexts_.insert(context.get());
@@ -554,32 +554,32 @@ std::shared_ptr<IQueueClientContext> TGRpcClientLow::CreateContext() {
         context->CQ = CQS_[0].get();
     }
     return context;
-}
-
-void TGRpcClientLow::ForgetContext(TContextImpl* context) {
-    bool shutdown = false;
-
+} 
+ 
+void TGRpcClientLow::ForgetContext(TContextImpl* context) { 
+    bool shutdown = false; 
+ 
     {
         std::unique_lock<std::mutex> guard(Mtx_);
 
-        if (!Contexts_.erase(context)) {
-            Y_FAIL("Unexpected ForgetContext(%p)", context);
-        }
-
-        if (Contexts_.empty()) {
-            if (IsStopping()) {
-                shutdown = true;
-            }
-
+        if (!Contexts_.erase(context)) { 
+            Y_FAIL("Unexpected ForgetContext(%p)", context); 
+        } 
+ 
+        if (Contexts_.empty()) { 
+            if (IsStopping()) { 
+                shutdown = true; 
+            } 
+ 
             ContextsEmpty_.notify_all();
-        }
-    }
-
-    if (shutdown) {
-        // This was the last context, shutdown CQ
+        } 
+    } 
+ 
+    if (shutdown) { 
+        // This was the last context, shutdown CQ 
         for (auto& cq : CQS_) {
             cq->Shutdown();
-        }
+        } 
     }
 }