event_engine.h

Go to the documentation of this file.

// Copyright 2021 The gRPC Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#ifndef GRPC_EVENT_ENGINE_EVENT_ENGINE_H
#define GRPC_EVENT_ENGINE_EVENT_ENGINE_H
 
#include <vector>
 
#include "absl/functional/any_invocable.h"
#include "absl/status/status.h"
#include "absl/status/statusor.h"
 
#include <grpc/event_engine/endpoint_config.h>
#include <grpc/event_engine/extensible.h>
#include <grpc/event_engine/memory_allocator.h>
#include <grpc/event_engine/port.h>
#include <grpc/event_engine/slice_buffer.h>
#include <grpc/support/port_platform.h>
 
// TODO(vigneshbabu): Define the Endpoint::Write metrics collection system
namespace grpc_event_engine {
namespace experimental {
 
class EventEngine : public std::enable_shared_from_this<EventEngine>,
                    public Extensible {
 public:
  using Duration = std::chrono::duration<int64_t, std::nano>;
 
  class Closure {
   public:
    Closure() = default;
    // Closure's are an interface, and thus non-copyable.
    Closure(const Closure&) = delete;
    Closure& operator=(const Closure&) = delete;
    // Polymorphic type => virtual destructor
    virtual ~Closure() = default;
    // Run the contained code.
    virtual void Run() = 0;
  };
  struct TaskHandle {
    intptr_t keys[2];
    static const GRPC_DLL TaskHandle kInvalid;
  };
  struct ConnectionHandle {
    intptr_t keys[2];
    static const GRPC_DLL ConnectionHandle kInvalid;
  };
  class ResolvedAddress {
   public:
    static constexpr socklen_t MAX_SIZE_BYTES = 128;
 
    ResolvedAddress(const sockaddr* address, socklen_t size);
    ResolvedAddress() = default;
    ResolvedAddress(const ResolvedAddress&) = default;
    const struct sockaddr* address() const;
    socklen_t size() const;
 
   private:
    char address_[MAX_SIZE_BYTES] = {};
    socklen_t size_ = 0;
  };
 
  class Endpoint : public Extensible {
   public:
    virtual ~Endpoint() = default;
    struct ReadArgs {
      // A suggestion to the endpoint implementation to read at-least the
      // specified number of bytes over the network connection before marking
      // the endpoint read operation as complete. gRPC may use this argument
      // to minimize the number of endpoint read API calls over the lifetime
      // of a connection.
      int64_t read_hint_bytes;
    };
    virtual bool Read(absl::AnyInvocable<void(absl::Status)> on_read,
                      SliceBuffer* buffer, const ReadArgs* args) = 0;
    struct WriteArgs {
      // Represents private information that may be passed by gRPC for
      // select endpoints expected to be used only within google.
      void* google_specific = nullptr;
      // A suggestion to the endpoint implementation to group data to be written
      // into frames of the specified max_frame_size. gRPC may use this
      // argument to dynamically control the max sizes of frames sent to a
      // receiver in response to high receiver memory pressure.
      int64_t max_frame_size;
    };
    virtual bool Write(absl::AnyInvocable<void(absl::Status)> on_writable,
                       SliceBuffer* data, const WriteArgs* args) = 0;
    virtual const ResolvedAddress& GetPeerAddress() const = 0;
    virtual const ResolvedAddress& GetLocalAddress() const = 0;
  };
 
  using OnConnectCallback =
      absl::AnyInvocable<void(absl::StatusOr<std::unique_ptr<Endpoint>>)>;
 
  class Listener : public Extensible {
   public:
    using AcceptCallback = absl::AnyInvocable<void(
        std::unique_ptr<Endpoint>, MemoryAllocator memory_allocator)>;
    virtual ~Listener() = default;
    virtual absl::StatusOr<int> Bind(const ResolvedAddress& addr) = 0;
    virtual absl::Status Start() = 0;
  };
 
  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;
  virtual ConnectionHandle Connect(OnConnectCallback on_connect,
                                   const ResolvedAddress& addr,
                                   const EndpointConfig& args,
                                   MemoryAllocator memory_allocator,
                                   Duration timeout) = 0;
 
  virtual bool CancelConnect(ConnectionHandle handle) = 0;
  class DNSResolver {
   public:
    struct ResolverOptions {
      std::string dns_server;
    };
    struct SRVRecord {
      std::string host;
      int port = 0;
      int priority = 0;
      int weight = 0;
    };
    using LookupHostnameCallback =
        absl::AnyInvocable<void(absl::StatusOr<std::vector<ResolvedAddress>>)>;
    using LookupSRVCallback =
        absl::AnyInvocable<void(absl::StatusOr<std::vector<SRVRecord>>)>;
    using LookupTXTCallback =
        absl::AnyInvocable<void(absl::StatusOr<std::vector<std::string>>)>;
 
    virtual ~DNSResolver() = default;
 
    virtual void LookupHostname(LookupHostnameCallback on_resolve,
                                absl::string_view name,
                                absl::string_view default_port) = 0;
    virtual void LookupSRV(LookupSRVCallback on_resolve,
                           absl::string_view name) = 0;
    virtual void LookupTXT(LookupTXTCallback on_resolve,
                           absl::string_view name) = 0;
  };
 
  virtual ~EventEngine() = default;
 
  // TODO(nnoble): consider whether we can remove this method before we
  // de-experimentalize this API.
  virtual bool IsWorkerThread() = 0;
 
  virtual absl::StatusOr<std::unique_ptr<DNSResolver>> GetDNSResolver(
      const DNSResolver::ResolverOptions& options) = 0;
 
  virtual void Run(Closure* closure) = 0;
  virtual void Run(absl::AnyInvocable<void()> closure) = 0;
  virtual TaskHandle RunAfter(Duration when, Closure* closure) = 0;
  virtual TaskHandle RunAfter(Duration when,
                              absl::AnyInvocable<void()> closure) = 0;
  virtual bool Cancel(TaskHandle handle) = 0;
};
 
void SetEventEngineFactory(
    absl::AnyInvocable<std::unique_ptr<EventEngine>()> factory);
 
void EventEngineFactoryReset();
std::unique_ptr<EventEngine> CreateEventEngine();
 
bool operator==(const EventEngine::TaskHandle& lhs,
                const EventEngine::TaskHandle& rhs);
bool operator!=(const EventEngine::TaskHandle& lhs,
                const EventEngine::TaskHandle& rhs);
std::ostream& operator<<(std::ostream& out,
                         const EventEngine::TaskHandle& handle);
bool operator==(const EventEngine::ConnectionHandle& lhs,
                const EventEngine::ConnectionHandle& rhs);
bool operator!=(const EventEngine::ConnectionHandle& lhs,
                const EventEngine::ConnectionHandle& rhs);
std::ostream& operator<<(std::ostream& out,
                         const EventEngine::ConnectionHandle& handle);
 
namespace detail {
std::string FormatHandleString(uint64_t key1, uint64_t key2);
}
 
template <typename Sink>
void AbslStringify(Sink& out, const EventEngine::ConnectionHandle& handle) {
  out.Append(detail::FormatHandleString(handle.keys[0], handle.keys[1]));
}
 
template <typename Sink>
void AbslStringify(Sink& out, const EventEngine::TaskHandle& handle) {
  out.Append(detail::FormatHandleString(handle.keys[0], handle.keys[1]));
}
 
}  // namespace experimental
}  // namespace grpc_event_engine
 
#endif  // GRPC_EVENT_ENGINE_EVENT_ENGINE_H