1 /****************************************************************************** 2 ******************************************************************************* 3 ** 4 ** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved. 5 ** Copyright (C) 2004-2005 Red Hat, Inc. All rights reserved. 6 ** 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. 10 ** 11 ******************************************************************************* 12 ******************************************************************************/ 13 14 #include "dlm_internal.h" 15 #include "lockspace.h" 16 #include "member.h" 17 #include "lowcomms.h" 18 #include "rcom.h" 19 #include "config.h" 20 #include "memory.h" 21 #include "recover.h" 22 #include "util.h" 23 #include "lock.h" 24 #include "dir.h" 25 26 27 static void put_free_de(struct dlm_ls *ls, struct dlm_direntry *de) 28 { 29 spin_lock(&ls->ls_recover_list_lock); 30 list_add(&de->list, &ls->ls_recover_list); 31 spin_unlock(&ls->ls_recover_list_lock); 32 } 33 34 static struct dlm_direntry *get_free_de(struct dlm_ls *ls, int len) 35 { 36 int found = 0; 37 struct dlm_direntry *de; 38 39 spin_lock(&ls->ls_recover_list_lock); 40 list_for_each_entry(de, &ls->ls_recover_list, list) { 41 if (de->length == len) { 42 list_del(&de->list); 43 de->master_nodeid = 0; 44 memset(de->name, 0, len); 45 found = 1; 46 break; 47 } 48 } 49 spin_unlock(&ls->ls_recover_list_lock); 50 51 if (!found) 52 de = kzalloc(sizeof(struct dlm_direntry) + len, 53 ls->ls_allocation); 54 return de; 55 } 56 57 void dlm_clear_free_entries(struct dlm_ls *ls) 58 { 59 struct dlm_direntry *de; 60 61 spin_lock(&ls->ls_recover_list_lock); 62 while (!list_empty(&ls->ls_recover_list)) { 63 de = list_entry(ls->ls_recover_list.next, struct dlm_direntry, 64 list); 65 list_del(&de->list); 66 kfree(de); 67 } 68 spin_unlock(&ls->ls_recover_list_lock); 69 } 70 71 /* 72 * We use the upper 16 bits of the hash value to select the directory node. 73 * Low bits are used for distribution of rsb's among hash buckets on each node. 74 * 75 * To give the exact range wanted (0 to num_nodes-1), we apply a modulus of 76 * num_nodes to the hash value. This value in the desired range is used as an 77 * offset into the sorted list of nodeid's to give the particular nodeid. 78 */ 79 80 int dlm_hash2nodeid(struct dlm_ls *ls, uint32_t hash) 81 { 82 struct list_head *tmp; 83 struct dlm_member *memb = NULL; 84 uint32_t node, n = 0; 85 int nodeid; 86 87 if (ls->ls_num_nodes == 1) { 88 nodeid = dlm_our_nodeid(); 89 goto out; 90 } 91 92 if (ls->ls_node_array) { 93 node = (hash >> 16) % ls->ls_total_weight; 94 nodeid = ls->ls_node_array[node]; 95 goto out; 96 } 97 98 /* make_member_array() failed to kmalloc ls_node_array... */ 99 100 node = (hash >> 16) % ls->ls_num_nodes; 101 102 list_for_each(tmp, &ls->ls_nodes) { 103 if (n++ != node) 104 continue; 105 memb = list_entry(tmp, struct dlm_member, list); 106 break; 107 } 108 109 DLM_ASSERT(memb , printk("num_nodes=%u n=%u node=%u\n", 110 ls->ls_num_nodes, n, node);); 111 nodeid = memb->nodeid; 112 out: 113 return nodeid; 114 } 115 116 int dlm_dir_nodeid(struct dlm_rsb *r) 117 { 118 return dlm_hash2nodeid(r->res_ls, r->res_hash); 119 } 120 121 static inline uint32_t dir_hash(struct dlm_ls *ls, char *name, int len) 122 { 123 uint32_t val; 124 125 val = jhash(name, len, 0); 126 val &= (ls->ls_dirtbl_size - 1); 127 128 return val; 129 } 130 131 static void add_entry_to_hash(struct dlm_ls *ls, struct dlm_direntry *de) 132 { 133 uint32_t bucket; 134 135 bucket = dir_hash(ls, de->name, de->length); 136 list_add_tail(&de->list, &ls->ls_dirtbl[bucket].list); 137 } 138 139 static struct dlm_direntry *search_bucket(struct dlm_ls *ls, char *name, 140 int namelen, uint32_t bucket) 141 { 142 struct dlm_direntry *de; 143 144 list_for_each_entry(de, &ls->ls_dirtbl[bucket].list, list) { 145 if (de->length == namelen && !memcmp(name, de->name, namelen)) 146 goto out; 147 } 148 de = NULL; 149 out: 150 return de; 151 } 152 153 void dlm_dir_remove_entry(struct dlm_ls *ls, int nodeid, char *name, int namelen) 154 { 155 struct dlm_direntry *de; 156 uint32_t bucket; 157 158 bucket = dir_hash(ls, name, namelen); 159 160 spin_lock(&ls->ls_dirtbl[bucket].lock); 161 162 de = search_bucket(ls, name, namelen, bucket); 163 164 if (!de) { 165 log_error(ls, "remove fr %u none", nodeid); 166 goto out; 167 } 168 169 if (de->master_nodeid != nodeid) { 170 log_error(ls, "remove fr %u ID %u", nodeid, de->master_nodeid); 171 goto out; 172 } 173 174 list_del(&de->list); 175 kfree(de); 176 out: 177 spin_unlock(&ls->ls_dirtbl[bucket].lock); 178 } 179 180 void dlm_dir_clear(struct dlm_ls *ls) 181 { 182 struct list_head *head; 183 struct dlm_direntry *de; 184 int i; 185 186 DLM_ASSERT(list_empty(&ls->ls_recover_list), ); 187 188 for (i = 0; i < ls->ls_dirtbl_size; i++) { 189 spin_lock(&ls->ls_dirtbl[i].lock); 190 head = &ls->ls_dirtbl[i].list; 191 while (!list_empty(head)) { 192 de = list_entry(head->next, struct dlm_direntry, list); 193 list_del(&de->list); 194 put_free_de(ls, de); 195 } 196 spin_unlock(&ls->ls_dirtbl[i].lock); 197 } 198 } 199 200 int dlm_recover_directory(struct dlm_ls *ls) 201 { 202 struct dlm_member *memb; 203 struct dlm_direntry *de; 204 char *b, *last_name = NULL; 205 int error = -ENOMEM, last_len, count = 0; 206 uint16_t namelen; 207 208 log_debug(ls, "dlm_recover_directory"); 209 210 if (dlm_no_directory(ls)) 211 goto out_status; 212 213 dlm_dir_clear(ls); 214 215 last_name = kmalloc(DLM_RESNAME_MAXLEN, ls->ls_allocation); 216 if (!last_name) 217 goto out; 218 219 list_for_each_entry(memb, &ls->ls_nodes, list) { 220 memset(last_name, 0, DLM_RESNAME_MAXLEN); 221 last_len = 0; 222 223 for (;;) { 224 int left; 225 error = dlm_recovery_stopped(ls); 226 if (error) 227 goto out_free; 228 229 error = dlm_rcom_names(ls, memb->nodeid, 230 last_name, last_len); 231 if (error) 232 goto out_free; 233 234 schedule(); 235 236 /* 237 * pick namelen/name pairs out of received buffer 238 */ 239 240 b = ls->ls_recover_buf->rc_buf; 241 left = ls->ls_recover_buf->rc_header.h_length; 242 left -= sizeof(struct dlm_rcom); 243 244 for (;;) { 245 __be16 v; 246 247 error = -EINVAL; 248 if (left < sizeof(__be16)) 249 goto out_free; 250 251 memcpy(&v, b, sizeof(__be16)); 252 namelen = be16_to_cpu(v); 253 b += sizeof(__be16); 254 left -= sizeof(__be16); 255 256 /* namelen of 0xFFFFF marks end of names for 257 this node; namelen of 0 marks end of the 258 buffer */ 259 260 if (namelen == 0xFFFF) 261 goto done; 262 if (!namelen) 263 break; 264 265 if (namelen > left) 266 goto out_free; 267 268 if (namelen > DLM_RESNAME_MAXLEN) 269 goto out_free; 270 271 error = -ENOMEM; 272 de = get_free_de(ls, namelen); 273 if (!de) 274 goto out_free; 275 276 de->master_nodeid = memb->nodeid; 277 de->length = namelen; 278 last_len = namelen; 279 memcpy(de->name, b, namelen); 280 memcpy(last_name, b, namelen); 281 b += namelen; 282 left -= namelen; 283 284 add_entry_to_hash(ls, de); 285 count++; 286 } 287 } 288 done: 289 ; 290 } 291 292 out_status: 293 error = 0; 294 dlm_set_recover_status(ls, DLM_RS_DIR); 295 log_debug(ls, "dlm_recover_directory %d entries", count); 296 out_free: 297 kfree(last_name); 298 out: 299 dlm_clear_free_entries(ls); 300 return error; 301 } 302 303 static int get_entry(struct dlm_ls *ls, int nodeid, char *name, 304 int namelen, int *r_nodeid) 305 { 306 struct dlm_direntry *de, *tmp; 307 uint32_t bucket; 308 309 bucket = dir_hash(ls, name, namelen); 310 311 spin_lock(&ls->ls_dirtbl[bucket].lock); 312 de = search_bucket(ls, name, namelen, bucket); 313 if (de) { 314 *r_nodeid = de->master_nodeid; 315 spin_unlock(&ls->ls_dirtbl[bucket].lock); 316 if (*r_nodeid == nodeid) 317 return -EEXIST; 318 return 0; 319 } 320 321 spin_unlock(&ls->ls_dirtbl[bucket].lock); 322 323 if (namelen > DLM_RESNAME_MAXLEN) 324 return -EINVAL; 325 326 de = kzalloc(sizeof(struct dlm_direntry) + namelen, ls->ls_allocation); 327 if (!de) 328 return -ENOMEM; 329 330 de->master_nodeid = nodeid; 331 de->length = namelen; 332 memcpy(de->name, name, namelen); 333 334 spin_lock(&ls->ls_dirtbl[bucket].lock); 335 tmp = search_bucket(ls, name, namelen, bucket); 336 if (tmp) { 337 kfree(de); 338 de = tmp; 339 } else { 340 list_add_tail(&de->list, &ls->ls_dirtbl[bucket].list); 341 } 342 *r_nodeid = de->master_nodeid; 343 spin_unlock(&ls->ls_dirtbl[bucket].lock); 344 return 0; 345 } 346 347 int dlm_dir_lookup(struct dlm_ls *ls, int nodeid, char *name, int namelen, 348 int *r_nodeid) 349 { 350 return get_entry(ls, nodeid, name, namelen, r_nodeid); 351 } 352 353 static struct dlm_rsb *find_rsb_root(struct dlm_ls *ls, char *name, int len) 354 { 355 struct dlm_rsb *r; 356 357 down_read(&ls->ls_root_sem); 358 list_for_each_entry(r, &ls->ls_root_list, res_root_list) { 359 if (len == r->res_length && !memcmp(name, r->res_name, len)) { 360 up_read(&ls->ls_root_sem); 361 return r; 362 } 363 } 364 up_read(&ls->ls_root_sem); 365 return NULL; 366 } 367 368 /* Find the rsb where we left off (or start again), then send rsb names 369 for rsb's we're master of and whose directory node matches the requesting 370 node. inbuf is the rsb name last sent, inlen is the name's length */ 371 372 void dlm_copy_master_names(struct dlm_ls *ls, char *inbuf, int inlen, 373 char *outbuf, int outlen, int nodeid) 374 { 375 struct list_head *list; 376 struct dlm_rsb *r; 377 int offset = 0, dir_nodeid; 378 __be16 be_namelen; 379 380 down_read(&ls->ls_root_sem); 381 382 if (inlen > 1) { 383 r = find_rsb_root(ls, inbuf, inlen); 384 if (!r) { 385 inbuf[inlen - 1] = '\0'; 386 log_error(ls, "copy_master_names from %d start %d %s", 387 nodeid, inlen, inbuf); 388 goto out; 389 } 390 list = r->res_root_list.next; 391 } else { 392 list = ls->ls_root_list.next; 393 } 394 395 for (offset = 0; list != &ls->ls_root_list; list = list->next) { 396 r = list_entry(list, struct dlm_rsb, res_root_list); 397 if (r->res_nodeid) 398 continue; 399 400 dir_nodeid = dlm_dir_nodeid(r); 401 if (dir_nodeid != nodeid) 402 continue; 403 404 /* 405 * The block ends when we can't fit the following in the 406 * remaining buffer space: 407 * namelen (uint16_t) + 408 * name (r->res_length) + 409 * end-of-block record 0x0000 (uint16_t) 410 */ 411 412 if (offset + sizeof(uint16_t)*2 + r->res_length > outlen) { 413 /* Write end-of-block record */ 414 be_namelen = cpu_to_be16(0); 415 memcpy(outbuf + offset, &be_namelen, sizeof(__be16)); 416 offset += sizeof(__be16); 417 goto out; 418 } 419 420 be_namelen = cpu_to_be16(r->res_length); 421 memcpy(outbuf + offset, &be_namelen, sizeof(__be16)); 422 offset += sizeof(__be16); 423 memcpy(outbuf + offset, r->res_name, r->res_length); 424 offset += r->res_length; 425 } 426 427 /* 428 * If we've reached the end of the list (and there's room) write a 429 * terminating record. 430 */ 431 432 if ((list == &ls->ls_root_list) && 433 (offset + sizeof(uint16_t) <= outlen)) { 434 be_namelen = cpu_to_be16(0xFFFF); 435 memcpy(outbuf + offset, &be_namelen, sizeof(__be16)); 436 offset += sizeof(__be16); 437 } 438 439 out: 440 up_read(&ls->ls_root_sem); 441 } 442 443