1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
|
// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#pragma once
#include <vespa/storage/distributor/persistencemessagetracker.h>
namespace document { class Document; }
namespace storage::api {
class UpdateCommand;
class CreateBucketReply;
}
namespace storage::distributor {
class DistributorBucketSpace;
class UpdateMetricSet;
class UpdateOperation : public Operation
{
public:
UpdateOperation(const DistributorNodeContext& node_ctx,
DistributorStripeOperationContext& op_ctx,
DistributorBucketSpace& bucketSpace,
const std::shared_ptr<api::UpdateCommand>& msg,
std::vector<BucketDatabase::Entry> entries,
UpdateMetricSet& metric);
~UpdateOperation() override;
void onStart(DistributorStripeMessageSender& sender) override;
const char* getName() const noexcept override { return "update"; };
std::string getStatus() const override { return ""; };
void onReceive(DistributorStripeMessageSender& sender, const std::shared_ptr<api::StorageReply> & msg) override;
void onClose(DistributorStripeMessageSender& sender) override;
std::pair<document::BucketId, uint16_t> getNewestTimestampLocation() const {
return _newestTimestampLocation;
}
private:
PersistenceMessageTracker _tracker;
std::shared_ptr<api::UpdateCommand> _msg;
std::vector<BucketDatabase::Entry> _entries;
const api::Timestamp _new_timestamp;
const bool _is_auto_create_update;
const DistributorNodeContext& _node_ctx;
DistributorStripeOperationContext& _op_ctx;
DistributorBucketSpace& _bucketSpace;
std::pair<document::BucketId, uint16_t> _newestTimestampLocation;
api::BucketInfo _infoAtSendTime; // Should be same across all replicas
bool anyStorageNodesAvailable() const;
class PreviousDocumentVersion {
public:
PreviousDocumentVersion(document::BucketId b, const api::BucketInfo& info, uint64_t o, uint16_t node) noexcept :
bucketId(b), bucketInfo(info), oldTs(o), nodeId(node) {}
document::BucketId bucketId;
api::BucketInfo bucketInfo;
uint64_t oldTs;
uint16_t nodeId;
};
std::vector<PreviousDocumentVersion> _results;
UpdateMetricSet& _metrics;
api::Timestamp adjusted_received_old_timestamp(api::Timestamp old_ts_from_node) const;
void log_inconsistency_warning(const api::UpdateReply& reply,
const PreviousDocumentVersion& highest_timestamped_version,
const PreviousDocumentVersion& low_timestamped_version);
};
}
|
