blob: 9f0111b32f9c320770aca15e56963a1759ad4737 (
plain) (
blame)
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
|
// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#pragma once
#include <vespa/storageframework/generic/thread/runnable.h>
#include <vespa/storage/common/content_bucket_space_repo.h>
#include <vespa/storage/common/storagecomponent.h>
#include <vespa/storage/common/servicelayercomponent.h>
#include <vespa/storage/common/storagelink.h>
#include <vespa/storage/config/config-stor-server.h>
#include <vespa/storage/persistence/messages.h>
#include <vespa/storage/persistence/types.h>
#include <vespa/document/bucket/bucketidlist.h>
#include <vespa/config/helper/ifetchercallback.h>
namespace config {
class ConfigUri;
class ConfigFetcher;
}
namespace storage {
namespace spi { struct PersistenceProvider; }
class ModifiedBucketChecker
: public StorageLink,
public framework::Runnable,
public Types
{
public:
using StorServerConfig = vespa::config::content::core::StorServerConfig;
using UP = std::unique_ptr<ModifiedBucketChecker>;
ModifiedBucketChecker(ServiceLayerComponentRegister& compReg,
spi::PersistenceProvider& provide,
const StorServerConfig& bootstrap_config);
~ModifiedBucketChecker() override;
void on_configure(const vespa::config::content::core::StorServerConfig&);
void run(framework::ThreadHandle& thread) override;
bool tick();
void onOpen() override;
void onClose() override;
void setUnitTestingSingleThreadedMode() {
_singleThreadMode = true;
}
private:
bool onInternalReply(const std::shared_ptr<api::InternalReply>&) override;
bool currentChunkFinished() const {
return _pendingRequests == 0;
}
bool moreChunksRemaining() const {
return !_rechecksNotStarted.empty();
}
bool requestModifiedBucketsFromProvider(document::BucketSpace bucketSpace);
void nextRecheckChunk(std::vector<RecheckBucketInfoCommand::SP>&);
void dispatchAllToPersistenceQueues(const std::vector<RecheckBucketInfoCommand::SP>&);
class CyclicBucketSpaceIterator {
private:
ContentBucketSpaceRepo::BucketSpaces _bucketSpaces;
size_t _idx;
public:
using UP = std::unique_ptr<CyclicBucketSpaceIterator>;
explicit CyclicBucketSpaceIterator(ContentBucketSpaceRepo::BucketSpaces bucketSpaces);
document::BucketSpace next() {
return _bucketSpaces[(_idx++)%_bucketSpaces.size()];
}
};
class BucketIdListResult {
private:
document::BucketSpace _bucketSpace;
document::bucket::BucketIdList _buckets;
public:
BucketIdListResult();
void reset(document::BucketSpace bucketSpace,
document::bucket::BucketIdList &buckets);
const document::BucketSpace &bucketSpace() const { return _bucketSpace; }
size_t size() const { return _buckets.size(); }
bool empty() const { return _buckets.empty(); }
const document::BucketId &back() const { return _buckets.back(); }
void pop_back() { _buckets.pop_back(); }
};
spi::PersistenceProvider & _provider;
ServiceLayerComponent::UP _component;
std::unique_ptr<framework::Thread> _thread;
std::mutex _monitor;
std::condition_variable _cond;
std::mutex _stateLock;
CyclicBucketSpaceIterator::UP _bucketSpaces;
BucketIdListResult _rechecksNotStarted;
size_t _pendingRequests;
size_t _maxPendingChunkSize;
bool _singleThreadMode; // For unit testing only
};
} // ns storage
|
