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