workbench/devs/xpipe_handler.c

   1 /*
   2     Copyright © 2008, The AROS Development Team. All rights reserved.
   3     $Id$
   4 */
   5
   6 /* Basic assumptions:
   7
   8 - buffered one-directional pipes
   9 - both ends opened at once with FSA_PIPE
  10 - FSA_OPEN and FSA_OPEN_FILE used only for duplicating locks (file name "")
  11 - no support for multiple readers (however there may be multiple duplicated pipe ends)
  12
  13 */
  14
  15 #define DEBUG 0
  16 #include <aros/debug.h>
  17
  18 #include <exec/errors.h>
  19 #include <exec/resident.h>
  20 #include <exec/memory.h>
  21 #include <exec/semaphores.h>
  22 #include <exec/lists.h>
  23 #include <exec/nodes.h>
  24 #include <proto/exec.h>
  25 #include <utility/tagitem.h>
  26 #include <dos/dosextens.h>
  27 #include <dos/filesystem.h>
  28 #include <clib/macros.h>
  29 #include <proto/dos.h>
  30 #include <aros/symbolsets.h>
  31 #include <aros/asmcall.h>
  32 #if defined(__GNUC__) || defined(__INTEL_COMPILER)
  33 #include "xpipe_handler_gcc.h"
  34 #endif
  35
  36 #include LC_LIBDEFS_FILE
  37
  38 #include <string.h>
  39 #include <stddef.h>
  40
  41 AROS_UFP3(LONG, xpipeproc,
  42     AROS_UFPA(char *,argstr,A0),
  43     AROS_UFPA(ULONG,argsize,D0),
  44     AROS_UFPA(struct ExecBase *,SysBase,A6));
  45
  46 struct XPipeMessage
  47 {
  48     struct Message msg;
  49     struct IOFileSys *iofs;
  50     LONG curlen;
  51 };
  52
  53 struct XPipe
  54 {
  55     struct Node node;
  56     struct List readers; /* List of XPipeEnd structures */
  57     struct List writers; /* List of XPipeEnd structures */
  58     struct List pendingreads;
  59     struct List pendingwrites;
  60     char buffer[4096];
  61     int bufstart;
  62     int bufend;
  63 };
  64
  65 struct XPipeUnit
  66 {
  67     struct Node node;
  68     struct List pipes;
  69 };
  70
  71 struct XPipeEnd
  72 {
  73     struct Node node;
  74     struct XPipe *pipe;
  75     ULONG mode;
  76     unsigned int bufpos; /* Used only for readers */
  77 };
  78
  79 enum { NONE, PROCESS_READS, PROCESS_WRITES };
  80 enum { NT_PIPEROOT, NT_PIPEEND };
  81
  82 static ULONG            SendRequest  (struct XPipeBase *xpipebase, struct IOFileSys *iofs, BOOL abort);
  83
  84 static int GM_UNIQUENAME(Init)(LIBBASETYPEPTR xpipebase)
  85 {
  86     DOSBase =  (struct DosLibrary *)OpenLibrary("dos.library",39);
  87
  88     if(DOSBase)
  89     {
  90         struct TagItem taglist[]=
  91         {
  92          {NP_Entry,              (IPTR)xpipeproc},
  93          {NP_Name, (IPTR)"xpipe.handler process"},
  94          {NP_UserData,           (IPTR)xpipebase},
  95          {TAG_DONE,                           0}
  96         };
  97
  98         xpipebase->proc = CreateNewProc(taglist);
  99
 100         if (xpipebase->proc)
 101             return TRUE;
 102
 103         CloseLibrary((struct Library *)DOSBase);
 104     }
 105
 106     return FALSE;
 107 }
 108
 109 static int GM_UNIQUENAME(Open)
 110 (
 111     LIBBASETYPEPTR xpipebase,
 112     struct IOFileSys *iofs,
 113     ULONG unitnum,
 114     ULONG flags
 115 )
 116 {
 117     struct XPipeUnit *un;
 118
 119     /* Mark message as replied. */
 120     iofs->IOFS.io_Message.mn_Node.ln_Type=NT_REPLYMSG;
 121
 122     un = AllocVec(sizeof(struct XPipeUnit), MEMF_PUBLIC | MEMF_CLEAR);
 123     if(un)
 124     {
 125         NEWLIST(&un->pipes);
 126         un->node.ln_Type = NT_PIPEROOT;
 127         iofs->IOFS.io_Unit=(struct Unit *)un;
 128         iofs->IOFS.io_Device=&xpipebase->device;
 129         return TRUE;
 130     }
 131
 132     iofs->io_DosError=ERROR_NO_FREE_STORE;
 133     iofs->IOFS.io_Error=IOERR_OPENFAIL;
 134
 135     return FALSE;
 136 }
 137
 138 static int GM_UNIQUENAME(Close)
 139 (
 140     LIBBASETYPEPTR xpipebase,
 141     struct IOFileSys *iofs
 142 )
 143 {
 144     struct XPipeUnit *un = (struct XPipeUnit *)iofs->IOFS.io_Unit;
 145
 146     if(!IsListEmpty(&un->pipes))
 147     {
 148         iofs->io_DosError=ERROR_OBJECT_IN_USE;
 149         return FALSE;
 150     }
 151
 152     FreeVec(un);
 153
 154     iofs->io_DosError=0;
 155
 156     return TRUE;
 157 }
 158
 159 static int GM_UNIQUENAME(Expunge)(LIBBASETYPEPTR xpipebase)
 160 {
 161     /*
 162         This function is single-threaded by exec by calling Forbid.
 163         Never break the Forbid() or strange things might happen.
 164     */
 165
 166     SendRequest(xpipebase, NULL, TRUE);
 167
 168     /* Free all resources */
 169     CloseLibrary((struct Library *)xpipebase->dosbase);
 170
 171     return TRUE;
 172 }
 173
 174 AROS_LH1(void, beginio,
 175  AROS_LHA(struct IOFileSys *, iofs, A1),
 176            struct XPipeBase *, xpipebase, 5, Xpipe)
 177 {
 178     AROS_LIBFUNC_INIT
 179     LONG error=0;
 180     BOOL enqueued = FALSE;
 181
 182     D(bug("[xpipe] COMMAND %d\n", iofs->IOFS.io_Command));
 183     switch(iofs->IOFS.io_Command)
 184     {
 185         case FSA_OPEN:
 186         case FSA_OPEN_FILE:
 187         case FSA_PIPE:
 188         case FSA_EXAMINE:
 189         case FSA_READ:
 190         case FSA_WRITE:
 191         case FSA_CLOSE:
 192         case FSA_FILE_MODE:
 193             error = SendRequest(xpipebase, iofs, FALSE);
 194             enqueued = !error;
 195             break;
 196
 197         case FSA_SEEK:
 198             error = ERROR_SEEK_ERROR;
 199             break;
 200         case FSA_IS_FILESYSTEM:
 201             iofs->io_Union.io_IS_FILESYSTEM.io_IsFilesystem = FALSE;
 202             break;
 203         case FSA_CREATE_DIR:
 204         case FSA_DELETE_OBJECT:
 205         case FSA_SET_FILE_SIZE:
 206         case FSA_EXAMINE_ALL:
 207         case FSA_CREATE_HARDLINK:
 208         case FSA_CREATE_SOFTLINK:
 209         case FSA_RENAME:
 210         case FSA_EXAMINE_NEXT:
 211             error = ERROR_NOT_IMPLEMENTED;
 212             break;
 213
 214         default:
 215             error = ERROR_ACTION_NOT_KNOWN;
 216             break;
 217     }
 218
 219     /* Set error code */
 220     iofs->io_DosError=error;
 221
 222     /* If the quick bit is not set and the request hasn't been redirected
 223        send the message to the port
 224     */
 225     if(!(iofs->IOFS.io_Flags&IOF_QUICK) && !enqueued)
 226         ReplyMsg(&iofs->IOFS.io_Message);
 227
 228     AROS_LIBFUNC_EXIT
 229 }
 230
 231 ADD2INITLIB(GM_UNIQUENAME(Init), 0)
 232 ADD2OPENDEV(GM_UNIQUENAME(Open), 0)
 233 ADD2CLOSEDEV(GM_UNIQUENAME(Close), 0)
 234 ADD2EXPUNGELIB(GM_UNIQUENAME(Expunge), 0)
 235
 236 AROS_LH1(LONG, abortio,
 237 AROS_LHA(struct IOFileSys *, iofs, A1),
 238 struct XPipeBase *, xpipebase, 6, Xpipe)
 239 {
 240     AROS_LIBFUNC_INIT
 241
 242     return SendRequest(xpipebase, iofs, TRUE);
 243
 244     AROS_LIBFUNC_EXIT
 245 }
 246
 247 static ULONG SendRequest(struct XPipeBase *xpipebase, struct IOFileSys *iofs, BOOL abort)
 248 {
 249     struct XPipeMessage *msg = AllocVec(sizeof(*msg), MEMF_PUBLIC);
 250
 251     if (msg)
 252     {
 253         msg->msg.mn_Node.ln_Type = NT_MESSAGE;
 254         msg->msg.mn_Node.ln_Name = "XPIPEMSG";
 255         msg->msg.mn_Length       = sizeof(struct XPipeMessage);
 256         msg->iofs                = iofs;
 257         msg->curlen              = abort;
 258
 259         if (iofs)
 260         {
 261             iofs->IOFS.io_Message.mn_Node.ln_Type = NT_MESSAGE;
 262             iofs->IOFS.io_Flags &= ~IOF_QUICK;
 263         }
 264
 265         PutMsg(&xpipebase->proc->pr_MsgPort, (struct Message *)msg);
 266
 267         return 0;
 268     }
 269
 270     return ERROR_NO_FREE_STORE;
 271 }
 272
 273 /* The helper process */
 274
 275 #define SendBack(msg, err)                   \
 276 {                                            \
 277     msg->iofs->io_DosError = err;            \
 278     ReplyMsg(&(msg)->iofs->IOFS.io_Message); \
 279     FreeVec(msg);                            \
 280 }
 281
 282 LONG DuplicatePipeEnd(struct XPipeEnd **oldend, ULONG mode)
 283 {
 284     struct XPipeEnd *newend;
 285     struct XPipe *pipe = (*oldend)->pipe;
 286
 287     if ((newend = AllocVec(sizeof(struct XPipeEnd), MEMF_PUBLIC | MEMF_CLEAR)) == NULL)
 288     {
 289         return ERROR_NO_FREE_STORE;;
 290     }
 291
 292     newend->node.ln_Type = NT_PIPEEND;
 293     newend->mode = mode;
 294     newend->pipe = (*oldend)->pipe;
 295     newend->bufpos = (*oldend)->bufpos;
 296
 297     if (mode & FMF_WRITE)
 298     {
 299         D(bug("[xpipe] Cloned pipe end is a writer\n"));
 300         ADDTAIL(&pipe->writers, newend);
 301     }
 302     if (mode & FMF_READ)
 303     {
 304         D(bug("[xpipe] Cloned pipe end is a reader\n"));
 305         ADDTAIL(&pipe->readers, newend);
 306     }
 307
 308     *oldend = newend;
 309     return 0;
 310 }
 311
 312 ULONG ReadWouldBlock (struct XPipe *pipe, ULONG length)
 313 {
 314     int curlen;
 315     int numwriters;
 316     ListLength (&pipe->writers, numwriters);
 317
 318     /* If there are no writers, we won't block */
 319     if(numwriters == 0)
 320         return FALSE;
 321
 322     if(pipe->bufstart <= pipe->bufend)
 323         curlen = pipe->bufend - pipe->bufstart;
 324     else
 325         curlen = (pipe->bufend + sizeof(pipe->buffer) - pipe->bufstart) % sizeof(pipe->buffer);
 326
 327     if(length > curlen)
 328         return TRUE;
 329     else
 330         return FALSE;
 331 }
 332
 333 ULONG WriteWouldBlock (struct XPipe *pipe, ULONG length)
 334 {
 335     int curlen;
 336     if(pipe->bufstart <= pipe->bufend)
 337         curlen = pipe->bufend - pipe->bufstart;
 338     else
 339         curlen = (pipe->bufend + sizeof(pipe->buffer) - pipe->bufstart) % sizeof(pipe->buffer);
 340
 341     if(sizeof(pipe->buffer) - curlen - 1 < length)
 342         return TRUE;
 343     else
 344         return FALSE;
 345 }
 346
 347 LONG ReadFromPipe (struct XPipe *pipe, APTR buffer, ULONG length)
 348 {
 349     D(bug("[xpipe] ReadFromPipe(%p, %p, %d)\n", pipe, buffer, length));
 350     D(bug("[xpipe] buffer before read from %d to %d\n", pipe->bufstart, pipe->bufend));
 351     if(pipe->bufstart <= pipe->bufend)
 352     {
 353         /* read bytes from between bufstart and bufend */
 354         int bytestoread = MIN((pipe->bufend - pipe->bufstart), length);
 355         CopyMem((APTR)((IPTR)pipe->buffer + pipe->bufstart), buffer, bytestoread);
 356         /* advance buffer */
 357         pipe->bufstart += bytestoread;
 358         D(bug("[xpipe] buffer after read from %d to %d\n", pipe->bufstart, pipe->bufend));
 359         return bytestoread;
 360     }
 361     else
 362     {
 363         /* read bytes from bufstart to the end of the buffer */
 364         int bytestoread1 = MIN((sizeof(pipe->buffer) - pipe->bufstart), length);
 365         CopyMem((APTR)((IPTR)pipe->buffer + pipe->bufstart), buffer, bytestoread1);
 366         pipe->bufstart = (pipe->bufstart + bytestoread1) % sizeof(pipe->buffer);
 367         if(bytestoread1 == length)
 368         {
 369             D(bug("[xpipe] buffer after read from %d to %d\n", pipe->bufstart, pipe->bufend));
 370             return length;
 371         }
 372         /*  and if it's not enough then from the beginning of the buffer to
 373             bufend */
 374         int bytestoread2 = MIN(pipe->bufend, (length - bytestoread1));
 375         CopyMem(pipe->buffer, (APTR)((IPTR) buffer + bytestoread1), bytestoread2);
 376         pipe->bufstart = bytestoread2;
 377         D(bug("[xpipe] buffer after read from %d to %d\n", pipe->bufstart, pipe->bufend));
 378         return bytestoread1 + bytestoread2;
 379     }
 380 }
 381
 382 LONG WriteToPipe (struct XPipe *pipe, APTR buffer, ULONG length)
 383 {
 384     D(bug("[xpipe] WriteToPipe(%p, %p, %d)\n", pipe, buffer, length));
 385     D(bug("[xpipe] buffer before write from %d to %d\n", pipe->bufstart, pipe->bufend));
 386     if(pipe->bufstart > pipe->bufend)
 387     {
 388         /* write between bufend and bufstart, left one byte */
 389         int bytestowrite = MIN((pipe->bufstart - pipe->bufend - 1), length);
 390         CopyMem(buffer, (APTR)((IPTR)pipe->buffer + pipe->bufend), bytestowrite);
 391         /* advance buffer */
 392         pipe->bufend += bytestowrite;
 393         D(bug("[xpipe] buffer after write from %d to %d\n", pipe->bufstart, pipe->bufend));
 394         return bytestowrite;
 395     }
 396     else
 397     {
 398         /* write bytes from bufend to the end of the buffer */
 399         int bytestowrite1 = MIN((sizeof(pipe->buffer) - pipe->bufend), length);
 400         /* if there's no space left at the beginning then make sure we leave
 401            at least one byte free at the end */
 402         if(pipe->bufstart == 0 && bytestowrite1 + pipe->bufend == sizeof(pipe->buffer)) bytestowrite1--;
 403         CopyMem(buffer, (APTR)((IPTR)pipe->buffer + pipe->bufend), bytestowrite1);
 404         pipe->bufend = (pipe->bufend + bytestowrite1) % sizeof(pipe->buffer);
 405
 406         D(bug("[xpipe] wrote %d bytes in first part\n", bytestowrite1));
 407         /* skip the second part if there's no space free at the beginning */
 408         if(bytestowrite1 == length || pipe->bufstart == 0)
 409         {
 410             D(bug("[xpipe] buffer after write from %d to %d\n", pipe->bufstart, pipe->bufend));
 411             return bytestowrite1;
 412         }
 413         /*  and if it's not enough then from the beginning of the buffer to
 414             bufstart - 1 (one byte free left) */
 415         int bytestowrite2 = MIN(pipe->bufstart - 1, (length - bytestowrite1));
 416         D(bug("[xpipe] wrote %d bytes in second part\n", bytestowrite2));
 417         CopyMem((APTR)((IPTR) buffer + bytestowrite1), pipe->buffer, bytestowrite2);
 418         pipe->bufend = bytestowrite2;
 419         D(bug("[xpipe] buffer after write from %d to %d\n", pipe->bufstart, pipe->bufend));
 420         return bytestowrite1 + bytestowrite2;
 421     }
 422 }
 423
 424 /* Pump data from write request buffers to read request buffers through
 425    the pipe buffer */
 426 void pump(struct XPipe *pipe, int operation)
 427 {
 428     while(operation != NONE)
 429     {
 430         switch(operation)
 431         {
 432             case PROCESS_READS:
 433             {
 434                 D(bug("[xpipe] Processing pending reads\n"));
 435                 operation = NONE;
 436                 /* Ok, we have some new data, use them to continue with pending reads */
 437                 struct XPipeMessage *readmsg;
 438                 struct Node *tempnode;
 439                 ForeachNodeSafe (&pipe->pendingreads, readmsg, tempnode)
 440                 {
 441                     struct XPipeEnd *reader = (struct XPipeEnd*) readmsg->iofs->IOFS.io_Unit;
 442                     struct XPipe *pipe = reader->pipe;
 443                     int nread = readmsg->curlen;
 444
 445                     readmsg->curlen += ReadFromPipe (
 446                         pipe,
 447                         (APTR) ((IPTR) readmsg->iofs->io_Union.io_READ_WRITE.io_Buffer + readmsg->curlen),
 448                         readmsg->iofs->io_Union.io_READ_WRITE.io_Length - readmsg->curlen
 449                     );
 450                     nread = readmsg->curlen - nread;
 451                     D(bug("[xpipe] managed to pump %d bytes from pipe (%d from requested %d)\n", nread, readmsg->curlen, readmsg->iofs->io_Union.io_READ_WRITE.io_Length));
 452                     if(nread > 0)
 453                     {
 454                         /* We managed to free some buffer space, now we can
 455                            write some data */
 456                         operation = PROCESS_WRITES;
 457                     }
 458
 459                     if(readmsg->iofs->io_Union.io_READ_WRITE.io_Length == readmsg->curlen)
 460                     {
 461                         D(bug("[xpipe] completed read request %p\n", readmsg));
 462                         /* hooray, we managed to read all data */
 463                         Remove ((struct Node*) readmsg);
 464                         SendBack (readmsg, 0);
 465                     }
 466                     else
 467                     {
 468                         /* Since we didn't manage to complete this request,
 469                            there's no point in processing other */
 470                         break;
 471                     }
 472                 }
 473                 break;
 474             }
 475             case PROCESS_WRITES:
 476             {
 477                 D(bug("[xpipe] Processing pending writes\n"));
 478                 operation = NONE;
 479                 /* Ok, we have some new data, use them to continue with pending reads */
 480                 struct XPipeMessage *writemsg;
 481                 struct Node *tempnode;
 482                 ForeachNodeSafe (&pipe->pendingwrites, writemsg, tempnode)
 483                 {
 484                     struct XPipeEnd *writer = (struct XPipeEnd*) writemsg->iofs->IOFS.io_Unit;
 485                     struct XPipe *pipe = writer->pipe;
 486                     int nwrote = writemsg->curlen;
 487
 488                     writemsg->curlen += WriteToPipe (
 489                         pipe,
 490                         (APTR) ((IPTR) writemsg->iofs->io_Union.io_READ_WRITE.io_Buffer + writemsg->curlen),
 491                         writemsg->iofs->io_Union.io_READ_WRITE.io_Length - writemsg->curlen
 492                     );
 493                     nwrote = writemsg->curlen - nwrote;
 494                     D(bug("[xpipe] managed to pump %d bytes to pipe (%d from requested %d)\n", nwrote, writemsg->curlen, writemsg->iofs->io_Union.io_READ_WRITE.io_Length));
 495                     if(nwrote > 0)
 496                     {
 497                         /* We managed to write some data, now we can process
 498                            some pending reads. */
 499                         operation = PROCESS_READS;
 500                     }
 501
 502                     if(writemsg->iofs->io_Union.io_READ_WRITE.io_Length == writemsg->curlen)
 503                     {
 504                         D(bug("[xpipe] completed write request %p\n", writemsg));
 505                         /* hooray, we managed to write all data */
 506                         Remove ((struct Node*) writemsg);
 507                         SendBack (writemsg, 0);
 508                     }
 509                     else
 510                     {
 511                         /* Since we didn't manage to complete this request,
 512                            there's no point in processing other */
 513                         break;
 514                     }
 515                 }
 516                 break;
 517             }
 518         }
 519     }
 520 }
 521
 522 AROS_UFH3(LONG, xpipeproc,
 523     AROS_UFHA(char *,argstr,A0),
 524     AROS_UFHA(ULONG,argsize,D0),
 525     AROS_UFHA(struct ExecBase *,SysBase,A6))
 526 {
 527     AROS_USERFUNC_INIT
 528
 529     struct Process       *me;
 530     struct XPipeBase     *xpipebase;
 531     struct XPipeMessage  *msg;
 532     struct Node          *pn;
 533     BOOL cont = TRUE;
 534
 535     me         = (struct Process *)FindTask(0);
 536     xpipebase = me->pr_Task.tc_UserData;
 537
 538
 539     do
 540     {
 541         WaitPort (&(me->pr_MsgPort));
 542
 543         while
 544         (
 545             (msg =(struct XPipeMessage *)GetMsg (&(me->pr_MsgPort))) &&
 546             (cont = (msg->iofs != 0))
 547         )
 548         {
 549             D(bug("[xpipe] Message received.\n"));
 550
 551             pn = (struct Node *)msg->iofs->IOFS.io_Unit;
 552
 553             switch (msg->iofs->IOFS.io_Command)
 554             {
 555                 case FSA_OPEN:
 556                 {
 557                     D(bug("[xpipe] Cmd is FSA_OPEN\n"));
 558
 559                     if(pn->ln_Type != NT_PIPEEND)
 560                     {
 561                         SendBack (msg, ERROR_OBJECT_NOT_FOUND);
 562                         break;
 563                     }
 564
 565                     if (msg->iofs->io_Union.io_OPEN.io_Filename[0])
 566                     {
 567                         SendBack (msg, ERROR_OBJECT_NOT_FOUND);
 568                         break;
 569                     }
 570
 571                     D(bug("[xpipe] Cloning pipe end: %p with mode \n", msg->iofs->IOFS.io_Unit, msg->iofs->io_Union.io_OPEN.io_FileMode));
 572
 573                     LONG error = DuplicatePipeEnd (
 574                         (struct XPipeEnd**) &msg->iofs->IOFS.io_Unit,
 575                         msg->iofs->io_Union.io_OPEN.io_FileMode
 576                     );
 577
 578                     D(bug("[xpipe] Cloned pipe end: %p\n", msg->iofs->IOFS.io_Unit));
 579
 580                     SendBack (msg, error);
 581                     break;
 582                 }
 583                 case FSA_OPEN_FILE:
 584                 {
 585                     D(bug("[xpipe] Cmd is FSA_OPEN_FILE\n"));
 586
 587                     if(pn->ln_Type != NT_PIPEEND)
 588                     {
 589                         SendBack (msg, ERROR_OBJECT_NOT_FOUND);
 590                         break;
 591                     }
 592
 593                     if (msg->iofs->io_Union.io_OPEN_FILE.io_Filename[0])
 594                     {
 595                         SendBack (msg, ERROR_OBJECT_NOT_FOUND);
 596                         break;
 597                     }
 598
 599                     D(bug("[xpipe] Cloning pipe end: %p with mode %d\n", msg->iofs->IOFS.io_Unit, msg->iofs->io_Union.io_OPEN_FILE.io_FileMode));
 600
 601                     LONG error = DuplicatePipeEnd (
 602                         (struct XPipeEnd**) &msg->iofs->IOFS.io_Unit,
 603                         msg->iofs->io_Union.io_OPEN_FILE.io_FileMode
 604                     );
 605                     D(bug("[xpipe] Cloned pipe end: %p\n", msg->iofs->IOFS.io_Unit));
 606                     SendBack (msg, error);
 607                     break;
 608                 }
 609                 case FSA_PIPE:
 610                 {
 611                     struct XPipeEnd *reader, *writer;
 612                     struct XPipe *pipe;
 613                     struct XPipeUnit *un;
 614
 615                     D(bug("[xpipe] Cmd is FSA_PIPE\n"));
 616
 617                     if(pn->ln_Type != NT_PIPEROOT)
 618                     {
 619                         SendBack (msg, ERROR_OBJECT_WRONG_TYPE);
 620                         break;
 621                     }
 622                     un = (struct XPipeUnit*) pn;
 623
 624                     if ((reader = AllocVec (sizeof(struct XPipeEnd), MEMF_PUBLIC | MEMF_CLEAR)) == NULL)
 625                     {
 626                         SendBack (msg, ERROR_NO_FREE_STORE);
 627                         break;
 628                     }
 629                     reader->node.ln_Type = NT_PIPEEND;
 630                     reader->mode = FMF_READ;
 631                     reader->bufpos = 0;
 632
 633                     if ((writer = AllocVec (sizeof(struct XPipeEnd), MEMF_PUBLIC | MEMF_CLEAR)) == NULL)
 634                     {
 635                         FreeVec (reader);
 636                         SendBack (msg, ERROR_NO_FREE_STORE);
 637                         break;
 638                     }
 639                     writer->node.ln_Type = NT_PIPEEND;
 640                     writer->mode = FMF_WRITE;
 641                     writer->bufpos = 0;
 642
 643                     if ((pipe = AllocVec (sizeof(struct XPipe), MEMF_PUBLIC | MEMF_CLEAR)) == NULL)
 644                     {
 645                         FreeVec (reader);
 646                         FreeVec (writer);
 647                         SendBack (msg, ERROR_NO_FREE_STORE);
 648                         break;
 649                     }
 650                     reader->pipe = pipe;
 651                     writer->pipe = pipe;
 652                     NEWLIST (&pipe->readers);
 653                     NEWLIST (&pipe->pendingreads);
 654                     ADDTAIL (&pipe->readers, reader);
 655                     NEWLIST (&pipe->writers);
 656                     NEWLIST (&pipe->pendingwrites);
 657                     ADDTAIL (&pipe->writers, writer);
 658                     ADDTAIL (&un->pipes, pipe);
 659
 660                     D(bug("[xpipe] Opened pipe with read end: %p and write end: %p\n", reader, writer));
 661
 662                     msg->iofs->IOFS.io_Unit = (struct Unit *) reader;
 663                     msg->iofs->io_Union.io_PIPE.io_Writer = (struct Unit *) writer;
 664
 665                     SendBack (msg, 0);
 666
 667                     break;
 668                 }
 669
 670                 case FSA_CLOSE:
 671                 {
 672                     D(bug("[xpipe] Cmd is FSA_CLOSE\n"));
 673
 674                     if(pn->ln_Type != NT_PIPEEND)
 675                     {
 676                         SendBack (msg, ERROR_OBJECT_NOT_FOUND);
 677                         break;
 678                     }
 679                     struct XPipeEnd *pipeend = (struct XPipeEnd*) pn;
 680                     struct XPipe *pipe = pipeend->pipe;
 681
 682                     D(bug("[xpipe] Closing pipe end %p\n", pipeend));
 683
 684                     Remove ((struct Node*) pipeend);
 685                     FreeVec (pipeend);
 686
 687                     int numreaders, numwriters;
 688                     ListLength (&pipe->readers, numreaders);
 689                     ListLength (&pipe->writers, numwriters);
 690
 691                     /* If all writing ends are closed we have EOF, so finish all pending read requests */
 692                     if(numwriters == 0)
 693                     {
 694                         D(bug("[xpipe] Processing pending reads\n"));
 695                         struct XPipeMessage *readmsg;
 696                         struct Node *tempnode;
 697                         ForeachNodeSafe (&pipe->pendingreads, readmsg, tempnode)
 698                         {
 699                             D(bug("[xpipe] Pending read msg %p\n", readmsg));
 700                             readmsg->iofs->io_Union.io_READ_WRITE.io_Length = readmsg->curlen;
 701                             Remove((struct Node *)readmsg);
 702                             SendBack (readmsg, 0);
 703                         }
 704                     }
 705
 706                     /* If there are no pipe ends left, close the pipe */
 707                     if (numreaders == 0 && numwriters == 0)
 708                     {
 709                         D(bug("[xpipe] No ends left, closing the pipe\n"));
 710                         Remove ((struct Node*) pipe);
 711                         FreeVec (pipe);
 712                     }
 713
 714                     SendBack (msg, 0);
 715                     break;
 716                 }
 717                 case FSA_EXAMINE:
 718                 {
 719                     struct ExAllData  *ead        = msg->iofs->io_Union.io_EXAMINE.io_ead;
 720                     const ULONG        type       = msg->iofs->io_Union.io_EXAMINE.io_Mode;
 721                     const ULONG        size       = msg->iofs->io_Union.io_EXAMINE.io_Size;
 722                     STRPTR             next, end;
 723
 724                     static const ULONG sizes[]=
 725                     {
 726                         0,
 727                         offsetof(struct ExAllData,ed_Type),
 728                         offsetof(struct ExAllData,ed_Size),
 729                         offsetof(struct ExAllData,ed_Prot),
 730                         offsetof(struct ExAllData,ed_Days),
 731                         offsetof(struct ExAllData,ed_Comment),
 732                         offsetof(struct ExAllData,ed_OwnerUID),
 733                         sizeof(struct ExAllData)
 734                     };
 735
 736                     D(bug("[xpipe] Cmd is EXAMINE\n"));
 737
 738                     if (type > ED_OWNER)
 739                     {
 740                         D(bug("[xpipe] The user requested an invalid type\n"));
 741                         SendBack (msg, ERROR_BAD_NUMBER);
 742                         break;
 743                     }
 744
 745                     next = (STRPTR)ead + sizes[type];
 746                     end  = (STRPTR)ead + size;
 747
 748                     if(next>end)  /* > is correct. Not >= */
 749                     {
 750                         SendBack (msg, ERROR_BUFFER_OVERFLOW);
 751                         break;
 752                     }
 753
 754                     switch(type)
 755                     {
 756                         case ED_OWNER:
 757                         ead->ed_OwnerUID = 0;
 758                         ead->ed_OwnerGID = 0;
 759
 760                         /* Fall through */
 761                         case ED_COMMENT:
 762                             ead->ed_Comment = NULL;
 763
 764                         /* Fall through */
 765                         case ED_DATE:
 766                             ead->ed_Days  = 0;
 767                             ead->ed_Mins  = 0;
 768                             ead->ed_Ticks = 0;
 769
 770                         /* Fall through */
 771                         case ED_PROTECTION:
 772                             ead->ed_Prot = 0;
 773
 774                         /* Fall through */
 775                         case ED_SIZE:
 776                             ead->ed_Size = 0;
 777
 778                         /* Fall through */
 779                         case ED_TYPE:
 780                             ead->ed_Type = ST_PIPEFILE;
 781
 782                         /* Fall through */
 783                         case ED_NAME:
 784                         {
 785                             ead->ed_Name = next;
 786                             ead->ed_Name[0] = '\0';
 787                         }
 788                     }
 789
 790                     ead->ed_Next = NULL;
 791
 792                     SendBack (msg, 0);
 793                     break;
 794                 }
 795                 case FSA_WRITE:
 796                 {
 797                     D(bug("[xpipe] Cmd is FSA_WRITE.\n"));
 798
 799                     if(pn->ln_Type != NT_PIPEEND)
 800                     {
 801                         SendBack (msg, ERROR_OBJECT_NOT_FOUND);
 802                         break;
 803                     }
 804                     struct XPipeEnd *writer = (struct XPipeEnd *) pn;
 805
 806                     D(bug("[xpipe] Writer end %p\n", writer));
 807                     struct XPipe *pipe = writer->pipe;
 808                     int numreaders;
 809                     int length = msg->iofs->io_Union.io_READ_WRITE.io_Length;
 810                     D(bug("[xpipe] length is %d.\n", length));
 811
 812                     if (!(writer->mode & FMF_WRITE))
 813                     {
 814                         D(bug("[xpipe] User tried to write to the wrong end of the pipe.\n"));
 815                         SendBack (msg, ERROR_WRITE_PROTECTED);
 816                         break;
 817                     }
 818
 819                     ListLength (&pipe->readers, numreaders);
 820                     if (numreaders == 0)
 821                     {
 822                         D(bug("[xpipe] There are no open read ends: PIPE BROKEN.\n"));
 823                         SendBack (msg, ERROR_BROKEN_PIPE);
 824                         break;
 825                     }
 826
 827                     if (WriteWouldBlock (pipe, length))
 828                     {
 829                         if(writer->mode & FMF_NONBLOCK)
 830                         {
 831                             D(bug("[xpipe] There is not enough space and the pipe is in nonblocking mode, so return EWOULDBLOCK\n"));
 832                             SendBack (msg, ERROR_WOULD_BLOCK);
 833                             break;
 834                         }
 835
 836                         /* Write as much as we can, enqueue the request and reply when it's finished */
 837                         D(bug("[xpipe] Enqueing the message\n"));
 838                         AddTail (&pipe->pendingwrites, (struct Node *)msg);
 839                     }
 840
 841                     msg->curlen = WriteToPipe (
 842                         pipe,
 843                         msg->iofs->io_Union.io_READ_WRITE.io_Buffer,
 844                         length
 845                     );
 846                     D(bug("[xpipe] Wrote %d bytes from requested %d\n", msg->curlen, length));
 847
 848                     if(length == msg->curlen)
 849                     {
 850                         /* Managed to write everything */
 851                         SendBack(msg, 0);
 852                     }
 853
 854                     if(msg->curlen != 0)
 855                         pump(pipe, PROCESS_READS);
 856
 857                     break;
 858                 }
 859                 case FSA_READ:
 860                 {
 861                     D(bug("[xpipe] Cmd is FSA_READ.\n"));
 862
 863                     if(pn->ln_Type != NT_PIPEEND)
 864                     {
 865                         SendBack (msg, ERROR_OBJECT_NOT_FOUND);
 866                         break;
 867                     }
 868                     struct XPipeEnd *reader = (struct XPipeEnd *) pn;
 869
 870                     D(bug("[xpipe] Reader end %p\n", reader));
 871                     struct XPipe *pipe = reader->pipe;
 872                     int numwriters;
 873                     ListLength(&pipe->writers, numwriters);
 874                     int length = msg->iofs->io_Union.io_READ_WRITE.io_Length;
 875                     D(bug("[xpipe] length is %d.\n", length));
 876
 877                     if (!(reader->mode & FMF_READ))
 878                     {
 879                         D(bug("[xpipe] User tried to read from the wrong end of the pipe.\n"));
 880                         SendBack(msg, ERROR_READ_PROTECTED);
 881                         break;
 882                     }
 883
 884                     if (ReadWouldBlock (pipe, length))
 885                     {
 886                         if(reader->mode & FMF_NONBLOCK)
 887                         {
 888                             D(bug("[xpipe] There is not enough data to read and the pipe is in nonblocking mode, so return EWOULDBLOCK\n"));
 889                             SendBack(msg, ERROR_WOULD_BLOCK);
 890                             break;
 891                         }
 892
 893                         /* Read as much as we can, enqueue the request and reply when it's finished */
 894                         D(bug("[xpipe] Enqueing the message\n"));
 895                         AddTail (&pipe->pendingreads, (struct Node *)msg);
 896                     }
 897
 898                     msg->curlen = ReadFromPipe (
 899                         pipe,
 900                         (APTR) msg->iofs->io_Union.io_READ_WRITE.io_Buffer,
 901                         length
 902                     );
 903                     D(bug("[xpipe] Read %d bytes\n", msg->curlen));
 904
 905                     if(length == msg->curlen || numwriters == 0)
 906                     {
 907                         /* Managed to read everything or there are no more writers (EOF) */
 908                         msg->iofs->io_Union.io_READ_WRITE.io_Length = msg->curlen;
 909                         SendBack (msg, 0);
 910                     }
 911
 912                     if(msg->curlen != 0)
 913                         pump(pipe, PROCESS_WRITES);
 914
 915                     break;
 916                 }
 917             }
 918         }
 919     } while (cont);
 920
 921     return 0;
 922
 923     AROS_USERFUNC_EXIT
 924 }