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 "DiskIO/DiskThreads/CommIO.h"
  37 #include "DiskThreads.h"
  38 #include "fd.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->freeOne(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->freeOne(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     // Initialize the thread I/O pipes before creating any threads
 300     // see bug 3189 comment 5 about race conditions.
 301     CommIO::Initialize();
 302
 303     /* Create threads and get them to sit in their wait loop */
 304     squidaio_thread_pool = memPoolCreate("aio_thread", sizeof(squidaio_thread_t));
 305
 306     assert(NUMTHREADS);
 307
 308     for (i = 0; i < NUMTHREADS; ++i) {
 309         threadp = (squidaio_thread_t *)squidaio_thread_pool->alloc();
 310         threadp->status = _THREAD_STARTING;
 311         threadp->current_req = NULL;
 312         threadp->requests = 0;
 313         threadp->next = threads;
 314         threads = threadp;
 315
 316         if ((threadp->thread = CreateThread(NULL,                   /* no security attributes */
 317                                             0,                      /* use default stack size */
 318                                             squidaio_thread_loop,   /* thread function */
 319                                             threadp,                /* argument to thread function */
 320                                             0,                      /* use default creation flags */
 321                                             &(threadp->dwThreadId)) /* returns the thread identifier */
 322             ) == NULL) {
 323             fprintf(stderr, "Thread creation failed\n");
 324             threadp->status = _THREAD_FAILED;
 325             continue;
 326         }
 327
 328         /* Set the new thread priority above parent process */
 329         SetThreadPriority(threadp->thread,THREAD_PRIORITY_ABOVE_NORMAL);
 330     }
 331
 332     /* Create request pool */
 333     squidaio_request_pool = memPoolCreate("aio_request", sizeof(squidaio_request_t));
 334
 335     squidaio_large_bufs = memPoolCreate("squidaio_large_bufs", AIO_LARGE_BUFS);
 336
 337     squidaio_medium_bufs = memPoolCreate("squidaio_medium_bufs", AIO_MEDIUM_BUFS);
 338
 339     squidaio_small_bufs = memPoolCreate("squidaio_small_bufs", AIO_SMALL_BUFS);
 340
 341     squidaio_tiny_bufs = memPoolCreate("squidaio_tiny_bufs", AIO_TINY_BUFS);
 342
 343     squidaio_micro_bufs = memPoolCreate("squidaio_micro_bufs", AIO_MICRO_BUFS);
 344
 345     squidaio_initialised = 1;
 346 }
 347
 348 void
 349 squidaio_shutdown(void)
 350 {
 351     squidaio_thread_t *threadp;
 352     int i;
 353     HANDLE * hthreads;
 354
 355     if (!squidaio_initialised)
 356         return;
 357
 358     /* This is the same as in squidaio_sync */
 359     do {
 360         squidaio_poll_queues();
 361     } while (request_queue_len > 0);
 362
 363     hthreads = (HANDLE *) xcalloc (NUMTHREADS, sizeof (HANDLE));
 364
 365     threadp = threads;
 366
 367     for (i = 0; i < NUMTHREADS; ++i) {
 368         threadp->exit = 1;
 369         hthreads[i] = threadp->thread;
 370         threadp = threadp->next;
 371     }
 372
 373     ReleaseMutex(request_queue.mutex);
 374     ResetEvent(request_queue.cond);
 375     ReleaseMutex(done_queue.mutex);
 376     ResetEvent(done_queue.cond);
 377     Sleep(0);
 378
 379     WaitForMultipleObjects(NUMTHREADS, hthreads, TRUE, 2000);
 380
 381     for (i = 0; i < NUMTHREADS; ++i) {
 382         CloseHandle(hthreads[i]);
 383     }
 384
 385     CloseHandle(main_thread);
 386     CommIO::NotifyIOClose();
 387
 388     squidaio_initialised = 0;
 389     xfree(hthreads);
 390 }
 391
 392 static DWORD WINAPI
 393 squidaio_thread_loop(LPVOID lpParam)
 394 {
 395     squidaio_thread_t *threadp = (squidaio_thread_t *)lpParam;
 396     squidaio_request_t *request;
 397     HANDLE cond; /* local copy of the event queue because win32 event handles
 398                               * don't atomically release the mutex as cond variables do. */
 399
 400     /* lock the thread info */
 401
 402     if (WAIT_FAILED == WaitForSingleObject(request_queue.mutex, INFINITE)) {
 403         fatal("Can't get ownership of mutex\n");
 404     }
 405
 406     /* duplicate the handle */
 407     if (!DuplicateHandle(GetCurrentProcess(),    /* pseudo handle, don't close */
 408                          request_queue.cond,     /* handle to copy */
 409                          GetCurrentProcess(),    /* pseudo handle, don't close */
 410                          &cond,
 411                          0,                      /* required access */
 412                          FALSE,                  /* child process's don't inherit the handle */
 413                          DUPLICATE_SAME_ACCESS))
 414         fatal("Can't duplicate mutex handle\n");
 415
 416     if (!ReleaseMutex(request_queue.mutex)) {
 417         CloseHandle(cond);
 418         fatal("Can't release mutex\n");
 419     }
 420
 421     Sleep(0);
 422
 423     while (1) {
 424         DWORD rv;
 425         threadp->current_req = request = NULL;
 426         request = NULL;
 427         /* Get a request to process */
 428         threadp->status = _THREAD_WAITING;
 429
 430         if (threadp->exit) {
 431             CloseHandle(request_queue.mutex);
 432             CloseHandle(cond);
 433             return 0;
 434         }
 435
 436         rv = WaitForSingleObject(request_queue.mutex, INFINITE);
 437
 438         if (rv == WAIT_FAILED) {
 439             CloseHandle(cond);
 440             return 1;
 441         }
 442
 443         while (!request_queue.head) {
 444             if (!ReleaseMutex(request_queue.mutex)) {
 445                 CloseHandle(cond);
 446                 threadp->status = _THREAD_FAILED;
 447                 return 1;
 448             }
 449
 450             Sleep(0);
 451             rv = WaitForSingleObject(cond, INFINITE);
 452
 453             if (rv == WAIT_FAILED) {
 454                 CloseHandle(cond);
 455                 return 1;
 456             }
 457
 458             rv = WaitForSingleObject(request_queue.mutex, INFINITE);
 459
 460             if (rv == WAIT_FAILED) {
 461                 CloseHandle(cond);
 462                 return 1;
 463             }
 464         }
 465
 466         request = request_queue.head;
 467
 468         if (request)
 469             request_queue.head = request->next;
 470
 471         if (!request_queue.head)
 472             request_queue.tailp = &request_queue.head;
 473
 474         if (!ReleaseMutex(request_queue.mutex)) {
 475             CloseHandle(cond);
 476             return 1;
 477         }
 478
 479         Sleep(0);
 480
 481         /* process the request */
 482         threadp->status = _THREAD_BUSY;
 483
 484         request->next = NULL;
 485
 486         threadp->current_req = request;
 487
 488         errno = 0;
 489
 490         if (!request->cancelled) {
 491             switch (request->request_type) {
 492
 493             case _AIO_OP_OPEN:
 494                 squidaio_do_open(request);
 495                 break;
 496
 497             case _AIO_OP_READ:
 498                 squidaio_do_read(request);
 499                 break;
 500
 501             case _AIO_OP_WRITE:
 502                 squidaio_do_write(request);
 503                 break;
 504
 505             case _AIO_OP_CLOSE:
 506                 squidaio_do_close(request);
 507                 break;
 508
 509             case _AIO_OP_UNLINK:
 510                 squidaio_do_unlink(request);
 511                 break;
 512
 513 #if AIO_OPENDIR                 /* Opendir not implemented yet */
 514
 515             case _AIO_OP_OPENDIR:
 516                 squidaio_do_opendir(request);
 517                 break;
 518 #endif
 519
 520             case _AIO_OP_STAT:
 521                 squidaio_do_stat(request);
 522                 break;
 523
 524             default:
 525                 request->ret = -1;
 526                 request->err = EINVAL;
 527                 break;
 528             }
 529         } else {                /* cancelled */
 530             request->ret = -1;
 531             request->err = EINTR;
 532         }
 533
 534         threadp->status = _THREAD_DONE;
 535         /* put the request in the done queue */
 536         rv = WaitForSingleObject(done_queue.mutex, INFINITE);
 537
 538         if (rv == WAIT_FAILED) {
 539             CloseHandle(cond);
 540             return 1;
 541         }
 542
 543         *done_queue.tailp = request;
 544         done_queue.tailp = &request->next;
 545
 546         if (!ReleaseMutex(done_queue.mutex)) {
 547             CloseHandle(cond);
 548             return 1;
 549         }
 550
 551         CommIO::NotifyIOCompleted();
 552         Sleep(0);
 553         ++ threadp->requests;
 554     }                           /* while forever */
 555
 556     CloseHandle(cond);
 557
 558     return 0;
 559 }                               /* squidaio_thread_loop */
 560
 561 static void
 562 squidaio_queue_request(squidaio_request_t * request)
 563 {
 564     static int high_start = 0;
 565     debugs(43, 9, "squidaio_queue_request: " << request << " type=" << request->request_type << " result=" << request->resultp);
 566     /* Mark it as not executed (failing result, no error) */
 567     request->ret = -1;
 568     request->err = 0;
 569     /* Internal housekeeping */
 570     request_queue_len += 1;
 571     request->resultp->_data = request;
 572     /* Play some tricks with the request_queue2 queue */
 573     request->next = NULL;
 574
 575     if (WaitForSingleObject(request_queue.mutex, 0) == WAIT_OBJECT_0) {
 576         if (request_queue2.head) {
 577             /* Grab blocked requests */
 578             *request_queue.tailp = request_queue2.head;
 579             request_queue.tailp = request_queue2.tailp;
 580         }
 581
 582         /* Enqueue request */
 583         *request_queue.tailp = request;
 584
 585         request_queue.tailp = &request->next;
 586
 587         if (!SetEvent(request_queue.cond))
 588             fatal("Couldn't push queue");
 589
 590         if (!ReleaseMutex(request_queue.mutex)) {
 591             /* unexpected error */
 592             fatal("Couldn't push queue");
 593         }
 594
 595         Sleep(0);
 596
 597         if (request_queue2.head) {
 598             /* Clear queue of blocked requests */
 599             request_queue2.head = NULL;
 600             request_queue2.tailp = &request_queue2.head;
 601         }
 602     } else {
 603         /* Oops, the request queue is blocked, use request_queue2 */
 604         *request_queue2.tailp = request;
 605         request_queue2.tailp = &request->next;
 606     }
 607
 608     if (request_queue2.head) {
 609         static int filter = 0;
 610         static int filter_limit = 8;
 611
 612         if (++filter >= filter_limit) {
 613             filter_limit += filter;
 614             filter = 0;
 615             debugs(43, DBG_IMPORTANT, "squidaio_queue_request: WARNING - Queue congestion");
 616         }
 617     }
 618
 619     /* Warn if out of threads */
 620     if (request_queue_len > MAGIC1) {
 621         static int last_warn = 0;
 622         static int queue_high, queue_low;
 623
 624         if (high_start == 0) {
 625             high_start = (int)squid_curtime;
 626             queue_high = request_queue_len;
 627             queue_low = request_queue_len;
 628         }
 629
 630         if (request_queue_len > queue_high)
 631             queue_high = request_queue_len;
 632
 633         if (request_queue_len < queue_low)
 634             queue_low = request_queue_len;
 635
 636         if (squid_curtime >= (last_warn + 15) &&
 637                 squid_curtime >= (high_start + 5)) {
 638             debugs(43, DBG_IMPORTANT, "squidaio_queue_request: WARNING - Disk I/O overloading");
 639
 640             if (squid_curtime >= (high_start + 15))
 641                 debugs(43, DBG_IMPORTANT, "squidaio_queue_request: Queue Length: current=" <<
 642                        request_queue_len << ", high=" << queue_high <<
 643                        ", low=" << queue_low << ", duration=" <<
 644                        (long int) (squid_curtime - high_start));
 645
 646             last_warn = (int)squid_curtime;
 647         }
 648     } else {
 649         high_start = 0;
 650     }
 651
 652     /* Warn if seriously overloaded */
 653     if (request_queue_len > RIDICULOUS_LENGTH) {
 654         debugs(43, DBG_CRITICAL, "squidaio_queue_request: Async request queue growing uncontrollably!");
 655         debugs(43, DBG_CRITICAL, "squidaio_queue_request: Syncing pending I/O operations.. (blocking)");
 656         squidaio_sync();
 657         debugs(43, DBG_CRITICAL, "squidaio_queue_request: Synced");
 658     }
 659 }                               /* squidaio_queue_request */
 660
 661 static void
 662 squidaio_cleanup_request(squidaio_request_t * requestp)
 663 {
 664     squidaio_result_t *resultp = requestp->resultp;
 665     int cancelled = requestp->cancelled;
 666
 667     /* Free allocated structures and copy data back to user space if the */
 668     /* request hasn't been cancelled */
 669
 670     switch (requestp->request_type) {
 671
 672     case _AIO_OP_STAT:
 673
 674         if (!cancelled && requestp->ret == 0)
 675             memcpy(requestp->statp, requestp->tmpstatp, sizeof(struct stat));
 676
 677         squidaio_xfree(requestp->tmpstatp, sizeof(struct stat));
 678
 679         squidaio_xstrfree(requestp->path);
 680
 681         break;
 682
 683     case _AIO_OP_OPEN:
 684         if (cancelled && requestp->ret >= 0)
 685             /* The open() was cancelled but completed */
 686             close(requestp->ret);
 687
 688         squidaio_xstrfree(requestp->path);
 689
 690         break;
 691
 692     case _AIO_OP_CLOSE:
 693         if (cancelled && requestp->ret < 0)
 694             /* The close() was cancelled and never got executed */
 695             close(requestp->fd);
 696
 697         break;
 698
 699     case _AIO_OP_UNLINK:
 700
 701     case _AIO_OP_OPENDIR:
 702         squidaio_xstrfree(requestp->path);
 703
 704         break;
 705
 706     case _AIO_OP_READ:
 707         break;
 708
 709     case _AIO_OP_WRITE:
 710         break;
 711
 712     default:
 713         break;
 714     }
 715
 716     if (resultp != NULL && !cancelled) {
 717         resultp->aio_return = requestp->ret;
 718         resultp->aio_errno = requestp->err;
 719     }
 720
 721     squidaio_request_pool->freeOne(requestp);
 722 }                               /* squidaio_cleanup_request */
 723
 724 int
 725 squidaio_cancel(squidaio_result_t * resultp)
 726 {
 727     squidaio_request_t *request = (squidaio_request_t *)resultp->_data;
 728
 729     if (request && request->resultp == resultp) {
 730         debugs(43, 9, "squidaio_cancel: " << request << " type=" << request->request_type << " result=" << request->resultp);
 731         request->cancelled = 1;
 732         request->resultp = NULL;
 733         resultp->_data = NULL;
 734         resultp->result_type = _AIO_OP_NONE;
 735         return 0;
 736     }
 737
 738     return 1;
 739 }                               /* squidaio_cancel */
 740
 741 int
 742 squidaio_open(const char *path, int oflag, mode_t mode, squidaio_result_t * resultp)
 743 {
 744     squidaio_init();
 745     squidaio_request_t *requestp;
 746
 747     requestp = (squidaio_request_t *)squidaio_request_pool->alloc();
 748
 749     requestp->path = (char *) squidaio_xstrdup(path);
 750
 751     requestp->oflag = oflag;
 752
 753     requestp->mode = mode;
 754
 755     requestp->resultp = resultp;
 756
 757     requestp->request_type = _AIO_OP_OPEN;
 758
 759     requestp->cancelled = 0;
 760
 761     resultp->result_type = _AIO_OP_OPEN;
 762
 763     squidaio_queue_request(requestp);
 764
 765     return 0;
 766 }
 767
 768 static void
 769 squidaio_do_open(squidaio_request_t * requestp)
 770 {
 771     requestp->ret = open(requestp->path, requestp->oflag, requestp->mode);
 772     requestp->err = errno;
 773 }
 774
 775 int
 776 squidaio_read(int fd, char *bufp, size_t bufs, off_t offset, int whence, squidaio_result_t * resultp)
 777 {
 778     squidaio_request_t *requestp;
 779
 780     requestp = (squidaio_request_t *)squidaio_request_pool->alloc();
 781
 782     requestp->fd = fd;
 783
 784     requestp->bufferp = bufp;
 785
 786     requestp->buflen = bufs;
 787
 788     requestp->offset = offset;
 789
 790     requestp->whence = whence;
 791
 792     requestp->resultp = resultp;
 793
 794     requestp->request_type = _AIO_OP_READ;
 795
 796     requestp->cancelled = 0;
 797
 798     resultp->result_type = _AIO_OP_READ;
 799
 800     squidaio_queue_request(requestp);
 801
 802     return 0;
 803 }
 804
 805 static void
 806 squidaio_do_read(squidaio_request_t * requestp)
 807 {
 808     lseek(requestp->fd, requestp->offset, requestp->whence);
 809
 810     if (!ReadFile((HANDLE)_get_osfhandle(requestp->fd), requestp->bufferp,
 811                   requestp->buflen, (LPDWORD)&requestp->ret, NULL)) {
 812         WIN32_maperror(GetLastError());
 813         requestp->ret = -1;
 814     }
 815
 816     requestp->err = errno;
 817 }
 818
 819 int
 820 squidaio_write(int fd, char *bufp, size_t bufs, off_t offset, int whence, squidaio_result_t * resultp)
 821 {
 822     squidaio_request_t *requestp;
 823
 824     requestp = (squidaio_request_t *)squidaio_request_pool->alloc();
 825
 826     requestp->fd = fd;
 827
 828     requestp->bufferp = bufp;
 829
 830     requestp->buflen = bufs;
 831
 832     requestp->offset = offset;
 833
 834     requestp->whence = whence;
 835
 836     requestp->resultp = resultp;
 837
 838     requestp->request_type = _AIO_OP_WRITE;
 839
 840     requestp->cancelled = 0;
 841
 842     resultp->result_type = _AIO_OP_WRITE;
 843
 844     squidaio_queue_request(requestp);
 845
 846     return 0;
 847 }
 848
 849 static void
 850 squidaio_do_write(squidaio_request_t * requestp)
 851 {
 852     if (!WriteFile((HANDLE)_get_osfhandle(requestp->fd), requestp->bufferp,
 853                    requestp->buflen, (LPDWORD)&requestp->ret, NULL)) {
 854         WIN32_maperror(GetLastError());
 855         requestp->ret = -1;
 856     }
 857
 858     requestp->err = errno;
 859 }
 860
 861 int
 862 squidaio_close(int fd, squidaio_result_t * resultp)
 863 {
 864     squidaio_request_t *requestp;
 865
 866     requestp = (squidaio_request_t *)squidaio_request_pool->alloc();
 867
 868     requestp->fd = fd;
 869
 870     requestp->resultp = resultp;
 871
 872     requestp->request_type = _AIO_OP_CLOSE;
 873
 874     requestp->cancelled = 0;
 875
 876     resultp->result_type = _AIO_OP_CLOSE;
 877
 878     squidaio_queue_request(requestp);
 879
 880     return 0;
 881 }
 882
 883 static void
 884 squidaio_do_close(squidaio_request_t * requestp)
 885 {
 886     if ((requestp->ret = close(requestp->fd)) < 0) {
 887         debugs(43, DBG_CRITICAL, "squidaio_do_close: FD " << requestp->fd << ", errno " << errno);
 888         close(requestp->fd);
 889     }
 890
 891     requestp->err = errno;
 892 }
 893
 894 int
 895
 896 squidaio_stat(const char *path, struct stat *sb, squidaio_result_t * resultp)
 897 {
 898     squidaio_init();
 899     squidaio_request_t *requestp;
 900
 901     requestp = (squidaio_request_t *)squidaio_request_pool->alloc();
 902
 903     requestp->path = (char *) squidaio_xstrdup(path);
 904
 905     requestp->statp = sb;
 906
 907     requestp->tmpstatp = (struct stat *) squidaio_xmalloc(sizeof(struct stat));
 908
 909     requestp->resultp = resultp;
 910
 911     requestp->request_type = _AIO_OP_STAT;
 912
 913     requestp->cancelled = 0;
 914
 915     resultp->result_type = _AIO_OP_STAT;
 916
 917     squidaio_queue_request(requestp);
 918
 919     return 0;
 920 }
 921
 922 static void
 923 squidaio_do_stat(squidaio_request_t * requestp)
 924 {
 925     requestp->ret = stat(requestp->path, requestp->tmpstatp);
 926     requestp->err = errno;
 927 }
 928
 929 int
 930 squidaio_unlink(const char *path, squidaio_result_t * resultp)
 931 {
 932     squidaio_init();
 933     squidaio_request_t *requestp;
 934
 935     requestp = (squidaio_request_t *)squidaio_request_pool->alloc();
 936
 937     requestp->path = squidaio_xstrdup(path);
 938
 939     requestp->resultp = resultp;
 940
 941     requestp->request_type = _AIO_OP_UNLINK;
 942
 943     requestp->cancelled = 0;
 944
 945     resultp->result_type = _AIO_OP_UNLINK;
 946
 947     squidaio_queue_request(requestp);
 948
 949     return 0;
 950 }
 951
 952 static void
 953 squidaio_do_unlink(squidaio_request_t * requestp)
 954 {
 955     requestp->ret = unlink(requestp->path);
 956     requestp->err = errno;
 957 }
 958
 959 #if AIO_OPENDIR
 960 /* XXX squidaio_opendir NOT implemented yet.. */
 961
 962 int
 963 squidaio_opendir(const char *path, squidaio_result_t * resultp)
 964 {
 965     squidaio_request_t *requestp;
 966     int len;
 967
 968     requestp = squidaio_request_pool->alloc();
 969
 970     resultp->result_type = _AIO_OP_OPENDIR;
 971
 972     return -1;
 973 }
 974
 975 static void
 976 squidaio_do_opendir(squidaio_request_t * requestp)
 977 {
 978     /* NOT IMPLEMENTED */
 979 }
 980
 981 #endif
 982
 983 static void
 984 squidaio_poll_queues(void)
 985 {
 986     /* kick "overflow" request queue */
 987
 988     if (request_queue2.head &&
 989             (WaitForSingleObject(request_queue.mutex, 0 )== WAIT_OBJECT_0)) {
 990         *request_queue.tailp = request_queue2.head;
 991         request_queue.tailp = request_queue2.tailp;
 992
 993         if (!SetEvent(request_queue.cond))
 994             fatal("couldn't push queue\n");
 995
 996         if (!ReleaseMutex(request_queue.mutex)) {
 997             /* unexpected error */
 998         }
 999
1000         Sleep(0);
1001         request_queue2.head = NULL;
1002         request_queue2.tailp = &request_queue2.head;
1003     }
1004
1005     /* poll done queue */
1006     if (done_queue.head &&
1007             (WaitForSingleObject(done_queue.mutex, 0)==WAIT_OBJECT_0)) {
1008
1009         struct squidaio_request_t *requests = done_queue.head;
1010         done_queue.head = NULL;
1011         done_queue.tailp = &done_queue.head;
1012
1013         if (!ReleaseMutex(done_queue.mutex)) {
1014             /* unexpected error */
1015         }
1016
1017         Sleep(0);
1018         *done_requests.tailp = requests;
1019         request_queue_len -= 1;
1020
1021         while (requests->next) {
1022             requests = requests->next;
1023             request_queue_len -= 1;
1024         }
1025
1026         done_requests.tailp = &requests->next;
1027     }
1028 }
1029
1030 squidaio_result_t *
1031 squidaio_poll_done(void)
1032 {
1033     squidaio_request_t *request;
1034     squidaio_result_t *resultp;
1035     int cancelled;
1036     int polled = 0;
1037
1038 AIO_REPOLL:
1039     request = done_requests.head;
1040
1041     if (request == NULL && !polled) {
1042         CommIO::ResetNotifications();
1043         squidaio_poll_queues();
1044         polled = 1;
1045         request = done_requests.head;
1046     }
1047
1048     if (!request) {
1049         return NULL;
1050     }
1051
1052     debugs(43, 9, "squidaio_poll_done: " << request << " type=" << request->request_type << " result=" << request->resultp);
1053     done_requests.head = request->next;
1054
1055     if (!done_requests.head)
1056         done_requests.tailp = &done_requests.head;
1057
1058     resultp = request->resultp;
1059
1060     cancelled = request->cancelled;
1061
1062     squidaio_debug(request);
1063
1064     debugs(43, 5, "DONE: " << request->ret << " -> " << request->err);
1065
1066     squidaio_cleanup_request(request);
1067
1068     if (cancelled)
1069         goto AIO_REPOLL;
1070
1071     return resultp;
1072 }                               /* squidaio_poll_done */
1073
1074 int
1075 squidaio_operations_pending(void)
1076 {
1077     return request_queue_len + (done_requests.head ? 1 : 0);
1078 }
1079
1080 int
1081 squidaio_sync(void)
1082 {
1083     /* XXX This might take a while if the queue is large.. */
1084
1085     do {
1086         squidaio_poll_queues();
1087     } while (request_queue_len > 0);
1088
1089     return squidaio_operations_pending();
1090 }
1091
1092 int
1093 squidaio_get_queue_len(void)
1094 {
1095     return request_queue_len;
1096 }
1097
1098 static void
1099 squidaio_debug(squidaio_request_t * request)
1100 {
1101     switch (request->request_type) {
1102
1103     case _AIO_OP_OPEN:
1104         debugs(43, 5, "OPEN of " << request->path << " to FD " << request->ret);
1105         break;
1106
1107     case _AIO_OP_READ:
1108         debugs(43, 5, "READ on fd: " << request->fd);
1109         break;
1110
1111     case _AIO_OP_WRITE:
1112         debugs(43, 5, "WRITE on fd: " << request->fd);
1113         break;
1114
1115     case _AIO_OP_CLOSE:
1116         debugs(43, 5, "CLOSE of fd: " << request->fd);
1117         break;
1118
1119     case _AIO_OP_UNLINK:
1120         debugs(43, 5, "UNLINK of " << request->path);
1121         break;
1122
1123     default:
1124         break;
1125     }
1126 }
1127
1128 void
1129 squidaio_stats(StoreEntry * sentry)
1130 {
1131     squidaio_thread_t *threadp;
1132     int i;
1133
1134     if (!squidaio_initialised)
1135         return;
1136
1137     storeAppendPrintf(sentry, "\n\nThreads Status:\n");
1138
1139     storeAppendPrintf(sentry, "#\tID\t# Requests\n");
1140
1141     threadp = threads;
1142
1143     for (i = 0; i < NUMTHREADS; ++i) {
1144         storeAppendPrintf(sentry, "%i\t0x%lx\t%ld\n", i + 1, threadp->dwThreadId, threadp->requests);
1145         threadp = threadp->next;
1146     }
1147 }