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