DEBUG: Add HashTable::operator<<, expand StoredValue::operator<<
[ep-engine.git] / src / linked_list.cc
1 /* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
2 /*
3  *     Copyright 2017 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
18 #include "linked_list.h"
19 #include <mutex>
20
21 BasicLinkedList::BasicLinkedList(uint16_t vbucketId, EPStats& st)
22     : SequenceList(),
23       readRange(0, 0),
24       staleSize(0),
25       staleMetaDataSize(0),
26       highSeqno(0),
27       highestDedupedSeqno(0),
28       highestPurgedDeletedSeqno(0),
29       numStaleItems(0),
30       numDeletedItems(0),
31       vbid(vbucketId),
32       st(st) {
33 }
34
35 BasicLinkedList::~BasicLinkedList() {
36     /* Delete stale items here, other items are deleted by the hash
37        table */
38     std::lock_guard<std::mutex> writeGuard(writeLock);
39     seqList.remove_and_dispose_if(
40             [&writeGuard](const OrderedStoredValue& v) {
41                 return v.isStale(writeGuard);
42             },
43             [this](OrderedStoredValue* v) {
44                 this->st.currentSize.fetch_sub(v->metaDataSize());
45                 delete v;
46             });
47
48     /* Erase all the list elements (does not destroy elements, just removes
49        them from the list) */
50     seqList.clear();
51 }
52
53 void BasicLinkedList::appendToList(std::lock_guard<std::mutex>& seqLock,
54                                    OrderedStoredValue& v) {
55     /* Allow only one write to the list at a time */
56     std::lock_guard<std::mutex> lckGd(writeLock);
57
58     seqList.push_back(v);
59 }
60
61 SequenceList::UpdateStatus BasicLinkedList::updateListElem(
62         std::lock_guard<std::mutex>& seqLock, OrderedStoredValue& v) {
63     /* Allow only one write to the list at a time */
64     std::lock_guard<std::mutex> lckGd(writeLock);
65
66     /* Lock that needed for consistent read of SeqRange 'readRange' */
67     std::lock_guard<SpinLock> lh(rangeLock);
68
69     if (readRange.fallsInRange(v.getBySeqno())) {
70         /* Range read is in middle of a point-in-time snapshot, hence we cannot
71            move the element to the end of the list. Return a temp failure */
72         return UpdateStatus::Append;
73     }
74
75     /* Since there is no other reads or writes happenning in this range, we can
76        move the item to the end of the list */
77     auto it = seqList.iterator_to(v);
78     seqList.erase(it);
79     seqList.push_back(v);
80
81     return UpdateStatus::Success;
82 }
83
84 std::pair<ENGINE_ERROR_CODE, std::vector<UniqueItemPtr>>
85 BasicLinkedList::rangeRead(seqno_t start, seqno_t end) {
86     std::vector<UniqueItemPtr> empty;
87     if ((start > end) || (start <= 0)) {
88         LOG(EXTENSION_LOG_WARNING,
89             "BasicLinkedList::rangeRead(): "
90             "(vb:%d) ERANGE: start %" PRIi64 " > end %" PRIi64,
91             vbid,
92             start,
93             end);
94         return {ENGINE_ERANGE, std::move(empty)
95                 /* MSVC not happy with std::vector<UniqueItemPtr>() */};
96     }
97
98     /* Allows only 1 rangeRead for now */
99     std::lock_guard<std::mutex> lckGd(rangeReadLock);
100
101     {
102         std::lock_guard<SpinLock> lh(rangeLock);
103         if (start > highSeqno) {
104             LOG(EXTENSION_LOG_WARNING,
105                 "BasicLinkedList::rangeRead(): "
106                 "(vb:%d) ERANGE: start %" PRIi64 " > highSeqno %" PRIi64,
107                 vbid,
108                 start,
109                 static_cast<seqno_t>(highSeqno));
110             /* If the request is for an invalid range, return before iterating
111                through the list */
112             return {ENGINE_ERANGE, std::move(empty)};
113         }
114
115         /* Mark the initial read range */
116         end = std::min(end, static_cast<seqno_t>(highSeqno));
117         end = std::max(end, static_cast<seqno_t>(highestDedupedSeqno));
118         readRange = SeqRange(1, end);
119     }
120
121     /* Read items in the range */
122     std::vector<UniqueItemPtr> items;
123
124     for (const auto& osv : seqList) {
125         if (osv.getBySeqno() > end) {
126             /* we are done */
127             break;
128         }
129
130         {
131             std::lock_guard<SpinLock> lh(rangeLock);
132             readRange.setBegin(osv.getBySeqno()); /* [EPHE TODO]: should we
133                                                      update the min every time ?
134                                                    */
135         }
136
137         if (osv.getBySeqno() < start) {
138             /* skip this item */
139             continue;
140         }
141
142         try {
143             items.push_back(UniqueItemPtr(osv.toItem(false, vbid)));
144         } catch (const std::bad_alloc&) {
145             /* [EPHE TODO]: Do we handle backfill in a better way ?
146                             Perhaps do backfilling partially (that is
147                             backfill ==> stream; backfill ==> stream ..so on )?
148              */
149             LOG(EXTENSION_LOG_WARNING,
150                 "BasicLinkedList::rangeRead(): "
151                 "(vb %d) ENOMEM while trying to copy "
152                 "item with seqno %" PRIi64 "before streaming it",
153                 vbid,
154                 osv.getBySeqno());
155             return {ENGINE_ENOMEM, std::move(empty)};
156         }
157     }
158
159     /* Done with range read, reset the range */
160     {
161         std::lock_guard<SpinLock> lh(rangeLock);
162         readRange.reset();
163     }
164
165     /* Return all the range read items */
166     return {ENGINE_SUCCESS, std::move(items)};
167 }
168
169 void BasicLinkedList::updateHighSeqno(const OrderedStoredValue& v) {
170     std::lock_guard<SpinLock> lh(rangeLock);
171     highSeqno = v.getBySeqno();
172     if (v.isDeleted()) {
173         ++numDeletedItems;
174     }
175 }
176
177 void BasicLinkedList::markItemStale(StoredValue::UniquePtr ownedSv) {
178     /* Release the StoredValue as BasicLinkedList does not want it to be of
179        owned type */
180     StoredValue* v = ownedSv.release();
181
182     /* Update the stats tracking the memory owned by the list */
183     staleSize.fetch_add(v->size());
184     staleMetaDataSize.fetch_add(v->metaDataSize());
185     st.currentSize.fetch_add(v->metaDataSize());
186
187     ++numStaleItems;
188     {
189         std::lock_guard<std::mutex> writeGuard(writeLock);
190         v->toOrderedStoredValue()->markStale(writeGuard);
191     }
192 }
193
194 size_t BasicLinkedList::purgeTombstones() {
195     // Purge items marked as stale from the seqList.
196     //
197     // Strategy - we try to ensure that this function does not block
198     // frontend-writes (adding new OrderedStoredValues (OSVs) to the seqList).
199     // To achieve this (safely),
200     // we (try to) acquire the rangeReadLock and setup a 'read' range for the
201     // whole of the seqList. This prevents any other readers from iterating
202     // the list (and accessing stale items) while we purge on it; but permits
203     // front-end operations to continue as they:
204     //   a) Only read/modify non-stale items (we only change stale items) and
205     //   b) Do not change the list membership of anything within the read-range.
206     // However, we do need to be careful about what members of OSVs we access
207     // here - the only OSVs we can safely access are ones marked stale as they
208     // are no longer in the HashTable (and hence subject to HashTable locks).
209     // To check if an item is stale we need to acquire the writeLock
210     // (OSV::stale is guarded by it) for each list item. While this isn't
211     // ideal (that's the same lock needed by front-end operations), we can
212     // release the lock between each element so front-end operations can
213     // have the opportunity to acquire it.
214     //
215     // Attempt to acquire the readRangeLock, to block anyone else concurrently
216     // reading from the list while we remove elements from it.
217     std::unique_lock<std::mutex> rrGuard(rangeReadLock, std::try_to_lock);
218     if (!rrGuard) {
219         // If we cannot acquire the lock then another thread is
220         // running a range read. Given these are typically long-running,
221         // return without blocking.
222         return 0;
223     }
224
225     // Determine the start and end iterators.
226     OrderedLL::iterator startIt;
227     OrderedLL::iterator endIt;
228     {
229         std::lock_guard<std::mutex> writeGuard(writeLock);
230         if (seqList.empty()) {
231             // Nothing in sequence list - nothing to purge.
232             return 0;
233         }
234         // Determine the {start} and {end} iterator (inclusive). Note
235         // that (at most) one item is added to the seqList before its
236         // sequence number is set (as the seqno comes from Ckpt
237         // manager); if that is the case (such an item is guaranteed
238         // to not be stale), then move end to the previous item
239         // (i.e. we don't consider this "in-flight" item), as long as
240         // there is at least two elements.
241         startIt = seqList.begin();
242         endIt = std::prev(seqList.end());
243         // Need rangeLock for highSeqno & readRange
244         std::lock_guard<SpinLock> rangeGuard(rangeLock);
245         if ((startIt != endIt) && (!endIt->isStale(writeGuard))) {
246             endIt = std::prev(endIt);
247         }
248         readRange = SeqRange(startIt->getBySeqno(), endIt->getBySeqno());
249     }
250
251     // Iterate across all but the last item in the seqList, looking
252     // for stale items.
253     // Note the for() loop terminates one element before endIt - we
254     // actually want an inclusive iteration but as we are comparing
255     // essentially random addresses (and we don't want to 'touch' the
256     // element after endIt), we loop to one before endIt, then handle
257     // endIt explicilty at the end.
258     // Note(2): Iterator is manually incremented outside the for() loop as it
259     // is invalidated when we erase items.
260     size_t purgedCount = 0;
261     bool stale;
262     for (auto it = startIt; it != endIt;) {
263         {
264             std::lock_guard<std::mutex> writeGuard(writeLock);
265             stale = it->isStale(writeGuard);
266         }
267         // Only stale items are purged.
268         if (!stale) {
269             ++it;
270             continue;
271         }
272
273         // Checks pass, remove from list and delete.
274         it = purgeListElem(it);
275         ++purgedCount;
276     }
277     // Handle the last element.
278     {
279         std::lock_guard<std::mutex> writeGuard(writeLock);
280         stale = endIt->isStale(writeGuard);
281     }
282     if (stale) {
283         purgeListElem(endIt);
284         ++purgedCount;
285     }
286
287     // Complete; reset the readRange.
288     {
289         std::lock_guard<SpinLock> lh(rangeLock);
290         readRange.reset();
291     }
292     return purgedCount;
293 }
294
295 uint64_t BasicLinkedList::getNumStaleItems() const {
296     return numStaleItems;
297 }
298
299 size_t BasicLinkedList::getStaleValueBytes() const {
300     return staleSize;
301 }
302
303 size_t BasicLinkedList::getStaleMetadataBytes() const {
304     return staleMetaDataSize;
305 }
306
307 uint64_t BasicLinkedList::getNumDeletedItems() const {
308     std::lock_guard<std::mutex> lckGd(writeLock);
309     return numDeletedItems;
310 }
311
312 uint64_t BasicLinkedList::getNumItems() const {
313     std::lock_guard<std::mutex> lckGd(writeLock);
314     return seqList.size();
315 }
316
317 uint64_t BasicLinkedList::getHighSeqno() const {
318     std::lock_guard<SpinLock> lh(rangeLock);
319     return highSeqno;
320 }
321
322 uint64_t BasicLinkedList::getHighestDedupedSeqno() const {
323     std::lock_guard<std::mutex> lckGd(writeLock);
324     return highestDedupedSeqno;
325 }
326
327 seqno_t BasicLinkedList::getHighestPurgedDeletedSeqno() const {
328     return highestPurgedDeletedSeqno;
329 }
330
331 uint64_t BasicLinkedList::getRangeReadBegin() const {
332     std::lock_guard<SpinLock> lh(rangeLock);
333     return readRange.getBegin();
334 }
335
336 uint64_t BasicLinkedList::getRangeReadEnd() const {
337     std::lock_guard<SpinLock> lh(rangeLock);
338     return readRange.getEnd();
339 }
340
341 void BasicLinkedList::dump() const {
342     std::cerr << *this << std::endl;
343 }
344
345 std::ostream& operator <<(std::ostream& os, const BasicLinkedList& ll) {
346     os << "BasicLinkedList[" << &ll << "] with numItems:" << ll.seqList.size()
347        << " deletedItems:" << ll.numDeletedItems
348        << " staleItems:" << ll.getNumStaleItems()
349        << " highPurgeSeqno:" << ll.getHighestPurgedDeletedSeqno()
350        << " elements:[" << std::endl;
351     for (const auto& val : ll.seqList) {
352         os << "    " << val << std::endl;
353     }
354     os << "]";
355     return os;
356 }
357
358 OrderedLL::iterator BasicLinkedList::purgeListElem(OrderedLL::iterator it) {
359     StoredValue::UniquePtr purged(&*it);
360     {
361         std::lock_guard<std::mutex> lckGd(writeLock);
362         it = seqList.erase(it);
363     }
364
365     /* Update the stats tracking the memory owned by the list */
366     staleSize.fetch_sub(purged->size());
367     staleMetaDataSize.fetch_sub(purged->metaDataSize());
368     st.currentSize.fetch_sub(purged->metaDataSize());
369
370     // Similary for the item counts:
371     --numStaleItems;
372     if (purged->isDeleted()) {
373         --numDeletedItems;
374     }
375
376     if (purged->isDeleted()) {
377         highestPurgedDeletedSeqno = std::max(seqno_t(highestPurgedDeletedSeqno),
378                                              purged->getBySeqno());
379     }
380     return it;
381 }