-
Notifications
You must be signed in to change notification settings - Fork 0
/
writer.go
156 lines (131 loc) · 3.34 KB
/
writer.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
package chunkbuffer
import (
"github.com/bitantics/chunkbuffer/pile"
"github.com/bitantics/moreio/meters"
"github.com/bitantics/moreio/sharedbuffer"
"io"
)
// writer end of ChunkBuffer
type writer struct {
name string
pile pile.Pile
chunkSize int64
buf sharedbuffer.SharedBuffer
wm meters.WriteMeter
nextErrs chan error
lastChunk pile.ChunkWriter
}
// newWriter creates new writer end of ChunkBuffer
func newWriter(name string, pile pile.Pile, chunkSize int) *writer {
w := writer{
name: name,
pile: pile,
chunkSize: int64(chunkSize),
buf: *sharedbuffer.New(),
nextErrs: make(chan error),
lastChunk: nil,
}
w.wm = *meters.NewWriteMeter(&w.buf)
return &w
}
// written byte count so far
func (w *writer) written() int64 {
return w.wm.Reading()
}
// Write data into the ChunkBuffer, blocking until complete.
// Will block for all chunks which are completed as part of this write. This
// means it will block for the previously incomplete, last chunk, but not for
// the newly incomplete, last chunk.
func (w *writer) Write(p []byte) (n int, err error) {
n, err = w.wm.Write(p)
if err != nil {
return
}
cs := w.chunkSize
old_written := w.written() - int64(n)
next, last := w.nextChunk(old_written), w.written()/cs
errors := w.nextErrs
w.nextErrs = make(chan error)
// wait on this write's last chunk if it's complete
if w.written()%cs == 0 {
last += 1
}
// initialize source readers for complete chunks
srcsN := 0
if last-next > 0 {
srcsN = int(last - next)
}
srcs := make([]io.ReadCloser, srcsN)
for c := next; c < last; c += 1 {
srcs[c-next], err = w.sourceChunk(c)
if err != nil {
return
}
}
// start writing chunks from source readers
for c := next; c < last; c += 1 {
go w.writeChunk(srcs[c-next], c, errors)
}
// consider waiting on previously incomplete chunk
if old_written%cs != 0 {
next -= 1
}
// wait for complete chunk writes and check for errors
for c := next; c < last; c += 1 {
if e := <-errors; e != nil && err == nil {
err = e
}
}
if err != nil {
return
}
// start next incomplete chunk
if w.nextChunk(w.written()) > w.nextChunk(old_written) {
var src io.ReadCloser
src, err = w.sourceChunk(last)
if err == nil {
go w.writeChunk(src, last, w.nextErrs)
}
}
return
}
// Close the writer, flushing the rest of the buffer to a partial chunk
func (w *writer) Close() error {
w.buf.Close()
var err error
if w.written()%w.chunkSize != 0 {
err = <-w.nextErrs
}
w.nextErrs = nil
if err == nil && w.lastChunk != nil {
err = w.lastChunk.SetLast()
}
return err
}
// writeChunk of fixed size (specified in ChunkBuffer) to the pile
func (w *writer) writeChunk(src io.ReadCloser, n int64, err chan<- error) {
ch, writeErr := w.pile.ChunkWriter(w.name, int(n))
if writeErr != nil {
err <- writeErr
return
}
w.lastChunk = ch
_, writeErr = io.Copy(ch, io.LimitReader(src, w.chunkSize))
src.Close()
if closeErr := ch.Close(); writeErr == nil {
writeErr = closeErr
}
err <- writeErr
}
// sourceChunk produces a partial reader for the source buffer
func (w *writer) sourceChunk(n int64) (io.ReadCloser, error) {
if ra, err := w.buf.NewReaderAt(w.chunkSize * n); err != nil {
return nil, err
} else {
return ra, nil
}
}
// nextChunk number
func (w writer) nextChunk(bytes int64) int64 {
return (bytes + w.chunkSize - 1) / w.chunkSize
}