<<>> <> <> <> <> DIRECTORY Basics, XCommunications, UXStrings, IO, Rope; XCommunicationsTCPPCedarMonitored: CEDAR MONITOR LOCKS d USING d: REF MyDataRec IMPORTS Basics, IO, XCommunications--, UXStrings ~ BEGIN STREAM: TYPE = IO.STREAM; <> <> <> <> <> --C-- HostEnt: TYPE = MACHINE DEPENDENT RECORD [ --see chapter 9.2, page 233 hName: UXStrings.UnixString, --official name of host hAliases: POINTER TO PACKED ARRAY [0..0) OF UXStrings.UnixString, --alias list hAddrtype: AFTypes, --host address type (e.g. AFŹINET) hLength: INT32, --length af address in bytes hAddrList: POINTER TO PACKED ARRAY [0..0) OF POINTER TO Basics.RawBytes --list of addresses, list null terminated; each address is hLength bytes long ]; --C-- PortNumber: TYPE = MACHINE DEPENDENT RECORD [val: Basics.HWORD]; --C-- InAddr: TYPE = PACKED ARRAY [0..4) OF BYTE; --internet address --C-- SockAddrIn: TYPE = MACHINE DEPENDENT RECORD [ --see chapter 8.7, page 203 ... sinFamily: AFTypes16, sinPort: PortNumber, sinAddr: InAddr, sinZero: PACKED ARRAY [0..8) OF BYTE ]; --C-- SockAddr: TYPE = MACHINE DEPENDENT RECORD [ SELECT OVERLAID * FROM in => [in: SockAddrIn], ENDCASE ]; IncludeUIO: PROC = TRUSTED MACHINE CODE { """/usr/cedar/threads/xr/threads/UIO.h""." }; IncludeTypes: PROC = TRUSTED MACHINE CODE { "." }; IncludeSocket: PROC = TRUSTED MACHINE CODE { "." }; IncludeIn: PROC = TRUSTED MACHINE CODE { "." }; IncludeNetDB: PROC = TRUSTED MACHINE CODE { "." }; IncludeSTDIO: PROC = TRUSTED MACHINE CODE { "." }; AFTypes16: TYPE = INT16; AFTypes: TYPE = INT32; SockTypes: TYPE = INT32; AfInet: PROC RETURNS [AFTypes] = TRUSTED MACHINE CODE { "@AF_INET" }; SockStream: PROC RETURNS [SockTypes] = TRUSTED MACHINE CODE { "@SOCK_STREAM" }; CreateSocket: PROC [domain: AFTypes, type: SockTypes, protocol: INT32 Ź 0] RETURNS [INT32] = TRUSTED MACHINE CODE { "XR_USocket" }; <> <> <<"gethostbyname">> <<};>> <

> <<};>> <<>> Connect: PROC [socket: INT32, server: POINTER TO SockAddr, bytes: INT32] RETURNS [errCode: INT] = TRUSTED MACHINE CODE { "XR_UConnect" }; Close: PROC [socket: INT] = TRUSTED MACHINE CODE { "XR_UClose" }; Write: PROC [socket: INT32, buf: POINTER, nbytes: INT32] RETURNS [bytesWritten: INT] = TRUSTED MACHINE CODE { "XR_UWrite" }; Read: PROC [socket: INT32, buf: POINTER, nbytes: INT32] RETURNS [errorCode: INT] = TRUSTED MACHINE CODE { "XR_URead" }; XrUDup: PROC [socket: INT32] RETURNS [INT] = TRUSTED MACHINE CODE { "XR_UDup" }; Recv: PROC [socket: INT32, buf: POINTER, nbytes: INT32, flags: INT32 Ź 0] RETURNS [errorCode: INT] = TRUSTED MACHINE CODE { "XR_URecv" }; myStreamProcs: REF IO.StreamProcs Ź IO.CreateStreamProcs[ variety: inputOutput, class: $MyTCP, getChar: MyGetChar, endOf: MyEndOf, unsafeGetBlock: MyUnsafeGetBlock, putChar: MyPutChar, unsafePutBlock: MyUnsafePutBlock, flush: SendNow, close: MyClose ]; bufferSize: INT = 2000; bufferSizeM1: INT = 1999; MyDataRec: TYPE = MONITORED RECORD [ socket1: INT, --writer... socket2: INT, --reader... closed: BOOL Ź FALSE, bufCnt: INT Ź 0, server: SockAddrIn, hp: POINTER TO HostEnt, outBuffer: PACKED ARRAY [0..bufferSize) OF CHAR ]; MyEndOf: PROC [self: STREAM] RETURNS [BOOL] = { RETURN [FALSE]; }; MyPutChar: PROC [self: STREAM, char: CHAR] = { d: REF MyDataRec = NARROW[self.streamData]; IF d.closed THEN ERROR IO.Error[StreamClosed, self]; EntryPutChar[d, char]; }; MyUnsafePutBlock: PROC [self: STREAM, block: Basics.UnsafeBlock] = { d: REF MyDataRec = NARROW[self.streamData]; IF d.closed THEN ERROR IO.Error[StreamClosed, self]; EntryUnsafePutBlock[d, block]; }; MoveBytes: UNSAFE PROC [dest: POINTER, src: POINTER, len: CARDINAL] ~ UNCHECKED MACHINE CODE{ "XR_MoveBytesDisjoint" }; EntryPutChar: ENTRY PROC [d: REF MyDataRec, char: CHAR] = { ENABLE UNWIND => NULL; IF d.bufCnt>=bufferSize THEN InternalFlush[d]; d.outBuffer[d.bufCnt] Ź char; d.bufCnt Ź d.bufCnt+1; IF d.bufCnt>=bufferSize THEN InternalFlush[d]; }; EntryUnsafePutBlock: ENTRY PROC [d: REF MyDataRec, block: Basics.UnsafeBlock] = TRUSTED { ENABLE UNWIND => NULL; IF d.bufCnt>=bufferSize THEN InternalFlush[d]; DO cnt: INT Ź MIN[bufferSize-d.bufCnt, block.count]; MoveBytes[@d.outBuffer+d.bufCnt, block.base+block.startIndex, cnt]; d.bufCnt Ź d.bufCnt + cnt; block.startIndex Ź block.startIndex + cnt; block.count Ź block.count - cnt; IF block.count>0 THEN InternalFlush[d] ELSE EXIT ENDLOOP; IF d.bufCnt>=bufferSize THEN InternalFlush[d]; }; MyGetChar: PROC [self: STREAM] RETURNS [ch: CHAR] = TRUSTED { d: REF MyDataRec = NARROW[self.streamData]; n: INT; buff: PACKED ARRAY [0..BYTES[WORD]) OF CHAR; IF d.closed THEN ERROR IO.Error[StreamClosed, self]; n Ź Read[d.socket2, @buff, 1]; IF n = 0 THEN ERROR IO.EndOfStream[self]; IF n < 0 THEN ERROR; RETURN [buff[0]]; }; MyUnsafeGetBlock: UNSAFE PROC [self: STREAM, block: Basics.UnsafeBlock] RETURNS [nBytesRead: INT Ź 0] = { d: REF MyDataRec = NARROW[self.streamData]; IF d.closed THEN ERROR IO.Error[StreamClosed, self]; WHILE block.count>0 DO n: INT Ź Read[d.socket2, LOOPHOLE[block.base+block.startIndex+nBytesRead], block.count]; IF n<=0 THEN { IF n=0 THEN RETURN; ERROR; }; block.count Ź block.count-n; nBytesRead Ź nBytesRead+n; ENDLOOP }; MyClose: PROC [self: STREAM, abort: BOOL] = { d: REF MyDataRec = NARROW[self.streamData]; IF d.closed THEN RETURN; IF ~abort THEN SendNow[self]; d.closed Ź TRUE; Close[d.socket1]; Close[d.socket2]; }; InternalFlush: INTERNAL PROC [d: REF MyDataRec] = TRUSTED { addr: POINTER Ź @d.outBuffer; cnt: INT Ź d.bufCnt; d.bufCnt Ź 0; WHILE cnt>0 DO n: INT Ź Write[d.socket1, addr, cnt]; IF n<=0 THEN ERROR; cnt Ź cnt-n; addr Ź addr+n; ENDLOOP }; EntryFlush: ENTRY PROC [d: REF MyDataRec] = { ENABLE UNWIND => NULL; InternalFlush[d]; }; SendNow: PROC [self: STREAM] = { d: REF MyDataRec = NARROW[self.streamData]; EntryFlush[d]; }; Create: PROC [base: Rope.ROPE, port: REF ANY] RETURNS [sd: XCommunications.StreamData] = TRUSTED { portNo: INT; d: REF MyDataRec Ź NEW[MyDataRec]; WITH port SELECT FROM ri: REF INT => portNo Ź ri; ENDCASE => {sd.errorMsg Ź "bad port type"; GOTO Oops}; <<--create socket>> d.socket1 Ź CreateSocket[AfInet[], SockStream[]]; IF d.socket1<0 THEN {sd.errorMsg Ź "socket not created"; GOTO Oops}; <<--bind socket to name>> d.server.sinFamily Ź AfInet[]; <> <> <> <> d.server.sinAddr Ź [13, 1, 100, 140]; --clarissa d.server.sinPort Ź [Basics.HFromInt16[portNo]]; <<--connect socket with remote socket>> IF Connect[d.socket1, LOOPHOLE[@d.server], BYTES[SockAddrIn]] < 0 THEN { sd.errorMsg Ź "not connected"; GOTO Oops }; <<>> d.socket2 Ź XrUDup[d.socket1]; IF d.socket2=0 THEN { Close[d.socket1]; sd.errorMsg Ź "not connected: duplication of socket failed"; GOTO Oops }; <<>> <<--create stream>> sd.in Ź sd.out Ź IO.CreateStream[streamProcs: myStreamProcs, streamData: d]; sd.success Ź TRUE; sd.errorFromStream Ź ErrorFromStream; EXITS Oops => {} }; ErrorFromStream: PROC [s: IO.STREAM] RETURNS [reason: Rope.ROPE] = { reason Ź "failure"; }; <<--get include files>> IncludeUIO[]; IncludeSocket[]; IncludeIn[]; IncludeNetDB[]; IncludeSTDIO[]; XCommunications.RegisterCommunication[[create: Create, protocol: $TCP]]; END.