control: Limit memory use by control connections

Signed-off-by: Mobin "Hojjat" Aydinfar <mobin@mobintestserver.ir>

src/control.cc

@@ -1135,7 +1135,7 @@ void control_conn_t::service_event(service_record *service, service_event_t even
 
 bool control_conn_t::queue_packet(const char *pkt, unsigned size) noexcept
 {
-    int in_flag = bad_conn_close ? 0 : IN_EVENTS;
+    int in_flag = bad_conn_close || input_memlimit() ? 0 : IN_EVENTS;
     bool was_empty = outbuf.empty();
 
     // If the queue is empty, we can try to write the packet out now rather than queueing it.
@@ -1190,7 +1190,7 @@ bool control_conn_t::queue_packet(const char *pkt, unsigned size) noexcept
 // make them extraordinary difficult to combine into a single method.
 bool control_conn_t::queue_packet(std::vector<char> &&pkt) noexcept
 {
-    int in_flag = bad_conn_close ? 0 : IN_EVENTS;
+    int in_flag = bad_conn_close || input_memlimit() ? 0 : IN_EVENTS;
     bool was_empty = outbuf.empty();
 
     if (was_empty) {
@@ -1275,8 +1275,9 @@ bool control_conn_t::data_ready() noexcept
         iob.set_watches(OUT_EVENTS);
     }
     else {
+        int in_flags = !input_memlimit() ? IN_EVENTS : 0;
         int out_flags = (bad_conn_close || !outbuf.empty()) ? OUT_EVENTS : 0;
-        iob.set_watches(IN_EVENTS | out_flags);
+        iob.set_watches(in_flags | out_flags);
     }
 
     return false;
@@ -1324,14 +1325,31 @@ bool control_conn_t::send_data() noexcept
             if (bad_conn_close) {
                 return true;
             }
-            iob.set_watches(IN_EVENTS);
+            int in_flag = !input_memlimit() ? IN_EVENTS : 0;
+            iob.set_watches(in_flag);
         }
     }
 
     // more to send
     return false;
 }
 
+bool control_conn_t::input_memlimit() noexcept
+{
+    if (outbuf.size() >= 1000) {
+        // Outbuf contains much data, Temporary disable "IN_EVENTS" to reduce it
+        if (!in_memlimit) {
+            in_memlimit = true;
+        }
+        return true;
+    }
+    else if (in_memlimit) {
+        // if "IN_EVENTS" disabled in previous packet and outbuf size has reduced, enable it again
+        in_memlimit = false;
+    }
+    return false;
+}
+
 control_conn_t::~control_conn_t() noexcept
 {
     int fd = iob.get_watched_fd();

src/includes/control.h

@@ -113,6 +113,9 @@ class control_conn_t : private service_listener
     list<vector<char>> outbuf;
     // Current index within the first outgoing packet (all previous bytes have been sent).
     unsigned outpkt_index = 0;
+    // If outbuf is almost full, We will disable "IN_EVENTS" and set this variable as "true"
+    // and vice-versa
+    bool in_memlimit = false;
 
     // Queue a packet to be sent
     //  Returns:  false if the packet could not be queued and a suitable error packet
@@ -181,6 +184,9 @@ class control_conn_t : private service_listener
     bool data_ready() noexcept;
 
     bool send_data() noexcept;
+
+    // Check the output buffer size and limit inputs if necessary
+    bool input_memlimit() noexcept;
 
     // Check if any dependents will be affected by stopping a service, generate a response packet if so.
     // had_dependents will be set true if the service should not be stopped, false otherwise.