func (r *Runner) ResumeAll() (err error) {
	debugf("Resuming all unfinished transactions")
	iter := r.tc.Find(bson.D{{"s", bson.D{{"$in", []state{tpreparing, tprepared, tapplying}}}}}).Iter()
	var t transaction
	for iter.Next(&t) {
		if t.State == tapplied || t.State == taborted {
			continue
		}
		debugf("Resuming %s from %q", t.Id, t.State)
		if err := flush(r, &t); err != nil {
			return err
		}
		if !t.done() {
			panic(fmt.Errorf("invalid state for %s after flush: %q", &t, t.State))
		}
	}
	return nil
}
ResumeAll resumes all pending transactions. All ErrAborted errors
from individual transactions are ignored.
package txn
import (
	"encoding/binary"
	"fmt"
	"labix.org/v2/mgo"
	"labix.org/v2/mgo/bson"
	"reflect"
	"sort"
	"sync"
	crand "crypto/rand"
	mrand "math/rand"
)
type state int
const (
	tpreparing state = 1  
	tprepared  state = 2  
	taborting  state = 3  
	tapplying  state = 4  
	taborted   state = 5  
	tapplied   state = 6  
)
func (s state) String() string {
	switch s {
	case tpreparing:
		return "preparing"
	case tprepared:
		return "prepared"
	case taborting:
		return "aborting"
	case tapplying:
		return "applying"
	case taborted:
		return "aborted"
	case tapplied:
		return "applied"
	}
	panic(fmt.Errorf("unknown state: %d", s))
}
var rand *mrand.Rand
var randmu sync.Mutex
func init() {
	var seed int64
	err := binary.Read(crand.Reader, binary.BigEndian, &seed)
	if err != nil {
		panic(err)
	}
	rand = mrand.New(mrand.NewSource(seed))
}
type transaction struct {
	Id     bson.ObjectId `bson:"_id"`
	State  state         `bson:"s"`
	Info   interface{}   `bson:"i,omitempty"`
	Ops    []Op          `bson:"o"`
	Nonce  string        `bson:"n,omitempty"`
	Revnos []int64       `bson:"r,omitempty"`
	docKeysCached docKeys
}
func (t *transaction) String() string {
	if t.Nonce == "" {
		return t.Id.Hex()
	}
	return string(t.token())
}
func (t *transaction) done() bool {
	return t.State == tapplied || t.State == taborted
}
func (t *transaction) token() token {
	if t.Nonce == "" {
		panic("transaction has no nonce")
	}
	return tokenFor(t)
}
func (t *transaction) docKeys() docKeys {
	if t.docKeysCached != nil {
		return t.docKeysCached
	}
	dkeys := make(docKeys, 0, len(t.Ops))
NextOp:
	for _, op := range t.Ops {
		dkey := op.docKey()
		for i := range dkeys {
			if dkey == dkeys[i] {
				continue NextOp
			}
		}
		dkeys = append(dkeys, dkey)
	}
	sort.Sort(dkeys)
	t.docKeysCached = dkeys
	return dkeys
}
func tokenFor(t *transaction) token {
	nonce := t.Nonce
	if nonce == "" {
		nonce = newNonce()
	}
	return token(t.Id.Hex() + "_" + nonce)
}
func newNonce() string {
	randmu.Lock()
	r := rand.Uint32()
	randmu.Unlock()
	n := make([]byte, 8)
	for i := uint(0); i < 8; i++ {
		n[i] = "0123456789abcdef"[(r>>(4*i))&0xf]
	}
	return string(n)
}
type token string
func (tt token) id() bson.ObjectId { return bson.ObjectIdHex(string(tt[:24])) }
func (tt token) nonce() string     { return string(tt[25:]) }
type Op struct {
	C  string      `bson:"c"`
	Id interface{} `bson:"d"`
	Assert interface{} `bson:"a,omitempty"`
	Insert interface{} `bson:"i,omitempty"`
	Update interface{} `bson:"u,omitempty"`
	Remove bool        `bson:"r,omitempty"`
}
func (op *Op) isChange() bool {
	return op.Update != nil || op.Insert != nil || op.Remove
}
func (op *Op) docKey() docKey {
	return docKey{op.C, op.Id}
}
func (op *Op) name() string {
	switch {
	case op.Update != nil:
		return "update"
	case op.Insert != nil:
		return "insert"
	case op.Remove:
		return "remove"
	case op.Assert != nil:
		return "assert"
	}
	return "none"
}
const (
	DocExists  = "d+"
	DocMissing = "d-"
)
type Runner struct {
	tc *mgo.Collection  
	sc *mgo.Collection  
	lc *mgo.Collection  
}
func NewRunner(tc *mgo.Collection) *Runner {
	return &Runner{tc, tc.Database.C(tc.Name + ".stash"), nil}
}
var ErrAborted = fmt.Errorf("transaction aborted")
func (r *Runner) Run(ops []Op, id bson.ObjectId, info interface{}) (err error) {
	const efmt = "error in transaction op %d: %s"
	for i := range ops {
		op := &ops[i]
		if op.C == "" || op.Id == nil {
			return fmt.Errorf(efmt, i, "C or Id missing")
		}
		changes := 0
		if op.Insert != nil {
			changes++
		}
		if op.Update != nil {
			changes++
		}
		if op.Remove {
			changes++
		}
		if changes > 1 {
			return fmt.Errorf(efmt, i, "more than one of Insert/Update/Remove set")
		}
		if changes == 0 && op.Assert == nil {
			return fmt.Errorf(efmt, i, "none of Assert/Insert/Update/Remove set")
		}
	}
	if id == "" {
		id = bson.NewObjectId()
	}
	t := transaction{
		Id:    id,
		Ops:   ops,
		State: tpreparing,
		Info:  info,
	}
	if err = r.tc.Insert(&t); err != nil {
		return err
	}
	if err = flush(r, &t); err != nil {
		return err
	}
	if t.State == taborted {
		return ErrAborted
	} else if t.State != tapplied {
		panic(fmt.Errorf("invalid state for %s after flush: %q", &t, t.State))
	}
	return nil
}