Distributed lock (3)–Distributed lock implementation based on Etcd

Preface

The distributed lock implemented based on Redis has strong performance and many functions. However, due to the characteristics of Redis itself, the distributed lock implemented based on Redis It may not be unique, so it cannot be applied in scenarios that require distributed locks to be absolutely unique. For this reason, distributed locks can only be implemented based on other services. However, the overall idea of their implementation is the same, but there are some differences in the calling methods. This article will introduce how to implement complete distribution based on Etcd style lock.

1. Why implement distributed locks based on Etcd

Currently, in addition to Redis, common distributed lock implementations include Etcd and Zookeeper. Since the consistency of Redis is only It satisfies AP, so the distributed lock implemented based on Redis cannot guarantee absolute uniqueness, and the consistency of Etcd and Zookeeper both satisfies CP. Data consistency is ensured through similar consistency algorithms, so that distributed locks based on Etcd and Zookeeper can guarantee absolute uniqueness. The functions of Etcd and Zookeeper are similar, but in some aspects Etcd even surpasses ZooKeeper, such as Raft protocol used by >Etcd is simpler and easier to understand than the Zab protocol used by ZooKeeper, and its performance is also better than Zookeeper is high, so Etcd is usually chosen as the implementation of distributed locks.

The comparison between the officially provided Etcd and other components of the same type is as follows:

You can see from the figure that Etcd is slightly better than other components. In addition, it provides Watch, Lease, Revision and The Prefix mechanism makes it very convenient to implement distributed locks based on Etcd. At the same time, it provides the corresponding API through HTTP/gRPC, so its client implementation is also more convenient. Of course, if the current system relies on Zookepeer, then you must still choose Zookerpeer, unless there are performance requirements.

2. Implementation of distributed lock

As mentioned earlier, Etcd‘s Watch, Lease, Revision and Prefix mechanisms grant EtcdThe capabilities of distributed locks. To implement distributed locks, you need to understand their functions. Their functions are as follows:

• Lease mechanism: It is similar to expiration in Redis. Etcd can set the lease time for Key-Value through Lease (That is, the expiration time), Key-Value will be deleted when the lease expires.
• Revision mechanism: Etcd will assign a version to each Key-Value, and when updating Key-Value, its version will also occur Variety. For example, when Key1 is inserted, its version is 0, and when Key2 is subsequently inserted, its version is 1. Through this mechanism, the order of write operations can be known.
• Prefix mechanism: Etcd provides the ability to perform the same operation on Keys with the same prefix through the Prefix mechanism, such as a lock named /lock. When two clients perform write operations, the Key actually written are key1="/lock/UUID1" and key2=/lock/ UUID2"", where UUID is the meaning of token, which is used to ensure the uniqueness of each lock. However, unlike the distributed lock implemented in Redis, both locks in Etcd will be written successfully, but the returned Revision It is different. The application needs to query the data through the /lock/ prefix, and then the results of key1 and key2 and Revision, and then use the result of Revision to determine which Key is the one that acquired the lock.
• Watch mechanism: Etcd‘s Watch mechanism can monitor a batch of Key in batches, when the monitored Key changes. , the client will receive a notification. When implementing distributed locks, if the lock grab fails, the Key list returned by the Prefix mechanism can be used to obtain a Revision that is smaller than itself and has the smallest difference. Key (called Pre-Key), monitors Pre-Key, because only when it releases the lock can it obtain the lock. If If you listen to the deletion event of Pre-Key, it means that Pre-Key has been released and you already hold the lock.

After understanding these mechanisms, you can find that through these mechanisms, you can quickly implement a simple distributed lock based on Etcd, as shown below:

The red square in the picture represents the Etcd server, the blue square represents the Etcd client, and the white square represents the current Etcd server. The data is divided into 4 steps in the figure, separated by dotted lines. The first is step 1. client1 and client2 will push /demo/{uuid} to put command. code>Etcd.

Then comes step 2, where the key-value pair information starting with /demo is obtained through Prefix.

Next is step 3. This step will determine whether the smallest Revision version in the key-value pair information returned in step 2 is itself. If it is, it means that it has obtained the lock. If not, it will pass The Watch mechanism monitors the deletion events of key-value pairs that Revision is 1 smaller than itself.

The last step is step 4. When client1 that acquired the lock has completed the task and deleted /demo/uuid1, then Etcd will notify client2Client, it has acquired the lock and can perform tasks.

Through these mechanisms, you can find that the distributed lock implemented based on Etcd is very convenient. At the same time, due to the role of Revision, the implemented distributed lock has fair lock by default. function, let’s start with the implementation of Etcd distributed lock. The first is the logic of acquiring the lock:

class Lock(object):
    def __init__(self, name: str, client: aetcd.Client, ttl: int) -> None:
        self._prefix = name.encode()
        self._name = self._prefix + b'/' + uuid1().hex.encode()
        self._client = client
        self._ttl = ttl
        self.lease = None
        self._watch_dog: Optional[asyncio.Task]= None

    async def acquire(self) -> bool:
        # 1.Create renewal
        self.lease = await self._client.lease(self._ttl)
        # 2. Write data
        create_result = await self._client.put(self._name, b"", lease=self.lease)
        self_revision = create_result.header.revision
        # 3. Query data
        range_result = await self._client.get_prefix(self._prefix, sort_target="create")
        create_revision = range_result.kvs[0].create_revision
        if create_revision == self_revision:
            return True

        # 4. Query the key that needs to be monitored
        watch_key = b""
        for index, item in enumerate(range_result.kvs):
            if item.create_revision == self_revision:
                watch_key = range_result.kvs[index - 1].key

        # 5. Monitor Key
        watch = await self._client.watch(
            watch_key,
            kind=aetcd.rtypes.EventKind.DELETE,
        )
        async for event in watch:
            return True

The sample code will be initialized first, and the name parameter passed in will be named prefix, and the name in the lock will be The splicing of prefix and token will make it easier to find the contents of the same lock through the Prefix mechanism, and also allow the Watch mechanism to be used independently To monitor the status of the corresponding lock and prevent other locks from releasing their own locks.

After initialization, it is the implementation of acquiring the lock. The first step to acquire the lock is to create the relevant method of renewal. The renewal mechanism of Etcd is different from the expiration time of Redis. Yes, Etcd needs to create a lease first, and then use the lease to bind the corresponding Key. If the lease expires, then the Key< corresponding to the lease /code> will expire at the same time, so you need to create a lease first, and then bind it to the Key in the put method. Next, the steps are the same as in the picture. The data will be pushed to Etcd first, and then the data of all locks will be obtained according to get_prefix. It should be noted that needs to be defined here. sort_target is create, so that the returned result set is sorted according to the version number created by Key. This enables you to quickly determine whether you have successfully acquired the lock, and it is also easy to infer the value of the previous Key. Finally, the watch method is used to monitor the deletion event of the previous Key. If the deletion event is received, it means that the lock is obtained.

class Lock(object):
    ...
    async def __aenter__(self) -> "Lock":
        self._watch_dog = asyncio.create_task(self.watch_dog())
        await self.acquire()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        if self._watch_dog and not self._watch_dog.done():
            self._watch_dog.cancel()
        await self.release()

    async def watch_dog(self):
        while True:
            if self.lease is not None:
                try:
                    await self.refresh()
                except ValueError:
                    pass
            await asyncio.sleep(self._ttl / 3)
    async def refresh(self):
        """Refresh the time to live on this lock."""
        if self.lease is not None:
            return await self.lease.refresh()

        raise ValueError(f'no lease associated with this lock: {self._name!r}')

    async def release(self) -> None:
        await self._client.delete(self._name)

def my_print(msg: str):
    print(f"Timestamp:{time.time()} Task:{id(asyncio.current_task())}, {msg}")

async def sub(client: aetcd.Client, cnt: int) -> None:
    my_print(f"cnt:{cnt} wait")
    async with Lock("demo", client, 1):
        my_print(f"cnt:{cnt} run")
        await asyncio.sleep(2)
    my_print(f"cnt:{cnt} done")

async def main() -> None:
    client = aetcd.Client()
    await client.delete_prefix(b"demo")
    tasks = []
    tasks.append(asyncio.create_task(sub(client, 1)))
    await asyncio.sleep(0.1)
    tasks.append(asyncio.create_task(sub(client, 2)))
    await asyncio.sleep(0.1)
    tasks.append(asyncio.create_task(sub(client, 3)))
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(main())

Timestamp:1694361604.7456405 Task:140007816152864, cnt:1 wait
Timestamp:1694361604.7522662 Task:140007816152864, cnt:1 run
Timestamp:1694361604.846592 Task:140007816153184, cnt:2 wait
Timestamp:1694361604.947299 Task:140007816154624, cnt:3 wait
Timestamp:1694361606.7529566 Task:140007816152864, cnt:1 done
Timestamp:1694361606.7571979 Task:140007816153184, cnt:2 run
Timestamp:1694361608.7577982 Task:140007816153184, cnt:2 done
Timestamp:1694361608.7609887 Task:140007816154624, cnt:3 run
Timestamp:1694361610.7621803 Task:140007816154624, cnt:3 done