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
}
func ( f * flusher ) prepare ( t * transaction , force bool ) ( revnos [ ] int64 , err error ) { if t . State != tpreparing { return f . rescan ( t , force ) } f . debugf ( " " , 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 { { " " , bson . D { { " " , 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 ( " " , 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 ( " " , dkey , info . Revno , info . Queue ) } else { f . debugf ( " " , 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 { { " " , t . Id } , { " " , tpreparing } } udoc := bson . D { { " " , bson . D { { " " , tprepared } , { " " , nonce } } } } chaos ( " " ) err = f . tc . Update ( qdoc , udoc ) if err == nil { t . State = tprepared t . Nonce = nonce } else if err == mgo . ErrNotFound { f . debugf ( " " , tt ) if err := f . reload ( t ) ; err != nil { return nil , err } else if t . State == tpreparing { panic ( " " ) } 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 ( " " , tt ) return nil , errPreReqs } revnos = assembledRevnos ( t . Ops , revno ) if ! prereqs { f . debugf ( " " , tt , revnos ) } else { f . debugf ( " " , 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.
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 .