lustre/utils/ofd_access_batch.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  *
  22  * Copyright 2020, DataDirect Networks Storage.
  23  *
  24  * This file is part of Lustre, http://www.lustre.org/
  25  *
  26  * Author: John L. Hammond <jhammond@whamcloud.com>
  27  *
  28  * lustre/utils/ofd_access_batch.c
  29  *
  30  * Access log entry batching for ofd_access_log_reader.
  31  */
  32 #include <stdlib.h>
  33 #include <stdbool.h>
  34 #include <stddef.h>
  35 #include <assert.h>
  36 #include <malloc.h>
  37 #include <unistd.h>
  38 #include <pthread.h>
  39 #include <linux/lustre/lustre_access_log.h>
  40 #include <linux/lustre/lustre_fid.h>
  41 #include <linux/lustre/lustre_idl.h>
  42 #include <libcfs/util/hash.h>
  43 #include <libcfs/util/list.h>
  44 #include <lustre/lustreapi.h>
  45 #include "lstddef.h"
  46 #include "ofd_access_batch.h"
  47
  48 struct fid_hash_node {
  49         struct list_head fhn_node;
  50         struct lu_fid fhn_fid;
  51 };
  52
  53 static inline bool fid_eq(const struct lu_fid *f1, const struct lu_fid *f2)
  54 {
  55         return f1->f_seq == f2->f_seq && f1->f_oid == f2->f_oid &&
  56                f1->f_ver == f2->f_ver;
  57 }
  58
  59 static void fhn_init(struct fid_hash_node *fhn, const struct lu_fid *fid)
  60 {
  61         INIT_LIST_HEAD(&fhn->fhn_node);
  62         fhn->fhn_fid = *fid;
  63 }
  64
  65 static bool fhn_is_hashed(struct fid_hash_node *fhn)
  66 {
  67         return !list_empty(&fhn->fhn_node);
  68 }
  69
  70 static void fhn_del_init(struct fid_hash_node *fhn)
  71 {
  72         if (fhn_is_hashed(fhn))
  73                 list_del_init(&fhn->fhn_node);
  74 }
  75
  76 static inline void fhn_replace_init(struct fid_hash_node *old_fhn,
  77                                 struct fid_hash_node *new_fhn)
  78 {
  79         list_add(&new_fhn->fhn_node, &old_fhn->fhn_node);
  80         list_del_init(&old_fhn->fhn_node);
  81 }
  82
  83 static void fid_hash_add(struct list_head *head, unsigned int shift,
  84                          struct fid_hash_node *fhn)
  85 {
  86         assert(!fhn_is_hashed(fhn));
  87
  88         list_add(&fhn->fhn_node, &head[llapi_fid_hash(&fhn->fhn_fid, shift)]);
  89 }
  90
  91 static struct fid_hash_node *
  92 fid_hash_insert(struct list_head *head, unsigned int shift, struct fid_hash_node *new_fhn)
  93 {
  94         struct list_head *list;
  95         struct fid_hash_node *old_fhn, *next;
  96
  97         list = &head[llapi_fid_hash(&new_fhn->fhn_fid, shift)];
  98         list_for_each_entry_safe(old_fhn, next, list, fhn_node) {
  99                 assert(fhn_is_hashed(old_fhn));
 100
 101                 if (fid_eq(&old_fhn->fhn_fid, &new_fhn->fhn_fid))
 102                         return old_fhn;
 103         }
 104
 105         list_add(&new_fhn->fhn_node, list);
 106
 107         return new_fhn;
 108 }
 109
 110 static int fid_hash_init(struct list_head **phead, unsigned int *pshift,
 111                          unsigned int shift)
 112 {
 113         struct list_head *new_head;
 114         unsigned int i;
 115
 116         new_head = malloc(sizeof(*new_head) << shift);
 117         if (new_head == NULL)
 118                 return -1;
 119
 120         for (i = 0; i < (1 << shift); i++)
 121                 INIT_LIST_HEAD(&new_head[i]);
 122
 123         *phead = new_head;
 124         *pshift = shift;
 125
 126         return 0;
 127 }
 128
 129 static int fid_hash_resize(struct list_head **phead, unsigned int *pshift,
 130                            unsigned int new_shift)
 131 {
 132         struct list_head *new_head;
 133         unsigned int i;
 134         int rc;
 135
 136         if (*pshift == new_shift)
 137                 return 0;
 138
 139         rc = fid_hash_init(&new_head, &new_shift, new_shift);
 140         if (rc < 0)
 141                 return rc;
 142
 143         for (i = 0; i < (1 << *pshift); i++) {
 144                 struct list_head *list = &(*phead)[i];
 145                 struct fid_hash_node *fhn, *next;
 146
 147                 list_for_each_entry_safe(fhn, next, list, fhn_node) {
 148                         fhn_del_init(fhn);
 149                         fid_hash_add(new_head, new_shift, fhn);
 150                 }
 151         }
 152
 153         free(*phead);
 154         *phead = new_head;
 155         *pshift = new_shift;
 156
 157         return 0;
 158 }
 159
 160 enum alr_rw {
 161         ALR_READ = 0,
 162         ALR_WRITE = 1,
 163         ALR_RW_MAX
 164 };
 165
 166 /* Entry in the batching hash. */
 167 struct alr_entry {
 168         struct fid_hash_node alre_fid_hash_node;
 169         time_t alre_time[ALR_RW_MAX]; /* Not strictly needed. */
 170         __u64 alre_begin[ALR_RW_MAX];
 171         __u64 alre_end[ALR_RW_MAX];
 172         __u64 alre_size[ALR_RW_MAX];
 173         __u64 alre_segment_count[ALR_RW_MAX];
 174         __u64 alre_count[ALR_RW_MAX];
 175         char alre_obd_name[];
 176 };
 177
 178 enum {
 179         ALR_BATCH_HASH_SHIFT_DEFAULT = 10,
 180         ALR_BATCH_HASH_SHIFT_MAX = 30,
 181 };
 182
 183 struct alr_batch {
 184         struct list_head *alrb_hash;
 185         unsigned int alrb_hash_shift;
 186         unsigned int alrb_count;
 187 };
 188
 189 static void alre_del_init(struct alr_entry *alre)
 190 {
 191         fhn_del_init(&alre->alre_fid_hash_node);
 192 }
 193
 194 static void alre_update(struct alr_entry *alre, time_t time, __u64 begin,
 195                         __u64 end, __u32 size, __u32 segment_count, __u32 flags)
 196 {
 197         enum alr_rw d = (flags & OFD_ACCESS_READ) ? ALR_READ : ALR_WRITE;
 198
 199         alre->alre_time[d] = max_t(time_t, alre->alre_time[d], time);
 200         alre->alre_begin[d] = min_t(__u64, alre->alre_begin[d], begin);
 201         alre->alre_end[d] = max_t(__u64, alre->alre_end[d], end);
 202         alre->alre_size[d] += size;
 203         alre->alre_segment_count[d] += segment_count;
 204         alre->alre_count[d] += 1;
 205 }
 206
 207 int alr_batch_add(struct alr_batch *alrb, const char *obd_name,
 208                 const struct lu_fid *pfid, time_t time, __u64 begin, __u64 end,
 209                 __u32 size, __u32 segment_count, __u32 flags)
 210 {
 211         struct fid_hash_node fhn, *p;
 212         struct alr_entry *alre;
 213         int rc;
 214
 215         if (alrb == NULL)
 216                 return 0;
 217
 218         assert(sizeof(time_t) == sizeof(__u64));
 219
 220         fhn_init(&fhn, pfid);
 221
 222         /* Find old or insert sentinel (fhn). Replace sentinel if returned. */
 223         p = fid_hash_insert(alrb->alrb_hash, alrb->alrb_hash_shift, &fhn);
 224         if (p == &fhn) {
 225                 size_t alre_size = sizeof(*alre) + strlen(obd_name) + 1;
 226
 227                 alre = calloc(1, alre_size);
 228                 if (alre == NULL) {
 229                         rc = -1;
 230                         goto out;
 231                 }
 232
 233                 fhn_init(&alre->alre_fid_hash_node, pfid);
 234                 strcpy(alre->alre_obd_name, obd_name);
 235                 fhn_replace_init(&fhn, &alre->alre_fid_hash_node);
 236                 alrb->alrb_count++;
 237         } else {
 238                 alre = container_of(p, struct alr_entry, alre_fid_hash_node);
 239         }
 240
 241         alre_update(alre, time, begin, end, size, segment_count, flags);
 242         rc = 0;
 243 out:
 244         fhn_del_init(&fhn);
 245
 246         return rc;
 247 }
 248
 249 static int sort_compare(const void *a1, const void *a2)
 250 {
 251         int l = *(const int*)a1;
 252         int r = *(const int *)a2;
 253         if (l > r) return -1;
 254         if (l < r) return  1;
 255         return 0;
 256 }
 257
 258 static void alre_printf(FILE *f, struct alr_entry *alre, enum alr_rw d)
 259 {
 260         fprintf(f, "o=%s f="DFID" t=%lld b=%llu e=%llu s=%llu g=%llu n=%llu d=%c\n",
 261                 alre->alre_obd_name,
 262                 PFID(&alre->alre_fid_hash_node.fhn_fid),
 263                 (long long)alre->alre_time[d],
 264                 (unsigned long long)alre->alre_begin[d],
 265                 (unsigned long long)alre->alre_end[d],
 266                 (unsigned long long)alre->alre_size[d],
 267                 (unsigned long long)alre->alre_segment_count[d],
 268                 (unsigned long long)alre->alre_count[d],
 269                 (d == ALR_READ) ? 'r' : 'w');
 270 }
 271
 272 struct alr_thread_arg {
 273         struct list_head list;
 274         int fraction;
 275         FILE *file;
 276         pthread_mutex_t *file_mutex;
 277 };
 278
 279 /* Fraction < 100 */
 280 static void *alr_sort_and_print_thread(void *arg)
 281 {
 282         struct alr_entry *alre, *next;
 283         struct alr_thread_arg *aa = arg;
 284         struct list_head *tmp = &aa->list;
 285         int *sa = NULL;
 286         int rc, i, nr = 0;
 287         enum alr_rw d;
 288         unsigned long cut;
 289
 290         list_for_each_entry(alre, tmp, alre_fid_hash_node.fhn_node) {
 291                 if (alre->alre_count[ALR_READ] > 0)
 292                         nr++;
 293                 if (alre->alre_count[ALR_WRITE] > 0)
 294                         nr++;
 295         }
 296
 297         if (nr == 0)
 298                 goto out;
 299
 300         sa = calloc(nr, sizeof(*sa));
 301         if (!sa) {
 302                 fprintf(stderr, "cannot allocate memory for sorting\n");
 303                 exit(1);
 304         }
 305
 306         i = 0;
 307         list_for_each_entry(alre, tmp, alre_fid_hash_node.fhn_node) {
 308                 if (alre->alre_count[ALR_READ] > 0)
 309                         sa[i++] = alre->alre_count[ALR_READ];
 310                 if (alre->alre_count[ALR_WRITE] > 0)
 311                         sa[i++] = alre->alre_count[ALR_WRITE];
 312         }
 313
 314         qsort(sa, nr, sizeof(*sa), sort_compare);
 315         i = nr * aa->fraction / 100;
 316
 317         cut = sa[i];
 318         if (cut < 1)
 319                 cut = 1;
 320         free(sa);
 321
 322         /* Prevent jumbled output from multiple concurrent sort and
 323          * print threads. */
 324         rc = pthread_mutex_lock(aa->file_mutex);
 325         if (rc != 0) {
 326                 fprintf(stderr, "cannot lock batch file: %s\n",
 327                         strerror(rc));
 328                 exit(1);
 329         }
 330
 331         /* there might be lots of items at @cut, but we want to limit total
 332          * output. so the first loop dumps all items > @cut and the second
 333          * loop dumps items=@cut so that total number (@i) is not exceeeded.
 334          * XXX: possible optimization - move items=@cut to another list, so
 335          * that 2nd pass takes < O(n) */
 336         list_for_each_entry(alre, tmp, alre_fid_hash_node.fhn_node) {
 337                 for (d = 0; d < ALR_RW_MAX; d++) {
 338                         if (alre->alre_count[d] <= cut)
 339                                 continue;
 340                         alre_printf(aa->file, alre, d);
 341                         i--;
 342                 }
 343         }
 344
 345         list_for_each_entry(alre, tmp, alre_fid_hash_node.fhn_node) {
 346                 for (d = 0; d < ALR_RW_MAX && i > 0; d++) {
 347                         if (alre->alre_count[d] != cut)
 348                                 continue;
 349                         alre_printf(aa->file, alre, d);
 350                         i--;
 351                 }
 352         }
 353
 354         rc = pthread_mutex_unlock(aa->file_mutex);
 355         if (rc != 0) {
 356                 fprintf(stderr, "cannot unlock batch file: %s\n",
 357                         strerror(rc));
 358                 exit(1);
 359         }
 360
 361 out:
 362         fflush(aa->file);
 363
 364         list_for_each_entry_safe(alre, next, tmp, alre_fid_hash_node.fhn_node) {
 365                 alre_del_init(alre);
 366                 free(alre);
 367         }
 368
 369         free(aa);
 370
 371         return NULL;
 372 }
 373
 374 /* Fraction == 100 */
 375 static void *alr_print_thread_fraction_100(void *arg)
 376 {
 377         struct alr_entry *alre, *next;
 378         struct alr_thread_arg *aa = arg;
 379         int rc;
 380
 381         /* Prevent jumbled output from multiple concurrent sort and
 382          * print threads. */
 383         rc = pthread_mutex_lock(aa->file_mutex);
 384         if (rc != 0) {
 385                 fprintf(stderr, "cannot lock batch file: %s\n", strerror(rc));
 386                 exit(1);
 387         }
 388
 389         list_for_each_entry(alre, &aa->list, alre_fid_hash_node.fhn_node) {
 390                 enum alr_rw d;
 391
 392                 for (d = 0; d < ALR_RW_MAX; d++) {
 393                         if (alre->alre_count[d] != 0)
 394                                 alre_printf(aa->file, alre, d);
 395                 }
 396         }
 397
 398         rc = pthread_mutex_unlock(aa->file_mutex);
 399         if (rc != 0) {
 400                 fprintf(stderr, "cannot unlock batch file: %s\n", strerror(rc));
 401                 exit(1);
 402         }
 403
 404         fflush(aa->file);
 405
 406         list_for_each_entry_safe(alre, next, &aa->list, alre_fid_hash_node.fhn_node) {
 407                 alre_del_init(alre);
 408                 free(alre);
 409         }
 410
 411         free(aa);
 412
 413         return NULL;
 414 }
 415
 416 /* Print, clear, and resize the batch. */
 417 int alr_batch_print(struct alr_batch *alrb, FILE *file,
 418                     pthread_mutex_t *file_mutex, int fraction)
 419 {
 420         unsigned int new_hash_shift;
 421         pthread_attr_t attr, *pattr = NULL;
 422         struct alr_thread_arg *aa = NULL;
 423         pthread_t pid;
 424         int i, rc;
 425
 426         if (alrb == NULL)
 427                 return 0;
 428
 429         aa = calloc(1, sizeof(*aa));
 430         if (aa == NULL)
 431                 return -ENOMEM;
 432
 433         /* move all collected items to the temp list */
 434         INIT_LIST_HEAD(&aa->list);
 435         for (i = 0; i < (1 << alrb->alrb_hash_shift); i++) {
 436                 if (list_empty(&alrb->alrb_hash[i]))
 437                         continue;
 438                 list_splice(&alrb->alrb_hash[i], &aa->list);
 439                 INIT_LIST_HEAD(&alrb->alrb_hash[i]);
 440         }
 441         aa->file = file;
 442         aa->file_mutex = file_mutex;
 443         aa->fraction = fraction;
 444
 445         rc = pthread_attr_init(&attr);
 446         if (rc != 0)
 447                 goto out;
 448
 449         pattr = &attr;
 450
 451         rc = pthread_attr_setdetachstate(pattr, PTHREAD_CREATE_DETACHED);
 452         if (rc != 0)
 453                 goto out;
 454
 455         /* as sorting may take time and we don't want to lose access
 456          * records we better do sorting and printing in a different thread */
 457
 458         if (fraction >= 100) /* Print all 100% records */
 459                 rc = pthread_create(&pid, pattr, &alr_print_thread_fraction_100, aa);
 460         else
 461                 rc = pthread_create(&pid, pattr, &alr_sort_and_print_thread, aa);
 462         if (rc != 0)
 463                 goto out;
 464
 465         aa = NULL; /* Sort and print thread owns it now. */
 466 out:
 467         /* Resize hash based on previous count. */
 468         new_hash_shift = alrb->alrb_hash_shift;
 469
 470         while (new_hash_shift < ALR_BATCH_HASH_SHIFT_MAX &&
 471                (1 << new_hash_shift) < alrb->alrb_count)
 472                 new_hash_shift++;
 473
 474         fid_hash_resize(&alrb->alrb_hash, &alrb->alrb_hash_shift,
 475                         new_hash_shift);
 476
 477         alrb->alrb_count = 0;
 478
 479         if (pattr != NULL)
 480                 pthread_attr_destroy(pattr);
 481
 482         if (aa != NULL) {
 483                 struct alr_entry *alre, *next;
 484
 485                 list_for_each_entry_safe(alre, next, &aa->list,
 486                                          alre_fid_hash_node.fhn_node) {
 487                         alre_del_init(alre);
 488                         free(alre);
 489                 }
 490         }
 491
 492         free(aa);
 493
 494         if (rc > 0)
 495                 rc = -rc; /* Fixup pthread return conventions. */
 496
 497         return rc;
 498 }
 499
 500 struct alr_batch *alr_batch_create(unsigned int shift)
 501 {
 502         struct alr_batch *alrb;
 503         int rc;
 504
 505         if (shift == -1U)
 506                 shift = ALR_BATCH_HASH_SHIFT_DEFAULT;
 507
 508         alrb = calloc(1, sizeof(*alrb));
 509         if (alrb == NULL)
 510                 return NULL;
 511
 512         rc = fid_hash_init(&alrb->alrb_hash, &alrb->alrb_hash_shift, shift);
 513         if (rc < 0) {
 514                 free(alrb);
 515                 return NULL;
 516         }
 517
 518         return alrb;
 519 }
 520
 521 void alr_batch_destroy(struct alr_batch *alrb)
 522 {
 523         unsigned int i;
 524
 525         if (alrb == NULL)
 526                 return;
 527
 528         for (i = 0; i < (1 << alrb->alrb_hash_shift); i++) {
 529                 struct list_head *list = &alrb->alrb_hash[i];
 530                 struct alr_entry *alre, *next;
 531
 532                 list_for_each_entry_safe(alre, next, list, alre_fid_hash_node.fhn_node) {
 533                         alre_del_init(alre);
 534                         free(alre);
 535                 }
 536         }
 537
 538         free(alrb->alrb_hash);
 539         free(alrb);
 540 }