couchbase / moss / 234

pending completion

Build # 234

Build Type

push

travis-ci

Committed by

abhinavdangeti

Commit Message

MB-23560: Stats to track ref-counted old files

New stats will be available in store.Stats() API:
+ num_files             ... number of files in directory
+ num_files_open        ... number of tracked open files
+ files                 ... map interface of tracked files

Change-Id: Ifd513f752
Reviewed-on: http://review.couchbase.org/75928
Reviewed-by: Steve Yen <steve.yen@gmail.com>
Tested-by: abhinav dangeti <abhinav@couchbase.com>

Run Details

46 of 46 new or added lines in 3 files covered. (100.0%)

2897 of 3507 relevant lines covered (82.61%)

12052.93 hits per line

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

86.71

/file.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 (
        "io"
        "os"
        "sync"
)

// An InitCloser holds onto an io.Closer, and is used for chaining
// io.Closer's.  That is, we often want the closing of one resource to
// close related resources.
type InitCloser interface {
        InitCloser(io.Closer) error
}

// The File interface is implemented by os.File.  App specific
// implementations may add concurrency, caching, stats, fuzzing, etc.
type File interface {
        io.ReaderAt
        io.WriterAt
        io.Closer
        Stat() (os.FileInfo, error)
        Sync() error
        Truncate(size int64) error
}

// The OpenFile func signature is similar to os.OpenFile().
type OpenFile func(name string, flag int, perm os.FileMode) (File, error)

// FileRef provides a ref-counting wrapper around a File.
type FileRef struct {
        file File
        m    sync.Mutex // Protects the fields that follow.
        refs int

        beforeCloseCallbacks []func() // Optional callbacks invoked before final close.
        afterCloseCallbacks  []func() // Optional callbacks invoked after final close.
}

type ioResult struct {
        kind string // Kind of io attempted.
        want int    // Num bytes expected to be written or read.
        got  int    // Num bytes actually written or read.
        err  error
}

// --------------------------------------------------------

// OnBeforeClose registers event callback func's that are invoked before the
// file is closed.
func (r *FileRef) OnBeforeClose(cb func()) {
        r.m.Lock()
        r.beforeCloseCallbacks = append(r.beforeCloseCallbacks, cb)
        r.m.Unlock()
}

// OnAfterClose registers event callback func's that are invoked after the
// file is closed.
func (r *FileRef) OnAfterClose(cb func()) {
        r.m.Lock()
        r.afterCloseCallbacks = append(r.afterCloseCallbacks, cb)
        r.m.Unlock()
}

// AddRef increases the ref-count on the file ref.
func (r *FileRef) AddRef() File {
        if r == nil {
                return nil
        }

        r.m.Lock()
        r.refs++
        file := r.file
        r.m.Unlock()

        return file
}

// DecRef decreases the ref-count on the file ref, and closing the
// underlying file when the ref-count reaches zero.
func (r *FileRef) DecRef() (err error) {
        if r == nil {
                return nil
        }

        r.m.Lock()

        r.refs--
        if r.refs <= 0 {
                for _, cb := range r.beforeCloseCallbacks {
                        cb()
                }
                r.beforeCloseCallbacks = nil

                err = r.file.Close()

                for _, cb := range r.afterCloseCallbacks {
                        cb()
                }
                r.afterCloseCallbacks = nil

                r.file = nil
        }

        r.m.Unlock()

        return err
}

// Close allows the FileRef to implement the io.Closer interface.  It actually
// just performs what should be the final DecRef() call which takes the
// reference count to 0.  Once 0, it allows the file to actually be closed.
func (r *FileRef) Close() error {
        return r.DecRef()
}

// FetchRefCount fetches the ref-count on the file ref.
func (r *FileRef) FetchRefCount() int {
        if r == nil {
                return 0
        }

        r.m.Lock()
        ref := r.refs
        r.m.Unlock()

        return ref
}

// --------------------------------------------------------

// OsFile interface allows conversion from a File to an os.File.
type OsFile interface {
        OsFile() *os.File
}

// ToOsFile provides the underlying os.File for a File, if available.
func ToOsFile(f File) *os.File {
        if osFile, ok := f.(*os.File); ok {
                return osFile
        }
        if osFile2, ok := f.(OsFile); ok {
                return osFile2.OsFile()
        }
        return nil
}

// --------------------------------------------------------

type bufferedSectionWriter struct {
        err error
        w   io.WriterAt
        beg int64 // Start position where we started writing in file.
        cur int64 // Current write-at position in file.
        max int64 // When > 0, max number of bytes we can write.
        buf []byte
        n   int

        stopCh chan struct{}
        doneCh chan struct{}
        reqCh  chan ioBuf
        resCh  chan ioBuf
}

type ioBuf struct {
        buf []byte
        pos int64
        err error
}

// newBufferedSectionWriter converts incoming Write() requests into
// buffered, asynchronous WriteAt()'s in a section of a file.
func newBufferedSectionWriter(w io.WriterAt, begPos, maxBytes int64,
        bufSize int) *bufferedSectionWriter {
        stopCh := make(chan struct{})
        doneCh := make(chan struct{})
        reqCh := make(chan ioBuf)
        resCh := make(chan ioBuf)

        go func() {
                defer close(doneCh)
                defer close(resCh)

                buf := make([]byte, bufSize)
                var pos int64
                var err error

                for {
                        select {
                        case <-stopCh:
                                return
                        case resCh <- ioBuf{buf: buf, pos: pos, err: err}:
                        }

                        req, ok := <-reqCh
                        if ok {
                                buf, pos = req.buf, req.pos
                                if len(buf) > 0 {
                                        _, err = w.WriteAt(buf, pos)
                                }
                        }
                }
        }()

        return &bufferedSectionWriter{
                w:   w,
                beg: begPos,
                cur: begPos,
                max: maxBytes,
                buf: make([]byte, bufSize),

                stopCh: stopCh,
                doneCh: doneCh,
                reqCh:  reqCh,
                resCh:  resCh,
        }
}

// Offset returns the byte offset into the file where the
// bufferedSectionWriter is currently logically positioned.
func (b *bufferedSectionWriter) Offset() int64 { return b.cur + int64(b.n) }

// Written returns the logical number of bytes written to this
// bufferedSectionWriter; or, the sum of bytes to Write() calls.
func (b *bufferedSectionWriter) Written() int64 { return b.Offset() - b.beg }

func (b *bufferedSectionWriter) Write(p []byte) (nn int, err error) {
        if b.max > 0 && b.Written()+int64(len(p)) > b.max {
                return 0, io.ErrShortBuffer // Would go over b.max.
        }
        for len(p) > 0 && b.err == nil {
                n := copy(b.buf[b.n:], p)
                b.n += n
                nn += n
                if n < len(p) {
                        b.err = b.Flush()
                }
                p = p[n:]
        }
        return nn, b.err
}

func (b *bufferedSectionWriter) Flush() error {
        if b.err != nil {
                return b.err
        }
        if b.n <= 0 {
                return nil
        }

        prevWrite := <-b.resCh
        b.err = prevWrite.err
        if b.err != nil {
                return b.err
        }

        b.reqCh <- ioBuf{buf: b.buf[0:b.n], pos: b.cur}

        b.cur += int64(b.n)
        b.buf = prevWrite.buf[:]
        b.n = 0

        return nil
}

func (b *bufferedSectionWriter) Stop() error {
        if b.stopCh != nil {
                close(b.stopCh)
                close(b.reqCh)
                <-b.doneCh
                b.stopCh = nil
        }
        return b.err
}

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	"io"
16	"os"
17	"sync"
18	)
19
20	// An InitCloser holds onto an io.Closer, and is used for chaining
21	// io.Closer's. That is, we often want the closing of one resource to
22	// close related resources.
23	type InitCloser interface {
24	InitCloser(io.Closer) error
25	}
26
27	// The File interface is implemented by os.File. App specific
28	// implementations may add concurrency, caching, stats, fuzzing, etc.
29	type File interface {
30	io.ReaderAt
31	io.WriterAt
32	io.Closer
33	Stat() (os.FileInfo, error)
34	Sync() error
35	Truncate(size int64) error
36	}
37
38	// The OpenFile func signature is similar to os.OpenFile().
39	type OpenFile func(name string, flag int, perm os.FileMode) (File, error)
40
41	// FileRef provides a ref-counting wrapper around a File.
42	type FileRef struct {
43	file File
44	m sync.Mutex // Protects the fields that follow.
45	refs int
46
47	beforeCloseCallbacks []func() // Optional callbacks invoked before final close.
48	afterCloseCallbacks []func() // Optional callbacks invoked after final close.
49	}
50
51	type ioResult struct {
52	kind string // Kind of io attempted.
53	want int // Num bytes expected to be written or read.
54	got int // Num bytes actually written or read.
55	err error
56	}
57
58	// --------------------------------------------------------
59
60	// OnBeforeClose registers event callback func's that are invoked before the
61	// file is closed.
62	func (r *FileRef) OnBeforeClose(cb func()) {	1✔
63	r.m.Lock()	1✔
64	r.beforeCloseCallbacks = append(r.beforeCloseCallbacks, cb)	1✔
65	r.m.Unlock()	1✔
66	}	1✔
67
68	// OnAfterClose registers event callback func's that are invoked after the
69	// file is closed.
70	func (r *FileRef) OnAfterClose(cb func()) {	70✔
71	r.m.Lock()	70✔
72	r.afterCloseCallbacks = append(r.afterCloseCallbacks, cb)	70✔
73	r.m.Unlock()	70✔
74	}	70✔
75
76	// AddRef increases the ref-count on the file ref.
77	func (r *FileRef) AddRef() File {	209✔
78	if r == nil {	209✔
79	return nil	×
80	}	×
81
82	r.m.Lock()	209✔
83	r.refs++	209✔
84	file := r.file	209✔
85	r.m.Unlock()	209✔
86		209✔
87	return file	209✔
88	}
89
90	// DecRef decreases the ref-count on the file ref, and closing the
91	// underlying file when the ref-count reaches zero.
92	func (r *FileRef) DecRef() (err error) {	265✔
93	if r == nil {	266✔
94	return nil	1✔
95	}	1✔
96
97	r.m.Lock()	264✔
98		264✔
99	r.refs--	264✔
100	if r.refs <= 0 {	354✔
101	for _, cb := range r.beforeCloseCallbacks {	91✔
102	cb()	1✔
103	}	1✔
104	r.beforeCloseCallbacks = nil	90✔
105		90✔
106	err = r.file.Close()	90✔
107		90✔
108	for _, cb := range r.afterCloseCallbacks {	160✔
109	cb()	70✔
110	}	70✔
111	r.afterCloseCallbacks = nil	90✔
112		90✔
113	r.file = nil	90✔
114	}
115
116	r.m.Unlock()	264✔
117		264✔
118	return err	264✔
119	}
120
121	// Close allows the FileRef to implement the io.Closer interface. It actually
122	// just performs what should be the final DecRef() call which takes the
123	// reference count to 0. Once 0, it allows the file to actually be closed.
124	func (r *FileRef) Close() error {	×
125	return r.DecRef()	×
126	}	×
127
128	// FetchRefCount fetches the ref-count on the file ref.
129	func (r *FileRef) FetchRefCount() int {	4✔
130	if r == nil {	4✔
131	return 0	×
132	}	×
133
134	r.m.Lock()	4✔
135	ref := r.refs	4✔
136	r.m.Unlock()	4✔
137		4✔
138	return ref	4✔
139	}
140
141	// --------------------------------------------------------
142
143	// OsFile interface allows conversion from a File to an os.File.
144	type OsFile interface {
145	OsFile() *os.File
146	}
147
148	// ToOsFile provides the underlying os.File for a File, if available.
149	func ToOsFile(f File) *os.File {	179✔
150	if osFile, ok := f.(*os.File); ok {	358✔
151	return osFile	179✔
152	}	179✔
153	if osFile2, ok := f.(OsFile); ok {	×
154	return osFile2.OsFile()	×
155	}	×
156	return nil	×
157	}
158
159	// --------------------------------------------------------
160
161	type bufferedSectionWriter struct {
162	err error
163	w io.WriterAt
164	beg int64 // Start position where we started writing in file.
165	cur int64 // Current write-at position in file.
166	max int64 // When > 0, max number of bytes we can write.
167	buf []byte
168	n int
169
170	stopCh chan struct{}
171	doneCh chan struct{}
172	reqCh chan ioBuf
173	resCh chan ioBuf
174	}
175
176	type ioBuf struct {
177	buf []byte
178	pos int64
179	err error
180	}
181
182	// newBufferedSectionWriter converts incoming Write() requests into
183	// buffered, asynchronous WriteAt()'s in a section of a file.
184	func newBufferedSectionWriter(w io.WriterAt, begPos, maxBytes int64,
185	bufSize int) *bufferedSectionWriter {	150✔
186	stopCh := make(chan struct{})	150✔
187	doneCh := make(chan struct{})	150✔
188	reqCh := make(chan ioBuf)	150✔
189	resCh := make(chan ioBuf)	150✔
190		150✔
191	go func() {	300✔
192	defer close(doneCh)	150✔
193	defer close(resCh)	150✔
194		150✔
195	buf := make([]byte, bufSize)	150✔
196	var pos int64	150✔
197	var err error	150✔
198		150✔
199	for {	440✔
200	select {	290✔
201	case <-stopCh:	150✔
202	return	150✔
203	case resCh <- ioBuf{buf: buf, pos: pos, err: err}:	140✔
204	}
205
206	req, ok := <-reqCh	140✔
207	if ok {	280✔
208	buf, pos = req.buf, req.pos	140✔
209	if len(buf) > 0 {	280✔
210	_, err = w.WriteAt(buf, pos)	140✔
211	}	140✔
212	}
213	}
214	}()
215
216	return &bufferedSectionWriter{	150✔
217	w: w,	150✔
218	beg: begPos,	150✔
219	cur: begPos,	150✔
220	max: maxBytes,	150✔
221	buf: make([]byte, bufSize),	150✔
222		150✔
223	stopCh: stopCh,	150✔
224	doneCh: doneCh,	150✔
225	reqCh: reqCh,	150✔
226	resCh: resCh,	150✔
227	}	150✔
228	}
229
230	// Offset returns the byte offset into the file where the
231	// bufferedSectionWriter is currently logically positioned.
232	func (b *bufferedSectionWriter) Offset() int64 { return b.cur + int64(b.n) }	56,808✔
233
234	// Written returns the logical number of bytes written to this
235	// bufferedSectionWriter; or, the sum of bytes to Write() calls.
236	func (b *bufferedSectionWriter) Written() int64 { return b.Offset() - b.beg }	56,658✔
237
238	func (b *bufferedSectionWriter) Write(p []byte) (nn int, err error) {	169,974✔
239	if b.max > 0 && b.Written()+int64(len(p)) > b.max {	169,974✔
240	return 0, io.ErrShortBuffer // Would go over b.max.	×
241	}	×
242	for len(p) > 0 && b.err == nil {	339,798✔
243	n := copy(b.buf[b.n:], p)	169,824✔
244	b.n += n	169,824✔
245	nn += n	169,824✔
246	if n < len(p) {	169,824✔
247	b.err = b.Flush()	×
248	}	×
249	p = p[n:]	169,824✔
250	}
251	return nn, b.err	169,974✔
252	}
253
254	func (b *bufferedSectionWriter) Flush() error {	150✔
255	if b.err != nil {	150✔
256	return b.err	×
257	}	×
258	if b.n <= 0 {	160✔
259	return nil	10✔
260	}	10✔
261
262	prevWrite := <-b.resCh	140✔
263	b.err = prevWrite.err	140✔
264	if b.err != nil {	140✔
265	return b.err	×
266	}	×
267
268	b.reqCh <- ioBuf{buf: b.buf[0:b.n], pos: b.cur}	140✔
269		140✔
270	b.cur += int64(b.n)	140✔
271	b.buf = prevWrite.buf[:]	140✔
272	b.n = 0	140✔
273		140✔
274	return nil	140✔
275	}
276
277	func (b *bufferedSectionWriter) Stop() error {	150✔
278	if b.stopCh != nil {	300✔
279	close(b.stopCh)	150✔
280	close(b.reqCh)	150✔
281	<-b.doneCh	150✔
282	b.stopCh = nil	150✔
283	}	150✔
284	return b.err	150✔
285	}

couchbase / moss / 234

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