@cryptotaxi247 / kubo / commits / 5474e15e9

blockstore locks return unlocker object now

License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>

Jeromy committed Jan 15, 2016 at 15:14 UTC 5474e15e93e0103b887b09c0c4c96b2c99c5862e

16 files changed +228 -126

blocks/blockstore/blockstore.go

+19 -6

@@ -43,13 +43,13 @@ type GCBlockstore interface {

43 // GCLock locks the blockstore for garbage collection. No operations

44 // that expect to finish with a pin should ocurr simultaneously.

45 // Reading during GC is safe, and requires no lock.

46 - GCLock() func()

46 + GCLock() Unlocker

48 // PinLock locks the blockstore for sequences of puts expected to finish

49 // with a pin (before GC). Multiple put->pin sequences can write through

50 // at the same time, but no GC should not happen simulatenously.

51 // Reading during Pinning is safe, and requires no lock.

52 - PinLock() func()

52 + PinLock() Unlocker

54 // GcRequested returns true if GCLock has been called and is waiting to

55 // take the lock

@@ -198,16 +198,29 @@ func (bs *blockstore) AllKeysChan(ctx context.Context) (<-chan key.Key, error) {

198 return output, nil

199 }

200

201 -func (bs *blockstore) GCLock() func() {

201 +type Unlocker interface {

202 + Unlock()

203 +}

204 +

205 +type unlocker struct {

206 + unlock func()

207 +}

208 +

209 +func (u *unlocker) Unlock() {

210 + u.unlock()

211 + u.unlock = nil // ensure its not called twice

212 +}

213 +

214 +func (bs *blockstore) GCLock() Unlocker {

215 atomic.AddInt32(&bs.gcreq, 1)

216 bs.lk.Lock()

217 atomic.AddInt32(&bs.gcreq, -1)

205 - return bs.lk.Unlock

218 + return &unlocker{bs.lk.Unlock}

219 }

220

208 -func (bs *blockstore) PinLock() func() {

221 +func (bs *blockstore) PinLock() Unlocker {

222 bs.lk.RLock()

210 - return bs.lk.RUnlock

223 + return &unlocker{bs.lk.RUnlock}

224 }

225

226 func (bs *blockstore) GCRequested() bool {

blocks/blockstore/write_cache.go

+2 -2

@@ -59,11 +59,11 @@ func (w *writecache) AllKeysChan(ctx context.Context) (<-chan key.Key, error) {

59 return w.blockstore.AllKeysChan(ctx)

60 }

62 -func (w *writecache) GCLock() func() {

62 +func (w *writecache) GCLock() Unlocker {

63 return w.blockstore.(GCBlockstore).GCLock()

64 }

66 -func (w *writecache) PinLock() func() {

66 +func (w *writecache) PinLock() Unlocker {

67 return w.blockstore.(GCBlockstore).PinLock()

68 }

core/commands/files/files.go

+17 -15

@@ -109,8 +109,7 @@ func statNode(ds dag.DAGService, fsn mfs.FSNode) (*Object, error) {

109 return nil, err

110 }

111

112 - // add to dagserv to ensure its available

113 - k, err := ds.Add(nd)

112 + k, err := nd.Key()

113 if err != nil {

114 return nil, err

115 }

@@ -159,6 +158,11 @@ var FilesCpCmd = &cmds.Command{

158 return

159 }

160

161 + flush, found, _ := req.Option("flush").Bool()

162 + if !found {

163 + flush = true

164 + }

165 +

166 src, err := checkPath(req.Arguments()[0])

167 if err != nil {

168 res.SetError(err, cmds.ErrNormal)

@@ -181,6 +185,14 @@ var FilesCpCmd = &cmds.Command{

185 res.SetError(err, cmds.ErrNormal)

186 return

187 }

188 +

189 + if flush {

190 + err := mfs.FlushPath(node.FilesRoot, dst)

191 + if err != nil {

192 + res.SetError(err, cmds.ErrNormal)

193 + return

194 + }

195 + }

196 },

197 }

198

@@ -501,8 +513,8 @@ Warning:

513

514 create, _, _ := req.Option("create").Bool()

515 trunc, _, _ := req.Option("truncate").Bool()

504 - flush, set, _ := req.Option("flush").Bool()

505 - if !set {

516 + flush, fset, _ := req.Option("flush").Bool()

517 + if !fset {

518 flush = true

519 }

520

@@ -529,14 +541,7 @@ Warning:

541 }

542

543 if flush {

532 - defer func() {

533 - fi.Close()

534 - err := mfs.FlushPath(nd.FilesRoot, path)

535 - if err != nil {

536 - res.SetError(err, cmds.ErrNormal)

537 - return

538 - }

539 - }()

544 + defer fi.Close()

545 } else {

546 defer fi.Sync()

547 }

@@ -653,9 +658,6 @@ are run with the '--flush=false'.

658 return

659 }

660

656 - // take the lock and defer the unlock

657 - defer nd.Blockstore.PinLock()()

658 -

661 path := "/"

662 if len(req.Arguments()) > 0 {

663 path = req.Arguments()[0]

core/commands/pin.go

+1 -2

@@ -54,8 +54,7 @@ on disk.

54 return

55 }

57 - unlock := n.Blockstore.PinLock()

58 - defer unlock()

57 + defer n.Blockstore.PinLock().Unlock()

59 // set recursive flag

60 recursive, found, err := req.Option("recursive").Bool()

core/core.go

-17

@@ -474,23 +474,6 @@ func (n *IpfsNode) loadBootstrapPeers() ([]peer.PeerInfo, error) {

474 func (n *IpfsNode) loadFilesRoot() error {

475 dsk := ds.NewKey("/local/filesroot")

476 pf := func(ctx context.Context, k key.Key) error {

477 - ds := n.Repo.Datastore()

478 - if old, err := ds.Get(dsk); err == nil && old != nil {

479 - _ = n.Pinning.Unpin(ctx, key.Key(old.([]byte)), true)

480 - }

481 - nnd, err := n.DAG.Get(ctx, k)

482 - if err != nil {

483 - return err

484 - }

485 -

486 - if err := n.Pinning.Pin(ctx, nnd, true); err != nil {

487 - return err

488 - }

489 -

490 - if err := n.Pinning.Flush(); err != nil {

491 - return err

492 - }

493 -

477 return n.Repo.Datastore().Put(dsk, []byte(k))

478 }

479

core/coreunix/add.go

+11 -11

@@ -20,6 +20,7 @@ import (

20 "github.com/ipfs/go-ipfs/pin"

21 context "gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context"

23 + bs "github.com/ipfs/go-ipfs/blocks/blockstore"

24 "github.com/ipfs/go-ipfs/commands/files"

25 core "github.com/ipfs/go-ipfs/core"

26 dag "github.com/ipfs/go-ipfs/merkledag"

@@ -100,7 +101,7 @@ type Adder struct {

101 Chunker string

102 root *dag.Node

103 mr *mfs.Root

103 - unlock func()

104 + unlocker bs.Unlocker

105 tempRoot key.Key

106 }

107

@@ -225,8 +226,7 @@ func (adder *Adder) outputDirs(path string, nd *dag.Node) error {

226 // Add builds a merkledag from the a reader, pinning all objects to the local

227 // datastore. Returns a key representing the root node.

228 func Add(n *core.IpfsNode, r io.Reader) (string, error) {

228 - unlock := n.Blockstore.PinLock()

229 - defer unlock()

229 + defer n.Blockstore.PinLock().Unlock()

230

231 fileAdder, err := NewAdder(n.Context(), n, nil)

232 if err != nil {

@@ -247,8 +247,7 @@ func Add(n *core.IpfsNode, r io.Reader) (string, error) {

247

248 // AddR recursively adds files in |path|.

249 func AddR(n *core.IpfsNode, root string) (key string, err error) {

250 - unlock := n.Blockstore.PinLock()

251 - defer unlock()

250 + n.Blockstore.PinLock().Unlock()

251

252 stat, err := os.Lstat(root)

253 if err != nil {

@@ -296,8 +295,7 @@ func AddWrapped(n *core.IpfsNode, r io.Reader, filename string) (string, *dag.No

295 }

296 fileAdder.Wrap = true

297

299 - unlock := n.Blockstore.PinLock()

300 - defer unlock()

298 + defer n.Blockstore.PinLock().Unlock()

299

300 err = fileAdder.addFile(file)

301 if err != nil {

@@ -347,8 +345,10 @@ func (adder *Adder) addNode(node *dag.Node, path string) error {

345

346 // Add the given file while respecting the adder.

347 func (adder *Adder) AddFile(file files.File) error {

350 - adder.unlock = adder.node.Blockstore.PinLock()

351 - defer adder.unlock()

348 + adder.unlocker = adder.node.Blockstore.PinLock()

349 + defer func() {

350 + adder.unlocker.Unlock()

351 + }()

352

353 return adder.addFile(file)

354 }

@@ -434,8 +434,8 @@ func (adder *Adder) maybePauseForGC() error {

434 return err

435 }

436

437 - adder.unlock()

438 - adder.unlock = adder.node.Blockstore.PinLock()

437 + adder.unlocker.Unlock()

438 + adder.unlocker = adder.node.Blockstore.PinLock()

439 }

440 return nil

441 }

mfs/dir.go

+20 -16

@@ -51,17 +51,20 @@ func NewDirectory(ctx context.Context, name string, node *dag.Node, parent child

52 // closeChild updates the child by the given name to the dag node 'nd'

53 // and changes its own dag node

54 -func (d *Directory) closeChild(name string, nd *dag.Node) error {

55 - mynd, err := d.closeChildUpdate(name, nd)

54 +func (d *Directory) closeChild(name string, nd *dag.Node, sync bool) error {

55 + mynd, err := d.closeChildUpdate(name, nd, sync)

56 if err != nil {

57 return err

58 }

60 - return d.parent.closeChild(d.name, mynd)

60 + if sync {

61 + return d.parent.closeChild(d.name, mynd, true)

62 + }

63 + return nil

64 }

66 // closeChildUpdate is the portion of closeChild that needs to be locked around

64 -func (d *Directory) closeChildUpdate(name string, nd *dag.Node) (*dag.Node, error) {

67 +func (d *Directory) closeChildUpdate(name string, nd *dag.Node, sync bool) (*dag.Node, error) {

68 d.lock.Lock()

69 defer d.lock.Unlock()

@@ -70,7 +73,10 @@ func (d *Directory) closeChildUpdate(name string, nd *dag.Node) (*dag.Node, erro

73 return nil, err

74 }

73 - return d.flushCurrentNode()

76 + if sync {

77 + return d.flushCurrentNode()

78 + }

79 + return nil, nil

80 }

82 func (d *Directory) flushCurrentNode() (*dag.Node, error) {

@@ -295,12 +301,15 @@ func (d *Directory) Unlink(name string) error {

301 }

302

303 func (d *Directory) Flush() error {

304 + d.lock.Lock()

305 nd, err := d.flushCurrentNode()

306 if err != nil {

307 + d.lock.Unlock()

308 return err

309 }

310 + d.lock.Unlock()

311

303 - return d.parent.closeChild(d.name, nd)

312 + return d.parent.closeChild(d.name, nd, true)

313 }

314

315 // AddChild adds the node 'nd' under this directory giving it the name 'name'

@@ -335,11 +344,6 @@ func (d *Directory) sync() error {

344 return err

345 }

346

338 - _, err = d.dserv.Add(nd)

339 - if err != nil {

340 - return err

341 - }

342 -

347 err = d.updateChild(name, nd)

348 if err != nil {

349 return err

@@ -352,11 +356,6 @@ func (d *Directory) sync() error {

356 return err

357 }

358

355 - _, err = d.dserv.Add(nd)

356 - if err != nil {

357 - return err

358 - }

359 -

359 err = d.updateChild(name, nd)

360 if err != nil {

361 return err

@@ -385,6 +384,11 @@ func (d *Directory) GetNode() (*dag.Node, error) {

384 return nil, err

385 }

386

387 + _, err = d.dserv.Add(d.node)

388 + if err != nil {

389 + return nil, err

390 + }

391 +

392 return d.node.Copy(), nil

393 }

394

mfs/file.go

+6 -8

@@ -73,7 +73,7 @@ func (fi *File) Close() error {

74 // explicitly stay locked for flushUp call,

75 // it will manage the lock for us

76 - return fi.flushUp()

76 + return fi.flushUp(true)

77 }

78 fi.Unlock()

@@ -82,7 +82,7 @@ func (fi *File) Close() error {

83 // flushUp syncs the file and adds it to the dagservice

84 // it *must* be called with the File's lock taken

85 -func (fi *File) flushUp() error {

85 +func (fi *File) flushUp(fullsync bool) error {

86 nd, err := fi.mod.GetNode()

87 if err != nil {

88 fi.Unlock()

@@ -95,20 +95,18 @@ func (fi *File) flushUp() error {

95 return err

96 }

98 - //name := fi.name

99 - //parent := fi.parent

98 + name := fi.name

99 + parent := fi.parent

100

101 // explicit unlock *only* before closeChild call

102 fi.Unlock()

103 - return nil

104 - //return parent.closeChild(name, nd)

103 + return parent.closeChild(name, nd, fullsync)

104 }

105

106 // Sync flushes the changes in the file to disk

107 func (fi *File) Sync() error {

108 fi.Lock()

110 - defer fi.Unlock()

111 - return fi.mod.Sync()

109 + return fi.flushUp(false)

110 }

111

112 // Seek implements io.Seeker

mfs/mfs_test.go

+69 -14

@@ -576,15 +576,15 @@ func actorRemoveFile(d *Directory) error {

576 return d.Unlink(re.Name)

577 }

578

579 -func actorReadFile(d *Directory) error {

579 +func randomFile(d *Directory) (*File, error) {

580 d, err := randomWalk(d, rand.Intn(6))

581 if err != nil {

582 - return err

582 + return nil, err

583 }

584

585 ents, err := d.List()

586 if err != nil {

587 - return err

587 + return nil, err

588 }

589

590 var files []string

@@ -595,18 +595,61 @@ func actorReadFile(d *Directory) error {

595 }

596

597 if len(files) == 0 {

598 - return nil

598 + return nil, nil

599 }

600

601 fname := files[rand.Intn(len(files))]

602 fsn, err := d.Child(fname)

603 if err != nil {

604 - return err

604 + return nil, err

605 }

606

607 fi, ok := fsn.(*File)

608 if !ok {

609 - return errors.New("file wasnt a file, race?")

609 + return nil, errors.New("file wasnt a file, race?")

610 + }

611 +

612 + return fi, nil

613 +}

614 +

615 +func actorWriteFile(d *Directory) error {

616 + fi, err := randomFile(d)

617 + if err != nil {

618 + return err

619 + }

620 + if fi == nil {

621 + return nil

622 + }

623 +

624 + size := rand.Intn(1024)

625 + buf := make([]byte, size)

626 + randbo.New().Read(buf)

627 +

628 + s, err := fi.Size()

629 + if err != nil {

630 + return err

631 + }

632 +

633 + offset := rand.Int63n(s)

634 +

635 + n, err := fi.WriteAt(buf, offset)

636 + if err != nil {

637 + return err

638 + }

639 + if n != size {

640 + return fmt.Errorf("didnt write enough")

641 + }

642 +

643 + return fi.Close()

644 +}

645 +

646 +func actorReadFile(d *Directory) error {

647 + fi, err := randomFile(d)

648 + if err != nil {

649 + return err

650 + }

651 + if fi == nil {

652 + return nil

653 }

654

655 _, err = fi.Size()

@@ -637,12 +680,7 @@ func testActor(rt *Root, iterations int, errs chan error) {

680 return

681 }

682 case 3:

640 - continue

641 - // randomly deleting things

642 - // doesnt really give us any sort of useful test results.

643 - // you will never have this in a real environment where

644 - // you expect anything productive to happen...

645 - if err := actorRemoveFile(d); err != nil {

683 + if err := actorWriteFile(d); err != nil {

684 errs <- err

685 return

686 }

@@ -698,6 +736,9 @@ func TestFlushing(t *testing.T) {

736 if err := e.AddChild("TEST", nd1); err != nil {

737 t.Fatal(err)

738 }

739 + if err := dir.AddChild("FILE", nd1); err != nil {

740 + t.Fatal(err)

741 + }

742

743 if err := FlushPath(rt, "/a/b/c/TEST"); err != nil {

744 t.Fatal(err)

@@ -711,17 +752,31 @@ func TestFlushing(t *testing.T) {

752 t.Fatal(err)

753 }

754

755 + if err := FlushPath(rt, "/FILE"); err != nil {

756 + t.Fatal(err)

757 + }

758 +

759 rnd, err := dir.GetNode()

760 if err != nil {

761 t.Fatal(err)

762 }

763

764 + fsnode, err := ft.FSNodeFromBytes(rnd.Data)

765 + if err != nil {

766 + t.Fatal(err)

767 + }

768 +

769 + if fsnode.Type != ft.TDirectory {

770 + t.Fatal("root wasnt a directory")

771 + }

772 +

773 rnk, err := rnd.Key()

774 if err != nil {

775 t.Fatal(err)

776 }

777

724 - if rnk.B58String() != "QmWcvrHUFk7LQRrA4WqKjqy7ZyRGFLVagtgNxbEodTEzQ4" {

725 - t.Fatal("dag looks wrong")

778 + exp := "QmWMVyhTuyxUrXX3ynz171jq76yY3PktfY9Bxiph7b9ikr"

779 + if rnk.B58String() != exp {

780 + t.Fatalf("dag looks wrong, expected %s, but got %s", exp, rnk.B58String())

781 }

782 }

mfs/ops.go

+21 -17

@@ -230,11 +230,6 @@ func flushPathRec(d *Directory, parts []string) (*dag.Node, error) {

230 return nil, err

231 }

232

233 - _, err = d.dserv.Add(nd)

234 - if err != nil {

235 - return nil, err

236 - }

237 -

233 return nd, nil

234 }

235

@@ -247,6 +242,7 @@ func flushPathRec(d *Directory, parts []string) (*dag.Node, error) {

242 return nil, err

243 }

244

245 + var ndagnode *dag.Node

246 switch next := next.(type) {

247 case *Directory:

248 nd, err := flushPathRec(next, parts[1:])

@@ -254,25 +250,33 @@ func flushPathRec(d *Directory, parts []string) (*dag.Node, error) {

250 return nil, err

251 }

252

257 - newnode, err := d.node.UpdateNodeLink(parts[0], nd)

258 - if err != nil {

259 - return nil, err

260 - }

261 -

262 - _, err = d.dserv.Add(newnode)

263 - if err != nil {

264 - return nil, err

265 - }

253 + ndagnode = nd

254

267 - d.node = newnode

268 - return newnode, nil

255 case *File:

256 if len(parts) > 1 {

257 return nil, fmt.Errorf("%s is a file, not a directory", parts[0])

258 }

259

274 - return next.GetNode()

260 + child, err := next.GetNode()

261 + if err != nil {

262 + return nil, err

263 + }

264 +

265 + ndagnode = child

266 default:

267 return nil, fmt.Errorf("unrecognized FSNode type: %#v", next)

268 }

269 +

270 + newnode, err := d.node.UpdateNodeLink(parts[0], ndagnode)

271 + if err != nil {

272 + return nil, err

273 + }

274 +

275 + _, err = d.dserv.Add(newnode)

276 + if err != nil {

277 + return nil, err

278 + }

279 +

280 + d.node = newnode

281 + return newnode, nil

282 }

mfs/system.go

+4 -4

@@ -29,7 +29,7 @@ var log = logging.Logger("mfs")

29 var ErrIsDirectory = errors.New("error: is a directory")

31 type childCloser interface {

32 - closeChild(string, *dag.Node) error

32 + closeChild(string, *dag.Node, bool) error

33 }

35 type NodeType int

@@ -115,7 +115,7 @@ func (kr *Root) Flush() error {

115 return err

116 }

117

118 - k, err := kr.dserv.Add(nd)

118 + k, err := nd.Key()

119 if err != nil {

120 return err

121 }

@@ -128,7 +128,7 @@ func (kr *Root) Flush() error {

128

129 // closeChild implements the childCloser interface, and signals to the publisher that

130 // there are changes ready to be published

131 -func (kr *Root) closeChild(name string, nd *dag.Node) error {

131 +func (kr *Root) closeChild(name string, nd *dag.Node, sync bool) error {

132 k, err := kr.dserv.Add(nd)

133 if err != nil {

134 return err

@@ -146,7 +146,7 @@ func (kr *Root) Close() error {

146 return err

147 }

148

149 - k, err := kr.dserv.Add(nd)

149 + k, err := nd.Key()

150 if err != nil {

151 return err

152 }

pin/gc/gc.go

+2 -2

@@ -23,7 +23,7 @@ var log = logging.Logger("gc")

23 // The routine then iterates over every block in the blockstore and

24 // deletes any block that is not found in the marked set.

25 func GC(ctx context.Context, bs bstore.GCBlockstore, pn pin.Pinner) (<-chan key.Key, error) {

26 - unlock := bs.GCLock()

26 + unlocker := bs.GCLock()

28 bsrv := bserv.New(bs, offline.Exchange(bs))

29 ds := dag.NewDAGService(bsrv)

@@ -41,7 +41,7 @@ func GC(ctx context.Context, bs bstore.GCBlockstore, pn pin.Pinner) (<-chan key.

41 output := make(chan key.Key)

42 go func() {

43 defer close(output)

44 - defer unlock()

44 + defer unlocker.Unlock()

45 for {

46 select {

47 case k, ok := <-keychan:

pin/pin.go

+3 -1

@@ -123,6 +123,8 @@ func (p *pinner) Pin(ctx context.Context, node *mdag.Node, recurse bool) error {

123 return nil

124 }

125

126 +var ErrNotPinned = fmt.Errorf("not pinned")

127 +

128 // Unpin a given key

129 func (p *pinner) Unpin(ctx context.Context, k key.Key, recursive bool) error {

130 p.lock.Lock()

@@ -132,7 +134,7 @@ func (p *pinner) Unpin(ctx context.Context, k key.Key, recursive bool) error {

134 return err

135 }

136 if !pinned {

135 - return fmt.Errorf("%s is not pinned", k)

137 + return ErrNotPinned

138 }

139 switch reason {

140 case "recursive":

test/sharness/t0250-files-api.sh

-5

@@ -340,11 +340,6 @@ test_files_api() {

340 ipfs files flush /

341 '

342

343 - test_expect_success "root hash is pinned after flush" '

344 - ipfs pin ls > pins &&

345 - grep $EXP_ROOT_HASH pins || (cat pins && exit 1)

346 - '

347 -

343 # test mv

344 test_expect_success "can mv dir" '

345 ipfs files mv /cats/this/is /cats/

test/sharness/t0251-files-flushing.sh new

+53

@@ -0,0 +1,53 @@

1 +#!/bin/sh

2 +#

4 +# MIT Licensed; see the LICENSE file in this repository.

5 +#

6 +

7 +test_description="test the unix files api flushing"

8 +

9 +. lib/test-lib.sh

10 +

11 +test_init_ipfs

12 +

13 +verify_path_exists() {

14 + # simply running ls on a file should be a good 'check'

15 + ipfs files ls $1

16 +}

17 +

18 +verify_dir_contents() {

19 + dir=$1

20 + shift

21 + rm -f expected

22 + touch expected

23 + for e in $@

24 + do

25 + echo $e >> expected

26 + done

27 +

28 + test_expect_success "can list dir" '

29 + ipfs files ls $dir > output

30 + '

31 +

32 + test_expect_success "dir entries look good" '

33 + test_sort_cmp output expected

34 + '

35 +}

36 +

37 +test_launch_ipfs_daemon

38 +

39 +test_expect_success "can copy a file in" '

40 + HASH=$(echo "foo" | ipfs add -q) &&

41 + ipfs files cp /ipfs/$HASH /file

42 +'

43 +

44 +test_kill_ipfs_daemon

45 +test_launch_ipfs_daemon

46 +

47 +test_expect_success "file is still there" '

48 + verify_path_exists /file

49 +'

50 +

51 +test_kill_ipfs_daemon

52 +

53 +test_done

unixfs/mod/dagmodifier.go

-6

@@ -169,12 +169,6 @@ func (dm *DagModifier) Sync() error {

169 // Number of bytes we're going to write

170 buflen := dm.wrBuf.Len()

171

172 - // Grab key for unpinning after mod operation

173 - _, err := dm.curNode.Key()

174 - if err != nil {

175 - return err

176 - }

177 -

172 // overwrite existing dag nodes

173 thisk, done, err := dm.modifyDag(dm.curNode, dm.writeStart, dm.wrBuf)

174 if err != nil {