1/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
2/*
3 *     Copyright 2016 Couchbase, Inc.
4 *
5 *   Licensed under the Apache License, Version 2.0 (the "License");
6 *   you may not use this file except in compliance with the License.
7 *   You may obtain a copy of the License at
8 *
9 *       http://www.apache.org/licenses/LICENSE-2.0
10 *
11 *   Unless required by applicable law or agreed to in writing, software
12 *   distributed under the License is distributed on an "AS IS" BASIS,
13 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 *   See the License for the specific language governing permissions and
15 *   limitations under the License.
16 */
17#include "scrubber_task.h"
18
19#include "default_engine_internal.h"
20#include "engine_manager.h"
21
22static void scrubber_task_main(void* arg) {
23    ScrubberTask* task = reinterpret_cast<ScrubberTask*>(arg);
24    task->run();
25}
26
27ScrubberTask::ScrubberTask(EngineManager& manager)
28    : state(State::Idle),
29      shuttingdown(false),
30      engineManager(manager) {
31    std::unique_lock<std::mutex> lck(lock);
32    if (cb_create_named_thread(&scrubberThread, &scrubber_task_main, this, 0,
33                               "mc:item scrub") != 0) {
34        throw std::runtime_error("Error creating 'mc:item scrub' thread");
35    }
36}
37
38void ScrubberTask::shutdown() {
39    std::unique_lock<std::mutex> lck(lock);
40    shuttingdown = true;
41    // Serialize with ::run
42    cvar.notify_one();
43}
44
45void ScrubberTask::joinThread() {
46    cb_join_thread(scrubberThread);
47}
48
49void ScrubberTask::placeOnWorkQueue(struct default_engine* engine,
50                                    bool destroy) {
51    std::lock_guard<std::mutex> lck(lock);
52    if (!shuttingdown) {
53        engine->scrubber.force_delete = destroy;
54        workQueue.push_back(std::make_pair(engine, destroy));
55        cvar.notify_one();
56    }
57}
58
59void ScrubberTask::run() {
60    std::unique_lock<std::mutex> lck(lock);
61    while (!shuttingdown) {
62        if (!workQueue.empty()) {
63            auto engine = workQueue.front();
64            workQueue.pop_front();
65            state = State::Scrubbing;
66            lck.unlock();
67            // Run the task without holding the lock
68            item_scrubber_main(engine.first);
69            engineManager.notifyScrubComplete(engine.first, engine.second);
70
71            // relock so lck can safely unlock when destroyed at loop end.
72            lck.lock();
73        } else {
74            state = State::Idle;
75            cvar.wait(lck);
76        }
77    }
78    state = State::Stopped;
79}
80