aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/lib
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/lib')
-rw-r--r--src/core/lib/gpr/arena.cc77
-rw-r--r--src/core/lib/gprpp/fork.cc74
-rw-r--r--src/core/lib/gprpp/fork.h17
-rw-r--r--src/core/lib/iomgr/ev_epoll1_linux.cc72
-rw-r--r--src/core/lib/iomgr/ev_poll_posix.cc114
-rw-r--r--src/core/lib/iomgr/ev_posix.cc7
-rw-r--r--src/core/lib/iomgr/exec_ctx.cc6
-rw-r--r--src/core/lib/iomgr/exec_ctx.h2
-rw-r--r--src/core/lib/iomgr/fork_posix.cc13
-rw-r--r--src/core/lib/iomgr/port.h12
-rw-r--r--src/core/lib/iomgr/socket_mutator.cc2
-rw-r--r--src/core/lib/iomgr/socket_mutator.h2
-rw-r--r--src/core/lib/iomgr/timer.h5
-rw-r--r--src/core/lib/security/credentials/jwt/json_token.h2
-rw-r--r--src/core/lib/security/credentials/jwt/jwt_verifier.cc2
-rw-r--r--src/core/lib/security/credentials/oauth2/oauth2_credentials.cc2
-rw-r--r--src/core/lib/surface/call.cc2
-rw-r--r--src/core/lib/surface/init.h1
-rw-r--r--src/core/lib/transport/service_config.cc4
-rw-r--r--src/core/lib/transport/service_config.h6
20 files changed, 318 insertions, 104 deletions
diff --git a/src/core/lib/gpr/arena.cc b/src/core/lib/gpr/arena.cc
index e30b297aea..77f9357146 100644
--- a/src/core/lib/gpr/arena.cc
+++ b/src/core/lib/gpr/arena.cc
@@ -77,16 +77,16 @@ void* gpr_arena_alloc(gpr_arena* arena, size_t size) {
// would allow us to use the alignment actually needed by the caller.
typedef struct zone {
- size_t size_begin; // All the space we have set aside for allocations up
- // until this zone.
- size_t size_end; // size_end = size_begin plus all the space we set aside for
- // allocations in zone z itself.
zone* next;
} zone;
struct gpr_arena {
- gpr_atm size_so_far;
+ // Keep track of the total used size. We use this in our call sizing
+ // historesis.
+ gpr_atm total_used;
+ size_t initial_zone_size;
zone initial_zone;
+ zone* last_zone;
gpr_mu arena_growth_mutex;
};
@@ -100,14 +100,15 @@ gpr_arena* gpr_arena_create(size_t initial_size) {
initial_size = GPR_ROUND_UP_TO_ALIGNMENT_SIZE(initial_size);
gpr_arena* a = static_cast<gpr_arena*>(zalloc_aligned(
GPR_ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(gpr_arena)) + initial_size));
- a->initial_zone.size_end = initial_size;
+ a->initial_zone_size = initial_size;
+ a->last_zone = &a->initial_zone;
gpr_mu_init(&a->arena_growth_mutex);
return a;
}
size_t gpr_arena_destroy(gpr_arena* arena) {
gpr_mu_destroy(&arena->arena_growth_mutex);
- gpr_atm size = gpr_atm_no_barrier_load(&arena->size_so_far);
+ gpr_atm size = gpr_atm_no_barrier_load(&arena->total_used);
zone* z = arena->initial_zone.next;
gpr_free_aligned(arena);
while (z) {
@@ -120,55 +121,25 @@ size_t gpr_arena_destroy(gpr_arena* arena) {
void* gpr_arena_alloc(gpr_arena* arena, size_t size) {
size = GPR_ROUND_UP_TO_ALIGNMENT_SIZE(size);
- size_t previous_size_of_arena_allocations = static_cast<size_t>(
- gpr_atm_no_barrier_fetch_add(&arena->size_so_far, size));
- size_t updated_size_of_arena_allocations =
- previous_size_of_arena_allocations + size;
- zone* z = &arena->initial_zone;
- // Check to see if the allocation isn't able to end in the initial zone.
- // This statement is true only in the uncommon case because of our arena
- // sizing historesis (that is, most calls should have a large enough initial
- // zone and will not need to grow the arena).
- if (updated_size_of_arena_allocations > z->size_end) {
- // Find a zone to fit this allocation
+ size_t begin = gpr_atm_no_barrier_fetch_add(&arena->total_used, size);
+ if (begin + size <= arena->initial_zone_size) {
+ return reinterpret_cast<char*>(arena) +
+ GPR_ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(gpr_arena)) + begin;
+ } else {
+ // If the allocation isn't able to end in the initial zone, create a new
+ // zone for this allocation, and any unused space in the initial zone is
+ // wasted. This overflowing and wasting is uncommon because of our arena
+ // sizing historesis (that is, most calls should have a large enough initial
+ // zone and will not need to grow the arena).
gpr_mu_lock(&arena->arena_growth_mutex);
- while (updated_size_of_arena_allocations > z->size_end) {
- if (z->next == nullptr) {
- // Note that we do an extra increment of size_so_far to prevent multiple
- // simultaneous callers from stepping on each other. However, this extra
- // increment means some space in the arena is wasted.
- // So whenever we need to allocate x bytes and there are x - n (where
- // n > 0) remaining in the current zone, we will waste x bytes (x - n
- // in the current zone and n in the new zone).
- previous_size_of_arena_allocations = static_cast<size_t>(
- gpr_atm_no_barrier_fetch_add(&arena->size_so_far, size));
- updated_size_of_arena_allocations =
- previous_size_of_arena_allocations + size;
- size_t next_z_size = updated_size_of_arena_allocations;
- z->next = static_cast<zone*>(zalloc_aligned(
- GPR_ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(zone)) + next_z_size));
- z->next->size_begin = z->size_end;
- z->next->size_end = z->size_end + next_z_size;
- }
- z = z->next;
- }
+ zone* z = static_cast<zone*>(
+ zalloc_aligned(GPR_ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(zone)) + size));
+ arena->last_zone->next = z;
+ arena->last_zone = z;
gpr_mu_unlock(&arena->arena_growth_mutex);
+ return reinterpret_cast<char*>(z) +
+ GPR_ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(zone));
}
- GPR_ASSERT(previous_size_of_arena_allocations >= z->size_begin);
- GPR_ASSERT(updated_size_of_arena_allocations <= z->size_end);
- // Skip the first part of the zone, which just contains tracking information.
- // For the initial zone, this is the gpr_arena struct and for any other zone,
- // it's the zone struct.
- char* start_of_allocation_space =
- (z == &arena->initial_zone)
- ? reinterpret_cast<char*>(arena) +
- GPR_ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(gpr_arena))
- : reinterpret_cast<char*>(z) +
- GPR_ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(zone));
- // previous_size_of_arena_allocations - size_begin is how many bytes have been
- // allocated into the current zone
- return start_of_allocation_space + previous_size_of_arena_allocations -
- z->size_begin;
}
#endif // SIMPLE_ARENA_FOR_DEBUGGING
diff --git a/src/core/lib/gprpp/fork.cc b/src/core/lib/gprpp/fork.cc
index f6d9a87d2c..3b9c16510a 100644
--- a/src/core/lib/gprpp/fork.cc
+++ b/src/core/lib/gprpp/fork.cc
@@ -157,11 +157,11 @@ class ThreadState {
} // namespace
void Fork::GlobalInit() {
- if (!overrideEnabled_) {
+ if (!override_enabled_) {
#ifdef GRPC_ENABLE_FORK_SUPPORT
- supportEnabled_ = true;
+ support_enabled_ = true;
#else
- supportEnabled_ = false;
+ support_enabled_ = false;
#endif
bool env_var_set = false;
char* env = gpr_getenv("GRPC_ENABLE_FORK_SUPPORT");
@@ -172,7 +172,7 @@ void Fork::GlobalInit() {
"False", "FALSE", "0"};
for (size_t i = 0; i < GPR_ARRAY_SIZE(truthy); i++) {
if (0 == strcmp(env, truthy[i])) {
- supportEnabled_ = true;
+ support_enabled_ = true;
env_var_set = true;
break;
}
@@ -180,7 +180,7 @@ void Fork::GlobalInit() {
if (!env_var_set) {
for (size_t i = 0; i < GPR_ARRAY_SIZE(falsey); i++) {
if (0 == strcmp(env, falsey[i])) {
- supportEnabled_ = false;
+ support_enabled_ = false;
env_var_set = true;
break;
}
@@ -189,72 +189,80 @@ void Fork::GlobalInit() {
gpr_free(env);
}
}
- if (supportEnabled_) {
- execCtxState_ = grpc_core::New<internal::ExecCtxState>();
- threadState_ = grpc_core::New<internal::ThreadState>();
+ if (support_enabled_) {
+ exec_ctx_state_ = grpc_core::New<internal::ExecCtxState>();
+ thread_state_ = grpc_core::New<internal::ThreadState>();
}
}
void Fork::GlobalShutdown() {
- if (supportEnabled_) {
- grpc_core::Delete(execCtxState_);
- grpc_core::Delete(threadState_);
+ if (support_enabled_) {
+ grpc_core::Delete(exec_ctx_state_);
+ grpc_core::Delete(thread_state_);
}
}
-bool Fork::Enabled() { return supportEnabled_; }
+bool Fork::Enabled() { return support_enabled_; }
// Testing Only
void Fork::Enable(bool enable) {
- overrideEnabled_ = true;
- supportEnabled_ = enable;
+ override_enabled_ = true;
+ support_enabled_ = enable;
}
void Fork::IncExecCtxCount() {
- if (supportEnabled_) {
- execCtxState_->IncExecCtxCount();
+ if (support_enabled_) {
+ exec_ctx_state_->IncExecCtxCount();
}
}
void Fork::DecExecCtxCount() {
- if (supportEnabled_) {
- execCtxState_->DecExecCtxCount();
+ if (support_enabled_) {
+ exec_ctx_state_->DecExecCtxCount();
}
}
+void Fork::SetResetChildPollingEngineFunc(
+ Fork::child_postfork_func reset_child_polling_engine) {
+ reset_child_polling_engine_ = reset_child_polling_engine;
+}
+Fork::child_postfork_func Fork::GetResetChildPollingEngineFunc() {
+ return reset_child_polling_engine_;
+}
+
bool Fork::BlockExecCtx() {
- if (supportEnabled_) {
- return execCtxState_->BlockExecCtx();
+ if (support_enabled_) {
+ return exec_ctx_state_->BlockExecCtx();
}
return false;
}
void Fork::AllowExecCtx() {
- if (supportEnabled_) {
- execCtxState_->AllowExecCtx();
+ if (support_enabled_) {
+ exec_ctx_state_->AllowExecCtx();
}
}
void Fork::IncThreadCount() {
- if (supportEnabled_) {
- threadState_->IncThreadCount();
+ if (support_enabled_) {
+ thread_state_->IncThreadCount();
}
}
void Fork::DecThreadCount() {
- if (supportEnabled_) {
- threadState_->DecThreadCount();
+ if (support_enabled_) {
+ thread_state_->DecThreadCount();
}
}
void Fork::AwaitThreads() {
- if (supportEnabled_) {
- threadState_->AwaitThreads();
+ if (support_enabled_) {
+ thread_state_->AwaitThreads();
}
}
-internal::ExecCtxState* Fork::execCtxState_ = nullptr;
-internal::ThreadState* Fork::threadState_ = nullptr;
-bool Fork::supportEnabled_ = false;
-bool Fork::overrideEnabled_ = false;
-
+internal::ExecCtxState* Fork::exec_ctx_state_ = nullptr;
+internal::ThreadState* Fork::thread_state_ = nullptr;
+bool Fork::support_enabled_ = false;
+bool Fork::override_enabled_ = false;
+Fork::child_postfork_func Fork::reset_child_polling_engine_ = nullptr;
} // namespace grpc_core
diff --git a/src/core/lib/gprpp/fork.h b/src/core/lib/gprpp/fork.h
index 123e22c4c6..5a7404f0d9 100644
--- a/src/core/lib/gprpp/fork.h
+++ b/src/core/lib/gprpp/fork.h
@@ -33,6 +33,8 @@ class ThreadState;
class Fork {
public:
+ typedef void (*child_postfork_func)(void);
+
static void GlobalInit();
static void GlobalShutdown();
@@ -46,6 +48,12 @@ class Fork {
// Decrement the count of active ExecCtxs
static void DecExecCtxCount();
+ // Provide a function that will be invoked in the child's postfork handler to
+ // reset the polling engine's internal state.
+ static void SetResetChildPollingEngineFunc(
+ child_postfork_func reset_child_polling_engine);
+ static child_postfork_func GetResetChildPollingEngineFunc();
+
// Check if there is a single active ExecCtx
// (the one used to invoke this function). If there are more,
// return false. Otherwise, return true and block creation of
@@ -68,10 +76,11 @@ class Fork {
static void Enable(bool enable);
private:
- static internal::ExecCtxState* execCtxState_;
- static internal::ThreadState* threadState_;
- static bool supportEnabled_;
- static bool overrideEnabled_;
+ static internal::ExecCtxState* exec_ctx_state_;
+ static internal::ThreadState* thread_state_;
+ static bool support_enabled_;
+ static bool override_enabled_;
+ static child_postfork_func reset_child_polling_engine_;
};
} // namespace grpc_core
diff --git a/src/core/lib/iomgr/ev_epoll1_linux.cc b/src/core/lib/iomgr/ev_epoll1_linux.cc
index 66e0f1fd6d..aa5016bd8f 100644
--- a/src/core/lib/iomgr/ev_epoll1_linux.cc
+++ b/src/core/lib/iomgr/ev_epoll1_linux.cc
@@ -131,6 +131,13 @@ static void epoll_set_shutdown() {
* Fd Declarations
*/
+/* Only used when GRPC_ENABLE_FORK_SUPPORT=1 */
+struct grpc_fork_fd_list {
+ grpc_fd* fd;
+ grpc_fd* next;
+ grpc_fd* prev;
+};
+
struct grpc_fd {
int fd;
@@ -141,6 +148,9 @@ struct grpc_fd {
struct grpc_fd* freelist_next;
grpc_iomgr_object iomgr_object;
+
+ /* Only used when GRPC_ENABLE_FORK_SUPPORT=1 */
+ grpc_fork_fd_list* fork_fd_list;
};
static void fd_global_init(void);
@@ -256,6 +266,10 @@ static bool append_error(grpc_error** composite, grpc_error* error,
static grpc_fd* fd_freelist = nullptr;
static gpr_mu fd_freelist_mu;
+/* Only used when GRPC_ENABLE_FORK_SUPPORT=1 */
+static grpc_fd* fork_fd_list_head = nullptr;
+static gpr_mu fork_fd_list_mu;
+
static void fd_global_init(void) { gpr_mu_init(&fd_freelist_mu); }
static void fd_global_shutdown(void) {
@@ -269,6 +283,38 @@ static void fd_global_shutdown(void) {
gpr_mu_destroy(&fd_freelist_mu);
}
+static void fork_fd_list_add_grpc_fd(grpc_fd* fd) {
+ if (grpc_core::Fork::Enabled()) {
+ gpr_mu_lock(&fork_fd_list_mu);
+ fd->fork_fd_list =
+ static_cast<grpc_fork_fd_list*>(gpr_malloc(sizeof(grpc_fork_fd_list)));
+ fd->fork_fd_list->next = fork_fd_list_head;
+ fd->fork_fd_list->prev = nullptr;
+ if (fork_fd_list_head != nullptr) {
+ fork_fd_list_head->fork_fd_list->prev = fd;
+ }
+ fork_fd_list_head = fd;
+ gpr_mu_unlock(&fork_fd_list_mu);
+ }
+}
+
+static void fork_fd_list_remove_grpc_fd(grpc_fd* fd) {
+ if (grpc_core::Fork::Enabled()) {
+ gpr_mu_lock(&fork_fd_list_mu);
+ if (fork_fd_list_head == fd) {
+ fork_fd_list_head = fd->fork_fd_list->next;
+ }
+ if (fd->fork_fd_list->prev != nullptr) {
+ fd->fork_fd_list->prev->fork_fd_list->next = fd->fork_fd_list->next;
+ }
+ if (fd->fork_fd_list->next != nullptr) {
+ fd->fork_fd_list->next->fork_fd_list->prev = fd->fork_fd_list->prev;
+ }
+ gpr_free(fd->fork_fd_list);
+ gpr_mu_unlock(&fork_fd_list_mu);
+ }
+}
+
static grpc_fd* fd_create(int fd, const char* name, bool track_err) {
grpc_fd* new_fd = nullptr;
@@ -295,6 +341,7 @@ static grpc_fd* fd_create(int fd, const char* name, bool track_err) {
char* fd_name;
gpr_asprintf(&fd_name, "%s fd=%d", name, fd);
grpc_iomgr_register_object(&new_fd->iomgr_object, fd_name);
+ fork_fd_list_add_grpc_fd(new_fd);
#ifndef NDEBUG
if (grpc_trace_fd_refcount.enabled()) {
gpr_log(GPR_DEBUG, "FD %d %p create %s", fd, new_fd, fd_name);
@@ -361,6 +408,7 @@ static void fd_orphan(grpc_fd* fd, grpc_closure* on_done, int* release_fd,
GRPC_CLOSURE_SCHED(on_done, GRPC_ERROR_REF(error));
grpc_iomgr_unregister_object(&fd->iomgr_object);
+ fork_fd_list_remove_grpc_fd(fd);
fd->read_closure->DestroyEvent();
fd->write_closure->DestroyEvent();
fd->error_closure->DestroyEvent();
@@ -1190,6 +1238,10 @@ static void shutdown_engine(void) {
fd_global_shutdown();
pollset_global_shutdown();
epoll_set_shutdown();
+ if (grpc_core::Fork::Enabled()) {
+ gpr_mu_destroy(&fork_fd_list_mu);
+ grpc_core::Fork::SetResetChildPollingEngineFunc(nullptr);
+ }
}
static const grpc_event_engine_vtable vtable = {
@@ -1227,6 +1279,21 @@ static const grpc_event_engine_vtable vtable = {
shutdown_engine,
};
+/* Called by the child process's post-fork handler to close open fds, including
+ * the global epoll fd. This allows gRPC to shutdown in the child process
+ * without interfering with connections or RPCs ongoing in the parent. */
+static void reset_event_manager_on_fork() {
+ gpr_mu_lock(&fork_fd_list_mu);
+ while (fork_fd_list_head != nullptr) {
+ close(fork_fd_list_head->fd);
+ fork_fd_list_head->fd = -1;
+ fork_fd_list_head = fork_fd_list_head->fork_fd_list->next;
+ }
+ gpr_mu_unlock(&fork_fd_list_mu);
+ shutdown_engine();
+ grpc_init_epoll1_linux(true);
+}
+
/* It is possible that GLIBC has epoll but the underlying kernel doesn't.
* Create epoll_fd (epoll_set_init() takes care of that) to make sure epoll
* support is available */
@@ -1248,6 +1315,11 @@ const grpc_event_engine_vtable* grpc_init_epoll1_linux(bool explicit_request) {
return nullptr;
}
+ if (grpc_core::Fork::Enabled()) {
+ gpr_mu_init(&fork_fd_list_mu);
+ grpc_core::Fork::SetResetChildPollingEngineFunc(
+ reset_event_manager_on_fork);
+ }
return &vtable;
}
diff --git a/src/core/lib/iomgr/ev_poll_posix.cc b/src/core/lib/iomgr/ev_poll_posix.cc
index fb4c71ef71..16562538a6 100644
--- a/src/core/lib/iomgr/ev_poll_posix.cc
+++ b/src/core/lib/iomgr/ev_poll_posix.cc
@@ -60,6 +60,19 @@ typedef struct grpc_fd_watcher {
grpc_fd* fd;
} grpc_fd_watcher;
+typedef struct grpc_cached_wakeup_fd grpc_cached_wakeup_fd;
+
+/* Only used when GRPC_ENABLE_FORK_SUPPORT=1 */
+struct grpc_fork_fd_list {
+ /* Only one of fd or cached_wakeup_fd will be set. The unused field will be
+ set to nullptr. */
+ grpc_fd* fd;
+ grpc_cached_wakeup_fd* cached_wakeup_fd;
+
+ grpc_fork_fd_list* next;
+ grpc_fork_fd_list* prev;
+};
+
struct grpc_fd {
int fd;
/* refst format:
@@ -108,8 +121,18 @@ struct grpc_fd {
grpc_closure* on_done_closure;
grpc_iomgr_object iomgr_object;
+
+ /* Only used when GRPC_ENABLE_FORK_SUPPORT=1 */
+ grpc_fork_fd_list* fork_fd_list;
};
+/* True when GRPC_ENABLE_FORK_SUPPORT=1. We do not support fork with poll-cv */
+static bool track_fds_for_fork = false;
+
+/* Only used when GRPC_ENABLE_FORK_SUPPORT=1 */
+static grpc_fork_fd_list* fork_fd_list_head = nullptr;
+static gpr_mu fork_fd_list_mu;
+
/* Begin polling on an fd.
Registers that the given pollset is interested in this fd - so that if read
or writability interest changes, the pollset can be kicked to pick up that
@@ -156,6 +179,9 @@ static void fd_unref(grpc_fd* fd);
typedef struct grpc_cached_wakeup_fd {
grpc_wakeup_fd fd;
struct grpc_cached_wakeup_fd* next;
+
+ /* Only used when GRPC_ENABLE_FORK_SUPPORT=1 */
+ grpc_fork_fd_list* fork_fd_list;
} grpc_cached_wakeup_fd;
struct grpc_pollset_worker {
@@ -281,9 +307,61 @@ poll_hash_table poll_cache;
grpc_cv_fd_table g_cvfds;
/*******************************************************************************
- * fd_posix.c
+ * functions to track opened fds. No-ops unless track_fds_for_fork is true.
*/
+static void fork_fd_list_remove_node(grpc_fork_fd_list* node) {
+ if (track_fds_for_fork) {
+ gpr_mu_lock(&fork_fd_list_mu);
+ if (fork_fd_list_head == node) {
+ fork_fd_list_head = node->next;
+ }
+ if (node->prev != nullptr) {
+ node->prev->next = node->next;
+ }
+ if (node->next != nullptr) {
+ node->next->prev = node->prev;
+ }
+ gpr_free(node);
+ gpr_mu_unlock(&fork_fd_list_mu);
+ }
+}
+
+static void fork_fd_list_add_node(grpc_fork_fd_list* node) {
+ gpr_mu_lock(&fork_fd_list_mu);
+ node->next = fork_fd_list_head;
+ node->prev = nullptr;
+ if (fork_fd_list_head != nullptr) {
+ fork_fd_list_head->prev = node;
+ }
+ fork_fd_list_head = node;
+ gpr_mu_unlock(&fork_fd_list_mu);
+}
+
+static void fork_fd_list_add_grpc_fd(grpc_fd* fd) {
+ if (track_fds_for_fork) {
+ fd->fork_fd_list =
+ static_cast<grpc_fork_fd_list*>(gpr_malloc(sizeof(grpc_fork_fd_list)));
+ fd->fork_fd_list->fd = fd;
+ fd->fork_fd_list->cached_wakeup_fd = nullptr;
+ fork_fd_list_add_node(fd->fork_fd_list);
+ }
+}
+
+static void fork_fd_list_add_wakeup_fd(grpc_cached_wakeup_fd* fd) {
+ if (track_fds_for_fork) {
+ fd->fork_fd_list =
+ static_cast<grpc_fork_fd_list*>(gpr_malloc(sizeof(grpc_fork_fd_list)));
+ fd->fork_fd_list->cached_wakeup_fd = fd;
+ fd->fork_fd_list->fd = nullptr;
+ fork_fd_list_add_node(fd->fork_fd_list);
+ }
+}
+
+ /*******************************************************************************
+ * fd_posix.c
+ */
+
#ifndef NDEBUG
#define REF_BY(fd, n, reason) ref_by(fd, n, reason, __FILE__, __LINE__)
#define UNREF_BY(fd, n, reason) unref_by(fd, n, reason, __FILE__, __LINE__)
@@ -319,6 +397,7 @@ static void unref_by(grpc_fd* fd, int n) {
if (old == n) {
gpr_mu_destroy(&fd->mu);
grpc_iomgr_unregister_object(&fd->iomgr_object);
+ fork_fd_list_remove_node(fd->fork_fd_list);
if (fd->shutdown) GRPC_ERROR_UNREF(fd->shutdown_error);
gpr_free(fd);
} else {
@@ -347,6 +426,7 @@ static grpc_fd* fd_create(int fd, const char* name, bool track_err) {
gpr_asprintf(&name2, "%s fd=%d", name, fd);
grpc_iomgr_register_object(&r->iomgr_object, name2);
gpr_free(name2);
+ fork_fd_list_add_grpc_fd(r);
return r;
}
@@ -822,6 +902,7 @@ static void pollset_destroy(grpc_pollset* pollset) {
GPR_ASSERT(!pollset_has_workers(pollset));
while (pollset->local_wakeup_cache) {
grpc_cached_wakeup_fd* next = pollset->local_wakeup_cache->next;
+ fork_fd_list_remove_node(pollset->local_wakeup_cache->fork_fd_list);
grpc_wakeup_fd_destroy(&pollset->local_wakeup_cache->fd);
gpr_free(pollset->local_wakeup_cache);
pollset->local_wakeup_cache = next;
@@ -895,6 +976,7 @@ static grpc_error* pollset_work(grpc_pollset* pollset,
worker.wakeup_fd = static_cast<grpc_cached_wakeup_fd*>(
gpr_malloc(sizeof(*worker.wakeup_fd)));
error = grpc_wakeup_fd_init(&worker.wakeup_fd->fd);
+ fork_fd_list_add_wakeup_fd(worker.wakeup_fd);
if (error != GRPC_ERROR_NONE) {
GRPC_LOG_IF_ERROR("pollset_work", GRPC_ERROR_REF(error));
return error;
@@ -1705,6 +1787,10 @@ static void shutdown_engine(void) {
if (grpc_cv_wakeup_fds_enabled()) {
global_cv_fd_table_shutdown();
}
+ if (track_fds_for_fork) {
+ gpr_mu_destroy(&fork_fd_list_mu);
+ grpc_core::Fork::SetResetChildPollingEngineFunc(nullptr);
+ }
}
static const grpc_event_engine_vtable vtable = {
@@ -1742,6 +1828,26 @@ static const grpc_event_engine_vtable vtable = {
shutdown_engine,
};
+/* Called by the child process's post-fork handler to close open fds, including
+ * worker wakeup fds. This allows gRPC to shutdown in the child process without
+ * interfering with connections or RPCs ongoing in the parent. */
+static void reset_event_manager_on_fork() {
+ gpr_mu_lock(&fork_fd_list_mu);
+ while (fork_fd_list_head != nullptr) {
+ if (fork_fd_list_head->fd != nullptr) {
+ close(fork_fd_list_head->fd->fd);
+ fork_fd_list_head->fd->fd = -1;
+ } else {
+ close(fork_fd_list_head->cached_wakeup_fd->fd.read_fd);
+ fork_fd_list_head->cached_wakeup_fd->fd.read_fd = -1;
+ close(fork_fd_list_head->cached_wakeup_fd->fd.write_fd);
+ fork_fd_list_head->cached_wakeup_fd->fd.write_fd = -1;
+ }
+ fork_fd_list_head = fork_fd_list_head->next;
+ }
+ gpr_mu_unlock(&fork_fd_list_mu);
+}
+
const grpc_event_engine_vtable* grpc_init_poll_posix(bool explicit_request) {
if (!grpc_has_wakeup_fd()) {
gpr_log(GPR_ERROR, "Skipping poll because of no wakeup fd.");
@@ -1750,6 +1856,12 @@ const grpc_event_engine_vtable* grpc_init_poll_posix(bool explicit_request) {
if (!GRPC_LOG_IF_ERROR("pollset_global_init", pollset_global_init())) {
return nullptr;
}
+ if (grpc_core::Fork::Enabled()) {
+ track_fds_for_fork = true;
+ gpr_mu_init(&fork_fd_list_mu);
+ grpc_core::Fork::SetResetChildPollingEngineFunc(
+ reset_event_manager_on_fork);
+ }
return &vtable;
}
diff --git a/src/core/lib/iomgr/ev_posix.cc b/src/core/lib/iomgr/ev_posix.cc
index b8fe017ce7..0205363d5c 100644
--- a/src/core/lib/iomgr/ev_posix.cc
+++ b/src/core/lib/iomgr/ev_posix.cc
@@ -59,7 +59,14 @@ grpc_core::DebugOnlyTraceFlag grpc_polling_api_trace(false, "polling_api");
/** Default poll() function - a pointer so that it can be overridden by some
* tests */
+#ifndef GPR_AIX
grpc_poll_function_type grpc_poll_function = poll;
+#else
+int aix_poll(struct pollfd fds[], nfds_t nfds, int timeout) {
+ return poll(fds, nfds, timeout);
+}
+grpc_poll_function_type grpc_poll_function = aix_poll;
+#endif
grpc_wakeup_fd grpc_global_wakeup_fd;
diff --git a/src/core/lib/iomgr/exec_ctx.cc b/src/core/lib/iomgr/exec_ctx.cc
index 5d5c355ff9..d68fa0714b 100644
--- a/src/core/lib/iomgr/exec_ctx.cc
+++ b/src/core/lib/iomgr/exec_ctx.cc
@@ -109,6 +109,12 @@ grpc_closure_scheduler* grpc_schedule_on_exec_ctx = &exec_ctx_scheduler;
namespace grpc_core {
GPR_TLS_CLASS_DEF(ExecCtx::exec_ctx_);
+// WARNING: for testing purposes only!
+void ExecCtx::TestOnlyGlobalInit(gpr_timespec new_val) {
+ g_start_time = new_val;
+ gpr_tls_init(&exec_ctx_);
+}
+
void ExecCtx::GlobalInit(void) {
g_start_time = gpr_now(GPR_CLOCK_MONOTONIC);
gpr_tls_init(&exec_ctx_);
diff --git a/src/core/lib/iomgr/exec_ctx.h b/src/core/lib/iomgr/exec_ctx.h
index 8ddab0d381..f3528d527a 100644
--- a/src/core/lib/iomgr/exec_ctx.h
+++ b/src/core/lib/iomgr/exec_ctx.h
@@ -192,6 +192,8 @@ class ExecCtx {
now_is_valid_ = true;
}
+ static void TestOnlyGlobalInit(gpr_timespec new_val);
+
/** Global initialization for ExecCtx. Called by iomgr. */
static void GlobalInit(void);
diff --git a/src/core/lib/iomgr/fork_posix.cc b/src/core/lib/iomgr/fork_posix.cc
index b37384b8db..e957bad73d 100644
--- a/src/core/lib/iomgr/fork_posix.cc
+++ b/src/core/lib/iomgr/fork_posix.cc
@@ -25,6 +25,7 @@
#include <string.h>
#include <grpc/fork.h>
+#include <grpc/grpc.h>
#include <grpc/support/log.h>
#include "src/core/lib/gpr/env.h"
@@ -34,7 +35,6 @@
#include "src/core/lib/iomgr/executor.h"
#include "src/core/lib/iomgr/timer_manager.h"
#include "src/core/lib/iomgr/wakeup_fd_posix.h"
-#include "src/core/lib/surface/init.h"
/*
* NOTE: FORKING IS NOT GENERALLY SUPPORTED, THIS IS ONLY INTENDED TO WORK
@@ -58,6 +58,12 @@ void grpc_prefork() {
"environment variable GRPC_ENABLE_FORK_SUPPORT=1");
return;
}
+ if (strcmp(grpc_get_poll_strategy_name(), "epoll1") != 0 &&
+ strcmp(grpc_get_poll_strategy_name(), "poll") != 0) {
+ gpr_log(GPR_ERROR,
+ "Fork support is only compatible with the epoll1 and poll polling "
+ "strategies");
+ }
if (!grpc_core::Fork::BlockExecCtx()) {
gpr_log(GPR_INFO,
"Other threads are currently calling into gRPC, skipping fork() "
@@ -84,6 +90,11 @@ void grpc_postfork_child() {
if (!skipped_handler) {
grpc_core::Fork::AllowExecCtx();
grpc_core::ExecCtx exec_ctx;
+ grpc_core::Fork::child_postfork_func reset_polling_engine =
+ grpc_core::Fork::GetResetChildPollingEngineFunc();
+ if (reset_polling_engine != nullptr) {
+ reset_polling_engine();
+ }
grpc_timer_manager_set_threading(true);
grpc_executor_set_threading(true);
}
diff --git a/src/core/lib/iomgr/port.h b/src/core/lib/iomgr/port.h
index 1d0ecff802..066417b93c 100644
--- a/src/core/lib/iomgr/port.h
+++ b/src/core/lib/iomgr/port.h
@@ -140,6 +140,18 @@
#define GRPC_POSIX_SOCKET 1
#define GRPC_POSIX_SOCKETUTILS 1
#define GRPC_POSIX_WAKEUP_FD 1
+#elif defined(GPR_SOLARIS)
+#define GRPC_HAVE_UNIX_SOCKET 1
+#define GRPC_POSIX_NO_SPECIAL_WAKEUP_FD 1
+#define GRPC_POSIX_SOCKET 1
+#define GRPC_POSIX_SOCKETUTILS 1
+#define GRPC_POSIX_WAKEUP_FD 1
+#elif defined(GPR_AIX)
+#define GRPC_HAVE_UNIX_SOCKET 1
+#define GRPC_POSIX_NO_SPECIAL_WAKEUP_FD 1
+#define GRPC_POSIX_SOCKET 1
+#define GRPC_POSIX_SOCKETUTILS 1
+#define GRPC_POSIX_WAKEUP_FD 1
#elif defined(GPR_NACL)
#define GRPC_HAVE_ARPA_NAMESER 1
#define GRPC_POSIX_NO_SPECIAL_WAKEUP_FD 1
diff --git a/src/core/lib/iomgr/socket_mutator.cc b/src/core/lib/iomgr/socket_mutator.cc
index b9b8eaf4ad..a448c9f61c 100644
--- a/src/core/lib/iomgr/socket_mutator.cc
+++ b/src/core/lib/iomgr/socket_mutator.cc
@@ -57,7 +57,7 @@ int grpc_socket_mutator_compare(grpc_socket_mutator* a,
void grpc_socket_mutator_unref(grpc_socket_mutator* mutator) {
if (gpr_unref(&mutator->refcount)) {
- mutator->vtable->destory(mutator);
+ mutator->vtable->destroy(mutator);
}
}
diff --git a/src/core/lib/iomgr/socket_mutator.h b/src/core/lib/iomgr/socket_mutator.h
index 6c7781c51d..8742a3ba61 100644
--- a/src/core/lib/iomgr/socket_mutator.h
+++ b/src/core/lib/iomgr/socket_mutator.h
@@ -33,7 +33,7 @@ typedef struct {
/** Compare socket mutator \a a and \a b */
int (*compare)(grpc_socket_mutator* a, grpc_socket_mutator* b);
/** Destroys the socket mutator instance */
- void (*destory)(grpc_socket_mutator* mutator);
+ void (*destroy)(grpc_socket_mutator* mutator);
} grpc_socket_mutator_vtable;
/** The Socket Mutator interface allows changes on socket options */
diff --git a/src/core/lib/iomgr/timer.h b/src/core/lib/iomgr/timer.h
index 7f534476df..17e933b865 100644
--- a/src/core/lib/iomgr/timer.h
+++ b/src/core/lib/iomgr/timer.h
@@ -61,10 +61,11 @@ typedef struct grpc_timer_vtable {
/* Initialize *timer. When expired or canceled, closure will be called with
error set to indicate if it expired (GRPC_ERROR_NONE) or was canceled
- (GRPC_ERROR_CANCELLED). timer_cb is guaranteed to be called exactly once, and
+ (GRPC_ERROR_CANCELLED). *closure is guaranteed to be called exactly once, and
application code should check the error to determine how it was invoked. The
application callback is also responsible for maintaining information about
- when to free up any user-level state. */
+ when to free up any user-level state. Behavior is undefined for a deadline of
+ GRPC_MILLIS_INF_FUTURE. */
void grpc_timer_init(grpc_timer* timer, grpc_millis deadline,
grpc_closure* closure);
diff --git a/src/core/lib/security/credentials/jwt/json_token.h b/src/core/lib/security/credentials/jwt/json_token.h
index d0fb4ebd0a..3ed990140d 100644
--- a/src/core/lib/security/credentials/jwt/json_token.h
+++ b/src/core/lib/security/credentials/jwt/json_token.h
@@ -21,6 +21,8 @@
#include <grpc/support/port_platform.h>
+#include "src/core/tsi/grpc_shadow_boringssl.h"
+
#include <grpc/slice.h>
#include <openssl/rsa.h>
diff --git a/src/core/lib/security/credentials/jwt/jwt_verifier.cc b/src/core/lib/security/credentials/jwt/jwt_verifier.cc
index 5c47276e32..c7d1b36ff0 100644
--- a/src/core/lib/security/credentials/jwt/jwt_verifier.cc
+++ b/src/core/lib/security/credentials/jwt/jwt_verifier.cc
@@ -18,6 +18,8 @@
#include <grpc/support/port_platform.h>
+#include "src/core/tsi/grpc_shadow_boringssl.h"
+
#include "src/core/lib/security/credentials/jwt/jwt_verifier.h"
#include <limits.h>
diff --git a/src/core/lib/security/credentials/oauth2/oauth2_credentials.cc b/src/core/lib/security/credentials/oauth2/oauth2_credentials.cc
index 43dd68e874..44b093557f 100644
--- a/src/core/lib/security/credentials/oauth2/oauth2_credentials.cc
+++ b/src/core/lib/security/credentials/oauth2/oauth2_credentials.cc
@@ -235,7 +235,7 @@ static void on_oauth2_token_fetcher_http_response(void* user_data,
access_token_md);
} else {
error = GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
- "Error occured when fetching oauth2 token.", &error, 1);
+ "Error occurred when fetching oauth2 token.", &error, 1);
}
GRPC_CLOSURE_SCHED(pending_request->on_request_metadata, error);
grpc_polling_entity_del_from_pollset_set(
diff --git a/src/core/lib/surface/call.cc b/src/core/lib/surface/call.cc
index 52053e686b..2923a86646 100644
--- a/src/core/lib/surface/call.cc
+++ b/src/core/lib/surface/call.cc
@@ -1951,7 +1951,7 @@ done:
return error;
done_with_error:
- /* reverse any mutations that occured */
+ /* reverse any mutations that occurred */
if (stream_op->send_initial_metadata) {
call->sent_initial_metadata = false;
grpc_metadata_batch_clear(&call->metadata_batch[0][0]);
diff --git a/src/core/lib/surface/init.h b/src/core/lib/surface/init.h
index 9353208332..193f51447d 100644
--- a/src/core/lib/surface/init.h
+++ b/src/core/lib/surface/init.h
@@ -22,6 +22,5 @@
void grpc_register_security_filters(void);
void grpc_security_pre_init(void);
void grpc_security_init(void);
-int grpc_is_initialized(void);
#endif /* GRPC_CORE_LIB_SURFACE_INIT_H */
diff --git a/src/core/lib/transport/service_config.cc b/src/core/lib/transport/service_config.cc
index e1a55d98ab..405e336028 100644
--- a/src/core/lib/transport/service_config.cc
+++ b/src/core/lib/transport/service_config.cc
@@ -65,8 +65,8 @@ const char* ServiceConfig::GetLoadBalancingPolicyName() const {
return lb_policy_name;
}
-size_t ServiceConfig::CountNamesInMethodConfig(grpc_json* json) {
- size_t num_names = 0;
+int ServiceConfig::CountNamesInMethodConfig(grpc_json* json) {
+ int num_names = 0;
for (grpc_json* field = json->child; field != nullptr; field = field->next) {
if (field->key != nullptr && strcmp(field->key, "name") == 0) {
if (field->type != GRPC_JSON_ARRAY) return -1;
diff --git a/src/core/lib/transport/service_config.h b/src/core/lib/transport/service_config.h
index a65b267d46..2c0dd75845 100644
--- a/src/core/lib/transport/service_config.h
+++ b/src/core/lib/transport/service_config.h
@@ -103,7 +103,7 @@ class ServiceConfig {
ServiceConfig(UniquePtr<char> json_string, grpc_json* json_tree);
// Returns the number of names specified in the method config \a json.
- static size_t CountNamesInMethodConfig(grpc_json* json);
+ static int CountNamesInMethodConfig(grpc_json* json);
// Returns a path string for the JSON name object specified by \a json.
// Returns null on error.
@@ -188,9 +188,9 @@ ServiceConfig::CreateMethodConfigTable(CreateValue<T> create_value) {
// Find number of entries.
for (grpc_json* method = field->child; method != nullptr;
method = method->next) {
- size_t count = CountNamesInMethodConfig(method);
+ int count = CountNamesInMethodConfig(method);
if (count <= 0) return nullptr;
- num_entries += count;
+ num_entries += static_cast<size_t>(count);
}
// Populate method config table entries.
entries = static_cast<typename SliceHashTable<RefCountedPtr<T>>::Entry*>(
generated by cgit v1.2.3 (git 2.39.1) at 2024-07-02 06:40:08 +0000