fs/dlm/dir.c

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