aboutsummaryrefslogtreecommitdiffstats
path: root/library/cpp/actors/core/actorsystem.cpp
blob: c58698a2061fd69b1b1b92f1426e6cae09859acc (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
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
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
#include "defs.h"
#include "actorsystem.h"
#include "callstack.h"
#include "cpu_manager.h"
#include "mailbox.h"
#include "events.h"
#include "interconnect.h"
#include "servicemap.h"
#include "scheduler_queue.h"
#include "scheduler_actor.h"
#include "log.h"
#include "probes.h"
#include "ask.h"
#include <library/cpp/actors/util/affinity.h>
#include <library/cpp/actors/util/datetime.h>
#include <util/generic/hash.h>
#include <util/system/rwlock.h>
#include <util/random/random.h>

namespace NActors {
    LWTRACE_USING(ACTORLIB_PROVIDER);

    struct TActorSystem::TServiceMap : TNonCopyable {
        NActors::TServiceMap<TActorId, TActorId, TActorId::THash> LocalMap;
        TTicketLock Lock;

        TActorId RegisterLocalService(const TActorId& serviceId, const TActorId& actorId) {
            TTicketLock::TGuard guard(&Lock);
            const TActorId old = LocalMap.Update(serviceId, actorId);
            return old;
        }

        TActorId LookupLocal(const TActorId& x) {
            return LocalMap.Find(x);
        }
    };

    TActorSystem::TActorSystem(THolder<TActorSystemSetup>& setup, void* appData,
                               TIntrusivePtr<NLog::TSettings> loggerSettings)
        : NodeId(setup->NodeId)
        , CpuManager(new TCpuManager(setup))
        , ExecutorPoolCount(CpuManager->GetExecutorsCount())
        , Scheduler(setup->Scheduler)
        , InterconnectCount((ui32)setup->Interconnect.ProxyActors.size())
        , CurrentTimestamp(0)
        , CurrentMonotonic(0)
        , CurrentIDCounter(RandomNumber<ui64>())
        , SystemSetup(setup.Release())
        , DefSelfID(NodeId, "actorsystem")
        , AppData0(appData)
        , LoggerSettings0(loggerSettings)
        , StartExecuted(false)
        , StopExecuted(false)
        , CleanupExecuted(false)
    {
        ServiceMap.Reset(new TServiceMap());
    }

    TActorSystem::~TActorSystem() {
        Cleanup();
    }

    bool TActorSystem::Send(TAutoPtr<IEventHandle> ev) const {
        if (Y_UNLIKELY(!ev))
            return false;

#ifdef USE_ACTOR_CALLSTACK
        ev->Callstack.TraceIfEmpty();
#endif

        TActorId recipient = ev->GetRecipientRewrite();
        const ui32 recpNodeId = recipient.NodeId();

        if (recpNodeId != NodeId && recpNodeId != 0) {
            // if recipient is not local one - rewrite with forward instruction
            Y_VERIFY_DEBUG(!ev->HasEvent() || ev->GetBase()->IsSerializable());
            Y_VERIFY(ev->Recipient == recipient,
                "Event rewrite from %s to %s would be lost via interconnect",
                ev->Recipient.ToString().c_str(),
                recipient.ToString().c_str());
            recipient = InterconnectProxy(recpNodeId);
            ev->Rewrite(TEvInterconnect::EvForward, recipient);
        }
        if (recipient.IsService()) {
            TActorId target = ServiceMap->LookupLocal(recipient);
            if (!target && IsInterconnectProxyId(recipient) && ProxyWrapperFactory) {
                const TActorId actorId = ProxyWrapperFactory(const_cast<TActorSystem*>(this),
                    GetInterconnectProxyNode(recipient));
                with_lock(ProxyCreationLock) {
                    target = ServiceMap->LookupLocal(recipient);
                    if (!target) {
                        target = actorId;
                        ServiceMap->RegisterLocalService(recipient, target);
                    }
                }
                if (target != actorId) {
                    // a race has occured, terminate newly created actor
                    Send(new IEventHandle(TEvents::TSystem::Poison, 0, actorId, {}, nullptr, 0));
                }
            }
            recipient = target;
            ev->Rewrite(ev->GetTypeRewrite(), recipient);
        }

        Y_VERIFY_DEBUG(recipient == ev->GetRecipientRewrite());
        const ui32 recpPool = recipient.PoolID();
        if (recipient && recpPool < ExecutorPoolCount) {
            if (CpuManager->GetExecutorPool(recpPool)->Send(ev)) {
                return true;
            }
        }

        Send(ev->ForwardOnNondelivery(TEvents::TEvUndelivered::ReasonActorUnknown));
        return false;
    }

    bool TActorSystem::Send(const TActorId& recipient, IEventBase* ev, ui32 flags) const {
        return this->Send(new IEventHandle(recipient, DefSelfID, ev, flags));
    }

    void TActorSystem::Schedule(TInstant deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie) const {
        Schedule(deadline - Timestamp(), ev, cookie);
    }

    void TActorSystem::Schedule(TMonotonic deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie) const {
        const auto current = Monotonic();
        if (deadline < current)
            deadline = current;

        TTicketLock::TGuard guard(&ScheduleLock);
        ScheduleQueue->Writer.Push(deadline.MicroSeconds(), ev.Release(), cookie);
    }

    void TActorSystem::Schedule(TDuration delta, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie) const {
        const auto deadline = Monotonic() + delta;

        TTicketLock::TGuard guard(&ScheduleLock);
        ScheduleQueue->Writer.Push(deadline.MicroSeconds(), ev.Release(), cookie);
    }

    TActorId TActorSystem::Register(IActor* actor, TMailboxType::EType mailboxType, ui32 executorPool, ui64 revolvingCounter,
                                    const TActorId& parentId) {
        Y_VERIFY(executorPool < ExecutorPoolCount, "executorPool# %" PRIu32 ", ExecutorPoolCount# %" PRIu32,
                 (ui32)executorPool, (ui32)ExecutorPoolCount);
        return CpuManager->GetExecutorPool(executorPool)->Register(actor, mailboxType, revolvingCounter, parentId);
    }

    NThreading::TFuture<THolder<IEventBase>> TActorSystem::AskGeneric(TMaybe<ui32> expectedEventType,
                                                                      TActorId recipient, THolder<IEventBase> event,
                                                                      TDuration timeout) {
        auto promise = NThreading::NewPromise<THolder<IEventBase>>();
        Register(MakeAskActor(expectedEventType, recipient, std::move(event), timeout, promise).Release());
        return promise.GetFuture();
    }

    ui64 TActorSystem::AllocateIDSpace(ui64 count) {
        Y_VERIFY_DEBUG(count < Max<ui32>() / 65536);

        static_assert(sizeof(TAtomic) == sizeof(ui64), "expect sizeof(TAtomic) == sizeof(ui64)");

        // get high 32 bits as seconds from epoch
        // it could wrap every century, but we don't expect any actor-reference to live this long so such wrap will do no harm
        const ui64 timeFromEpoch = TInstant::MicroSeconds(RelaxedLoad(&CurrentTimestamp)).Seconds();

        // get low 32 bits as counter value
        ui32 lowPartEnd = (ui32)(AtomicAdd(CurrentIDCounter, count));
        while (lowPartEnd < count) // if our request crosses 32bit boundary - retry
            lowPartEnd = (ui32)(AtomicAdd(CurrentIDCounter, count));

        const ui64 lowPart = lowPartEnd - count;
        const ui64 ret = (timeFromEpoch << 32) | lowPart;

        return ret;
    }

    TActorId TActorSystem::InterconnectProxy(ui32 destinationNode) const {
        if (destinationNode < InterconnectCount)
            return Interconnect[destinationNode];
        else if (destinationNode != NodeId)
            return MakeInterconnectProxyId(destinationNode);
        else
            return TActorId();
    }

    ui32 TActorSystem::BroadcastToProxies(const std::function<IEventHandle*(const TActorId&)>& eventFabric) {
        // TODO: get rid of this method
        for (ui32 i = 0; i < InterconnectCount; ++i) {
            Send(eventFabric(Interconnect[i]));
        }
        return InterconnectCount;
    }

    TActorId TActorSystem::LookupLocalService(const TActorId& x) const {
        return ServiceMap->LookupLocal(x);
    }

    TActorId TActorSystem::RegisterLocalService(const TActorId& serviceId, const TActorId& actorId) {
        // TODO: notify old actor about demotion
        return ServiceMap->RegisterLocalService(serviceId, actorId);
    }

    void TActorSystem::GetPoolStats(ui32 poolId, TExecutorPoolStats& poolStats, TVector<TExecutorThreadStats>& statsCopy) const {
        CpuManager->GetPoolStats(poolId, poolStats, statsCopy);
    }

    void TActorSystem::Start() {
        Y_VERIFY(StartExecuted == false);
        StartExecuted = true;

        ScheduleQueue.Reset(new NSchedulerQueue::TQueueType());
        TVector<NSchedulerQueue::TReader*> scheduleReaders;
        scheduleReaders.push_back(&ScheduleQueue->Reader);
        CpuManager->PrepareStart(scheduleReaders, this);
        Scheduler->Prepare(this, &CurrentTimestamp, &CurrentMonotonic);
        Scheduler->PrepareSchedules(&scheduleReaders.front(), (ui32)scheduleReaders.size());

        // setup interconnect proxies
        {
            const TInterconnectSetup& setup = SystemSetup->Interconnect;
            Interconnect.Reset(new TActorId[InterconnectCount + 1]);
            for (ui32 i = 0, e = InterconnectCount; i != e; ++i) {
                const TActorSetupCmd& x = setup.ProxyActors[i];
                if (x.Actor) {
                    Interconnect[i] = Register(x.Actor, x.MailboxType, x.PoolId, i);
                    Y_VERIFY(!!Interconnect[i]);
                }
            }
            ProxyWrapperFactory = std::move(SystemSetup->Interconnect.ProxyWrapperFactory);
        }

        // setup local services
        {
            for (ui32 i = 0, e = (ui32)SystemSetup->LocalServices.size(); i != e; ++i) {
                const std::pair<TActorId, TActorSetupCmd>& x = SystemSetup->LocalServices[i];
                const TActorId xid = Register(x.second.Actor, x.second.MailboxType, x.second.PoolId, i);
                Y_VERIFY(!!xid);
                if (!!x.first)
                    RegisterLocalService(x.first, xid);
            }
        }

        // ok, setup complete, we could destroy setup config
        SystemSetup.Destroy();

        Scheduler->PrepareStart();
        CpuManager->Start();
        Send(MakeSchedulerActorId(), new TEvSchedulerInitialize(scheduleReaders, &CurrentTimestamp, &CurrentMonotonic));
        Scheduler->Start();
    }

    void TActorSystem::Stop() {
        if (StopExecuted || !StartExecuted)
            return;

        StopExecuted = true;

        for (auto&& fn : std::exchange(DeferredPreStop, {})) {
            fn();
        }

        Scheduler->PrepareStop();
        CpuManager->PrepareStop();
        Scheduler->Stop();
        CpuManager->Shutdown();
    }

    void TActorSystem::Cleanup() {
        Stop();
        if (CleanupExecuted || !StartExecuted)
            return;
        CleanupExecuted = true;
        CpuManager->Cleanup();
        Scheduler.Destroy();
    }

    ui32 TActorSystem::MemProfActivityBase;
}