src/DiskIO/DiskThreads/aiops_win32.cc

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