|
18 | 18 | #include <cstdlib> |
19 | 19 | #include <functional> |
20 | 20 | #include <kj/function.h> |
| 21 | +#include <kj/io.h> |
21 | 22 | #include <map> |
22 | 23 | #include <memory> |
23 | 24 | #include <optional> |
@@ -213,6 +214,9 @@ class Logger |
213 | 214 |
|
214 | 215 | std::string LongThreadName(const char* exe_name); |
215 | 216 |
|
| 217 | +//! Wrap a socket file descriptor as an async stream, taking ownership of the fd. |
| 218 | +Stream MakeStream(EventLoop&loop, SocketId socket); |
| 219 | + |
216 | 220 | //! Event loop implementation. |
217 | 221 | //! |
218 | 222 | //! Cap'n Proto threading model is very simple: all I/O operations are |
@@ -311,11 +315,12 @@ class EventLoop |
311 | 315 | //! Callback functions to run on async thread. |
312 | 316 | std::optional<CleanupList> m_async_fns MP_GUARDED_BY(m_mutex); |
313 | 317 |
|
314 | | - //! Pipe read handle used to wake up the event loop thread. |
315 | | - int m_wait_fd = -1; |
| 318 | + //! Socket pair used to post and wait for wakeups to the event loop thread. |
| 319 | + kj::Own<kj::AsyncIoStream> m_wait_stream; |
| 320 | + kj::Own<kj::AsyncIoStream> m_post_stream; |
316 | 321 |
|
317 | | - //! Pipe write handle used to wake up the event loop thread. |
318 | | - int m_post_fd = -1; |
| 322 | + //! Synchronous writer used to write to m_post_stream. |
| 323 | + kj::Own<kj::OutputStream> m_post_writer; |
319 | 324 |
|
320 | 325 | //! Number of EventLoopRef instances referencing this event loop. This is a |
321 | 326 | //! sum of the number of client and server objects (Connection, ProxyClient, |
@@ -451,7 +456,7 @@ class Connection |
451 | 456 | //! Capability::Client handles owned by ProxyClient objects), then schedules |
452 | 457 | //! asynchronous cleanup functions to run in a worker thread (to run |
453 | 458 | //! destructors of m_impl instances owned by ProxyServer objects). |
454 | | - ~Connection(); |
| 459 | + ~Connection() noexcept(false); |
455 | 460 |
|
456 | 461 | //! Register synchronous cleanup function to run on event loop thread (with |
457 | 462 | //! access to capnp thread local variables) when disconnect() is called. |
@@ -824,17 +829,15 @@ kj::Promise<T> ProxyServer<Thread>::post(Fn&& fn) |
824 | 829 | return ret; |
825 | 830 | } |
826 | 831 |
|
827 | | -//! Given stream file descriptor, make a new ProxyClient object to send requests |
828 | | -//! over the stream. Also create a new Connection object embedded in the |
829 | | -//! client that is freed when the client is closed. |
| 832 | +//! Given a stream, make a new ProxyClient object to send requests over it. |
| 833 | +//! Also create a new Connection object embedded in the client that is freed |
| 834 | +//! when the client is closed. |
830 | 835 | template <typename InitInterface> |
831 | | -std::unique_ptr<ProxyClient<InitInterface>> ConnectStream(EventLoop& loop, int fd) |
| 836 | +std::unique_ptr<ProxyClient<InitInterface>> ConnectStream(EventLoop& loop, Stream stream) |
832 | 837 | { |
833 | 838 | typename InitInterface::Client init_client(nullptr); |
834 | 839 | std::unique_ptr<Connection> connection; |
835 | 840 | loop.sync([&] { |
836 | | - auto stream = |
837 | | - loop.m_io_context.lowLevelProvider->wrapSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP); |
838 | 841 | connection = std::make_unique<Connection>(loop, kj::mv(stream)); |
839 | 842 | init_client = connection->m_rpc_system->bootstrap(ServerVatId().vat_id).castAs<InitInterface>(); |
840 | 843 | Connection* connection_ptr = connection.get(); |
@@ -905,22 +908,18 @@ void _Listen(const std::shared_ptr<Listener>& listener, EventLoop& loop, InitImp |
905 | 908 | })); |
906 | 909 | } |
907 | 910 |
|
908 | | -//! Given stream file descriptor and an init object, handle requests on the |
909 | | -//! stream by calling methods on the Init object. |
| 911 | +//! Given a stream and an init object, handle requests on the stream by calling |
| 912 | +//! methods on the Init object. |
910 | 913 | template <typename InitInterface, typename InitImpl> |
911 | | -void ServeStream(EventLoop& loop, int fd, InitImpl& init) |
| 914 | +void ServeStream(EventLoop& loop, Stream stream, InitImpl& init) |
912 | 915 | { |
913 | | - _Serve<InitInterface>( |
914 | | - loop, |
915 | | - loop.m_io_context.lowLevelProvider->wrapSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP), |
916 | | - init, |
917 | | - [] {}); |
| 916 | + _Serve<InitInterface>(loop, kj::mv(stream), init, [] {}); |
918 | 917 | } |
919 | 918 |
|
920 | | -//! Given listening socket file descriptor and an init object, handle incoming |
| 919 | +//! Given listening socket identifier and an init object, handle incoming |
921 | 920 | //! connections and requests by calling methods on the Init object. |
922 | 921 | template <typename InitInterface, typename InitImpl> |
923 | | -void ListenConnections(EventLoop& loop, int fd, InitImpl& init, std::optional<size_t> max_connections = std::nullopt) |
| 922 | +void ListenConnections(EventLoop& loop, SocketId fd, InitImpl& init, std::optional<size_t> max_connections = std::nullopt) |
924 | 923 | { |
925 | 924 | loop.sync([&]() { |
926 | 925 | auto listener{std::make_shared<Listener>( |
|
0 commit comments