threads/threadio.c

   1 /*
   2  * threadCalls.c
   3  * Support for threaded ops which may block (read, write, connect, etc.).
   4  *
   5  * Copyright (c) 1996 T. J. Wilkinson & Associates, London, UK.
   6  *
   7  * See the file "license.terms" for information on usage and redistribution
   8  * of this file, and for a DISCLAIMER OF ALL WARRANTIES.
   9  *
  10  * Written by Tim Wilkinson <tim@tjwassoc.demon.co.uk>, 1996.
  11  */
  12
  13 #define DBG(s)
  14
  15 #include "sysdep/defines.h"
  16
  17 #include <sys/types.h>
  18 #include <sys/time.h>
  19 #include <sys/socket.h>
  20 #include <fcntl.h>
  21 #include <errno.h>
  22 #include <assert.h>
  23 #include <unistd.h>
  24
  25 #include "thread.h"
  26
  27 /*
  28  * We only need this stuff is we are using the internal thread system.
  29  */
  30 #if defined(USE_INTERNAL_THREADS)
  31
  32 #define TH_READ         0
  33 #define TH_WRITE        1
  34 #define TH_ACCEPT       TH_READ
  35 #define TH_CONNECT      TH_WRITE
  36
  37 static int maxFd;
  38 static fd_set readsPending;
  39 static fd_set writesPending;
  40 static thread* readQ[FD_SETSIZE];
  41 static thread* writeQ[FD_SETSIZE];
  42 static struct timeval tm = { 0, 0 };
  43
  44 void blockOnFile(int, int);
  45 void waitOnEvents(void);
  46
  47 extern thread* currentThread;
  48
  49 /* These are undefined because we do not yet support async I/O */
  50 #undef  F_SETOWN
  51 #undef  FIOSETOWN
  52 #undef  O_ASYNC
  53 #undef  FIOASYNC
  54
  55 /*
  56  * Create a threaded file descriptor.
  57  */
  58 int
  59 threadedFileDescriptor(int fd)
  60 {
  61 #if !defined(BLOCKING_CALLS)
  62     int r;
  63     int on = 1;
  64     int pid;
  65
  66     /* Make non-blocking */
  67 #if defined(HAVE_FCNTL) && defined(O_NONBLOCK)
  68     r = fcntl(fd, F_GETFL, 0);
  69     r = fcntl(fd, F_SETFL, r|O_NONBLOCK);
  70 #elif defined(HAVE_IOCTL) && defined(FIONBIO)
  71     r = ioctl(fd, FIONBIO, &on);
  72 #else
  73     r = 0;
  74 #endif
  75     if (r < 0)
  76     {
  77         return (r);
  78     }
  79
  80     /* Allow socket to signal this process when new data is available */
  81     pid = getpid();
  82 #if defined(HAVE_FCNTL) && defined(F_SETOWN)
  83     r = fcntl(fd, F_SETOWN, pid);
  84 #elif defined(HAVE_IOCTL) && defined(FIOSETOWN)
  85     r = ioctl(fd, FIOSETOWN, &pid);
  86 #else
  87     r = 0;
  88 #endif
  89     if (r < 0)
  90     {
  91         return (r);
  92     }
  93
  94 #if defined(HAVE_FCNTL) && defined(O_ASYNC)
  95     r = fcntl(fd, F_GETFL, 0);
  96     r = fcntl(fd, F_SETFL, r|O_ASYNC);
  97 #elif defined(HAVE_IOCTL) && defined(FIOASYNC)
  98     r = ioctl(fd, FIOASYNC, &on);
  99 #else
 100     r = 0;
 101 #endif
 102     if (r < 0)
 103     {
 104         return (r);
 105     }
 106 #endif
 107     return (fd);
 108 }
 109
 110 void clear_thread_flags(void)
 111 {
 112 #if !defined(BLOCKING_CALLS)
 113         int fl, fd;
 114
 115 #if defined(HAVE_FCNTL) && defined(O_NONBLOCK)
 116         fd = fileno(stdin);
 117     fl = fcntl(fd, F_GETFL, 0);
 118     fl = fcntl(fd, F_SETFL, fl & (~O_NONBLOCK));
 119
 120         fd = fileno(stdout);
 121     fl = fcntl(fd, F_GETFL, 0);
 122     fl = fcntl(fd, F_SETFL, fl & (~O_NONBLOCK));
 123
 124         fd = fileno(stderr);
 125     fl = fcntl(fd, F_GETFL, 0);
 126     fl = fcntl(fd, F_SETFL, fl & (~O_NONBLOCK));
 127 #elif defined(HAVE_IOCTL) && defined(FIONBIO)
 128         fl = 0;
 129         fd = fileno(stdin);
 130     (void) ioctl(fd, FIONBIO, &fl);
 131
 132         fd = fileno(stdout);
 133     (void) ioctl(fd, FIONBIO, &fl);
 134
 135         fd = fileno(stderr);
 136     (void) ioctl(fd, FIONBIO, &fl);
 137 #endif
 138
 139 #endif
 140 fflush (stdout);
 141 fflush (stderr);
 142 }
 143
 144
 145 /*
 146  * Threaded create socket.
 147  */
 148 int
 149 threadedSocket(int af, int type, int proto)
 150 {
 151     int fd;
 152     int r;
 153     int on = 1;
 154     int pid;
 155
 156     fd = socket(af, type, proto);
 157     return (threadedFileDescriptor(fd));
 158 }
 159
 160 /*
 161  * Threaded file open.
 162  */
 163 int
 164 threadedOpen(char* path, int flags, int mode)
 165 {
 166     int fd;
 167     int r;
 168     int on = 1;
 169     int pid;
 170
 171     fd = open(path, flags, mode);
 172     return (threadedFileDescriptor(fd));
 173 }
 174
 175 /*
 176  * Threaded socket connect.
 177  */
 178 int
 179 threadedConnect(int fd, struct sockaddr* addr, int len)
 180 {
 181     int r;
 182
 183     r = connect(fd, addr, len);
 184 #if !defined(BLOCKING_CALLS)
 185     if ((r < 0)
 186         && (errno == EINPROGRESS || errno == EALREADY
 187             || errno == EWOULDBLOCK)) {
 188         blockOnFile(fd, TH_CONNECT);
 189         r = 0; /* Assume it's okay when we get released */
 190     }
 191 #endif
 192
 193     return (r);
 194 }
 195
 196 /*
 197  * Threaded socket accept.
 198  */
 199 int
 200 threadedAccept(int fd, struct sockaddr* addr, int* len)
 201 {
 202     int r;
 203     int on = 1;
 204
 205     for (;;)
 206     {
 207 #if defined(BLOCKING_CALLS)
 208         blockOnFile(fd, TH_ACCEPT);
 209 #endif
 210         r = accept(fd, addr, len);
 211         if (r >= 0
 212             || !(errno == EINPROGRESS || errno == EALREADY
 213                  || errno == EWOULDBLOCK))
 214         {
 215             break;
 216         }
 217 #if !defined(BLOCKING_CALLS)
 218         blockOnFile(fd, TH_ACCEPT);
 219 #endif
 220     }
 221     return (threadedFileDescriptor(r));
 222 }
 223
 224 /*
 225  * Read but only if we can.
 226  */
 227 int
 228 threadedRead(int fd, char* buf, int len)
 229 {
 230     int r;
 231
 232     DBG(   printf("threadedRead\n");          )
 233
 234 #if defined(BLOCKING_CALLS)
 235     blockOnFile(fd, TH_READ);
 236 #endif
 237     for (;;)
 238     {
 239         r = read(fd, buf, len);
 240         if (r < 0
 241             && (errno == EAGAIN || errno == EWOULDBLOCK
 242                 || errno == EINTR))
 243         {
 244             blockOnFile(fd, TH_READ);
 245             continue;
 246         }
 247         return (r);
 248     }
 249 }
 250
 251 /*
 252  * Write but only if we can.
 253  */
 254 int
 255 threadedWrite(int fd, char* buf, int len)
 256 {
 257     int r;
 258     char* ptr;
 259
 260     ptr = buf;
 261     r = 1;
 262
 263     DBG(    printf("threadedWrite %dbytes\n",len);      )
 264
 265     while (len > 0 && r > 0)
 266     {
 267 #if defined(BLOCKING_CALLS)
 268         blockOnFile(fd, TH_WRITE);
 269 #endif
 270         r = write(fd, ptr, len);
 271         if (r < 0
 272             && (errno == EAGAIN || errno == EWOULDBLOCK
 273                 || errno == EINTR))
 274         {
 275 #if !defined(BLOCKING_CALLS)
 276             blockOnFile(fd, TH_WRITE);
 277 #endif
 278             r = 1;
 279         }
 280         else
 281         {
 282             ptr += r;
 283             len -= r;
 284         }
 285     }
 286     return (ptr - buf);
 287 }
 288
 289 /*
 290  * An attempt to access a file would block, so suspend the thread until
 291  * it will happen.
 292  */
 293 void
 294 blockOnFile(int fd, int op)
 295 {
 296 DBG(    printf("blockOnFile()\n");                                      )
 297
 298     intsDisable();
 299
 300     if (fd > maxFd)
 301     {
 302         maxFd = fd;
 303     }
 304     if (op == TH_READ)
 305     {
 306         FD_SET(fd, &readsPending);
 307         suspendOnQThread(currentThread, &readQ[fd]);
 308         FD_CLR(fd, &readsPending);
 309     }
 310     else
 311     {
 312         FD_SET(fd, &writesPending);
 313         suspendOnQThread(currentThread, &writeQ[fd]);
 314         FD_CLR(fd, &writesPending);
 315     }
 316
 317     intsRestore();
 318 }
 319
 320 /*
 321  * Check if some file descriptor or other event to become ready.
 322  * Block if required (but make sure we can still take timer interrupts).
 323  */
 324 void
 325 checkEvents(bool block)
 326 {
 327     int r;
 328     fd_set rd;
 329     fd_set wr;
 330     thread* tid;
 331     thread* ntid;
 332     int i;
 333     int b;
 334
 335 DBG(    printf("checkEvents block:%d\n", block);                        )
 336
 337 #if defined(FD_COPY)
 338     FD_COPY(&readsPending, &rd);
 339     FD_COPY(&writesPending, &wr);
 340 #else
 341     memcpy(&rd, &readsPending, sizeof(rd));
 342     memcpy(&wr, &writesPending, sizeof(wr));
 343 #endif
 344
 345     /*
 346      * If select() is called with indefinite wait, we have to make sure
 347      * we can get interrupted by timer events.
 348      */
 349     if (block == true)
 350     {
 351         b = blockInts;
 352         blockInts = 0;
 353         r = select(maxFd+1, &rd, &wr, 0, 0);
 354         blockInts = b;
 355     }
 356     else
 357     {
 358         r = select(maxFd+1, &rd, &wr, 0, &tm);
 359     }
 360
 361     /* We must be holding off interrupts before we start playing with
 362      * the read and write queues.  This should be already done but a
 363      * quick check never hurt anyone.
 364      */
 365     assert(blockInts > 0);
 366
 367 DBG(    printf("Select returns %d\n", r);                               )
 368
 369     for (i = 0; r > 0 && i <= maxFd; i++)
 370     {
 371         if (readQ[i] != 0 && FD_ISSET(i, &rd))
 372         {
 373             for (tid = readQ[i]; tid != 0; tid = ntid)
 374             {
 375                 ntid = tid->next;
 376                 iresumeThread(tid);
 377             }
 378             readQ[i] = 0;
 379             r--;
 380         }
 381         if (writeQ[i] != 0 && FD_ISSET(i, &wr))
 382         {
 383             for (tid = writeQ[i]; tid != 0; tid = ntid)
 384             {
 385                 ntid = tid->next;
 386                 iresumeThread(tid);
 387             }
 388             writeQ[i] = 0;
 389             r--;
 390         }
 391     }
 392 }
 393 #endif