KeyDB/src/AsyncWorkQueue.cpp at main · Snapchat/KeyDB · GitHub

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
#include "AsyncWorkQueue.h"
#include "server.h"

AsyncWorkQueue::AsyncWorkQueue(int nthreads)
{
    for (int i = 0; i < nthreads; ++i)
    {
        m_vecthreads.emplace_back([&]{
            WorkerThreadMain();
        });
    }
}

void AsyncWorkQueue::WorkerThreadMain()
{
    redisServerThreadVars vars;
    serverTL = &vars;

    vars.clients_pending_asyncwrite = listCreate();

    m_mutex.lock();
    m_vecpthreadVars.push_back(&vars);
    m_mutex.unlock();

    while (!m_fQuitting)
    {
        std::unique_lock<std::mutex> lock(m_mutex);
        if (m_workqueue.empty())
            m_cvWakeup.wait(lock);

        aeThreadOnline();
        while (!m_workqueue.empty())
        {
            WorkItem task = std::move(m_workqueue.front());
            m_workqueue.pop_front();

            lock.unlock();
            serverTL->gcEpoch = g_pserver->garbageCollector.startEpoch();
            task.fnAsync();
            g_pserver->garbageCollector.endEpoch(serverTL->gcEpoch);
            lock.lock();
        }

        lock.unlock();
        serverTL->gcEpoch = g_pserver->garbageCollector.startEpoch();
        if (listLength(serverTL->clients_pending_asyncwrite)) {
            aeAcquireLock();
            ProcessPendingAsyncWrites();
            aeReleaseLock();
        }
        g_pserver->garbageCollector.endEpoch(serverTL->gcEpoch);
        serverTL->gcEpoch.reset();
        aeThreadOffline();
    }

    listRelease(vars.clients_pending_asyncwrite);

    std::unique_lock<fastlock> lockf(serverTL->lockPendingWrite);
    serverTL->vecclientsProcess.clear();
    serverTL->clients_pending_write.clear();
    std::atomic_thread_fence(std::memory_order_seq_cst);
}

bool AsyncWorkQueue::removeClientAsyncWrites(client *c)
{
    bool fFound = false;
    aeAcquireLock();
    m_mutex.lock();
    for (auto pvars : m_vecpthreadVars)
    {
        listIter li;
        listNode *ln;
        listRewind(pvars->clients_pending_asyncwrite, &li);
        while ((ln = listNext(&li)) != nullptr)
        {
            if (c == listNodeValue(ln))
            {
                listDelNode(pvars->clients_pending_asyncwrite, ln);
                fFound = true;
            }
        }
    }
    m_mutex.unlock();
    aeReleaseLock();
    return fFound;
}

void AsyncWorkQueue::shutdown()
{
    std::unique_lock<std::mutex> lock(m_mutex);
    serverAssert(!GlobalLocksAcquired());
    m_fQuitting = true;
    m_cvWakeup.notify_all();
    lock.unlock();

    for (auto &thread : m_vecthreads)
        thread.join();
}

void AsyncWorkQueue::abandonThreads()
{
    std::unique_lock<std::mutex> lock(m_mutex);
    m_fQuitting = true;
    m_cvWakeup.notify_all();
    for (auto &thread : m_vecthreads)
    {
        thread.detach();
    }
    m_vecthreads.clear();
}

AsyncWorkQueue::~AsyncWorkQueue()
{
    serverAssert(!GlobalLocksAcquired() || m_vecthreads.empty());
    std::unique_lock<std::mutex> lock(m_mutex);
    m_fQuitting = true;
    m_cvWakeup.notify_all();
    lock.unlock();

    abandonThreads();
}

void AsyncWorkQueue::AddWorkFunction(std::function<void()> &&fnAsync, bool fHiPri)
{
    std::unique_lock<std::mutex> lock(m_mutex);
    if (fHiPri)
        m_workqueue.emplace_front(std::move(fnAsync));
    else
        m_workqueue.emplace_back(std::move(fnAsync));
    m_cvWakeup.notify_one();
}