scheduling threads using disk I/O

[jonas]

Hi,

I was playing around with pth trying to find out how to use it with a
server using disk I/O. The program below is based on the example found
in the manual page. If I comment out the call to pth_yield in the
handler, the program will serialize all requests. What is the "right"
way to make pth schedule other threads, like the main thread accepting
new connections?

(I am using pth 1.4.0 on FreeBSD 4.4RC)


regards,
        jonas

--8<-------8<------------------------------------------------------------------------------
#include <pth.h>

#include <sys/param.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/stat.h>

#include <netinet/in.h>
#include <netdb.h>
#include <stdlib.h>
#include <time.h>
#include <unistd.h>
#include <stdio.h>
#include <fcntl.h>

#define PORT 1234
#define BLOCK_SIZE 16*1024
#define BUF_SIZE 16*1024*1024

/* the socket connection handler thread */
static void *handler(void *_arg)
{
  int fd = (int)_arg;
  char path[32], buf[32];
  int i, rfd, wfd, rsize, val;
  char *rbuf;

  rbuf = (char*)malloc(BLOCK_SIZE);
  for (i = 0; i < 10; i++) {
    snprintf(path, 32, "h%d: %d\n", fd, i);
    pth_write(fd, path, strlen(path));
    path[strlen(path) - 1] = '\0';

    rfd = open("/dev/zero", O_RDONLY);
    wfd = open(path, O_WRONLY | O_CREAT, S_IRUSR | S_IWUSR);

    rsize = 0;
    while (rsize < BUF_SIZE) {
      val = pth_read(rfd, rbuf, BLOCK_SIZE);
      rsize += val;

      snprintf(buf, 32, ".");
      pth_write(fd, buf, strlen(buf));

      val = pth_write(wfd, rbuf, val);
 
      /*      pth_yield(NULL); */
    }

    close(rfd);
    close(wfd);
    
    unlink(path);

  }

  close(fd);
  return NULL;
}

/* the stderr time ticker thread */
static void *ticker(void *_arg)
{
  time_t now;
  char *ct;
  float load;

  for (;;) {
    pth_sleep(5);
    now = time(NULL);
    ct = ctime(&now);
    ct[strlen(ct)-1] = '\0';
    pth_ctrl(PTH_CTRL_GETAVLOAD, &load);
    printf("ticker: time: %s, average load: %.2f\n", ct, load);
  }
}

/* the main thread/procedure */
int main(int argc, char *argv[])
{
  pth_attr_t attr;
  struct sockaddr_in sar;
  struct protoent *pe;
  struct sockaddr_in peer_addr;
  int peer_len;
  int sa, sw;
  int opt;

  pth_init();
  signal(SIGPIPE, SIG_IGN);

  attr = pth_attr_new();
  pth_attr_set(attr, PTH_ATTR_NAME, "ticker");
  pth_attr_set(attr, PTH_ATTR_STACK_SIZE, 64*1024);
  pth_attr_set(attr, PTH_ATTR_JOINABLE, FALSE);
  pth_spawn(attr, ticker, NULL);

  pe = getprotobyname("tcp");
  sa = socket(AF_INET, SOCK_STREAM, pe->p_proto);
  if (setsockopt(sa, SOL_SOCKET,  SO_REUSEPORT, &opt, sizeof(opt)) ==
-1) {
    perror("setsockopt");
    return -1;
  }
    
  sar.sin_family = AF_INET;
  sar.sin_addr.s_addr = INADDR_ANY;
  sar.sin_port = htons(PORT);
  bind(sa, (struct sockaddr *)&sar, sizeof(struct sockaddr_in));
  listen(sa, 10);

  pth_attr_set(attr, PTH_ATTR_NAME, "handler");
  for (;;) {
    peer_len = sizeof(peer_addr);
    sw = pth_accept(sa, (struct sockaddr *)&peer_addr, &peer_len);
    pth_spawn(attr, handler, (void *)sw);
  }
}
______________________________________________________________________
GNU Portable Threads (Pth)            http://www.gnu.org/software/pth/
User Support Mailing List                            [EMAIL PROTECTED]
Automated List Manager (Majordomo)           [EMAIL PROTECTED]