rpc: call ->rio_sent() unconditionally. (#980)

rpc promises to call m0_rpc_item_ops::rio_sent() when item is placed on the
wire, or when sending process completes.

Rpc users depend on ->rio_sent() being called in case of error, for all
types of items (requests, one-ways, replies):

   - rpc itself (conn_terminate_reply_sent_cb());

   - ha/link.c (ha_link_outgoing_item_sent()).

Rpc fails to invoke ->rio_sent() in the following cases:

   - item was cancelled (item_sent()),
   - item sent failed and item was not one-way (m0_rpc_item_failed()).

Fix all these places to invoke ->rio_sent().

Signed-off-by: Nikita Danilov <nikita.danilov@seagate.com>

rpc/frmops.c

@@ -502,6 +502,7 @@ static void item_sent(struct m0_rpc_item *item)
 		 */
 		M0_ASSERT(m0_rpc_item_is_request(item));
 		M0_ASSERT(item->ri_error == -ECANCELED);
+		m0_rpc_item_sent_invoke(item);
 		return;
 	}
 
@@ -516,8 +517,7 @@ static void item_sent(struct m0_rpc_item *item)
 
 	if (item->ri_nr_sent == 1) { /* not resent */
 		stats->rs_nr_sent_items_uniq++;
-		if (item->ri_ops != NULL && item->ri_ops->rio_sent != NULL)
-			item->ri_ops->rio_sent(item);
+		m0_rpc_item_sent_invoke(item);
 	} else if (item->ri_nr_sent == 2) {
 		/* item with ri_nr_sent >= 2 are counted as 1 in
 		   rs_nr_resent_items i.e. rs_nr_resent_items counts number

rpc/item.c

@@ -800,10 +800,7 @@ M0_INTERNAL void m0_rpc_item_failed(struct m0_rpc_item *item, int32_t rc)
 	m0_rpc_item_change_state(item, M0_RPC_ITEM_FAILED);
 	m0_rpc_item_timer_stop(item);
 	/* XXX ->rio_sent() can be called multiple times (due to cancel). */
-	if (m0_rpc_item_is_oneway(item) &&
-	    item->ri_ops != NULL && item->ri_ops->rio_sent != NULL)
-		item->ri_ops->rio_sent(item);
-
+	m0_rpc_item_sent_invoke(item);
 	m0_rpc_session_item_failed(item);
 	/*
 	 * Reference release done here is for the reference taken
@@ -1766,6 +1763,12 @@ M0_INTERNAL void m0_rpc_item_replied_invoke(struct m0_rpc_item *req)
 	}
 }
 
+M0_INTERNAL void m0_rpc_item_sent_invoke(struct m0_rpc_item *item)
+{
+	if (item->ri_ops != NULL && item->ri_ops->rio_sent != NULL)
+		item->ri_ops->rio_sent(item);
+}
+
 #undef M0_TRACE_SUBSYSTEM
 /** @} end of rpc group */
 

rpc/item_internal.h

@@ -80,6 +80,7 @@ M0_INTERNAL void m0_rpc_item_send_reply(struct m0_rpc_item *req,
 					struct m0_rpc_item *reply);
 
 M0_INTERNAL void m0_rpc_item_replied_invoke(struct m0_rpc_item *item);
+M0_INTERNAL void m0_rpc_item_sent_invoke(struct m0_rpc_item *item);
 
 /** @} */
 #endif /* __MOTR_RPC_ITEM_INT_H__ */