Merge branch 'master' into cstd

BUILD

@@ -355,6 +355,7 @@ grpc_cc_library(
         "gpr",
         "grpc",
         "grpc++_base",
+        "grpc_cfstream",
         "grpc++_codegen_base",
         "grpc++_codegen_base_src",
         "grpc++_codegen_proto",
@@ -618,10 +619,6 @@ grpc_cc_library(
 
 grpc_cc_library(
     name = "atomic",
-    hdrs = [
-        "src/core/lib/gprpp/atomic_with_atm.h",
-        "src/core/lib/gprpp/atomic_with_std.h",
-    ],
     language = "c++",
     public_hdrs = [
         "src/core/lib/gprpp/atomic.h",
@@ -677,6 +674,7 @@ grpc_cc_library(
     language = "c++",
     public_hdrs = ["src/core/lib/gprpp/ref_counted.h"],
     deps = [
+        "atomic",
         "debug_location",
         "gpr_base",
         "grpc_trace",

build.yaml

@@ -192,8 +192,6 @@ filegroups:
   - src/core/lib/gpr/useful.h
   - src/core/lib/gprpp/abstract.h
   - src/core/lib/gprpp/atomic.h
-  - src/core/lib/gprpp/atomic_with_atm.h
-  - src/core/lib/gprpp/atomic_with_std.h
   - src/core/lib/gprpp/fork.h
   - src/core/lib/gprpp/manual_constructor.h
   - src/core/lib/gprpp/memory.h

gRPC-C++.podspec

@@ -251,8 +251,6 @@ Pod::Spec.new do |s|
                       'src/core/lib/gpr/useful.h',
                       'src/core/lib/gprpp/abstract.h',
                       'src/core/lib/gprpp/atomic.h',
-                      'src/core/lib/gprpp/atomic_with_atm.h',
-                      'src/core/lib/gprpp/atomic_with_std.h',
                       'src/core/lib/gprpp/fork.h',
                       'src/core/lib/gprpp/manual_constructor.h',
                       'src/core/lib/gprpp/memory.h',
@@ -567,8 +565,6 @@ Pod::Spec.new do |s|
                               'src/core/lib/gpr/useful.h',
                               'src/core/lib/gprpp/abstract.h',
                               'src/core/lib/gprpp/atomic.h',
-                              'src/core/lib/gprpp/atomic_with_atm.h',
-                              'src/core/lib/gprpp/atomic_with_std.h',
                               'src/core/lib/gprpp/fork.h',
                               'src/core/lib/gprpp/manual_constructor.h',
                               'src/core/lib/gprpp/memory.h',

gRPC-Core.podspec

@@ -206,8 +206,6 @@ Pod::Spec.new do |s|
                       'src/core/lib/gpr/useful.h',
                       'src/core/lib/gprpp/abstract.h',
                       'src/core/lib/gprpp/atomic.h',
-                      'src/core/lib/gprpp/atomic_with_atm.h',
-                      'src/core/lib/gprpp/atomic_with_std.h',
                       'src/core/lib/gprpp/fork.h',
                       'src/core/lib/gprpp/manual_constructor.h',
                       'src/core/lib/gprpp/memory.h',
@@ -875,8 +873,6 @@ Pod::Spec.new do |s|
                               'src/core/lib/gpr/useful.h',
                               'src/core/lib/gprpp/abstract.h',
                               'src/core/lib/gprpp/atomic.h',
-                              'src/core/lib/gprpp/atomic_with_atm.h',
-                              'src/core/lib/gprpp/atomic_with_std.h',
                               'src/core/lib/gprpp/fork.h',
                               'src/core/lib/gprpp/manual_constructor.h',
                               'src/core/lib/gprpp/memory.h',

grpc.gemspec

@@ -100,8 +100,6 @@ Gem::Specification.new do |s|
   s.files += %w( src/core/lib/gpr/useful.h )
   s.files += %w( src/core/lib/gprpp/abstract.h )
   s.files += %w( src/core/lib/gprpp/atomic.h )
-  s.files += %w( src/core/lib/gprpp/atomic_with_atm.h )
-  s.files += %w( src/core/lib/gprpp/atomic_with_std.h )
   s.files += %w( src/core/lib/gprpp/fork.h )
   s.files += %w( src/core/lib/gprpp/manual_constructor.h )
   s.files += %w( src/core/lib/gprpp/memory.h )

include/grpcpp/impl/codegen/client_interceptor.h

@@ -76,7 +76,7 @@ class ClientRpcInfo {
     UNKNOWN  // UNKNOWN is not API and will be removed later
   };
 
-  ~ClientRpcInfo(){};
+  ~ClientRpcInfo() {}
 
   // Delete copy constructor but allow default move constructor
   ClientRpcInfo(const ClientRpcInfo&) = delete;

include/grpcpp/impl/codegen/interceptor.h

@@ -90,7 +90,7 @@ enum class InterceptionHookPoints {
 ///   5. Set some fields of an RPC at each interception point, when possible
 class InterceptorBatchMethods {
  public:
-  virtual ~InterceptorBatchMethods(){};
+  virtual ~InterceptorBatchMethods() {}
   /// Determine whether the current batch has an interception hook point
   /// of type \a type
   virtual bool QueryInterceptionHookPoint(InterceptionHookPoints type) = 0;

include/grpcpp/impl/codegen/server_callback.h

@@ -102,7 +102,7 @@ class ServerCallbackWriter {
     // Default implementation that can/should be overridden
     Write(msg, std::move(options));
     Finish(std::move(s));
-  };
+  }
 
  protected:
   template <class Request>
@@ -125,7 +125,7 @@ class ServerCallbackReaderWriter {
     // Default implementation that can/should be overridden
     Write(msg, std::move(options));
     Finish(std::move(s));
-  };
+  }
 
  protected:
   void BindReactor(ServerBidiReactor<Request, Response>* reactor) {

include/grpcpp/impl/codegen/server_interceptor.h

@@ -60,7 +60,7 @@ class ServerRpcInfo {
   /// Type categorizes RPCs by unary or streaming type
   enum class Type { UNARY, CLIENT_STREAMING, SERVER_STREAMING, BIDI_STREAMING };
 
-  ~ServerRpcInfo(){};
+  ~ServerRpcInfo() {}
 
   // Delete all copy and move constructors and assignments
   ServerRpcInfo(const ServerRpcInfo&) = delete;

include/grpcpp/security/credentials.h

@@ -95,7 +95,7 @@ class ChannelCredentials : private GrpcLibraryCodegen {
           std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>
           interceptor_creators) {
     return nullptr;
-  };
+  }
 };
 
 /// A call credentials object encapsulates the state needed by a client to

include/grpcpp/server.h

@@ -189,7 +189,7 @@ class Server : public ServerInterface, private GrpcLibraryCodegen {
   /// \param num_cqs How many completion queues does \a cqs hold.
   void Start(ServerCompletionQueue** cqs, size_t num_cqs) override;
 
-  grpc_server* server() override { return server_; };
+  grpc_server* server() override { return server_; }
 
  private:
   std::vector<std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>*
@@ -223,7 +223,7 @@ class Server : public ServerInterface, private GrpcLibraryCodegen {
 
   int max_receive_message_size() const override {
     return max_receive_message_size_;
-  };
+  }
 
   CompletionQueue* CallbackCQ() override;
 

package.xml

@@ -105,8 +105,6 @@
     <file baseinstalldir="/" name="src/core/lib/gpr/useful.h" role="src" />
     <file baseinstalldir="/" name="src/core/lib/gprpp/abstract.h" role="src" />
     <file baseinstalldir="/" name="src/core/lib/gprpp/atomic.h" role="src" />
-    <file baseinstalldir="/" name="src/core/lib/gprpp/atomic_with_atm.h" role="src" />
-    <file baseinstalldir="/" name="src/core/lib/gprpp/atomic_with_std.h" role="src" />
     <file baseinstalldir="/" name="src/core/lib/gprpp/fork.h" role="src" />
     <file baseinstalldir="/" name="src/core/lib/gprpp/manual_constructor.h" role="src" />
     <file baseinstalldir="/" name="src/core/lib/gprpp/memory.h" role="src" />

src/compiler/protobuf_plugin.h

@@ -108,11 +108,11 @@ class ProtoBufService : public grpc_generator::Service {
 
   grpc::string name() const { return service_->name(); }
 
-  int method_count() const { return service_->method_count(); };
+  int method_count() const { return service_->method_count(); }
   std::unique_ptr<const grpc_generator::Method> method(int i) const {
     return std::unique_ptr<const grpc_generator::Method>(
         new ProtoBufMethod(service_->method(i)));
-  };
+  }
 
   grpc::string GetLeadingComments(const grpc::string prefix) const {
     return GetCommentsHelper(service_, true, prefix);
@@ -166,7 +166,7 @@ class ProtoBufFile : public grpc_generator::File {
 
   grpc::string additional_headers() const { return ""; }
 
-  int service_count() const { return file_->service_count(); };
+  int service_count() const { return file_->service_count(); }
   std::unique_ptr<const grpc_generator::Service> service(int i) const {
     return std::unique_ptr<const grpc_generator::Service>(
         new ProtoBufService(file_->service(i)));

src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc

@@ -596,7 +596,7 @@ void GrpcLb::BalancerCallState::StartQuery() {
   call_error = grpc_call_start_batch_and_execute(
       lb_call_, ops, (size_t)(op - ops), &lb_on_balancer_status_received_);
   GPR_ASSERT(GRPC_CALL_OK == call_error);
-};
+}
 
 void GrpcLb::BalancerCallState::ScheduleNextClientLoadReportLocked() {
   const grpc_millis next_client_load_report_time =

src/core/ext/filters/load_reporting/server_load_reporting_filter.cc

@@ -342,8 +342,8 @@ bool MaybeAddServerLoadReportingFilter(const grpc_channel_args& args) {
 // time if we build with the filter target.
 struct ServerLoadReportingFilterStaticRegistrar {
   ServerLoadReportingFilterStaticRegistrar() {
-    static std::atomic_bool registered{false};
+    static grpc_core::Atomic<bool> registered{false};
-    if (registered) return;
+    if (registered.Load(grpc_core::MemoryOrder::ACQUIRE)) return;
     RegisterChannelFilter<ServerLoadReportingChannelData,
                           ServerLoadReportingCallData>(
         "server_load_reporting", GRPC_SERVER_CHANNEL, INT_MAX,
@@ -356,7 +356,7 @@ struct ServerLoadReportingFilterStaticRegistrar {
     ::grpc::load_reporter::MeasureEndBytesReceived();
     ::grpc::load_reporter::MeasureEndLatencyMs();
     ::grpc::load_reporter::MeasureOtherCallMetric();
-    registered = true;
+    registered.Store(true, grpc_core::MemoryOrder::RELEASE);
   }
 } server_load_reporting_filter_static_registrar;
 

src/core/lib/gprpp/atomic.h

@@ -21,10 +21,78 @@
 
 #include <grpc/support/port_platform.h>
 
-#ifdef GPR_HAS_CXX11_ATOMIC
+#include <atomic>
-#include "src/core/lib/gprpp/atomic_with_std.h"
+
-#else
+namespace grpc_core {
-#include "src/core/lib/gprpp/atomic_with_atm.h"
+
-#endif
+enum class MemoryOrder {
+  RELAXED = std::memory_order_relaxed,
+  CONSUME = std::memory_order_consume,
+  ACQUIRE = std::memory_order_acquire,
+  RELEASE = std::memory_order_release,
+  ACQ_REL = std::memory_order_acq_rel,
+  SEQ_CST = std::memory_order_seq_cst
+};
+
+template <typename T>
+class Atomic {
+ public:
+  explicit Atomic(T val = T()) : storage_(val) {}
+
+  T Load(MemoryOrder order) const {
+    return storage_.load(static_cast<std::memory_order>(order));
+  }
+
+  void Store(T val, MemoryOrder order) {
+    storage_.store(val, static_cast<std::memory_order>(order));
+  }
+
+  bool CompareExchangeWeak(T* expected, T desired, MemoryOrder success,
+                           MemoryOrder failure) {
+    return GPR_ATM_INC_CAS_THEN(
+        storage_.compare_exchange_weak(*expected, desired, success, failure));
+  }
+
+  bool CompareExchangeStrong(T* expected, T desired, MemoryOrder success,
+                             MemoryOrder failure) {
+    return GPR_ATM_INC_CAS_THEN(storage_.compare_exchange_weak(
+        *expected, desired, static_cast<std::memory_order>(success),
+        static_cast<std::memory_order>(failure)));
+  }
+
+  template <typename Arg>
+  T FetchAdd(Arg arg, MemoryOrder order = MemoryOrder::SEQ_CST) {
+    return GPR_ATM_INC_ADD_THEN(storage_.fetch_add(
+        static_cast<Arg>(arg), static_cast<std::memory_order>(order)));
+  }
+
+  template <typename Arg>
+  T FetchSub(Arg arg, MemoryOrder order = MemoryOrder::SEQ_CST) {
+    return GPR_ATM_INC_ADD_THEN(storage_.fetch_sub(
+        static_cast<Arg>(arg), static_cast<std::memory_order>(order)));
+  }
+
+  // Atomically increment a counter only if the counter value is not zero.
+  // Returns true if increment took place; false if counter is zero.
+  bool IncrementIfNonzero(MemoryOrder load_order = MemoryOrder::ACQ_REL) {
+    T count = storage_.load(static_cast<std::memory_order>(load_order));
+    do {
+      // If zero, we are done (without an increment). If not, we must do a CAS
+      // to maintain the contract: do not increment the counter if it is already
+      // zero
+      if (count == 0) {
+        return false;
+      }
+    } while (!storage_.AtomicCompareExchangeWeak(
+        &count, count + 1, static_cast<std::memory_order>(MemoryOrder::ACQ_REL),
+        static_cast<std::memory_order>(load_order)));
+    return true;
+  }
+
+ private:
+  std::atomic<T> storage_;
+};
+
+}  // namespace grpc_core
 
 #endif /* GRPC_CORE_LIB_GPRPP_ATOMIC_H */

src/core/lib/gprpp/atomic_with_atm.h

@@ -1,57 +0,0 @@
-/*
- *
- * Copyright 2017 gRPC authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-#ifndef GRPC_CORE_LIB_GPRPP_ATOMIC_WITH_ATM_H
-#define GRPC_CORE_LIB_GPRPP_ATOMIC_WITH_ATM_H
-
-#include <grpc/support/port_platform.h>
-
-#include <grpc/support/atm.h>
-
-namespace grpc_core {
-
-enum MemoryOrderRelaxed { memory_order_relaxed };
-
-template <class T>
-class atomic;
-
-template <>
-class atomic<bool> {
- public:
-  atomic() { gpr_atm_no_barrier_store(&x_, static_cast<gpr_atm>(false)); }
-  explicit atomic(bool x) {
-    gpr_atm_no_barrier_store(&x_, static_cast<gpr_atm>(x));
-  }
-
-  bool compare_exchange_strong(bool& expected, bool update, MemoryOrderRelaxed,
-                               MemoryOrderRelaxed) {
-    if (!gpr_atm_no_barrier_cas(&x_, static_cast<gpr_atm>(expected),
-                                static_cast<gpr_atm>(update))) {
-      expected = gpr_atm_no_barrier_load(&x_) != 0;
-      return false;
-    }
-    return true;
-  }
-
- private:
-  gpr_atm x_;
-};
-
-}  // namespace grpc_core
-
-#endif /* GRPC_CORE_LIB_GPRPP_ATOMIC_WITH_ATM_H */

src/core/lib/gprpp/atomic_with_std.h

@@ -1,35 +0,0 @@
-/*
- *
- * Copyright 2017 gRPC authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-#ifndef GRPC_CORE_LIB_GPRPP_ATOMIC_WITH_STD_H
-#define GRPC_CORE_LIB_GPRPP_ATOMIC_WITH_STD_H
-
-#include <grpc/support/port_platform.h>
-
-#include <atomic>
-
-namespace grpc_core {
-
-template <class T>
-using atomic = std::atomic<T>;
-
-typedef std::memory_order memory_order;
-
-}  // namespace grpc_core
-
-#endif /* GRPC_CORE_LIB_GPRPP_ATOMIC_WITH_STD_H */

src/core/lib/gprpp/ref_counted.h

@@ -31,6 +31,7 @@
 
 #include "src/core/lib/debug/trace.h"
 #include "src/core/lib/gprpp/abstract.h"
+#include "src/core/lib/gprpp/atomic.h"
 #include "src/core/lib/gprpp/debug_location.h"
 #include "src/core/lib/gprpp/memory.h"
 #include "src/core/lib/gprpp/ref_counted_ptr.h"
@@ -88,9 +89,7 @@ class RefCount {
   }
 
   // Increases the ref-count by `n`.
-  void Ref(Value n = 1) {
+  void Ref(Value n = 1) { value_.FetchAdd(n, MemoryOrder::RELAXED); }
-    GPR_ATM_INC_ADD_THEN(value_.fetch_add(n, std::memory_order_relaxed));
-  }
   void Ref(const DebugLocation& location, const char* reason, Value n = 1) {
 #ifndef NDEBUG
     if (location.Log() && trace_flag_ != nullptr && trace_flag_->enabled()) {
@@ -106,8 +105,7 @@ class RefCount {
   // Similar to Ref() with an assert on the ref-count being non-zero.
   void RefNonZero() {
 #ifndef NDEBUG
-    const Value prior =
+    const Value prior = value_.FetchAdd(1, MemoryOrder::RELAXED);
-        GPR_ATM_INC_ADD_THEN(value_.fetch_add(1, std::memory_order_relaxed));
     assert(prior > 0);
 #else
     Ref();
@@ -127,8 +125,7 @@ class RefCount {
 
   // Decrements the ref-count and returns true if the ref-count reaches 0.
   bool Unref() {
-    const Value prior =
+    const Value prior = value_.FetchSub(1, MemoryOrder::ACQ_REL);
-        GPR_ATM_INC_ADD_THEN(value_.fetch_sub(1, std::memory_order_acq_rel));
     GPR_DEBUG_ASSERT(prior > 0);
     return prior == 1;
   }
@@ -145,12 +142,12 @@ class RefCount {
   }
 
  private:
-  Value get() const { return value_.load(std::memory_order_relaxed); }
+  Value get() const { return value_.Load(MemoryOrder::RELAXED); }
 
 #ifndef NDEBUG
   TraceFlag* trace_flag_;
 #endif
-  std::atomic<Value> value_;
+  Atomic<Value> value_;
 };
 
 // A base class for reference-counted objects.

src/core/lib/gprpp/thd.h

@@ -112,19 +112,22 @@ class Thread {
   }
 
   /// The destructor is strictly optional; either the thread never came to life
-  /// and the constructor itself killed it or it has already been joined and
+  /// and the constructor itself killed it, or it has already been joined and
-  /// the Join function kills it. The destructor shouldn't have to do anything.
+  /// the Join function kills it, or it was detached (non-joinable) and it has
-  ~Thread() { GPR_ASSERT(impl_ == nullptr); }
+  /// run to completion and is now killing itself. The destructor shouldn't have
+  /// to do anything.
+  ~Thread() { GPR_ASSERT(!options_.joinable() || impl_ == nullptr); }
 
   void Start() {
     if (impl_ != nullptr) {
       GPR_ASSERT(state_ == ALIVE);
       state_ = STARTED;
       impl_->Start();
-      if (!options_.joinable()) {
+      // If the Thread is not joinable, then the impl_ will cause the deletion
-        state_ = DONE;
+      // of this Thread object when the thread function completes. Since no
-        impl_ = nullptr;
+      // other operation is allowed to a detached thread after Start, there is
-      }
+      // no need to change the value of the impl_ or state_ . The next operation
+      // on this object will be the deletion, which will trigger the destructor.
     } else {
       GPR_ASSERT(state_ == FAILED);
     }
@@ -140,7 +143,7 @@ class Thread {
     } else {
       GPR_ASSERT(state_ == FAILED);
     }
-  };
+  }
 
  private:
   Thread(const Thread&) = delete;

src/core/lib/iomgr/buffer_list.h

@@ -160,6 +160,6 @@ void grpc_tcp_set_write_timestamps_callback(void (*fn)(void*,
                                                        grpc_core::Timestamps*,
                                                        grpc_error* error));
 
-}; /* namespace grpc_core */
+} /* namespace grpc_core */
 
 #endif /* GRPC_CORE_LIB_IOMGR_BUFFER_LIST_H */

src/core/lib/surface/lame_client.cc

@@ -25,10 +25,9 @@
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 
-#include "src/core/lib/gprpp/atomic.h"
-
 #include "src/core/lib/channel/channel_stack.h"
 #include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gprpp/atomic.h"
 #include "src/core/lib/surface/api_trace.h"
 #include "src/core/lib/surface/call.h"
 #include "src/core/lib/surface/channel.h"
@@ -43,7 +42,7 @@ struct CallData {
   grpc_call_combiner* call_combiner;
   grpc_linked_mdelem status;
   grpc_linked_mdelem details;
-  grpc_core::atomic<bool> filled_metadata;
+  grpc_core::Atomic<bool> filled_metadata;
 };
 
 struct ChannelData {
@@ -54,9 +53,8 @@ struct ChannelData {
 static void fill_metadata(grpc_call_element* elem, grpc_metadata_batch* mdb) {
   CallData* calld = static_cast<CallData*>(elem->call_data);
   bool expected = false;
-  if (!calld->filled_metadata.compare_exchange_strong(
+  if (!calld->filled_metadata.CompareExchangeStrong(
-          expected, true, grpc_core::memory_order_relaxed,
+          &expected, true, MemoryOrder::RELAXED, MemoryOrder::RELAXED)) {
-          grpc_core::memory_order_relaxed)) {
     return;
   }
   ChannelData* chand = static_cast<ChannelData*>(elem->channel_data);

src/cpp/common/core_codegen.cc

@@ -81,7 +81,7 @@ void CoreCodegen::gpr_free(void* p) { return ::gpr_free(p); }
 void CoreCodegen::grpc_init() { ::grpc_init(); }
 void CoreCodegen::grpc_shutdown() { ::grpc_shutdown(); }
 
-void CoreCodegen::gpr_mu_init(gpr_mu* mu) { ::gpr_mu_init(mu); };
+void CoreCodegen::gpr_mu_init(gpr_mu* mu) { ::gpr_mu_init(mu); }
 void CoreCodegen::gpr_mu_destroy(gpr_mu* mu) { ::gpr_mu_destroy(mu); }
 void CoreCodegen::gpr_mu_lock(gpr_mu* mu) { ::gpr_mu_lock(mu); }
 void CoreCodegen::gpr_mu_unlock(gpr_mu* mu) { ::gpr_mu_unlock(mu); }

src/cpp/server/load_reporter/get_cpu_stats_linux.cc

@@ -32,7 +32,10 @@ std::pair<uint64_t, uint64_t> GetCpuStatsImpl() {
   FILE* fp;
   fp = fopen("/proc/stat", "r");
   uint64_t user, nice, system, idle;
-  fscanf(fp, "cpu %lu %lu %lu %lu", &user, &nice, &system, &idle);
+  if (fscanf(fp, "cpu %lu %lu %lu %lu", &user, &nice, &system, &idle) != 4) {
+    // Something bad happened with the information, so assume it's all invalid
+    user = nice = system = idle = 0;
+  }
   fclose(fp);
   busy = user + nice + system;
   total = busy + idle;

src/cpp/server/server_cc.cc

@@ -1251,6 +1251,6 @@ CompletionQueue* Server::CallbackCQ() {
     shutdown_callback->TakeCQ(callback_cq_);
   }
   return callback_cq_;
-};
+}
 
 }  // namespace grpc

src/python/grpcio_health_checking/grpc_health/v1/health.py

@@ -82,6 +82,7 @@ class HealthServicer(_health_pb2_grpc.HealthServicer):
         self._send_response_callbacks = {}
         self.Watch.__func__.experimental_non_blocking = experimental_non_blocking
         self.Watch.__func__.experimental_thread_pool = experimental_thread_pool
+        self._gracefully_shutting_down = False
 
     def _on_close_callback(self, send_response_callback, service):
 
@@ -135,9 +136,30 @@ class HealthServicer(_health_pb2_grpc.HealthServicer):
             the service
         """
         with self._lock:
+            if self._gracefully_shutting_down:
+                return
+            else:
                 self._server_status[service] = status
                 if service in self._send_response_callbacks:
                     for send_response_callback in self._send_response_callbacks[
                             service]:
                         send_response_callback(
                             _health_pb2.HealthCheckResponse(status=status))
+
+    def enter_graceful_shutdown(self):
+        """Permanently sets the status of all services to NOT_SERVING.
+
+        This should be invoked when the server is entering a graceful shutdown
+        period. After this method is invoked, future attempts to set the status
+        of a service will be ignored.
+
+        This is an EXPERIMENTAL API.
+        """
+        with self._lock:
+            if self._gracefully_shutting_down:
+                return
+            else:
+                for service in self._server_status:
+                    self.set(service,
+                             _health_pb2.HealthCheckResponse.NOT_SERVING)  # pylint: disable=no-member
+                self._gracefully_shutting_down = True

src/python/grpcio_tests/tests/health_check/_health_servicer_test.py

@@ -176,6 +176,7 @@ class BaseWatchTests(object):
             self.assertTrue(response_queue1.empty())
             self.assertTrue(response_queue2.empty())
 
+        @unittest.skip("https://github.com/grpc/grpc/issues/18127")
         def test_cancelled_watch_removed_from_watch_list(self):
             request = health_pb2.HealthCheckRequest(service=_WATCH_SERVICE)
             response_queue = queue.Queue()
@@ -194,7 +195,7 @@ class BaseWatchTests(object):
             thread.join()
 
             # Wait, if necessary, for serving thread to process client cancellation
-            timeout = time.time() + test_constants.SHORT_TIMEOUT
+            timeout = time.time() + test_constants.TIME_ALLOWANCE
             while time.time(
             ) < timeout and self._servicer._send_response_callbacks[_WATCH_SERVICE]:
                 time.sleep(1)
@@ -203,6 +204,30 @@ class BaseWatchTests(object):
                 'watch set should be empty')
             self.assertTrue(response_queue.empty())
 
+        def test_graceful_shutdown(self):
+            request = health_pb2.HealthCheckRequest(service='')
+            response_queue = queue.Queue()
+            rendezvous = self._stub.Watch(request)
+            thread = threading.Thread(
+                target=_consume_responses, args=(rendezvous, response_queue))
+            thread.start()
+
+            response = response_queue.get(timeout=test_constants.SHORT_TIMEOUT)
+            self.assertEqual(health_pb2.HealthCheckResponse.SERVING,
+                             response.status)
+
+            self._servicer.enter_graceful_shutdown()
+            response = response_queue.get(timeout=test_constants.SHORT_TIMEOUT)
+            self.assertEqual(health_pb2.HealthCheckResponse.NOT_SERVING,
+                             response.status)
+
+            # This should be a no-op.
+            self._servicer.set('', health_pb2.HealthCheckResponse.SERVING)
+
+            rendezvous.cancel()
+            thread.join()
+            self.assertTrue(response_queue.empty())
+
 
 class HealthServicerTest(BaseWatchTests.WatchTests):
 

test/cpp/end2end/BUILD

@@ -638,7 +638,6 @@ grpc_cc_test(
         "//:gpr",
         "//:grpc",
         "//:grpc++",
-        "//:grpc_cfstream",
         "//src/proto/grpc/testing:echo_messages_proto",
         "//src/proto/grpc/testing:echo_proto",
         "//src/proto/grpc/testing:simple_messages_proto",

test/cpp/qps/client_callback.cc

@@ -253,18 +253,20 @@ class CallbackStreamingPingPongReactor final
       : client_(client), ctx_(std::move(ctx)), messages_issued_(0) {}
 
   void StartNewRpc() {
-    if (client_->ThreadCompleted()) return;
     ctx_->stub_->experimental_async()->StreamingCall(&(ctx_->context_), this);
     write_time_ = UsageTimer::Now();
     StartWrite(client_->request());
+    writes_done_started_.clear();
     StartCall();
   }
 
   void OnWriteDone(bool ok) override {
-    if (!ok || client_->ThreadCompleted()) {
+    if (!ok) {
-      if (!ok) gpr_log(GPR_ERROR, "Error writing RPC");
+      gpr_log(GPR_ERROR, "Error writing RPC");
+    }
+    if ((!ok || client_->ThreadCompleted()) &&
+        !writes_done_started_.test_and_set()) {
       StartWritesDone();
-      return;
     }
     StartRead(&ctx_->response_);
   }
@@ -278,7 +280,9 @@ class CallbackStreamingPingPongReactor final
       if (!ok) {
         gpr_log(GPR_ERROR, "Error reading RPC");
       }
+      if (!writes_done_started_.test_and_set()) {
         StartWritesDone();
+      }
       return;
     }
     write_time_ = UsageTimer::Now();
@@ -295,8 +299,6 @@ class CallbackStreamingPingPongReactor final
   }
 
   void ScheduleRpc() {
-    if (client_->ThreadCompleted()) return;
-
     if (!client_->IsClosedLoop()) {
       gpr_timespec next_issue_time = client_->NextRPCIssueTime();
       // Start an alarm callback to run the internal callback after
@@ -312,6 +314,7 @@ class CallbackStreamingPingPongReactor final
 
   CallbackStreamingPingPongClient* client_;
   std::unique_ptr<CallbackClientRpcContext> ctx_;
+  std::atomic_flag writes_done_started_;
   Client::Thread* thread_ptr_;  // Needed to update histogram entries
   double write_time_;           // Track ping-pong round start time
   int messages_issued_;         // Messages issued by this stream

third_party/cares/cares.BUILD

@@ -3,6 +3,11 @@ config_setting(
     values = {"cpu": "darwin"},
 )
 
+config_setting(
+    name = "darwin_x86_64",
+    values = {"cpu": "darwin_x86_64"},
+)
+
 config_setting(
     name = "windows",
     values = {"cpu": "x64_windows"},
@@ -54,6 +59,7 @@ genrule(
         ":ios_armv7s": ["@com_github_grpc_grpc//third_party/cares:config_darwin/ares_config.h"],
         ":ios_arm64": ["@com_github_grpc_grpc//third_party/cares:config_darwin/ares_config.h"],
         ":darwin": ["@com_github_grpc_grpc//third_party/cares:config_darwin/ares_config.h"],
+        ":darwin_x86_64": ["@com_github_grpc_grpc//third_party/cares:config_darwin/ares_config.h"],
         ":windows": ["@com_github_grpc_grpc//third_party/cares:config_windows/ares_config.h"],
         ":android": ["@com_github_grpc_grpc//third_party/cares:config_android/ares_config.h"],
         "//conditions:default": ["@com_github_grpc_grpc//third_party/cares:config_linux/ares_config.h"],

tools/doxygen/Doxyfile.c++.internal

@@ -1068,8 +1068,6 @@ src/core/lib/gpr/tmpfile.h \
 src/core/lib/gpr/useful.h \
 src/core/lib/gprpp/abstract.h \
 src/core/lib/gprpp/atomic.h \
-src/core/lib/gprpp/atomic_with_atm.h \
-src/core/lib/gprpp/atomic_with_std.h \
 src/core/lib/gprpp/debug_location.h \
 src/core/lib/gprpp/fork.h \
 src/core/lib/gprpp/inlined_vector.h \

tools/doxygen/Doxyfile.core.internal

@@ -1157,8 +1157,6 @@ src/core/lib/gpr/wrap_memcpy.cc \
 src/core/lib/gprpp/README.md \
 src/core/lib/gprpp/abstract.h \
 src/core/lib/gprpp/atomic.h \
-src/core/lib/gprpp/atomic_with_atm.h \
-src/core/lib/gprpp/atomic_with_std.h \
 src/core/lib/gprpp/debug_location.h \
 src/core/lib/gprpp/fork.cc \
 src/core/lib/gprpp/fork.h \

tools/http2_interop/doc.go

@@ -1,3 +1,17 @@
+// Copyright 2019 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 // http2interop project doc.go
 
 /*

tools/http2_interop/frame.go

@@ -1,3 +1,17 @@
+// Copyright 2019 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package http2interop
 
 import (

tools/http2_interop/frameheader.go

@@ -1,3 +1,17 @@
+// Copyright 2019 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package http2interop
 
 import (

tools/http2_interop/goaway.go

@@ -1,3 +1,17 @@
+// Copyright 2019 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package http2interop
 
 import (

tools/http2_interop/http1frame.go

@@ -1,3 +1,17 @@
+// Copyright 2019 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package http2interop
 
 import (

tools/http2_interop/http2interop.go

@@ -1,3 +1,17 @@
+// Copyright 2019 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package http2interop
 
 import (

tools/http2_interop/http2interop_test.go

@@ -1,3 +1,17 @@
+// Copyright 2019 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package http2interop
 
 import (

tools/http2_interop/ping.go

@@ -1,3 +1,17 @@
+// Copyright 2019 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package http2interop
 
 import (

tools/http2_interop/s6.5.go

@@ -1,3 +1,17 @@
+// Copyright 2019 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package http2interop
 
 import (

tools/http2_interop/s6.5_test.go

@@ -1,3 +1,17 @@
+// Copyright 2019 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package http2interop
 
 import (

tools/http2_interop/settings.go

@@ -1,3 +1,17 @@
+// Copyright 2019 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package http2interop
 
 import (

tools/http2_interop/testsuite.go

@@ -1,3 +1,17 @@
+// Copyright 2019 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package http2interop
 
 import (

tools/http2_interop/unknownframe.go

@@ -1,3 +1,17 @@
+// Copyright 2019 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package http2interop
 
 import (

tools/internal_ci/linux/grpc_flaky_network_in_docker.sh

@@ -28,4 +28,4 @@ cd /var/local/git/grpc/test/cpp/end2end
 # iptables is used to drop traffic between client and server
 apt-get install -y iptables
 
-bazel test --spawn_strategy=standalone --genrule_strategy=standalone --test_output=all :flaky_network_test
+bazel test --spawn_strategy=standalone --genrule_strategy=standalone --test_output=all :flaky_network_test --test_env=GRPC_VERBOSITY=debug --test_env=GRPC_TRACE=channel,client_channel,call_error,connectivity_state

tools/run_tests/generated/sources_and_headers.json

@@ -9288,8 +9288,6 @@
       "src/core/lib/gpr/useful.h", 
       "src/core/lib/gprpp/abstract.h", 
       "src/core/lib/gprpp/atomic.h", 
-      "src/core/lib/gprpp/atomic_with_atm.h", 
-      "src/core/lib/gprpp/atomic_with_std.h", 
       "src/core/lib/gprpp/fork.h", 
       "src/core/lib/gprpp/manual_constructor.h", 
       "src/core/lib/gprpp/memory.h", 
@@ -9336,8 +9334,6 @@
       "src/core/lib/gpr/useful.h", 
       "src/core/lib/gprpp/abstract.h", 
       "src/core/lib/gprpp/atomic.h", 
-      "src/core/lib/gprpp/atomic_with_atm.h", 
-      "src/core/lib/gprpp/atomic_with_std.h", 
       "src/core/lib/gprpp/fork.h", 
       "src/core/lib/gprpp/manual_constructor.h", 
       "src/core/lib/gprpp/memory.h",