src/DiskIO/DiskThreads/aiops_win32.cc

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