src/DiskIO/DiskThreads/aiops_win32.cc

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