**Describe the situation**
```
$ ./clickhouse-21.3.3 client --query "SELECT * FROM system.numbers_mt FORMAT Native" | pv > /dev/null
^C,1GiB 0:00:45 [ 710MiB/s] [                                                                                                                                                                                               <=>                      ]

$ ./clickhouse-21.3.3 client --compress 0 --query "SELECT * FROM system.numbers_mt FORMAT Native" | pv > /dev/null
^C,2GiB 0:00:12 [1,48GiB/s] [                                                   <=>                                                                                                                                                                  ]
```

std::lock_guard metrics_lock(metrics_mutex);
                do_sleep = true;
                ++error_count;
                sleep_time = std::min(
                    std::chrono::milliseconds{Int64(default_sleep_time.count() * std::exp2(error_count))},
                    max_sleep_time);
                tryLogCurrentException(getLoggerName().data());
                last_exception = std::current_exception();
            }
        }
        else
        {
            LOG_DEBUG(log, "Skipping send data over distributed table.");
        }
        const auto now = std::chrono::system_clock::now();
        if (now - last_decrease_time > decrease_error_count_period)
        {
            std::lock_guard metrics_lock(metrics_mutex);
            error_count /= 2;
            last_decrease_time = now;
        }
        if (do_sleep)
            break;
    }
    
    getFiles();
    if (!quit && do_sleep)
        task_handle->scheduleAfter(sleep_time.count());
}
ConnectionPoolPtr StorageDistributedDirectoryMonitor::createPool(const std::string & name, const StorageDistributed & storage)