src/WINNT/afsd/cm_buf.c

   1 /*
   2  * Copyright 2000, International Business Machines Corporation and others.
   3  * All Rights Reserved.
   4  *
   5  * This software has been released under the terms of the IBM Public
   6  * License.  For details, see the LICENSE file in the top-level source
   7  * directory or online at http://www.openafs.org/dl/license10.html
   8  */
   9
  10 /* Copyright (C) 1994 Cazamar Systems, Inc. */
  11
  12 #include <afs/param.h>
  13 #include <afs/stds.h>
  14
  15 #include <windows.h>
  16 #include <osi.h>
  17 #include <stdio.h>
  18 #include <assert.h>
  19 #include <strsafe.h>
  20 #include <math.h>
  21
  22 #include "afsd.h"
  23 #include "cm_memmap.h"
  24
  25 #ifdef DEBUG
  26 #define TRACE_BUFFER 1
  27 #endif
  28
  29 extern void afsi_log(char *pattern, ...);
  30
  31 /* This module implements the buffer package used by the local transaction
  32  * system (cm).  It is initialized by calling cm_Init, which calls buf_Init;
  33  * it must be initalized before any of its main routines are called.
  34  *
  35  * Each buffer is hashed into a hash table by file ID and offset, and if its
  36  * reference count is zero, it is also in a free list.
  37  *
  38  * There are two locks involved in buffer processing.  The global lock
  39  * buf_globalLock protects all of the global variables defined in this module,
  40  * the reference counts and hash pointers in the actual cm_buf_t structures,
  41  * and the LRU queue pointers in the buffer structures.
  42  *
  43  * The mutexes in the buffer structures protect the remaining fields in the
  44  * buffers, as well the data itself.
  45  *
  46  * The locking hierarchy here is this:
  47  *
  48  * - resv multiple simul. buffers reservation
  49  * - lock buffer I/O flags
  50  * - lock buffer's mutex
  51  * - lock buf_globalLock
  52  *
  53  */
  54
  55 /* global debugging log */
  56 osi_log_t *buf_logp = NULL;
  57
  58 /* Global lock protecting hash tables and free lists */
  59 osi_rwlock_t buf_globalLock;
  60
  61 /* ptr to head of the free list (most recently used) and the
  62  * tail (the guy to remove first).  We use osi_Q* functions
  63  * to put stuff in buf_freeListp, and maintain the end
  64  * pointer manually
  65  */
  66
  67 /* a pointer to a list of all buffers, just so that we can find them
  68  * easily for debugging, and for the incr syncer.  Locked under
  69  * the global lock.
  70  */
  71
  72 /* defaults setup; these variables may be manually assigned into
  73  * before calling cm_Init, as a way of changing these defaults.
  74  */
  75
  76 /* callouts for reading and writing data, etc */
  77 cm_buf_ops_t *cm_buf_opsp;
  78
  79 #ifdef DISKCACHE95
  80 /* for experimental disk caching support in Win95 client */
  81 cm_buf_t *buf_diskFreeListp;
  82 cm_buf_t *buf_diskFreeListEndp;
  83 cm_buf_t *buf_diskAllp;
  84 extern int cm_diskCacheEnabled;
  85 #endif /* DISKCACHE95 */
  86
  87 /* set this to 1 when we are terminating to prevent access attempts */
  88 static int buf_ShutdownFlag = 0;
  89
  90 /* hold a reference to an already held buffer */
  91 void buf_Hold(cm_buf_t *bp)
  92 {
  93     osi_assert(bp->magic == CM_BUF_MAGIC);
  94     lock_ObtainWrite(&buf_globalLock);
  95     bp->refCount++;
  96     lock_ReleaseWrite(&buf_globalLock);
  97 }
  98
  99 /* incremental sync daemon.  Writes 1/10th of all the buffers every 5000 ms */
 100 void buf_IncrSyncer(long parm)
 101 {
 102     cm_buf_t *bp;                       /* buffer we're hacking on; held */
 103     long i;                             /* counter */
 104     long nAtOnce;                       /* how many to do at once */
 105     cm_req_t req;
 106
 107     lock_ObtainWrite(&buf_globalLock);
 108     bp = cm_data.buf_allp;
 109     bp->refCount++;
 110     lock_ReleaseWrite(&buf_globalLock);
 111     nAtOnce = (long)sqrt((double)cm_data.buf_nbuffers);
 112     while (buf_ShutdownFlag == 0) {
 113         i = SleepEx(5000, 1);
 114         if (i != 0) continue;
 115
 116         if (buf_ShutdownFlag == 1)
 117             return;
 118
 119         /* now go through our percentage of the buffers */
 120         for (i=0; i<nAtOnce; i++) {
 121             /* don't want its identity changing while we're
 122              * messing with it, so must do all of this with
 123              * bp held.
 124              */
 125
 126             /* start cleaning the buffer; don't touch log pages since
 127              * the log code counts on knowing exactly who is writing
 128              * a log page at any given instant.
 129              */
 130             cm_InitReq(&req);
 131             req.flags |= CM_REQ_NORETRY;
 132             buf_CleanAsync(bp, &req);
 133
 134             /* now advance to the next buffer; the allp chain never changes,
 135              * and so can be followed even when holding no locks.
 136              */
 137             lock_ObtainWrite(&buf_globalLock);
 138             buf_LockedRelease(bp);
 139             bp = bp->allp;
 140             if (!bp)
 141                 bp = cm_data.buf_allp;
 142             bp->refCount++;
 143             lock_ReleaseWrite(&buf_globalLock);
 144         }       /* for loop over a bunch of buffers */
 145     }           /* whole daemon's while loop */
 146 }
 147
 148 long
 149 buf_ValidateBuffers(void)
 150 {
 151     cm_buf_t * bp, *bpf, *bpa, *bpb;
 152     afs_uint64 countb = 0, countf = 0, counta = 0;
 153
 154     if (cm_data.buf_freeListp == NULL && cm_data.buf_freeListEndp != NULL ||
 155          cm_data.buf_freeListp != NULL && cm_data.buf_freeListEndp == NULL) {
 156         afsi_log("cm_ValidateBuffers failure: inconsistent free list pointers");
 157         fprintf(stderr, "cm_ValidateBuffers failure: inconsistent free list pointers\n");
 158         return -9;
 159     }
 160
 161     for (bp = cm_data.buf_freeListEndp; bp; bp=(cm_buf_t *) osi_QPrev(&bp->q)) {
 162         if (bp->magic != CM_BUF_MAGIC) {
 163             afsi_log("cm_ValidateBuffers failure: bp->magic != CM_BUF_MAGIC");
 164             fprintf(stderr, "cm_ValidateBuffers failure: bp->magic != CM_BUF_MAGIC\n");
 165             return -1;
 166         }
 167         countb++;
 168         bpb = bp;
 169
 170         if (countb > cm_data.buf_nbuffers) {
 171             afsi_log("cm_ValidateBuffers failure: countb > cm_data.buf_nbuffers");
 172             fprintf(stderr, "cm_ValidateBuffers failure: countb > cm_data.buf_nbuffers\n");
 173             return -6;
 174         }
 175     }
 176
 177     for (bp = cm_data.buf_freeListp; bp; bp=(cm_buf_t *) osi_QNext(&bp->q)) {
 178         if (bp->magic != CM_BUF_MAGIC) {
 179             afsi_log("cm_ValidateBuffers failure: bp->magic != CM_BUF_MAGIC");
 180             fprintf(stderr, "cm_ValidateBuffers failure: bp->magic != CM_BUF_MAGIC\n");
 181             return -2;
 182         }
 183         countf++;
 184         bpf = bp;
 185
 186         if (countf > cm_data.buf_nbuffers) {
 187             afsi_log("cm_ValidateBuffers failure: countf > cm_data.buf_nbuffers");
 188             fprintf(stderr, "cm_ValidateBuffers failure: countf > cm_data.buf_nbuffers\n");
 189             return -7;
 190         }
 191     }
 192
 193     for (bp = cm_data.buf_allp; bp; bp=bp->allp) {
 194         if (bp->magic != CM_BUF_MAGIC) {
 195             afsi_log("cm_ValidateBuffers failure: bp->magic != CM_BUF_MAGIC");
 196             fprintf(stderr, "cm_ValidateBuffers failure: bp->magic != CM_BUF_MAGIC\n");
 197             return -3;
 198         }
 199         counta++;
 200         bpa = bp;
 201
 202         if (counta > cm_data.buf_nbuffers) {
 203             afsi_log("cm_ValidateBuffers failure: counta > cm_data.buf_nbuffers");
 204             fprintf(stderr, "cm_ValidateBuffers failure: counta > cm_data.buf_nbuffers\n");
 205             return -8;
 206         }
 207     }
 208
 209     if (countb != countf) {
 210         afsi_log("cm_ValidateBuffers failure: countb != countf");
 211         fprintf(stderr, "cm_ValidateBuffers failure: countb != countf\n");
 212         return -4;
 213     }
 214
 215     if (counta != cm_data.buf_nbuffers) {
 216         afsi_log("cm_ValidateBuffers failure: counta != cm_data.buf_nbuffers");
 217         fprintf(stderr, "cm_ValidateBuffers failure: counta != cm_data.buf_nbuffers\n");
 218         return -5;
 219     }
 220
 221     return 0;
 222 }
 223
 224 void buf_Shutdown(void)
 225 {
 226     buf_ShutdownFlag = 1;
 227 }
 228
 229 /* initialize the buffer package; called with no locks
 230  * held during the initialization phase.
 231  */
 232 long buf_Init(int newFile, cm_buf_ops_t *opsp, afs_uint64 nbuffers)
 233 {
 234     static osi_once_t once;
 235     cm_buf_t *bp;
 236     thread_t phandle;
 237     long i;
 238     unsigned long pid;
 239     char *data;
 240
 241     if ( newFile ) {
 242         if (nbuffers)
 243             cm_data.buf_nbuffers = nbuffers;
 244
 245         /* Have to be able to reserve a whole chunk */
 246         if (((cm_data.buf_nbuffers - 3) * cm_data.buf_blockSize) < cm_chunkSize)
 247             return CM_ERROR_TOOFEWBUFS;
 248     }
 249
 250     /* recall for callouts */
 251     cm_buf_opsp = opsp;
 252
 253     if (osi_Once(&once)) {
 254         /* initialize global locks */
 255         lock_InitializeRWLock(&buf_globalLock, "Global buffer lock");
 256
 257         if ( newFile ) {
 258             /* remember this for those who want to reset it */
 259             cm_data.buf_nOrigBuffers = cm_data.buf_nbuffers;
 260
 261             /* lower hash size to a prime number */
 262             cm_data.buf_hashSize = osi_PrimeLessThan(CM_BUF_HASHSIZE);
 263
 264             /* create hash table */
 265             memset((void *)cm_data.buf_hashTablepp, 0, cm_data.buf_hashSize * sizeof(cm_buf_t *));
 266
 267             /* another hash table */
 268             memset((void *)cm_data.buf_fileHashTablepp, 0, cm_data.buf_hashSize * sizeof(cm_buf_t *));
 269
 270             /* create buffer headers and put in free list */
 271             bp = cm_data.bufHeaderBaseAddress;
 272             data = cm_data.bufDataBaseAddress;
 273             cm_data.buf_allp = NULL;
 274
 275             for (i=0; i<cm_data.buf_nbuffers; i++) {
 276                 osi_assert(bp >= cm_data.bufHeaderBaseAddress && bp < (cm_buf_t *)cm_data.bufDataBaseAddress);
 277                 osi_assert(data >= cm_data.bufDataBaseAddress && data < cm_data.bufEndOfData);
 278
 279                 /* allocate and zero some storage */
 280                 memset(bp, 0, sizeof(cm_buf_t));
 281                 bp->magic = CM_BUF_MAGIC;
 282                 /* thread on list of all buffers */
 283                 bp->allp = cm_data.buf_allp;
 284                 cm_data.buf_allp = bp;
 285
 286                 osi_QAdd((osi_queue_t **)&cm_data.buf_freeListp, &bp->q);
 287                 bp->flags |= CM_BUF_INLRU;
 288                 lock_InitializeMutex(&bp->mx, "Buffer mutex");
 289
 290                 /* grab appropriate number of bytes from aligned zone */
 291                 bp->datap = data;
 292
 293                 /* setup last buffer pointer */
 294                 if (i == 0)
 295                     cm_data.buf_freeListEndp = bp;
 296
 297                 /* next */
 298                 bp++;
 299                 data += cm_data.buf_blockSize;
 300             }
 301
 302             /* none reserved at first */
 303             cm_data.buf_reservedBufs = 0;
 304
 305             /* just for safety's sake */
 306             cm_data.buf_maxReservedBufs = cm_data.buf_nbuffers - 3;
 307         } else {
 308             bp = cm_data.bufHeaderBaseAddress;
 309             data = cm_data.bufDataBaseAddress;
 310
 311             for (i=0; i<cm_data.buf_nbuffers; i++) {
 312                 lock_InitializeMutex(&bp->mx, "Buffer mutex");
 313                 bp->userp = NULL;
 314                 bp->waitCount = 0;
 315                 bp->waitRequests = 0;
 316                 bp->flags &= ~CM_BUF_WAITING;
 317                 bp++;
 318             }
 319         }
 320
 321 #ifdef TESTING
 322         buf_ValidateBufQueues();
 323 #endif /* TESTING */
 324
 325 #ifdef TRACE_BUFFER
 326         /* init the buffer trace log */
 327         buf_logp = osi_LogCreate("buffer", 1000);
 328         osi_LogEnable(buf_logp);
 329 #endif
 330
 331         osi_EndOnce(&once);
 332
 333         /* and create the incr-syncer */
 334         phandle = thrd_Create(0, 0,
 335                                (ThreadFunc) buf_IncrSyncer, 0, 0, &pid,
 336                                "buf_IncrSyncer");
 337
 338         osi_assertx(phandle != NULL, "buf: can't create incremental sync proc");
 339         CloseHandle(phandle);
 340     }
 341
 342 #ifdef TESTING
 343     buf_ValidateBufQueues();
 344 #endif /* TESTING */
 345     return 0;
 346 }
 347
 348 /* add nbuffers to the buffer pool, if possible.
 349  * Called with no locks held.
 350  */
 351 long buf_AddBuffers(afs_uint64 nbuffers)
 352 {
 353     /* The size of a virtual cache cannot be changed after it has
 354      * been created.  Subsequent calls to MapViewofFile() with
 355      * an existing mapping object name would not allow the
 356      * object to be resized.  Return failure immediately.
 357      *
 358      * A similar problem now occurs with the persistent cache
 359      * given that the memory mapped file now contains a complex
 360      * data structure.
 361      */
 362     afsi_log("request to add %d buffers to the existing cache of size %d denied",
 363               nbuffers, cm_data.buf_nbuffers);
 364
 365     return CM_ERROR_INVAL;
 366 }
 367
 368 /* interface to set the number of buffers to an exact figure.
 369  * Called with no locks held.
 370  */
 371 long buf_SetNBuffers(afs_uint64 nbuffers)
 372 {
 373     if (nbuffers < 10)
 374         return CM_ERROR_INVAL;
 375     if (nbuffers == cm_data.buf_nbuffers)
 376         return 0;
 377     else if (nbuffers > cm_data.buf_nbuffers)
 378         return buf_AddBuffers(nbuffers - cm_data.buf_nbuffers);
 379     else
 380         return CM_ERROR_INVAL;
 381 }
 382
 383 /* release a buffer.  Buffer must be referenced, but unlocked. */
 384 void buf_Release(cm_buf_t *bp)
 385 {
 386     lock_ObtainWrite(&buf_globalLock);
 387     buf_LockedRelease(bp);
 388     lock_ReleaseWrite(&buf_globalLock);
 389 }
 390
 391 /* wait for reading or writing to clear; called with write-locked
 392  * buffer and unlocked scp and returns with locked buffer.
 393  */
 394 void buf_WaitIO(cm_scache_t * scp, cm_buf_t *bp)
 395 {
 396     if (scp)
 397         osi_assert(scp->magic == CM_SCACHE_MAGIC);
 398     osi_assert(bp->magic == CM_BUF_MAGIC);
 399
 400     while (1) {
 401         /* if no IO is happening, we're done */
 402         if (!(bp->flags & (CM_BUF_READING | CM_BUF_WRITING)))
 403             break;
 404
 405         /* otherwise I/O is happening, but some other thread is waiting for
 406          * the I/O already.  Wait for that guy to figure out what happened,
 407          * and then check again.
 408          */
 409         if ( bp->flags & CM_BUF_WAITING ) {
 410             bp->waitCount++;
 411             bp->waitRequests++;
 412             osi_Log1(afsd_logp, "buf_WaitIO CM_BUF_WAITING already set for 0x%p", bp);
 413         } else {
 414             osi_Log1(afsd_logp, "buf_WaitIO CM_BUF_WAITING set for 0x%p", bp);
 415             bp->flags |= CM_BUF_WAITING;
 416             bp->waitCount = bp->waitRequests = 1;
 417         }
 418         osi_SleepM((LONG_PTR)bp, &bp->mx);
 419
 420         smb_UpdateServerPriority();
 421
 422         lock_ObtainMutex(&bp->mx);
 423         osi_Log1(afsd_logp, "buf_WaitIO conflict wait done for 0x%p", bp);
 424         bp->waitCount--;
 425         if (bp->waitCount == 0) {
 426             osi_Log1(afsd_logp, "buf_WaitIO CM_BUF_WAITING reset for 0x%p", bp);
 427             bp->flags &= ~CM_BUF_WAITING;
 428             bp->waitRequests = 0;
 429         }
 430
 431         if ( !scp ) {
 432             scp = cm_FindSCache(&bp->fid);
 433         }
 434         if ( scp ) {
 435             lock_ObtainMutex(&scp->mx);
 436             if (scp->flags & CM_SCACHEFLAG_WAITING) {
 437                 osi_Log1(afsd_logp, "buf_WaitIO waking scp 0x%p", scp);
 438                 osi_Wakeup((LONG_PTR)&scp->flags);
 439             }
 440             lock_ReleaseMutex(&scp->mx);
 441         }
 442     }
 443
 444     /* if we get here, the IO is done, but we may have to wakeup people waiting for
 445      * the I/O to complete.  Do so.
 446      */
 447     if (bp->flags & CM_BUF_WAITING) {
 448         osi_Log1(afsd_logp, "buf_WaitIO Waking bp 0x%p", bp);
 449         osi_Wakeup((LONG_PTR) bp);
 450     }
 451     osi_Log1(afsd_logp, "WaitIO finished wait for bp 0x%p", bp);
 452 }
 453
 454 /* code to drop reference count while holding buf_globalLock */
 455 void buf_LockedRelease(cm_buf_t *bp)
 456 {
 457     /* ensure that we're in the LRU queue if our ref count is 0 */
 458     osi_assert(bp->refCount > 0);
 459     if (--bp->refCount == 0) {
 460         if (!(bp->flags & CM_BUF_INLRU)) {
 461             osi_QAdd((osi_queue_t **) &cm_data.buf_freeListp, &bp->q);
 462
 463             /* watch for transition from empty to one element */
 464             if (!cm_data.buf_freeListEndp)
 465                 cm_data.buf_freeListEndp = cm_data.buf_freeListp;
 466             bp->flags |= CM_BUF_INLRU;
 467         }
 468     }
 469 }
 470
 471 /* find a buffer, if any, for a particular file ID and offset.  Assumes
 472  * that buf_globalLock is write locked when called.
 473  */
 474 cm_buf_t *buf_LockedFind(struct cm_scache *scp, osi_hyper_t *offsetp)
 475 {
 476     long i;
 477     cm_buf_t *bp;
 478
 479     i = BUF_HASH(&scp->fid, offsetp);
 480     for(bp = cm_data.buf_hashTablepp[i]; bp; bp=bp->hashp) {
 481         if (cm_FidCmp(&scp->fid, &bp->fid) == 0
 482              && offsetp->LowPart == bp->offset.LowPart
 483              && offsetp->HighPart == bp->offset.HighPart) {
 484             bp->refCount++;
 485             break;
 486         }
 487     }
 488
 489     /* return whatever we found, if anything */
 490     return bp;
 491 }
 492
 493 /* find a buffer with offset *offsetp for vnode *scp.  Called
 494  * with no locks held.
 495  */
 496 cm_buf_t *buf_Find(struct cm_scache *scp, osi_hyper_t *offsetp)
 497 {
 498     cm_buf_t *bp;
 499
 500     lock_ObtainWrite(&buf_globalLock);
 501     bp = buf_LockedFind(scp, offsetp);
 502     lock_ReleaseWrite(&buf_globalLock);
 503
 504     return bp;
 505 }
 506
 507 /* start cleaning I/O on this buffer.  Buffer must be write locked, and is returned
 508  * write-locked.
 509  *
 510  * Makes sure that there's only one person writing this block
 511  * at any given time, and also ensures that the log is forced sufficiently far,
 512  * if this buffer contains logged data.
 513  */
 514 void buf_LockedCleanAsync(cm_buf_t *bp, cm_req_t *reqp)
 515 {
 516     long code = 0;
 517
 518     osi_assert(bp->magic == CM_BUF_MAGIC);
 519
 520     while ((bp->flags & CM_BUF_DIRTY) == CM_BUF_DIRTY) {
 521         lock_ReleaseMutex(&bp->mx);
 522
 523         osi_Log1(afsd_logp, "buf_LockedCleanAsync starts I/O on 0x%p", bp);
 524         code = (*cm_buf_opsp->Writep)(&bp->fid, &bp->offset,
 525                                        cm_data.buf_blockSize, 0, bp->userp,
 526                                        reqp);
 527         osi_Log2(afsd_logp, "buf_LockedCleanAsync I/O on 0x%p, done=%d", bp, code);
 528
 529         lock_ObtainMutex(&bp->mx);
 530         if (code)
 531             break;
 532
 533 #ifdef DISKCACHE95
 534         /* Disk cache support */
 535         /* write buffer to disk cache (synchronous for now) */
 536         diskcache_Update(bp->dcp, bp->datap, cm_data.buf_blockSize, bp->dataVersion);
 537 #endif /* DISKCACHE95 */
 538     };
 539
 540     /* do logging after call to GetLastError, or else */
 541
 542     /* if someone was waiting for the I/O that just completed or failed,
 543      * wake them up.
 544      */
 545     if (bp->flags & CM_BUF_WAITING) {
 546         /* turn off flags and wakeup users */
 547         osi_Log1(buf_logp, "buf_WaitIO Waking bp 0x%p", bp);
 548         osi_Wakeup((LONG_PTR) bp);
 549     }
 550 }
 551
 552 /* Called with a zero-ref count buffer and with the buf_globalLock write locked.
 553  * recycles the buffer, and leaves it ready for reuse with a ref count of 1.
 554  * The buffer must already be clean, and no I/O should be happening to it.
 555  */
 556 void buf_Recycle(cm_buf_t *bp)
 557 {
 558     int i;
 559     cm_buf_t **lbpp;
 560     cm_buf_t *tbp;
 561     cm_buf_t *prevBp, *nextBp;
 562
 563     osi_assert(bp->magic == CM_BUF_MAGIC);
 564
 565     /* if we get here, we know that the buffer still has a 0 ref count,
 566      * and that it is clean and has no currently pending I/O.  This is
 567      * the dude to return.
 568      * Remember that as long as the ref count is 0, we know that we won't
 569      * have any lock conflicts, so we can grab the buffer lock out of
 570      * order in the locking hierarchy.
 571      */
 572     osi_Log2( buf_logp, "buf_Recycle recycles 0x%p, off 0x%x",
 573               bp, bp->offset.LowPart);
 574
 575     osi_assert(bp->refCount == 0);
 576     osi_assert(!(bp->flags & (CM_BUF_READING | CM_BUF_WRITING | CM_BUF_DIRTY)));
 577     lock_AssertWrite(&buf_globalLock);
 578
 579     if (bp->flags & CM_BUF_INHASH) {
 580         /* Remove from hash */
 581
 582         i = BUF_HASH(&bp->fid, &bp->offset);
 583         lbpp = &(cm_data.buf_hashTablepp[i]);
 584         for(tbp = *lbpp; tbp; lbpp = &tbp->hashp, tbp = *lbpp) {
 585             if (tbp == bp) break;
 586         }
 587
 588         /* we better find it */
 589         osi_assertx(tbp != NULL, "buf_Recycle: hash table screwup");
 590
 591         *lbpp = bp->hashp;      /* hash out */
 592
 593         /* Remove from file hash */
 594
 595         i = BUF_FILEHASH(&bp->fid);
 596         prevBp = bp->fileHashBackp;
 597         nextBp = bp->fileHashp;
 598         if (prevBp)
 599             prevBp->fileHashp = nextBp;
 600         else
 601             cm_data.buf_fileHashTablepp[i] = nextBp;
 602         if (nextBp)
 603             nextBp->fileHashBackp = prevBp;
 604
 605         bp->flags &= ~CM_BUF_INHASH;
 606     }
 607
 608     /* bump the soft reference counter now, to invalidate softRefs; no
 609      * wakeup is required since people don't sleep waiting for this
 610      * counter to change.
 611      */
 612     bp->idCounter++;
 613
 614     /* make the fid unrecognizable */
 615     memset(&bp->fid, 0, sizeof(cm_fid_t));
 616 }
 617
 618 /* recycle a buffer, removing it from the free list, hashing in its new identity
 619  * and returning it write-locked so that no one can use it.  Called without
 620  * any locks held, and can return an error if it loses the race condition and
 621  * finds that someone else created the desired buffer.
 622  *
 623  * If success is returned, the buffer is returned write-locked.
 624  *
 625  * May be called with null scp and offsetp, if we're just trying to reclaim some
 626  * space from the buffer pool.  In that case, the buffer will be returned
 627  * without being hashed into the hash table.
 628  */
 629 long buf_GetNewLocked(struct cm_scache *scp, osi_hyper_t *offsetp, cm_buf_t **bufpp)
 630 {
 631     cm_buf_t *bp;               /* buffer we're dealing with */
 632     cm_buf_t *nextBp;   /* next buffer in file hash chain */
 633     long i;                     /* temp */
 634     cm_req_t req;
 635
 636     cm_InitReq(&req);   /* just in case */
 637
 638 #ifdef TESTING
 639     buf_ValidateBufQueues();
 640 #endif /* TESTING */
 641
 642     while(1) {
 643       retry:
 644         lock_ObtainWrite(&buf_globalLock);
 645         /* check to see if we lost the race */
 646         if (scp) {
 647             if (bp = buf_LockedFind(scp, offsetp)) {
 648                 bp->refCount--;
 649                 lock_ReleaseWrite(&buf_globalLock);
 650                 return CM_BUF_EXISTS;
 651             }
 652         }
 653
 654         /* does this fix the problem below?  it's a simple solution. */
 655         if (!cm_data.buf_freeListEndp)
 656             {
 657             lock_ReleaseWrite(&buf_globalLock);
 658             Sleep(200);
 659             goto retry;
 660             }
 661
 662         /* for debugging, assert free list isn't empty, although we
 663          * really should try waiting for a running tranasction to finish
 664          * instead of this; or better, we should have a transaction
 665          * throttler prevent us from entering this situation.
 666          */
 667         osi_assertx(cm_data.buf_freeListEndp != NULL, "buf_GetNewLocked: no free buffers");
 668
 669         /* look at all buffers in free list, some of which may temp.
 670          * have high refcounts and which then should be skipped,
 671          * starting cleaning I/O for those which are dirty.  If we find
 672          * a clean buffer, we rehash it, lock it and return it.
 673          */
 674         for(bp = cm_data.buf_freeListEndp; bp; bp=(cm_buf_t *) osi_QPrev(&bp->q)) {
 675             /* check to see if it really has zero ref count.  This
 676              * code can bump refcounts, at least, so it may not be
 677              * zero.
 678              */
 679             if (bp->refCount > 0)
 680                 continue;
 681
 682             /* we don't have to lock buffer itself, since the ref
 683              * count is 0 and we know it will stay zero as long as
 684              * we hold the global lock.
 685              */
 686
 687             /* don't recycle someone in our own chunk */
 688             if (!cm_FidCmp(&bp->fid, &scp->fid)
 689                  && (bp->offset.LowPart & (-cm_chunkSize))
 690                  == (offsetp->LowPart & (-cm_chunkSize)))
 691                 continue;
 692
 693             /* if this page is being filled (!) or cleaned, see if
 694              * the I/O has completed.  If not, skip it, otherwise
 695              * do the final processing for the I/O.
 696              */
 697             if (bp->flags & (CM_BUF_READING | CM_BUF_WRITING)) {
 698                 /* probably shouldn't do this much work while
 699                  * holding the big lock?  Watch for contention
 700                  * here.
 701                  */
 702                 continue;
 703             }
 704
 705             if (bp->flags & CM_BUF_DIRTY) {
 706                 /* if the buffer is dirty, start cleaning it and
 707                  * move on to the next buffer.  We do this with
 708                  * just the lock required to minimize contention
 709                  * on the big lock.
 710                  */
 711                 bp->refCount++;
 712                 lock_ReleaseWrite(&buf_globalLock);
 713
 714                 /* grab required lock and clean; this only
 715                  * starts the I/O.  By the time we're back,
 716                  * it'll still be marked dirty, but it will also
 717                  * have the WRITING flag set, so we won't get
 718                  * back here.
 719                  */
 720                 buf_CleanAsync(bp, &req);
 721
 722                 /* now put it back and go around again */
 723                 buf_Release(bp);
 724                 goto retry;
 725             }
 726
 727             /* if we get here, we know that the buffer still has a 0
 728              * ref count, and that it is clean and has no currently
 729              * pending I/O.  This is the dude to return.
 730              * Remember that as long as the ref count is 0, we know
 731              * that we won't have any lock conflicts, so we can grab
 732              * the buffer lock out of order in the locking hierarchy.
 733              */
 734             buf_Recycle(bp);
 735
 736             /* clean up junk flags */
 737             bp->flags &= ~(CM_BUF_EOF | CM_BUF_ERROR);
 738             bp->dataVersion = -1;       /* unknown so far */
 739
 740             /* now hash in as our new buffer, and give it the
 741              * appropriate label, if requested.
 742              */
 743             if (scp) {
 744                 bp->flags |= CM_BUF_INHASH;
 745                 bp->fid = scp->fid;
 746                 bp->offset = *offsetp;
 747                 i = BUF_HASH(&scp->fid, offsetp);
 748                 bp->hashp = cm_data.buf_hashTablepp[i];
 749                 cm_data.buf_hashTablepp[i] = bp;
 750                 i = BUF_FILEHASH(&scp->fid);
 751                 nextBp = cm_data.buf_fileHashTablepp[i];
 752                 bp->fileHashp = nextBp;
 753                 bp->fileHashBackp = NULL;
 754                 if (nextBp)
 755                     nextBp->fileHashBackp = bp;
 756                 cm_data.buf_fileHashTablepp[i] = bp;
 757             }
 758
 759             /* prepare to return it.  Start by giving it a good
 760              * refcount */
 761             bp->refCount = 1;
 762
 763             /* and since it has a non-zero ref count, we should move
 764              * it from the lru queue.  It better be still there,
 765              * since we've held the global (big) lock since we found
 766              * it there.
 767              */
 768             osi_assertx(bp->flags & CM_BUF_INLRU,
 769                          "buf_GetNewLocked: LRU screwup");
 770             if (cm_data.buf_freeListEndp == bp) {
 771                 /* we're the last guy in this queue, so maintain it */
 772                 cm_data.buf_freeListEndp = (cm_buf_t *) osi_QPrev(&bp->q);
 773             }
 774             osi_QRemove((osi_queue_t **) &cm_data.buf_freeListp, &bp->q);
 775             bp->flags &= ~CM_BUF_INLRU;
 776
 777             /* finally, grab the mutex so that people don't use it
 778              * before the caller fills it with data.  Again, no one
 779              * should have been able to get to this dude to lock it.
 780              */
 781             osi_assertx(lock_TryMutex(&bp->mx),
 782                          "buf_GetNewLocked: TryMutex failed");
 783
 784             lock_ReleaseWrite(&buf_globalLock);
 785             *bufpp = bp;
 786
 787 #ifdef TESTING
 788             buf_ValidateBufQueues();
 789 #endif /* TESTING */
 790             return 0;
 791         } /* for all buffers in lru queue */
 792         lock_ReleaseWrite(&buf_globalLock);
 793     }   /* while loop over everything */
 794     /* not reached */
 795 } /* the proc */
 796
 797 /* get a page, returning it held but unlocked.  Doesn't fill in the page
 798  * with I/O, since we're going to write the whole thing new.
 799  */
 800 long buf_GetNew(struct cm_scache *scp, osi_hyper_t *offsetp, cm_buf_t **bufpp)
 801 {
 802     cm_buf_t *bp;
 803     long code;
 804     osi_hyper_t pageOffset;
 805     int created;
 806
 807     created = 0;
 808     pageOffset.HighPart = offsetp->HighPart;
 809     pageOffset.LowPart = offsetp->LowPart & ~(cm_data.buf_blockSize-1);
 810     while (1) {
 811         bp = buf_Find(scp, &pageOffset);
 812         if (bp) {
 813             /* lock it and break out */
 814             lock_ObtainMutex(&bp->mx);
 815             break;
 816         }
 817
 818         /* otherwise, we have to create a page */
 819         code = buf_GetNewLocked(scp, &pageOffset, &bp);
 820
 821         /* check if the buffer was created in a race condition branch.
 822          * If so, go around so we can hold a reference to it.
 823          */
 824         if (code == CM_BUF_EXISTS)
 825             continue;
 826
 827         /* something else went wrong */
 828         if (code != 0)
 829             return code;
 830
 831         /* otherwise, we have a locked buffer that we just created */
 832         created = 1;
 833         break;
 834     } /* big while loop */
 835
 836     /* wait for reads */
 837     if (bp->flags & CM_BUF_READING)
 838         buf_WaitIO(scp, bp);
 839
 840     /* once it has been read once, we can unlock it and return it, still
 841      * with its refcount held.
 842      */
 843     lock_ReleaseMutex(&bp->mx);
 844     *bufpp = bp;
 845     osi_Log3(buf_logp, "buf_GetNew returning bp 0x%p for scp 0x%p, offset 0x%x",
 846               bp, scp, offsetp->LowPart);
 847     return 0;
 848 }
 849
 850 /* get a page, returning it held but unlocked.  Make sure it is complete */
 851 /* The scp must be unlocked when passed to this function */
 852 long buf_Get(struct cm_scache *scp, osi_hyper_t *offsetp, cm_buf_t **bufpp)
 853 {
 854     cm_buf_t *bp;
 855     long code;
 856     osi_hyper_t pageOffset;
 857     unsigned long tcount;
 858     int created;
 859     long lcount = 0;
 860 #ifdef DISKCACHE95
 861     cm_diskcache_t *dcp;
 862 #endif /* DISKCACHE95 */
 863
 864     created = 0;
 865     pageOffset.HighPart = offsetp->HighPart;
 866     pageOffset.LowPart = offsetp->LowPart & ~(cm_data.buf_blockSize-1);
 867     while (1) {
 868         lcount++;
 869 #ifdef TESTING
 870         buf_ValidateBufQueues();
 871 #endif /* TESTING */
 872
 873         bp = buf_Find(scp, &pageOffset);
 874         if (bp) {
 875             /* lock it and break out */
 876             lock_ObtainMutex(&bp->mx);
 877             break;
 878
 879 #ifdef DISKCACHE95
 880             /* touch disk chunk to update LRU info */
 881             diskcache_Touch(bp->dcp);
 882 #endif /* DISKCACHE95 */
 883         }
 884
 885         /* otherwise, we have to create a page */
 886         code = buf_GetNewLocked(scp, &pageOffset, &bp);
 887
 888         /* check if the buffer was created in a race condition branch.
 889          * If so, go around so we can hold a reference to it.
 890          */
 891         if (code == CM_BUF_EXISTS)
 892             continue;
 893
 894         /* something else went wrong */
 895         if (code != 0) {
 896 #ifdef TESTING
 897             buf_ValidateBufQueues();
 898 #endif /* TESTING */
 899             return code;
 900         }
 901
 902         /* otherwise, we have a locked buffer that we just created */
 903         created = 1;
 904         break;
 905     } /* big while loop */
 906
 907     /* if we get here, we have a locked buffer that may have just been
 908      * created, in which case it needs to be filled with data.
 909      */
 910     if (created) {
 911         /* load the page; freshly created pages should be idle */
 912         osi_assert(!(bp->flags & (CM_BUF_READING | CM_BUF_WRITING)));
 913
 914         /* setup offset, event */
 915         bp->over.Offset = bp->offset.LowPart;
 916         bp->over.OffsetHigh = bp->offset.HighPart;
 917
 918         /* start the I/O; may drop lock */
 919         bp->flags |= CM_BUF_READING;
 920         code = (*cm_buf_opsp->Readp)(bp, cm_data.buf_blockSize, &tcount, NULL);
 921
 922 #ifdef DISKCACHE95
 923         code = diskcache_Get(&bp->fid, &bp->offset, bp->datap, cm_data.buf_blockSize, &bp->dataVersion, &tcount, &dcp);
 924         bp->dcp = dcp;    /* pointer to disk cache struct. */
 925 #endif /* DISKCACHE95 */
 926
 927         if (code != 0) {
 928             /* failure or queued */
 929             if (code != ERROR_IO_PENDING) {
 930                 bp->error = code;
 931                 bp->flags |= CM_BUF_ERROR;
 932                 bp->flags &= ~CM_BUF_READING;
 933                 if (bp->flags & CM_BUF_WAITING) {
 934                     osi_Log1(buf_logp, "buf_Get Waking bp 0x%p", bp);
 935                     osi_Wakeup((LONG_PTR) bp);
 936                 }
 937                 lock_ReleaseMutex(&bp->mx);
 938                 buf_Release(bp);
 939 #ifdef TESTING
 940                 buf_ValidateBufQueues();
 941 #endif /* TESTING */
 942                 return code;
 943             }
 944         } else {
 945             /* otherwise, I/O completed instantly and we're done, except
 946              * for padding the xfr out with 0s and checking for EOF
 947              */
 948             if (tcount < (unsigned long) cm_data.buf_blockSize) {
 949                 memset(bp->datap+tcount, 0, cm_data.buf_blockSize - tcount);
 950                 if (tcount == 0)
 951                     bp->flags |= CM_BUF_EOF;
 952             }
 953             bp->flags &= ~CM_BUF_READING;
 954             if (bp->flags & CM_BUF_WAITING) {
 955                 osi_Log1(buf_logp, "buf_Get Waking bp 0x%p", bp);
 956                 osi_Wakeup((LONG_PTR) bp);
 957             }
 958         }
 959
 960     } /* if created */
 961
 962     /* wait for reads, either that which we started above, or that someone
 963      * else started.  We don't care if we return a buffer being cleaned.
 964      */
 965     if (bp->flags & CM_BUF_READING)
 966         buf_WaitIO(scp, bp);
 967
 968     /* once it has been read once, we can unlock it and return it, still
 969      * with its refcount held.
 970      */
 971     lock_ReleaseMutex(&bp->mx);
 972     *bufpp = bp;
 973
 974     /* now remove from queue; will be put in at the head (farthest from
 975      * being recycled) when we're done in buf_Release.
 976      */
 977     lock_ObtainWrite(&buf_globalLock);
 978     if (bp->flags & CM_BUF_INLRU) {
 979         if (cm_data.buf_freeListEndp == bp)
 980             cm_data.buf_freeListEndp = (cm_buf_t *) osi_QPrev(&bp->q);
 981         osi_QRemove((osi_queue_t **) &cm_data.buf_freeListp, &bp->q);
 982         bp->flags &= ~CM_BUF_INLRU;
 983     }
 984     lock_ReleaseWrite(&buf_globalLock);
 985
 986     osi_Log3(buf_logp, "buf_Get returning bp 0x%p for scp 0x%p, offset 0x%x",
 987               bp, scp, offsetp->LowPart);
 988 #ifdef TESTING
 989     buf_ValidateBufQueues();
 990 #endif /* TESTING */
 991     return 0;
 992 }
 993
 994 /* count # of elements in the free list;
 995  * we don't bother doing the proper locking for accessing dataVersion or flags
 996  * since it is a pain, and this is really just an advisory call.  If you need
 997  * to do better at some point, rewrite this function.
 998  */
 999 long buf_CountFreeList(void)
1000 {
1001     long count;
1002     cm_buf_t *bufp;
1003
1004     count = 0;
1005     lock_ObtainRead(&buf_globalLock);
1006     for(bufp = cm_data.buf_freeListp; bufp; bufp = (cm_buf_t *) osi_QNext(&bufp->q)) {
1007         /* if the buffer doesn't have an identity, or if the buffer
1008          * has been invalidate (by having its DV stomped upon), then
1009          * count it as free, since it isn't really being utilized.
1010          */
1011         if (!(bufp->flags & CM_BUF_INHASH) || bufp->dataVersion <= 0)
1012             count++;
1013     }
1014     lock_ReleaseRead(&buf_globalLock);
1015     return count;
1016 }
1017
1018 /* clean a buffer synchronously */
1019 void buf_CleanAsync(cm_buf_t *bp, cm_req_t *reqp)
1020 {
1021     osi_assert(bp->magic == CM_BUF_MAGIC);
1022
1023     lock_ObtainMutex(&bp->mx);
1024     buf_LockedCleanAsync(bp, reqp);
1025     lock_ReleaseMutex(&bp->mx);
1026 }
1027
1028 /* wait for a buffer's cleaning to finish */
1029 void buf_CleanWait(cm_scache_t * scp, cm_buf_t *bp)
1030 {
1031     osi_assert(bp->magic == CM_BUF_MAGIC);
1032
1033     lock_ObtainMutex(&bp->mx);
1034     if (bp->flags & CM_BUF_WRITING) {
1035         buf_WaitIO(scp, bp);
1036     }
1037     lock_ReleaseMutex(&bp->mx);
1038 }
1039
1040 /* set the dirty flag on a buffer, and set associated write-ahead log,
1041  * if there is one.  Allow one to be added to a buffer, but not changed.
1042  *
1043  * The buffer must be locked before calling this routine.
1044  */
1045 void buf_SetDirty(cm_buf_t *bp)
1046 {
1047     osi_assert(bp->magic == CM_BUF_MAGIC);
1048     osi_assert(bp->refCount > 0);
1049
1050     osi_Log1(buf_logp, "buf_SetDirty 0x%p", bp);
1051
1052     /* set dirty bit */
1053     bp->flags |= CM_BUF_DIRTY;
1054
1055     /* and turn off EOF flag, since it has associated data now */
1056     bp->flags &= ~CM_BUF_EOF;
1057 }
1058
1059 /* clean all buffers, reset log pointers and invalidate all buffers.
1060  * Called with no locks held, and returns with same.
1061  *
1062  * This function is guaranteed to clean and remove the log ptr of all the
1063  * buffers that were dirty or had non-zero log ptrs before the call was
1064  * made.  That's sufficient to clean up any garbage left around by recovery,
1065  * which is all we're counting on this for; there may be newly created buffers
1066  * added while we're running, but that should be OK.
1067  *
1068  * In an environment where there are no transactions (artificially imposed, for
1069  * example, when switching the database to raw mode), this function is used to
1070  * make sure that all updates have been written to the disk.  In that case, we don't
1071  * really require that we forget the log association between pages and logs, but
1072  * it also doesn't hurt.  Since raw mode I/O goes through this buffer package, we don't
1073  * have to worry about invalidating data in the buffers.
1074  *
1075  * This function is used at the end of recovery as paranoia to get the recovered
1076  * database out to disk.  It removes all references to the recovery log and cleans
1077  * all buffers.
1078  */
1079 long buf_CleanAndReset(void)
1080 {
1081     long i;
1082     cm_buf_t *bp;
1083     cm_req_t req;
1084
1085     lock_ObtainWrite(&buf_globalLock);
1086     for(i=0; i<cm_data.buf_hashSize; i++) {
1087         for(bp = cm_data.buf_hashTablepp[i]; bp; bp = bp->hashp) {
1088             if ((bp->flags & CM_BUF_DIRTY) == CM_BUF_DIRTY) {
1089                 bp->refCount++;
1090                 lock_ReleaseWrite(&buf_globalLock);
1091
1092                 /* now no locks are held; clean buffer and go on */
1093                 cm_InitReq(&req);
1094                 buf_CleanAsync(bp, &req);
1095                 buf_CleanWait(NULL, bp);
1096
1097                 /* relock and release buffer */
1098                 lock_ObtainWrite(&buf_globalLock);
1099                 buf_LockedRelease(bp);
1100             } /* dirty */
1101         } /* over one bucket */
1102     }   /* for loop over all hash buckets */
1103
1104     /* release locks */
1105     lock_ReleaseWrite(&buf_globalLock);
1106
1107 #ifdef TESTING
1108     buf_ValidateBufQueues();
1109 #endif /* TESTING */
1110
1111     /* and we're done */
1112     return 0;
1113 }
1114
1115 /* called without global lock being held, reserves buffers for callers
1116  * that need more than one held (not locked) at once.
1117  */
1118 void buf_ReserveBuffers(afs_uint64 nbuffers)
1119 {
1120     lock_ObtainWrite(&buf_globalLock);
1121     while (1) {
1122         if (cm_data.buf_reservedBufs + nbuffers > cm_data.buf_maxReservedBufs) {
1123             cm_data.buf_reserveWaiting = 1;
1124             osi_Log1(buf_logp, "buf_ReserveBuffers waiting for %d bufs", nbuffers);
1125             osi_SleepW((LONG_PTR) &cm_data.buf_reservedBufs, &buf_globalLock);
1126             lock_ObtainWrite(&buf_globalLock);
1127         }
1128         else {
1129             cm_data.buf_reservedBufs += nbuffers;
1130             break;
1131         }
1132     }
1133     lock_ReleaseWrite(&buf_globalLock);
1134 }
1135
1136 int buf_TryReserveBuffers(afs_uint64 nbuffers)
1137 {
1138     int code;
1139
1140     lock_ObtainWrite(&buf_globalLock);
1141     if (cm_data.buf_reservedBufs + nbuffers > cm_data.buf_maxReservedBufs) {
1142         code = 0;
1143     }
1144     else {
1145         cm_data.buf_reservedBufs += nbuffers;
1146         code = 1;
1147     }
1148     lock_ReleaseWrite(&buf_globalLock);
1149     return code;
1150 }
1151
1152 /* called without global lock held, releases reservation held by
1153  * buf_ReserveBuffers.
1154  */
1155 void buf_UnreserveBuffers(afs_uint64 nbuffers)
1156 {
1157     lock_ObtainWrite(&buf_globalLock);
1158     cm_data.buf_reservedBufs -= nbuffers;
1159     if (cm_data.buf_reserveWaiting) {
1160         cm_data.buf_reserveWaiting = 0;
1161         osi_Wakeup((LONG_PTR) &cm_data.buf_reservedBufs);
1162     }
1163     lock_ReleaseWrite(&buf_globalLock);
1164 }
1165
1166 /* truncate the buffers past sizep, zeroing out the page, if we don't
1167  * end on a page boundary.
1168  *
1169  * Requires cm_bufCreateLock to be write locked.
1170  */
1171 long buf_Truncate(cm_scache_t *scp, cm_user_t *userp, cm_req_t *reqp,
1172                    osi_hyper_t *sizep)
1173 {
1174     cm_buf_t *bufp;
1175     cm_buf_t *nbufp;                    /* next buffer, if didRelease */
1176     osi_hyper_t bufEnd;
1177     long code;
1178     long bufferPos;
1179     int didRelease;
1180     long i;
1181
1182     /* assert that cm_bufCreateLock is held in write mode */
1183     lock_AssertWrite(&scp->bufCreateLock);
1184
1185     i = BUF_FILEHASH(&scp->fid);
1186
1187     lock_ObtainWrite(&buf_globalLock);
1188     bufp = cm_data.buf_fileHashTablepp[i];
1189     if (bufp == NULL) {
1190         lock_ReleaseWrite(&buf_globalLock);
1191         return 0;
1192     }
1193
1194     bufp->refCount++;
1195     lock_ReleaseWrite(&buf_globalLock);
1196     for(; bufp; bufp = nbufp) {
1197         didRelease = 0;
1198         lock_ObtainMutex(&bufp->mx);
1199
1200         bufEnd.HighPart = 0;
1201         bufEnd.LowPart = cm_data.buf_blockSize;
1202         bufEnd = LargeIntegerAdd(bufEnd, bufp->offset);
1203
1204         if (cm_FidCmp(&bufp->fid, &scp->fid) == 0 &&
1205              LargeIntegerLessThan(*sizep, bufEnd)) {
1206             buf_WaitIO(scp, bufp);
1207         }
1208         lock_ObtainMutex(&scp->mx);
1209
1210         /* make sure we have a callback (so we have the right value for
1211          * the length), and wait for it to be safe to do a truncate.
1212          */
1213         code = cm_SyncOp(scp, bufp, userp, reqp, 0,
1214                           CM_SCACHESYNC_NEEDCALLBACK
1215                           | CM_SCACHESYNC_GETSTATUS
1216                           | CM_SCACHESYNC_SETSIZE
1217                           | CM_SCACHESYNC_BUFLOCKED);
1218         /* if we succeeded in our locking, and this applies to the right
1219          * file, and the truncate request overlaps the buffer either
1220          * totally or partially, then do something.
1221          */
1222         if (code == 0 && cm_FidCmp(&bufp->fid, &scp->fid) == 0
1223              && LargeIntegerLessThan(*sizep, bufEnd)) {
1224
1225             lock_ObtainWrite(&buf_globalLock);
1226
1227             /* destroy the buffer, turning off its dirty bit, if
1228              * we're truncating the whole buffer.  Otherwise, set
1229              * the dirty bit, and clear out the tail of the buffer
1230              * if we just overlap some.
1231              */
1232             if (LargeIntegerLessThanOrEqualTo(*sizep, bufp->offset)) {
1233                 /* truncating the entire page */
1234                 bufp->flags &= ~CM_BUF_DIRTY;
1235                 bufp->dataVersion = -1; /* known bad */
1236                 bufp->dirtyCounter++;
1237             }
1238             else {
1239                 /* don't set dirty, since dirty implies
1240                  * currently up-to-date.  Don't need to do this,
1241                  * since we'll update the length anyway.
1242                  *
1243                  * Zero out remainder of the page, in case we
1244                  * seek and write past EOF, and make this data
1245                  * visible again.
1246                  */
1247                 bufferPos = sizep->LowPart & (cm_data.buf_blockSize - 1);
1248                 osi_assert(bufferPos != 0);
1249                 memset(bufp->datap + bufferPos, 0,
1250                         cm_data.buf_blockSize - bufferPos);
1251             }
1252
1253             lock_ReleaseWrite(&buf_globalLock);
1254         }
1255
1256         lock_ReleaseMutex(&scp->mx);
1257         lock_ReleaseMutex(&bufp->mx);
1258         if (!didRelease) {
1259             lock_ObtainWrite(&buf_globalLock);
1260             nbufp = bufp->fileHashp;
1261             if (nbufp) nbufp->refCount++;
1262             buf_LockedRelease(bufp);
1263             lock_ReleaseWrite(&buf_globalLock);
1264         }
1265
1266         /* bail out early if we fail */
1267         if (code) {
1268             /* at this point, nbufp is held; bufp has already been
1269              * released.
1270              */
1271             if (nbufp)
1272                 buf_Release(nbufp);
1273
1274 #ifdef TESTING
1275             buf_ValidateBufQueues();
1276 #endif /* TESTING */
1277
1278             return code;
1279         }
1280     }
1281
1282 #ifdef TESTING
1283     buf_ValidateBufQueues();
1284 #endif /* TESTING */
1285
1286     /* success */
1287     return 0;
1288 }
1289
1290 long buf_FlushCleanPages(cm_scache_t *scp, cm_user_t *userp, cm_req_t *reqp)
1291 {
1292     long code;
1293     cm_buf_t *bp;               /* buffer we're hacking on */
1294     cm_buf_t *nbp;
1295     int didRelease;
1296     long i;
1297
1298     i = BUF_FILEHASH(&scp->fid);
1299
1300     code = 0;
1301     lock_ObtainWrite(&buf_globalLock);
1302     bp = cm_data.buf_fileHashTablepp[i];
1303     if (bp)
1304         bp->refCount++;
1305     lock_ReleaseWrite(&buf_globalLock);
1306     for (; bp; bp = nbp) {
1307         didRelease = 0; /* haven't released this buffer yet */
1308
1309         /* clean buffer synchronously */
1310         if (cm_FidCmp(&bp->fid, &scp->fid) == 0) {
1311             lock_ObtainMutex(&bp->mx);
1312
1313             /* start cleaning the buffer, and wait for it to finish */
1314             buf_LockedCleanAsync(bp, reqp);
1315             buf_WaitIO(scp, bp);
1316             lock_ReleaseMutex(&bp->mx);
1317
1318             code = (*cm_buf_opsp->Stabilizep)(scp, userp, reqp);
1319             if (code)
1320                 goto skip;
1321
1322             lock_ObtainWrite(&buf_globalLock);
1323             /* actually, we only know that buffer is clean if ref
1324              * count is 1, since we don't have buffer itself locked.
1325              */
1326             if (!(bp->flags & CM_BUF_DIRTY)) {
1327                 if (bp->refCount == 1) {        /* bp is held above */
1328                     buf_LockedRelease(bp);
1329                     nbp = bp->fileHashp;
1330                     if (nbp)
1331                         nbp->refCount++;
1332                     didRelease = 1;
1333                     buf_Recycle(bp);
1334                 }
1335             }
1336             lock_ReleaseWrite(&buf_globalLock);
1337
1338             (*cm_buf_opsp->Unstabilizep)(scp, userp);
1339         }
1340
1341       skip:
1342         if (!didRelease) {
1343             lock_ObtainWrite(&buf_globalLock);
1344             if (nbp = bp->fileHashp)
1345                 nbp->refCount++;
1346             buf_LockedRelease(bp);
1347             lock_ReleaseWrite(&buf_globalLock);
1348         }
1349     }   /* for loop over a bunch of buffers */
1350
1351 #ifdef TESTING
1352             buf_ValidateBufQueues();
1353 #endif /* TESTING */
1354
1355     /* done */
1356     return code;
1357 }
1358
1359 long buf_CleanVnode(struct cm_scache *scp, cm_user_t *userp, cm_req_t *reqp)
1360 {
1361     long code;
1362     cm_buf_t *bp;               /* buffer we're hacking on */
1363     cm_buf_t *nbp;              /* next one */
1364     long i;
1365
1366     i = BUF_FILEHASH(&scp->fid);
1367
1368     code = 0;
1369     lock_ObtainWrite(&buf_globalLock);
1370     bp = cm_data.buf_fileHashTablepp[i];
1371     if (bp)
1372         bp->refCount++;
1373     lock_ReleaseWrite(&buf_globalLock);
1374     for (; bp; bp = nbp) {
1375         /* clean buffer synchronously */
1376         if (cm_FidCmp(&bp->fid, &scp->fid) == 0) {
1377             if (userp) {
1378                 cm_HoldUser(userp);
1379                 lock_ObtainMutex(&bp->mx);
1380                 if (bp->userp)
1381                     cm_ReleaseUser(bp->userp);
1382                 bp->userp = userp;
1383                 lock_ReleaseMutex(&bp->mx);
1384             }
1385             buf_CleanAsync(bp, reqp);
1386             buf_CleanWait(scp, bp);
1387             lock_ObtainMutex(&bp->mx);
1388             if (bp->flags & CM_BUF_ERROR) {
1389                 if (code == 0 || code == -1)
1390                     code = bp->error;
1391                 if (code == 0)
1392                     code = -1;
1393             }
1394             lock_ReleaseMutex(&bp->mx);
1395         }
1396
1397         lock_ObtainWrite(&buf_globalLock);
1398         buf_LockedRelease(bp);
1399         nbp = bp->fileHashp;
1400         if (nbp)
1401             nbp->refCount++;
1402         lock_ReleaseWrite(&buf_globalLock);
1403     }   /* for loop over a bunch of buffers */
1404
1405 #ifdef TESTING
1406     buf_ValidateBufQueues();
1407 #endif /* TESTING */
1408
1409     /* done */
1410     return code;
1411 }
1412
1413 #ifdef TESTING
1414 void
1415 buf_ValidateBufQueues(void)
1416 {
1417     cm_buf_t * bp, *bpb, *bpf, *bpa;
1418     afs_uint32 countf=0, countb=0, counta=0;
1419
1420     lock_ObtainRead(&buf_globalLock);
1421     for (bp = cm_data.buf_freeListEndp; bp; bp=(cm_buf_t *) osi_QPrev(&bp->q)) {
1422         if (bp->magic != CM_BUF_MAGIC)
1423             DebugBreak();
1424         countb++;
1425         bpb = bp;
1426     }
1427
1428     for (bp = cm_data.buf_freeListp; bp; bp=(cm_buf_t *) osi_QNext(&bp->q)) {
1429         if (bp->magic != CM_BUF_MAGIC)
1430             DebugBreak();
1431         countf++;
1432         bpf = bp;
1433     }
1434
1435     for (bp = cm_data.buf_allp; bp; bp=bp->allp) {
1436         if (bp->magic != CM_BUF_MAGIC)
1437             DebugBreak();
1438         counta++;
1439         bpa = bp;
1440     }
1441     lock_ReleaseRead(&buf_globalLock);
1442
1443     if (countb != countf)
1444         DebugBreak();
1445
1446     if (counta != cm_data.buf_nbuffers)
1447         DebugBreak();
1448 }
1449 #endif /* TESTING */
1450
1451 /* dump the contents of the buf_hashTablepp. */
1452 int cm_DumpBufHashTable(FILE *outputFile, char *cookie, int lock)
1453 {
1454     int zilch;
1455     cm_buf_t *bp;
1456     char output[1024];
1457     int i;
1458
1459     if (cm_data.buf_hashTablepp == NULL)
1460         return -1;
1461
1462     if (lock)
1463         lock_ObtainRead(&buf_globalLock);
1464
1465     StringCbPrintfA(output, sizeof(output), "%s - dumping buf_HashTable - buf_hashSize=%d\n",
1466                     cookie, cm_data.buf_hashSize);
1467     WriteFile(outputFile, output, (DWORD)strlen(output), &zilch, NULL);
1468
1469     for (i = 0; i < cm_data.buf_hashSize; i++)
1470     {
1471         for (bp = cm_data.buf_hashTablepp[i]; bp; bp=bp->hashp)
1472         {
1473             if (bp->refCount)
1474             {
1475                 StringCbPrintfA(output, sizeof(output), "vnode=%d, unique=%d), size=%d refCount=%d\n",
1476                         cookie, (void *)bp, i, bp->fid.cell, bp->fid.volume,
1477                         bp->fid.vnode, bp->fid.unique, bp->size, bp->refCount);
1478                 WriteFile(outputFile, output, (DWORD)strlen(output), &zilch, NULL);
1479             }
1480         }
1481     }
1482
1483     StringCbPrintfA(output, sizeof(output), "%s - Done dumping buf_HashTable.\n", cookie);
1484     WriteFile(outputFile, output, (DWORD)strlen(output), &zilch, NULL);
1485
1486     if (lock)
1487         lock_ReleaseRead(&buf_globalLock);
1488     return 0;
1489 }
1490
1491 void buf_ForceTrace(BOOL flush)
1492 {
1493     HANDLE handle;
1494     int len;
1495     char buf[256];
1496
1497     if (!buf_logp)
1498         return;
1499
1500     len = GetTempPath(sizeof(buf)-10, buf);
1501     StringCbCopyA(&buf[len], sizeof(buf)-len, "/afs-buffer.log");
1502     handle = CreateFile(buf, GENERIC_WRITE, FILE_SHARE_READ,
1503                             NULL, CREATE_ALWAYS, FILE_ATTRIBUTE_NORMAL, NULL);
1504     if (handle == INVALID_HANDLE_VALUE) {
1505         osi_panic("Cannot create log file", __FILE__, __LINE__);
1506     }
1507     osi_LogPrint(buf_logp, handle);
1508     if (flush)
1509         FlushFileBuffers(handle);
1510     CloseHandle(handle);
1511 }