Attempt to solve problem with verify while ydb is shutting down pqv0 25-1-3 (#21729)

Author: alexnick88

ydb/library/grpc/server/grpc_server.cpp

@@ -273,6 +273,7 @@ void TGRpcServer::Start() {
 }
 
 void TGRpcServer::Stop() {
+ NYdbGrpc::GrpcDead = 1;
  for (auto& service : Services_) {
  service->StopService();
  }
@@ -304,8 +305,8 @@ void TGRpcServer::Stop() {
  }
 
  if (!unsafe && spent > Options_.GRpcShutdownDeadline.SecondsFloat()) {
- Cerr << "GRpc shutdown warning: failed to shutdown all connections, left infly: " << infly << ", spent: " << spent << " sec"
- << Endl;
+ Cerr << "GRpc shutdown warning: failed to shutdown all connections, left infly: " << infly << ", spent: " << spent
+ << " sec. GRpcShutdownDeadline: " << Options_.GRpcShutdownDeadline.SecondsFloat() << Endl;
  break;
  }
  Sleep(TDuration::MilliSeconds(10));

ydb/library/grpc/server/grpc_server.h

@@ -24,6 +24,8 @@ namespace NMonitoring {
 
 namespace NYdbGrpc {
 
+static std::atomic<int> GrpcDead = 0;
+
 struct TSslData {
  TString Cert;
  TString Key;

ydb/services/deprecated/persqueue_v0/grpc_pq_read.h

@@ -79,6 +79,11 @@ class TPQReadService : public IPQClustersUpdaterCallback, public std::enable_sha
 
  void StopService() {
  AtomicSet(ShuttingDown_, 1);
+ auto g(Guard(Lock));
+ for (auto it = Sessions.begin(); it != Sessions.end();) {
+ auto jt = it++;
+ jt->second->DestroyStream("Grpc server is dead", NPersQueue::NErrorCode::BAD_REQUEST);
+ }
  }
 
  bool IsShuttingDown() const {

ydb/services/deprecated/persqueue_v0/grpc_pq_session.h

@@ -118,14 +118,18 @@ class ISession : public ISessionHandler<TResponse>
  Session->HaveWriteInflight = false;
  if (Session->NeedFinish) {
  lock.Release();
- Session->Stream.Finish(Status::OK, new TFinishDone(Session));
+ if (!NYdbGrpc::GrpcDead) {
+ Session->Stream.Finish(Status::OK, new TFinishDone(Session));
+ }
  }
  } else {
  auto resp = std::move(Session->Responses.front());
  Session->Responses.pop();
  lock.Release();
  ui64 sz = resp.ByteSize();
- Session->Stream.Write(resp, new TWriteDone(Session, sz));
+ if (!NYdbGrpc::GrpcDead) {
+ Session->Stream.Write(resp, new TWriteDone(Session, sz));
+ }
  }
 
  return false;
@@ -271,7 +275,9 @@ class ISession : public ISessionHandler<TResponse>
  HaveWriteInflight = true;
  }
 
- Stream.Finish(Status::OK, new TFinishDone(this));
+ if (!NYdbGrpc::GrpcDead) {
+ Stream.Finish(Status::OK, new TFinishDone(this));
+ }
  }
 
  /// Send reply to client.
@@ -289,7 +295,9 @@ class ISession : public ISessionHandler<TResponse>
  }
 
  ui64 size = resp.ByteSize();
- Stream.Write(resp, new TWriteDone(this, size));
+ if (!NYdbGrpc::GrpcDead) {
+ Stream.Write(resp, new TWriteDone(this, size));
+ }
  }
 
  void ReadyForNextRead() override {
@@ -301,7 +309,9 @@ class ISession : public ISessionHandler<TResponse>
  }
 
  auto read = new TReadDone(this);
- Stream.Read(&read->Request, read);
+ if (!NYdbGrpc::GrpcDead) {
+ Stream.Read(&read->Request, read);
+ }
  }
 
 protected:

ydb/services/deprecated/persqueue_v0/grpc_pq_write.h

@@ -69,6 +69,11 @@ class TPQWriteServiceImpl : public IPQClustersUpdaterCallback, public std::enabl
 
  void StopService() {
  AtomicSet(ShuttingDown_, 1);
+ auto g(Guard(Lock));
+ for (auto it = Sessions.begin(); it != Sessions.end();) {
+ auto jt = it++;
+ jt->second->DestroyStream("Grpc server is dead", NPersQueue::NErrorCode::BAD_REQUEST);
+ }
  }
 
  bool IsShuttingDown() const {