===================================================================
RCS file: /home/cvs/OpenXM_contrib2/asir2000/io/tcpf.c,v
retrieving revision 1.39
retrieving revision 1.50
diff -u -p -r1.39 -r1.50
--- OpenXM_contrib2/asir2000/io/tcpf.c	2003/03/07 06:39:57	1.39
+++ OpenXM_contrib2/asir2000/io/tcpf.c	2003/12/13 08:03:56	1.50
@@ -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/tcpf.c,v 1.38 2003/03/07 03:12:28 noro Exp $ 
+ * $OpenXM: OpenXM_contrib2/asir2000/io/tcpf.c,v 1.49 2003/12/12 04:59:59 noro Exp $ 
 */
 #include "ca.h"
 #include "parse.h"
@@ -79,6 +79,8 @@ static struct m_c {
 static int m_c_i,m_c_s;
 int I_am_server;
 
+extern int little_endian;
+
 #if defined(MPI)
 extern int mpi_nprocs;
 #define valid_mctab_index(ind)\
@@ -94,6 +96,12 @@ if((ind)<0||(ind)>=m_c_i||\
 ((m_c_tab[ind].m<0)&&(m_c_tab[ind].c<0))){(ind)=-1;}
 #endif
 
+struct IOFP iofp_102[MAXIOFP];
+int nserver_102;
+int myrank_102;
+
+int register_102(int s,int rank, int is_master);
+
 int register_server();
 int get_mcindex(int);
 
@@ -119,6 +127,14 @@ void Pgenerate_port(),Ptry_bind_listen(),Ptry_connect(
 void Pregister_server();
 void Pox_get_serverinfo();
 void Pox_mpi_myid(), Pox_mpi_nprocs();
+void Pnd_exec_f4_red();
+void Pox_tcp_accept_102(),Pox_tcp_connect_102();
+void Pox_send_102(),Pox_recv_102();
+void Pox_set_rank_102();
+void Pox_get_rank_102();
+void Pox_reset_102();
+void Pox_bcast_102();
+void Pox_reduce_102();
 
 void ox_launch_generic();
 
@@ -129,6 +145,19 @@ struct ftab tcp_tab[] = {
 	{"ox_recv_raw_cmo",Pox_recv_raw_cmo,1},
 	{"ox_get_serverinfo",Pox_get_serverinfo,-1},
 	{"generate_port",Pgenerate_port,-1},
+
+	/* from master to client */
+	{"ox_set_rank_102",Pox_set_rank_102,3},
+	{"ox_tcp_accept_102",Pox_tcp_accept_102,3},
+	{"ox_tcp_connect_102",Pox_tcp_connect_102,4},
+	{"ox_reset_102",Pox_reset_102,1},
+
+	{"ox_get_rank_102",Pox_get_rank_102,0},
+	{"ox_send_102",Pox_send_102,2},
+	{"ox_recv_102",Pox_recv_102,1},
+	{"ox_bcast_102",Pox_bcast_102,-2},
+	{"ox_reduce_102",Pox_reduce_102,-3},
+
 	{"try_bind_listen",Ptry_bind_listen,1},
 	{"try_connect",Ptry_connect,2},
 	{"try_accept",Ptry_accept,2},
@@ -173,7 +202,9 @@ struct ftab tcp_tab[] = {
 	{"ox_execute_string",Pox_execute_string,2},
 	{"ox_execute_function",Pox_execute_function,3},
 
+	{"nd_exec_f4_red",Pnd_exec_f4_red,0},
 
+
 	{0,0,0},
 };
 
@@ -183,6 +214,12 @@ extern int ox_exchange_mathcap;
 
 char *getenv();
 
+void Pnd_exec_f4_red(Q *rp)
+{
+	nd_exec_f4_red_dist();
+	*rp = 0;
+}
+
 #if defined(MPI)
 extern int mpi_myid, mpi_nprocs;
 
@@ -255,7 +292,84 @@ void Pgenerate_port(NODE arg,Obj *rp)
 	}
 }
 
+void Pox_reset_102(NODE arg,Q *rp)
+{
+	int s;
+	int	index = QTOS((Q)ARG0(arg));
+
+	valid_mctab_index(index);
+	s = m_c_tab[index].c;
+	ox_send_cmd(s,SM_reset_102);
+	ox_flush_stream_force(s);
+	*rp = 0;
+}
+
+void Pox_get_rank_102(LIST *rp)
+{
+	Q n,r;
+	NODE node;
+
+	STOQ(nserver_102,n);
+	STOQ(myrank_102,r);
+	node = mknode(2,n,r);
+	MKLIST(*rp,node);
+}
+
+void Pox_set_rank_102(NODE arg,Q *rp)
+{
+	Q nserver,rank;
+	int s;
+	int	index = QTOS((Q)ARG0(arg));
+
+	valid_mctab_index(index);
+	s = m_c_tab[index].c;
+	nserver = (Q)ARG1(arg);	
+	rank = (Q)ARG2(arg);	
+	ox_send_data(s,nserver);
+	ox_send_data(s,rank);
+	ox_send_cmd(s,SM_set_rank_102);
+	ox_flush_stream_force(s);
+	*rp = 0;
+}
+
+/* ox_tcp_accept_102(server,port,rank) */
+
+void Pox_tcp_accept_102(NODE arg,Q *rp)
+{
+	int s;
+	int	index = QTOS((Q)ARG0(arg));
+
+	valid_mctab_index(index);
+	s = m_c_tab[index].c;
+
+	ox_send_data(s,ARG1(arg));
+	ox_send_data(s,ARG2(arg));
+	ox_send_cmd(s,SM_tcp_accept_102);
+	ox_flush_stream_force(s);
+	*rp = 0;
+}
+
 /*
+ ox_tcp_connect_102(server,host,port,rank)
+*/
+
+void Pox_tcp_connect_102(NODE arg,Q *rp)
+{
+	int s;
+	int	index = QTOS((Q)ARG0(arg));
+
+	valid_mctab_index(index);
+	s = m_c_tab[index].c;
+
+	ox_send_data(s,ARG1(arg));
+	ox_send_data(s,ARG2(arg));
+	ox_send_data(s,ARG3(arg));
+	ox_send_cmd(s,SM_tcp_connect_102);
+	ox_flush_stream_force(s);
+	*rp = 0;	
+}
+
+/*
  try_bind_listen(port)
 */
 
@@ -425,17 +539,23 @@ void ox_launch_generic(char *host,char *launcher,char 
 		generate_port(use_unix,server_port_str);
 		if ( !conn_to_serv ) {
 			cs = try_bind_listen(use_unix,control_port_str);
+			if ( cs < 0 ) continue;
 			ss = try_bind_listen(use_unix,server_port_str);
+			if ( ss < 0 ) continue;
 		}
 		spawn_server(host,launcher,server,
 			use_unix,use_ssh,use_x,conn_to_serv,
 				control_port_str,server_port_str);
 		if ( conn_to_serv ) {
 			cs = try_connect(use_unix,host,control_port_str);
+			if ( cs < 0 ) continue;
 			ss = try_connect(use_unix,host,server_port_str);
+			if ( ss < 0 ) continue;
 		} else {
 			cs = try_accept(use_unix,cs);
+			if ( cs < 0 ) continue;
 			ss = try_accept(use_unix,ss);
+			if ( ss < 0 ) continue;
 		}
 	} while ( cs < 0 || ss < 0 );
 
@@ -879,13 +999,66 @@ void Pox_recv_raw_cmo(NODE arg,Obj *rp)
 	ox_read_cmo(s,rp);
 }
 
+void Pox_send_102(NODE arg,Obj *rp)
+{
+	int rank = QTOS((Q)ARG0(arg));
+
+	ox_send_data_102(rank,(Obj)ARG1(arg));
+	*rp = 0;
+}
+
+void Pox_recv_102(NODE arg,Obj *rp)
+{
+	int id;
+	int rank = QTOS((Q)ARG0(arg));
+
+	ox_recv_102(rank,&id,rp);
+}
+
+void Pox_bcast_102(NODE arg,Obj *rp)
+{
+	int rank = QTOS((Q)ARG0(arg));
+	Obj data;
+
+	if ( argc(arg) > 1 )
+		asir_push_one((Obj)ARG1(arg));
+	ox_bcast_102(rank);
+	*rp = (Obj)asir_pop_one();
+}
+
+void Pox_reduce_102(NODE arg,Obj *rp)
+{
+	int root = QTOS((Q)ARG0(arg));
+	STRING op;
+	char *opname;
+	void (*func)();
+
+	op = (STRING)ARG1(arg);
+	asir_assert(op,O_STR,"ox_reduce_102");
+	opname = BDY(op);
+	if ( !strcmp(opname,"+") )
+		func = arf_add;
+	else if ( !strcmp(opname,"*") )
+		func = arf_mul;
+	else {
+		error("ox_reduce_102 : operation not supported");
+	}
+	if ( argc(arg) > 2 )
+		asir_push_one((Obj)ARG2(arg));
+	ox_reduce_102(root,func);
+	*rp = (Obj)asir_pop_one();
+}
+
 void Pox_push_local(NODE arg,Obj *rp)
 {
 	int s;
 	struct oLIST dummy;
 	VL vl;
-	int index = QTOS((Q)ARG0(arg));
+	int index;
 
+	if ( !arg )
+		error("ox_push_local : too few arguments.");
+	index = QTOS((Q)ARG0(arg));
 	valid_mctab_index(index);
 	s = m_c_tab[index].c; arg = NEXT(arg);
 
@@ -901,8 +1074,11 @@ void Pox_push_local(NODE arg,Obj *rp)
 void Pox_push_cmo(NODE arg,Obj *rp)
 {
 	int s;
-	int index = QTOS((Q)ARG0(arg));
+	int index;
 
+	if ( !arg )
+		error("ox_push_cmo : too few arguments.");
+	index = QTOS((Q)ARG0(arg));
 	valid_mctab_index(index);
 	s = m_c_tab[index].c; arg = NEXT(arg);
 	for ( ; arg; arg = NEXT(arg) )
@@ -1283,3 +1459,52 @@ int validate_ox_plot_stream(int index)
 #endif
 	return i;
 }
+
+int get_ox_server_id(int index)
+{
+	valid_mctab_index(index);
+	return m_c_tab[index].c;
+}
+
+int register_102(int s1,int rank,int is_master)
+{
+	unsigned char c,rc;
+
+	if ( rank >= MAXIOFP ) return -1;
+	iofp_102[rank].s = s1;
+#if defined(VISUAL)
+	iofp_102[rank].in = WSIO_open(s1,"r");
+	iofp_102[rank].out = WSIO_open(s1,"w");
+#else
+	iofp_102[rank].in = fdopen(s1,"r");
+	iofp_102[rank].out = fdopen(s1,"w");
+#if !defined(__CYGWIN__)
+	setbuffer(iofp_102[rank].in,iofp_102[rank].inbuf = 
+		(char *)GC_malloc_atomic(LBUFSIZ),LBUFSIZ);
+	setbuffer(iofp_102[rank].out,iofp_102[rank].outbuf = 
+		(char *)GC_malloc_atomic(LBUFSIZ),LBUFSIZ);
+#endif
+#endif
+	if ( little_endian )
+		c = 1;
+	else
+		c = 0xff;
+	if ( is_master ) {
+		/* server : write -> read */
+		write_char((FILE *)iofp_102[rank].out,&c);
+		ox_flush_stream_force_102(rank);
+		read_char((FILE *)iofp_102[rank].in,&rc);
+	} else {
+		/* client : read -> write */
+		read_char((FILE *)iofp_102[rank].in,&rc);
+		/* special care for a failure of spawing a server */
+		if ( rc !=0 && rc != 1 && rc != 0xff )
+			return -1;	
+		write_char((FILE *)iofp_102[rank].out,&c);
+		ox_flush_stream_force_102(rank);
+	}
+	iofp_102[rank].conv = c == rc ? 0 : 1;
+	iofp_102[rank].socket = 0;
+	return 0;
+}
+