Rdb loader tasks continue running on failure

[dranikpg]

Currently, the rdb loader collects items to be inserted and schedules them for insertion inside the shard with FlushShardAsync. Because those tasks are added to the shard_set and not executed immediately, it might happen, that RdbLoader::Load exits early with an error and those tasks will continue running. They can access the loaders fields and will crash if the loader was already deleted after reporting an error. We should make sure on every return that those tasks finished executing, like it is done on the very bottom of the function.

I would also suggest replacing the stop_early_ flag with Cancellation from commons 🙂

[romange]

I think a simpler thing would be to wrap the final part of the function, i.e.

  fibers_ext::BlockingCounter bc(shard_set->size());
  for (unsigned i = 0; i < shard_set->size(); ++i) {
    // Flush the remaining items.
    FlushShardAsync(i);

    // Send sentinel callbacks to ensure that all previous messages have been processed.
    shard_set->Add(i, [bc]() mutable { bc.Dec(); });
  }
  bc.Wait();  // wait for sentinels to report.

  absl::Duration dur = absl::Now() - start;
  load_time_ = double(absl::ToInt64Milliseconds(dur)) / 1000;
  keys_loaded_ = keys_loaded;

with absl::Cleanup