aboutsummaryrefslogtreecommitdiffstats
path: root/library/cpp/messagebus/scheduler/scheduler.cpp
blob: 149d12c41befcd0cfc28fc3d290c2f5206382d0f (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
#include "scheduler.h"

#include <util/datetime/base.h>
#include <util/generic/algorithm.h>
#include <util/generic/yexception.h>

//#include "dummy_debugger.h"

using namespace NBus;
using namespace NBus::NPrivate;

class TScheduleDeadlineCompare {
public:
    bool operator()(const IScheduleItemAutoPtr& i1, const IScheduleItemAutoPtr& i2) const noexcept {
        return i1->GetScheduleTime() > i2->GetScheduleTime();
    }
};

TScheduler::TScheduler()
    : StopThread(false)
    , Thread([&] { this->SchedulerThread(); })
{
}

TScheduler::~TScheduler() {
    Y_ABORT_UNLESS(StopThread, "state check");
}

size_t TScheduler::Size() const {
    TGuard<TLock> guard(Lock);
    return Items.size() + (!!NextItem ? 1 : 0);
}

void TScheduler::Stop() {
    {
        TGuard<TLock> guard(Lock);
        Y_ABORT_UNLESS(!StopThread, "Scheduler already stopped");
        StopThread = true;
        CondVar.Signal();
    }
    Thread.Get();

    if (!!NextItem) {
        NextItem.Destroy();
    }

    for (auto& item : Items) {
        item.Destroy();
    }
}

void TScheduler::Schedule(TAutoPtr<IScheduleItem> i) {
    TGuard<TLock> lock(Lock);
    if (StopThread)
        return;

    if (!!NextItem) {
        if (i->GetScheduleTime() < NextItem->GetScheduleTime()) {
            DoSwap(i, NextItem);
        }
    }

    Items.push_back(i);
    PushHeap(Items.begin(), Items.end(), TScheduleDeadlineCompare());

    FillNextItem();

    CondVar.Signal();
}

void TScheduler::FillNextItem() {
    if (!NextItem && !Items.empty()) {
        PopHeap(Items.begin(), Items.end(), TScheduleDeadlineCompare());
        NextItem = Items.back();
        Items.erase(Items.end() - 1);
    }
}

void TScheduler::SchedulerThread() {
    for (;;) {
        IScheduleItemAutoPtr current;

        {
            TGuard<TLock> guard(Lock);

            if (StopThread) {
                break;
            }

            if (!!NextItem) {
                CondVar.WaitD(Lock, NextItem->GetScheduleTime());
            } else {
                CondVar.WaitI(Lock);
            }

            if (StopThread) {
                break;
            }

            // signal comes if either scheduler is to be stopped of there's work to do
            Y_ABORT_UNLESS(!!NextItem, "state check");

            if (TInstant::Now() < NextItem->GetScheduleTime()) {
                // NextItem is updated since WaitD
                continue;
            }

            current = NextItem.Release();
        }

        current->Do();
        current.Destroy();

        {
            TGuard<TLock> guard(Lock);
            FillNextItem();
        }
    }
}