;       SCCSID = @(#)pipe.asm   6.1 90/11/17
;      SCCSID = @(#)pipe.asm    1.14 90/09/17
        PAGE    ,132
TITLE   Pipes - CPDOS interprocess I/O streams
NAME    PIPE
.286p

;***    pipe.asm - MTDOS interprocess I/O streams
;
;       IBM/Microsoft Confidential
;
;       Copyright (c) IBM Corporation 1987, 1989
;       Copyright (c) Microsoft Corporation 1987, 1989
;
;       All Rights Reserved
;
;       This file contains the source to implement
;       pipes, a form of interprocess communication
;       styled after XENIX pipes.
;
;       A 'pipe' is a connection between (among) file
;       handles (JFN's).  Data written to the 'write end'
;       of the pipe are made available for reading on the
;       'read end'.  The $Pipe system call creates a pipe
;       and returns two file handles, one for the read end
;       and one for the write end.  These handles are manipulated
;       in the same way as normal file handles; they may be
;       'dup'ed and are inherited in the same way.
;       Data are written into a pipe via a 'write' system call
;       on the write end of the pipe.  Likewise, data are
;       read from the pipe via a 'read' call on the read end.
;
;       Data that are written to a pipe are captured in a
;       circular buffer.  The size of the buffer is specified
;       when the pipe is created; if no size is specified,
;       a default size is used.
;
;       The circular buffer is described by an 'ioblock'.
;       The ioblock is the buffer's header; the circular
;       buffer proper follows the ioblock in a heap memory
;       object (mapped by a GDT selector) allocated when the
;       pipe is created.  The ioblock contains all of the
;       per-pipe information, such as reader, writer, and
;       reference counts, and also holds the pointers into
;       the circular buffer proper.
;
;       When the in and out pointers are equal, the circular buffer
;       is empty.  When the in pointer trails the out pointer
;       by 1, the buffer is full.  Thus, a 512 byte buffer can hold
;       only 511 bytes; one byte is lost so that full and empty
;       conditions can be distinguished. So that the user can
;       put 512 bytes in a pipe that they created with a size
;       of 512, we allow for this byte lost when allocating
;       the segment.
;
;       w_MakePipe
;       palloc
;       aps
;       ClosePipe
;       pfree
;       ReadPipe
;       readiob
;       rcb
;       WritePipe
;       writeiob
;       wcb
;       SBlk
;       SWait
;       SSig
;
;       8/10/88 jsf     increment pipe buffer size to match user expectation
;Defect 105267 11/17/94  ChangeTeam - Roll defect 89305 into r207 sparse tree.
;                        Removed SFLinkFree call from ClosePipe.

.xlist
        include dosseg.inc
        include basemaca.inc
        include osmaca.inc
        include oldmaca.inc
        include mi.inc
        include defdbug.inc
        include dossymc.inc
        include error.inc
        include filemode.inc
        include vmowner.inc
        include seldesc.inc
        include virtsym.inc
        include sft.inc
        include tcbdata.inc
        include fsc.inc
        include osconst.inc
        include pgexport.inc
        include selxport.inc
        include vmexport.inc
.list

        ModName pipe

;*      Extern functions called from this file

        EXTRNFAR        ProcBlock               ; block this process
        EXTRNFAR        ProcRun                 ; unblock some process
        EXTRNFAR        SFFromHandle

        EXTRN32         _SELAllocKHB
        EXTRN32         _SELFreeKHB

        extrn           JFNFindFree:near
        extrn           SFFromSFN:near
        extrn           SFLinkFree:near
        extrn           pJFNFromHandle:near
        extrn           SFNAlloc:near
        extrn           SuBuff:near
        extrn           FuBuff:near

;*      External Per-Task variables referenced in this file

        ReferTask       CurrTCB,WORD            ; current thread control block
        ReferTask       JFN_Flg_Ptr, DWORD      ; pointer to per file
                                                ;       descriptor flags

;*      ioblock - structure that describes a circular buffer

ioblock struc
        io_inprogcnt    db      ?               ; count of read/wrts in progress
        io_refcnt       db      ?               ; count of references
        io_rdrcnt       db      ?               ; count of readers
        io_wtrcnt       db      ?               ; count of writers
        io_selector     dw      ?               ; buffer selector
        io_first        dw      ?               ; ptr to base of circular buffer
        io_in           dw      ?               ; ptr to next free byte
        io_out          dw      ?               ; ptr to next byte of data
        io_last         dw      ?               ; ptr to end+1 of buffer
        io_rdlksem      dw      ?               ; read lock semaphore
        io_wtlksem      dw      ?               ; write lock semaphore
        io_rdsem        dw      ?               ; read sync semaphore
        io_wtsem        dw      ?               ; write sync semaphore
ioblock ends

;*      Local manifest constants

DEFSIZE equ 512                                 ; default buffer size

BUFFERHEAP      equ     VM_HKH_PUB_RESRW

          FARCODE HIGH2
HIGH2CODE segment
    assume  ss:TASKAREA

BREAK   <w_MakePipe - Create a new Pipe>

;***    w_MakePipe - set up an anonymous pipe
;
;       This routine handles the 'pipe()' system call
;
;       We need to set up two jfns, each pointing to its own sft.
;       Both sft's point to the pipe structure.
;
;       Because this is an anonymous pipe, we need make no MFT
;       entry for it.  Likewise, it will never have more than
;       two sft's pointing to it.
;
;       ENTRY:  (cx) = size of circular buffer in bytes
;
;       EXIT:   via either sys_ret_ok, or sys_ret_err
;               sys_ret_ok:
;                       (ax) = read handle (aka JFN)
;                       (bx) = write handle (aka JFN)
;               sys_ret_err:
;                       (ax) = error code
;       USES:   ALL
;
        ASSUME  DS:NOTHING, ES:NOTHING

Procedure w_MakePipe,hybrid
        PMONLY

;       allocate pipe segment

        DBPRT   40h,40h,<w_MakePipe entered, PTDA=$x\n>,ss

        call    palloc                  ; (dx) = selector for header
        LJC     per0                    ; couldn't get memory

;       allocate reader sft

                                        ; mode in ax for reader
        mov     ax,open_for_read OR sharing_deny_both
        call    aps                     ; allocate and initialize sft
        LJC     per1                    ; no free sft

;       allocate reader JFN

        call    JFNFindFree             ; (bx) = JFN, (es:di) = ptr to slot
        jc      per2                    ; no free jfn
        push    bx                      ; save reader JFN on stack
        mov     es:[di],cx              ; JFN now points to the reader sft
        les     di,[JFN_Flg_Ptr]        ; get pointer to flags
        mov     byte ptr es:[di+bx],0   ; initialize flags

;       allocate writer SFT

                                        ; mode in ax for writer
        mov     ax,open_for_write OR sharing_deny_both

        call    aps                     ; allocate and initialize sft
        jc      per3                    ; no free sft

;       allocate writer JFN

        call    JFNFindFree             ; (bx) = JFN, (es:di) = ptr to slot
        jc      per4                    ; no free jfn
        mov     es:[di],cx              ; JFN now points to the writer sft
        les     di,[JFN_Flg_Ptr]        ; get pointer to flags
        mov     byte ptr es:[di+bx],0   ; initialize flags

;       all set to return successfully

                                        ; (bx) = write handle
        pop     ax                      ; (ax) = read handle
        DBPRT   40h,40h,<w_MakePipe success rd $x wr $x\n>,<ax,bx>

;       MMIOPH Performance Hook Removed
;       Maj/Min = 18E4, Data = bx, ax (write handle, read handle)

        ret                             ; return

;*      error returns - free any already allocate resources, return
;
;       writer SFT, reader JFN, reader SFT, pipe are allocated
;        (cx)  = writer SFN,
;        (tos) = reader JFN
;        (dx)  = handle to pipe segment
;        (ax) = error code

per4:   mov     bx,cx                   ; (bx) = writer SFN
        call    SFFromSFN               ; (es:di) = ptr to writer SFT
        dec     es:[di].sf_ref_count    ; zero ref count - free the sft
        call    SFLinkFree              ; put SFT on free list
                                        ; fall thru to free rest of stuff

;       reader JFN, reader SFT, pipe, are allocated
;        (tos) = reader JFN
;        (dx)  = handle to pipe segment
;        (ax) = error code

per3:   pop     bx                      ; (bx) = reader JFN
        push    ax                      ; preserve error code
        call    pJFNFromHandle          ; (es:di) = ptr to JFN slot
        pop     ax                      ; restore error code
        mov     cx,SFN_NULL
        xchg    cx,es:[di]              ; (cx) = reader SFN, free the JFN

;       reader SFT, pipe, are allocated
;        (cx)  = reader SFN
;        (ax) = error code

per2:   mov     bx,cx                   ; (bx) = reader SFN
        call    SFFromSFN               ; (es:di) = ptr to reader SFT
        dec     es:[di].sf_ref_count    ; zero ref count - free the sft
        call    SFLinkFree              ; put SFT on free list

;       pipe segment is still allocated
;        (dx) = selector of pipe segment
;        (ax) = error code

per1:   push    ax                      ; preserve error code
        mov     ax,dx                   ; ax = selector for pipe segment
        call    pfree                   ; free the pipe
        pop     ax                      ; restore error code

;       nothing is allocated, return error

per0:   DBPRT   40h,40h,<w_MakePipe failure ptda $x\n>,ss
        stc                             ; indicate error
        ret                             ; . . . and return

EndProc   w_MakePipe


;***    palloc - allocate and initialize a pipe segment
;
;       We allocate the pipe segment from the kernel heap.
;       We also initialize the pipe structure at the head of it
;       to describe the buffer, and to contain reference counts, etc.
;
;       ENTRY:  (cx) = size in bytes of circular buffer
;
;               if cx=0 then size defaults to "DEFSIZE" bytes
;
;       EXIT:   'c' clear:
;                       (dx) = segment selector, segment is initialized
;               'c' set:
;                       (ax) = error code
;
;       USES:   ALL
;
        ASSUME  DS:NOTHING, ES:NOTHING

Procedure palloc,near,local
        LocalVar        pal_ac,<size vmac_s>

        EnterProc
        or      cx,cx                   ; size 0?
        jnz     pal0                    ; no, it's ok
        mov     cx,DEFSIZE              ; yes, use default size

;       allocate the header and buffer

pal0:   mov     bx,cx

;       Allow for lost byte in circular buffer when allocating segment.

        add     bx,size ioblock +1      ; allocate header and buffer together
        jnc     pal1                    ; size < 64k
;
; Buffer requested is too big. Make the segent allocated 64K-1 so that the
; comparisons for end of buffer in wcb and rcb work correctly.
;
        mov     bx,0FFFFH               ; force to 64k-1; AX:BX = 0x0000FFFF

;       _SELAllocKHB(BUFFERHEAP,bx,0,0,FSNMPIPEOWNER,pAC)

pal1:   CpuMode 386
        FlatContext <ds,es>,ax
        SSToDS  ecx,[pal_ac]
        movzx   ebx,bx
        CCALL32 _SELAllocKHB,<BUFFERHEAP,ebx,0,0,FSPIPEOWNER,ecx>
        ASSUME  DS:NOTHING,ES:NOTHING
        or      eax,eax                 ; clears carry
        CpuMode reset
        jz      short pal2
        mov     ax,ERROR_NOT_ENOUGH_MEMORY
        stc
        jmp     short pal3
pal2:
        xor     cx,cx
        mov     ax,[pal_ac].ac_sel      ; ax = selector for buffer
        mov     es,ax                   ; (es:0) = ptr to header
        mov     es:[io_selector],ax     ; save buffer selector in ioblock
        mov     es:[io_rdrcnt],1        ; set reader count
        mov     es:[io_wtrcnt],1        ; set writer count
        mov     es:[io_refcnt],2        ; set reference count
        mov     es:[io_inprogcnt],cl    ; clear in progress counter
        mov     dx,ax                   ; (dx) = segment selector
        mov     ax,size ioblock         ; size of header
        mov     es:[io_first],ax        ; initialize first byte available
        mov     es:[io_in],ax           ; initialize first byte available
        mov     es:[io_out],ax          ; no data available
        mov     es:[io_last],bx         ; set the pointer to the end of buffer
        mov     es:[io_rdlksem],cx
        mov     es:[io_wtlksem],cx
        mov     es:[io_rdsem],cx
        mov     es:[io_wtsem],cx        ; zero-init sempahore fields

pal3:   LeaveProc
        ret

EndProc   palloc


;***    aps - Allocate Pipe SFN
;
;       Aps allocates and initializes an SFN for a pipe.
;
;       ENTRY   (ax) = sf_fsi.sfi_mode value
;               (dx) = pipe header selector
;
;       EXIT    'C' set if error
;                 (AX) = error code
;               'C' clear if OK
;                 (cx) = sfn
;                 sf_fsi.sfi_hVPB = (dx)
;                 sf_fsi.sfi_mode = (al)
;                 sf_flags = SF_PIPE
;                 sf_ref_count = 1
;                 sf_MFT  = SF_MFT_NULL
;                 (cx) = SFN
;
;       USES    AX, BX, CX, DI, DS, ES, Flags
;
        ASSUME  DS:NOTHING, ES:NOTHING

Procedure aps,near,local

        call    SFNAlloc                ; (bx) = sfn, (es:di) = ptr to sft
        jc      aps1                    ; error then exit
        mov     es:[di].sf_fsi.sfi_mode,ax      ; set access mode in sft entry
        mov     es:[di].sf_fsi.sfi_hVPB,dx      ; set handle to pipe in sft entry
        mov     es:[di].sf_flags,SF_PIPE; mark sft as pipe
        mov     es:[di].sf_ref_count,1  ; set the reference count

        CpuMode 386
        mov     es:[di].sf_MFT,SF_MFT_NULL; clear MFT entry
        CpuMode reset

; force FSD_ procedures to dispatch pipe to DOS_ procedures

        mov     es:[di].sf_FSC.Offst,FSC_LOCAL
        mov     es:[di].sf_FSC.Segmt,0
        mov     cx,bx                   ; (cx) = sfn
aps1:   ret

EndProc   aps



;***    ClosePipe - close an SFT that points to a pipe
;
;       Close the SFT, which points to a pipe.  We need to decrement
;       reference counts, and if the pipe is no longer referenced,
;       we need to free it.
;
;       If all the pipe handles have been closed and there still is
;       a read or write in progress, the actual removal of the pipe
;       header/buffer is delayed until the all the reads and writes
;       finish.
;
;       Pipes do read/write serialization or read/write in progress
;       processing themselves and don't and can't use the SFT
;       serialization.
;
;       ENTRY:  (ES:DI) = ptr to SFT
;
;       EXIT:   Reference counts adjusted as needed
;
;       USES:   ALL
;
        ASSUME  DS:NOTHING, ES:NOTHING
Procedure ClosePipe,hybrid
        PMONLY
        DBPRT   40h,40h,<Closepipe- ptda $x sft $x:$x\n>,<ss,es,di>

        push    es
        pop     ds                      ; (DS:DI) = SFT pointer
        cmp     ds:[di].sf_ref_count,0
        jz      clpx                    ; if sft is already freed

        dec     ds:[di].sf_ref_count    ; one fewer JFN's point at this SFT
        jnz     clpx                    ; still at least one ref to SFT

;       This SFT no longer has any references; we need to
;       adjust the pipe structure to reflect the fact that
;       this SFT is now free.

        DBPRT   40h,80h,<Closepipe- freeing SFT\n>,<>

        mov     ax,ds:[di].sf_fsi.sfi_hVPB      ; (ax) = pipe segment selector
;89305  mov     ds:[di].sf_fsi.sfi_hVPB,0       ; clear field so no confusion
;89305  push    ds
;89305  pop     es                      ; (ES:DI) -> SFT
;89305  call    SFLinkFree              ; put SFT back on free list
        mov     es,ax                   ; (ES:0) = ptr to pipe struct

;       Adjust the reference count, free seg if no more references and
;       no one is doing a read or write of a pipe.

        DBPRT   40h,80h,<Closepipe- header at $x\n>,es
        dec     es:[io_refcnt]
        jnz     clp1                    ; still at least one reference to pipe

        cmp     es:[io_inprogcnt],0     ; check for any reads or writes in
        jne     clpx                    ;  progress. Jump if there are any.

;       Free pipe header and buffer.

        DBPRT   40h,80h,<Closepipe- freeing header\n>,<>

;       (AX) = pipe buffer selector

        call    pfree                   ; free the header and buffer
        jmp     short clpx              ; EXIT

;       Adjust reader and writer counts

clp1:   mov     bx,ds:[di].sf_fsi.sfi_mode      ; (BX) = file access mode
        .errnz  open_for_read - 0       ; make sure code below works
        .errnz  open_for_write - 1
        .errnz  open_for_both - 2
        and     bl,open_access          ; strip off sharing info
        inc     bl                      ; please see block comment below

;*      Decrement reader and writer counts
;
;       the open modes are:
;               0 = read-only
;               1 = write only
;               2 = both read and write
;
;       By incrementing this value, we get one bit that indicates
;       whether or not the SFT is open for read, another bit
;       that indicates if it is open for write
;
;       bx = 1 -> open read-only
;       bx = 2 -> open write-only
;       bx = 3 -> open read/write
;       N.B. bit zero set if open for reading
;            bit one set if open for writing

;       reader count first

        shr     bx,1
        jnc     clp2                    ; not open for reading
        dec     es:[io_rdrcnt]          ; open for reading, decrement reader cnt
        jnz     clp2                    ; at least one reader left

        push    bx                      ; we were the last reader
        mov     bx,offset io_wtsem      ; so we must wake up writer
        call    SSig                    ; so he can get SIG_PIPE
        pop     bx

;       now the writer count

clp2:   shr     bl,1
        jnc     clpx                    ; not open for writing
        dec     es:[io_wtrcnt]          ; open for writing, 1 fewer writers
        jnz     clpx                    ; at least one writer left

        push    bx                      ; we were last writer
        mov     bx,offset io_rdsem      ; we should wake up
        call    SSig                    ; reader so he can get EOF
        pop     bx

clpx:   DBPRT   40h,40h,<Closepipe return\n>,<>
        clc
        ret

EndProc     ClosePipe



;***    pfree - free the memory associated with a pipe
;
;       Free the circular buffer and its header.
;
;       ENTRY:  ax = selector of buffer/header
;
;       EXIT:   memory for pipe is freed
;
;       USES:   EAX,ECX,EDX,DS,ES,FLAGS
;
Procedure pfree,near,local
        ASSUME  DS:NOTHING, ES:NOTHING

; _SELFreeKHB(BUFFERHEAP,0,selector)

        CpuMode 386
        movzx   eax,ax                  ; eax has selector to free
        FlatContext <ds,es>,cx,noassume
        CCALL32 _SELFreeKHB,<BUFFERHEAP,0,eax>
        CpuMode reset
        ret

EndProc     pfree


BREAK   <Reading from pipes>
;***    ReadPipe - read data from a pipe
;
;       Most of work is getting ready for a call to readiob,
;       which does the work of getting the data from the
;       circular buffer.
;
;       We lock the pipe for reading; this prevents simultaneous
;       reads of more than the buffer length from 'interspersing'
;       data among the reads.  This interleaving might also
;       occur if the reads are short but block when the buffer
;       becomes empty.
;
;       Reads on empty pipes with no writers return EOF (0 bytes read)
;       Reads on empty pipes with writers block until data are present.
;
;       A read can return fewer bytes than requested if:
;         * there are less than the requested bytes available
;         * the read is interrupted by a signal
;         * there are no writers on the pipe
;
;       If all the pipe handles have been closed and there still is
;       a read or write in progress, the actual removal of the pipe
;       header/buffer is delayed until the all the reads and writes
;       finish.
;
;       Pipes do read/write serialization or read/write in progress
;       processing themselves and don't and can't use the SFT
;       serialization.
;
;       ENTRY:  (es:di) = ptr to SFT
;               [ss:dmaadd] = ptr to data destination
;               (cx) = byte count
;
;       EXIT:   (cx) = number of bytes read
;
;       USES:   ALL but ds
;
        ASSUME  DS:NOTHING, ES:NOTHING
Procedure ReadPipe,faronly
        PMONLY
        DBPRT   40h,40h,<ReadPipe entered ptda $x\n>,ss
        push    ds                              ; preserve DS

        mov     es,es:[di].sf_fsi.sfi_hVPB      ; (es:0) -> pipe segment
        inc     es:[io_inprogcnt]               ; indicate someone reading
        mov     bx,offset io_rdlksem            ; (es:bx) = ptr to read lock sem
        xor     dh,dh                           ; (dh) = interrruptible
        call    SWait                           ; entering critical section
        mov     dx,cx                           ; (dx) = original byte count
        jc      rdp1                            ; jump if interrupted block

        push    es
        pop     ds                              ; (ds) = segment of pipe header
        sub     si,si                           ; (ds:si) = ptr to pipe header
        mov     di,CurrTCB
        mov     bx,ss:[di].TCBDMAAdd.Segmt      ; (bx:di) = ptr to dest of data
        mov     di,ss:[di].TCBDMAAdd.Offst
        push    dx                              ; save original byte count
        call    readiob                         ; get the data
        pop     dx                              ; (dx) = original byte count

        push    ds                              ; (tos) = segment of pipe header
        pop     es                              ; (es) = segment of header

        push    ax                              ; preserve possible error code
        pushf                                   ; preserve flags

        mov     bx,offset io_rdlksem            ; (es:bx) = ptr to read lock sem
        call    SSig                            ; leaving critical section

        popf                                    ; restore flags (ints are on)
        pop     ax                              ; restore error code

rdp1:   push    ax                              ; preserve possible error code
        pushf                                   ; preserve flags

        sub     cx,dx
        neg     cx                              ; (cx) = # of bytes read

        dec     es:[io_inprogcnt]               ; done with read
        jnz     rdp3
        cmp     es:[io_refcnt],0                ; does this pipe header need
        jnz     rdp3                            ;  to be removed?

        mov     ax,es:[io_selector]             ; (AX) = pipe buffer selector
        push    cx
        call    pfree
        pop     cx

rdp3:   popf                                    ; restore flags (ints are on)
        pop     ax                              ; restore error code

        pop     ds
        DBPRT   40h,40h,<ReadPipe return count $x ptda $x\n>,<ss,cx>
        ret

EndProc   ReadPipe



;***    readiob - read from an ioblock
;
;       ENTRY:  (ds:si) = ptr to struct ioblock (in fixed or locked segment)
;               (bx:di) = ptr to dest of data
;               (cx) = byte count
;
;       EXIT:   (cx) = residual byte count
;               carry set = interrupted read
;
;       USES:   ALL
;
        ASSUME  DS:NOTHING, ES:NOTHING

Procedure readiob,near,local

;       transfer as much as we can without blocking
;       (bx:di) = ptr to dest of data
;       (ds:si) = ptr to ioblock
;       (cx) = byte count

ri1:    push    cx                      ; (tos) requested byte count
        call    rcb                     ; (cx) = number of bytes remaining
        jc      rie                     ; rcb failed

;       wake up any waiting writer

        SaveReg <bx>
        push    ds
        pop     es
        lea     bx,[si].io_wtsem        ; (es:bx) = offset of wtr sem
        call    SSig                    ; wake up writer that may be waiting
        RestoreReg <bx>                 ; (bx) = dest segment

        pop     ax                      ; requested number of bytes
        sub     ax,cx                   ; (ax) = bytes read
        jnz     rix                     ; if we read some we return

;       if no bytes were available in the buffer . . .
;       we must wait for more data

        cmp     ds:[si].io_wtrcnt,0
        jz      rix                     ; no writers => no more data forever

        DBPRT   40h,40h,<ri - sleeping for data\n>,<>
        SaveReg <bx>                    ; save dest segment
        lea     bx,[si].io_rdsem
        push    ds
        pop     es                      ; (es:bx) = ptr to read semaphore
        xor     dh,dh                   ; interruptible
        call    SBlk                    ; wait for more data
        DBPRT   40h,40h,<readiob - got awakened\n>,<>
        push    es
        pop     ds                      ; (ds:0) = ptr to ioblock
        RestoreReg <bx>                 ; (bx) = dest segment
        jnc     ri1                     ; go try again

;       Interrupted block.

rix:    ret

;*  Fault accessing user segment
rie:    pop     cx
        jmp     rix

EndProc   readiob



;***    rcb - circular buffer read
;
;       We read as much data out of the circular buffer as we can
;       without blocking.  In any case, we never transfer more
;       bytes than requested.
;
;       ENTRY:  (ds:si) = ptr to ioblock
;               (bx:di) = ptr to dest of data
;               (cx) = byte count
;
;       EXIT:   (ds:si) = ptr to ioblock
;               (bx:di) = adjusted dest ptr
;               (cx) = residual byte count
;
;       USES:   ALL
;
        ASSUME  DS:NOTHING, ES:NOTHING

Procedure rcb,near,local

        DBPRT   40h,80h,<rcb - entered\n>,<>
        mov     dx,cx                   ; (bx) = byte count
rcb0:   mov     cx,ds:[si].io_in
        sub     cx,ds:[si].io_out
        jz      rcb5                    ; buffer empty, go return
        jnc     rcb2                    ; can transfer to in ptr
        mov     cx,ds:[si].io_last
        sub     cx,ds:[si].io_out       ; can only xfer to end of buffer

rcb2:   cmp     cx,dx
        jbe     rcb3                    ; xfer len smaller than byte count
        mov     cx,dx                   ; don't xfer more byte count

rcb3:   push    si                      ; (tos) = ioblock ptr
        mov     si,ds:[si].io_out       ; (si) = offset part of buffer ptr
                                        ; (ds:si) = ptr to source
                                        ; (es:di) = ptr to dest for copy
                                        ; (cx) = byte count for this copy
        sub     dx,cx                   ; (dx) = bytes remaining after copy
        DBPRT   40h,80h,<rcb - xfer len $x\n>,cx
        SaveReg <bx,ds>
        call    SuBuff                  ; (ds:si) is adjusted
                                        ; (bx:di) is adjusted
                                        ; (cx) = zero
        RestoreReg <ds,bx>
        mov     cx,si                   ; (cx) = offset of new out ptr
        pop     si                      ;
        jc      rcb5                    ; SuBuff failed, DS = ES = 0

        cmp     cx,ds:[si].io_last
        jb      rcb4                    ; didn't hit end of buffer
        mov     cx,ds:[si].io_first     ; hit end, pointer wraps around

rcb4:   mov     ds:[si].io_out,cx       ; save the new pointer
        or      dx,dx
        jnz     rcb0                    ; more bytes to xfer

rcb5:   mov     cx,dx                   ; (cx) = residual byte count
        DBPRT   40h,80h,<rcb exit - $x bytes remaining\n>,cx
        ret

EndProc   rcb


BREAK   <Writing into pipes>
;***    WritePipe - write data into a pipe
;
;       Most of work is getting ready for a call to writeiob,
;       which does the work of getting the data into the
;       circular buffer.
;
;       We lock the pipe for writing; this prevents simultaneous
;       writes of more than the buffer length from 'interspersing'
;       data among the writes.  This interleaving might also
;       occur if the writes are short but block when the buffer
;       becomes full.
;
;       A signal will interrupt a writer waiting for space or while
;       is waiting for the critical section.
;
;       Writes on pipes with no readers return error ERROR_BROKEN_PIPE.
;       Writes on pipes with readers but no data block until space is free.
;
;       A write may return with less than the requested number of bytes
;       written if:
;         * the call is interrupted by a signal
;         * there are no readers for the pipe
;
;       If all the pipe handles have been closed and there still is
;       a read or write in progress, the actual removal of the pipe
;       header/buffer is delayed until the all the reads and writes
;       finish.
;
;       Pipes do read/write serialization or read/write in progress
;       processing themselves and don't and can't use the SFT
;       serialization.
;
;       ENTRY:  (es:di) = ptr to SFT
;               ss:[dmaadd] = source pointer
;               (cx) = byte count
;
;       EXIT:   (cx) = # of bytes written
;
;       USES:   ALL but ds
;
        ASSUME  DS:NOTHING, ES:NOTHING
Procedure WritePipe,faronly
        PMONLY
        DBPRT   40h, 40h, <Writepipe entered ptda $x\n>, ss
        push    ds                              ; preserve DS

        mov     es,es:[di].sf_fsi.sfi_hVPB      ; (es:0) -> pipe segment
        inc     es:[io_inprogcnt]               ; indicate someone writing

        mov     bx,offset io_wtlksem            ; (es:bx) = ptr to write lock
        xor     dh,dh                           ; (dh) = interruptible
        call    SWait                           ; entering critical section
        mov     dx,cx                           ; (dx) = original byte count
        jc      wtp1                            ; jump if interrupted blocked

        push    es
        pop     ds
        xor     si,si                           ; (ds:si) = ptr to pipe header
        mov     di,CurrTCB
        mov     bx,ss:[di].TCBDMAAdd.Segmt      ; (bx:di) = ptr to src of data
        mov     di,ss:[di].TCBDMAAdd.Offst
        push    dx                              ; save original byte count
        call    writeiob                        ; get the data
        pop     dx                              ; (dx) = original byte count

        push    ds                              ; (tos) = segment of header
        pop     es                              ; (es) = segment of header

        push    ax                              ; preserve error code
        pushf                                   ;  and flags

        mov     bx,offset io_wtlksem            ; (es:bx) = ptr to write lock
        call    SSig                            ; leaving critical section

        popf                                    ; restore flags (ints are on)
        pop     ax                              ;  and error code.

wtp1:   push    ax                              ; preserve error code
        pushf                                   ;  and flags.

        sub     cx,dx
        neg     cx                              ; (cx) = # of bytes written

        dec     es:[io_inprogcnt]               ; done with write
        jnz     wtp3
        cmp     es:[io_refcnt],0                ; does this pipe header need
        jnz     wtp3                            ;  to be removed?

        mov     ax,es:[io_selector]             ; (AX) = pipe header selector
        push    cx
        call    pfree
        pop     cx
wtp3:   popf                                    ; restore flags (ints are on)
        pop     ax                              ; and error code

        pop     ds
        DBPRT   40h,40h,<WritePipe return count $x ptda $x\n>,<cx,ss>
        ret

EndProc   WritePipe


;***    writeiob - write to an ioblock
;
;       ENTRY:  (ds:si) = ptr to struct ioblock (in fixed or locked segment)
;               (bx:di) = ptr to src of data
;               (cx) = byte count
;
;       EXIT:   (cx) = residual byte count
;               'C' set iff error
;
;       USES:   ALL
;
        ASSUME  DS:NOTHING, ES:NOTHING

Procedure writeiob,near,local

wi0:    cmp     ds:[si].io_rdrcnt,0
        jz      wie                     ; no readers, go force error

;       transfer as much data as we can without blocking
;       (bx:di) = ptr to src of data
;       (ds:si) = ptr to ioblock
;       (cx) = byte count

        call    wcb                     ; (cx) = number of bytes remaining
        jc      wix                     ;   Error

;       we've put in some data, so notify any waiting readers

        SaveReg <bx>
        push    ds
        pop     es
        lea     bx,[si].io_rdsem        ; (es:bx) = ptr to rdr sem
        call    SSig                    ; wake up reader who may be waiting
        RestoreReg <bx>                 ; (bx) = src segment
        clc                             ; Indicate success, for exit path

        jcxz    wix                     ; no more to write, we're done

;       wait for space

        DBPRT   40h,40h,<wi - sleeping for space\n>,<>
        SaveReg <bx>
        lea     bx,[si].io_wtsem
        push    ds
        pop     es                      ; (es:bx) = ptr to write semaphore
        xor     dh,dh                   ; interruptible
        call    SBlk                    ; wait for space
        DBPRT   40h,40h,<wi - awakened\n>,<>
        push    es
        pop     ds                      ; (ds:0) = ptr to ioblock
        RestoreReg <bx>                 ; (bx) = src segment
        jnc     wi0                     ; SBlk returned normally

wix:    ret

;*  return 'pipe broken' code to this proc

wie:    mov     ax,ERROR_BROKEN_PIPE    ; set error code
        stc                             ; indicate error
        jmp     wix

EndProc   writeiob



;***    wcb - circular buffer write
;
;       We write as much data into the circular buffer as we can
;       without blocking.  In any case, we never transfer more
;       bytes than requested.
;
;       ENTRY:  (ds:si) = ptr to ioblock
;               (bx:di) = ptr to src of data
;               (cx) = byte count
;
;       EXIT:   (ds:si) = ptr to ioblock
;               (bx:di) = adjusted src ptr
;               (cx) = residual byte count
;
;       USES:   ALL
;
        ASSUME  DS:NOTHING, ES:NOTHING

Procedure wcb,near,local

        DBPRT   40h,80h,<wcb - entered\n>, <>

;       exchange src and dest pointers
;
;       (ES) on input is don't care; (BX) has user selector

        mov     ax,ds
        mov     es,ax
        xchg    si,di

        mov     bp,cx                   ; (bp) = byte count

;*      calculate length of transfer
;       there are two cases:
;       1. in ptr is < out ptr
;               in this case, we can only put (out - in) - 1 bytes in
;       2. in ptr >= out ptr
;               In this case, we can only xfer to end of buffer.
;               If in == out == first, we must leave last byte
;               empty.

wcb0:   mov     cx,es:[di].io_in
        mov     dx,es:[di].io_out
        sub     cx,dx                   ; (cx) = -(out - in)
        jb      wcb1                    ; in < out, can only go to (in - 1)
        add     cx,dx                   ; undo subtraction (cx) = in
        sub     cx,es:[di].io_last      ;
        cmp     dx,es:[di].io_first
        jne     wcb2

wcb1:   inc     cx                      ; 'subtract' from negated count

wcb2:   neg     cx                      ; (cx) = transfer count

        jcxz    wcb5                    ; no room, return
        cmp     cx,bp
        jbe     wcb3                    ; xfer len smaller than request
        mov     cx,bp                   ; don't xfer more than requested

;       (es:di) = ptr to ioblock
;       (cx) = byte count for this xfer
;       (bx:si) = src ptr
;       (ax) = segment of circular buffer

wcb3:   push    di

        mov     di,es:[di].io_in        ; si = in ptr
                                        ; (es:di) = dest ptr for copy
                                        ; (cx) = byte count for copy
        sub     bp,cx                   ; (bp) = bytes remaining after copy
        DBPRT   40h,80h,<wcb - xfer len $x\n>,cx
        SaveReg <bx,es>
        call    FuBuff                  ; (bx:si) is adjusted
                                        ; (es:di) is adjusted
                                        ; (cx) = zero
        RestoreReg <es,bx>
        mov     cx,di                   ; (cx) = offset part of new out ptr
        pop     di                      ; (es:di) = ptr to ioblock
        jc      wcb5                    ; FuBuff failed, DS = ES = 0

        cmp     cx,es:[di].io_last
        jb      wcb4                    ; didn't hit end of buffer
        mov     cx,es:[di].io_first     ; hit end, handle wrap around

wcb4:   mov     es:[di].io_in,cx        ; and save the new pointer
        or      bp,bp
        jnz     wcb0                    ; more bytes to xfer

wcb5:   mov     cx,bp                   ; (cx) = byte count
        push    ds
        push    es
        pop     ds                      ; (ds) = ioblock
        pop     es                      ; (es) = 0
        xchg    si,di
        DBPRT   40h,80h,<wcb exit - $x bytes remaining\n>,cx
        ret

EndProc   wcb

BREAK   <Semaphore Subroutines>

;***    SBlk - set a semaphore and wait on it
;
;       ENTRY:  (es:bx) = ptr to semaphore word
;               (dh) = 1 if not interruptble, 0 otherwise
;
;       EXIT:   thru SWait
;
;       USES:   AX, flags, I-flag
;
        ASSUME  DS:NOTHING, ES:NOTHING

Procedure SBlk,near,local

        mov     byte ptr es:[bx],1              ; set the flag
IF2
        .errnz  (SWait - $)                     ; fall thru into SWait
ENDIF
EndProc   SBlk

;**     SWait - Wait for a semaphore
;
;       This routine
;       operates on a semaphore made up of a pair of bytes- an "in
;       use" flag and a "someone is wait" flag.
;
;       DB      busyflag
;       DB      waitingflag
;
;       If busyflag is clear then
;               set busyflag
;               return
;       else
;               set waitingflag (to notify owner to do a wakeup when he's done)
;               ProcBlock on address of busyflag
;
;       Note that when ProcBlock returns interrupts will be enabled.
;       Since the entire procblock interval was an "interrupts enabled"
;       window we figure we must be called with interrupts on and
;       thus don't mess with PUSHF/popf but just jam the interrupts
;       on and off as necessary.  (In keeping with good practice we
;       keep them on as much as possible.)
;
;       ENTRY   (ES:BX) = address of semaphore word
;               (dh) = 1 if not interruptble, 0 otherwise
;
;       EXIT    Carry clear
;                   When semaphore owned by caller
;               semaphore set
;               enables interrupts
;               Carry set
;                   (AX) = error code
;                               ERROR_INTERRUPT
;                   ProcBlock interrupted
;                   Interrupts enabled
;
;       USES    AX, I-flag, Flags
;
        ASSUME  DS:NOTHING, ES:NOTHING

Procedure SWait,near,local

;       retest semaphore
;
;       (es:bx) = address
;       (TOS) = caller's AX

swa0:   sti
        mov     al,1
        xchg    al,BYTE PTR es:[bx]     ; set semaphore

;       If busyflag was set then its still set and (al) = 1
;       If busyflag was clear then its set now and (al) = 0

        or      al,al                   ; (OR clears carry)
        jz      swa1                    ; done  - we've got it

;       The thing is set.  We're going to have to sleep on it
;       We have to do it under "cli" so that we don't have a race with
;       the semaphore owner releasing it while we're blocking
;
;       (al) = 1

        cli
        test    byte ptr es:[bx],0ffh
        jz      swa0                            ; just unlocked!  try again
        mov     byte ptr es:1[bx],al            ; set waiting

        mov     ax,es:[io_selector]             ; AX:BX -> semaphore to block.

        SaveReg <cx,di>
        mov     cx,-1                           ; cx:di = -1 no timeout
        mov     di,cx
        CALLFAR ProcBlock                       ; wait till semaphore released
        RestoreReg <di,cx>

        jnc     swa0                            ; loop if normal awakening

;       ProcBlock was interrupted.  Return with carry set.

        mov     ax,ERROR_INTERRUPT
swa1:   ret
EndProc   SWait


;**     SSig - Signal a semaphore
;
;       SSig releases a semaphore (it is presumed that the caller indeed
;       owns it!) and, if anyone was waiting, wakes them up.
;
;       ENTRY   (ES:BX) = address of semaphore word (see SWait, above)
;
;       EXIT    NONE
;
;       USES    AX,Flags
;
        ASSUME  DS:NOTHING, ES:NOTHING

Procedure SSig,near,local

        xor     ax,ax
        xchg    ax,es:[bx]              ; clear semaphore and wait flag

;       (al) = semaphore flag
;       (ah) = wait flag

        and     ah,ah
        jz      ssig2                   ; nobody waiting, don't ProcRun

;       Someone is waiting on this semaphore.  Wake him/them up.
;
;       (ES:BX) = semaphore address = block/run code value

        mov     ax,es:[io_selector]     ; AX:BX -> semaphore to block.
        CALLFAR ProcRun                 ; let 'em run.

ssig2:  ret

EndProc   SSig

HIGH2CODE        ends


        FARCODE INIT
INITCODE segment

;***    QueryPipe - Query Pipe size
;
;       See if a pipe is empty
;
;       ENTRY:  (BX) = pipe handle
;
;       EXIT:   AX =  0 if pipe empty
;               AX != 0 if pipe not empty
;       USES:   ALL
;
        ASSUME  DS:NOTHING, ES:NOTHING

Procedure QueryPipe,near
        PMONLY

        CALLFAR SFFromHandle                    ; get the sft
        jc      qpip1
        mov     es,es:[di].sf_fsi.sfi_hVPB      ; (es:0) -> pipe segment
        mov     ax,es:[io_in]
        sub     ax,es:[io_out]
        stc
qpip1:  ret
EndProc     QueryPipe
INITCODE ends
        END
