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
|
#include "services.h"
#include <library/cpp/actors/core/actorsystem.h>
#include <library/cpp/actors/core/hfunc.h>
#include <library/cpp/actors/core/interconnect.h>
#include <library/cpp/actors/core/actor_bootstrapped.h>
#include <util/generic/set.h>
#include <util/generic/vector.h>
class TExampleLookupRequestActor : public TActor<TExampleLookupRequestActor> {
const TActorId Owner;
const TActorId Replica;
const TString Key;
void Registered(TActorSystem* sys, const TActorId&) override {
const auto flags = IEventHandle::FlagTrackDelivery | IEventHandle::FlagSubscribeOnSession;
sys->Send(new IEventHandle(Replica, SelfId(), new TEvExample::TEvReplicaLookup(Key), flags));
}
void PassAway() override {
const ui32 replicaNode = Replica.NodeId();
if (replicaNode != SelfId().NodeId()) {
const TActorId &interconnectProxy = TlsActivationContext->ExecutorThread.ActorSystem->InterconnectProxy(Replica.NodeId());
Send(interconnectProxy, new TEvents::TEvUnsubscribe());
}
return IActor::PassAway();
}
void Handle(TEvExample::TEvReplicaInfo::TPtr &ev) {
Send(Owner, ev->Release().Release());
return PassAway();
}
void HandleUndelivered() {
Send(Owner, new TEvExample::TEvReplicaInfo(Key));
return PassAway();
}
public:
static constexpr IActor::EActivityType ActorActivityType() {
// define app-specific activity tag to track elapsed cpu | handled events | actor count in Solomon
return EActorActivity::ACTORLIB_COMMON;
}
TExampleLookupRequestActor(TActorId owner, TActorId replica, const TString &key)
: TActor(&TThis::StateWork)
, Owner(owner)
, Replica(replica)
, Key(key)
{}
STFUNC(StateWork) {
Y_UNUSED(ctx);
switch (ev->GetTypeRewrite()) {
hFunc(TEvExample::TEvReplicaInfo, Handle);
cFunc(TEvents::TEvUndelivered::EventType, HandleUndelivered);
cFunc(TEvInterconnect::TEvNodeDisconnected::EventType, HandleUndelivered);
default:
break;
}
}
};
class TExampleLookupActor : public TActorBootstrapped<TExampleLookupActor> {
TIntrusiveConstPtr<TExampleStorageConfig> Config;
const TString Key;
const TActorId ReplyTo;
TVector<TActorId> RequestActors;
ui32 TotalReplicas = 0;
ui32 RepliedSuccess = 0;
ui32 RepliedError = 0;
TSet<TString> Payloads;
void Handle(TEvExample::TEvReplicaInfo::TPtr &ev) {
NActorsExample::TEvReplicaInfo &record = ev->Get()->Record;
if (record.PayloadSize()) {
++RepliedSuccess;
for (const TString &payload : record.GetPayload()) {
Payloads.insert(payload);
}
}
else {
++RepliedError;
}
const ui32 majority = (TotalReplicas / 2 + 1);
if (RepliedSuccess == majority || (RepliedSuccess + RepliedError == TotalReplicas))
return ReplyAndDie();
}
void ReplyAndDie() {
TVector<TString> replyPayloads(Payloads.begin(), Payloads.end());
Send(ReplyTo, new TEvExample::TEvInfo(Key, std::move(replyPayloads)));
return PassAway();
}
public:
static constexpr IActor::EActivityType ActorActivityType() {
// define app-specific activity tag to track elapsed cpu | handled events | actor count in Solomon
return EActorActivity::ACTORLIB_COMMON;
}
TExampleLookupActor(TExampleStorageConfig *config, const TString &key, TActorId replyTo)
: Config(config)
, Key(key)
, ReplyTo(replyTo)
{}
void Bootstrap() {
Y_VERIFY(Config->Replicas.size() > 0);
TotalReplicas = Config->Replicas.size();
RequestActors.reserve(TotalReplicas);
for (const auto &replica : Config->Replicas) {
const TActorId requestActor = Register(new TExampleLookupRequestActor(SelfId(), replica, Key));
RequestActors.emplace_back(requestActor);
}
Become(&TThis::StateWork);
}
STFUNC(StateWork) {
Y_UNUSED(ctx);
switch (ev->GetTypeRewrite()) {
hFunc(TEvExample::TEvReplicaInfo, Handle);
default:
break;
}
}
};
IActor* CreateLookupActor(TExampleStorageConfig *config, const TString &key, TActorId replyTo) {
return new TExampleLookupActor(config, key, replyTo);
}
|
