database: check timout before applying writes

Attempting to apply timed-out writes is a wasted effort. The coordinator
have already given up on the write and reported it as failed to the
client. Any cycles spent on this write is a waste at this point.
We currently only check the timeout if the write is blocked on memory,
otherwise, if the system is not under pressure, we will happily apply
timed out writes. If the system is under pressure we will make it worse
by wasting cycles on processing a timed out write.

Prevent this by checking the timeout as early as possible in
`database::apply()` and `database::apply_counter_update()`.

This patch doesn't solve all our problems related to timed out writes.
They can still sit and accumulate in various queues without expiring, a
prominent example being the smp queues. It is however a good first step
towards reducing wasted effort spent on them.

Refs: #5055
Ref #5251

Signed-off-by: Botond Dénes <bdenes@scylladb.com>
Message-Id: <20200129093007.550250-1-bdenes@scylladb.com>

database.cc

@@ -1431,6 +1431,10 @@ future<> database::apply_in_memory(const mutation& m, column_family& cf, db::rp_
 }
 
 future<mutation> database::apply_counter_update(schema_ptr s, const frozen_mutation& m, db::timeout_clock::time_point timeout, tracing::trace_state_ptr trace_state) {
+    if (timeout <= db::timeout_clock::now()) {
+        update_write_metrics_for_timed_out_write();
+        return make_exception_future<mutation>(timed_out_error{});
+    }
   return update_write_metrics(seastar::futurize_apply([&] {
     if (!s->is_synced()) {
         throw std::runtime_error(format("attempted to mutate using not synced schema of {}.{}, version={}",
@@ -1527,10 +1531,20 @@ Future database::update_write_metrics(Future&& f) {
     });
 }
 
+void database::update_write_metrics_for_timed_out_write() {
+    ++_stats->total_writes;
+    ++_stats->total_writes_failed;
+    ++_stats->total_writes_timedout;
+}
+
 future<> database::apply(schema_ptr s, const frozen_mutation& m, db::commitlog::force_sync sync, db::timeout_clock::time_point timeout) {
     if (dblog.is_enabled(logging::log_level::trace)) {
         dblog.trace("apply {}", m.pretty_printer(s));
     }
+    if (timeout <= db::timeout_clock::now()) {
+        update_write_metrics_for_timed_out_write();
+        return make_exception_future<>(timed_out_error{});
+    }
     return update_write_metrics(_apply_stage(this, std::move(s), seastar::cref(m), timeout, sync));
 }
 

database.hh

@@ -1361,6 +1361,7 @@ private:
 
     template<typename Future>
     Future update_write_metrics(Future&& f);
+    void update_write_metrics_for_timed_out_write();
 public:
     static utils::UUID empty_version;