Overview of data replication/syncing

Question

How does the IRI process sync its data between peers? Besides reading the code, is there any information that can explain the process?

Answer 1

Learning without thought is labor lost; thought without learning is perilous - Confucius

Jump to the bottom if you want an immediate review. If not, continue along and maybe you might learn a thing or two. As of this writing iri version 1.4.1.2 is the latest release. So the details that follow will be based on this release.

Before we begin, lets start with a visual of the iri process that relates to data replication. We start with the initialization and starting of threads.

The next image displays a rough workflow of basic external to internal communication. It is no by no means complete. As we progress through learning the inner working of data replication for the iri process, additional information will be presented in a clear concise way.

Entry Points

Our discussion begins with the com.iota.iri.network.UDPReceiver::spawnReceiverThread and com.iota.iri.network.replicator.Replicator::replicatorSourcePool threads. These threads listen on their respective ports defined by the values UDP_RECEIVER_PORT and TCP_RECEIVER_PORT. Their main role is to accept data, unpack, and send to the main receiveQueue.

NOTE: This thread initializes a threadPool the size of 1 or 4*availableProcessors(). The variable for threadPool is called processor. A DatagramPacket object is created called receivedPacket with byte buffer and length each 1650.

com.iota.iri.network.UDPReceiver::spawnReceiverThread

The spawnReceiverThread thread, after initialization creates packet counters processed and dropped.

int processed = 0, dropped = 0;

The packet counters are reset when processed+dropped = 50,000

This thread will block until data is received on the port defined by UDP_RECEIVER_PORT. Packet is size is than validated against expected transaction packet size.

if (receivingPacket.getLength() == TRANSACTION_PACKET_SIZE) {

NOTE: The TRANSACTION_PACKET_SIZE equals 1650

Once validated, a copy of the payload from the received packet and peer address are submitted to the processor pool to be executed by com.iota.iri.network.Node::preProcessReceivedData method.

processor.submit(() -> node.preProcessReceivedData(bytes, address, "udp"));

After submission, processed counter is incremented. If the packet received is invalid, it's rejected and dropped counter is incremented. Given the size of the packet 1650, this is greater than most mtu values. Thus, packet fragmentation will occur. The following line is used to ensure fragmented data is reconstructed into proper transaction packet size.

receivingPacket.setLength(TRANSACTION_PACKET_SIZE);

com.iota.iri.network.replicator.Replicator::replicatorSourcePool

The replicatorSourcePool will listen on the TCP_RECEIVER_PORT defined port. An ExecutorService thread pool of 32 threads is created prior to entering main loop. The thread will block until data is received. Once data is received, a new ReplicatorSourceProcessor is created and submitted to the pool for execution.

com.iota.iri.network.replicator.ReplicatorSourceProcessor:

The ReplicatorSourceProcessor thread handles the reading and verification of peer packet received. It starts by creating a data byte array with a 2000 byte capacity. It then moves into verifiying the remote peer connection to ensure it's an existing neighbor.

existingNeighbor = false; List neighbors = node.getNeighbors(); neighbors.stream().filter(n -> n instanceof TCPNeighbor) .map(n -> ((TCPNeighbor) n)) .forEach(n -> { String hisAddress = inet_socket_address.getAddress().getHostAddress(); if (n.getHostAddress().equals(hisAddress)) { existingNeighbor = true; neighbor = n; } } );

The value of MAX_PEERS is tested, including connection from unauthorized neighbors. If either of these occur, a log message is reported and connection is closed. Otherwise, neighbor is added to the neighbors array and counter for neighbors is incremented. This is followed by a verification check to ensure an existing neighbor socket connection already exists, if not set one for neighbor object.

final TCPNeighbor fresh_neighbor = new TCPNeighbor(inet_socket_address,false); node.getNeighbors().add(fresh_neighbor); neighbor = fresh_neighbor; Neighbor.incNumPeers();

The next section of the code deals with handling/accepting a peer's request. As you'll see in com.iota.iri.network.ReplicatorSinkProcessor, an initial 10 byte array is sent which contains the listening service port. It is followed by main payload (1650 byte length) and it's 16 byte checksum value (prepended with "0" to fill 16 bytes).

NOTE: ReplicatorSinkPool.PORT_BYTES = 10

offset = 0; while (((count = stream.read(data, offset, ReplicatorSinkPool.PORT_BYTES - offset)) != -1) && (offset < ReplicatorSinkPool.PORT_BYTES)) { offset += count; }

Thread exit, if the first 10 bytes of the payload received is not the listener service port or connection is closed. The port value received will be used by the ReplicatorSinkProcessor thread to send information back to peer.

The final section of this thread focuses on ensuring proper payload was sent. The full payload in proper order is 10 bytes (peer listener port), 1650 bytes (main payload), and 16 bytes (checksum). The data received is unpacked, checksum validated for main payload, and then submitted to com.iota.iri.network.Node::preProcessReceivedData method for processing.

if (!crcError) { node.preProcessReceivedData(data, address, "tcp"); }

Summary

The iri process listens on UDP_RECEIVER_PORT and TCP_RECEIVER_PORT to accept transactions from its peers. Each transaction passes through validation prior to being added to the ledger.

Syncing

A node starting up for the first time will attempt to initialize its copy of the tangle by reading everything stored in the database. Once database restore is complete, it will check and see if it's current latestSolidSubtangleMilestone matches latestMilestone. If they match, random tip selection would ensue unless transactionsToRequest queue has increased whereby requests will need to be satisfied by it's peers.

A peer that receives the node's request will look through its own copy of the tangle and if request can be fulfilled, will increment numberOfRandomTransactionRequests counter and send transaction to node.

A node that receives a transaction from it's peer that it did not have in it's ledger will process transaction and increment numberOfNewTransactions. Since the transaction is considered new, it will also be placed in the broadcastQueue to be sent to the node's peers.

During the sync'ing process the transactionsToRequest will increase and only decrease as it approaches ledger conformity with its peers. Also through this process, the node will receive transactions that do not match its latestSolidSubtangleMilestone, these transactions are simiply queued until they can be processed.

This overview is being published early. Although far from complete, the intent is to help the operator gain an early understanding until additional information can be presented.