Add very first approach of automatic concurrency control

This is a very naive approach that works relatively well when all resolvers
are 100 % reliable (e.g. subdomain enumeration against authoritative nameservers).
Starting at a concurrency of 1, it doubles the concurrency in steps of
--interval when no timeouts occurred.

CMakeLists.txt

@@ -4,6 +4,6 @@ project(massdns)
 set(CMAKE_C_STANDARD 11)
 
 set(SOURCE_FILES src/main.c src/list.h src/hashmap.h src/massdns.h src/security.h src/net.h src/string.h src/buffers.h src/dns.h
-        src/timed_ring.h src/random.h src/cmd.h src/flow.h)
+        src/timed_ring.h src/random.h src/cmd.h src/flow.h src/auto_concurrency.h)
 set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/bin)
 add_executable(massdns ${SOURCE_FILES})

src/auto_concurrency.h

@@ -0,0 +1,90 @@
+// SPDX-License-Identifier: GPL-3.0-only
+
+#ifndef MASSDNS_AUTO_CONCURRENCY_H
+#define MASSDNS_AUTO_CONCURRENCY_H
+
+#include "massdns.h"
+
+typedef struct
+{
+    size_t current_concurrency;
+    struct timespec last_update;
+    size_t iteration;
+    size_t last_numtimeouts;
+} auto_concurrency_state_t;
+
+auto_concurrency_state_t concurrency_state;
+
+void init_concurrency_controller()
+{
+    bzero(&concurrency_state, sizeof(concurrency_state));
+    concurrency_state.current_concurrency = context.cmd_args.auto_concurrency ? 1 : context.cmd_args.hashmap_size;
+    clock_gettime(CLOCK_REALTIME, &concurrency_state.last_update);
+}
+
+/* Return 1 if a > b, -1 if a < b and 0 if a == b */ 
+int timespec_compare(const struct timespec *a, const struct timespec *b)
+{
+    if (a->tv_sec == b->tv_sec && a->tv_nsec == b->tv_nsec)
+    {
+        return 0;
+    }
+    if (a->tv_sec != b->tv_sec)
+    {
+        return 2 * (a->tv_sec > b->tv_sec) - 1;
+    }
+    return 2 * (a->tv_nsec > b->tv_nsec) - 1;
+}
+
+void timespec_diff(const struct timespec *old, const struct timespec *now, struct timespec *result)
+{
+    result->tv_sec = now->tv_sec - old->tv_sec;
+    if(result->tv_sec > 0 && now->tv_nsec >= old->tv_nsec)
+    {
+        result->tv_sec--;
+    }
+    result->tv_nsec = now->tv_nsec >= old->tv_nsec ?
+            now->tv_nsec - old->tv_nsec
+            : (1000000000L - now->tv_nsec) + (1000000000L - old->tv_nsec);
+}
+
+bool elapsed_ms(const struct timespec *old, const struct timespec *now, size_t ms)
+{
+    struct timespec diff;
+    timespec_diff(old, now, &diff);
+
+    struct timespec ms_timespec = {.tv_sec = ms / 1000, .tv_nsec = (ms % 1000) * 1000000};
+    if(timespec_compare(&diff, &ms_timespec) >= 0)
+    {
+        return true;
+    }
+    return false;
+}
+
+void auto_concurrency_handle(const struct timespec *now)
+{
+    if (!context.cmd_args.auto_concurrency)
+    {
+        return;
+    }
+    struct timespec new_now;
+    if (now == NULL)
+    {
+        now = &new_now;
+        clock_gettime(CLOCK_REALTIME, &new_now);
+    }
+    if (!elapsed_ms(&concurrency_state.last_update, now, context.cmd_args.interval_ms))
+    {
+        return;
+    }
+    concurrency_state.last_update = *now;
+
+    if (context.stats.numtimeouts - concurrency_state.last_numtimeouts == 0)
+    {
+        concurrency_state.current_concurrency *= 2;
+    }
+
+    concurrency_state.last_numtimeouts = context.stats.numtimeouts;
+}
+
+#endif //MASSDNS_AUTO_CONCURRENCY_H

src/main.c

@@ -14,6 +14,7 @@
 #include "dns.h"
 #include "list.h"
 #include "flow.h"
+#include "auto_concurrency.h"
 #include <unistd.h>
 #include <pwd.h>
 #include <grp.h>
@@ -108,6 +109,7 @@ void print_help()
 
 /* The default real-time status output, human reeadable, very granular stats */
 static const char* stats_fmt_ansi = "\033[H\033[2J" // Clear screen (probably simplest and most portable solution)
+        "Concurrency: %zu\n"
         "Processed queries: %zu\n"
         "Received packets: %zu\n"
         "Progress: %.2f%% (%02lld h %02lld min %02lld sec / %02lld h %02lld min %02lld sec)\n"
@@ -126,6 +128,7 @@ static const char* stats_fmt_ansi = "\033[H\033[2J" // Clear screen (probably si
 /* Optional real-time status output, all stats on a single line as valid JSON */
 static const char* stats_fmt_json =
     "{"
+        "\"concurrency\":%zu,"
         "\"processed_queries\":%zu,"
         "\"received_packets\":%zu,"
         "\"progress\":"
@@ -915,6 +918,7 @@ void check_progress()
 
         fprintf(stderr,
                 context.status_fmt,
+                concurrency_state.current_concurrency,
                 context.stats.numdomains,
                 context.stats.numreplies,
                 progress * 100, h, min, sec, prog_h, prog_min, prog_sec, rate_pps, average_pps,
@@ -983,6 +987,7 @@ void check_progress()
 
         fprintf(stderr,
                 context.status_fmt,
+                concurrency_state.current_concurrency,  // TODO: This is only the concurrency of the main process.
                 context.stat_messages[0].numdomains,
                 context.stat_messages[0].numreplies,
                 progress * 100, h, min, sec, prog_h, prog_min, prog_sec, rate_pps, average_pps,
@@ -1033,7 +1038,8 @@ void can_send()
     dedicated_resolvers_t *dedicated_resolvers = NULL;
     dns_record_type rtype;
 
-    while (hashmapSize(context.map) < context.cmd_args.hashmap_size && context.state <= STATE_QUERYING)
+    while (hashmapSize(context.map) < min(context.cmd_args.hashmap_size, concurrency_state.current_concurrency)
+        && context.state <= STATE_QUERYING)
     {
         if(!next_query(&qname, &dedicated_resolvers, &rtype))
         {
@@ -1112,7 +1118,10 @@ void ring_timeout(void *param)
         return;
     }
 
+    auto_concurrency_handle(NULL);
+
     lookup_t *lookup = param;
+    context.stats.numtimeouts++;
     if(!retry(lookup))
     {
         write_exhausted_tries(lookup, "TIMEOUT");
@@ -1419,6 +1428,7 @@ void pcap_callback(u_char *arg, const struct pcap_pkthdr *header, const u_char *
         return;
     }
     do_read((uint8_t*)frame, len, &addr);
+    auto_concurrency_handle(NULL);
 }
 
 void pcap_can_read()
@@ -1442,8 +1452,8 @@ void can_read(socket_info_t *info)
     {
         return;
     }
-
     do_read(readbuf, (size_t)num_received, &recvaddr);
+    auto_concurrency_handle(NULL);
 }
 
 bool cmp_lookup(void *lookup1, void *lookup2)
@@ -1874,6 +1884,7 @@ void run()
         make_query_sockets_nonblocking();
     }
 
+    init_concurrency_controller();
 
     clock_gettime(CLOCK_MONOTONIC, &context.stats.start_time);
     check_progress();
@@ -2317,7 +2328,16 @@ void parse_cmd(int argc, char **argv)
         }
         else if (strcmp(argv[i], "--hashmap-size") == 0 || strcmp(argv[i], "-s") == 0)
         {
-            context.cmd_args.hashmap_size = (size_t) expect_arg_nonneg(i++, 1, SIZE_MAX);
+            if (strcmp(argv[i+1], "auto") == 0)
+            {
+                context.cmd_args.auto_concurrency = true;
+                context.cmd_args.hashmap_size = 100000;
+            }
+            else
+            {
+                context.cmd_args.hashmap_size = (size_t) expect_arg_nonneg(i, 1, SIZE_MAX);
+            }
+            i++;
         }
         else if (strcmp(argv[i], "--processes") == 0)
         {

src/massdns.h

@@ -187,6 +187,7 @@ typedef struct
         size_t socket_count;
         bool busypoll;
         bool extended_input;
+        bool auto_concurrency;
     } cmd_args;
 
     struct
@@ -222,6 +223,7 @@ typedef struct
         size_t numreplies;
         size_t numparsed;
         size_t numdomains;
+        size_t numtimeouts;
         struct timespec last_print;
         size_t current_rate;
         size_t success_rate;
@@ -233,6 +235,7 @@ typedef struct
         size_t mismatch_id;
         size_t mismatch_domain;
     } stats;
+
     stats_exchange_t *stat_messages;
 #ifdef PCAP_SUPPORT
     pcap_t *pcap;