How to implement read-after-write with raft-rs?

[tisonkun]

Ref - https://github.com/tisonkun/mephisto

I'm trying to understand this crate by creating a simple DataTree storage system (as linked above).

By implementing the basic PUT/GET (Create/GetData) methods, I'm expecting the following requests get linearizable results:

Create k0 v0
GetData k0 => v0
Create k0 v1
GetData k0 => v1
Create k1 v2
GetData k1 => v2
Create k1 v3
GetData k1 => v3

The raftstore example exactly performs these actions.

I implement it with MsgPropose for every "Create" request, and issue a MsgReadIndex for every "GetData" request. But it seems that the read index returns the index just when it's issued instead of after the former proposal committed.

Now, since the "Create" requests go with Raft append entries, it returns properly. But "GetData" can return previous data:

2023-08-21T15:02:01.054544Z  INFO raftstore: rx.recv()[0]=Ok(FatReply { header: ReplyHeader { req_id: 0, txn_id: 0, err: 0 }, reply: Create(CreateReply { path: "mephisto-0" }) })
2023-08-21T15:02:01.054593Z  INFO raftstore: rx.recv()[1]=Ok(FatReply { header: ReplyHeader { req_id: 1, txn_id: 2, err: 0 }, reply: GetData(GetDataReply { data: b"value-0" }) })
2023-08-21T15:02:01.054611Z  INFO raftstore: rx.recv()[2]=Ok(FatReply { header: ReplyHeader { req_id: 2, txn_id: 2, err: 0 }, reply: Create(CreateReply { path: "mephisto-0" }) })
2023-08-21T15:02:01.054624Z  INFO raftstore: rx.recv()[3]=Ok(FatReply { header: ReplyHeader { req_id: 3, txn_id: 2, err: 0 }, reply: GetData(GetDataReply { data: b"value-0" }) })
2023-08-21T15:02:01.054638Z  INFO raftstore: rx.recv()[4]=Ok(FatReply { header: ReplyHeader { req_id: 4, txn_id: 3, err: 0 }, reply: Create(CreateReply { path: "mephisto-1" }) })
2023-08-21T15:02:01.054697Z  INFO raftstore: rx.recv()[5]=Ok(FatReply { header: ReplyHeader { req_id: 5, txn_id: 2, err: 0 }, reply: GetData(GetDataReply { data: b"" }) })
2023-08-21T15:02:01.054731Z  INFO raftstore: rx.recv()[6]=Ok(FatReply { header: ReplyHeader { req_id: 6, txn_id: 4, err: 0 }, reply: Create(CreateReply { path: "mephisto-1" }) })
2023-08-21T15:02:01.054749Z  INFO raftstore: rx.recv()[7]=Ok(FatReply { header: ReplyHeader { req_id: 7, txn_id: 2, err: 0 }, reply: GetData(GetDataReply { data: b"" }) })
2023-08-21T15:02:01.054762Z  INFO raftstore: rx.recv()[8]=Ok(FatReply { header: ReplyHeader { req_id: 8, txn_id: 5, err: 0 }, reply: Create(CreateReply { path: "mephisto-2" }) })
2023-08-21T15:02:01.054775Z  INFO raftstore: rx.recv()[9]=Ok(FatReply { header: ReplyHeader { req_id: 9, txn_id: 6, err: 0 }, reply: GetData(GetDataReply { data: b"value-4" }) })
2023-08-21T15:02:01.054819Z  INFO raftstore: rx.recv()[10]=Ok(FatReply { header: ReplyHeader { req_id: 10, txn_id: 6, err: 0 }, reply: Create(CreateReply { path: "mephisto-2" }) })
2023-08-21T15:02:01.054846Z  INFO raftstore: rx.recv()[11]=Ok(FatReply { header: ReplyHeader { req_id: 11, txn_id: 6, err: 0 }, reply: GetData(GetDataReply { data: b"value-4" }) })
2023-08-21T15:02:01.054863Z  INFO raftstore: rx.recv()[12]=Ok(FatReply { header: ReplyHeader { req_id: 12, txn_id: 7, err: 0 }, reply: Create(CreateReply { path: "mephisto-3" }) })
2023-08-21T15:02:01.054876Z  INFO raftstore: rx.recv()[13]=Ok(FatReply { header: ReplyHeader { req_id: 13, txn_id: 6, err: 0 }, reply: GetData(GetDataReply { data: b"" }) })
2023-08-21T15:02:01.054888Z  INFO raftstore: rx.recv()[14]=Ok(FatReply { header: ReplyHeader { req_id: 14, txn_id: 8, err: 0 }, reply: Create(CreateReply { path: "mephisto-3" }) })
2023-08-21T15:02:01.054905Z  INFO raftstore: rx.recv()[15]=Ok(FatReply { header: ReplyHeader { req_id: 15, txn_id: 6, err: 0 }, reply: GetData(GetDataReply { data: b"" }) })
2023-08-21T15:02:01.054916Z  INFO raftstore: rx.recv()[16]=Ok(FatReply { header: ReplyHeader { req_id: 16, txn_id: 9, err: 0 }, reply: Create(CreateReply { path: "mephisto-4" }) })
2023-08-21T15:02:01.054928Z  INFO raftstore: rx.recv()[17]=Ok(FatReply { header: ReplyHeader { req_id: 17, txn_id: 6, err: 0 }, reply: GetData(GetDataReply { data: b"" }) })
2023-08-21T15:02:01.054939Z  INFO raftstore: rx.recv()[18]=Ok(FatReply { header: ReplyHeader { req_id: 18, txn_id: 10, err: 0 }, reply: Create(CreateReply { path: "mephisto-4" }) })
2023-08-21T15:02:01.055018Z  INFO raftstore: rx.recv()[19]=Ok(FatReply { header: ReplyHeader { req_id: 19, txn_id: 6, err: 0 }, reply: GetData(GetDataReply { data: b"" }) })

I don't know how we can implement read-after-write without inserting "GetData" request with MsgPropose.

I read this post but am still unclear what to do.

cc @BusyJay

[tisonkun]

Blocking read and write on client side can ensure the order. But I'm still wondering what kind of read/write concurrency we can do.

Also, blocking read and write doesn't seems need "ReadIndex" at all.

[BusyJay]

raft-rs only guarantees ReadIndex can read the latest data that it has told client successfully processed. In theory, no system can guarantee the order without extra synchronization. For example, the previous create request can be delay by network, and client has no idea when the request reaches the server and it's totally OK that the create is out of order with the following read index request.

On the other hand, using committed index is good for latency. In most cases, when the read index is returned, the index is probably already applied.

[tisonkun]

Thanks for your sharing!

May I ask how TiKV implements read-after-write upon raft-rs in a high-level? Or what is its "extra synchronization"?

[tisonkun]

... and in this definition, what is the proper situation for "ReadIndex", cause it only tell the node the last commit index among the cluster but I have no idea how I can make use of this information

[BusyJay]

TiKV also uses read index except it uses lease when it's a leader read. After a ReadIndex is returned, application should wait till the log at the given index is applied.

[tisonkun]

@BusyJay Thanks for your explanation! Then I saw two kinds of read logics:

• Among different clients, each read op issues a ReadIndex request (leader can use leader lease) to ensure that they read fresh data from the cluster point of view.
• Inside the same clients, application should implement its own synchronization for ensuring read-after-write or other semantic.

For example,

If client C1 and C2 modify the same key K1, a possible series is:

• C1 updates K1 to V1
• C2 updates K1 to V2
• C1 reads K1

C1's synchronization ensures that the read at least read V1. And ReadIndex ensures that if the commit index beyond the one causes K1 to V2 when C1's read happens, the read will read V2. But if the commit index is before V2's or even V1's entry, without extra synchronization, the read may read V1 or empty.

[tisonkun]

Implemented step by step. But I may find better structure for that mess of code.. tisonkun/mephisto@5aeeb5f

Fixup that read-index can lose in raft-rs tisonkun/mephisto@721981a we should always retry to ensure that read would eventually return. The condition here is:

• A write whose seq_id > read's wait_write_seq will force the previous read to flush. Otherwise we read forward data.
• If no following write drive the read to return, keep retry read_index so that we can eventually return.