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. 9 ** 10 ******************************************************************************* 11 ******************************************************************************/ 12 13 #include "dlm_internal.h" 14 #include "member.h" 15 #include "lock.h" 16 #include "dir.h" 17 #include "config.h" 18 #include "requestqueue.h" 19 20 struct rq_entry { 21 struct list_head list; 22 uint32_t recover_seq; 23 int nodeid; 24 struct dlm_message request; 25 }; 26 27 /* 28 * Requests received while the lockspace is in recovery get added to the 29 * request queue and processed when recovery is complete. This happens when 30 * the lockspace is suspended on some nodes before it is on others, or the 31 * lockspace is enabled on some while still suspended on others. 32 */ 33 34 void dlm_add_requestqueue(struct dlm_ls *ls, int nodeid, struct dlm_message *ms) 35 { 36 struct rq_entry *e; 37 int length = ms->m_header.h_length - sizeof(struct dlm_message); 38 39 e = kmalloc(sizeof(struct rq_entry) + length, GFP_NOFS); 40 if (!e) { 41 log_print("dlm_add_requestqueue: out of memory len %d", length); 42 return; 43 } 44 45 e->recover_seq = ls->ls_recover_seq & 0xFFFFFFFF; 46 e->nodeid = nodeid; 47 memcpy(&e->request, ms, ms->m_header.h_length); 48 49 mutex_lock(&ls->ls_requestqueue_mutex); 50 list_add_tail(&e->list, &ls->ls_requestqueue); 51 mutex_unlock(&ls->ls_requestqueue_mutex); 52 } 53 54 /* 55 * Called by dlm_recoverd to process normal messages saved while recovery was 56 * happening. Normal locking has been enabled before this is called. dlm_recv 57 * upon receiving a message, will wait for all saved messages to be drained 58 * here before processing the message it got. If a new dlm_ls_stop() arrives 59 * while we're processing these saved messages, it may block trying to suspend 60 * dlm_recv if dlm_recv is waiting for us in dlm_wait_requestqueue. In that 61 * case, we don't abort since locking_stopped is still 0. If dlm_recv is not 62 * waiting for us, then this processing may be aborted due to locking_stopped. 63 */ 64 65 int dlm_process_requestqueue(struct dlm_ls *ls) 66 { 67 struct rq_entry *e; 68 int error = 0; 69 70 mutex_lock(&ls->ls_requestqueue_mutex); 71 72 for (;;) { 73 if (list_empty(&ls->ls_requestqueue)) { 74 mutex_unlock(&ls->ls_requestqueue_mutex); 75 error = 0; 76 break; 77 } 78 e = list_entry(ls->ls_requestqueue.next, struct rq_entry, list); 79 mutex_unlock(&ls->ls_requestqueue_mutex); 80 81 dlm_receive_message_saved(ls, &e->request, e->recover_seq); 82 83 mutex_lock(&ls->ls_requestqueue_mutex); 84 list_del(&e->list); 85 kfree(e); 86 87 if (dlm_locking_stopped(ls)) { 88 log_debug(ls, "process_requestqueue abort running"); 89 mutex_unlock(&ls->ls_requestqueue_mutex); 90 error = -EINTR; 91 break; 92 } 93 schedule(); 94 } 95 96 return error; 97 } 98 99 /* 100 * After recovery is done, locking is resumed and dlm_recoverd takes all the 101 * saved requests and processes them as they would have been by dlm_recv. At 102 * the same time, dlm_recv will start receiving new requests from remote nodes. 103 * We want to delay dlm_recv processing new requests until dlm_recoverd has 104 * finished processing the old saved requests. We don't check for locking 105 * stopped here because dlm_ls_stop won't stop locking until it's suspended us 106 * (dlm_recv). 107 */ 108 109 void dlm_wait_requestqueue(struct dlm_ls *ls) 110 { 111 for (;;) { 112 mutex_lock(&ls->ls_requestqueue_mutex); 113 if (list_empty(&ls->ls_requestqueue)) 114 break; 115 mutex_unlock(&ls->ls_requestqueue_mutex); 116 schedule(); 117 } 118 mutex_unlock(&ls->ls_requestqueue_mutex); 119 } 120 121 static int purge_request(struct dlm_ls *ls, struct dlm_message *ms, int nodeid) 122 { 123 uint32_t type = ms->m_type; 124 125 /* the ls is being cleaned up and freed by release_lockspace */ 126 if (!ls->ls_count) 127 return 1; 128 129 if (dlm_is_removed(ls, nodeid)) 130 return 1; 131 132 /* directory operations are always purged because the directory is 133 always rebuilt during recovery and the lookups resent */ 134 135 if (type == DLM_MSG_REMOVE || 136 type == DLM_MSG_LOOKUP || 137 type == DLM_MSG_LOOKUP_REPLY) 138 return 1; 139 140 if (!dlm_no_directory(ls)) 141 return 0; 142 143 /* with no directory, the master is likely to change as a part of 144 recovery; requests to/from the defunct master need to be purged */ 145 146 switch (type) { 147 case DLM_MSG_REQUEST: 148 case DLM_MSG_CONVERT: 149 case DLM_MSG_UNLOCK: 150 case DLM_MSG_CANCEL: 151 /* we're no longer the master of this resource, the sender 152 will resend to the new master (see waiter_needs_recovery) */ 153 154 if (dlm_hash2nodeid(ls, ms->m_hash) != dlm_our_nodeid()) 155 return 1; 156 break; 157 158 case DLM_MSG_REQUEST_REPLY: 159 case DLM_MSG_CONVERT_REPLY: 160 case DLM_MSG_UNLOCK_REPLY: 161 case DLM_MSG_CANCEL_REPLY: 162 case DLM_MSG_GRANT: 163 /* this reply is from the former master of the resource, 164 we'll resend to the new master if needed */ 165 166 if (dlm_hash2nodeid(ls, ms->m_hash) != nodeid) 167 return 1; 168 break; 169 } 170 171 return 0; 172 } 173 174 void dlm_purge_requestqueue(struct dlm_ls *ls) 175 { 176 struct dlm_message *ms; 177 struct rq_entry *e, *safe; 178 179 mutex_lock(&ls->ls_requestqueue_mutex); 180 list_for_each_entry_safe(e, safe, &ls->ls_requestqueue, list) { 181 ms = &e->request; 182 183 if (purge_request(ls, ms, e->nodeid)) { 184 list_del(&e->list); 185 kfree(e); 186 } 187 } 188 mutex_unlock(&ls->ls_requestqueue_mutex); 189 } 190 191