library/cpp/messagebus/actor/tasks.h


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

#pragma once

#include <util/system/atomic.h>
#include <util/system/yassert.h>

namespace NActor {
    class TTasks {
        enum {
            // order of values is important
            E_WAITING,
            E_RUNNING_NO_TASKS,
            E_RUNNING_GOT_TASKS,
        };

    private:
        TAtomic State;

    public:
        TTasks()
            : State(E_WAITING)
        {
        }

        // @return true iff caller have to either schedule task or execute it
        bool AddTask() {
            // High contention case optimization: AtomicGet is cheaper than AtomicSwap.
            if (E_RUNNING_GOT_TASKS == AtomicGet(State)) {
                return false;
            }

            TAtomicBase oldState = AtomicSwap(&State, E_RUNNING_GOT_TASKS);
            return oldState == E_WAITING;
        }

        // called by executor
        // @return true iff we have to recheck queues
        bool FetchTask() {
            TAtomicBase newState = AtomicDecrement(State);
            if (newState == E_RUNNING_NO_TASKS) {
                return true;
            } else if (newState == E_WAITING) {
                return false;
            }
            Y_FAIL("unknown");
        }
    };

}