Refactor and update the IO/TCP to make it's behaviour more consistent…

… and reliable (windows to come)
Fastcode · Aug 9, 2023 · 44b35ca · 44b35ca
1 parent 59fb065
commit 44b35ca
Show file tree

Hide file tree

Showing 7 changed files with 286 additions and 223 deletions.
diff --git a/src/dsl/word/IO.hpp b/src/dsl/word/IO.hpp
@@ -32,11 +32,18 @@ namespace dsl {
     namespace word {
 
         struct IOConfiguration {
+            IOConfiguration(fd_t fd, int events, const std::shared_ptr<threading::Reaction>& reaction)
+                : fd(fd), events(events), reaction(reaction) {}
             fd_t fd;
             int events;
             std::shared_ptr<threading::Reaction> reaction;
         };
 
+        struct IOFinished {
+            IOFinished(const uint64_t& id) : id(id) {}
+            uint64_t id;
+        };
+
         /**
          * @brief
          *  This is used to trigger reactions based on standard I/O operations using file descriptors.
@@ -68,15 +75,25 @@ namespace dsl {
          * @par Implements
          *  Bind
          */
-        struct IO : public Single {
+        struct IO {
 
 // On windows we use different wait events
 #ifdef _WIN32
             // NOLINTNEXTLINE(google-runtime-int)
-            enum EventType : short{READ = FD_READ | FD_OOB | FD_ACCEPT, WRITE = FD_WRITE, CLOSE = FD_CLOSE, ERROR = 0};
+            enum EventType : short {
+                READ  = FD_READ | FD_OOB | FD_ACCEPT,
+                WRITE = FD_WRITE,
+                CLOSE = FD_CLOSE,
+                ERROR = 0,
+            };
 #else
             // NOLINTNEXTLINE(google-runtime-int)
-            enum EventType : short { READ = POLLIN, WRITE = POLLOUT, CLOSE = POLLHUP, ERROR = POLLNVAL | POLLERR };
+            enum EventType : short {
+                READ  = POLLIN,
+                WRITE = POLLOUT,
+                CLOSE = POLLHUP,
+                ERROR = POLLNVAL | POLLERR,
+            };
 #endif
 
             struct Event {
@@ -114,6 +131,11 @@ namespace dsl {
                 // Otherwise return an invalid event
                 return Event{INVALID_SOCKET, 0};
             }
+
+            template <typename DSL>
+            static inline void postcondition(threading::ReactionTask& task) {
+                task.parent.reactor.emit<emit::Direct>(std::make_unique<IOFinished>(task.parent.id));
+            }
         };
 
     }  // namespace word

diff --git a/src/dsl/word/TCP.hpp b/src/dsl/word/TCP.hpp
@@ -58,7 +58,7 @@ namespace dsl {
          * @par Implements
          *  Bind
          */
-        struct TCP {
+        struct TCP : public IO {
 
             struct Connection {
 
@@ -124,10 +124,8 @@ namespace dsl {
                 port = ntohs(address.sin_port);
 
                 // Generate a reaction for the IO system that closes on death
-                const fd_t cfd = fd;
-                reaction->unbinders.push_back([](const threading::Reaction& r) {
-                    r.reactor.emit<emit::Direct>(std::make_unique<operation::Unbind<IO>>(r.id));
-                });
+                const fd_t cfd = fd.release();
+
                 reaction->unbinders.push_back([cfd](const threading::Reaction&) {
                     ::shutdown(cfd, SHUT_RDWR);
 #ifdef _WIN32
@@ -137,10 +135,8 @@ namespace dsl {
 #endif
                 });
 
-                auto io_config = std::make_unique<IOConfiguration>(IOConfiguration{fd.release(), IO::READ, reaction});
-
-                // Send our configuration out
-                reaction->reactor.emit<emit::Direct>(io_config);
+                // Bind using the IO system
+                IO::bind<DSL>(reaction, cfd, IO::READ | IO::CLOSE);
 
                 // Return our handles
                 return std::make_tuple(port, cfd);

diff --git a/src/dsl/word/UDP.hpp b/src/dsl/word/UDP.hpp
@@ -58,7 +58,7 @@ namespace dsl {
          * @par Implements
          *  Bind
          */
-        struct UDP {
+        struct UDP : public IO {
 
             struct Packet {
                 Packet() = default;
@@ -148,16 +148,9 @@ namespace dsl {
                 port = ntohs(address.sin_port);
 
                 // Generate a reaction for the IO system that closes on death
-                const fd_t cfd = fd;
-                reaction->unbinders.push_back([](const threading::Reaction& r) {
-                    r.reactor.emit<emit::Direct>(std::make_unique<operation::Unbind<IO>>(r.id));
-                });
-                reaction->unbinders.push_back([cfd](const threading::Reaction&) { close(cfd); });
-
-                auto io_config = std::make_unique<IOConfiguration>(IOConfiguration{fd.release(), IO::READ, reaction});
-
-                // Send our configuration out
-                reaction->reactor.emit<emit::Direct>(io_config);
+                const fd_t cfd = fd.release();
+                reaction->unbinders.push_back([cfd](const threading::Reaction&) { ::close(cfd); });
+                IO::bind<DSL>(reaction, cfd, IO::READ | IO::CLOSE);
 
                 // Return our handles and our bound port
                 return std::make_tuple(port, cfd);
@@ -248,7 +241,7 @@ namespace dsl {
                 return p;
             }
 
-            struct Broadcast {
+            struct Broadcast : public IO {
 
                 template <typename DSL>
                 static inline std::tuple<in_port_t, fd_t> bind(const std::shared_ptr<threading::Reaction>& reaction,
@@ -306,17 +299,9 @@ namespace dsl {
                     port = ntohs(address.sin_port);
 
                     // Generate a reaction for the IO system that closes on death
-                    const fd_t cfd = fd;
-                    reaction->unbinders.push_back([](const threading::Reaction& r) {
-                        r.reactor.emit<emit::Direct>(std::make_unique<operation::Unbind<IO>>(r.id));
-                    });
-                    reaction->unbinders.push_back([cfd](const threading::Reaction&) { close(cfd); });
-
-                    auto io_config =
-                        std::make_unique<IOConfiguration>(IOConfiguration{fd.release(), IO::READ, reaction});
-
-                    // Send our configuration out
-                    reaction->reactor.emit<emit::Direct>(io_config);
+                    const fd_t cfd = fd.release();
+                    reaction->unbinders.push_back([cfd](const threading::Reaction&) { ::close(cfd); });
+                    IO::bind<DSL>(reaction, cfd, IO::READ | IO::CLOSE);
 
                     // Return our handles and our bound port
                     return std::make_tuple(port, cfd);
@@ -328,7 +313,7 @@ namespace dsl {
                 }
             };
 
-            struct Multicast {
+            struct Multicast : public IO {
 
                 template <typename DSL>
                 static inline std::tuple<in_port_t, fd_t> bind(const std::shared_ptr<threading::Reaction>& reaction,
@@ -412,17 +397,9 @@ namespace dsl {
                     }
 
                     // Generate a reaction for the IO system that closes on death
-                    const fd_t cfd = fd;
-                    reaction->unbinders.push_back([](const threading::Reaction& r) {
-                        r.reactor.emit<emit::Direct>(std::make_unique<operation::Unbind<IO>>(r.id));
-                    });
+                    const fd_t cfd = fd.release();
                     reaction->unbinders.push_back([cfd](const threading::Reaction&) { close(cfd); });
-
-                    auto io_config =
-                        std::make_unique<IOConfiguration>(IOConfiguration{fd.release(), IO::READ, reaction});
-
-                    // Send our configuration out for each file descriptor (same reaction)
-                    reaction->reactor.emit<emit::Direct>(io_config);
+                    IO::bind<DSL>(reaction, cfd, IO::READ | IO::CLOSE);
 
                     // Return our handles
                     return std::make_tuple(port, cfd);