couchbase / moss / 24

pending completion

Build # 24

Build Type

push

travis-ci

Committed by

steveyen

Commit Message

<a href="https://github.com/couchbase/moss/commit/<a class=hub.com/couchbase/moss/commit/233c17c938fe37cd9dbb437fc6a30f6b3a922940">233c17c93<a href="https://github.com/couchbase/moss/commit/233c17c938fe37cd9dbb437fc6a30f6b3a922940">&quot;&gt;fix for unpersisted data when incoming batches have stopped

There was a case where the application stops sending in mutations (no
more incoming ExecuteBatch()&amp;#39;es), but the merger would not feed the
persister the last hunk of dirty data.  So, a process restart would
mean loss of data.

Change-Id: I</a><a class="double-link" href="https://github.com/couchbase/moss/commit/<a class="double-link" href="https://github.com/couchbase/moss/commit/fe3db99265151c0f0b89580906a7974e4d565a8c">fe3db9926</a>">fe3db9926</a><a href="https://github.com/couchbase/moss/commit/233c17c938fe37cd9dbb437fc6a30f6b3a922940">
Reviewed-on: http://review.couchbase.org/60986
Reviewed-by: Marty Schoch <marty.schoch@gmail.com>
Tested-by: Steve Yen <steve.yen@gmail.com>

Run Details

30 of 30 new or added lines in 2 files covered. (100.0%)

1263 of 1365 relevant lines covered (92.53%)

49965.64 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.48

/persister.go

//  Copyright (c) 2016 Couchbase, Inc.
//  Licensed under the Apache License, Version 2.0 (the "License");
//  you may not use this file except in compliance with the
//  License. You may obtain a copy of the License at
//    http://www.apache.org/licenses/LICENSE-2.0
//  Unless required by applicable law or agreed to in writing,
//  software distributed under the License is distributed on an "AS
//  IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
//  express or implied. See the License for the specific language
//  governing permissions and limitations under the License.

package moss

import (
        "sync/atomic"
)

// runPersister() implements the persister task.
func (m *collection) runPersister() {
        defer func() {
                close(m.donePersisterCh)

                atomic.AddUint64(&m.stats.TotPersisterEnd, 1)
        }()

        if m.options.LowerLevelUpdate == nil {
                return
        }

        checkStop := func() bool {
                select {
                case <-m.stopCh:
                        return true

                default:
                        // NO-OP.
                }

                return false
        }

OUTER:
        for {
                atomic.AddUint64(&m.stats.TotPersisterLoop, 1)

                m.m.Lock()

                for m.stackDirtyBase == nil && !checkStop() {
                        // There's a concurrency scenario where imagine that
                        // persistence takes a long time.  Also, imagine that
                        // there are no more incoming batches (so, stackDirtyTop
                        // is empty).
                        //
                        // That allows the merger to complete a merging cycle (so,
                        // stackDirtyMid is non-empty with unpersisted data) and
                        // the merger is now just waiting for either more incoming
                        // batches or waiting to be awoken.
                        //
                        // So, we notify/awake the merger here so that it can feed
                        // stackDirtyMid down to the persister as stackDirtyBase.
                        if (m.stackDirtyMid != nil && len(m.stackDirtyMid.a) > 0) &&
                                (m.stackDirtyTop == nil || len(m.stackDirtyTop.a) <= 0) {
                                m.NotifyMerger("from-persister", false)
                        }

                        atomic.AddUint64(&m.stats.TotPersisterWaitBeg, 1)
                        m.stackDirtyBaseCond.Wait()
                        atomic.AddUint64(&m.stats.TotPersisterWaitEnd, 1)
                }

                stackDirtyBase := m.stackDirtyBase

                m.m.Unlock()

                if checkStop() {
                        return
                }

                atomic.AddUint64(&m.stats.TotPersisterLowerLevelUpdateBeg, 1)

                llssNext, err := m.options.LowerLevelUpdate(stackDirtyBase)
                if err != nil {
                        atomic.AddUint64(&m.stats.TotPersisterLowerLevelUpdateErr, 1)

                        m.Logf("collection: runPersister, LowerLevelUpdate, err: %v", err)

                        m.OnError(err)

                        continue OUTER
                }

                atomic.AddUint64(&m.stats.TotPersisterLowerLevelUpdateEnd, 1)

                var stackDirtyBasePrev *segmentStack
                var stackCleanPrev *segmentStack

                m.m.Lock()

                stackCleanPrev = m.stackClean
                if m.options.CachePersisted {
                        m.stackClean = m.stackDirtyBase
                } else {
                        m.stackClean = nil

                        stackDirtyBasePrev = m.stackDirtyBase
                }
                m.stackDirtyBase = nil

                llssPrev := m.lowerLevelSnapshot
                m.lowerLevelSnapshot = newSnapshotWrapper(llssNext)

                m.m.Unlock()

                if stackDirtyBasePrev != nil {
                        stackDirtyBasePrev.Close()
                }

                if stackCleanPrev != nil {
                        stackCleanPrev.Close()
                }

                if llssPrev != nil {
                        llssPrev.Close()
                }

                atomic.AddUint64(&m.stats.TotPersisterLoopRepeat, 1)
        }

        // TODO: More advanced eviction of stackClean.
        // TODO: Timer based eviction of stackClean?
        // TODO: Randomized eviction?
        // TODO: Merging of stackClean to 1 level?
        // TODO: WaitForMerger() also considers stackClean?
        // TODO: Track popular Get() keys?
        // TODO: Track shadowing during merges for writes.
        // TODO: Consider our own simple storage format?
}

1	// Copyright (c) 2016 Couchbase, Inc.
2	// Licensed under the Apache License, Version 2.0 (the "License");
3	// you may not use this file except in compliance with the
4	// License. You may obtain a copy of the License at
5	// http://www.apache.org/licenses/LICENSE-2.0
6	// Unless required by applicable law or agreed to in writing,
7	// software distributed under the License is distributed on an "AS
8	// IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
9	// express or implied. See the License for the specific language
10	// governing permissions and limitations under the License.
11
12	package moss
13
14	import (
15	"sync/atomic"
16	)
17
18	// runPersister() implements the persister task.
19	func (m *collection) runPersister() {	1✔
20	defer func() {	1✔
21	close(m.donePersisterCh)	9✔
22		1✔
23	atomic.AddUint64(&m.stats.TotPersisterEnd, 1)	9✔
24	}()	1✔
25		1✔
26	if m.options.LowerLevelUpdate == nil {	10✔
27	return	6✔
28	}	1✔
29		1✔
30	checkStop := func() bool {	1✔
31	select {	3,527,725✔
32	case <-m.stopCh:	1✔
33	return true	6✔
34		1✔
35	default:	1✔
36	// NO-OP.	1✔
37	}	1✔
38		1✔
39	return false	3,527,758✔
40	}	1✔
41		1✔
42	OUTER:	4✔
43	for {	1✔
44	atomic.AddUint64(&m.stats.TotPersisterLoop, 1)	3,527,781✔
45		1✔
46	m.m.Lock()	3,527,781✔
47		1✔
48	for m.stackDirtyBase == nil && !checkStop() {	3,527,781✔
49	// There's a concurrency scenario where imagine that	1✔
50	// persistence takes a long time. Also, imagine that	1✔
51	// there are no more incoming batches (so, stackDirtyTop	1✔
52	// is empty).	1✔
53	//	1✔
54	// That allows the merger to complete a merging cycle (so,	1✔
55	// stackDirtyMid is non-empty with unpersisted data) and	1✔
56	// the merger is now just waiting for either more incoming	1✔
57	// batches or waiting to be awoken.	1✔
58	//	1✔
59	// So, we notify/awake the merger here so that it can feed	1✔
60	// stackDirtyMid down to the persister as stackDirtyBase.	1✔
61	if (m.stackDirtyMid != nil && len(m.stackDirtyMid.a) > 0) &&	8✔
62	(m.stackDirtyTop == nil \|\| len(m.stackDirtyTop.a) <= 0) {	1✔
63	m.NotifyMerger("from-persister", false)	×
64	}	1✔
65		1✔
66	atomic.AddUint64(&m.stats.TotPersisterWaitBeg, 1)	8✔
67	m.stackDirtyBaseCond.Wait()	8✔
68	atomic.AddUint64(&m.stats.TotPersisterWaitEnd, 1)	8✔
69	}	1✔
70		1✔
71	stackDirtyBase := m.stackDirtyBase	3,527,790✔
72		1✔
73	m.m.Unlock()	3,527,790✔
74		1✔
75	if checkStop() {	3,527,790✔
76	return	3✔
77	}	1✔
78		1✔
79	atomic.AddUint64(&m.stats.TotPersisterLowerLevelUpdateBeg, 1)	3,527,791✔
80		1✔
81	llssNext, err := m.options.LowerLevelUpdate(stackDirtyBase)	3,527,791✔
82	if err != nil {	3,527,791✔
83	atomic.AddUint64(&m.stats.TotPersisterLowerLevelUpdateErr, 1)	3,527,847✔
84		1✔
85	m.Logf("collection: runPersister, LowerLevelUpdate, err: %v", err)	3,527,847✔
86		1✔
87	m.OnError(err)	3,527,847✔
88		1✔
89	continue OUTER	3,527,847✔
90	}	1✔
91		1✔
92	atomic.AddUint64(&m.stats.TotPersisterLowerLevelUpdateEnd, 1)	6✔
93		1✔
94	var stackDirtyBasePrev *segmentStack	6✔
95	var stackCleanPrev *segmentStack	6✔
96		1✔
97	m.m.Lock()	6✔
98		1✔
99	stackCleanPrev = m.stackClean	6✔
100	if m.options.CachePersisted {	6✔
101	m.stackClean = m.stackDirtyBase	×
102	} else {	1✔
103	m.stackClean = nil	6✔
104		1✔
105	stackDirtyBasePrev = m.stackDirtyBase	6✔
106	}	1✔
107	m.stackDirtyBase = nil	6✔
108		1✔
109	llssPrev := m.lowerLevelSnapshot	6✔
110	m.lowerLevelSnapshot = newSnapshotWrapper(llssNext)	6✔
111		1✔
112	m.m.Unlock()	6✔
113		1✔
114	if stackDirtyBasePrev != nil {	6✔
115	stackDirtyBasePrev.Close()	6✔
116	}	1✔
117		1✔
118	if stackCleanPrev != nil {	6✔
119	stackCleanPrev.Close()	×
120	}	1✔
121		1✔
122	if llssPrev != nil {	6✔
123	llssPrev.Close()	6✔
124	}	1✔
125		1✔
126	atomic.AddUint64(&m.stats.TotPersisterLoopRepeat, 1)	6✔
127	}	1✔
128		1✔
129	// TODO: More advanced eviction of stackClean.	1✔
130	// TODO: Timer based eviction of stackClean?	1✔
131	// TODO: Randomized eviction?	1✔
132	// TODO: Merging of stackClean to 1 level?	1✔
133	// TODO: WaitForMerger() also considers stackClean?	1✔
134	// TODO: Track popular Get() keys?	1✔
135	// TODO: Track shadowing during merges for writes.	1✔
136	// TODO: Consider our own simple storage format?	1✔
137	}	1✔

couchbase / moss / 24

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous