src/DiskIO/DiskThreads/aiops_win32.cc

   1 /*
   2  * $Id: aiops_win32.cc,v 1.2 2006/09/09 15:29:59 serassio Exp $
   3  *
   4  * DEBUG: section 43    Windows AIOPS
   5  * AUTHOR: Stewart Forster <slf@connect.com.au>
   6  * AUTHOR: Robert Collins <robertc@squid-cache.org>
   7  * AUTHOR: Guido Serassio <serassio@squid-cache.org>
   8  *
   9  * SQUID Web Proxy Cache          http://www.squid-cache.org/
  10  * ----------------------------------------------------------
  11  *
  12  *  Squid is the result of efforts by numerous individuals from
  13  *  the Internet community; see the CONTRIBUTORS file for full
  14  *  details.   Many organizations have provided support for Squid's
  15  *  development; see the SPONSORS file for full details.  Squid is
  16  *  Copyrighted (C) 2001 by the Regents of the University of
  17  *  California; see the COPYRIGHT file for full details.  Squid
  18  *  incorporates software developed and/or copyrighted by other
  19  *  sources; see the CREDITS file for full details.
  20  *
  21  *  This program is free software; you can redistribute it and/or modify
  22  *  it under the terms of the GNU General Public License as published by
  23  *  the Free Software Foundation; either version 2 of the License, or
  24  *  (at your option) any later version.
  25  *
  26  *  This program is distributed in the hope that it will be useful,
  27  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  28  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  29  *  GNU General Public License for more details.
  30  *
  31  *  You should have received a copy of the GNU General Public License
  32  *  along with this program; if not, write to the Free Software
  33  *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111, USA.
  34  *
  35  */
  36
  37 #include "squid.h"
  38 #include "squid_windows.h"
  39 #include "DiskThreads.h"
  40
  41 #include        <stdio.h>
  42 #include        <sys/types.h>
  43 #include        <sys/stat.h>
  44 #include        <fcntl.h>
  45 #include        <errno.h>
  46 #include        <dirent.h>
  47 #include        <signal.h>
  48 #include "CommIO.h"
  49 #include "SquidTime.h"
  50 #include "Store.h"
  51
  52 #define RIDICULOUS_LENGTH       4096
  53
  54 enum _squidaio_thread_status {
  55     _THREAD_STARTING = 0,
  56     _THREAD_WAITING,
  57     _THREAD_BUSY,
  58     _THREAD_FAILED,
  59     _THREAD_DONE
  60 };
  61 typedef enum _squidaio_thread_status squidaio_thread_status;
  62
  63 typedef struct squidaio_request_t
  64 {
  65
  66     struct squidaio_request_t *next;
  67     squidaio_request_type request_type;
  68     int cancelled;
  69     char *path;
  70     int oflag;
  71     mode_t mode;
  72     int fd;
  73     char *bufferp;
  74     char *tmpbufp;
  75     int buflen;
  76     off_t offset;
  77     int whence;
  78     int ret;
  79     int err;
  80
  81     struct stat *tmpstatp;
  82
  83     struct stat *statp;
  84     squidaio_result_t *resultp;
  85 }
  86
  87 squidaio_request_t;
  88
  89 typedef struct squidaio_request_queue_t
  90 {
  91     HANDLE mutex;
  92     HANDLE cond; /* See Event objects */
  93     squidaio_request_t *volatile head;
  94     squidaio_request_t *volatile *volatile tailp;
  95     unsigned long requests;
  96     unsigned long blocked;      /* main failed to lock the queue */
  97 }
  98
  99 squidaio_request_queue_t;
 100
 101 typedef struct squidaio_thread_t squidaio_thread_t;
 102
 103 struct squidaio_thread_t
 104 {
 105     squidaio_thread_t *next;
 106     HANDLE thread;
 107     DWORD dwThreadId; /* thread ID */
 108     squidaio_thread_status status;
 109
 110     struct squidaio_request_t *current_req;
 111     unsigned long requests;
 112     int volatile exit;
 113 };
 114
 115 static void squidaio_queue_request(squidaio_request_t *);
 116 static void squidaio_cleanup_request(squidaio_request_t *);
 117 static DWORD WINAPI squidaio_thread_loop( LPVOID lpParam );
 118 static void squidaio_do_open(squidaio_request_t *);
 119 static void squidaio_do_read(squidaio_request_t *);
 120 static void squidaio_do_write(squidaio_request_t *);
 121 static void squidaio_do_close(squidaio_request_t *);
 122 static void squidaio_do_stat(squidaio_request_t *);
 123 #if USE_TRUNCATE
 124 static void squidaio_do_truncate(squidaio_request_t *);
 125 #else
 126 static void squidaio_do_unlink(squidaio_request_t *);
 127 #endif
 128 #if AIO_OPENDIR
 129 static void *squidaio_do_opendir(squidaio_request_t *);
 130 #endif
 131 static void squidaio_debug(squidaio_request_t *);
 132 static void squidaio_poll_queues(void);
 133
 134 static squidaio_thread_t *threads = NULL;
 135 static int squidaio_initialised = 0;
 136
 137
 138 #define AIO_LARGE_BUFS  16384
 139 #define AIO_MEDIUM_BUFS AIO_LARGE_BUFS >> 1
 140 #define AIO_SMALL_BUFS  AIO_LARGE_BUFS >> 2
 141 #define AIO_TINY_BUFS   AIO_LARGE_BUFS >> 3
 142 #define AIO_MICRO_BUFS  128
 143
 144 static MemAllocator *squidaio_large_bufs = NULL;        /* 16K */
 145 static MemAllocator *squidaio_medium_bufs = NULL;       /* 8K */
 146 static MemAllocator *squidaio_small_bufs = NULL;        /* 4K */
 147 static MemAllocator *squidaio_tiny_bufs = NULL; /* 2K */
 148 static MemAllocator *squidaio_micro_bufs = NULL;        /* 128K */
 149
 150 static int request_queue_len = 0;
 151 static MemAllocator *squidaio_request_pool = NULL;
 152 static MemAllocator *squidaio_thread_pool = NULL;
 153 static squidaio_request_queue_t request_queue;
 154
 155 static struct
 156 {
 157     squidaio_request_t *head, **tailp;
 158 }
 159
 160 request_queue2 = {
 161
 162                      NULL, &request_queue2.head
 163                  };
 164 static squidaio_request_queue_t done_queue;
 165
 166 static struct
 167 {
 168     squidaio_request_t *head, **tailp;
 169 }
 170
 171 done_requests = {
 172
 173                     NULL, &done_requests.head
 174                 };
 175
 176 static HANDLE main_thread;
 177
 178 static MemAllocator *
 179 squidaio_get_pool(int size)
 180 {
 181     if (size <= AIO_LARGE_BUFS) {
 182         if (size <= AIO_MICRO_BUFS)
 183             return squidaio_micro_bufs;
 184         else if (size <= AIO_TINY_BUFS)
 185             return squidaio_tiny_bufs;
 186         else if (size <= AIO_SMALL_BUFS)
 187             return squidaio_small_bufs;
 188         else if (size <= AIO_MEDIUM_BUFS)
 189             return squidaio_medium_bufs;
 190         else
 191             return squidaio_large_bufs;
 192     }
 193
 194     return NULL;
 195 }
 196
 197 void *
 198 squidaio_xmalloc(int size)
 199 {
 200     void *p;
 201     MemAllocator *pool;
 202
 203     if ((pool = squidaio_get_pool(size)) != NULL) {
 204         p = pool->alloc();
 205     } else
 206         p = xmalloc(size);
 207
 208     return p;
 209 }
 210
 211 static char *
 212 squidaio_xstrdup(const char *str)
 213 {
 214     char *p;
 215     int len = strlen(str) + 1;
 216
 217     p = (char *)squidaio_xmalloc(len);
 218     strncpy(p, str, len);
 219
 220     return p;
 221 }
 222
 223 void
 224 squidaio_xfree(void *p, int size)
 225 {
 226     MemAllocator *pool;
 227
 228     if ((pool = squidaio_get_pool(size)) != NULL) {
 229         pool->free(p);
 230     } else
 231         xfree(p);
 232 }
 233
 234 static void
 235 squidaio_xstrfree(char *str)
 236 {
 237     MemAllocator *pool;
 238     int len = strlen(str) + 1;
 239
 240     if ((pool = squidaio_get_pool(len)) != NULL) {
 241         pool->free(str);
 242     } else
 243         xfree(str);
 244 }
 245
 246 void
 247 squidaio_init(void)
 248 {
 249     int i;
 250     squidaio_thread_t *threadp;
 251
 252     if (squidaio_initialised)
 253         return;
 254
 255     if (!DuplicateHandle(GetCurrentProcess(), /* pseudo handle, don't close */
 256                          GetCurrentThread(),  /* pseudo handle to copy */
 257                          GetCurrentProcess(), /* pseudo handle, don't close */
 258                          &main_thread,
 259                          0,                   /* required access */
 260                          FALSE,               /* child process's don't inherit the handle */
 261                          DUPLICATE_SAME_ACCESS)) {
 262         /* spit errors */
 263         fatal("Couldn't get current thread handle");
 264     }
 265
 266     /* Initialize request queue */
 267     if ((request_queue.mutex = CreateMutex(NULL,    /* no inheritance */
 268                                            FALSE,   /* start unowned (as per mutex_init) */
 269                                            NULL)    /* no name */
 270         ) == NULL) {
 271         fatal("Failed to create mutex");
 272     }
 273
 274     if ((request_queue.cond = CreateEvent(NULL,     /* no inheritance */
 275                                           FALSE,    /* auto signal reset - which I think is pthreads like ? */
 276                                           FALSE,    /* start non signaled */
 277                                           NULL)     /* no name */
 278         ) == NULL) {
 279         fatal("Failed to create condition variable");
 280     }
 281
 282     request_queue.head = NULL;
 283
 284     request_queue.tailp = &request_queue.head;
 285
 286     request_queue.requests = 0;
 287
 288     request_queue.blocked = 0;
 289
 290     /* Initialize done queue */
 291
 292     if ((done_queue.mutex = CreateMutex(NULL,  /* no inheritance */
 293                                         FALSE, /* start unowned (as per mutex_init) */
 294                                         NULL)  /* no name */
 295         ) == NULL) {
 296         fatal("Failed to create mutex");
 297     }
 298
 299     if ((done_queue.cond = CreateEvent(NULL,  /* no inheritance */
 300                                        TRUE,  /* manually signaled - which I think is pthreads like ? */
 301                                        FALSE, /* start non signaled */
 302                                        NULL)  /* no name */
 303         ) == NULL) {
 304         fatal("Failed to create condition variable");
 305     }
 306
 307     done_queue.head = NULL;
 308
 309     done_queue.tailp = &done_queue.head;
 310
 311     done_queue.requests = 0;
 312
 313     done_queue.blocked = 0;
 314
 315     CommIO::NotifyIOCompleted();
 316
 317     /* Create threads and get them to sit in their wait loop */
 318     squidaio_thread_pool = memPoolCreate("aio_thread", sizeof(squidaio_thread_t));
 319
 320     assert(NUMTHREADS);
 321
 322     for (i = 0; i < NUMTHREADS; i++) {
 323         threadp = (squidaio_thread_t *)squidaio_thread_pool->alloc();
 324         threadp->status = _THREAD_STARTING;
 325         threadp->current_req = NULL;
 326         threadp->requests = 0;
 327         threadp->next = threads;
 328         threads = threadp;
 329
 330         if ((threadp->thread = CreateThread(NULL,                   /* no security attributes */
 331                                             0,                      /* use default stack size */
 332                                             squidaio_thread_loop,   /* thread function */
 333                                             threadp,                /* argument to thread function */
 334                                             0,                      /* use default creation flags */
 335                                             &(threadp->dwThreadId)) /* returns the thread identifier */
 336             ) == NULL) {
 337             fprintf(stderr, "Thread creation failed\n");
 338             threadp->status = _THREAD_FAILED;
 339             continue;
 340         }
 341
 342         /* Set the new thread priority above parent process */
 343         SetThreadPriority(threadp->thread,THREAD_PRIORITY_ABOVE_NORMAL);
 344     }
 345
 346     /* Create request pool */
 347     squidaio_request_pool = memPoolCreate("aio_request", sizeof(squidaio_request_t));
 348
 349     squidaio_large_bufs = memPoolCreate("squidaio_large_bufs", AIO_LARGE_BUFS);
 350
 351     squidaio_medium_bufs = memPoolCreate("squidaio_medium_bufs", AIO_MEDIUM_BUFS);
 352
 353     squidaio_small_bufs = memPoolCreate("squidaio_small_bufs", AIO_SMALL_BUFS);
 354
 355     squidaio_tiny_bufs = memPoolCreate("squidaio_tiny_bufs", AIO_TINY_BUFS);
 356
 357     squidaio_micro_bufs = memPoolCreate("squidaio_micro_bufs", AIO_MICRO_BUFS);
 358
 359     squidaio_initialised = 1;
 360 }
 361
 362 void
 363 squidaio_shutdown(void)
 364 {
 365     squidaio_thread_t *threadp;
 366     int i;
 367     HANDLE * hthreads;
 368
 369     if (!squidaio_initialised)
 370         return;
 371
 372     /* This is the same as in squidaio_sync */
 373     do {
 374         squidaio_poll_queues();
 375     } while (request_queue_len > 0);
 376
 377     hthreads = (HANDLE *) xcalloc (NUMTHREADS, sizeof (HANDLE));
 378
 379     threadp = threads;
 380
 381     for (i = 0; i < NUMTHREADS; i++) {
 382         threadp->exit = 1;
 383         hthreads[i] = threadp->thread;
 384         threadp = threadp->next;
 385     }
 386
 387     ReleaseMutex(request_queue.mutex);
 388     ResetEvent(request_queue.cond);
 389     ReleaseMutex(done_queue.mutex);
 390     ResetEvent(done_queue.cond);
 391     Sleep(0);
 392
 393     WaitForMultipleObjects(NUMTHREADS, hthreads, TRUE, 2000);
 394
 395     for (i = 0; i < NUMTHREADS; i++) {
 396         CloseHandle(hthreads[i]);
 397     }
 398
 399     CloseHandle(main_thread);
 400     CommIO::NotifyIOClose();
 401
 402     squidaio_initialised = 0;
 403     xfree(hthreads);
 404 }
 405
 406 static DWORD WINAPI
 407 squidaio_thread_loop(LPVOID lpParam)
 408 {
 409     squidaio_thread_t *threadp = (squidaio_thread_t *)lpParam;
 410     squidaio_request_t *request;
 411     HANDLE cond; /* local copy of the event queue because win32 event handles
 412                               * don't atomically release the mutex as cond variables do. */
 413
 414     /* lock the thread info */
 415
 416     if (WAIT_FAILED == WaitForSingleObject(request_queue.mutex, INFINITE)) {
 417         fatal("Can't get ownership of mutex\n");
 418     }
 419
 420     /* duplicate the handle */
 421     if (!DuplicateHandle(GetCurrentProcess(),    /* pseudo handle, don't close */
 422                          request_queue.cond,     /* handle to copy */
 423                          GetCurrentProcess(),    /* pseudo handle, don't close */
 424                          &cond,
 425                          0,                      /* required access */
 426                          FALSE,                  /* child process's don't inherit the handle */
 427                          DUPLICATE_SAME_ACCESS))
 428         fatal("Can't duplicate mutex handle\n");
 429
 430     if (!ReleaseMutex(request_queue.mutex)) {
 431         CloseHandle(cond);
 432         fatal("Can't release mutex\n");
 433     }
 434
 435     Sleep(0);
 436
 437     while (1) {
 438         DWORD rv;
 439         threadp->current_req = request = NULL;
 440         request = NULL;
 441         /* Get a request to process */
 442         threadp->status = _THREAD_WAITING;
 443
 444         if (threadp->exit) {
 445             CloseHandle(request_queue.mutex);
 446             CloseHandle(cond);
 447             return 0;
 448         }
 449
 450         rv = WaitForSingleObject(request_queue.mutex, INFINITE);
 451
 452         if (rv == WAIT_FAILED) {
 453             CloseHandle(cond);
 454             return 1;
 455         }
 456
 457         while (!request_queue.head) {
 458             if (!ReleaseMutex(request_queue.mutex)) {
 459                 CloseHandle(cond);
 460                 threadp->status = _THREAD_FAILED;
 461                 return 1;
 462             }
 463
 464             Sleep(0);
 465             rv = WaitForSingleObject(cond, INFINITE);
 466
 467             if (rv == WAIT_FAILED) {
 468                 CloseHandle(cond);
 469                 return 1;
 470             }
 471
 472             rv = WaitForSingleObject(request_queue.mutex, INFINITE);
 473
 474             if (rv == WAIT_FAILED) {
 475                 CloseHandle(cond);
 476                 return 1;
 477             }
 478         }
 479
 480         request = request_queue.head;
 481
 482         if (request)
 483             request_queue.head = request->next;
 484
 485         if (!request_queue.head)
 486             request_queue.tailp = &request_queue.head;
 487
 488         if (!ReleaseMutex(request_queue.mutex)) {
 489             CloseHandle(cond);
 490             return 1;
 491         }
 492
 493         Sleep(0);
 494
 495         /* process the request */
 496         threadp->status = _THREAD_BUSY;
 497
 498         request->next = NULL;
 499
 500         threadp->current_req = request;
 501
 502         errno = 0;
 503
 504         if (!request->cancelled) {
 505             switch (request->request_type) {
 506
 507             case _AIO_OP_OPEN:
 508                 squidaio_do_open(request);
 509                 break;
 510
 511             case _AIO_OP_READ:
 512                 squidaio_do_read(request);
 513                 break;
 514
 515             case _AIO_OP_WRITE:
 516                 squidaio_do_write(request);
 517                 break;
 518
 519             case _AIO_OP_CLOSE:
 520                 squidaio_do_close(request);
 521                 break;
 522
 523 #if USE_TRUNCATE
 524
 525             case _AIO_OP_TRUNCATE:
 526                 squidaio_do_truncate(request);
 527                 break;
 528 #else
 529
 530             case _AIO_OP_UNLINK:
 531                 squidaio_do_unlink(request);
 532                 break;
 533
 534 #endif
 535 #if AIO_OPENDIR                 /* Opendir not implemented yet */
 536
 537             case _AIO_OP_OPENDIR:
 538                 squidaio_do_opendir(request);
 539                 break;
 540 #endif
 541
 542             case _AIO_OP_STAT:
 543                 squidaio_do_stat(request);
 544                 break;
 545
 546             default:
 547                 request->ret = -1;
 548                 request->err = EINVAL;
 549                 break;
 550             }
 551         } else {                /* cancelled */
 552             request->ret = -1;
 553             request->err = EINTR;
 554         }
 555
 556         threadp->status = _THREAD_DONE;
 557         /* put the request in the done queue */
 558         rv = WaitForSingleObject(done_queue.mutex, INFINITE);
 559
 560         if (rv == WAIT_FAILED) {
 561             CloseHandle(cond);
 562             return 1;
 563         }
 564
 565         *done_queue.tailp = request;
 566         done_queue.tailp = &request->next;
 567
 568         if (!ReleaseMutex(done_queue.mutex)) {
 569             CloseHandle(cond);
 570             return 1;
 571         }
 572
 573         CommIO::NotifyIOCompleted();
 574         Sleep(0);
 575         threadp->requests++;
 576     }                           /* while forever */
 577
 578     CloseHandle(cond);
 579
 580     return 0;
 581 }                               /* squidaio_thread_loop */
 582
 583 static void
 584 squidaio_queue_request(squidaio_request_t * request)
 585 {
 586     static int high_start = 0;
 587     debug(43, 9) ("squidaio_queue_request: %p type=%d result=%p\n",
 588                   request, request->request_type, request->resultp);
 589     /* Mark it as not executed (failing result, no error) */
 590     request->ret = -1;
 591     request->err = 0;
 592     /* Internal housekeeping */
 593     request_queue_len += 1;
 594     request->resultp->_data = request;
 595     /* Play some tricks with the request_queue2 queue */
 596     request->next = NULL;
 597
 598     if (WaitForSingleObject(request_queue.mutex, 0) == WAIT_OBJECT_0) {
 599         if (request_queue2.head) {
 600             /* Grab blocked requests */
 601             *request_queue.tailp = request_queue2.head;
 602             request_queue.tailp = request_queue2.tailp;
 603         }
 604
 605         /* Enqueue request */
 606         *request_queue.tailp = request;
 607
 608         request_queue.tailp = &request->next;
 609
 610         if (!SetEvent(request_queue.cond))
 611             fatal("Couldn't push queue");
 612
 613         if (!ReleaseMutex(request_queue.mutex)) {
 614             /* unexpected error */
 615             fatal("Couldn't push queue");
 616         }
 617
 618         Sleep(0);
 619
 620         if (request_queue2.head) {
 621             /* Clear queue of blocked requests */
 622             request_queue2.head = NULL;
 623             request_queue2.tailp = &request_queue2.head;
 624         }
 625     } else {
 626         /* Oops, the request queue is blocked, use request_queue2 */
 627         *request_queue2.tailp = request;
 628         request_queue2.tailp = &request->next;
 629     }
 630
 631     if (request_queue2.head) {
 632         static int filter = 0;
 633         static int filter_limit = 8;
 634
 635         if (++filter >= filter_limit) {
 636             filter_limit += filter;
 637             filter = 0;
 638             debug(43, 1) ("squidaio_queue_request: WARNING - Queue congestion\n");
 639         }
 640     }
 641
 642     /* Warn if out of threads */
 643     if (request_queue_len > MAGIC1) {
 644         static int last_warn = 0;
 645         static int queue_high, queue_low;
 646
 647         if (high_start == 0) {
 648             high_start = (int)squid_curtime;
 649             queue_high = request_queue_len;
 650             queue_low = request_queue_len;
 651         }
 652
 653         if (request_queue_len > queue_high)
 654             queue_high = request_queue_len;
 655
 656         if (request_queue_len < queue_low)
 657             queue_low = request_queue_len;
 658
 659         if (squid_curtime >= (last_warn + 15) &&
 660                 squid_curtime >= (high_start + 5)) {
 661             debug(43, 1) ("squidaio_queue_request: WARNING - Disk I/O overloading\n");
 662
 663             if (squid_curtime >= (high_start + 15))
 664                 debug(43, 1) ("squidaio_queue_request: Queue Length: current=%d, high=%d, low=%d, duration=%ld\n",
 665                               request_queue_len, queue_high, queue_low, (long int) (squid_curtime - high_start));
 666
 667             last_warn = (int)squid_curtime;
 668         }
 669     } else {
 670         high_start = 0;
 671     }
 672
 673     /* Warn if seriously overloaded */
 674     if (request_queue_len > RIDICULOUS_LENGTH) {
 675         debug(43, 0) ("squidaio_queue_request: Async request queue growing uncontrollably!\n");
 676         debug(43, 0) ("squidaio_queue_request: Syncing pending I/O operations.. (blocking)\n");
 677         squidaio_sync();
 678         debug(43, 0) ("squidaio_queue_request: Synced\n");
 679     }
 680 }                               /* squidaio_queue_request */
 681
 682 static void
 683 squidaio_cleanup_request(squidaio_request_t * requestp)
 684 {
 685     squidaio_result_t *resultp = requestp->resultp;
 686     int cancelled = requestp->cancelled;
 687
 688     /* Free allocated structures and copy data back to user space if the */
 689     /* request hasn't been cancelled */
 690
 691     switch (requestp->request_type) {
 692
 693     case _AIO_OP_STAT:
 694
 695         if (!cancelled && requestp->ret == 0)
 696
 697             xmemcpy(requestp->statp, requestp->tmpstatp, sizeof(struct stat));
 698
 699         squidaio_xfree(requestp->tmpstatp, sizeof(struct stat));
 700
 701         squidaio_xstrfree(requestp->path);
 702
 703         break;
 704
 705     case _AIO_OP_OPEN:
 706         if (cancelled && requestp->ret >= 0)
 707             /* The open() was cancelled but completed */
 708             close(requestp->ret);
 709
 710         squidaio_xstrfree(requestp->path);
 711
 712         break;
 713
 714     case _AIO_OP_CLOSE:
 715         if (cancelled && requestp->ret < 0)
 716             /* The close() was cancelled and never got executed */
 717             close(requestp->fd);
 718
 719         break;
 720
 721     case _AIO_OP_UNLINK:
 722
 723     case _AIO_OP_TRUNCATE:
 724
 725     case _AIO_OP_OPENDIR:
 726         squidaio_xstrfree(requestp->path);
 727
 728         break;
 729
 730     case _AIO_OP_READ:
 731         break;
 732
 733     case _AIO_OP_WRITE:
 734         break;
 735
 736     default:
 737         break;
 738     }
 739
 740     if (resultp != NULL && !cancelled) {
 741         resultp->aio_return = requestp->ret;
 742         resultp->aio_errno = requestp->err;
 743     }
 744
 745     squidaio_request_pool->free(requestp);
 746 }                               /* squidaio_cleanup_request */
 747
 748
 749 int
 750 squidaio_cancel(squidaio_result_t * resultp)
 751 {
 752     squidaio_request_t *request = (squidaio_request_t *)resultp->_data;
 753
 754     if (request && request->resultp == resultp) {
 755         debug(43, 9) ("squidaio_cancel: %p type=%d result=%p\n",
 756                       request, request->request_type, request->resultp);
 757         request->cancelled = 1;
 758         request->resultp = NULL;
 759         resultp->_data = NULL;
 760         resultp->result_type = _AIO_OP_NONE;
 761         return 0;
 762     }
 763
 764     return 1;
 765 }                               /* squidaio_cancel */
 766
 767
 768 int
 769 squidaio_open(const char *path, int oflag, mode_t mode, squidaio_result_t * resultp)
 770 {
 771     squidaio_init();
 772     squidaio_request_t *requestp;
 773
 774     requestp = (squidaio_request_t *)squidaio_request_pool->alloc();
 775
 776     requestp->path = (char *) squidaio_xstrdup(path);
 777
 778     requestp->oflag = oflag;
 779
 780     requestp->mode = mode;
 781
 782     requestp->resultp = resultp;
 783
 784     requestp->request_type = _AIO_OP_OPEN;
 785
 786     requestp->cancelled = 0;
 787
 788     resultp->result_type = _AIO_OP_OPEN;
 789
 790     squidaio_queue_request(requestp);
 791
 792     return 0;
 793 }
 794
 795
 796 static void
 797 squidaio_do_open(squidaio_request_t * requestp)
 798 {
 799     requestp->ret = open(requestp->path, requestp->oflag, requestp->mode);
 800     requestp->err = errno;
 801 }
 802
 803
 804 int
 805 squidaio_read(int fd, char *bufp, int bufs, off_t offset, int whence, squidaio_result_t * resultp)
 806 {
 807     squidaio_request_t *requestp;
 808
 809     requestp = (squidaio_request_t *)squidaio_request_pool->alloc();
 810
 811     requestp->fd = fd;
 812
 813     requestp->bufferp = bufp;
 814
 815     requestp->buflen = bufs;
 816
 817     requestp->offset = offset;
 818
 819     requestp->whence = whence;
 820
 821     requestp->resultp = resultp;
 822
 823     requestp->request_type = _AIO_OP_READ;
 824
 825     requestp->cancelled = 0;
 826
 827     resultp->result_type = _AIO_OP_READ;
 828
 829     squidaio_queue_request(requestp);
 830
 831     return 0;
 832 }
 833
 834
 835 static void
 836 squidaio_do_read(squidaio_request_t * requestp)
 837 {
 838     lseek(requestp->fd, requestp->offset, requestp->whence);
 839
 840     if (!ReadFile((HANDLE)_get_osfhandle(requestp->fd), requestp->bufferp,
 841                   requestp->buflen, (LPDWORD)&requestp->ret, NULL)) {
 842         WIN32_maperror(GetLastError());
 843         requestp->ret = -1;
 844     }
 845
 846     requestp->err = errno;
 847 }
 848
 849
 850 int
 851 squidaio_write(int fd, char *bufp, int bufs, off_t offset, int whence, squidaio_result_t * resultp)
 852 {
 853     squidaio_request_t *requestp;
 854
 855     requestp = (squidaio_request_t *)squidaio_request_pool->alloc();
 856
 857     requestp->fd = fd;
 858
 859     requestp->bufferp = bufp;
 860
 861     requestp->buflen = bufs;
 862
 863     requestp->offset = offset;
 864
 865     requestp->whence = whence;
 866
 867     requestp->resultp = resultp;
 868
 869     requestp->request_type = _AIO_OP_WRITE;
 870
 871     requestp->cancelled = 0;
 872
 873     resultp->result_type = _AIO_OP_WRITE;
 874
 875     squidaio_queue_request(requestp);
 876
 877     return 0;
 878 }
 879
 880
 881 static void
 882 squidaio_do_write(squidaio_request_t * requestp)
 883 {
 884     if (!WriteFile((HANDLE)_get_osfhandle(requestp->fd), requestp->bufferp,
 885                    requestp->buflen, (LPDWORD)&requestp->ret, NULL)) {
 886         WIN32_maperror(GetLastError());
 887         requestp->ret = -1;
 888     }
 889
 890     requestp->err = errno;
 891 }
 892
 893
 894 int
 895 squidaio_close(int fd, squidaio_result_t * resultp)
 896 {
 897     squidaio_request_t *requestp;
 898
 899     requestp = (squidaio_request_t *)squidaio_request_pool->alloc();
 900
 901     requestp->fd = fd;
 902
 903     requestp->resultp = resultp;
 904
 905     requestp->request_type = _AIO_OP_CLOSE;
 906
 907     requestp->cancelled = 0;
 908
 909     resultp->result_type = _AIO_OP_CLOSE;
 910
 911     squidaio_queue_request(requestp);
 912
 913     return 0;
 914 }
 915
 916
 917 static void
 918 squidaio_do_close(squidaio_request_t * requestp)
 919 {
 920     if((requestp->ret = close(requestp->fd)) < 0) {
 921         debug(43, 0) ("squidaio_do_close: FD %d, errno %d\n", requestp->fd, errno);
 922         close(requestp->fd);
 923     }
 924
 925     requestp->err = errno;
 926 }
 927
 928
 929 int
 930
 931 squidaio_stat(const char *path, struct stat *sb, squidaio_result_t * resultp)
 932 {
 933     squidaio_init();
 934     squidaio_request_t *requestp;
 935
 936     requestp = (squidaio_request_t *)squidaio_request_pool->alloc();
 937
 938     requestp->path = (char *) squidaio_xstrdup(path);
 939
 940     requestp->statp = sb;
 941
 942     requestp->tmpstatp = (struct stat *) squidaio_xmalloc(sizeof(struct stat));
 943
 944     requestp->resultp = resultp;
 945
 946     requestp->request_type = _AIO_OP_STAT;
 947
 948     requestp->cancelled = 0;
 949
 950     resultp->result_type = _AIO_OP_STAT;
 951
 952     squidaio_queue_request(requestp);
 953
 954     return 0;
 955 }
 956
 957
 958 static void
 959 squidaio_do_stat(squidaio_request_t * requestp)
 960 {
 961     requestp->ret = stat(requestp->path, requestp->tmpstatp);
 962     requestp->err = errno;
 963 }
 964
 965
 966 #if USE_TRUNCATE
 967 int
 968 squidaio_truncate(const char *path, off_t length, squidaio_result_t * resultp)
 969 {
 970     squidaio_init();
 971     squidaio_request_t *requestp;
 972
 973     requestp = (squidaio_request_t *)squidaio_request_pool->alloc();
 974
 975     requestp->path = (char *) squidaio_xstrdup(path);
 976
 977     requestp->offset = length;
 978
 979     requestp->resultp = resultp;
 980
 981     requestp->request_type = _AIO_OP_TRUNCATE;
 982
 983     requestp->cancelled = 0;
 984
 985     resultp->result_type = _AIO_OP_TRUNCATE;
 986
 987     squidaio_queue_request(requestp);
 988
 989     return 0;
 990 }
 991
 992
 993 static void
 994 squidaio_do_truncate(squidaio_request_t * requestp)
 995 {
 996     requestp->ret = truncate(requestp->path, requestp->offset);
 997     requestp->err = errno;
 998 }
 999
1000
1001 #else
1002 int
1003 squidaio_unlink(const char *path, squidaio_result_t * resultp)
1004 {
1005     squidaio_init();
1006     squidaio_request_t *requestp;
1007
1008     requestp = (squidaio_request_t *)squidaio_request_pool->alloc();
1009
1010     requestp->path = squidaio_xstrdup(path);
1011
1012     requestp->resultp = resultp;
1013
1014     requestp->request_type = _AIO_OP_UNLINK;
1015
1016     requestp->cancelled = 0;
1017
1018     resultp->result_type = _AIO_OP_UNLINK;
1019
1020     squidaio_queue_request(requestp);
1021
1022     return 0;
1023 }
1024
1025
1026 static void
1027 squidaio_do_unlink(squidaio_request_t * requestp)
1028 {
1029     requestp->ret = unlink(requestp->path);
1030     requestp->err = errno;
1031 }
1032
1033 #endif
1034
1035 #if AIO_OPENDIR
1036 /* XXX squidaio_opendir NOT implemented yet.. */
1037
1038 int
1039 squidaio_opendir(const char *path, squidaio_result_t * resultp)
1040 {
1041     squidaio_request_t *requestp;
1042     int len;
1043
1044     requestp = squidaio_request_pool->alloc();
1045
1046     resultp->result_type = _AIO_OP_OPENDIR;
1047
1048     return -1;
1049 }
1050
1051 static void
1052 squidaio_do_opendir(squidaio_request_t * requestp)
1053 {
1054     /* NOT IMPLEMENTED */
1055 }
1056
1057 #endif
1058
1059 static void
1060 squidaio_poll_queues(void)
1061 {
1062     /* kick "overflow" request queue */
1063
1064     if (request_queue2.head &&
1065             (WaitForSingleObject(request_queue.mutex, 0 )== WAIT_OBJECT_0)) {
1066         *request_queue.tailp = request_queue2.head;
1067         request_queue.tailp = request_queue2.tailp;
1068
1069         if (!SetEvent(request_queue.cond))
1070             fatal("couldn't push queue\n");
1071
1072         if (!ReleaseMutex(request_queue.mutex)) {
1073             /* unexpected error */
1074         }
1075
1076         Sleep(0);
1077         request_queue2.head = NULL;
1078         request_queue2.tailp = &request_queue2.head;
1079     }
1080
1081     /* poll done queue */
1082     if (done_queue.head &&
1083             (WaitForSingleObject(done_queue.mutex, 0)==WAIT_OBJECT_0)) {
1084
1085         struct squidaio_request_t *requests = done_queue.head;
1086         done_queue.head = NULL;
1087         done_queue.tailp = &done_queue.head;
1088
1089         if (!ReleaseMutex(done_queue.mutex)) {
1090             /* unexpected error */
1091         }
1092
1093         Sleep(0);
1094         *done_requests.tailp = requests;
1095         request_queue_len -= 1;
1096
1097         while (requests->next) {
1098             requests = requests->next;
1099             request_queue_len -= 1;
1100         }
1101
1102         done_requests.tailp = &requests->next;
1103     }
1104 }
1105
1106 squidaio_result_t *
1107 squidaio_poll_done(void)
1108 {
1109     squidaio_request_t *request;
1110     squidaio_result_t *resultp;
1111     int cancelled;
1112     int polled = 0;
1113
1114 AIO_REPOLL:
1115     request = done_requests.head;
1116
1117     if (request == NULL && !polled) {
1118         CommIO::ResetNotifications();
1119         squidaio_poll_queues();
1120         polled = 1;
1121         request = done_requests.head;
1122     }
1123
1124     if (!request) {
1125         return NULL;
1126     }
1127
1128     debug(43, 9) ("squidaio_poll_done: %p type=%d result=%p\n",
1129                   request, request->request_type, request->resultp);
1130     done_requests.head = request->next;
1131
1132     if (!done_requests.head)
1133         done_requests.tailp = &done_requests.head;
1134
1135     resultp = request->resultp;
1136
1137     cancelled = request->cancelled;
1138
1139     squidaio_debug(request);
1140
1141     debug(43, 5) ("DONE: %d -> %d\n", request->ret, request->err);
1142
1143     squidaio_cleanup_request(request);
1144
1145     if (cancelled)
1146         goto AIO_REPOLL;
1147
1148     return resultp;
1149 }                               /* squidaio_poll_done */
1150
1151 int
1152 squidaio_operations_pending(void)
1153 {
1154     return request_queue_len + (done_requests.head ? 1 : 0);
1155 }
1156
1157 int
1158 squidaio_sync(void)
1159 {
1160     /* XXX This might take a while if the queue is large.. */
1161
1162     do {
1163         squidaio_poll_queues();
1164     } while (request_queue_len > 0);
1165
1166     return squidaio_operations_pending();
1167 }
1168
1169 int
1170 squidaio_get_queue_len(void)
1171 {
1172     return request_queue_len;
1173 }
1174
1175 static void
1176 squidaio_debug(squidaio_request_t * request)
1177 {
1178     switch (request->request_type) {
1179
1180     case _AIO_OP_OPEN:
1181         debug(43, 5) ("OPEN of %s to FD %d\n", request->path, request->ret);
1182         break;
1183
1184     case _AIO_OP_READ:
1185         debug(43, 5) ("READ on fd: %d\n", request->fd);
1186         break;
1187
1188     case _AIO_OP_WRITE:
1189         debug(43, 5) ("WRITE on fd: %d\n", request->fd);
1190         break;
1191
1192     case _AIO_OP_CLOSE:
1193         debug(43, 5) ("CLOSE of fd: %d\n", request->fd);
1194         break;
1195
1196     case _AIO_OP_UNLINK:
1197         debug(43, 5) ("UNLINK of %s\n", request->path);
1198         break;
1199
1200     case _AIO_OP_TRUNCATE:
1201         debug(43, 5) ("UNLINK of %s\n", request->path);
1202         break;
1203
1204     default:
1205         break;
1206     }
1207 }
1208
1209 void
1210 squidaio_stats(StoreEntry * sentry)
1211 {
1212     squidaio_thread_t *threadp;
1213     int i;
1214
1215     if (!squidaio_initialised)
1216         return;
1217
1218     storeAppendPrintf(sentry, "\n\nThreads Status:\n");
1219
1220     storeAppendPrintf(sentry, "#\tID\t# Requests\n");
1221
1222     threadp = threads;
1223
1224     for (i = 0; i < NUMTHREADS; i++) {
1225         storeAppendPrintf(sentry, "%i\t0x%lx\t%ld\n", i + 1, threadp->dwThreadId, threadp->requests);
1226         threadp = threadp->next;
1227     }
1228 }