Thread: Understanding pipes...

Going back to my gasp project in a little bit and planning to rewrite it from the ground up after getting to grips with pipes, it is that "getting to grips" part that I'm having trouble with however, here's what I have (ignore the win32 code as I haven't even checked the reference yet for those).

Code:

#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <pthread.h>

typedef struct smem
{
	size_t have, want;
	void *data;
} smem_t;

typedef struct smemv
{
	size_t have;
	smem_t *data;
} smemv_t;

#ifdef _WIN32
typedef HANDLE pipe_t;
#define INVALID_PIPE NULL
#else
typedef int pipe_t;
#define INVALID_PIPE -1
#endif

int open_pipes( pipe_t pipes[2] );
void shut_pipes( pipe_t pipes[2] );
int pipe_err( pipe_t pipe );
ssize_t rdpipe( pipe_t pipe, void *data, size_t size );
ssize_t wrpipe( pipe_t pipe, void *data, size_t size );

#define INVALID_TID -1

typedef struct worker
{
	int num;
	pthread_t tid;
	pipe_t pipes[2];
	pthread_attr_t pthread_attr;
} worker_t;

typedef void * (*Worker_t)( void *arg );

void * rdworker( worker_t *worker );
void * wrworker( worker_t *worker );

int main()
{
	int i;
	worker_t workers[2] = {{0}}, *worker, *worker_ret[2];
	for ( i = 0; i < 2; ++i )
	{
		worker = &(workers[i]);
		worker->num = i + 1;
		worker->tid = INVALID_TID;
		worker->pipes[0] = INVALID_PIPE;
		worker->pipes[1] = INVALID_PIPE;
		if ( pthread_attr_init( &(worker->pthread_attr) ) != 0 )
			goto cleanup;
	}
	
	if ( pthread_create
		(
			&(workers[0].tid)
			, &(workers[0].pthread_attr)
			, (Worker_t)rdworker
			, &(workers[0])
		) != 0
	) goto cleanup;
	
	if ( pthread_create
		(
			&(workers[1].tid)
			, &(workers[1].pthread_attr)
			, (Worker_t)wrworker
			, &(workers[1])
		) != 0
	) goto cleanup;
	
	cleanup:
	
	while ( i > 0 )
	{
		--i;
		worker = &(workers[i]);
		if ( worker->tid != INVALID_TID )
		{
			pthread_join( worker->tid, (void**)&(worker_ret[i]) );
		}
		pthread_attr_destroy( &(worker->pthread_attr) );
		worker->pipes[1] = INVALID_PIPE;
		worker->pipes[0] = INVALID_PIPE;
		worker->tid = INVALID_TID;
	}
	
	printf("Hello World!\n");
	return 0;
}

int open_pipes( pipe_t pipes[2] )
{
	pipes[0] = pipes[1] = INVALID_PIPE;
#ifdef _WIN32
#error "Unchecked calls"
	pipes[0] = CreatePipe( FILE_READ, 0 );
	pipes[1] = CreatePipe( FILE_WRITE, 0 );
	return 0;
#else
	return pipe( pipes );
#endif
}
void shut_pipes( pipe_t pipes[2] )
{
#ifdef _WIN32
	CloseHandle( pipes[1] );
	CloseHandle( pipes[0] );
#else
	close( pipes[1] );
	close( pipes[0] );
#endif
	pipes[0] = pipes[1] = INVALID_PIPE;
}
int pipe_err( pipe_t pipe )
{
	return error( pipe );
}
ssize_t rdpipe( pipe_t pipe, void *data, size_t size )
{
#ifdef _WIN32
	DWORD bytes = 0;
	ReadFile( pipe, data, size, &bytes );
	return bytes;
#else
	return read( pipe, data, size );
#endif
}
ssize_t wrpipe( pipe_t pipe, void *data, size_t size )
{
#ifdef _WIN32
	DWORD bytes = 0;
	WriteFile( pipe, data, size, &bytes );
	return bytes;
#else
	return write( pipe, data, size );
#endif
}

void * rdworker( worker_t *worker )
{
	int num = 0;
	ssize_t bytes;
	
	while (bytes = rdpipe( worker->pipes[0], &num, sizeof(int) ));
	
	printf( "Worker %d: Read %zd bytes, num = %d\n", worker->num, bytes, num );
	
	return worker;
}

void * wrworker( worker_t *worker )
{
	int num = 0x12345678;
	size_t bytes = wrpipe( worker->pipes[1], &num, sizeof(int) );
	
	printf( "Worker %d: Wrote %zd bytes, num = %d\n", worker->num, bytes, num );
	
	return worker;
}

I compiled under "cc --std=c89 %f -lpthread && ./a.out", can someone explain what I'm doing wrong please? Take your time if you want as I'm going to bed, I'll read it 2mw