Golang implements redis (10): local atomic transactions

Keep creating , Accelerate growth ！ This is my participation 「 Nuggets day new plan · 6 Yuegengwen challenge 」 Of the 27 God , Click to see the event details

In order to support the atomic execution of multiple commands Redis Provides a transaction mechanism . Redis The official document says that the transaction has the following two important guarantees ：

• A transaction is a separate isolation operation ： All commands in the transaction are serialized 、 To execute in order . Transaction is in the process of execution , Will not be interrupted by command requests from other clients .
• A transaction is an atomic operation ： The commands in the transaction are either all executed , Or none of it

redis.io/docs/manual…

We may encounter two types of errors when using transactions :

1. Syntax error during command queue
2. A runtime error occurred during command execution , For example, yes. string Type of key Conduct lpush operation

When encountering syntax errors Redis Will abort the order to join the queue and discard the transaction . When encountering a runtime error Redis Only an error will be reported, and then continue to execute the remaining commands in the transaction , Transactions are not rolled back like most databases . Regarding this ,Redis The official explanation is ：

Redis The command will only fail because of the wrong Syntax （ And these problems can't be found when entering the team ）, Or the command is used on the wrong type of key ： That means , From a practical point of view , Failed commands are caused by programming errors , And these errors should be found in the development process , Not in the production environment .

Because there is no need to support rollback , therefore Redis The interior can be kept simple and fast . There is a view that Redis The practice of dealing with affairs produces bug , However, it should be noted that , In general , Rollback doesn't solve the problem of programming errors . for instance , If you wanted to pass INCR Command to add the value of the key to 1 , But accidentally added 2 , Or the wrong type of key is executed INCR , There is no way to handle these situations with rollback . Given that there is no mechanism to avoid mistakes made by programmers themselves , And this kind of error usually doesn't appear in a production environment , therefore Redis It's simpler 、 A faster way to handle transactions without rollback .

emmmm, Next, let's try in Godis The implementation has atomicity 、 Isolated Affairs .

The atomicity of transactions has two characteristics ：1. Transaction execution cannot be by other transactions ( Threads ) Insert 2. The transaction is either completely successful or not executed at all , There is no status of partial success Transaction isolation refers to whether the results of operations in a transaction are visible to other concurrent transactions . because KV There is no unreal reading problem in the database , Therefore, we need to avoid dirty reading and non repeatability problems .

Analysis of transaction mechanism

lock

And Redis Our single threaded engine is different godis Our storage engine is parallel , Therefore, it is necessary to design a locking mechanism to ensure the atomicity and isolation when executing multiple commands .

We are Realize memory database Mentioned in the article :

To implement a regular command, you need to provide 3 A function :

• ExecFunc Is the function that actually executes the command
• PrepareFunc stay ExecFunc Pre execution , Responsible for analyzing what the command line reads and writes key Easy to lock
• UndoFunc Used only in transactions , Be responsible for preparing undo logs In case of errors during transaction execution, rollback is required .

Among them PrepareFunc It will analyze the command line and return the data to be read and written key, With prepareMSet For example :

// return writtenKeys, readKeys
func prepareMSet(args [][]byte) ([]string, []string) {
	size := len(args) / 2
	keys := make([]string, size)
	for i := 0; i < size; i++ {
		keys[i] = string(args[2*i])
	}
	return keys, nil
}

combination Realize memory database Mentioned in LockMap You can complete locking . Due to other collaborative processes, relevant information cannot be obtained key Therefore, it is impossible to insert the lock into the transaction , So we implemented the non insertable feature in atomicity .

Business needs to put all key Complete locking at one time , Unlock only when the transaction is committed or rolled back . You can't use one key Just add the lock once and unlock it after use , This method can lead to dirty reading ：

Time	Business 1	Business 2
t1	lock key A
t2	modify key A
t3	Unlock key A
t4		lock key A
t4		Read key A
t5		Unlock key A
t6	Submit

As shown in the figure above t4 moment , Business 2 Read the business 1 Uncommitted data , A dirty read exception occurred .

Roll back

In order to roll back the transaction when a runtime error is encountered （ Atomicity ）, There are two rollback methods available ：

• Save the modified value, When rolling back, use the modified value To cover
• Use the rollback command to undo the impact of the original command . for instance ： key A The original value is 1, Called Incr A And then it became 2, We can do it again Set A 1 Order to revoke incr command .

In order to save memory, we finally chose the second scheme . such as HSet The command only needs another HSet take field Change back to the original value , If saving is adopted value We need to save the whole HashMap. There are similar situations LPushRPop Wait for the order .

Some commands may require multiple commands to roll back , For example, rollback Del It is not only necessary to restore the corresponding key-value You need to recover TTL data . perhaps Del The command deleted multiple key when , You also need multiple commands to roll back . To sum up, we give UndoFunc The definition of ：

// UndoFunc returns undo logs for the given command line
// execute from head to tail when undo
type UndoFunc func(db *DB, args [][]byte) []CmdLine

We can roll back any operation rollbackGivenKeys Take an example to illustrate , Of course use rollbackGivenKeys High cost of , When possible, try to achieve targeted undo log.

func rollbackGivenKeys(db *DB, keys ...string) []CmdLine {
	var undoCmdLines [][][]byte
	for _, key := range keys {
		entity, ok := db.GetEntity(key)
		if !ok {
			//  It didn't exist  key  Delete 
			undoCmdLines = append(undoCmdLines,
				utils.ToCmdLine("DEL", key),
			)
		} else {
			undoCmdLines = append(undoCmdLines,
				utils.ToCmdLine("DEL", key), //  First put the new  key  Delete the 
				aof.EntityToCmd(key, entity).Args, //  hold  DataEntity  Serialize into command line 
				toTTLCmd(db, key).Args,
			)
		}
	}
	return undoCmdLines
}

Let's take a look EntityToCmd, It's very easy to understand ：

func EntityToCmd(key string, entity *database.DataEntity) *protocol.MultiBulkReply {
	if entity == nil {
		return nil
	}
	var cmd *protocol.MultiBulkReply
	switch val := entity.Data.(type) {
	case []byte:
		cmd = stringToCmd(key, val)
	case *List.LinkedList:
		cmd = listToCmd(key, val)
	case *set.Set:
		cmd = setToCmd(key, val)
	case dict.Dict:
		cmd = hashToCmd(key, val)
	case *SortedSet.SortedSet:
		cmd = zSetToCmd(key, val)
	}
	return cmd
}

var hMSetCmd = []byte("HMSET")

func hashToCmd(key string, hash dict.Dict) *protocol.MultiBulkReply {
	args := make([][]byte, 2+hash.Len()*2)
	args[0] = hMSetCmd
	args[1] = []byte(key)
	i := 0
	hash.ForEach(func(field string, val interface{}) bool {
		bytes, _ := val.([]byte)
		args[2+i*2] = []byte(field)
		args[3+i*2] = bytes
		i++
		return true
	})
	return protocol.MakeMultiBulkReply(args)
}

Watch

Redis Watch The command is used to monitor a ( Or more ) key , If before the transaction is executed ( Or these ) key Altered by other orders , Then the transaction will be abandoned .

Realization Watch The core of the command is to find key Is it changed , We use a simple and reliable version number scheme ： For each key Store a version number , Description of version number change key It was modified :

// database/single_db.go
func (db *DB) GetVersion(key string) uint32 {
	entity, ok := db.versionMap.Get(key)
	if !ok {
		return 0
	}
	return entity.(uint32)
}

// database/transaciton.go
func Watch(db *DB, conn redis.Connection, args [][]byte) redis.Reply {
	watching := conn.GetWatching()
	for _, bkey := range args {
		key := string(bkey)
		watching[key] = db.GetVersion(key) //  Save the current version number  conn  In the object 
	}
	return protocol.MakeOkReply()
}

Compare version numbers before executing transactions ：

// database/transaciton.go
func isWatchingChanged(db *DB, watching map[string]uint32) bool {
	for key, ver := range watching {
		currentVersion := db.GetVersion(key)
		if ver != currentVersion {
			return true
		}
	}
	return false
}

Source guide

After understanding the transaction related mechanism , Let's take a look at the core code of transaction execution ExecMulti

func (db *DB) ExecMulti(conn redis.Connection, watching map[string]uint32, cmdLines []CmdLine) redis.Reply {
	//  Preparation stage 
	//  Use  prepareFunc  Get the transaction to read and write  key
	writeKeys := make([]string, 0) // may contains duplicate
	readKeys := make([]string, 0)
	for _, cmdLine := range cmdLines {
		cmdName := strings.ToLower(string(cmdLine[0]))
		cmd := cmdTable[cmdName]
		prepare := cmd.prepare
		write, read := prepare(cmdLine[1:])
		writeKeys = append(writeKeys, write...)
		readKeys = append(readKeys, read...)
	}
	watchingKeys := make([]string, 0, len(watching))
	for key := range watching {
		watchingKeys = append(watchingKeys, key)
	}
	readKeys = append(readKeys, watchingKeys...)
	//  About to read and write  key  And quilt  watch  Of  key  Lock together 
	db.RWLocks(writeKeys, readKeys)
	defer db.RWUnLocks(writeKeys, readKeys)

	//  Check to be  watch  Of  key  Has it changed 
	if isWatchingChanged(db, watching) { // watching keys changed, abort
		return protocol.MakeEmptyMultiBulkReply()
	}

	//  Execution phase 
	results := make([]redis.Reply, 0, len(cmdLines))
	aborted := false
	undoCmdLines := make([][]CmdLine, 0, len(cmdLines))
	for _, cmdLine := range cmdLines {
		//  Prepare before the command is executed  undo log,  Only in this way can we ensure, for example, that  decr  Roll back  incr  The implementation of the command can work normally 
		undoCmdLines = append(undoCmdLines, db.GetUndoLogs(cmdLine))
		result := db.execWithLock(cmdLine)
		if protocol.IsErrorReply(result) {
			aborted = true
			// don't rollback failed commands
			undoCmdLines = undoCmdLines[:len(undoCmdLines)-1]
			break
		}
		results = append(results, result)
	}
	//  Successful implementation 
	if !aborted { 
		db.addVersion(writeKeys...)
		return protocol.MakeMultiRawReply(results)
	}
	//  The transaction failed and rolled back 
	size := len(undoCmdLines)
	for i := size - 1; i >= 0; i-- {
		curCmdLines := undoCmdLines[i]
		if len(curCmdLines) == 0 {
			continue
		}
		for _, cmdLine := range curCmdLines {
			db.execWithLock(cmdLine)
		}
	}
	return protocol.MakeErrReply("EXECABORT Transaction discarded because of previous errors.")
}