func ViewUserInfoTxn(t Query, tx ETransaction) (*Result, error) {
	uk := UserKey(t.U1)
	urec, err := tx.Read(uk)
	if err != nil {
		if err == ESTASH {
			dlog.Printf("User  %v stashed\n", t.U1)
			return nil, ESTASH
		} else if err == EABORT {
			return nil, EABORT
		} else if err == ENOKEY {
			dlog.Printf("No user? %v\n", t.U1)
			if tx.Commit() == 0 {
				return nil, EABORT
			} else {
				return nil, ENORETRY
			}
		} else {
			log.Fatalf("view user err: %v\n", err)
		}
	}
	_ = urec.Value().(*User)
	if tx.Commit() == 0 {
		return nil, EABORT
	}
	var r *Result = nil
	if *Allocate {
		r = &Result{urec.Value()}
	}
	return r, nil
}

func BenchmarkRead(b *testing.B) {
	runtime.GOMAXPROCS(4)
	b.StopTimer()
	nb := 10000
	np := 100
	n := 4

	s := NewStore()
	// Load
	for i := 0; i < np; i++ {
		s.CreateKey(ProductKey(i), int32(0), SUM)
	}
	for i := 0; i < nb; i++ {
		s.CreateKey(UserKey(uint64(i)), "x", WRITE)
	}

	c := NewCoordinator(n, s)
	val := make([]int32, np)
	read_rate := 50

	var wg sync.WaitGroup
	b.StartTimer()
	for p := 0; p < n; p++ {
		wg.Add(1)
		go func(id int) {
			w := c.Workers[id]
			for i := 0; i < b.N/3; i++ {
				p := ProductKey(i % np)
				u := UserKey(uint64(i % nb))
				amt := int32(rand.Intn(100))
				var tx Query
				rr := rand.Intn(100)
				if rr >= read_rate {
					tx = Query{TXN: D_BUY, K1: u, K2: p, A: amt, W: nil, T: 0}
					_, err := w.One(tx)
					if err == nil {
						atomic.AddInt32(&val[i%np], amt)
					}
				} else {
					tx = Query{TXN: D_READ_ONE, K1: p, W: make(chan struct {
						R *Result
						E error
					}), T: 0}
					_, err := w.One(tx)
					if err == ESTASH {
						dlog.Printf("client [%v] waiting for %v; epoch %v\n", w.ID, i%np, w.epoch)
						<-tx.W
					}
				}
			}
			wg.Done()
		}(p)
	}
	dlog.Printf("Waiting on outer\n")
	wg.Wait()
	b.StopTimer()
	c.Finish()
	Validate(c, s, nb, np, val, b.N)
}

func PutCommentTxn(t Query, tx ETransaction) (*Result, error) {
	var r *Result = nil
	touser := t.U1
	item := t.U2
	tok := UserKey(touser)
	torec, err := tx.Read(tok)
	if err != nil {
		if err == ESTASH {
			dlog.Printf("User key for user %v stashed\n", touser)
			return nil, ESTASH
		} else if err == EABORT {
			return nil, EABORT
		} else if err == ENOKEY {
			dlog.Printf("No user? %v\n", touser)
			if tx.Commit() == 0 {
				return nil, EABORT
			} else {
				return nil, ENORETRY
			}
		} else {
			log.Fatalf("err %v\n", err)
		}
	}
	nickname := torec.Value().(*User).Nickname
	ik := ItemKey(item)
	irec, err := tx.Read(ik)
	if err != nil {
		if err == ESTASH {
			dlog.Printf("Item key  %v stashed\n", item)
			return nil, ESTASH
		} else if err == EABORT {
			return nil, EABORT
		} else if err == ENOKEY {
			dlog.Printf("PutCommentTxn: No item? %v\n", item)
			if tx.Commit() == 0 {
				return nil, EABORT
			} else {
				return nil, ENORETRY
			}
		} else {
			log.Fatalf("err %v\n", err)
		}
	}
	itemname := irec.Value().(*Item).Name
	if tx.Commit() == 0 {
		return r, EABORT
	}
	if *Allocate {
		r = &Result{
			&struct {
				nick  string
				iname string
			}{nickname, itemname},
		}
	}
	return r, nil
}

func (b *Buy) Populate(s *ddtxn.Store, ex *ddtxn.ETransaction) {
	for i := 0; i < b.nbidders; i++ {
		k := ddtxn.ProductKey(i)
		s.CreateKey(k, int32(0), ddtxn.SUM)
	}
	dlog.Printf("Created %v products; np: %v\n", b.nbidders, b.nproducts)
	for i := 0; i < b.nbidders; i++ {
		k := ddtxn.UserKey(uint64(i))
		s.CreateKey(k, "x", ddtxn.WRITE)
	}
	dlog.Printf("Created %v bidders\n", b.nbidders)
	dlog.Printf("Done with Populate")
}

func RegisterUserTxn(t Query, tx ETransaction) (*Result, error) {
	region := t.U1
	nickname := t.U2
	var r *Result = nil

	var n uint64
	var nick Key

	if !*Allocate || nickname == 0 {
		n = tx.UID('u')
		nick = NicknameKey(tx.UID('d'))
	} else {
		n = nickname
		nick = NicknameKey(n)
	}
	u := UserKey(n)
	user := &User{
		ID:       n,
		Name:     "xxxxxxx",
		Nickname: string(nickname),
		Region:   region,
	}
	tx.MaybeWrite(nick)
	br, err := tx.Read(nick)
	var val uint64 = 0
	if br != nil && br.exists {
		val = br.Value().(uint64)
	}

	if err != ENOKEY && val != 0 {
		// Someone else is using this nickname
		dlog.Printf("Nickname taken %v %v\n", nickname, nick)
		tx.Abort()
		return nil, ENORETRY
	}
	tx.Write(u, user, WRITE)
	tx.Write(nick, nickname, WRITE)

	if tx.Commit() == 0 {
		dlog.Printf("RegisterUser() Abort\n")
		return nil, EABORT
	}
	if *Allocate {
		r = &Result{uint64(n)}
		// dlog.Printf("Registered user %v %v\n", nickname, n)
	}
	return r, nil
}

// Used during "merge" phase, along with br.mu
func (br *BRecord) Apply(val Value) {
	if br == nil {
		dlog.Printf("Nil record %v %v\n", val, br)
	}
	switch br.key_type {
	case SUM:
		delta := val.(int32)
		atomic.AddInt32(&br.int_value, delta)
	case MAX:
		delta := val.(int32)
		br.mu.Lock()
		defer br.mu.Unlock()
		if br.int_value < delta {
			br.int_value = delta
		}
	case WRITE:
		br.mu.Lock()
		defer br.mu.Unlock()
		br.value = val
	case LIST:
		br.mu.Lock()
		defer br.mu.Unlock()
		entries := val.([]Entry)
		br.listApply(entries)
	case OOWRITE:
		br.mu.Lock()
		defer br.mu.Unlock()
		x := val.(Overwrite)
		if br.int_value < x.i {
			br.int_value = x.i
			br.value = x.v
		}
	}
}

func NewCoordinator(n int, s *Store) *Coordinator {
	c := &Coordinator{
		n:                     n,
		Workers:               make([]*Worker, n),
		epochTID:              EPOCH_INCR,
		wepoch:                make([]chan TID, n),
		wsafe:                 make([]chan TID, n),
		wgo:                   make([]chan TID, n),
		wdone:                 make([]chan TID, n),
		Done:                  make(chan chan bool),
		Accelerate:            make(chan bool),
		Coordinate:            false,
		PotentialPhaseChanges: 0,
		to_remove:             make(map[Key]bool),
		Finished:              make([]bool, n),
	}
	for i := 0; i < n; i++ {
		c.wepoch[i] = make(chan TID)
		c.wsafe[i] = make(chan TID)
		c.wgo[i] = make(chan TID)
		c.wdone[i] = make(chan TID)
		c.Finished[i] = false
		c.Workers[i] = NewWorker(i, s, c)
	}
	c.Finished = make([]bool, n)
	dlog.Printf("[coordinator] %v workers\n", n)
	go c.Process()
	return c
}

func (b *Buy) Validate(s *ddtxn.Store, nitr int) bool {
	good := true
	zero_cnt := 0
	for j := 0; j < b.nproducts; j++ {
		var x int32
		k := ddtxn.ProductKey(j)
		v, err := s.Get(k)
		if err != nil {
			if b.validate[j] != 0 {
				fmt.Printf("Validating key %v failed; store: none should have: %v\n", k, b.validate[j])
				good = false
			}
			continue
		}
		x = v.Value().(int32)
		if x != b.validate[j] {
			fmt.Printf("Validating key %v failed; store: %v should have: %v\n", k, x, b.validate[j])
			good = false
		}
		if x == 0 {
			zero_cnt++
		}
	}
	if zero_cnt == b.nproducts && nitr > 10 {
		fmt.Printf("Bad: all zeroes!\n")
		dlog.Printf("Bad: all zeroes!\n")
		good = false
	}
	return good
}

func BenchmarkBuy(b *testing.B) {
	runtime.GOMAXPROCS(8)
	b.StopTimer()
	nb := 10000
	np := 100
	n := 8
	s := NewStore()
	// Load
	for i := 0; i < np; i++ {
		s.CreateKey(ProductKey(i), int32(0), MAX)
	}
	for i := 0; i < nb; i++ {
		s.CreateKey(UserKey(uint64(i)), "x", WRITE)
	}
	c := NewCoordinator(n, s)
	val := make([]int32, np)

	var wg sync.WaitGroup
	b.StartTimer()
	for p := 0; p < n; p++ {
		wg.Add(1)
		go func(id int) {
			w := c.Workers[id]
			for i := 0; i < b.N/3; i++ {
				p := ProductKey(i % np)
				u := UserKey(uint64(i % nb))
				amt := int32(rand.Intn(100))
				tx := Query{TXN: D_BUY, K1: u, A: amt, K2: p, W: nil, T: 0}
				_, err := w.One(tx)
				if err == nil {
					atomic.AddInt32(&val[i%np], amt)
				}
			}
			wg.Done()
		}(p)
	}
	dlog.Printf("Waiting on outer\n")
	wg.Wait()
	dlog.Printf("done\n")
	b.StopTimer()
	c.Finish()
	Validate(c, s, nb, np, val, b.N)
	//PrintLockCounts(s, nb, np, false)
}

func TestRandN(t *testing.T) {
	var seed uint32 = uint32(1)
	dlog.Printf("seed %v\n", seed)
	for i := 0; i < 1000; i++ {
		x := RandN(&seed, 10)
		// No idea how to test a random number generator, just look at the results for now.
		dlog.Println(x, seed)
		_ = x
	}
}

func StoreCommentTxn(t Query, tx ETransaction) (*Result, error) {
	touser := t.U1
	fromuser := t.U2
	item := t.U3
	comment_s := t.S1
	rating := t.U4

	n := tx.UID('c')
	com := CommentKey(n)
	comment := &Comment{
		ID:      n,
		From:    fromuser,
		To:      touser,
		Rating:  rating,
		Comment: comment_s,
		Item:    item,
		Date:    11,
	}
	tx.Write(com, comment, WRITE)

	rkey := RatingKey(touser)
	err := tx.WriteInt32(rkey, int32(rating), SUM)
	if err != nil {
		dlog.Printf("Comment abort %v\n", t)
		tx.Abort()
		return nil, err
	}

	if tx.Commit() == 0 {
		dlog.Printf("Comment abort %v\n", t)
		return nil, EABORT
	}
	var r *Result = nil
	if *Allocate {
		r = &Result{uint64(n)}
		dlog.Printf("%v Comment %v %v\n", touser, fromuser, item)
	}
	return r, nil
}

func Test2RandN(t *testing.T) {
	n := 0
	var local_seed uint32 = uint32(n + 1)
	portion_sz := 100
	dlog.Printf("LOCAL: %v\n", local_seed)
	j := 0
	for {
		select {
		default:
			var bidder int
			rand := RandN(&local_seed, uint32(portion_sz))
			lb := int(rand)
			bidder = lb + n*portion_sz
			amt := int(RandN(&local_seed, 10))
			dlog.Printf("%v rand: %v bidder: %v local: %v amt: %v\n", n, rand, bidder, local_seed, amt)
			j++
			if j > 100 {
				return
			}
		}
	}
}

func (s *Store) CreateMuRLockedKey(k Key, kt KeyType) (*BRecord, error) {
	br := MakeBR(k, nil, kt)
	br.SRLock()
	if *GStore {
		ok := s.gstore.PutIfMissing(gotomic.Key(k), unsafe.Pointer(br))
		if !ok {
			dlog.Printf("Key already exists %v\n", k)
			return nil, EEXISTS
		}
	} else {
		chunk := s.store[k[0]]
		chunk.Lock()
		_, ok := chunk.rows[k]
		if ok {
			chunk.Unlock()
			dlog.Printf("Key already exists %v\n", k)
			return nil, EEXISTS
		}
		chunk.rows[k] = br
		chunk.Unlock()
	}
	return br, nil
}

func Validate(c *Coordinator, s *Store, nkeys int, nproducts int, val []int32, n int) bool {
	good := true
	dlog.Printf("Validate start, store at %x\n", c.GetEpoch())
	zero_cnt := 0
	for j := 0; j < nproducts; j++ {
		var x int32
		k := ProductKey(j)
		v, err := s.getKey(k, nil)
		if err != nil {
			if val[j] != 0 {
				fmt.Printf("Validating key %v failed; store: none should have: %v\n", k, val[j])
				dlog.Printf("Validating key %v failed; store: none should have: %v\n", k, val[j])
				good = false
			}
			continue
		}
		x = v.Value().(int32)
		if x != val[j] {
			dlog.Printf("Validating key %v failed; store: %v should have: %v\n", k, x, val[j])
			good = false
		}
		if x == 0 {
			//dlog.Printf("Saying x is zero %v %v\n", x, zero_cnt)
			zero_cnt++
		}
	}
	if zero_cnt == nproducts && n > 10 {
		fmt.Printf("Bad: all zeroes!\n")
		dlog.Printf("Bad: all zeroes!\n")
		good = false
	}
	dlog.Printf("Done validating\n")
	if !good {
		fmt.Printf("Validating failed\n")
	}
	return good
}

func compute(w *Worker, txn int) (int64, int64) {
	var total int64
	var sum int64
	var i int64
	for i = 0; i < TIMES; i++ {
		total = total + w.times[txn][i]
		sum = sum + (w.times[txn][i] * i)
	}
	total = total + w.tooLong[txn]
	sum = sum + w.tooLong[txn]*10000000
	var x99 int64 = int64(float64(total) * .99)
	var y99 int64
	var v99 int64
	var buckets [TIMES / 1000]int64
	for i = 0; i < TIMES; i++ {
		buckets[i/1000] += w.times[txn][i]
		y99 = y99 + w.times[txn][i]
		if y99 >= x99 {
			v99 = i
			break
		}
	}
	if total == 0 {
		log.Fatalf("No latency recorded\n")
	}
	dlog.Printf("%v avg: %v us; 99: %v us, x99: %v, sum: %v, total: %v \n", txn, sum/total, v99, x99, sum, total)

	var one int64
	var ten int64
	var hundred int64
	var more int64
	for i = 0; i < TIMES/1000; i++ {
		if i == 0 {
			one += buckets[i]
		} else if i < 10 {
			ten += buckets[i]
		} else if i < 100 {
			hundred += buckets[i]
		} else {
			more += buckets[i]
		}
	}

	fmt.Printf("Txn %v\n Less than 1ms: %v\n 1-10ms: %v\n 10-100ms: %v\n 100ms-10s: %v\n Greater than 10s: %v\n", txn, one, ten, hundred, more, w.tooLong[txn])
	total_time_in_ms := (one/2 + 5*ten + 55*100 + 15000*more)
	fmt.Printf("Rough total time in ms: %v\n", total_time_in_ms)
	return sum / total, v99
}

func (b *Buy) Init(np, nb, nw, rr, ngo int, ncrr, zipfd float64) {
	b.nproducts = np
	b.nbidders = nb
	b.nworkers = nw
	b.ngo = ngo
	b.read_rate = rr
	b.ncontended_rate = int(ncrr * float64(rr))
	b.validate = make([]int32, nb)
	b.sp = uint32(nb / nw)
	if zipfd != -1 {
		b.z = make([]*ddtxn.Zipf, nw)
		for i := 0; i < nw; i++ {
			r := rand.New(rand.NewSource(int64(i * 38748767)))
			b.z[i] = ddtxn.NewZipf(r, zipfd, 1, uint64(b.nbidders-1))
		}
	}
	b.zipfd = zipfd
	dlog.Printf("Read rate %v, not contended: %v\n", b.read_rate, b.ncontended_rate)
}

func (s *Store) getKey(k Key, ld *gotomic.LocalData) (*BRecord, error) {
	if len(k) == 0 {
		debug.PrintStack()
		log.Fatalf("[store] getKey(): Empty key\n")
	}
	if *GStore {
		var x unsafe.Pointer
		var ok bool
		hc, present := s.hash_codes[k]
		if ld == nil {
			x, ok = s.gstore.Get(gotomic.Key(k))
		} else if !present {
			x, ok = s.gstore.GetHC(gotomic.Key(k).HashCode(), gotomic.Key(k), ld)
		} else {
			x, ok = s.gstore.GetHC(hc, gotomic.Key(k), ld)
		}
		if !ok {
			dlog.Printf("Not in hash map. %v %v %v\n", k, x, ok)
			return nil, ENOKEY
		} else {
			if x == nil {
				fmt.Printf("Nil brecord! %v\n", k)
			}
			return (*BRecord)(x), nil
		}
	}
	if !*UseRLocks {
		x, err := s.getKeyStatic(k)
		return x, err
	}
	chunk := s.store[k[0]]
	if chunk == nil {
		log.Fatalf("[store] Didn't initialize chunk for key %v byte %v\n", k, k[0])
	}
	chunk.RLock()
	vr, ok := chunk.rows[k]
	if !ok || vr == nil {
		chunk.RUnlock()
		return vr, ENOKEY
	}
	chunk.RUnlock()
	return vr, nil
}

func (c *Coordinator) Process() {
	tm := time.NewTicker(time.Duration(*PhaseLength) * time.Millisecond).C

	// More frequently, check if the workers are demanding a phase
	// change due to long stashed queue lengths.
	check_trigger := time.NewTicker(time.Duration(*PhaseLength) * time.Microsecond * 10).C

	for {
		select {
		case x := <-c.Done:
			if *SysType == DOPPEL && c.n > 1 && c.Workers[0].store.any_dd {
				c.IncrementEpoch(true)
			}
			for i := 0; i < c.n; i++ {
				c.Workers[i].done <- true
			}
			x <- true
			return
		case <-tm:
			if *SysType == DOPPEL && c.n > 1 {
				c.IncrementEpoch(false)
			}
		case <-check_trigger:
			if *SysType == DOPPEL && c.n > 1 {
				x := atomic.LoadInt32(&c.trigger)
				if x == int32(c.n) {
					Nfast++
					atomic.StoreInt32(&c.trigger, 0)
					c.IncrementEpoch(true)
				}
			}
		case <-c.Accelerate:
			if *SysType == DOPPEL && c.n > 1 {
				dlog.Printf("Accelerating\n")
				c.IncrementEpoch(true)
			}
		}
	}
}

func TestBasic(t *testing.T) {
	s := NewStore()
	c := NewCoordinator(1, s)
	w := c.Workers[0]
	s.CreateKey(ProductKey(4), int32(0), SUM)
	s.CreateKey(ProductKey(5), int32(0), WRITE)
	s.CreateKey(UserKey(1), "u1", WRITE)
	s.CreateKey(UserKey(2), "u2", WRITE)
	s.CreateKey(UserKey(3), "u3", WRITE)
	tx := Query{TXN: D_BUY, K1: UserKey(1), A: int32(5), K2: ProductKey(4), W: nil, T: 0}

	r, err := w.One(tx)
	_ = err
	// Fresh read test
	tx = Query{TXN: D_READ_ONE, K1: ProductKey(4), W: make(chan struct {
		R *Result
		E error
	}), T: 0}
	r, err = w.One(tx)
	dlog.Printf("[test] Returned from one\n")
	if r.V.(int32) != 5 {
		t.Errorf("Wrong answer %v\n", r)
	}
}

// TODO: Check and see if I need more tx.MaybeWrite()s
func StoreBidTxn(t Query, tx ETransaction) (*Result, error) {
	var r *Result = nil
	user := t.U1
	item := t.U2
	price := int32(t.U3)
	if price < 0 {
		log.Fatalf("price %v %v", price, t.U3)
	}
	// insert bid
	n := tx.UID('b')
	bid_key := BidKey(n)
	bid := &Bid{
		ID:     uint64(n),
		Item:   item,
		Bidder: user,
		Price:  price,
	}
	tx.Write(bid_key, bid, WRITE)

	// update # bids per item
	err := tx.WriteInt32(NumBidsKey(item), 1, SUM)
	if err != nil {
		tx.RelinquishKey(n, 'b')
		tx.Abort()
		dlog.Printf("StoreBidTxn(): Couldn't write numbids for item %v; %v\n", item, err)
		return nil, err
	}

	// update max bid?
	high := MaxBidKey(item)
	tx.MaybeWrite(high)
	err = tx.WriteInt32(high, price, MAX)
	if err != nil {
		tx.RelinquishKey(n, 'b')
		dlog.Println("Aborting because of max")
		tx.Abort()
		dlog.Printf("StoreBidTxn(): Couldn't write maxbid for item %v; %v\n", item, err)
		return nil, err
	}
	bidder := MaxBidBidderKey(item)
	err = tx.WriteOO(bidder, price, user, OOWRITE)
	if err != nil {
		tx.RelinquishKey(n, 'b')
		dlog.Println("Aborting because of max oowrite")
		tx.Abort()
		dlog.Printf("StoreBidTxn(): Couldn't write maxbidder for item %v; %v\n", item, err)
		return nil, err
	}
	// add to item's bid list
	e := Entry{int(bid.Price), bid_key, 0}
	err = tx.WriteList(BidsPerItemKey(item), e, LIST)
	if err != nil {
		tx.RelinquishKey(n, 'b')
		tx.Abort()
		dlog.Printf("StoreBidTxn(): Error adding to bids per item key %v! %v\n", item, err)
		return nil, err
	}
	if tx.Commit() == 0 {
		tx.RelinquishKey(n, 'b')
		dlog.Printf("StoreBidTxn(): Abort item %v\n", item)
		return r, EABORT
	}

	if *Allocate {
		r = &Result{uint64(n)}
		// dlog.Printf("User %v Bid on item %v for %v dollars\n", user, item, price)
	}
	return r, nil
}