projects / linux-2.6-block.git / blame
? search:
summary | shortlog | log | commit | commitdiff | tree
blob | blame (incremental) | history | HEAD
md-cluster: use %pU to print UUIDs
[linux-2.6-block.git] / drivers / md / md-cluster.c
CommitLineData
8e854e9c
GR		 1/*
2 * Copyright (C) 2015, SUSE
3 *
4 * This program is free software; you can redistribute it and/or modify
5 * it under the terms of the GNU General Public License as published by
6 * the Free Software Foundation; either version 2, or (at your option)
7 * any later version.
8 *
9 */
10
11
12#include <linux/module.h>
47741b7c
GR		 13#include <linux/dlm.h>
14#include <linux/sched.h>
1aee41f6 15#include <linux/raid/md_p.h>
47741b7c 16#include "md.h"
e94987db 17#include "bitmap.h"
edb39c9d 18#include "md-cluster.h"
47741b7c
GR		 19
20#define LVB_SIZE 64
1aee41f6 21#define NEW_DEV_TIMEOUT 5000
47741b7c
GR		 22
23struct dlm_lock_resource {
24 dlm_lockspace_t *ls;
25 struct dlm_lksb lksb;
26 char *name; /* lock name. */
27 uint32_t flags; /* flags to pass to dlm_lock() */
47741b7c 28 struct completion completion; /* completion for synchronized locking */
c4ce867f
GR		 29 void (*bast)(void *arg, int mode); /* blocking AST function pointer*/
30 struct mddev *mddev; /* pointing back to mddev. */
31};
32
96ae923a
GR		 33struct suspend_info {
34 int slot;
35 sector_t lo;
36 sector_t hi;
37 struct list_head list;
38};
39
40struct resync_info {
41 __le64 lo;
42 __le64 hi;
43};
44
fa8259da
GR		 45/* md_cluster_info flags */
46#define MD_CLUSTER_WAITING_FOR_NEWDISK 1
90382ed9 47#define MD_CLUSTER_SUSPEND_READ_BALANCING 2
fa8259da
GR		 48
49
c4ce867f
GR		 50struct md_cluster_info {
51 /* dlm lock space and resources for clustered raid. */
52 dlm_lockspace_t *lockspace;
cf921cc1
GR		 53 int slot_number;
54 struct completion completion;
c4ce867f
GR		 55 struct dlm_lock_resource *sb_lock;
56 struct mutex sb_mutex;
54519c5f 57 struct dlm_lock_resource *bitmap_lockres;
96ae923a
GR		 58 struct list_head suspend_list;
59 spinlock_t suspend_lock;
e94987db
GR		 60 struct md_thread *recovery_thread;
61 unsigned long recovery_map;
4664680c
GR		 62 /* communication loc resources */
63 struct dlm_lock_resource *ack_lockres;
64 struct dlm_lock_resource *message_lockres;
65 struct dlm_lock_resource *token_lockres;
1aee41f6 66 struct dlm_lock_resource *no_new_dev_lockres;
4664680c 67 struct md_thread *recv_thread;
1aee41f6 68 struct completion newdisk_completion;
fa8259da 69 unsigned long state;
4664680c
GR		 70};
71
72enum msg_type {
73 METADATA_UPDATED = 0,
74 RESYNCING,
1aee41f6 75 NEWDISK,
88bcfef7 76 REMOVE,
97f6cd39 77 RE_ADD,
4664680c
GR		 78};
79
80struct cluster_msg {
81 int type;
82 int slot;
1aee41f6 83 /* TODO: Unionize this for smaller footprint */
4664680c
GR		 84 sector_t low;
85 sector_t high;
1aee41f6
GR		 86 char uuid[16];
87 int raid_slot;
47741b7c
GR		 88};
89
90static void sync_ast(void *arg)
91{
92 struct dlm_lock_resource *res;
93
94 res = (struct dlm_lock_resource *) arg;
95 complete(&res->completion);
96}
97
98static int dlm_lock_sync(struct dlm_lock_resource *res, int mode)
99{
100 int ret = 0;
101
102 init_completion(&res->completion);
103 ret = dlm_lock(res->ls, mode, &res->lksb,
104 res->flags, res->name, strlen(res->name),
105 0, sync_ast, res, res->bast);
106 if (ret)
107 return ret;
108 wait_for_completion(&res->completion);
109 return res->lksb.sb_status;
110}
111
112static int dlm_unlock_sync(struct dlm_lock_resource *res)
113{
114 return dlm_lock_sync(res, DLM_LOCK_NL);
115}
116
c4ce867f 117static struct dlm_lock_resource *lockres_init(struct mddev *mddev,
47741b7c
GR		 118 char *name, void (*bastfn)(void *arg, int mode), int with_lvb)
119{
120 struct dlm_lock_resource *res = NULL;
121 int ret, namelen;
c4ce867f 122 struct md_cluster_info *cinfo = mddev->cluster_info;
47741b7c
GR		 123
124 res = kzalloc(sizeof(struct dlm_lock_resource), GFP_KERNEL);
125 if (!res)
126 return NULL;
c4ce867f
GR		 127 res->ls = cinfo->lockspace;
128 res->mddev = mddev;
47741b7c
GR		 129 namelen = strlen(name);
130 res->name = kzalloc(namelen + 1, GFP_KERNEL);
131 if (!res->name) {
132 pr_err("md-cluster: Unable to allocate resource name for resource %s\n", name);
133 goto out_err;
134 }
135 strlcpy(res->name, name, namelen + 1);
136 if (with_lvb) {
137 res->lksb.sb_lvbptr = kzalloc(LVB_SIZE, GFP_KERNEL);
138 if (!res->lksb.sb_lvbptr) {
139 pr_err("md-cluster: Unable to allocate LVB for resource %s\n", name);
140 goto out_err;
141 }
142 res->flags = DLM_LKF_VALBLK;
143 }
144
145 if (bastfn)
146 res->bast = bastfn;
147
148 res->flags |= DLM_LKF_EXPEDITE;
149
150 ret = dlm_lock_sync(res, DLM_LOCK_NL);
151 if (ret) {
152 pr_err("md-cluster: Unable to lock NL on new lock resource %s\n", name);
153 goto out_err;
154 }
155 res->flags &= ~DLM_LKF_EXPEDITE;
156 res->flags |= DLM_LKF_CONVERT;
157
158 return res;
159out_err:
160 kfree(res->lksb.sb_lvbptr);
161 kfree(res->name);
162 kfree(res);
163 return NULL;
164}
165
166static void lockres_free(struct dlm_lock_resource *res)
167{
168 if (!res)
169 return;
170
171 init_completion(&res->completion);
172 dlm_unlock(res->ls, res->lksb.sb_lkid, 0, &res->lksb, res);
173 wait_for_completion(&res->completion);
174
175 kfree(res->name);
176 kfree(res->lksb.sb_lvbptr);
177 kfree(res);
178}
8e854e9c 179
96ae923a
GR		 180static void add_resync_info(struct mddev *mddev, struct dlm_lock_resource *lockres,
181 sector_t lo, sector_t hi)
182{
183 struct resync_info *ri;
184
185 ri = (struct resync_info *)lockres->lksb.sb_lvbptr;
186 ri->lo = cpu_to_le64(lo);
187 ri->hi = cpu_to_le64(hi);
188}
189
190static struct suspend_info *read_resync_info(struct mddev *mddev, struct dlm_lock_resource *lockres)
191{
192 struct resync_info ri;
193 struct suspend_info *s = NULL;
194 sector_t hi = 0;
195
196 dlm_lock_sync(lockres, DLM_LOCK_CR);
197 memcpy(&ri, lockres->lksb.sb_lvbptr, sizeof(struct resync_info));
198 hi = le64_to_cpu(ri.hi);
199 if (ri.hi > 0) {
200 s = kzalloc(sizeof(struct suspend_info), GFP_KERNEL);
201 if (!s)
202 goto out;
203 s->hi = hi;
204 s->lo = le64_to_cpu(ri.lo);
205 }
206 dlm_unlock_sync(lockres);
207out:
208 return s;
209}
210
6dc69c9c 211static void recover_bitmaps(struct md_thread *thread)
e94987db
GR		 212{
213 struct mddev *mddev = thread->mddev;
214 struct md_cluster_info *cinfo = mddev->cluster_info;
215 struct dlm_lock_resource *bm_lockres;
216 char str[64];
217 int slot, ret;
218 struct suspend_info *s, *tmp;
219 sector_t lo, hi;
220
221 while (cinfo->recovery_map) {
222 slot = fls64((u64)cinfo->recovery_map) - 1;
223
224 /* Clear suspend_area associated with the bitmap */
225 spin_lock_irq(&cinfo->suspend_lock);
226 list_for_each_entry_safe(s, tmp, &cinfo->suspend_list, list)
227 if (slot == s->slot) {
228 list_del(&s->list);
229 kfree(s);
230 }
231 spin_unlock_irq(&cinfo->suspend_lock);
232
233 snprintf(str, 64, "bitmap%04d", slot);
234 bm_lockres = lockres_init(mddev, str, NULL, 1);
235 if (!bm_lockres) {
236 pr_err("md-cluster: Cannot initialize bitmaps\n");
237 goto clear_bit;
238 }
239
240 ret = dlm_lock_sync(bm_lockres, DLM_LOCK_PW);
241 if (ret) {
242 pr_err("md-cluster: Could not DLM lock %s: %d\n",
243 str, ret);
244 goto clear_bit;
245 }
97f6cd39 246 ret = bitmap_copy_from_slot(mddev, slot, &lo, &hi, true);
4b26a08a 247 if (ret) {
e94987db 248 pr_err("md-cluster: Could not copy data from bitmap %d\n", slot);
4b26a08a
GR		 249 goto dlm_unlock;
250 }
251 if (hi > 0) {
252 /* TODO:Wait for current resync to get over */
253 set_bit(MD_RECOVERY_NEEDED, &mddev->recovery);
254 if (lo < mddev->recovery_cp)
255 mddev->recovery_cp = lo;
256 md_check_recovery(mddev);
257 }
258dlm_unlock:
e94987db
GR		 259 dlm_unlock_sync(bm_lockres);
260clear_bit:
261 clear_bit(slot, &cinfo->recovery_map);
262 }
263}
264
cf921cc1
GR		 265static void recover_prep(void *arg)
266{
90382ed9
GR		 267 struct mddev *mddev = arg;
268 struct md_cluster_info *cinfo = mddev->cluster_info;
269 set_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state);
cf921cc1
GR		 270}
271
272static void recover_slot(void *arg, struct dlm_slot *slot)
273{
274 struct mddev *mddev = arg;
275 struct md_cluster_info *cinfo = mddev->cluster_info;
276
277 pr_info("md-cluster: %s Node %d/%d down. My slot: %d. Initiating recovery.\n",
278 mddev->bitmap_info.cluster_name,
279 slot->nodeid, slot->slot,
280 cinfo->slot_number);
e94987db
GR		 281 set_bit(slot->slot - 1, &cinfo->recovery_map);
282 if (!cinfo->recovery_thread) {
283 cinfo->recovery_thread = md_register_thread(recover_bitmaps,
284 mddev, "recover");
285 if (!cinfo->recovery_thread) {
286 pr_warn("md-cluster: Could not create recovery thread\n");
287 return;
288 }
289 }
290 md_wakeup_thread(cinfo->recovery_thread);
cf921cc1
GR		 291}
292
293static void recover_done(void *arg, struct dlm_slot *slots,
294 int num_slots, int our_slot,
295 uint32_t generation)
296{
297 struct mddev *mddev = arg;
298 struct md_cluster_info *cinfo = mddev->cluster_info;
299
300 cinfo->slot_number = our_slot;
301 complete(&cinfo->completion);
90382ed9 302 clear_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state);
cf921cc1
GR		 303}
304
305static const struct dlm_lockspace_ops md_ls_ops = {
306 .recover_prep = recover_prep,
307 .recover_slot = recover_slot,
308 .recover_done = recover_done,
309};
310
4664680c
GR		 311/*
312 * The BAST function for the ack lock resource
313 * This function wakes up the receive thread in
314 * order to receive and process the message.
315 */
316static void ack_bast(void *arg, int mode)
317{
318 struct dlm_lock_resource *res = (struct dlm_lock_resource *)arg;
319 struct md_cluster_info *cinfo = res->mddev->cluster_info;
320
321 if (mode == DLM_LOCK_EX)
322 md_wakeup_thread(cinfo->recv_thread);
323}
324
e59721cc
GR		 325static void __remove_suspend_info(struct md_cluster_info *cinfo, int slot)
326{
327 struct suspend_info *s, *tmp;
328
329 list_for_each_entry_safe(s, tmp, &cinfo->suspend_list, list)
330 if (slot == s->slot) {
331 pr_info("%s:%d Deleting suspend_info: %d\n",
332 __func__, __LINE__, slot);
333 list_del(&s->list);
334 kfree(s);
335 break;
336 }
337}
338
339static void remove_suspend_info(struct md_cluster_info *cinfo, int slot)
340{
341 spin_lock_irq(&cinfo->suspend_lock);
342 __remove_suspend_info(cinfo, slot);
343 spin_unlock_irq(&cinfo->suspend_lock);
344}
345
346
347static void process_suspend_info(struct md_cluster_info *cinfo,
348 int slot, sector_t lo, sector_t hi)
349{
350 struct suspend_info *s;
351
352 if (!hi) {
353 remove_suspend_info(cinfo, slot);
354 return;
355 }
356 s = kzalloc(sizeof(struct suspend_info), GFP_KERNEL);
357 if (!s)
358 return;
359 s->slot = slot;
360 s->lo = lo;
361 s->hi = hi;
362 spin_lock_irq(&cinfo->suspend_lock);
363 /* Remove existing entry (if exists) before adding */
364 __remove_suspend_info(cinfo, slot);
365 list_add(&s->list, &cinfo->suspend_list);
366 spin_unlock_irq(&cinfo->suspend_lock);
367}
368
1aee41f6
GR		 369static void process_add_new_disk(struct mddev *mddev, struct cluster_msg *cmsg)
370{
371 char disk_uuid[64];
372 struct md_cluster_info *cinfo = mddev->cluster_info;
373 char event_name[] = "EVENT=ADD_DEVICE";
374 char raid_slot[16];
375 char *envp[] = {event_name, disk_uuid, raid_slot, NULL};
376 int len;
377
378 len = snprintf(disk_uuid, 64, "DEVICE_UUID=");
b89f704a 379 sprintf(disk_uuid + len, "%pU", cmsg->uuid);
1aee41f6
GR		 380 snprintf(raid_slot, 16, "RAID_DISK=%d", cmsg->raid_slot);
381 pr_info("%s:%d Sending kobject change with %s and %s\n", __func__, __LINE__, disk_uuid, raid_slot);
382 init_completion(&cinfo->newdisk_completion);
fa8259da 383 set_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state);
1aee41f6
GR		 384 kobject_uevent_env(&disk_to_dev(mddev->gendisk)->kobj, KOBJ_CHANGE, envp);
385 wait_for_completion_timeout(&cinfo->newdisk_completion,
386 NEW_DEV_TIMEOUT);
fa8259da 387 clear_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state);
1aee41f6
GR		 388}
389
390
391static void process_metadata_update(struct mddev *mddev, struct cluster_msg *msg)
392{
393 struct md_cluster_info *cinfo = mddev->cluster_info;
394
395 md_reload_sb(mddev);
396 dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR);
397}
398
88bcfef7
GR		 399static void process_remove_disk(struct mddev *mddev, struct cluster_msg *msg)
400{
401 struct md_rdev *rdev = md_find_rdev_nr_rcu(mddev, msg->raid_slot);
402
403 if (rdev)
404 md_kick_rdev_from_array(rdev);
405 else
406 pr_warn("%s: %d Could not find disk(%d) to REMOVE\n", __func__, __LINE__, msg->raid_slot);
407}
408
97f6cd39
GR		 409static void process_readd_disk(struct mddev *mddev, struct cluster_msg *msg)
410{
411 struct md_rdev *rdev = md_find_rdev_nr_rcu(mddev, msg->raid_slot);
412
413 if (rdev && test_bit(Faulty, &rdev->flags))
414 clear_bit(Faulty, &rdev->flags);
415 else
416 pr_warn("%s: %d Could not find disk(%d) which is faulty", __func__, __LINE__, msg->raid_slot);
417}
418
4664680c
GR		 419static void process_recvd_msg(struct mddev *mddev, struct cluster_msg *msg)
420{
421 switch (msg->type) {
422 case METADATA_UPDATED:
423 pr_info("%s: %d Received message: METADATA_UPDATE from %d\n",
424 __func__, __LINE__, msg->slot);
1aee41f6 425 process_metadata_update(mddev, msg);
4664680c
GR		 426 break;
427 case RESYNCING:
428 pr_info("%s: %d Received message: RESYNCING from %d\n",
429 __func__, __LINE__, msg->slot);
e59721cc
GR		 430 process_suspend_info(mddev->cluster_info, msg->slot,
431 msg->low, msg->high);
4664680c 432 break;
1aee41f6
GR		 433 case NEWDISK:
434 pr_info("%s: %d Received message: NEWDISK from %d\n",
435 __func__, __LINE__, msg->slot);
436 process_add_new_disk(mddev, msg);
88bcfef7
GR		 437 break;
438 case REMOVE:
439 pr_info("%s: %d Received REMOVE from %d\n",
440 __func__, __LINE__, msg->slot);
441 process_remove_disk(mddev, msg);
442 break;
97f6cd39
GR		 443 case RE_ADD:
444 pr_info("%s: %d Received RE_ADD from %d\n",
445 __func__, __LINE__, msg->slot);
446 process_readd_disk(mddev, msg);
447 break;
88bcfef7
GR		 448 default:
449 pr_warn("%s:%d Received unknown message from %d\n",
450 __func__, __LINE__, msg->slot);
09dd1af2 451 }
4664680c
GR		 452}
453
454/*
455 * thread for receiving message
456 */
457static void recv_daemon(struct md_thread *thread)
458{
459 struct md_cluster_info *cinfo = thread->mddev->cluster_info;
460 struct dlm_lock_resource *ack_lockres = cinfo->ack_lockres;
461 struct dlm_lock_resource *message_lockres = cinfo->message_lockres;
462 struct cluster_msg msg;
463
464 /*get CR on Message*/
465 if (dlm_lock_sync(message_lockres, DLM_LOCK_CR)) {
466 pr_err("md/raid1:failed to get CR on MESSAGE\n");
467 return;
468 }
469
470 /* read lvb and wake up thread to process this message_lockres */
471 memcpy(&msg, message_lockres->lksb.sb_lvbptr, sizeof(struct cluster_msg));
472 process_recvd_msg(thread->mddev, &msg);
473
474 /*release CR on ack_lockres*/
475 dlm_unlock_sync(ack_lockres);
476 /*up-convert to EX on message_lockres*/
477 dlm_lock_sync(message_lockres, DLM_LOCK_EX);
478 /*get CR on ack_lockres again*/
479 dlm_lock_sync(ack_lockres, DLM_LOCK_CR);
480 /*release CR on message_lockres*/
481 dlm_unlock_sync(message_lockres);
482}
483
601b515c
GR		 484/* lock_comm()
485 * Takes the lock on the TOKEN lock resource so no other
486 * node can communicate while the operation is underway.
487 */
488static int lock_comm(struct md_cluster_info *cinfo)
489{
490 int error;
491
492 error = dlm_lock_sync(cinfo->token_lockres, DLM_LOCK_EX);
493 if (error)
494 pr_err("md-cluster(%s:%d): failed to get EX on TOKEN (%d)\n",
495 __func__, __LINE__, error);
496 return error;
497}
498
499static void unlock_comm(struct md_cluster_info *cinfo)
500{
501 dlm_unlock_sync(cinfo->token_lockres);
502}
503
504/* __sendmsg()
505 * This function performs the actual sending of the message. This function is
506 * usually called after performing the encompassing operation
507 * The function:
508 * 1. Grabs the message lockresource in EX mode
509 * 2. Copies the message to the message LVB
510 * 3. Downconverts message lockresource to CR
511 * 4. Upconverts ack lock resource from CR to EX. This forces the BAST on other nodes
512 * and the other nodes read the message. The thread will wait here until all other
513 * nodes have released ack lock resource.
514 * 5. Downconvert ack lockresource to CR
515 */
516static int __sendmsg(struct md_cluster_info *cinfo, struct cluster_msg *cmsg)
517{
518 int error;
519 int slot = cinfo->slot_number - 1;
520
521 cmsg->slot = cpu_to_le32(slot);
522 /*get EX on Message*/
523 error = dlm_lock_sync(cinfo->message_lockres, DLM_LOCK_EX);
524 if (error) {
525 pr_err("md-cluster: failed to get EX on MESSAGE (%d)\n", error);
526 goto failed_message;
527 }
528
529 memcpy(cinfo->message_lockres->lksb.sb_lvbptr, (void *)cmsg,
530 sizeof(struct cluster_msg));
531 /*down-convert EX to CR on Message*/
532 error = dlm_lock_sync(cinfo->message_lockres, DLM_LOCK_CR);
533 if (error) {
534 pr_err("md-cluster: failed to convert EX to CR on MESSAGE(%d)\n",
535 error);
536 goto failed_message;
537 }
538
539 /*up-convert CR to EX on Ack*/
540 error = dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_EX);
541 if (error) {
542 pr_err("md-cluster: failed to convert CR to EX on ACK(%d)\n",
543 error);
544 goto failed_ack;
545 }
546
547 /*down-convert EX to CR on Ack*/
548 error = dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_CR);
549 if (error) {
550 pr_err("md-cluster: failed to convert EX to CR on ACK(%d)\n",
551 error);
552 goto failed_ack;
553 }
554
555failed_ack:
556 dlm_unlock_sync(cinfo->message_lockres);
557failed_message:
558 return error;
559}
560
561static int sendmsg(struct md_cluster_info *cinfo, struct cluster_msg *cmsg)
562{
563 int ret;
564
565 lock_comm(cinfo);
566 ret = __sendmsg(cinfo, cmsg);
567 unlock_comm(cinfo);
568 return ret;
569}
570
96ae923a
GR		 571static int gather_all_resync_info(struct mddev *mddev, int total_slots)
572{
573 struct md_cluster_info *cinfo = mddev->cluster_info;
574 int i, ret = 0;
575 struct dlm_lock_resource *bm_lockres;
576 struct suspend_info *s;
577 char str[64];
578
579
580 for (i = 0; i < total_slots; i++) {
581 memset(str, '\0', 64);
582 snprintf(str, 64, "bitmap%04d", i);
583 bm_lockres = lockres_init(mddev, str, NULL, 1);
584 if (!bm_lockres)
585 return -ENOMEM;
586 if (i == (cinfo->slot_number - 1))
587 continue;
588
589 bm_lockres->flags |= DLM_LKF_NOQUEUE;
590 ret = dlm_lock_sync(bm_lockres, DLM_LOCK_PW);
591 if (ret == -EAGAIN) {
592 memset(bm_lockres->lksb.sb_lvbptr, '\0', LVB_SIZE);
593 s = read_resync_info(mddev, bm_lockres);
594 if (s) {
595 pr_info("%s:%d Resync[%llu..%llu] in progress on %d\n",
596 __func__, __LINE__,
597 (unsigned long long) s->lo,
598 (unsigned long long) s->hi, i);
599 spin_lock_irq(&cinfo->suspend_lock);
600 s->slot = i;
601 list_add(&s->list, &cinfo->suspend_list);
602 spin_unlock_irq(&cinfo->suspend_lock);
603 }
604 ret = 0;
605 lockres_free(bm_lockres);
606 continue;
607 }
608 if (ret)
609 goto out;
610 /* TODO: Read the disk bitmap sb and check if it needs recovery */
611 dlm_unlock_sync(bm_lockres);
612 lockres_free(bm_lockres);
613 }
614out:
615 return ret;
616}
617
edb39c9d
GR		 618static int join(struct mddev *mddev, int nodes)
619{
c4ce867f 620 struct md_cluster_info *cinfo;
cf921cc1 621 int ret, ops_rv;
c4ce867f
GR		 622 char str[64];
623
624 if (!try_module_get(THIS_MODULE))
625 return -ENOENT;
626
627 cinfo = kzalloc(sizeof(struct md_cluster_info), GFP_KERNEL);
628 if (!cinfo)
629 return -ENOMEM;
630
cf921cc1
GR		 631 init_completion(&cinfo->completion);
632
633 mutex_init(&cinfo->sb_mutex);
634 mddev->cluster_info = cinfo;
635
c4ce867f 636 memset(str, 0, 64);
b89f704a 637 sprintf(str, "%pU", mddev->uuid);
cf921cc1
GR		 638 ret = dlm_new_lockspace(str, mddev->bitmap_info.cluster_name,
639 DLM_LSFL_FS, LVB_SIZE,
640 &md_ls_ops, mddev, &ops_rv, &cinfo->lockspace);
c4ce867f
GR		 641 if (ret)
642 goto err;
cf921cc1 643 wait_for_completion(&cinfo->completion);
8c58f02e
GJ		 644 if (nodes < cinfo->slot_number) {
645 pr_err("md-cluster: Slot allotted(%d) is greater than available slots(%d).",
646 cinfo->slot_number, nodes);
b97e9257
GR		 647 ret = -ERANGE;
648 goto err;
649 }
c4ce867f
GR		 650 cinfo->sb_lock = lockres_init(mddev, "cmd-super",
651 NULL, 0);
652 if (!cinfo->sb_lock) {
653 ret = -ENOMEM;
654 goto err;
655 }
4664680c
GR		 656 /* Initiate the communication resources */
657 ret = -ENOMEM;
658 cinfo->recv_thread = md_register_thread(recv_daemon, mddev, "cluster_recv");
659 if (!cinfo->recv_thread) {
660 pr_err("md-cluster: cannot allocate memory for recv_thread!\n");
661 goto err;
662 }
663 cinfo->message_lockres = lockres_init(mddev, "message", NULL, 1);
664 if (!cinfo->message_lockres)
665 goto err;
666 cinfo->token_lockres = lockres_init(mddev, "token", NULL, 0);
667 if (!cinfo->token_lockres)
668 goto err;
669 cinfo->ack_lockres = lockres_init(mddev, "ack", ack_bast, 0);
670 if (!cinfo->ack_lockres)
671 goto err;
1aee41f6
GR		 672 cinfo->no_new_dev_lockres = lockres_init(mddev, "no-new-dev", NULL, 0);
673 if (!cinfo->no_new_dev_lockres)
674 goto err;
675
4664680c
GR		 676 /* get sync CR lock on ACK. */
677 if (dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_CR))
678 pr_err("md-cluster: failed to get a sync CR lock on ACK!(%d)\n",
679 ret);
1aee41f6
GR		 680 /* get sync CR lock on no-new-dev. */
681 if (dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR))
682 pr_err("md-cluster: failed to get a sync CR lock on no-new-dev!(%d)\n", ret);
683
54519c5f
GR		 684
685 pr_info("md-cluster: Joined cluster %s slot %d\n", str, cinfo->slot_number);
686 snprintf(str, 64, "bitmap%04d", cinfo->slot_number - 1);
687 cinfo->bitmap_lockres = lockres_init(mddev, str, NULL, 1);
688 if (!cinfo->bitmap_lockres)
689 goto err;
690 if (dlm_lock_sync(cinfo->bitmap_lockres, DLM_LOCK_PW)) {
691 pr_err("Failed to get bitmap lock\n");
692 ret = -EINVAL;
693 goto err;
694 }
695
96ae923a
GR		 696 INIT_LIST_HEAD(&cinfo->suspend_list);
697 spin_lock_init(&cinfo->suspend_lock);
698
699 ret = gather_all_resync_info(mddev, nodes);
700 if (ret)
701 goto err;
702
edb39c9d 703 return 0;
c4ce867f 704err:
4664680c
GR		 705 lockres_free(cinfo->message_lockres);
706 lockres_free(cinfo->token_lockres);
707 lockres_free(cinfo->ack_lockres);
1aee41f6 708 lockres_free(cinfo->no_new_dev_lockres);
96ae923a
GR		 709 lockres_free(cinfo->bitmap_lockres);
710 lockres_free(cinfo->sb_lock);
c4ce867f
GR		 711 if (cinfo->lockspace)
712 dlm_release_lockspace(cinfo->lockspace, 2);
cf921cc1 713 mddev->cluster_info = NULL;
c4ce867f
GR		 714 kfree(cinfo);
715 module_put(THIS_MODULE);
716 return ret;
edb39c9d
GR		 717}
718
719static int leave(struct mddev *mddev)
720{
c4ce867f
GR		 721 struct md_cluster_info *cinfo = mddev->cluster_info;
722
723 if (!cinfo)
724 return 0;
e94987db 725 md_unregister_thread(&cinfo->recovery_thread);
4664680c
GR		 726 md_unregister_thread(&cinfo->recv_thread);
727 lockres_free(cinfo->message_lockres);
728 lockres_free(cinfo->token_lockres);
729 lockres_free(cinfo->ack_lockres);
1aee41f6 730 lockres_free(cinfo->no_new_dev_lockres);
c4ce867f 731 lockres_free(cinfo->sb_lock);
54519c5f 732 lockres_free(cinfo->bitmap_lockres);
c4ce867f 733 dlm_release_lockspace(cinfo->lockspace, 2);
edb39c9d
GR		 734 return 0;
735}
736
cf921cc1
GR		 737/* slot_number(): Returns the MD slot number to use
738 * DLM starts the slot numbers from 1, wheras cluster-md
739 * wants the number to be from zero, so we deduct one
740 */
741static int slot_number(struct mddev *mddev)
742{
743 struct md_cluster_info *cinfo = mddev->cluster_info;
744
745 return cinfo->slot_number - 1;
746}
747
96ae923a
GR		 748static void resync_info_update(struct mddev *mddev, sector_t lo, sector_t hi)
749{
750 struct md_cluster_info *cinfo = mddev->cluster_info;
751
752 add_resync_info(mddev, cinfo->bitmap_lockres, lo, hi);
753 /* Re-acquire the lock to refresh LVB */
754 dlm_lock_sync(cinfo->bitmap_lockres, DLM_LOCK_PW);
755}
756
293467aa
GR		 757static int metadata_update_start(struct mddev *mddev)
758{
759 return lock_comm(mddev->cluster_info);
760}
761
762static int metadata_update_finish(struct mddev *mddev)
763{
764 struct md_cluster_info *cinfo = mddev->cluster_info;
765 struct cluster_msg cmsg;
766 int ret;
767
768 memset(&cmsg, 0, sizeof(cmsg));
769 cmsg.type = cpu_to_le32(METADATA_UPDATED);
770 ret = __sendmsg(cinfo, &cmsg);
771 unlock_comm(cinfo);
772 return ret;
773}
774
775static int metadata_update_cancel(struct mddev *mddev)
776{
777 struct md_cluster_info *cinfo = mddev->cluster_info;
778
779 return dlm_unlock_sync(cinfo->token_lockres);
780}
781
965400eb
GR		 782static int resync_send(struct mddev *mddev, enum msg_type type,
783 sector_t lo, sector_t hi)
784{
785 struct md_cluster_info *cinfo = mddev->cluster_info;
786 struct cluster_msg cmsg;
787 int slot = cinfo->slot_number - 1;
788
789 pr_info("%s:%d lo: %llu hi: %llu\n", __func__, __LINE__,
790 (unsigned long long)lo,
791 (unsigned long long)hi);
792 resync_info_update(mddev, lo, hi);
793 cmsg.type = cpu_to_le32(type);
794 cmsg.slot = cpu_to_le32(slot);
795 cmsg.low = cpu_to_le64(lo);
796 cmsg.high = cpu_to_le64(hi);
797 return sendmsg(cinfo, &cmsg);
798}
799
800static int resync_start(struct mddev *mddev, sector_t lo, sector_t hi)
801{
802 pr_info("%s:%d\n", __func__, __LINE__);
803 return resync_send(mddev, RESYNCING, lo, hi);
804}
805
806static void resync_finish(struct mddev *mddev)
807{
808 pr_info("%s:%d\n", __func__, __LINE__);
809 resync_send(mddev, RESYNCING, 0, 0);
810}
811
90382ed9
GR		 812static int area_resyncing(struct mddev *mddev, int direction,
813 sector_t lo, sector_t hi)
589a1c49
GR		 814{
815 struct md_cluster_info *cinfo = mddev->cluster_info;
816 int ret = 0;
817 struct suspend_info *s;
818
90382ed9
GR		 819 if ((direction == READ) &&
820 test_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state))
821 return 1;
822
589a1c49
GR		 823 spin_lock_irq(&cinfo->suspend_lock);
824 if (list_empty(&cinfo->suspend_list))
825 goto out;
826 list_for_each_entry(s, &cinfo->suspend_list, list)
827 if (hi > s->lo && lo < s->hi) {
828 ret = 1;
829 break;
830 }
831out:
832 spin_unlock_irq(&cinfo->suspend_lock);
833 return ret;
834}
835
1aee41f6
GR		 836static int add_new_disk_start(struct mddev *mddev, struct md_rdev *rdev)
837{
838 struct md_cluster_info *cinfo = mddev->cluster_info;
839 struct cluster_msg cmsg;
840 int ret = 0;
841 struct mdp_superblock_1 *sb = page_address(rdev->sb_page);
842 char *uuid = sb->device_uuid;
843
844 memset(&cmsg, 0, sizeof(cmsg));
845 cmsg.type = cpu_to_le32(NEWDISK);
846 memcpy(cmsg.uuid, uuid, 16);
847 cmsg.raid_slot = rdev->desc_nr;
848 lock_comm(cinfo);
849 ret = __sendmsg(cinfo, &cmsg);
850 if (ret)
851 return ret;
852 cinfo->no_new_dev_lockres->flags |= DLM_LKF_NOQUEUE;
853 ret = dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_EX);
854 cinfo->no_new_dev_lockres->flags &= ~DLM_LKF_NOQUEUE;
855 /* Some node does not "see" the device */
856 if (ret == -EAGAIN)
857 ret = -ENOENT;
858 else
859 dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR);
860 return ret;
861}
862
863static int add_new_disk_finish(struct mddev *mddev)
864{
865 struct cluster_msg cmsg;
866 struct md_cluster_info *cinfo = mddev->cluster_info;
867 int ret;
868 /* Write sb and inform others */
869 md_update_sb(mddev, 1);
870 cmsg.type = METADATA_UPDATED;
871 ret = __sendmsg(cinfo, &cmsg);
872 unlock_comm(cinfo);
873 return ret;
874}
875
fa8259da 876static int new_disk_ack(struct mddev *mddev, bool ack)
1aee41f6
GR		 877{
878 struct md_cluster_info *cinfo = mddev->cluster_info;
879
fa8259da
GR		 880 if (!test_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state)) {
881 pr_warn("md-cluster(%s): Spurious cluster confirmation\n", mdname(mddev));
882 return -EINVAL;
883 }
884
1aee41f6
GR		 885 if (ack)
886 dlm_unlock_sync(cinfo->no_new_dev_lockres);
887 complete(&cinfo->newdisk_completion);
fa8259da 888 return 0;
1aee41f6
GR		 889}
890
88bcfef7
GR		 891static int remove_disk(struct mddev *mddev, struct md_rdev *rdev)
892{
893 struct cluster_msg cmsg;
894 struct md_cluster_info *cinfo = mddev->cluster_info;
895 cmsg.type = REMOVE;
896 cmsg.raid_slot = rdev->desc_nr;
897 return __sendmsg(cinfo, &cmsg);
898}
899
97f6cd39
GR		 900static int gather_bitmaps(struct md_rdev *rdev)
901{
902 int sn, err;
903 sector_t lo, hi;
904 struct cluster_msg cmsg;
905 struct mddev *mddev = rdev->mddev;
906 struct md_cluster_info *cinfo = mddev->cluster_info;
907
908 cmsg.type = RE_ADD;
909 cmsg.raid_slot = rdev->desc_nr;
910 err = sendmsg(cinfo, &cmsg);
911 if (err)
912 goto out;
913
914 for (sn = 0; sn < mddev->bitmap_info.nodes; sn++) {
915 if (sn == (cinfo->slot_number - 1))
916 continue;
917 err = bitmap_copy_from_slot(mddev, sn, &lo, &hi, false);
918 if (err) {
919 pr_warn("md-cluster: Could not gather bitmaps from slot %d", sn);
920 goto out;
921 }
922 if ((hi > 0) && (lo < mddev->recovery_cp))
923 mddev->recovery_cp = lo;
924 }
925out:
926 return err;
927}
928
edb39c9d
GR		 929static struct md_cluster_operations cluster_ops = {
930 .join = join,
931 .leave = leave,
cf921cc1 932 .slot_number = slot_number,
96ae923a 933 .resync_info_update = resync_info_update,
965400eb
GR		 934 .resync_start = resync_start,
935 .resync_finish = resync_finish,
293467aa
GR		 936 .metadata_update_start = metadata_update_start,
937 .metadata_update_finish = metadata_update_finish,
938 .metadata_update_cancel = metadata_update_cancel,
589a1c49 939 .area_resyncing = area_resyncing,
1aee41f6
GR		 940 .add_new_disk_start = add_new_disk_start,
941 .add_new_disk_finish = add_new_disk_finish,
942 .new_disk_ack = new_disk_ack,
88bcfef7 943 .remove_disk = remove_disk,
97f6cd39 944 .gather_bitmaps = gather_bitmaps,
edb39c9d
GR		 945};
946
8e854e9c
GR		 947static int __init cluster_init(void)
948{
949 pr_warn("md-cluster: EXPERIMENTAL. Use with caution\n");
950 pr_info("Registering Cluster MD functions\n");
edb39c9d 951 register_md_cluster_operations(&cluster_ops, THIS_MODULE);
8e854e9c
GR		 952 return 0;
953}
954
955static void cluster_exit(void)
956{
edb39c9d 957 unregister_md_cluster_operations();
8e854e9c
GR		 958}
959
960module_init(cluster_init);
961module_exit(cluster_exit);
962MODULE_LICENSE("GPL");
963MODULE_DESCRIPTION("Clustering support for MD");
Linux 6.x block layer and io_uring tree(s)