===================================================================
RCS file: /home/cvs/OpenXM_contrib2/asir2000/io/tcpf.c,v
retrieving revision 1.38
retrieving revision 1.43
diff -u -p -r1.38 -r1.43
--- OpenXM_contrib2/asir2000/io/tcpf.c	2003/03/07 03:12:28	1.38
+++ OpenXM_contrib2/asir2000/io/tcpf.c	2003/12/03 09:32:36	1.43
@@ -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.42 2003/09/19 02:33:14 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,11 @@ if((ind)<0||(ind)>=m_c_i||\
 ((m_c_tab[ind].m<0)&&(m_c_tab[ind].c<0))){(ind)=-1;}
 #endif
 
+static struct IOFP iofp_102[MAXIOFP];
+
+void flush_stream_102(int rank), flush_stream_force_102(int rank);
+int register_102(int s,int rank, int is_master);
+
 int register_server();
 int get_mcindex(int);
 
@@ -119,6 +126,9 @@ 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 Paccept_102(),Pconnect_102();
+void Pox_send_cmo_102(),Pox_recv_cmo_102();
 
 void ox_launch_generic();
 
@@ -129,6 +139,10 @@ 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},
+	{"ox_send_cmo_102",Pox_send_cmo_102,2},
+	{"ox_recv_cmo_102",Pox_recv_cmo_102,1},
+	{"accept_102",Paccept_102,2},
+	{"connect_102",Pconnect_102,3},
 	{"try_bind_listen",Ptry_bind_listen,1},
 	{"try_connect",Ptry_connect,2},
 	{"try_accept",Ptry_accept,2},
@@ -143,11 +157,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 +187,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 +199,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 +277,58 @@ void Pgenerate_port(NODE arg,Obj *rp)
 	}
 }
 
+/* accept_102(port,rank) */
+
+void Paccept_102(NODE arg,Q *rp)
+{
+	char port_str[BUFSIZ];
+	int port,s,use_unix,rank;
+
+	if ( IS_CYGWIN || !ARG0(arg) || NUM(ARG0(arg)) ) {
+		port = QTOS((Q)ARG0(arg));
+		sprintf(port_str,"%d",port);
+		use_unix = 0;
+	} else {
+		strcpy(port_str,BDY((STRING)ARG0(arg)));
+		use_unix = 1;
+	}
+	s = try_bind_listen(use_unix,port_str);
+	s = try_accept(use_unix,s);
+	rank = QTOS((Q)ARG1(arg));
+	if ( register_102(s,rank,1) < 0 )
+		STOQ(-1,*rp);
+	else
+		*rp = 0;
+}
+
 /*
+ connect_102(host,port,rank)
+*/
+
+void Pconnect_102(NODE arg,Q *rp)
+{
+	char port_str[BUFSIZ];
+	char *host;
+	int port,s,use_unix,rank;
+
+	if ( IS_CYGWIN || !ARG1(arg) || NUM(ARG1(arg)) ) {
+		port = QTOS((Q)ARG1(arg));
+		sprintf(port_str,"%d",port);
+		use_unix = 0;
+	} else {
+		strcpy(port_str,BDY((STRING)ARG1(arg)));
+		use_unix = 1;
+	}
+	host = BDY((STRING)ARG0(arg));
+	s = try_connect(use_unix,host,port_str);
+	rank = QTOS((Q)ARG2(arg));
+	if ( register_102(s,rank,0) < 0 )
+		STOQ(-1,*rp);
+	else
+		*rp = 0;
+}
+
+/*
  try_bind_listen(port)
 */
 
@@ -425,17 +498,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 +850,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 +958,37 @@ void Pox_recv_raw_cmo(NODE arg,Obj *rp)
 	ox_read_cmo(s,rp);
 }
 
+void Pox_send_cmo_102(NODE arg,Obj *rp)
+{
+	FILE *fp;
+	int rank = QTOS((Q)ARG0(arg));
+
+	fp = iofp_102[rank].out;
+	write_cmo(fp,(Obj)ARG1(arg));
+	/* flush always */
+	flush_stream_102(rank);
+	*rp = 0;
+}
+
+void Pox_recv_cmo_102(NODE arg,Obj *rp)
+{
+	FILE *fp;
+	int rank = QTOS((Q)ARG0(arg));
+
+	fp = iofp_102[rank].in;
+	read_cmo(fp,rp);
+}
+
 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 +1004,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 +1301,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 +1341,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
@@ -1282,4 +1388,72 @@ int validate_ox_plot_stream(int index)
 	ox_flush_stream_force(m_c_tab[i].c);
 #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);
+		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);
+		flush_stream_force_102(rank);
+	}
+	iofp_102[rank].conv = c == rc ? 0 : 1;
+	iofp_102[rank].socket = 0;
+	return 0;
+}
+
+extern int ox_batch;
+
+void flush_stream_102(int rank)
+{
+	if ( !ox_batch )
+		flush_stream_force_102(rank);
+}
+
+void 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
 }