Merge 216da0b7 (automated)

1 files changed, 605 insertions, 0 deletions

diff --git a/pkg/state/decode.go b/pkg/state/decode.go
new file mode 100644
index 000000000..73a59f871
--- /dev/null
+++ b/pkg/state/decode.go
@@ -0,0 +1,605 @@
+// Copyright 2018 The gVisor Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package state
+
+import (
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"io"
+	"reflect"
+	"sort"
+
+	"github.com/golang/protobuf/proto"
+	pb "gvisor.googlesource.com/gvisor/pkg/state/object_go_proto"
+)
+
+// objectState represents an object that may be in the process of being
+// decoded. Specifically, it represents either a decoded object, or an an
+// interest in a future object that will be decoded. When that interest is
+// registered (via register), the storage for the object will be created, but
+// it will not be decoded until the object is encountered in the stream.
+type objectState struct {
+	// id is the id for this object.
+	//
+	// If this field is zero, then this is an anonymous (unregistered,
+	// non-reference primitive) object. This is immutable.
+	id uint64
+
+	// obj is the object. This may or may not be valid yet, depending on
+	// whether complete returns true. However, regardless of whether the
+	// object is valid, obj contains a final storage location for the
+	// object. This is immutable.
+	//
+	// Note that this must be addressable (obj.Addr() must not panic).
+	//
+	// The obj passed to the decode methods below will equal this obj only
+	// in the case of decoding the top-level object. However, the passed
+	// obj may represent individual fields, elements of a slice, etc. that
+	// are effectively embedded within the reflect.Value below but with
+	// distinct types.
+	obj reflect.Value
+
+	// blockedBy is the number of dependencies this object has.
+	blockedBy int
+
+	// blocking is a list of the objects blocked by this one.
+	blocking []*objectState
+
+	// callbacks is a set of callbacks to execute on load.
+	callbacks []func()
+
+	// path is the decoding path to the object.
+	path recoverable
+}
+
+// complete indicates the object is complete.
+func (os *objectState) complete() bool {
+	return os.blockedBy == 0 && len(os.callbacks) == 0
+}
+
+// checkComplete checks for completion. If the object is complete, pending
+// callbacks will be executed and checkComplete will be called on downstream
+// objects (those depending on this one).
+func (os *objectState) checkComplete(stats *Stats) {
+	if os.blockedBy > 0 {
+		return
+	}
+	stats.Start(os.obj)
+
+	// Fire all callbacks.
+	for _, fn := range os.callbacks {
+		fn()
+	}
+	os.callbacks = nil
+
+	// Clear all blocked objects.
+	for _, other := range os.blocking {
+		other.blockedBy--
+		other.checkComplete(stats)
+	}
+	os.blocking = nil
+	stats.Done()
+}
+
+// waitFor queues a dependency on the given object.
+func (os *objectState) waitFor(other *objectState, callback func()) {
+	os.blockedBy++
+	other.blocking = append(other.blocking, os)
+	if callback != nil {
+		other.callbacks = append(other.callbacks, callback)
+	}
+}
+
+// findCycleFor returns when the given object is found in the blocking set.
+func (os *objectState) findCycleFor(target *objectState) []*objectState {
+	for _, other := range os.blocking {
+		if other == target {
+			return []*objectState{target}
+		} else if childList := other.findCycleFor(target); childList != nil {
+			return append(childList, other)
+		}
+	}
+	return nil
+}
+
+// findCycle finds a dependency cycle.
+func (os *objectState) findCycle() []*objectState {
+	return append(os.findCycleFor(os), os)
+}
+
+// decodeState is a graph of objects in the process of being decoded.
+//
+// The decode process involves loading the breadth-first graph generated by
+// encode. This graph is read in it's entirety, ensuring that all object
+// storage is complete.
+//
+// As the graph is being serialized, a set of completion callbacks are
+// executed. These completion callbacks should form a set of acyclic subgraphs
+// over the original one. After decoding is complete, the objects are scanned
+// to ensure that all callbacks are executed, otherwise the callback graph was
+// not acyclic.
+type decodeState struct {
+	// objectByID is the set of objects in progress.
+	objectsByID map[uint64]*objectState
+
+	// deferred are objects that have been read, by no interest has been
+	// registered yet. These will be decoded once interest in registered.
+	deferred map[uint64]*pb.Object
+
+	// outstanding is the number of outstanding objects.
+	outstanding uint32
+
+	// r is the input stream.
+	r io.Reader
+
+	// stats is the passed stats object.
+	stats *Stats
+
+	// recoverable is the panic recover facility.
+	recoverable
+}
+
+// lookup looks up an object in decodeState or returns nil if no such object
+// has been previously registered.
+func (ds *decodeState) lookup(id uint64) *objectState {
+	return ds.objectsByID[id]
+}
+
+// wait registers a dependency on an object.
+//
+// As a special case, we always allow _useable_ references back to the first
+// decoding object because it may have fields that are already decoded. We also
+// allow trivial self reference, since they can be handled internally.
+func (ds *decodeState) wait(waiter *objectState, id uint64, callback func()) {
+	switch id {
+	case 0:
+		// Nil pointer; nothing to wait for.
+		fallthrough
+	case waiter.id:
+		// Trivial self reference.
+		fallthrough
+	case 1:
+		// Root object; see above.
+		if callback != nil {
+			callback()
+		}
+		return
+	}
+
+	// No nil can be returned here.
+	waiter.waitFor(ds.lookup(id), callback)
+}
+
+// waitObject notes a blocking relationship.
+func (ds *decodeState) waitObject(os *objectState, p *pb.Object, callback func()) {
+	if rv, ok := p.Value.(*pb.Object_RefValue); ok {
+		// Refs can encode pointers and maps.
+		ds.wait(os, rv.RefValue, callback)
+	} else if sv, ok := p.Value.(*pb.Object_SliceValue); ok {
+		// See decodeObject; we need to wait for the array (if non-nil).
+		ds.wait(os, sv.SliceValue.RefValue, callback)
+	} else if iv, ok := p.Value.(*pb.Object_InterfaceValue); ok {
+		// It's an interface (wait recurisvely).
+		ds.waitObject(os, iv.InterfaceValue.Value, callback)
+	} else if callback != nil {
+		// Nothing to wait for: execute the callback immediately.
+		callback()
+	}
+}
+
+// register registers a decode with a type.
+//
+// This type is only used to instantiate a new object if it has not been
+// registered previously.
+func (ds *decodeState) register(id uint64, typ reflect.Type) *objectState {
+	os, ok := ds.objectsByID[id]
+	if ok {
+		return os
+	}
+
+	// Record in the object index.
+	if typ.Kind() == reflect.Map {
+		os = &objectState{id: id, obj: reflect.MakeMap(typ), path: ds.recoverable.copy()}
+	} else {
+		os = &objectState{id: id, obj: reflect.New(typ).Elem(), path: ds.recoverable.copy()}
+	}
+	ds.objectsByID[id] = os
+
+	if o, ok := ds.deferred[id]; ok {
+		// There is a deferred object.
+		delete(ds.deferred, id) // Free memory.
+		ds.decodeObject(os, os.obj, o, "", nil)
+	} else {
+		// There is no deferred object.
+		ds.outstanding++
+	}
+
+	return os
+}
+
+// decodeStruct decodes a struct value.
+func (ds *decodeState) decodeStruct(os *objectState, obj reflect.Value, s *pb.Struct) {
+	// Set the fields.
+	m := Map{newInternalMap(nil, ds, os)}
+	defer internalMapPool.Put(m.internalMap)
+	for _, field := range s.Fields {
+		m.data = append(m.data, entry{
+			name:   field.Name,
+			object: field.Value,
+		})
+	}
+
+	// Sort the fields for efficient searching.
+	//
+	// Technically, these should already appear in sorted order in the
+	// state ordering, so this cost is effectively a single scan to ensure
+	// that the order is correct.
+	if len(m.data) > 1 {
+		sort.Slice(m.data, func(i, j int) bool {
+			return m.data[i].name < m.data[j].name
+		})
+	}
+
+	// Invoke the load; this will recursively decode other objects.
+	fns, ok := registeredTypes.lookupFns(obj.Addr().Type())
+	if ok {
+		// Invoke the loader.
+		fns.invokeLoad(obj.Addr(), m)
+	} else if obj.NumField() == 0 {
+		// Allow anonymous empty structs.
+		return
+	} else {
+		// Propagate an error.
+		panic(fmt.Errorf("unregistered type %s", obj.Type()))
+	}
+}
+
+// decodeMap decodes a map value.
+func (ds *decodeState) decodeMap(os *objectState, obj reflect.Value, m *pb.Map) {
+	if obj.IsNil() {
+		obj.Set(reflect.MakeMap(obj.Type()))
+	}
+	for i := 0; i < len(m.Keys); i++ {
+		// Decode the objects.
+		kv := reflect.New(obj.Type().Key()).Elem()
+		vv := reflect.New(obj.Type().Elem()).Elem()
+		ds.decodeObject(os, kv, m.Keys[i], ".(key %d)", i)
+		ds.decodeObject(os, vv, m.Values[i], "[%#v]", kv.Interface())
+		ds.waitObject(os, m.Keys[i], nil)
+		ds.waitObject(os, m.Values[i], nil)
+
+		// Set in the map.
+		obj.SetMapIndex(kv, vv)
+	}
+}
+
+// decodeArray decodes an array value.
+func (ds *decodeState) decodeArray(os *objectState, obj reflect.Value, a *pb.Array) {
+	if len(a.Contents) != obj.Len() {
+		panic(fmt.Errorf("mismatching array length expect=%d, actual=%d", obj.Len(), len(a.Contents)))
+	}
+	// Decode the contents into the array.
+	for i := 0; i < len(a.Contents); i++ {
+		ds.decodeObject(os, obj.Index(i), a.Contents[i], "[%d]", i)
+		ds.waitObject(os, a.Contents[i], nil)
+	}
+}
+
+// decodeInterface decodes an interface value.
+func (ds *decodeState) decodeInterface(os *objectState, obj reflect.Value, i *pb.Interface) {
+	// Is this a nil value?
+	if i.Type == "" {
+		return // Just leave obj alone.
+	}
+
+	// Get the dispatchable type. This may not be used if the given
+	// reference has already been resolved, but if not we need to know the
+	// type to create.
+	t, ok := registeredTypes.lookupType(i.Type)
+	if !ok {
+		panic(fmt.Errorf("no valid type for %q", i.Type))
+	}
+
+	if obj.Kind() != reflect.Map {
+		// Set the obj to be the given typed value; this actually sets
+		// obj to be a non-zero value -- namely, it inserts type
+		// information. There's no need to do this for maps.
+		obj.Set(reflect.Zero(t))
+	}
+
+	// Decode the dereferenced element; there is no need to wait here, as
+	// the interface object shares the current object state.
+	ds.decodeObject(os, obj, i.Value, ".(%s)", i.Type)
+}
+
+// decodeObject decodes a object value.
+func (ds *decodeState) decodeObject(os *objectState, obj reflect.Value, object *pb.Object, format string, param interface{}) {
+	ds.push(false, format, param)
+	ds.stats.Add(obj)
+	ds.stats.Start(obj)
+
+	switch x := object.GetValue().(type) {
+	case *pb.Object_BoolValue:
+		obj.SetBool(x.BoolValue)
+	case *pb.Object_StringValue:
+		obj.SetString(string(x.StringValue))
+	case *pb.Object_Int64Value:
+		obj.SetInt(x.Int64Value)
+		if obj.Int() != x.Int64Value {
+			panic(fmt.Errorf("signed integer truncated in %v for %s", object, obj.Type()))
+		}
+	case *pb.Object_Uint64Value:
+		obj.SetUint(x.Uint64Value)
+		if obj.Uint() != x.Uint64Value {
+			panic(fmt.Errorf("unsigned integer truncated in %v for %s", object, obj.Type()))
+		}
+	case *pb.Object_DoubleValue:
+		obj.SetFloat(x.DoubleValue)
+		if obj.Float() != x.DoubleValue {
+			panic(fmt.Errorf("float truncated in %v for %s", object, obj.Type()))
+		}
+	case *pb.Object_RefValue:
+		// Resolve the pointer itself, even though the object may not
+		// be decoded yet. You need to use wait() in order to ensure
+		// that is the case. See wait above, and Map.Barrier.
+		if id := x.RefValue; id != 0 {
+			// Decoding the interface should have imparted type
+			// information, so from this point it's safe to resolve
+			// and use this dynamic information for actually
+			// creating the object in register.
+			//
+			// (For non-interfaces this is a no-op).
+			dyntyp := reflect.TypeOf(obj.Interface())
+			if dyntyp.Kind() == reflect.Map {
+				// Remove the map object count here to avoid
+				// double counting, as this object will be
+				// counted again when it gets processed later.
+				// We do not add a reference count as the
+				// reference is artificial.
+				ds.stats.Remove(obj)
+				obj.Set(ds.register(id, dyntyp).obj)
+			} else if dyntyp.Kind() == reflect.Ptr {
+				ds.push(true /* dereference */, "", nil)
+				obj.Set(ds.register(id, dyntyp.Elem()).obj.Addr())
+				ds.pop()
+			} else {
+				obj.Set(ds.register(id, dyntyp.Elem()).obj.Addr())
+			}
+		} else {
+			// We leave obj alone here. That's because if obj
+			// represents an interface, it may have been embued
+			// with type information in decodeInterface, and we
+			// don't want to destroy that information.
+		}
+	case *pb.Object_SliceValue:
+		// It's okay to slice the array here, since the contents will
+		// still be provided later on. These semantics are a bit
+		// strange but they are handled in the Map.Barrier properly.
+		//
+		// The special semantics of zero ref apply here too.
+		if id := x.SliceValue.RefValue; id != 0 && x.SliceValue.Capacity > 0 {
+			v := reflect.ArrayOf(int(x.SliceValue.Capacity), obj.Type().Elem())
+			obj.Set(ds.register(id, v).obj.Slice3(0, int(x.SliceValue.Length), int(x.SliceValue.Capacity)))
+		}
+	case *pb.Object_ArrayValue:
+		ds.decodeArray(os, obj, x.ArrayValue)
+	case *pb.Object_StructValue:
+		ds.decodeStruct(os, obj, x.StructValue)
+	case *pb.Object_MapValue:
+		ds.decodeMap(os, obj, x.MapValue)
+	case *pb.Object_InterfaceValue:
+		ds.decodeInterface(os, obj, x.InterfaceValue)
+	case *pb.Object_ByteArrayValue:
+		copyArray(obj, reflect.ValueOf(x.ByteArrayValue))
+	case *pb.Object_Uint16ArrayValue:
+		// 16-bit slices are serialized as 32-bit slices.
+		// See object.proto for details.
+		s := x.Uint16ArrayValue.Values
+		t := obj.Slice(0, obj.Len()).Interface().([]uint16)
+		if len(t) != len(s) {
+			panic(fmt.Errorf("mismatching array length expect=%d, actual=%d", len(t), len(s)))
+		}
+		for i := range s {
+			t[i] = uint16(s[i])
+		}
+	case *pb.Object_Uint32ArrayValue:
+		copyArray(obj, reflect.ValueOf(x.Uint32ArrayValue.Values))
+	case *pb.Object_Uint64ArrayValue:
+		copyArray(obj, reflect.ValueOf(x.Uint64ArrayValue.Values))
+	case *pb.Object_UintptrArrayValue:
+		copyArray(obj, castSlice(reflect.ValueOf(x.UintptrArrayValue.Values), reflect.TypeOf(uintptr(0))))
+	case *pb.Object_Int8ArrayValue:
+		copyArray(obj, castSlice(reflect.ValueOf(x.Int8ArrayValue.Values), reflect.TypeOf(int8(0))))
+	case *pb.Object_Int16ArrayValue:
+		// 16-bit slices are serialized as 32-bit slices.
+		// See object.proto for details.
+		s := x.Int16ArrayValue.Values
+		t := obj.Slice(0, obj.Len()).Interface().([]int16)
+		if len(t) != len(s) {
+			panic(fmt.Errorf("mismatching array length expect=%d, actual=%d", len(t), len(s)))
+		}
+		for i := range s {
+			t[i] = int16(s[i])
+		}
+	case *pb.Object_Int32ArrayValue:
+		copyArray(obj, reflect.ValueOf(x.Int32ArrayValue.Values))
+	case *pb.Object_Int64ArrayValue:
+		copyArray(obj, reflect.ValueOf(x.Int64ArrayValue.Values))
+	case *pb.Object_BoolArrayValue:
+		copyArray(obj, reflect.ValueOf(x.BoolArrayValue.Values))
+	case *pb.Object_Float64ArrayValue:
+		copyArray(obj, reflect.ValueOf(x.Float64ArrayValue.Values))
+	case *pb.Object_Float32ArrayValue:
+		copyArray(obj, reflect.ValueOf(x.Float32ArrayValue.Values))
+	default:
+		// Shoud not happen, not propagated as an error.
+		panic(fmt.Sprintf("unknown object %v for %s", object, obj.Type()))
+	}
+
+	ds.stats.Done()
+	ds.pop()
+}
+
+func copyArray(dest reflect.Value, src reflect.Value) {
+	if dest.Len() != src.Len() {
+		panic(fmt.Errorf("mismatching array length expect=%d, actual=%d", dest.Len(), src.Len()))
+	}
+	reflect.Copy(dest, castSlice(src, dest.Type().Elem()))
+}
+
+// Deserialize deserializes the object state.
+//
+// This function may panic and should be run in safely().
+func (ds *decodeState) Deserialize(obj reflect.Value) {
+	ds.objectsByID[1] = &objectState{id: 1, obj: obj, path: ds.recoverable.copy()}
+	ds.outstanding = 1 // The root object.
+
+	// Decode all objects in the stream.
+	//
+	// See above, we never process objects while we have no outstanding
+	// interests (other than the very first object).
+	for id := uint64(1); ds.outstanding > 0; id++ {
+		os := ds.lookup(id)
+		ds.stats.Start(os.obj)
+
+		o, err := ds.readObject()
+		if err != nil {
+			panic(err)
+		}
+
+		if os != nil {
+			// Decode the object.
+			ds.from = &os.path
+			ds.decodeObject(os, os.obj, o, "", nil)
+			ds.outstanding--
+		} else {
+			// If an object hasn't had interest registered
+			// previously, we deferred decoding until interest is
+			// registered.
+			ds.deferred[id] = o
+		}
+
+		ds.stats.Done()
+	}
+
+	// Check the zero-length header at the end.
+	length, object, err := ReadHeader(ds.r)
+	if err != nil {
+		panic(err)
+	}
+	if length != 0 {
+		panic(fmt.Sprintf("expected zero-length terminal, got %d", length))
+	}
+	if object {
+		panic("expected non-object terminal")
+	}
+
+	// Check if we have any deferred objects.
+	if count := len(ds.deferred); count > 0 {
+		// Shoud not happen, not propagated as an error.
+		panic(fmt.Sprintf("still have %d deferred objects", count))
+	}
+
+	// Scan and fire all callbacks.
+	for _, os := range ds.objectsByID {
+		os.checkComplete(ds.stats)
+	}
+
+	// Check if we have any remaining dependency cycles.
+	for _, os := range ds.objectsByID {
+		if !os.complete() {
+			// This must be the result of a dependency cycle.
+			cycle := os.findCycle()
+			var buf bytes.Buffer
+			buf.WriteString("dependency cycle: {")
+			for i, cycleOS := range cycle {
+				if i > 0 {
+					buf.WriteString(" => ")
+				}
+				buf.WriteString(fmt.Sprintf("%s", cycleOS.obj.Type()))
+			}
+			buf.WriteString("}")
+			// Panic as an error; propagate to the caller.
+			panic(errors.New(string(buf.Bytes())))
+		}
+	}
+}
+
+type byteReader struct {
+	io.Reader
+}
+
+// ReadByte implements io.ByteReader.
+func (br byteReader) ReadByte() (byte, error) {
+	var b [1]byte
+	n, err := br.Reader.Read(b[:])
+	if n > 0 {
+		return b[0], nil
+	} else if err != nil {
+		return 0, err
+	} else {
+		return 0, io.ErrUnexpectedEOF
+	}
+}
+
+// ReadHeader reads an object header.
+//
+// Each object written to the statefile is prefixed with a header. See
+// WriteHeader for more information; these functions are exported to allow
+// non-state writes to the file to play nice with debugging tools.
+func ReadHeader(r io.Reader) (length uint64, object bool, err error) {
+	// Read the header.
+	length, err = binary.ReadUvarint(byteReader{r})
+	if err != nil {
+		return
+	}
+
+	// Decode whether the object is valid.
+	object = length&0x1 != 0
+	length = length >> 1
+	return
+}
+
+// readObject reads an object from the stream.
+func (ds *decodeState) readObject() (*pb.Object, error) {
+	// Read the header.
+	length, object, err := ReadHeader(ds.r)
+	if err != nil {
+		return nil, err
+	}
+	if !object {
+		return nil, fmt.Errorf("invalid object header")
+	}
+
+	// Read the object.
+	buf := make([]byte, length)
+	for done := 0; done < len(buf); {
+		n, err := ds.r.Read(buf[done:])
+		done += n
+		if n == 0 && err != nil {
+			return nil, err
+		}
+	}
+
+	// Unmarshal.
+	obj := new(pb.Object)
+	if err := proto.Unmarshal(buf, obj); err != nil {
+		return nil, err
+	}
+
+	return obj, nil
+}