=================================================================== RCS file: /home/cvs/OpenXM_contrib2/asir2000/io/ox.c,v retrieving revision 1.20 retrieving revision 1.23 diff -u -p -r1.20 -r1.23 --- OpenXM_contrib2/asir2000/io/ox.c 2003/03/07 03:12:28 1.20 +++ OpenXM_contrib2/asir2000/io/ox.c 2003/12/10 07:37:40 1.23 @@ -44,7 +44,7 @@ * 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.19 2003/02/14 22:29:15 ohara Exp $ + * $OpenXM: OpenXM_contrib2/asir2000/io/ox.c,v 1.22 2003/12/10 05:39:58 noro Exp $ */ #include "ca.h" #include "parse.h" @@ -56,6 +56,7 @@ extern Obj VOIDobj; +extern int nserver_102, myrank_102; extern int ox_need_conv; int ox_usr1_sent, ox_int_received, critical_when_signal; unsigned int ox_serial; @@ -144,6 +145,7 @@ static int ox_asir_available_sm[] = { 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, 0 }; @@ -480,7 +482,27 @@ 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) { @@ -498,6 +520,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 *rp) +{ + Obj data; + int r,mask,id,src,dst; + + data = *rp; + r = myrank_102-root; + 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); + } + *rp = data; +} + +/* func : an arithmetic funcion func(vl,a,b,*c) */ + +void ox_reduce_102(int root,void (*func)(),Obj data,Obj *rp) +{ + Obj data0,t; + int r,mask,id,src,dst; + + r = myrank_102-root; + if ( r < 0 ) r += nserver_102; + 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; + } + } + if ( !r ) + *rp = data; + else + *rp = 0; +} + void ox_send_cmd(int s,int id) { if ( ox_check && !check_sm_by_mc(s,id) ) @@ -519,6 +607,15 @@ 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(); @@ -530,6 +627,17 @@ void ox_send_local_data(int s,Obj p) 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(); @@ -541,6 +649,17 @@ void ox_send_local_ring(int s,VL vl) end_critical(); } +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; @@ -570,6 +689,35 @@ unsigned int ox_recv(int s, int *id, Obj *p) return serial; } +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; @@ -600,12 +748,25 @@ void ox_read_int(int s, int *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((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; @@ -626,18 +787,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((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((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; @@ -754,4 +947,22 @@ void ox_flush_stream_force(int s) else #endif 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 }