Update lz4, lz5 and zstd to latest stable series

- added some more error checking in zstdmt also

C/zstdmt/README.md

@@ -1,66 +1,11 @@
 
 # Multithreading Library for [LZ4], [LZ5] and [ZStandard]
 
+- this library is a temporary solution to the threading problem with zstd, lz4 and lz5
+- future version will support threading in some way, I think ;)
+- it's based on my zstdmt version @ github...
 
-### Compression
-```
-typedef struct {
-	void *buf;		/* ptr to data */
-	size_t size;		/* current filled in buf */
-	size_t allocated;	/* length of buf */
-} LZ4MT_Buffer;
+zstdmt_compress.c
 
-/**
- * reading and writing functions
- * - you can use stdio functions or plain read/write ...
- * - a sample is given in 7-Zip ZS or lz4mt.c
- */
-typedef int (fn_read) (void *args, LZ4MT_Buffer * in);
-typedef int (fn_write) (void *args, LZ4MT_Buffer * out);
+/TR 2016-12-22
 
-typedef struct {
-	fn_read *fn_read;
-	void *arg_read;
-	fn_write *fn_write;
-	void *arg_write;
-} LZ4MT_RdWr_t;
-
-typedef struct LZ4MT_CCtx_s LZ4MT_CCtx;
-
-/* 1) allocate new cctx */
-LZ4MT_CCtx *LZ4MT_createCCtx(int threads, int level, int inputsize);
-
-/* 2) threaded compression */
-size_t LZ4MT_CompressCCtx(LZ4MT_CCtx * ctx, LZ4MT_RdWr_t * rdwr);
-
-/* 3) get some statistic */
-size_t LZ4MT_GetFramesCCtx(LZ4MT_CCtx * ctx);
-size_t LZ4MT_GetInsizeCCtx(LZ4MT_CCtx * ctx);
-size_t LZ4MT_GetOutsizeCCtx(LZ4MT_CCtx * ctx);
-
-/* 4) free cctx */
-void LZ4MT_freeCCtx(LZ4MT_CCtx * ctx);
-```
-
-### Decompression
-```
-typedef struct LZ4MT_DCtx_s LZ4MT_DCtx;
-
-/* 1) allocate new cctx */
-LZ4MT_DCtx *LZ4MT_createDCtx(int threads, int inputsize);
-
-/* 2) threaded compression */
-size_t LZ4MT_DecompressDCtx(LZ4MT_DCtx * ctx, LZ4MT_RdWr_t * rdwr);
-
-/* 3) get some statistic */
-size_t LZ4MT_GetFramesDCtx(LZ4MT_DCtx * ctx);
-size_t LZ4MT_GetInsizeDCtx(LZ4MT_DCtx * ctx);
-size_t LZ4MT_GetOutsizeDCtx(LZ4MT_DCtx * ctx);
-
-/* 4) free cctx */
-void LZ4MT_freeDCtx(LZ4MT_DCtx * ctx);
-```
-
-## Todo
-
-- add Makefile

C/zstdmt/lz4mt.h

@@ -47,6 +47,7 @@ typedef enum {
   LZ4MT_error_frame_decompress,
   LZ4MT_error_compressionParameter_unsupported,
   LZ4MT_error_compression_library,
+  LZ4MT_error_canceled,
   LZ4MT_error_maxCode
 } LZ4MT_ErrorCode;
 
@@ -107,7 +108,7 @@ LZ4MT_CCtx *LZ4MT_createCCtx(int threads, int level, int inputsize);
  * 2) threaded compression
  * - errorcheck via 
  */
-size_t LZ4MT_CompressCCtx(LZ4MT_CCtx * ctx, LZ4MT_RdWr_t * rdwr);
+size_t LZ4MT_compressCCtx(LZ4MT_CCtx * ctx, LZ4MT_RdWr_t * rdwr);
 
 /**
  * 3) get some statistic
@@ -141,7 +142,7 @@ LZ4MT_DCtx *LZ4MT_createDCtx(int threads, int inputsize);
  * 2) threaded compression
  * - return -1 on error
  */
-size_t LZ4MT_DecompressDCtx(LZ4MT_DCtx * ctx, LZ4MT_RdWr_t * rdwr);
+size_t LZ4MT_decompressDCtx(LZ4MT_DCtx * ctx, LZ4MT_RdWr_t * rdwr);
 
 /**
  * 3) get some statistic

C/zstdmt/lz4mt_common.c

@@ -32,12 +32,11 @@ unsigned LZ4MT_isError(size_t code)
  */
 const char *LZ4MT_getErrorString(size_t code)
 {
-	static const char *notErrorCode = "Unspecified error lz4mt code";
-
 	if (LZ4F_isError(lz4mt_errcode))
 		return LZ4F_getErrorName(lz4mt_errcode);
 
-	switch ((LZ4MT_ErrorCode)(0-code)) {
+	static const char *notErrorCode = "Unspecified error lz4mt code";
+	switch ((LZ4MT_ErrorCode) (0 - code)) {
 	case PREFIX(no_error):
 		return "No error detected";
 	case PREFIX(memory_allocation):

C/zstdmt/lz4mt_compress.c

@@ -155,6 +155,24 @@ LZ4MT_CCtx *LZ4MT_createCCtx(int threads, int level, int inputsize)
 	return 0;
 }
 
+/**
+ * mt_error - return mt lib specific error code
+ */
+static size_t mt_error(int rv)
+{
+	switch (rv) {
+	case -1:
+		return ERROR(read_fail);
+	case -2:
+		return ERROR(canceled);
+	case -3:
+		return ERROR(memory_allocation);
+	}
+
+	/* XXX, some catch all other errors */
+	return ERROR(read_fail);
+}
+
 /**
  * pt_write - queue for compressed output
  */
@@ -175,8 +193,8 @@ static size_t pt_write(LZ4MT_CCtx * ctx, struct writelist *wl)
 		wl = list_entry(entry, struct writelist, node);
 		if (wl->frame == ctx->curframe) {
 			int rv = ctx->fn_write(ctx->arg_write, &wl->out);
-			if (rv == -1)
-				return ERROR(write_fail);
+			if (rv < 0)
+				return mt_error(rv);
 			ctx->outsize += wl->out.size;
 			ctx->curframe++;
 			list_move(entry, &ctx->writelist_free);
@@ -239,9 +257,9 @@ static void *pt_compress(void *arg)
 		pthread_mutex_lock(&ctx->read_mutex);
 		in.size = ctx->inputsize;
 		rv = ctx->fn_read(ctx->arg_read, &in);
-		if (rv == -1) {
+		if (rv < 0) {
 			pthread_mutex_unlock(&ctx->read_mutex);
-			return (void *)ERROR(read_fail);
+			return (void*)mt_error(rv);
 		}
 
 		/* eof */
@@ -277,8 +295,7 @@ static void *pt_compress(void *arg)
 		MEM_writeLE32((unsigned char *)wl->out.buf + 0,
 			      LZ4FMT_MAGIC_SKIPPABLE);
 		MEM_writeLE32((unsigned char *)wl->out.buf + 4, 4);
-		MEM_writeLE32((unsigned char *)wl->out.buf + 8,
-			      (U32)result);
+		MEM_writeLE32((unsigned char *)wl->out.buf + 8, (U32) result);
 		wl->out.size = result + 12;
 
 		/* write result */
@@ -293,7 +310,7 @@ static void *pt_compress(void *arg)
 	return 0;
 }
 
-size_t LZ4MT_CompressCCtx(LZ4MT_CCtx * ctx, LZ4MT_RdWr_t * rdwr)
+size_t LZ4MT_compressCCtx(LZ4MT_CCtx * ctx, LZ4MT_RdWr_t * rdwr)
 {
 	int t;
 

C/zstdmt/lz4mt_decompress.c

@@ -141,6 +141,24 @@ LZ4MT_DCtx *LZ4MT_createDCtx(int threads, int inputsize)
 	return 0;
 }
 
+/**
+ * mt_error - return mt lib specific error code
+ */
+static size_t mt_error(int rv)
+{
+	switch (rv) {
+	case -1:
+		return ERROR(read_fail);
+	case -2:
+		return ERROR(canceled);
+	case -3:
+		return ERROR(memory_allocation);
+	}
+
+	/* XXX, some catch all other errors */
+	return ERROR(read_fail);
+}
+
 /**
  * pt_write - queue for decompressed output
  */
@@ -156,8 +174,8 @@ static size_t pt_write(LZ4MT_DCtx * ctx, struct writelist *wl)
 		wl = list_entry(entry, struct writelist, node);
 		if (wl->frame == ctx->curframe) {
 			int rv = ctx->fn_write(ctx->arg_write, &wl->out);
-			if (rv == -1)
-				return ERROR(write_fail);
+			if (rv < 0)
+				return mt_error(rv);
 			ctx->outsize += wl->out.size;
 			ctx->curframe++;
 			list_move(entry, &ctx->writelist_free);
@@ -185,8 +203,10 @@ static size_t pt_read(LZ4MT_DCtx * ctx, LZ4MT_Buffer * in, size_t * frame)
 		hdr.buf = hdrbuf + 4;
 		hdr.size = 8;
 		rv = ctx->fn_read(ctx->arg_read, &hdr);
-		if (rv == -1)
-			goto error_read;
+		if (rv < 0) {
+			pthread_mutex_unlock(&ctx->read_mutex);
+			return mt_error(rv);
+		}
 		if (hdr.size != 8)
 			goto error_read;
 		hdr.buf = hdrbuf;
@@ -194,8 +214,10 @@ static size_t pt_read(LZ4MT_DCtx * ctx, LZ4MT_Buffer * in, size_t * frame)
 		hdr.buf = hdrbuf;
 		hdr.size = 12;
 		rv = ctx->fn_read(ctx->arg_read, &hdr);
-		if (rv == -1)
-			goto error_read;
+		if (rv < 0) {
+			pthread_mutex_unlock(&ctx->read_mutex);
+			return mt_error(rv);
+		}
 		/* eof reached ? */
 		if (hdr.size == 0) {
 			pthread_mutex_unlock(&ctx->read_mutex);
@@ -231,8 +253,10 @@ static size_t pt_read(LZ4MT_DCtx * ctx, LZ4MT_Buffer * in, size_t * frame)
 		in->size = toRead;
 		rv = ctx->fn_read(ctx->arg_read, in);
 		/* generic read failure! */
-		if (rv == -1)
-			goto error_read;
+		if (rv < 0) {
+			pthread_mutex_unlock(&ctx->read_mutex);
+			return mt_error(rv);
+		}
 		/* needed more bytes! */
 		if (in->size != toRead)
 			goto error_data;
@@ -406,10 +430,10 @@ static size_t st_decompress(void *arg)
 		/* read new input */
 		in->size = nextToLoad;
 		rv = ctx->fn_read(ctx->arg_read, in);
-		if (rv == -1) {
+		if (rv < 0) {
 			free(in->buf);
 			free(out->buf);
-			return ERROR(read_fail);
+			return mt_error(rv);
 		}
 
 		/* done, eof reached */
@@ -435,10 +459,10 @@ static size_t st_decompress(void *arg)
 			/* have some output */
 			if (out->size) {
 				rv = ctx->fn_write(ctx->arg_write, out);
-				if (rv == -1) {
+				if (rv < 0) {
 					free(in->buf);
 					free(out->buf);
-					return ERROR(write_fail);
+					return mt_error(rv);
 				}
 			}
 
@@ -455,7 +479,7 @@ static size_t st_decompress(void *arg)
 	return 0;
 }
 
-size_t LZ4MT_DecompressDCtx(LZ4MT_DCtx * ctx, LZ4MT_RdWr_t * rdwr)
+size_t LZ4MT_decompressDCtx(LZ4MT_DCtx * ctx, LZ4MT_RdWr_t * rdwr)
 {
 	unsigned char buf[4];
 	int t, rv;
@@ -475,8 +499,8 @@ size_t LZ4MT_DecompressDCtx(LZ4MT_DCtx * ctx, LZ4MT_RdWr_t * rdwr)
 	in->buf = buf;
 	in->size = 4;
 	rv = ctx->fn_read(ctx->arg_read, in);
-	if (rv == -1)
-		return ERROR(read_fail);
+	if (rv < 0)
+		return mt_error(rv);
 	if (in->size != 4)
 		return ERROR(data_error);
 

C/zstdmt/lz5mt.h

@@ -47,6 +47,7 @@ typedef enum {
   LZ5MT_error_frame_decompress,
   LZ5MT_error_compressionParameter_unsupported,
   LZ5MT_error_compression_library,
+  LZ5MT_error_canceled,
   LZ5MT_error_maxCode
 } LZ5MT_ErrorCode;
 
@@ -105,7 +106,7 @@ LZ5MT_CCtx *LZ5MT_createCCtx(int threads, int level, int inputsize);
  * 2) threaded compression
  * - return -1 on error
  */
-size_t LZ5MT_CompressCCtx(LZ5MT_CCtx * ctx, LZ5MT_RdWr_t * rdwr);
+size_t LZ5MT_compressCCtx(LZ5MT_CCtx * ctx, LZ5MT_RdWr_t * rdwr);
 
 /**
  * 3) get some statistic
@@ -141,7 +142,7 @@ LZ5MT_DCtx *LZ5MT_createDCtx(int threads, int inputsize);
  * 2) threaded compression
  * - return -1 on error
  */
-size_t LZ5MT_DecompressDCtx(LZ5MT_DCtx * ctx, LZ5MT_RdWr_t * rdwr);
+size_t LZ5MT_decompressDCtx(LZ5MT_DCtx * ctx, LZ5MT_RdWr_t * rdwr);
 
 /**
  * 3) get some statistic

C/zstdmt/lz5mt_common.c

@@ -32,12 +32,11 @@ unsigned LZ5MT_isError(size_t code)
  */
 const char *LZ5MT_getErrorString(size_t code)
 {
-	static const char *notErrorCode = "Unspecified error lz4mt code";
-
 	if (LZ5F_isError(lz5mt_errcode))
 		return LZ5F_getErrorName(lz5mt_errcode);
 
-	switch ((LZ5MT_ErrorCode)(0-code)) {
+	static const char *notErrorCode = "Unspecified error lz4mt code";
+	switch ((LZ5MT_ErrorCode) (0 - code)) {
 	case PREFIX(no_error):
 		return "No error detected";
 	case PREFIX(memory_allocation):

C/zstdmt/lz5mt_compress.c

@@ -155,6 +155,24 @@ LZ5MT_CCtx *LZ5MT_createCCtx(int threads, int level, int inputsize)
 	return 0;
 }
 
+/**
+ * mt_error - return mt lib specific error code
+ */
+static size_t mt_error(int rv)
+{
+	switch (rv) {
+	case -1:
+		return ERROR(read_fail);
+	case -2:
+		return ERROR(canceled);
+	case -3:
+		return ERROR(memory_allocation);
+	}
+
+	/* XXX, some catch all other errors */
+	return ERROR(read_fail);
+}
+
 /**
  * pt_write - queue for compressed output
  */
@@ -175,8 +193,8 @@ static size_t pt_write(LZ5MT_CCtx * ctx, struct writelist *wl)
 		wl = list_entry(entry, struct writelist, node);
 		if (wl->frame == ctx->curframe) {
 			int rv = ctx->fn_write(ctx->arg_write, &wl->out);
-			if (rv == -1)
-				return ERROR(write_fail);
+			if (rv < 0)
+				return mt_error(rv);
 			ctx->outsize += wl->out.size;
 			ctx->curframe++;
 			list_move(entry, &ctx->writelist_free);
@@ -239,9 +257,9 @@ static void *pt_compress(void *arg)
 		pthread_mutex_lock(&ctx->read_mutex);
 		in.size = ctx->inputsize;
 		rv = ctx->fn_read(ctx->arg_read, &in);
-		if (rv == -1) {
+		if (rv < 0) {
 			pthread_mutex_unlock(&ctx->read_mutex);
-			return (void *)ERROR(read_fail);
+			return (void*)mt_error(rv);
 		}
 
 		/* eof */
@@ -277,8 +295,7 @@ static void *pt_compress(void *arg)
 		MEM_writeLE32((unsigned char *)wl->out.buf + 0,
 			      LZ5FMT_MAGIC_SKIPPABLE);
 		MEM_writeLE32((unsigned char *)wl->out.buf + 4, 4);
-		MEM_writeLE32((unsigned char *)wl->out.buf + 8,
-			      (U32)result);
+		MEM_writeLE32((unsigned char *)wl->out.buf + 8, (U32) result);
 		wl->out.size = result + 12;
 
 		/* write result */
@@ -293,7 +310,7 @@ static void *pt_compress(void *arg)
 	return 0;
 }
 
-size_t LZ5MT_CompressCCtx(LZ5MT_CCtx * ctx, LZ5MT_RdWr_t * rdwr)
+size_t LZ5MT_compressCCtx(LZ5MT_CCtx * ctx, LZ5MT_RdWr_t * rdwr)
 {
 	int t;
 

C/zstdmt/lz5mt_decompress.c

@@ -141,6 +141,24 @@ LZ5MT_DCtx *LZ5MT_createDCtx(int threads, int inputsize)
 	return 0;
 }
 
+/**
+ * mt_error - return mt lib specific error code
+ */
+static size_t mt_error(int rv)
+{
+	switch (rv) {
+	case -1:
+		return ERROR(read_fail);
+	case -2:
+		return ERROR(canceled);
+	case -3:
+		return ERROR(memory_allocation);
+	}
+
+	/* XXX, some catch all other errors */
+	return ERROR(read_fail);
+}
+
 /**
  * pt_write - queue for decompressed output
  */
@@ -156,8 +174,8 @@ static size_t pt_write(LZ5MT_DCtx * ctx, struct writelist *wl)
 		wl = list_entry(entry, struct writelist, node);
 		if (wl->frame == ctx->curframe) {
 			int rv = ctx->fn_write(ctx->arg_write, &wl->out);
-			if (rv == -1)
-				return ERROR(write_fail);
+			if (rv < 0)
+				return mt_error(rv);
 			ctx->outsize += wl->out.size;
 			ctx->curframe++;
 			list_move(entry, &ctx->writelist_free);
@@ -185,8 +203,10 @@ static size_t pt_read(LZ5MT_DCtx * ctx, LZ5MT_Buffer * in, size_t * frame)
 		hdr.buf = hdrbuf + 4;
 		hdr.size = 8;
 		rv = ctx->fn_read(ctx->arg_read, &hdr);
-		if (rv == -1)
-			goto error_read;
+		if (rv < 0) {
+			pthread_mutex_unlock(&ctx->read_mutex);
+			return mt_error(rv);
+		}
 		if (hdr.size != 8)
 			goto error_read;
 		hdr.buf = hdrbuf;
@@ -194,8 +214,10 @@ static size_t pt_read(LZ5MT_DCtx * ctx, LZ5MT_Buffer * in, size_t * frame)
 		hdr.buf = hdrbuf;
 		hdr.size = 12;
 		rv = ctx->fn_read(ctx->arg_read, &hdr);
-		if (rv == -1)
-			goto error_read;
+		if (rv < 0) {
+			pthread_mutex_unlock(&ctx->read_mutex);
+			return mt_error(rv);
+		}
 		/* eof reached ? */
 		if (hdr.size == 0) {
 			pthread_mutex_unlock(&ctx->read_mutex);
@@ -231,8 +253,10 @@ static size_t pt_read(LZ5MT_DCtx * ctx, LZ5MT_Buffer * in, size_t * frame)
 		in->size = toRead;
 		rv = ctx->fn_read(ctx->arg_read, in);
 		/* generic read failure! */
-		if (rv == -1)
-			goto error_read;
+		if (rv < 0) {
+			pthread_mutex_unlock(&ctx->read_mutex);
+			return mt_error(rv);
+		}
 		/* needed more bytes! */
 		if (in->size != toRead)
 			goto error_data;
@@ -293,15 +317,14 @@ static void *pt_decompress(void *arg)
 
 		/* zero should not happen here! */
 		result = pt_read(ctx, in, &wl->frame);
-		if (in->size == 0)
-			break;
-
 		if (LZ5MT_isError(result)) {
 			list_move(&wl->node, &ctx->writelist_free);
-
 			goto error_lock;
 		}
 
+		if (in->size == 0)
+			break;
+
 		{
 			/* get frame size for output buffer */
 			unsigned char *src = (unsigned char *)in->buf + 6;
@@ -407,10 +430,10 @@ static size_t st_decompress(void *arg)
 		/* read new input */
 		in->size = nextToLoad;
 		rv = ctx->fn_read(ctx->arg_read, in);
-		if (rv == -1) {
+		if (rv < 0) {
 			free(in->buf);
 			free(out->buf);
-			return ERROR(read_fail);
+			return mt_error(rv);
 		}
 
 		/* done, eof reached */
@@ -436,10 +459,10 @@ static size_t st_decompress(void *arg)
 			/* have some output */
 			if (out->size) {
 				rv = ctx->fn_write(ctx->arg_write, out);
-				if (rv == -1) {
+				if (rv < 0) {
 					free(in->buf);
 					free(out->buf);
-					return ERROR(write_fail);
+					return mt_error(rv);
 				}
 			}
 
@@ -456,7 +479,7 @@ static size_t st_decompress(void *arg)
 	return 0;
 }
 
-size_t LZ5MT_DecompressDCtx(LZ5MT_DCtx * ctx, LZ5MT_RdWr_t * rdwr)
+size_t LZ5MT_decompressDCtx(LZ5MT_DCtx * ctx, LZ5MT_RdWr_t * rdwr)
 {
 	unsigned char buf[4];
 	int t, rv;
@@ -476,8 +499,8 @@ size_t LZ5MT_DecompressDCtx(LZ5MT_DCtx * ctx, LZ5MT_RdWr_t * rdwr)
 	in->buf = buf;
 	in->size = 4;
 	rv = ctx->fn_read(ctx->arg_read, in);
-	if (rv == -1)
-		return ERROR(read_fail);
+	if (rv < 0)
+		return mt_error(rv);
 	if (in->size != 4)
 		return ERROR(data_error);
 

C/zstdmt/threading.c

@@ -52,12 +52,10 @@ pthread_create(pthread_t * thread, const void *unused,
 
 int _pthread_join(pthread_t * thread, void **value_ptr)
 {
-	DWORD result;
-
 	if (!thread->handle)
 		return 0;
 
-	result = WaitForSingleObject(thread->handle, INFINITE);
+	DWORD result = WaitForSingleObject(thread->handle, INFINITE);
 	switch (result) {
 	case WAIT_OBJECT_0:
 		if (value_ptr)

C/zstdmt/threading.h

@@ -31,7 +31,7 @@ extern "C" {
 #include <windows.h>
 
 /* mutex */
-#define pthread_mutex_t           CRITICAL_SECTION
+#define pthread_mutex_t CRITICAL_SECTION
 #define pthread_mutex_init(a,b)   InitializeCriticalSection((a))
 #define pthread_mutex_destroy(a)  DeleteCriticalSection((a))
 #define pthread_mutex_lock        EnterCriticalSection

C/zstdmt/zstdmt_compress.c

@@ -154,11 +154,29 @@ ZSTDMT_CCtx *ZSTDMT_createCCtx(int threads, int level, int inputsize)
 
 	return ctx;
 
-err_ctx:
+ err_ctx:
 	free(ctx);
 	return 0;
 }
 
+/**
+ * mt_error - return mt lib specific error code
+ */
+static size_t mt_error(int rv)
+{
+	switch (rv) {
+	case -1:
+		return ZSTDMT_ERROR(read_fail);
+	case -2:
+		return ZSTDMT_ERROR(canceled);
+	case -3:
+		return ZSTDMT_ERROR(memory_allocation);
+	}
+
+	/* XXX, some catch all other errors */
+	return ZSTDMT_ERROR(read_fail);
+}
+
 /**
  * pt_write - queue for compressed output
  */
@@ -178,10 +196,8 @@ static size_t pt_write(ZSTDMT_CCtx * ctx, struct writelist *wl)
 		b.buf = frame0;
 		b.size = 9;
 		rv = ctx->fn_write(ctx->arg_write, &b);
-		if (rv == -1)
-			return ZSTDMT_ERROR(write_fail);
-		if (rv == -2)
-			return ZSTDMT_ERROR(canceled);
+		if (rv < 0)
+			return mt_error(rv);
 		if (b.size != 9)
 			return ZSTDMT_ERROR(write_fail);
 		ctx->outsize += 9;
@@ -197,10 +213,8 @@ static size_t pt_write(ZSTDMT_CCtx * ctx, struct writelist *wl)
 		wl = list_entry(entry, struct writelist, node);
 		if (wl->frame == ctx->curframe) {
 			rv = ctx->fn_write(ctx->arg_write, &wl->out);
-			if (rv == -1)
-				return ZSTDMT_ERROR(write_fail);
-			if (rv == -2)
-				return ZSTDMT_ERROR(canceled);
+			if (rv < 0)
+				return mt_error(rv);
 			ctx->outsize += wl->out.size;
 			ctx->curframe++;
 			list_move(entry, &ctx->writelist_free);
@@ -264,13 +278,9 @@ static void *pt_compress(void *arg)
 		pthread_mutex_lock(&ctx->read_mutex);
 		in.size = ctx->inputsize;
 		rv = ctx->fn_read(ctx->arg_read, &in);
-		if (rv == -1) {
+		if (rv < 0) {
 			pthread_mutex_unlock(&ctx->read_mutex);
-			result = ZSTDMT_ERROR(read_fail);
-			goto error;
-		} else if (rv == -2) {
-			pthread_mutex_unlock(&ctx->read_mutex);
-			result = ZSTDMT_ERROR(canceled);
+			result = mt_error(rv);
 			goto error;
 		}
 
@@ -291,13 +301,15 @@ static void *pt_compress(void *arg)
 
 		/* compress whole frame */
 		{
-		unsigned char *outbuf = out->buf;
-		result = ZSTD_compress(outbuf + 12, out->size - 12, in.buf, in.size, ctx->level);
-		if (ZSTD_isError(result)) {
-			zstdmt_errcode = result;
-			result = ZSTDMT_ERROR(compression_library);
-			goto error;
-		}
+			unsigned char *outbuf = out->buf;
+			result =
+			    ZSTD_compress(outbuf + 12, out->size - 12, in.buf,
+					  in.size, ctx->level);
+			if (ZSTD_isError(result)) {
+				zstdmt_errcode = result;
+				result = ZSTDMT_ERROR(compression_library);
+				goto error;
+			}
 		}
 
 		/* write skippable frame */

C/zstdmt/zstdmt_decompress.c

@@ -158,6 +158,24 @@ static int IsZstd_Skippable(unsigned char *buf)
 	return (MEM_readLE32(buf) == ZSTDMT_MAGIC_SKIPPABLE);
 }
 
+/**
+ * mt_error - return mt lib specific error code
+ */
+static size_t mt_error(int rv)
+{
+	switch (rv) {
+	case -1:
+		return ZSTDMT_ERROR(read_fail);
+	case -2:
+		return ZSTDMT_ERROR(canceled);
+	case -3:
+		return ZSTDMT_ERROR(memory_allocation);
+	}
+
+	/* XXX, some catch all other errors */
+	return ZSTDMT_ERROR(read_fail);
+}
+
 /**
  * pt_write - queue for decompressed output
  */
@@ -173,10 +191,8 @@ static size_t pt_write(ZSTDMT_DCtx * ctx, struct writelist *wl)
 		wl = list_entry(entry, struct writelist, node);
 		if (wl->frame == ctx->curframe) {
 			int rv = ctx->fn_write(ctx->arg_write, &wl->out);
-			if (rv == -1)
-				return ZSTDMT_ERROR(write_fail);
-			if (rv == -2)
-				return ZSTDMT_ERROR(canceled);
+			if (rv < 0)
+				return mt_error(rv);
 			ctx->outsize += wl->out.size;
 			ctx->curframe++;
 			list_move(entry, &ctx->writelist_free);
@@ -217,10 +233,10 @@ static size_t pt_read(ZSTDMT_DCtx * ctx, ZSTDMT_Buffer * in, size_t * frame)
 			hdr.buf = hdrbuf + 7;
 			hdr.size = 5;
 			rv = ctx->fn_read(ctx->arg_read, &hdr);
-			if (rv == -1)
-				goto error_read;
-			if (rv == -2)
-				goto error_read;
+			if (rv < 0) {
+				pthread_mutex_unlock(&ctx->read_mutex);
+				return mt_error(rv);
+			}
 			if (hdr.size != 5)
 				goto error_data;
 			hdr.buf = hdrbuf;
@@ -234,8 +250,10 @@ static size_t pt_read(ZSTDMT_DCtx * ctx, ZSTDMT_Buffer * in, size_t * frame)
 				goto error_nomem;
 			in->allocated = in->size;
 			rv = ctx->fn_read(ctx->arg_read, in);
-			if (rv == -1)
-				goto error_read;
+			if (rv < 0) {
+				pthread_mutex_unlock(&ctx->read_mutex);
+				return mt_error(rv);
+			}
 			if (in->size != toRead)
 				goto error_data;
 			ctx->insize += in->size;
@@ -264,8 +282,10 @@ static size_t pt_read(ZSTDMT_DCtx * ctx, ZSTDMT_Buffer * in, size_t * frame)
 			in->buf = start + 4;
 			in->size = toRead - 4;
 			rv = ctx->fn_read(ctx->arg_read, in);
-			if (rv == -1)
-				goto error_read;
+			if (rv < 0) {
+				pthread_mutex_unlock(&ctx->read_mutex);
+				return mt_error(rv);
+			}
 			if (in->size != toRead - 4)
 				goto error_data;
 			ctx->insize += in->size;
@@ -286,10 +306,10 @@ static size_t pt_read(ZSTDMT_DCtx * ctx, ZSTDMT_Buffer * in, size_t * frame)
 	hdr.buf = hdrbuf;
 	hdr.size = 12;
 	rv = ctx->fn_read(ctx->arg_read, &hdr);
-	if (rv == -1)
-		goto error_read;
-	if (rv == -2)
-		goto error_canceled;
+	if (rv < 0) {
+		pthread_mutex_unlock(&ctx->read_mutex);
+		return mt_error(rv);
+	}
 
 	/* eof reached ? */
 	if (unlikely(hdr.size == 0)) {
@@ -325,7 +345,7 @@ static size_t pt_read(ZSTDMT_DCtx * ctx, ZSTDMT_Buffer * in, size_t * frame)
 		if (rv == -1)
 			goto error_read;
 		if (rv == -2)
- 			goto error_canceled;
+			goto error_canceled;
 		/* needed more bytes! */
 		if (in->size != toRead)
 			goto error_data;
@@ -361,12 +381,12 @@ static void *pt_decompress(void *arg)
 	size_t result = 0;
 	ZSTDMT_Buffer collect;
 
-		/* init dstream stream */
-		result = ZSTD_initDStream(w->dctx);
-		if (ZSTD_isError(result)) {
-			zstdmt_errcode = result;
-			return (void *)ZSTDMT_ERROR(compression_library);
-		}
+	/* init dstream stream */
+	result = ZSTD_initDStream(w->dctx);
+	if (ZSTD_isError(result)) {
+		zstdmt_errcode = result;
+		return (void *)ZSTDMT_ERROR(compression_library);
+	}
 
 	collect.buf = 0;
 	collect.size = 0;
@@ -452,7 +472,8 @@ static void *pt_decompress(void *arg)
 					bnew = malloc(collect.size + zOut.pos);
 					if (!bnew) {
 						result =
-						    ZSTDMT_ERROR(memory_allocation);
+						    ZSTDMT_ERROR
+						    (memory_allocation);
 						goto error_lock;
 					}
 					memcpy((char *)bnew, collect.buf,
@@ -496,7 +517,8 @@ static void *pt_decompress(void *arg)
 				pthread_mutex_unlock(&ctx->write_mutex);
 				out->buf = realloc(out->buf, out->size);
 				if (!out->buf) {
-					result = ZSTDMT_ERROR(memory_allocation);
+					result =
+					    ZSTDMT_ERROR(memory_allocation);
 					goto error_lock;
 				}
 				out->allocated = out->size;
@@ -581,8 +603,8 @@ static size_t st_decompress(void *arg)
 
 		/* read more bytes, to fill buffer */
 		rv = ctx->fn_read(ctx->arg_read, in);
-		if (rv == -1) {
-			result = ZSTDMT_ERROR(read_fail);
+		if (rv < 0) {
+			result = mt_error(rv);
 			goto error;
 		}
 
@@ -621,6 +643,10 @@ static size_t st_decompress(void *arg)
 					return ZSTDMT_ERROR(canceled);
 					goto error_clib;
 				}
+				if (rv == -3) {
+					return ZSTDMT_ERROR(memory_allocation);
+					goto error_clib;
+				}
 				ctx->outsize += zOut.pos;
 			}
 
@@ -643,8 +669,8 @@ static size_t st_decompress(void *arg)
 		/* read next input */
 		in->size = in->allocated;
 		rv = ctx->fn_read(ctx->arg_read, in);
-		if (rv == -1) {
-			result = ZSTDMT_ERROR(read_fail);
+		if (rv < 0) {
+			result = mt_error(rv);
 			goto error;
 		}
 
@@ -706,8 +732,8 @@ size_t ZSTDMT_decompressDCtx(ZSTDMT_DCtx * ctx, ZSTDMT_RdWr_t * rdwr)
 	in->buf = buf;
 	in->size = 16;
 	rv = ctx->fn_read(ctx->arg_read, in);
-	if (rv == -1)
-		return ZSTDMT_ERROR(read_fail);
+	if (rv < 0)
+		return mt_error(rv);
 
 	/* must be single threaded standard zstd, when smaller 16 bytes */
 	if (in->size < 16) {