sys/src/9/port/devpipe.c

   1 #include        "u.h"
   2 #include        "../port/lib.h"
   3 #include        "mem.h"
   4 #include        "dat.h"
   5 #include        "fns.h"
   6 #include        "../port/error.h"
   7
   8 #include        "netif.h"
   9
  10 typedef struct Pipe     Pipe;
  11 struct Pipe
  12 {
  13         QLock;
  14         Pipe    *next;
  15         int     ref;
  16         ulong   path;
  17         long    perm;
  18         Queue   *q[2];
  19         int     qref[2];
  20 };
  21
  22 struct
  23 {
  24         Lock;
  25         ulong   path;
  26 } pipealloc;
  27
  28 enum
  29 {
  30         Qdir,
  31         Qdata0,
  32         Qdata1,
  33 };
  34
  35 Dirtab pipedir[] =
  36 {
  37         ".",            {Qdir,0,QTDIR}, 0,              DMDIR|0500,
  38         "data",         {Qdata0},       0,              0600,
  39         "data1",        {Qdata1},       0,              0600,
  40 };
  41 #define NPIPEDIR 3
  42
  43 static void
  44 pipeinit(void)
  45 {
  46         if(conf.pipeqsize == 0){
  47                 if(conf.nmach > 1)
  48                         conf.pipeqsize = 256*1024;
  49                 else
  50                         conf.pipeqsize = 32*1024;
  51         }
  52 }
  53
  54 /*
  55  *  create a pipe, no streams are created until an open
  56  */
  57 static Chan*
  58 pipeattach(char *spec)
  59 {
  60         Pipe *p;
  61         Chan *c;
  62
  63         c = devattach('|', spec);
  64         if(waserror()){
  65                 chanfree(c);
  66                 nexterror();
  67         }
  68         p = malloc(sizeof(Pipe));
  69         if(p == 0)
  70                 exhausted("memory");
  71         p->ref = 1;
  72
  73         p->q[0] = qopen(conf.pipeqsize, 0, 0, 0);
  74         if(p->q[0] == 0){
  75                 free(p);
  76                 exhausted("memory");
  77         }
  78         p->q[1] = qopen(conf.pipeqsize, 0, 0, 0);
  79         if(p->q[1] == 0){
  80                 qfree(p->q[0]);
  81                 free(p);
  82                 exhausted("memory");
  83         }
  84         poperror();
  85
  86         lock(&pipealloc);
  87         p->path = ++pipealloc.path;
  88         unlock(&pipealloc);
  89         p->perm = pipedir[Qdata0].perm;
  90
  91         mkqid(&c->qid, NETQID(2*p->path, Qdir), 0, QTDIR);
  92         c->aux = p;
  93         c->dev = 0;
  94         return c;
  95 }
  96
  97 static int
  98 pipegen(Chan *c, char*, Dirtab *tab, int ntab, int i, Dir *dp)
  99 {
 100         Qid q;
 101         int len;
 102         Pipe *p;
 103
 104         if(i == DEVDOTDOT){
 105                 devdir(c, c->qid, "#|", 0, eve, 0555, dp);
 106                 return 1;
 107         }
 108         i++;    /* skip . */
 109         if(tab==0 || i>=ntab)
 110                 return -1;
 111
 112         tab += i;
 113         p = c->aux;
 114         switch((ulong)tab->qid.path){
 115         case Qdata0:
 116                 len = qlen(p->q[0]);
 117                 break;
 118         case Qdata1:
 119                 len = qlen(p->q[1]);
 120                 break;
 121         default:
 122                 len = tab->length;
 123                 break;
 124         }
 125         mkqid(&q, NETQID(NETID(c->qid.path), tab->qid.path), 0, QTFILE);
 126         devdir(c, q, tab->name, len, eve, p->perm, dp);
 127         return 1;
 128 }
 129
 130
 131 static Walkqid*
 132 pipewalk(Chan *c, Chan *nc, char **name, int nname)
 133 {
 134         Walkqid *wq;
 135         Pipe *p;
 136
 137         wq = devwalk(c, nc, name, nname, pipedir, NPIPEDIR, pipegen);
 138         if(wq != nil && wq->clone != nil && wq->clone != c){
 139                 p = c->aux;
 140                 qlock(p);
 141                 p->ref++;
 142                 if(c->flag & COPEN){
 143                         print("channel open in pipewalk\n");
 144                         switch(NETTYPE(c->qid.path)){
 145                         case Qdata0:
 146                                 p->qref[0]++;
 147                                 break;
 148                         case Qdata1:
 149                                 p->qref[1]++;
 150                                 break;
 151                         }
 152                 }
 153                 qunlock(p);
 154         }
 155         return wq;
 156 }
 157
 158 static int
 159 pipestat(Chan *c, uchar *db, int n)
 160 {
 161         Pipe *p;
 162         Dir dir;
 163
 164         p = c->aux;
 165
 166         switch(NETTYPE(c->qid.path)){
 167         case Qdir:
 168                 devdir(c, c->qid, ".", 0, eve, 0555, &dir);
 169                 break;
 170         case Qdata0:
 171                 devdir(c, c->qid, "data", qlen(p->q[0]), eve, p->perm, &dir);
 172                 break;
 173         case Qdata1:
 174                 devdir(c, c->qid, "data1", qlen(p->q[1]), eve, p->perm, &dir);
 175                 break;
 176         default:
 177                 panic("pipestat");
 178         }
 179         n = convD2M(&dir, db, n);
 180         if(n < BIT16SZ)
 181                 error(Eshortstat);
 182         return n;
 183 }
 184
 185 static int
 186 pipewstat(Chan* c, uchar* db, int n)
 187 {
 188         int m;
 189         Dir *dir;
 190         Pipe *p;
 191
 192         p = c->aux;
 193         if(strcmp(up->user, eve) != 0)
 194                 error(Eperm);
 195         if(NETTYPE(c->qid.path) == Qdir)
 196                 error(Eisdir);
 197
 198         dir = smalloc(sizeof(Dir)+n);
 199         if(waserror()){
 200                 free(dir);
 201                 nexterror();
 202         }
 203         m = convM2D(db, n, &dir[0], (char*)&dir[1]);
 204         if(m == 0)
 205                 error(Eshortstat);
 206         if(!emptystr(dir[0].uid))
 207                 error("can't change owner");
 208         if(dir[0].mode != ~0UL)
 209                 p->perm = dir[0].mode;
 210         poperror();
 211         free(dir);
 212         return m;
 213 }
 214
 215 /*
 216  *  if the stream doesn't exist, create it
 217  */
 218 static Chan*
 219 pipeopen(Chan *c, int omode)
 220 {
 221         Pipe *p;
 222
 223         if(c->qid.type & QTDIR){
 224                 if(omode != OREAD)
 225                         error(Ebadarg);
 226                 c->mode = omode;
 227                 c->flag |= COPEN;
 228                 c->offset = 0;
 229                 return c;
 230         }
 231
 232         p = c->aux;
 233         qlock(p);
 234         switch(NETTYPE(c->qid.path)){
 235         case Qdata0:
 236                 p->qref[0]++;
 237                 break;
 238         case Qdata1:
 239                 p->qref[1]++;
 240                 break;
 241         }
 242         qunlock(p);
 243
 244         c->mode = openmode(omode);
 245         c->flag |= COPEN;
 246         c->offset = 0;
 247         c->iounit = qiomaxatomic;
 248         return c;
 249 }
 250
 251 static void
 252 pipeclose(Chan *c)
 253 {
 254         Pipe *p;
 255
 256         p = c->aux;
 257         qlock(p);
 258
 259         if(c->flag & COPEN){
 260                 /*
 261                  *  closing either side hangs up the stream
 262                  */
 263                 switch(NETTYPE(c->qid.path)){
 264                 case Qdata0:
 265                         p->qref[0]--;
 266                         if(p->qref[0] == 0){
 267                                 qhangup(p->q[1], 0);
 268                                 qclose(p->q[0]);
 269                         }
 270                         break;
 271                 case Qdata1:
 272                         p->qref[1]--;
 273                         if(p->qref[1] == 0){
 274                                 qhangup(p->q[0], 0);
 275                                 qclose(p->q[1]);
 276                         }
 277                         break;
 278                 }
 279         }
 280
 281
 282         /*
 283          *  if both sides are closed, they are reusable
 284          */
 285         if(p->qref[0] == 0 && p->qref[1] == 0){
 286                 qreopen(p->q[0]);
 287                 qreopen(p->q[1]);
 288         }
 289
 290         /*
 291          *  free the structure on last close
 292          */
 293         p->ref--;
 294         if(p->ref == 0){
 295                 qunlock(p);
 296                 free(p->q[0]);
 297                 free(p->q[1]);
 298                 free(p);
 299         } else
 300                 qunlock(p);
 301 }
 302
 303 static long
 304 piperead(Chan *c, void *va, long n, vlong)
 305 {
 306         Pipe *p;
 307
 308         p = c->aux;
 309
 310         switch(NETTYPE(c->qid.path)){
 311         case Qdir:
 312                 return devdirread(c, va, n, pipedir, NPIPEDIR, pipegen);
 313         case Qdata0:
 314                 return qread(p->q[0], va, n);
 315         case Qdata1:
 316                 return qread(p->q[1], va, n);
 317         default:
 318                 panic("piperead");
 319         }
 320         return -1;      /* not reached */
 321 }
 322
 323 static Block*
 324 pipebread(Chan *c, long n, ulong offset)
 325 {
 326         Pipe *p;
 327
 328         p = c->aux;
 329
 330         switch(NETTYPE(c->qid.path)){
 331         case Qdata0:
 332                 return qbread(p->q[0], n);
 333         case Qdata1:
 334                 return qbread(p->q[1], n);
 335         }
 336
 337         return devbread(c, n, offset);
 338 }
 339
 340 /*
 341  *  a write to a closed pipe causes a note to be sent to
 342  *  the process.
 343  */
 344 static long
 345 pipewrite(Chan *c, void *va, long n, vlong)
 346 {
 347         Pipe *p;
 348
 349         if(!islo())
 350                 print("pipewrite hi %#p\n", getcallerpc(&c));
 351         if(waserror()) {
 352                 /* avoid notes when pipe is a mounted queue */
 353                 if((c->flag & CMSG) == 0)
 354                         postnote(up, 1, "sys: write on closed pipe", NUser);
 355                 nexterror();
 356         }
 357
 358         p = c->aux;
 359
 360         switch(NETTYPE(c->qid.path)){
 361         case Qdata0:
 362                 n = qwrite(p->q[1], va, n);
 363                 break;
 364
 365         case Qdata1:
 366                 n = qwrite(p->q[0], va, n);
 367                 break;
 368
 369         default:
 370                 panic("pipewrite");
 371         }
 372
 373         poperror();
 374         return n;
 375 }
 376
 377 static long
 378 pipebwrite(Chan *c, Block *bp, ulong)
 379 {
 380         long n;
 381         Pipe *p;
 382
 383         if(waserror()) {
 384                 /* avoid notes when pipe is a mounted queue */
 385                 if((c->flag & CMSG) == 0)
 386                         postnote(up, 1, "sys: write on closed pipe", NUser);
 387                 nexterror();
 388         }
 389
 390         p = c->aux;
 391         switch(NETTYPE(c->qid.path)){
 392         case Qdata0:
 393                 n = qbwrite(p->q[1], bp);
 394                 break;
 395
 396         case Qdata1:
 397                 n = qbwrite(p->q[0], bp);
 398                 break;
 399
 400         default:
 401                 n = 0;
 402                 panic("pipebwrite");
 403         }
 404
 405         poperror();
 406         return n;
 407 }
 408
 409 Dev pipedevtab = {
 410         '|',
 411         "pipe",
 412
 413         devreset,
 414         pipeinit,
 415         devshutdown,
 416         pipeattach,
 417         pipewalk,
 418         pipestat,
 419         pipeopen,
 420         devcreate,
 421         pipeclose,
 422         piperead,
 423         pipebread,
 424         pipewrite,
 425         pipebwrite,
 426         devremove,
 427         pipewstat,
 428 };