Index: ChangeLog =================================================================== RCS file: /home/igarashi/kl1/cvsroot/klic/ChangeLog,v retrieving revision 1.1.1.1 retrieving revision 1.1.1.1.4.1 diff -u -r1.1.1.1 -r1.1.1.1.4.1 --- ChangeLog 2000/07/28 02:01:40 1.1.1.1 +++ ChangeLog 2000/10/11 07:38:38 1.1.1.1.4.1 @@ -1,3 +1,12 @@ +Wed Oct 11 16:33:41 2000 Yusuke TAKAGI + + * runtime/termio.kl1: support prolog-like IO for socket stream. + +Wed Oct 11 16:31:13 2000 Hiroshi IGARASHI + + * runtime/asyncio.c, runtime/gunix.kl1, runtime/signal.c: + support full-duplex IO for socket stream. + Sat Mar 27 11:14:18 1999 Sekita Daigo * macro.kl1: fix the bug in replaceClase for the :- Dec form clause. Index: runtime/asyncio.c =================================================================== RCS file: /home/igarashi/kl1/cvsroot/klic/runtime/asyncio.c,v retrieving revision 1.1.1.1 retrieving revision 1.1.1.1.4.1 diff -u -r1.1.1.1 -r1.1.1.1.4.1 --- runtime/asyncio.c 2000/07/28 02:01:41 1.1.1.1 +++ runtime/asyncio.c 2000/10/11 07:38:38 1.1.1.1.4.1 @@ -263,6 +263,7 @@ { #ifdef USESIG add_sigio_handler(fd, 0, KLIC_SIGIO_NONE); + asyncio_streams[fd] = 0; #endif } void register_asynchronous_io_stream(fd, stream) Index: runtime/gunix.kl1 =================================================================== RCS file: /home/igarashi/kl1/cvsroot/klic/runtime/gunix.kl1,v retrieving revision 1.1.1.1 retrieving revision 1.1.1.1.4.1 diff -u -r1.1.1.1 -r1.1.1.1.4.1 --- runtime/gunix.kl1 2000/07/28 02:01:41 1.1.1.1 +++ runtime/gunix.kl1 2000/10/11 07:38:38 1.1.1.1.4.1 @@ -81,8 +81,6 @@ MakeFilePointer(F, file); \\ } -static char buffer[4096]; - struct iobuf { unsigned char *ptr, *lim, *buf; int fd; @@ -105,6 +103,33 @@ return iob; } +struct biobuf { + struct iobuf ibuf; + struct iobuf obuf; +}; + +static struct biobuf *make_biobuf(ifd, ofd, size) + int ifd, ofd, size; +{ + struct biobuf *biob = (struct biobuf *)malloc_check(sizeof(struct biobuf)); + struct iobuf *ibuf = &(biob->ibuf); + struct iobuf *obuf = &(biob->obuf); + + ibuf->fd = ifd; + ibuf->bufsize = size; + ibuf->buf = (unsigned char *)malloc_check(ibuf->bufsize); + ibuf->ptr = ibuf->buf; + ibuf->lim = ibuf->buf; + + obuf->fd = ofd; + obuf->bufsize = size; + obuf->buf = (unsigned char *)malloc_check(obuf->bufsize); + obuf->ptr = obuf->buf; + obuf->lim = obuf->buf+size; + + return biob; +} + #define MakeInBuf(klicvar, fd, size) \\ { \\ klicvar = gd_new_pointer((q)make_iobuf(fd,size,0), allocp); \\ @@ -120,6 +145,15 @@ #define IOBuf(x) \\ ((struct iobuf*)((struct pointer_object *)(data_objectp(x)))->pointer) +#define MakeBIOBuf(klicvar, ifd, ofd, size) \\ +{ \\ + klicvar = gd_new_pointer((q)make_biobuf(ifd,ofd,size), allocp); \\ + allocp = heapp; \\ +} + +#define BIOBuf(x) \\ + ((struct biobuf*)((struct pointer_object *)(data_objectp(x)))->pointer) + static int fill_buf(iob) struct iobuf *iob; { @@ -128,6 +162,10 @@ if (result < 0) { switch (errno) { case EINTR: continue; +#ifdef EPIPE + case EPIPE: + return -2; +#endif /* EPIPE */ #ifdef ASYNCIO case EAGAIN: #ifdef EWOULDBLOCK @@ -159,6 +197,10 @@ if (result < 0) { switch (errno) { case EINTR: continue; +#ifdef EPIPE + case EPIPE: + return -2; +#endif /* EPIPE */ #ifdef ASYNCIO case EAGAIN: #ifdef EWOULDBLOCK @@ -203,7 +245,7 @@ /* End of Patch */ #endif default: - fatalp(\"write\", \"Error in asynchronous input\"); + fatalp(\"write\", \"Error in asynchronous output\"); } } else { wp += result; @@ -259,6 +301,7 @@ unix([update_open(P,R)|Cmnds]) :- update_open(P,R0), cont(R0,R,Cmnds). unix([signal_stream(S,R)|Cmnds]):- signal_stream(S,R0), cont(R0,R,Cmnds). unix([connect(A,R)|Cmnds]) :- connect(A,R0), cont(R0,R,Cmnds). +unix([connect2(A,R)|Cmnds]) :- connect2(A,R0), cont(R0,R,Cmnds). unix([bind(A,R)|Cmnds]) :- bind(A,R0), cont(R0,R,Cmnds). unix([pipe(R)|Cmnds]) :- pipe(R0), cont(R0,R,Cmnds). unix([system(S,R)|Cmnds]) :- system(S,R0), cont(R0,R,Cmnds). @@ -488,6 +531,72 @@ connect_sub(FD,Addr,Async,[],R) :- connect_sub(FD,Addr,Async,R). +connect2(Spec,R) :- + net_convert(Spec,Fam,Addr), + connect2(Fam,Addr,_Async,R). + +connect2(Fam,Addr,Async,R) :- inline:" +{ +#ifndef NO_USESOCKET + int family = intval(%0); + int sock = socket(family, SOCK_STREAM, 0); + + if (sock < 0) { fatalp(\"socket\", \"Socket creation error\"); } +#ifdef ASYNCIO + init_asynchronous_io(); + setasync(sock, \"connection\"); + register_asynchronous_io_stream(sock, %1); +#endif + %2 = makeint(sock); +#else + goto %f; +#endif +}":[Fam+int,Async+any,FD-int] | + connect2_sub(FD,Addr,Async,R). + +connect2_sub(FD,Addr,Async,R) :- inline:" +{ +#ifndef NO_USESOCKET + int sock = intval(%0); + struct sockaddr *addr = (struct sockaddr*) + ((struct pointer_object *)(data_objectp(%1)))->pointer; + + again2: + if (connect(sock, addr, sizeof(struct sockaddr)) < 0) { +#ifdef ASYNCIO + if (errno == EINTR) goto again2; + if (errno != EISCONN) { + if (errno == EINPROGRESS || errno == EALREADY) goto %f; +#endif + fatalp(\"connect\", \"Socket connection error\"); +#ifdef ASYNCIO + } +#endif + } + free(addr); + MakeBIOBuf(%2, sock, sock, 4096); +#else + goto %f; +#endif +}":[FD+int,Addr+object(pointer),IOB-object(pointer)] | + R=normal(SIn,SOut), + async_input(SIn,Async,IOB,0), + async_output(SOut,Async,IOB,0). +otherwise. +connect2_sub(FD,Addr,Async,R) :- inline:"{ +#ifdef NO_USE_SOCKET + goto %f; +#endif +}":[] | + timer:instantiate_after(time(0,3,0), Timer), + connect2_sub(FD, Addr, Async, Timer, R). + +connect2_sub(FD,Addr,[_|Async],_,R) :- + connect2_sub(FD,Addr,Async,R). +otherwise. +connect2_sub(FD,Addr,Async,[],R) :- + connect2_sub(FD,Addr,Async,R). + bind(unix(Path),R) :- net_convert(unix(Path),Fam,Addr), bind(Fam,Addr,_Async,R). @@ -590,6 +699,37 @@ bound_sock(S,FD,[_|Async],Async1) :- bound_sock(S,FD,Async,Async1). +bound_sock([accept2(R)|S],FD,Async,Async1) :- inline:" +{ +#ifndef NO_USESOCKET + int sock; + struct sockaddr addr; + int socklen = sizeof(addr); +#ifdef ASYNCIO + if (!poll_read_available(intval(%0))) goto %f; +#endif + while (1) { + if ((sock = accept(intval(%0), &addr, &socklen)) > 0) break; + if (errno != EINTR) { + fatalp(\"accept\", \"Error in accept\"); + } + } +#ifdef ASYNCIO + setasync(sock, \"accepted socket\"); + register_asynchronous_io_stream(sock,%1); +#endif + MakeBIOBuf(%2, sock, sock, 4096); +#else + goto %f; +#endif +}":[FD+int,Async1+any,IOB-object(pointer)] | + R=normal(SIn,SOut), + async_input(SIn,Async1,IOB,0), + async_output(SOut,Async1,IOB,0), + bound_sock(S,FD,Async,_). +bound_sock(S,FD,[_|Async],Async1) :- + bound_sock(S,FD,Async,Async1). + pipe(R) :- inline:" { int fd[2]; @@ -650,6 +790,7 @@ if (iob->ptr == iob->lim) { switch (fill_buf(iob)) { case -1: goto %f; + case -2: goto %f; case 0: c = -1; break; case 1: c = *iob->ptr++; break; } @@ -674,7 +815,8 @@ q str; if (ready_bytes==0) { switch (fill_buf(iob)) { - case -1: + case -1: goto %f; + case -2: goto %f; case 0: goto %f; case 1: break; } @@ -713,6 +855,7 @@ if (iob->ptr == iob->lim) { switch (fill_buf(iob)) { case -1: goto %f; + case -2: goto %f; case 0: %1 = makeint(1); break; case 1: %1 = makeint(0); break; } @@ -789,6 +932,252 @@ alternatively. async_io(S,[_|Async],InB,OutB,LC) :- async_io(S,Async,InB,OutB,LC). + +async_input([],_Async,IOB,_LC) :- inline:" +{ +#ifdef ASYNCIO + struct biobuf *biob = BIOBuf(%0); + struct iobuf *inb = &(biob->ibuf); + struct iobuf *outb = &(biob->obuf); + if (inb->fd != outb->fd) { + if (close(inb->fd) != 0) { + fatalp(\"close\", \"Error in closing asynchronous input\"); + } + close_asynchronous_io_stream(inb->fd); + } + if (outb->fd == -1) { + free(inb->buf); + free(outb->buf); + } + inb->fd = -1; +#else + goto %f; +#endif +}":[IOB+object(pointer)] | true. +async_input([getc(C)|S],Async,IOB,LC0) :- inline:" +{ +#ifdef ASYNCIO + struct biobuf *biob = BIOBuf(%0); + struct iobuf *iob = &(biob->ibuf); + int c; + if (iob->ptr == iob->lim) { + switch (fill_buf(iob)) { + case -1: goto %f; + case -2: goto %f; + case 0: c = -1; break; + case 1: c = *iob->ptr++; break; + } + } else { + c = *iob->ptr++; + } + %3 = makeint(intval(%2) + ( c=='\\n' ? 1 : 0 )); + %1 = makeint(c); +#else + goto %f; +#endif +}":[IOB+object(pointer),C0-int,LC0+int,LC-int] | + C=C0, + async_input(S,Async,IOB,LC). +async_input([fread(N,R)|S],Async,IOB,LC0) :- inline:" +{ +#ifdef ASYNCIO + struct biobuf *biob = BIOBuf(%0); + struct iobuf *iob = &(biob->ibuf); + int toread = intval(%1); + int k, nnl; + int ready_bytes = iob->lim - iob->ptr; + q str; + if (ready_bytes==0) { + switch (fill_buf(iob)) { + case -1: goto %f; + case -2: goto %f; + case 0: goto %f; + case 1: break; + } + } + if (toread > iob->lim - iob->ptr) toread = iob->lim - iob->ptr; + if ((char *)allocp+ + sizeof(struct byte_string_object)+toread+sizeof(long) >= + (char *)real_heaplimit) { + allocp = real_heaplimit; + goto async__input_4_ext_interrupt; + } + for (k=0, nnl=0; kptr[k] == '\\n') nnl++; + } + str = BC2KLIC(iob->ptr, toread, allocp); + if (isref(str)) { + fatal(\"internal error: string allocation failure for fread\"); + } + %2 = str; + allocp = heapp; + iob->ptr += toread; + %4 = makeint(intval(%3)+nnl); +#else + goto %f; +#endif +}":[IOB+object(pointer),N+int,R0-object(byte_string),LC0+int,LC-int] | + R=R0, + async_input(S,Async,IOB,LC). +async_input([linecount(N)|S],Async,IOB,LC) :- + N=LC, + async_input(S,Async,IOB,LC). +async_input([feof(R)|S],Async,IOB,LC) :- inline:" +{ +#ifdef ASYNCIO + struct biobuf *biob = BIOBuf(%0); + struct iobuf *iob = &(biob->ibuf); + if (iob->ptr == iob->lim) { + switch (fill_buf(iob)) { + case -1: goto %f; + case -2: goto %f; + case 0: %1 = makeint(1); break; + case 1: %1 = makeint(0); break; + } + } else { + %1 = makeint(0); + } +#else + goto %f; +#endif +}":[IOB+object(pointer),R0-int] | + R=R0, + async_input(S,Async,IOB,LC). +alternatively. +async_input(S,[_|Async],IOB,LC) :- + async_input(S,Async,IOB,LC). + +async_output([],_Async,IOB,_LC) :- inline:" +{ +#ifdef ASYNCIO + struct biobuf *biob = BIOBuf(%0); + struct iobuf *inb = &(biob->ibuf); + struct iobuf *outb = &(biob->obuf); + if (outb->ptr != outb->buf) { + switch (write_buf(outb)) { + case -1: goto %f; + case -2: break; + case 1: break; + } + } + if (inb->fd != outb->fd) { + if (close(outb->fd) != 0) { + fatalp(\"close\", \"Error in closing asynchronous output\"); + } + close_asynchronous_io_stream(outb->fd); + } + if (inb->fd == -1) { + free(inb->buf); + free(outb->buf); + } + outb->fd = -1; +#else + goto %f; +#endif +}":[IOB+object(pointer)] | true. +async_output([C|S],Async,IOB,LC) :- integer(C), inline:" +{ +#ifdef ASYNCIO + struct biobuf *biob = BIOBuf(%0); + struct iobuf *iob = &(biob->obuf); + if (iob->ptr == iob->lim) { + switch (write_buf(iob)) { + case -1: goto %f; + case -2: goto %f; + case 1: break; + } + } + *iob->ptr++ = intval(%1); +#else + goto %f; +#endif +}":[IOB+object(pointer),C+int] | + async_output(S,Async,IOB,LC). +async_output([linecount(N)|S],Async,IOB,LC) :- + N=LC, + async_output(S,Async,IOB,LC). +async_output([putc(C)|S],Async,IOB,LC) :- inline:" +#ifndef ASYNCIO +#endif +":[] | + async_output([C|S],Async,IOB,LC). +async_output([fwrite(X,R)|S],Async,IOB,LC) :- string(X,L,8), inline:" +{ +#ifdef ASYNCIO + struct biobuf *biob = BIOBuf(%0); + struct iobuf *iob = &(biob->obuf); + char *str = KLIC2C(%1); + int len =intval(%2); + int room = iob->lim - iob->ptr; + while (iob->ptr + len >= iob->lim) { + BCOPY(str, iob->ptr, room); + len -= room; + str += room; + iob->ptr += room; + switch (write_buf(iob)) { + case -1: goto %f; + case -2: goto %f; + case 1: break; + } + } + BCOPY(str, iob->ptr, len); + iob->ptr += len; +#else + goto %f; +#endif +}":[IOB+object(pointer),X+object(byte_string),L+int] | + R=L, + async_output(S,Async,IOB,LC). +otherwise. +async_output([fwrite(X,R)|S],Async,IOB,LC) :- true | + R = -1, + async_output(S,Async,IOB,LC). +async_output([fwrite(X)|S],Async,IOB,LC) :- + async_output([fwrite(X,_)|S],Async,IOB,LC). +async_output([fflush(R)|S],Async,IOB,LC) :- inline:" +{ +#ifdef ASYNCIO + struct biobuf *biob = BIOBuf(%0); + struct iobuf *iob = &(biob->obuf); + if (iob->ptr != iob->buf) { + switch (write_buf(iob)) { + case -1: goto %f; + case -2: goto %f; + case 1: break; + } + } +#else + goto %f; +#endif +}":[IOB+object(pointer)] | + R = 0, + async_output(S,Async,IOB,LC). +otherwise. +async_output([fflush(R)|S],Async,IOB,LC) :- true | + R = -1, + async_output(S,Async,IOB,LC). +async_output([sync(R)|S],Async,IOB,LC) :- inline:" +#ifdef ASYNCIO +{ + struct biobuf *biob = BIOBuf(%0); + struct iobuf *outb = &(biob->obuf); + switch (write_buf(outb)) { + case -1: goto %f; + case -2: goto %f; + case 1: break; + } +} +#endif +":[IOB+object(pointer)] | + R=0, + async_output(S,Async,IOB,LC). +otherwise. +async_output([sync(R)|S],Async,IOB,LC) :- true | + R = -1, + async_output(S,Async,IOB,LC). +alternatively. +async_output(S,[_|Async],IOB,LC) :- + async_output(S,Async,IOB,LC). /**************************************** SYSTEM CALLS, etc @@ -905,12 +1294,12 @@ In1-object(pointer),Out1-object(pointer), In2-object(pointer),Out2-object(pointer)] | ( - PID=:=0 -> + PID=\=0 -> R=parent(PID,In,Out), generic:new(file_io,In,In1,"pipe-input",[],""), generic:new(file_io,Out,[],"",Out2,"pipe-output") ; - PID=\=0 -> + PID=:=0 -> R=child(In,Out), generic:new(file_io,In,In2,"pipe-input",[],""), generic:new(file_io,Out,[],"",Out1,"pipe-output") Index: runtime/signal.c =================================================================== RCS file: /home/igarashi/kl1/cvsroot/klic/runtime/signal.c,v retrieving revision 1.1.1.1 retrieving revision 1.1.1.1.4.1 diff -u -r1.1.1.1 -r1.1.1.1.4.1 --- runtime/signal.c 2000/07/28 02:01:41 1.1.1.1 +++ runtime/signal.c 2000/10/11 07:38:38 1.1.1.1.4.1 @@ -190,6 +190,18 @@ register_gc_hook(gc_signal_streams); } +/**************************************** + DEFAULT SIGPIPE HANDLER +****************************************/ + +static int default_sigpipe_handler(allocp, sig) + q *allocp; + int sig; +{ + declare_globals; + return 0; +} + /* GENERAL INITIATION */ @@ -199,6 +211,7 @@ init_general_signal_handling(); init_streamed_signal_handling(); add_signal_handler(SIGINT, default_sigint_handler); + add_signal_handler(SIGPIPE, default_sigpipe_handler); } #endif Index: runtime/termio.kl1 =================================================================== RCS file: /home/igarashi/kl1/cvsroot/klic/runtime/termio.kl1,v retrieving revision 1.1.1.1 retrieving revision 1.1.1.1.4.1 diff -u -r1.1.1.1 -r1.1.1.1.4.1 --- runtime/termio.kl1 2000/07/28 02:01:41 1.1.1.1 +++ runtime/termio.kl1 2000/10/11 07:38:38 1.1.1.1.4.1 @@ -29,6 +29,10 @@ send_unix(update_open(Path,RU),U,UT), termupdt(RU,R), klicio(S,UT). klicio([connect(What,R)|S],U) :- send_unix(connect(What,RU),U,UT), termupdt(RU,R), klicio(S,UT). +klicio([connect2(What,R)|S],U) :- + send_unix(connect2(What,RU),U,UT), termupdt(RU,R), klicio(S,UT). +klicio([bind(What,R)|S],U) :- + send_unix(bind(What,RU),U,UT), bind(RU,R), klicio(S,UT). otherwise. klicio([Msg|S],U) :- send_unix(Msg,U,UT), klicio(S,UT). @@ -59,6 +63,18 @@ updt(SI,SU,s(Ops,0)). otherwise. termupdt(R0,R) :- R=R0. + +bind(normal(SU),R) :- + R=normal(SI), + bound_sock(SI,SU). +otherwise. +bind(R0,R) :- R=R0. + +bound_sock([],U) :- U=[]. +bound_sock([accept(R)|S],U) :- + U=[accept(RU)|UT], termupdt(RU,R), bound_sock(S,UT). +bound_sock([accept2(R)|S],U) :- + U=[accept2(RU)|UT], termupdt(RU,R), bound_sock(S,UT). in([],U,_Ops) :- U=[]. in([gett(Term)|S],U,Stat) :-