=================================================================== RCS file: /home/cvs/OpenXM_contrib2/asir2000/io/ox.c,v retrieving revision 1.1.1.1 retrieving revision 1.28 diff -u -p -r1.1.1.1 -r1.28 --- OpenXM_contrib2/asir2000/io/ox.c 1999/12/03 07:39:11 1.1.1.1 +++ OpenXM_contrib2/asir2000/io/ox.c 2009/02/13 11:25:48 1.28 @@ -1,25 +1,70 @@ -/* $OpenXM: OpenXM/src/asir99/io/ox.c,v 1.7 1999/11/19 04:15:49 noro Exp $ */ +/* + * Copyright (c) 1994-2000 FUJITSU LABORATORIES LIMITED + * All rights reserved. + * + * FUJITSU LABORATORIES LIMITED ("FLL") hereby grants you a limited, + * non-exclusive and royalty-free license to use, copy, modify and + * redistribute, solely for non-commercial and non-profit purposes, the + * computer program, "Risa/Asir" ("SOFTWARE"), subject to the terms and + * conditions of this Agreement. For the avoidance of doubt, you acquire + * only a limited right to use the SOFTWARE hereunder, and FLL or any + * third party developer retains all rights, including but not limited to + * copyrights, in and to the SOFTWARE. + * + * (1) FLL does not grant you a license in any way for commercial + * purposes. You may use the SOFTWARE only for non-commercial and + * non-profit purposes only, such as academic, research and internal + * business use. + * (2) The SOFTWARE is protected by the Copyright Law of Japan and + * international copyright treaties. If you make copies of the SOFTWARE, + * with or without modification, as permitted hereunder, you shall affix + * to all such copies of the SOFTWARE the above copyright notice. + * (3) An explicit reference to this SOFTWARE and its copyright owner + * shall be made on your publication or presentation in any form of the + * results obtained by use of the SOFTWARE. + * (4) In the event that you modify the SOFTWARE, you shall notify FLL by + * e-mail at risa-admin@sec.flab.fujitsu.co.jp of the detailed specification + * for such modification or the source code of the modified part of the + * SOFTWARE. + * + * THE SOFTWARE IS PROVIDED AS IS WITHOUT ANY WARRANTY OF ANY KIND. FLL + * MAKES ABSOLUTELY NO WARRANTIES, EXPRESSED, IMPLIED OR STATUTORY, AND + * EXPRESSLY DISCLAIMS ANY IMPLIED WARRANTY OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE OR NONINFRINGEMENT OF THIRD PARTIES' + * RIGHTS. NO FLL DEALER, AGENT, EMPLOYEES IS AUTHORIZED TO MAKE ANY + * MODIFICATIONS, EXTENSIONS, OR ADDITIONS TO THIS WARRANTY. + * UNDER NO CIRCUMSTANCES AND UNDER NO LEGAL THEORY, TORT, CONTRACT, + * OR OTHERWISE, SHALL FLL BE LIABLE TO YOU OR ANY OTHER PERSON FOR ANY + * DIRECT, INDIRECT, SPECIAL, INCIDENTAL, PUNITIVE OR CONSEQUENTIAL + * DAMAGES OF ANY CHARACTER, INCLUDING, WITHOUT LIMITATION, DAMAGES + * ARISING OUT OF OR RELATING TO THE SOFTWARE OR THIS AGREEMENT, DAMAGES + * FOR LOSS OF GOODWILL, WORK STOPPAGE, OR LOSS OF DATA, OR FOR ANY + * DAMAGES, EVEN IF FLL SHALL HAVE BEEN INFORMED OF THE POSSIBILITY OF + * SUCH DAMAGES, OR FOR ANY CLAIM BY ANY OTHER PARTY. EVEN IF A PART + * OF THE SOFTWARE HAS BEEN DEVELOPED BY A THIRD PARTY, THE THIRD PARTY + * DEVELOPER SHALL HAVE NO LIABILITY IN CONNECTION WITH THE USE, + * PERFORMANCE OR NON-PERFORMANCE OF THE SOFTWARE. + * $OpenXM: OpenXM_contrib2/asir2000/io/ox.c,v 1.27 2006/02/08 02:11:19 noro Exp $ +*/ #include "ca.h" #include "parse.h" +#include "signal.h" #include "wsio.h" #include "ox.h" #define ISIZ sizeof(int) -void ox_flush_stream(),ox_write_int(),ox_write_cmo(); -void ox_read_int(),ox_read_cmo(); -void mclist_to_mc(); -void ox_read_local(); - extern Obj VOIDobj; +extern int nserver_102, myrank_102; extern int ox_need_conv; -extern int ox_usr1_sent, ox_int_received, critical_when_signal; +int ox_usr1_sent, ox_int_received, critical_when_signal; unsigned int ox_serial; int ox_flushing; int ox_batch; int ox_check=1; -jmp_buf ox_env; +int ox_exchange_mathcap=1; +JMP_BUF ox_env; MATHCAP my_mathcap; @@ -45,11 +90,28 @@ static struct mathcap my_mc; static struct mathcap *remote_mc; static int remote_mc_len; -void ox_resetenv(s) -char *s; +void mclist_to_mc(LIST mclist,struct mathcap *mc); + +#if defined(VISUAL) +/* XXX : mainly used in engine2000/io.c, but declared here */ +HANDLE hStreamNotify,hStreamNotify_Ack; + +void cleanup_events() { + /* ox_watch_stream may be waiting for hStreamNotify_Ack to be set */ + + ResetEvent(hStreamNotify); + SetEvent(hStreamNotify_Ack); +} +#endif + +void ox_resetenv(char *s) +{ +#if defined(VISUAL) + cleanup_events(); +#endif fprintf(stderr,"%s\n",s); - longjmp(ox_env,1); + LONGJMP(ox_env,1); } static int available_cmo[] = { @@ -58,14 +120,16 @@ static int available_cmo[] = { CMO_ZZ, CMO_QQ, CMO_ZERO, CMO_DMS_GENERIC, CMO_DMS_OF_N_VARIABLES, CMO_RING_BY_NAME, CMO_DISTRIBUTED_POLYNOMIAL, + CMO_RATIONAL, CMO_RECURSIVE_POLYNOMIAL, CMO_UNIVARIATE_POLYNOMIAL, CMO_INDETERMINATE, + CMO_TREE, 0 }; -static int available_sm[] = { +static int asir_available_sm[] = { SM_dupErrors, SM_getsp, SM_popSerializedLocalObject, - SM_popCMO, SM_popString, SM_setName, + SM_popCMO, SM_popString, SM_pushCMOtag, SM_setName, SM_evalName, SM_executeStringByLocalParser, SM_executeStringByLocalParserInBatchMode, SM_executeFunction, SM_shutdown, SM_pops, @@ -74,6 +138,28 @@ static int available_sm[] = { 0 }; +static int ox_asir_available_sm[] = { + SM_dupErrors, SM_getsp, SM_popSerializedLocalObject, + SM_popCMO, SM_popString, SM_pushCMOtag, SM_setName, + SM_evalName, SM_executeStringByLocalParser, + SM_executeStringByLocalParserInBatchMode, + SM_executeFunction, SM_shutdown, SM_pops, + SM_mathcap, SM_setMathcap, SM_nop, + SM_beginBlock, SM_endBlock, + SM_set_rank_102, SM_tcp_accept_102, SM_tcp_connect_102, SM_reset_102, + SM_bcast_102, SM_reduce_102, + 0 +}; + +static int ox_plot_available_sm[] = { + SM_dupErrors, SM_getsp, SM_popSerializedLocalObject, + SM_popCMO, SM_popString, SM_setName, + SM_evalName, SM_executeStringByLocalParser, + SM_executeFunction, SM_shutdown, SM_pops, + SM_mathcap, SM_setMathcap, SM_nop, + 0 +}; + /* mathcap = [ @@ -92,7 +178,7 @@ void create_my_mathcap(char *system) NODE n,n0; int i,k; STRING str; - LIST sname,smlist,oxlist,cmolist,asirlist,oxtag,oxasir,r; + LIST sname,smlist,oxlist,cmolist,asirlist,oxasir,r; USINT tag,t,t1; if ( my_mathcap ) @@ -102,11 +188,24 @@ void create_my_mathcap(char *system) MKUSINT(t,OX_VERSION); n0 = mknode(2,t,str); MKLIST(sname,n0); - /* cmo tag */ - for ( n0 = 0, i = 0; k = available_sm[i]; i++ ) { - NEXTNODE(n0,n); MKUSINT(t,k); BDY(n) = (pointer)t; + /* sm tag */ + n0 = 0; + if ( !strcmp(system,"asir") ) { + for ( i = 0; k = asir_available_sm[i]; i++ ) { + NEXTNODE(n0,n); MKUSINT(t,k); BDY(n) = (pointer)t; + } + } else if ( !strcmp(system,"ox_asir") ) { + for ( i = 0; k = ox_asir_available_sm[i]; i++ ) { + NEXTNODE(n0,n); MKUSINT(t,k); BDY(n) = (pointer)t; + } + NEXT(n) = 0; + } else if ( !strcmp(system,"ox_plot") ) { + for ( i = 0; k = ox_plot_available_sm[i]; i++ ) { + NEXTNODE(n0,n); MKUSINT(t,k); BDY(n) = (pointer)t; + } + NEXT(n) = 0; } - NEXT(n) = 0; MKLIST(smlist,n0); + MKLIST(smlist,n0); /* creation of [OX_DATA,CMO list] */ /* ox tag */ @@ -185,8 +284,8 @@ void store_remote_mathcap(int s,MATHCAP mc) void mclist_to_mc(LIST mclist,struct mathcap *mc) { - int id,l,i,j; - NODE n,t,oxcmo,ox,cap; + int l,i,j; + NODE n,t,oxcmo,cap; int *ptr; /* @@ -232,9 +331,7 @@ void mclist_to_mc(LIST mclist,struct mathcap *mc) } } -int check_sm_by_mc(s,smtag) -int s; -unsigned int smtag; +int check_sm_by_mc(int s,unsigned int smtag) { struct mathcap *rmc; int nsmcap,i; @@ -257,9 +354,7 @@ unsigned int smtag; return 1; } -int check_by_mc(s,oxtag,cmotag) -int s; -unsigned int oxtag,cmotag; +int check_by_mc(int s,unsigned int oxtag,unsigned int cmotag) { struct mathcap *rmc; int noxcap,ncap,i,j; @@ -297,18 +392,23 @@ void begin_critical() { void end_critical() { critical_when_signal = 0; if ( ox_usr1_sent ) { - ox_usr1_sent = 0; ox_usr1_handler(); + ox_usr1_sent = 0; +#if !defined(VISUAL) + ox_usr1_handler(SIGUSR1); +#else + ox_usr1_handler(0); +#endif } if ( ox_int_received ) { ox_int_received = 0; int_handler(SIGINT); } } -void ox_usr1_handler(sig) -int sig; +extern NODE user_int_handler; + +void ox_usr1_handler(int sig) { - extern jmp_buf env; - unsigned int cmd; + NODE t; #if !defined(VISUAL) signal(SIGUSR1,ox_usr1_handler); @@ -318,17 +418,19 @@ int sig; ox_usr1_sent = 1; } else { ox_flushing = 1; + if ( user_int_handler ) { + fprintf(stderr, + "usr1 : calling the registered exception handlers..."); + for ( t = user_int_handler; t; t = NEXT(t) ) + bevalf((FUNC)BDY(t),0); + fprintf(stderr, "done.\n"); + } ox_resetenv("usr1 : return to toplevel by SIGUSR1"); } } void clear_readbuffer() { - char c; - fd_set r,w,e; - struct timeval interval; - int n,sock; - #if defined(linux) iofp[0].in->_IO_read_ptr = iofp[0].in->_IO_read_end; #elif defined(__FreeBSD__) @@ -360,6 +462,11 @@ void wait_for_data(int s) { return; } + +void wait_for_data_102(int rank) +{ + return; +} #else int ox_data_is_available(int s) { @@ -375,7 +482,7 @@ void wait_for_data(int s) #if defined(VISUAL) sock = iofp[s].in->fildes; FD_ZERO(&r); - FD_SET(sock,&r); + FD_SET((unsigned int)sock,&r); select(0,&r,NULL,NULL,NULL); #else sock = fileno(iofp[s].in); @@ -385,12 +492,36 @@ void wait_for_data(int s) #endif } } + +void wait_for_data_102(int rank) +{ + fd_set r; + int sock; + + if ( !FP_DATA_IS_AVAILABLE(iofp_102[rank].in) ) { +#if defined(VISUAL) + sock = iofp_102[rank].in->fildes; + FD_ZERO(&r); + FD_SET((unsigned int)sock,&r); + select(0,&r,NULL,NULL,NULL); +#else + sock = fileno(iofp_102[rank].in); + FD_ZERO(&r); + FD_SET(sock,&r); + select(FD_SETSIZE,&r,NULL,NULL,NULL); #endif + } +} +#endif void ox_send_data(int s,pointer p) { - if ( ox_check && !ox_check_cmo(s,(Obj)p) ) - error("ox_send_data : Mathcap violation"); + ERR err; + + if ( ox_check && !ox_check_cmo(s,(Obj)p) ) { + create_error(&err,ox_serial,"ox_send_data : Mathcap violation",0); + p = (pointer)err; + } begin_critical(); ox_write_int(s,OX_DATA); ox_write_int(s,ox_serial++); @@ -399,6 +530,72 @@ void ox_send_data(int s,pointer p) end_critical(); } +void ox_send_data_102(int rank,pointer p) +{ + ERR err; + + begin_critical(); + ox_write_int_102(rank,OX_DATA); + ox_write_int_102(rank,ox_serial++); + ox_write_cmo_102(rank,p); + ox_flush_stream_102(rank); + end_critical(); +} + +void ox_bcast_102(int root) +{ + Obj data; + int r,mask,id,src,dst; + + r = myrank_102-root; + if ( r == 0 ) + data = (Obj)asir_pop_one(); + + if ( r < 0 ) r += nserver_102; + for ( mask = 1; mask < nserver_102; mask <<= 1 ) + if ( r&mask ) { + src = myrank_102-mask; + if ( src < 0 ) src += nserver_102; + ox_recv_102(src,&id,&data); + break; + } + for ( mask >>= 1; mask > 0; mask >>= 1 ) + if ( (r+mask) < nserver_102 ) { + dst = myrank_102+mask; + if ( dst >= nserver_102 ) dst -= nserver_102; + ox_send_data_102(dst,data); + } + asir_push_one(data); +} + +/* func : an arithmetic funcion func(vl,a,b,*c) */ + +void ox_reduce_102(int root,void (*func)()) +{ + Obj data,data0,t; + int r,mask,id,src,dst; + + r = myrank_102-root; + if ( r < 0 ) r += nserver_102; + data = (Obj)asir_pop_one(); + for ( mask = 1; mask < nserver_102; mask <<= 1 ) + if ( r&mask ) { + dst = (r-mask)+root; + if ( dst >= nserver_102 ) dst -= nserver_102; + ox_send_data_102(dst,data); + break; + } else { + src = r+mask; + if ( src < nserver_102 ) { + src += root; + if ( src >= nserver_102 ) src -= nserver_102; + ox_recv_102(src,&id,&data0); + (*func)(CO,data,data0,&t); data = t; + } + } + asir_push_one(r?0:data); +} + void ox_send_cmd(int s,int id) { if ( ox_check && !check_sm_by_mc(s,id) ) @@ -420,30 +617,61 @@ void ox_send_sync(int s) end_critical(); } +void ox_send_sync_102(int rank) +{ + begin_critical(); + ox_write_int_102(rank,OX_SYNC_BALL); + ox_write_int_102(rank,ox_serial++); + ox_flush_stream_102(rank); + end_critical(); +} + void ox_send_local_data(int s,Obj p) { begin_critical(); ox_write_int(s,OX_LOCAL_OBJECT_ASIR); ox_write_int(s,ox_serial++); ox_write_int(s,ASIR_OBJ); - saveobj(iofp[s].out,p); + saveobj((FILE *)iofp[s].out,p); ox_flush_stream(s); end_critical(); } +void ox_send_local_data_102(int rank,Obj p) +{ + begin_critical(); + ox_write_int_102(rank,OX_LOCAL_OBJECT_ASIR); + ox_write_int_102(rank,ox_serial++); + ox_write_int_102(rank,ASIR_OBJ); + saveobj((FILE *)iofp_102[rank].out,p); + ox_flush_stream_102(rank); + end_critical(); +} + void ox_send_local_ring(int s,VL vl) { begin_critical(); ox_write_int(s,OX_LOCAL_OBJECT_ASIR); ox_write_int(s,ox_serial++); ox_write_int(s,ASIR_VL); - savevl(iofp[s].out,vl); + savevl((FILE *)iofp[s].out,vl); ox_flush_stream(s); end_critical(); } -unsigned int ox_recv(int s, int *id, pointer *p) +void ox_send_local_ring_102(int rank,VL vl) { + begin_critical(); + ox_write_int_102(rank,OX_LOCAL_OBJECT_ASIR); + ox_write_int_102(rank,ox_serial++); + ox_write_int_102(rank,ASIR_VL); + savevl((FILE *)iofp_102[rank].out,vl); + ox_flush_stream_102(rank); + end_critical(); +} + +unsigned int ox_recv(int s, int *id, Obj *p) +{ unsigned int cmd,serial; USINT ui; @@ -455,7 +683,7 @@ unsigned int ox_recv(int s, int *id, pointer *p) case OX_COMMAND: ox_read_int(s,&cmd); MKUSINT(ui,cmd); - *p = (pointer)ui; + *p = (Obj)ui; break; case OX_DATA: ox_read_cmo(s,p); @@ -471,10 +699,37 @@ unsigned int ox_recv(int s, int *id, pointer *p) return serial; } -void ox_get_result(s,rp) -int s; -Obj *rp; +unsigned int ox_recv_102(int rank, int *id, Obj *p) { + unsigned int cmd,serial; + USINT ui; + + wait_for_data_102(rank); + begin_critical(); + ox_read_int_102(rank,id); + ox_read_int_102(rank,&serial); + switch ( *id ) { + case OX_COMMAND: + ox_read_int_102(rank,&cmd); + MKUSINT(ui,cmd); + *p = (Obj)ui; + break; + case OX_DATA: + ox_read_cmo_102(rank,p); + break; + case OX_LOCAL_OBJECT_ASIR: + ox_read_local_102(rank,p); + break; + default: + *p = 0; + break; + } + end_critical(); + return serial; +} + +void ox_get_result(int s,Obj *rp) +{ int id; Obj obj,r; int level; @@ -482,7 +737,7 @@ Obj *rp; level = 0; r = 0; do { - ox_recv(s,&id,(pointer *)&obj); + ox_recv(s,&id,&obj); if ( id == OX_COMMAND ) { switch ( ((USINT)obj)->body ) { case SM_beginBlock: @@ -500,28 +755,41 @@ Obj *rp; void ox_read_int(int s, int *n) { ox_need_conv = iofp[s].conv; - read_int(iofp[s].in,n); + read_int((FILE *)iofp[s].in,n); } +void ox_read_int_102(int rank, int *n) +{ + ox_need_conv = iofp_102[rank].conv; + read_int((FILE *)iofp_102[rank].in,n); +} + void ox_read_cmo(int s, Obj *rp) { ox_need_conv = iofp[s].conv; - read_cmo(iofp[s].in,rp); + read_cmo((FILE *)iofp[s].in,rp); } +void ox_read_cmo_102(int rank, Obj *rp) +{ + ox_need_conv = iofp_102[rank].conv; + read_cmo((FILE *)iofp_102[rank].in,rp); +} + + void ox_read_local(int s, Obj *rp) { int id; ox_need_conv = iofp[s].conv; - read_int(iofp[s].in,&id); + read_int((FILE *)iofp[s].in,&id); switch ( id ) { case ASIR_VL: - loadvl(iofp[s].in); + loadvl((FILE *)iofp[s].in); *rp = VOIDobj; break; case ASIR_OBJ: - loadobj(iofp[s].in,rp); + loadobj((FILE *)iofp[s].in,rp); break; default: error("ox_read_local : unsupported id"); @@ -529,18 +797,50 @@ void ox_read_local(int s, Obj *rp) } } +void ox_read_local_102(int rank, Obj *rp) +{ + int id; + + ox_need_conv = iofp_102[rank].conv; + read_int((FILE *)iofp_102[rank].in,&id); + switch ( id ) { + case ASIR_VL: + loadvl((FILE *)iofp_102[rank].in); + *rp = VOIDobj; + break; + case ASIR_OBJ: + loadobj((FILE *)iofp_102[rank].in,rp); + break; + default: + error("ox_read_local_102 : unsupported id"); + break; + } +} + void ox_write_int(int s, int n) { ox_need_conv = iofp[s].conv; - write_int(iofp[s].out,&n); + write_int((FILE *)iofp[s].out,&n); } +void ox_write_int_102(int rank, int n) +{ + ox_need_conv = iofp_102[rank].conv; + write_int((FILE *)iofp_102[rank].out,&n); +} + void ox_write_cmo(int s, Obj obj) { ox_need_conv = iofp[s].conv; - write_cmo(iofp[s].out,obj); + write_cmo((FILE *)iofp[s].out,obj); } +void ox_write_cmo_102(int rank, Obj obj) +{ + ox_need_conv = iofp_102[rank].conv; + write_cmo((FILE *)iofp_102[rank].out,obj); +} + int ox_check_cmo(int s, Obj obj) { NODE m; @@ -549,6 +849,7 @@ int ox_check_cmo(int s, Obj obj) return 1; switch ( OID(obj) ) { case O_MATHCAP: case O_STR: case O_ERR: case O_USINT: case O_VOID: + case O_BYTEARRAY: return 1; case O_P: if ( !check_by_mc(s,OX_DATA,CMO_RECURSIVE_POLYNOMIAL) ) @@ -565,18 +866,25 @@ int ox_check_cmo(int s, Obj obj) case O_DP: return ox_check_cmo_dp(s,(DP)obj); case O_N: - if ( NID((Num)obj) == N_Q ) { - if ( INT((Q)obj) ) - return check_by_mc(s,OX_DATA,CMO_ZZ); - else - return check_by_mc(s,OX_DATA,CMO_QQ); - } else - return 0; + switch ( NID((Num)obj) ) { + case N_Q: + if ( INT((Q)obj) ) + return check_by_mc(s,OX_DATA,CMO_ZZ); + else + return check_by_mc(s,OX_DATA,CMO_QQ); + case N_R: + return 1; + default: + return 0; + } + break; case O_LIST: for ( m = BDY((LIST)obj); m; m = NEXT(m) ) if ( !ox_check_cmo(s,(BDY(m))) ) return 0; return 1; + case O_QUOTE: /* XXX */ + return 1; default: return 0; } @@ -591,6 +899,12 @@ void ox_get_serverinfo(int s, LIST *rp) } } +char *ox_get_servername(int s) +{ + return (remote_mc && remote_mc[s].servername)?remote_mc[s].servername:0; +} + + int ox_check_cmo_p(int s, P p) { DCP dc; @@ -615,8 +929,7 @@ int ox_check_cmo_dp(int s, DP p) return 1; } -void ox_flush_stream(s) -int s; +void ox_flush_stream(int s) { if ( ox_batch ) return; @@ -629,11 +942,10 @@ int s; cflush(iofp[s].out); else #endif - fflush(iofp[s].out); + fflush((FILE *)iofp[s].out); } -void ox_flush_stream_force(s) -int s; +void ox_flush_stream_force(int s) { #if defined(VISUAL) if ( _fileno(&iofp[s].out->fp) < 0 ) @@ -644,5 +956,23 @@ int s; cflush(iofp[s].out); else #endif - fflush(iofp[s].out); + fflush((FILE *)iofp[s].out); +} + +void ox_flush_stream_102(int rank) +{ + if ( !ox_batch ) + ox_flush_stream_force_102(rank); +} + +void ox_flush_stream_force_102(int rank) +{ + if ( iofp_102[rank].out ) +#if defined(VISUAL) + cflush(iofp_102[rank].out); +#elif MPI + cflush(iofp_102[rank].out); +#else + fflush(iofp_102[rank].out); +#endif }