1 /*
2 *
3 * Copyright 2019 gRPC authors.
4 *
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
8 *
9 * http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 *
17 */
18
19 #include <grpc/support/port_platform.h>
20
21 #include "src/core/lib/iomgr/poller/eventmanager_libuv.h"
22
23 #include <grpc/support/time.h>
24
Options()25 grpc::experimental::LibuvEventManager::Options::Options() : num_workers_(-1) {}
Options(int num_workers)26 grpc::experimental::LibuvEventManager::Options::Options(int num_workers)
27 : num_workers_(num_workers) {}
28
LibuvEventManager(const Options & options)29 grpc::experimental::LibuvEventManager::LibuvEventManager(const Options& options)
30 : options_(options) {
31 int num_workers = options_.num_workers();
32 // Number of workers can't be 0 if we do not accept thread donation.
33 // TODO(guantaol): replaces the hard-coded number with a flag.
34 if (num_workers <= 0) num_workers = 32;
35
36 for (int i = 0; i < num_workers; i++) {
37 workers_.emplace_back(
38 options_.thread_name_prefix().c_str(),
39 [](void* em) { static_cast<LibuvEventManager*>(em)->RunWorkerLoop(); },
40 this);
41 workers_.back().Start();
42 }
43 }
44
~LibuvEventManager()45 grpc::experimental::LibuvEventManager::~LibuvEventManager() {
46 Shutdown();
47 for (auto& th : workers_) {
48 th.Join();
49 }
50 }
51
RunWorkerLoop()52 void grpc::experimental::LibuvEventManager::RunWorkerLoop() {
53 while (true) {
54 // TODO(guantaol): extend the worker loop with real work.
55 if (ShouldStop()) return;
56 gpr_sleep_until(gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC),
57 gpr_time_from_micros(10, GPR_TIMESPAN)));
58 }
59 }
60
ShouldStop()61 bool grpc::experimental::LibuvEventManager::ShouldStop() {
62 return should_stop_.Load(grpc_core::MemoryOrder::ACQUIRE) != 0;
63 }
64
Shutdown()65 void grpc::experimental::LibuvEventManager::Shutdown() {
66 if (should_stop_.Load(grpc_core::MemoryOrder::ACQUIRE)) {
67 return; // Already shut down.
68 }
69
70 {
71 grpc_core::MutexLock lock(&shutdown_mu_);
72 while (shutdown_refcount_.Load(grpc_core::MemoryOrder::ACQUIRE) > 0) {
73 shutdown_cv_.Wait(&shutdown_mu_);
74 }
75 }
76 should_stop_.Store(true, grpc_core::MemoryOrder::RELEASE);
77 }
78
ShutdownRef()79 void grpc::experimental::LibuvEventManager::ShutdownRef() {
80 shutdown_refcount_.FetchAdd(1, grpc_core::MemoryOrder::RELAXED);
81 }
82
ShutdownUnref()83 void grpc::experimental::LibuvEventManager::ShutdownUnref() {
84 if (shutdown_refcount_.FetchSub(1, grpc_core::MemoryOrder::ACQ_REL) == 1) {
85 grpc_core::MutexLock lock(&shutdown_mu_);
86 shutdown_cv_.Signal();
87 }
88 }
89