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
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
|
// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#pragma once
#include "feed_handler_stats.h"
#include "i_inc_serial_num.h"
#include "i_operation_storer.h"
#include "idocumentmovehandler.h"
#include "igetserialnum.h"
#include "iheartbeathandler.h"
#include "ipruneremoveddocumentshandler.h"
#include "tlswriter.h"
#include "transactionlogmanager.h"
#include <vespa/document/bucket/bucketid.h>
#include <vespa/searchcore/proton/common/doctypename.h>
#include <vespa/searchcore/proton/common/feedtoken.h>
#include <vespa/searchlib/transactionlog/client_common.h>
#include <shared_mutex>
namespace searchcorespi::index { struct IThreadingService; }
namespace document { class DocumentTypeRepo; }
namespace proton {
struct ConfigStore;
class CreateBucketOperation;
class DDBState;
class DeleteBucketOperation;
struct FeedConfigStore;
class FeedState;
class IDocumentDBOwner;
struct IFeedHandlerOwner;
class IFeedView;
struct IResourceWriteFilter;
class IReplayConfig;
class JoinBucketsOperation;
class PutOperation;
class RemoveOperation;
class ReplayThrottlingPolicy;
class SplitBucketOperation;
class UpdateOperation;
namespace bucketdb { class IBucketDBHandler; }
/**
* Class handling all aspects of feeding for a document database.
* In addition to regular feeding this also includes handling the transaction log.
*/
class FeedHandler: private search::transactionlog::client::Callback,
public IDocumentMoveHandler,
public IPruneRemovedDocumentsHandler,
public IHeartBeatHandler,
public IOperationStorer,
public IGetSerialNum,
public IIncSerialNum
{
private:
using Packet = search::transactionlog::Packet;
using RPC = search::transactionlog::client::RPC;
using SerialNum = search::SerialNum;
using BucketId = document::BucketId;
using FeedStateSP = std::shared_ptr<FeedState>;
using FeedOperationUP = std::unique_ptr<FeedOperation>;
using ReadGuard = std::shared_lock<std::shared_mutex>;
using WriteGuard = std::unique_lock<std::shared_mutex>;
using IThreadingService = searchcorespi::index::IThreadingService;
using TlsWriterFactory = search::transactionlog::WriterFactory;
IThreadingService &_writeService;
DocTypeName _docTypeName;
IFeedHandlerOwner &_owner;
const IResourceWriteFilter &_writeFilter;
IReplayConfig &_replayConfig;
TransactionLogManager _tlsMgr;
const TlsWriterFactory &_tlsWriterfactory;
std::unique_ptr<TlsWriter> _tlsMgrWriter;
TlsWriter *_tlsWriter;
TlsReplayProgress::UP _tlsReplayProgress;
// the serial num of the last feed operation processed by feed handler.
std::atomic<SerialNum> _serialNum;
// the serial num considered to be fully procssessed and flushed to stable storage. Used to prune transaction log.
SerialNum _prunedSerialNum;
// the serial num of the last feed operation in the transaction log at startup before replay
SerialNum _replay_end_serial_num;
uint64_t _prepare_serial_num;
FeedOperationCounter _numOperations;
bool _delayedPrune;
mutable std::shared_mutex _feedLock;
FeedStateSP _feedState;
// used by master write thread tasks
IFeedView *_activeFeedView;
const document::DocumentTypeRepo *_repo;
const document::DocumentType *_documentType;
bucketdb::IBucketDBHandler *_bucketDBHandler;
std::mutex _syncLock;
SerialNum _syncedSerialNum;
bool _allowSync; // Sanity check
std::atomic<vespalib::steady_time> _heart_beat_time;
mutable std::mutex _stats_lock;
mutable FeedHandlerStats _stats;
/**
* Delayed handling of feed operations, in master write thread.
* The current feed state is sampled here.
*/
void doHandleOperation(FeedToken token, FeedOperationUP op);
bool considerWriteOperationForRejection(FeedToken & token, const FeedOperation &op);
bool considerUpdateOperationForRejection(FeedToken &token, UpdateOperation &op);
/**
* Delayed execution of feed operations against feed view, in
* master write thread.
*/
void performPut(FeedToken token, PutOperation &op);
void performUpdate(FeedToken token, UpdateOperation &op);
void performInternalUpdate(FeedToken token, UpdateOperation &op);
void createNonExistingDocument(FeedToken, const UpdateOperation &op);
void performRemove(FeedToken token, RemoveOperation &op);
void performGarbageCollect(FeedToken token);
void performCreateBucket(FeedToken token, CreateBucketOperation &op);
void performDeleteBucket(FeedToken token, DeleteBucketOperation &op);
void performSplit(FeedToken token, SplitBucketOperation &op);
void performJoin(FeedToken token, JoinBucketsOperation &op);
void performEof();
/**
* Used when flushing is done
*/
void performFlushDone(SerialNum flushedSerial);
void performPrune(SerialNum flushedSerial);
FeedStateSP getFeedState() const;
void changeFeedState(FeedStateSP newState);
void doChangeFeedState(FeedStateSP newState);
void onCommitDone(size_t numPendingAtStart, vespalib::steady_time start_time);
void initiateCommit(vespalib::steady_time start_time);
void enqueCommitTask();
public:
FeedHandler(const FeedHandler &) = delete;
FeedHandler & operator = (const FeedHandler &) = delete;
/**
* Create a new feed handler.
*
* @param writeService The thread service used for all write tasks.
* @param tlsSpec The spec to connect to the transaction log server.
* @param docTypeName The name and version of the document type we are feed handler for.
* @param owner Reference to the owner of this feed handler.
* @param replayConfig Reference to interface used for replaying config changes.
* @param writer Inject writer for tls, or nullptr to use internal.
*/
FeedHandler(IThreadingService &writeService,
const vespalib::string &tlsSpec,
const DocTypeName &docTypeName,
IFeedHandlerOwner &owner,
const IResourceWriteFilter &writerFilter,
IReplayConfig &replayConfig,
const TlsWriterFactory & writer,
TlsWriter * tlsWriter = nullptr);
~FeedHandler() override;
/**
* Init this feed handler.
*
* @param oldestConfigSerial The serial number of the oldest config snapshot.
*/
void init(SerialNum oldestConfigSerial);
/**
* Close this feed handler and its components.
*/
void close();
/**
* Start replay of the transaction log.
*
* @param flushedIndexMgrSerial The flushed serial number of the
* index manager.
* @param flushedSummaryMgrSerial The flushed serial number of the
* document store.
* @param config_store Reference to the config store.
*/
void
replayTransactionLog(SerialNum flushedIndexMgrSerial,
SerialNum flushedSummaryMgrSerial,
SerialNum oldestFlushedSerial,
SerialNum newestFlushedSerial,
ConfigStore &config_store,
const ReplayThrottlingPolicy& replay_throttling_policy);
/**
* Called when a flush is done and allows pruning of the transaction log.
*
* @param flushedSerial serial number flushed for all relevant flush targets.
*/
void flushDone(SerialNum flushedSerial);
/**
* Used to flip between normal and recovery feed states.
*/
void changeToNormalFeedState();
/**
* Update the active feed view.
* Always called by the master write thread so locking is not needed.
*/
void setActiveFeedView(IFeedView *feedView);
void setBucketDBHandler(bucketdb::IBucketDBHandler *bucketDBHandler) {
_bucketDBHandler = bucketDBHandler;
}
// Must only be called from writer thread:
void setSerialNum(SerialNum serialNum) { _serialNum.store(serialNum, std::memory_order_relaxed); }
SerialNum inc_serial_num() override {
const auto post_inc = _serialNum.load(std::memory_order_relaxed) + 1u;
_serialNum.store(post_inc, std::memory_order_relaxed);
return post_inc;
}
// May be called from non-writer threads:
SerialNum getSerialNum() const override { return _serialNum.load(std::memory_order_relaxed); }
// The two following methods are used when saving initial config
SerialNum get_replay_end_serial_num() const { return _replay_end_serial_num; }
SerialNum inc_replay_end_serial_num() { return ++_replay_end_serial_num; }
SerialNum getPrunedSerialNum() const { return _prunedSerialNum; }
uint64_t inc_prepare_serial_num() { return ++_prepare_serial_num; }
bool isDoingReplay() const;
float getReplayProgress() const {
return _tlsReplayProgress ? _tlsReplayProgress->getProgress() : 0;
}
bool getTransactionLogReplayDone() const;
vespalib::string getDocTypeName() const { return _docTypeName.getName(); }
void tlsPrune(SerialNum oldest_to_keep);
void performOperation(FeedToken token, FeedOperationUP op);
void handleOperation(FeedToken token, FeedOperationUP op);
void handleMove(MoveOperation &op, std::shared_ptr<vespalib::IDestructorCallback> moveDoneCtx) override;
void heartBeat() override;
RPC::Result receive(const Packet &packet) override;
void eof() override;
void performPruneRemovedDocuments(PruneRemovedDocumentsOperation &pruneOp) override;
void syncTls(SerialNum syncTo);
void appendOperation(const FeedOperation &op, DoneCallback onDone) override;
[[nodiscard]] CommitResult startCommit(DoneCallback onDone) override;
[[nodiscard]] CommitResult storeOperationSync(const FeedOperation & op);
void considerDelayedPrune();
vespalib::steady_time get_heart_beat_time() const;
FeedHandlerStats get_stats(bool reset_min_max) const;
};
} // namespace proton
|
