summaryrefslogtreecommitdiff
path: root/src/quotprint.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/quotprint.c')
-rw-r--r--src/quotprint.c167
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;
}
}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-27 11:22:29 +0000