#include <u.h>#include <libc.h>#include <thread.h>#include <mpipe.h>#include <filt.h> Mpipe *p; void *emalloc(ulong size){ void *a; a = malloc(size); if(a == 0) sysfatal(0, "no memory"); memset(a, 0, size); return a;} static voidmpout(void *v){ Mpipe *p; Ioproc *io; int n, h, len; char *buf, *b; p = v; buf = emalloc(8192); n = 0; io = ioproc();start: while((n = ioread(io, p->infd, buf+n, 8192)) > 0){ b = buf; while(n > 0){ h = p->sep(p, b, n, &len); if(len == -1){ memmove(buf, b, n); goto start; } iowrite(io,p->fds[h], b, len); n -= len; b+=len; } } free(buf);}enum { Maxfd = 256,}; Channel *c[Maxfd]; typedef struct Buffer Buffer;struct Buffer { int n; char buf[1];}; voidmpreader(void *v){ int fd, n, len; Ioproc *io; char *buf, *bufbeg; Buffer *b; SET(n); fd = (int)v; io = ioproc(); bufbeg = buf = malloc(8192); while((n = ioread(io, fd, buf+n, 8192)) > 0){ while(p->sep(p, buf, n, &len) != -1){ b = malloc(sizeof(Buffer)+len); b->n = len; memmove(b->buf, buf, len); sendp(c[fd], b); buf += len; n -= len; } buf = bufbeg; }} voidmpin(void *v){ int i, n; Mpipe *p; Ioproc *io; Buffer *b; Alt *a; p = v; io = ioproc(); a = emalloc(p->npipe * sizeof(Alt*)); for(i = 0; i < p->npipe; i++){ c[p->fds[i]] = chancreate(sizeof(Buffer*), 0); a[i].c = c[p->fds[i]]; a[i].op = CHANRCV; threadcreate(mpreader, (void*)p->fds[i], 8192); } a[i+1].c = nil; a[i+1].op = CHANEND; for(;;){ i = alt(a); b = recvp(a[i].c); n = iowrite(io, 1, b->buf, b->n); /* what about partial writes? */ if(n < 0) break; free(b); } free(a);} voidthreadmain(int argc, char **argv){ int i; int *fds; ARGBEGIN{ }ARGEND; fds = malloc(argc*sizeof(int)); for(i = 0; i < argc; i++) fds[i] = atoi(argv[i]); p = emalloc(sizeof(Mpipe)); p->fds = fds; p->npipe = argc; p->infd = 0; p->sep = filtmain; threadcreate(mpout, p, 8192); threadcreate(mpin, p, 8192);}
#include <u.h>
#include <libc.h>
#include <thread.h>
#include <mpipe.h>
#include <filt.h>
Mpipe *p;
void *
emalloc(ulong size)
{
void *a;
a = malloc(size);
if(a == 0)
sysfatal(0, "no memory");
memset(a, 0, size);
return a;
}
static void
mpout(void *v)
{
Mpipe *p;
Ioproc *io;
int n, h, len;
char *buf, *b;
p = v;
buf = emalloc(8192);
n = 0;
io = ioproc();
start:
while((n = ioread(io, p->infd, buf+n, 8192)) > 0){
b = buf;
while(n > 0){
h = p->sep(p, b, n, &len);
if(len == -1){
memmove(buf, b, n);
goto start;
}
iowrite(io,p->fds[h], b, len);
n -= len;
b+=len;
}
}
free(buf);
}
enum {
Maxfd = 256,
};
Channel *c[Maxfd];
typedef struct Buffer Buffer;
struct Buffer {
int n;
char buf[1];
};
void
mpreader(void *v)
{
int fd, n, len;
Ioproc *io;
char *buf, *bufbeg;
Buffer *b;
SET(n);
fd = (int)v;
io = ioproc();
bufbeg = buf = malloc(8192);
while((n = ioread(io, fd, buf+n, 8192)) > 0){
while(p->sep(p, buf, n, &len) != -1){
b = malloc(sizeof(Buffer)+len);
b->n = len;
memmove(b->buf, buf, len);
sendp(c[fd], b);
buf += len;
n -= len;
}
buf = bufbeg;
}
}
void
mpin(void *v)
{
int i, n;
Mpipe *p;
Ioproc *io;
Buffer *b;
Alt *a;
p = v;
io = ioproc();
a = emalloc(p->npipe * sizeof(Alt*));
for(i = 0; i < p->npipe; i++){
c[p->fds[i]] = chancreate(sizeof(Buffer*), 0);
a[i].c = c[p->fds[i]];
a[i].op = CHANRCV;
threadcreate(mpreader, (void*)p->fds[i], 8192);
}
a[i+1].c = nil;
a[i+1].op = CHANEND;
for(;;){
i = alt(a);
b = recvp(a[i].c);
n = iowrite(io, 1, b->buf, b->n);
/* what about partial writes? */
if(n < 0)
break;
free(b);
}
free(a);
}
void
threadmain(int argc, char **argv)
{
int i;
int *fds;
ARGBEGIN{
}ARGEND;
fds = malloc(argc*sizeof(int));
for(i = 0; i < argc; i++)
fds[i] = atoi(argv[i]);
p = emalloc(sizeof(Mpipe));
p->fds = fds;
p->npipe = argc;
p->infd = 0;
p->sep = filtmain;
threadcreate(mpout, p, 8192);
threadcreate(mpin, p, 8192);
}