blob: 3a9071fda2352f28a2ecb458b0166dcb0d2c3b03 (
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
|
// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#pragma once
#include <vespa/persistence/spi/result.h>
#include <vespa/persistence/spi/operationcomplete.h>
#include <vespa/searchcore/proton/common/feedtoken.h>
#include <vespa/vespalib/util/sequence.h>
#include <vespa/vespalib/util/count_down_latch.h>
#include <mutex>
namespace proton {
/**
* Base implementation for merging results from multiple sources.
*/
class TransportMerger : public feedtoken::ITransport {
public:
using Result = storage::spi::Result;
static Result mergeErrorResults(const Result &lhs, const Result &rhs);
protected:
TransportMerger(bool needLocking);
~TransportMerger() override;
void mergeResult(ResultUP result, bool documentWasFound);
virtual void completeIfDone() { } // Called with lock held if necessary on every merge
virtual ResultUP merge(ResultUP accum, ResultUP incoming, bool documentWasFound);
ResultUP _result;
private:
void mergeWithLock(ResultUP result, bool documentWasFound);
std::unique_ptr<std::mutex> _lock;
};
/**
* Implementation of FeedToken::ITransport for handling the async reply for an operation.
* Uses an internal count down latch to keep track the number of outstanding replies.
*/
class TransportLatch : public TransportMerger {
private:
using UpdateResult = storage::spi::UpdateResult;
using RemoveResult = storage::spi::RemoveResult;
vespalib::CountDownLatch _latch;
public:
TransportLatch(uint32_t cnt);
~TransportLatch() override;
void send(ResultUP result, bool documentWasFound) override;
void await() {
_latch.await();
}
const Result &getResult() const {
return *_result;
}
};
/**
* Implementation of FeedToken::ITransport for async handling of the async reply for an operation.
* Uses an internal count to keep track the number of the outstanding replies.
*/
class AsyncTransportContext : public TransportMerger {
private:
using Result = storage::spi::Result;
using OperationComplete = storage::spi::OperationComplete;
int _countDown;
OperationComplete::UP _onComplete;
void completeIfDone() override;
public:
AsyncTransportContext(uint32_t cnt, OperationComplete::UP);
~AsyncTransportContext() override;
void send(ResultUP result, bool documentWasFound) override;
};
class AsyncRemoveTransportContext : public AsyncTransportContext {
public:
using AsyncTransportContext::AsyncTransportContext;
protected:
ResultUP merge(ResultUP accum, ResultUP incoming, bool documentWasFound) override;
};
}
|
