New option f-setpipe-sz

CHANGES

@@ -161,6 +161,11 @@ Features:
 	ABSTRACT_RECVFROM_CLIENT_BIND_TEMPNAME ABSTRACT_RECVFROM_SENDTO_BIND_TEMPNAME
 	Thanks to Kai Lüke for sending an initial patch.
 
+	New option f-setpipe-sz (pipesz) sets the pipe size on systems that
+	provide ioctl F_SETIPE_SZ.
+	Filan prints the current value.
+	Tests: STDIN_F_SETPIPE_SZ EXEC_F_SETPIPE_SZ
+
 Corrections:
 	When a sub process (EXEC, SYSTEM) terminated with exit code other than
 	0, its last sent data might have been lost depending on timing of read/

VERSION

@@ -1 +1 @@
-"1.7.4.5+"
+"1.7.4.5+setpipe"

doc/socat.yo

@@ -1984,6 +1984,21 @@ enddit()
 startdit()enddit()nl()
 
 
+label(GROUP_PIPE)em(bf(PIPE options))
+
+These options may be applied to pipes (fifos).
+
+startdit()
+label(OPTION_F_SETPIPE_SZ)dit(bf(tt(f-setpipe-sz=<int>)))
+dit(bf(tt(setpipe=<int>)))
+   Set the number of bytes a pipe can buffer. Where more bytes are written the
+   writing process might block. When more bytes are written in a single
+   tt(write()) the writing process blocks and might never recover.
+enddit()
+
+startdit()enddit()nl()
+
+
 label(GROUP_ADDRS)em(bf(General address options))
 
 These options may be applied to all address types. They change some process

filan.c

@@ -146,10 +146,10 @@ int filan_fd(int fd, FILE *outfile) {
 	    ;
 #if HAVE_POLL
 	 if (Poll(&ufds, 1, 0) < 0) {
-	    Warn4("poll({%d, %hd, %hd}, 1, 0): %s",
+	    Warn4("\tpoll({%d, %hd, %hd}, 1, 0): %s",
 		   ufds.fd, ufds.events, ufds.revents, strerror(errno));
 	 } else {
-	    fputs("poll: ", outfile);
+	    fputs("\tpoll: ", outfile);
 	    if (ufds.revents & POLLIN)   fputs("IN,", outfile);
 	    if (ufds.revents & POLLPRI)  fputs("PRI,", outfile);
 	    if (ufds.revents & POLLOUT)  fputs("OUT,", outfile);
@@ -378,6 +378,9 @@ int filan_stat(
   if (statfd >= 0) { /*!indent */
    switch (buf->st_mode&S_IFMT) {
    case (S_IFIFO):	/* 1, FIFO */
+#if defined(F_GETPIPE_SZ)
+      fprintf(outfile, "\tF_GETPIPE_SZ=%d", Fcntl(statfd, F_GETPIPE_SZ));
+#endif
       break;
    case (S_IFCHR):	/* 2, character device */
       cdevan(statfd, outfile);

sycls.c

@@ -574,6 +574,7 @@ ssize_t Write(int fd, const void *buf, size_t count) {
    return result;
 }
 
+/* fcntl() without value */
 int Fcntl(int fd, int cmd) {
    int result, _errno;
    if (!diag_in_handler) diag_flush();
@@ -581,7 +582,8 @@ int Fcntl(int fd, int cmd) {
    Debug2("fcntl(%d, %d)", fd, cmd);
 #endif /* WITH_SYCLS */
    result = fcntl(fd, cmd);
-   if (!diag_in_handler) diag_flush();
+   if (!diag_in_handler)
+      diag_flush();
 #if WITH_SYCLS
    _errno = errno;
    Debug1("fcntl() -> 0x%x", result);
@@ -590,6 +592,25 @@ int Fcntl(int fd, int cmd) {
    return result;
 }
 
+/* fcntl() with int value */
+int Fcntl_i(int fd, int cmd, int arg) {
+   int result, _errno;
+   if (!diag_in_handler) diag_flush();
+#if WITH_SYCLS
+   Debug3("fcntl(%d, %d, 0x%x)", fd, cmd, arg);
+#endif /* WITH_SYCLS */
+   result = fcntl(fd, cmd, arg);
+   _errno = errno;
+   if (!diag_in_handler)
+      diag_flush();
+#if WITH_SYCLS
+   Debug1("fcntl() -> 0x%x", result);
+#endif /* WITH_SYCLS */
+   errno = _errno;
+   return result;
+}
+
+/* fcntl() with long value */
 int Fcntl_l(int fd, int cmd, long arg) {
    int result, _errno;
    if (!diag_in_handler) diag_flush();
@@ -598,7 +619,8 @@ int Fcntl_l(int fd, int cmd, long arg) {
 #endif /* WITH_SYCLS */
    result = fcntl(fd, cmd, arg);
    _errno = errno;
-   if (!diag_in_handler) diag_flush();
+   if (!diag_in_handler)
+      diag_flush();
 #if WITH_SYCLS
    Debug1("fcntl() -> 0x%x", result);
 #endif /* WITH_SYCLS */

sycls.h

@@ -61,6 +61,7 @@ int Pipe(int filedes[2]);
 ssize_t Read(int fd, void *buf, size_t count);
 ssize_t Write(int fd, const void *buf, size_t count);
 int Fcntl(int fd, int cmd);
+int Fcntl_i(int fd, int cmd, int arg);
 int Fcntl_l(int fd, int cmd, long arg);
 int Fcntl_lock(int fd, int cmd, struct flock *l);
 #if WITH_SYCLS

sysutils.c

@@ -44,7 +44,7 @@ ssize_t writefull(int fd, const void *buff, size_t bytes) {
 	 default: return -1;
 	 }
       } else if (writt+chk < bytes) {
-	 Warn4("write(%d, %p, "F_Zu"): only wrote "F_Zu" bytes, trying to continue ",
+	 Warn4("write(%d, %p, "F_Zu"): only wrote "F_Zu" bytes, trying to continue (rev.direction is blocked)",
 	       fd, (const char *)buff+writt, bytes-writt, chk);
 	 writt += chk;
       } else {
@@ -52,6 +52,7 @@ ssize_t writefull(int fd, const void *buff, size_t bytes) {
 	 break;
       }
    }
+   Notice3("write(%d, %p, "F_Zu") completed", fd, (const char *)buff, bytes);
    return writt;
 }
 

test.sh

@@ -7652,24 +7652,26 @@ touch "${tf}2"
 sleep 1
 # read from the first file
 sh -c "$CMD" 2>"$te"
-if [ $? -ne 0 ]; then # command failed
+rc=$?
-    $PRINTF "${FAILED}:\n"
+if [ $rc -ne 0 ]; then # command failed
+    $PRINTF "${FAILED} (rc=$rc):\n"
     echo "$CMD"
-    cat "$te"
+    cat "$te" >&2
     numFAIL=$((numFAIL+1))
     listFAIL="$listFAIL $N"
 else
 # check which file has a later atime stamp
 if [ $(ls -ltu "${tf}1" "${tf}2" |head -1 |sed 's/.* //') != "${tf}2" ];
 then
-   $PRINTF "$FAILED: $TRACE $SOCAT:\n"
+    $PRINTF "$FAILED (bad order):\n"
-   echo "$CMD"
+    echo "$CMD" >&2
     cat "$te"
     numFAIL=$((numFAIL+1))
     listFAIL="$listFAIL $N"
 else
     $PRINTF "$OK\n"
-   if [ -n "$debug" ]; then cat "$te"; fi
+    if [ "$VERBOSE" ]; then echo "$CMD"; fi
+    if [ "$DEBUG" ];   then cat "${te}" >&2; fi
     numOK=$((numOK+1))
 fi # wrong time stamps
 fi # command ok
@@ -17998,6 +18000,133 @@ PORT=$((PORT+1))
 N=$((N+1))
 
 
+# Test the new f-setpipe-sz option on a STDIN pipe
+NAME=STDIN_F_SETPIPE_SZ
+case "$TESTS" in
+*%$N%*|*%functions%*|*%filan%*|*%dual%*|*%stdio%*|*%exec%*|*%pipe%*|*%f-setpipe-sz%*|*%$NAME%*)
+TEST="$NAME: f-setpipe-sz on STDIN"
+# Start Socat in a shell pipe and have it calling Filan via EXEC and nofork
+# Check Filan output if pipe size of its input pipe is modified.
+if ! eval $NUMCOND; then :;
+# Remove unneeded checks, adapt lists of the remaining ones
+elif ! $(type true >/dev/null 2>&1); then
+    $PRINTF "test $F_n $TEST... ${YELLOW}true not available${NORMAL}\n" $N
+    numCANT=$((numCANT+1))
+    listCANT="$listCANT $N"
+elif ! F=$(testfeats STDIO EXEC); then
+    $PRINTF "test $F_n $TEST... ${YELLOW}Feature $F not configured in $SOCAT${NORMAL}\n" $N
+    numCANT=$((numCANT+1))
+    listCANT="$listCANT $N"
+elif ! A=$(testaddrs STDIO EXEC); then
+    $PRINTF "test $F_n $TEST... ${YELLOW}Address $A not available in $SOCAT${NORMAL}\n" $N
+    numCANT=$((numCANT+1))
+    listCANT="$listCANT $N"
+elif ! o=$(testoptions f-setpipe-sz nofork) >/dev/null; then
+    $PRINTF "test $F_n $TEST... ${YELLOW}Option $o not available in $SOCAT${NORMAL}\n" $N
+    numCANT=$((numCANT+1))
+    listCANT="$listCANT $N"
+else
+tf="$td/test$N.stdout"
+te="$td/test$N.stderr"
+tdiff="$td/test$N.diff"
+da="test$N $(date) $RANDOM"
+newport tcp4 	# or whatever proto, or drop this line
+# Find the default pipe size
+PIPESZ="$(echo |$FILAN -n 0 |grep "0:" |head -n 1 |sed 's/.*F_GETPIPE_SZ=\([0-9][0-9]*\).*/\1/')"
+PIPESZ2=$((2*PIPESZ))
+CMD0="$TRACE $SOCAT $opts STDIN,f-setpipe-sz=$PIPESZ2!!STDOUT EXEC:$FILAN,nofork"
+printf "test $F_n $TEST... " $N
+true |$CMD0 >"${tf}" 2>"${te}0"
+rc0=$?
+PIPESZ2b="$(cat "$tf" |grep "0:" |head -n 1 |sed 's/.*F_GETPIPE_SZ=\([0-9][0-9]*\).*/\1/')"
+if [ "$rc0" -ne 0 ]; then
+    $PRINTF "$FAILED\n"
+    echo "$CMD0"
+    cat "${te}0" >&2
+    numFAIL=$((numFAIL+1))
+    listFAIL="$listFAIL $N"
+    namesFAIL="$namesFAIL $NAME"
+elif ! diff <(echo $PIPESZ2) <(echo $PIPESZ2b) >$tdiff; then
+    $PRINTF "$FAILED\n"
+    echo "$CMD0"
+    cat "${te}0" >&2
+    echo "diff:" >&2
+    cat "$tdiff" >&2
+    numFAIL=$((numFAIL+1))
+    listFAIL="$listFAIL $N"
+    namesFAIL="$namesFAIL $NAME"
+else
+    $PRINTF "$OK\n"
+    if [ "$VERBOSE" ]; then echo "$CMD0"; fi
+    if [ "$DEBUG" ];   then cat "${te}0" >&2; fi
+    numOK=$((numOK+1))
+fi
+fi # NUMCOND
+ ;;
+esac
+N=$((N+1))
+
+# Test the new f-setpipe-sz option on EXEC with pipes
+NAME=EXEC_F_SETPIPE_SZ
+case "$TESTS" in
+*%$N%*|*%functions%*|*%filan%*|*%stdio%*|*%exec%*|*%pipe%*|*%f-setpipe-sz%*|*%$NAME%*)
+TEST="$NAME: f-setpipe-sz on EXEC with pipes"
+# Start Socat calling Filan via EXEC and pipes and f-setpipe-sz
+# Check Filan output if pipe size of both pipes is modified.
+if ! eval $NUMCOND; then :;
+# Remove unneeded checks, adapt lists of the remaining ones
+elif ! F=$(testfeats STDIO EXEC); then
+    $PRINTF "test $F_n $TEST... ${YELLOW}Feature $F not configured in $SOCAT${NORMAL}\n" $N
+    numCANT=$((numCANT+1))
+    listCANT="$listCANT $N"
+elif ! A=$(testaddrs STDIO EXEC); then
+    $PRINTF "test $F_n $TEST... ${YELLOW}Address $A not available in $SOCAT${NORMAL}\n" $N
+    numCANT=$((numCANT+1))
+    listCANT="$listCANT $N"
+elif ! o=$(testoptions pipes f-setpipe-sz) >/dev/null; then
+    $PRINTF "test $F_n $TEST... ${YELLOW}Option $o not available in $SOCAT${NORMAL}\n" $N
+    numCANT=$((numCANT+1))
+    listCANT="$listCANT $N"
+else
+tf="$td/test$N.stdout"
+te="$td/test$N.stderr"
+tdiff="$td/test$N.diff"
+# Find the default pipe size
+PIPESZ="$(echo |$FILAN -n 0 |grep "0:" |head -n 1 |sed 's/.*F_GETPIPE_SZ=\([0-9][0-9]*\).*/\1/')"
+PIPESZ2=$((2*PIPESZ))
+CMD0="$TRACE $SOCAT $opts STDIO EXEC:$FILAN,pipes,f-setpipe-sz=$PIPESZ2"
+printf "test $F_n $TEST... " $N
+$CMD0 >"$tf" 2>"${te}0"
+rc0=$?
+PIPESZ2b="$(cat "$tf" |grep "0:" |head -n 1 |sed 's/.*F_GETPIPE_SZ=\([0-9][0-9]*\).*/\1/')"
+if [ "$rc0" -ne 0 ]; then
+    $PRINTF "$FAILED\n"
+    echo "$CMD0"
+    cat "${te}0" >&2
+    numFAIL=$((numFAIL+1))
+    listFAIL="$listFAIL $N"
+    namesFAIL="$namesFAIL $NAME"
+elif ! diff <(echo $PIPESZ2) <(echo $PIPESZ2b) >$tdiff; then
+    $PRINTF "$FAILED\n"
+    echo "$CMD0 &"
+    cat "${te}0" >&2
+    echo "diff:" >&2
+    cat "$tdiff" >&2
+    numFAIL=$((numFAIL+1))
+    listFAIL="$listFAIL $N"
+    namesFAIL="$namesFAIL $NAME"
+else
+    $PRINTF "$OK\n"
+    if [ "$VERBOSE" ]; then echo "$CMD0"; fi
+    if [ "$DEBUG" ];   then cat "${te}0" >&2; fi
+    numOK=$((numOK+1))
+fi
+fi # NUMCOND
+ ;;
+esac
+N=$((N+1))
+
+
 # end of common tests
 
 ##################################################################################

xio-pipe.c

@@ -20,6 +20,9 @@ static int xioopen_fifo_unnamed(xiofile_t *sock, struct opt *opts);
 
 const struct addrdesc xioaddr_pipe   = { "PIPE",   3, xioopen_fifo,  GROUP_FD|GROUP_NAMED|GROUP_OPEN|GROUP_FIFO, 0, 0, 0 HELP("[:<filename>]") };
 
+#if defined(F_SETPIPE_SZ)
+const struct optdesc opt_f_setpipe_sz = { "f-setpipe-sz", "pipesz", OPT_F_SETPIPE_SZ, GROUP_FD, PH_FD, TYPE_INT, OFUNC_FCNTL, F_SETPIPE_SZ, 0 };
+#endif
 
 /* process an unnamed bidirectional "pipe" or "fifo" or "echo" argument with
    options */

xio-pipe.h

@@ -7,4 +7,6 @@
 
 extern const struct addrdesc xioaddr_pipe;
 
+extern const struct optdesc opt_f_setpipe_sz;
+
 #endif /* !defined(__xio_pipe_h_included) */

xio-progcall.c

@@ -291,6 +291,7 @@ int _xioopen_foxec(int xioflags,	/* XIO_RDONLY etc. */
 
    if (usepipes) {
       /* withfork usepipes */
+      struct opt *popts2 = NULL;
 
       if (rw == XIO_RDWR)
 	 sfd->dtype = XIODATA_2PIPE;
@@ -309,6 +310,8 @@ int _xioopen_foxec(int xioflags,	/* XIO_RDONLY etc. */
 	 return -1;
       }
       popts = opts;
+      if (sfd->dtype == XIODATA_2PIPE)
+	 popts2 = copyopts(popts, GROUP_ALL);
 
       if (rw != XIO_WRONLY) {
 	 applyopts_cloexec(rdpip[0], popts);
@@ -326,8 +329,11 @@ int _xioopen_foxec(int xioflags,	/* XIO_RDONLY etc. */
       /* wrpip[1]: write by socat; wrpip[0]: read by child */
       if (rw != XIO_RDONLY) {
 	 applyopts_cloexec(wrpip[1], popts);
-	 applyopts(sfd, wrpip[1], popts, PH_FD);
+	 if (sfd->dtype == XIODATA_2PIPE)
-	 applyopts(sfd, wrpip[0], copts, PH_FD);
+	    applyopts(NULL, wrpip[1], popts, PH_FD);
+	 else
+	    applyopts(NULL, wrpip[1], popts, PH_FD);
+	 applyopts(NULL, wrpip[0], copts, PH_FD);
       }
       if (sfd->howtoend == END_UNSPEC) {
 	 sfd->howtoend = END_CLOSE_KILL;

xioopts.c

@@ -548,6 +548,9 @@ const struct optname optionnames[] = {
 	IF_ANY 	  ("f-setlkw",	&opt_f_setlkw_wr)
 	IF_ANY 	  ("f-setlkw-rd",	&opt_f_setlkw_rd)
 	IF_ANY 	  ("f-setlkw-wr",	&opt_f_setlkw_wr)
+#if defined(F_SETPIPE_SZ)
+	IF_ANY    ("f-setpipe-sz",	&opt_f_setpipe_sz)
+#endif
 	IF_EXEC   ("fdin",	&opt_fdin)
 	IF_EXEC   ("fdout",	&opt_fdout)
 #ifdef FFDLY
@@ -1305,6 +1308,9 @@ const struct optname optionnames[] = {
 	IF_SOCKET ("pf",	&opt_protocol_family)
 	IF_EXEC   ("pgid",	&opt_setpgid)
 	IF_EXEC   ("pipes",	&opt_pipes)
+#if defined(F_SETPIPE_SZ)
+	IF_ANY    ("pipesz",	&opt_f_setpipe_sz)
+#endif
 #ifdef IP_PKTINFO
 	IF_IP     ("pktinfo",	&opt_ip_pktinfo)
 #endif
@@ -3375,23 +3381,34 @@ int applyopt_fcntl(
 {
 	int flag;
 
-	/* retrieve existing flag setttings */
+	if (opt->desc->type == TYPE_BOOL) {
+	   /* Retrieve existing flag settings */
 	   if ((flag = Fcntl(fd, opt->desc->major-1)) < 0) {
 		Error3("fcntl(%d, %d): %s",
 		       fd, opt->desc->major, strerror(errno));
 		return -1;
-	} else {
+	   }
 	   if (opt->value.u_bool) {
 	      flag |= opt->desc->minor;
 	   } else {
 	      flag &= ~opt->desc->minor;
 	   }
-		if (Fcntl_l(fd, opt->desc->major, flag) < 0) {
+	   if (Fcntl_i(fd, opt->desc->major, flag) < 0) {
-			Error4("fcntl(%d, %d, %d): %s",
+	      Error4("fcntl(%d, %d, 0x%x): %s",
-			       fd, opt->desc->major, flag,
+		     fd, opt->desc->major, flag, strerror(errno));
-			       strerror(errno));
 	      return -1;
 	   }
+
+	} else if (opt->desc->type == TYPE_INT) {
+	   if (Fcntl_i(fd, opt->desc->major, opt->value.u_int) < 0) {
+	      Error4("fcntl(%d, %d, 0x%x): %s",
+		     fd, opt->desc->major, opt->value.u_int, strerror(errno));
+	      return -1;
+	   }
+	} else {
+	   Error2("applyopt_fcntl(\"%s\", ...): INTERNAL: type %d not implemented",
+		  opt->desc->defname, opt->desc->type);
+	   return -1;
 	}
 	return 0;
 }

xioopts.h

@@ -351,6 +351,7 @@ enum e_optcode {
    OPT_F_SETLKW_WR,	/* fcntl with struct flock - write-lock, wait */
    OPT_F_SETLK_RD,	/* fcntl with struct flock - read-lock */
    OPT_F_SETLK_WR,	/* fcntl with struct flock - write-lock */
+   OPT_F_SETPIPE_SZ, 	/* pipe == fifo */
    OPT_GROUP,
    OPT_GROUP_EARLY,
    OPT_GROUP_LATE,

xiowrite.c

@@ -96,12 +96,11 @@ ssize_t xiowrite(xiofile_t *file, const void *buff, size_t bytes) {
       }
       if ((size_t)writt < bytes) {
 	 char infobuff[256];
-	 Warn7("sendto(%d, %p, "F_Zu", 0, %s, "F_socklen") only wrote "F_Zu" of "F_Zu" bytes",
+	 Warn7("sendto(%d, %p, "F_Zu", 0, %s, "F_socklen") only sent "F_Zu" of "F_Zu" bytes",
 	       pipe->fd, buff, bytes,
 	       sockaddr_info(&pipe->peersa.soa, pipe->salen,
 			     infobuff, sizeof(infobuff)),
 	       pipe->salen, writt, bytes);
-      } else {
       }
       {
 	 char infobuff[256];