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
|
// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#pragma once
#include "i_operation_storer.h"
#include "idocumentmovehandler.h"
#include "igetserialnum.h"
#include "iheartbeathandler.h"
#include "ipruneremoveddocumentshandler.h"
#include "tlswriter.h"
#include "transactionlogmanager.h"
#include <persistence/spi/types.h>
#include <vespa/searchcore/proton/common/doctypename.h>
#include <vespa/searchcore/proton/common/feedtoken.h>
#include <vespa/searchlib/transactionlog/translogclient.h>
namespace searchcorespi { namespace index { class IThreadingService; } }
namespace proton {
class ConfigStore;
class CreateBucketOperation;
class DDBState;
class DeleteBucketOperation;
class FeedConfigStore;
class FeedState;
class IDocumentDBOwner;
class IFeedHandlerOwner;
class IFeedView;
class IResourceWriteFilter;
class IReplayConfig;
class JoinBucketsOperation;
class PutOperation;
class RemoveOperation;
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::TransLogClient::Session::Callback,
public IDocumentMoveHandler,
public IPruneRemovedDocumentsHandler,
public IHeartBeatHandler,
public IOperationStorer,
public IGetSerialNum
{
private:
typedef search::transactionlog::Packet Packet;
typedef search::transactionlog::RPC RPC;
typedef search::SerialNum SerialNum;
typedef storage::spi::Timestamp Timestamp;
typedef document::BucketId BucketId;
using FeedStateSP = std::shared_ptr<FeedState>;
using FeedOperationUP = std::unique_ptr<FeedOperation>;
class TlsMgrWriter : public TlsWriter {
TransactionLogManager &_tls_mgr;
search::transactionlog::Writer *_tlsDirectWriter;
public:
TlsMgrWriter(TransactionLogManager &tls_mgr,
search::transactionlog::Writer * tlsDirectWriter) :
_tls_mgr(tls_mgr),
_tlsDirectWriter(tlsDirectWriter)
{ }
void storeOperation(const FeedOperation &op) override;
bool erase(SerialNum oldest_to_keep) override;
SerialNum sync(SerialNum syncTo) override;
};
typedef searchcorespi::index::IThreadingService IThreadingService;
IThreadingService &_writeService;
DocTypeName _docTypeName;
DDBState &_state;
IFeedHandlerOwner &_owner;
const IResourceWriteFilter &_writeFilter;
IReplayConfig &_replayConfig;
TransactionLogManager _tlsMgr;
TlsMgrWriter _tlsMgrWriter;
TlsWriter &_tlsWriter;
TlsReplayProgress::UP _tlsReplayProgress;
// the serial num of the last message in the transaction log
SerialNum _serialNum;
SerialNum _prunedSerialNum;
bool _delayedPrune;
vespalib::Lock _feedLock;
FeedStateSP _feedState;
// used by master write thread tasks
IFeedView *_activeFeedView;
bucketdb::IBucketDBHandler *_bucketDBHandler;
vespalib::Lock _syncLock;
SerialNum _syncedSerialNum;
bool _allowSync; // Sanity check
/**
* 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);
/**
* 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 performSync();
void performEof();
/**
* Used when flushing is done
*/
void performFlushDone(SerialNum flushedSerial);
void performPrune(SerialNum flushedSerial);
FeedStateSP getFeedState() const;
void changeFeedState(FeedStateSP newState);
void changeFeedState(FeedStateSP newState, const vespalib::LockGuard &feedGuard);
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 state Document db state
* @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,
DDBState &state,
IFeedHandlerOwner &owner,
const IResourceWriteFilter &writerFilter,
IReplayConfig &replayConfig,
search::transactionlog::Writer & 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);
/**
* 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) {
_activeFeedView = feedView;
}
void setBucketDBHandler(bucketdb::IBucketDBHandler *bucketDBHandler) {
_bucketDBHandler = bucketDBHandler;
}
void setSerialNum(SerialNum serialNum) { _serialNum = serialNum; }
SerialNum incSerialNum() { return ++_serialNum; }
SerialNum getSerialNum() const override { return _serialNum; }
SerialNum getPrunedSerialNum() const { return _prunedSerialNum; }
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<search::IDestructorCallback> moveDoneCtx) override;
void heartBeat() override;
virtual void sync();
RPC::Result receive(const Packet &packet) override;
void eof() override;
void performPruneRemovedDocuments(PruneRemovedDocumentsOperation &pruneOp) override;
void syncTls(SerialNum syncTo);
void storeOperation(FeedOperation &op) override;
void considerDelayedPrune();
};
} // namespace proton
|
