Initial scaffolding

include/grpc/grpc.h

@@ -450,6 +450,10 @@ GRPCAPI void grpc_resource_quota_unref(grpc_resource_quota* resource_quota);
 GRPCAPI void grpc_resource_quota_resize(grpc_resource_quota* resource_quota,
                                         size_t new_size);
 
+/** Update the size of the maximum number of threads allowed */
+GRPCAPI void grpc_resource_quota_set_max_threads(
+    grpc_resource_quota* resource_quota, int new_max_threads);
+
 /** Fetch a vtable for a grpc_channel_arg that points to a grpc_resource_quota
  */
 GRPCAPI const grpc_arg_pointer_vtable* grpc_resource_quota_arg_vtable(void);

include/grpcpp/resource_quota.h

@@ -44,6 +44,16 @@ class ResourceQuota final : private GrpcLibraryCodegen {
   /// No time bound is given for this to occur however.
   ResourceQuota& Resize(size_t new_size);
 
+  /// Set the max number of threads that can be allocated from this
+  /// ResourceQuota object.
+  ///
+  /// If the new_max_threads value is smaller than the current value, no new
+  /// threads are allocated until the number of active threads fall below
+  /// new_max_threads. There is no time bound on when this may happen i.e none
+  /// of the current threads are forcefully destroyed and all threads run their
+  /// normal course.
+  ResourceQuota& SetMaxThreads(int new_max_threads);
+
   grpc_resource_quota* c_resource_quota() const { return impl_; }
 
  private:

src/core/lib/iomgr/resource_quota.cc

@@ -96,6 +96,9 @@ struct grpc_resource_user {
      list, false otherwise */
   bool added_to_free_pool;
 
+  /* The number of threads currently allocated to this resource user */
+  gpr_atm num_threads;
+
   /* Reclaimers: index 0 is the benign reclaimer, 1 is the destructive reclaimer
    */
   grpc_closure* reclaimers[2];
@@ -135,12 +138,21 @@ struct grpc_resource_quota {
 
   gpr_atm last_size;
 
+  /* Max number of threads allowed */
+  int max_threads;
+
+  /* Number of threads currently allocated via this resource_quota object */
+  gpr_atm num_threads;
+
   /* Has rq_step been scheduled to occur? */
   bool step_scheduled;
+
   /* Are we currently reclaiming memory */
   bool reclaiming;
+
   /* Closure around rq_step */
   grpc_closure rq_step_closure;
+
   /* Closure around rq_reclamation_done */
   grpc_closure rq_reclamation_done_closure;
 
@@ -594,6 +606,8 @@ grpc_resource_quota* grpc_resource_quota_create(const char* name) {
   resource_quota->free_pool = INT64_MAX;
   resource_quota->size = INT64_MAX;
   gpr_atm_no_barrier_store(&resource_quota->last_size, GPR_ATM_MAX);
+  resource_quota->max_threads = INT_MAX;
+  gpr_atm_no_barrier_store(&resource_quota->num_threads, 0);
   resource_quota->step_scheduled = false;
   resource_quota->reclaiming = false;
   gpr_atm_no_barrier_store(&resource_quota->memory_usage_estimation, 0);
@@ -646,6 +660,10 @@ double grpc_resource_quota_get_memory_pressure(
          (static_cast<double>(MEMORY_USAGE_ESTIMATION_MAX));
 }
 
+/* Public API */
+void grpc_resource_quota_set_max_threads(grpc_resource_quota* resource_quota,
+                                         int new_max_threads) {}
+
 /* Public API */
 void grpc_resource_quota_resize(grpc_resource_quota* resource_quota,
                                 size_t size) {
@@ -731,6 +749,7 @@ grpc_resource_user* grpc_resource_user_create(
   grpc_closure_list_init(&resource_user->on_allocated);
   resource_user->allocating = false;
   resource_user->added_to_free_pool = false;
+  gpr_atm_no_barrier_store(&resource_user->num_threads, 0);
   resource_user->reclaimers[0] = nullptr;
   resource_user->reclaimers[1] = nullptr;
   resource_user->new_reclaimers[0] = nullptr;
@@ -785,6 +804,14 @@ void grpc_resource_user_shutdown(grpc_resource_user* resource_user) {
   }
 }
 
+bool grpc_resource_user_alloc_threads(grpc_resource_user* resource_user,
+                                      int thd_count) {
+  return true;
+}
+
+void grpc_resource_user_free_threads(grpc_resource_user* resource_user,
+                                     int thd_count) {}
+
 void grpc_resource_user_alloc(grpc_resource_user* resource_user, size_t size,
                               grpc_closure* optional_on_done) {
   gpr_mu_lock(&resource_user->mu);

src/core/lib/iomgr/resource_quota.h

@@ -93,6 +93,22 @@ void grpc_resource_user_ref(grpc_resource_user* resource_user);
 void grpc_resource_user_unref(grpc_resource_user* resource_user);
 void grpc_resource_user_shutdown(grpc_resource_user* resource_user);
 
+/* Attempts to get quota (from the resource_user) to create 'thd_count' number
+ * of threads. Returns true if successful (i.e the caller is now free to create
+ * 'thd_count' number of threads or false if quota is not available */
+bool grpc_resource_user_alloc_threads(grpc_resource_user* resource_user,
+                                      int thd_count);
+/* Releases 'thd_count' worth of quota back to the resource user. The quota
+ * should have been previously obtained successfully by calling
+ * grpc_resource_user_alloc_threads().
+ *
+ * Note: There need not be an exact one-to-one correspondence between
+ * grpc_resource_user_alloc_threads() and grpc_resource_user_free_threads()
+ * calls. The only requirement is that the number of threads allocated should
+ * all be eventually released */
+void grpc_resource_user_free_threads(grpc_resource_user* resource_user,
+                                     int thd_count);
+
 /* Allocate from the resource user (and its quota).
    If optional_on_done is NULL, then allocate immediately. This may push the
    quota over-limit, at which point reclamation will kick in.

src/cpp/common/resource_quota_cc.cc

@@ -33,4 +33,8 @@ ResourceQuota& ResourceQuota::Resize(size_t new_size) {
   return *this;
 }
 
+ResourceQuota& ResourceQuota::SetMaxThreads(int new_max_threads) {
+  grpc_resource_quota_set_max_threads(impl_, new_max_threads);
+  return *this;
+}
 }  // namespace grpc