pipe with threads

[Gennady Proskurin]

I use usual scenario for writing data to stdin of some other program:
1. pipe, fork
2. Child: dup2, exec. Parent: write

When my program is single-threaded (or may be multithreaded, with only
one thread running this scenario), all works fine.
But when this scenario executed concurently by many threads, the reading
process sometimes doesn't see, that pipe was closed and reading process
is stuck in read() (piperd wchan), and write process is stuck in waitpid.

Is it a bug somewhere or I missing something?

Test program attached.
I compile it with "cc -o pipetest pipetest.c -pthread"
And run: "while [ 1 ]; do ./pipetest ; done"

My system: fresh 9-currend/amd64, smp (2 CPU)

-------------- next part --------------
#include <stdio.h>
#include <unistd.h>
#include <signal.h>
#include <sys/wait.h>
#include <pthread.h>

static char data[10];

void* thr_func(void* arg)
{
	int pipe_fd[2];
        int pid;
	const char* buf = data;
	int write_remain = sizeof(data);
	int waitpid_status;
        int waitpid_res;
        int bar_res;

	if (pipe(pipe_fd)<0) {
		perror("pipe()");
		_exit(1);
	}

        printf("fd: %i %i\n", pipe_fd[0], pipe_fd[1]);

	pid = fork();
	switch(pid) {
		case -1:
			perror("fork()");
			_exit(1);
		case 0:
			if (close(pipe_fd[1])) {
                                perror("close()");
                                _exit(1);
                        }
			if (dup2(pipe_fd[0], STDIN_FILENO) == -1) {
				perror("dup2()");
				_exit(1);
			}
			execl("/bin/sh", "sh", "-c", "cat > /dev/null", 0);
			perror("exec()");
			_exit(1);
	}
	if (close(pipe_fd[0])) {
                perror("close()");
                _exit(1);
        }

        // write data
	while (write_remain>0) {
		int w = write(pipe_fd[1], buf, write_remain);
		if (w>0) {
			buf += w;
			write_remain -= w;
                        printf("write:%i\n", w);
		} else {
			perror("write()");
                        _exit(1);
		}
	}
	if (close(pipe_fd[1])) {
                perror("close()");
                _exit(1);
        }

        // wait child
	waitpid_res = waitpid(pid, &waitpid_status, 0);
	if (waitpid_res != pid) {
		perror("waitpid()");
		_exit(1);
	}

	return 0;
}

int main()
{
        pthread_t thr1, thr2;
	sigset_t mask;

        // block SIGCHLD
	sigemptyset(&mask);
	sigaddset(&mask, SIGCHLD);
        if (sigprocmask(SIG_BLOCK, &mask, 0)) {
		perror("sigprocmask()");
		_exit(1);
	}

        // create threads
        if (pthread_create(&thr1,0,thr_func,0) || pthread_create(&thr2,0,thr_func,0)) {
                perror("pthread_create()");
                _exit(1);
        }

        // join threads
        if (pthread_join(thr1,0) || pthread_join(thr2,0)) {
                perror("pthread_join()");
                _exit(1);
        }

        printf("OK\n");

	return 0;
}