ocfs2: Added post handler callable function in o2net message handler

Currently o2net allows one handler function per message type. This
patch adds the ability to call another function to be called after
the handler has returned the message to the other node.

Handlers are now given the option of returning a context (in the form of a
void **) which will be passed back into the post message handler function.

Signed-off-by: Kurt Hackel <kurt.hackel@oracle.com>
Signed-off-by: Sunil Mushran <sunil.mushran@oracle.com>
Signed-off-by: Mark Fasheh <mark.fasheh@oracle.com>

8 files changed, 90 insertions, 54 deletions

diff --git a/fs/ocfs2/dlm/dlmast.c b/fs/ocfs2/dlm/dlmast.c
index ad5e7e1fa1ff..241cad342a48 100644
--- a/fs/ocfs2/dlm/dlmast.c
+++ b/fs/ocfs2/dlm/dlmast.c
@@ -263,7 +263,8 @@ void dlm_do_local_bast(struct dlm_ctxt *dlm, struct dlm_lock_resource *res,
 
 
 
-int dlm_proxy_ast_handler(struct o2net_msg *msg, u32 len, void *data)
+int dlm_proxy_ast_handler(struct o2net_msg *msg, u32 len, void *data,
+                          void **ret_data)
 {
         int ret;
         unsigned int locklen;
diff --git a/fs/ocfs2/dlm/dlmcommon.h b/fs/ocfs2/dlm/dlmcommon.h
index e95ecb2aaf14..2df6fde3e652 100644
--- a/fs/ocfs2/dlm/dlmcommon.h
+++ b/fs/ocfs2/dlm/dlmcommon.h
@@ -707,16 +707,20 @@ void dlm_lock_put(struct dlm_lock *lock);
 void dlm_lock_attach_lockres(struct dlm_lock *lock,
                              struct dlm_lock_resource *res);
 
-int dlm_create_lock_handler(struct o2net_msg *msg, u32 len, void *data);
-int dlm_convert_lock_handler(struct o2net_msg *msg, u32 len, void *data);
-int dlm_proxy_ast_handler(struct o2net_msg *msg, u32 len, void *data);
+int dlm_create_lock_handler(struct o2net_msg *msg, u32 len, void *data,
+                            void **ret_data);
+int dlm_convert_lock_handler(struct o2net_msg *msg, u32 len, void *data,
+                             void **ret_data);
+int dlm_proxy_ast_handler(struct o2net_msg *msg, u32 len, void *data,
+                          void **ret_data);
 
 void dlm_revert_pending_convert(struct dlm_lock_resource *res,
                                 struct dlm_lock *lock);
 void dlm_revert_pending_lock(struct dlm_lock_resource *res,
                              struct dlm_lock *lock);
 
-int dlm_unlock_lock_handler(struct o2net_msg *msg, u32 len, void *data);
+int dlm_unlock_lock_handler(struct o2net_msg *msg, u32 len, void *data,
+                            void **ret_data);
 void dlm_commit_pending_cancel(struct dlm_lock_resource *res,
                                struct dlm_lock *lock);
 void dlm_commit_pending_unlock(struct dlm_lock_resource *res,
@@ -871,16 +875,26 @@ void dlm_lockres_release_ast(struct dlm_ctxt *dlm,
                              struct dlm_lock_resource *res);
 void __dlm_lockres_reserve_ast(struct dlm_lock_resource *res);
 
-int dlm_master_request_handler(struct o2net_msg *msg, u32 len, void *data);
-int dlm_assert_master_handler(struct o2net_msg *msg, u32 len, void *data);
-int dlm_deref_lockres_handler(struct o2net_msg *msg, u32 len, void *data);
-int dlm_migrate_request_handler(struct o2net_msg *msg, u32 len, void *data);
-int dlm_mig_lockres_handler(struct o2net_msg *msg, u32 len, void *data);
-int dlm_master_requery_handler(struct o2net_msg *msg, u32 len, void *data);
-int dlm_request_all_locks_handler(struct o2net_msg *msg, u32 len, void *data);
-int dlm_reco_data_done_handler(struct o2net_msg *msg, u32 len, void *data);
-int dlm_begin_reco_handler(struct o2net_msg *msg, u32 len, void *data);
-int dlm_finalize_reco_handler(struct o2net_msg *msg, u32 len, void *data);
+int dlm_master_request_handler(struct o2net_msg *msg, u32 len, void *data,
+                               void **ret_data);
+int dlm_assert_master_handler(struct o2net_msg *msg, u32 len, void *data,
+                              void **ret_data);
+int dlm_deref_lockres_handler(struct o2net_msg *msg, u32 len, void *data,
+                              void **ret_data);
+int dlm_migrate_request_handler(struct o2net_msg *msg, u32 len, void *data,
+                                void **ret_data);
+int dlm_mig_lockres_handler(struct o2net_msg *msg, u32 len, void *data,
+                            void **ret_data);
+int dlm_master_requery_handler(struct o2net_msg *msg, u32 len, void *data,
+                               void **ret_data);
+int dlm_request_all_locks_handler(struct o2net_msg *msg, u32 len, void *data,
+                                  void **ret_data);
+int dlm_reco_data_done_handler(struct o2net_msg *msg, u32 len, void *data,
+                               void **ret_data);
+int dlm_begin_reco_handler(struct o2net_msg *msg, u32 len, void *data,
+                           void **ret_data);
+int dlm_finalize_reco_handler(struct o2net_msg *msg, u32 len, void *data,
+                              void **ret_data);
 int dlm_do_master_requery(struct dlm_ctxt *dlm, struct dlm_lock_resource *res,
                           u8 nodenum, u8 *real_master);
 
diff --git a/fs/ocfs2/dlm/dlmconvert.c b/fs/ocfs2/dlm/dlmconvert.c
index 59fb63da8b65..ecb4d997221e 100644
--- a/fs/ocfs2/dlm/dlmconvert.c
+++ b/fs/ocfs2/dlm/dlmconvert.c
@@ -418,7 +418,8 @@ static enum dlm_status dlm_send_remote_convert_request(struct dlm_ctxt *dlm,
  * returns: DLM_NORMAL, DLM_IVLOCKID, DLM_BADARGS,
  *          status from __dlmconvert_master
  */
-int dlm_convert_lock_handler(struct o2net_msg *msg, u32 len, void *data)
+int dlm_convert_lock_handler(struct o2net_msg *msg, u32 len, void *data,
+                             void **ret_data)
 {
         struct dlm_ctxt *dlm = data;
         struct dlm_convert_lock *cnv = (struct dlm_convert_lock *)msg->buf;
diff --git a/fs/ocfs2/dlm/dlmdomain.c b/fs/ocfs2/dlm/dlmdomain.c
index 3995de360264..8a208b06fdd7 100644
--- a/fs/ocfs2/dlm/dlmdomain.c
+++ b/fs/ocfs2/dlm/dlmdomain.c
@@ -95,10 +95,14 @@ static DECLARE_WAIT_QUEUE_HEAD(dlm_domain_events);
 
 #define DLM_DOMAIN_BACKOFF_MS 200
 
-static int dlm_query_join_handler(struct o2net_msg *msg, u32 len, void *data);
-static int dlm_assert_joined_handler(struct o2net_msg *msg, u32 len, void *data);
-static int dlm_cancel_join_handler(struct o2net_msg *msg, u32 len, void *data);
-static int dlm_exit_domain_handler(struct o2net_msg *msg, u32 len, void *data);
+static int dlm_query_join_handler(struct o2net_msg *msg, u32 len, void *data,
+                                  void **ret_data);
+static int dlm_assert_joined_handler(struct o2net_msg *msg, u32 len, void *data,
+                                     void **ret_data);
+static int dlm_cancel_join_handler(struct o2net_msg *msg, u32 len, void *data,
+                                   void **ret_data);
+static int dlm_exit_domain_handler(struct o2net_msg *msg, u32 len, void *data,
+                                   void **ret_data);
 
 static void dlm_unregister_domain_handlers(struct dlm_ctxt *dlm);
 
@@ -466,7 +470,8 @@ static void __dlm_print_nodes(struct dlm_ctxt *dlm)
         printk("\n");
 }
 
-static int dlm_exit_domain_handler(struct o2net_msg *msg, u32 len, void *data)
+static int dlm_exit_domain_handler(struct o2net_msg *msg, u32 len, void *data,
+                                   void **ret_data)
 {
         struct dlm_ctxt *dlm = data;
         unsigned int node;
@@ -630,7 +635,8 @@ void dlm_unregister_domain(struct dlm_ctxt *dlm)
 }
 EXPORT_SYMBOL_GPL(dlm_unregister_domain);
 
-static int dlm_query_join_handler(struct o2net_msg *msg, u32 len, void *data)
+static int dlm_query_join_handler(struct o2net_msg *msg, u32 len, void *data,
+                                  void **ret_data)
 {
         struct dlm_query_join_request *query;
         enum dlm_query_join_response response;
@@ -707,7 +713,8 @@ respond:
         return response;
 }
 
-static int dlm_assert_joined_handler(struct o2net_msg *msg, u32 len, void *data)
+static int dlm_assert_joined_handler(struct o2net_msg *msg, u32 len, void *data,
+                                     void **ret_data)
 {
         struct dlm_assert_joined *assert;
         struct dlm_ctxt *dlm = NULL;
@@ -744,7 +751,8 @@ static int dlm_assert_joined_handler(struct o2net_msg *msg, u32 len, void *data)
         return 0;
 }
 
-static int dlm_cancel_join_handler(struct o2net_msg *msg, u32 len, void *data)
+static int dlm_cancel_join_handler(struct o2net_msg *msg, u32 len, void *data,
+                                   void **ret_data)
 {
         struct dlm_cancel_join *cancel;
         struct dlm_ctxt *dlm = NULL;
@@ -1086,105 +1094,105 @@ static int dlm_register_domain_handlers(struct dlm_ctxt *dlm)
         status = o2net_register_handler(DLM_MASTER_REQUEST_MSG, dlm->key,
                                         sizeof(struct dlm_master_request),
                                         dlm_master_request_handler,
-                                        dlm, &dlm->dlm_domain_handlers);
+                                        dlm, NULL, &dlm->dlm_domain_handlers);
         if (status)
                 goto bail;
 
         status = o2net_register_handler(DLM_ASSERT_MASTER_MSG, dlm->key,
                                         sizeof(struct dlm_assert_master),
                                         dlm_assert_master_handler,
-                                        dlm, &dlm->dlm_domain_handlers);
+                                        dlm, NULL, &dlm->dlm_domain_handlers);
         if (status)
                 goto bail;
 
         status = o2net_register_handler(DLM_CREATE_LOCK_MSG, dlm->key,
                                         sizeof(struct dlm_create_lock),
                                         dlm_create_lock_handler,
-                                        dlm, &dlm->dlm_domain_handlers);
+                                        dlm, NULL, &dlm->dlm_domain_handlers);
         if (status)
                 goto bail;
 
         status = o2net_register_handler(DLM_CONVERT_LOCK_MSG, dlm->key,
                                         DLM_CONVERT_LOCK_MAX_LEN,
                                         dlm_convert_lock_handler,
-                                        dlm, &dlm->dlm_domain_handlers);
+                                        dlm, NULL, &dlm->dlm_domain_handlers);
         if (status)
                 goto bail;
 
         status = o2net_register_handler(DLM_UNLOCK_LOCK_MSG, dlm->key,
                                         DLM_UNLOCK_LOCK_MAX_LEN,
                                         dlm_unlock_lock_handler,
-                                        dlm, &dlm->dlm_domain_handlers);
+                                        dlm, NULL, &dlm->dlm_domain_handlers);
         if (status)
                 goto bail;
 
         status = o2net_register_handler(DLM_PROXY_AST_MSG, dlm->key,
                                         DLM_PROXY_AST_MAX_LEN,
                                         dlm_proxy_ast_handler,
-                                        dlm, &dlm->dlm_domain_handlers);
+                                        dlm, NULL, &dlm->dlm_domain_handlers);
         if (status)
                 goto bail;
 
         status = o2net_register_handler(DLM_EXIT_DOMAIN_MSG, dlm->key,
                                         sizeof(struct dlm_exit_domain),
                                         dlm_exit_domain_handler,
-                                        dlm, &dlm->dlm_domain_handlers);
+                                        dlm, NULL, &dlm->dlm_domain_handlers);
         if (status)
                 goto bail;
 
         status = o2net_register_handler(DLM_DEREF_LOCKRES_MSG, dlm->key,
                                         sizeof(struct dlm_deref_lockres),
                                         dlm_deref_lockres_handler,
-                                        dlm, &dlm->dlm_domain_handlers);
+                                        dlm, NULL, &dlm->dlm_domain_handlers);
         if (status)
                 goto bail;
 
         status = o2net_register_handler(DLM_MIGRATE_REQUEST_MSG, dlm->key,
                                         sizeof(struct dlm_migrate_request),
                                         dlm_migrate_request_handler,
-                                        dlm, &dlm->dlm_domain_handlers);
+                                        dlm, NULL, &dlm->dlm_domain_handlers);
         if (status)
                 goto bail;
 
         status = o2net_register_handler(DLM_MIG_LOCKRES_MSG, dlm->key,
                                         DLM_MIG_LOCKRES_MAX_LEN,
                                         dlm_mig_lockres_handler,
-                                        dlm, &dlm->dlm_domain_handlers);
+                                        dlm, NULL, &dlm->dlm_domain_handlers);
         if (status)
                 goto bail;
 
         status = o2net_register_handler(DLM_MASTER_REQUERY_MSG, dlm->key,
                                         sizeof(struct dlm_master_requery),
                                         dlm_master_requery_handler,
-                                        dlm, &dlm->dlm_domain_handlers);
+                                        dlm, NULL, &dlm->dlm_domain_handlers);
         if (status)
                 goto bail;
 
         status = o2net_register_handler(DLM_LOCK_REQUEST_MSG, dlm->key,
                                         sizeof(struct dlm_lock_request),
                                         dlm_request_all_locks_handler,
-                                        dlm, &dlm->dlm_domain_handlers);
+                                        dlm, NULL, &dlm->dlm_domain_handlers);
         if (status)
                 goto bail;
 
         status = o2net_register_handler(DLM_RECO_DATA_DONE_MSG, dlm->key,
                                         sizeof(struct dlm_reco_data_done),
                                         dlm_reco_data_done_handler,
-                                        dlm, &dlm->dlm_domain_handlers);
+                                        dlm, NULL, &dlm->dlm_domain_handlers);
         if (status)
                 goto bail;
 
         status = o2net_register_handler(DLM_BEGIN_RECO_MSG, dlm->key,
                                         sizeof(struct dlm_begin_reco),
                                         dlm_begin_reco_handler,
-                                        dlm, &dlm->dlm_domain_handlers);
+                                        dlm, NULL, &dlm->dlm_domain_handlers);
         if (status)
                 goto bail;
 
         status = o2net_register_handler(DLM_FINALIZE_RECO_MSG, dlm->key,
                                         sizeof(struct dlm_finalize_reco),
                                         dlm_finalize_reco_handler,
-                                        dlm, &dlm->dlm_domain_handlers);
+                                        dlm, NULL, &dlm->dlm_domain_handlers);
         if (status)
                 goto bail;
 
@@ -1478,21 +1486,21 @@ static int dlm_register_net_handlers(void)
         status = o2net_register_handler(DLM_QUERY_JOIN_MSG, DLM_MOD_KEY,
                                         sizeof(struct dlm_query_join_request),
                                         dlm_query_join_handler,
-                                        NULL, &dlm_join_handlers);
+                                        NULL, NULL, &dlm_join_handlers);
         if (status)
                 goto bail;
 
         status = o2net_register_handler(DLM_ASSERT_JOINED_MSG, DLM_MOD_KEY,
                                         sizeof(struct dlm_assert_joined),
                                         dlm_assert_joined_handler,
-                                        NULL, &dlm_join_handlers);
+                                        NULL, NULL, &dlm_join_handlers);
         if (status)
                 goto bail;
 
         status = o2net_register_handler(DLM_CANCEL_JOIN_MSG, DLM_MOD_KEY,
                                         sizeof(struct dlm_cancel_join),
                                         dlm_cancel_join_handler,
-                                        NULL, &dlm_join_handlers);
+                                        NULL, NULL, &dlm_join_handlers);
 
 bail:
         if (status < 0)
diff --git a/fs/ocfs2/dlm/dlmlock.c b/fs/ocfs2/dlm/dlmlock.c
index ac91a76b1e78..52578d907d9a 100644
--- a/fs/ocfs2/dlm/dlmlock.c
+++ b/fs/ocfs2/dlm/dlmlock.c
@@ -441,7 +441,8 @@ struct dlm_lock * dlm_new_lock(int type, u8 node, u64 cookie,
  *   held on exit:  none
  * returns: DLM_NORMAL, DLM_SYSERR, DLM_IVLOCKID, DLM_NOTQUEUED
  */
-int dlm_create_lock_handler(struct o2net_msg *msg, u32 len, void *data)
+int dlm_create_lock_handler(struct o2net_msg *msg, u32 len, void *data,
+                            void **ret_data)
 {
         struct dlm_ctxt *dlm = data;
         struct dlm_create_lock *create = (struct dlm_create_lock *)msg->buf;
diff --git a/fs/ocfs2/dlm/dlmmaster.c b/fs/ocfs2/dlm/dlmmaster.c
index 6cfbdf282d46..bd1268778b66 100644
--- a/fs/ocfs2/dlm/dlmmaster.c
+++ b/fs/ocfs2/dlm/dlmmaster.c
@@ -1469,7 +1469,8 @@ out:
  *
  * if possible, TRIM THIS DOWN!!!
  */
-int dlm_master_request_handler(struct o2net_msg *msg, u32 len, void *data)
+int dlm_master_request_handler(struct o2net_msg *msg, u32 len, void *data,
+                               void **ret_data)
 {
         u8 response = DLM_MASTER_RESP_MAYBE;
         struct dlm_ctxt *dlm = data;
@@ -1800,7 +1801,8 @@ again:
  *
  * if possible, TRIM THIS DOWN!!!
  */
-int dlm_assert_master_handler(struct o2net_msg *msg, u32 len, void *data)
+int dlm_assert_master_handler(struct o2net_msg *msg, u32 len, void *data,
+                              void **ret_data)
 {
         struct dlm_ctxt *dlm = data;
         struct dlm_master_list_entry *mle = NULL;
@@ -2265,7 +2267,8 @@ int dlm_drop_lockres_ref(struct dlm_ctxt *dlm, struct dlm_lock_resource *res)
         return ret;
 }
 
-int dlm_deref_lockres_handler(struct o2net_msg *msg, u32 len, void *data)
+int dlm_deref_lockres_handler(struct o2net_msg *msg, u32 len, void *data,
+                              void **ret_data)
 {
         struct dlm_ctxt *dlm = data;
         struct dlm_deref_lockres *deref = (struct dlm_deref_lockres *)msg->buf;
@@ -2948,7 +2951,8 @@ static int dlm_do_migrate_request(struct dlm_ctxt *dlm,
  * we will have no mle in the list to start with.  now we can add an mle for
  * the migration and this should be the only one found for those scanning the
  * list.  */
-int dlm_migrate_request_handler(struct o2net_msg *msg, u32 len, void *data)
+int dlm_migrate_request_handler(struct o2net_msg *msg, u32 len, void *data,
+                                void **ret_data)
 {
         struct dlm_ctxt *dlm = data;
         struct dlm_lock_resource *res = NULL;
diff --git a/fs/ocfs2/dlm/dlmrecovery.c b/fs/ocfs2/dlm/dlmrecovery.c
index 38d714645309..6d4a83d50152 100644
--- a/fs/ocfs2/dlm/dlmrecovery.c
+++ b/fs/ocfs2/dlm/dlmrecovery.c
@@ -818,7 +818,8 @@ static int dlm_request_all_locks(struct dlm_ctxt *dlm, u8 request_from,
 
 }
 
-int dlm_request_all_locks_handler(struct o2net_msg *msg, u32 len, void *data)
+int dlm_request_all_locks_handler(struct o2net_msg *msg, u32 len, void *data,
+                                  void **ret_data)
 {
         struct dlm_ctxt *dlm = data;
         struct dlm_lock_request *lr = (struct dlm_lock_request *)msg->buf;
@@ -975,7 +976,8 @@ static int dlm_send_all_done_msg(struct dlm_ctxt *dlm, u8 dead_node, u8 send_to)
 }
 
 
-int dlm_reco_data_done_handler(struct o2net_msg *msg, u32 len, void *data)
+int dlm_reco_data_done_handler(struct o2net_msg *msg, u32 len, void *data,
+                               void **ret_data)
 {
         struct dlm_ctxt *dlm = data;
         struct dlm_reco_data_done *done = (struct dlm_reco_data_done *)msg->buf;
@@ -1331,7 +1333,8 @@ error:
  * do we spin?  returning an error only delays the problem really
  */
 
-int dlm_mig_lockres_handler(struct o2net_msg *msg, u32 len, void *data)
+int dlm_mig_lockres_handler(struct o2net_msg *msg, u32 len, void *data,
+                            void **ret_data)
 {
         struct dlm_ctxt *dlm = data;
         struct dlm_migratable_lockres *mres =
@@ -1624,7 +1627,8 @@ int dlm_do_master_requery(struct dlm_ctxt *dlm, struct dlm_lock_resource *res,
 /* this function cannot error, so unless the sending
  * or receiving of the message failed, the owner can
  * be trusted */
-int dlm_master_requery_handler(struct o2net_msg *msg, u32 len, void *data)
+int dlm_master_requery_handler(struct o2net_msg *msg, u32 len, void *data,
+                               void **ret_data)
 {
         struct dlm_ctxt *dlm = data;
         struct dlm_master_requery *req = (struct dlm_master_requery *)msg->buf;
@@ -2600,7 +2604,8 @@ retry:
         return ret;
 }
 
-int dlm_begin_reco_handler(struct o2net_msg *msg, u32 len, void *data)
+int dlm_begin_reco_handler(struct o2net_msg *msg, u32 len, void *data,
+                           void **ret_data)
 {
         struct dlm_ctxt *dlm = data;
         struct dlm_begin_reco *br = (struct dlm_begin_reco *)msg->buf;
@@ -2728,7 +2733,8 @@ stage2:
         return ret;
 }
 
-int dlm_finalize_reco_handler(struct o2net_msg *msg, u32 len, void *data)
+int dlm_finalize_reco_handler(struct o2net_msg *msg, u32 len, void *data,
+                              void **ret_data)
 {
         struct dlm_ctxt *dlm = data;
         struct dlm_finalize_reco *fr = (struct dlm_finalize_reco *)msg->buf;
diff --git a/fs/ocfs2/dlm/dlmunlock.c b/fs/ocfs2/dlm/dlmunlock.c
index fc8baa3e9539..86ca085ef324 100644
--- a/fs/ocfs2/dlm/dlmunlock.c
+++ b/fs/ocfs2/dlm/dlmunlock.c
@@ -383,7 +383,8 @@ static enum dlm_status dlm_send_remote_unlock_request(struct dlm_ctxt *dlm,
  * returns: DLM_NORMAL, DLM_BADARGS, DLM_IVLOCKID,
  *          return value from dlmunlock_master
  */
-int dlm_unlock_lock_handler(struct o2net_msg *msg, u32 len, void *data)
+int dlm_unlock_lock_handler(struct o2net_msg *msg, u32 len, void *data,
+                            void **ret_data)
 {
         struct dlm_ctxt *dlm = data;
         struct dlm_unlock_lock *unlock = (struct dlm_unlock_lock *)msg->buf;