Check for multiple pinned blocks in a single pass.

Provide a new method, Pinner.CheckIfPinned(), which will check if
any of the arguments are pinned.  Previously IsPinned would need to be
called once for each block.  The new method will speed up the checking
of multiple pinned blocks from O(p*n) to O(p) (where p is the number
of pinned blocks and n is the number of blocks to be check)

Use the new method in "block rm".

License: MIT
Signed-off-by: Kevin Atkinson <k@kevina.org>

core/commands/block.go

@@ -228,7 +228,10 @@ It takes a list of base58 encoded multihashs to remove.
 			if ignorePins {
 				pinning = nil
 			}
-			rmBlocks(n.Blockstore, pinning, outChan, keys)
+			err := rmBlocks(n.Blockstore, pinning, outChan, keys)
+			if err != nil {
+				outChan <- &RemovedBlock{Error: err.Error()}
+			}
 		}()
 		return
 	},
@@ -246,7 +249,10 @@ It takes a list of base58 encoded multihashs to remove.
 		someFailed := false
 		for out := range outChan {
 			o := out.(*RemovedBlock)
-			if o.Error != "" {
+			if o.Hash == "" && o.Error != "" {
+				res.SetError(fmt.Errorf("aborted: %s", o.Error), cmds.ErrNormal)
+				return
+			} else if o.Error != "" {
 				someFailed = true
 				fmt.Fprintf(res.Stderr(), "cannot remove %s: %s\n", o.Hash, o.Error)
 			} else {
@@ -261,12 +267,12 @@ It takes a list of base58 encoded multihashs to remove.
 }
 
 type RemovedBlock struct {
-	Hash string
+	Hash  string `json:",omitempty"`
 	Error string `json:",omitempty"`
 }
 
 // pins may be nil
-func rmBlocks(blocks bs.GCBlockstore, pins pin.Pinner, out chan<- interface{}, keys []key.Key) {
+func rmBlocks(blocks bs.GCBlockstore, pins pin.Pinner, out chan<- interface{}, keys []key.Key) error {
 	var unlocker bs.Unlocker
 	defer func() {
 		if unlocker != nil {
@@ -278,33 +284,44 @@ func rmBlocks(blocks bs.GCBlockstore, pins pin.Pinner, out chan<- interface{}, k
 		// Need to make sure that some operation that is
 		// finishing with a pin is ocurr simultaneously.
 		unlocker = blocks.GCLock()
-		stillOkay = checkIfPinned(pins, keys, out)
+		var err error
+		stillOkay, err = checkIfPinned(pins, keys, out)
+		if err != nil {
+			return fmt.Errorf("pin check failed: %s", err)
+		}
 	}
 	for _, k := range stillOkay {
 		err := blocks.DeleteBlock(k)
 		if err != nil {
-			out <- &RemovedBlock{ Hash: k.String(), Error: err.Error()}
+			out <- &RemovedBlock{Hash: k.String(), Error: err.Error()}
 		} else {
-			out <- &RemovedBlock{ Hash: k.String() }
+			out <- &RemovedBlock{Hash: k.String()}
 		}
 	}
+	return nil
 }
 
-func checkIfPinned(pins pin.Pinner, keys []key.Key, out chan<- interface{}) []key.Key {
+func checkIfPinned(pins pin.Pinner, keys []key.Key, out chan<- interface{}) ([]key.Key, error) {
 	stillOkay := make([]key.Key, 0, len(keys))
-	for _, k := range keys {
-		reason, pinned, err := pins.IsPinned(k)
-		if err != nil {
-			out <- &RemovedBlock {
-				Hash: k.String(),
-				Error: fmt.Sprintf("pin check failed %s", err.Error()) }
-		} else if pinned {
-			out <- &RemovedBlock {
-				Hash: k.String(),
-				Error: fmt.Sprintf("pinned via %s", reason) }
-		} else {
-			stillOkay = append(stillOkay, k)
+	res, err := pins.CheckIfPinned(keys...)
+	if err != nil {
+		return nil, err
+	}
+	for _, r := range res {
+		switch r.Mode {
+		case pin.NotPinned:
+			stillOkay = append(stillOkay, r.Key)
+		case pin.Indirect:
+			out <- &RemovedBlock{
+				Hash:  r.Key.String(),
+				Error: fmt.Sprintf("pinned via %s", r.Via)}
+		default:
+			modeStr, _ := pin.PinModeToString(r.Mode)
+			out <- &RemovedBlock{
+				Hash:  r.Key.String(),
+				Error: fmt.Sprintf("pinned: %s", modeStr)}
+
 		}
 	}
-	return stillOkay
+	return stillOkay, nil
 }

pin/pin.go

@@ -75,6 +75,10 @@ type Pinner interface {
 	Pin(context.Context, *mdag.Node, bool) error
 	Unpin(context.Context, key.Key, bool) error
 
+	// Check if a set of keys are pinned, more efficient than
+	// calling IsPinned for each key
+	CheckIfPinned(keys ...key.Key) ([]Pinned, error)
+
 	// PinWithMode is for manually editing the pin structure. Use with
 	// care! If used improperly, garbage collection may not be
 	// successful.
@@ -90,6 +94,12 @@ type Pinner interface {
 	InternalPins() []key.Key
 }
 
+type Pinned struct {
+	Key  key.Key
+	Mode PinMode
+	Via  key.Key
+}
+
 // pinner implements the Pinner interface
 type pinner struct {
 	lock       sync.RWMutex
@@ -255,6 +265,70 @@ func (p *pinner) isPinnedWithType(k key.Key, mode PinMode) (string, bool, error)
 	return "", false, nil
 }
 
+func (p *pinner) CheckIfPinned(keys ...key.Key) ([]Pinned, error) {
+	p.lock.RLock()
+	defer p.lock.RUnlock()
+	pinned := make([]Pinned, 0, len(keys))
+	toCheck := make(map[key.Key]struct{})
+
+	// First check for non-Indirect pins directly
+	for _, k := range keys {
+		if p.recursePin.HasKey(k) {
+			pinned = append(pinned, Pinned{Key: k, Mode: Recursive})
+		} else if p.directPin.HasKey(k) {
+			pinned = append(pinned, Pinned{Key: k, Mode: Direct})
+		} else if p.isInternalPin(k) {
+			pinned = append(pinned, Pinned{Key: k, Mode: Internal})
+		} else {
+			toCheck[k] = struct{}{}
+		}
+	}
+
+	// Now walk all recursive pins to check for indirect pins
+	var checkChildren func(key.Key, key.Key) error
+	checkChildren = func(rk key.Key, parentKey key.Key) error {
+		parent, err := p.dserv.Get(context.Background(), parentKey)
+		if err != nil {
+			return err
+		}
+		for _, lnk := range parent.Links {
+			k := key.Key(lnk.Hash)
+
+			if _, found := toCheck[k]; found {
+				pinned = append(pinned,
+					Pinned{Key: k, Mode: Indirect, Via: rk})
+				delete(toCheck, k)
+			}
+
+			err := checkChildren(rk, k)
+			if err != nil {
+				return err
+			}
+
+			if len(toCheck) == 0 {
+				return nil
+			}
+		}
+		return nil
+	}
+	for _, rk := range p.recursePin.GetKeys() {
+		err := checkChildren(rk, rk)
+		if err != nil {
+			return nil, err
+		}
+		if len(toCheck) == 0 {
+			break
+		}
+	}
+
+	// Anything left in toCheck is not pinned
+	for k, _ := range toCheck {
+		pinned = append(pinned, Pinned{Key: k, Mode: NotPinned})
+	}
+
+	return pinned, nil
+}
+
 func (p *pinner) RemovePinWithMode(key key.Key, mode PinMode) {
 	p.lock.Lock()
 	defer p.lock.Unlock()

test/sharness/t0050-block.sh

@@ -71,15 +71,15 @@ test_expect_success "can't remove pinned block" '
 '
 
 test_expect_success "can't remove pinned block: output looks good" '
-  grep -q "$DIRHASH: pinned via recursive" block_rm_err 
+  grep -q "$DIRHASH: pinned: recursive" block_rm_err
 '
 
 test_expect_success "can't remove indirectly pinned block" '
   test_must_fail ipfs block rm $FILE1HASH 2> block_rm_err
 '
 
 test_expect_success "can't remove indirectly pinned block: output looks good" '
-  grep -q "$FILE1HASH: pinned via $DIRHASH" block_rm_err 
+  grep -q "$FILE1HASH: pinned via $DIRHASH" block_rm_err
 '
 
 test_expect_success "multi-block 'ipfs block rm --ignore-pins' succeeds" '
@@ -100,10 +100,9 @@ test_expect_success "multi-block 'ipfs block rm' succeeds" '
 '
 
 test_expect_success "multi-block 'ipfs block rm' output looks good" '
-  echo "removed $FILE1HASH" > expected_rm &&
-  echo "removed $FILE2HASH" >> expected_rm &&
-  echo "removed $FILE3HASH" >> expected_rm &&
-  test_cmp expected_rm actual_rm
+  grep -F -q "removed $FILE1HASH" actual_rm &&
+  grep -F -q "removed $FILE2HASH" actual_rm &&
+  grep -F -q "removed $FILE3HASH" actual_rm
 '
 
 test_expect_success "'ipfs block stat' with nothing from stdin doesnt crash" '