aboutsummaryrefslogblamecommitdiffstats
path: root/library/cpp/messagebus/scheduler/scheduler.cpp
blob: 5a5fe528943541f18b99d60f325959a6a9f488d5 (plain) (tree)
1
2
3
4
5
6
7
8
9
10




                                    
                             

                               

                                
                                                                                                    
                                                             


                        
                       
                                              


                           
                                        
 



                                               
                         
                                  
                                                           


                          



                           
                              


                                                      
                             
               



                                                                 
     
 
                       











                                                                        
 
                                    

                                     
         








                                                                 
             

                             
             
                                                                                      
                                                




                                                                
                                         
         






                                      
     
#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_VERIFY(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_VERIFY(!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_VERIFY(!!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();
        }
    }
}