diff --git a/fs/dlm/dir.c b/fs/dlm/dir.c index b6692f81ec83..fb1981654bb2 100644 --- a/fs/dlm/dir.c +++ b/fs/dlm/dir.c @@ -101,7 +101,7 @@ int dlm_recover_directory(struct dlm_ls *ls) */ b = ls->ls_recover_buf->rc_buf; - left = ls->ls_recover_buf->rc_header.h_length; + left = le16_to_cpu(ls->ls_recover_buf->rc_header.h_length); left -= sizeof(struct dlm_rcom); for (;;) { diff --git a/fs/dlm/dlm_internal.h b/fs/dlm/dlm_internal.h index 74a9590a4dd5..776c3ed519f0 100644 --- a/fs/dlm/dlm_internal.h +++ b/fs/dlm/dlm_internal.h @@ -379,15 +379,15 @@ static inline int rsb_flag(struct dlm_rsb *r, enum rsb_flags flag) #define DLM_FIN 5 struct dlm_header { - uint32_t h_version; + __le32 h_version; union { /* for DLM_MSG and DLM_RCOM */ - uint32_t h_lockspace; + __le32 h_lockspace; /* for DLM_ACK and DLM_OPTS */ - uint32_t h_seq; + __le32 h_seq; } u; - uint32_t h_nodeid; /* nodeid of sender */ - uint16_t h_length; + __le32 h_nodeid; /* nodeid of sender */ + __le16 h_length; uint8_t h_cmd; /* DLM_MSG, DLM_RCOM */ uint8_t h_pad; }; @@ -409,24 +409,24 @@ struct dlm_header { struct dlm_message { struct dlm_header m_header; - uint32_t m_type; /* DLM_MSG_ */ - uint32_t m_nodeid; - uint32_t m_pid; - uint32_t m_lkid; /* lkid on sender */ - uint32_t m_remid; /* lkid on receiver */ - uint32_t m_parent_lkid; - uint32_t m_parent_remid; - uint32_t m_exflags; - uint32_t m_sbflags; - uint32_t m_flags; - uint32_t m_lvbseq; - uint32_t m_hash; - int m_status; - int m_grmode; - int m_rqmode; - int m_bastmode; - int m_asts; - int m_result; /* 0 or -EXXX */ + __le32 m_type; /* DLM_MSG_ */ + __le32 m_nodeid; + __le32 m_pid; + __le32 m_lkid; /* lkid on sender */ + __le32 m_remid; /* lkid on receiver */ + __le32 m_parent_lkid; + __le32 m_parent_remid; + __le32 m_exflags; + __le32 m_sbflags; + __le32 m_flags; + __le32 m_lvbseq; + __le32 m_hash; + __le32 m_status; + __le32 m_grmode; + __le32 m_rqmode; + __le32 m_bastmode; + __le32 m_asts; + __le32 m_result; /* 0 or -EXXX */ char m_extra[]; /* name or lvb */ }; @@ -451,18 +451,18 @@ struct dlm_message { struct dlm_rcom { struct dlm_header rc_header; - uint32_t rc_type; /* DLM_RCOM_ */ - int rc_result; /* multi-purpose */ - uint64_t rc_id; /* match reply with request */ - uint64_t rc_seq; /* sender's ls_recover_seq */ - uint64_t rc_seq_reply; /* remote ls_recover_seq */ + __le32 rc_type; /* DLM_RCOM_ */ + __le32 rc_result; /* multi-purpose */ + __le64 rc_id; /* match reply with request */ + __le64 rc_seq; /* sender's ls_recover_seq */ + __le64 rc_seq_reply; /* remote ls_recover_seq */ char rc_buf[]; }; struct dlm_opt_header { - uint16_t t_type; - uint16_t t_length; - uint32_t t_pad; + __le16 t_type; + __le16 t_length; + __le32 t_pad; /* need to be 8 byte aligned */ char t_value[]; }; @@ -472,8 +472,8 @@ struct dlm_opts { struct dlm_header o_header; uint8_t o_nextcmd; uint8_t o_pad; - uint16_t o_optlen; - uint32_t o_pad2; + __le16 o_optlen; + __le32 o_pad2; char o_opts[]; }; diff --git a/fs/dlm/lock.c b/fs/dlm/lock.c index bdb51d209ba2..226822f49d30 100644 --- a/fs/dlm/lock.c +++ b/fs/dlm/lock.c @@ -350,10 +350,12 @@ static void put_rsb(struct dlm_rsb *r) { struct dlm_ls *ls = r->res_ls; uint32_t bucket = r->res_bucket; + int rv; - spin_lock(&ls->ls_rsbtbl[bucket].lock); - kref_put(&r->res_ref, toss_rsb); - spin_unlock(&ls->ls_rsbtbl[bucket].lock); + rv = kref_put_lock(&r->res_ref, toss_rsb, + &ls->ls_rsbtbl[bucket].lock); + if (rv) + spin_unlock(&ls->ls_rsbtbl[bucket].lock); } void dlm_put_rsb(struct dlm_rsb *r) @@ -602,7 +604,6 @@ static int find_rsb_dir(struct dlm_ls *ls, char *name, int len, */ kref_get(&r->res_ref); - error = 0; goto out_unlock; @@ -880,6 +881,88 @@ static int validate_master_nodeid(struct dlm_ls *ls, struct dlm_rsb *r, } } +static void __dlm_master_lookup(struct dlm_ls *ls, struct dlm_rsb *r, int our_nodeid, + int from_nodeid, bool toss_list, unsigned int flags, + int *r_nodeid, int *result) +{ + int fix_master = (flags & DLM_LU_RECOVER_MASTER); + int from_master = (flags & DLM_LU_RECOVER_DIR); + + if (r->res_dir_nodeid != our_nodeid) { + /* should not happen, but may as well fix it and carry on */ + log_error(ls, "%s res_dir %d our %d %s", __func__, + r->res_dir_nodeid, our_nodeid, r->res_name); + r->res_dir_nodeid = our_nodeid; + } + + if (fix_master && dlm_is_removed(ls, r->res_master_nodeid)) { + /* Recovery uses this function to set a new master when + * the previous master failed. Setting NEW_MASTER will + * force dlm_recover_masters to call recover_master on this + * rsb even though the res_nodeid is no longer removed. + */ + + r->res_master_nodeid = from_nodeid; + r->res_nodeid = from_nodeid; + rsb_set_flag(r, RSB_NEW_MASTER); + + if (toss_list) { + /* I don't think we should ever find it on toss list. */ + log_error(ls, "%s fix_master on toss", __func__); + dlm_dump_rsb(r); + } + } + + if (from_master && (r->res_master_nodeid != from_nodeid)) { + /* this will happen if from_nodeid became master during + * a previous recovery cycle, and we aborted the previous + * cycle before recovering this master value + */ + + log_limit(ls, "%s from_master %d master_nodeid %d res_nodeid %d first %x %s", + __func__, from_nodeid, r->res_master_nodeid, + r->res_nodeid, r->res_first_lkid, r->res_name); + + if (r->res_master_nodeid == our_nodeid) { + log_error(ls, "from_master %d our_master", from_nodeid); + dlm_dump_rsb(r); + goto ret_assign; + } + + r->res_master_nodeid = from_nodeid; + r->res_nodeid = from_nodeid; + rsb_set_flag(r, RSB_NEW_MASTER); + } + + if (!r->res_master_nodeid) { + /* this will happen if recovery happens while we're looking + * up the master for this rsb + */ + + log_debug(ls, "%s master 0 to %d first %x %s", __func__, + from_nodeid, r->res_first_lkid, r->res_name); + r->res_master_nodeid = from_nodeid; + r->res_nodeid = from_nodeid; + } + + if (!from_master && !fix_master && + (r->res_master_nodeid == from_nodeid)) { + /* this can happen when the master sends remove, the dir node + * finds the rsb on the keep list and ignores the remove, + * and the former master sends a lookup + */ + + log_limit(ls, "%s from master %d flags %x first %x %s", + __func__, from_nodeid, flags, r->res_first_lkid, + r->res_name); + } + + ret_assign: + *r_nodeid = r->res_master_nodeid; + if (result) + *result = DLM_LU_MATCH; +} + /* * We're the dir node for this res and another node wants to know the * master nodeid. During normal operation (non recovery) this is only @@ -914,10 +997,8 @@ int dlm_master_lookup(struct dlm_ls *ls, int from_nodeid, char *name, int len, { struct dlm_rsb *r = NULL; uint32_t hash, b; - int from_master = (flags & DLM_LU_RECOVER_DIR); - int fix_master = (flags & DLM_LU_RECOVER_MASTER); int our_nodeid = dlm_our_nodeid(); - int dir_nodeid, error, toss_list = 0; + int dir_nodeid, error; if (len > DLM_RESNAME_MAXLEN) return -EINVAL; @@ -949,12 +1030,21 @@ int dlm_master_lookup(struct dlm_ls *ls, int from_nodeid, char *name, int len, error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r); if (!error) { /* because the rsb is active, we need to lock_rsb before - checking/changing re_master_nodeid */ + * checking/changing re_master_nodeid + */ hold_rsb(r); spin_unlock(&ls->ls_rsbtbl[b].lock); lock_rsb(r); - goto found; + + __dlm_master_lookup(ls, r, our_nodeid, from_nodeid, false, + flags, r_nodeid, result); + + /* the rsb was active */ + unlock_rsb(r); + put_rsb(r); + + return 0; } error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r); @@ -962,90 +1052,16 @@ int dlm_master_lookup(struct dlm_ls *ls, int from_nodeid, char *name, int len, goto not_found; /* because the rsb is inactive (on toss list), it's not refcounted - and lock_rsb is not used, but is protected by the rsbtbl lock */ + * and lock_rsb is not used, but is protected by the rsbtbl lock + */ - toss_list = 1; - found: - if (r->res_dir_nodeid != our_nodeid) { - /* should not happen, but may as well fix it and carry on */ - log_error(ls, "dlm_master_lookup res_dir %d our %d %s", - r->res_dir_nodeid, our_nodeid, r->res_name); - r->res_dir_nodeid = our_nodeid; - } + __dlm_master_lookup(ls, r, our_nodeid, from_nodeid, true, flags, + r_nodeid, result); - if (fix_master && dlm_is_removed(ls, r->res_master_nodeid)) { - /* Recovery uses this function to set a new master when - the previous master failed. Setting NEW_MASTER will - force dlm_recover_masters to call recover_master on this - rsb even though the res_nodeid is no longer removed. */ + r->res_toss_time = jiffies; + /* the rsb was inactive (on toss list) */ + spin_unlock(&ls->ls_rsbtbl[b].lock); - r->res_master_nodeid = from_nodeid; - r->res_nodeid = from_nodeid; - rsb_set_flag(r, RSB_NEW_MASTER); - - if (toss_list) { - /* I don't think we should ever find it on toss list. */ - log_error(ls, "dlm_master_lookup fix_master on toss"); - dlm_dump_rsb(r); - } - } - - if (from_master && (r->res_master_nodeid != from_nodeid)) { - /* this will happen if from_nodeid became master during - a previous recovery cycle, and we aborted the previous - cycle before recovering this master value */ - - log_limit(ls, "dlm_master_lookup from_master %d " - "master_nodeid %d res_nodeid %d first %x %s", - from_nodeid, r->res_master_nodeid, r->res_nodeid, - r->res_first_lkid, r->res_name); - - if (r->res_master_nodeid == our_nodeid) { - log_error(ls, "from_master %d our_master", from_nodeid); - dlm_dump_rsb(r); - goto out_found; - } - - r->res_master_nodeid = from_nodeid; - r->res_nodeid = from_nodeid; - rsb_set_flag(r, RSB_NEW_MASTER); - } - - if (!r->res_master_nodeid) { - /* this will happen if recovery happens while we're looking - up the master for this rsb */ - - log_debug(ls, "dlm_master_lookup master 0 to %d first %x %s", - from_nodeid, r->res_first_lkid, r->res_name); - r->res_master_nodeid = from_nodeid; - r->res_nodeid = from_nodeid; - } - - if (!from_master && !fix_master && - (r->res_master_nodeid == from_nodeid)) { - /* this can happen when the master sends remove, the dir node - finds the rsb on the keep list and ignores the remove, - and the former master sends a lookup */ - - log_limit(ls, "dlm_master_lookup from master %d flags %x " - "first %x %s", from_nodeid, flags, - r->res_first_lkid, r->res_name); - } - - out_found: - *r_nodeid = r->res_master_nodeid; - if (result) - *result = DLM_LU_MATCH; - - if (toss_list) { - r->res_toss_time = jiffies; - /* the rsb was inactive (on toss list) */ - spin_unlock(&ls->ls_rsbtbl[b].lock); - } else { - /* the rsb was active */ - unlock_rsb(r); - put_rsb(r); - } return 0; not_found: @@ -1076,7 +1092,6 @@ int dlm_master_lookup(struct dlm_ls *ls, int from_nodeid, char *name, int len, if (result) *result = DLM_LU_ADD; *r_nodeid = from_nodeid; - error = 0; out_unlock: spin_unlock(&ls->ls_rsbtbl[b].lock); return error; @@ -1253,9 +1268,11 @@ static void kill_lkb(struct kref *kref) static int __put_lkb(struct dlm_ls *ls, struct dlm_lkb *lkb) { uint32_t lkid = lkb->lkb_id; + int rv; - spin_lock(&ls->ls_lkbidr_spin); - if (kref_put(&lkb->lkb_ref, kill_lkb)) { + rv = kref_put_lock(&lkb->lkb_ref, kill_lkb, + &ls->ls_lkbidr_spin); + if (rv) { idr_remove(&ls->ls_lkbidr, lkid); spin_unlock(&ls->ls_lkbidr_spin); @@ -1265,11 +1282,9 @@ static int __put_lkb(struct dlm_ls *ls, struct dlm_lkb *lkb) if (lkb->lkb_lvbptr && is_master_copy(lkb)) dlm_free_lvb(lkb->lkb_lvbptr); dlm_free_lkb(lkb); - return 1; - } else { - spin_unlock(&ls->ls_lkbidr_spin); - return 0; } + + return rv; } int dlm_put_lkb(struct dlm_lkb *lkb) @@ -1306,13 +1321,17 @@ static inline void unhold_lkb(struct dlm_lkb *lkb) static void lkb_add_ordered(struct list_head *new, struct list_head *head, int mode) { - struct dlm_lkb *lkb = NULL; + struct dlm_lkb *lkb = NULL, *iter; - list_for_each_entry(lkb, head, lkb_statequeue) - if (lkb->lkb_rqmode < mode) + list_for_each_entry(iter, head, lkb_statequeue) + if (iter->lkb_rqmode < mode) { + lkb = iter; + list_add_tail(new, &iter->lkb_statequeue); break; + } - __list_add(new, lkb->lkb_statequeue.prev, &lkb->lkb_statequeue); + if (!lkb) + list_add_tail(new, head); } /* add/remove lkb to rsb's grant/convert/wait queue */ @@ -1559,6 +1578,7 @@ static int _remove_from_waiters(struct dlm_lkb *lkb, int mstype, lkb->lkb_wait_type = 0; lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL; lkb->lkb_wait_count--; + unhold_lkb(lkb); goto out_del; } @@ -1571,8 +1591,8 @@ static int _remove_from_waiters(struct dlm_lkb *lkb, int mstype, } log_error(ls, "remwait error %x remote %d %x msg %d flags %x no wait", - lkb->lkb_id, ms ? ms->m_header.h_nodeid : 0, lkb->lkb_remid, - mstype, lkb->lkb_flags); + lkb->lkb_id, ms ? le32_to_cpu(ms->m_header.h_nodeid) : 0, + lkb->lkb_remid, mstype, lkb->lkb_flags); return -1; out_del: @@ -1585,6 +1605,7 @@ static int _remove_from_waiters(struct dlm_lkb *lkb, int mstype, log_error(ls, "remwait error %x reply %d wait_type %d overlap", lkb->lkb_id, mstype, lkb->lkb_wait_type); lkb->lkb_wait_count--; + unhold_lkb(lkb); lkb->lkb_wait_type = 0; } @@ -1617,10 +1638,10 @@ static int remove_from_waiters_ms(struct dlm_lkb *lkb, struct dlm_message *ms) struct dlm_ls *ls = lkb->lkb_resource->res_ls; int error; - if (ms->m_flags != DLM_IFL_STUB_MS) + if (ms->m_flags != cpu_to_le32(DLM_IFL_STUB_MS)) mutex_lock(&ls->ls_waiters_mutex); - error = _remove_from_waiters(lkb, ms->m_type, ms); - if (ms->m_flags != DLM_IFL_STUB_MS) + error = _remove_from_waiters(lkb, le32_to_cpu(ms->m_type), ms); + if (ms->m_flags != cpu_to_le32(DLM_IFL_STUB_MS)) mutex_unlock(&ls->ls_waiters_mutex); return error; } @@ -1795,7 +1816,6 @@ static void shrink_bucket(struct dlm_ls *ls, int b) memcpy(ls->ls_remove_name, name, DLM_RESNAME_MAXLEN); spin_unlock(&ls->ls_remove_spin); spin_unlock(&ls->ls_rsbtbl[b].lock); - wake_up(&ls->ls_remove_wait); send_remove(r); @@ -1804,6 +1824,7 @@ static void shrink_bucket(struct dlm_ls *ls, int b) ls->ls_remove_len = 0; memset(ls->ls_remove_name, 0, DLM_RESNAME_MAXLEN); spin_unlock(&ls->ls_remove_spin); + wake_up(&ls->ls_remove_wait); dlm_free_rsb(r); } @@ -1866,7 +1887,7 @@ static void del_timeout(struct dlm_lkb *lkb) void dlm_scan_timeout(struct dlm_ls *ls) { struct dlm_rsb *r; - struct dlm_lkb *lkb; + struct dlm_lkb *lkb = NULL, *iter; int do_cancel, do_warn; s64 wait_us; @@ -1877,27 +1898,28 @@ void dlm_scan_timeout(struct dlm_ls *ls) do_cancel = 0; do_warn = 0; mutex_lock(&ls->ls_timeout_mutex); - list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list) { + list_for_each_entry(iter, &ls->ls_timeout, lkb_time_list) { wait_us = ktime_to_us(ktime_sub(ktime_get(), - lkb->lkb_timestamp)); + iter->lkb_timestamp)); - if ((lkb->lkb_exflags & DLM_LKF_TIMEOUT) && - wait_us >= (lkb->lkb_timeout_cs * 10000)) + if ((iter->lkb_exflags & DLM_LKF_TIMEOUT) && + wait_us >= (iter->lkb_timeout_cs * 10000)) do_cancel = 1; - if ((lkb->lkb_flags & DLM_IFL_WATCH_TIMEWARN) && + if ((iter->lkb_flags & DLM_IFL_WATCH_TIMEWARN) && wait_us >= dlm_config.ci_timewarn_cs * 10000) do_warn = 1; if (!do_cancel && !do_warn) continue; - hold_lkb(lkb); + hold_lkb(iter); + lkb = iter; break; } mutex_unlock(&ls->ls_timeout_mutex); - if (!do_cancel && !do_warn) + if (!lkb) break; r = lkb->lkb_resource; @@ -2051,7 +2073,7 @@ static void set_lvb_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb, if (len > r->res_ls->ls_lvblen) len = r->res_ls->ls_lvblen; memcpy(lkb->lkb_lvbptr, ms->m_extra, len); - lkb->lkb_lvbseq = ms->m_lvbseq; + lkb->lkb_lvbseq = le32_to_cpu(ms->m_lvbseq); } } @@ -2182,10 +2204,10 @@ static void munge_demoted(struct dlm_lkb *lkb) static void munge_altmode(struct dlm_lkb *lkb, struct dlm_message *ms) { - if (ms->m_type != DLM_MSG_REQUEST_REPLY && - ms->m_type != DLM_MSG_GRANT) { + if (ms->m_type != cpu_to_le32(DLM_MSG_REQUEST_REPLY) && + ms->m_type != cpu_to_le32(DLM_MSG_GRANT)) { log_print("munge_altmode %x invalid reply type %d", - lkb->lkb_id, ms->m_type); + lkb->lkb_id, le32_to_cpu(ms->m_type)); return; } @@ -2912,7 +2934,8 @@ static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb, if (lkb->lkb_status != DLM_LKSTS_GRANTED) goto out; - if (lkb->lkb_wait_type) + /* lock not allowed if there's any op in progress */ + if (lkb->lkb_wait_type || lkb->lkb_wait_count) goto out; if (is_overlap(lkb)) @@ -3563,13 +3586,13 @@ static int _create_message(struct dlm_ls *ls, int mb_len, ms = (struct dlm_message *) mb; - ms->m_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR); - ms->m_header.u.h_lockspace = ls->ls_global_id; - ms->m_header.h_nodeid = dlm_our_nodeid(); - ms->m_header.h_length = mb_len; + ms->m_header.h_version = cpu_to_le32(DLM_HEADER_MAJOR | DLM_HEADER_MINOR); + ms->m_header.u.h_lockspace = cpu_to_le32(ls->ls_global_id); + ms->m_header.h_nodeid = cpu_to_le32(dlm_our_nodeid()); + ms->m_header.h_length = cpu_to_le16(mb_len); ms->m_header.h_cmd = DLM_MSG; - ms->m_type = mstype; + ms->m_type = cpu_to_le32(mstype); *mh_ret = mh; *ms_ret = ms; @@ -3608,7 +3631,6 @@ static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb, static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms) { - dlm_message_out(ms); dlm_midcomms_commit_mhandle(mh); return 0; } @@ -3616,40 +3638,40 @@ static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms) static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb, struct dlm_message *ms) { - ms->m_nodeid = lkb->lkb_nodeid; - ms->m_pid = lkb->lkb_ownpid; - ms->m_lkid = lkb->lkb_id; - ms->m_remid = lkb->lkb_remid; - ms->m_exflags = lkb->lkb_exflags; - ms->m_sbflags = lkb->lkb_sbflags; - ms->m_flags = lkb->lkb_flags; - ms->m_lvbseq = lkb->lkb_lvbseq; - ms->m_status = lkb->lkb_status; - ms->m_grmode = lkb->lkb_grmode; - ms->m_rqmode = lkb->lkb_rqmode; - ms->m_hash = r->res_hash; + ms->m_nodeid = cpu_to_le32(lkb->lkb_nodeid); + ms->m_pid = cpu_to_le32(lkb->lkb_ownpid); + ms->m_lkid = cpu_to_le32(lkb->lkb_id); + ms->m_remid = cpu_to_le32(lkb->lkb_remid); + ms->m_exflags = cpu_to_le32(lkb->lkb_exflags); + ms->m_sbflags = cpu_to_le32(lkb->lkb_sbflags); + ms->m_flags = cpu_to_le32(lkb->lkb_flags); + ms->m_lvbseq = cpu_to_le32(lkb->lkb_lvbseq); + ms->m_status = cpu_to_le32(lkb->lkb_status); + ms->m_grmode = cpu_to_le32(lkb->lkb_grmode); + ms->m_rqmode = cpu_to_le32(lkb->lkb_rqmode); + ms->m_hash = cpu_to_le32(r->res_hash); /* m_result and m_bastmode are set from function args, not from lkb fields */ if (lkb->lkb_bastfn) - ms->m_asts |= DLM_CB_BAST; + ms->m_asts |= cpu_to_le32(DLM_CB_BAST); if (lkb->lkb_astfn) - ms->m_asts |= DLM_CB_CAST; + ms->m_asts |= cpu_to_le32(DLM_CB_CAST); /* compare with switch in create_message; send_remove() doesn't use send_args() */ switch (ms->m_type) { - case DLM_MSG_REQUEST: - case DLM_MSG_LOOKUP: + case cpu_to_le32(DLM_MSG_REQUEST): + case cpu_to_le32(DLM_MSG_LOOKUP): memcpy(ms->m_extra, r->res_name, r->res_length); break; - case DLM_MSG_CONVERT: - case DLM_MSG_UNLOCK: - case DLM_MSG_REQUEST_REPLY: - case DLM_MSG_CONVERT_REPLY: - case DLM_MSG_GRANT: + case cpu_to_le32(DLM_MSG_CONVERT): + case cpu_to_le32(DLM_MSG_UNLOCK): + case cpu_to_le32(DLM_MSG_REQUEST_REPLY): + case cpu_to_le32(DLM_MSG_CONVERT_REPLY): + case cpu_to_le32(DLM_MSG_GRANT): if (!lkb->lkb_lvbptr) break; memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen); @@ -3699,8 +3721,8 @@ static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb) /* down conversions go without a reply from the master */ if (!error && down_conversion(lkb)) { remove_from_waiters(lkb, DLM_MSG_CONVERT_REPLY); - r->res_ls->ls_stub_ms.m_flags = DLM_IFL_STUB_MS; - r->res_ls->ls_stub_ms.m_type = DLM_MSG_CONVERT_REPLY; + r->res_ls->ls_stub_ms.m_flags = cpu_to_le32(DLM_IFL_STUB_MS); + r->res_ls->ls_stub_ms.m_type = cpu_to_le32(DLM_MSG_CONVERT_REPLY); r->res_ls->ls_stub_ms.m_result = 0; __receive_convert_reply(r, lkb, &r->res_ls->ls_stub_ms); } @@ -3757,7 +3779,7 @@ static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode) send_args(r, lkb, ms); - ms->m_bastmode = mode; + ms->m_bastmode = cpu_to_le32(mode); error = send_message(mh, ms); out: @@ -3805,7 +3827,7 @@ static int send_remove(struct dlm_rsb *r) goto out; memcpy(ms->m_extra, r->res_name, r->res_length); - ms->m_hash = r->res_hash; + ms->m_hash = cpu_to_le32(r->res_hash); error = send_message(mh, ms); out: @@ -3827,7 +3849,7 @@ static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, send_args(r, lkb, ms); - ms->m_result = rv; + ms->m_result = cpu_to_le32(to_dlm_errno(rv)); error = send_message(mh, ms); out: @@ -3860,15 +3882,15 @@ static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in, struct dlm_rsb *r = &ls->ls_stub_rsb; struct dlm_message *ms; struct dlm_mhandle *mh; - int error, nodeid = ms_in->m_header.h_nodeid; + int error, nodeid = le32_to_cpu(ms_in->m_header.h_nodeid); error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh); if (error) goto out; ms->m_lkid = ms_in->m_lkid; - ms->m_result = rv; - ms->m_nodeid = ret_nodeid; + ms->m_result = cpu_to_le32(to_dlm_errno(rv)); + ms->m_nodeid = cpu_to_le32(ret_nodeid); error = send_message(mh, ms); out: @@ -3881,25 +3903,26 @@ static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in, static void receive_flags(struct dlm_lkb *lkb, struct dlm_message *ms) { - lkb->lkb_exflags = ms->m_exflags; - lkb->lkb_sbflags = ms->m_sbflags; + lkb->lkb_exflags = le32_to_cpu(ms->m_exflags); + lkb->lkb_sbflags = le32_to_cpu(ms->m_sbflags); lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) | - (ms->m_flags & 0x0000FFFF); + (le32_to_cpu(ms->m_flags) & 0x0000FFFF); } static void receive_flags_reply(struct dlm_lkb *lkb, struct dlm_message *ms) { - if (ms->m_flags == DLM_IFL_STUB_MS) + if (ms->m_flags == cpu_to_le32(DLM_IFL_STUB_MS)) return; - lkb->lkb_sbflags = ms->m_sbflags; + lkb->lkb_sbflags = le32_to_cpu(ms->m_sbflags); lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) | - (ms->m_flags & 0x0000FFFF); + (le32_to_cpu(ms->m_flags) & 0x0000FFFF); } static int receive_extralen(struct dlm_message *ms) { - return (ms->m_header.h_length - sizeof(struct dlm_message)); + return (le16_to_cpu(ms->m_header.h_length) - + sizeof(struct dlm_message)); } static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb, @@ -3933,14 +3956,14 @@ static void fake_astfn(void *astparam) static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb, struct dlm_message *ms) { - lkb->lkb_nodeid = ms->m_header.h_nodeid; - lkb->lkb_ownpid = ms->m_pid; - lkb->lkb_remid = ms->m_lkid; + lkb->lkb_nodeid = le32_to_cpu(ms->m_header.h_nodeid); + lkb->lkb_ownpid = le32_to_cpu(ms->m_pid); + lkb->lkb_remid = le32_to_cpu(ms->m_lkid); lkb->lkb_grmode = DLM_LOCK_IV; - lkb->lkb_rqmode = ms->m_rqmode; + lkb->lkb_rqmode = le32_to_cpu(ms->m_rqmode); - lkb->lkb_bastfn = (ms->m_asts & DLM_CB_BAST) ? &fake_bastfn : NULL; - lkb->lkb_astfn = (ms->m_asts & DLM_CB_CAST) ? &fake_astfn : NULL; + lkb->lkb_bastfn = (ms->m_asts & cpu_to_le32(DLM_CB_BAST)) ? &fake_bastfn : NULL; + lkb->lkb_astfn = (ms->m_asts & cpu_to_le32(DLM_CB_CAST)) ? &fake_astfn : NULL; if (lkb->lkb_exflags & DLM_LKF_VALBLK) { /* lkb was just created so there won't be an lvb yet */ @@ -3961,8 +3984,8 @@ static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb, if (receive_lvb(ls, lkb, ms)) return -ENOMEM; - lkb->lkb_rqmode = ms->m_rqmode; - lkb->lkb_lvbseq = ms->m_lvbseq; + lkb->lkb_rqmode = le32_to_cpu(ms->m_rqmode); + lkb->lkb_lvbseq = le32_to_cpu(ms->m_lvbseq); return 0; } @@ -3981,8 +4004,8 @@ static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb, static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms) { struct dlm_lkb *lkb = &ls->ls_stub_lkb; - lkb->lkb_nodeid = ms->m_header.h_nodeid; - lkb->lkb_remid = ms->m_lkid; + lkb->lkb_nodeid = le32_to_cpu(ms->m_header.h_nodeid); + lkb->lkb_remid = le32_to_cpu(ms->m_lkid); } /* This is called after the rsb is locked so that we can safely inspect @@ -3990,11 +4013,12 @@ static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms) static int validate_message(struct dlm_lkb *lkb, struct dlm_message *ms) { - int from = ms->m_header.h_nodeid; + int from = le32_to_cpu(ms->m_header.h_nodeid); int error = 0; /* currently mixing of user/kernel locks are not supported */ - if (ms->m_flags & DLM_IFL_USER && ~lkb->lkb_flags & DLM_IFL_USER) { + if (ms->m_flags & cpu_to_le32(DLM_IFL_USER) && + ~lkb->lkb_flags & DLM_IFL_USER) { log_error(lkb->lkb_resource->res_ls, "got user dlm message for a kernel lock"); error = -EINVAL; @@ -4002,23 +4026,23 @@ static int validate_message(struct dlm_lkb *lkb, struct dlm_message *ms) } switch (ms->m_type) { - case DLM_MSG_CONVERT: - case DLM_MSG_UNLOCK: - case DLM_MSG_CANCEL: + case cpu_to_le32(DLM_MSG_CONVERT): + case cpu_to_le32(DLM_MSG_UNLOCK): + case cpu_to_le32(DLM_MSG_CANCEL): if (!is_master_copy(lkb) || lkb->lkb_nodeid != from) error = -EINVAL; break; - case DLM_MSG_CONVERT_REPLY: - case DLM_MSG_UNLOCK_REPLY: - case DLM_MSG_CANCEL_REPLY: - case DLM_MSG_GRANT: - case DLM_MSG_BAST: + case cpu_to_le32(DLM_MSG_CONVERT_REPLY): + case cpu_to_le32(DLM_MSG_UNLOCK_REPLY): + case cpu_to_le32(DLM_MSG_CANCEL_REPLY): + case cpu_to_le32(DLM_MSG_GRANT): + case cpu_to_le32(DLM_MSG_BAST): if (!is_process_copy(lkb) || lkb->lkb_nodeid != from) error = -EINVAL; break; - case DLM_MSG_REQUEST_REPLY: + case cpu_to_le32(DLM_MSG_REQUEST_REPLY): if (!is_process_copy(lkb)) error = -EINVAL; else if (lkb->lkb_nodeid != -1 && lkb->lkb_nodeid != from) @@ -4033,8 +4057,8 @@ static int validate_message(struct dlm_lkb *lkb, struct dlm_message *ms) if (error) log_error(lkb->lkb_resource->res_ls, "ignore invalid message %d from %d %x %x %x %d", - ms->m_type, from, lkb->lkb_id, lkb->lkb_remid, - lkb->lkb_flags, lkb->lkb_nodeid); + le32_to_cpu(ms->m_type), from, lkb->lkb_id, + lkb->lkb_remid, lkb->lkb_flags, lkb->lkb_nodeid); return error; } @@ -4079,22 +4103,23 @@ static void send_repeat_remove(struct dlm_ls *ls, char *ms_name, int len) memcpy(ls->ls_remove_name, name, DLM_RESNAME_MAXLEN); spin_unlock(&ls->ls_remove_spin); spin_unlock(&ls->ls_rsbtbl[b].lock); - wake_up(&ls->ls_remove_wait); rv = _create_message(ls, sizeof(struct dlm_message) + len, dir_nodeid, DLM_MSG_REMOVE, &ms, &mh); if (rv) - return; + goto out; memcpy(ms->m_extra, name, len); - ms->m_hash = hash; + ms->m_hash = cpu_to_le32(hash); send_message(mh, ms); +out: spin_lock(&ls->ls_remove_spin); ls->ls_remove_len = 0; memset(ls->ls_remove_name, 0, DLM_RESNAME_MAXLEN); spin_unlock(&ls->ls_remove_spin); + wake_up(&ls->ls_remove_wait); } static int receive_request(struct dlm_ls *ls, struct dlm_message *ms) @@ -4104,7 +4129,7 @@ static int receive_request(struct dlm_ls *ls, struct dlm_message *ms) int from_nodeid; int error, namelen = 0; - from_nodeid = ms->m_header.h_nodeid; + from_nodeid = le32_to_cpu(ms->m_header.h_nodeid); error = create_lkb(ls, &lkb); if (error) @@ -4177,7 +4202,7 @@ static int receive_request(struct dlm_ls *ls, struct dlm_message *ms) if (error != -ENOTBLK) { log_limit(ls, "receive_request %x from %d %d", - ms->m_lkid, from_nodeid, error); + le32_to_cpu(ms->m_lkid), from_nodeid, error); } if (namelen && error == -EBADR) { @@ -4196,15 +4221,16 @@ static int receive_convert(struct dlm_ls *ls, struct dlm_message *ms) struct dlm_rsb *r; int error, reply = 1; - error = find_lkb(ls, ms->m_remid, &lkb); + error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb); if (error) goto fail; - if (lkb->lkb_remid != ms->m_lkid) { + if (lkb->lkb_remid != le32_to_cpu(ms->m_lkid)) { log_error(ls, "receive_convert %x remid %x recover_seq %llu " "remote %d %x", lkb->lkb_id, lkb->lkb_remid, (unsigned long long)lkb->lkb_recover_seq, - ms->m_header.h_nodeid, ms->m_lkid); + le32_to_cpu(ms->m_header.h_nodeid), + le32_to_cpu(ms->m_lkid)); error = -ENOENT; dlm_put_lkb(lkb); goto fail; @@ -4251,14 +4277,15 @@ static int receive_unlock(struct dlm_ls *ls, struct dlm_message *ms) struct dlm_rsb *r; int error; - error = find_lkb(ls, ms->m_remid, &lkb); + error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb); if (error) goto fail; - if (lkb->lkb_remid != ms->m_lkid) { + if (lkb->lkb_remid != le32_to_cpu(ms->m_lkid)) { log_error(ls, "receive_unlock %x remid %x remote %d %x", lkb->lkb_id, lkb->lkb_remid, - ms->m_header.h_nodeid, ms->m_lkid); + le32_to_cpu(ms->m_header.h_nodeid), + le32_to_cpu(ms->m_lkid)); error = -ENOENT; dlm_put_lkb(lkb); goto fail; @@ -4302,7 +4329,7 @@ static int receive_cancel(struct dlm_ls *ls, struct dlm_message *ms) struct dlm_rsb *r; int error; - error = find_lkb(ls, ms->m_remid, &lkb); + error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb); if (error) goto fail; @@ -4338,7 +4365,7 @@ static int receive_grant(struct dlm_ls *ls, struct dlm_message *ms) struct dlm_rsb *r; int error; - error = find_lkb(ls, ms->m_remid, &lkb); + error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb); if (error) return error; @@ -4369,7 +4396,7 @@ static int receive_bast(struct dlm_ls *ls, struct dlm_message *ms) struct dlm_rsb *r; int error; - error = find_lkb(ls, ms->m_remid, &lkb); + error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb); if (error) return error; @@ -4382,8 +4409,8 @@ static int receive_bast(struct dlm_ls *ls, struct dlm_message *ms) if (error) goto out; - queue_bast(r, lkb, ms->m_bastmode); - lkb->lkb_highbast = ms->m_bastmode; + queue_bast(r, lkb, le32_to_cpu(ms->m_bastmode)); + lkb->lkb_highbast = le32_to_cpu(ms->m_bastmode); out: unlock_rsb(r); put_rsb(r); @@ -4395,7 +4422,7 @@ static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms) { int len, error, ret_nodeid, from_nodeid, our_nodeid; - from_nodeid = ms->m_header.h_nodeid; + from_nodeid = le32_to_cpu(ms->m_header.h_nodeid); our_nodeid = dlm_our_nodeid(); len = receive_extralen(ms); @@ -4418,7 +4445,7 @@ static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms) uint32_t hash, b; int rv, len, dir_nodeid, from_nodeid; - from_nodeid = ms->m_header.h_nodeid; + from_nodeid = le32_to_cpu(ms->m_header.h_nodeid); len = receive_extralen(ms); @@ -4428,7 +4455,7 @@ static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms) return; } - dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash); + dir_nodeid = dlm_hash2nodeid(ls, le32_to_cpu(ms->m_hash)); if (dir_nodeid != dlm_our_nodeid()) { log_error(ls, "receive_remove from %d bad nodeid %d", from_nodeid, dir_nodeid); @@ -4501,7 +4528,7 @@ static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms) static void receive_purge(struct dlm_ls *ls, struct dlm_message *ms) { - do_purge(ls, ms->m_nodeid, ms->m_pid); + do_purge(ls, le32_to_cpu(ms->m_nodeid), le32_to_cpu(ms->m_pid)); } static int receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms) @@ -4509,9 +4536,9 @@ static int receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms) struct dlm_lkb *lkb; struct dlm_rsb *r; int error, mstype, result; - int from_nodeid = ms->m_header.h_nodeid; + int from_nodeid = le32_to_cpu(ms->m_header.h_nodeid); - error = find_lkb(ls, ms->m_remid, &lkb); + error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb); if (error) return error; @@ -4527,7 +4554,8 @@ static int receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms) error = remove_from_waiters(lkb, DLM_MSG_REQUEST_REPLY); if (error) { log_error(ls, "receive_request_reply %x remote %d %x result %d", - lkb->lkb_id, from_nodeid, ms->m_lkid, ms->m_result); + lkb->lkb_id, from_nodeid, le32_to_cpu(ms->m_lkid), + from_dlm_errno(le32_to_cpu(ms->m_result))); dlm_dump_rsb(r); goto out; } @@ -4541,7 +4569,7 @@ static int receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms) } /* this is the value returned from do_request() on the master */ - result = ms->m_result; + result = from_dlm_errno(le32_to_cpu(ms->m_result)); switch (result) { case -EAGAIN: @@ -4555,7 +4583,7 @@ static int receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms) case 0: /* request was queued or granted on remote master */ receive_flags_reply(lkb, ms); - lkb->lkb_remid = ms->m_lkid; + lkb->lkb_remid = le32_to_cpu(ms->m_lkid); if (is_altmode(lkb)) munge_altmode(lkb, ms); if (result) { @@ -4628,7 +4656,7 @@ static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, struct dlm_message *ms) { /* this is the value returned from do_convert() on the master */ - switch (ms->m_result) { + switch (from_dlm_errno(le32_to_cpu(ms->m_result))) { case -EAGAIN: /* convert would block (be queued) on remote master */ queue_cast(r, lkb, -EAGAIN); @@ -4661,8 +4689,9 @@ static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, default: log_error(r->res_ls, "receive_convert_reply %x remote %d %x %d", - lkb->lkb_id, ms->m_header.h_nodeid, ms->m_lkid, - ms->m_result); + lkb->lkb_id, le32_to_cpu(ms->m_header.h_nodeid), + le32_to_cpu(ms->m_lkid), + from_dlm_errno(le32_to_cpu(ms->m_result))); dlm_print_rsb(r); dlm_print_lkb(lkb); } @@ -4696,7 +4725,7 @@ static int receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms) struct dlm_lkb *lkb; int error; - error = find_lkb(ls, ms->m_remid, &lkb); + error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb); if (error) return error; @@ -4724,7 +4753,7 @@ static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms) /* this is the value returned from do_unlock() on the master */ - switch (ms->m_result) { + switch (from_dlm_errno(le32_to_cpu(ms->m_result))) { case -DLM_EUNLOCK: receive_flags_reply(lkb, ms); remove_lock_pc(r, lkb); @@ -4734,7 +4763,7 @@ static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms) break; default: log_error(r->res_ls, "receive_unlock_reply %x error %d", - lkb->lkb_id, ms->m_result); + lkb->lkb_id, from_dlm_errno(le32_to_cpu(ms->m_result))); } out: unlock_rsb(r); @@ -4746,7 +4775,7 @@ static int receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms) struct dlm_lkb *lkb; int error; - error = find_lkb(ls, ms->m_remid, &lkb); + error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb); if (error) return error; @@ -4774,7 +4803,7 @@ static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms) /* this is the value returned from do_cancel() on the master */ - switch (ms->m_result) { + switch (from_dlm_errno(le32_to_cpu(ms->m_result))) { case -DLM_ECANCEL: receive_flags_reply(lkb, ms); revert_lock_pc(r, lkb); @@ -4784,7 +4813,8 @@ static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms) break; default: log_error(r->res_ls, "receive_cancel_reply %x error %d", - lkb->lkb_id, ms->m_result); + lkb->lkb_id, + from_dlm_errno(le32_to_cpu(ms->m_result))); } out: unlock_rsb(r); @@ -4796,7 +4826,7 @@ static int receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms) struct dlm_lkb *lkb; int error; - error = find_lkb(ls, ms->m_remid, &lkb); + error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb); if (error) return error; @@ -4812,9 +4842,10 @@ static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms) int error, ret_nodeid; int do_lookup_list = 0; - error = find_lkb(ls, ms->m_lkid, &lkb); + error = find_lkb(ls, le32_to_cpu(ms->m_lkid), &lkb); if (error) { - log_error(ls, "receive_lookup_reply no lkid %x", ms->m_lkid); + log_error(ls, "%s no lkid %x", __func__, + le32_to_cpu(ms->m_lkid)); return; } @@ -4829,7 +4860,7 @@ static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms) if (error) goto out; - ret_nodeid = ms->m_nodeid; + ret_nodeid = le32_to_cpu(ms->m_nodeid); /* We sometimes receive a request from the dir node for this rsb before we've received the dir node's loookup_reply for it. @@ -4841,8 +4872,8 @@ static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms) /* This should never happen */ log_error(ls, "receive_lookup_reply %x from %d ret %d " "master %d dir %d our %d first %x %s", - lkb->lkb_id, ms->m_header.h_nodeid, ret_nodeid, - r->res_master_nodeid, r->res_dir_nodeid, + lkb->lkb_id, le32_to_cpu(ms->m_header.h_nodeid), + ret_nodeid, r->res_master_nodeid, r->res_dir_nodeid, dlm_our_nodeid(), r->res_first_lkid, r->res_name); } @@ -4854,7 +4885,7 @@ static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms) } else if (ret_nodeid == -1) { /* the remote node doesn't believe it's the dir node */ log_error(ls, "receive_lookup_reply %x from %d bad ret_nodeid", - lkb->lkb_id, ms->m_header.h_nodeid); + lkb->lkb_id, le32_to_cpu(ms->m_header.h_nodeid)); r->res_master_nodeid = 0; r->res_nodeid = -1; lkb->lkb_nodeid = -1; @@ -4888,10 +4919,12 @@ static void _receive_message(struct dlm_ls *ls, struct dlm_message *ms, { int error = 0, noent = 0; - if (!dlm_is_member(ls, ms->m_header.h_nodeid)) { + if (!dlm_is_member(ls, le32_to_cpu(ms->m_header.h_nodeid))) { log_limit(ls, "receive %d from non-member %d %x %x %d", - ms->m_type, ms->m_header.h_nodeid, ms->m_lkid, - ms->m_remid, ms->m_result); + le32_to_cpu(ms->m_type), + le32_to_cpu(ms->m_header.h_nodeid), + le32_to_cpu(ms->m_lkid), le32_to_cpu(ms->m_remid), + from_dlm_errno(le32_to_cpu(ms->m_result))); return; } @@ -4899,77 +4932,78 @@ static void _receive_message(struct dlm_ls *ls, struct dlm_message *ms, /* messages sent to a master node */ - case DLM_MSG_REQUEST: + case cpu_to_le32(DLM_MSG_REQUEST): error = receive_request(ls, ms); break; - case DLM_MSG_CONVERT: + case cpu_to_le32(DLM_MSG_CONVERT): error = receive_convert(ls, ms); break; - case DLM_MSG_UNLOCK: + case cpu_to_le32(DLM_MSG_UNLOCK): error = receive_unlock(ls, ms); break; - case DLM_MSG_CANCEL: + case cpu_to_le32(DLM_MSG_CANCEL): noent = 1; error = receive_cancel(ls, ms); break; /* messages sent from a master node (replies to above) */ - case DLM_MSG_REQUEST_REPLY: + case cpu_to_le32(DLM_MSG_REQUEST_REPLY): error = receive_request_reply(ls, ms); break; - case DLM_MSG_CONVERT_REPLY: + case cpu_to_le32(DLM_MSG_CONVERT_REPLY): error = receive_convert_reply(ls, ms); break; - case DLM_MSG_UNLOCK_REPLY: + case cpu_to_le32(DLM_MSG_UNLOCK_REPLY): error = receive_unlock_reply(ls, ms); break; - case DLM_MSG_CANCEL_REPLY: + case cpu_to_le32(DLM_MSG_CANCEL_REPLY): error = receive_cancel_reply(ls, ms); break; /* messages sent from a master node (only two types of async msg) */ - case DLM_MSG_GRANT: + case cpu_to_le32(DLM_MSG_GRANT): noent = 1; error = receive_grant(ls, ms); break; - case DLM_MSG_BAST: + case cpu_to_le32(DLM_MSG_BAST): noent = 1; error = receive_bast(ls, ms); break; /* messages sent to a dir node */ - case DLM_MSG_LOOKUP: + case cpu_to_le32(DLM_MSG_LOOKUP): receive_lookup(ls, ms); break; - case DLM_MSG_REMOVE: + case cpu_to_le32(DLM_MSG_REMOVE): receive_remove(ls, ms); break; /* messages sent from a dir node (remove has no reply) */ - case DLM_MSG_LOOKUP_REPLY: + case cpu_to_le32(DLM_MSG_LOOKUP_REPLY): receive_lookup_reply(ls, ms); break; /* other messages */ - case DLM_MSG_PURGE: + case cpu_to_le32(DLM_MSG_PURGE): receive_purge(ls, ms); break; default: - log_error(ls, "unknown message type %d", ms->m_type); + log_error(ls, "unknown message type %d", + le32_to_cpu(ms->m_type)); } /* @@ -4985,22 +5019,26 @@ static void _receive_message(struct dlm_ls *ls, struct dlm_message *ms, if (error == -ENOENT && noent) { log_debug(ls, "receive %d no %x remote %d %x saved_seq %u", - ms->m_type, ms->m_remid, ms->m_header.h_nodeid, - ms->m_lkid, saved_seq); + le32_to_cpu(ms->m_type), le32_to_cpu(ms->m_remid), + le32_to_cpu(ms->m_header.h_nodeid), + le32_to_cpu(ms->m_lkid), saved_seq); } else if (error == -ENOENT) { log_error(ls, "receive %d no %x remote %d %x saved_seq %u", - ms->m_type, ms->m_remid, ms->m_header.h_nodeid, - ms->m_lkid, saved_seq); + le32_to_cpu(ms->m_type), le32_to_cpu(ms->m_remid), + le32_to_cpu(ms->m_header.h_nodeid), + le32_to_cpu(ms->m_lkid), saved_seq); - if (ms->m_type == DLM_MSG_CONVERT) - dlm_dump_rsb_hash(ls, ms->m_hash); + if (ms->m_type == cpu_to_le32(DLM_MSG_CONVERT)) + dlm_dump_rsb_hash(ls, le32_to_cpu(ms->m_hash)); } if (error == -EINVAL) { log_error(ls, "receive %d inval from %d lkid %x remid %x " "saved_seq %u", - ms->m_type, ms->m_header.h_nodeid, - ms->m_lkid, ms->m_remid, saved_seq); + le32_to_cpu(ms->m_type), + le32_to_cpu(ms->m_header.h_nodeid), + le32_to_cpu(ms->m_lkid), le32_to_cpu(ms->m_remid), + saved_seq); } } @@ -5021,7 +5059,7 @@ static void dlm_receive_message(struct dlm_ls *ls, struct dlm_message *ms, lockspace generation before we left. */ if (!ls->ls_generation) { log_limit(ls, "receive %d from %d ignore old gen", - ms->m_type, nodeid); + le32_to_cpu(ms->m_type), nodeid); return; } @@ -5054,30 +5092,30 @@ void dlm_receive_buffer(union dlm_packet *p, int nodeid) switch (hd->h_cmd) { case DLM_MSG: - dlm_message_in(&p->message); - type = p->message.m_type; + type = le32_to_cpu(p->message.m_type); break; case DLM_RCOM: - dlm_rcom_in(&p->rcom); - type = p->rcom.rc_type; + type = le32_to_cpu(p->rcom.rc_type); break; default: log_print("invalid h_cmd %d from %u", hd->h_cmd, nodeid); return; } - if (hd->h_nodeid != nodeid) { + if (le32_to_cpu(hd->h_nodeid) != nodeid) { log_print("invalid h_nodeid %d from %d lockspace %x", - hd->h_nodeid, nodeid, hd->u.h_lockspace); + le32_to_cpu(hd->h_nodeid), nodeid, + le32_to_cpu(hd->u.h_lockspace)); return; } - ls = dlm_find_lockspace_global(hd->u.h_lockspace); + ls = dlm_find_lockspace_global(le32_to_cpu(hd->u.h_lockspace)); if (!ls) { if (dlm_config.ci_log_debug) { printk_ratelimited(KERN_DEBUG "dlm: invalid lockspace " "%u from %d cmd %d type %d\n", - hd->u.h_lockspace, nodeid, hd->h_cmd, type); + le32_to_cpu(hd->u.h_lockspace), nodeid, + hd->h_cmd, type); } if (hd->h_cmd == DLM_RCOM && type == DLM_RCOM_STATUS) @@ -5104,10 +5142,10 @@ static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb, if (middle_conversion(lkb)) { hold_lkb(lkb); memset(ms_stub, 0, sizeof(struct dlm_message)); - ms_stub->m_flags = DLM_IFL_STUB_MS; - ms_stub->m_type = DLM_MSG_CONVERT_REPLY; - ms_stub->m_result = -EINPROGRESS; - ms_stub->m_header.h_nodeid = lkb->lkb_nodeid; + ms_stub->m_flags = cpu_to_le32(DLM_IFL_STUB_MS); + ms_stub->m_type = cpu_to_le32(DLM_MSG_CONVERT_REPLY); + ms_stub->m_result = cpu_to_le32(to_dlm_errno(-EINPROGRESS)); + ms_stub->m_header.h_nodeid = cpu_to_le32(lkb->lkb_nodeid); _receive_convert_reply(lkb, ms_stub); /* Same special case as in receive_rcom_lock_args() */ @@ -5226,10 +5264,10 @@ void dlm_recover_waiters_pre(struct dlm_ls *ls) case DLM_MSG_UNLOCK: hold_lkb(lkb); memset(ms_stub, 0, sizeof(struct dlm_message)); - ms_stub->m_flags = DLM_IFL_STUB_MS; - ms_stub->m_type = DLM_MSG_UNLOCK_REPLY; - ms_stub->m_result = stub_unlock_result; - ms_stub->m_header.h_nodeid = lkb->lkb_nodeid; + ms_stub->m_flags = cpu_to_le32(DLM_IFL_STUB_MS); + ms_stub->m_type = cpu_to_le32(DLM_MSG_UNLOCK_REPLY); + ms_stub->m_result = cpu_to_le32(to_dlm_errno(stub_unlock_result)); + ms_stub->m_header.h_nodeid = cpu_to_le32(lkb->lkb_nodeid); _receive_unlock_reply(lkb, ms_stub); dlm_put_lkb(lkb); break; @@ -5237,10 +5275,10 @@ void dlm_recover_waiters_pre(struct dlm_ls *ls) case DLM_MSG_CANCEL: hold_lkb(lkb); memset(ms_stub, 0, sizeof(struct dlm_message)); - ms_stub->m_flags = DLM_IFL_STUB_MS; - ms_stub->m_type = DLM_MSG_CANCEL_REPLY; - ms_stub->m_result = stub_cancel_result; - ms_stub->m_header.h_nodeid = lkb->lkb_nodeid; + ms_stub->m_flags = cpu_to_le32(DLM_IFL_STUB_MS); + ms_stub->m_type = cpu_to_le32(DLM_MSG_CANCEL_REPLY); + ms_stub->m_result = cpu_to_le32(to_dlm_errno(stub_cancel_result)); + ms_stub->m_header.h_nodeid = cpu_to_le32(lkb->lkb_nodeid); _receive_cancel_reply(lkb, ms_stub); dlm_put_lkb(lkb); break; @@ -5257,21 +5295,18 @@ void dlm_recover_waiters_pre(struct dlm_ls *ls) static struct dlm_lkb *find_resend_waiter(struct dlm_ls *ls) { - struct dlm_lkb *lkb; - int found = 0; + struct dlm_lkb *lkb = NULL, *iter; mutex_lock(&ls->ls_waiters_mutex); - list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) { - if (lkb->lkb_flags & DLM_IFL_RESEND) { - hold_lkb(lkb); - found = 1; + list_for_each_entry(iter, &ls->ls_waiters, lkb_wait_reply) { + if (iter->lkb_flags & DLM_IFL_RESEND) { + hold_lkb(iter); + lkb = iter; break; } } mutex_unlock(&ls->ls_waiters_mutex); - if (!found) - lkb = NULL; return lkb; } @@ -5331,11 +5366,16 @@ int dlm_recover_waiters_post(struct dlm_ls *ls) lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK; lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL; lkb->lkb_wait_type = 0; - lkb->lkb_wait_count = 0; + /* drop all wait_count references we still + * hold a reference for this iteration. + */ + while (lkb->lkb_wait_count) { + lkb->lkb_wait_count--; + unhold_lkb(lkb); + } mutex_lock(&ls->ls_waiters_mutex); list_del_init(&lkb->lkb_wait_reply); mutex_unlock(&ls->ls_waiters_mutex); - unhold_lkb(lkb); /* for waiters list */ if (oc || ou) { /* do an unlock or cancel instead of resending */ @@ -5605,7 +5645,7 @@ static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb, { struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf; - lkb->lkb_nodeid = rc->rc_header.h_nodeid; + lkb->lkb_nodeid = le32_to_cpu(rc->rc_header.h_nodeid); lkb->lkb_ownpid = le32_to_cpu(rl->rl_ownpid); lkb->lkb_remid = le32_to_cpu(rl->rl_lkid); lkb->lkb_exflags = le32_to_cpu(rl->rl_exflags); @@ -5620,8 +5660,8 @@ static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb, lkb->lkb_astfn = (rl->rl_asts & DLM_CB_CAST) ? &fake_astfn : NULL; if (lkb->lkb_exflags & DLM_LKF_VALBLK) { - int lvblen = rc->rc_header.h_length - sizeof(struct dlm_rcom) - - sizeof(struct rcom_lock); + int lvblen = le16_to_cpu(rc->rc_header.h_length) - + sizeof(struct dlm_rcom) - sizeof(struct rcom_lock); if (lvblen > ls->ls_lvblen) return -EINVAL; lkb->lkb_lvbptr = dlm_allocate_lvb(ls); @@ -5657,7 +5697,7 @@ int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc) struct dlm_rsb *r; struct dlm_lkb *lkb; uint32_t remid = 0; - int from_nodeid = rc->rc_header.h_nodeid; + int from_nodeid = le32_to_cpu(rc->rc_header.h_nodeid); int error; if (rl->rl_parent_lkid) { @@ -5707,7 +5747,6 @@ int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc) attach_lkb(r, lkb); add_lkb(r, lkb, rl->rl_status); - error = 0; ls->ls_recover_locks_in++; if (!list_empty(&r->res_waitqueue) || !list_empty(&r->res_convertqueue)) @@ -5747,7 +5786,8 @@ int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc) error = find_lkb(ls, lkid, &lkb); if (error) { log_error(ls, "dlm_recover_process_copy no %x remote %d %x %d", - lkid, rc->rc_header.h_nodeid, remid, result); + lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid, + result); return error; } @@ -5757,7 +5797,8 @@ int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc) if (!is_process_copy(lkb)) { log_error(ls, "dlm_recover_process_copy bad %x remote %d %x %d", - lkid, rc->rc_header.h_nodeid, remid, result); + lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid, + result); dlm_dump_rsb(r); unlock_rsb(r); put_rsb(r); @@ -5772,7 +5813,8 @@ int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc) a barrier between recover_masters and recover_locks. */ log_debug(ls, "dlm_recover_process_copy %x remote %d %x %d", - lkid, rc->rc_header.h_nodeid, remid, result); + lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid, + result); dlm_send_rcom_lock(r, lkb); goto out; @@ -5782,7 +5824,8 @@ int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc) break; default: log_error(ls, "dlm_recover_process_copy %x remote %d %x %d unk", - lkid, rc->rc_header.h_nodeid, remid, result); + lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid, + result); } /* an ack for dlm_recover_locks() which waits for replies from @@ -5925,37 +5968,36 @@ int dlm_user_adopt_orphan(struct dlm_ls *ls, struct dlm_user_args *ua_tmp, int mode, uint32_t flags, void *name, unsigned int namelen, unsigned long timeout_cs, uint32_t *lkid) { - struct dlm_lkb *lkb; + struct dlm_lkb *lkb = NULL, *iter; struct dlm_user_args *ua; int found_other_mode = 0; - int found = 0; int rv = 0; mutex_lock(&ls->ls_orphans_mutex); - list_for_each_entry(lkb, &ls->ls_orphans, lkb_ownqueue) { - if (lkb->lkb_resource->res_length != namelen) + list_for_each_entry(iter, &ls->ls_orphans, lkb_ownqueue) { + if (iter->lkb_resource->res_length != namelen) continue; - if (memcmp(lkb->lkb_resource->res_name, name, namelen)) + if (memcmp(iter->lkb_resource->res_name, name, namelen)) continue; - if (lkb->lkb_grmode != mode) { + if (iter->lkb_grmode != mode) { found_other_mode = 1; continue; } - found = 1; - list_del_init(&lkb->lkb_ownqueue); - lkb->lkb_flags &= ~DLM_IFL_ORPHAN; - *lkid = lkb->lkb_id; + lkb = iter; + list_del_init(&iter->lkb_ownqueue); + iter->lkb_flags &= ~DLM_IFL_ORPHAN; + *lkid = iter->lkb_id; break; } mutex_unlock(&ls->ls_orphans_mutex); - if (!found && found_other_mode) { + if (!lkb && found_other_mode) { rv = -EAGAIN; goto out; } - if (!found) { + if (!lkb) { rv = -ENOENT; goto out; } @@ -6307,8 +6349,8 @@ static int send_purge(struct dlm_ls *ls, int nodeid, int pid) DLM_MSG_PURGE, &ms, &mh); if (error) return error; - ms->m_nodeid = nodeid; - ms->m_pid = pid; + ms->m_nodeid = cpu_to_le32(nodeid); + ms->m_pid = cpu_to_le32(pid); return send_message(mh, ms); } diff --git a/fs/dlm/lockspace.c b/fs/dlm/lockspace.c index 0d3833a124a3..19ed41a5da93 100644 --- a/fs/dlm/lockspace.c +++ b/fs/dlm/lockspace.c @@ -922,3 +922,15 @@ void dlm_stop_lockspaces(void) log_print("dlm user daemon left %d lockspaces", count); } +void dlm_stop_lockspaces_check(void) +{ + struct dlm_ls *ls; + + spin_lock(&lslist_lock); + list_for_each_entry(ls, &lslist, ls_list) { + if (WARN_ON(!rwsem_is_locked(&ls->ls_in_recovery) || + !dlm_locking_stopped(ls))) + break; + } + spin_unlock(&lslist_lock); +} diff --git a/fs/dlm/lockspace.h b/fs/dlm/lockspace.h index a78d853b9342..306fc4f4ea15 100644 --- a/fs/dlm/lockspace.h +++ b/fs/dlm/lockspace.h @@ -19,6 +19,7 @@ struct dlm_ls *dlm_find_lockspace_local(void *id); struct dlm_ls *dlm_find_lockspace_device(int minor); void dlm_put_lockspace(struct dlm_ls *ls); void dlm_stop_lockspaces(void); +void dlm_stop_lockspaces_check(void); #endif /* __LOCKSPACE_DOT_H__ */ diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c index e284d696c1fd..19e82f08c0e0 100644 --- a/fs/dlm/lowcomms.c +++ b/fs/dlm/lowcomms.c @@ -1303,6 +1303,10 @@ static struct dlm_msg *dlm_lowcomms_new_msg_con(struct connection *con, int len, return msg; } +/* avoid false positive for nodes_srcu, unlock happens in + * dlm_lowcomms_commit_msg which is a must call if success + */ +#ifndef __CHECKER__ struct dlm_msg *dlm_lowcomms_new_msg(int nodeid, int len, gfp_t allocation, char **ppc, void (*cb)(void *data), void *data) @@ -1336,6 +1340,7 @@ struct dlm_msg *dlm_lowcomms_new_msg(int nodeid, int len, gfp_t allocation, msg->idx = idx; return msg; } +#endif static void _dlm_lowcomms_commit_msg(struct dlm_msg *msg) { @@ -1362,11 +1367,16 @@ static void _dlm_lowcomms_commit_msg(struct dlm_msg *msg) return; } +/* avoid false positive for nodes_srcu, lock was happen in + * dlm_lowcomms_new_msg + */ +#ifndef __CHECKER__ void dlm_lowcomms_commit_msg(struct dlm_msg *msg) { _dlm_lowcomms_commit_msg(msg); srcu_read_unlock(&connections_srcu, msg->idx); } +#endif void dlm_lowcomms_put_msg(struct dlm_msg *msg) { @@ -1789,7 +1799,7 @@ static int dlm_listen_for_all(void) SOCK_STREAM, dlm_proto_ops->proto, &sock); if (result < 0) { log_print("Can't create comms socket: %d", result); - goto out; + return result; } sock_set_mark(sock->sk, dlm_config.ci_mark); diff --git a/fs/dlm/member.c b/fs/dlm/member.c index 61f906e705db..98084e0cfccf 100644 --- a/fs/dlm/member.c +++ b/fs/dlm/member.c @@ -20,7 +20,7 @@ int dlm_slots_version(struct dlm_header *h) { - if ((h->h_version & 0x0000FFFF) < DLM_HEADER_SLOTS) + if ((le32_to_cpu(h->h_version) & 0x0000FFFF) < DLM_HEADER_SLOTS) return 0; return 1; } @@ -120,18 +120,13 @@ int dlm_slots_copy_in(struct dlm_ls *ls) ro0 = (struct rcom_slot *)(rc->rc_buf + sizeof(struct rcom_config)); - for (i = 0, ro = ro0; i < num_slots; i++, ro++) { - ro->ro_nodeid = le32_to_cpu(ro->ro_nodeid); - ro->ro_slot = le16_to_cpu(ro->ro_slot); - } - log_slots(ls, gen, num_slots, ro0, NULL, 0); list_for_each_entry(memb, &ls->ls_nodes, list) { for (i = 0, ro = ro0; i < num_slots; i++, ro++) { - if (ro->ro_nodeid != memb->nodeid) + if (le32_to_cpu(ro->ro_nodeid) != memb->nodeid) continue; - memb->slot = ro->ro_slot; + memb->slot = le16_to_cpu(ro->ro_slot); memb->slot_prev = memb->slot; break; } diff --git a/fs/dlm/midcomms.c b/fs/dlm/midcomms.c index 3635e42b0669..6489bc22ad61 100644 --- a/fs/dlm/midcomms.c +++ b/fs/dlm/midcomms.c @@ -135,6 +135,7 @@ #include #include "dlm_internal.h" +#include "lockspace.h" #include "lowcomms.h" #include "config.h" #include "memory.h" @@ -380,13 +381,12 @@ static int dlm_send_ack(int nodeid, uint32_t seq) m_header = (struct dlm_header *)ppc; - m_header->h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR); - m_header->h_nodeid = dlm_our_nodeid(); - m_header->h_length = mb_len; + m_header->h_version = cpu_to_le32(DLM_HEADER_MAJOR | DLM_HEADER_MINOR); + m_header->h_nodeid = cpu_to_le32(dlm_our_nodeid()); + m_header->h_length = cpu_to_le16(mb_len); m_header->h_cmd = DLM_ACK; - m_header->u.h_seq = seq; + m_header->u.h_seq = cpu_to_le32(seq); - header_out(m_header); dlm_lowcomms_commit_msg(msg); dlm_lowcomms_put_msg(msg); @@ -409,13 +409,11 @@ static int dlm_send_fin(struct midcomms_node *node, m_header = (struct dlm_header *)ppc; - m_header->h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR); - m_header->h_nodeid = dlm_our_nodeid(); - m_header->h_length = mb_len; + m_header->h_version = cpu_to_le32(DLM_HEADER_MAJOR | DLM_HEADER_MINOR); + m_header->h_nodeid = cpu_to_le32(dlm_our_nodeid()); + m_header->h_length = cpu_to_le16(mb_len); m_header->h_cmd = DLM_FIN; - header_out(m_header); - pr_debug("sending fin msg to node %d\n", node->nodeid); dlm_midcomms_commit_mhandle(mh); set_bit(DLM_NODE_FLAG_STOP_TX, &node->flags); @@ -574,14 +572,14 @@ dlm_midcomms_recv_node_lookup(int nodeid, const union dlm_packet *p, return NULL; } - switch (le32_to_cpu(p->rcom.rc_type)) { - case DLM_RCOM_NAMES: + switch (p->rcom.rc_type) { + case cpu_to_le32(DLM_RCOM_NAMES): fallthrough; - case DLM_RCOM_NAMES_REPLY: + case cpu_to_le32(DLM_RCOM_NAMES_REPLY): fallthrough; - case DLM_RCOM_STATUS: + case cpu_to_le32(DLM_RCOM_STATUS): fallthrough; - case DLM_RCOM_STATUS_REPLY: + case cpu_to_le32(DLM_RCOM_STATUS_REPLY): node = nodeid2node(nodeid, 0); if (node) { spin_lock(&node->state_lock); @@ -741,14 +739,14 @@ static void dlm_midcomms_receive_buffer_3_2(union dlm_packet *p, int nodeid) * * length already checked. */ - switch (le32_to_cpu(p->rcom.rc_type)) { - case DLM_RCOM_NAMES: + switch (p->rcom.rc_type) { + case cpu_to_le32(DLM_RCOM_NAMES): fallthrough; - case DLM_RCOM_NAMES_REPLY: + case cpu_to_le32(DLM_RCOM_NAMES_REPLY): fallthrough; - case DLM_RCOM_STATUS: + case cpu_to_le32(DLM_RCOM_STATUS): fallthrough; - case DLM_RCOM_STATUS_REPLY: + case cpu_to_le32(DLM_RCOM_STATUS_REPLY): break; default: log_print("unsupported rcom type received: %u, will skip this message from node %d", @@ -1020,11 +1018,10 @@ static void dlm_fill_opts_header(struct dlm_opts *opts, uint16_t inner_len, uint32_t seq) { opts->o_header.h_cmd = DLM_OPTS; - opts->o_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR); - opts->o_header.h_nodeid = dlm_our_nodeid(); - opts->o_header.h_length = DLM_MIDCOMMS_OPT_LEN + inner_len; - opts->o_header.u.h_seq = seq; - header_out(&opts->o_header); + opts->o_header.h_version = cpu_to_le32(DLM_HEADER_MAJOR | DLM_HEADER_MINOR); + opts->o_header.h_nodeid = cpu_to_le32(dlm_our_nodeid()); + opts->o_header.h_length = cpu_to_le16(DLM_MIDCOMMS_OPT_LEN + inner_len); + opts->o_header.u.h_seq = cpu_to_le32(seq); } static void midcomms_new_msg_cb(void *data) @@ -1062,6 +1059,10 @@ static struct dlm_msg *dlm_midcomms_get_msg_3_2(struct dlm_mhandle *mh, int node return msg; } +/* avoid false positive for nodes_srcu, unlock happens in + * dlm_midcomms_commit_mhandle which is a must call if success + */ +#ifndef __CHECKER__ struct dlm_mhandle *dlm_midcomms_get_mhandle(int nodeid, int len, gfp_t allocation, char **ppc) { @@ -1127,6 +1128,7 @@ struct dlm_mhandle *dlm_midcomms_get_mhandle(int nodeid, int len, srcu_read_unlock(&nodes_srcu, idx); return NULL; } +#endif static void dlm_midcomms_commit_msg_3_2(struct dlm_mhandle *mh) { @@ -1136,6 +1138,10 @@ static void dlm_midcomms_commit_msg_3_2(struct dlm_mhandle *mh) dlm_lowcomms_commit_msg(mh->msg); } +/* avoid false positive for nodes_srcu, lock was happen in + * dlm_midcomms_get_mhandle + */ +#ifndef __CHECKER__ void dlm_midcomms_commit_mhandle(struct dlm_mhandle *mh) { switch (mh->node->version) { @@ -1157,6 +1163,7 @@ void dlm_midcomms_commit_mhandle(struct dlm_mhandle *mh) break; } } +#endif int dlm_midcomms_start(void) { @@ -1406,6 +1413,8 @@ int dlm_midcomms_close(int nodeid) if (nodeid == dlm_our_nodeid()) return 0; + dlm_stop_lockspaces_check(); + idx = srcu_read_lock(&nodes_srcu); /* Abort pending close/remove operation */ node = nodeid2node(nodeid, 0); @@ -1455,7 +1464,7 @@ static void midcomms_new_rawmsg_cb(void *data) switch (h->h_cmd) { case DLM_OPTS: if (!h->u.h_seq) - h->u.h_seq = rd->node->seq_send++; + h->u.h_seq = cpu_to_le32(rd->node->seq_send++); break; default: break; diff --git a/fs/dlm/plock.c b/fs/dlm/plock.c index c38b2b8ffd1d..0993eebf2060 100644 --- a/fs/dlm/plock.c +++ b/fs/dlm/plock.c @@ -13,27 +13,27 @@ #include "dlm_internal.h" #include "lockspace.h" -static spinlock_t ops_lock; -static struct list_head send_list; -static struct list_head recv_list; -static wait_queue_head_t send_wq; -static wait_queue_head_t recv_wq; +static DEFINE_SPINLOCK(ops_lock); +static LIST_HEAD(send_list); +static LIST_HEAD(recv_list); +static DECLARE_WAIT_QUEUE_HEAD(send_wq); +static DECLARE_WAIT_QUEUE_HEAD(recv_wq); + +struct plock_async_data { + void *fl; + void *file; + struct file_lock flc; + int (*callback)(struct file_lock *fl, int result); +}; struct plock_op { struct list_head list; int done; struct dlm_plock_info info; + /* if set indicates async handling */ + struct plock_async_data *data; }; -struct plock_xop { - struct plock_op xop; - int (*callback)(struct file_lock *fl, int result); - void *fl; - void *file; - struct file_lock flc; -}; - - static inline void set_version(struct dlm_plock_info *info) { info->version[0] = DLM_PLOCK_VERSION_MAJOR; @@ -58,10 +58,15 @@ static int check_version(struct dlm_plock_info *info) return 0; } +static void dlm_release_plock_op(struct plock_op *op) +{ + kfree(op->data); + kfree(op); +} + static void send_op(struct plock_op *op) { set_version(&op->info); - INIT_LIST_HEAD(&op->list); spin_lock(&ops_lock); list_add_tail(&op->list, &send_list); spin_unlock(&ops_lock); @@ -101,22 +106,21 @@ static void do_unlock_close(struct dlm_ls *ls, u64 number, int dlm_posix_lock(dlm_lockspace_t *lockspace, u64 number, struct file *file, int cmd, struct file_lock *fl) { + struct plock_async_data *op_data; struct dlm_ls *ls; struct plock_op *op; - struct plock_xop *xop; int rv; ls = dlm_find_lockspace_local(lockspace); if (!ls) return -EINVAL; - xop = kzalloc(sizeof(*xop), GFP_NOFS); - if (!xop) { + op = kzalloc(sizeof(*op), GFP_NOFS); + if (!op) { rv = -ENOMEM; goto out; } - op = &xop->xop; op->info.optype = DLM_PLOCK_OP_LOCK; op->info.pid = fl->fl_pid; op->info.ex = (fl->fl_type == F_WRLCK); @@ -125,46 +129,49 @@ int dlm_posix_lock(dlm_lockspace_t *lockspace, u64 number, struct file *file, op->info.number = number; op->info.start = fl->fl_start; op->info.end = fl->fl_end; + /* async handling */ if (fl->fl_lmops && fl->fl_lmops->lm_grant) { + op_data = kzalloc(sizeof(*op_data), GFP_NOFS); + if (!op_data) { + dlm_release_plock_op(op); + rv = -ENOMEM; + goto out; + } + /* fl_owner is lockd which doesn't distinguish processes on the nfs client */ op->info.owner = (__u64) fl->fl_pid; - xop->callback = fl->fl_lmops->lm_grant; - locks_init_lock(&xop->flc); - locks_copy_lock(&xop->flc, fl); - xop->fl = fl; - xop->file = file; + op_data->callback = fl->fl_lmops->lm_grant; + locks_init_lock(&op_data->flc); + locks_copy_lock(&op_data->flc, fl); + op_data->fl = fl; + op_data->file = file; + + op->data = op_data; + + send_op(op); + rv = FILE_LOCK_DEFERRED; + goto out; } else { op->info.owner = (__u64)(long) fl->fl_owner; - xop->callback = NULL; } send_op(op); - if (xop->callback == NULL) { - rv = wait_event_interruptible(recv_wq, (op->done != 0)); - if (rv == -ERESTARTSYS) { - log_debug(ls, "dlm_posix_lock: wait killed %llx", - (unsigned long long)number); - spin_lock(&ops_lock); - list_del(&op->list); - spin_unlock(&ops_lock); - kfree(xop); - do_unlock_close(ls, number, file, fl); - goto out; - } - } else { - rv = FILE_LOCK_DEFERRED; + rv = wait_event_interruptible(recv_wq, (op->done != 0)); + if (rv == -ERESTARTSYS) { + spin_lock(&ops_lock); + list_del(&op->list); + spin_unlock(&ops_lock); + log_print("%s: wait interrupted %x %llx, op removed", + __func__, ls->ls_global_id, + (unsigned long long)number); + dlm_release_plock_op(op); + do_unlock_close(ls, number, file, fl); goto out; } - spin_lock(&ops_lock); - if (!list_empty(&op->list)) { - log_error(ls, "dlm_posix_lock: op on list %llx", - (unsigned long long)number); - list_del(&op->list); - } - spin_unlock(&ops_lock); + WARN_ON(!list_empty(&op->list)); rv = op->info.rv; @@ -174,7 +181,7 @@ int dlm_posix_lock(dlm_lockspace_t *lockspace, u64 number, struct file *file, (unsigned long long)number); } - kfree(xop); + dlm_release_plock_op(op); out: dlm_put_lockspace(ls); return rv; @@ -184,26 +191,20 @@ EXPORT_SYMBOL_GPL(dlm_posix_lock); /* Returns failure iff a successful lock operation should be canceled */ static int dlm_plock_callback(struct plock_op *op) { + struct plock_async_data *op_data = op->data; struct file *file; struct file_lock *fl; struct file_lock *flc; int (*notify)(struct file_lock *fl, int result) = NULL; - struct plock_xop *xop = (struct plock_xop *)op; int rv = 0; - spin_lock(&ops_lock); - if (!list_empty(&op->list)) { - log_print("dlm_plock_callback: op on list %llx", - (unsigned long long)op->info.number); - list_del(&op->list); - } - spin_unlock(&ops_lock); + WARN_ON(!list_empty(&op->list)); /* check if the following 2 are still valid or make a copy */ - file = xop->file; - flc = &xop->flc; - fl = xop->fl; - notify = xop->callback; + file = op_data->file; + flc = &op_data->flc; + fl = op_data->fl; + notify = op_data->callback; if (op->info.rv) { notify(fl, op->info.rv); @@ -234,7 +235,7 @@ static int dlm_plock_callback(struct plock_op *op) } out: - kfree(xop); + dlm_release_plock_op(op); return rv; } @@ -290,13 +291,7 @@ int dlm_posix_unlock(dlm_lockspace_t *lockspace, u64 number, struct file *file, send_op(op); wait_event(recv_wq, (op->done != 0)); - spin_lock(&ops_lock); - if (!list_empty(&op->list)) { - log_error(ls, "dlm_posix_unlock: op on list %llx", - (unsigned long long)number); - list_del(&op->list); - } - spin_unlock(&ops_lock); + WARN_ON(!list_empty(&op->list)); rv = op->info.rv; @@ -304,7 +299,7 @@ int dlm_posix_unlock(dlm_lockspace_t *lockspace, u64 number, struct file *file, rv = 0; out_free: - kfree(op); + dlm_release_plock_op(op); out: dlm_put_lockspace(ls); fl->fl_flags = fl_flags; @@ -344,13 +339,7 @@ int dlm_posix_get(dlm_lockspace_t *lockspace, u64 number, struct file *file, send_op(op); wait_event(recv_wq, (op->done != 0)); - spin_lock(&ops_lock); - if (!list_empty(&op->list)) { - log_error(ls, "dlm_posix_get: op on list %llx", - (unsigned long long)number); - list_del(&op->list); - } - spin_unlock(&ops_lock); + WARN_ON(!list_empty(&op->list)); /* info.rv from userspace is 1 for conflict, 0 for no-conflict, -ENOENT if there are no locks on the file */ @@ -370,7 +359,7 @@ int dlm_posix_get(dlm_lockspace_t *lockspace, u64 number, struct file *file, rv = 0; } - kfree(op); + dlm_release_plock_op(op); out: dlm_put_lockspace(ls); return rv; @@ -406,7 +395,7 @@ static ssize_t dev_read(struct file *file, char __user *u, size_t count, (the process did not make an unlock call). */ if (op->info.flags & DLM_PLOCK_FL_CLOSE) - kfree(op); + dlm_release_plock_op(op); if (copy_to_user(u, &info, sizeof(info))) return -EFAULT; @@ -418,9 +407,9 @@ static ssize_t dev_read(struct file *file, char __user *u, size_t count, static ssize_t dev_write(struct file *file, const char __user *u, size_t count, loff_t *ppos) { + struct plock_op *op = NULL, *iter; struct dlm_plock_info info; - struct plock_op *op; - int found = 0, do_callback = 0; + int do_callback = 0; if (count != sizeof(info)) return -EINVAL; @@ -432,31 +421,30 @@ static ssize_t dev_write(struct file *file, const char __user *u, size_t count, return -EINVAL; spin_lock(&ops_lock); - list_for_each_entry(op, &recv_list, list) { - if (op->info.fsid == info.fsid && - op->info.number == info.number && - op->info.owner == info.owner) { - struct plock_xop *xop = (struct plock_xop *)op; - list_del_init(&op->list); - memcpy(&op->info, &info, sizeof(info)); - if (xop->callback) + list_for_each_entry(iter, &recv_list, list) { + if (iter->info.fsid == info.fsid && + iter->info.number == info.number && + iter->info.owner == info.owner) { + list_del_init(&iter->list); + memcpy(&iter->info, &info, sizeof(info)); + if (iter->data) do_callback = 1; else - op->done = 1; - found = 1; + iter->done = 1; + op = iter; break; } } spin_unlock(&ops_lock); - if (found) { + if (op) { if (do_callback) dlm_plock_callback(op); else wake_up(&recv_wq); } else - log_print("dev_write no op %x %llx", info.fsid, - (unsigned long long)info.number); + log_print("%s: no op %x %llx - may got interrupted?", __func__, + info.fsid, (unsigned long long)info.number); return count; } @@ -492,12 +480,6 @@ int dlm_plock_init(void) { int rv; - spin_lock_init(&ops_lock); - INIT_LIST_HEAD(&send_list); - INIT_LIST_HEAD(&recv_list); - init_waitqueue_head(&send_wq); - init_waitqueue_head(&recv_wq); - rv = misc_register(&plock_dev_misc); if (rv) log_print("dlm_plock_init: misc_register failed %d", rv); diff --git a/fs/dlm/rcom.c b/fs/dlm/rcom.c index 5821b777a1a7..f19860315043 100644 --- a/fs/dlm/rcom.c +++ b/fs/dlm/rcom.c @@ -34,16 +34,16 @@ static void _create_rcom(struct dlm_ls *ls, int to_nodeid, int type, int len, rc = (struct dlm_rcom *) mb; - rc->rc_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR); - rc->rc_header.u.h_lockspace = ls->ls_global_id; - rc->rc_header.h_nodeid = dlm_our_nodeid(); - rc->rc_header.h_length = mb_len; + rc->rc_header.h_version = cpu_to_le32(DLM_HEADER_MAJOR | DLM_HEADER_MINOR); + rc->rc_header.u.h_lockspace = cpu_to_le32(ls->ls_global_id); + rc->rc_header.h_nodeid = cpu_to_le32(dlm_our_nodeid()); + rc->rc_header.h_length = cpu_to_le16(mb_len); rc->rc_header.h_cmd = DLM_RCOM; - rc->rc_type = type; + rc->rc_type = cpu_to_le32(type); spin_lock(&ls->ls_recover_lock); - rc->rc_seq = ls->ls_recover_seq; + rc->rc_seq = cpu_to_le64(ls->ls_recover_seq); spin_unlock(&ls->ls_recover_lock); *rc_ret = rc; @@ -91,13 +91,11 @@ static int create_rcom_stateless(struct dlm_ls *ls, int to_nodeid, int type, static void send_rcom(struct dlm_mhandle *mh, struct dlm_rcom *rc) { - dlm_rcom_out(rc); dlm_midcomms_commit_mhandle(mh); } static void send_rcom_stateless(struct dlm_msg *msg, struct dlm_rcom *rc) { - dlm_rcom_out(rc); dlm_lowcomms_commit_msg(msg); dlm_lowcomms_put_msg(msg); } @@ -127,10 +125,10 @@ static int check_rcom_config(struct dlm_ls *ls, struct dlm_rcom *rc, int nodeid) { struct rcom_config *rf = (struct rcom_config *) rc->rc_buf; - if ((rc->rc_header.h_version & 0xFFFF0000) != DLM_HEADER_MAJOR) { + if ((le32_to_cpu(rc->rc_header.h_version) & 0xFFFF0000) != DLM_HEADER_MAJOR) { log_error(ls, "version mismatch: %x nodeid %d: %x", DLM_HEADER_MAJOR | DLM_HEADER_MINOR, nodeid, - rc->rc_header.h_version); + le32_to_cpu(rc->rc_header.h_version)); return -EPROTO; } @@ -145,10 +143,10 @@ static int check_rcom_config(struct dlm_ls *ls, struct dlm_rcom *rc, int nodeid) return 0; } -static void allow_sync_reply(struct dlm_ls *ls, uint64_t *new_seq) +static void allow_sync_reply(struct dlm_ls *ls, __le64 *new_seq) { spin_lock(&ls->ls_rcom_spin); - *new_seq = ++ls->ls_rcom_seq; + *new_seq = cpu_to_le64(++ls->ls_rcom_seq); set_bit(LSFL_RCOM_WAIT, &ls->ls_flags); spin_unlock(&ls->ls_rcom_spin); } @@ -182,7 +180,7 @@ int dlm_rcom_status(struct dlm_ls *ls, int nodeid, uint32_t status_flags) if (nodeid == dlm_our_nodeid()) { rc = ls->ls_recover_buf; - rc->rc_result = dlm_recover_status(ls); + rc->rc_result = cpu_to_le32(dlm_recover_status(ls)); goto out; } @@ -208,7 +206,7 @@ int dlm_rcom_status(struct dlm_ls *ls, int nodeid, uint32_t status_flags) rc = ls->ls_recover_buf; - if (rc->rc_result == -ESRCH) { + if (rc->rc_result == cpu_to_le32(-ESRCH)) { /* we pretend the remote lockspace exists with 0 status */ log_debug(ls, "remote node %d not ready", nodeid); rc->rc_result = 0; @@ -227,7 +225,7 @@ static void receive_rcom_status(struct dlm_ls *ls, struct dlm_rcom *rc_in) struct dlm_rcom *rc; struct rcom_status *rs; uint32_t status; - int nodeid = rc_in->rc_header.h_nodeid; + int nodeid = le32_to_cpu(rc_in->rc_header.h_nodeid); int len = sizeof(struct rcom_config); struct dlm_msg *msg; int num_slots = 0; @@ -259,7 +257,7 @@ static void receive_rcom_status(struct dlm_ls *ls, struct dlm_rcom *rc_in) rc->rc_id = rc_in->rc_id; rc->rc_seq_reply = rc_in->rc_seq; - rc->rc_result = status; + rc->rc_result = cpu_to_le32(status); set_rcom_config(ls, (struct rcom_config *)rc->rc_buf, num_slots); @@ -287,14 +285,16 @@ static void receive_sync_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in) { spin_lock(&ls->ls_rcom_spin); if (!test_bit(LSFL_RCOM_WAIT, &ls->ls_flags) || - rc_in->rc_id != ls->ls_rcom_seq) { + le64_to_cpu(rc_in->rc_id) != ls->ls_rcom_seq) { log_debug(ls, "reject reply %d from %d seq %llx expect %llx", - rc_in->rc_type, rc_in->rc_header.h_nodeid, - (unsigned long long)rc_in->rc_id, + le32_to_cpu(rc_in->rc_type), + le32_to_cpu(rc_in->rc_header.h_nodeid), + (unsigned long long)le64_to_cpu(rc_in->rc_id), (unsigned long long)ls->ls_rcom_seq); goto out; } - memcpy(ls->ls_recover_buf, rc_in, rc_in->rc_header.h_length); + memcpy(ls->ls_recover_buf, rc_in, + le16_to_cpu(rc_in->rc_header.h_length)); set_bit(LSFL_RCOM_READY, &ls->ls_flags); clear_bit(LSFL_RCOM_WAIT, &ls->ls_flags); wake_up(&ls->ls_wait_general); @@ -336,8 +336,9 @@ static void receive_rcom_names(struct dlm_ls *ls, struct dlm_rcom *rc_in) int error, inlen, outlen, nodeid; struct dlm_msg *msg; - nodeid = rc_in->rc_header.h_nodeid; - inlen = rc_in->rc_header.h_length - sizeof(struct dlm_rcom); + nodeid = le32_to_cpu(rc_in->rc_header.h_nodeid); + inlen = le16_to_cpu(rc_in->rc_header.h_length) - + sizeof(struct dlm_rcom); outlen = DLM_MAX_APP_BUFSIZE - sizeof(struct dlm_rcom); error = create_rcom_stateless(ls, nodeid, DLM_RCOM_NAMES_REPLY, outlen, @@ -364,7 +365,7 @@ int dlm_send_rcom_lookup(struct dlm_rsb *r, int dir_nodeid) if (error) goto out; memcpy(rc->rc_buf, r->res_name, r->res_length); - rc->rc_id = (unsigned long) r->res_id; + rc->rc_id = cpu_to_le64(r->res_id); send_rcom(mh, rc); out: @@ -375,11 +376,12 @@ static void receive_rcom_lookup(struct dlm_ls *ls, struct dlm_rcom *rc_in) { struct dlm_rcom *rc; struct dlm_mhandle *mh; - int error, ret_nodeid, nodeid = rc_in->rc_header.h_nodeid; - int len = rc_in->rc_header.h_length - sizeof(struct dlm_rcom); + int error, ret_nodeid, nodeid = le32_to_cpu(rc_in->rc_header.h_nodeid); + int len = le16_to_cpu(rc_in->rc_header.h_length) - + sizeof(struct dlm_rcom); /* Old code would send this special id to trigger a debug dump. */ - if (rc_in->rc_id == 0xFFFFFFFF) { + if (rc_in->rc_id == cpu_to_le64(0xFFFFFFFF)) { log_error(ls, "receive_rcom_lookup dump from %d", nodeid); dlm_dump_rsb_name(ls, rc_in->rc_buf, len); return; @@ -393,7 +395,7 @@ static void receive_rcom_lookup(struct dlm_ls *ls, struct dlm_rcom *rc_in) DLM_LU_RECOVER_MASTER, &ret_nodeid, NULL); if (error) ret_nodeid = error; - rc->rc_result = ret_nodeid; + rc->rc_result = cpu_to_le32(ret_nodeid); rc->rc_id = rc_in->rc_id; rc->rc_seq_reply = rc_in->rc_seq; @@ -452,7 +454,7 @@ int dlm_send_rcom_lock(struct dlm_rsb *r, struct dlm_lkb *lkb) rl = (struct rcom_lock *) rc->rc_buf; pack_rcom_lock(r, lkb, rl); - rc->rc_id = (unsigned long) r; + rc->rc_id = cpu_to_le64((uintptr_t)r); send_rcom(mh, rc); out: @@ -464,7 +466,7 @@ static void receive_rcom_lock(struct dlm_ls *ls, struct dlm_rcom *rc_in) { struct dlm_rcom *rc; struct dlm_mhandle *mh; - int error, nodeid = rc_in->rc_header.h_nodeid; + int error, nodeid = le32_to_cpu(rc_in->rc_header.h_nodeid); dlm_recover_master_copy(ls, rc_in); @@ -500,21 +502,20 @@ int dlm_send_ls_not_ready(int nodeid, struct dlm_rcom *rc_in) rc = (struct dlm_rcom *) mb; - rc->rc_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR); + rc->rc_header.h_version = cpu_to_le32(DLM_HEADER_MAJOR | DLM_HEADER_MINOR); rc->rc_header.u.h_lockspace = rc_in->rc_header.u.h_lockspace; - rc->rc_header.h_nodeid = dlm_our_nodeid(); - rc->rc_header.h_length = mb_len; + rc->rc_header.h_nodeid = cpu_to_le32(dlm_our_nodeid()); + rc->rc_header.h_length = cpu_to_le16(mb_len); rc->rc_header.h_cmd = DLM_RCOM; - rc->rc_type = DLM_RCOM_STATUS_REPLY; + rc->rc_type = cpu_to_le32(DLM_RCOM_STATUS_REPLY); rc->rc_id = rc_in->rc_id; rc->rc_seq_reply = rc_in->rc_seq; - rc->rc_result = -ESRCH; + rc->rc_result = cpu_to_le32(-ESRCH); rf = (struct rcom_config *) rc->rc_buf; rf->rf_lvblen = cpu_to_le32(~0U); - dlm_rcom_out(rc); dlm_midcomms_commit_mhandle(mh); return 0; @@ -573,27 +574,27 @@ void dlm_receive_rcom(struct dlm_ls *ls, struct dlm_rcom *rc, int nodeid) uint64_t seq; switch (rc->rc_type) { - case DLM_RCOM_STATUS_REPLY: + case cpu_to_le32(DLM_RCOM_STATUS_REPLY): reply = 1; break; - case DLM_RCOM_NAMES: + case cpu_to_le32(DLM_RCOM_NAMES): names = 1; break; - case DLM_RCOM_NAMES_REPLY: + case cpu_to_le32(DLM_RCOM_NAMES_REPLY): names = 1; reply = 1; break; - case DLM_RCOM_LOOKUP: + case cpu_to_le32(DLM_RCOM_LOOKUP): lookup = 1; break; - case DLM_RCOM_LOOKUP_REPLY: + case cpu_to_le32(DLM_RCOM_LOOKUP_REPLY): lookup = 1; reply = 1; break; - case DLM_RCOM_LOCK: + case cpu_to_le32(DLM_RCOM_LOCK): lock = 1; break; - case DLM_RCOM_LOCK_REPLY: + case cpu_to_le32(DLM_RCOM_LOCK_REPLY): lock = 1; reply = 1; break; @@ -605,10 +606,10 @@ void dlm_receive_rcom(struct dlm_ls *ls, struct dlm_rcom *rc, int nodeid) seq = ls->ls_recover_seq; spin_unlock(&ls->ls_recover_lock); - if (stop && (rc->rc_type != DLM_RCOM_STATUS)) + if (stop && (rc->rc_type != cpu_to_le32(DLM_RCOM_STATUS))) goto ignore; - if (reply && (rc->rc_seq_reply != seq)) + if (reply && (le64_to_cpu(rc->rc_seq_reply) != seq)) goto ignore; if (!(status & DLM_RS_NODES) && (names || lookup || lock)) @@ -618,59 +619,60 @@ void dlm_receive_rcom(struct dlm_ls *ls, struct dlm_rcom *rc, int nodeid) goto ignore; switch (rc->rc_type) { - case DLM_RCOM_STATUS: + case cpu_to_le32(DLM_RCOM_STATUS): receive_rcom_status(ls, rc); break; - case DLM_RCOM_NAMES: + case cpu_to_le32(DLM_RCOM_NAMES): receive_rcom_names(ls, rc); break; - case DLM_RCOM_LOOKUP: + case cpu_to_le32(DLM_RCOM_LOOKUP): receive_rcom_lookup(ls, rc); break; - case DLM_RCOM_LOCK: - if (rc->rc_header.h_length < lock_size) + case cpu_to_le32(DLM_RCOM_LOCK): + if (le16_to_cpu(rc->rc_header.h_length) < lock_size) goto Eshort; receive_rcom_lock(ls, rc); break; - case DLM_RCOM_STATUS_REPLY: + case cpu_to_le32(DLM_RCOM_STATUS_REPLY): receive_sync_reply(ls, rc); break; - case DLM_RCOM_NAMES_REPLY: + case cpu_to_le32(DLM_RCOM_NAMES_REPLY): receive_sync_reply(ls, rc); break; - case DLM_RCOM_LOOKUP_REPLY: + case cpu_to_le32(DLM_RCOM_LOOKUP_REPLY): receive_rcom_lookup_reply(ls, rc); break; - case DLM_RCOM_LOCK_REPLY: - if (rc->rc_header.h_length < lock_size) + case cpu_to_le32(DLM_RCOM_LOCK_REPLY): + if (le16_to_cpu(rc->rc_header.h_length) < lock_size) goto Eshort; dlm_recover_process_copy(ls, rc); break; default: - log_error(ls, "receive_rcom bad type %d", rc->rc_type); + log_error(ls, "receive_rcom bad type %d", + le32_to_cpu(rc->rc_type)); } return; ignore: log_limit(ls, "dlm_receive_rcom ignore msg %d " "from %d %llu %llu recover seq %llu sts %x gen %u", - rc->rc_type, + le32_to_cpu(rc->rc_type), nodeid, - (unsigned long long)rc->rc_seq, - (unsigned long long)rc->rc_seq_reply, + (unsigned long long)le64_to_cpu(rc->rc_seq), + (unsigned long long)le64_to_cpu(rc->rc_seq_reply), (unsigned long long)seq, status, ls->ls_generation); return; Eshort: log_error(ls, "recovery message %d from %d is too short", - rc->rc_type, nodeid); + le32_to_cpu(rc->rc_type), nodeid); } diff --git a/fs/dlm/recover.c b/fs/dlm/recover.c index 8928e99dfd47..ccff1791803f 100644 --- a/fs/dlm/recover.c +++ b/fs/dlm/recover.c @@ -114,7 +114,7 @@ static int wait_status_all(struct dlm_ls *ls, uint32_t wait_status, if (save_slots) dlm_slot_save(ls, rc, memb); - if (rc->rc_result & wait_status) + if (le32_to_cpu(rc->rc_result) & wait_status) break; if (delay < 1000) delay += 20; @@ -141,7 +141,7 @@ static int wait_status_low(struct dlm_ls *ls, uint32_t wait_status, if (error) break; - if (rc->rc_result & wait_status) + if (le32_to_cpu(rc->rc_result) & wait_status) break; if (delay < 1000) delay += 20; @@ -568,14 +568,14 @@ int dlm_recover_master_reply(struct dlm_ls *ls, struct dlm_rcom *rc) struct dlm_rsb *r; int ret_nodeid, new_master; - r = recover_idr_find(ls, rc->rc_id); + r = recover_idr_find(ls, le64_to_cpu(rc->rc_id)); if (!r) { log_error(ls, "dlm_recover_master_reply no id %llx", - (unsigned long long)rc->rc_id); + (unsigned long long)le64_to_cpu(rc->rc_id)); goto out; } - ret_nodeid = rc->rc_result; + ret_nodeid = le32_to_cpu(rc->rc_result); if (ret_nodeid == dlm_our_nodeid()) new_master = 0; @@ -732,10 +732,9 @@ void dlm_recovered_lock(struct dlm_rsb *r) static void recover_lvb(struct dlm_rsb *r) { - struct dlm_lkb *lkb, *high_lkb = NULL; + struct dlm_lkb *big_lkb = NULL, *iter, *high_lkb = NULL; uint32_t high_seq = 0; int lock_lvb_exists = 0; - int big_lock_exists = 0; int lvblen = r->res_ls->ls_lvblen; if (!rsb_flag(r, RSB_NEW_MASTER2) && @@ -751,37 +750,37 @@ static void recover_lvb(struct dlm_rsb *r) /* we are the new master, so figure out if VALNOTVALID should be set, and set the rsb lvb from the best lkb available. */ - list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue) { - if (!(lkb->lkb_exflags & DLM_LKF_VALBLK)) + list_for_each_entry(iter, &r->res_grantqueue, lkb_statequeue) { + if (!(iter->lkb_exflags & DLM_LKF_VALBLK)) continue; lock_lvb_exists = 1; - if (lkb->lkb_grmode > DLM_LOCK_CR) { - big_lock_exists = 1; + if (iter->lkb_grmode > DLM_LOCK_CR) { + big_lkb = iter; goto setflag; } - if (((int)lkb->lkb_lvbseq - (int)high_seq) >= 0) { - high_lkb = lkb; - high_seq = lkb->lkb_lvbseq; + if (((int)iter->lkb_lvbseq - (int)high_seq) >= 0) { + high_lkb = iter; + high_seq = iter->lkb_lvbseq; } } - list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue) { - if (!(lkb->lkb_exflags & DLM_LKF_VALBLK)) + list_for_each_entry(iter, &r->res_convertqueue, lkb_statequeue) { + if (!(iter->lkb_exflags & DLM_LKF_VALBLK)) continue; lock_lvb_exists = 1; - if (lkb->lkb_grmode > DLM_LOCK_CR) { - big_lock_exists = 1; + if (iter->lkb_grmode > DLM_LOCK_CR) { + big_lkb = iter; goto setflag; } - if (((int)lkb->lkb_lvbseq - (int)high_seq) >= 0) { - high_lkb = lkb; - high_seq = lkb->lkb_lvbseq; + if (((int)iter->lkb_lvbseq - (int)high_seq) >= 0) { + high_lkb = iter; + high_seq = iter->lkb_lvbseq; } } @@ -790,7 +789,7 @@ static void recover_lvb(struct dlm_rsb *r) goto out; /* lvb is invalidated if only NL/CR locks remain */ - if (!big_lock_exists) + if (!big_lkb) rsb_set_flag(r, RSB_VALNOTVALID); if (!r->res_lvbptr) { @@ -799,9 +798,9 @@ static void recover_lvb(struct dlm_rsb *r) goto out; } - if (big_lock_exists) { - r->res_lvbseq = lkb->lkb_lvbseq; - memcpy(r->res_lvbptr, lkb->lkb_lvbptr, lvblen); + if (big_lkb) { + r->res_lvbseq = big_lkb->lkb_lvbseq; + memcpy(r->res_lvbptr, big_lkb->lkb_lvbptr, lvblen); } else if (high_lkb) { r->res_lvbseq = high_lkb->lkb_lvbseq; memcpy(r->res_lvbptr, high_lkb->lkb_lvbptr, lvblen); diff --git a/fs/dlm/requestqueue.c b/fs/dlm/requestqueue.c index ccb5307c21e9..036a9a0078f6 100644 --- a/fs/dlm/requestqueue.c +++ b/fs/dlm/requestqueue.c @@ -14,6 +14,7 @@ #include "dir.h" #include "config.h" #include "requestqueue.h" +#include "util.h" struct rq_entry { struct list_head list; @@ -32,7 +33,8 @@ struct rq_entry { void dlm_add_requestqueue(struct dlm_ls *ls, int nodeid, struct dlm_message *ms) { struct rq_entry *e; - int length = ms->m_header.h_length - sizeof(struct dlm_message); + int length = le16_to_cpu(ms->m_header.h_length) - + sizeof(struct dlm_message); e = kmalloc(sizeof(struct rq_entry) + length, GFP_NOFS); if (!e) { @@ -42,7 +44,7 @@ void dlm_add_requestqueue(struct dlm_ls *ls, int nodeid, struct dlm_message *ms) e->recover_seq = ls->ls_recover_seq & 0xFFFFFFFF; e->nodeid = nodeid; - memcpy(&e->request, ms, ms->m_header.h_length); + memcpy(&e->request, ms, le16_to_cpu(ms->m_header.h_length)); atomic_inc(&ls->ls_requestqueue_cnt); mutex_lock(&ls->ls_requestqueue_mutex); @@ -82,8 +84,10 @@ int dlm_process_requestqueue(struct dlm_ls *ls) log_limit(ls, "dlm_process_requestqueue msg %d from %d " "lkid %x remid %x result %d seq %u", - ms->m_type, ms->m_header.h_nodeid, - ms->m_lkid, ms->m_remid, ms->m_result, + le32_to_cpu(ms->m_type), + le32_to_cpu(ms->m_header.h_nodeid), + le32_to_cpu(ms->m_lkid), le32_to_cpu(ms->m_remid), + from_dlm_errno(le32_to_cpu(ms->m_result)), e->recover_seq); dlm_receive_message_saved(ls, &e->request, e->recover_seq); @@ -124,7 +128,7 @@ void dlm_wait_requestqueue(struct dlm_ls *ls) static int purge_request(struct dlm_ls *ls, struct dlm_message *ms, int nodeid) { - uint32_t type = ms->m_type; + __le32 type = ms->m_type; /* the ls is being cleaned up and freed by release_lockspace */ if (!atomic_read(&ls->ls_count)) @@ -136,9 +140,9 @@ static int purge_request(struct dlm_ls *ls, struct dlm_message *ms, int nodeid) /* directory operations are always purged because the directory is always rebuilt during recovery and the lookups resent */ - if (type == DLM_MSG_REMOVE || - type == DLM_MSG_LOOKUP || - type == DLM_MSG_LOOKUP_REPLY) + if (type == cpu_to_le32(DLM_MSG_REMOVE) || + type == cpu_to_le32(DLM_MSG_LOOKUP) || + type == cpu_to_le32(DLM_MSG_LOOKUP_REPLY)) return 1; if (!dlm_no_directory(ls)) diff --git a/fs/dlm/user.c b/fs/dlm/user.c index e5cefa90b1ce..1060b24f18d4 100644 --- a/fs/dlm/user.c +++ b/fs/dlm/user.c @@ -108,11 +108,11 @@ static void compat_input(struct dlm_write_request *kb, kb->i.lock.parent = kb32->i.lock.parent; kb->i.lock.xid = kb32->i.lock.xid; kb->i.lock.timeout = kb32->i.lock.timeout; - kb->i.lock.castparam = (void *)(long)kb32->i.lock.castparam; - kb->i.lock.castaddr = (void *)(long)kb32->i.lock.castaddr; - kb->i.lock.bastparam = (void *)(long)kb32->i.lock.bastparam; - kb->i.lock.bastaddr = (void *)(long)kb32->i.lock.bastaddr; - kb->i.lock.lksb = (void *)(long)kb32->i.lock.lksb; + kb->i.lock.castparam = (__user void *)(long)kb32->i.lock.castparam; + kb->i.lock.castaddr = (__user void *)(long)kb32->i.lock.castaddr; + kb->i.lock.bastparam = (__user void *)(long)kb32->i.lock.bastparam; + kb->i.lock.bastaddr = (__user void *)(long)kb32->i.lock.bastaddr; + kb->i.lock.lksb = (__user void *)(long)kb32->i.lock.lksb; memcpy(kb->i.lock.lvb, kb32->i.lock.lvb, DLM_USER_LVB_LEN); memcpy(kb->i.lock.name, kb32->i.lock.name, namelen); } @@ -127,9 +127,9 @@ static void compat_output(struct dlm_lock_result *res, res32->version[1] = res->version[1]; res32->version[2] = res->version[2]; - res32->user_astaddr = (__u32)(long)res->user_astaddr; - res32->user_astparam = (__u32)(long)res->user_astparam; - res32->user_lksb = (__u32)(long)res->user_lksb; + res32->user_astaddr = (__u32)(__force long)res->user_astaddr; + res32->user_astparam = (__u32)(__force long)res->user_astparam; + res32->user_lksb = (__u32)(__force long)res->user_lksb; res32->bast_mode = res->bast_mode; res32->lvb_offset = res->lvb_offset; diff --git a/fs/dlm/util.c b/fs/dlm/util.c index 58acbcc2081a..f2bc401f312f 100644 --- a/fs/dlm/util.c +++ b/fs/dlm/util.c @@ -20,28 +20,10 @@ #define DLM_ERRNO_ETIMEDOUT 110 #define DLM_ERRNO_EINPROGRESS 115 -void header_out(struct dlm_header *hd) -{ - hd->h_version = cpu_to_le32(hd->h_version); - /* does it for others u32 in union as well */ - hd->u.h_lockspace = cpu_to_le32(hd->u.h_lockspace); - hd->h_nodeid = cpu_to_le32(hd->h_nodeid); - hd->h_length = cpu_to_le16(hd->h_length); -} - -void header_in(struct dlm_header *hd) -{ - hd->h_version = le32_to_cpu(hd->h_version); - /* does it for others u32 in union as well */ - hd->u.h_lockspace = le32_to_cpu(hd->u.h_lockspace); - hd->h_nodeid = le32_to_cpu(hd->h_nodeid); - hd->h_length = le16_to_cpu(hd->h_length); -} - /* higher errno values are inconsistent across architectures, so select one set of values for on the wire */ -static int to_dlm_errno(int err) +int to_dlm_errno(int err) { switch (err) { case -EDEADLK: @@ -62,7 +44,7 @@ static int to_dlm_errno(int err) return err; } -static int from_dlm_errno(int err) +int from_dlm_errno(int err) { switch (err) { case -DLM_ERRNO_EDEADLK: @@ -82,73 +64,3 @@ static int from_dlm_errno(int err) } return err; } - -void dlm_message_out(struct dlm_message *ms) -{ - header_out(&ms->m_header); - - ms->m_type = cpu_to_le32(ms->m_type); - ms->m_nodeid = cpu_to_le32(ms->m_nodeid); - ms->m_pid = cpu_to_le32(ms->m_pid); - ms->m_lkid = cpu_to_le32(ms->m_lkid); - ms->m_remid = cpu_to_le32(ms->m_remid); - ms->m_parent_lkid = cpu_to_le32(ms->m_parent_lkid); - ms->m_parent_remid = cpu_to_le32(ms->m_parent_remid); - ms->m_exflags = cpu_to_le32(ms->m_exflags); - ms->m_sbflags = cpu_to_le32(ms->m_sbflags); - ms->m_flags = cpu_to_le32(ms->m_flags); - ms->m_lvbseq = cpu_to_le32(ms->m_lvbseq); - ms->m_hash = cpu_to_le32(ms->m_hash); - ms->m_status = cpu_to_le32(ms->m_status); - ms->m_grmode = cpu_to_le32(ms->m_grmode); - ms->m_rqmode = cpu_to_le32(ms->m_rqmode); - ms->m_bastmode = cpu_to_le32(ms->m_bastmode); - ms->m_asts = cpu_to_le32(ms->m_asts); - ms->m_result = cpu_to_le32(to_dlm_errno(ms->m_result)); -} - -void dlm_message_in(struct dlm_message *ms) -{ - header_in(&ms->m_header); - - ms->m_type = le32_to_cpu(ms->m_type); - ms->m_nodeid = le32_to_cpu(ms->m_nodeid); - ms->m_pid = le32_to_cpu(ms->m_pid); - ms->m_lkid = le32_to_cpu(ms->m_lkid); - ms->m_remid = le32_to_cpu(ms->m_remid); - ms->m_parent_lkid = le32_to_cpu(ms->m_parent_lkid); - ms->m_parent_remid = le32_to_cpu(ms->m_parent_remid); - ms->m_exflags = le32_to_cpu(ms->m_exflags); - ms->m_sbflags = le32_to_cpu(ms->m_sbflags); - ms->m_flags = le32_to_cpu(ms->m_flags); - ms->m_lvbseq = le32_to_cpu(ms->m_lvbseq); - ms->m_hash = le32_to_cpu(ms->m_hash); - ms->m_status = le32_to_cpu(ms->m_status); - ms->m_grmode = le32_to_cpu(ms->m_grmode); - ms->m_rqmode = le32_to_cpu(ms->m_rqmode); - ms->m_bastmode = le32_to_cpu(ms->m_bastmode); - ms->m_asts = le32_to_cpu(ms->m_asts); - ms->m_result = from_dlm_errno(le32_to_cpu(ms->m_result)); -} - -void dlm_rcom_out(struct dlm_rcom *rc) -{ - header_out(&rc->rc_header); - - rc->rc_type = cpu_to_le32(rc->rc_type); - rc->rc_result = cpu_to_le32(rc->rc_result); - rc->rc_id = cpu_to_le64(rc->rc_id); - rc->rc_seq = cpu_to_le64(rc->rc_seq); - rc->rc_seq_reply = cpu_to_le64(rc->rc_seq_reply); -} - -void dlm_rcom_in(struct dlm_rcom *rc) -{ - header_in(&rc->rc_header); - - rc->rc_type = le32_to_cpu(rc->rc_type); - rc->rc_result = le32_to_cpu(rc->rc_result); - rc->rc_id = le64_to_cpu(rc->rc_id); - rc->rc_seq = le64_to_cpu(rc->rc_seq); - rc->rc_seq_reply = le64_to_cpu(rc->rc_seq_reply); -} diff --git a/fs/dlm/util.h b/fs/dlm/util.h index d46f23c7a6a0..b6a4b8adca8d 100644 --- a/fs/dlm/util.h +++ b/fs/dlm/util.h @@ -11,12 +11,8 @@ #ifndef __UTIL_DOT_H__ #define __UTIL_DOT_H__ -void dlm_message_out(struct dlm_message *ms); -void dlm_message_in(struct dlm_message *ms); -void dlm_rcom_out(struct dlm_rcom *rc); -void dlm_rcom_in(struct dlm_rcom *rc); -void header_out(struct dlm_header *hd); -void header_in(struct dlm_header *hd); +int to_dlm_errno(int err); +int from_dlm_errno(int err); #endif