GRPC C++  1.71.0
event_engine.h
Go to the documentation of this file.
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 
20 #include <grpc/event_engine/port.h>
23 
24 #include <vector>
25 
26 #include "absl/functional/any_invocable.h"
27 #include "absl/status/status.h"
28 #include "absl/status/statusor.h"
29 
30 // TODO(vigneshbabu): Define the Endpoint::Write metrics collection system
31 // TODO(hork): remove all references to the factory methods.
32 namespace grpc_event_engine {
33 namespace experimental {
34 
103 class EventEngine : public std::enable_shared_from_this<EventEngine>,
104  public Extensible {
105  public:
110  using Duration = std::chrono::duration<int64_t, std::nano>;
117 
118  class Closure {
119  public:
120  Closure() = default;
121  // Closure's are an interface, and thus non-copyable.
122  Closure(const Closure&) = delete;
123  Closure& operator=(const Closure&) = delete;
124  // Polymorphic type => virtual destructor
125  virtual ~Closure() = default;
126  // Run the contained code.
127  virtual void Run() = 0;
128  };
133  struct TaskHandle {
134  intptr_t keys[2];
136  };
141  intptr_t keys[2];
143  };
152  public:
153  static constexpr socklen_t MAX_SIZE_BYTES = 128;
154 
155  ResolvedAddress(const sockaddr* address, socklen_t size);
156  ResolvedAddress() = default;
157  ResolvedAddress(const ResolvedAddress&) = default;
158  const struct sockaddr* address() const;
159  socklen_t size() const;
160 
161  private:
162  char address_[MAX_SIZE_BYTES] = {};
163  socklen_t size_ = 0;
164  };
165 
174  class Endpoint : public Extensible {
175  public:
178  virtual ~Endpoint() = default;
183  struct ReadArgs {
184  // A suggestion to the endpoint implementation to read at-least the
185  // specified number of bytes over the network connection before marking
186  // the endpoint read operation as complete. gRPC may use this argument
187  // to minimize the number of endpoint read API calls over the lifetime
188  // of a connection.
190  };
214  virtual bool Read(absl::AnyInvocable<void(absl::Status)> on_read,
215  SliceBuffer* buffer, const ReadArgs* args) = 0;
220  struct WriteArgs {
221  // Represents private information that may be passed by gRPC for
222  // select endpoints expected to be used only within google.
223  void* google_specific = nullptr;
224  // A suggestion to the endpoint implementation to group data to be written
225  // into frames of the specified max_frame_size. gRPC may use this
226  // argument to dynamically control the max sizes of frames sent to a
227  // receiver in response to high receiver memory pressure.
228  int64_t max_frame_size;
229  };
250  virtual bool Write(absl::AnyInvocable<void(absl::Status)> on_writable,
251  SliceBuffer* data, const WriteArgs* args) = 0;
254  virtual const ResolvedAddress& GetPeerAddress() const = 0;
255  virtual const ResolvedAddress& GetLocalAddress() const = 0;
256  };
257 
264  using OnConnectCallback =
265  absl::AnyInvocable<void(absl::StatusOr<std::unique_ptr<Endpoint>>)>;
266 
269  class Listener : public Extensible {
270  public:
272  using AcceptCallback = absl::AnyInvocable<void(
273  std::unique_ptr<Endpoint>, MemoryAllocator memory_allocator)>;
274  virtual ~Listener() = default;
280  virtual absl::StatusOr<int> Bind(const ResolvedAddress& addr) = 0;
281  virtual absl::Status Start() = 0;
282  };
283 
299  virtual absl::StatusOr<std::unique_ptr<Listener>> CreateListener(
300  Listener::AcceptCallback on_accept,
301  absl::AnyInvocable<void(absl::Status)> on_shutdown,
302  const EndpointConfig& config,
303  std::unique_ptr<MemoryAllocatorFactory> memory_allocator_factory) = 0;
315  virtual ConnectionHandle Connect(OnConnectCallback on_connect,
316  const ResolvedAddress& addr,
317  const EndpointConfig& args,
318  MemoryAllocator memory_allocator,
319  Duration timeout) = 0;
320 
329  virtual bool CancelConnect(ConnectionHandle handle) = 0;
337  class DNSResolver {
338  public:
343  std::string dns_server;
344  };
346  struct SRVRecord {
347  std::string host;
348  int port = 0;
349  int priority = 0;
350  int weight = 0;
351  };
354  using LookupHostnameCallback =
355  absl::AnyInvocable<void(absl::StatusOr<std::vector<ResolvedAddress>>)>;
357  using LookupSRVCallback =
358  absl::AnyInvocable<void(absl::StatusOr<std::vector<SRVRecord>>)>;
360  using LookupTXTCallback =
361  absl::AnyInvocable<void(absl::StatusOr<std::vector<std::string>>)>;
362 
363  virtual ~DNSResolver() = default;
364 
375  virtual void LookupHostname(LookupHostnameCallback on_resolve,
376  absl::string_view name,
377  absl::string_view default_port) = 0;
382  virtual void LookupSRV(LookupSRVCallback on_resolve,
383  absl::string_view name) = 0;
388  virtual void LookupTXT(LookupTXTCallback on_resolve,
389  absl::string_view name) = 0;
390  };
391 
399  virtual ~EventEngine() = default;
400 
401  // TODO(nnoble): consider whether we can remove this method before we
402  // de-experimentalize this API.
403  virtual bool IsWorkerThread() = 0;
404 
410  virtual absl::StatusOr<std::unique_ptr<DNSResolver>> GetDNSResolver(
411  const DNSResolver::ResolverOptions& options) = 0;
412 
422  virtual void Run(Closure* closure) = 0;
435  virtual void Run(absl::AnyInvocable<void()> closure) = 0;
447  virtual TaskHandle RunAfter(Duration when, Closure* closure) = 0;
462  virtual TaskHandle RunAfter(Duration when,
463  absl::AnyInvocable<void()> closure) = 0;
472  virtual bool Cancel(TaskHandle handle) = 0;
473 };
474 
484 // TODO(hork): delete once all known users have migrated away
486  absl::AnyInvocable<std::shared_ptr<EventEngine>()> factory);
487 
494 //
495 // TODO(hork): delete once all known users have migrated away
497 
499 std::shared_ptr<EventEngine> CreateEventEngine();
500 
510 void SetDefaultEventEngine(std::shared_ptr<EventEngine> engine);
511 
520 std::shared_ptr<EventEngine> GetDefaultEventEngine();
521 
534 
535 bool operator==(const EventEngine::TaskHandle& lhs,
536  const EventEngine::TaskHandle& rhs);
537 bool operator!=(const EventEngine::TaskHandle& lhs,
538  const EventEngine::TaskHandle& rhs);
539 std::ostream& operator<<(std::ostream& out,
540  const EventEngine::TaskHandle& handle);
542  const EventEngine::ConnectionHandle& rhs);
544  const EventEngine::ConnectionHandle& rhs);
545 std::ostream& operator<<(std::ostream& out,
546  const EventEngine::ConnectionHandle& handle);
547 
548 namespace detail {
549 std::string FormatHandleString(uint64_t key1, uint64_t key2);
550 }
551 
552 template <typename Sink>
553 void AbslStringify(Sink& out, const EventEngine::ConnectionHandle& handle) {
554  out.Append(detail::FormatHandleString(handle.keys[0], handle.keys[1]));
555 }
556 
557 template <typename Sink>
558 void AbslStringify(Sink& out, const EventEngine::TaskHandle& handle) {
559  out.Append(detail::FormatHandleString(handle.keys[0], handle.keys[1]));
560 }
561 
562 } // namespace experimental
563 } // namespace grpc_event_engine
564 
565 #endif // GRPC_EVENT_ENGINE_EVENT_ENGINE_H
grpc_event_engine::experimental::EventEngine::OnConnectCallback
absl::AnyInvocable< void(absl::StatusOr< std::unique_ptr< Endpoint > >)> OnConnectCallback
Called when a new connection is established.
Definition: event_engine.h:265
grpc_event_engine::experimental::EventEngine::~EventEngine
virtual ~EventEngine()=default
At time of destruction, the EventEngine must have no active responsibilities.
grpc_event_engine::experimental::operator<<
std::ostream & operator<<(std::ostream &out, const EventEngine::TaskHandle &handle)
grpc_event_engine::experimental::EventEngine::ResolvedAddress::size
socklen_t size() const
grpc_event_engine::experimental::ShutdownDefaultEventEngine
void ShutdownDefaultEventEngine()
Resets gRPC to use one of the default internal EventEngines for all new GetDefaultEventEngine request...
grpc_event_engine::experimental::EventEngine::ConnectionHandle
A handle to a cancellable connection attempt.
Definition: event_engine.h:140
port.h
grpc_event_engine::experimental::EventEngine::Endpoint::WriteArgs
A struct representing optional arguments that may be provided to an EventEngine Endpoint Write API ca...
Definition: event_engine.h:220
grpc_event_engine::experimental::GetDefaultEventEngine
std::shared_ptr< EventEngine > GetDefaultEventEngine()
Returns the default EventEngine instance.
grpc_event_engine::experimental::EventEngine::ResolvedAddress::MAX_SIZE_BYTES
static constexpr socklen_t MAX_SIZE_BYTES
Definition: event_engine.h:153
extensible.h
grpc_event_engine::experimental::EventEngine::RunAfter
virtual TaskHandle RunAfter(Duration when, Closure *closure)=0
Synonymous with scheduling an alarm to run after duration when.
grpc_event_engine::experimental::EventEngine::Listener::AcceptCallback
absl::AnyInvocable< void(std::unique_ptr< Endpoint >, MemoryAllocator memory_allocator)> AcceptCallback
Called when the listener has accepted a new client connection.
Definition: event_engine.h:273
grpc_event_engine::experimental::EventEngine
The EventEngine Interface.
Definition: event_engine.h:103
grpc_event_engine::experimental::EventEngine::Endpoint::ReadArgs::read_hint_bytes
int64_t read_hint_bytes
Definition: event_engine.h:189
grpc_event_engine::experimental::EventEngine::Endpoint::~Endpoint
virtual ~Endpoint()=default
Shuts down all connections and invokes all pending read or write callbacks with an error status.
grpc_event_engine::experimental::EventEngine::IsWorkerThread
virtual bool IsWorkerThread()=0
grpc_event_engine::experimental::EventEngine::DNSResolver::~DNSResolver
virtual ~DNSResolver()=default
grpc_event_engine::experimental::EventEngine::Endpoint::Read
virtual bool Read(absl::AnyInvocable< void(absl::Status)> on_read, SliceBuffer *buffer, const ReadArgs *args)=0
Reads data from the Endpoint.
grpc_event_engine::experimental::EventEngine::DNSResolver::LookupHostname
virtual void LookupHostname(LookupHostnameCallback on_resolve, absl::string_view name, absl::string_view default_port)=0
Asynchronously resolve an address.
grpc_event_engine::experimental::AbslStringify
void AbslStringify(Sink &out, const EventEngine::ConnectionHandle &handle)
Definition: event_engine.h:553
grpc_event_engine::experimental::EventEngine::DNSResolver::LookupSRVCallback
absl::AnyInvocable< void(absl::StatusOr< std::vector< SRVRecord > >)> LookupSRVCallback
Called with a collection of SRV records.
Definition: event_engine.h:358
grpc_event_engine::experimental::EventEngine::DNSResolver::SRVRecord::port
int port
Definition: event_engine.h:348
grpc_event_engine::experimental::EventEngine::Endpoint::ReadArgs
A struct representing optional arguments that may be provided to an EventEngine Endpoint Read API cal...
Definition: event_engine.h:183
grpc_event_engine::experimental::EventEngine::Listener::Bind
virtual absl::StatusOr< int > Bind(const ResolvedAddress &addr)=0
Bind an address/port to this Listener.
grpc_event_engine::experimental::EventEngine::DNSResolver::SRVRecord::host
std::string host
Definition: event_engine.h:347
grpc_event_engine::experimental::EventEngine::ConnectionHandle::keys
intptr_t keys[2]
Definition: event_engine.h:141
grpc_event_engine::experimental::EventEngineFactoryReset
void EventEngineFactoryReset()
[DEPRECATED] Reset gRPC's EventEngine factory to the built-in default.
grpc_event_engine::experimental::EventEngine::DNSResolver::LookupHostnameCallback
absl::AnyInvocable< void(absl::StatusOr< std::vector< ResolvedAddress > >)> LookupHostnameCallback
Called with the collection of sockaddrs that were resolved from a given target address.
Definition: event_engine.h:355
grpc_event_engine::experimental::EventEngine::Endpoint::Write
virtual bool Write(absl::AnyInvocable< void(absl::Status)> on_writable, SliceBuffer *data, const WriteArgs *args)=0
Writes data out on the connection.
grpc_event_engine::experimental::EventEngine::CancelConnect
virtual bool CancelConnect(ConnectionHandle handle)=0
Request cancellation of a connection attempt.
grpc_event_engine::experimental::operator!=
bool operator!=(const EventEngine::TaskHandle &lhs, const EventEngine::TaskHandle &rhs)
grpc_event_engine::experimental::EventEngine::DNSResolver::ResolverOptions
Optional configuration for DNSResolvers.
Definition: event_engine.h:340
grpc_event_engine::experimental::EventEngine::Endpoint::WriteArgs::max_frame_size
int64_t max_frame_size
Definition: event_engine.h:228
grpc_event_engine::experimental::EventEngine::DNSResolver::SRVRecord
DNS SRV record type.
Definition: event_engine.h:346
grpc_event_engine::experimental::EventEngine::DNSResolver::LookupTXT
virtual void LookupTXT(LookupTXTCallback on_resolve, absl::string_view name)=0
Asynchronously perform a TXT record lookup.
grpc_event_engine::experimental::EndpointConfig
Collection of parameters used to configure client and server endpoints.
Definition: endpoint_config.h:31
grpc_event_engine::experimental::EventEngine::ResolvedAddress
Thin wrapper around a platform-specific sockaddr type.
Definition: event_engine.h:151
grpc_event_engine::experimental::EventEngine::Listener
Listens for incoming connection requests from gRPC clients and initiates request processing once conn...
Definition: event_engine.h:269
grpc_event_engine::experimental::SliceBuffer
A Wrapper around grpc_slice_buffer pointer.
Definition: slice_buffer.h:50
grpc_event_engine::experimental::detail::FormatHandleString
std::string FormatHandleString(uint64_t key1, uint64_t key2)
grpc_event_engine::experimental::EventEngine::Closure
A custom closure type for EventEngine task execution.
Definition: event_engine.h:118
grpc_event_engine::experimental::EventEngine::Closure::Closure
Closure()=default
grpc_event_engine::experimental::EventEngine::ResolvedAddress::ResolvedAddress
ResolvedAddress()=default
grpc_event_engine::experimental::EventEngine::ResolvedAddress::address
const struct sockaddr * address() const
grpc_event_engine::experimental::EventEngine::Listener::Start
virtual absl::Status Start()=0
grpc_event_engine::experimental::EventEngine::Closure::operator=
Closure & operator=(const Closure &)=delete
grpc_event_engine::experimental::EventEngine::Closure::~Closure
virtual ~Closure()=default
grpc_event_engine::experimental::EventEngine::Run
virtual void Run(Closure *closure)=0
Asynchronously executes a task as soon as possible.
slice_buffer.h
grpc_event_engine::experimental::EventEngine::Endpoint::WriteArgs::google_specific
void * google_specific
Definition: event_engine.h:223
grpc_event_engine::experimental::EventEngine::DNSResolver::ResolverOptions::dns_server
std::string dns_server
If empty, default DNS servers will be used.
Definition: event_engine.h:343
grpc_event_engine::experimental::EventEngine::DNSResolver::SRVRecord::weight
int weight
Definition: event_engine.h:350
grpc_event_engine::experimental::SetEventEngineFactory
void SetEventEngineFactory(absl::AnyInvocable< std::shared_ptr< EventEngine >()> factory)
[DEPRECATED] Replace gRPC's default EventEngine factory.
grpc_event_engine::experimental::EventEngine::Endpoint::GetPeerAddress
virtual const ResolvedAddress & GetPeerAddress() const =0
Returns an address in the format described in DNSResolver.
grpc_event_engine::experimental::EventEngine::Closure::Run
virtual void Run()=0
grpc_event_engine::experimental::EventEngine::DNSResolver
Provides asynchronous resolution.
Definition: event_engine.h:337
grpc_event_engine::experimental::Extensible
Definition: extensible.h:25
grpc_event_engine::experimental::EventEngine::TaskHandle
Represents a scheduled task.
Definition: event_engine.h:133
grpc_event_engine::experimental::EventEngine::Cancel
virtual bool Cancel(TaskHandle handle)=0
Request cancellation of a task.
grpc_event_engine::experimental::SetDefaultEventEngine
void SetDefaultEventEngine(std::shared_ptr< EventEngine > engine)
Set the default EventEngine instance, which will be used throughout gRPC.
grpc_event_engine::experimental::CreateEventEngine
std::shared_ptr< EventEngine > CreateEventEngine()
Create a new EventEngine instance.
grpc_event_engine::experimental::EventEngine::DNSResolver::SRVRecord::priority
int priority
Definition: event_engine.h:349
grpc_event_engine::experimental::EventEngine::GetDNSResolver
virtual absl::StatusOr< std::unique_ptr< DNSResolver > > GetDNSResolver(const DNSResolver::ResolverOptions &options)=0
Creates and returns an instance of a DNSResolver, optionally configured by the options struct.
grpc_event_engine::experimental::EventEngine::TaskHandle::kInvalid
static const GRPC_DLL TaskHandle kInvalid
Definition: event_engine.h:135
grpc_event_engine
Definition: endpoint_config.h:24
grpc_event_engine::experimental::EventEngine::DNSResolver::LookupTXTCallback
absl::AnyInvocable< void(absl::StatusOr< std::vector< std::string > >)> LookupTXTCallback
Called with the result of a TXT record lookup.
Definition: event_engine.h:361
grpc_event_engine::experimental::EventEngine::ConnectionHandle::kInvalid
static const GRPC_DLL ConnectionHandle kInvalid
Definition: event_engine.h:142
grpc_event_engine::experimental::EventEngine::Endpoint
One end of a connection between a gRPC client and server.
Definition: event_engine.h:174
grpc_event_engine::experimental::MemoryAllocator
Definition: memory_allocator.h:33
grpc_event_engine::experimental::EventEngine::Connect
virtual ConnectionHandle Connect(OnConnectCallback on_connect, const ResolvedAddress &addr, const EndpointConfig &args, MemoryAllocator memory_allocator, Duration timeout)=0
Creates a client network connection to a remote network listener.
memory_allocator.h
GRPC_DLL
#define GRPC_DLL
Definition: port_platform.h:127
grpc_event_engine::experimental::EventEngine::TaskHandle::keys
intptr_t keys[2]
Definition: event_engine.h:134
endpoint_config.h
grpc_event_engine::experimental::EventEngine::Listener::~Listener
virtual ~Listener()=default
grpc::protobuf::util::Status
::absl::Status Status
Definition: config_protobuf.h:107
grpc_event_engine::experimental::EventEngine::DNSResolver::LookupSRV
virtual void LookupSRV(LookupSRVCallback on_resolve, absl::string_view name)=0
Asynchronously perform an SRV record lookup.
grpc_event_engine::experimental::EventEngine::CreateListener
virtual absl::StatusOr< std::unique_ptr< Listener > > CreateListener(Listener::AcceptCallback on_accept, absl::AnyInvocable< void(absl::Status)> on_shutdown, const EndpointConfig &config, std::unique_ptr< MemoryAllocatorFactory > memory_allocator_factory)=0
Factory method to create a network listener / server.
grpc_event_engine::experimental::EventEngine::Duration
std::chrono::duration< int64_t, std::nano > Duration
A duration between two events.
Definition: event_engine.h:110
grpc_event_engine::experimental::EventEngine::Endpoint::GetLocalAddress
virtual const ResolvedAddress & GetLocalAddress() const =0
grpc_event_engine::experimental::operator==
bool operator==(const EventEngine::TaskHandle &lhs, const EventEngine::TaskHandle &rhs)
port_platform.h