1 /******************************************************************************
2 *******************************************************************************
4 ** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
5 ** Copyright (C) 2005-2008 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
, ls
->ls_allocation
, &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
;
59 spin_lock(&ls
->ls_recover_lock
);
60 rc
->rc_seq
= ls
->ls_recover_seq
;
61 spin_unlock(&ls
->ls_recover_lock
);
68 static void send_rcom(struct dlm_ls
*ls
, struct dlm_mhandle
*mh
,
72 dlm_lowcomms_commit_buffer(mh
);
75 /* When replying to a status request, a node also sends back its
76 configuration values. The requesting node then checks that the remote
77 node is configured the same way as itself. */
79 static void make_config(struct dlm_ls
*ls
, struct rcom_config
*rf
)
81 rf
->rf_lvblen
= cpu_to_le32(ls
->ls_lvblen
);
82 rf
->rf_lsflags
= cpu_to_le32(ls
->ls_exflags
);
85 static int check_config(struct dlm_ls
*ls
, struct dlm_rcom
*rc
, int nodeid
)
87 struct rcom_config
*rf
= (struct rcom_config
*) rc
->rc_buf
;
89 if ((rc
->rc_header
.h_version
& 0xFFFF0000) != DLM_HEADER_MAJOR
) {
90 log_error(ls
, "version mismatch: %x nodeid %d: %x",
91 DLM_HEADER_MAJOR
| DLM_HEADER_MINOR
, nodeid
,
92 rc
->rc_header
.h_version
);
96 if (le32_to_cpu(rf
->rf_lvblen
) != ls
->ls_lvblen
||
97 le32_to_cpu(rf
->rf_lsflags
) != ls
->ls_exflags
) {
98 log_error(ls
, "config mismatch: %d,%x nodeid %d: %d,%x",
99 ls
->ls_lvblen
, ls
->ls_exflags
, nodeid
,
100 le32_to_cpu(rf
->rf_lvblen
),
101 le32_to_cpu(rf
->rf_lsflags
));
107 static void allow_sync_reply(struct dlm_ls
*ls
, uint64_t *new_seq
)
109 spin_lock(&ls
->ls_rcom_spin
);
110 *new_seq
= ++ls
->ls_rcom_seq
;
111 set_bit(LSFL_RCOM_WAIT
, &ls
->ls_flags
);
112 spin_unlock(&ls
->ls_rcom_spin
);
115 static void disallow_sync_reply(struct dlm_ls
*ls
)
117 spin_lock(&ls
->ls_rcom_spin
);
118 clear_bit(LSFL_RCOM_WAIT
, &ls
->ls_flags
);
119 clear_bit(LSFL_RCOM_READY
, &ls
->ls_flags
);
120 spin_unlock(&ls
->ls_rcom_spin
);
123 int dlm_rcom_status(struct dlm_ls
*ls
, int nodeid
)
126 struct dlm_mhandle
*mh
;
129 ls
->ls_recover_nodeid
= nodeid
;
131 if (nodeid
== dlm_our_nodeid()) {
132 rc
= (struct dlm_rcom
*) ls
->ls_recover_buf
;
133 rc
->rc_result
= dlm_recover_status(ls
);
137 error
= create_rcom(ls
, nodeid
, DLM_RCOM_STATUS
, 0, &rc
, &mh
);
141 allow_sync_reply(ls
, &rc
->rc_id
);
142 memset(ls
->ls_recover_buf
, 0, dlm_config
.ci_buffer_size
);
144 send_rcom(ls
, mh
, rc
);
146 error
= dlm_wait_function(ls
, &rcom_response
);
147 disallow_sync_reply(ls
);
151 rc
= (struct dlm_rcom
*) ls
->ls_recover_buf
;
153 if (rc
->rc_result
== -ESRCH
) {
154 /* we pretend the remote lockspace exists with 0 status */
155 log_debug(ls
, "remote node %d not ready", nodeid
);
158 error
= check_config(ls
, rc
, nodeid
);
159 /* the caller looks at rc_result for the remote recovery status */
164 static void receive_rcom_status(struct dlm_ls
*ls
, struct dlm_rcom
*rc_in
)
167 struct dlm_mhandle
*mh
;
168 int error
, nodeid
= rc_in
->rc_header
.h_nodeid
;
170 error
= create_rcom(ls
, nodeid
, DLM_RCOM_STATUS_REPLY
,
171 sizeof(struct rcom_config
), &rc
, &mh
);
174 rc
->rc_id
= rc_in
->rc_id
;
175 rc
->rc_seq_reply
= rc_in
->rc_seq
;
176 rc
->rc_result
= dlm_recover_status(ls
);
177 make_config(ls
, (struct rcom_config
*) rc
->rc_buf
);
179 send_rcom(ls
, mh
, rc
);
182 static void receive_sync_reply(struct dlm_ls
*ls
, struct dlm_rcom
*rc_in
)
184 spin_lock(&ls
->ls_rcom_spin
);
185 if (!test_bit(LSFL_RCOM_WAIT
, &ls
->ls_flags
) ||
186 rc_in
->rc_id
!= ls
->ls_rcom_seq
) {
187 log_debug(ls
, "reject reply %d from %d seq %llx expect %llx",
188 rc_in
->rc_type
, rc_in
->rc_header
.h_nodeid
,
189 (unsigned long long)rc_in
->rc_id
,
190 (unsigned long long)ls
->ls_rcom_seq
);
193 memcpy(ls
->ls_recover_buf
, rc_in
, rc_in
->rc_header
.h_length
);
194 set_bit(LSFL_RCOM_READY
, &ls
->ls_flags
);
195 clear_bit(LSFL_RCOM_WAIT
, &ls
->ls_flags
);
196 wake_up(&ls
->ls_wait_general
);
198 spin_unlock(&ls
->ls_rcom_spin
);
201 int dlm_rcom_names(struct dlm_ls
*ls
, int nodeid
, char *last_name
, int last_len
)
204 struct dlm_mhandle
*mh
;
205 int error
= 0, len
= sizeof(struct dlm_rcom
);
207 ls
->ls_recover_nodeid
= nodeid
;
209 if (nodeid
== dlm_our_nodeid()) {
210 dlm_copy_master_names(ls
, last_name
, last_len
,
211 ls
->ls_recover_buf
+ len
,
212 dlm_config
.ci_buffer_size
- len
, nodeid
);
216 error
= create_rcom(ls
, nodeid
, DLM_RCOM_NAMES
, last_len
, &rc
, &mh
);
219 memcpy(rc
->rc_buf
, last_name
, last_len
);
221 allow_sync_reply(ls
, &rc
->rc_id
);
222 memset(ls
->ls_recover_buf
, 0, dlm_config
.ci_buffer_size
);
224 send_rcom(ls
, mh
, rc
);
226 error
= dlm_wait_function(ls
, &rcom_response
);
227 disallow_sync_reply(ls
);
232 static void receive_rcom_names(struct dlm_ls
*ls
, struct dlm_rcom
*rc_in
)
235 struct dlm_mhandle
*mh
;
236 int error
, inlen
, outlen
, nodeid
;
238 nodeid
= rc_in
->rc_header
.h_nodeid
;
239 inlen
= rc_in
->rc_header
.h_length
- sizeof(struct dlm_rcom
);
240 outlen
= dlm_config
.ci_buffer_size
- sizeof(struct dlm_rcom
);
242 error
= create_rcom(ls
, nodeid
, DLM_RCOM_NAMES_REPLY
, outlen
, &rc
, &mh
);
245 rc
->rc_id
= rc_in
->rc_id
;
246 rc
->rc_seq_reply
= rc_in
->rc_seq
;
248 dlm_copy_master_names(ls
, rc_in
->rc_buf
, inlen
, rc
->rc_buf
, outlen
,
250 send_rcom(ls
, mh
, rc
);
253 int dlm_send_rcom_lookup(struct dlm_rsb
*r
, int dir_nodeid
)
256 struct dlm_mhandle
*mh
;
257 struct dlm_ls
*ls
= r
->res_ls
;
260 error
= create_rcom(ls
, dir_nodeid
, DLM_RCOM_LOOKUP
, r
->res_length
,
264 memcpy(rc
->rc_buf
, r
->res_name
, r
->res_length
);
265 rc
->rc_id
= (unsigned long) r
;
267 send_rcom(ls
, mh
, rc
);
272 static void receive_rcom_lookup(struct dlm_ls
*ls
, struct dlm_rcom
*rc_in
)
275 struct dlm_mhandle
*mh
;
276 int error
, ret_nodeid
, nodeid
= rc_in
->rc_header
.h_nodeid
;
277 int len
= rc_in
->rc_header
.h_length
- sizeof(struct dlm_rcom
);
279 error
= create_rcom(ls
, nodeid
, DLM_RCOM_LOOKUP_REPLY
, 0, &rc
, &mh
);
283 error
= dlm_dir_lookup(ls
, nodeid
, rc_in
->rc_buf
, len
, &ret_nodeid
);
286 rc
->rc_result
= ret_nodeid
;
287 rc
->rc_id
= rc_in
->rc_id
;
288 rc
->rc_seq_reply
= rc_in
->rc_seq
;
290 send_rcom(ls
, mh
, rc
);
293 static void receive_rcom_lookup_reply(struct dlm_ls
*ls
, struct dlm_rcom
*rc_in
)
295 dlm_recover_master_reply(ls
, rc_in
);
298 static void pack_rcom_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
,
299 struct rcom_lock
*rl
)
301 memset(rl
, 0, sizeof(*rl
));
303 rl
->rl_ownpid
= cpu_to_le32(lkb
->lkb_ownpid
);
304 rl
->rl_lkid
= cpu_to_le32(lkb
->lkb_id
);
305 rl
->rl_exflags
= cpu_to_le32(lkb
->lkb_exflags
);
306 rl
->rl_flags
= cpu_to_le32(lkb
->lkb_flags
);
307 rl
->rl_lvbseq
= cpu_to_le32(lkb
->lkb_lvbseq
);
308 rl
->rl_rqmode
= lkb
->lkb_rqmode
;
309 rl
->rl_grmode
= lkb
->lkb_grmode
;
310 rl
->rl_status
= lkb
->lkb_status
;
311 rl
->rl_wait_type
= cpu_to_le16(lkb
->lkb_wait_type
);
313 if (lkb
->lkb_bastaddr
)
314 rl
->rl_asts
|= AST_BAST
;
315 if (lkb
->lkb_astaddr
)
316 rl
->rl_asts
|= AST_COMP
;
318 rl
->rl_namelen
= cpu_to_le16(r
->res_length
);
319 memcpy(rl
->rl_name
, r
->res_name
, r
->res_length
);
321 /* FIXME: might we have an lvb without DLM_LKF_VALBLK set ?
322 If so, receive_rcom_lock_args() won't take this copy. */
325 memcpy(rl
->rl_lvb
, lkb
->lkb_lvbptr
, r
->res_ls
->ls_lvblen
);
328 int dlm_send_rcom_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
330 struct dlm_ls
*ls
= r
->res_ls
;
332 struct dlm_mhandle
*mh
;
333 struct rcom_lock
*rl
;
334 int error
, len
= sizeof(struct rcom_lock
);
337 len
+= ls
->ls_lvblen
;
339 error
= create_rcom(ls
, r
->res_nodeid
, DLM_RCOM_LOCK
, len
, &rc
, &mh
);
343 rl
= (struct rcom_lock
*) rc
->rc_buf
;
344 pack_rcom_lock(r
, lkb
, rl
);
345 rc
->rc_id
= (unsigned long) r
;
347 send_rcom(ls
, mh
, rc
);
352 static void receive_rcom_lock(struct dlm_ls
*ls
, struct dlm_rcom
*rc_in
)
355 struct dlm_mhandle
*mh
;
356 int error
, nodeid
= rc_in
->rc_header
.h_nodeid
;
358 dlm_recover_master_copy(ls
, rc_in
);
360 error
= create_rcom(ls
, nodeid
, DLM_RCOM_LOCK_REPLY
,
361 sizeof(struct rcom_lock
), &rc
, &mh
);
365 /* We send back the same rcom_lock struct we received, but
366 dlm_recover_master_copy() has filled in rl_remid and rl_result */
368 memcpy(rc
->rc_buf
, rc_in
->rc_buf
, sizeof(struct rcom_lock
));
369 rc
->rc_id
= rc_in
->rc_id
;
370 rc
->rc_seq_reply
= rc_in
->rc_seq
;
372 send_rcom(ls
, mh
, rc
);
375 /* If the lockspace doesn't exist then still send a status message
376 back; it's possible that it just doesn't have its global_id yet. */
378 int dlm_send_ls_not_ready(int nodeid
, struct dlm_rcom
*rc_in
)
381 struct rcom_config
*rf
;
382 struct dlm_mhandle
*mh
;
384 int mb_len
= sizeof(struct dlm_rcom
) + sizeof(struct rcom_config
);
386 mh
= dlm_lowcomms_get_buffer(nodeid
, mb_len
, GFP_NOFS
, &mb
);
389 memset(mb
, 0, mb_len
);
391 rc
= (struct dlm_rcom
*) mb
;
393 rc
->rc_header
.h_version
= (DLM_HEADER_MAJOR
| DLM_HEADER_MINOR
);
394 rc
->rc_header
.h_lockspace
= rc_in
->rc_header
.h_lockspace
;
395 rc
->rc_header
.h_nodeid
= dlm_our_nodeid();
396 rc
->rc_header
.h_length
= mb_len
;
397 rc
->rc_header
.h_cmd
= DLM_RCOM
;
399 rc
->rc_type
= DLM_RCOM_STATUS_REPLY
;
400 rc
->rc_id
= rc_in
->rc_id
;
401 rc
->rc_seq_reply
= rc_in
->rc_seq
;
402 rc
->rc_result
= -ESRCH
;
404 rf
= (struct rcom_config
*) rc
->rc_buf
;
405 rf
->rf_lvblen
= cpu_to_le32(~0U);
408 dlm_lowcomms_commit_buffer(mh
);
413 static int is_old_reply(struct dlm_ls
*ls
, struct dlm_rcom
*rc
)
418 switch (rc
->rc_type
) {
419 case DLM_RCOM_STATUS_REPLY
:
420 case DLM_RCOM_NAMES_REPLY
:
421 case DLM_RCOM_LOOKUP_REPLY
:
422 case DLM_RCOM_LOCK_REPLY
:
423 spin_lock(&ls
->ls_recover_lock
);
424 seq
= ls
->ls_recover_seq
;
425 spin_unlock(&ls
->ls_recover_lock
);
426 if (rc
->rc_seq_reply
!= seq
) {
427 log_debug(ls
, "ignoring old reply %x from %d "
428 "seq_reply %llx expect %llx",
429 rc
->rc_type
, rc
->rc_header
.h_nodeid
,
430 (unsigned long long)rc
->rc_seq_reply
,
431 (unsigned long long)seq
);
438 /* Called by dlm_recv; corresponds to dlm_receive_message() but special
439 recovery-only comms are sent through here. */
441 void dlm_receive_rcom(struct dlm_ls
*ls
, struct dlm_rcom
*rc
, int nodeid
)
443 if (dlm_recovery_stopped(ls
) && (rc
->rc_type
!= DLM_RCOM_STATUS
)) {
444 log_debug(ls
, "ignoring recovery message %x from %d",
445 rc
->rc_type
, nodeid
);
449 if (is_old_reply(ls
, rc
))
452 switch (rc
->rc_type
) {
453 case DLM_RCOM_STATUS
:
454 receive_rcom_status(ls
, rc
);
458 receive_rcom_names(ls
, rc
);
461 case DLM_RCOM_LOOKUP
:
462 receive_rcom_lookup(ls
, rc
);
466 receive_rcom_lock(ls
, rc
);
469 case DLM_RCOM_STATUS_REPLY
:
470 receive_sync_reply(ls
, rc
);
473 case DLM_RCOM_NAMES_REPLY
:
474 receive_sync_reply(ls
, rc
);
477 case DLM_RCOM_LOOKUP_REPLY
:
478 receive_rcom_lookup_reply(ls
, rc
);
481 case DLM_RCOM_LOCK_REPLY
:
482 dlm_recover_process_copy(ls
, rc
);
486 log_error(ls
, "receive_rcom bad type %d", rc
->rc_type
);