// tx inserts/updates a set of resources, then deletes a set of resources, then
// increases the cache's version number atomically if the cache is actually
// changed.
// The version after updating the set is returned.
196047
func (c *Cache) tx(typeURL string, upsertedResources map[string]proto.Message, deletedNames []string, force bool) (version uint64, updated bool, revert ResourceMutatorRevertFunc) {
	c.locker.Lock()
	defer c.locker.Unlock()

	cacheIsUpdated := force
	newVersion := c.version + 1

	cacheLog := log.WithFields(logrus.Fields{
		logfields.XDSTypeURL:     typeURL,
		logfields.XDSVersionInfo: newVersion,
	})

	cacheLog.Debugf("preparing new cache transaction: upserting %d entries, deleting %d entries",
		len(upsertedResources), len(deletedNames))

	// The parameters to pass to tx in revertFunc.
	var revertUpsertedResources map[string]proto.Message
	var revertDeletedNames []string

	k := cacheKey{
		typeURL: typeURL,
	}

	v := cacheValue{
		lastModifiedVersion: newVersion,
	}

	for name, value := range upsertedResources {
		k.resourceName = name
		oldV, found := c.resources[k]
		// If the value is unchanged, don't update the entry, to preserve its
		// lastModifiedVersion. This allows minimizing the frequency of
		// responses in GetResources.
		// Calling proto.Message.String is not very cheap, but we assume that
		// the reduced churn between the clients and the server is worth it.
		if !found || oldV.resource.String() != value.String() {
			if found {
				cacheLog.WithField(logfields.XDSResourceName, name).Debug("updating resource in cache")

				if revertUpsertedResources == nil {
					revertUpsertedResources = make(map[string]proto.Message, len(upsertedResources)+len(deletedNames))
				}
				revertUpsertedResources[name] = oldV.resource
			} else {
				cacheLog.WithField(logfields.XDSResourceName, name).Debug("inserting resource into cache")

				if revertDeletedNames == nil {
					revertDeletedNames = make([]string, 0, len(upsertedResources))
				}
				revertDeletedNames = append(revertDeletedNames, name)
			}
			cacheIsUpdated = true
			v.resource = value
			c.resources[k] = v
		}
		if force {
			cacheIsUpdated = true
		}
	}

	for _, name := range deletedNames {
		k.resourceName = name
		oldV, found := c.resources[k]
		if found {
			cacheLog.WithField(logfields.XDSResourceName, name).
				Debug("deleting resource from cache")

			if revertUpsertedResources == nil {
				revertUpsertedResources = make(map[string]proto.Message, len(upsertedResources)+len(deletedNames))
			}
			revertUpsertedResources[name] = oldV.resource

			cacheIsUpdated = true
			delete(c.resources, k)
		}
		if force {
			cacheIsUpdated = true
		}
	}

	if cacheIsUpdated {
		cacheLog.Debug("committing cache transaction and notifying of new version")
		c.version = newVersion
		c.NotifyNewResourceVersionRLocked(typeURL, c.version)
	} else {
		cacheLog.Debug("cache unmodified by transaction; aborting")
	}

	revertFunc := func(force bool) (version uint64, updated bool) {
		version, updated, _ = c.tx(typeURL, revertUpsertedResources, revertDeletedNames, force)
		return
	}

	return c.version, cacheIsUpdated, revertFunc
}
180948
func MaybeCommitTx(ctx context.Context, storage interface{}) error {
	txnStorage, transactional := storage.(Transactional)
	if transactional {
		return txnStorage.Commit(ctx)
	} else {
		return nil
	}
}
320060
func (t *Tx) Init(tx, rx []byte) {
	t.tx.Init(tx)
	t.rx.Init(rx)
}
1337434
func (tm *transactionManager) transaction(
	key string, keyType int) *transaction {

	sm := tm.transactions
	if keyType == 1 {
		sm = tm.index
	}

	v, ok := sm.Get(key)
	if !ok {
		return nil
	}

	return v.(*transaction)
}
889253
func (t *txn) Delete(key string) keyval.BytesTxn {
	t.updates = append(t.updates, &update{
		key:   []byte(key),
		value: nil,
	})
	return t
}
357394
func update(ctx context.Context, db transactor, fn func(*bolt.Tx) error) error {
	tx, ok := ctx.Value(transactionKey{}).(*bolt.Tx)
	if !ok {
		return db.Update(fn)
	} else if !tx.Writable() {
		return errors.Wrap(bolt.ErrTxNotWritable, "unable to use transaction from context")
	}
	return fn(tx)
}
1017061
func (pgb *ChainDB) Transaction(txHash string) ([]*dbtypes.Tx, error) {
	ctx, cancel := context.WithTimeout(pgb.ctx, pgb.queryTimeout)
	defer cancel()
	_, dbTxs, err := RetrieveDbTxsByHash(ctx, pgb.db, txHash)
	return dbTxs, pgb.replaceCancelError(err)
}
326127
func txKey(tx types.Tx) [sha256.Size]byte {
	return sha256.Sum256(tx)
}
22085
func (tx *Tx) Hash() binary.HexBytes {
	if tx == nil {
		return nil
	}
	if tx.txHash == nil {
		return tx.Rehash()
	}
	return tx.txHash
}
152159
func (qre *QueryExecutor) txFetch(conn *TxConnection, parsedQuery *sqlparser.ParsedQuery, bindVars map[string]*querypb.BindVariable, extras map[string]sqlparser.Encodable, buildStreamComment string, wantfields, record bool) (*sqltypes.Result, error) {
	sql, _, err := qre.generateFinalSQL(parsedQuery, bindVars, extras, buildStreamComment)
	if err != nil {
		return nil, err
	}
	qr, err := qre.execSQL(conn, sql, wantfields)
	if err != nil {
		return nil, err
	}
	// Only record successful queries.
	if record {
		conn.RecordQuery(sql)
	}
	return qr, nil
}
1054294
func (b *bucket) Tx() walletdb.ReadWriteTx {
	return &transaction{
		(*bbolt.Bucket)(b).Tx(),
	}
}
1171728
func (t *Transaction) Commit() error {
	if !t.closed {
		t.closed = true
		if t.dbmap.logger != nil {
			now := time.Now()
			defer t.dbmap.trace(now, "commit;")
		}
		return t.tx.Commit()
	}

	return sql.ErrTxDone
}
461120
func (tx *Tx) write() error {
	// Sort pages by id.
	pages := make(pages, 0, len(tx.pages))
	for _, p := range tx.pages {
		pages = append(pages, p)
	}
	// Clear out page cache early.
	tx.pages = make(map[pgid]*page)
	sort.Sort(pages)

	// Write pages to disk in order.
	for _, p := range pages {
		size := (int(p.overflow) + 1) * tx.db.pageSize
		offset := int64(p.id) * int64(tx.db.pageSize)

		// Write out page in "max allocation" sized chunks.
		ptr := (*[maxAllocSize]byte)(unsafe.Pointer(p))
		for {
			// Limit our write to our max allocation size.
			sz := size
			if sz > maxAllocSize-1 {
				sz = maxAllocSize - 1
			}

			// Write chunk to disk.
			buf := ptr[:sz]
			if _, err := tx.db.ops.writeAt(buf, offset); err != nil {
				return err
			}

			// Update statistics.
			tx.stats.Write++

			// Exit inner for loop if we've written all the chunks.
			size -= sz
			if size == 0 {
				break
			}

			// Otherwise move offset forward and move pointer to next chunk.
			offset += int64(sz)
			ptr = (*[maxAllocSize]byte)(unsafe.Pointer(&ptr[sz]))
		}
	}

	// Ignore file sync if flag is set on DB.
	if !tx.db.NoSync || IgnoreNoSync {
		if err := fdatasync(tx.db); err != nil {
			return err
		}
	}

	// Put small pages back to page pool.
	for _, p := range pages {
		// Ignore page sizes over 1 page.
		// These are allocated using make() instead of the page pool.
		if int(p.overflow) != 0 {
			continue
		}

		buf := (*[maxAllocSize]byte)(unsafe.Pointer(p))[:tx.db.pageSize]

		// See https://go.googlesource.com/go/+/f03c9202c43e0abb130669852082117ca50aa9b1
		for i := range buf {
			buf[i] = 0
		}
		tx.db.pagePool.Put(buf)
	}

	return nil
}
1202801
func (t *Tx) Commit() ([]ProcessedPacket, *ReplaySet, error) {
	if t.batch.IsCommitted {
		return t.packets, t.batch.ReplaySet, nil
	}

	rs, err := t.router.log.PutBatch(t.batch)

	return t.packets, rs, err
}
263198
func (d *AddressCacheItem) Transactions(N, offset int, txnView dbtypes.AddrTxnViewType) (interface{}, *BlockID, error) {
	if offset < 0 || N < 0 {
		return nil, nil, fmt.Errorf("invalid offset (%d) or N (%d)", offset, N)
	}

	if d == nil {
		return nil, nil, fmt.Errorf("uninitialized AddressCacheItem")
	}

	d.mtx.RLock()
	defer d.mtx.RUnlock()
	merged, err := txnView.IsMerged()
	if err != nil {
		return nil, nil, fmt.Errorf("invalid transaction view: %v", txnView)
	}

	// Identify cache miss by nil rows.
	if d.rows == nil {
		// Cache miss is not an error, but rows return type must be consistent
		// with requested view for the sanity checking in AddressCache
		// (TransactionsCompact or TransactionsMerged).
		if merged {
			return []dbtypes.AddressRowMerged(nil), nil, nil
		}
		return []dbtypes.AddressRowCompact(nil), nil, nil
	}

	blockID := d.blockID()
	numRows := len(d.rows)

	if N == 0 || numRows == 0 || offset >= numRows {
		// Not a cache miss, just no requested or matching data.
		if merged {
			return []dbtypes.AddressRowMerged{}, blockID, nil
		}
		return []dbtypes.AddressRowCompact{}, blockID, nil
	}

	switch txnView {
	case dbtypes.AddrTxnAll:
		// []dbtypes.AddressRowCompact
		return addressRows(d.rows, N, offset), blockID, nil
	case dbtypes.AddrTxnCredit:
		return creditAddressRows(d.rows, N, offset), blockID, nil
	case dbtypes.AddrTxnDebit:
		return debitAddressRows(d.rows, N, offset), blockID, nil
	case dbtypes.AddrMergedTxn, dbtypes.AddrMergedTxnCredit, dbtypes.AddrMergedTxnDebit:
		// []dbtypes.AddressRowMerged
		return dbtypes.MergeRowsCompactRange(d.rows, N, offset, txnView), blockID, nil
	default:
		// This should already be caught by IsMerged err check.
		return nil, nil, fmt.Errorf("unrecognized address transaction view: %v", txnView)
	}
}
924893
func (t *Tx) Bytes() (tx, rx []byte) {
	return t.tx.Bytes(), t.rx.Bytes()
}
958502
func (tx *txn) Commit(ctx context.Context) error {
	ok, resp, _, err := tx.kv.Txn(tx.ops, nil)
	if err != nil {
		return err
	} else if !ok {
		return fmt.Errorf("transaction failed: %v", resp)
	}
	return nil
}