99efd4c2d8b9180ea206069f2336be3a572aadc0
[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 <afsconfig.h>
13 #include <afs/param.h>
14 #include <roken.h>
15
16 #include <afs/stds.h>
17
18 #include <windows.h>
19 #include <osi.h>
20 #include <stdio.h>
21 #include <assert.h>
22 #include <strsafe.h>
23 #include <math.h>
24
25 #include "afsd.h"
26 #include "cm_memmap.h"
27
28 #ifdef DEBUG
29 #define TRACE_BUFFER 1
30 #endif
31
32 extern void afsi_log(char *pattern, ...);
33
34 /* This module implements the buffer package used by the local transaction
35  * system (cm).  It is initialized by calling cm_Init, which calls buf_Init;
36  * it must be initalized before any of its main routines are called.
37  *
38  * Each buffer is hashed into a hash table by file ID and offset, and if its
39  * reference count is zero, it is also in a free list.
40  *
41  * There are two locks involved in buffer processing.  The global lock
42  * buf_globalLock protects all of the global variables defined in this module,
43  * the reference counts and hash pointers in the actual cm_buf_t structures,
44  * and the LRU queue pointers in the buffer structures.
45  *
46  * The mutexes in the buffer structures protect the remaining fields in the
47  * buffers, as well the data itself.
48  * 
49  * The locking hierarchy here is this:
50  * 
51  * - resv multiple simul. buffers reservation
52  * - lock buffer I/O flags
53  * - lock buffer's mutex
54  * - lock buf_globalLock
55  *
56  */
57
58 /* global debugging log */
59 osi_log_t *buf_logp = NULL;
60
61 /* Global lock protecting hash tables and free lists */
62 osi_rwlock_t buf_globalLock;
63
64 /* ptr to head of the free list (most recently used) and the
65  * tail (the guy to remove first).  We use osi_Q* functions
66  * to put stuff in buf_freeListp, and maintain the end
67  * pointer manually
68  */
69
70 /* a pointer to a list of all buffers, just so that we can find them
71  * easily for debugging, and for the incr syncer.  Locked under
72  * the global lock.
73  */
74
75 /* defaults setup; these variables may be manually assigned into
76  * before calling cm_Init, as a way of changing these defaults.
77  */
78
79 /* callouts for reading and writing data, etc */
80 cm_buf_ops_t *cm_buf_opsp;
81
82 #ifdef DISKCACHE95
83 /* for experimental disk caching support in Win95 client */
84 cm_buf_t *buf_diskFreeListp;
85 cm_buf_t *buf_diskFreeListEndp;
86 cm_buf_t *buf_diskAllp;
87 extern int cm_diskCacheEnabled;
88 #endif /* DISKCACHE95 */
89
90 /* set this to 1 when we are terminating to prevent access attempts */
91 static int buf_ShutdownFlag = 0;
92
93 #ifdef DEBUG_REFCOUNT
94 void buf_HoldLockedDbg(cm_buf_t *bp, char *file, long line)
95 #else
96 void buf_HoldLocked(cm_buf_t *bp)
97 #endif
98 {
99     afs_int32 refCount;
100
101     osi_assertx(bp->magic == CM_BUF_MAGIC,"incorrect cm_buf_t magic");
102     refCount = InterlockedIncrement(&bp->refCount);
103 #ifdef DEBUG_REFCOUNT
104     osi_Log2(afsd_logp,"buf_HoldLocked bp 0x%p ref %d",bp, refCount);
105     afsi_log("%s:%d buf_HoldLocked bp 0x%p, ref %d", file, line, bp, refCount);
106 #endif
107 }
108
109 /* hold a reference to an already held buffer */
110 #ifdef DEBUG_REFCOUNT
111 void buf_HoldDbg(cm_buf_t *bp, char *file, long line)
112 #else
113 void buf_Hold(cm_buf_t *bp)
114 #endif
115 {
116     afs_int32 refCount;
117
118     lock_ObtainRead(&buf_globalLock);
119     osi_assertx(bp->magic == CM_BUF_MAGIC,"incorrect cm_buf_t magic");
120     refCount = InterlockedIncrement(&bp->refCount);
121 #ifdef DEBUG_REFCOUNT
122     osi_Log2(afsd_logp,"buf_Hold bp 0x%p ref %d",bp, refCount);
123     afsi_log("%s:%d buf_Hold bp 0x%p, ref %d", file, line, bp, refCount);
124 #endif
125     lock_ReleaseRead(&buf_globalLock);
126 }
127
128 /* code to drop reference count while holding buf_globalLock */
129 #ifdef DEBUG_REFCOUNT
130 void buf_ReleaseLockedDbg(cm_buf_t *bp, afs_uint32 writeLocked, char *file, long line)
131 #else
132 void buf_ReleaseLocked(cm_buf_t *bp, afs_uint32 writeLocked)
133 #endif
134 {
135     afs_int32 refCount;
136
137     if (writeLocked)
138         lock_AssertWrite(&buf_globalLock);
139     else
140         lock_AssertRead(&buf_globalLock);
141
142     /* ensure that we're in the LRU queue if our ref count is 0 */
143     osi_assertx(bp->magic == CM_BUF_MAGIC,"incorrect cm_buf_t magic");
144
145     refCount = InterlockedDecrement(&bp->refCount);
146 #ifdef DEBUG_REFCOUNT
147     osi_Log3(afsd_logp,"buf_ReleaseLocked %s bp 0x%p ref %d",writeLocked?"write":"read", bp, refCount);
148     afsi_log("%s:%d buf_ReleaseLocked %s bp 0x%p, ref %d", file, line, writeLocked?"write":"read", bp, refCount);
149 #endif
150 #ifdef DEBUG
151     if (refCount < 0)
152         osi_panic("buf refcount 0",__FILE__,__LINE__);;
153 #else
154     osi_assertx(refCount >= 0, "cm_buf_t refCount == 0");
155 #endif
156     if (refCount == 0) {
157         /* 
158          * If we are read locked there could be a race condition
159          * with buf_Find() so we must obtain a write lock and
160          * double check that the refCount is actually zero
161          * before we remove the buffer from the LRU queue.
162          */
163         if (!writeLocked)
164             lock_ConvertRToW(&buf_globalLock);
165
166         if (bp->refCount == 0 &&
167             !(bp->qFlags & CM_BUF_QINLRU)) {
168             osi_QAdd((osi_queue_t **) &cm_data.buf_freeListp, &bp->q);
169
170             /* watch for transition from empty to one element */
171             if (!cm_data.buf_freeListEndp)
172                 cm_data.buf_freeListEndp = cm_data.buf_freeListp;
173             bp->qFlags |= CM_BUF_QINLRU;
174         }
175
176         if (!writeLocked)
177             lock_ConvertWToR(&buf_globalLock);
178     }
179 }       
180
181 /* release a buffer.  Buffer must be referenced, but unlocked. */
182 #ifdef DEBUG_REFCOUNT
183 void buf_ReleaseDbg(cm_buf_t *bp, char *file, long line)
184 #else
185 void buf_Release(cm_buf_t *bp)
186 #endif
187 {
188     afs_int32 refCount;
189
190     /* ensure that we're in the LRU queue if our ref count is 0 */
191     osi_assertx(bp->magic == CM_BUF_MAGIC,"incorrect cm_buf_t magic");
192
193     refCount = InterlockedDecrement(&bp->refCount);
194 #ifdef DEBUG_REFCOUNT
195     osi_Log2(afsd_logp,"buf_Release bp 0x%p ref %d", bp, refCount);
196     afsi_log("%s:%d buf_ReleaseLocked bp 0x%p, ref %d", file, line, bp, refCount);
197 #endif
198 #ifdef DEBUG
199     if (refCount < 0)
200         osi_panic("buf refcount 0",__FILE__,__LINE__);;
201 #else
202     osi_assertx(refCount >= 0, "cm_buf_t refCount == 0");
203 #endif
204     if (refCount == 0) {
205         lock_ObtainWrite(&buf_globalLock);
206         if (bp->refCount == 0 && 
207             !(bp->qFlags & CM_BUF_QINLRU)) {
208             osi_QAdd((osi_queue_t **) &cm_data.buf_freeListp, &bp->q);
209
210             /* watch for transition from empty to one element */
211             if (!cm_data.buf_freeListEndp)
212                 cm_data.buf_freeListEndp = cm_data.buf_freeListp;
213             bp->qFlags |= CM_BUF_QINLRU;
214         }
215         lock_ReleaseWrite(&buf_globalLock);
216     }
217 }
218
219 long 
220 buf_Sync(int quitOnShutdown) 
221 {
222     cm_buf_t **bpp, *bp, *prevbp;
223     afs_uint32 wasDirty = 0;
224     cm_req_t req;
225
226     /* go through all of the dirty buffers */
227     lock_ObtainRead(&buf_globalLock);
228     for (bpp = &cm_data.buf_dirtyListp, prevbp = NULL; bp = *bpp; ) {
229         if (quitOnShutdown && buf_ShutdownFlag)
230             break;
231
232         lock_ReleaseRead(&buf_globalLock);
233         /* all dirty buffers are held when they are added to the
234         * dirty list.  No need for an additional hold.
235         */
236         lock_ObtainMutex(&bp->mx);
237
238         if (bp->flags & CM_BUF_DIRTY && !(bp->qFlags & CM_BUF_QREDIR)) {
239             /* start cleaning the buffer; don't touch log pages since
240              * the log code counts on knowing exactly who is writing
241              * a log page at any given instant.
242              *
243              * only attempt to write the buffer if the volume might
244              * be online.
245              */
246             afs_uint32 dirty;
247             cm_volume_t * volp;
248
249             volp = cm_GetVolumeByFID(&bp->fid);
250             switch (cm_GetVolumeStatus(volp, bp->fid.volume)) {
251             case vl_online:
252             case vl_unknown:
253                 cm_InitReq(&req);
254                 req.flags |= CM_REQ_NORETRY;
255                 buf_CleanAsyncLocked(NULL, bp, &req, 0, &dirty);
256                 wasDirty |= dirty;
257             }
258             cm_PutVolume(volp);
259         }
260
261         /* the buffer may or may not have been dirty
262         * and if dirty may or may not have been cleaned
263         * successfully.  check the dirty flag again.  
264         */
265         if (!(bp->flags & CM_BUF_DIRTY)) {
266             /* remove the buffer from the dirty list */
267             lock_ObtainWrite(&buf_globalLock);
268 #ifdef DEBUG_REFCOUNT
269             if (bp->dirtyp == NULL && bp != cm_data.buf_dirtyListEndp) {
270                 osi_Log1(afsd_logp,"buf_IncrSyncer bp 0x%p list corruption",bp);
271                 afsi_log("buf_IncrSyncer bp 0x%p list corruption", bp);
272             }
273 #endif
274             *bpp = bp->dirtyp;
275             bp->dirtyp = NULL;
276             bp->qFlags &= ~CM_BUF_QINDL;
277             if (cm_data.buf_dirtyListp == NULL)
278                 cm_data.buf_dirtyListEndp = NULL;
279             else if (cm_data.buf_dirtyListEndp == bp)
280                 cm_data.buf_dirtyListEndp = prevbp;
281             buf_ReleaseLocked(bp, TRUE);
282             lock_ConvertWToR(&buf_globalLock);
283         } else {
284             if (buf_ShutdownFlag) {
285                 cm_cell_t *cellp;
286                 cm_volume_t *volp;
287                 char volstr[VL_MAXNAMELEN+12]="";
288                 char *ext = "";
289
290                 volp = cm_GetVolumeByFID(&bp->fid);
291                 if (volp) {
292                     cellp = volp->cellp;
293                     if (bp->fid.volume == volp->vol[RWVOL].ID)
294                         ext = "";
295                     else if (bp->fid.volume == volp->vol[ROVOL].ID)
296                         ext = ".readonly";
297                     else if (bp->fid.volume == volp->vol[BACKVOL].ID)
298                         ext = ".backup";
299                     else
300                         ext = ".nomatch";
301                     snprintf(volstr, sizeof(volstr), "%s%s", volp->namep, ext);
302                 } else {
303                     cellp = cm_FindCellByID(bp->fid.cell, CM_FLAG_NOPROBE);
304                     snprintf(volstr, sizeof(volstr), "%u", bp->fid.volume);
305                 }
306
307                 LogEvent(EVENTLOG_INFORMATION_TYPE, MSG_DIRTY_BUFFER_AT_SHUTDOWN, 
308                          cellp->name, volstr, bp->fid.vnode, bp->fid.unique, 
309                          bp->offset.QuadPart+bp->dirty_offset, bp->dirty_length);
310             }
311
312             /* advance the pointer so we don't loop forever */
313             lock_ObtainRead(&buf_globalLock);
314             bpp = &bp->dirtyp;
315             prevbp = bp;
316         }
317         lock_ReleaseMutex(&bp->mx);
318     }   /* for loop over a bunch of buffers */
319     lock_ReleaseRead(&buf_globalLock);
320
321     return wasDirty;
322 }
323
324 /* incremental sync daemon.  Writes all dirty buffers every 5000 ms */
325 void buf_IncrSyncer(long parm)
326 {
327     long wasDirty = 0;
328     long i;
329
330     while (buf_ShutdownFlag == 0) {
331
332         if (!wasDirty) {
333             i = SleepEx(5000, 1);
334             if (i != 0) 
335                 continue;
336         } else {
337             Sleep(50);
338         }
339
340         wasDirty = buf_Sync(1);
341     } /* whole daemon's while loop */
342 }
343
344 long
345 buf_ValidateBuffers(void)
346 {
347     cm_buf_t * bp, *bpf, *bpa, *bpb;
348     afs_uint64 countb = 0, countf = 0, counta = 0;
349
350     if (cm_data.buf_freeListp == NULL && cm_data.buf_freeListEndp != NULL ||
351          cm_data.buf_freeListp != NULL && cm_data.buf_freeListEndp == NULL) {
352         afsi_log("cm_ValidateBuffers failure: inconsistent free list pointers");
353         fprintf(stderr, "cm_ValidateBuffers failure: inconsistent free list pointers\n");
354         return -9;                  
355     }
356
357     for (bp = cm_data.buf_freeListEndp; bp; bp=(cm_buf_t *) osi_QPrev(&bp->q)) { 
358         if (bp->magic != CM_BUF_MAGIC) {
359             afsi_log("cm_ValidateBuffers failure: bp->magic != CM_BUF_MAGIC");
360             fprintf(stderr, "cm_ValidateBuffers failure: bp->magic != CM_BUF_MAGIC\n");
361             return -1;                  
362         }
363         countb++;                                                                
364         bpb = bp;     
365
366         if (countb > cm_data.buf_nbuffers) {
367             afsi_log("cm_ValidateBuffers failure: countb > cm_data.buf_nbuffers");
368             fprintf(stderr, "cm_ValidateBuffers failure: countb > cm_data.buf_nbuffers\n");
369             return -6;                   
370         }
371     }
372
373     for (bp = cm_data.buf_freeListp; bp; bp=(cm_buf_t *) osi_QNext(&bp->q)) { 
374         if (bp->magic != CM_BUF_MAGIC) {
375             afsi_log("cm_ValidateBuffers failure: bp->magic != CM_BUF_MAGIC");
376             fprintf(stderr, "cm_ValidateBuffers failure: bp->magic != CM_BUF_MAGIC\n");
377             return -2;                  
378         }
379         countf++;                                                             
380         bpf = bp;    
381
382         if (countf > cm_data.buf_nbuffers) {
383             afsi_log("cm_ValidateBuffers failure: countf > cm_data.buf_nbuffers");
384             fprintf(stderr, "cm_ValidateBuffers failure: countf > cm_data.buf_nbuffers\n");
385             return -7;
386         }
387     }                                                                         
388                                                                               
389     for (bp = cm_data.buf_allp; bp; bp=bp->allp) {                            
390         if (bp->magic != CM_BUF_MAGIC) {
391             afsi_log("cm_ValidateBuffers failure: bp->magic != CM_BUF_MAGIC");
392             fprintf(stderr, "cm_ValidateBuffers failure: bp->magic != CM_BUF_MAGIC\n");
393             return -3;                  
394         }
395         counta++;                                                             
396         bpa = bp;                                                             
397
398         if (counta > cm_data.buf_nbuffers) {
399             afsi_log("cm_ValidateBuffers failure: counta > cm_data.buf_nbuffers");
400             fprintf(stderr, "cm_ValidateBuffers failure: counta > cm_data.buf_nbuffers\n");
401             return -8;                   
402         }
403     }                                                                         
404                                                                               
405     if (countb != countf) {
406         afsi_log("cm_ValidateBuffers failure: countb != countf");
407         fprintf(stderr, "cm_ValidateBuffers failure: countb != countf\n");
408         return -4;         
409     }
410                                                                               
411     if (counta != cm_data.buf_nbuffers) {
412         afsi_log("cm_ValidateBuffers failure: counta != cm_data.buf_nbuffers");
413         fprintf(stderr, "cm_ValidateBuffers failure: counta != cm_data.buf_nbuffers\n");
414         return -5;                       
415     }
416                                                                               
417     return 0;                                                                 
418 }
419
420 void buf_Shutdown(void)  
421 {  
422     /* disable the buf_IncrSyncer() threads */
423     buf_ShutdownFlag = 1;
424
425     /* then force all dirty buffers to the file servers */
426     buf_Sync(0);
427 }                        
428
429 /* initialize the buffer package; called with no locks
430  * held during the initialization phase.
431  */
432 long buf_Init(int newFile, cm_buf_ops_t *opsp, afs_uint64 nbuffers)
433 {
434     static osi_once_t once;
435     cm_buf_t *bp;
436     thread_t phandle;
437     long i;
438     unsigned long pid;
439     char *data;
440
441     if ( newFile ) {
442         if (nbuffers) 
443             cm_data.buf_nbuffers = nbuffers;
444
445         /* Have to be able to reserve a whole chunk */
446         if (((cm_data.buf_nbuffers - 3) * cm_data.buf_blockSize) < cm_chunkSize)
447             return CM_ERROR_TOOFEWBUFS;
448     }
449
450     /* recall for callouts */
451     cm_buf_opsp = opsp;
452
453     if (osi_Once(&once)) {
454         /* initialize global locks */
455         lock_InitializeRWLock(&buf_globalLock, "Global buffer lock", LOCK_HIERARCHY_BUF_GLOBAL);
456
457         if ( newFile ) {
458             /* remember this for those who want to reset it */
459             cm_data.buf_nOrigBuffers = cm_data.buf_nbuffers;
460  
461             /* lower hash size to a prime number */
462             cm_data.buf_hashSize = osi_PrimeLessThan((afs_uint32)(cm_data.buf_nbuffers/7 + 1));
463  
464             /* create hash table */
465             memset((void *)cm_data.buf_scacheHashTablepp, 0, cm_data.buf_hashSize * sizeof(cm_buf_t *));
466             
467             /* another hash table */
468             memset((void *)cm_data.buf_fileHashTablepp, 0, cm_data.buf_hashSize * sizeof(cm_buf_t *));
469
470             /* create buffer headers and put in free list */
471             bp = cm_data.bufHeaderBaseAddress;
472             data = cm_data.bufDataBaseAddress;
473             cm_data.buf_allp = NULL;
474             
475             for (i=0; i<cm_data.buf_nbuffers; i++) {
476                 osi_assertx(bp >= cm_data.bufHeaderBaseAddress && bp < (cm_buf_t *)cm_data.bufDataBaseAddress, 
477                             "invalid cm_buf_t address");
478                 osi_assertx(data >= cm_data.bufDataBaseAddress && data < cm_data.bufEndOfData,
479                             "invalid cm_buf_t data address");
480                 
481                 /* allocate and zero some storage */
482                 memset(bp, 0, sizeof(cm_buf_t));
483                 bp->magic = CM_BUF_MAGIC;
484                 /* thread on list of all buffers */
485                 bp->allp = cm_data.buf_allp;
486                 cm_data.buf_allp = bp;
487                 
488                 osi_QAdd((osi_queue_t **)&cm_data.buf_freeListp, &bp->q);
489                 bp->qFlags |= CM_BUF_QINLRU;
490                 lock_InitializeMutex(&bp->mx, "Buffer mutex", LOCK_HIERARCHY_BUFFER);
491                 
492                 /* grab appropriate number of bytes from aligned zone */
493                 bp->datap = data;
494                 
495                 /* setup last buffer pointer */
496                 if (i == 0)
497                     cm_data.buf_freeListEndp = bp;
498                     
499                 /* next */
500                 bp++;
501                 data += cm_data.buf_blockSize;
502             }       
503  
504             /* none reserved at first */
505             cm_data.buf_reservedBufs = 0;
506  
507             /* just for safety's sake */
508             cm_data.buf_maxReservedBufs = cm_data.buf_nbuffers - 3;
509         } else {
510             bp = cm_data.bufHeaderBaseAddress;
511             data = cm_data.bufDataBaseAddress;
512             
513             for (i=0; i<cm_data.buf_nbuffers; i++) {
514                 lock_InitializeMutex(&bp->mx, "Buffer mutex", LOCK_HIERARCHY_BUFFER);
515                 bp->userp = NULL;
516                 bp->waitCount = 0;
517                 bp->waitRequests = 0;
518                 bp->flags &= ~CM_BUF_WAITING;
519                 bp++;
520             }       
521         }
522  
523 #ifdef TESTING
524         buf_ValidateBufQueues();
525 #endif /* TESTING */
526
527 #ifdef TRACE_BUFFER
528         /* init the buffer trace log */
529         buf_logp = osi_LogCreate("buffer", 1000);
530         osi_LogEnable(buf_logp);
531 #endif
532
533         osi_EndOnce(&once);
534
535         /* and create the incr-syncer */
536         phandle = thrd_Create(0, 0,
537                                (ThreadFunc) buf_IncrSyncer, 0, 0, &pid,
538                                "buf_IncrSyncer");
539
540         osi_assertx(phandle != NULL, "buf: can't create incremental sync proc");
541         CloseHandle(phandle);
542     }
543
544 #ifdef TESTING
545     buf_ValidateBufQueues();
546 #endif /* TESTING */
547     return 0;
548 }
549
550 /* add nbuffers to the buffer pool, if possible.
551  * Called with no locks held.
552  */
553 long buf_AddBuffers(afs_uint64 nbuffers)
554 {
555     /* The size of a virtual cache cannot be changed after it has
556      * been created.  Subsequent calls to MapViewofFile() with
557      * an existing mapping object name would not allow the 
558      * object to be resized.  Return failure immediately.
559      *
560      * A similar problem now occurs with the persistent cache
561      * given that the memory mapped file now contains a complex
562      * data structure.
563      */
564     afsi_log("request to add %d buffers to the existing cache of size %d denied",
565               nbuffers, cm_data.buf_nbuffers);
566
567     return CM_ERROR_INVAL;
568 }       
569
570 /* interface to set the number of buffers to an exact figure.
571  * Called with no locks held.
572  */
573 long buf_SetNBuffers(afs_uint64 nbuffers)
574 {
575     if (nbuffers < 10) 
576         return CM_ERROR_INVAL;
577     if (nbuffers == cm_data.buf_nbuffers) 
578         return 0;
579     else if (nbuffers > cm_data.buf_nbuffers)
580         return buf_AddBuffers(nbuffers - cm_data.buf_nbuffers);
581     else 
582         return CM_ERROR_INVAL;
583 }
584
585 /* wait for reading or writing to clear; called with write-locked
586  * buffer and unlocked scp and returns with locked buffer.
587  */
588 void buf_WaitIO(cm_scache_t * scp, cm_buf_t *bp)
589 {
590     int release = 0;
591
592     if (scp)
593         osi_assertx(scp->magic == CM_SCACHE_MAGIC, "invalid cm_scache_t magic");
594     osi_assertx(bp->magic == CM_BUF_MAGIC, "invalid cm_buf_t magic");
595
596     while (1) {
597         /* if no IO is happening, we're done */
598         if (!(bp->flags & (CM_BUF_READING | CM_BUF_WRITING)))
599             break;
600                 
601         /* otherwise I/O is happening, but some other thread is waiting for
602          * the I/O already.  Wait for that guy to figure out what happened,
603          * and then check again.
604          */
605         if ( bp->flags & CM_BUF_WAITING ) {
606             bp->waitCount++;
607             bp->waitRequests++;
608             osi_Log1(buf_logp, "buf_WaitIO CM_BUF_WAITING already set for 0x%p", bp);
609         } else {
610             osi_Log1(buf_logp, "buf_WaitIO CM_BUF_WAITING set for 0x%p", bp);
611             bp->flags |= CM_BUF_WAITING;
612             bp->waitCount = bp->waitRequests = 1;
613         }
614         osi_SleepM((LONG_PTR)bp, &bp->mx);
615
616         cm_UpdateServerPriority();
617
618         lock_ObtainMutex(&bp->mx);
619         osi_Log1(buf_logp, "buf_WaitIO conflict wait done for 0x%p", bp);
620         bp->waitCount--;
621         if (bp->waitCount == 0) {
622             osi_Log1(buf_logp, "buf_WaitIO CM_BUF_WAITING reset for 0x%p", bp);
623             bp->flags &= ~CM_BUF_WAITING;
624             bp->waitRequests = 0;
625         }
626
627         if ( !scp ) {
628             if (scp = cm_FindSCache(&bp->fid))
629                  release = 1;
630         }
631         if ( scp ) {
632             lock_ObtainRead(&scp->rw);
633             if (scp->flags & CM_SCACHEFLAG_WAITING) {
634                 osi_Log1(buf_logp, "buf_WaitIO waking scp 0x%p", scp);
635                 osi_Wakeup((LONG_PTR)&scp->flags);
636             }
637             lock_ReleaseRead(&scp->rw);
638         }
639     }
640         
641     /* if we get here, the IO is done, but we may have to wakeup people waiting for
642      * the I/O to complete.  Do so.
643      */
644     if (bp->flags & CM_BUF_WAITING) {
645         osi_Log1(buf_logp, "buf_WaitIO Waking bp 0x%p", bp);
646         osi_Wakeup((LONG_PTR) bp);
647     }
648     osi_Log1(buf_logp, "WaitIO finished wait for bp 0x%p", bp);
649
650     if (scp && release)
651         cm_ReleaseSCache(scp);
652 }
653
654 /* find a buffer, if any, for a particular file ID and offset.  Assumes
655  * that buf_globalLock is write locked when called.
656  */
657 cm_buf_t *buf_FindLocked(struct cm_scache *scp, osi_hyper_t *offsetp)
658 {
659     afs_uint32 i;
660     cm_buf_t *bp;
661
662     i = BUF_HASH(&scp->fid, offsetp);
663     for(bp = cm_data.buf_scacheHashTablepp[i]; bp; bp=bp->hashp) {
664         if (cm_FidCmp(&scp->fid, &bp->fid) == 0
665              && offsetp->LowPart == bp->offset.LowPart
666              && offsetp->HighPart == bp->offset.HighPart) {
667             buf_HoldLocked(bp);
668             break;
669         }
670     }
671         
672     /* return whatever we found, if anything */
673     return bp;
674 }
675
676 /* find a buffer with offset *offsetp for vnode *scp.  Called
677  * with no locks held.
678  */
679 cm_buf_t *buf_Find(struct cm_scache *scp, osi_hyper_t *offsetp)
680 {
681     cm_buf_t *bp;
682
683     lock_ObtainRead(&buf_globalLock);
684     bp = buf_FindLocked(scp, offsetp);
685     lock_ReleaseRead(&buf_globalLock);
686
687     return bp;
688 }       
689
690 /* find a buffer, if any, for a particular file ID and offset.  Assumes
691  * that buf_globalLock is write locked when called.  Uses the all buffer
692  * list.
693  */
694 cm_buf_t *buf_FindAllLocked(struct cm_scache *scp, osi_hyper_t *offsetp, afs_uint32 flags)
695 {
696     cm_buf_t *bp;
697
698     if (flags == 0) {
699         for(bp = cm_data.buf_allp; bp; bp=bp->allp) {
700             if (cm_FidCmp(&scp->fid, &bp->fid) == 0
701                  && offsetp->LowPart == bp->offset.LowPart
702                  && offsetp->HighPart == bp->offset.HighPart) {
703                 buf_HoldLocked(bp);
704                 break;
705             }
706         }
707     } else {
708         for(bp = cm_data.buf_allp; bp; bp=bp->allp) {
709             if (cm_FidCmp(&scp->fid, &bp->fid) == 0) {
710                 char * fileOffset;
711                 
712                 fileOffset = offsetp->QuadPart + cm_data.baseAddress;
713                 if (fileOffset == bp->datap) {
714                     buf_HoldLocked(bp);
715                     break;
716                 }
717             }
718         }
719     }
720     /* return whatever we found, if anything */
721     return bp;
722 }
723
724 /* find a buffer with offset *offsetp for vnode *scp.  Called
725  * with no locks held.  Use the all buffer list.
726  */
727 cm_buf_t *buf_FindAll(struct cm_scache *scp, osi_hyper_t *offsetp, afs_uint32 flags)
728 {
729     cm_buf_t *bp;
730
731     lock_ObtainRead(&buf_globalLock);
732     bp = buf_FindAllLocked(scp, offsetp, flags);
733     lock_ReleaseRead(&buf_globalLock);
734
735     return bp;
736 }       
737
738 /* start cleaning I/O on this buffer.  Buffer must be write locked, and is returned
739  * write-locked.
740  *
741  * Makes sure that there's only one person writing this block
742  * at any given time, and also ensures that the log is forced sufficiently far,
743  * if this buffer contains logged data.
744  *
745  * Returns non-zero if the buffer was dirty.
746  */
747 afs_uint32 buf_CleanAsyncLocked(cm_scache_t *scp, cm_buf_t *bp, cm_req_t *reqp,
748                                 afs_uint32 flags, afs_uint32 *pisdirty)
749 {
750     afs_uint32 code = 0;
751     afs_uint32 isdirty = 0;
752     osi_hyper_t offset;
753     int release_scp = 0;
754
755     osi_assertx(bp->magic == CM_BUF_MAGIC, "invalid cm_buf_t magic");
756
757     if (scp = cm_FindSCache(&bp->fid))
758         release_scp = 1;
759
760     if (!scp) {
761         osi_Log1(buf_logp, "buf_CleanAsyncLocked unable to start I/O - scp not found buf 0x%p", bp);
762         code = CM_ERROR_NOSUCHFILE;
763     }
764
765     while ((bp->flags & CM_BUF_DIRTY) == CM_BUF_DIRTY) {
766         isdirty = 1;
767         lock_ReleaseMutex(&bp->mx);
768
769         osi_Log2(buf_logp, "buf_CleanAsyncLocked starts I/O on scp 0x%p buf 0x%p", scp, bp);
770
771         offset = bp->offset;
772         LargeIntegerAdd(offset, ConvertLongToLargeInteger(bp->dirty_offset));
773         code = (*cm_buf_opsp->Writep)(scp, &offset,
774 #if 1
775                                        /* we might as well try to write all of the contiguous
776                                        * dirty buffers in one RPC
777                                        */
778                                        cm_chunkSize,
779 #else
780                                        bp->dirty_length,
781 #endif
782                                        flags, bp->userp, reqp);
783         osi_Log3(buf_logp, "buf_CleanAsyncLocked I/O on scp 0x%p buf 0x%p, done=%d", scp, bp, code);
784
785         lock_ObtainMutex(&bp->mx);
786         /* if the Write routine returns No Such File, clear the dirty flag
787          * because we aren't going to be able to write this data to the file
788          * server.
789          */
790         if (code == CM_ERROR_NOSUCHFILE || code == CM_ERROR_BADFD || code == CM_ERROR_NOACCESS || 
791             code == CM_ERROR_QUOTA || code == CM_ERROR_SPACE || code == CM_ERROR_TOOBIG || 
792             code == CM_ERROR_READONLY || code == CM_ERROR_NOSUCHPATH){
793             bp->flags &= ~CM_BUF_DIRTY;
794             bp->flags |= CM_BUF_ERROR;
795             bp->dirty_offset = 0;
796             bp->dirty_length = 0;
797             bp->error = code;
798             bp->dataVersion = CM_BUF_VERSION_BAD;
799             bp->dirtyCounter++;
800             break;
801         }
802
803 #ifdef DISKCACHE95
804         /* Disk cache support */
805         /* write buffer to disk cache (synchronous for now) */
806         diskcache_Update(bp->dcp, bp->datap, cm_data.buf_blockSize, bp->dataVersion);
807 #endif /* DISKCACHE95 */
808
809         /* if we get here and retries are not permitted 
810          * then we need to exit this loop regardless of 
811          * whether or not we were able to clear the dirty bit
812          */
813         if (reqp->flags & CM_REQ_NORETRY)
814             break;
815
816         /* Ditto if the hardDeadTimeout or idleTimeout was reached */
817         if (code == CM_ERROR_TIMEDOUT || code == CM_ERROR_ALLDOWN ||
818             code == CM_ERROR_ALLBUSY || code == CM_ERROR_ALLOFFLINE ||
819             code == CM_ERROR_CLOCKSKEW) {
820             break;
821         }
822     }
823
824     if (release_scp)
825         cm_ReleaseSCache(scp);
826
827     /* if someone was waiting for the I/O that just completed or failed,
828      * wake them up.
829      */
830     if (bp->flags & CM_BUF_WAITING) {
831         /* turn off flags and wakeup users */
832         osi_Log1(buf_logp, "buf_WaitIO Waking bp 0x%p", bp);
833         osi_Wakeup((LONG_PTR) bp);
834     }
835
836     if (pisdirty)
837         *pisdirty = isdirty;
838
839     return code;
840 }
841
842 /* Called with a zero-ref count buffer and with the buf_globalLock write locked.
843  * recycles the buffer, and leaves it ready for reuse with a ref count of 1.
844  * The buffer must already be clean, and no I/O should be happening to it.
845  */
846 void buf_Recycle(cm_buf_t *bp)
847 {
848     afs_uint32 i;
849     cm_buf_t **lbpp;
850     cm_buf_t *tbp;
851     cm_buf_t *prevBp, *nextBp;
852
853     osi_assertx(bp->magic == CM_BUF_MAGIC, "invalid cm_buf_t magic");
854
855     /* if we get here, we know that the buffer still has a 0 ref count,
856      * and that it is clean and has no currently pending I/O.  This is
857      * the dude to return.
858      * Remember that as long as the ref count is 0, we know that we won't
859      * have any lock conflicts, so we can grab the buffer lock out of
860      * order in the locking hierarchy.
861      */
862     osi_Log3( buf_logp, "buf_Recycle recycles 0x%p, off 0x%x:%08x",
863               bp, bp->offset.HighPart, bp->offset.LowPart);
864
865     osi_assertx(bp->refCount == 0, "cm_buf_t refcount != 0");
866     osi_assertx(!(bp->flags & (CM_BUF_READING | CM_BUF_WRITING | CM_BUF_DIRTY)),
867                 "incorrect cm_buf_t flags");
868     lock_AssertWrite(&buf_globalLock);
869
870     if (bp->qFlags & CM_BUF_QINHASH) {
871         /* Remove from hash */
872
873         i = BUF_HASH(&bp->fid, &bp->offset);
874         lbpp = &(cm_data.buf_scacheHashTablepp[i]);
875         for(tbp = *lbpp; tbp; lbpp = &tbp->hashp, tbp = *lbpp) {
876             if (tbp == bp) 
877                 break;
878         }
879
880         /* we better find it */
881         osi_assertx(tbp != NULL, "buf_Recycle: hash table screwup");
882
883         *lbpp = bp->hashp;      /* hash out */
884         bp->hashp = NULL;
885
886         /* Remove from file hash */
887
888         i = BUF_FILEHASH(&bp->fid);
889         prevBp = bp->fileHashBackp;
890         bp->fileHashBackp = NULL;
891         nextBp = bp->fileHashp;
892         bp->fileHashp = NULL;
893         if (prevBp)
894             prevBp->fileHashp = nextBp;
895         else
896             cm_data.buf_fileHashTablepp[i] = nextBp;
897         if (nextBp)
898             nextBp->fileHashBackp = prevBp;
899
900         bp->qFlags &= ~CM_BUF_QINHASH;
901     }
902
903     /* make the fid unrecognizable */
904     memset(&bp->fid, 0, sizeof(cm_fid_t));
905
906     /* clean up junk flags */
907     bp->flags &= ~(CM_BUF_EOF | CM_BUF_ERROR);
908     bp->dataVersion = CM_BUF_VERSION_BAD;       /* unknown so far */
909 }       
910
911 /* recycle a buffer, removing it from the free list, hashing in its new identity
912  * and returning it write-locked so that no one can use it.  Called without
913  * any locks held, and can return an error if it loses the race condition and 
914  * finds that someone else created the desired buffer.
915  *
916  * If success is returned, the buffer is returned write-locked.
917  *
918  * May be called with null scp and offsetp, if we're just trying to reclaim some
919  * space from the buffer pool.  In that case, the buffer will be returned
920  * without being hashed into the hash table.
921  */
922 long buf_GetNewLocked(struct cm_scache *scp, osi_hyper_t *offsetp, cm_req_t *reqp, cm_buf_t **bufpp)
923 {
924     cm_buf_t *bp;       /* buffer we're dealing with */
925     cm_buf_t *nextBp;   /* next buffer in file hash chain */
926     afs_uint32 i;       /* temp */
927
928 #ifdef TESTING
929     buf_ValidateBufQueues();
930 #endif /* TESTING */
931
932     while(1) {
933       retry:
934         lock_ObtainRead(&scp->bufCreateLock);
935         lock_ObtainWrite(&buf_globalLock);
936         /* check to see if we lost the race */
937         if (scp) {
938             if (bp = buf_FindLocked(scp, offsetp)) {
939                 /* Do not call buf_ReleaseLocked() because we 
940                  * do not want to allow the buffer to be added
941                  * to the free list.
942                  */
943                 afs_int32 refCount = InterlockedDecrement(&bp->refCount);
944 #ifdef DEBUG_REFCOUNT
945                 osi_Log2(afsd_logp,"buf_GetNewLocked bp 0x%p ref %d", bp, refCount);
946                 afsi_log("%s:%d buf_GetNewLocked bp 0x%p, ref %d", __FILE__, __LINE__, bp, refCount);
947 #endif
948                 lock_ReleaseWrite(&buf_globalLock);
949                 lock_ReleaseRead(&scp->bufCreateLock);
950                 return CM_BUF_EXISTS;
951             }
952         }
953
954         /* does this fix the problem below?  it's a simple solution. */
955         if (!cm_data.buf_freeListEndp)
956         {
957             lock_ReleaseWrite(&buf_globalLock);
958             lock_ReleaseRead(&scp->bufCreateLock);
959             osi_Log0(afsd_logp, "buf_GetNewLocked: Free Buffer List is empty - sleeping 200ms");
960             Sleep(200);
961             goto retry;
962         }
963
964         /* for debugging, assert free list isn't empty, although we
965          * really should try waiting for a running tranasction to finish
966          * instead of this; or better, we should have a transaction
967          * throttler prevent us from entering this situation.
968          */
969         osi_assertx(cm_data.buf_freeListEndp != NULL, "buf_GetNewLocked: no free buffers");
970
971         /* look at all buffers in free list, some of which may temp.
972          * have high refcounts and which then should be skipped,
973          * starting cleaning I/O for those which are dirty.  If we find
974          * a clean buffer, we rehash it, lock it and return it.
975          */
976         for(bp = cm_data.buf_freeListEndp; bp; bp=(cm_buf_t *) osi_QPrev(&bp->q)) {
977             /* check to see if it really has zero ref count.  This
978              * code can bump refcounts, at least, so it may not be
979              * zero.
980              */
981             if (bp->refCount > 0) 
982                 continue;
983                         
984             /* we don't have to lock buffer itself, since the ref
985              * count is 0 and we know it will stay zero as long as
986              * we hold the global lock.
987              */
988
989             /* Don't recycle a buffer held by the redirector. */
990             if (bp->qFlags & CM_BUF_QREDIR)
991                 continue;
992
993             /* don't recycle someone in our own chunk */
994             if (!cm_FidCmp(&bp->fid, &scp->fid)
995                  && (bp->offset.LowPart & (-cm_chunkSize))
996                  == (offsetp->LowPart & (-cm_chunkSize)))
997                 continue;
998
999             /* if this page is being filled (!) or cleaned, see if
1000              * the I/O has completed.  If not, skip it, otherwise
1001              * do the final processing for the I/O.
1002              */
1003             if (bp->flags & (CM_BUF_READING | CM_BUF_WRITING)) {
1004                 /* probably shouldn't do this much work while
1005                  * holding the big lock?  Watch for contention
1006                  * here.
1007                  */
1008                 continue;
1009             }
1010                         
1011             if (bp->flags & CM_BUF_DIRTY) {
1012                 /* if the buffer is dirty, start cleaning it and
1013                  * move on to the next buffer.  We do this with
1014                  * just the lock required to minimize contention
1015                  * on the big lock.
1016                  */
1017                 buf_HoldLocked(bp);
1018                 lock_ReleaseWrite(&buf_globalLock);
1019                 lock_ReleaseRead(&scp->bufCreateLock);
1020
1021                 /* grab required lock and clean; this only
1022                  * starts the I/O.  By the time we're back,
1023                  * it'll still be marked dirty, but it will also
1024                  * have the WRITING flag set, so we won't get
1025                  * back here.
1026                  */
1027                 buf_CleanAsync(scp, bp, reqp, 0, NULL);
1028
1029                 /* now put it back and go around again */
1030                 buf_Release(bp);
1031                 goto retry;
1032             }
1033
1034             /* if we get here, we know that the buffer still has a 0
1035              * ref count, and that it is clean and has no currently
1036              * pending I/O.  This is the dude to return.
1037              * Remember that as long as the ref count is 0, we know
1038              * that we won't have any lock conflicts, so we can grab
1039              * the buffer lock out of order in the locking hierarchy.
1040              */
1041             buf_Recycle(bp);
1042
1043             /* now hash in as our new buffer, and give it the
1044              * appropriate label, if requested.
1045              */
1046             if (scp) {
1047                 bp->qFlags |= CM_BUF_QINHASH;
1048                 bp->fid = scp->fid;
1049 #ifdef DEBUG
1050                 bp->scp = scp;
1051 #endif
1052                 bp->offset = *offsetp;
1053                 i = BUF_HASH(&scp->fid, offsetp);
1054                 bp->hashp = cm_data.buf_scacheHashTablepp[i];
1055                 cm_data.buf_scacheHashTablepp[i] = bp;
1056                 i = BUF_FILEHASH(&scp->fid);
1057                 nextBp = cm_data.buf_fileHashTablepp[i];
1058                 bp->fileHashp = nextBp;
1059                 bp->fileHashBackp = NULL;
1060                 if (nextBp)
1061                     nextBp->fileHashBackp = bp;
1062                 cm_data.buf_fileHashTablepp[i] = bp;
1063             }
1064
1065             /* we should move it from the lru queue.  It better still be there,
1066              * since we've held the global (big) lock since we found it there.
1067              */
1068             osi_assertx(bp->qFlags & CM_BUF_QINLRU,
1069                          "buf_GetNewLocked: LRU screwup");
1070
1071             if (cm_data.buf_freeListEndp == bp) {
1072                 /* we're the last guy in this queue, so maintain it */
1073                 cm_data.buf_freeListEndp = (cm_buf_t *) osi_QPrev(&bp->q);
1074             }
1075             osi_QRemove((osi_queue_t **) &cm_data.buf_freeListp, &bp->q);
1076             bp->qFlags &= ~CM_BUF_QINLRU;
1077
1078             /* prepare to return it.  Give it a refcount */
1079             bp->refCount = 1;
1080 #ifdef DEBUG_REFCOUNT
1081             osi_Log2(afsd_logp,"buf_GetNewLocked bp 0x%p ref %d", bp, 1);
1082             afsi_log("%s:%d buf_GetNewLocked bp 0x%p, ref %d", __FILE__, __LINE__, bp, 1);
1083 #endif
1084             /* grab the mutex so that people don't use it
1085              * before the caller fills it with data.  Again, no one     
1086              * should have been able to get to this dude to lock it.
1087              */
1088             if (!lock_TryMutex(&bp->mx)) {
1089                 osi_Log2(afsd_logp, "buf_GetNewLocked bp 0x%p cannot be mutex locked.  refCount %d should be 0",
1090                          bp, bp->refCount);
1091                 osi_panic("buf_GetNewLocked: TryMutex failed",__FILE__,__LINE__);
1092             }
1093
1094             lock_ReleaseWrite(&buf_globalLock);
1095             lock_ReleaseRead(&scp->bufCreateLock);
1096
1097             *bufpp = bp;
1098
1099 #ifdef TESTING
1100             buf_ValidateBufQueues();
1101 #endif /* TESTING */
1102             return 0;
1103         } /* for all buffers in lru queue */
1104         lock_ReleaseWrite(&buf_globalLock);
1105         lock_ReleaseRead(&scp->bufCreateLock);
1106         osi_Log0(afsd_logp, "buf_GetNewLocked: Free Buffer List has no buffers with a zero refcount - sleeping 100ms");
1107         Sleep(100);             /* give some time for a buffer to be freed */
1108     }   /* while loop over everything */
1109     /* not reached */
1110 } /* the proc */
1111
1112 /* get a page, returning it held but unlocked.  Doesn't fill in the page
1113  * with I/O, since we're going to write the whole thing new.
1114  */
1115 long buf_GetNew(struct cm_scache *scp, osi_hyper_t *offsetp, cm_req_t *reqp, cm_buf_t **bufpp)
1116 {
1117     cm_buf_t *bp;
1118     long code;
1119     osi_hyper_t pageOffset;
1120     int created;
1121
1122     created = 0;
1123     pageOffset.HighPart = offsetp->HighPart;
1124     pageOffset.LowPart = offsetp->LowPart & ~(cm_data.buf_blockSize-1);
1125     while (1) {
1126         bp = buf_Find(scp, &pageOffset);
1127         if (bp) {
1128             /* lock it and break out */
1129             lock_ObtainMutex(&bp->mx);
1130             break;
1131         }
1132
1133         /* otherwise, we have to create a page */
1134         code = buf_GetNewLocked(scp, &pageOffset, reqp, &bp);
1135
1136         /* check if the buffer was created in a race condition branch.
1137          * If so, go around so we can hold a reference to it. 
1138          */
1139         if (code == CM_BUF_EXISTS) 
1140             continue;
1141
1142         /* something else went wrong */
1143         if (code != 0) 
1144             return code;
1145
1146         /* otherwise, we have a locked buffer that we just created */
1147         created = 1;
1148         break;
1149     } /* big while loop */
1150
1151     /* wait for reads */
1152     if (bp->flags & CM_BUF_READING)
1153         buf_WaitIO(scp, bp);
1154
1155     /* once it has been read once, we can unlock it and return it, still
1156      * with its refcount held.
1157      */
1158     lock_ReleaseMutex(&bp->mx);
1159     *bufpp = bp;
1160     osi_Log4(buf_logp, "buf_GetNew returning bp 0x%p for scp 0x%p, offset 0x%x:%08x",
1161               bp, scp, offsetp->HighPart, offsetp->LowPart);
1162     return 0;
1163 }
1164
1165 /* get a page, returning it held but unlocked.  Make sure it is complete */
1166 /* The scp must be unlocked when passed to this function */
1167 long buf_Get(struct cm_scache *scp, osi_hyper_t *offsetp, cm_req_t *reqp, cm_buf_t **bufpp)
1168 {
1169     cm_buf_t *bp;
1170     long code;
1171     osi_hyper_t pageOffset;
1172     unsigned long tcount;
1173     int created;
1174     long lcount = 0;
1175 #ifdef DISKCACHE95
1176     cm_diskcache_t *dcp;
1177 #endif /* DISKCACHE95 */
1178
1179     created = 0;
1180     pageOffset.HighPart = offsetp->HighPart;
1181     pageOffset.LowPart = offsetp->LowPart & ~(cm_data.buf_blockSize-1);
1182     while (1) {
1183         lcount++;
1184 #ifdef TESTING
1185         buf_ValidateBufQueues();
1186 #endif /* TESTING */
1187
1188         bp = buf_Find(scp, &pageOffset);
1189         if (bp) {
1190             /* lock it and break out */
1191             lock_ObtainMutex(&bp->mx);
1192
1193 #ifdef DISKCACHE95
1194             /* touch disk chunk to update LRU info */
1195             diskcache_Touch(bp->dcp);
1196 #endif /* DISKCACHE95 */
1197             break;
1198         }
1199
1200         /* otherwise, we have to create a page */
1201         code = buf_GetNewLocked(scp, &pageOffset, reqp, &bp);
1202         /* bp->mx is now held */
1203
1204         /* check if the buffer was created in a race condition branch.
1205          * If so, go around so we can hold a reference to it. 
1206          */
1207         if (code == CM_BUF_EXISTS) 
1208             continue;
1209
1210         /* something else went wrong */
1211         if (code != 0) { 
1212 #ifdef TESTING
1213             buf_ValidateBufQueues();
1214 #endif /* TESTING */
1215             return code;
1216         }
1217                 
1218         /* otherwise, we have a locked buffer that we just created */
1219         created = 1;
1220         break;
1221     } /* big while loop */
1222
1223     /* if we get here, we have a locked buffer that may have just been
1224      * created, in which case it needs to be filled with data.
1225      */
1226     if (created) {
1227         /* load the page; freshly created pages should be idle */
1228         osi_assertx(!(bp->flags & (CM_BUF_READING | CM_BUF_WRITING)), "incorrect cm_buf_t flags");
1229
1230         /* start the I/O; may drop lock */
1231         bp->flags |= CM_BUF_READING;
1232         code = (*cm_buf_opsp->Readp)(bp, cm_data.buf_blockSize, &tcount, NULL);
1233
1234 #ifdef DISKCACHE95
1235         code = diskcache_Get(&bp->fid, &bp->offset, bp->datap, cm_data.buf_blockSize, &bp->dataVersion, &tcount, &dcp);
1236         bp->dcp = dcp;    /* pointer to disk cache struct. */
1237 #endif /* DISKCACHE95 */
1238
1239         if (code != 0) {
1240             /* failure or queued */
1241             if (code != ERROR_IO_PENDING) {
1242                 bp->error = code;
1243                 bp->flags |= CM_BUF_ERROR;
1244                 bp->flags &= ~CM_BUF_READING;
1245                 if (bp->flags & CM_BUF_WAITING) {
1246                     osi_Log1(buf_logp, "buf_Get Waking bp 0x%p", bp);
1247                     osi_Wakeup((LONG_PTR) bp);
1248                 }
1249                 lock_ReleaseMutex(&bp->mx);
1250                 buf_Release(bp);
1251 #ifdef TESTING
1252                 buf_ValidateBufQueues();
1253 #endif /* TESTING */
1254                 return code;
1255             }
1256         } else {
1257             /* otherwise, I/O completed instantly and we're done, except
1258              * for padding the xfr out with 0s and checking for EOF
1259              */
1260             if (tcount < (unsigned long) cm_data.buf_blockSize) {
1261                 memset(bp->datap+tcount, 0, cm_data.buf_blockSize - tcount);
1262                 if (tcount == 0)
1263                     bp->flags |= CM_BUF_EOF;
1264             }
1265             bp->flags &= ~CM_BUF_READING;
1266             if (bp->flags & CM_BUF_WAITING) {
1267                 osi_Log1(buf_logp, "buf_Get Waking bp 0x%p", bp);
1268                 osi_Wakeup((LONG_PTR) bp);
1269             }
1270         }
1271
1272     } /* if created */
1273
1274     /* wait for reads, either that which we started above, or that someone
1275      * else started.  We don't care if we return a buffer being cleaned.
1276      */
1277     if (bp->flags & CM_BUF_READING)
1278         buf_WaitIO(scp, bp);
1279
1280     /* once it has been read once, we can unlock it and return it, still
1281      * with its refcount held.
1282      */
1283     lock_ReleaseMutex(&bp->mx);
1284     *bufpp = bp;
1285
1286     /* now remove from queue; will be put in at the head (farthest from
1287      * being recycled) when we're done in buf_Release.
1288      */
1289     lock_ObtainWrite(&buf_globalLock);
1290     if (bp->qFlags & CM_BUF_QINLRU) {
1291         if (cm_data.buf_freeListEndp == bp)
1292             cm_data.buf_freeListEndp = (cm_buf_t *) osi_QPrev(&bp->q);
1293         osi_QRemove((osi_queue_t **) &cm_data.buf_freeListp, &bp->q);
1294         bp->qFlags &= ~CM_BUF_QINLRU;
1295     }
1296     lock_ReleaseWrite(&buf_globalLock);
1297
1298     osi_Log4(buf_logp, "buf_Get returning bp 0x%p for scp 0x%p, offset 0x%x:%08x",
1299               bp, scp, offsetp->HighPart, offsetp->LowPart);
1300 #ifdef TESTING
1301     buf_ValidateBufQueues();
1302 #endif /* TESTING */
1303     return 0;
1304 }
1305
1306 /* count # of elements in the free list;
1307  * we don't bother doing the proper locking for accessing dataVersion or flags
1308  * since it is a pain, and this is really just an advisory call.  If you need
1309  * to do better at some point, rewrite this function.
1310  */
1311 long buf_CountFreeList(void)
1312 {
1313     long count;
1314     cm_buf_t *bufp;
1315
1316     count = 0;
1317     lock_ObtainRead(&buf_globalLock);
1318     for(bufp = cm_data.buf_freeListp; bufp; bufp = (cm_buf_t *) osi_QNext(&bufp->q)) {
1319         /* if the buffer doesn't have an identity, or if the buffer
1320          * has been invalidate (by having its DV stomped upon), then
1321          * count it as free, since it isn't really being utilized.
1322          */
1323         if (!(bufp->qFlags & CM_BUF_QINHASH) || bufp->dataVersion == CM_BUF_VERSION_BAD)
1324             count++;
1325     }       
1326     lock_ReleaseRead(&buf_globalLock);
1327     return count;
1328 }
1329
1330 /* clean a buffer synchronously */
1331 afs_uint32 buf_CleanAsync(cm_scache_t *scp, cm_buf_t *bp, cm_req_t *reqp, afs_uint32 flags, afs_uint32 *pisdirty)
1332 {
1333     long code;
1334     osi_assertx(bp->magic == CM_BUF_MAGIC, "invalid cm_buf_t magic");
1335     osi_assertx(!(flags & CM_BUF_WRITE_SCP_LOCKED), "scp->rw must not be held when calling buf_CleanAsync");
1336
1337     lock_ObtainMutex(&bp->mx);
1338     code = buf_CleanAsyncLocked(scp, bp, reqp, flags, pisdirty);
1339     lock_ReleaseMutex(&bp->mx);
1340
1341     return code;
1342 }       
1343
1344 /* wait for a buffer's cleaning to finish */
1345 void buf_CleanWait(cm_scache_t * scp, cm_buf_t *bp, afs_uint32 locked)
1346 {
1347     osi_assertx(bp->magic == CM_BUF_MAGIC, "invalid cm_buf_t magic");
1348
1349     if (!locked)
1350         lock_ObtainMutex(&bp->mx);
1351     if (bp->flags & CM_BUF_WRITING) {
1352         buf_WaitIO(scp, bp);
1353     }
1354     if (!locked)
1355         lock_ReleaseMutex(&bp->mx);
1356 }       
1357
1358 /* set the dirty flag on a buffer, and set associated write-ahead log,
1359  * if there is one.  Allow one to be added to a buffer, but not changed.
1360  *
1361  * The buffer must be locked before calling this routine.
1362  */
1363 void buf_SetDirty(cm_buf_t *bp, afs_uint32 offset, afs_uint32 length, cm_user_t *userp)
1364 {
1365     osi_assertx(bp->magic == CM_BUF_MAGIC, "invalid cm_buf_t magic");
1366     osi_assertx(bp->refCount > 0, "cm_buf_t refcount 0");
1367
1368     if (bp->flags & CM_BUF_DIRTY) {
1369
1370         osi_Log1(buf_logp, "buf_SetDirty 0x%p already dirty", bp);
1371
1372         if (bp->dirty_offset <= offset) {
1373             if (bp->dirty_offset + bp->dirty_length >= offset + length) {
1374                 /* dirty_length remains the same */
1375             } else {
1376                 bp->dirty_length = offset + length - bp->dirty_offset;
1377             }
1378         } else /* bp->dirty_offset > offset */ {
1379             if (bp->dirty_offset + bp->dirty_length >= offset + length) {
1380                 bp->dirty_length = bp->dirty_offset + bp->dirty_length - offset;
1381             } else {
1382                 bp->dirty_length = length;
1383             }
1384             bp->dirty_offset = offset;
1385         }
1386     } else {
1387         osi_Log1(buf_logp, "buf_SetDirty 0x%p", bp);
1388
1389         /* set dirty bit */
1390         bp->flags |= CM_BUF_DIRTY;
1391
1392         /* and turn off EOF flag, since it has associated data now */
1393         bp->flags &= ~CM_BUF_EOF;
1394
1395         bp->dirty_offset = offset;
1396         bp->dirty_length = length;
1397
1398         /* and add to the dirty list.  
1399          * we obtain a hold on the buffer for as long as it remains 
1400          * in the list.  buffers are only removed from the list by 
1401          * the buf_IncrSyncer function regardless of when else the
1402          * dirty flag might be cleared.
1403          *
1404          * This should never happen but just in case there is a bug
1405          * elsewhere, never add to the dirty list if the buffer is 
1406          * already there.
1407          */
1408         lock_ObtainWrite(&buf_globalLock);
1409         if (!(bp->qFlags & CM_BUF_QINDL)) {
1410             buf_HoldLocked(bp);
1411             if (!cm_data.buf_dirtyListp) {
1412                 cm_data.buf_dirtyListp = cm_data.buf_dirtyListEndp = bp;
1413             } else {
1414                 cm_data.buf_dirtyListEndp->dirtyp = bp;
1415                 cm_data.buf_dirtyListEndp = bp;
1416             }
1417             bp->dirtyp = NULL;
1418             bp->qFlags |= CM_BUF_QINDL;
1419         }
1420         lock_ReleaseWrite(&buf_globalLock);
1421     }
1422
1423     /* and record the last writer */
1424     if (bp->userp != userp) {
1425         cm_HoldUser(userp);
1426         if (bp->userp) 
1427             cm_ReleaseUser(bp->userp);
1428         bp->userp = userp;
1429     }
1430 }
1431
1432 /* clean all buffers, reset log pointers and invalidate all buffers.
1433  * Called with no locks held, and returns with same.
1434  *
1435  * This function is guaranteed to clean and remove the log ptr of all the
1436  * buffers that were dirty or had non-zero log ptrs before the call was
1437  * made.  That's sufficient to clean up any garbage left around by recovery,
1438  * which is all we're counting on this for; there may be newly created buffers
1439  * added while we're running, but that should be OK.
1440  *
1441  * In an environment where there are no transactions (artificially imposed, for
1442  * example, when switching the database to raw mode), this function is used to
1443  * make sure that all updates have been written to the disk.  In that case, we don't
1444  * really require that we forget the log association between pages and logs, but
1445  * it also doesn't hurt.  Since raw mode I/O goes through this buffer package, we don't
1446  * have to worry about invalidating data in the buffers.
1447  *
1448  * This function is used at the end of recovery as paranoia to get the recovered
1449  * database out to disk.  It removes all references to the recovery log and cleans
1450  * all buffers.
1451  */
1452 long buf_CleanAndReset(void)
1453 {
1454     afs_uint32 i;
1455     cm_buf_t *bp;
1456     cm_req_t req;
1457
1458     lock_ObtainRead(&buf_globalLock);
1459     for(i=0; i<cm_data.buf_hashSize; i++) {
1460         for(bp = cm_data.buf_scacheHashTablepp[i]; bp; bp = bp->hashp) {
1461             if ((bp->flags & CM_BUF_DIRTY) == CM_BUF_DIRTY) {
1462                 buf_HoldLocked(bp);
1463                 lock_ReleaseRead(&buf_globalLock);
1464
1465                 /* now no locks are held; clean buffer and go on */
1466                 cm_InitReq(&req);
1467                 req.flags |= CM_REQ_NORETRY;
1468
1469                 buf_CleanAsync(NULL, bp, &req, 0, NULL);
1470                 buf_CleanWait(NULL, bp, FALSE);
1471
1472                 /* relock and release buffer */
1473                 lock_ObtainRead(&buf_globalLock);
1474                 buf_ReleaseLocked(bp, FALSE);
1475             } /* dirty */
1476         } /* over one bucket */
1477     }   /* for loop over all hash buckets */
1478
1479     /* release locks */
1480     lock_ReleaseRead(&buf_globalLock);
1481
1482 #ifdef TESTING
1483     buf_ValidateBufQueues();
1484 #endif /* TESTING */
1485     
1486     /* and we're done */
1487     return 0;
1488 }       
1489
1490 /* called without global lock being held, reserves buffers for callers
1491  * that need more than one held (not locked) at once.
1492  */
1493 void buf_ReserveBuffers(afs_uint64 nbuffers)
1494 {
1495     lock_ObtainWrite(&buf_globalLock);
1496     while (1) {
1497         if (cm_data.buf_reservedBufs + nbuffers > cm_data.buf_maxReservedBufs) {
1498             cm_data.buf_reserveWaiting = 1;
1499             osi_Log1(buf_logp, "buf_ReserveBuffers waiting for %d bufs", nbuffers);
1500             osi_SleepW((LONG_PTR) &cm_data.buf_reservedBufs, &buf_globalLock);
1501             lock_ObtainWrite(&buf_globalLock);
1502         }
1503         else {
1504             cm_data.buf_reservedBufs += nbuffers;
1505             break;
1506         }
1507     }
1508     lock_ReleaseWrite(&buf_globalLock);
1509 }
1510
1511 int buf_TryReserveBuffers(afs_uint64 nbuffers)
1512 {
1513     int code;
1514
1515     lock_ObtainWrite(&buf_globalLock);
1516     if (cm_data.buf_reservedBufs + nbuffers > cm_data.buf_maxReservedBufs) {
1517         code = 0;
1518     }
1519     else {
1520         cm_data.buf_reservedBufs += nbuffers;
1521         code = 1;
1522     }
1523     lock_ReleaseWrite(&buf_globalLock);
1524     return code;
1525 }       
1526
1527 /* called without global lock held, releases reservation held by
1528  * buf_ReserveBuffers.
1529  */
1530 void buf_UnreserveBuffers(afs_uint64 nbuffers)
1531 {
1532     lock_ObtainWrite(&buf_globalLock);
1533     cm_data.buf_reservedBufs -= nbuffers;
1534     if (cm_data.buf_reserveWaiting) {
1535         cm_data.buf_reserveWaiting = 0;
1536         osi_Wakeup((LONG_PTR) &cm_data.buf_reservedBufs);
1537     }
1538     lock_ReleaseWrite(&buf_globalLock);
1539 }       
1540
1541 /* truncate the buffers past sizep, zeroing out the page, if we don't
1542  * end on a page boundary.
1543  *
1544  * Requires cm_bufCreateLock to be write locked.
1545  */
1546 long buf_Truncate(cm_scache_t *scp, cm_user_t *userp, cm_req_t *reqp,
1547                    osi_hyper_t *sizep)
1548 {
1549     cm_buf_t *bufp;
1550     cm_buf_t *nbufp;                    /* next buffer, if didRelease */
1551     osi_hyper_t bufEnd;
1552     long code;
1553     long bufferPos;
1554     afs_uint32 i;
1555
1556     /* assert that cm_bufCreateLock is held in write mode */
1557     lock_AssertWrite(&scp->bufCreateLock);
1558
1559     i = BUF_FILEHASH(&scp->fid);
1560
1561     lock_ObtainRead(&buf_globalLock);
1562     bufp = cm_data.buf_fileHashTablepp[i];
1563     if (bufp == NULL) {
1564         lock_ReleaseRead(&buf_globalLock);
1565         return 0;
1566     }
1567
1568     buf_HoldLocked(bufp);
1569     lock_ReleaseRead(&buf_globalLock);
1570     while (bufp) {
1571         lock_ObtainMutex(&bufp->mx);
1572
1573         bufEnd.HighPart = 0;
1574         bufEnd.LowPart = cm_data.buf_blockSize;
1575         bufEnd = LargeIntegerAdd(bufEnd, bufp->offset);
1576
1577         if (cm_FidCmp(&bufp->fid, &scp->fid) == 0 &&
1578              LargeIntegerLessThan(*sizep, bufEnd)) {
1579             buf_WaitIO(scp, bufp);
1580         }
1581         lock_ObtainWrite(&scp->rw);
1582         
1583         /* make sure we have a callback (so we have the right value for
1584          * the length), and wait for it to be safe to do a truncate.
1585          */
1586         code = cm_SyncOp(scp, bufp, userp, reqp, 0,
1587                           CM_SCACHESYNC_NEEDCALLBACK
1588                           | CM_SCACHESYNC_GETSTATUS
1589                           | CM_SCACHESYNC_SETSIZE
1590                           | CM_SCACHESYNC_BUFLOCKED);
1591
1592         
1593         /* if we succeeded in our locking, and this applies to the right
1594          * file, and the truncate request overlaps the buffer either
1595          * totally or partially, then do something.
1596          */
1597         if (code == 0 && cm_FidCmp(&bufp->fid, &scp->fid) == 0
1598              && LargeIntegerLessThan(*sizep, bufEnd)) {
1599
1600
1601             /* destroy the buffer, turning off its dirty bit, if
1602              * we're truncating the whole buffer.  Otherwise, set
1603              * the dirty bit, and clear out the tail of the buffer
1604              * if we just overlap some.
1605              */
1606             if (LargeIntegerLessThanOrEqualTo(*sizep, bufp->offset)) {
1607                 /* truncating the entire page */
1608                 bufp->flags &= ~CM_BUF_DIRTY;
1609                 bufp->dirty_offset = 0;
1610                 bufp->dirty_length = 0;
1611                 bufp->dataVersion = CM_BUF_VERSION_BAD; /* known bad */
1612                 bufp->dirtyCounter++;
1613             }
1614             else {
1615                 /* don't set dirty, since dirty implies
1616                  * currently up-to-date.  Don't need to do this,
1617                  * since we'll update the length anyway.
1618                  *
1619                  * Zero out remainder of the page, in case we
1620                  * seek and write past EOF, and make this data
1621                  * visible again.
1622                  */
1623                 bufferPos = sizep->LowPart & (cm_data.buf_blockSize - 1);
1624                 osi_assertx(bufferPos != 0, "non-zero bufferPos");
1625                 memset(bufp->datap + bufferPos, 0,
1626                         cm_data.buf_blockSize - bufferPos);
1627             }
1628         }
1629                 
1630         cm_SyncOpDone( scp, bufp, 
1631                        CM_SCACHESYNC_NEEDCALLBACK | CM_SCACHESYNC_GETSTATUS
1632                        | CM_SCACHESYNC_SETSIZE | CM_SCACHESYNC_BUFLOCKED);
1633
1634         lock_ReleaseWrite(&scp->rw);
1635         lock_ReleaseMutex(&bufp->mx);
1636     
1637         if (!code) {
1638             nbufp = bufp->fileHashp;
1639             if (nbufp) 
1640                 buf_Hold(nbufp);
1641         } else {
1642             /* This forces the loop to end and the error code
1643              * to be returned. */
1644             nbufp = NULL;
1645         }
1646         buf_Release(bufp);
1647         bufp = nbufp;
1648     }
1649
1650 #ifdef TESTING
1651     buf_ValidateBufQueues();
1652 #endif /* TESTING */
1653
1654     /* done */
1655     return code;
1656 }
1657
1658 long buf_FlushCleanPages(cm_scache_t *scp, cm_user_t *userp, cm_req_t *reqp)
1659 {
1660     long code;
1661     cm_buf_t *bp;               /* buffer we're hacking on */
1662     cm_buf_t *nbp;
1663     int didRelease;
1664     afs_uint32 i;
1665     afs_uint32 stable = 0;
1666
1667     i = BUF_FILEHASH(&scp->fid);
1668
1669     code = 0;
1670     lock_ObtainRead(&buf_globalLock);
1671     bp = cm_data.buf_fileHashTablepp[i];
1672     if (bp) 
1673         buf_HoldLocked(bp);
1674     lock_ReleaseRead(&buf_globalLock);
1675
1676     for (; bp; bp = nbp) {
1677         didRelease = 0; /* haven't released this buffer yet */
1678
1679         /* clean buffer synchronously */
1680         if (cm_FidCmp(&bp->fid, &scp->fid) == 0) {
1681
1682             if (code == 0 && !stable && (bp->flags & CM_BUF_DIRTY)) {
1683                 /*
1684                  * we must stabilize the object to ensure that buffer
1685                  * changes cannot occur while the flush is performed.
1686                  * However, we do not want to Stabilize if we do not
1687                  * need to because Stabilize obtains a callback.
1688                  */
1689                 code = (*cm_buf_opsp->Stabilizep)(scp, userp, reqp);
1690                 stable = (code == 0);
1691             }
1692
1693             if (code == CM_ERROR_BADFD) {
1694                 /* if the scp's FID is bad its because we received VNOVNODE 
1695                  * when attempting to FetchStatus before the write.  This
1696                  * page therefore contains data that can no longer be stored.
1697                  */
1698                 lock_ObtainMutex(&bp->mx);
1699                 bp->flags &= ~CM_BUF_DIRTY;
1700                 bp->flags |= CM_BUF_ERROR;
1701                 bp->error = CM_ERROR_BADFD;
1702                 bp->dirty_offset = 0;
1703                 bp->dirty_length = 0;
1704                 bp->dataVersion = CM_BUF_VERSION_BAD;   /* known bad */
1705                 bp->dirtyCounter++;
1706                 lock_ReleaseMutex(&bp->mx);
1707             } else if (!(scp->flags & CM_SCACHEFLAG_RO)) {
1708                 if (code) {
1709                     goto skip;
1710                 }
1711
1712                 lock_ObtainMutex(&bp->mx);
1713
1714                 /* start cleaning the buffer, and wait for it to finish */
1715                 buf_CleanAsyncLocked(scp, bp, reqp, 0, NULL);
1716                 buf_WaitIO(scp, bp);
1717
1718                 lock_ReleaseMutex(&bp->mx);
1719             }
1720
1721             /* actually, we only know that buffer is clean if ref
1722              * count is 1, since we don't have buffer itself locked.
1723              */
1724             if (!(bp->flags & CM_BUF_DIRTY)) {
1725                 lock_ObtainWrite(&buf_globalLock);
1726                 if (bp->refCount == 1) {        /* bp is held above */
1727                     nbp = bp->fileHashp;
1728                     if (nbp) 
1729                         buf_HoldLocked(nbp);
1730                     buf_ReleaseLocked(bp, TRUE);
1731                     didRelease = 1;
1732                     buf_Recycle(bp);
1733                 }
1734                 lock_ReleaseWrite(&buf_globalLock);
1735             }
1736         }
1737
1738       skip:
1739         if (!didRelease) {
1740             lock_ObtainRead(&buf_globalLock);
1741             nbp = bp->fileHashp;
1742             if (nbp)
1743                 buf_HoldLocked(nbp);
1744             buf_ReleaseLocked(bp, FALSE);
1745             lock_ReleaseRead(&buf_globalLock);
1746         }
1747     }   /* for loop over a bunch of buffers */
1748
1749     if (stable)
1750         (*cm_buf_opsp->Unstabilizep)(scp, userp);
1751
1752 #ifdef TESTING
1753     buf_ValidateBufQueues();
1754 #endif /* TESTING */
1755
1756     /* done */
1757     return code;
1758 }       
1759
1760 /* Must be called with scp->rw held */
1761 long buf_ForceDataVersion(cm_scache_t * scp, afs_uint64 fromVersion, afs_uint64 toVersion)
1762 {
1763     cm_buf_t * bp;
1764     afs_uint32 i;
1765     int found = 0;
1766
1767     lock_AssertAny(&scp->rw);
1768
1769     i = BUF_FILEHASH(&scp->fid);
1770
1771     lock_ObtainRead(&buf_globalLock);
1772
1773     for (bp = cm_data.buf_fileHashTablepp[i]; bp; bp = bp->fileHashp) {
1774         if (cm_FidCmp(&bp->fid, &scp->fid) == 0) {
1775             if (bp->dataVersion == fromVersion) {
1776                 bp->dataVersion = toVersion;
1777                 found = 1;
1778             }
1779         }
1780     }
1781     lock_ReleaseRead(&buf_globalLock);
1782
1783     if (found)
1784         return 0;
1785     else
1786         return ENOENT;
1787 }
1788
1789 long buf_CleanVnode(struct cm_scache *scp, cm_user_t *userp, cm_req_t *reqp)
1790 {
1791     long code = 0;
1792     long wasDirty = 0;
1793     cm_buf_t *bp;               /* buffer we're hacking on */
1794     cm_buf_t *nbp;              /* next one */
1795     afs_uint32 i;
1796
1797     i = BUF_FILEHASH(&scp->fid);
1798
1799     lock_ObtainRead(&buf_globalLock);
1800     bp = cm_data.buf_fileHashTablepp[i];
1801     if (bp) 
1802         buf_HoldLocked(bp);
1803     lock_ReleaseRead(&buf_globalLock);
1804     for (; bp; bp = nbp) {
1805         /* clean buffer synchronously */
1806         if (cm_FidCmp(&bp->fid, &scp->fid) == 0) {
1807             lock_ObtainMutex(&bp->mx);
1808             if (bp->flags & CM_BUF_DIRTY) {
1809                 if (userp && userp != bp->userp) {
1810                     cm_HoldUser(userp);
1811                     if (bp->userp) 
1812                         cm_ReleaseUser(bp->userp);
1813                     bp->userp = userp;
1814                 }   
1815
1816                 switch (code) {
1817                 case CM_ERROR_NOSUCHFILE:
1818                 case CM_ERROR_BADFD:
1819                 case CM_ERROR_NOACCESS:
1820                 case CM_ERROR_QUOTA:
1821                 case CM_ERROR_SPACE:
1822                 case CM_ERROR_TOOBIG:
1823                 case CM_ERROR_READONLY:
1824                 case CM_ERROR_NOSUCHPATH:
1825                     /* 
1826                      * Apply the previous fatal error to this buffer.
1827                      * Do not waste the time attempting to store to
1828                      * the file server when we know it will fail.
1829                      */
1830                     bp->flags &= ~CM_BUF_DIRTY;
1831                     bp->flags |= CM_BUF_ERROR;
1832                     bp->dirty_offset = 0;
1833                     bp->dirty_length = 0;
1834                     bp->error = code;
1835                     bp->dataVersion = CM_BUF_VERSION_BAD;
1836                     bp->dirtyCounter++;
1837                     break;
1838                 case CM_ERROR_TIMEDOUT:
1839                 case CM_ERROR_ALLDOWN:
1840                 case CM_ERROR_ALLBUSY:
1841                 case CM_ERROR_ALLOFFLINE:
1842                 case CM_ERROR_CLOCKSKEW:
1843                     /* do not mark the buffer in error state but do
1844                      * not attempt to complete the rest either.
1845                      */
1846                     break;
1847                 default:
1848                     code = buf_CleanAsyncLocked(scp, bp, reqp, 0, &wasDirty);
1849                     if (bp->flags & CM_BUF_ERROR) {
1850                         code = bp->error;
1851                         if (code == 0)
1852                             code = -1;
1853                     }
1854                 }
1855                 buf_CleanWait(scp, bp, TRUE);
1856             }
1857             lock_ReleaseMutex(&bp->mx);
1858         }
1859
1860         lock_ObtainRead(&buf_globalLock);
1861         nbp = bp->fileHashp;
1862         if (nbp) 
1863             buf_HoldLocked(nbp);
1864         buf_ReleaseLocked(bp, FALSE);
1865         lock_ReleaseRead(&buf_globalLock);
1866     }   /* for loop over a bunch of buffers */
1867
1868 #ifdef TESTING
1869     buf_ValidateBufQueues();
1870 #endif /* TESTING */
1871
1872     /* done */
1873     return code;
1874 }
1875
1876 #ifdef TESTING
1877 void
1878 buf_ValidateBufQueues(void)
1879 {
1880     cm_buf_t * bp, *bpb, *bpf, *bpa;
1881     afs_uint32 countf=0, countb=0, counta=0;
1882
1883     lock_ObtainRead(&buf_globalLock);
1884     for (bp = cm_data.buf_freeListEndp; bp; bp=(cm_buf_t *) osi_QPrev(&bp->q)) {
1885         if (bp->magic != CM_BUF_MAGIC)
1886             osi_panic("buf magic error",__FILE__,__LINE__);
1887         countb++;
1888         bpb = bp;
1889     }
1890
1891     for (bp = cm_data.buf_freeListp; bp; bp=(cm_buf_t *) osi_QNext(&bp->q)) {
1892         if (bp->magic != CM_BUF_MAGIC)
1893             osi_panic("buf magic error",__FILE__,__LINE__);
1894         countf++;
1895         bpf = bp;
1896     }
1897
1898     for (bp = cm_data.buf_allp; bp; bp=bp->allp) {
1899         if (bp->magic != CM_BUF_MAGIC)
1900             osi_panic("buf magic error",__FILE__,__LINE__);
1901         counta++;
1902         bpa = bp;
1903     }
1904     lock_ReleaseRead(&buf_globalLock);
1905
1906     if (countb != countf)
1907         osi_panic("buf magic error",__FILE__,__LINE__);
1908
1909     if (counta != cm_data.buf_nbuffers)
1910         osi_panic("buf magic error",__FILE__,__LINE__);
1911 }
1912 #endif /* TESTING */
1913
1914 /* dump the contents of the buf_scacheHashTablepp. */
1915 int cm_DumpBufHashTable(FILE *outputFile, char *cookie, int lock)
1916 {
1917     int zilch;
1918     cm_buf_t *bp;
1919     char output[1024];
1920     afs_uint32 i;
1921   
1922     if (cm_data.buf_scacheHashTablepp == NULL)
1923         return -1;
1924
1925     if (lock)
1926         lock_ObtainRead(&buf_globalLock);
1927   
1928     StringCbPrintfA(output, sizeof(output), "%s - dumping buf_HashTable - buf_hashSize=%d\r\n", 
1929                     cookie, cm_data.buf_hashSize);
1930     WriteFile(outputFile, output, (DWORD)strlen(output), &zilch, NULL);
1931   
1932     for (i = 0; i < cm_data.buf_hashSize; i++)
1933     {
1934         for (bp = cm_data.buf_scacheHashTablepp[i]; bp; bp=bp->hashp) 
1935         {
1936             StringCbPrintfA(output, sizeof(output), 
1937                             "%s bp=0x%08X, hash=%d, fid (cell=%d, volume=%d, "
1938                             "vnode=%d, unique=%d), offset=%x:%08x, dv=%I64d, "
1939                             "flags=0x%x, qFlags=0x%x cmFlags=0x%x, error=0x%x, refCount=%d\r\n",
1940                              cookie, (void *)bp, i, bp->fid.cell, bp->fid.volume, 
1941                              bp->fid.vnode, bp->fid.unique, bp->offset.HighPart, 
1942                              bp->offset.LowPart, bp->dataVersion, bp->flags, bp->qFlags,
1943                              bp->cmFlags, bp->error, bp->refCount);
1944             WriteFile(outputFile, output, (DWORD)strlen(output), &zilch, NULL);
1945         }
1946     }
1947   
1948     StringCbPrintfA(output, sizeof(output), "%s - Done dumping buf_HashTable.\r\n", cookie);
1949     WriteFile(outputFile, output, (DWORD)strlen(output), &zilch, NULL);
1950
1951     StringCbPrintfA(output, sizeof(output), "%s - dumping buf_freeListEndp\r\n", cookie);
1952     WriteFile(outputFile, output, (DWORD)strlen(output), &zilch, NULL);
1953     for(bp = cm_data.buf_freeListEndp; bp; bp=(cm_buf_t *) osi_QPrev(&bp->q)) {
1954         StringCbPrintfA(output, sizeof(output), 
1955                          "%s bp=0x%08X, fid (cell=%d, volume=%d, "
1956                          "vnode=%d, unique=%d), offset=%x:%08x, dv=%I64d, "
1957                          "flags=0x%x, qFlags=0x%x, cmFlags=0x%x, error=0x%x, refCount=%d\r\n",
1958                          cookie, (void *)bp, bp->fid.cell, bp->fid.volume, 
1959                          bp->fid.vnode, bp->fid.unique, bp->offset.HighPart, 
1960                          bp->offset.LowPart, bp->dataVersion, bp->flags, bp->qFlags,
1961                          bp->cmFlags, bp->error, bp->refCount);
1962         WriteFile(outputFile, output, (DWORD)strlen(output), &zilch, NULL);
1963     }
1964     StringCbPrintfA(output, sizeof(output), "%s - Done dumping buf_FreeListEndp.\r\n", cookie);
1965     WriteFile(outputFile, output, (DWORD)strlen(output), &zilch, NULL);
1966
1967     StringCbPrintfA(output, sizeof(output), "%s - dumping buf_dirtyListp\r\n", cookie);
1968     WriteFile(outputFile, output, (DWORD)strlen(output), &zilch, NULL);
1969     for(bp = cm_data.buf_dirtyListp; bp; bp=bp->dirtyp) {
1970         StringCbPrintfA(output, sizeof(output), 
1971                          "%s bp=0x%08X, fid (cell=%d, volume=%d, "
1972                          "vnode=%d, unique=%d), offset=%x:%08x, dv=%I64d, "
1973                          "flags=0x%x, qFlags=0x%x, cmFlags=0x%x, error=0x%x, refCount=%d\r\n",
1974                          cookie, (void *)bp, bp->fid.cell, bp->fid.volume, 
1975                          bp->fid.vnode, bp->fid.unique, bp->offset.HighPart, 
1976                          bp->offset.LowPart, bp->dataVersion, bp->flags, bp->qFlags,
1977                          bp->cmFlags, bp->error, bp->refCount);
1978         WriteFile(outputFile, output, (DWORD)strlen(output), &zilch, NULL);
1979     }
1980     StringCbPrintfA(output, sizeof(output), "%s - Done dumping buf_dirtyListp.\r\n", cookie);
1981     WriteFile(outputFile, output, (DWORD)strlen(output), &zilch, NULL);
1982
1983     if (lock)
1984         lock_ReleaseRead(&buf_globalLock);
1985     return 0;
1986 }
1987
1988 void buf_ForceTrace(BOOL flush)
1989 {
1990     HANDLE handle;
1991     int len;
1992     char buf[256];
1993
1994     if (!buf_logp) 
1995         return;
1996
1997     len = GetTempPath(sizeof(buf)-10, buf);
1998     StringCbCopyA(&buf[len], sizeof(buf)-len, "/afs-buffer.log");
1999     handle = CreateFile(buf, GENERIC_WRITE, FILE_SHARE_READ,
2000                             NULL, CREATE_ALWAYS, FILE_ATTRIBUTE_NORMAL, NULL);
2001     if (handle == INVALID_HANDLE_VALUE) {
2002         osi_panic("Cannot create log file", __FILE__, __LINE__);
2003     }
2004     osi_LogPrint(buf_logp, handle);
2005     if (flush)
2006         FlushFileBuffers(handle);
2007     CloseHandle(handle);
2008 }
2009
2010 long buf_DirtyBuffersExist(cm_fid_t *fidp)
2011 {
2012     cm_buf_t *bp;
2013     afs_uint32 bcount = 0;
2014     afs_uint32 i;
2015
2016     i = BUF_FILEHASH(fidp);
2017
2018     for (bp = cm_data.buf_fileHashTablepp[i]; bp; bp=bp->fileHashp, bcount++) {
2019         if (!cm_FidCmp(fidp, &bp->fid) && (bp->flags & CM_BUF_DIRTY))
2020             return 1;
2021     }
2022     return 0;
2023 }
2024
2025 #if 0
2026 long buf_CleanDirtyBuffers(cm_scache_t *scp)
2027 {
2028     cm_buf_t *bp;
2029     afs_uint32 bcount = 0;
2030     cm_fid_t * fidp = &scp->fid;
2031
2032     for (bp = cm_data.buf_allp; bp; bp=bp->allp, bcount++) {
2033         if (!cm_FidCmp(fidp, &bp->fid) && (bp->flags & CM_BUF_DIRTY)) {
2034             buf_Hold(bp);
2035             lock_ObtainMutex(&bp->mx);
2036             bp->cmFlags &= ~CM_BUF_CMSTORING;
2037             bp->flags &= ~CM_BUF_DIRTY;
2038             bp->dirty_offset = 0;
2039             bp->dirty_length = 0;
2040             bp->flags |= CM_BUF_ERROR;
2041             bp->error = VNOVNODE;
2042             bp->dataVersion = CM_BUF_VERSION_BAD; /* bad */
2043             bp->dirtyCounter++;
2044             if (bp->flags & CM_BUF_WAITING) {
2045                 osi_Log2(buf_logp, "BUF CleanDirtyBuffers Waking [scp 0x%x] bp 0x%x", scp, bp);
2046                 osi_Wakeup((long) &bp);
2047             }
2048             lock_ReleaseMutex(&bp->mx);
2049             buf_Release(bp);
2050         }
2051     }
2052     return 0;
2053 }
2054 #endif