Fix bugs in streaming and enable support for gap filling (#9214)

This PR adds (inactive) support that we will use to fill the gaps on chart when a receiving agent goes offline and the sender reconnects. The streaming component has been reworked to make the connection bi-directional and fix several outstanding bugs in the area. 

* Fixed an incorrect case of version negotiation. Removed fatal() on exhaustion of fds. 
* Fixed cases that fell through to polling the socket after closing. 
* Fixed locking of data related to sender and receiver in the host structure. 
* Added fine-grained locks to reduce contention.
* Added circular buffer to sender to prevent starvation in high-latency conditions. 
* Fixed case where agent is a proxy and negotiated different streaming versions with sender and receiver. 
* Changed interface to new parser to put the buffering code in streaming. 
* Fixed the bug that stopped senders from reconnecting after their socket times out - this was part of the scaling fixes that provide an early shortcut path for rejecting connections without lock contention. 
* Uses fine-grained locking and a different approach to thread shutdown instead. 
* Added liveness detection to connections to allow selection of the best connection.

1 files changed, 1 insertions, 33 deletions

diff --git a/parser/parser.c b/parser/parser.c
index 84f36a84ea..21d7fb3fc1 100644
--- a/parser/parser.c
+++ b/parser/parser.c
@@ -209,42 +209,10 @@ int parser_next(PARSER *parser)
         return 0;
     }
 
-#ifdef ENABLE_HTTPS
-    int normalread = 1;
-    if (netdata_srv_ctx) {
-        if (parser->host->stream_ssl.conn && !parser->host->stream_ssl.flags) {
-            tmp = parser->buffer;
-            if (!parser->bytesleft) {
-                parser->readfrom = parser->tmpbuffer;
-                parser->bytesleft =
-                    pluginsd_update_buffer(parser->readfrom, parser->host->stream_ssl.conn);
-                if (parser->bytesleft <= 0) {
-                    return 1;
-                }
-            }
-
-            parser->readfrom = pluginsd_get_from_buffer(
-                parser->buffer, &parser->bytesleft, parser->readfrom,
-                parser->host->stream_ssl.conn, parser->tmpbuffer);
-
-            if (!parser->readfrom)
-                tmp = NULL;
-
-            normalread = 0;
-        }
-    }
-    if (normalread) {
-        if (unlikely(parser->read_function))
-            tmp = parser->read_function(parser->buffer, PLUGINSD_LINE_MAX, parser->input);
-        else
-            tmp = fgets(parser->buffer, PLUGINSD_LINE_MAX, (FILE *)parser->input);
-    }
-#else
     if (unlikely(parser->read_function))
         tmp = parser->read_function(parser->buffer, PLUGINSD_LINE_MAX, parser->input);
     else
         tmp = fgets(parser->buffer, PLUGINSD_LINE_MAX, (FILE *)parser->input);
-#endif
 
     if (unlikely(!tmp)) {
         if (unlikely(parser->eof_function)) {
@@ -320,7 +288,7 @@ inline int parser_action(PARSER *parser, char *input)
         else
             rc = PARSER_RC_ERROR;
 #ifdef NETDATA_INTERNAL_CHECKS
-        error("Unknown keyword [%s]", words[0]);
+        error("Unknown keyword [%s]", input);
 #endif
     }
     else {