lustre/ptlrpc/sec_bulk.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * Copyright (C) 2006-2007 Cluster File Systems, Inc.
   5  *   Author: Eric Mei <ericm@clusterfs.com>
   6  *
   7  *   This file is part of Lustre, http://www.lustre.org.
   8  *
   9  *   Lustre is free software; you can redistribute it and/or
  10  *   modify it under the terms of version 2 of the GNU General Public
  11  *   License as published by the Free Software Foundation.
  12  *
  13  *   Lustre is distributed in the hope that it will be useful,
  14  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16  *   GNU General Public License for more details.
  17  *
  18  *   You should have received a copy of the GNU General Public License
  19  *   along with Lustre; if not, write to the Free Software
  20  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  21  */
  22
  23 #ifndef EXPORT_SYMTAB
  24 #define EXPORT_SYMTAB
  25 #endif
  26 #define DEBUG_SUBSYSTEM S_SEC
  27
  28 #include <libcfs/libcfs.h>
  29 #ifndef __KERNEL__
  30 #include <liblustre.h>
  31 #include <libcfs/list.h>
  32 #else
  33 #include <linux/crypto.h>
  34 #endif
  35
  36 #include <obd.h>
  37 #include <obd_cksum.h>
  38 #include <obd_class.h>
  39 #include <obd_support.h>
  40 #include <lustre_net.h>
  41 #include <lustre_import.h>
  42 #include <lustre_dlm.h>
  43 #include <lustre_sec.h>
  44
  45 #include "ptlrpc_internal.h"
  46
  47 /****************************************
  48  * bulk encryption page pools           *
  49  ****************************************/
  50
  51 #ifdef __KERNEL__
  52
  53 #define PTRS_PER_PAGE   (CFS_PAGE_SIZE / sizeof(void *))
  54 #define PAGES_PER_POOL  (PTRS_PER_PAGE)
  55
  56 #define IDLE_IDX_MAX            (100)
  57 #define IDLE_IDX_WEIGHT         (3)
  58
  59 #define CACHE_QUIESCENT_PERIOD  (20)
  60
  61 static struct ptlrpc_enc_page_pool {
  62         /*
  63          * constants
  64          */
  65         unsigned long    epp_max_pages;   /* maximum pages can hold, const */
  66         unsigned int     epp_max_pools;   /* number of pools, const */
  67
  68         /*
  69          * wait queue in case of not enough free pages.
  70          */
  71         cfs_waitq_t      epp_waitq;       /* waiting threads */
  72         unsigned int     epp_waitqlen;    /* wait queue length */
  73         unsigned long    epp_pages_short; /* # of pages wanted of in-q users */
  74         unsigned int     epp_growing:1;   /* during adding pages */
  75
  76         /*
  77          * indicating how idle the pools are, from 0 to MAX_IDLE_IDX
  78          * this is counted based on each time when getting pages from
  79          * the pools, not based on time. which means in case that system
  80          * is idled for a while but the idle_idx might still be low if no
  81          * activities happened in the pools.
  82          */
  83         unsigned long    epp_idle_idx;
  84
  85         /* last shrink time due to mem tight */
  86         long             epp_last_shrink;
  87         long             epp_last_access;
  88
  89         /*
  90          * in-pool pages bookkeeping
  91          */
  92         spinlock_t       epp_lock;        /* protect following fields */
  93         unsigned long    epp_total_pages; /* total pages in pools */
  94         unsigned long    epp_free_pages;  /* current pages available */
  95
  96         /*
  97          * statistics
  98          */
  99         unsigned long    epp_st_max_pages;      /* # of pages ever reached */
 100         unsigned int     epp_st_grows;          /* # of grows */
 101         unsigned int     epp_st_grow_fails;     /* # of add pages failures */
 102         unsigned int     epp_st_shrinks;        /* # of shrinks */
 103         unsigned long    epp_st_access;         /* # of access */
 104         unsigned long    epp_st_missings;       /* # of cache missing */
 105         unsigned long    epp_st_lowfree;        /* lowest free pages reached */
 106         unsigned int     epp_st_max_wqlen;      /* highest waitqueue length */
 107         cfs_time_t       epp_st_max_wait;       /* in jeffies */
 108         /*
 109          * pointers to pools
 110          */
 111         cfs_page_t    ***epp_pools;
 112 } page_pools;
 113
 114 /*
 115  * memory shrinker
 116  */
 117 const int pools_shrinker_seeks = DEFAULT_SEEKS;
 118 static struct shrinker *pools_shrinker = NULL;
 119
 120
 121 /*
 122  * /proc/fs/lustre/sptlrpc/encrypt_page_pools
 123  */
 124 int sptlrpc_proc_read_enc_pool(char *page, char **start, off_t off, int count,
 125                                int *eof, void *data)
 126 {
 127         int     rc;
 128
 129         spin_lock(&page_pools.epp_lock);
 130
 131         rc = snprintf(page, count,
 132                       "physical pages:          %lu\n"
 133                       "pages per pool:          %lu\n"
 134                       "max pages:               %lu\n"
 135                       "max pools:               %u\n"
 136                       "total pages:             %lu\n"
 137                       "total free:              %lu\n"
 138                       "idle index:              %lu/100\n"
 139                       "last shrink:             %lds\n"
 140                       "last access:             %lds\n"
 141                       "max pages reached:       %lu\n"
 142                       "grows:                   %u\n"
 143                       "grows failure:           %u\n"
 144                       "shrinks:                 %u\n"
 145                       "cache access:            %lu\n"
 146                       "cache missing:           %lu\n"
 147                       "low free mark:           %lu\n"
 148                       "max waitqueue depth:     %u\n"
 149                       "max wait time:           "CFS_TIME_T"/%u\n"
 150                       ,
 151                       num_physpages,
 152                       PAGES_PER_POOL,
 153                       page_pools.epp_max_pages,
 154                       page_pools.epp_max_pools,
 155                       page_pools.epp_total_pages,
 156                       page_pools.epp_free_pages,
 157                       page_pools.epp_idle_idx,
 158                       cfs_time_current_sec() - page_pools.epp_last_shrink,
 159                       cfs_time_current_sec() - page_pools.epp_last_access,
 160                       page_pools.epp_st_max_pages,
 161                       page_pools.epp_st_grows,
 162                       page_pools.epp_st_grow_fails,
 163                       page_pools.epp_st_shrinks,
 164                       page_pools.epp_st_access,
 165                       page_pools.epp_st_missings,
 166                       page_pools.epp_st_lowfree,
 167                       page_pools.epp_st_max_wqlen,
 168                       page_pools.epp_st_max_wait, HZ
 169                      );
 170
 171         spin_unlock(&page_pools.epp_lock);
 172         return rc;
 173 }
 174
 175 static void enc_pools_release_free_pages(long npages)
 176 {
 177         int     p_idx, g_idx;
 178         int     p_idx_max1, p_idx_max2;
 179
 180         LASSERT(npages > 0);
 181         LASSERT(npages <= page_pools.epp_free_pages);
 182         LASSERT(page_pools.epp_free_pages <= page_pools.epp_total_pages);
 183
 184         /* max pool index before the release */
 185         p_idx_max2 = (page_pools.epp_total_pages - 1) / PAGES_PER_POOL;
 186
 187         page_pools.epp_free_pages -= npages;
 188         page_pools.epp_total_pages -= npages;
 189
 190         /* max pool index after the release */
 191         p_idx_max1 = page_pools.epp_total_pages == 0 ? 0 :
 192                      ((page_pools.epp_total_pages - 1) / PAGES_PER_POOL);
 193
 194         p_idx = page_pools.epp_free_pages / PAGES_PER_POOL;
 195         g_idx = page_pools.epp_free_pages % PAGES_PER_POOL;
 196         LASSERT(page_pools.epp_pools[p_idx]);
 197
 198         while (npages--) {
 199                 LASSERT(page_pools.epp_pools[p_idx]);
 200                 LASSERT(page_pools.epp_pools[p_idx][g_idx] != NULL);
 201
 202                 cfs_free_page(page_pools.epp_pools[p_idx][g_idx]);
 203                 page_pools.epp_pools[p_idx][g_idx] = NULL;
 204
 205                 if (++g_idx == PAGES_PER_POOL) {
 206                         p_idx++;
 207                         g_idx = 0;
 208                 }
 209         };
 210
 211         /* free unused pools */
 212         while (p_idx_max1 < p_idx_max2) {
 213                 LASSERT(page_pools.epp_pools[p_idx_max2]);
 214                 OBD_FREE(page_pools.epp_pools[p_idx_max2], CFS_PAGE_SIZE);
 215                 page_pools.epp_pools[p_idx_max2] = NULL;
 216                 p_idx_max2--;
 217         }
 218 }
 219
 220 /*
 221  * could be called frequently for query (@nr_to_scan == 0)
 222  */
 223 static int enc_pools_shrink(int nr_to_scan, unsigned int gfp_mask)
 224 {
 225         unsigned long   ret;
 226
 227         spin_lock(&page_pools.epp_lock);
 228
 229         if (nr_to_scan > page_pools.epp_free_pages)
 230                 nr_to_scan = page_pools.epp_free_pages;
 231
 232         if (nr_to_scan > 0) {
 233                 enc_pools_release_free_pages(nr_to_scan);
 234                 CDEBUG(D_SEC, "released %d pages, %ld left\n",
 235                        nr_to_scan, page_pools.epp_free_pages);
 236
 237                 page_pools.epp_st_shrinks++;
 238                 page_pools.epp_last_shrink = cfs_time_current_sec();
 239         }
 240
 241         /*
 242          * try to keep at least PTLRPC_MAX_BRW_PAGES pages in the pool
 243          */
 244         if (page_pools.epp_free_pages <= PTLRPC_MAX_BRW_PAGES) {
 245                 ret = 0;
 246                 goto out_unlock;
 247         }
 248
 249         /*
 250          * if no pool access for a long time, we consider it's fully idle
 251          */
 252         if (cfs_time_current_sec() - page_pools.epp_last_access >
 253             CACHE_QUIESCENT_PERIOD)
 254                 page_pools.epp_idle_idx = IDLE_IDX_MAX;
 255
 256         LASSERT(page_pools.epp_idle_idx <= IDLE_IDX_MAX);
 257         ret = (page_pools.epp_free_pages * page_pools.epp_idle_idx /
 258                IDLE_IDX_MAX);
 259         if (page_pools.epp_free_pages - ret < PTLRPC_MAX_BRW_PAGES)
 260                 ret = page_pools.epp_free_pages - PTLRPC_MAX_BRW_PAGES;
 261
 262 out_unlock:
 263         spin_unlock(&page_pools.epp_lock);
 264         return ret;
 265 }
 266
 267 static inline
 268 int npages_to_npools(unsigned long npages)
 269 {
 270         return (int) ((npages + PAGES_PER_POOL - 1) / PAGES_PER_POOL);
 271 }
 272
 273 /*
 274  * return how many pages cleaned up.
 275  */
 276 static unsigned long enc_pools_cleanup(cfs_page_t ***pools, int npools)
 277 {
 278         unsigned long cleaned = 0;
 279         int           i, j;
 280
 281         for (i = 0; i < npools; i++) {
 282                 if (pools[i]) {
 283                         for (j = 0; j < PAGES_PER_POOL; j++) {
 284                                 if (pools[i][j]) {
 285                                         cfs_free_page(pools[i][j]);
 286                                         cleaned++;
 287                                 }
 288                         }
 289                         OBD_FREE(pools[i], CFS_PAGE_SIZE);
 290                         pools[i] = NULL;
 291                 }
 292         }
 293
 294         return cleaned;
 295 }
 296
 297 /*
 298  * merge @npools pointed by @pools which contains @npages new pages
 299  * into current pools.
 300  *
 301  * we have options to avoid most memory copy with some tricks. but we choose
 302  * the simplest way to avoid complexity. It's not frequently called.
 303  */
 304 static void enc_pools_insert(cfs_page_t ***pools, int npools, int npages)
 305 {
 306         int     freeslot;
 307         int     op_idx, np_idx, og_idx, ng_idx;
 308         int     cur_npools, end_npools;
 309
 310         LASSERT(npages > 0);
 311         LASSERT(page_pools.epp_total_pages+npages <= page_pools.epp_max_pages);
 312         LASSERT(npages_to_npools(npages) == npools);
 313
 314         spin_lock(&page_pools.epp_lock);
 315
 316         /*
 317          * (1) fill all the free slots of current pools.
 318          */
 319         /* free slots are those left by rent pages, and the extra ones with
 320          * index >= total_pages, locate at the tail of last pool. */
 321         freeslot = page_pools.epp_total_pages % PAGES_PER_POOL;
 322         if (freeslot != 0)
 323                 freeslot = PAGES_PER_POOL - freeslot;
 324         freeslot += page_pools.epp_total_pages - page_pools.epp_free_pages;
 325
 326         op_idx = page_pools.epp_free_pages / PAGES_PER_POOL;
 327         og_idx = page_pools.epp_free_pages % PAGES_PER_POOL;
 328         np_idx = npools - 1;
 329         ng_idx = (npages - 1) % PAGES_PER_POOL;
 330
 331         while (freeslot) {
 332                 LASSERT(page_pools.epp_pools[op_idx][og_idx] == NULL);
 333                 LASSERT(pools[np_idx][ng_idx] != NULL);
 334
 335                 page_pools.epp_pools[op_idx][og_idx] = pools[np_idx][ng_idx];
 336                 pools[np_idx][ng_idx] = NULL;
 337
 338                 freeslot--;
 339
 340                 if (++og_idx == PAGES_PER_POOL) {
 341                         op_idx++;
 342                         og_idx = 0;
 343                 }
 344                 if (--ng_idx < 0) {
 345                         if (np_idx == 0)
 346                                 break;
 347                         np_idx--;
 348                         ng_idx = PAGES_PER_POOL - 1;
 349                 }
 350         }
 351
 352         /*
 353          * (2) add pools if needed.
 354          */
 355         cur_npools = (page_pools.epp_total_pages + PAGES_PER_POOL - 1) /
 356                      PAGES_PER_POOL;
 357         end_npools = (page_pools.epp_total_pages + npages + PAGES_PER_POOL -1) /
 358                      PAGES_PER_POOL;
 359         LASSERT(end_npools <= page_pools.epp_max_pools);
 360
 361         np_idx = 0;
 362         while (cur_npools < end_npools) {
 363                 LASSERT(page_pools.epp_pools[cur_npools] == NULL);
 364                 LASSERT(np_idx < npools);
 365                 LASSERT(pools[np_idx] != NULL);
 366
 367                 page_pools.epp_pools[cur_npools++] = pools[np_idx];
 368                 pools[np_idx++] = NULL;
 369         }
 370
 371         page_pools.epp_total_pages += npages;
 372         page_pools.epp_free_pages += npages;
 373         page_pools.epp_st_lowfree = page_pools.epp_free_pages;
 374
 375         if (page_pools.epp_total_pages > page_pools.epp_st_max_pages)
 376                 page_pools.epp_st_max_pages = page_pools.epp_total_pages;
 377
 378         CDEBUG(D_SEC, "add %d pages to total %lu\n", npages,
 379                page_pools.epp_total_pages);
 380
 381         spin_unlock(&page_pools.epp_lock);
 382 }
 383
 384 static int enc_pools_add_pages(int npages)
 385 {
 386         static DECLARE_MUTEX(sem_add_pages);
 387         cfs_page_t   ***pools;
 388         int             npools, alloced = 0;
 389         int             i, j, rc = -ENOMEM;
 390
 391         if (npages < PTLRPC_MAX_BRW_PAGES)
 392                 npages = PTLRPC_MAX_BRW_PAGES;
 393
 394         down(&sem_add_pages);
 395
 396         if (npages + page_pools.epp_total_pages > page_pools.epp_max_pages)
 397                 npages = page_pools.epp_max_pages - page_pools.epp_total_pages;
 398         LASSERT(npages > 0);
 399
 400         page_pools.epp_st_grows++;
 401
 402         npools = npages_to_npools(npages);
 403         OBD_ALLOC(pools, npools * sizeof(*pools));
 404         if (pools == NULL)
 405                 goto out;
 406
 407         for (i = 0; i < npools; i++) {
 408                 OBD_ALLOC(pools[i], CFS_PAGE_SIZE);
 409                 if (pools[i] == NULL)
 410                         goto out_pools;
 411
 412                 for (j = 0; j < PAGES_PER_POOL && alloced < npages; j++) {
 413                         pools[i][j] = cfs_alloc_page(CFS_ALLOC_IO |
 414                                                      CFS_ALLOC_HIGH);
 415                         if (pools[i][j] == NULL)
 416                                 goto out_pools;
 417
 418                         alloced++;
 419                 }
 420         }
 421
 422         enc_pools_insert(pools, npools, npages);
 423         CDEBUG(D_SEC, "added %d pages into pools\n", npages);
 424         rc = 0;
 425
 426 out_pools:
 427         enc_pools_cleanup(pools, npools);
 428         OBD_FREE(pools, npools * sizeof(*pools));
 429 out:
 430         if (rc) {
 431                 page_pools.epp_st_grow_fails++;
 432                 CERROR("Failed to allocate %d enc pages\n", npages);
 433         }
 434
 435         up(&sem_add_pages);
 436         return rc;
 437 }
 438
 439 static inline void enc_pools_wakeup(void)
 440 {
 441         if (unlikely(page_pools.epp_waitqlen)) {
 442                 LASSERT(page_pools.epp_waitqlen > 0);
 443                 LASSERT(cfs_waitq_active(&page_pools.epp_waitq));
 444                 cfs_waitq_broadcast(&page_pools.epp_waitq);
 445         }
 446 }
 447
 448 static int enc_pools_should_grow(int page_needed, long now)
 449 {
 450         /* don't grow if someone else is growing the pools right now,
 451          * or the pools has reached its full capacity
 452          */
 453         if (page_pools.epp_growing ||
 454             page_pools.epp_total_pages == page_pools.epp_max_pages)
 455                 return 0;
 456
 457         /* if total pages is not enough, we need to grow */
 458         if (page_pools.epp_total_pages < page_needed)
 459                 return 1;
 460
 461         /* if we just did a shrink due to memory tight, we'd better
 462          * wait a while to grow again.
 463          */
 464         if (now - page_pools.epp_last_shrink < 2)
 465                 return 0;
 466
 467         /*
 468          * here we perhaps need consider other factors like wait queue
 469          * length, idle index, etc. ?
 470          */
 471
 472         /* grow the pools in any other cases */
 473         return 1;
 474 }
 475
 476 /*
 477  * we allocate the requested pages atomically.
 478  */
 479 int sptlrpc_enc_pool_get_pages(struct ptlrpc_bulk_desc *desc)
 480 {
 481         cfs_waitlink_t  waitlink;
 482         unsigned long   this_idle = -1;
 483         cfs_time_t      tick = 0;
 484         long            now;
 485         int             p_idx, g_idx;
 486         int             i;
 487
 488         LASSERT(desc->bd_max_iov > 0);
 489         LASSERT(desc->bd_max_iov <= page_pools.epp_max_pages);
 490
 491         /* resent bulk, enc pages might have been allocated previously */
 492         if (desc->bd_enc_pages != NULL)
 493                 return 0;
 494
 495         OBD_ALLOC(desc->bd_enc_pages,
 496                   desc->bd_max_iov * sizeof(*desc->bd_enc_pages));
 497         if (desc->bd_enc_pages == NULL)
 498                 return -ENOMEM;
 499
 500         spin_lock(&page_pools.epp_lock);
 501
 502         page_pools.epp_st_access++;
 503 again:
 504         if (unlikely(page_pools.epp_free_pages < desc->bd_max_iov)) {
 505                 if (tick == 0)
 506                         tick = cfs_time_current();
 507
 508                 now = cfs_time_current_sec();
 509
 510                 page_pools.epp_st_missings++;
 511                 page_pools.epp_pages_short += desc->bd_max_iov;
 512
 513                 if (enc_pools_should_grow(desc->bd_max_iov, now)) {
 514                         page_pools.epp_growing = 1;
 515
 516                         spin_unlock(&page_pools.epp_lock);
 517                         enc_pools_add_pages(page_pools.epp_pages_short / 2);
 518                         spin_lock(&page_pools.epp_lock);
 519
 520                         page_pools.epp_growing = 0;
 521                 } else {
 522                         if (++page_pools.epp_waitqlen >
 523                             page_pools.epp_st_max_wqlen)
 524                                 page_pools.epp_st_max_wqlen =
 525                                                 page_pools.epp_waitqlen;
 526
 527                         set_current_state(TASK_UNINTERRUPTIBLE);
 528                         cfs_waitlink_init(&waitlink);
 529                         cfs_waitq_add(&page_pools.epp_waitq, &waitlink);
 530
 531                         spin_unlock(&page_pools.epp_lock);
 532                         cfs_schedule();
 533                         spin_lock(&page_pools.epp_lock);
 534
 535                         LASSERT(page_pools.epp_waitqlen > 0);
 536                         page_pools.epp_waitqlen--;
 537                 }
 538
 539                 LASSERT(page_pools.epp_pages_short >= desc->bd_max_iov);
 540                 page_pools.epp_pages_short -= desc->bd_max_iov;
 541
 542                 this_idle = 0;
 543                 goto again;
 544         }
 545
 546         /* record max wait time */
 547         if (unlikely(tick != 0)) {
 548                 tick = cfs_time_current() - tick;
 549                 if (tick > page_pools.epp_st_max_wait)
 550                         page_pools.epp_st_max_wait = tick;
 551         }
 552
 553         /* proceed with rest of allocation */
 554         page_pools.epp_free_pages -= desc->bd_max_iov;
 555
 556         p_idx = page_pools.epp_free_pages / PAGES_PER_POOL;
 557         g_idx = page_pools.epp_free_pages % PAGES_PER_POOL;
 558
 559         for (i = 0; i < desc->bd_max_iov; i++) {
 560                 LASSERT(page_pools.epp_pools[p_idx][g_idx] != NULL);
 561                 desc->bd_enc_pages[i] = page_pools.epp_pools[p_idx][g_idx];
 562                 page_pools.epp_pools[p_idx][g_idx] = NULL;
 563
 564                 if (++g_idx == PAGES_PER_POOL) {
 565                         p_idx++;
 566                         g_idx = 0;
 567                 }
 568         }
 569
 570         if (page_pools.epp_free_pages < page_pools.epp_st_lowfree)
 571                 page_pools.epp_st_lowfree = page_pools.epp_free_pages;
 572
 573         /*
 574          * new idle index = (old * weight + new) / (weight + 1)
 575          */
 576         if (this_idle == -1) {
 577                 this_idle = page_pools.epp_free_pages * IDLE_IDX_MAX /
 578                             page_pools.epp_total_pages;
 579         }
 580         page_pools.epp_idle_idx = (page_pools.epp_idle_idx * IDLE_IDX_WEIGHT +
 581                                    this_idle) /
 582                                   (IDLE_IDX_WEIGHT + 1);
 583
 584         page_pools.epp_last_access = cfs_time_current_sec();
 585
 586         spin_unlock(&page_pools.epp_lock);
 587         return 0;
 588 }
 589 EXPORT_SYMBOL(sptlrpc_enc_pool_get_pages);
 590
 591 void sptlrpc_enc_pool_put_pages(struct ptlrpc_bulk_desc *desc)
 592 {
 593         int     p_idx, g_idx;
 594         int     i;
 595
 596         if (desc->bd_enc_pages == NULL)
 597                 return;
 598         if (desc->bd_max_iov == 0)
 599                 return;
 600
 601         spin_lock(&page_pools.epp_lock);
 602
 603         p_idx = page_pools.epp_free_pages / PAGES_PER_POOL;
 604         g_idx = page_pools.epp_free_pages % PAGES_PER_POOL;
 605
 606         LASSERT(page_pools.epp_free_pages + desc->bd_max_iov <=
 607                 page_pools.epp_total_pages);
 608         LASSERT(page_pools.epp_pools[p_idx]);
 609
 610         for (i = 0; i < desc->bd_max_iov; i++) {
 611                 LASSERT(desc->bd_enc_pages[i] != NULL);
 612                 LASSERT(g_idx != 0 || page_pools.epp_pools[p_idx]);
 613                 LASSERT(page_pools.epp_pools[p_idx][g_idx] == NULL);
 614
 615                 page_pools.epp_pools[p_idx][g_idx] = desc->bd_enc_pages[i];
 616
 617                 if (++g_idx == PAGES_PER_POOL) {
 618                         p_idx++;
 619                         g_idx = 0;
 620                 }
 621         }
 622
 623         page_pools.epp_free_pages += desc->bd_max_iov;
 624
 625         enc_pools_wakeup();
 626
 627         spin_unlock(&page_pools.epp_lock);
 628
 629         OBD_FREE(desc->bd_enc_pages,
 630                  desc->bd_max_iov * sizeof(*desc->bd_enc_pages));
 631         desc->bd_enc_pages = NULL;
 632 }
 633 EXPORT_SYMBOL(sptlrpc_enc_pool_put_pages);
 634
 635 /*
 636  * we don't do much stuff for add_user/del_user anymore, except adding some
 637  * initial pages in add_user() if current pools are empty, rest would be
 638  * handled by the pools's self-adaption.
 639  */
 640 int sptlrpc_enc_pool_add_user(void)
 641 {
 642         int     need_grow = 0;
 643
 644         spin_lock(&page_pools.epp_lock);
 645         if (page_pools.epp_growing == 0 && page_pools.epp_total_pages == 0) {
 646                 page_pools.epp_growing = 1;
 647                 need_grow = 1;
 648         }
 649         spin_unlock(&page_pools.epp_lock);
 650
 651         if (need_grow) {
 652                 enc_pools_add_pages(PTLRPC_MAX_BRW_PAGES);
 653
 654                 spin_lock(&page_pools.epp_lock);
 655                 page_pools.epp_growing = 0;
 656                 enc_pools_wakeup();
 657                 spin_unlock(&page_pools.epp_lock);
 658         }
 659         return 0;
 660 }
 661 EXPORT_SYMBOL(sptlrpc_enc_pool_add_user);
 662
 663 int sptlrpc_enc_pool_del_user(void)
 664 {
 665         return 0;
 666 }
 667 EXPORT_SYMBOL(sptlrpc_enc_pool_del_user);
 668
 669 static inline void enc_pools_alloc(void)
 670 {
 671         LASSERT(page_pools.epp_max_pools);
 672         /*
 673          * on system with huge memory but small page size, this might lead to
 674          * high-order allocation. but it's not common, and we suppose memory
 675          * be not too much fragmented at module loading time.
 676          */
 677         OBD_ALLOC(page_pools.epp_pools,
 678                   page_pools.epp_max_pools * sizeof(*page_pools.epp_pools));
 679 }
 680
 681 static inline void enc_pools_free(void)
 682 {
 683         LASSERT(page_pools.epp_max_pools);
 684         LASSERT(page_pools.epp_pools);
 685
 686         OBD_FREE(page_pools.epp_pools,
 687                  page_pools.epp_max_pools * sizeof(*page_pools.epp_pools));
 688 }
 689
 690 int sptlrpc_enc_pool_init(void)
 691 {
 692         /*
 693          * maximum capacity is 1/8 of total physical memory.
 694          * is the 1/8 a good number?
 695          */
 696         page_pools.epp_max_pages = num_physpages / 8;
 697         page_pools.epp_max_pools = npages_to_npools(page_pools.epp_max_pages);
 698
 699         cfs_waitq_init(&page_pools.epp_waitq);
 700         page_pools.epp_waitqlen = 0;
 701         page_pools.epp_pages_short = 0;
 702
 703         page_pools.epp_growing = 0;
 704
 705         page_pools.epp_idle_idx = 0;
 706         page_pools.epp_last_shrink = cfs_time_current_sec();
 707         page_pools.epp_last_access = cfs_time_current_sec();
 708
 709         spin_lock_init(&page_pools.epp_lock);
 710         page_pools.epp_total_pages = 0;
 711         page_pools.epp_free_pages = 0;
 712
 713         page_pools.epp_st_max_pages = 0;
 714         page_pools.epp_st_grows = 0;
 715         page_pools.epp_st_grow_fails = 0;
 716         page_pools.epp_st_shrinks = 0;
 717         page_pools.epp_st_access = 0;
 718         page_pools.epp_st_missings = 0;
 719         page_pools.epp_st_lowfree = 0;
 720         page_pools.epp_st_max_wqlen = 0;
 721         page_pools.epp_st_max_wait = 0;
 722
 723         enc_pools_alloc();
 724         if (page_pools.epp_pools == NULL)
 725                 return -ENOMEM;
 726
 727         pools_shrinker = set_shrinker(pools_shrinker_seeks, enc_pools_shrink);
 728         if (pools_shrinker == NULL) {
 729                 enc_pools_free();
 730                 return -ENOMEM;
 731         }
 732
 733         return 0;
 734 }
 735
 736 void sptlrpc_enc_pool_fini(void)
 737 {
 738         unsigned long cleaned, npools;
 739
 740         LASSERT(pools_shrinker);
 741         LASSERT(page_pools.epp_pools);
 742         LASSERT(page_pools.epp_total_pages == page_pools.epp_free_pages);
 743
 744         remove_shrinker(pools_shrinker);
 745
 746         npools = npages_to_npools(page_pools.epp_total_pages);
 747         cleaned = enc_pools_cleanup(page_pools.epp_pools, npools);
 748         LASSERT(cleaned == page_pools.epp_total_pages);
 749
 750         enc_pools_free();
 751
 752         if (page_pools.epp_st_access > 0) {
 753                 CWARN("max pages %lu, grows %u, grow fails %u, shrinks %u, "
 754                       "access %lu, missing %lu, max qlen %u, max wait "
 755                       CFS_TIME_T"/%d\n",
 756                       page_pools.epp_st_max_pages, page_pools.epp_st_grows,
 757                       page_pools.epp_st_grow_fails,
 758                       page_pools.epp_st_shrinks, page_pools.epp_st_access,
 759                       page_pools.epp_st_missings, page_pools.epp_st_max_wqlen,
 760                       page_pools.epp_st_max_wait, HZ);
 761         }
 762 }
 763
 764 #else /* !__KERNEL__ */
 765
 766 int sptlrpc_enc_pool_get_pages(struct ptlrpc_bulk_desc *desc)
 767 {
 768         return 0;
 769 }
 770
 771 void sptlrpc_enc_pool_put_pages(struct ptlrpc_bulk_desc *desc)
 772 {
 773 }
 774
 775 int sptlrpc_enc_pool_init(void)
 776 {
 777         return 0;
 778 }
 779
 780 void sptlrpc_enc_pool_fini(void)
 781 {
 782 }
 783 #endif
 784
 785 /****************************************
 786  * Helpers to assist policy modules to  *
 787  * implement checksum funcationality    *
 788  ****************************************/
 789
 790 static struct sptlrpc_hash_type hash_types[] = {
 791         [BULK_HASH_ALG_NULL]    = { "null",     "null",         0 },
 792         [BULK_HASH_ALG_ADLER32] = { "adler32",  "adler32",      4 },
 793         [BULK_HASH_ALG_CRC32]   = { "crc32",    "crc32",        4 },
 794         [BULK_HASH_ALG_MD5]     = { "md5",      "md5",          16 },
 795         [BULK_HASH_ALG_SHA1]    = { "sha1",     "sha1",         20 },
 796         [BULK_HASH_ALG_SHA256]  = { "sha256",   "sha256",       32 },
 797         [BULK_HASH_ALG_SHA384]  = { "sha384",   "sha384",       48 },
 798         [BULK_HASH_ALG_SHA512]  = { "sha512",   "sha512",       64 },
 799         [BULK_HASH_ALG_WP256]   = { "wp256",    "wp256",        32 },
 800         [BULK_HASH_ALG_WP384]   = { "wp384",    "wp384",        48 },
 801         [BULK_HASH_ALG_WP512]   = { "wp512",    "wp512",        64 },
 802 };
 803
 804 const struct sptlrpc_hash_type *sptlrpc_get_hash_type(__u8 hash_alg)
 805 {
 806         struct sptlrpc_hash_type *ht;
 807
 808         if (hash_alg < BULK_HASH_ALG_MAX) {
 809                 ht = &hash_types[hash_alg];
 810                 if (ht->sht_tfm_name)
 811                         return ht;
 812         }
 813         return NULL;
 814 }
 815 EXPORT_SYMBOL(sptlrpc_get_hash_type);
 816
 817 const char * sptlrpc_get_hash_name(__u8 hash_alg)
 818 {
 819         const struct sptlrpc_hash_type *ht;
 820
 821         ht = sptlrpc_get_hash_type(hash_alg);
 822         if (ht)
 823                 return ht->sht_name;
 824         else
 825                 return "unknown";
 826 }
 827 EXPORT_SYMBOL(sptlrpc_get_hash_name);
 828
 829 int bulk_sec_desc_size(__u8 hash_alg, int request, int read)
 830 {
 831         int size = sizeof(struct ptlrpc_bulk_sec_desc);
 832
 833         LASSERT(hash_alg < BULK_HASH_ALG_MAX);
 834
 835         /* read request don't need extra data */
 836         if (!(read && request))
 837                 size += hash_types[hash_alg].sht_size;
 838
 839         return size;
 840 }
 841 EXPORT_SYMBOL(bulk_sec_desc_size);
 842
 843 int bulk_sec_desc_unpack(struct lustre_msg *msg, int offset)
 844 {
 845         struct ptlrpc_bulk_sec_desc *bsd;
 846         int    size = msg->lm_buflens[offset];
 847
 848         bsd = lustre_msg_buf(msg, offset, sizeof(*bsd));
 849         if (bsd == NULL) {
 850                 CERROR("Invalid bulk sec desc: size %d\n", size);
 851                 return -EINVAL;
 852         }
 853
 854         /* nothing to swab */
 855
 856         if (unlikely(bsd->bsd_version != 0)) {
 857                 CERROR("Unexpected version %u\n", bsd->bsd_version);
 858                 return -EPROTO;
 859         }
 860
 861         if (unlikely(bsd->bsd_flags != 0)) {
 862                 CERROR("Unexpected flags %x\n", bsd->bsd_flags);
 863                 return -EPROTO;
 864         }
 865
 866         if (unlikely(!sptlrpc_get_hash_type(bsd->bsd_hash_alg))) {
 867                 CERROR("Unsupported checksum algorithm %u\n",
 868                        bsd->bsd_hash_alg);
 869                 return -EINVAL;
 870         }
 871
 872         if (unlikely(!sptlrpc_get_ciph_type(bsd->bsd_ciph_alg))) {
 873                 CERROR("Unsupported cipher algorithm %u\n",
 874                        bsd->bsd_ciph_alg);
 875                 return -EINVAL;
 876         }
 877
 878         if (unlikely(size > sizeof(*bsd)) &&
 879             size < sizeof(*bsd) + hash_types[bsd->bsd_hash_alg].sht_size) {
 880                 CERROR("Mal-formed checksum data: csum alg %u, size %d\n",
 881                        bsd->bsd_hash_alg, size);
 882                 return -EINVAL;
 883         }
 884
 885         return 0;
 886 }
 887 EXPORT_SYMBOL(bulk_sec_desc_unpack);
 888
 889 #ifdef __KERNEL__
 890
 891 #ifdef HAVE_ADLER
 892 static int do_bulk_checksum_adler32(struct ptlrpc_bulk_desc *desc, void *buf)
 893 {
 894         struct page    *page;
 895         int             off;
 896         char           *ptr;
 897         __u32           adler32 = 1;
 898         int             len, i;
 899
 900         for (i = 0; i < desc->bd_iov_count; i++) {
 901                 page = desc->bd_iov[i].kiov_page;
 902                 off = desc->bd_iov[i].kiov_offset & ~CFS_PAGE_MASK;
 903                 ptr = cfs_kmap(page) + off;
 904                 len = desc->bd_iov[i].kiov_len;
 905
 906                 adler32 = adler32(adler32, ptr, len);
 907
 908                 cfs_kunmap(page);
 909         }
 910
 911         adler32 = cpu_to_le32(adler32);
 912         memcpy(buf, &adler32, sizeof(adler32));
 913         return 0;
 914 }
 915 #endif
 916
 917 static int do_bulk_checksum_crc32(struct ptlrpc_bulk_desc *desc, void *buf)
 918 {
 919         struct page    *page;
 920         int             off;
 921         char           *ptr;
 922         __u32           crc32 = ~0;
 923         int             len, i;
 924
 925         for (i = 0; i < desc->bd_iov_count; i++) {
 926                 page = desc->bd_iov[i].kiov_page;
 927                 off = desc->bd_iov[i].kiov_offset & ~CFS_PAGE_MASK;
 928                 ptr = cfs_kmap(page) + off;
 929                 len = desc->bd_iov[i].kiov_len;
 930
 931                 crc32 = crc32_le(crc32, ptr, len);
 932
 933                 cfs_kunmap(page);
 934         }
 935
 936         crc32 = cpu_to_le32(crc32);
 937         memcpy(buf, &crc32, sizeof(crc32));
 938         return 0;
 939 }
 940
 941 static int do_bulk_checksum(struct ptlrpc_bulk_desc *desc, __u32 alg, void *buf)
 942 {
 943         struct hash_desc    hdesc;
 944         struct scatterlist *sl;
 945         int i, rc = 0, bytes = 0;
 946
 947         LASSERT(alg > BULK_HASH_ALG_NULL &&
 948                 alg < BULK_HASH_ALG_MAX);
 949
 950         switch (alg) {
 951         case BULK_HASH_ALG_ADLER32:
 952 #ifdef HAVE_ADLER
 953                 return do_bulk_checksum_adler32(desc, buf);
 954 #else
 955                 CERROR("Adler32 not supported\n");
 956                 return -EINVAL;
 957 #endif
 958         case BULK_HASH_ALG_CRC32:
 959                 return do_bulk_checksum_crc32(desc, buf);
 960         }
 961
 962         hdesc.tfm = ll_crypto_alloc_hash(hash_types[alg].sht_tfm_name, 0, 0);
 963         if (hdesc.tfm == NULL) {
 964                 CERROR("Unable to allocate TFM %s\n", hash_types[alg].sht_name);
 965                 return -ENOMEM;
 966         }
 967         hdesc.flags = 0;
 968
 969         OBD_ALLOC(sl, sizeof(*sl) * desc->bd_iov_count);
 970         if (sl == NULL) {
 971                 rc = -ENOMEM;
 972                 goto out_tfm;
 973         }
 974
 975         for (i = 0; i < desc->bd_iov_count; i++) {
 976                 sl[i].page = desc->bd_iov[i].kiov_page;
 977                 sl[i].offset = desc->bd_iov[i].kiov_offset & ~CFS_PAGE_MASK;
 978                 sl[i].length = desc->bd_iov[i].kiov_len;
 979                 bytes += desc->bd_iov[i].kiov_len;
 980         }
 981
 982         ll_crypto_hash_init(&hdesc);
 983         ll_crypto_hash_update(&hdesc, sl, bytes);
 984         ll_crypto_hash_final(&hdesc, buf);
 985
 986         OBD_FREE(sl, sizeof(*sl) * desc->bd_iov_count);
 987
 988 out_tfm:
 989         ll_crypto_free_hash(hdesc.tfm);
 990         return rc;
 991 }
 992
 993 #else /* !__KERNEL__ */
 994
 995 static int do_bulk_checksum(struct ptlrpc_bulk_desc *desc, __u32 alg, void *buf)
 996 {
 997         __u32   csum32;
 998         int     i;
 999
1000         LASSERT(alg == BULK_HASH_ALG_ADLER32 || alg == BULK_HASH_ALG_CRC32);
1001
1002         if (alg == BULK_HASH_ALG_ADLER32)
1003                 csum32 = 1;
1004         else
1005                 csum32 = ~0;
1006
1007         for (i = 0; i < desc->bd_iov_count; i++) {
1008                 char *ptr = desc->bd_iov[i].iov_base;
1009                 int len = desc->bd_iov[i].iov_len;
1010
1011                 switch (alg) {
1012                 case BULK_HASH_ALG_ADLER32:
1013 #ifdef HAVE_ADLER
1014                         csum32 = adler32(csum32, ptr, len);
1015 #else
1016                         CERROR("Adler32 not supported\n");
1017                         return -EINVAL;
1018 #endif
1019                         break;
1020                 case BULK_HASH_ALG_CRC32:
1021                         csum32 = crc32_le(csum32, ptr, len);
1022                         break;
1023                 }
1024         }
1025
1026         csum32 = cpu_to_le32(csum32);
1027         memcpy(buf, &csum32, sizeof(csum32));
1028         return 0;
1029 }
1030
1031 #endif /* __KERNEL__ */
1032
1033 /*
1034  * perform algorithm @alg checksum on @desc, store result in @buf.
1035  * if anything goes wrong, leave 'alg' be BULK_HASH_ALG_NULL.
1036  */
1037 static
1038 int generate_bulk_csum(struct ptlrpc_bulk_desc *desc, __u32 alg,
1039                        struct ptlrpc_bulk_sec_desc *bsd, int bsdsize)
1040 {
1041         int rc;
1042
1043         LASSERT(bsd);
1044         LASSERT(alg < BULK_HASH_ALG_MAX);
1045
1046         bsd->bsd_hash_alg = BULK_HASH_ALG_NULL;
1047
1048         if (alg == BULK_HASH_ALG_NULL)
1049                 return 0;
1050
1051         LASSERT(bsdsize >= sizeof(*bsd) + hash_types[alg].sht_size);
1052
1053         rc = do_bulk_checksum(desc, alg, bsd->bsd_csum);
1054         if (rc == 0)
1055                 bsd->bsd_hash_alg = alg;
1056
1057         return rc;
1058 }
1059
1060 static
1061 int verify_bulk_csum(struct ptlrpc_bulk_desc *desc, int read,
1062                      struct ptlrpc_bulk_sec_desc *bsdv, int bsdvsize,
1063                      struct ptlrpc_bulk_sec_desc *bsdr, int bsdrsize)
1064 {
1065         char *csum_p;
1066         char *buf = NULL;
1067         int   csum_size, rc = 0;
1068
1069         LASSERT(bsdv);
1070         LASSERT(bsdv->bsd_hash_alg < BULK_HASH_ALG_MAX);
1071
1072         if (bsdr)
1073                 bsdr->bsd_hash_alg = BULK_HASH_ALG_NULL;
1074
1075         if (bsdv->bsd_hash_alg == BULK_HASH_ALG_NULL)
1076                 return 0;
1077
1078         /* for all supported algorithms */
1079         csum_size = hash_types[bsdv->bsd_hash_alg].sht_size;
1080
1081         if (bsdvsize < sizeof(*bsdv) + csum_size) {
1082                 CERROR("verifier size %d too small, require %d\n",
1083                        bsdvsize, (int) sizeof(*bsdv) + csum_size);
1084                 return -EINVAL;
1085         }
1086
1087         if (bsdr) {
1088                 LASSERT(bsdrsize >= sizeof(*bsdr) + csum_size);
1089                 csum_p = (char *) bsdr->bsd_csum;
1090         } else {
1091                 OBD_ALLOC(buf, csum_size);
1092                 if (buf == NULL)
1093                         return -EINVAL;
1094                 csum_p = buf;
1095         }
1096
1097         rc = do_bulk_checksum(desc, bsdv->bsd_hash_alg, csum_p);
1098
1099         if (memcmp(bsdv->bsd_csum, csum_p, csum_size)) {
1100                 CERROR("BAD %s CHECKSUM (%s), data mutated during "
1101                        "transfer!\n", read ? "READ" : "WRITE",
1102                        hash_types[bsdv->bsd_hash_alg].sht_name);
1103                 rc = -EINVAL;
1104         } else {
1105                 CDEBUG(D_SEC, "bulk %s checksum (%s) verified\n",
1106                       read ? "read" : "write",
1107                       hash_types[bsdv->bsd_hash_alg].sht_name);
1108         }
1109
1110         if (bsdr) {
1111                 bsdr->bsd_hash_alg = bsdv->bsd_hash_alg;
1112                 memcpy(bsdr->bsd_csum, csum_p, csum_size);
1113         } else {
1114                 LASSERT(buf);
1115                 OBD_FREE(buf, csum_size);
1116         }
1117
1118         return rc;
1119 }
1120
1121 int bulk_csum_cli_request(struct ptlrpc_bulk_desc *desc, int read,
1122                           __u32 alg, struct lustre_msg *rmsg, int roff)
1123 {
1124         struct ptlrpc_bulk_sec_desc *bsdr;
1125         int    rsize, rc = 0;
1126
1127         rsize = rmsg->lm_buflens[roff];
1128         bsdr = lustre_msg_buf(rmsg, roff, sizeof(*bsdr));
1129
1130         LASSERT(bsdr);
1131         LASSERT(rsize >= sizeof(*bsdr));
1132         LASSERT(alg < BULK_HASH_ALG_MAX);
1133
1134         if (read) {
1135                 bsdr->bsd_hash_alg = alg;
1136         } else {
1137                 rc = generate_bulk_csum(desc, alg, bsdr, rsize);
1138                 if (rc)
1139                         CERROR("bulk write: client failed to compute "
1140                                "checksum: %d\n", rc);
1141
1142                 /* For sending we only compute the wrong checksum instead
1143                  * of corrupting the data so it is still correct on a redo */
1144                 if (rc == 0 && OBD_FAIL_CHECK(OBD_FAIL_OSC_CHECKSUM_SEND) &&
1145                     bsdr->bsd_hash_alg != BULK_HASH_ALG_NULL)
1146                         bsdr->bsd_csum[0] ^= 0x1;
1147         }
1148
1149         return rc;
1150 }
1151 EXPORT_SYMBOL(bulk_csum_cli_request);
1152
1153 int bulk_csum_cli_reply(struct ptlrpc_bulk_desc *desc, int read,
1154                         struct lustre_msg *rmsg, int roff,
1155                         struct lustre_msg *vmsg, int voff)
1156 {
1157         struct ptlrpc_bulk_sec_desc *bsdv, *bsdr;
1158         int    rsize, vsize;
1159
1160         rsize = rmsg->lm_buflens[roff];
1161         vsize = vmsg->lm_buflens[voff];
1162         bsdr = lustre_msg_buf(rmsg, roff, 0);
1163         bsdv = lustre_msg_buf(vmsg, voff, 0);
1164
1165         if (bsdv == NULL || vsize < sizeof(*bsdv)) {
1166                 CERROR("Invalid checksum verifier from server: size %d\n",
1167                        vsize);
1168                 return -EINVAL;
1169         }
1170
1171         LASSERT(bsdr);
1172         LASSERT(rsize >= sizeof(*bsdr));
1173         LASSERT(vsize >= sizeof(*bsdv));
1174
1175         if (bsdr->bsd_hash_alg != bsdv->bsd_hash_alg) {
1176                 CERROR("bulk %s: checksum algorithm mismatch: client request "
1177                        "%s but server reply with %s. try to use the new one "
1178                        "for checksum verification\n",
1179                        read ? "read" : "write",
1180                        hash_types[bsdr->bsd_hash_alg].sht_name,
1181                        hash_types[bsdv->bsd_hash_alg].sht_name);
1182         }
1183
1184         if (read)
1185                 return verify_bulk_csum(desc, 1, bsdv, vsize, NULL, 0);
1186         else {
1187                 char *cli, *srv, *new = NULL;
1188                 int csum_size = hash_types[bsdr->bsd_hash_alg].sht_size;
1189
1190                 LASSERT(bsdr->bsd_hash_alg < BULK_HASH_ALG_MAX);
1191                 if (bsdr->bsd_hash_alg == BULK_HASH_ALG_NULL)
1192                         return 0;
1193
1194                 if (vsize < sizeof(*bsdv) + csum_size) {
1195                         CERROR("verifier size %d too small, require %d\n",
1196                                vsize, (int) sizeof(*bsdv) + csum_size);
1197                         return -EINVAL;
1198                 }
1199
1200                 cli = (char *) (bsdr + 1);
1201                 srv = (char *) (bsdv + 1);
1202
1203                 if (!memcmp(cli, srv, csum_size)) {
1204                         /* checksum confirmed */
1205                         CDEBUG(D_SEC, "bulk write checksum (%s) confirmed\n",
1206                                hash_types[bsdr->bsd_hash_alg].sht_name);
1207                         return 0;
1208                 }
1209
1210                 /* checksum mismatch, re-compute a new one and compare with
1211                  * others, give out proper warnings. */
1212                 OBD_ALLOC(new, csum_size);
1213                 if (new == NULL)
1214                         return -ENOMEM;
1215
1216                 do_bulk_checksum(desc, bsdr->bsd_hash_alg, new);
1217
1218                 if (!memcmp(new, srv, csum_size)) {
1219                         CERROR("BAD WRITE CHECKSUM (%s): pages were mutated "
1220                                "on the client after we checksummed them\n",
1221                                hash_types[bsdr->bsd_hash_alg].sht_name);
1222                 } else if (!memcmp(new, cli, csum_size)) {
1223                         CERROR("BAD WRITE CHECKSUM (%s): pages were mutated "
1224                                "in transit\n",
1225                                hash_types[bsdr->bsd_hash_alg].sht_name);
1226                 } else {
1227                         CERROR("BAD WRITE CHECKSUM (%s): pages were mutated "
1228                                "in transit, and the current page contents "
1229                                "don't match the originals and what the server "
1230                                "received\n",
1231                                hash_types[bsdr->bsd_hash_alg].sht_name);
1232                 }
1233                 OBD_FREE(new, csum_size);
1234
1235                 return -EINVAL;
1236         }
1237 }
1238 EXPORT_SYMBOL(bulk_csum_cli_reply);
1239
1240 #ifdef __KERNEL__
1241 static void corrupt_bulk_data(struct ptlrpc_bulk_desc *desc)
1242 {
1243         char           *ptr;
1244         unsigned int    off, i;
1245
1246         for (i = 0; i < desc->bd_iov_count; i++) {
1247                 if (desc->bd_iov[i].kiov_len == 0)
1248                         continue;
1249
1250                 ptr = cfs_kmap(desc->bd_iov[i].kiov_page);
1251                 off = desc->bd_iov[i].kiov_offset & ~CFS_PAGE_MASK;
1252                 ptr[off] ^= 0x1;
1253                 cfs_kunmap(desc->bd_iov[i].kiov_page);
1254                 return;
1255         }
1256 }
1257 #else
1258 static void corrupt_bulk_data(struct ptlrpc_bulk_desc *desc)
1259 {
1260 }
1261 #endif /* __KERNEL__ */
1262
1263 int bulk_csum_svc(struct ptlrpc_bulk_desc *desc, int read,
1264                   struct ptlrpc_bulk_sec_desc *bsdv, int vsize,
1265                   struct ptlrpc_bulk_sec_desc *bsdr, int rsize)
1266 {
1267         int    rc;
1268
1269         LASSERT(vsize >= sizeof(*bsdv));
1270         LASSERT(rsize >= sizeof(*bsdr));
1271         LASSERT(bsdv && bsdr);
1272
1273         if (read) {
1274                 rc = generate_bulk_csum(desc, bsdv->bsd_hash_alg, bsdr, rsize);
1275                 if (rc)
1276                         CERROR("bulk read: server failed to generate %s "
1277                                "checksum: %d\n",
1278                                hash_types[bsdv->bsd_hash_alg].sht_name, rc);
1279
1280                 /* corrupt the data after we compute the checksum, to
1281                  * simulate an OST->client data error */
1282                 if (rc == 0 && OBD_FAIL_CHECK(OBD_FAIL_OSC_CHECKSUM_RECEIVE))
1283                         corrupt_bulk_data(desc);
1284         } else {
1285                 rc = verify_bulk_csum(desc, 0, bsdv, vsize, bsdr, rsize);
1286         }
1287
1288         return rc;
1289 }
1290 EXPORT_SYMBOL(bulk_csum_svc);
1291
1292 /****************************************
1293  * Helpers to assist policy modules to  *
1294  * implement encryption funcationality  *
1295  ****************************************/
1296
1297 /* FIXME */
1298 #ifndef __KERNEL__
1299 #define CRYPTO_TFM_MODE_ECB     (0)
1300 #define CRYPTO_TFM_MODE_CBC     (1)
1301 #endif
1302
1303 static struct sptlrpc_ciph_type cipher_types[] = {
1304         [BULK_CIPH_ALG_NULL]    = {
1305                 "null",         "null",       0,                   0,  0
1306         },
1307         [BULK_CIPH_ALG_ARC4]    = {
1308                 "arc4",         "ecb(arc4)",       0, 0,  16
1309         },
1310         [BULK_CIPH_ALG_AES128]  = {
1311                 "aes128",       "cbc(aes)",        0, 16, 16
1312         },
1313         [BULK_CIPH_ALG_AES192]  = {
1314                 "aes192",       "cbc(aes)",        0, 16, 24
1315         },
1316         [BULK_CIPH_ALG_AES256]  = {
1317                 "aes256",       "cbc(aes)",        0, 16, 32
1318         },
1319         [BULK_CIPH_ALG_CAST128] = {
1320                 "cast128",      "cbc(cast5)",      0, 8,  16
1321         },
1322         [BULK_CIPH_ALG_CAST256] = {
1323                 "cast256",      "cbc(cast6)",      0, 16, 32
1324         },
1325         [BULK_CIPH_ALG_TWOFISH128] = {
1326                 "twofish128",   "cbc(twofish)",    0, 16, 16
1327         },
1328         [BULK_CIPH_ALG_TWOFISH256] = {
1329                 "twofish256",   "cbc(twofish)",    0, 16, 32
1330         },
1331 };
1332
1333 const struct sptlrpc_ciph_type *sptlrpc_get_ciph_type(__u8 ciph_alg)
1334 {
1335         struct sptlrpc_ciph_type *ct;
1336
1337         if (ciph_alg < BULK_CIPH_ALG_MAX) {
1338                 ct = &cipher_types[ciph_alg];
1339                 if (ct->sct_tfm_name)
1340                         return ct;
1341         }
1342         return NULL;
1343 }
1344 EXPORT_SYMBOL(sptlrpc_get_ciph_type);
1345
1346 const char *sptlrpc_get_ciph_name(__u8 ciph_alg)
1347 {
1348         const struct sptlrpc_ciph_type *ct;
1349
1350         ct = sptlrpc_get_ciph_type(ciph_alg);
1351         if (ct)
1352                 return ct->sct_name;
1353         else
1354                 return "unknown";
1355 }
1356 EXPORT_SYMBOL(sptlrpc_get_ciph_name);