lock.c (b434eda6fda5bcdcc2dd918e5ffbf7184f2d4e17) lock.c (c36258b5925e6cf6bf72904635100593573bfcff)
1/******************************************************************************
2*******************************************************************************
3**
4** Copyright (C) 2005-2007 Red Hat, Inc. All rights reserved.
5**
6** This copyrighted material is made available to anyone wishing to use,
7** modify, copy, or redistribute it subject to the terms and conditions
8** of the GNU General Public License v.2.

--- 3624 unchanged lines hidden (view full) ---

3633 if (!ret_nodeid)
3634 process_lookup_list(r);
3635 out:
3636 unlock_rsb(r);
3637 put_rsb(r);
3638 dlm_put_lkb(lkb);
3639}
3640
1/******************************************************************************
2*******************************************************************************
3**
4** Copyright (C) 2005-2007 Red Hat, Inc. All rights reserved.
5**
6** This copyrighted material is made available to anyone wishing to use,
7** modify, copy, or redistribute it subject to the terms and conditions
8** of the GNU General Public License v.2.

--- 3624 unchanged lines hidden (view full) ---

3633 if (!ret_nodeid)
3634 process_lookup_list(r);
3635 out:
3636 unlock_rsb(r);
3637 put_rsb(r);
3638 dlm_put_lkb(lkb);
3639}
3640
3641int dlm_receive_message(struct dlm_header *hd, int nodeid, int recovery)
3641static void _receive_message(struct dlm_ls *ls, struct dlm_message *ms)
3642{
3642{
3643 struct dlm_message *ms = (struct dlm_message *) hd;
3644 struct dlm_ls *ls;
3645 int error = 0;
3646
3647 if (!recovery)
3648 dlm_message_in(ms);
3649
3650 ls = dlm_find_lockspace_global(hd->h_lockspace);
3651 if (!ls) {
3652 log_print("drop message %d from %d for unknown lockspace %d",
3653 ms->m_type, nodeid, hd->h_lockspace);
3654 return -EINVAL;
3655 }
3656
3657 /* recovery may have just ended leaving a bunch of backed-up requests
3658 in the requestqueue; wait while dlm_recoverd clears them */
3659
3660 if (!recovery)
3661 dlm_wait_requestqueue(ls);
3662
3663 /* recovery may have just started while there were a bunch of
3664 in-flight requests -- save them in requestqueue to be processed
3665 after recovery. we can't let dlm_recvd block on the recovery
3666 lock. if dlm_recoverd is calling this function to clear the
3667 requestqueue, it needs to be interrupted (-EINTR) if another
3668 recovery operation is starting. */
3669
3670 while (1) {
3671 if (dlm_locking_stopped(ls)) {
3672 if (recovery) {
3673 error = -EINTR;
3674 goto out;
3675 }
3676 error = dlm_add_requestqueue(ls, nodeid, hd);
3677 if (error == -EAGAIN)
3678 continue;
3679 else {
3680 error = -EINTR;
3681 goto out;
3682 }
3683 }
3684
3685 if (dlm_lock_recovery_try(ls))
3686 break;
3687 schedule();
3688 }
3689
3690 switch (ms->m_type) {
3691
3692 /* messages sent to a master node */
3693
3694 case DLM_MSG_REQUEST:
3695 receive_request(ls, ms);
3696 break;
3697

--- 58 unchanged lines hidden (view full) ---

3756 case DLM_MSG_PURGE:
3757 receive_purge(ls, ms);
3758 break;
3759
3760 default:
3761 log_error(ls, "unknown message type %d", ms->m_type);
3762 }
3763
3643 switch (ms->m_type) {
3644
3645 /* messages sent to a master node */
3646
3647 case DLM_MSG_REQUEST:
3648 receive_request(ls, ms);
3649 break;
3650

--- 58 unchanged lines hidden (view full) ---

3709 case DLM_MSG_PURGE:
3710 receive_purge(ls, ms);
3711 break;
3712
3713 default:
3714 log_error(ls, "unknown message type %d", ms->m_type);
3715 }
3716
3764 dlm_unlock_recovery(ls);
3765 out:
3766 dlm_put_lockspace(ls);
3767 dlm_astd_wake();
3717 dlm_astd_wake();
3768 return error;
3769}
3770
3718}
3719
3720/* If the lockspace is in recovery mode (locking stopped), then normal
3721 messages are saved on the requestqueue for processing after recovery is
3722 done. When not in recovery mode, we wait for dlm_recoverd to drain saved
3723 messages off the requestqueue before we process new ones. This occurs right
3724 after recovery completes when we transition from saving all messages on
3725 requestqueue, to processing all the saved messages, to processing new
3726 messages as they arrive. */
3771
3727
3772/*
3773 * Recovery related
3774 */
3728static void dlm_receive_message(struct dlm_ls *ls, struct dlm_message *ms,
3729 int nodeid)
3730{
3731 if (dlm_locking_stopped(ls)) {
3732 dlm_add_requestqueue(ls, nodeid, (struct dlm_header *) ms);
3733 } else {
3734 dlm_wait_requestqueue(ls);
3735 _receive_message(ls, ms);
3736 }
3737}
3775
3738
3739/* This is called by dlm_recoverd to process messages that were saved on
3740 the requestqueue. */
3741
3742void dlm_receive_message_saved(struct dlm_ls *ls, struct dlm_message *ms)
3743{
3744 _receive_message(ls, ms);
3745}
3746
3747/* This is called by the midcomms layer when something is received for
3748 the lockspace. It could be either a MSG (normal message sent as part of
3749 standard locking activity) or an RCOM (recovery message sent as part of
3750 lockspace recovery). */
3751
3752void dlm_receive_buffer(struct dlm_header *hd, int nodeid)
3753{
3754 struct dlm_message *ms = (struct dlm_message *) hd;
3755 struct dlm_rcom *rc = (struct dlm_rcom *) hd;
3756 struct dlm_ls *ls;
3757 int type = 0;
3758
3759 switch (hd->h_cmd) {
3760 case DLM_MSG:
3761 dlm_message_in(ms);
3762 type = ms->m_type;
3763 break;
3764 case DLM_RCOM:
3765 dlm_rcom_in(rc);
3766 type = rc->rc_type;
3767 break;
3768 default:
3769 log_print("invalid h_cmd %d from %u", hd->h_cmd, nodeid);
3770 return;
3771 }
3772
3773 if (hd->h_nodeid != nodeid) {
3774 log_print("invalid h_nodeid %d from %d lockspace %x",
3775 hd->h_nodeid, nodeid, hd->h_lockspace);
3776 return;
3777 }
3778
3779 ls = dlm_find_lockspace_global(hd->h_lockspace);
3780 if (!ls) {
3781 log_print("invalid h_lockspace %x from %d cmd %d type %d",
3782 hd->h_lockspace, nodeid, hd->h_cmd, type);
3783
3784 if (hd->h_cmd == DLM_RCOM && type == DLM_RCOM_STATUS)
3785 dlm_send_ls_not_ready(nodeid, rc);
3786 return;
3787 }
3788
3789 /* this rwsem allows dlm_ls_stop() to wait for all dlm_recv threads to
3790 be inactive (in this ls) before transitioning to recovery mode */
3791
3792 down_read(&ls->ls_recv_active);
3793 if (hd->h_cmd == DLM_MSG)
3794 dlm_receive_message(ls, ms, nodeid);
3795 else
3796 dlm_receive_rcom(ls, rc, nodeid);
3797 up_read(&ls->ls_recv_active);
3798
3799 dlm_put_lockspace(ls);
3800}
3801
3776static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb)
3777{
3778 if (middle_conversion(lkb)) {
3779 hold_lkb(lkb);
3780 ls->ls_stub_ms.m_type = DLM_MSG_CONVERT_REPLY;
3781 ls->ls_stub_ms.m_result = -EINPROGRESS;
3782 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
3783 _receive_convert_reply(lkb, &ls->ls_stub_ms);

--- 969 unchanged lines hidden --- 		3802static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb)
3803{
3804 if (middle_conversion(lkb)) {
3805 hold_lkb(lkb);
3806 ls->ls_stub_ms.m_type = DLM_MSG_CONVERT_REPLY;
3807 ls->ls_stub_ms.m_result = -EINPROGRESS;
3808 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
3809 _receive_convert_reply(lkb, &ls->ls_stub_ms);

--- 969 unchanged lines hidden ---