1 /******************************************************************************
2 *******************************************************************************
4 ** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
5 ** Copyright (C) 2005 Red Hat, Inc. All rights reserved.
7 ** This copyrighted material is made available to anyone wishing to use,
8 ** modify, copy, or redistribute it subject to the terms and conditions
9 ** of the GNU General Public License v.2.
11 *******************************************************************************
12 ******************************************************************************/
14 #include "dlm_internal.h"
15 #include "lockspace.h"
28 static int rcom_response(struct dlm_ls
*ls
)
30 return test_bit(LSFL_RCOM_READY
, &ls
->ls_flags
);
33 static int create_rcom(struct dlm_ls
*ls
, int to_nodeid
, int type
, int len
,
34 struct dlm_rcom
**rc_ret
, struct dlm_mhandle
**mh_ret
)
37 struct dlm_mhandle
*mh
;
39 int mb_len
= sizeof(struct dlm_rcom
) + len
;
41 mh
= dlm_lowcomms_get_buffer(to_nodeid
, mb_len
, GFP_KERNEL
, &mb
);
43 log_print("create_rcom to %d type %d len %d ENOBUFS",
44 to_nodeid
, type
, len
);
47 memset(mb
, 0, mb_len
);
49 rc
= (struct dlm_rcom
*) mb
;
51 rc
->rc_header
.h_version
= (DLM_HEADER_MAJOR
| DLM_HEADER_MINOR
);
52 rc
->rc_header
.h_lockspace
= ls
->ls_global_id
;
53 rc
->rc_header
.h_nodeid
= dlm_our_nodeid();
54 rc
->rc_header
.h_length
= mb_len
;
55 rc
->rc_header
.h_cmd
= DLM_RCOM
;
64 static void send_rcom(struct dlm_ls
*ls
, struct dlm_mhandle
*mh
,
68 dlm_lowcomms_commit_buffer(mh
);
71 /* When replying to a status request, a node also sends back its
72 configuration values. The requesting node then checks that the remote
73 node is configured the same way as itself. */
75 static void make_config(struct dlm_ls
*ls
, struct rcom_config
*rf
)
77 rf
->rf_lvblen
= ls
->ls_lvblen
;
78 rf
->rf_lsflags
= ls
->ls_exflags
;
81 static int check_config(struct dlm_ls
*ls
, struct rcom_config
*rf
, int nodeid
)
83 if (rf
->rf_lvblen
!= ls
->ls_lvblen
||
84 rf
->rf_lsflags
!= ls
->ls_exflags
) {
85 log_error(ls
, "config mismatch: %d,%x nodeid %d: %d,%x",
86 ls
->ls_lvblen
, ls
->ls_exflags
,
87 nodeid
, rf
->rf_lvblen
, rf
->rf_lsflags
);
93 static void allow_sync_reply(struct dlm_ls
*ls
, uint64_t *new_seq
)
95 spin_lock(&ls
->ls_rcom_spin
);
96 *new_seq
= ++ls
->ls_rcom_seq
;
97 set_bit(LSFL_RCOM_WAIT
, &ls
->ls_flags
);
98 spin_unlock(&ls
->ls_rcom_spin
);
101 static void disallow_sync_reply(struct dlm_ls
*ls
)
103 spin_lock(&ls
->ls_rcom_spin
);
104 clear_bit(LSFL_RCOM_WAIT
, &ls
->ls_flags
);
105 clear_bit(LSFL_RCOM_READY
, &ls
->ls_flags
);
106 spin_unlock(&ls
->ls_rcom_spin
);
109 int dlm_rcom_status(struct dlm_ls
*ls
, int nodeid
)
112 struct dlm_mhandle
*mh
;
115 ls
->ls_recover_nodeid
= nodeid
;
117 if (nodeid
== dlm_our_nodeid()) {
118 rc
= (struct dlm_rcom
*) ls
->ls_recover_buf
;
119 rc
->rc_result
= dlm_recover_status(ls
);
123 error
= create_rcom(ls
, nodeid
, DLM_RCOM_STATUS
, 0, &rc
, &mh
);
127 allow_sync_reply(ls
, &rc
->rc_id
);
128 memset(ls
->ls_recover_buf
, 0, dlm_config
.buffer_size
);
130 send_rcom(ls
, mh
, rc
);
132 error
= dlm_wait_function(ls
, &rcom_response
);
133 disallow_sync_reply(ls
);
137 rc
= (struct dlm_rcom
*) ls
->ls_recover_buf
;
139 if (rc
->rc_result
== -ESRCH
) {
140 /* we pretend the remote lockspace exists with 0 status */
141 log_debug(ls
, "remote node %d not ready", nodeid
);
144 error
= check_config(ls
, (struct rcom_config
*) rc
->rc_buf
,
146 /* the caller looks at rc_result for the remote recovery status */
151 static void receive_rcom_status(struct dlm_ls
*ls
, struct dlm_rcom
*rc_in
)
154 struct dlm_mhandle
*mh
;
155 int error
, nodeid
= rc_in
->rc_header
.h_nodeid
;
157 error
= create_rcom(ls
, nodeid
, DLM_RCOM_STATUS_REPLY
,
158 sizeof(struct rcom_config
), &rc
, &mh
);
161 rc
->rc_id
= rc_in
->rc_id
;
162 rc
->rc_result
= dlm_recover_status(ls
);
163 make_config(ls
, (struct rcom_config
*) rc
->rc_buf
);
165 send_rcom(ls
, mh
, rc
);
168 static void receive_sync_reply(struct dlm_ls
*ls
, struct dlm_rcom
*rc_in
)
170 spin_lock(&ls
->ls_rcom_spin
);
171 if (!test_bit(LSFL_RCOM_WAIT
, &ls
->ls_flags
) ||
172 rc_in
->rc_id
!= ls
->ls_rcom_seq
) {
173 log_debug(ls
, "reject reply %d from %d seq %llx expect %llx",
174 rc_in
->rc_type
, rc_in
->rc_header
.h_nodeid
,
175 (unsigned long long)rc_in
->rc_id
,
176 (unsigned long long)ls
->ls_rcom_seq
);
179 memcpy(ls
->ls_recover_buf
, rc_in
, rc_in
->rc_header
.h_length
);
180 set_bit(LSFL_RCOM_READY
, &ls
->ls_flags
);
181 clear_bit(LSFL_RCOM_WAIT
, &ls
->ls_flags
);
182 wake_up(&ls
->ls_wait_general
);
184 spin_unlock(&ls
->ls_rcom_spin
);
187 static void receive_rcom_status_reply(struct dlm_ls
*ls
, struct dlm_rcom
*rc_in
)
189 receive_sync_reply(ls
, rc_in
);
192 int dlm_rcom_names(struct dlm_ls
*ls
, int nodeid
, char *last_name
, int last_len
)
195 struct dlm_mhandle
*mh
;
196 int error
= 0, len
= sizeof(struct dlm_rcom
);
198 ls
->ls_recover_nodeid
= nodeid
;
200 if (nodeid
== dlm_our_nodeid()) {
201 dlm_copy_master_names(ls
, last_name
, last_len
,
202 ls
->ls_recover_buf
+ len
,
203 dlm_config
.buffer_size
- len
, nodeid
);
207 error
= create_rcom(ls
, nodeid
, DLM_RCOM_NAMES
, last_len
, &rc
, &mh
);
210 memcpy(rc
->rc_buf
, last_name
, last_len
);
212 allow_sync_reply(ls
, &rc
->rc_id
);
213 memset(ls
->ls_recover_buf
, 0, dlm_config
.buffer_size
);
215 send_rcom(ls
, mh
, rc
);
217 error
= dlm_wait_function(ls
, &rcom_response
);
218 disallow_sync_reply(ls
);
223 static void receive_rcom_names(struct dlm_ls
*ls
, struct dlm_rcom
*rc_in
)
226 struct dlm_mhandle
*mh
;
227 int error
, inlen
, outlen
;
228 int nodeid
= rc_in
->rc_header
.h_nodeid
;
229 uint32_t status
= dlm_recover_status(ls
);
232 * We can't run dlm_dir_rebuild_send (which uses ls_nodes) while
233 * dlm_recoverd is running ls_nodes_reconfig (which changes ls_nodes).
234 * It could only happen in rare cases where we get a late NAMES
235 * message from a previous instance of recovery.
238 if (!(status
& DLM_RS_NODES
)) {
239 log_debug(ls
, "ignoring RCOM_NAMES from %u", nodeid
);
243 nodeid
= rc_in
->rc_header
.h_nodeid
;
244 inlen
= rc_in
->rc_header
.h_length
- sizeof(struct dlm_rcom
);
245 outlen
= dlm_config
.buffer_size
- sizeof(struct dlm_rcom
);
247 error
= create_rcom(ls
, nodeid
, DLM_RCOM_NAMES_REPLY
, outlen
, &rc
, &mh
);
250 rc
->rc_id
= rc_in
->rc_id
;
252 dlm_copy_master_names(ls
, rc_in
->rc_buf
, inlen
, rc
->rc_buf
, outlen
,
254 send_rcom(ls
, mh
, rc
);
257 static void receive_rcom_names_reply(struct dlm_ls
*ls
, struct dlm_rcom
*rc_in
)
259 receive_sync_reply(ls
, rc_in
);
262 int dlm_send_rcom_lookup(struct dlm_rsb
*r
, int dir_nodeid
)
265 struct dlm_mhandle
*mh
;
266 struct dlm_ls
*ls
= r
->res_ls
;
269 error
= create_rcom(ls
, dir_nodeid
, DLM_RCOM_LOOKUP
, r
->res_length
,
273 memcpy(rc
->rc_buf
, r
->res_name
, r
->res_length
);
274 rc
->rc_id
= (unsigned long) r
;
276 send_rcom(ls
, mh
, rc
);
281 static void receive_rcom_lookup(struct dlm_ls
*ls
, struct dlm_rcom
*rc_in
)
284 struct dlm_mhandle
*mh
;
285 int error
, ret_nodeid
, nodeid
= rc_in
->rc_header
.h_nodeid
;
286 int len
= rc_in
->rc_header
.h_length
- sizeof(struct dlm_rcom
);
288 error
= create_rcom(ls
, nodeid
, DLM_RCOM_LOOKUP_REPLY
, 0, &rc
, &mh
);
292 error
= dlm_dir_lookup(ls
, nodeid
, rc_in
->rc_buf
, len
, &ret_nodeid
);
295 rc
->rc_result
= ret_nodeid
;
296 rc
->rc_id
= rc_in
->rc_id
;
298 send_rcom(ls
, mh
, rc
);
301 static void receive_rcom_lookup_reply(struct dlm_ls
*ls
, struct dlm_rcom
*rc_in
)
303 dlm_recover_master_reply(ls
, rc_in
);
306 static void pack_rcom_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
,
307 struct rcom_lock
*rl
)
309 memset(rl
, 0, sizeof(*rl
));
311 rl
->rl_ownpid
= lkb
->lkb_ownpid
;
312 rl
->rl_lkid
= lkb
->lkb_id
;
313 rl
->rl_exflags
= lkb
->lkb_exflags
;
314 rl
->rl_flags
= lkb
->lkb_flags
;
315 rl
->rl_lvbseq
= lkb
->lkb_lvbseq
;
316 rl
->rl_rqmode
= lkb
->lkb_rqmode
;
317 rl
->rl_grmode
= lkb
->lkb_grmode
;
318 rl
->rl_status
= lkb
->lkb_status
;
319 rl
->rl_wait_type
= lkb
->lkb_wait_type
;
321 if (lkb
->lkb_bastaddr
)
322 rl
->rl_asts
|= AST_BAST
;
323 if (lkb
->lkb_astaddr
)
324 rl
->rl_asts
|= AST_COMP
;
326 rl
->rl_namelen
= r
->res_length
;
327 memcpy(rl
->rl_name
, r
->res_name
, r
->res_length
);
329 /* FIXME: might we have an lvb without DLM_LKF_VALBLK set ?
330 If so, receive_rcom_lock_args() won't take this copy. */
333 memcpy(rl
->rl_lvb
, lkb
->lkb_lvbptr
, r
->res_ls
->ls_lvblen
);
336 int dlm_send_rcom_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
338 struct dlm_ls
*ls
= r
->res_ls
;
340 struct dlm_mhandle
*mh
;
341 struct rcom_lock
*rl
;
342 int error
, len
= sizeof(struct rcom_lock
);
345 len
+= ls
->ls_lvblen
;
347 error
= create_rcom(ls
, r
->res_nodeid
, DLM_RCOM_LOCK
, len
, &rc
, &mh
);
351 rl
= (struct rcom_lock
*) rc
->rc_buf
;
352 pack_rcom_lock(r
, lkb
, rl
);
353 rc
->rc_id
= (unsigned long) r
;
355 send_rcom(ls
, mh
, rc
);
360 static void receive_rcom_lock(struct dlm_ls
*ls
, struct dlm_rcom
*rc_in
)
363 struct dlm_mhandle
*mh
;
364 int error
, nodeid
= rc_in
->rc_header
.h_nodeid
;
366 dlm_recover_master_copy(ls
, rc_in
);
368 error
= create_rcom(ls
, nodeid
, DLM_RCOM_LOCK_REPLY
,
369 sizeof(struct rcom_lock
), &rc
, &mh
);
373 /* We send back the same rcom_lock struct we received, but
374 dlm_recover_master_copy() has filled in rl_remid and rl_result */
376 memcpy(rc
->rc_buf
, rc_in
->rc_buf
, sizeof(struct rcom_lock
));
377 rc
->rc_id
= rc_in
->rc_id
;
379 send_rcom(ls
, mh
, rc
);
382 static void receive_rcom_lock_reply(struct dlm_ls
*ls
, struct dlm_rcom
*rc_in
)
384 uint32_t status
= dlm_recover_status(ls
);
386 if (!(status
& DLM_RS_DIR
)) {
387 log_debug(ls
, "ignoring RCOM_LOCK_REPLY from %u",
388 rc_in
->rc_header
.h_nodeid
);
392 dlm_recover_process_copy(ls
, rc_in
);
395 static int send_ls_not_ready(int nodeid
, struct dlm_rcom
*rc_in
)
398 struct rcom_config
*rf
;
399 struct dlm_mhandle
*mh
;
401 int mb_len
= sizeof(struct dlm_rcom
) + sizeof(struct rcom_config
);
403 mh
= dlm_lowcomms_get_buffer(nodeid
, mb_len
, GFP_KERNEL
, &mb
);
406 memset(mb
, 0, mb_len
);
408 rc
= (struct dlm_rcom
*) mb
;
410 rc
->rc_header
.h_version
= (DLM_HEADER_MAJOR
| DLM_HEADER_MINOR
);
411 rc
->rc_header
.h_lockspace
= rc_in
->rc_header
.h_lockspace
;
412 rc
->rc_header
.h_nodeid
= dlm_our_nodeid();
413 rc
->rc_header
.h_length
= mb_len
;
414 rc
->rc_header
.h_cmd
= DLM_RCOM
;
416 rc
->rc_type
= DLM_RCOM_STATUS_REPLY
;
417 rc
->rc_id
= rc_in
->rc_id
;
418 rc
->rc_result
= -ESRCH
;
420 rf
= (struct rcom_config
*) rc
->rc_buf
;
424 dlm_lowcomms_commit_buffer(mh
);
429 /* Called by dlm_recvd; corresponds to dlm_receive_message() but special
430 recovery-only comms are sent through here. */
432 void dlm_receive_rcom(struct dlm_header
*hd
, int nodeid
)
434 struct dlm_rcom
*rc
= (struct dlm_rcom
*) hd
;
439 /* If the lockspace doesn't exist then still send a status message
440 back; it's possible that it just doesn't have its global_id yet. */
442 ls
= dlm_find_lockspace_global(hd
->h_lockspace
);
444 log_print("lockspace %x from %d type %x not found",
445 hd
->h_lockspace
, nodeid
, rc
->rc_type
);
446 if (rc
->rc_type
== DLM_RCOM_STATUS
)
447 send_ls_not_ready(nodeid
, rc
);
451 if (dlm_recovery_stopped(ls
) && (rc
->rc_type
!= DLM_RCOM_STATUS
)) {
452 log_error(ls
, "ignoring recovery message %x from %d",
453 rc
->rc_type
, nodeid
);
457 if (nodeid
!= rc
->rc_header
.h_nodeid
) {
458 log_error(ls
, "bad rcom nodeid %d from %d",
459 rc
->rc_header
.h_nodeid
, nodeid
);
463 switch (rc
->rc_type
) {
464 case DLM_RCOM_STATUS
:
465 receive_rcom_status(ls
, rc
);
469 receive_rcom_names(ls
, rc
);
472 case DLM_RCOM_LOOKUP
:
473 receive_rcom_lookup(ls
, rc
);
477 receive_rcom_lock(ls
, rc
);
480 case DLM_RCOM_STATUS_REPLY
:
481 receive_rcom_status_reply(ls
, rc
);
484 case DLM_RCOM_NAMES_REPLY
:
485 receive_rcom_names_reply(ls
, rc
);
488 case DLM_RCOM_LOOKUP_REPLY
:
489 receive_rcom_lookup_reply(ls
, rc
);
492 case DLM_RCOM_LOCK_REPLY
:
493 receive_rcom_lock_reply(ls
, rc
);
497 DLM_ASSERT(0, printk("rc_type=%x\n", rc
->rc_type
););
500 dlm_put_lockspace(ls
);