28  * In all states except DEDUPING, there is a single data_vio, called the lock agent, performing the
29  * asynchronous operations on behalf of the lock. The agent will change during the lifetime of the
31  * on a wait queue. Viewed a different way, the agent holds the lock exclusively until the lock
33  * new data_vios that arrive) use to share a PBN lock. In state DEDUPING, there is no agent. When
34  * the last data_vio in the lock calls back in DEDUPING, it becomes the agent and the lock becomes
42  * data_vios in the lock, so the lock will send the agent to the duplicate zone to acquire the PBN
48  * lock on the duplicate block (UNLOCKING), and if the agent is the last data_vio referencing the
94  * hash_lock's agent, added to the list of pending contexts, and then sent to the index. The
262 	struct data_vio *agent;  member
265 	 * Other data_vios with data identical to the agent who are currently waiting for the agent
391  * assert_hash_lock_agent() - Assert that a data_vio is the agent of its hash lock, and that this
393  * @data_vio: The data_vio expected to be the lock agent.
398 	/* Not safe to access the agent field except from the hash zone. */  in assert_hash_lock_agent()
400 	VDO_ASSERT_LOG_ONLY(data_vio == data_vio->hash_lock->agent,  in assert_hash_lock_agent()
401 			    "%s must be for the hash lock agent", where);  in assert_hash_lock_agent()
480 static void start_deduping(struct hash_lock *lock, struct data_vio *agent,
482 static void start_locking(struct hash_lock *lock, struct data_vio *agent);
483 static void start_writing(struct hash_lock *lock, struct data_vio *agent);
489  *                    longer needed to be an agent for the hash lock.
517  * retire_lock_agent() - Retire the active lock agent, replacing it with the first lock waiter, and
518  *                       make the retired agent exit the hash lock.
521  * Return: The new lock agent (which will be NULL if there was no waiter)
525 	struct data_vio *old_agent = lock->agent;  in retire_lock_agent()
528 	lock->agent = new_agent;  in retire_lock_agent()
545 	 * Make sure the agent doesn't block indefinitely in the packer since it now has at least  in wait_on_hash_lock()
548 	if ((lock->state != VDO_HASH_LOCK_WRITING) || !cancel_data_vio_compression(lock->agent))  in wait_on_hash_lock()
553 	 * packer to ensure the agent continues executing. This is safe because  in wait_on_hash_lock()
554 	 * cancel_vio_compression() guarantees the agent won't continue executing until this  in wait_on_hash_lock()
558 	data_vio->compression.lock_holder = lock->agent;  in wait_on_hash_lock()
576  * @agent: The data_vio acting as the agent for the lock.
581 static void start_bypassing(struct hash_lock *lock, struct data_vio *agent)  in start_bypassing()  argument
584 	exit_hash_lock(agent);  in start_bypassing()
596 	if (lock->agent == NULL) {  in vdo_clean_failed_hash_lock()
597 		lock->agent = data_vio;  in vdo_clean_failed_hash_lock()
598 	} else if (data_vio != lock->agent) {  in vdo_clean_failed_hash_lock()
611 		/* The agent must reference the duplicate zone to launch it. */  in vdo_clean_failed_hash_lock()
617 	lock->agent = NULL;  in vdo_clean_failed_hash_lock()
623  * finish_unlocking() - Handle the result of the agent for the lock releasing a read lock on
625  * @completion: The completion of the data_vio acting as the lock's agent.
631 	struct data_vio *agent = as_data_vio(completion);  in finish_unlocking()  local
632 	struct hash_lock *lock = agent->hash_lock;  in finish_unlocking()
634 	assert_hash_lock_agent(agent, __func__);  in finish_unlocking()
646 		start_writing(lock, agent);  in finish_unlocking()
659 		 * agent was releasing the PBN lock. The current agent exits and the waiter has to  in finish_unlocking()
662 		 * TODO: If we used the current agent to re-acquire the PBN lock we wouldn't need  in finish_unlocking()
665 		agent = retire_lock_agent(lock);  in finish_unlocking()
666 		start_locking(lock, agent);  in finish_unlocking()
671 	 * UNLOCKING -> BYPASSING transition: The agent is done with the lock and no other  in finish_unlocking()
674 	start_bypassing(lock, agent);  in finish_unlocking()
680  * @completion: The completion of the data_vio acting as the lock's agent.
687 	struct data_vio *agent = as_data_vio(completion);  in unlock_duplicate_pbn()  local
688 	struct hash_lock *lock = agent->hash_lock;  in unlock_duplicate_pbn()
690 	assert_data_vio_in_duplicate_zone(agent);  in unlock_duplicate_pbn()
694 	vdo_release_physical_zone_pbn_lock(agent->duplicate.zone, agent->duplicate.pbn,  in unlock_duplicate_pbn()
701 	launch_data_vio_hash_zone_callback(agent, finish_unlocking);  in unlock_duplicate_pbn()
708  * @agent: The data_vio currently acting as the agent for the lock.
710 static void start_unlocking(struct hash_lock *lock, struct data_vio *agent)  in start_unlocking()  argument
713 	launch_data_vio_duplicate_zone_callback(agent, unlock_duplicate_pbn);  in start_unlocking()
724 static void process_update_result(struct data_vio *agent)  in process_update_result()  argument
726 	struct dedupe_context *context = agent->dedupe_context;  in process_update_result()
732 	agent->dedupe_context = NULL;  in process_update_result()
737  * finish_updating() - Process the result of a UDS update performed by the agent for the lock.
744 	struct data_vio *agent = as_data_vio(completion);  in finish_updating()  local
745 	struct hash_lock *lock = agent->hash_lock;  in finish_updating()
747 	assert_hash_lock_agent(agent, __func__);  in finish_updating()
749 	process_update_result(agent);  in finish_updating()
760 		 * Send it on the verified dedupe path. The agent is done with the lock, but the  in finish_updating()
763 		start_deduping(lock, agent, true);  in finish_updating()
772 		start_unlocking(lock, agent);  in finish_updating()
780 	start_bypassing(lock, agent);  in finish_updating()
789  * @agent: The data_vio currently acting as the agent for the lock.
791 static void start_updating(struct hash_lock *lock, struct data_vio *agent)  in start_updating()  argument
798 	agent->last_async_operation = VIO_ASYNC_OP_UPDATE_DEDUPE_INDEX;  in start_updating()
799 	set_data_vio_hash_zone_callback(agent, finish_updating);  in start_updating()
800 	query_index(agent, UDS_UPDATE);  in start_updating()
811  * this makes the data_vio the lock agent and uses it to advance the state of the lock so it can
816 	struct data_vio *agent = data_vio;  in finish_deduping()  local
818 	VDO_ASSERT_LOG_ONLY(lock->agent == NULL, "shouldn't have an agent in DEDUPING");  in finish_deduping()
828 	/* The hash lock must have an agent for all other lock states. */  in finish_deduping()
829 	lock->agent = agent;  in finish_deduping()
834 		 * query agent didn't have an allocation. The UDS update was delayed in case there  in finish_deduping()
838 		start_updating(lock, agent);  in finish_deduping()
843 		 * arriving in the lock before the agent returns).  in finish_deduping()
845 		start_unlocking(lock, agent);  in finish_deduping()
936  * @new_agent: The data_vio that will be the agent for the new lock.
938  * Transfers the new agent and any lock waiters to a new hash lock instance which takes the place
961 	new_lock->agent = new_agent;  in fork_hash_lock()
977  * as the writing agent for that lock.
998  * @agent: The data_vio acting as the agent for the lock.
999  * @agent_is_done: true only if the agent has already written or deduplicated against its data.
1001  * If the agent itself needs to deduplicate, an increment for it must already have been claimed
1004 static void start_deduping(struct hash_lock *lock, struct data_vio *agent,  in start_deduping()  argument
1010 	 * We don't take the downgraded allocation lock from the agent unless we actually need to  in start_deduping()
1014 		VDO_ASSERT_LOG_ONLY(!vdo_is_state_compressed(agent->new_mapped.state),  in start_deduping()
1017 				    "agent must have written the new duplicate");  in start_deduping()
1018 		transfer_allocation_lock(agent);  in start_deduping()
1025 	 * This state is not like any of the other states. There is no designated agent--the agent  in start_deduping()
1029 	lock->agent = NULL;  in start_deduping()
1032 	 * Launch the agent (if not already deduplicated) and as many lock waiters as we have  in start_deduping()
1037 		launch_dedupe(lock, agent, true);  in start_deduping()
1038 		agent = NULL;  in start_deduping()
1046 		 * will continue to use the old agent to clean up this lock, and otherwise it just  in start_deduping()
1047 		 * lets the agent exit the lock.  in start_deduping()
1049 		finish_deduping(lock, agent);  in start_deduping()
1067  * finish_verifying() - Handle the result of the agent for the lock comparing its data to the
1075 	struct data_vio *agent = as_data_vio(completion);  in finish_verifying()  local
1076 	struct hash_lock *lock = agent->hash_lock;  in finish_verifying()
1078 	assert_hash_lock_agent(agent, __func__);  in finish_verifying()
1080 	lock->verified = agent->is_duplicate;  in finish_verifying()
1089 			increment_stat(&agent->hash_zone->statistics.dedupe_advice_valid);  in finish_verifying()
1091 			increment_stat(&agent->hash_zone->statistics.dedupe_advice_stale);  in finish_verifying()
1096 	 * claim a reference count increment for the agent.  in finish_verifying()
1099 		agent->is_duplicate = false;  in finish_verifying()
1108 		start_deduping(lock, agent, false);  in finish_verifying()
1113 		 * lock without an agent to release the PBN lock. In both cases, the data will have  in finish_verifying()
1115 		 * VERIFYING agent.  in finish_verifying()
1118 		start_unlocking(lock, agent);  in finish_verifying()
1136 	struct data_vio *agent = as_data_vio(completion);  in verify_callback()  local
1138 	agent->is_duplicate = blocks_equal(agent->vio.data, agent->scratch_block);  in verify_callback()
1139 	launch_data_vio_hash_zone_callback(agent, finish_verifying);  in verify_callback()
1144 	struct data_vio *agent = as_data_vio(completion);  in uncompress_and_verify()  local
1147 	result = uncompress_data_vio(agent, agent->duplicate.state,  in uncompress_and_verify()
1148 				     agent->scratch_block);  in uncompress_and_verify()
1154 	agent->is_duplicate = false;  in uncompress_and_verify()
1155 	launch_data_vio_hash_zone_callback(agent, finish_verifying);  in uncompress_and_verify()
1160 	struct data_vio *agent = vio_as_data_vio(bio->bi_private);  in verify_endio()  local
1165 		agent->is_duplicate = false;  in verify_endio()
1166 		launch_data_vio_hash_zone_callback(agent, finish_verifying);  in verify_endio()
1170 	if (vdo_is_state_compressed(agent->duplicate.state)) {  in verify_endio()
1171 		launch_data_vio_cpu_callback(agent, uncompress_and_verify,  in verify_endio()
1176 	launch_data_vio_cpu_callback(agent, verify_callback,  in verify_endio()
1183  * @agent: The data_vio to use to read and compare candidate data.
1185  * Continue the deduplication path for a hash lock by using the agent to read (and possibly
1186  * decompress) the data at the candidate duplicate location, comparing it to the data in the agent
1191 static void start_verifying(struct hash_lock *lock, struct data_vio *agent)  in start_verifying()  argument
1194 	struct vio *vio = &agent->vio;  in start_verifying()
1195 	char *buffer = (vdo_is_state_compressed(agent->duplicate.state) ?  in start_verifying()
1196 			(char *) agent->compression.block :  in start_verifying()
1197 			agent->scratch_block);  in start_verifying()
1202 	agent->last_async_operation = VIO_ASYNC_OP_VERIFY_DUPLICATION;  in start_verifying()
1204 			       agent->duplicate.pbn);  in start_verifying()
1206 		set_data_vio_hash_zone_callback(agent, finish_verifying);  in start_verifying()
1207 		continue_data_vio_with_error(agent, result);  in start_verifying()
1211 	set_data_vio_bio_zone_callback(agent, vdo_submit_vio);  in start_verifying()
1216  * finish_locking() - Handle the result of the agent for the lock attempting to obtain a PBN read
1224 	struct data_vio *agent = as_data_vio(completion);  in finish_locking()  local
1225 	struct hash_lock *lock = agent->hash_lock;  in finish_locking()
1227 	assert_hash_lock_agent(agent, __func__);  in finish_locking()
1229 	if (!agent->is_duplicate) {  in finish_locking()
1237 		increment_stat(&agent->hash_zone->statistics.dedupe_advice_stale);  in finish_locking()
1239 		start_writing(lock, agent);  in finish_locking()
1249 		 * the candidate duplicate and comparing it to the agent's data to decide whether  in finish_locking()
1252 		start_verifying(lock, agent);  in finish_locking()
1262 		agent->is_duplicate = false;  in finish_locking()
1265 		start_unlocking(lock, agent);  in finish_locking()
1273 	start_deduping(lock, agent, false);  in finish_locking()
1276 static bool acquire_provisional_reference(struct data_vio *agent, struct pbn_lock *lock,  in acquire_provisional_reference()  argument
1280 	struct vdo_slab *slab = vdo_get_slab(depot, agent->duplicate.pbn);  in acquire_provisional_reference()
1281 	int result = vdo_acquire_provisional_reference(slab, agent->duplicate.pbn, lock);  in acquire_provisional_reference()
1288 	agent->is_duplicate = false;  in acquire_provisional_reference()
1289 	vdo_release_physical_zone_pbn_lock(agent->duplicate.zone,  in acquire_provisional_reference()
1290 					   agent->duplicate.pbn, lock);  in acquire_provisional_reference()
1291 	continue_data_vio_with_error(agent, result);  in acquire_provisional_reference()
1311 	struct data_vio *agent = as_data_vio(completion);  in lock_duplicate_pbn()  local
1312 	struct slab_depot *depot = vdo_from_data_vio(agent)->depot;  in lock_duplicate_pbn()
1313 	struct physical_zone *zone = agent->duplicate.zone;  in lock_duplicate_pbn()
1315 	assert_data_vio_in_duplicate_zone(agent);  in lock_duplicate_pbn()
1317 	set_data_vio_hash_zone_callback(agent, finish_locking);  in lock_duplicate_pbn()
1323 	increment_limit = vdo_get_increment_limit(depot, agent->duplicate.pbn);  in lock_duplicate_pbn()
1329 		agent->is_duplicate = false;  in lock_duplicate_pbn()
1330 		continue_data_vio(agent);  in lock_duplicate_pbn()
1334 	result = vdo_attempt_physical_zone_pbn_lock(zone, agent->duplicate.pbn,  in lock_duplicate_pbn()
1337 		continue_data_vio_with_error(agent, result);  in lock_duplicate_pbn()
1361 		 * agent.  in lock_duplicate_pbn()
1377 		agent->is_duplicate = false;  in lock_duplicate_pbn()
1378 		continue_data_vio(agent);  in lock_duplicate_pbn()
1383 		if (!acquire_provisional_reference(agent, lock, depot))  in lock_duplicate_pbn()
1397 	set_duplicate_lock(agent->hash_lock, lock);  in lock_duplicate_pbn()
1403 	continue_data_vio(agent);  in lock_duplicate_pbn()
1408  *                   potential duplicate through its agent.
1410  * @agent: The data_vio bearing the dedupe advice.
1412 static void start_locking(struct hash_lock *lock, struct data_vio *agent)  in start_locking()  argument
1421 	 * accepting the advice, and don't explicitly change lock states (or use an agent-local  in start_locking()
1424 	agent->last_async_operation = VIO_ASYNC_OP_LOCK_DUPLICATE_PBN;  in start_locking()
1425 	launch_data_vio_duplicate_zone_callback(agent, lock_duplicate_pbn);  in start_locking()
1429  * finish_writing() - Re-entry point for the lock agent after it has finished writing or
1432  * @agent: The data_vio that wrote its data for the lock.
1434  * The agent will never need to dedupe against anything, so it's done with the lock, but the lock
1437  * If there are other lock holders, the agent will hand the job to one of them and exit, leaving
1439  * agent either exits (and later tears down the hash lock), or it remains the agent and updates
1442 static void finish_writing(struct hash_lock *lock, struct data_vio *agent)  in finish_writing()  argument
1445 	 * Dedupe against the data block or compressed block slot the agent wrote. Since we know  in finish_writing()
1448 	lock->duplicate = agent->new_mapped;  in finish_writing()
1463 		 * compress, so the PBN lock on the written copy was already transferred. The agent  in finish_writing()
1467 		start_deduping(lock, agent, true);  in finish_writing()
1472 	 * There are no waiters and the agent has successfully written, so take a step towards  in finish_writing()
1478 		 * retain the WRITING agent and use it to launch the update. The happens on  in finish_writing()
1479 		 * compression, rollover, or the QUERYING agent not having an allocation.  in finish_writing()
1481 		start_updating(lock, agent);  in finish_writing()
1487 		set_duplicate_location(agent, lock->duplicate);  in finish_writing()
1488 		start_unlocking(lock, agent);  in finish_writing()
1492 		 * duplicate lock held, so both the agent and lock have no more work to do. The  in finish_writing()
1493 		 * agent will release its allocation lock in cleanup.  in finish_writing()
1495 		start_bypassing(lock, agent);  in finish_writing()
1503  * If an allocation is found, swap agents, put the old agent at the head of the wait queue, then
1504  * return the new agent. Otherwise, just return the current agent.
1531 		 * The current agent is being replaced and will have to wait to dedupe; make it the  in select_writing_agent()
1534 		vdo_waitq_enqueue_waiter(&lock->waiters, &lock->agent->waiter);  in select_writing_agent()
1535 		lock->agent = data_vio;  in select_writing_agent()
1537 		/* No one has an allocation, so keep the current agent. */  in select_writing_agent()
1538 		data_vio = lock->agent;  in select_writing_agent()
1549  * @agent: The data_vio currently acting as the agent for the lock.
1552  * with an allocation as a new agent, if necessary, then resuming the agent on the data_vio write
1555 static void start_writing(struct hash_lock *lock, struct data_vio *agent)  in start_writing()  argument
1560 	 * The agent might not have received an allocation and so can't be used for writing, but  in start_writing()
1563 	if (!data_vio_has_allocation(agent)) {  in start_writing()
1564 		agent = select_writing_agent(lock);  in start_writing()
1566 		if (!data_vio_has_allocation(agent)) {  in start_writing()
1572 			 * if the agent and all waiters release.  in start_writing()
1574 			continue_data_vio_with_error(agent, VDO_NO_SPACE);  in start_writing()
1580 	 * If the agent compresses, it might wait indefinitely in the packer, which would be bad if  in start_writing()
1584 		cancel_data_vio_compression(agent);  in start_writing()
1587 	 * Send the agent to the compress/pack/write path in vioWrite. If it succeeds, it will  in start_writing()
1590 	launch_compress_data_vio(agent);  in start_writing()
1643 static void process_query_result(struct data_vio *agent)  in process_query_result()  argument
1645 	struct dedupe_context *context = agent->dedupe_context;  in process_query_result()
1651 		agent->is_duplicate = decode_uds_advice(context);  in process_query_result()
1652 		agent->dedupe_context = NULL;  in process_query_result()
1658  * finish_querying() - Process the result of a UDS query performed by the agent for the lock.
1665 	struct data_vio *agent = as_data_vio(completion);  in finish_querying()  local
1666 	struct hash_lock *lock = agent->hash_lock;  in finish_querying()
1668 	assert_hash_lock_agent(agent, __func__);  in finish_querying()
1670 	process_query_result(agent);  in finish_querying()
1672 	if (agent->is_duplicate) {  in finish_querying()
1673 		lock->duplicate = agent->duplicate;  in finish_querying()
1676 		 * QUERYING agent to start the hash lock on the unverified dedupe path, verifying  in finish_querying()
1679 		start_locking(lock, agent);  in finish_querying()
1682 		 * The agent will be used as the duplicate if has an allocation; if it does, that  in finish_querying()
1685 		lock->update_advice = !data_vio_has_allocation(agent);  in finish_querying()
1690 		start_writing(lock, agent);  in finish_querying()
1700  * requested it the agent, entering the QUERYING state, and using the agent to perform the UDS
1705 	lock->agent = data_vio;  in start_querying()
1745 		VDO_ASSERT_LOG_ONLY(data_vio == lock->agent,  in vdo_continue_hash_lock()
1746 				    "only the lock agent may continue the lock");  in vdo_continue_hash_lock()
2755 		     vdo_waitq_num_waiters(&lock->waiters), lock->agent);  in dump_hash_lock()