path: root/library/cpp/messagebus/actor/executor.cpp

#include "executor.h"

#include "thread_extra.h"
#include "what_thread_does.h"
#include "what_thread_does_guard.h"

#include <util/generic/utility.h>
#include <util/random/random.h>
#include <util/stream/str.h>
#include <util/system/tls.h>
#include <util/system/yassert.h>

#include <array>

using namespace NActor;
using namespace NActor::NPrivate;

namespace {
    struct THistoryInternal {
        struct TRecord {
            TAtomic MaxQueueSize;

            TRecord()
                : MaxQueueSize()
            {
            }

            TExecutorHistory::THistoryRecord Capture() {
                TExecutorHistory::THistoryRecord r;
                r.MaxQueueSize = AtomicGet(MaxQueueSize);
                return r;
            }
        };

        ui64 Start;
        ui64 LastTime;

        std::array<TRecord, 3600> Records;

        THistoryInternal() {
            Start = TInstant::Now().Seconds();
            LastTime = Start - 1;
        }

        TRecord& GetRecordForTime(ui64 time) {
            return Records[time % Records.size()];
        }

        TRecord& GetNowRecord(ui64 now) {
            for (ui64 t = LastTime + 1; t <= now; ++t) {
                GetRecordForTime(t) = TRecord();
            }
            LastTime = now;
            return GetRecordForTime(now);
        }

        TExecutorHistory Capture() {
            TExecutorHistory history;
            ui64 now = TInstant::Now().Seconds();
            ui64 lastHistoryRecord = now - 1;
            ui32 historySize = Min<ui32>(lastHistoryRecord - Start, Records.size() - 1);
            history.HistoryRecords.resize(historySize);
            for (ui32 i = 0; i < historySize; ++i) {
                history.HistoryRecords[i] = GetRecordForTime(lastHistoryRecord - historySize + i).Capture();
            }
            history.LastHistoryRecordSecond = lastHistoryRecord;
            return history;
        }
    };

}

Y_POD_STATIC_THREAD(TExecutor*)
ThreadCurrentExecutor;

static const char* NoLocation = "nowhere";

struct TExecutorWorkerThreadLocalData {
    ui32 MaxQueueSize;
};

static TExecutorWorkerThreadLocalData WorkerNoThreadLocalData;
Y_POD_STATIC_THREAD(TExecutorWorkerThreadLocalData)
WorkerThreadLocalData;

namespace NActor {
    struct TExecutorWorker {
        TExecutor* const Executor;
        TThread Thread;
        const char** WhatThreadDoesLocation;
        TExecutorWorkerThreadLocalData* ThreadLocalData;

        TExecutorWorker(TExecutor* executor)
            : Executor(executor)
            , Thread(RunThreadProc, this)
            , WhatThreadDoesLocation(&NoLocation)
            , ThreadLocalData(&::WorkerNoThreadLocalData)
        {
            Thread.Start();
        }

        void Run() {
            WhatThreadDoesLocation = ::WhatThreadDoesLocation();
            AtomicSet(ThreadLocalData, &::WorkerThreadLocalData);
            WHAT_THREAD_DOES_PUSH_POP_CURRENT_FUNC();
            Executor->RunWorker();
        }

        static void* RunThreadProc(void* thiz0) {
            TExecutorWorker* thiz = (TExecutorWorker*)thiz0;
            thiz->Run();
            return nullptr;
        }
    };

    struct TExecutor::TImpl {
        TExecutor* const Executor;
        THistoryInternal History;

        TSystemEvent HelperStopSignal;
        TThread HelperThread;

        TImpl(TExecutor* executor)
            : Executor(executor)
            , HelperThread(HelperThreadProc, this)
        {
        }

        void RunHelper() {
            ui64 nowSeconds = TInstant::Now().Seconds();
            for (;;) {
                TInstant nextStop = TInstant::Seconds(nowSeconds + 1) + TDuration::MilliSeconds(RandomNumber<ui32>(1000));

                if (HelperStopSignal.WaitD(nextStop)) {
                    return;
                }

                nowSeconds = nextStop.Seconds();

                THistoryInternal::TRecord& record = History.GetNowRecord(nowSeconds);

                ui32 maxQueueSize = Executor->GetMaxQueueSizeAndClear();
                if (maxQueueSize > record.MaxQueueSize) {
                    AtomicSet(record.MaxQueueSize, maxQueueSize);
                }
            }
        }

        static void* HelperThreadProc(void* impl0) {
            TImpl* impl = (TImpl*)impl0;
            impl->RunHelper();
            return nullptr;
        }
    };

}

static TExecutor::TConfig MakeConfig(unsigned workerCount) {
    TExecutor::TConfig config;
    config.WorkerCount = workerCount;
    return config;
}

TExecutor::TExecutor(size_t workerCount)
    : Config(MakeConfig(workerCount))
{
    Init();
}

TExecutor::TExecutor(const TExecutor::TConfig& config)
    : Config(config)
{
    Init();
}

void TExecutor::Init() {
    Impl.Reset(new TImpl(this));

    AtomicSet(ExitWorkers, 0);

    Y_VERIFY(Config.WorkerCount > 0); 

    for (size_t i = 0; i < Config.WorkerCount; i++) {
        WorkerThreads.push_back(new TExecutorWorker(this));
    }

    Impl->HelperThread.Start();
}

TExecutor::~TExecutor() {
    Stop();
}

void TExecutor::Stop() {
    AtomicSet(ExitWorkers, 1);

    Impl->HelperStopSignal.Signal();
    Impl->HelperThread.Join();

    {
        TWhatThreadDoesAcquireGuard<TMutex> guard(WorkMutex, "executor: acquiring lock for Stop");
        WorkAvailable.BroadCast();
    }

    for (size_t i = 0; i < WorkerThreads.size(); i++) {
        WorkerThreads[i]->Thread.Join();
    }

    // TODO: make queue empty at this point
    ProcessWorkQueueHere();
}

void TExecutor::EnqueueWork(TArrayRef<IWorkItem* const> wis) {
    if (wis.empty())
        return;

    if (Y_UNLIKELY(AtomicGet(ExitWorkers) != 0)) {
        Y_VERIFY(WorkItems.Empty(), "executor %s: cannot add tasks after queue shutdown", Config.Name); 
    }

    TWhatThreadDoesPushPop pp("executor: EnqueueWork");

    WorkItems.PushAll(wis);

    {
        if (wis.size() == 1) {
            TWhatThreadDoesAcquireGuard<TMutex> g(WorkMutex, "executor: acquiring lock for EnqueueWork");
            WorkAvailable.Signal();
        } else {
            TWhatThreadDoesAcquireGuard<TMutex> g(WorkMutex, "executor: acquiring lock for EnqueueWork");
            WorkAvailable.BroadCast();
        }
    }
}

size_t TExecutor::GetWorkQueueSize() const {
    return WorkItems.Size();
}

using namespace NTSAN;

ui32 TExecutor::GetMaxQueueSizeAndClear() const {
    ui32 max = 0;
    for (unsigned i = 0; i < WorkerThreads.size(); ++i) {
        TExecutorWorkerThreadLocalData* wtls = RelaxedLoad(&WorkerThreads[i]->ThreadLocalData);
        max = Max<ui32>(max, RelaxedLoad(&wtls->MaxQueueSize));
        RelaxedStore<ui32>(&wtls->MaxQueueSize, 0);
    }
    return max;
}

TString TExecutor::GetStatus() const {
    return GetStatusRecordInternal().Status;
}

TString TExecutor::GetStatusSingleLine() const {
    TStringStream ss;
    ss << "work items: " << GetWorkQueueSize();
    return ss.Str();
}

TExecutorStatus TExecutor::GetStatusRecordInternal() const {
    TExecutorStatus r;

    r.WorkQueueSize = GetWorkQueueSize();

    {
        TStringStream ss;
        ss << "work items:     " << GetWorkQueueSize() << "\n";
        ss << "workers:\n";
        for (unsigned i = 0; i < WorkerThreads.size(); ++i) {
            ss << "-- " << AtomicGet(*AtomicGet(WorkerThreads[i]->WhatThreadDoesLocation)) << "\n";
        }
        r.Status = ss.Str();
    }

    r.History = Impl->History.Capture();

    return r;
}

bool TExecutor::IsInExecutorThread() const {
    return ThreadCurrentExecutor == this;
}

TAutoPtr<IWorkItem> TExecutor::DequeueWork() {
    IWorkItem* wi = reinterpret_cast<IWorkItem*>(1);
    size_t queueSize = Max<size_t>();
    if (!WorkItems.TryPop(&wi, &queueSize)) {
        TWhatThreadDoesAcquireGuard<TMutex> g(WorkMutex, "executor: acquiring lock for DequeueWork");
        while (!WorkItems.TryPop(&wi, &queueSize)) {
            if (AtomicGet(ExitWorkers) != 0)
                return nullptr;

            TWhatThreadDoesPushPop pp("waiting for work on condvar");
            WorkAvailable.Wait(WorkMutex);
        }
    }

    auto& wtls = TlsRef(WorkerThreadLocalData);

    if (queueSize > RelaxedLoad(&wtls.MaxQueueSize)) {
        RelaxedStore<ui32>(&wtls.MaxQueueSize, queueSize);
    }

    return wi;
}

void TExecutor::RunWorkItem(TAutoPtr<IWorkItem> wi) {
    WHAT_THREAD_DOES_PUSH_POP_CURRENT_FUNC();
    wi.Release()->DoWork();
}

void TExecutor::ProcessWorkQueueHere() {
    IWorkItem* wi;
    while (WorkItems.TryPop(&wi)) {
        RunWorkItem(wi);
    }
}

void TExecutor::RunWorker() {
    Y_VERIFY(!ThreadCurrentExecutor, "state check"); 
    ThreadCurrentExecutor = this;

    SetCurrentThreadName("wrkr");

    for (;;) {
        TAutoPtr<IWorkItem> wi = DequeueWork();
        if (!wi) {
            break;
        }
        // Note for messagebus users: make sure program crashes
        // on uncaught exception in thread, otherewise messagebus may just hang on error.
        RunWorkItem(wi);
    }

    ThreadCurrentExecutor = (TExecutor*)nullptr;
}