开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > Internet/网络编程 > 代理服务器 > 查看源码
aiops.c
资源名称：squid-2.2.devel3-src.tar.gz　[点击查看]
上传用户：liugui
上传日期：2007-01-04
资源大小：822k
文件大小：22k
源码类别：

代理服务器

开发平台：

Unix_Linux

aiops.c：源码内容
  1. /*
  2.  * $Id: aiops.c,v 1.25 1999/01/24 02:26:20 wessels Exp $
  3.  *
  4.  * DEBUG: section 43    AIOPS
  5.  * AUTHOR: Stewart Forster <slf@connect.com.au>
  6.  *
  7.  * SQUID Internet Object Cache  http://squid.nlanr.net/Squid/
  8.  * ----------------------------------------------------------
  9.  *
  10.  *  Squid is the result of efforts by numerous individuals from the
  11.  *  Internet community.  Development is led by Duane Wessels of the
  12.  *  National Laboratory for Applied Network Research and funded by the
  13.  *  National Science Foundation.  Squid is Copyrighted (C) 1998 by
  14.  *  Duane Wessels and the University of California San Diego.  Please
  15.  *  see the COPYRIGHT file for full details.  Squid incorporates
  16.  *  software developed and/or copyrighted by other sources.  Please see
  17.  *  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.  */
  35. #include "squid.h"
  37. #include <stdio.h>
  38. #include <sys/types.h>
  39. #include <sys/stat.h>
  40. #include <fcntl.h>
  41. #include <pthread.h>
  42. #include <errno.h>
  43. #include <dirent.h>
  44. #include <signal.h>
  45. #if HAVE_SCHED_H
  46. #include <sched.h>
  47. #endif
  49. #ifndef NUMTHREADS
  50. #define NUMTHREADS 16
  51. #endif
  52. #define RIDICULOUS_LENGTH 4096
  54. enum _aio_thread_status {
  55.     _THREAD_STARTING = 0,
  56.     _THREAD_WAITING,
  57.     _THREAD_BUSY,
  58.     _THREAD_FAILED,
  59.     _THREAD_DONE
  60. };
  62. enum _aio_request_type {
  63.     _AIO_OP_NONE = 0,
  64.     _AIO_OP_OPEN,
  65.     _AIO_OP_READ,
  66.     _AIO_OP_WRITE,
  67.     _AIO_OP_CLOSE,
  68.     _AIO_OP_UNLINK,
  69.     _AIO_OP_OPENDIR,
  70.     _AIO_OP_STAT
  71. };
  73. typedef struct aio_request_t {
  74.     enum _aio_request_type request_type;
  75.     int cancelled;
  76.     char *path;
  77.     int oflag;
  78.     mode_t mode;
  79.     int fd;
  80.     char *bufferp;
  81.     char *tmpbufp;
  82.     int buflen;
  83.     off_t offset;
  84.     int whence;
  85.     int ret;
  86.     int err;
  87.     struct stat *tmpstatp;
  88.     struct stat *statp;
  89.     aio_result_t *resultp;
  90.     struct aio_request_t *next;
  91. } aio_request_t;
  94. typedef struct aio_thread_t {
  95.     pthread_t thread;
  96.     enum _aio_thread_status status;
  97.     pthread_mutex_t mutex; /* Mutex for testing condition variable */
  98.     pthread_cond_t cond; /* Condition variable */
  99.     struct aio_request_t *volatile req; /* set by main, cleared by thread */
  100.     struct aio_request_t *processed_req; /* reminder to main */
  101.     struct aio_thread_t *next;
  102. } aio_thread_t;
  105. int aio_cancel(aio_result_t *);
  106. int aio_open(const char *, int, mode_t, aio_result_t *);
  107. int aio_read(int, char *, int, off_t, int, aio_result_t *);
  108. int aio_write(int, char *, int, off_t, int, aio_result_t *);
  109. int aio_close(int, aio_result_t *);
  110. int aio_unlink(const char *, aio_result_t *);
  111. int aio_opendir(const char *, aio_result_t *);
  112. aio_result_t *aio_poll_done();
  114. static void aio_init(void);
  115. static void aio_queue_request(aio_request_t *);
  116. static void aio_process_request_queue(void);
  117. static void aio_cleanup_request(aio_request_t *);
  118. static void *aio_thread_loop(void *);
  119. static void aio_do_open(aio_request_t *);
  120. static void aio_do_read(aio_request_t *);
  121. static void aio_do_write(aio_request_t *);
  122. static void aio_do_close(aio_request_t *);
  123. static void aio_do_stat(aio_request_t *);
  124. static void aio_do_unlink(aio_request_t *);
  125. #if AIO_OPENDIR
  126. static void *aio_do_opendir(aio_request_t *);
  127. #endif
  128. static void aio_debug(aio_request_t *);
  129. static void aio_poll_threads(void);
  131. static aio_thread_t *threads;
  132. static int aio_initialised = 0;
  134. static int request_queue_len = 0;
  135. static MemPool *aio_request_pool = NULL;
  136. static aio_request_t *request_queue_head = NULL;
  137. static aio_request_t *request_queue_tail = NULL;
  138. static aio_request_t *request_done_head = NULL;
  139. static aio_request_t *request_done_tail = NULL;
  140. static aio_thread_t *wait_threads = NULL;
  141. static aio_thread_t *busy_threads_head = NULL;
  142. static aio_thread_t *busy_threads_tail = NULL;
  143. static pthread_attr_t globattr;
  144. static struct sched_param globsched;
  145. static pthread_t main_thread;
  147. static void
  148. aio_init(void)
  149. {
  150.     int i;
  151.     aio_thread_t *threadp;
  153.     if (aio_initialised)
  154. return;
  156.     pthread_attr_init(&globattr);
  157. #if HAVE_PTHREAD_ATTR_SETSCOPE
  158.     pthread_attr_setscope(&globattr, PTHREAD_SCOPE_SYSTEM);
  159. #endif
  160.     globsched.sched_priority = 1;
  161.     main_thread = pthread_self();
  162. #if HAVE_PTHREAD_SETSCHEDPARAM
  163.     pthread_setschedparam(main_thread, SCHED_OTHER, &globsched);
  164. #endif
  165.     globsched.sched_priority = 2;
  166. #if HAVE_PTHREAD_ATTR_SETSCHEDPARAM
  167.     pthread_attr_setschedparam(&globattr, &globsched);
  168. #endif
  170.     /* Create threads and get them to sit in their wait loop */
  171.     threads = xcalloc(NUMTHREADS, sizeof(aio_thread_t));
  173.     for (i = 0; i < NUMTHREADS; i++) {
  174. threadp = &threads[i];
  175. threadp->status = _THREAD_STARTING;
  176. if (pthread_mutex_init(&(threadp->mutex), NULL)) {
  177.     threadp->status = _THREAD_FAILED;
  178.     continue;
  179. }
  180. if (pthread_cond_init(&(threadp->cond), NULL)) {
  181.     threadp->status = _THREAD_FAILED;
  182.     continue;
  183. }
  184. threadp->req = NULL;
  185. threadp->processed_req = NULL;
  186. if (pthread_create(&(threadp->thread), &globattr, aio_thread_loop, threadp)) {
  187.     fprintf(stderr, "Thread creation failedn");
  188.     threadp->status = _THREAD_FAILED;
  189.     continue;
  190. }
  191. threadp->next = wait_threads;
  192. wait_threads = threadp;
  193. #if AIO_PROPER_MUTEX
  194. pthread_mutex_lock(&threadp->mutex);
  195. #endif
  196.     }
  198.     /* Create request pool */
  199.     aio_request_pool = memPoolCreate("aio_request", sizeof(aio_request_t));
  201.     aio_initialised = 1;
  202. }
  205. static void *
  206. aio_thread_loop(void *ptr)
  207. {
  208.     aio_thread_t *threadp = (aio_thread_t *) ptr;
  209.     aio_request_t *request;
  210.     sigset_t new;
  211. #if !AIO_PROPER_MUTEX
  212.     struct timespec wait_time;
  213. #endif
  215.     /* Make sure to ignore signals which may possibly get sent to the parent */
  216.     /* squid thread.  Causes havoc with mutex's and condition waits otherwise */
  218.     sigemptyset(&new);
  219.     sigaddset(&new, SIGPIPE);
  220.     sigaddset(&new, SIGCHLD);
  221. #ifdef _SQUID_LINUX_THREADS_
  222.     sigaddset(&new, SIGQUIT);
  223.     sigaddset(&new, SIGTRAP);
  224. #else
  225.     sigaddset(&new, SIGUSR1);
  226.     sigaddset(&new, SIGUSR2);
  227. #endif
  228.     sigaddset(&new, SIGHUP);
  229.     sigaddset(&new, SIGTERM);
  230.     sigaddset(&new, SIGINT);
  231.     sigaddset(&new, SIGALRM);
  232.     pthread_sigmask(SIG_BLOCK, &new, NULL);
  234.     pthread_mutex_lock(&threadp->mutex);
  235.     while (1) {
  236. #if AIO_PROPER_MUTEX
  237. while (threadp->req == NULL) {
  238.     threadp->status = _THREAD_WAITING;
  239.     pthread_cond_wait(&(threadp->cond), &(threadp->mutex));
  240. }
  241. #else
  242. /* The timeout is used to unlock the race condition where
  243.  * ->req is set between the check and pthread_cond_wait.
  244.  * The thread steps it's own clock on each timeout, to avoid a CPU
  245.  * spin situation if the main thread is suspended (paging), and
  246.  * squid_curtime is not being updated timely.
  247.  */
  248. wait_time.tv_sec = squid_curtime + 1; /* little quicker first time */
  249. wait_time.tv_nsec = 0;
  250. while (threadp->req == NULL) {
  251.     threadp->status = _THREAD_WAITING;
  252.     pthread_cond_timedwait(&(threadp->cond), &(threadp->mutex),
  253. &wait_time);
  254.     wait_time.tv_sec += 3; /* then wait 3 seconds between each check */
  255. }
  256. #endif
  257. request = threadp->req;
  258. errno = 0;
  259. if (!request->cancelled) {
  260.     switch (request->request_type) {
  261.     case _AIO_OP_OPEN:
  262. aio_do_open(request);
  263. break;
  264.     case _AIO_OP_READ:
  265. aio_do_read(request);
  266. break;
  267.     case _AIO_OP_WRITE:
  268. aio_do_write(request);
  269. break;
  270.     case _AIO_OP_CLOSE:
  271. aio_do_close(request);
  272. break;
  273.     case _AIO_OP_UNLINK:
  274. aio_do_unlink(request);
  275. break;
  276. #if AIO_OPENDIR /* Opendir not implemented yet */
  277.     case _AIO_OP_OPENDIR:
  278. aio_do_opendir(request);
  279. break;
  280. #endif
  281.     case _AIO_OP_STAT:
  282. aio_do_stat(request);
  283. break;
  284.     default:
  285. request->ret = -1;
  286. request->err = EINVAL;
  287. break;
  288.     }
  289. } else { /* cancelled */
  290.     request->ret = -1;
  291.     request->err = EINTR;
  292. }
  293. threadp->req = NULL; /* tells main thread that we are done */
  294.     } /* while */
  295. } /* aio_thread_loop */
  297. static void
  298. aio_do_request(aio_request_t * requestp)
  299. {
  300.     if (wait_threads == NULL && busy_threads_head == NULL) {
  301. fprintf(stderr, "PANIC: No threads to service requests with!n");
  302. exit(-1);
  303.     }
  304.     aio_queue_request(requestp);
  305. } /* aio_do_request */
  308. static void
  309. aio_queue_request(aio_request_t * requestp)
  310. {
  311.     aio_request_t *rp;
  312.     static int last_warn = 0;
  313.     static int high_start = 0;
  314.     static int queue_high, queue_low;
  315.     int i;
  317.     /* Mark it as not executed (failing result, no error) */
  318.     requestp->ret = -1;
  319.     requestp->err = 0;
  320.     /* Queue it on the request queue */
  321.     if (request_queue_head == NULL) {
  322. request_queue_head = requestp;
  323. request_queue_tail = requestp;
  324.     } else {
  325. request_queue_tail->next = requestp;
  326. request_queue_tail = requestp;
  327.     }
  328.     requestp->next = NULL;
  329.     request_queue_len += 1;
  330.     /* Poll done threads if needed */
  331.     if (wait_threads == NULL)
  332. aio_poll_threads();
  333.     /* Kick it rolling */
  334.     aio_process_request_queue();
  335.     /* Warn if out of threads */
  336.     if (request_queue_len > (NUMTHREADS >> 1)) {
  337. if (high_start == 0) {
  338.     high_start = squid_curtime;
  339.     queue_high = request_queue_len;
  340.     queue_low = request_queue_len;
  341. }
  342. if (request_queue_len > queue_high)
  343.     queue_high = request_queue_len;
  344. if (request_queue_len < queue_low)
  345.     queue_low = request_queue_len;
  346. if (squid_curtime >= (last_warn + 15) &&
  347.     squid_curtime >= (high_start + 1)) {
  348.     debug(43, 1) ("aio_queue_request: WARNING - Running out of I/O theadsn");
  349.     debug(43, 2) ("aio_queue_request: Queue Length: current=%d, high=%d, low=%d, duration=%dn",
  350. request_queue_len, queue_high, queue_low, squid_curtime - high_start);
  351.     debug(43, 1) ("aio_queue_request: Perhaps you should increase NUMTHREADSn");
  352.     debug(43, 1) ("aio_queue_request: Or install more disks to share the loadn");
  353.     debug(43, 3) ("aio_queue_request: First %d items on request queuen", NUMTHREADS);
  354.     rp = request_queue_head;
  355.     for (i = 1; i <= NUMTHREADS; i++) {
  356. switch (rp->request_type) {
  357. case _AIO_OP_OPEN:
  358.     debug(43, 3) ("aio_queue_request: %d : open -> %sn", i, rp->path);
  359.     break;
  360. case _AIO_OP_READ:
  361.     debug(43, 3) ("aio_queue_request: %d : read -> FD = %dn", i, rp->fd);
  362.     break;
  363. case _AIO_OP_WRITE:
  364.     debug(43, 3) ("aio_queue_request: %d : write -> FD = %dn", i, rp->fd);
  365.     break;
  366. case _AIO_OP_CLOSE:
  367.     debug(43, 3) ("aio_queue_request: %d : close -> FD = %dn", i, rp->fd);
  368.     break;
  369. case _AIO_OP_UNLINK:
  370.     debug(43, 3) ("aio_queue_request: %d : unlink -> %sn", i, rp->path);
  371.     break;
  372. case _AIO_OP_STAT:
  373.     debug(43, 3) ("aio_queue_request: %d : stat -> %sn", i, rp->path);
  374.     break;
  375. default:
  376.     debug(43, 1) ("aio_queue_request: %d : Unimplemented request type: %dn", i, rp->request_type);
  377.     break;
  378. }
  379. if ((rp = rp->next) == NULL)
  380.     break;
  381.     }
  382.     last_warn = squid_curtime;
  383. }
  384.     } else {
  385. high_start = 0;
  386.     }
  387.     if (request_queue_len > RIDICULOUS_LENGTH) {
  388. debug(43, 0) ("aio_queue_request: Async request queue growing uncontrollably!n");
  389. debug(43, 0) ("aio_queue_request: Possible infinite loop somewhere in squid. Restarting...n");
  390. abort();
  391.     }
  392. } /* aio_queue_request */
  395. static void
  396. aio_process_request_queue()
  397. {
  398.     aio_thread_t *threadp;
  399.     aio_request_t *requestp;
  401.     for (;;) {
  402. if (wait_threads == NULL || request_queue_head == NULL)
  403.     return;
  405. requestp = request_queue_head;
  406. if ((request_queue_head = requestp->next) == NULL)
  407.     request_queue_tail = NULL;
  408. requestp->next = NULL;
  409. request_queue_len--;
  411. if (requestp->cancelled) {
  412.     aio_cleanup_request(requestp);
  413.     continue;
  414. }
  415. threadp = wait_threads;
  416. wait_threads = threadp->next;
  417. threadp->next = NULL;
  419. if (busy_threads_head != NULL)
  420.     busy_threads_tail->next = threadp;
  421. else
  422.     busy_threads_head = threadp;
  423. busy_threads_tail = threadp;
  425. threadp->status = _THREAD_BUSY;
  426. threadp->req = threadp->processed_req = requestp;
  427. pthread_cond_signal(&(threadp->cond));
  428. #if AIO_PROPER_MUTEX
  429. pthread_mutex_unlock(&threadp->mutex);
  430. #endif
  431.     }
  432. } /* aio_process_request_queue */
  435. static void
  436. aio_cleanup_request(aio_request_t * requestp)
  437. {
  438.     aio_result_t *resultp = requestp->resultp;
  439.     int cancelled = requestp->cancelled;
  441.     /* Free allocated structures and copy data back to user space if the */
  442.     /* request hasn't been cancelled */
  443.     switch (requestp->request_type) {
  444.     case _AIO_OP_STAT:
  445. if (!cancelled && requestp->ret == 0)
  446.     xmemcpy(requestp->statp, requestp->tmpstatp, sizeof(struct stat));
  447. xfree(requestp->tmpstatp);
  448.     case _AIO_OP_OPEN:
  449. if (cancelled && requestp->ret >= 0)
  450.     /* The open() was cancelled but completed */
  451.     close(requestp->ret);
  452. xfree(requestp->path);
  453. break;
  454.     case _AIO_OP_CLOSE:
  455. if (cancelled && requestp->ret < 0)
  456.     /* The close() was cancelled and never got executed */
  457.     close(requestp->fd);
  458. break;
  459.     case _AIO_OP_UNLINK:
  460.     case _AIO_OP_OPENDIR:
  461. xfree(requestp->path);
  462. break;
  463.     case _AIO_OP_READ:
  464. if (!cancelled && requestp->ret > 0)
  465.     xmemcpy(requestp->bufferp, requestp->tmpbufp, requestp->ret);
  466.     case _AIO_OP_WRITE:
  467. xfree(requestp->tmpbufp);
  468. break;
  469.     default:
  470. break;
  471.     }
  472.     if (resultp != NULL && !cancelled) {
  473. resultp->aio_return = requestp->ret;
  474. resultp->aio_errno = requestp->err;
  475.     }
  476.     memPoolFree(aio_request_pool, requestp);
  477. } /* aio_cleanup_request */
  480. int
  481. aio_cancel(aio_result_t * resultp)
  482. {
  483.     aio_thread_t *threadp;
  484.     aio_request_t *requestp;
  486.     for (threadp = busy_threads_head; threadp != NULL; threadp = threadp->next)
  487. if (threadp->processed_req->resultp == resultp) {
  488.     threadp->processed_req->cancelled = 1;
  489.     threadp->processed_req->resultp = NULL;
  490.     return 0;
  491. }
  492.     for (requestp = request_queue_head; requestp != NULL; requestp = requestp->next)
  493. if (requestp->resultp == resultp) {
  494.     requestp->cancelled = 1;
  495.     requestp->resultp = NULL;
  496.     return 0;
  497. }
  498.     for (requestp = request_done_head; requestp != NULL; requestp = requestp->next)
  499. if (requestp->resultp == resultp) {
  500.     requestp->cancelled = 1;
  501.     requestp->resultp = NULL;
  502.     return 0;
  503. }
  504.     return 1;
  505. } /* aio_cancel */
  508. int
  509. aio_open(const char *path, int oflag, mode_t mode, aio_result_t * resultp)
  510. {
  511.     aio_request_t *requestp;
  512.     int len;
  514.     if (!aio_initialised)
  515. aio_init();
  516.     if ((requestp = memPoolAlloc(aio_request_pool)) == NULL) {
  517. errno = ENOMEM;
  518. return -1;
  519.     }
  520.     len = strlen(path) + 1;
  521.     if ((requestp->path = (char *) xmalloc(len)) == NULL) {
  522. memPoolFree(aio_request_pool, requestp);
  523. errno = ENOMEM;
  524. return -1;
  525.     }
  526.     strncpy(requestp->path, path, len);
  527.     requestp->oflag = oflag;
  528.     requestp->mode = mode;
  529.     requestp->resultp = resultp;
  530.     requestp->request_type = _AIO_OP_OPEN;
  531.     requestp->cancelled = 0;
  533.     aio_do_request(requestp);
  534.     return 0;
  535. }
  538. static void
  539. aio_do_open(aio_request_t * requestp)
  540. {
  541.     requestp->ret = open(requestp->path, requestp->oflag, requestp->mode);
  542.     requestp->err = errno;
  543. }
  546. int
  547. aio_read(int fd, char *bufp, int bufs, off_t offset, int whence, aio_result_t * resultp)
  548. {
  549.     aio_request_t *requestp;
  551.     if (!aio_initialised)
  552. aio_init();
  553.     if ((requestp = memPoolAlloc(aio_request_pool)) == NULL) {
  554. errno = ENOMEM;
  555. return -1;
  556.     }
  557.     requestp->fd = fd;
  558.     requestp->bufferp = bufp;
  559.     if ((requestp->tmpbufp = (char *) xmalloc(bufs)) == NULL) {
  560. memPoolFree(aio_request_pool, requestp);
  561. errno = ENOMEM;
  562. return -1;
  563.     }
  564.     requestp->buflen = bufs;
  565.     requestp->offset = offset;
  566.     requestp->whence = whence;
  567.     requestp->resultp = resultp;
  568.     requestp->request_type = _AIO_OP_READ;
  569.     requestp->cancelled = 0;
  571.     aio_do_request(requestp);
  572.     return 0;
  573. }
  576. static void
  577. aio_do_read(aio_request_t * requestp)
  578. {
  579.     lseek(requestp->fd, requestp->offset, requestp->whence);
  580.     requestp->ret = read(requestp->fd, requestp->tmpbufp, requestp->buflen);
  581.     requestp->err = errno;
  582. }
  585. int
  586. aio_write(int fd, char *bufp, int bufs, off_t offset, int whence, aio_result_t * resultp)
  587. {
  588.     aio_request_t *requestp;
  590.     if (!aio_initialised)
  591. aio_init();
  592.     if ((requestp = memPoolAlloc(aio_request_pool)) == NULL) {
  593. errno = ENOMEM;
  594. return -1;
  595.     }
  596.     requestp->fd = fd;
  597.     if ((requestp->tmpbufp = (char *) xmalloc(bufs)) == NULL) {
  598. memPoolFree(aio_request_pool, requestp);
  599. errno = ENOMEM;
  600. return -1;
  601.     }
  602.     xmemcpy(requestp->tmpbufp, bufp, bufs);
  603.     requestp->buflen = bufs;
  604.     requestp->offset = offset;
  605.     requestp->whence = whence;
  606.     requestp->resultp = resultp;
  607.     requestp->request_type = _AIO_OP_WRITE;
  608.     requestp->cancelled = 0;
  610.     aio_do_request(requestp);
  611.     return 0;
  612. }
  615. static void
  616. aio_do_write(aio_request_t * requestp)
  617. {
  618.     requestp->ret = write(requestp->fd, requestp->tmpbufp, requestp->buflen);
  619.     requestp->err = errno;
  620. }
  623. int
  624. aio_close(int fd, aio_result_t * resultp)
  625. {
  626.     aio_request_t *requestp;
  628.     if (!aio_initialised)
  629. aio_init();
  630.     if ((requestp = memPoolAlloc(aio_request_pool)) == NULL) {
  631. errno = ENOMEM;
  632. return -1;
  633.     }
  634.     requestp->fd = fd;
  635.     requestp->resultp = resultp;
  636.     requestp->request_type = _AIO_OP_CLOSE;
  637.     requestp->cancelled = 0;
  639.     aio_do_request(requestp);
  640.     return 0;
  641. }
  644. static void
  645. aio_do_close(aio_request_t * requestp)
  646. {
  647.     requestp->ret = close(requestp->fd);
  648.     requestp->err = errno;
  649. }
  652. int
  653. aio_stat(const char *path, struct stat *sb, aio_result_t * resultp)
  654. {
  655.     aio_request_t *requestp;
  656.     int len;
  658.     if (!aio_initialised)
  659. aio_init();
  660.     if ((requestp = memPoolAlloc(aio_request_pool)) == NULL) {
  661. errno = ENOMEM;
  662. return -1;
  663.     }
  664.     len = strlen(path) + 1;
  665.     if ((requestp->path = (char *) xmalloc(len)) == NULL) {
  666. memPoolFree(aio_request_pool, requestp);
  667. errno = ENOMEM;
  668. return -1;
  669.     }
  670.     strncpy(requestp->path, path, len);
  671.     requestp->statp = sb;
  672.     if ((requestp->tmpstatp = (struct stat *) xmalloc(sizeof(struct stat))) == NULL) {
  673. xfree(requestp->path);
  674. memPoolFree(aio_request_pool, requestp);
  675. errno = ENOMEM;
  676. return -1;
  677.     }
  678.     requestp->resultp = resultp;
  679.     requestp->request_type = _AIO_OP_STAT;
  680.     requestp->cancelled = 0;
  682.     aio_do_request(requestp);
  683.     return 0;
  684. }
  687. static void
  688. aio_do_stat(aio_request_t * requestp)
  689. {
  690.     requestp->ret = stat(requestp->path, requestp->tmpstatp);
  691.     requestp->err = errno;
  692. }
  695. int
  696. aio_unlink(const char *path, aio_result_t * resultp)
  697. {
  698.     aio_request_t *requestp;
  699.     int len;
  701.     if (!aio_initialised)
  702. aio_init();
  703.     if ((requestp = memPoolAlloc(aio_request_pool)) == NULL) {
  704. errno = ENOMEM;
  705. return -1;
  706.     }
  707.     len = strlen(path) + 1;
  708.     if ((requestp->path = (char *) xmalloc(len)) == NULL) {
  709. memPoolFree(aio_request_pool, requestp);
  710. errno = ENOMEM;
  711. return -1;
  712.     }
  713.     strncpy(requestp->path, path, len);
  714.     requestp->resultp = resultp;
  715.     requestp->request_type = _AIO_OP_UNLINK;
  716.     requestp->cancelled = 0;
  718.     aio_do_request(requestp);
  719.     return 0;
  720. }
  723. static void
  724. aio_do_unlink(aio_request_t * requestp)
  725. {
  726.     requestp->ret = unlink(requestp->path);
  727.     requestp->err = errno;
  728. }
  731. #if AIO_OPENDIR
  732. /* XXX aio_opendir NOT implemented? */
  734. int
  735. aio_opendir(const char *path, aio_result_t * resultp)
  736. {
  737.     aio_request_t *requestp;
  738.     int len;
  740.     if (!aio_initialised)
  741. aio_init();
  742.     if ((requestp = memPoolAlloc(aio_request_pool)) == NULL) {
  743. errno = ENOMEM;
  744. return -1;
  745.     }
  746.     return -1;
  747. }
  749. static void
  750. aio_do_opendir(aio_request_t * requestp)
  751. {
  752.     /* NOT IMPLEMENTED */
  753. }
  755. #endif
  758. void
  759. aio_poll_threads(void)
  760. {
  761.     aio_thread_t *prev;
  762.     aio_thread_t *threadp;
  763.     aio_request_t *requestp;
  765.     do { /* while found completed thread */
  766. prev = NULL;
  767. threadp = busy_threads_head;
  768. while (threadp) {
  769.     debug(43, 5) ("%d: %d -> %dn",
  770. threadp->thread,
  771. threadp->processed_req->request_type,
  772. threadp->status);
  773. #if AIO_PROPER_MUTEX
  774.     if (threadp->req == NULL)
  775. if (pthread_mutex_trylock(&threadp->mutex) == 0)
  776.     break;
  777. #else
  778.     if (threadp->req == NULL)
  779. break;
  780. #endif
  781.     prev = threadp;
  782.     threadp = threadp->next;
  783. }
  784. if (threadp == NULL)
  785.     break;
  787. if (prev == NULL)
  788.     busy_threads_head = busy_threads_head->next;
  789. else
  790.     prev->next = threadp->next;
  792. if (busy_threads_tail == threadp)
  793.     busy_threads_tail = prev;
  795. requestp = threadp->processed_req;
  796. threadp->processed_req = NULL;
  798. threadp->next = wait_threads;
  799. wait_threads = threadp;
  801. if (request_done_tail != NULL)
  802.     request_done_tail->next = requestp;
  803. else
  804.     request_done_head = requestp;
  805. request_done_tail = requestp;
  806.     } while (threadp);
  808.     aio_process_request_queue();
  809. } /* aio_poll_threads */
  811. aio_result_t *
  812. aio_poll_done()
  813. {
  814.     aio_request_t *requestp;
  815.     aio_result_t *resultp;
  816.     int cancelled;
  818.   AIO_REPOLL:
  819.     aio_poll_threads();
  820.     if (request_done_head == NULL) {
  821. return NULL;
  822.     }
  823.     requestp = request_done_head;
  824.     request_done_head = requestp->next;
  825.     if (!request_done_head)
  826. request_done_tail = NULL;
  828.     resultp = requestp->resultp;
  829.     cancelled = requestp->cancelled;
  830.     aio_debug(requestp);
  831.     debug(43, 5) ("DONE: %d -> %dn", requestp->ret, requestp->err);
  832.     aio_cleanup_request(requestp);
  833.     if (cancelled)
  834. goto AIO_REPOLL;
  835.     return resultp;
  836. } /* aio_poll_done */
  838. int
  839. aio_operations_pending(void)
  840. {
  841.     if (request_done_head)
  842. return 1;
  843.     if (busy_threads_head)
  844. return 1;
  845.     return 0;
  846. }
  848. static void
  849. aio_debug(aio_request_t * requestp)
  850. {
  851.     switch (requestp->request_type) {
  852.     case _AIO_OP_OPEN:
  853. debug(43, 5) ("OPEN of %s to FD %dn", requestp->path, requestp->ret);
  854. break;
  855.     case _AIO_OP_READ:
  856. debug(43, 5) ("READ on fd: %dn", requestp->fd);
  857. break;
  858.     case _AIO_OP_WRITE:
  859. debug(43, 5) ("WRITE on fd: %dn", requestp->fd);
  860. break;
  861.     case _AIO_OP_CLOSE:
  862. debug(43, 5) ("CLOSE of fd: %dn", requestp->fd);
  863. break;
  864.     case _AIO_OP_UNLINK:
  865. debug(43, 5) ("UNLINK of %sn", requestp->path);
  866. break;
  867.     default:
  868. break;
  869.     }
  870. }