blob: c64693150fc60caf72cff0be2322d06a99ae3d00 [file] [log] [blame]
// Copyright 2018 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include <map>
#include <utility>
#include <lib/callback/scoped_callback.h>
#include <lib/fit/defer.h>
#include <lib/fit/function.h>
#include <lib/fxl/memory/weak_ptr.h>
#include "peridot/bin/ledger/sync_helper/mutable.h"
namespace ledger {
// This class allows to register operations and synchronization callback.
// Operation are registered by wrapping the callback that they are expected to
// call when they are finished.
// A synchronization callback is an callback that takes no parameter and that
// will be called by this class when all operations registered before the
// synchronization callback have finished.
class SyncHelper {
SyncHelper(const SyncHelper&) = delete;
SyncHelper& operator=(const SyncHelper&) = delete;
// Sets the callback to be called every time the SyncHelper is empty.
// SyncHelper is empty when no operation is currently in progress.
void set_on_empty(fit::closure on_empty_callback) {
on_empty_callback_ = std::move(on_empty_callback);
// Returns whether there is currently no running operation.
bool empty() { return in_flight_operation_counts_per_sync_point_.empty(); }
// Registers a synchronization callback. |callback| will be called when all
// operation wrapped by |WrapOperation| before the call to
// |RegisterSynchronizationCallback| have finished.
void RegisterSynchronizationCallback(fit::function<void()> callback);
// Wraps |callback| and marks it as a live operation. No callback registered
// through |RegisterSynchronizationCallback| after this call will be called
// until the returned callback has been called at least once.
template <typename A>
auto WrapOperation(A callback) {
auto sync_point = current_sync_point_;
auto on_first_call = fit::defer(callback::MakeScoped(
weak_ptr_factory_.GetWeakPtr(), [this, sync_point] {
if (--in_flight_operation_counts_per_sync_point_[sync_point] == 0) {
// The lambda is not marked mutable, because the original callback might
// have a const operator, and this should not force the receiver to use
// only non-const operator.
// Because of this:
// - on_first_callcallback| must be wrap into a Mutable because calling it
// is not a const operation.
// - |callback| must be wrap into a Mutable because it might not have a
// const operator().
[callback = Mutable(std::move(callback)),
on_first_call = Mutable(std::move(on_first_call))](auto&&... params) {
// Calls all synchronization callbacks that are currently due.
void CallSynchronizationCallbacks();
// This class operates with a virtual timestamp.
// - Each time an operation is registered, it increases the number of
// operation at the current timestamp.
// - Each time a synchronization callback is registered, it is either
// immediately called if no operation is in progress, or it is associated
// with the current timestamp, and after this, the current timestamp is
// incremented.
// - Each time an operation terminates, it decrements the number of operations
// at the current timestamp. Then the algorithm looks at all timestamp in
// increasing order. Until it finds one for which there is still operation
// in progress, it calls the associated synchronization callback.
// The current timestamp.
int64_t current_sync_point_;
// The synchronization callbacks associated to their respective timestamp.
std::map<int64_t, fit::function<void()>> sync_callback_per_sync_points_;
// The number of operation in progress for each timestamp.
std::map<int64_t, int64_t> in_flight_operation_counts_per_sync_point_;
fit::closure on_empty_callback_;
// This must be the last member.
fxl::WeakPtrFactory<SyncHelper> weak_ptr_factory_;
} // namespace ledger