Get/Put path walk through

get_put_code.md

Put

API

• Starting: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_wm_object.erl#L619
• We create a new riak_object and populate the various fields with the headers, metadata supplied by the client.
• Big suprise, we eventually call riak_client:put: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_client.erl#L143
• If/when the client returns any errors these are handled in handle_common_errors and it is nice to return human readable errors to client :)

Riak_client

• The client creates a request ID and the normal path does not increment the vclock(yet).
• We then start_link a riak_kv_put_fsm: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_put_fsm.erl#L131
• Eventually the FSM gets back to us at wait_for_reqid which is where we wait until timout: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_client.erl#L607

FSM

Setup

• We start the FSM through sidejob(assuming overload protections isn't disabled) and sidejob may say we're overloaded.
• We then call gen_fsm:start_link which creates a new fsm at the init of riak_kv_put_fsm: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_put_fsm.erl#L174
• Start by initializing our state, notifying the monitor count of our start, and check if the riak_object is tombstone. We eventually end up in prepare state: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_put_fsm.erl#L210
• prepare gets the bucket properties, hashes the bucket/key(riak_core_util:chash_key(BKey)), and decides/validates the n_val. We then create the preflist for the hash: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_put_fsm.erl#L233
• We then check if the local node is in the preflist or if we need to forward to another node for coordination. If as-is is set, skip coordination all together.
• Forwarding to another node starts a brand new FSM which will eventually report back to the riak_client waiting on RequestID: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_put_fsm.erl#L255
• Handling coordination locally sends us to validate state(starting our FSM timeout): https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_put_fsm.erl#L295

Coordination

• validate checks all write specific request options(W,DW,PW,etc) and throws an error for those not met: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_put_fsm.erl#L324
• We then iterate the hooks in the bucket props, update out state, start the riak_core_put_core and enter precommit state(assuming there are precommit hooks to run): https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_put_fsm.erl#L374
• precommit iterates over all the hook functions, returns the result, and update the riak_object if necessary. We eventually end up at execute: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_put_fsm.erl#L390
• execute either immediately calls execute_remote which skips all coordination(in the case of as-is being set) or(common case) execute_local: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_put_fsm.erl#L402
• execute_local calls riak_kv_vnode:coord_put and then waits in waiting_local_vnode state: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_put_fsm.erl#L422
• riak_kv_vnode:coord_put returns a failure or returns success with or without a new riak_object depending on if an object existed. We then call execute_remote with/without an updated object: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_put_fsm.erl#L452
• execute_remote call riak_kv_vnode:put and then sits waiting for vnode responses and adding them to riak_kv_put_core:add_result: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_put_core.erl#L81

Riak_kv_put_core

• riak_kv_put_core:add_result updates the riak_kv_put_core state and eventually enough will return true or false: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_put_fsm.erl#L483
• riak_kv_put_core:response then returns the necessary response(success or failure with/without object). We then enter process_reply of the put fsm: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_put_fsm.erl#L575

Finalize

• process_reply calls client_reply which sends the results of the put back to the client: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_put_fsm.erl#L870
• We thenhen iterates over all postcommit hooks. If the client reply is not an error, we call postcommit state: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_put_fsm.erl#L495
• postcommit state runs all postcommit hooks, even if the client request has timed out, and eventually calls finish state: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_put_fsm.erl#L515
• finish state finalizes all stats and updates the riak_kv_stat module. We then call stop state which stops the fsm.

Vnode

Coordinating Put

• riak_kv_vnode:coord_put:https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_vnode.erl#L229
• This function starts a riak_core_vnode_master:command of type ?KV_PUT_REQ but only on IndexNode and with the special coord option set: https://github.com/basho/riak_core/blob/1.4.2/src/riak_core_vnode_master.erl#L74
• Since this is a list of one node, we only send one gen_fsm:send_event to the local vnode we are using as the coordinating put: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_vnode.erl#L395
• We then immediately reply(because not always DW) with riak_core_vnode:reply and then do_put: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_vnode.erl#L900
• do_put checks if this is a coordinating put by looking for the coord option, if true it makes sure to return the body of the put. We then call prepare_put: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_vnode.erl#L954
• prepare_put checks LWW and if true increments the vecotr clock for the coordinating put. If not true, it then calls the next prepare_put function: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_vnode.erl#L981
• prepare_put calls do_get_object which returns a not found or riak_object. If not found, we call prepare_new_putwhich simpy increments the vecotr clock for the coordinating put. If existing, we call put_merge: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_vnode.erl#L1154
• put_merge is the function to look at if interested in LWW and allow_mult. Basically, either a new_obj or an old_obj atom are returned back to prepare_put with an incremented vclock either way in the case of a coordinating put.
• Back in prepare_put we'll always be in the new_obj code because we are coordinating: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_vnode.erl#L1014
• Once we finish merging siblings and trimming vclocks we return a new state back to do_put. This then calls perform_put: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_vnode.erl#L1071
• For coordinating puts we will always be in the true function. We then encode_and_put: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_vnode.erl#L1670
• encode_and_put calls Mod:put or Mod:object_put depending on riak_object format used. Mod = riak_kv_X_backend.
• We then update our AAE hashtrees and reply back to the put FSM.

Normal Put

• Exactly the same as coordinating PUT except it does not have the coord option set. This operation is run on the remaining vnodes in the preflist after the coordinating put has been succesful.
• The main operations are do_put and prepare_put which eventually calls Mod:put responding back to the FSM before and after sending the put to the backend.

Backend

Riak_kv_bitcask_backend

• Entry point is riak_kv_bitcask_backend:put or put_object depending on the riak_object format used: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_bitcask_backend.erl#L166
• After a binary_to_term(b2t) call we hit bitcask:put: https://github.com/basho/bitcask/blob/1.6/src/bitcask.erl#L232
• This function call do_put and then returns to the riak_kv backend module.
• More work to do here, but it's all in the bitcask backend.

Riak_kv_eleveldb_backend

• Entry point is riak_kv_eleveldb_backend:put or put_object depending on the riak_object format used: https://github.com/basho/eleveldb/blob/develop/src/eleveldb.erl#L148
• We call the write function, which in turn calls the async_write function: https://github.com/basho/eleveldb/blob/develop/src/eleveldb.erl#L160
• The rest of the work here is in the leveldb NIF. Another day.

Riak_kv_memory_backend

• Entry point is riak_kv_memory_backend:put or put_object depending on the riak_object format used: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_memory_backend.erl#L198
• We eventually call do_put which calls ets:insert to insert the object: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_memory_backend.erl#L550
• We then update the index table either with a remove or add atom determining if we ets:delete or ets:insert: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_memory_backend.erl#L559

Riak_kv_multi_backend

• Entry point is riak_kv_multi_backend:put or put_object depending on the riak_object format used:https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_multi_backend.erl#L228
• We call get_backend: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_multi_backend.erl#L435
• This queries bucket properties, identifies the backend, and validates it.
• We then call the appropriate Module:put where Module is our backend: https://github.com/basho/riak_kv/blob/1.4.2/src/riak_kv_multi_backend.erl#L230

Some additional information on https://gist.github.com/bsparrow435/7818782#finalize.

Unlike the GET FSM, the PUT FSM will close without all remote responses being received (and without the request timeout being reached). Adding here (copied from postriak slack) as I got confused by this when looking at a potential sidejob issue, and the use of immediate timeouts to achieve this makes it not necessarily immediately obvious from reviewing the code.

Just for the record, I've been reading through the PUT_FSM code and I don't think one of the things I said before was correct. The critical question is - does the PUT_FSM stay active until all vnodes have responded to the PUT (or a timeout triggers).

The answer, I think is actually "no". The PUT FSM (unlike the GET FSM) can effectively finish/stop when enough (rather than all) vnodes have responded.

When enough responses have been received processReply is called https://github.com/basho/riak_kv/blob/2.1.7/src/riak_kv_put_fsm.erl#L638 - and this triggers a new_state_timeout/2 call https://github.com/basho/riak_kv/blob/2.1.7/src/riak_kv_put_fsm.erl#L792

This function call will set the state to post_commit, but also trigger a timeout message immediately so that the next timeout action will happen straight away.

Each postcommit hook is then processed, with the return setting the timeout to 0 after each one causing another immediate timeout trigger - https://github.com/basho/riak_kv/blob/2.1.7/src/riak_kv_put_fsm.erl#L655-L657

When the postcommit hooks have been exhausted - there is another new_state_timeout call which will trigger a timeout message on the finish state - https://github.com/basho/riak_kv/blob/2.1.7/src/riak_kv_put_fsm.erl#L646

Handling timeout in the finish state will always result in "stop" - https://github.com/basho/riak_kv/blob/2.1.7/src/riak_kv_put_fsm.erl#L676-L697

So although the code will handle late replys in all states (e.g. https://github.com/basho/riak_kv/blob/2.1.7/src/riak_kv_put_fsm.erl#L662) the FSM will close once enough replys have been replied and the postcommit hook functions have been called (which may have to wait). A late reply is either ignored as the FSM has terminated, or taken as a message and effectively ignored by the code