blob: 5156693707034051b796a06422d8e24b5aa7a93e [file] [log] [blame]
// Copyright 2016 The Upspin Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package tree // import "upspin.io/dir/server/tree"
// This file implements block reading and writing.
import (
"upspin.io/bind"
"upspin.io/client/clientutil"
"upspin.io/errors"
"upspin.io/path"
"upspin.io/upspin"
)
// store marshals kids and packs them for the given DirEntry, updating its
// Packdata and Blocks, and stores them to the StoreServer.
func (t *Tree) store(entry *upspin.DirEntry, kids map[string]*node) error {
// Get our store server.
storeServer, err := bind.StoreServer(t.config, t.config.StoreEndpoint())
if err != nil {
return err
}
// Use our preferred packing
packer := t.packer
// Can't pack a non-dir entry. Something went bad if we got here.
if !entry.IsDir() {
err := errors.E(errors.Internal, "can't pack non-dir entry")
return err
}
// Prepare the dirEntry
entry.Blocks = nil // if any blocks existed, their references are lost as we're packing dirEntry again.
entry.Packing = t.config.Packing()
entry.Time = upspin.Now()
entry.Writer = t.config.UserName()
// Sequence number is already up-to-date (see setNodeDirtyAt).
// Start packing.
bp, err := packer.Pack(t.config, entry)
if err != nil {
return err
}
// Pack and store child nodes, keeping blocks at ~BlockSize.
var data []byte
for _, kid := range kids {
if kid.dirty {
// We should write nodes from the bottom up, so this should never happen.
return errors.E(kid.entry.Name, errors.Internal, "kid node is dirty")
}
// Check whether there are any empty Blocks or locations that
// are non-empty dirs or files.
if len(kid.kids) != 0 {
for _, b := range kid.entry.Blocks {
if len(b.Location.Reference) == 0 || b.Size == 0 {
return errors.E(kid.entry.Name, errors.Internal, "empty directory block when there exist kid blocks")
}
}
}
block, err := kid.entry.Marshal()
if err != nil {
return err
}
// Don't let blocks grow too much (but we never split a large DirEntry in the middle).
if len(data) > 0 && len(data)+len(block) > upspin.BlockSize {
// Flush now.
err = storeBlock(storeServer, bp, data)
if err != nil {
return err
}
data = data[:0]
}
data = append(data, block...)
}
// Put remaining data.
if len(data) > 0 {
err = storeBlock(storeServer, bp, data)
if err != nil {
return err
}
}
err = bp.Close()
if err != nil {
return err
}
return nil
}
// storeBlock stores a single block of data to the StoreServer as part of a block packing operation.
func storeBlock(store upspin.StoreServer, bp upspin.BlockPacker, data []byte) error {
cipher, err := bp.Pack(data)
if err != nil {
return err
}
refdata, err := store.Put(cipher)
if err != nil {
return err
}
loc := upspin.Location{
Endpoint: store.Endpoint(),
Reference: refdata.Reference,
}
bp.SetLocation(loc)
return nil
}
// Early logs, now called version 0 logs, used a sequence number that varied per
// file. Each file's sequence number grew from a unique, random number in the
// high bits, reserving the low 23 bits for an actual sequence. This was to
// prevent certain race conditions, most important an attempt to write a file at
// a sequence number when the file had been deleted and recreated underfoot.
// From version 1 onward, the sequence numbers increment at the tree level, and
// such races are impossible. We therefore don't bother with random high bits
// any more. For consistency, though, we need to clear the high bits when we
// have sequence numbers recovered from version 0 logs. This constant is used in
// the loading of the DirEntry in loadKidsFromBlock to clear those bits.
const version0SeqMask = 1<<23 - 1
// load fetches the blocks for a DirEntry from the StoreServer and unmarshals
// them into a map of kids.
func (t *Tree) load(entry *upspin.DirEntry) (kids map[string]*node, err error) {
nodePath, err := path.Parse(entry.Name)
if err != nil {
return nil, err
}
data, err := clientutil.ReadAll(t.config, entry)
if err != nil {
return nil, err
}
// Load children for this node.
kids = make(map[string]*node)
elemPos := nodePath.NElem()
v1Transition := t.user.V1Transition()
for len(data) > 0 {
var kid upspin.DirEntry
remaining, err := kid.Unmarshal(data)
if err != nil {
return nil, err
}
data = remaining
// Process this entry.
p, err := path.Parse(kid.Name)
if err != nil {
return nil, err
}
// Is this an old entry? If so, clear the high bits of the sequence number.
if kid.Time < v1Transition {
kid.Sequence &= version0SeqMask
}
// elem is the next pathwise element to load. Normally, it's the
// next element in entryPath. But if it's a directory that
// doesn't conform with the parent name, we allow it, to support
// snapshots and other types of "redirected" directories. But we
// must patch it according to this tree's perspective.
var elem string
if p.Drop(1).Equal(nodePath) {
elem = p.Elem(elemPos)
} else {
// If the block being loaded is not a prefix of the
// parent, then it's a "redirection" such as a snapshot
// entry.
elem = p.Elem(p.NElem() - 1) // ok, never root.
// Patch the entry's Name, so it belongs to this tree.
kid.Name = path.Join(nodePath.Path(), elem)
}
if _, exists := kids[elem]; exists {
// Trying to re-add an existing child. Something is amiss.
return nil, errors.E(errors.Internal, kid.Name, "re-adding an existing element in the Tree")
}
kids[elem] = &node{entry: kid}
}
return kids, nil
}