include/grpc/event_engine/event_engine.h

   1 // Copyright 2021 The gRPC Authors
   2 //
   3 // Licensed under the Apache License, Version 2.0 (the "License");
   4 // you may not use this file except in compliance with the License.
   5 // You may obtain a copy of the License at
   6 //
   7 //     http://www.apache.org/licenses/LICENSE-2.0
   8 //
   9 // Unless required by applicable law or agreed to in writing, software
  10 // distributed under the License is distributed on an "AS IS" BASIS,
  11 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 // See the License for the specific language governing permissions and
  13 // limitations under the License.
  14 #ifndef GRPC_EVENT_ENGINE_EVENT_ENGINE_H
  15 #define GRPC_EVENT_ENGINE_EVENT_ENGINE_H
  16
  17 #include <grpc/support/port_platform.h>
  18
  19 #include <functional>
  20 #include <vector>
  21
  22 #include "absl/status/status.h"
  23 #include "absl/status/statusor.h"
  24 #include "absl/time/time.h"
  25
  26 #include "grpc/event_engine/endpoint_config.h"
  27 #include "grpc/event_engine/port.h"
  28 #include "grpc/event_engine/slice_allocator.h"
  29
  30 // TODO(hork): Define the Endpoint::Write metrics collection system
  31 namespace grpc_event_engine {
  32 namespace experimental {
  33
  34 ////////////////////////////////////////////////////////////////////////////////
  35 /// The EventEngine encapsulates all platform-specific behaviors related to low
  36 /// level network I/O, timers, asynchronous execution, and DNS resolution.
  37 ///
  38 /// This interface allows developers to provide their own event management and
  39 /// network stacks. Motivating uses cases for supporting custom EventEngines
  40 /// include the ability to hook into external event loops, and using different
  41 /// EventEngine instances for each channel to better insulate network I/O and
  42 /// callback processing from other channels.
  43 ///
  44 /// A default cross-platform EventEngine instance is provided by gRPC.
  45 ///
  46 /// LIFESPAN AND OWNERSHIP
  47 ///
  48 /// gRPC takes shared ownership of EventEngines via std::shared_ptrs to ensure
  49 /// that the engines remain available until they are no longer needed. Depending
  50 /// on the use case, engines may live until gRPC is shut down.
  51 ///
  52 /// EXAMPLE USAGE (Not yet implemented)
  53 ///
  54 /// Custom EventEngines can be specified per channel, and allow configuration
  55 /// for both clients and servers. To set a custom EventEngine for a client
  56 /// channel, you can do something like the following:
  57 ///
  58 ///    ChannelArguments args;
  59 ///    std::shared_ptr<EventEngine> engine = std::make_shared<MyEngine>(...);
  60 ///    args.SetEventEngine(engine);
  61 ///    MyAppClient client(grpc::CreateCustomChannel(
  62 ///        "localhost:50051", grpc::InsecureChannelCredentials(), args));
  63 ///
  64 /// A gRPC server can use a custom EventEngine by calling the
  65 /// ServerBuilder::SetEventEngine method:
  66 ///
  67 ///    ServerBuilder builder;
  68 ///    std::shared_ptr<EventEngine> engine = std::make_shared<MyEngine>(...);
  69 ///    builder.SetEventEngine(engine);
  70 ///    std::unique_ptr<Server> server(builder.BuildAndStart());
  71 ///    server->Wait();
  72 ///
  73 ////////////////////////////////////////////////////////////////////////////////
  74 class EventEngine {
  75  public:
  76   /// A basic callable function. The first argument to all callbacks is an
  77   /// absl::Status indicating the status of the operation associated with this
  78   /// callback. Each EventEngine method that takes a callback parameter, defines
  79   /// the expected sets and meanings of statuses for that use case.
  80   using Callback = std::function<void(absl::Status)>;
  81   /// A callback handle, used to cancel a callback.
  82   struct TaskHandle {
  83     intptr_t keys[2];
  84   };
  85   /// A thin wrapper around a platform-specific sockaddr type. A sockaddr struct
  86   /// exists on all platforms that gRPC supports.
  87   ///
  88   /// Platforms are expected to provide definitions for:
  89   /// * sockaddr
  90   /// * sockaddr_in
  91   /// * sockaddr_in6
  92   class ResolvedAddress {
  93    public:
  94     static constexpr socklen_t MAX_SIZE_BYTES = 128;
  95
  96     ResolvedAddress(const sockaddr* address, socklen_t size);
  97     ResolvedAddress() = default;
  98     ResolvedAddress(const ResolvedAddress&) = default;
  99     const struct sockaddr* address() const;
 100     socklen_t size() const;
 101
 102    private:
 103     char address_[MAX_SIZE_BYTES];
 104     socklen_t size_ = 0;
 105   };
 106
 107   /// An Endpoint represents one end of a connection between a gRPC client and
 108   /// server. Endpoints are created when connections are established, and
 109   /// Endpoint operations are gRPC's primary means of communication.
 110   ///
 111   /// Endpoints must use the provided SliceAllocator for all data buffer memory
 112   /// allocations. gRPC allows applications to set memory constraints per
 113   /// Channel or Server, and the implementation depends on all dynamic memory
 114   /// allocation being handled by the quota system.
 115   class Endpoint {
 116    public:
 117     /// The Endpoint destructor is responsible for shutting down all connections
 118     /// and invoking all pending read or write callbacks with an error status.
 119     virtual ~Endpoint() = default;
 120     /// Read data from the Endpoint.
 121     ///
 122     /// When data is available on the connection, that data is moved into the
 123     /// \a buffer, and the \a on_read callback is called. The caller must ensure
 124     /// that the callback has access to the buffer when executed later.
 125     /// Ownership of the buffer is not transferred. Valid slices *may* be placed
 126     /// into the buffer even if the callback is invoked with a non-OK Status.
 127     ///
 128     /// For failed read operations, implementations should pass the appropriate
 129     /// statuses to \a on_read. For example, callbacks might expect to receive
 130     /// CANCELLED on endpoint shutdown.
 131     virtual void Read(Callback on_read, SliceBuffer* buffer) = 0;
 132     /// Write data out on the connection.
 133     ///
 134     /// \a on_writable is called when the connection is ready for more data. The
 135     /// Slices within the \a data buffer may be mutated at will by the Endpoint
 136     /// until \a on_writable is called. The \a data SliceBuffer will remain
 137     /// valid after calling \a Write, but its state is otherwise undefined.
 138     ///
 139     /// For failed write operations, implementations should pass the appropriate
 140     /// statuses to \a on_writable. For example, callbacks might expect to
 141     /// receive CANCELLED on endpoint shutdown.
 142     virtual void Write(Callback on_writable, SliceBuffer* data) = 0;
 143     /// These methods return an address in the format described in DNSResolver.
 144     /// The returned values are owned by the Endpoint and are expected to remain
 145     /// valid for the life of the Endpoint.
 146     virtual const ResolvedAddress& GetPeerAddress() const = 0;
 147     virtual const ResolvedAddress& GetLocalAddress() const = 0;
 148   };
 149
 150   /// Called when a new connection is established.
 151   ///
 152   /// If the connection attempt was not successful, implementations should pass
 153   /// the appropriate statuses to this callback. For example, callbacks might
 154   /// expect to receive DEADLINE_EXCEEDED statuses when appropriate, or
 155   /// CANCELLED statuses on EventEngine shutdown.
 156   using OnConnectCallback =
 157       std::function<void(absl::StatusOr<std::unique_ptr<Endpoint>>)>;
 158
 159   /// An EventEngine Listener listens for incoming connection requests from gRPC
 160   /// clients and initiates request processing once connections are established.
 161   class Listener {
 162    public:
 163     /// Called when the listener has accepted a new client connection.
 164     using AcceptCallback = std::function<void(std::unique_ptr<Endpoint>)>;
 165     virtual ~Listener() = default;
 166     /// Bind an address/port to this Listener.
 167     ///
 168     /// It is expected that multiple addresses/ports can be bound to this
 169     /// Listener before Listener::Start has been called. Returns either the
 170     /// bound port or an appropriate error status.
 171     virtual absl::StatusOr<int> Bind(const ResolvedAddress& addr) = 0;
 172     virtual absl::Status Start() = 0;
 173   };
 174
 175   /// Factory method to create a network listener / server.
 176   ///
 177   /// Once a \a Listener is created and started, the \a on_accept callback will
 178   /// be called once asynchronously for each established connection. Note that
 179   /// unlike other callbacks, there is no status code parameter since the
 180   /// callback will only be called in healthy scenarios where connections can be
 181   /// accepted.
 182   ///
 183   /// This method may return a non-OK status immediately if an error was
 184   /// encountered in any synchronous steps required to create the Listener. In
 185   /// this case, \a on_shutdown will never be called.
 186   ///
 187   /// If this method returns a Listener, then \a on_shutdown will be invoked
 188   /// exactly once, when the Listener is shut down. The status passed to it will
 189   /// indicate if there was a problem during shutdown.
 190   ///
 191   /// The provided \a SliceAllocatorFactory is used to create \a SliceAllocators
 192   /// for Endpoint construction.
 193   virtual absl::StatusOr<std::unique_ptr<Listener>> CreateListener(
 194       Listener::AcceptCallback on_accept, Callback on_shutdown,
 195       const EndpointConfig& args,
 196       std::unique_ptr<SliceAllocatorFactory> slice_allocator_factory) = 0;
 197   /// Creates a client network connection to a remote network listener.
 198   ///
 199   /// \a Connect may return an error status immediately if there was a failure
 200   /// in the synchronous part of establishing a connection. In that event, the
 201   /// \a on_connect callback *will not* have been executed. Otherwise, it is
 202   /// expected that the \a on_connect callback will be asynchronously executed
 203   /// exactly once by the EventEngine.
 204   ///
 205   /// Implementation Note: it is important that the \a slice_allocator be used
 206   /// for all read/write buffer allocations in the EventEngine implementation.
 207   /// This allows gRPC's \a ResourceQuota system to monitor and control memory
 208   /// usage with graceful degradation mechanisms. Please see the \a
 209   /// SliceAllocator API for more information.
 210   virtual absl::Status Connect(OnConnectCallback on_connect,
 211                                const ResolvedAddress& addr,
 212                                const EndpointConfig& args,
 213                                std::unique_ptr<SliceAllocator> slice_allocator,
 214                                absl::Time deadline) = 0;
 215
 216   /// The DNSResolver that provides asynchronous resolution.
 217   class DNSResolver {
 218    public:
 219     /// A task handle for DNS Resolution requests.
 220     struct LookupTaskHandle {
 221       intptr_t key[2];
 222     };
 223     /// A DNS SRV record type.
 224     struct SRVRecord {
 225       std::string host;
 226       int port = 0;
 227       int priority = 0;
 228       int weight = 0;
 229     };
 230     /// Called with the collection of sockaddrs that were resolved from a given
 231     /// target address.
 232     using LookupHostnameCallback =
 233         std::function<void(absl::StatusOr<std::vector<ResolvedAddress>>)>;
 234     /// Called with a collection of SRV records.
 235     using LookupSRVCallback =
 236         std::function<void(absl::StatusOr<std::vector<SRVRecord>>)>;
 237     /// Called with the result of a TXT record lookup
 238     using LookupTXTCallback = std::function<void(absl::StatusOr<std::string>)>;
 239
 240     virtual ~DNSResolver() = default;
 241
 242     /// Asynchronously resolve an address.
 243     ///
 244     /// \a default_port may be a non-numeric named service port, and will only
 245     /// be used if \a address does not already contain a port component.
 246     ///
 247     /// When the lookup is complete, the \a on_resolve callback will be invoked
 248     /// with a status indicating the success or failure of the lookup.
 249     /// Implementations should pass the appropriate statuses to the callback.
 250     /// For example, callbacks might expect to receive DEADLINE_EXCEEDED when
 251     /// the deadline is exceeded or CANCELLED if the lookup was cancelled.
 252     virtual LookupTaskHandle LookupHostname(LookupHostnameCallback on_resolve,
 253                                             absl::string_view address,
 254                                             absl::string_view default_port,
 255                                             absl::Time deadline) = 0;
 256     /// Asynchronously perform an SRV record lookup.
 257     ///
 258     /// \a on_resolve has the same meaning and expectations as \a
 259     /// LookupHostname's \a on_resolve callback.
 260     virtual LookupTaskHandle LookupSRV(LookupSRVCallback on_resolve,
 261                                        absl::string_view name,
 262                                        absl::Time deadline) = 0;
 263     /// Asynchronously perform a TXT record lookup.
 264     ///
 265     /// \a on_resolve has the same meaning and expectations as \a
 266     /// LookupHostname's \a on_resolve callback.
 267     virtual LookupTaskHandle LookupTXT(LookupTXTCallback on_resolve,
 268                                        absl::string_view name,
 269                                        absl::Time deadline) = 0;
 270     /// Cancel an asynchronous lookup operation.
 271     virtual void TryCancelLookup(LookupTaskHandle handle) = 0;
 272   };
 273
 274   virtual ~EventEngine() = default;
 275
 276   // TODO(nnoble): consider whether we can remove this method before we
 277   // de-experimentalize this API.
 278   virtual bool IsWorkerThread() = 0;
 279
 280   // TODO(hork): define return status codes
 281   /// Retrieves an instance of a DNSResolver.
 282   virtual absl::StatusOr<std::unique_ptr<DNSResolver>> GetDNSResolver() = 0;
 283
 284   /// Intended for future expansion of Task run functionality.
 285   struct RunOptions {};
 286   /// Run a callback as soon as possible.
 287   ///
 288   /// The \a fn callback's \a status argument is used to indicate whether it was
 289   /// executed normally. For example, the status may be CANCELLED if
 290   /// \a TryCancel was called, or if the EventEngine is being shut down.
 291   virtual TaskHandle Run(Callback fn, RunOptions opts) = 0;
 292   /// Synonymous with scheduling an alarm to run at time \a when.
 293   ///
 294   /// The callback \a fn will execute when either when time \a when arrives
 295   /// (receiving status OK), or when the \a fn is cancelled (reveiving status
 296   /// CANCELLED). The callback is guaranteed to be called exactly once.
 297   virtual TaskHandle RunAt(absl::Time when, Callback fn, RunOptions opts) = 0;
 298   /// Immediately tries to cancel a callback.
 299   /// Note that this is a "best effort" cancellation. No guarantee is made that
 300   /// the callback will be cancelled, the call could be in any stage.
 301   ///
 302   /// There are three scenarios in which we may cancel a scheduled function:
 303   ///   1. We cancel the execution before it has run.
 304   ///   2. The callback has already run.
 305   ///   3. We can't cancel it because it is "in flight".
 306   ///
 307   /// In all cases, the cancellation is still considered successful, the
 308   /// callback will be run exactly once from either cancellation or from its
 309   /// activation.
 310   virtual void TryCancel(TaskHandle handle) = 0;
 311   /// Immediately run all callbacks with status indicating the shutdown. Every
 312   /// EventEngine is expected to shut down exactly once. No new callbacks/tasks
 313   /// should be scheduled after shutdown has begun, no new connections should be
 314   /// created.
 315   ///
 316   /// If the \a on_shutdown_complete callback is given a non-OK status, errors
 317   /// are expected to be unrecoverable. For example, an implementation could
 318   /// warn callers about leaks if memory cannot be freed within a certain
 319   /// timeframe.
 320   virtual void Shutdown(Callback on_shutdown_complete) = 0;
 321 };
 322
 323 // TODO(hork): finalize the API and document it. We need to firm up the story
 324 // around user-provided EventEngines.
 325 std::shared_ptr<EventEngine> DefaultEventEngineFactory();
 326
 327 }  // namespace experimental
 328 }  // namespace grpc_event_engine
 329
 330 #endif  // GRPC_EVENT_ENGINE_EVENT_ENGINE_H