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 #include "config.h"
  14
  15 #include "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
  43 void blockOnFile(int, int);
  44 void waitOnEvents(void);
  45
  46 extern thread* currentThread;
  47
  48 /* These are undefined because we do not yet support async I/O */
  49 #undef  F_SETOWN
  50 #undef  FIOSETOWN
  51 #undef  O_ASYNC
  52 #undef  FIOASYNC
  53
  54 /*
  55  * Create a threaded file descriptor.
  56  */
  57 int
  58 threadedFileDescriptor(int fd)
  59 {
  60 #if !defined(BLOCKING_CALLS)
  61     int r;
  62 #if defined(HAVE_IOCTL) && defined(FIOASYNC)
  63     int on = 1;
  64 #endif
  65     int pid;
  66
  67     /* Make non-blocking */
  68 #if defined(HAVE_FCNTL) && defined(O_NONBLOCK)
  69     r = fcntl(fd, F_GETFL, 0);
  70     r = fcntl(fd, F_SETFL, r|O_NONBLOCK);
  71 #elif defined(HAVE_IOCTL) && defined(FIONBIO)
  72     r = ioctl(fd, FIONBIO, &on);
  73 #else
  74     r = 0;
  75 #endif
  76     if (r < 0)
  77     {
  78         return (r);
  79     }
  80
  81     /* Allow socket to signal this process when new data is available */
  82     pid = getpid();
  83 #if defined(HAVE_FCNTL) && defined(F_SETOWN)
  84     r = fcntl(fd, F_SETOWN, pid);
  85 #elif defined(HAVE_IOCTL) && defined(FIOSETOWN)
  86     r = ioctl(fd, FIOSETOWN, &pid);
  87 #else
  88     r = 0;
  89 #endif
  90     if (r < 0)
  91     {
  92         return (r);
  93     }
  94
  95 #if defined(HAVE_FCNTL) && defined(O_ASYNC)
  96     r = fcntl(fd, F_GETFL, 0);
  97     r = fcntl(fd, F_SETFL, r|O_ASYNC);
  98 #elif defined(HAVE_IOCTL) && defined(FIOASYNC)
  99     r = ioctl(fd, FIOASYNC, &on);
 100 #else
 101     r = 0;
 102 #endif
 103     if (r < 0)
 104     {
 105         return (r);
 106     }
 107 #endif
 108     return (fd);
 109 }
 110
 111 void clear_thread_flags(void)
 112 {
 113 #if !defined(BLOCKING_CALLS)
 114         int fl, fd;
 115
 116 #if defined(HAVE_FCNTL) && defined(O_NONBLOCK)
 117         fd = fileno(stdin);
 118     fl = fcntl(fd, F_GETFL, 0);
 119     fl = fcntl(fd, F_SETFL, fl & (~O_NONBLOCK));
 120
 121         fd = fileno(stdout);
 122     fl = fcntl(fd, F_GETFL, 0);
 123     fl = fcntl(fd, F_SETFL, fl & (~O_NONBLOCK));
 124
 125         fd = fileno(stderr);
 126     fl = fcntl(fd, F_GETFL, 0);
 127     fl = fcntl(fd, F_SETFL, fl & (~O_NONBLOCK));
 128 #elif defined(HAVE_IOCTL) && defined(FIONBIO)
 129         fl = 0;
 130         fd = fileno(stdin);
 131     (void) ioctl(fd, FIONBIO, &fl);
 132
 133         fd = fileno(stdout);
 134     (void) ioctl(fd, FIONBIO, &fl);
 135
 136         fd = fileno(stderr);
 137     (void) ioctl(fd, FIONBIO, &fl);
 138 #endif
 139
 140 #endif
 141 fflush (stdout);
 142 fflush (stderr);
 143 }
 144
 145
 146 /*
 147  * Threaded create socket.
 148  */
 149 int
 150 threadedSocket(int af, int type, int proto)
 151 {
 152     int fd;
 153
 154     fd = socket(af, type, proto);
 155     return (threadedFileDescriptor(fd));
 156 }
 157
 158 /*
 159  * Threaded file open.
 160  */
 161 int
 162 threadedOpen(char* path, int flags, int mode)
 163 {
 164     int fd;
 165
 166     fd = open(path, flags, mode);
 167     return (threadedFileDescriptor(fd));
 168 }
 169
 170 /*
 171  * Threaded socket connect.
 172  */
 173 int
 174 threadedConnect(int fd, struct sockaddr* addr, int len)
 175 {
 176     int r;
 177
 178     r = connect(fd, addr, len);
 179 #if !defined(BLOCKING_CALLS)
 180     if ((r < 0)
 181         && (errno == EINPROGRESS || errno == EALREADY
 182             || errno == EWOULDBLOCK)) {
 183         blockOnFile(fd, TH_CONNECT);
 184         r = 0; /* Assume it's okay when we get released */
 185     }
 186 #endif
 187
 188     return (r);
 189 }
 190
 191 /*
 192  * Threaded socket accept.
 193  */
 194 int
 195 threadedAccept(int fd, struct sockaddr* addr, int* len)
 196 {
 197     int r;
 198
 199     for (;;)
 200     {
 201 #if defined(BLOCKING_CALLS)
 202         blockOnFile(fd, TH_ACCEPT);
 203 #endif
 204         r = accept(fd, addr, (int*)len);
 205         if (r >= 0
 206             || !(errno == EINPROGRESS || errno == EALREADY
 207                  || errno == EWOULDBLOCK))
 208         {
 209             break;
 210         }
 211 #if !defined(BLOCKING_CALLS)
 212         blockOnFile(fd, TH_ACCEPT);
 213 #endif
 214     }
 215     return (threadedFileDescriptor(r));
 216 }
 217
 218 /*
 219  * Read but only if we can.
 220  */
 221 int
 222 threadedRead(int fd, char* buf, int len)
 223 {
 224     int r;
 225
 226     DBG(   printf("threadedRead\n");          )
 227
 228 #if defined(BLOCKING_CALLS)
 229     blockOnFile(fd, TH_READ);
 230 #endif
 231     for (;;)
 232     {
 233         r = read(fd, buf, len);
 234         if (r < 0
 235             && (errno == EAGAIN || errno == EWOULDBLOCK
 236                 || errno == EINTR))
 237         {
 238             blockOnFile(fd, TH_READ);
 239             continue;
 240         }
 241         return (r);
 242     }
 243 }
 244
 245 /*
 246  * Write but only if we can.
 247  */
 248 int
 249 threadedWrite(int fd, char* buf, int len)
 250 {
 251     int r;
 252     char* ptr;
 253
 254     ptr = buf;
 255     r = 1;
 256
 257     DBG(    printf("threadedWrite %dbytes\n",len);      )
 258
 259     while (len > 0 && r > 0)
 260     {
 261 #if defined(BLOCKING_CALLS)
 262         blockOnFile(fd, TH_WRITE);
 263 #endif
 264         r = write(fd, ptr, len);
 265         if (r < 0
 266             && (errno == EAGAIN || errno == EWOULDBLOCK
 267                 || errno == EINTR))
 268         {
 269 #if !defined(BLOCKING_CALLS)
 270             blockOnFile(fd, TH_WRITE);
 271 #endif
 272             r = 1;
 273         }
 274         else
 275         {
 276             ptr += r;
 277             len -= r;
 278         }
 279     }
 280     return (ptr - buf);
 281 }
 282
 283 /*
 284  * Receive, but only if we can.
 285  */
 286 int
 287 threadedRecvfrom (int fd, void *buf, size_t len, int flags, struct sockaddr *addr, int *addrlen)
 288 {
 289     int r;
 290
 291     DBG(   printf("threadedRecvfrom\n");          )
 292
 293 #if defined(BLOCKING_CALLS)
 294     blockOnFile(fd, TH_READ);
 295 #endif
 296     for (;;)
 297     {
 298         r = recvfrom(fd, buf, len, flags, addr, addrlen);
 299         if (r < 0
 300             && (errno == EAGAIN || errno == EWOULDBLOCK
 301                 || errno == EINTR))
 302         {
 303             blockOnFile(fd, TH_READ);
 304             continue;
 305         }
 306         return (r);
 307     }
 308 }
 309
 310 /*
 311  * Send, but only if we can.
 312  */
 313 int
 314 threadedSendto (int fd, void *buf, size_t len, int flags, struct sockaddr *addr, int addrlen)
 315 {
 316     int r;
 317
 318     DBG(   printf("threadedSendto\n");          )
 319
 320 #if defined(BLOCKING_CALLS)
 321     blockOnFile(fd, TH_WRITE);
 322 #endif
 323     for (;;)
 324     {
 325         r = sendto(fd, buf, len, flags, addr, addrlen);
 326         if (r < 0
 327             && (errno == EAGAIN || errno == EWOULDBLOCK
 328                 || errno == EINTR))
 329         {
 330             blockOnFile(fd, TH_WRITE);
 331             continue;
 332         }
 333         return (r);
 334     }
 335 }
 336
 337 /*
 338  * An attempt to access a file would block, so suspend the thread until
 339  * it will happen.
 340  */
 341 void
 342 blockOnFile(int fd, int op)
 343 {
 344 DBG(    printf("blockOnFile()\n");                                      )
 345
 346     intsDisable();
 347
 348     if (fd > maxFd)
 349     {
 350         maxFd = fd;
 351     }
 352
 353     if (op == TH_READ)
 354     {
 355         FD_SET(fd, &readsPending);
 356         suspendOnQThread(currentThread, &readQ[fd]);
 357         FD_CLR(fd, &readsPending);
 358     }
 359     else
 360     {
 361         FD_SET(fd, &writesPending);
 362         suspendOnQThread(currentThread, &writeQ[fd]);
 363         FD_CLR(fd, &writesPending);
 364     }
 365
 366     intsRestore();
 367 }
 368
 369 /*
 370  * Check if some file descriptor or other event to become ready.
 371  * Block if required (but make sure we can still take timer interrupts).
 372  */
 373 void
 374 checkEvents(bool block)
 375 {
 376     int r;
 377     fd_set rd;
 378     fd_set wr;
 379     thread* tid;
 380     thread* ntid;
 381     int i;
 382     s8 time = -1;
 383     struct timeval tv;
 384     struct timeval *timeout;
 385
 386     assert(blockInts > 0);
 387
 388     DBG( printf("checkEvents block:%d\n", block); )
 389
 390     if (sleepThreads != 0)
 391     {
 392         time = currentTime();
 393         while (sleepThreads != 0 && time >= CONTEXT(sleepThreads).time)
 394         {
 395             tid = sleepThreads;
 396             sleepThreads = sleepThreads->next;
 397             tid->next = 0;
 398
 399             iresumeThread(tid);
 400         }
 401     }
 402
 403     if (block)
 404     {
 405         if (sleepThreads != 0)
 406         {
 407             s8 wait_time = CONTEXT(sleepThreads).time - time;
 408
 409             tv.tv_sec = wait_time / 1000;
 410             tv.tv_usec = (wait_time % 1000) * 1000;
 411             timeout = &tv;
 412         }
 413         else
 414             timeout = 0;
 415     }
 416     else
 417     {
 418         tv.tv_sec = 0;
 419         tv.tv_usec = 0;
 420         timeout = &tv;
 421     }
 422
 423 #if defined(FD_COPY)
 424     FD_COPY(&readsPending, &rd);
 425     FD_COPY(&writesPending, &wr);
 426 #else
 427     memcpy(&rd, &readsPending, sizeof(rd));
 428     memcpy(&wr, &writesPending, sizeof(wr));
 429 #endif
 430
 431     r = select(maxFd+1, &rd, &wr, 0, timeout);
 432
 433     /* We must be holding off interrupts before we start playing with
 434      * the read and write queues.  This should be already done but a
 435      * quick check never hurt anyone.
 436      */
 437     assert(blockInts > 0);
 438
 439     DBG( printf("Select returns %d\n", r); )
 440
 441     /* Some threads may have finished sleeping.
 442      */
 443     if (block && sleepThreads != 0)
 444     {
 445         time = currentTime();
 446         while (sleepThreads != 0 && time >= CONTEXT(sleepThreads).time)
 447         {
 448             tid = sleepThreads;
 449             sleepThreads = sleepThreads->next;
 450             tid->next = 0;
 451
 452             iresumeThread(tid);
 453         }
 454     }
 455
 456     for (i = 0; r > 0 && i <= maxFd; i++)
 457     {
 458         if (readQ[i] != 0 && FD_ISSET(i, &rd))
 459         {
 460             for (tid = readQ[i]; tid != 0; tid = ntid)
 461             {
 462                 ntid = tid->next;
 463                 iresumeThread(tid);
 464             }
 465             readQ[i] = 0;
 466             r--;
 467         }
 468         if (writeQ[i] != 0 && FD_ISSET(i, &wr))
 469         {
 470             for (tid = writeQ[i]; tid != 0; tid = ntid)
 471             {
 472                 ntid = tid->next;
 473                 iresumeThread(tid);
 474             }
 475             writeQ[i] = 0;
 476             r--;
 477         }
 478     }
 479 }
 480 #endif