Godeps/_workspace/src/gopkg.in/mgo.v2/txn/flusher.go
func (f *flusher) prepare(t *transaction, force bool) (revnos []int64, err error) {
	if t.State != tpreparing {
		return f.rescan(t, force)
	}
	f.debugf("Preparing %s", t)

	// dkeys being sorted means stable iteration across all runners. This
	// isn't strictly required, but reduces the chances of cycles.
	dkeys := t.docKeys()

	revno := make(map[docKey]int64)
	info := txnInfo{}
	tt := tokenFor(t)
NextDoc:
	for _, dkey := range dkeys {
		change := mgo.Change{
			Update:    bson.D{{"$addToSet", bson.D{{"txn-queue", tt}}}},
			ReturnNew: true,
		}
		c := f.tc.Database.C(dkey.C)
		cquery := c.FindId(dkey.Id).Select(txnFields)

	RetryDoc:
		change.Upsert = false
		chaos("")
		if _, err := cquery.Apply(change, &info); err == nil {
			if info.Remove == "" {
				// Fast path, unless workload is insert/remove heavy.
				revno[dkey] = info.Revno
				f.queue[dkey] = info.Queue
				f.debugf("[A] Prepared document %v with revno %d and queue: %v", dkey, info.Revno, info.Queue)
				continue NextDoc
			} else {
				// Handle remove in progress before preparing it.
				if err := f.loadAndApply(info.Remove); err != nil {
					return nil, err
				}
				goto RetryDoc
			}
		} else if err != mgo.ErrNotFound {
			return nil, err
		}

		// Document missing. Use stash collection.
		change.Upsert = true
		chaos("")
		_, err := f.sc.FindId(dkey).Apply(change, &info)
		if err != nil {
			return nil, err
		}
		if info.Insert != "" {
			// Handle insert in progress before preparing it.
			if err := f.loadAndApply(info.Insert); err != nil {
				return nil, err
			}
			goto RetryDoc
		}

		// Must confirm stash is still in use and is the same one
		// prepared, since applying a remove overwrites the stash.
		docFound := false
		stashFound := false
		if err = c.FindId(dkey.Id).Select(txnFields).One(&info); err == nil {
			docFound = true
		} else if err != mgo.ErrNotFound {
			return nil, err
		} else if err = f.sc.FindId(dkey).One(&info); err == nil {
			stashFound = true
			if info.Revno == 0 {
				// Missing revno in the stash only happens when it
				// has been upserted, in which case it defaults to -1.
				// Txn-inserted documents get revno -1 while in the stash
				// for the first time, and -revno-1 == 2 when they go live.
				info.Revno = -1
			}
		} else if err != mgo.ErrNotFound {
			return nil, err
		}

		if docFound && info.Remove == "" || stashFound && info.Insert == "" {
			for _, dtt := range info.Queue {
				if dtt != tt {
					continue
				}
				// Found tt properly prepared.
				if stashFound {
					f.debugf("[B] Prepared document %v on stash with revno %d and queue: %v", dkey, info.Revno, info.Queue)
				} else {
					f.debugf("[B] Prepared document %v with revno %d and queue: %v", dkey, info.Revno, info.Queue)
				}
				revno[dkey] = info.Revno
				f.queue[dkey] = info.Queue
				continue NextDoc
			}
		}

		// The stash wasn't valid and tt got overwriten. Try again.
		f.unstashToken(tt, dkey)
		goto RetryDoc
	}

	// Save the prepared nonce onto t.
	nonce := tt.nonce()
	qdoc := bson.D{{"_id", t.Id}, {"s", tpreparing}}
	udoc := bson.D{{"$set", bson.D{{"s", tprepared}, {"n", nonce}}}}
	chaos("set-prepared")
	err = f.tc.Update(qdoc, udoc)
	if err == nil {
		t.State = tprepared
		t.Nonce = nonce
	} else if err == mgo.ErrNotFound {
		f.debugf("Can't save nonce of %s: LOST RACE", tt)
		if err := f.reload(t); err != nil {
			return nil, err
		} else if t.State == tpreparing {
			panic("can't save nonce yet transaction is still preparing")
		} else if t.State != tprepared {
			return t.Revnos, nil
		}
		tt = t.token()
	} else if err != nil {
		return nil, err
	}

	prereqs, found := f.hasPreReqs(tt, dkeys)
	if !found {
		// Must only happen when reloading above.
		return f.rescan(t, force)
	} else if prereqs && !force {
		f.debugf("Prepared queue with %s [has prereqs & not forced].", tt)
		return nil, errPreReqs
	}
	revnos = assembledRevnos(t.Ops, revno)
	if !prereqs {
		f.debugf("Prepared queue with %s [no prereqs]. Revnos: %v", tt, revnos)
	} else {
		f.debugf("Prepared queue with %s [forced] Revnos: %v", tt, revnos)
	}
	return revnos, nil
}
func (f *flusher) prepare(t *transaction, force bool) (revnos []int64, err error) {
	if t.State != tpreparing {
		return f.rescan(t, force)
	}
	f.debugf("Preparing %s", t)

	// dkeys being sorted means stable iteration across all runners. This
	// isn't strictly required, but reduces the chances of cycles.
	dkeys := t.docKeys()

	revno := make(map[docKey]int64)
	info := txnInfo{}
	tt := tokenFor(t)
NextDoc:
	for _, dkey := range dkeys {
		change := mgo.Change{
			Update:    bson.D{{"$addToSet", bson.D{{"txn-queue", tt}}}},
			ReturnNew: true,
		}
		c := f.tc.Database.C(dkey.C)
		cquery := c.FindId(dkey.Id).Select(txnFields)

	RetryDoc:
		change.Upsert = false
		chaos("")
		if _, err := cquery.Apply(change, &info); err == nil {
			if info.Remove == "" {
				// Fast path, unless workload is insert/remove heavy.
				revno[dkey] = info.Revno
				f.queue[dkey] = info.Queue
				f.debugf("[A] Prepared document %v with revno %d and queue: %v", dkey, info.Revno, info.Queue)
				continue NextDoc
			} else {
				// Handle remove in progress before preparing it.
				if err := f.loadAndApply(info.Remove); err != nil {
					return nil, err
				}
				goto RetryDoc
			}
		} else if err != mgo.ErrNotFound {
			return nil, err
		}

		// Document missing. Use stash collection.
		change.Upsert = true
		chaos("")
		_, err := f.sc.FindId(dkey).Apply(change, &info)
		if err != nil {
			return nil, err
		}
		if info.Insert != "" {
			// Handle insert in progress before preparing it.
			if err := f.loadAndApply(info.Insert); err != nil {
				return nil, err
			}
			goto RetryDoc
		}

		// Must confirm stash is still in use and is the same one
		// prepared, since applying a remove overwrites the stash.
		docFound := false
		stashFound := false
		if err = c.FindId(dkey.Id).Select(txnFields).One(&info); err == nil {
			docFound = true
		} else if err != mgo.ErrNotFound {
			return nil, err
		} else if err = f.sc.FindId(dkey).One(&info); err == nil {
			stashFound = true
			if info.Revno == 0 {
				// Missing revno in the stash only happens when it
				// has been upserted, in which case it defaults to -1.
				// Txn-inserted documents get revno -1 while in the stash
				// for the first time, and -revno-1 == 2 when they go live.
				info.Revno = -1
			}
		} else if err != mgo.ErrNotFound {
			return nil, err
		}

		if docFound && info.Remove == "" || stashFound && info.Insert == "" {
			for _, dtt := range info.Queue {
				if dtt != tt {
					continue
				}
				// Found tt properly prepared.
				if stashFound {
					f.debugf("[B] Prepared document %v on stash with revno %d and queue: %v", dkey, info.Revno, info.Queue)
				} else {
					f.debugf("[B] Prepared document %v with revno %d and queue: %v", dkey, info.Revno, info.Queue)
				}
				revno[dkey] = info.Revno
				f.queue[dkey] = info.Queue
				continue NextDoc
			}
		}

		// The stash wasn't valid and tt got overwriten. Try again.
		f.unstashToken(tt, dkey)
		goto RetryDoc
	}

	// Save the prepared nonce onto t.
	nonce := tt.nonce()
	qdoc := bson.D{{"_id", t.Id}, {"s", tpreparing}}
	udoc := bson.D{{"$set", bson.D{{"s", tprepared}, {"n", nonce}}}}
	chaos("set-prepared")
	err = f.tc.Update(qdoc, udoc)
	if err == nil {
		t.State = tprepared
		t.Nonce = nonce
	} else if err == mgo.ErrNotFound {
		f.debugf("Can't save nonce of %s: LOST RACE", tt)
		if err := f.reload(t); err != nil {
			return nil, err
		} else if t.State == tpreparing {
			panic("can't save nonce yet transaction is still preparing")
		} else if t.State != tprepared {
			return t.Revnos, nil
		}
		tt = t.token()
	} else if err != nil {
		return nil, err
	}

	prereqs, found := f.hasPreReqs(tt, dkeys)
	if !found {
		// Must only happen when reloading above.
		return f.rescan(t, force)
	} else if prereqs && !force {
		f.debugf("Prepared queue with %s [has prereqs & not forced].", tt)
		return nil, errPreReqs
	}
	revnos = assembledRevnos(t.Ops, revno)
	if !prereqs {
		f.debugf("Prepared queue with %s [no prereqs]. Revnos: %v", tt, revnos)
	} else {
		f.debugf("Prepared queue with %s [forced] Revnos: %v", tt, revnos)
	}
	return revnos, nil
}
// prepare injects t's id onto txn-queue for all affected documents
// and collects the current txn-queue and txn-revno values during
// the process. If the prepared txn-queue indicates that there are
// pre-requisite transactions to be applied and the force parameter
// is false, errPreReqs will be returned. Otherwise, the current
// tip revision numbers for all the documents are returned.