md-cluster.c 34 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258
  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. #include <linux/module.h>
  11. #include <linux/dlm.h>
  12. #include <linux/sched.h>
  13. #include <linux/raid/md_p.h>
  14. #include "md.h"
  15. #include "bitmap.h"
  16. #include "md-cluster.h"
  17. #define LVB_SIZE 64
  18. #define NEW_DEV_TIMEOUT 5000
  19. struct dlm_lock_resource {
  20. dlm_lockspace_t *ls;
  21. struct dlm_lksb lksb;
  22. char *name; /* lock name. */
  23. uint32_t flags; /* flags to pass to dlm_lock() */
  24. struct completion completion; /* completion for synchronized locking */
  25. void (*bast)(void *arg, int mode); /* blocking AST function pointer*/
  26. struct mddev *mddev; /* pointing back to mddev. */
  27. int mode;
  28. };
  29. struct suspend_info {
  30. int slot;
  31. sector_t lo;
  32. sector_t hi;
  33. struct list_head list;
  34. };
  35. struct resync_info {
  36. __le64 lo;
  37. __le64 hi;
  38. };
  39. /* md_cluster_info flags */
  40. #define MD_CLUSTER_WAITING_FOR_NEWDISK 1
  41. #define MD_CLUSTER_SUSPEND_READ_BALANCING 2
  42. #define MD_CLUSTER_BEGIN_JOIN_CLUSTER 3
  43. /* Lock the send communication. This is done through
  44. * bit manipulation as opposed to a mutex in order to
  45. * accomodate lock and hold. See next comment.
  46. */
  47. #define MD_CLUSTER_SEND_LOCK 4
  48. /* If cluster operations (such as adding a disk) must lock the
  49. * communication channel, so as to perform extra operations
  50. * (update metadata) and no other operation is allowed on the
  51. * MD. Token needs to be locked and held until the operation
  52. * completes witha md_update_sb(), which would eventually release
  53. * the lock.
  54. */
  55. #define MD_CLUSTER_SEND_LOCKED_ALREADY 5
  56. /* We should receive message after node joined cluster and
  57. * set up all the related infos such as bitmap and personality */
  58. #define MD_CLUSTER_ALREADY_IN_CLUSTER 6
  59. #define MD_CLUSTER_PENDING_RECV_EVENT 7
  60. struct md_cluster_info {
  61. /* dlm lock space and resources for clustered raid. */
  62. dlm_lockspace_t *lockspace;
  63. int slot_number;
  64. struct completion completion;
  65. struct mutex recv_mutex;
  66. struct dlm_lock_resource *bitmap_lockres;
  67. struct dlm_lock_resource **other_bitmap_lockres;
  68. struct dlm_lock_resource *resync_lockres;
  69. struct list_head suspend_list;
  70. spinlock_t suspend_lock;
  71. struct md_thread *recovery_thread;
  72. unsigned long recovery_map;
  73. /* communication loc resources */
  74. struct dlm_lock_resource *ack_lockres;
  75. struct dlm_lock_resource *message_lockres;
  76. struct dlm_lock_resource *token_lockres;
  77. struct dlm_lock_resource *no_new_dev_lockres;
  78. struct md_thread *recv_thread;
  79. struct completion newdisk_completion;
  80. wait_queue_head_t wait;
  81. unsigned long state;
  82. /* record the region in RESYNCING message */
  83. sector_t sync_low;
  84. sector_t sync_hi;
  85. };
  86. enum msg_type {
  87. METADATA_UPDATED = 0,
  88. RESYNCING,
  89. NEWDISK,
  90. REMOVE,
  91. RE_ADD,
  92. BITMAP_NEEDS_SYNC,
  93. };
  94. struct cluster_msg {
  95. __le32 type;
  96. __le32 slot;
  97. /* TODO: Unionize this for smaller footprint */
  98. __le64 low;
  99. __le64 high;
  100. char uuid[16];
  101. __le32 raid_slot;
  102. };
  103. static void sync_ast(void *arg)
  104. {
  105. struct dlm_lock_resource *res;
  106. res = arg;
  107. complete(&res->completion);
  108. }
  109. static int dlm_lock_sync(struct dlm_lock_resource *res, int mode)
  110. {
  111. int ret = 0;
  112. ret = dlm_lock(res->ls, mode, &res->lksb,
  113. res->flags, res->name, strlen(res->name),
  114. 0, sync_ast, res, res->bast);
  115. if (ret)
  116. return ret;
  117. wait_for_completion(&res->completion);
  118. if (res->lksb.sb_status == 0)
  119. res->mode = mode;
  120. return res->lksb.sb_status;
  121. }
  122. static int dlm_unlock_sync(struct dlm_lock_resource *res)
  123. {
  124. return dlm_lock_sync(res, DLM_LOCK_NL);
  125. }
  126. static struct dlm_lock_resource *lockres_init(struct mddev *mddev,
  127. char *name, void (*bastfn)(void *arg, int mode), int with_lvb)
  128. {
  129. struct dlm_lock_resource *res = NULL;
  130. int ret, namelen;
  131. struct md_cluster_info *cinfo = mddev->cluster_info;
  132. res = kzalloc(sizeof(struct dlm_lock_resource), GFP_KERNEL);
  133. if (!res)
  134. return NULL;
  135. init_completion(&res->completion);
  136. res->ls = cinfo->lockspace;
  137. res->mddev = mddev;
  138. res->mode = DLM_LOCK_IV;
  139. namelen = strlen(name);
  140. res->name = kzalloc(namelen + 1, GFP_KERNEL);
  141. if (!res->name) {
  142. pr_err("md-cluster: Unable to allocate resource name for resource %s\n", name);
  143. goto out_err;
  144. }
  145. strlcpy(res->name, name, namelen + 1);
  146. if (with_lvb) {
  147. res->lksb.sb_lvbptr = kzalloc(LVB_SIZE, GFP_KERNEL);
  148. if (!res->lksb.sb_lvbptr) {
  149. pr_err("md-cluster: Unable to allocate LVB for resource %s\n", name);
  150. goto out_err;
  151. }
  152. res->flags = DLM_LKF_VALBLK;
  153. }
  154. if (bastfn)
  155. res->bast = bastfn;
  156. res->flags |= DLM_LKF_EXPEDITE;
  157. ret = dlm_lock_sync(res, DLM_LOCK_NL);
  158. if (ret) {
  159. pr_err("md-cluster: Unable to lock NL on new lock resource %s\n", name);
  160. goto out_err;
  161. }
  162. res->flags &= ~DLM_LKF_EXPEDITE;
  163. res->flags |= DLM_LKF_CONVERT;
  164. return res;
  165. out_err:
  166. kfree(res->lksb.sb_lvbptr);
  167. kfree(res->name);
  168. kfree(res);
  169. return NULL;
  170. }
  171. static void lockres_free(struct dlm_lock_resource *res)
  172. {
  173. int ret;
  174. if (!res)
  175. return;
  176. /* cancel a lock request or a conversion request that is blocked */
  177. res->flags |= DLM_LKF_CANCEL;
  178. retry:
  179. ret = dlm_unlock(res->ls, res->lksb.sb_lkid, 0, &res->lksb, res);
  180. if (unlikely(ret != 0)) {
  181. pr_info("%s: failed to unlock %s return %d\n", __func__, res->name, ret);
  182. /* if a lock conversion is cancelled, then the lock is put
  183. * back to grant queue, need to ensure it is unlocked */
  184. if (ret == -DLM_ECANCEL)
  185. goto retry;
  186. }
  187. res->flags &= ~DLM_LKF_CANCEL;
  188. wait_for_completion(&res->completion);
  189. kfree(res->name);
  190. kfree(res->lksb.sb_lvbptr);
  191. kfree(res);
  192. }
  193. static void add_resync_info(struct dlm_lock_resource *lockres,
  194. sector_t lo, sector_t hi)
  195. {
  196. struct resync_info *ri;
  197. ri = (struct resync_info *)lockres->lksb.sb_lvbptr;
  198. ri->lo = cpu_to_le64(lo);
  199. ri->hi = cpu_to_le64(hi);
  200. }
  201. static struct suspend_info *read_resync_info(struct mddev *mddev, struct dlm_lock_resource *lockres)
  202. {
  203. struct resync_info ri;
  204. struct suspend_info *s = NULL;
  205. sector_t hi = 0;
  206. dlm_lock_sync(lockres, DLM_LOCK_CR);
  207. memcpy(&ri, lockres->lksb.sb_lvbptr, sizeof(struct resync_info));
  208. hi = le64_to_cpu(ri.hi);
  209. if (hi > 0) {
  210. s = kzalloc(sizeof(struct suspend_info), GFP_KERNEL);
  211. if (!s)
  212. goto out;
  213. s->hi = hi;
  214. s->lo = le64_to_cpu(ri.lo);
  215. }
  216. dlm_unlock_sync(lockres);
  217. out:
  218. return s;
  219. }
  220. static void recover_bitmaps(struct md_thread *thread)
  221. {
  222. struct mddev *mddev = thread->mddev;
  223. struct md_cluster_info *cinfo = mddev->cluster_info;
  224. struct dlm_lock_resource *bm_lockres;
  225. char str[64];
  226. int slot, ret;
  227. struct suspend_info *s, *tmp;
  228. sector_t lo, hi;
  229. while (cinfo->recovery_map) {
  230. slot = fls64((u64)cinfo->recovery_map) - 1;
  231. /* Clear suspend_area associated with the bitmap */
  232. spin_lock_irq(&cinfo->suspend_lock);
  233. list_for_each_entry_safe(s, tmp, &cinfo->suspend_list, list)
  234. if (slot == s->slot) {
  235. list_del(&s->list);
  236. kfree(s);
  237. }
  238. spin_unlock_irq(&cinfo->suspend_lock);
  239. snprintf(str, 64, "bitmap%04d", slot);
  240. bm_lockres = lockres_init(mddev, str, NULL, 1);
  241. if (!bm_lockres) {
  242. pr_err("md-cluster: Cannot initialize bitmaps\n");
  243. goto clear_bit;
  244. }
  245. ret = dlm_lock_sync(bm_lockres, DLM_LOCK_PW);
  246. if (ret) {
  247. pr_err("md-cluster: Could not DLM lock %s: %d\n",
  248. str, ret);
  249. goto clear_bit;
  250. }
  251. ret = bitmap_copy_from_slot(mddev, slot, &lo, &hi, true);
  252. if (ret) {
  253. pr_err("md-cluster: Could not copy data from bitmap %d\n", slot);
  254. goto dlm_unlock;
  255. }
  256. if (hi > 0) {
  257. if (lo < mddev->recovery_cp)
  258. mddev->recovery_cp = lo;
  259. /* wake up thread to continue resync in case resync
  260. * is not finished */
  261. if (mddev->recovery_cp != MaxSector) {
  262. set_bit(MD_RECOVERY_NEEDED, &mddev->recovery);
  263. md_wakeup_thread(mddev->thread);
  264. }
  265. }
  266. dlm_unlock:
  267. dlm_unlock_sync(bm_lockres);
  268. clear_bit:
  269. lockres_free(bm_lockres);
  270. clear_bit(slot, &cinfo->recovery_map);
  271. }
  272. }
  273. static void recover_prep(void *arg)
  274. {
  275. struct mddev *mddev = arg;
  276. struct md_cluster_info *cinfo = mddev->cluster_info;
  277. set_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state);
  278. }
  279. static void __recover_slot(struct mddev *mddev, int slot)
  280. {
  281. struct md_cluster_info *cinfo = mddev->cluster_info;
  282. set_bit(slot, &cinfo->recovery_map);
  283. if (!cinfo->recovery_thread) {
  284. cinfo->recovery_thread = md_register_thread(recover_bitmaps,
  285. mddev, "recover");
  286. if (!cinfo->recovery_thread) {
  287. pr_warn("md-cluster: Could not create recovery thread\n");
  288. return;
  289. }
  290. }
  291. md_wakeup_thread(cinfo->recovery_thread);
  292. }
  293. static void recover_slot(void *arg, struct dlm_slot *slot)
  294. {
  295. struct mddev *mddev = arg;
  296. struct md_cluster_info *cinfo = mddev->cluster_info;
  297. pr_info("md-cluster: %s Node %d/%d down. My slot: %d. Initiating recovery.\n",
  298. mddev->bitmap_info.cluster_name,
  299. slot->nodeid, slot->slot,
  300. cinfo->slot_number);
  301. /* deduct one since dlm slot starts from one while the num of
  302. * cluster-md begins with 0 */
  303. __recover_slot(mddev, slot->slot - 1);
  304. }
  305. static void recover_done(void *arg, struct dlm_slot *slots,
  306. int num_slots, int our_slot,
  307. uint32_t generation)
  308. {
  309. struct mddev *mddev = arg;
  310. struct md_cluster_info *cinfo = mddev->cluster_info;
  311. cinfo->slot_number = our_slot;
  312. /* completion is only need to be complete when node join cluster,
  313. * it doesn't need to run during another node's failure */
  314. if (test_bit(MD_CLUSTER_BEGIN_JOIN_CLUSTER, &cinfo->state)) {
  315. complete(&cinfo->completion);
  316. clear_bit(MD_CLUSTER_BEGIN_JOIN_CLUSTER, &cinfo->state);
  317. }
  318. clear_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state);
  319. }
  320. /* the ops is called when node join the cluster, and do lock recovery
  321. * if node failure occurs */
  322. static const struct dlm_lockspace_ops md_ls_ops = {
  323. .recover_prep = recover_prep,
  324. .recover_slot = recover_slot,
  325. .recover_done = recover_done,
  326. };
  327. /*
  328. * The BAST function for the ack lock resource
  329. * This function wakes up the receive thread in
  330. * order to receive and process the message.
  331. */
  332. static void ack_bast(void *arg, int mode)
  333. {
  334. struct dlm_lock_resource *res = arg;
  335. struct md_cluster_info *cinfo = res->mddev->cluster_info;
  336. if (mode == DLM_LOCK_EX) {
  337. if (test_bit(MD_CLUSTER_ALREADY_IN_CLUSTER, &cinfo->state))
  338. md_wakeup_thread(cinfo->recv_thread);
  339. else
  340. set_bit(MD_CLUSTER_PENDING_RECV_EVENT, &cinfo->state);
  341. }
  342. }
  343. static void __remove_suspend_info(struct md_cluster_info *cinfo, int slot)
  344. {
  345. struct suspend_info *s, *tmp;
  346. list_for_each_entry_safe(s, tmp, &cinfo->suspend_list, list)
  347. if (slot == s->slot) {
  348. list_del(&s->list);
  349. kfree(s);
  350. break;
  351. }
  352. }
  353. static void remove_suspend_info(struct mddev *mddev, int slot)
  354. {
  355. struct md_cluster_info *cinfo = mddev->cluster_info;
  356. spin_lock_irq(&cinfo->suspend_lock);
  357. __remove_suspend_info(cinfo, slot);
  358. spin_unlock_irq(&cinfo->suspend_lock);
  359. mddev->pers->quiesce(mddev, 2);
  360. }
  361. static void process_suspend_info(struct mddev *mddev,
  362. int slot, sector_t lo, sector_t hi)
  363. {
  364. struct md_cluster_info *cinfo = mddev->cluster_info;
  365. struct suspend_info *s;
  366. if (!hi) {
  367. remove_suspend_info(mddev, slot);
  368. set_bit(MD_RECOVERY_NEEDED, &mddev->recovery);
  369. md_wakeup_thread(mddev->thread);
  370. return;
  371. }
  372. /*
  373. * The bitmaps are not same for different nodes
  374. * if RESYNCING is happening in one node, then
  375. * the node which received the RESYNCING message
  376. * probably will perform resync with the region
  377. * [lo, hi] again, so we could reduce resync time
  378. * a lot if we can ensure that the bitmaps among
  379. * different nodes are match up well.
  380. *
  381. * sync_low/hi is used to record the region which
  382. * arrived in the previous RESYNCING message,
  383. *
  384. * Call bitmap_sync_with_cluster to clear
  385. * NEEDED_MASK and set RESYNC_MASK since
  386. * resync thread is running in another node,
  387. * so we don't need to do the resync again
  388. * with the same section */
  389. bitmap_sync_with_cluster(mddev, cinfo->sync_low,
  390. cinfo->sync_hi,
  391. lo, hi);
  392. cinfo->sync_low = lo;
  393. cinfo->sync_hi = hi;
  394. s = kzalloc(sizeof(struct suspend_info), GFP_KERNEL);
  395. if (!s)
  396. return;
  397. s->slot = slot;
  398. s->lo = lo;
  399. s->hi = hi;
  400. mddev->pers->quiesce(mddev, 1);
  401. mddev->pers->quiesce(mddev, 0);
  402. spin_lock_irq(&cinfo->suspend_lock);
  403. /* Remove existing entry (if exists) before adding */
  404. __remove_suspend_info(cinfo, slot);
  405. list_add(&s->list, &cinfo->suspend_list);
  406. spin_unlock_irq(&cinfo->suspend_lock);
  407. mddev->pers->quiesce(mddev, 2);
  408. }
  409. static void process_add_new_disk(struct mddev *mddev, struct cluster_msg *cmsg)
  410. {
  411. char disk_uuid[64];
  412. struct md_cluster_info *cinfo = mddev->cluster_info;
  413. char event_name[] = "EVENT=ADD_DEVICE";
  414. char raid_slot[16];
  415. char *envp[] = {event_name, disk_uuid, raid_slot, NULL};
  416. int len;
  417. len = snprintf(disk_uuid, 64, "DEVICE_UUID=");
  418. sprintf(disk_uuid + len, "%pU", cmsg->uuid);
  419. snprintf(raid_slot, 16, "RAID_DISK=%d", le32_to_cpu(cmsg->raid_slot));
  420. pr_info("%s:%d Sending kobject change with %s and %s\n", __func__, __LINE__, disk_uuid, raid_slot);
  421. init_completion(&cinfo->newdisk_completion);
  422. set_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state);
  423. kobject_uevent_env(&disk_to_dev(mddev->gendisk)->kobj, KOBJ_CHANGE, envp);
  424. wait_for_completion_timeout(&cinfo->newdisk_completion,
  425. NEW_DEV_TIMEOUT);
  426. clear_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state);
  427. }
  428. static void process_metadata_update(struct mddev *mddev, struct cluster_msg *msg)
  429. {
  430. struct md_cluster_info *cinfo = mddev->cluster_info;
  431. mddev->good_device_nr = le32_to_cpu(msg->raid_slot);
  432. set_bit(MD_RELOAD_SB, &mddev->flags);
  433. dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR);
  434. md_wakeup_thread(mddev->thread);
  435. }
  436. static void process_remove_disk(struct mddev *mddev, struct cluster_msg *msg)
  437. {
  438. struct md_rdev *rdev = md_find_rdev_nr_rcu(mddev,
  439. le32_to_cpu(msg->raid_slot));
  440. if (rdev) {
  441. set_bit(ClusterRemove, &rdev->flags);
  442. set_bit(MD_RECOVERY_NEEDED, &mddev->recovery);
  443. md_wakeup_thread(mddev->thread);
  444. }
  445. else
  446. pr_warn("%s: %d Could not find disk(%d) to REMOVE\n",
  447. __func__, __LINE__, le32_to_cpu(msg->raid_slot));
  448. }
  449. static void process_readd_disk(struct mddev *mddev, struct cluster_msg *msg)
  450. {
  451. struct md_rdev *rdev = md_find_rdev_nr_rcu(mddev,
  452. le32_to_cpu(msg->raid_slot));
  453. if (rdev && test_bit(Faulty, &rdev->flags))
  454. clear_bit(Faulty, &rdev->flags);
  455. else
  456. pr_warn("%s: %d Could not find disk(%d) which is faulty",
  457. __func__, __LINE__, le32_to_cpu(msg->raid_slot));
  458. }
  459. static int process_recvd_msg(struct mddev *mddev, struct cluster_msg *msg)
  460. {
  461. int ret = 0;
  462. if (WARN(mddev->cluster_info->slot_number - 1 == le32_to_cpu(msg->slot),
  463. "node %d received it's own msg\n", le32_to_cpu(msg->slot)))
  464. return -1;
  465. switch (le32_to_cpu(msg->type)) {
  466. case METADATA_UPDATED:
  467. process_metadata_update(mddev, msg);
  468. break;
  469. case RESYNCING:
  470. process_suspend_info(mddev, le32_to_cpu(msg->slot),
  471. le64_to_cpu(msg->low),
  472. le64_to_cpu(msg->high));
  473. break;
  474. case NEWDISK:
  475. process_add_new_disk(mddev, msg);
  476. break;
  477. case REMOVE:
  478. process_remove_disk(mddev, msg);
  479. break;
  480. case RE_ADD:
  481. process_readd_disk(mddev, msg);
  482. break;
  483. case BITMAP_NEEDS_SYNC:
  484. __recover_slot(mddev, le32_to_cpu(msg->slot));
  485. break;
  486. default:
  487. ret = -1;
  488. pr_warn("%s:%d Received unknown message from %d\n",
  489. __func__, __LINE__, msg->slot);
  490. }
  491. return ret;
  492. }
  493. /*
  494. * thread for receiving message
  495. */
  496. static void recv_daemon(struct md_thread *thread)
  497. {
  498. struct md_cluster_info *cinfo = thread->mddev->cluster_info;
  499. struct dlm_lock_resource *ack_lockres = cinfo->ack_lockres;
  500. struct dlm_lock_resource *message_lockres = cinfo->message_lockres;
  501. struct cluster_msg msg;
  502. int ret;
  503. mutex_lock(&cinfo->recv_mutex);
  504. /*get CR on Message*/
  505. if (dlm_lock_sync(message_lockres, DLM_LOCK_CR)) {
  506. pr_err("md/raid1:failed to get CR on MESSAGE\n");
  507. mutex_unlock(&cinfo->recv_mutex);
  508. return;
  509. }
  510. /* read lvb and wake up thread to process this message_lockres */
  511. memcpy(&msg, message_lockres->lksb.sb_lvbptr, sizeof(struct cluster_msg));
  512. ret = process_recvd_msg(thread->mddev, &msg);
  513. if (ret)
  514. goto out;
  515. /*release CR on ack_lockres*/
  516. ret = dlm_unlock_sync(ack_lockres);
  517. if (unlikely(ret != 0))
  518. pr_info("unlock ack failed return %d\n", ret);
  519. /*up-convert to PR on message_lockres*/
  520. ret = dlm_lock_sync(message_lockres, DLM_LOCK_PR);
  521. if (unlikely(ret != 0))
  522. pr_info("lock PR on msg failed return %d\n", ret);
  523. /*get CR on ack_lockres again*/
  524. ret = dlm_lock_sync(ack_lockres, DLM_LOCK_CR);
  525. if (unlikely(ret != 0))
  526. pr_info("lock CR on ack failed return %d\n", ret);
  527. out:
  528. /*release CR on message_lockres*/
  529. ret = dlm_unlock_sync(message_lockres);
  530. if (unlikely(ret != 0))
  531. pr_info("unlock msg failed return %d\n", ret);
  532. mutex_unlock(&cinfo->recv_mutex);
  533. }
  534. /* lock_token()
  535. * Takes the lock on the TOKEN lock resource so no other
  536. * node can communicate while the operation is underway.
  537. */
  538. static int lock_token(struct md_cluster_info *cinfo)
  539. {
  540. int error;
  541. error = dlm_lock_sync(cinfo->token_lockres, DLM_LOCK_EX);
  542. if (error)
  543. pr_err("md-cluster(%s:%d): failed to get EX on TOKEN (%d)\n",
  544. __func__, __LINE__, error);
  545. /* Lock the receive sequence */
  546. mutex_lock(&cinfo->recv_mutex);
  547. return error;
  548. }
  549. /* lock_comm()
  550. * Sets the MD_CLUSTER_SEND_LOCK bit to lock the send channel.
  551. */
  552. static int lock_comm(struct md_cluster_info *cinfo)
  553. {
  554. wait_event(cinfo->wait,
  555. !test_and_set_bit(MD_CLUSTER_SEND_LOCK, &cinfo->state));
  556. return lock_token(cinfo);
  557. }
  558. static void unlock_comm(struct md_cluster_info *cinfo)
  559. {
  560. WARN_ON(cinfo->token_lockres->mode != DLM_LOCK_EX);
  561. mutex_unlock(&cinfo->recv_mutex);
  562. dlm_unlock_sync(cinfo->token_lockres);
  563. clear_bit(MD_CLUSTER_SEND_LOCK, &cinfo->state);
  564. wake_up(&cinfo->wait);
  565. }
  566. /* __sendmsg()
  567. * This function performs the actual sending of the message. This function is
  568. * usually called after performing the encompassing operation
  569. * The function:
  570. * 1. Grabs the message lockresource in EX mode
  571. * 2. Copies the message to the message LVB
  572. * 3. Downconverts message lockresource to CW
  573. * 4. Upconverts ack lock resource from CR to EX. This forces the BAST on other nodes
  574. * and the other nodes read the message. The thread will wait here until all other
  575. * nodes have released ack lock resource.
  576. * 5. Downconvert ack lockresource to CR
  577. */
  578. static int __sendmsg(struct md_cluster_info *cinfo, struct cluster_msg *cmsg)
  579. {
  580. int error;
  581. int slot = cinfo->slot_number - 1;
  582. cmsg->slot = cpu_to_le32(slot);
  583. /*get EX on Message*/
  584. error = dlm_lock_sync(cinfo->message_lockres, DLM_LOCK_EX);
  585. if (error) {
  586. pr_err("md-cluster: failed to get EX on MESSAGE (%d)\n", error);
  587. goto failed_message;
  588. }
  589. memcpy(cinfo->message_lockres->lksb.sb_lvbptr, (void *)cmsg,
  590. sizeof(struct cluster_msg));
  591. /*down-convert EX to CW on Message*/
  592. error = dlm_lock_sync(cinfo->message_lockres, DLM_LOCK_CW);
  593. if (error) {
  594. pr_err("md-cluster: failed to convert EX to CW on MESSAGE(%d)\n",
  595. error);
  596. goto failed_ack;
  597. }
  598. /*up-convert CR to EX on Ack*/
  599. error = dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_EX);
  600. if (error) {
  601. pr_err("md-cluster: failed to convert CR to EX on ACK(%d)\n",
  602. error);
  603. goto failed_ack;
  604. }
  605. /*down-convert EX to CR on Ack*/
  606. error = dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_CR);
  607. if (error) {
  608. pr_err("md-cluster: failed to convert EX to CR on ACK(%d)\n",
  609. error);
  610. goto failed_ack;
  611. }
  612. failed_ack:
  613. error = dlm_unlock_sync(cinfo->message_lockres);
  614. if (unlikely(error != 0)) {
  615. pr_err("md-cluster: failed convert to NL on MESSAGE(%d)\n",
  616. error);
  617. /* in case the message can't be released due to some reason */
  618. goto failed_ack;
  619. }
  620. failed_message:
  621. return error;
  622. }
  623. static int sendmsg(struct md_cluster_info *cinfo, struct cluster_msg *cmsg)
  624. {
  625. int ret;
  626. lock_comm(cinfo);
  627. ret = __sendmsg(cinfo, cmsg);
  628. unlock_comm(cinfo);
  629. return ret;
  630. }
  631. static int gather_all_resync_info(struct mddev *mddev, int total_slots)
  632. {
  633. struct md_cluster_info *cinfo = mddev->cluster_info;
  634. int i, ret = 0;
  635. struct dlm_lock_resource *bm_lockres;
  636. struct suspend_info *s;
  637. char str[64];
  638. sector_t lo, hi;
  639. for (i = 0; i < total_slots; i++) {
  640. memset(str, '\0', 64);
  641. snprintf(str, 64, "bitmap%04d", i);
  642. bm_lockres = lockres_init(mddev, str, NULL, 1);
  643. if (!bm_lockres)
  644. return -ENOMEM;
  645. if (i == (cinfo->slot_number - 1)) {
  646. lockres_free(bm_lockres);
  647. continue;
  648. }
  649. bm_lockres->flags |= DLM_LKF_NOQUEUE;
  650. ret = dlm_lock_sync(bm_lockres, DLM_LOCK_PW);
  651. if (ret == -EAGAIN) {
  652. memset(bm_lockres->lksb.sb_lvbptr, '\0', LVB_SIZE);
  653. s = read_resync_info(mddev, bm_lockres);
  654. if (s) {
  655. pr_info("%s:%d Resync[%llu..%llu] in progress on %d\n",
  656. __func__, __LINE__,
  657. (unsigned long long) s->lo,
  658. (unsigned long long) s->hi, i);
  659. spin_lock_irq(&cinfo->suspend_lock);
  660. s->slot = i;
  661. list_add(&s->list, &cinfo->suspend_list);
  662. spin_unlock_irq(&cinfo->suspend_lock);
  663. }
  664. ret = 0;
  665. lockres_free(bm_lockres);
  666. continue;
  667. }
  668. if (ret) {
  669. lockres_free(bm_lockres);
  670. goto out;
  671. }
  672. /* Read the disk bitmap sb and check if it needs recovery */
  673. ret = bitmap_copy_from_slot(mddev, i, &lo, &hi, false);
  674. if (ret) {
  675. pr_warn("md-cluster: Could not gather bitmaps from slot %d", i);
  676. lockres_free(bm_lockres);
  677. continue;
  678. }
  679. if ((hi > 0) && (lo < mddev->recovery_cp)) {
  680. set_bit(MD_RECOVERY_NEEDED, &mddev->recovery);
  681. mddev->recovery_cp = lo;
  682. md_check_recovery(mddev);
  683. }
  684. dlm_unlock_sync(bm_lockres);
  685. lockres_free(bm_lockres);
  686. }
  687. out:
  688. return ret;
  689. }
  690. static int join(struct mddev *mddev, int nodes)
  691. {
  692. struct md_cluster_info *cinfo;
  693. int ret, ops_rv;
  694. char str[64];
  695. cinfo = kzalloc(sizeof(struct md_cluster_info), GFP_KERNEL);
  696. if (!cinfo)
  697. return -ENOMEM;
  698. INIT_LIST_HEAD(&cinfo->suspend_list);
  699. spin_lock_init(&cinfo->suspend_lock);
  700. init_completion(&cinfo->completion);
  701. set_bit(MD_CLUSTER_BEGIN_JOIN_CLUSTER, &cinfo->state);
  702. init_waitqueue_head(&cinfo->wait);
  703. mutex_init(&cinfo->recv_mutex);
  704. mddev->cluster_info = cinfo;
  705. memset(str, 0, 64);
  706. sprintf(str, "%pU", mddev->uuid);
  707. ret = dlm_new_lockspace(str, mddev->bitmap_info.cluster_name,
  708. DLM_LSFL_FS, LVB_SIZE,
  709. &md_ls_ops, mddev, &ops_rv, &cinfo->lockspace);
  710. if (ret)
  711. goto err;
  712. wait_for_completion(&cinfo->completion);
  713. if (nodes < cinfo->slot_number) {
  714. pr_err("md-cluster: Slot allotted(%d) is greater than available slots(%d).",
  715. cinfo->slot_number, nodes);
  716. ret = -ERANGE;
  717. goto err;
  718. }
  719. /* Initiate the communication resources */
  720. ret = -ENOMEM;
  721. cinfo->recv_thread = md_register_thread(recv_daemon, mddev, "cluster_recv");
  722. if (!cinfo->recv_thread) {
  723. pr_err("md-cluster: cannot allocate memory for recv_thread!\n");
  724. goto err;
  725. }
  726. cinfo->message_lockres = lockres_init(mddev, "message", NULL, 1);
  727. if (!cinfo->message_lockres)
  728. goto err;
  729. cinfo->token_lockres = lockres_init(mddev, "token", NULL, 0);
  730. if (!cinfo->token_lockres)
  731. goto err;
  732. cinfo->no_new_dev_lockres = lockres_init(mddev, "no-new-dev", NULL, 0);
  733. if (!cinfo->no_new_dev_lockres)
  734. goto err;
  735. ret = dlm_lock_sync(cinfo->token_lockres, DLM_LOCK_EX);
  736. if (ret) {
  737. ret = -EAGAIN;
  738. pr_err("md-cluster: can't join cluster to avoid lock issue\n");
  739. goto err;
  740. }
  741. cinfo->ack_lockres = lockres_init(mddev, "ack", ack_bast, 0);
  742. if (!cinfo->ack_lockres)
  743. goto err;
  744. /* get sync CR lock on ACK. */
  745. if (dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_CR))
  746. pr_err("md-cluster: failed to get a sync CR lock on ACK!(%d)\n",
  747. ret);
  748. dlm_unlock_sync(cinfo->token_lockres);
  749. /* get sync CR lock on no-new-dev. */
  750. if (dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR))
  751. pr_err("md-cluster: failed to get a sync CR lock on no-new-dev!(%d)\n", ret);
  752. pr_info("md-cluster: Joined cluster %s slot %d\n", str, cinfo->slot_number);
  753. snprintf(str, 64, "bitmap%04d", cinfo->slot_number - 1);
  754. cinfo->bitmap_lockres = lockres_init(mddev, str, NULL, 1);
  755. if (!cinfo->bitmap_lockres)
  756. goto err;
  757. if (dlm_lock_sync(cinfo->bitmap_lockres, DLM_LOCK_PW)) {
  758. pr_err("Failed to get bitmap lock\n");
  759. ret = -EINVAL;
  760. goto err;
  761. }
  762. cinfo->resync_lockres = lockres_init(mddev, "resync", NULL, 0);
  763. if (!cinfo->resync_lockres)
  764. goto err;
  765. return 0;
  766. err:
  767. md_unregister_thread(&cinfo->recovery_thread);
  768. md_unregister_thread(&cinfo->recv_thread);
  769. lockres_free(cinfo->message_lockres);
  770. lockres_free(cinfo->token_lockres);
  771. lockres_free(cinfo->ack_lockres);
  772. lockres_free(cinfo->no_new_dev_lockres);
  773. lockres_free(cinfo->resync_lockres);
  774. lockres_free(cinfo->bitmap_lockres);
  775. if (cinfo->lockspace)
  776. dlm_release_lockspace(cinfo->lockspace, 2);
  777. mddev->cluster_info = NULL;
  778. kfree(cinfo);
  779. return ret;
  780. }
  781. static void load_bitmaps(struct mddev *mddev, int total_slots)
  782. {
  783. struct md_cluster_info *cinfo = mddev->cluster_info;
  784. /* load all the node's bitmap info for resync */
  785. if (gather_all_resync_info(mddev, total_slots))
  786. pr_err("md-cluster: failed to gather all resyn infos\n");
  787. set_bit(MD_CLUSTER_ALREADY_IN_CLUSTER, &cinfo->state);
  788. /* wake up recv thread in case something need to be handled */
  789. if (test_and_clear_bit(MD_CLUSTER_PENDING_RECV_EVENT, &cinfo->state))
  790. md_wakeup_thread(cinfo->recv_thread);
  791. }
  792. static void resync_bitmap(struct mddev *mddev)
  793. {
  794. struct md_cluster_info *cinfo = mddev->cluster_info;
  795. struct cluster_msg cmsg = {0};
  796. int err;
  797. cmsg.type = cpu_to_le32(BITMAP_NEEDS_SYNC);
  798. err = sendmsg(cinfo, &cmsg);
  799. if (err)
  800. pr_err("%s:%d: failed to send BITMAP_NEEDS_SYNC message (%d)\n",
  801. __func__, __LINE__, err);
  802. }
  803. static void unlock_all_bitmaps(struct mddev *mddev);
  804. static int leave(struct mddev *mddev)
  805. {
  806. struct md_cluster_info *cinfo = mddev->cluster_info;
  807. if (!cinfo)
  808. return 0;
  809. /* BITMAP_NEEDS_SYNC message should be sent when node
  810. * is leaving the cluster with dirty bitmap, also we
  811. * can only deliver it when dlm connection is available */
  812. if (cinfo->slot_number > 0 && mddev->recovery_cp != MaxSector)
  813. resync_bitmap(mddev);
  814. md_unregister_thread(&cinfo->recovery_thread);
  815. md_unregister_thread(&cinfo->recv_thread);
  816. lockres_free(cinfo->message_lockres);
  817. lockres_free(cinfo->token_lockres);
  818. lockres_free(cinfo->ack_lockres);
  819. lockres_free(cinfo->no_new_dev_lockres);
  820. lockres_free(cinfo->resync_lockres);
  821. lockres_free(cinfo->bitmap_lockres);
  822. unlock_all_bitmaps(mddev);
  823. dlm_release_lockspace(cinfo->lockspace, 2);
  824. return 0;
  825. }
  826. /* slot_number(): Returns the MD slot number to use
  827. * DLM starts the slot numbers from 1, wheras cluster-md
  828. * wants the number to be from zero, so we deduct one
  829. */
  830. static int slot_number(struct mddev *mddev)
  831. {
  832. struct md_cluster_info *cinfo = mddev->cluster_info;
  833. return cinfo->slot_number - 1;
  834. }
  835. /*
  836. * Check if the communication is already locked, else lock the communication
  837. * channel.
  838. * If it is already locked, token is in EX mode, and hence lock_token()
  839. * should not be called.
  840. */
  841. static int metadata_update_start(struct mddev *mddev)
  842. {
  843. struct md_cluster_info *cinfo = mddev->cluster_info;
  844. wait_event(cinfo->wait,
  845. !test_and_set_bit(MD_CLUSTER_SEND_LOCK, &cinfo->state) ||
  846. test_and_clear_bit(MD_CLUSTER_SEND_LOCKED_ALREADY, &cinfo->state));
  847. /* If token is already locked, return 0 */
  848. if (cinfo->token_lockres->mode == DLM_LOCK_EX)
  849. return 0;
  850. return lock_token(cinfo);
  851. }
  852. static int metadata_update_finish(struct mddev *mddev)
  853. {
  854. struct md_cluster_info *cinfo = mddev->cluster_info;
  855. struct cluster_msg cmsg;
  856. struct md_rdev *rdev;
  857. int ret = 0;
  858. int raid_slot = -1;
  859. memset(&cmsg, 0, sizeof(cmsg));
  860. cmsg.type = cpu_to_le32(METADATA_UPDATED);
  861. /* Pick up a good active device number to send.
  862. */
  863. rdev_for_each(rdev, mddev)
  864. if (rdev->raid_disk > -1 && !test_bit(Faulty, &rdev->flags)) {
  865. raid_slot = rdev->desc_nr;
  866. break;
  867. }
  868. if (raid_slot >= 0) {
  869. cmsg.raid_slot = cpu_to_le32(raid_slot);
  870. ret = __sendmsg(cinfo, &cmsg);
  871. } else
  872. pr_warn("md-cluster: No good device id found to send\n");
  873. clear_bit(MD_CLUSTER_SEND_LOCKED_ALREADY, &cinfo->state);
  874. unlock_comm(cinfo);
  875. return ret;
  876. }
  877. static void metadata_update_cancel(struct mddev *mddev)
  878. {
  879. struct md_cluster_info *cinfo = mddev->cluster_info;
  880. clear_bit(MD_CLUSTER_SEND_LOCKED_ALREADY, &cinfo->state);
  881. unlock_comm(cinfo);
  882. }
  883. static int resync_start(struct mddev *mddev)
  884. {
  885. struct md_cluster_info *cinfo = mddev->cluster_info;
  886. return dlm_lock_sync(cinfo->resync_lockres, DLM_LOCK_EX);
  887. }
  888. static int resync_info_update(struct mddev *mddev, sector_t lo, sector_t hi)
  889. {
  890. struct md_cluster_info *cinfo = mddev->cluster_info;
  891. struct resync_info ri;
  892. struct cluster_msg cmsg = {0};
  893. /* do not send zero again, if we have sent before */
  894. if (hi == 0) {
  895. memcpy(&ri, cinfo->bitmap_lockres->lksb.sb_lvbptr, sizeof(struct resync_info));
  896. if (le64_to_cpu(ri.hi) == 0)
  897. return 0;
  898. }
  899. add_resync_info(cinfo->bitmap_lockres, lo, hi);
  900. /* Re-acquire the lock to refresh LVB */
  901. dlm_lock_sync(cinfo->bitmap_lockres, DLM_LOCK_PW);
  902. cmsg.type = cpu_to_le32(RESYNCING);
  903. cmsg.low = cpu_to_le64(lo);
  904. cmsg.high = cpu_to_le64(hi);
  905. return sendmsg(cinfo, &cmsg);
  906. }
  907. static int resync_finish(struct mddev *mddev)
  908. {
  909. struct md_cluster_info *cinfo = mddev->cluster_info;
  910. dlm_unlock_sync(cinfo->resync_lockres);
  911. return resync_info_update(mddev, 0, 0);
  912. }
  913. static int area_resyncing(struct mddev *mddev, int direction,
  914. sector_t lo, sector_t hi)
  915. {
  916. struct md_cluster_info *cinfo = mddev->cluster_info;
  917. int ret = 0;
  918. struct suspend_info *s;
  919. if ((direction == READ) &&
  920. test_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state))
  921. return 1;
  922. spin_lock_irq(&cinfo->suspend_lock);
  923. if (list_empty(&cinfo->suspend_list))
  924. goto out;
  925. list_for_each_entry(s, &cinfo->suspend_list, list)
  926. if (hi > s->lo && lo < s->hi) {
  927. ret = 1;
  928. break;
  929. }
  930. out:
  931. spin_unlock_irq(&cinfo->suspend_lock);
  932. return ret;
  933. }
  934. /* add_new_disk() - initiates a disk add
  935. * However, if this fails before writing md_update_sb(),
  936. * add_new_disk_cancel() must be called to release token lock
  937. */
  938. static int add_new_disk(struct mddev *mddev, struct md_rdev *rdev)
  939. {
  940. struct md_cluster_info *cinfo = mddev->cluster_info;
  941. struct cluster_msg cmsg;
  942. int ret = 0;
  943. struct mdp_superblock_1 *sb = page_address(rdev->sb_page);
  944. char *uuid = sb->device_uuid;
  945. memset(&cmsg, 0, sizeof(cmsg));
  946. cmsg.type = cpu_to_le32(NEWDISK);
  947. memcpy(cmsg.uuid, uuid, 16);
  948. cmsg.raid_slot = cpu_to_le32(rdev->desc_nr);
  949. lock_comm(cinfo);
  950. ret = __sendmsg(cinfo, &cmsg);
  951. if (ret)
  952. return ret;
  953. cinfo->no_new_dev_lockres->flags |= DLM_LKF_NOQUEUE;
  954. ret = dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_EX);
  955. cinfo->no_new_dev_lockres->flags &= ~DLM_LKF_NOQUEUE;
  956. /* Some node does not "see" the device */
  957. if (ret == -EAGAIN)
  958. ret = -ENOENT;
  959. if (ret)
  960. unlock_comm(cinfo);
  961. else {
  962. dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR);
  963. /* Since MD_CHANGE_DEVS will be set in add_bound_rdev which
  964. * will run soon after add_new_disk, the below path will be
  965. * invoked:
  966. * md_wakeup_thread(mddev->thread)
  967. * -> conf->thread (raid1d)
  968. * -> md_check_recovery -> md_update_sb
  969. * -> metadata_update_start/finish
  970. * MD_CLUSTER_SEND_LOCKED_ALREADY will be cleared eventually.
  971. *
  972. * For other failure cases, metadata_update_cancel and
  973. * add_new_disk_cancel also clear below bit as well.
  974. * */
  975. set_bit(MD_CLUSTER_SEND_LOCKED_ALREADY, &cinfo->state);
  976. wake_up(&cinfo->wait);
  977. }
  978. return ret;
  979. }
  980. static void add_new_disk_cancel(struct mddev *mddev)
  981. {
  982. struct md_cluster_info *cinfo = mddev->cluster_info;
  983. clear_bit(MD_CLUSTER_SEND_LOCKED_ALREADY, &cinfo->state);
  984. unlock_comm(cinfo);
  985. }
  986. static int new_disk_ack(struct mddev *mddev, bool ack)
  987. {
  988. struct md_cluster_info *cinfo = mddev->cluster_info;
  989. if (!test_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state)) {
  990. pr_warn("md-cluster(%s): Spurious cluster confirmation\n", mdname(mddev));
  991. return -EINVAL;
  992. }
  993. if (ack)
  994. dlm_unlock_sync(cinfo->no_new_dev_lockres);
  995. complete(&cinfo->newdisk_completion);
  996. return 0;
  997. }
  998. static int remove_disk(struct mddev *mddev, struct md_rdev *rdev)
  999. {
  1000. struct cluster_msg cmsg = {0};
  1001. struct md_cluster_info *cinfo = mddev->cluster_info;
  1002. cmsg.type = cpu_to_le32(REMOVE);
  1003. cmsg.raid_slot = cpu_to_le32(rdev->desc_nr);
  1004. return sendmsg(cinfo, &cmsg);
  1005. }
  1006. static int lock_all_bitmaps(struct mddev *mddev)
  1007. {
  1008. int slot, my_slot, ret, held = 1, i = 0;
  1009. char str[64];
  1010. struct md_cluster_info *cinfo = mddev->cluster_info;
  1011. cinfo->other_bitmap_lockres = kzalloc((mddev->bitmap_info.nodes - 1) *
  1012. sizeof(struct dlm_lock_resource *),
  1013. GFP_KERNEL);
  1014. if (!cinfo->other_bitmap_lockres) {
  1015. pr_err("md: can't alloc mem for other bitmap locks\n");
  1016. return 0;
  1017. }
  1018. my_slot = slot_number(mddev);
  1019. for (slot = 0; slot < mddev->bitmap_info.nodes; slot++) {
  1020. if (slot == my_slot)
  1021. continue;
  1022. memset(str, '\0', 64);
  1023. snprintf(str, 64, "bitmap%04d", slot);
  1024. cinfo->other_bitmap_lockres[i] = lockres_init(mddev, str, NULL, 1);
  1025. if (!cinfo->other_bitmap_lockres[i])
  1026. return -ENOMEM;
  1027. cinfo->other_bitmap_lockres[i]->flags |= DLM_LKF_NOQUEUE;
  1028. ret = dlm_lock_sync(cinfo->other_bitmap_lockres[i], DLM_LOCK_PW);
  1029. if (ret)
  1030. held = -1;
  1031. i++;
  1032. }
  1033. return held;
  1034. }
  1035. static void unlock_all_bitmaps(struct mddev *mddev)
  1036. {
  1037. struct md_cluster_info *cinfo = mddev->cluster_info;
  1038. int i;
  1039. /* release other node's bitmap lock if they are existed */
  1040. if (cinfo->other_bitmap_lockres) {
  1041. for (i = 0; i < mddev->bitmap_info.nodes - 1; i++) {
  1042. if (cinfo->other_bitmap_lockres[i]) {
  1043. dlm_unlock_sync(cinfo->other_bitmap_lockres[i]);
  1044. lockres_free(cinfo->other_bitmap_lockres[i]);
  1045. }
  1046. }
  1047. kfree(cinfo->other_bitmap_lockres);
  1048. }
  1049. }
  1050. static int gather_bitmaps(struct md_rdev *rdev)
  1051. {
  1052. int sn, err;
  1053. sector_t lo, hi;
  1054. struct cluster_msg cmsg = {0};
  1055. struct mddev *mddev = rdev->mddev;
  1056. struct md_cluster_info *cinfo = mddev->cluster_info;
  1057. cmsg.type = cpu_to_le32(RE_ADD);
  1058. cmsg.raid_slot = cpu_to_le32(rdev->desc_nr);
  1059. err = sendmsg(cinfo, &cmsg);
  1060. if (err)
  1061. goto out;
  1062. for (sn = 0; sn < mddev->bitmap_info.nodes; sn++) {
  1063. if (sn == (cinfo->slot_number - 1))
  1064. continue;
  1065. err = bitmap_copy_from_slot(mddev, sn, &lo, &hi, false);
  1066. if (err) {
  1067. pr_warn("md-cluster: Could not gather bitmaps from slot %d", sn);
  1068. goto out;
  1069. }
  1070. if ((hi > 0) && (lo < mddev->recovery_cp))
  1071. mddev->recovery_cp = lo;
  1072. }
  1073. out:
  1074. return err;
  1075. }
  1076. static struct md_cluster_operations cluster_ops = {
  1077. .join = join,
  1078. .leave = leave,
  1079. .slot_number = slot_number,
  1080. .resync_start = resync_start,
  1081. .resync_finish = resync_finish,
  1082. .resync_info_update = resync_info_update,
  1083. .metadata_update_start = metadata_update_start,
  1084. .metadata_update_finish = metadata_update_finish,
  1085. .metadata_update_cancel = metadata_update_cancel,
  1086. .area_resyncing = area_resyncing,
  1087. .add_new_disk = add_new_disk,
  1088. .add_new_disk_cancel = add_new_disk_cancel,
  1089. .new_disk_ack = new_disk_ack,
  1090. .remove_disk = remove_disk,
  1091. .load_bitmaps = load_bitmaps,
  1092. .gather_bitmaps = gather_bitmaps,
  1093. .lock_all_bitmaps = lock_all_bitmaps,
  1094. .unlock_all_bitmaps = unlock_all_bitmaps,
  1095. };
  1096. static int __init cluster_init(void)
  1097. {
  1098. pr_warn("md-cluster: EXPERIMENTAL. Use with caution\n");
  1099. pr_info("Registering Cluster MD functions\n");
  1100. register_md_cluster_operations(&cluster_ops, THIS_MODULE);
  1101. return 0;
  1102. }
  1103. static void cluster_exit(void)
  1104. {
  1105. unregister_md_cluster_operations();
  1106. }
  1107. module_init(cluster_init);
  1108. module_exit(cluster_exit);
  1109. MODULE_AUTHOR("SUSE");
  1110. MODULE_LICENSE("GPL");
  1111. MODULE_DESCRIPTION("Clustering support for MD");