// A new seg.buf has been allocated.
return
}
- seg.flushing = nil
if err != nil {
// TODO: stall (or return errors from)
// subsequent writes until flushing
offsets := make([]int, 0, len(refs)) // location of segment's data within block
for _, ref := range refs {
seg := ref.fn.segments[ref.idx].(*memSegment)
- if seg.flushing != nil && !sync {
+ if !sync && seg.flushingUnfinished() {
// Let the other flushing goroutine finish. If
// it fails, we'll try again next time.
+ close(done)
return nil
} else {
// In sync mode, we proceed regardless of
defer close(errs)
locator, _, err := dn.fs.PutB(block)
dn.fs.throttle().Release()
- {
- defer func() {
- for _, seg := range segs {
- if seg.flushing == done {
- seg.flushing = nil
- }
- }
- }()
- }
if err != nil {
errs <- err
return
type memSegment struct {
buf []byte
- // If flushing is not nil, then a) buf is being shared by a
- // pruneMemSegments goroutine, and must be copied on write;
- // and b) the flushing channel will close when the goroutine
- // finishes, whether it succeeds or not.
+ // If flushing is not nil and not ready/closed, then a) buf is
+ // being shared by a pruneMemSegments goroutine, and must be
+ // copied on write; and b) the flushing channel will close
+ // when the goroutine finishes, whether it succeeds or not.
flushing <-chan struct{}
}
+func (me *memSegment) flushingUnfinished() bool {
+ if me.flushing == nil {
+ return false
+ }
+ select {
+ case <-me.flushing:
+ me.flushing = nil
+ return false
+ default:
+ return true
+ }
+}
+
func (me *memSegment) Len() int {
return len(me.buf)
}
"os"
"regexp"
"runtime"
+ "runtime/pprof"
"strings"
"sync"
"sync/atomic"
fs.Flush("", true)
}
+func (s *CollectionFSSuite) TestFlushStress(c *check.C) {
+ done := false
+ defer func() { done = true }()
+ time.AfterFunc(10*time.Second, func() {
+ if !done {
+ pprof.Lookup("goroutine").WriteTo(os.Stderr, 1)
+ panic("timeout")
+ }
+ })
+
+ wrote := 0
+ s.kc.onPut = func(p []byte) {
+ s.kc.Lock()
+ s.kc.blocks = map[string][]byte{}
+ wrote++
+ defer c.Logf("wrote block %d, %d bytes", wrote, len(p))
+ s.kc.Unlock()
+ time.Sleep(20 * time.Millisecond)
+ }
+
+ fs, err := (&Collection{}).FileSystem(s.client, s.kc)
+ c.Assert(err, check.IsNil)
+
+ data := make([]byte, 1<<20)
+ for i := 0; i < 3; i++ {
+ dir := fmt.Sprintf("dir%d", i)
+ fs.Mkdir(dir, 0755)
+ for j := 0; j < 200; j++ {
+ data[0] = byte(j)
+ f, err := fs.OpenFile(fmt.Sprintf("%s/file%d", dir, j), os.O_WRONLY|os.O_CREATE, 0)
+ c.Assert(err, check.IsNil)
+ _, err = f.Write(data)
+ c.Assert(err, check.IsNil)
+ f.Close()
+ fs.Flush(dir, false)
+ }
+ _, err := fs.MarshalManifest(".")
+ c.Check(err, check.IsNil)
+ }
+}
+
func (s *CollectionFSSuite) TestFlushShort(c *check.C) {
s.kc.onPut = func([]byte) {
s.kc.Lock()
f.Close()
fs.Flush(dir, false)
}
+ fs.Flush(dir, true)
+ _, err := fs.MarshalManifest(".")
+ c.Check(err, check.IsNil)
}
}