===================================================================
RCS file: /home/cvs/OpenXM_contrib2/asir2000/io/tcpf.c,v
retrieving revision 1.38
retrieving revision 1.46
diff -u -p -r1.38 -r1.46
--- OpenXM_contrib2/asir2000/io/tcpf.c	2003/03/07 03:12:28	1.38
+++ OpenXM_contrib2/asir2000/io/tcpf.c	2003/12/10 05:39:58	1.46
@@ -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.37 2003/02/14 22:29:15 ohara Exp $ 
+ * $OpenXM: OpenXM_contrib2/asir2000/io/tcpf.c,v 1.45 2003/12/10 02:16:08 noro Exp $ 
 */
 #include "ca.h"
 #include "parse.h"
@@ -79,7 +79,9 @@ static struct m_c {
 static int m_c_i,m_c_s;
 int I_am_server;
 
-#if MPI
+extern int little_endian;
+
+#if defined(MPI)
 extern int mpi_nprocs;
 #define valid_mctab_index(ind)\
 if((ind)<0||(ind)>=mpi_nprocs){error("invalid server id");}
@@ -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,12 @@ 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_reset_102();
+void Pox_bcast_102();
 
 void ox_launch_generic();
 
@@ -129,6 +143,17 @@ 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_send_102",Pox_send_102,2},
+	{"ox_recv_102",Pox_recv_102,1},
+	{"ox_bcast_102",Pox_bcast_102,-2},
+
 	{"try_bind_listen",Ptry_bind_listen,1},
 	{"try_connect",Ptry_connect,2},
 	{"try_accept",Ptry_accept,2},
@@ -143,11 +168,11 @@ struct ftab tcp_tab[] = {
 	{"ox_cmo_rpc",Pox_cmo_rpc,-99999999},
 
 	{"ox_sync",Pox_sync,1},
-#if MPI
+#if defined(MPI)
 	{"ox_mpi_myid",Pox_mpi_myid,0},
 	{"ox_mpi_nprocs",Pox_mpi_nprocs,0},
 #endif
-#if !MPI
+#if !defined(MPI)
 	{"ox_reset",Pox_reset,-2},
 	{"ox_intr",Pox_intr,1},
 	{"ox_select",Pox_select,-2},
@@ -173,7 +198,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,7 +210,13 @@ extern int ox_exchange_mathcap;
 
 char *getenv();
 
-#if MPI
+void Pnd_exec_f4_red(Q *rp)
+{
+	nd_exec_f4_red_dist();
+	*rp = 0;
+}
+
+#if defined(MPI)
 extern int mpi_myid, mpi_nprocs;
 
 void Pox_mpi_myid(Q *rp)
@@ -255,7 +288,72 @@ 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);
+}
+
+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 +523,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 );
 
@@ -771,7 +875,7 @@ int register_server(int af_unix,int m,int c)
 		}
 		m_c_s = INIT_TAB_SIZ;
 	}
-#if !MPI
+#if !defined(MPI)
 	for ( i = 0; i < m_c_i; i++ )
 		if ( (m_c_tab[i].m<0) && (m_c_tab[i].c<0) )
 			break;
@@ -879,13 +983,46 @@ 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 ( rank == myrank_102 ) {
+		if ( argc(arg) == 1 ) 
+			error("ox_bcast_102 : data should be given at the root");
+		data = (Obj)ARG1(arg);
+	}
+	ox_bcast_102(rank,&data);
+	*rp = data;
+}
+
 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 +1038,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) )
@@ -1195,7 +1335,7 @@ void Pox_shutdown(NODE arg,Q *rp)
 	free_iofp(s);
 	s = m_c_tab[index].c;
 	free_iofp(s);
-#if !MPI && !defined(VISUAL)
+#if !defined(MPI) && !defined(VISUAL)
 	if ( m_c_tab[index].af_unix )
 		wait(&status);
 #endif
@@ -1235,7 +1375,7 @@ void shutdown_all() {
 		free_iofp(s);
 		s = m_c_tab[index].c;
 		free_iofp(s);
-#if !MPI && !defined(VISUAL)
+#if !defined(MPI) && !defined(VISUAL)
 		if ( m_c_tab[index].af_unix )
 			wait(&status);
 #endif
@@ -1283,3 +1423,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;
+}
+