Python: building a distributed system with PySyncObj / Wargaming's blog

Imagine you have a class:

class MyCounter(object):
    def __init__(self):
        self.__counter = 0
    def incCounter(self):
        self.__counter += 1
    def getCounter(self):
        return self.__counter

And you want to make it distributed. Just inherit it from SyncObj (passing it a list of servers with which you want to synchronize) and mark the @replicated decorator with all the methods that change the internal state of the class:

class MyCounter(SyncObj):
    def __init__(self):
        super(MyCounter, self).__init__('serverA:4321', ['serverB:4321', 'serverC:4321'])
        self.__counter = 0
    @replicated
    def incCounter(self):
        self.__counter += 1
    def getCounter(self):
        return self.__counter

PySyncObj will automatically provide replication of your class between servers, fault tolerance (everything will work as long as more than half of the servers are alive), and also (if necessary) an asynchronous dump of contents to disk.
On the basis of PySyncObj, you can build various distributed systems, for example, a distributed mutex, decentralized databases, billing systems, and other similar things. All those where reliability and fault tolerance come first.

general description

PySyncObj uses the Raft algorithm for replication . Raft is a simple consensus building algorithm for a distributed system. Raft was developed as an easier replacement for the Paxos algorithm. In short, the raft algorithm works as follows. Among all the nodes, a leader is selected that pings the remaining nodes after a certain period of time. Each node selects a random period of time that it will wait for a ping from the leader. When the waiting time ends, but the ping from the leader has not arrived, the node believes that the leader has fallen and sends a message to the other nodes, in which it says that he himself became the leader. With a successful set of circumstances, this all ends (the other nodes agree). And in the event that two nodes want to become leaders at the same time, the leader selection procedure is repeated (but with other random values ​​of the waiting time). You can learn more about choosing a leader by looking at the visualization , or by reading a scientific article .

Once a leader is identified, he is responsible for maintaining a distributed journal. All actions that change the state of the system are written to the distributed journal. The action is applied to the system only if most nodes confirm receipt of the record - this ensures consistency. In order for the number of entries in the distributed log not to grow indefinitely, an operation called log compaction periodically occurs. The current log is thrown out, and instead the serialized state of the system at the current moment begins to be stored.

In order not to lose content (for example, when all servers are turned off at all), it must be periodically saved to disk. Since the amount of data can be very large, the content is stored asynchronously. To simultaneously be able to work with data and simultaneously save it to disk, PySyncObj uses CopyOnWrite through the fork of the process. After the fork, the parent and child processes share memory. Copying data is carried out by the operating system only if you try to overwrite this data.

PySyncObj is implemented entirely in Python (supported by Python 2 and Python 3) and does not use any external libraries. Networking is done using select or poll, depending on the platform.

Examples

And now a few examples.

Key-value storage

class KVStorage(SyncObj):
    def __init__(self, selfAddress, partnerAddrs, dumpFile):
        conf = SyncObjConf(
            fullDumpFile=dumpFile,
        )
        super(KVStorage, self).__init__(selfAddress, partnerAddrs, conf)
        self.__data = {}
    @replicated
    def set(self, key, value):
        self.__data[key] = value
    @replicated
    def pop(self, key):
        self.__data.pop(key, None)
    def get(self, key):
        return self.__data.get(key, None)

In general, everything is the same as with the counter. In order to periodically save data to disk, create SyncObjConf and transfer fullDumpFile to it.

Callback

PySyncObj supports callbacks - you can create methods that return some values, they will be automatically thrown into callback:

class Counter(SyncObj):
    def __init__(self):
        super(Counter, self).__init__('localhost:1234', ['localhost:1235', 'localhost:1236'])
        self.__counter = 0
    @replicated
    def incCounter(self):
        self.__counter += 1
        return self.__counter
def onAdd(res, err):
    print 'OnAdd: counter = %d:' % res
counter = Counter()
counter.incCounter(callback=onAdd)

Distributed lock

An example is a bit more complicated - distributed lock. You can look at all code on github , and here I will simply describe the main aspects of its work.

Let's start with the interface. Lock supports the following operations:

• tryAcquireLock - attempt to take a lock
• isAcquired - check if the lock is taken or released
• release - release lock

The first possible implementation of the lock is similar to the key-value repository. If there is something on the lockA key, then the lock is taken, otherwise it is free, and we can take it ourselves. But not so simple.

Firstly, if we simply use the kv-repository from the example above without any modifications, then the operations of checking for the presence of an element (whether the lock is taken) and writing the element (taking the lock) will not be atomic (that is, we can overwrite someone else’s lock ) Therefore, checking and taking locks should be a single operation, implemented within the replicated class (in this case, tryAcquireLock).

Secondly, if one of the clients who took the lock falls, the lock will hang forever (or until the client lifts up and releases it). In most cases, this is an undesirable behavior. Therefore, we will introduce a timeout after which the lock will be considered released. You will also have to add an operation confirming the capture of the lock (let's call it ping), which will be called with the timeout / 4 interval, and which will extend the life of the taken locks.

The third feature - replicated classes should provide identical behavior on all servers. This means that they should not use inside themselves any data that may differ. For example, a list of processes on the server, a random value, or time. Therefore, if we still want to use time, we will have to pass it as a parameter to all the methods of the class in which it is used.

With this in mind, the resulting implementation consists of two classes - LockImpl, which is a replicated object and also Lock, a wrapper over it. Inside Lock, we automatically add the current time to all operations on LockImpl and also periodically ping in order to confirm the taken locks. The resulting lock is just a minimal example, which can be modified taking into account the required functionality. For example, add callbacks informing us about taking and releasing locks.

Conclusion

We use PySyncObj in the WOT Blitz project to synchronize data between servers in different regions. For example, for the counter of the remaining tanks during the IS-3 Defender event . PySyncObj is a good alternative to existing storage mechanisms in distributed systems. The main analogues are various distributed databases, for example Apache Zookeeper , etcdand others. In contrast, PySyncObj is not a database. It is a lower-level tool and allows you to replicate complex state machines. In addition, it does not require external servers and is easily integrated into python applications. Among the drawbacks of the current version is potentially not the highest performance (now it is completely python code, there are plans to try to rewrite it as a c ++ extension) as well as the lack of separation into the server / client part - sometimes it may be necessary to have a large number of client nodes (often connected / disconnecting) and only a few constantly running servers.

References

• github.com/bakwc/PySyncObj - project sources
• pip install pysyncobj - install via pypi
• raft.github.io - raft protocol site (description and visualization)
• ramcloud.stanford.edu/raft.pdf - original raft publication (with a detailed description of implementation details)
• habrahabr.ru/post/222825 - Consensus in distributed systems. Paxos