1 /* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
2 /*
3  *     Copyright 2013 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 "config.h"
18 
19 #include "bgfetcher.h"
20 #include "ep_bucket.h"
21 #include "ep_engine.h"
22 #include "flusher.h"
23 #include "kvstore.h"
24 #include "tasks.h"
25 #include "warmup.h"
26 
27 #include <climits>
28 #include <type_traits>
29 
30 #include <phosphor/phosphor.h>
31 
32 static const double WORKLOAD_MONITOR_FREQ(5.0);
33 
run()34 bool FlusherTask::run() {
35     return flusher->step(this);
36 }
37 
CompactTask(EPBucket& bucket, compaction_ctx c, const void* ck, bool completeBeforeShutdown)38 CompactTask::CompactTask(EPBucket& bucket,
39                          compaction_ctx c,
40                          const void* ck,
41                          bool completeBeforeShutdown)
42     : GlobalTask(&bucket.getEPEngine(),
43                  TaskId::CompactVBucketTask,
44                  0,
45                  completeBeforeShutdown),
46       bucket(bucket),
47       compactCtx(c),
48       cookie(ck) {
49     desc = "Compact DB file " + std::to_string(c.db_file_id);
50 }
51 
run()52 bool CompactTask::run() {
53     TRACE_EVENT1(
54             "ep-engine/task", "CompactTask", "file_id", compactCtx.db_file_id);
55     return bucket.doCompact(&compactCtx, cookie);
56 }
57 
run()58 bool StatSnap::run() {
59     TRACE_EVENT0("ep-engine/task", "StatSnap");
60     engine->getKVBucket()->snapshotStats();
61     if (runOnce) {
62         return false;
63     }
64     ExecutorPool::get()->snooze(uid, 60);
65     return true;
66 }
67 
MultiBGFetcherTask(EventuallyPersistentEngine* e, BgFetcher* b)68 MultiBGFetcherTask::MultiBGFetcherTask(EventuallyPersistentEngine* e,
69                                        BgFetcher* b)
70     : GlobalTask(e,
71                  TaskId::MultiBGFetcherTask,
72                  /*sleeptime*/ INT_MAX,
73                  /*completeBeforeShutdown*/ false),
74       bgfetcher(b) {
75 }
76 
run()77 bool MultiBGFetcherTask::run() {
78     TRACE_EVENT0("ep-engine/task", "MultiBGFetcherTask");
79     return bgfetcher->run(this);
80 }
81 
run()82 bool VKeyStatBGFetchTask::run() {
83     TRACE_EVENT2("ep-engine/task",
84                  "VKeyStatBGFetchTask",
85                  "cookie",
86                  cookie,
87                  "vb",
88                  vbucket);
89     engine->getKVBucket()->completeStatsVKey(cookie, key, vbucket, bySeqNum);
90     return false;
91 }
92 
93 
run()94 bool SingleBGFetcherTask::run() {
95     TRACE_EVENT2("ep-engine/task",
96                  "SingleBGFetcherTask",
97                  "cookie",
98                  cookie,
99                  "vb",
100                  vbucket);
101     engine->getKVBucket()->completeBGFetch(key, vbucket, cookie, init,
102                                            metaFetch);
103     return false;
104 }
105 
WorkLoadMonitor(EventuallyPersistentEngine *e, bool completeBeforeShutdown)106 WorkLoadMonitor::WorkLoadMonitor(EventuallyPersistentEngine *e,
107                                  bool completeBeforeShutdown) :
108     GlobalTask(e, TaskId::WorkLoadMonitor, WORKLOAD_MONITOR_FREQ,
109                completeBeforeShutdown) {
110     prevNumMutations = getNumMutations();
111     prevNumGets = getNumGets();
112 }
113 
getNumMutations()114 size_t WorkLoadMonitor::getNumMutations() {
115     return engine->getEpStats().numOpsStore +
116            engine->getEpStats().numOpsDelete +
117            engine->getEpStats().numOpsSetMeta +
118            engine->getEpStats().numOpsDelMeta +
119            engine->getEpStats().numOpsSetRetMeta +
120            engine->getEpStats().numOpsDelRetMeta;
121 }
122 
getNumGets()123 size_t WorkLoadMonitor::getNumGets() {
124     return engine->getEpStats().numOpsGet +
125            engine->getEpStats().numOpsGetMeta;
126 }
127 
run()128 bool WorkLoadMonitor::run() {
129     size_t curr_num_mutations = getNumMutations();
130     size_t curr_num_gets = getNumGets();
131     double delta_mutations = static_cast<double>(curr_num_mutations -
132                                                  prevNumMutations);
133     double delta_gets = static_cast<double>(curr_num_gets - prevNumGets);
134     double total_delta_ops = delta_gets + delta_mutations;
135     double read_ratio = 0;
136 
137     if (total_delta_ops) {
138         read_ratio = delta_gets / total_delta_ops;
139         if (read_ratio < 0.4) {
140             engine->getWorkLoadPolicy().setWorkLoadPattern(WRITE_HEAVY);
141         } else if (read_ratio >= 0.4 && read_ratio <= 0.6) {
142             engine->getWorkLoadPolicy().setWorkLoadPattern(MIXED);
143         } else {
144             engine->getWorkLoadPolicy().setWorkLoadPattern(READ_HEAVY);
145         }
146     }
147     prevNumMutations = curr_num_mutations;
148     prevNumGets = curr_num_gets;
149 
150     snooze(WORKLOAD_MONITOR_FREQ);
151     if (engine->getEpStats().isShutdown) {
152         return false;
153     }
154     return true;
155 }
156