io fixes

TODO.FACTOR.txt

@@ -1,8 +1,6 @@
-- non-blocking client socket
 - input style after clicking link
 - fedit broken with listener
 - maple-like: press enter at old commands to evaluate there
-- sending ^C on socket
 - read#
 - enforce bottom-up in native bootstrap
 - fix up native file/line info

library/platform/native/stream.factor

@@ -88,7 +88,8 @@ USE: unparser
     <fd-stream> [ ":" swap unparse cat3 "client" set ] extend ;
 
 : <client> ( host port -- stream )
-    2dup client-socket <client-stream> ;
+    #! fflush yields until connection is established.
+    2dup client-socket <client-stream> dup fflush ;
 
 : accept ( server -- client )
     #! Accept a connection from a server socket.

native/fd.c

@@ -26,11 +26,25 @@ void primitive_can_read_line(void)
 /* Return true if something was read */
 bool read_step(PORT* port)
 {
-	FIXNUM amount = read(port->fd,
+	FIXNUM  amount = 0;
+
+	if(port->type == PORT_RECV)
+	{
+		/* try reading OOB data. */
+		amount = recv(port->fd,
+			port->buffer + 1,
+			port->buffer->capacity * 2,
+			MSG_OOB);
+	}
+
+	if(amount <= 0)
+	{
+		amount = read(port->fd,
 			port->buffer + 1,
 			port->buffer->capacity * 2);
+	}
 
-	if(amount == -1)
+	if(amount < 0)
 	{
 		if(errno != EAGAIN)
 			io_error(__FUNCTION__);

native/iomux.c

@@ -231,9 +231,13 @@ CELL next_io_task(void)
 	if(!reading && !writing)
 		critical_error("next_io_task() called with no IO tasks",0);
 
-	select(read_fd_count > write_fd_count ? read_fd_count : write_fd_count,
-		&read_fd_set,&write_fd_set,NULL,NULL);
+	set_up_fd_set(&except_fd_set,
+		read_fd_count,read_io_tasks);
 
+	select(read_fd_count > write_fd_count ? read_fd_count : write_fd_count,
+		&read_fd_set,&write_fd_set,&except_fd_set,NULL);
+	
+		for(i = 0; i < read_fd_count; i++) if(FD_ISSET(i,&except_fd_set)) exit(1);/* write(3,"FUBAR\n",6); */
 	callback = perform_io_tasks(&read_fd_set,read_fd_count,read_io_tasks);
 	if(callback != F)
 		return callback;

native/port.h

@@ -1,24 +1,27 @@
-typedef enum { PORT_READ, PORT_WRITE, PORT_SPECIAL } PORT_MODE;
+typedef enum { PORT_READ, PORT_RECV, PORT_WRITE, PORT_SPECIAL } PORT_MODE;
 
 typedef struct {
 	CELL header;
-	/* one of PORT_READ or PORT_WRITE */
+	/* one of PORT_READ, PORT_RECV, PORT_WRITE or PORT_SPECIAL */
 	PORT_MODE type;
 	FIXNUM fd;
 	STRING* buffer;
+
+	/* top of buffer */
+	CELL buf_fill;
+	/* current read/write position */
+	CELL buf_pos;
+
 	/* tagged partial line used by read_line_fd */
 	CELL line;
 	/* is it ready to be returned? */
 	bool line_ready;
+
 	/* tagged client info used by accept_fd */
 	CELL client_host;
 	CELL client_port;
 	/* untagged fd of accepted connection */
 	CELL client_socket;
-	/* top of buffer */
-	CELL buf_fill;
-	/* current read/write position */
-	CELL buf_pos;
 } PORT;
 
 PORT* untag_port(CELL tagged);

native/socket.c

@@ -25,8 +25,8 @@ int make_client_socket(const char* hostname, uint16_t port)
 	if(sock < 0)
 		io_error(__FUNCTION__);
 
-	/* if(fcntl(sock,F_SETFL,O_NONBLOCK,1) == -1)
-		io_error(__FUNCTION__); */
+	if(fcntl(sock,F_SETFL,O_NONBLOCK,1) == -1)
+		io_error(__FUNCTION__);
 
 	/* Connect to the server. */
 	init_sockaddr(&servername,hostname,port);
@@ -47,7 +47,7 @@ void primitive_client_socket(void)
 	uint16_t p = (uint16_t)to_fixnum(dpop());
 	char* host = to_c_string(untag_string(dpop()));
 	int sock = make_client_socket(host,p);
-	dpush(tag_object(port(PORT_READ,sock)));
+	dpush(tag_object(port(PORT_RECV,sock)));
 	dpush(tag_object(port(PORT_WRITE,sock)));
 }
 
@@ -99,7 +99,7 @@ CELL accept_connection(PORT* p)
 	struct sockaddr_in clientname;
 	size_t size = sizeof(clientname);
 	
-	int oobinline = 1;
+	/* int oobinline = 1; */
 
 	int new = accept(p->fd,(struct sockaddr *)&clientname,&size);
 	if(new < 0)
@@ -110,8 +110,8 @@ CELL accept_connection(PORT* p)
 			io_error(__FUNCTION__);
 	}
 
-	if(setsockopt(new,SOL_SOCKET,SO_OOBINLINE,&oobinline,sizeof(int)) < 0)
-		io_error(__FUNCTION__);
+	/* if(setsockopt(new,SOL_SOCKET,SO_OOBINLINE,&oobinline,sizeof(int)) < 0)
+		io_error(__FUNCTION__); */
 
 	p->client_host = tag_object(from_c_string(inet_ntoa(
 		clientname.sin_addr)));
@@ -126,6 +126,6 @@ void primitive_accept_fd(void)
 	PORT* p = untag_port(dpop());
 	dpush(p->client_host);
 	dpush(p->client_port);
-	dpush(tag_object(port(PORT_READ,p->client_socket)));
+	dpush(tag_object(port(PORT_RECV,p->client_socket)));
 	dpush(tag_object(port(PORT_WRITE,p->client_socket)));
 }