Looks like basic IO is workikng now. (TODO: do the actual encoding)

1 files changed, 127 insertions, 40 deletions

diff --git a/src/quotprint.c b/src/quotprint.c
index 902d265..75aa57f 100644
--- a/src/quotprint.c
+++ b/src/quotprint.c
@@ -28,6 +28,12 @@
 #	define REGISTER /*no-op*/
 #endif
 
+#if 0
+#	define LOGD(...) fprintf(stderr, __VA_ARGS__)
+#else
+#	define LOGD(...)
+#endif
+
 #define STR_QUOT_(S) #S
 #define STR_QUOT(S) STR_QUOT_(S)
 
@@ -76,6 +82,9 @@ typedef struct App {
 } App;
 
 
+PRIVATE(char) endOfFile;
+
+
 INLINE(void)
 printHelp( void ){
 	printf("\n"
@@ -134,34 +143,45 @@ mainOfOutputThread( void*_ ){
 	/* TODO shutdown if error on other thread */
 	int err;
 	#define BUF (this->dstBufReadyToFlush)
-	#define LEN *(((int*)0) \
+	#define LEN (*(int*)(((uintptr_t)0) \
 		+ ((uintptr_t)(BUF == this->dstBufOne.buf) * (uintptr_t)&this->dstBufOne.len) \
-		+ ((uintptr_t)(BUF != this->dstBufOne.buf) * (uintptr_t)&this->dstBufTwo.len) \
-	)
+		+ ((uintptr_t)(BUF == this->dstBufTwo.buf) * (uintptr_t)&this->dstBufTwo.len) \
+	))
 	DEFINE_App(this, _);
 	{
 		err = pthread_mutex_lock(&this->mutxDst);  assert(!err);
 	}awaitBuffer:{
+		LOGD("output - await BUF ...\n");
 		while( !BUF )
 			pthread_cond_wait(&this->condDst, &this->mutxDst);
+		if( BUF == &endOfFile )
+			{ goto onSrcEof; }
+		LOGD("output - BUF ready\n");
 		goto writeChunk;
 	}writeChunk:{
 		assert(LEN > 0);  assert(BUF);
 		err = fprintf(stdout, "%.*s", LEN, BUF);
-		if( err ){
+		if( err < 0 ){
 			err = -errno;
 			/*TODO maybe handle closed sink gracefully*/
 			this->flg |= FLG_isError;
 			fprintf(stderr, "ERRNO_%d: fprintf(stdout)\n", -err);
 		}
 		/*clear buf and set unassigned to inidicate we're ready for another one*/
+		LOGD("output - wrote %d bytes from dstBuf%s to the world\n",
+			err, (BUF==this->dstBufOne.buf)?"One":"Two");
 		LEN = 0;
 		BUF = NULL;
 		err = pthread_cond_broadcast(&this->condDst);  assert(!err);
 		goto awaitBuffer;
-	}{
+	}onSrcEof:{
+		LOGD("output - EOF received\n");
+		err = 0; goto resolveWithErr;
+	}resolveWithErr:{
+		assert(err == 0);
+		LOGD("RETURN from %s()\n", __func__);
 		err = pthread_mutex_unlock(&this->mutxDst);  assert(!err);
-		assert(!"TODO_3LHezbEJ2sjcryPR");
+		return NULL;
 	}
 	#undef BUF
 	#undef LEN
@@ -180,8 +200,10 @@ mainOfInputThread( void*_ ){
 		err = pthread_mutex_lock(&this->mutxSrc);  assert(!err);
 		goto awaitBuffer;
 	}awaitBuffer:{
+		LOGD("innput - await BUF ...\n");
 		while( !BUF )
 			pthread_cond_wait(&this->condSrc, &this->mutxSrc);
+		LOGD("innput - BUF ready\n");
 		goto readNextChunk;
 	}readNextChunk:{
 		#define SRC stdin
@@ -194,19 +216,30 @@ mainOfInputThread( void*_ ){
 			fprintf(stderr, "ERRNO_%d: " STR_QUOT(PROJECT_NAME) ": fread()\n", -err);
 			goto resolveVoid;
 		}
-		if( err == 0 && feof(SRC) ){
-			assert(!"TODO_JcpEQzXe09haSQBK EOF");
-		}
+		if( err == 0 && feof(SRC) )
+			{ goto onSrcEof; }
 		assert(err > 0);
 		/* mark buffer filled and unassing to indicate we're ready for another
 		 * one to fill */
+		LOGD("Filled %d bytes into srcBuf%s\n", err, (BUF==this->srcBufOne.buf)?"One":"Two");
 		LEN = err;
+		/*make sure slot is free for assignment*/
+		while( this->srcBufReadyToConsume ){
+			err = pthread_cond_wait(&this->condSrc, &this->mutxSrc);  assert(!err);
+		}
+		this->srcBufReadyToConsume = BUF;
 		BUF = NULL;
 		err = pthread_cond_broadcast(&this->condSrc);  assert(!err);
-		err = pthread_mutex_unlock(&this->mutxSrc);  assert(!err);
 		goto awaitBuffer;
 		#undef SRC
+	}onSrcEof:{
+		LOGD("innput - EOF\n");
+		this->srcBufReadyToConsume = &endOfFile;
+		err = pthread_cond_broadcast(&this->condSrc);  assert(!err);
+		goto resolveVoid;
 	}resolveVoid:{
+		LOGD("RETURN from %s()\n", __func__);
+		err = pthread_mutex_unlock(&this->mutxSrc);  assert(!err);
 		return NULL;
 	}
 	#undef BUF
@@ -214,39 +247,49 @@ mainOfInputThread( void*_ ){
 }
 
 
-INLINE(int)
-quotPrintEncode( App*this ){
-	int err, iSrc, iDst;
+INLINE(void*)
+mainOfEncode( void*_ ){
+	DEFINE_App(this, _);
+	int err, iSrc = 0, iDst = 0;
 	{
 		#define BUF_DST (this->dstBufReadyToFill)
-		#define LEN_DST *(((int*)0) \
+		#define DST_CAP BUF_CAP
+		#define LEN_DST (*(int*)(((uintptr_t)0) \
 			+ ((uintptr_t)(BUF_DST == this->dstBufOne.buf) * (uintptr_t)&this->dstBufOne.len) \
-			+ ((uintptr_t)(BUF_DST == this->dstBufTwo.buf) * (uintptr_t)&this->dstBufTwo.len) \
-		)
+			+ ((uintptr_t)(BUF_DST != this->dstBufOne.buf) * (uintptr_t)&this->dstBufTwo.len) \
+		))
 		#define BUF_SRC (this->srcBufReadyToConsume)
-		#define LEN_SRC *(((int*)0) \
+		#define SRC_CAP BUF_CAP
+		#define LEN_SRC (*(int*)(((uintptr_t)0) \
 			+ ((uintptr_t)(BUF_SRC == this->srcBufOne.buf) * (uintptr_t)&this->srcBufOne.len) \
 			+ ((uintptr_t)(BUF_SRC == this->srcBufTwo.buf) * (uintptr_t)&this->srcBufTwo.len) \
-		)
+		))
 	}awaitBuffers:{
+		LOGD("encoder - await DST buf\n");
 		err = pthread_mutex_lock(&this->mutxDst);  assert(!err);
 		while( !BUF_DST )
 			pthread_cond_wait(&this->condDst, &this->mutxDst);
 		err = pthread_mutex_unlock(&this->mutxDst);  assert(!err);
 	}{
+		LOGD("encoder - await SRC buf\n");
 		err = pthread_mutex_lock(&this->mutxSrc);  assert(!err);
 		while( !BUF_SRC )
 			pthread_cond_wait(&this->condSrc, &this->mutxSrc);
 		err = pthread_mutex_unlock(&this->mutxSrc);  assert(!err);
 	}{
-		/*both buffers are ready now*/
-		iSrc = iDst = 0;
+		LOGD("encoder - both buffers ready\n");
+		if( BUF_SRC == &endOfFile ){
+			/* input has ended. we MUST ensure that we really flush what
+			 * we maybe still have in dst. So assert dst is full and handle it
+			 * there. */
+			goto onDstBufFull;
+		}
 		for(;;){
-			if( iDst >= LEN_DST )
-				{ assert(!"TODO_HYo5e9nXVCpDf8UL"); }
+			if( iDst >= DST_CAP )
+				{ goto onDstBufFull; }
 			if( iSrc >= LEN_SRC )
-				{ assert(!"TODO_WUrQL6GwlSueGI6v"); }
-			uint_fast8_t c = BUF_SRC[iSrc];
+				{ goto onSrcBufDrained; }
+			uint_fast8_t c = BUF_SRC[iSrc++];
 			switch( c ){
 			case 0x00: case 0x01: case 0x02: case 0x03: case 0x04: case 0x05: case 0x06: case 0x07:
 			case 0x08: case 0x09: case 0x0A: case 0x0B: case 0x0C: case 0x0D: case 0x0E: case 0x0F:
@@ -283,6 +326,43 @@ quotPrintEncode( App*this ){
 				BUF_DST[iDst++] = c;
 			}
 		}
+		assert(!"unreachable");
+	}onDstBufFull:{
+		/* hand over dst buf as soon consumer is ready to take it */
+		err = pthread_mutex_lock(&this->mutxDst);  assert(!err);
+		while( this->dstBufReadyToFlush )
+			{ err = pthread_cond_wait(&this->condDst, &this->mutxDst);  assert(!err); }
+		LOGD("encoder - dstBufReadyToFlush := dstBuf%s\n",
+			(BUF_DST==this->dstBufOne.buf)?"One":"Two");
+		LOGD("encoder - dstBuf%s.len := %d\n", (BUF_DST==this->dstBufOne.buf)?"One":"Two", iDst);
+		LOGD("encoder - dstBuf%s := NULL\n", (BUF_DST==this->dstBufOne.buf)?"One":"Two");
+		this->dstBufReadyToFlush = BUF_DST;
+		LEN_DST = iDst;
+		BUF_DST = NULL;
+		err = pthread_cond_broadcast(&this->condDst);  assert(!err);
+		if( BUF_SRC == &endOfFile ){
+			/* src eof, so mark our eof too (as soon prev buffer is consumed) */
+			while( this->dstBufReadyToFlush )
+				{ err = pthread_cond_wait(&this->condDst, &this->mutxDst);  assert(!err); }
+			LOGD("encoder - src did report EOF. Therefore report EOF too\n");
+			this->dstBufReadyToFlush = &endOfFile;
+			err = pthread_cond_broadcast(&this->condDst);  assert(!err);
+			err = pthread_mutex_unlock(&this->mutxDst);  assert(!err);
+			err = 0; goto resolveWithErr;
+		}else{
+			err = pthread_mutex_unlock(&this->mutxDst);  assert(!err);
+		}
+		goto awaitBuffers;
+	}onSrcBufDrained:{
+		err = pthread_mutex_lock(&this->mutxSrc);  assert(!err);
+		while( this->srcBufReadyToFill )
+			{ pthread_cond_wait(&this->condSrc, &this->mutxSrc); }
+		this->srcBufReadyToFill = BUF_SRC;
+		LEN_SRC = 0;
+		BUF_SRC = NULL;
+		err = pthread_cond_broadcast(&this->condSrc);  assert(!err);
+		err = pthread_mutex_unlock(&this->mutxSrc);  assert(!err);
+		goto awaitBuffers;
 	}/*update buffer states*/{
 		err = pthread_mutex_lock(&this->mutxDst);  assert(!err);
 		LEN_DST = iDst;
@@ -294,12 +374,17 @@ quotPrintEncode( App*this ){
 		BUF_SRC = NULL;
 		err = pthread_mutex_unlock(&this->mutxSrc);  assert(!err);
 		goto awaitBuffers;
-	}/*resolveWithErr*/{
-		assert(err <= 0);
+	}resolveWithErr:{
+		assert(err == 0);
+		LOGD("RETURN from %s()\n", __func__);
 		pthread_mutex_unlock(&this->mutxSrc);
-		return err;
-		#undef BUF
-		#undef LEN
+		return NULL;
+		#undef BUF_DST
+		#undef DST_CAP
+		#undef LEN_DST
+		#undef BUF_SRC
+		#undef SRC_CAP
+		#undef LEN_SRC
 	}
 }
 
@@ -314,10 +399,17 @@ INLINE(int)
 run( App*this ){
 	int err;
 	/*assign initial buffers BEFORE we lanch threads, so we dont need mutx*/{
+		LOGD("Initial buffer assignments\n");
+		LOGD("srcBufReadyToFill := srcBufOne.buf\n");
+		LOGD("dstBufReadyToFill := dstBufOne.buf\n");
 		this->srcBufReadyToFill = this->srcBufOne.buf;
 		this->dstBufReadyToFill = this->dstBufOne.buf;
-	}/*launch a thread for writing*/{
-		err = pthread_create(&this->thrdWr, NULL, mainOfOutputThread, this);
+	}/*launch a thread for processing*/{
+		void *(*processor)(void*) = (this->flg & FLG_isEnc)
+			? mainOfEncode
+			: NULL //TODO mainOfDecode
+		;
+		err = pthread_create(&this->thrdWr, NULL, processor, this);
 		if( err ){
 			assert(err > 0);
 			err = -err; goto resolveWithErr;
@@ -328,16 +420,11 @@ run( App*this ){
 			assert(err > 0);
 			err = -err; goto resolveWithErr;
 		}
-	}/*use main thread to do the processing*/{
-		if( this->flg & FLG_isEnc ){
-			err = quotPrintEncode(this);  assert(err <= 0);
-			goto resolveWithErr;
-		}else if( this->flg & FLG_isDec ){
-			err = quotPrintDecode(this);  assert(err <= 0);
-			goto resolveWithErr;
-		}
-		assert(!"unreachable");
+	}/*use main thread to do the output writing*/{
+		mainOfOutputThread(this);
+		err = 0;
 	}resolveWithErr:{
+		/* TODO make sure to error out if ANY of the threads had an error*/
 		return err;
 	}
 }