Merge branch 'chan2' into chan3

.github/labeler.yml

@@ -51,6 +51,7 @@ lang/python:
       - all-globs-to-any-file:
           - src/python/**
           - '!src/python/grpcio/grpc_core_dependencies.py'
+          - '!src/python/grpcio_observability/observability_lib_deps.py'
 lang/ruby:
   - changed-files:
       - any-glob-to-any-file:

doc/command_line_tool.md

@@ -2,37 +2,44 @@
 
 ## Overview
 
-This document describes the command line tool that comes with gRPC repository. It is desirable to have command line
-tools written in other languages roughly follow the same syntax and flags.
+This document describes the command line tool that comes with gRPC repository.
+It is desirable to have command line tools written in other languages roughly
+follow the same syntax and flags.
 
-At this point, the tool needs to be built from source, and it should be moved out to grpc-tools repository as a stand
-alone application once it is mature enough.
+> [!NOTE]
+> At present, the tool needs to be built from source, and it should be moved out
+> to grpc-tools repository as a stand alone application once it is mature
+> enough. This tool in its current state though is not up to par in its
+> user-friendliness. Other tools in the ecosystem, for example,
+> [grpcurl](https://github.com/fullstorydev/grpcurl) are better maintained.
 
 ## Core functionality
 
 The command line tool can do the following things:
 
-- Send unary rpc.
-- Attach metadata and display received metadata.
-- Handle common authentication to server.
-- Infer request/response types from server reflection result.
-- Find the request/response types from a given proto file.
-- Read proto request in text form.
-- Read request in wire form (for protobuf messages, this means serialized binary form).
-- Display proto response in text form.
-- Write response in wire form to a file.
+-   Send unary rpc.
+-   Attach metadata and display received metadata.
+-   Handle common authentication to server.
+-   Infer request/response types from server reflection result.
+-   Find the request/response types from a given proto file.
+-   Read proto request in text form.
+-   Read request in wire form (for protobuf messages, this means serialized
+    binary form).
+-   Display proto response in text form.
+-   Write response in wire form to a file.
 
 The command line tool should support the following things:
 
-- List server services and methods through server reflection.
-- Fine-grained auth control (such as, use this oauth token to talk to the server).
-- Send streaming rpc.
+-   List server services and methods through server reflection.
+-   Fine-grained auth control (such as, use this oauth token to talk to the
+    server).
+-   Send streaming rpc.
 
 ## Code location
 
 To use the tool, you need to get the grpc repository and make sure your system
-has the prerequisites for building grpc from source, given in the [installation
-instructions](../BUILDING.md).
+has the prerequisites for building grpc from source, given in the
+[installation instructions](../BUILDING.md).
 
 In order to build the grpc command line tool from a fresh clone of the grpc
 repository, you need to run the following command to update submodules:
@@ -58,10 +65,13 @@ https://github.com/grpc/grpc/blob/master/test/cpp/util/grpc_cli.cc
 Most `grpc_cli` commands need the server to support server reflection. See
 guides for
 [Java](https://github.com/grpc/grpc-java/blob/master/documentation/server-reflection-tutorial.md#enable-server-reflection)
-, [C++](https://github.com/grpc/grpc/blob/master/doc/server_reflection_tutorial.md)
-and [Go](https://github.com/grpc/grpc-go/blob/master/Documentation/server-reflection-tutorial.md)
+,
+[C++](https://github.com/grpc/grpc/blob/master/doc/server_reflection_tutorial.md)
+and
+[Go](https://github.com/grpc/grpc-go/blob/master/Documentation/server-reflection-tutorial.md)
 
-Local proto files can be used as an alternative. See instructions [below](#Call-a-remote-method).
+Local proto files can be used as an alternative. See instructions
+[below](#Call-a-remote-method).
 
 ## Usage
 
@@ -176,8 +186,8 @@ We can send RPCs to a server and get responses using `grpc_cli call` command.
     ```
 
     If the proto file is not under the current directory, you can use
-    `--proto_path` to specify new search roots
-    (separated by colon on Mac/Linux/Cygwin or semicolon on Windows).
+    `--proto_path` to specify new search roots (separated by colon on
+    Mac/Linux/Cygwin or semicolon on Windows).
 
     Note that the tool will always attempt to use the reflection service first,
     falling back to local proto files if the service is not found. Use

src/core/BUILD

@@ -6288,11 +6288,13 @@ grpc_cc_library(
         "closure",
         "error",
         "error_utils",
+        "event_engine_tcp_socket_utils",
         "grpc_insecure_credentials",
         "handshaker_registry",
         "iomgr_fwd",
         "memory_quota",
         "pollset_set",
+        "posix_event_engine_tcp_socket_utils",
         "resolved_address",
         "resource_quota",
         "status_helper",

src/core/ext/filters/client_channel/retry_filter_legacy_call_data.cc

@@ -2020,8 +2020,10 @@ void RetryFilter::LegacyCallData::OnRetryTimer() {
 void RetryFilter::LegacyCallData::OnRetryTimerLocked(
     void* arg, grpc_error_handle /*error*/) {
   auto* calld = static_cast<RetryFilter::LegacyCallData*>(arg);
-  calld->retry_timer_handle_.reset();
-  calld->CreateCallAttempt(/*is_transparent_retry=*/false);
+  if (calld->retry_timer_handle_.has_value()) {
+    calld->retry_timer_handle_.reset();
+    calld->CreateCallAttempt(/*is_transparent_retry=*/false);
+  }
   GRPC_CALL_STACK_UNREF(calld->owning_call_, "OnRetryTimer");
 }
 

src/core/ext/transport/chttp2/server/chttp2_server.cc

@@ -56,6 +56,8 @@
 #include "src/core/lib/config/core_configuration.h"
 #include "src/core/lib/debug/trace.h"
 #include "src/core/lib/event_engine/channel_args_endpoint_config.h"
+#include "src/core/lib/event_engine/posix_engine/tcp_socket_utils.h"
+#include "src/core/lib/event_engine/tcp_socket_utils.h"
 #include "src/core/lib/gprpp/debug_location.h"
 #include "src/core/lib/gprpp/orphanable.h"
 #include "src/core/lib/gprpp/ref_counted_ptr.h"
@@ -95,6 +97,8 @@ namespace grpc_core {
 namespace {
 
 using ::grpc_event_engine::experimental::EventEngine;
+using ::grpc_event_engine::experimental::PosixSocketWrapper;
+using ::grpc_event_engine::experimental::ResolvedAddressToURI;
 
 const char kUnixUriPrefix[] = "unix:";
 const char kUnixAbstractUriPrefix[] = "unix-abstract:";
@@ -1076,12 +1080,25 @@ void grpc_server_add_channel_from_fd(grpc_server* server, int fd,
 
   grpc_core::ChannelArgs server_args = core_server->channel_args();
   std::string name = absl::StrCat("fd:", fd);
+  // The fd is going to be registered with the gRPC poller to be asynchronously
+  // notified for read/write/error events. It must be marked as NonBlocking
+  // to ensure that the poller/gRPC executor threads do not get blocked on
+  // the socket when attempting to read/write from it.
+  grpc_core::PosixSocketWrapper sock(fd);
+  (void)sock.SetSocketNonBlocking(true);
+  (void)sock.SetSocketCloexec(true);
+  (void)sock.SetSocketNoSigpipeIfPossible();
+  auto peer_address = sock.PeerAddress();
+  std::string addr_str = name;
+  if (peer_address.ok()) {
+    addr_str = *grpc_core::ResolvedAddressToURI(peer_address.value());
+  }
   auto memory_quota =
       server_args.GetObject<grpc_core::ResourceQuota>()->memory_quota();
   grpc_endpoint* server_endpoint = grpc_tcp_create_from_fd(
       grpc_fd_create(fd, name.c_str(), true),
       grpc_event_engine::experimental::ChannelArgsEndpointConfig(server_args),
-      name);
+      addr_str);
   grpc_core::Transport* transport = grpc_create_chttp2_transport(
       server_args, server_endpoint, false  // is_client
   );

src/python/grpcio/grpc/_cython/_cygrpc/aio/common.pyx.pxi

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import warnings
+
 from cpython.version cimport PY_MAJOR_VERSION, PY_MINOR_VERSION
 
 TYPE_METADATA_STRING = "Tuple[Tuple[str, Union[str, bytes]]...]"
@@ -181,7 +183,15 @@ if PY_MAJOR_VERSION >= 3 and PY_MINOR_VERSION >= 7:
         try:
             return asyncio.get_running_loop()
         except RuntimeError:
-            return asyncio.get_event_loop_policy().get_event_loop()
+            with warnings.catch_warnings():
+                # Convert DeprecationWarning to errors so we can capture them with except
+                warnings.simplefilter("error", DeprecationWarning)
+                try:
+                    return asyncio.get_event_loop_policy().get_event_loop()
+                # Since version 3.12, DeprecationWarning is emitted if there is no
+                # current event loop.
+                except DeprecationWarning:
+                    return asyncio.get_event_loop_policy().new_event_loop()
 else:
     def get_working_loop():
         """Returns a running event loop."""

src/python/grpcio_tests/tests/unit/_dynamic_stubs_test.py

@@ -29,7 +29,7 @@ _DATA_DIR = os.path.join("tests", "unit", "data")
 @contextlib.contextmanager
 def _grpc_tools_unimportable():
     original_sys_path = sys.path
-    sys.path = [path for path in sys.path if "grpcio_tools" not in path]
+    sys.path = [path for path in sys.path if "grpcio_tools" not in str(path)]
     try:
         import grpc_tools
     except ImportError:

test/core/http/httpcli_test.cc

@@ -475,11 +475,19 @@ TEST_F(HttpRequestTest, CallerPollentsAreNotReferencedAfterCallbackIsRan) {
   http_request->Start();
   exec_ctx.Flush();
   http_request.reset();  // cancel the request
+  // With iomgr polling:
   // Since the request was cancelled, the on_done callback should be flushed
   // out on the ExecCtx flush below. When the on_done callback is ran, it will
-  // eagerly destroy 'request_state.pollset_set_to_destroy_eagerly'. Thus, we
-  // can't poll on that pollset here.
+  // eagerly destroy 'request_state.pollset_set_to_destroy_eagerly'. PollUntil's
+  // predicate should return true immediately.
+  //
+  // With EventEngine polling:
+  // Since the callback will be run asynchronously in another thread, with an
+  // independent ExecCtx, PollUntil is used here to ensure this test does not
+  // finish before the callback is run.
   exec_ctx.Flush();
+  PollUntil([&request_state]() { return request_state.done; },
+            AbslDeadlineSeconds(60));
 }
 
 void CancelRequest(grpc_core::HttpRequest* req) {

test/core/iomgr/fd_conservation_posix_test.cc

@@ -39,9 +39,9 @@ int main(int argc, char** argv) {
     // set max # of file descriptors to a low value, and
     // verify we can create and destroy many more than this number
     // of descriptors
-    rlim.rlim_cur = rlim.rlim_max = 10;
+    rlim.rlim_cur = rlim.rlim_max = 1000;
     GPR_ASSERT(0 == setrlimit(RLIMIT_NOFILE, &rlim));
-    for (i = 0; i < 100; i++) {
+    for (i = 0; i < 10000; i++) {
       p = grpc_iomgr_create_endpoint_pair("test", nullptr);
       grpc_endpoint_destroy(p.client);
       grpc_endpoint_destroy(p.server);

tools/distrib/python/grpcio_tools/grpc_tools/protoc.py

@@ -18,7 +18,11 @@ import os
 import sys
 
 from grpc_tools import _protoc_compiler
-import pkg_resources
+
+if sys.version_info >= (3, 9, 0):
+    from importlib import resources
+else:
+    import pkg_resources
 
 _PROTO_MODULE_SUFFIX = "_pb2"
 _SERVICE_MODULE_SUFFIX = "_pb2_grpc"
@@ -37,6 +41,20 @@ def main(command_arguments):
     return _protoc_compiler.run_main(command_arguments)
 
 
+def _get_resource_file_name(
+    package_or_requirement: str, resource_name: str
+) -> str:
+    """Obtain the filename for a resource on the file system."""
+    if sys.version_info >= (3, 9, 0):
+        return (
+            resources.files(package_or_requirement) / resource_name
+        ).resolve()
+    else:
+        return pkg_resources.resource_filename(
+            package_or_requirement, resource_name
+        )
+
+
 # NOTE(rbellevi): importlib.abc is not supported on 3.4.
 if sys.version_info >= (3, 5, 0):
     import contextlib
@@ -63,9 +81,10 @@ if sys.version_info >= (3, 5, 0):
                         ),
                     ]
                 )
-                sys.path.append(
-                    pkg_resources.resource_filename("grpc_tools", "_proto")
-                )
+
+                proto_include = _get_resource_file_name("grpc_tools", "_proto")
+                sys.path.append(proto_include)
+
                 _FINDERS_INSTALLED = True
 
     def _module_name_to_proto_file(suffix, module_name):
@@ -185,5 +204,5 @@ if sys.version_info >= (3, 5, 0):
         _maybe_install_proto_finders()
 
 if __name__ == "__main__":
-    proto_include = pkg_resources.resource_filename("grpc_tools", "_proto")
+    proto_include = _get_resource_file_name("grpc_tools", "_proto")
     sys.exit(main(sys.argv + ["-I{}".format(proto_include)]))