Method Commit

tx.go:176–300

Commit commits the transaction, following these steps: 1. check the length of pendingWrites.If there are no writes, return immediately. 2. check if the ActiveFile has not enough space to store entry. if not, call rotateActiveFile function. 3. write pendingWrites to disk, if a non-nil error,return

()

//
// 6. Unlock the database and clear the db field.
func (tx *Tx) Commit() (err error) {
	defer func() {
		if err != nil {
			tx.handleErr(err)
		}

		tx.unlock()

		// Ensure the transaction is unregistered so active counts stay accurate
		tx.db.transactionMgr.UnregisterTx(tx.id)

		tx.db = nil

		tx.pendingWrites = nil
	}()

	if tx.isClosed() {
		return ErrCannotCommitAClosedTx
	}

	if tx.db == nil {
		tx.setStatusClosed()
		return ErrDBClosed
	}

	var curWriteCount int64

	// If the database is closing/closed, abort early to avoid touching released resources.
	if tx.db.statusMgr.isClosed() {
		return ErrDBClosed
	}

	if tx.db.opt.MaxWriteRecordCount > 0 {
		curWriteCount, err = tx.getNewAddRecordCount()
		if err != nil {
			return err
		}

		// judge all write records is whether more than the MaxWriteRecordCount
		if tx.db.RecordCount+curWriteCount > tx.db.opt.MaxWriteRecordCount {
			return ErrTxnExceedWriteLimit
		}
	}

	tx.setStatusCommitting()
	defer tx.setStatusClosed()

	writesBucketLen := len(tx.pendingBucketList)
	if tx.pendingWrites.size == 0 && writesBucketLen == 0 {
		return nil
	}

	buff := tx.allocCommitBuffer()
	defer tx.db.commitBuffer.Reset()

	var records []*core.Record

	pendingWriteList := tx.pendingWrites.toList()