|
|
|
|
@ -33,7 +33,9 @@ |
|
|
|
|
#include <grpc/event_engine/endpoint_config.h> |
|
|
|
|
#include <grpc/event_engine/event_engine.h> |
|
|
|
|
#include <grpc/event_engine/memory_allocator.h> |
|
|
|
|
#include <grpc/event_engine/slice_buffer.h> |
|
|
|
|
|
|
|
|
|
#include "src/core/lib/event_engine/posix.h" |
|
|
|
|
#include "src/core/lib/iomgr/port.h" |
|
|
|
|
|
|
|
|
|
#ifdef GRPC_POSIX_SOCKET_TCP |
|
|
|
|
@ -52,7 +54,7 @@ class PosixEngineListenerImpl |
|
|
|
|
: public std::enable_shared_from_this<PosixEngineListenerImpl> { |
|
|
|
|
public: |
|
|
|
|
PosixEngineListenerImpl( |
|
|
|
|
EventEngine::Listener::AcceptCallback on_accept, |
|
|
|
|
PosixEventEngineWithFdSupport::PosixAcceptCallback on_accept, |
|
|
|
|
absl::AnyInvocable<void(absl::Status)> on_shutdown, |
|
|
|
|
const grpc_event_engine::experimental::EndpointConfig& config, |
|
|
|
|
std::unique_ptr<grpc_event_engine::experimental::MemoryAllocatorFactory> |
|
|
|
|
@ -60,12 +62,17 @@ class PosixEngineListenerImpl |
|
|
|
|
PosixEventPoller* poller, std::shared_ptr<EventEngine> engine); |
|
|
|
|
// Binds an address to the listener. This creates a ListenerSocket
|
|
|
|
|
// and sets its fields appropriately.
|
|
|
|
|
absl::StatusOr<int> Bind(const EventEngine::ResolvedAddress& addr); |
|
|
|
|
absl::StatusOr<int> Bind( |
|
|
|
|
const EventEngine::ResolvedAddress& addr, |
|
|
|
|
PosixListenerWithFdSupport::OnPosixBindNewFdCallback on_bind_new_fd); |
|
|
|
|
// Signals event manager to listen for connections on all created sockets.
|
|
|
|
|
absl::Status Start(); |
|
|
|
|
// Trigger graceful shutdown of all asynchronous accept operations.
|
|
|
|
|
void TriggerShutdown(); |
|
|
|
|
|
|
|
|
|
absl::Status HandleExternalConnection(int listener_fd, int fd, |
|
|
|
|
SliceBuffer* pending_data); |
|
|
|
|
|
|
|
|
|
~PosixEngineListenerImpl(); |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
@ -119,9 +126,18 @@ class PosixEngineListenerImpl |
|
|
|
|
public: |
|
|
|
|
explicit ListenerAsyncAcceptors(PosixEngineListenerImpl* listener) |
|
|
|
|
: listener_(listener){}; |
|
|
|
|
|
|
|
|
|
void UpdateOnAppendCallback( |
|
|
|
|
PosixListenerWithFdSupport::OnPosixBindNewFdCallback on_append) { |
|
|
|
|
on_append_ = std::move(on_append); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void Append(ListenerSocket socket) override { |
|
|
|
|
acceptors_.push_back(new AsyncConnectionAcceptor( |
|
|
|
|
listener_->engine_, listener_->shared_from_this(), socket)); |
|
|
|
|
if (on_append_) { |
|
|
|
|
on_append_(socket.sock.Fd()); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
absl::StatusOr<ListenerSocket> Find( |
|
|
|
|
@ -148,6 +164,7 @@ class PosixEngineListenerImpl |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
PosixListenerWithFdSupport::OnPosixBindNewFdCallback on_append_; |
|
|
|
|
std::list<AsyncConnectionAcceptor*> acceptors_; |
|
|
|
|
PosixEngineListenerImpl* listener_; |
|
|
|
|
}; |
|
|
|
|
@ -163,7 +180,7 @@ class PosixEngineListenerImpl |
|
|
|
|
// operation.
|
|
|
|
|
ListenerAsyncAcceptors acceptors_ ABSL_GUARDED_BY(mu_); |
|
|
|
|
// Callback to be invoked upon accepting a connection.
|
|
|
|
|
EventEngine::Listener::AcceptCallback on_accept_; |
|
|
|
|
PosixEventEngineWithFdSupport::PosixAcceptCallback on_accept_; |
|
|
|
|
// Callback to be invoked upon shutdown of listener.
|
|
|
|
|
absl::AnyInvocable<void(absl::Status)> on_shutdown_; |
|
|
|
|
// Set to true when the listener has started listening for new connections.
|
|
|
|
|
@ -175,12 +192,10 @@ class PosixEngineListenerImpl |
|
|
|
|
memory_allocator_factory_; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
class PosixEngineListener |
|
|
|
|
: public grpc_event_engine::experimental::EventEngine::Listener { |
|
|
|
|
class PosixEngineListener : public PosixListenerWithFdSupport { |
|
|
|
|
public: |
|
|
|
|
PosixEngineListener( |
|
|
|
|
grpc_event_engine::experimental::EventEngine::Listener::AcceptCallback |
|
|
|
|
on_accept, |
|
|
|
|
PosixEventEngineWithFdSupport::PosixAcceptCallback on_accept, |
|
|
|
|
absl::AnyInvocable<void(absl::Status)> on_shutdown, |
|
|
|
|
const grpc_event_engine::experimental::EndpointConfig& config, |
|
|
|
|
std::unique_ptr<grpc_event_engine::experimental::MemoryAllocatorFactory> |
|
|
|
|
@ -189,16 +204,34 @@ class PosixEngineListener |
|
|
|
|
: impl_(std::make_shared<PosixEngineListenerImpl>( |
|
|
|
|
std::move(on_accept), std::move(on_shutdown), config, |
|
|
|
|
std::move(memory_allocator_factory), poller, std::move(engine))) {} |
|
|
|
|
~PosixEngineListener() override { impl_->TriggerShutdown(); }; |
|
|
|
|
~PosixEngineListener() override { ShutdownListeningFds(); }; |
|
|
|
|
absl::StatusOr<int> Bind( |
|
|
|
|
const grpc_event_engine::experimental::EventEngine::ResolvedAddress& addr) |
|
|
|
|
override { |
|
|
|
|
return impl_->Bind(addr); |
|
|
|
|
return impl_->Bind(addr, nullptr); |
|
|
|
|
} |
|
|
|
|
absl::StatusOr<int> BindWithFd( |
|
|
|
|
const EventEngine::ResolvedAddress& addr, |
|
|
|
|
PosixListenerWithFdSupport::OnPosixBindNewFdCallback on_bind_new_fd) |
|
|
|
|
override { |
|
|
|
|
return impl_->Bind(addr, std::move(on_bind_new_fd)); |
|
|
|
|
} |
|
|
|
|
absl::Status HandleExternalConnection(int listener_fd, int fd, |
|
|
|
|
SliceBuffer* pending_data) override { |
|
|
|
|
return impl_->HandleExternalConnection(listener_fd, fd, pending_data); |
|
|
|
|
} |
|
|
|
|
absl::Status Start() override { return impl_->Start(); } |
|
|
|
|
|
|
|
|
|
void ShutdownListeningFds() override { |
|
|
|
|
if (!shutdown_.exchange(true, std::memory_order_acq_rel)) { |
|
|
|
|
impl_->TriggerShutdown(); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
std::shared_ptr<PosixEngineListenerImpl> impl_; |
|
|
|
|
// Set to true when the listener had been explicitly shutdown.
|
|
|
|
|
std::atomic<bool> shutdown_{false}; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
#else // GRPC_POSIX_SOCKET_TCP
|
|
|
|
|
@ -219,6 +252,25 @@ class PosixEngineListener |
|
|
|
|
grpc_core::Crash( |
|
|
|
|
"EventEngine::Listener::Start not supported on this platform"); |
|
|
|
|
} |
|
|
|
|
absl::StatusOr<int> BindWithFd( |
|
|
|
|
const EventEngine::ResolvedAddress& /*addr*/, |
|
|
|
|
PosixListenerWithFdSupport::OnPosixBindNewFdCallback |
|
|
|
|
/*on_bind_new_fd*/) override { |
|
|
|
|
grpc_core::Crash( |
|
|
|
|
"PosixEngineListener::BindWithFd not supported on this " |
|
|
|
|
"platform"); |
|
|
|
|
} |
|
|
|
|
absl::Status HandleExternalConnection( |
|
|
|
|
int /*listener_fd*/, int /*fd*/, SliceBuffer* /*pending_data*/) override { |
|
|
|
|
grpc_core::Crash( |
|
|
|
|
"PosixEngineListener::HandleExternalConnection not " |
|
|
|
|
"supported on this platform"); |
|
|
|
|
} |
|
|
|
|
void ShutdownListeningFds() override { |
|
|
|
|
grpc_core::Crash( |
|
|
|
|
"PosixEngineListener::ShutdownListeningFds not supported on " |
|
|
|
|
"this platform"); |
|
|
|
|
} |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
#endif |
|
|
|
|
|
Vignesh Babu
2 years ago
committed by
GitHub