Fix errors de/compressing chunks

src/archiver.c

@@ -34,6 +34,7 @@
 #include <spawn.h>
 #include <sys/stat.h>
 #include <sys/wait.h>
+#include <time.h>
 #include <unistd.h>
 #endif
 
@@ -955,10 +956,104 @@ int read_fd_to_out_fd(FILE *in_fd, FILE *out_fd, char *read_buf,
   return SDAS_SUCCESS;
 }
 
+int try_write_to_decomp(int *to_dec_pipe, uint64_t *chunk_remaining, FILE *in_f,
+                        char *buf, const size_t buf_size, char *hold_buf,
+                        int *has_hold) {
+  if (*to_dec_pipe >= 0) {
+    if (*chunk_remaining > 0) {
+      if (*chunk_remaining > buf_size) {
+        if (*has_hold < 0) {
+          size_t fread_ret = fread(buf, 1, 1024, in_f);
+          if (fread_ret == 0) {
+            goto TRY_WRITE_TO_DECOMP_END;
+          } else {
+            ssize_t write_ret = write(*to_dec_pipe, buf, fread_ret);
+            if (write_ret < 0) {
+              if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                *has_hold = (int)fread_ret;
+                memcpy(hold_buf, buf, fread_ret);
+                return SDAS_SUCCESS;
+              } else {
+                return SDAS_INTERNAL_ERROR;
+              }
+            } else if (write_ret == 0) {
+              return SDAS_INTERNAL_ERROR;
+            } else {
+              *chunk_remaining -= (size_t)write_ret;
+            }
+          }
+        } else {
+          ssize_t write_ret = write(*to_dec_pipe, hold_buf, (size_t)*has_hold);
+          if (write_ret < 0) {
+            if (errno == EAGAIN || errno == EWOULDBLOCK) {
+              return SDAS_SUCCESS;
+            } else {
+              return SDAS_INTERNAL_ERROR;
+            }
+          } else if (write_ret == 0) {
+            return SDAS_INTERNAL_ERROR;
+          } else {
+            *chunk_remaining -= (size_t)*has_hold;
+            *has_hold = -1;
+          }
+        }
+      } else {
+        if (*has_hold < 0) {
+          size_t fread_ret = fread(buf, 1, *chunk_remaining, in_f);
+          if (fread_ret == 0) {
+            goto TRY_WRITE_TO_DECOMP_END;
+          } else {
+            ssize_t write_ret = write(*to_dec_pipe, buf, fread_ret);
+            if (write_ret < 0) {
+              if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                *has_hold = (int)fread_ret;
+                memcpy(hold_buf, buf, fread_ret);
+                return SDAS_SUCCESS;
+              } else {
+                return SDAS_INTERNAL_ERROR;
+              }
+            } else if (write_ret == 0) {
+              return SDAS_INTERNAL_ERROR;
+            } else if ((size_t)write_ret <= *chunk_remaining) {
+              *chunk_remaining -= (size_t)write_ret;
+            } else {
+              return SDAS_INTERNAL_ERROR;
+            }
+          }
+        } else {
+          ssize_t write_ret = write(*to_dec_pipe, hold_buf, (size_t)*has_hold);
+          if (write_ret < 0) {
+            if (errno == EAGAIN || errno == EWOULDBLOCK) {
+              return SDAS_SUCCESS;
+            } else {
+              return SDAS_INTERNAL_ERROR;
+            }
+          } else if (write_ret == 0) {
+            return SDAS_INTERNAL_ERROR;
+          } else {
+            *chunk_remaining -= (size_t)*has_hold;
+            *has_hold = -1;
+          }
+        }
+      }
+    }
+  }
+
+TRY_WRITE_TO_DECOMP_END:
+  if (*to_dec_pipe >= 0 && *chunk_remaining == 0) {
+    close(*to_dec_pipe);
+    *to_dec_pipe = -1;
+  }
+
+  return SDAS_SUCCESS;
+}
+
 /// Returns SDAS_SUCCESS on success.
 int read_decomp_to_out_file(const char *out_filename, int in_pipe,
                             char *read_buf, const size_t read_buf_size,
-                            const uint64_t file_size) {
+                            const uint64_t file_size, int *to_dec_pipe,
+                            uint64_t *chunk_remaining, FILE *in_f,
+                            char *hold_buf, int *has_hold) {
   __attribute__((cleanup(simple_archiver_helper_cleanup_FILE))) FILE *out_fd =
       NULL;
   if (out_filename) {
@@ -974,6 +1069,11 @@ int read_decomp_to_out_file(const char *out_filename, int in_pipe,
   ssize_t read_ret;
   size_t fwrite_ret;
   while (written_amt < file_size) {
+    int ret = try_write_to_decomp(to_dec_pipe, chunk_remaining, in_f, read_buf,
+                                  read_buf_size, hold_buf, has_hold);
+    if (ret != SDAS_SUCCESS) {
+      return ret;
+    }
     if (file_size - written_amt >= read_buf_size) {
       read_ret = read(in_pipe, read_buf, read_buf_size);
       if (read_ret > 0) {
@@ -1002,12 +1102,17 @@ int read_decomp_to_out_file(const char *out_filename, int in_pipe,
         } else {
           break;
         }
+      } else {
+        if (errno == EAGAIN || errno == EWOULDBLOCK) {
+          // Non-blocking read from pipe.
+          continue;
         } else {
           // Error.
           fprintf(stderr, "ERROR Failed to read from decompressor! (%lu)\n",
                   read_ret);
           return SDAS_INTERNAL_ERROR;
         }
+      }
     } else {
       read_ret = read(in_pipe, read_buf, file_size - written_amt);
       if (read_ret > 0) {
@@ -1036,6 +1141,10 @@ int read_decomp_to_out_file(const char *out_filename, int in_pipe,
         } else {
           break;
         }
+      } else {
+        if (errno == EAGAIN || errno == EWOULDBLOCK) {
+          // Non-blocking read from pipe.
+          continue;
         } else {
           // Error.
           fprintf(stderr, "ERROR Failed to read from decompressor! (%d)\n",
@@ -1044,6 +1153,7 @@ int read_decomp_to_out_file(const char *out_filename, int in_pipe,
         }
       }
     }
+  }
 
   return written_amt == file_size ? SDAS_SUCCESS : SDAS_INTERNAL_ERROR;
 }
@@ -1291,7 +1401,7 @@ void simple_archiver_internal_cleanup_int_fd(int *fd) {
 #if SIMPLE_ARCHIVER_PLATFORM == SIMPLE_ARCHIVER_PLATFORM_COSMOPOLITAN || \
     SIMPLE_ARCHIVER_PLATFORM == SIMPLE_ARCHIVER_PLATFORM_MAC ||          \
     SIMPLE_ARCHIVER_PLATFORM == SIMPLE_ARCHIVER_PLATFORM_LINUX
-void simple_archiver_internal_cleanup_decomp(pid_t *decomp_pid) {
+void simple_archiver_internal_cleanup_decomp_pid(pid_t *decomp_pid) {
   if (decomp_pid && *decomp_pid >= 0) {
     int decompressor_status;
     int decompressor_return_val;
@@ -1323,6 +1433,7 @@ void simple_archiver_internal_cleanup_decomp(pid_t *decomp_pid) {
               "WARNING: Exec failed (exec exit code unknown)! Invalid "
               "decompressor cmd?\n");
     }
+    *decomp_pid = -1;
   }
 }
 #endif
@@ -2001,12 +2112,15 @@ int simple_archiver_write_v1(FILE *out_f, SDArchiverState *state,
   uint64_t *non_c_chunk_size = non_compressing_chunk_size;
 
   SDArchiverLLNode *file_node = files_list->head;
+  uint64_t chunk_count = 0;
   for (SDArchiverLLNode *chunk_c_node = chunk_counts->head->next;
        chunk_c_node != chunk_counts->tail; chunk_c_node = chunk_c_node->next) {
+    fprintf(stderr, "CHUNK %3lu of %3lu\n", ++chunk_count, chunk_counts->count);
     // Write file count before iterating through files.
     if (non_c_chunk_size) {
       *non_c_chunk_size = 0;
     }
+
     u32 = (uint32_t)(*((uint64_t *)chunk_c_node->data));
     simple_archiver_helper_32_bit_be(&u32);
     if (fwrite(&u32, 4, 1, out_f) != 1) {
@@ -2116,7 +2230,9 @@ int simple_archiver_write_v1(FILE *out_f, SDArchiverState *state,
 
       int pipe_into_cmd[2];
       int pipe_outof_cmd[2];
-      pid_t compressor_pid;
+      __attribute__((cleanup(
+          simple_archiver_internal_cleanup_decomp_pid))) pid_t compressor_pid =
+          -1;
 
       if (pipe(pipe_into_cmd) != 0) {
         // Unable to create pipes.
@@ -2126,6 +2242,20 @@ int simple_archiver_write_v1(FILE *out_f, SDArchiverState *state,
         close(pipe_into_cmd[0]);
         close(pipe_into_cmd[1]);
         return SDAS_INTERNAL_ERROR;
+      } else if (fcntl(pipe_into_cmd[1], F_SETFL, O_NONBLOCK) != 0) {
+        // Unable to set non-blocking on into-write-pipe.
+        close(pipe_into_cmd[0]);
+        close(pipe_into_cmd[1]);
+        close(pipe_outof_cmd[0]);
+        close(pipe_outof_cmd[1]);
+        return SDAS_INTERNAL_ERROR;
+      } else if (fcntl(pipe_outof_cmd[0], F_SETFL, O_NONBLOCK) != 0) {
+        // Unable to set non-blocking on outof-read-pipe.
+        close(pipe_into_cmd[0]);
+        close(pipe_into_cmd[1]);
+        close(pipe_outof_cmd[0]);
+        close(pipe_outof_cmd[1]);
+        return SDAS_INTERNAL_ERROR;
       } else if (simple_archiver_de_compress(pipe_into_cmd, pipe_outof_cmd,
                                              state->parsed->compressor,
                                              &compressor_pid) != 0) {
@@ -2142,15 +2272,18 @@ int simple_archiver_write_v1(FILE *out_f, SDArchiverState *state,
       close(pipe_outof_cmd[1]);
 
       // Set up cleanup so that remaining open pipes in this side is cleaned up.
-      __attribute__((cleanup(
-          simple_archiver_internal_cleanup_int_fd))) int pipe_into_write =
-          pipe_into_cmd[1];
       __attribute__((cleanup(
           simple_archiver_internal_cleanup_int_fd))) int pipe_outof_read =
           pipe_outof_cmd[0];
+      __attribute__((cleanup(
+          simple_archiver_internal_cleanup_int_fd))) int pipe_into_write =
+          pipe_into_cmd[1];
 
+      int_fast8_t to_temp_finished = 0;
       for (uint64_t file_idx = 0; file_idx < *((uint64_t *)chunk_c_node->data);
            ++file_idx) {
+        fprintf(stderr, "  FILE %3lu of %3lu\n", file_idx + 1,
+                *(uint64_t *)chunk_c_node->data);
         file_node = file_node->next;
         if (file_node == files_list->tail) {
           return SDAS_INTERNAL_ERROR;
@@ -2158,36 +2291,99 @@ int simple_archiver_write_v1(FILE *out_f, SDArchiverState *state,
         const SDArchiverInternalFileInfo *file_info_struct = file_node->data;
         __attribute__((cleanup(simple_archiver_helper_cleanup_FILE))) FILE *fd =
             fopen(file_info_struct->filename, "rb");
-        while (!feof(fd)) {
+
+        int_fast8_t to_comp_finished = 0;
+        char hold_buf[1024];
+        int has_hold = -1;
+        while (!to_comp_finished) {
+          if (!to_comp_finished) {
+            // Write to compressor.
             if (ferror(fd)) {
               fprintf(stderr, "ERROR: Writing to chunk, file read error!\n");
               return SDAS_INTERNAL_ERROR;
             }
+            if (has_hold < 0) {
               size_t fread_ret = fread(buf, 1, 1024, fd);
               if (fread_ret > 0) {
                 ssize_t write_ret = write(pipe_into_write, buf, fread_ret);
                 if (write_ret < 0) {
-              fprintf(stderr,
+                  if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                    // Non-blocking write.
+                    has_hold = (int)fread_ret;
+                    memcpy(hold_buf, buf, fread_ret);
+                  } else {
+                    fprintf(
+                        stderr,
                         "ERROR: Writing to compressor, pipe write error!\n");
                     return SDAS_FAILED_TO_WRITE;
+                  }
                 } else if ((size_t)write_ret != fread_ret) {
-              fprintf(stderr,
+                  fprintf(
+                      stderr,
                       "ERROR: Writing to compressor, unable to write bytes!\n");
                   return SDAS_FAILED_TO_WRITE;
                 }
               }
+
+              if (feof(fd)) {
+                to_comp_finished = 1;
+              }
+            } else {
+              ssize_t write_ret =
+                  write(pipe_into_write, hold_buf, (size_t)has_hold);
+              if (write_ret < 0) {
+                if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                  // Non-blocking write.
+                } else {
+                  return SDAS_INTERNAL_ERROR;
+                }
+              } else if (write_ret != has_hold) {
+                return SDAS_INTERNAL_ERROR;
+              } else {
+                has_hold = -1;
+              }
             }
           }
 
-      // Close write to pipe to compressor as the chunk is written.
+          // Write compressed data to temp file.
-      simple_archiver_internal_cleanup_int_fd(&pipe_into_write);
-
-      // Read compressed data into temporary file.
-      do {
           ssize_t read_ret = read(pipe_outof_read, buf, 1024);
           if (read_ret < 0) {
-          fprintf(stderr, "ERROR: Reading from compressor, pipe read error!\n");
+            if (errno == EAGAIN || errno == EWOULDBLOCK) {
+              // Non-blocking read.
+            } else {
+              fprintf(stderr,
+                      "ERROR: Reading from compressor, pipe read error!\n");
               return SDAS_INTERNAL_ERROR;
+            }
+          } else if (read_ret == 0) {
+            // EOF.
+            to_temp_finished = 1;
+          } else {
+            size_t fwrite_ret = fwrite(buf, 1, (size_t)read_ret, temp_fd);
+            if (fwrite_ret != (size_t)read_ret) {
+              fprintf(stderr,
+                      "ERROR: Reading from compressor, failed to write to "
+                      "temporary file!\n");
+              return SDAS_INTERNAL_ERROR;
+            }
+          }
+        }
+      }
+
+      simple_archiver_internal_cleanup_int_fd(&pipe_into_write);
+
+      // Finish writing.
+      if (!to_temp_finished) {
+        while (1) {
+          ssize_t read_ret = read(pipe_outof_read, buf, 1024);
+          if (read_ret < 0) {
+            if (errno == EAGAIN || errno == EWOULDBLOCK) {
+              // Non-blocking read.
+            } else {
+              fprintf(stderr,
+                      "ERROR: Reading from compressor, pipe read error!\n");
+              return SDAS_INTERNAL_ERROR;
+            }
           } else if (read_ret == 0) {
             // EOF.
             break;
@@ -2200,13 +2396,8 @@ int simple_archiver_write_v1(FILE *out_f, SDArchiverState *state,
               return SDAS_INTERNAL_ERROR;
             }
           }
-      } while (1);
+        }
-
+      }
-      // Close read from pipe from compressor as chunk is fully compressed.
-      simple_archiver_internal_cleanup_int_fd(&pipe_outof_read);
-
-      // Wait on compressor to stop.
-      waitpid(compressor_pid, NULL, 0);
 
       long comp_chunk_size = ftell(temp_fd);
       if (comp_chunk_size < 0) {
@@ -2226,6 +2417,8 @@ int simple_archiver_write_v1(FILE *out_f, SDArchiverState *state,
         return SDAS_INTERNAL_ERROR;
       }
 
+      size_t written_size = 0;
+
       // Write compressed chunk.
       while (!feof(temp_fd)) {
         if (ferror(temp_fd)) {
@@ -2234,6 +2427,7 @@ int simple_archiver_write_v1(FILE *out_f, SDArchiverState *state,
         size_t fread_ret = fread(buf, 1, 1024, temp_fd);
         if (fread_ret > 0) {
           size_t fwrite_ret = fwrite(buf, 1, fread_ret, out_f);
+          written_size += fread_ret;
           if (fwrite_ret != fread_ret) {
             fprintf(stderr,
                     "ERROR: Partial write of read bytes from temp file to "
@@ -2243,6 +2437,12 @@ int simple_archiver_write_v1(FILE *out_f, SDArchiverState *state,
         }
       }
 
+      if (written_size != (size_t)comp_chunk_size) {
+        fprintf(stderr,
+                "ERROR: Written chunk size is not actual chunk size!\n");
+        return SDAS_FAILED_TO_WRITE;
+      }
+
       // Cleanup and remove temp_fd.
       simple_archiver_helper_cleanup_FILE(&temp_fd);
     } else {
@@ -2254,6 +2454,8 @@ int simple_archiver_write_v1(FILE *out_f, SDArchiverState *state,
       fwrite(non_c_chunk_size, 8, 1, out_f);
       for (uint64_t file_idx = 0; file_idx < *((uint64_t *)chunk_c_node->data);
            ++file_idx) {
+        fprintf(stderr, "  FILE %3lu of %3lu\n", file_idx + 1,
+                *(uint64_t *)chunk_c_node->data);
         file_node = file_node->next;
         if (file_node == files_list->tail) {
           return SDAS_INTERNAL_ERROR;
@@ -3543,6 +3745,7 @@ int simple_archiver_parse_archive_version_1(FILE *in_f, int_fast8_t do_extract,
     simple_archiver_helper_64_bit_be(&u64);
 
     const uint64_t chunk_size = u64;
+    uint64_t chunk_remaining = chunk_size;
     uint64_t chunk_idx = 0;
 
     SDArchiverLLNode *node = file_info_list->head;
@@ -3555,12 +3758,13 @@ int simple_archiver_parse_archive_version_1(FILE *in_f, int_fast8_t do_extract,
       // Start the decompressing process and read into files.
 
       // Handle SIGPIPE.
+      is_sig_pipe_occurred = 0;
       signal(SIGPIPE, handle_sig_pipe);
 
       int pipe_into_cmd[2];
       int pipe_outof_cmd[2];
       __attribute__((cleanup(
-          simple_archiver_internal_cleanup_decomp))) pid_t decompressor_pid;
+          simple_archiver_internal_cleanup_decomp_pid))) pid_t decompressor_pid;
       if (pipe(pipe_into_cmd) != 0) {
         // Unable to create pipes.
         break;
@@ -3569,6 +3773,20 @@ int simple_archiver_parse_archive_version_1(FILE *in_f, int_fast8_t do_extract,
         close(pipe_into_cmd[0]);
         close(pipe_into_cmd[1]);
         return SDAS_INTERNAL_ERROR;
+      } else if (fcntl(pipe_into_cmd[1], F_SETFL, O_NONBLOCK) != 0) {
+        // Unable to set non-blocking on into-write-pipe.
+        close(pipe_into_cmd[0]);
+        close(pipe_into_cmd[1]);
+        close(pipe_outof_cmd[0]);
+        close(pipe_outof_cmd[1]);
+        return SDAS_INTERNAL_ERROR;
+      } else if (fcntl(pipe_outof_cmd[0], F_SETFL, O_NONBLOCK) != 0) {
+        // Unable to set non-blocking on outof-read-pipe.
+        close(pipe_into_cmd[0]);
+        close(pipe_into_cmd[1]);
+        close(pipe_outof_cmd[0]);
+        close(pipe_outof_cmd[1]);
+        return SDAS_INTERNAL_ERROR;
       }
 
       if (state && state->parsed && state->parsed->decompressor) {
@@ -3599,12 +3817,12 @@ int simple_archiver_parse_archive_version_1(FILE *in_f, int_fast8_t do_extract,
       close(pipe_into_cmd[0]);
       close(pipe_outof_cmd[1]);
 
-      __attribute__((cleanup(
-          simple_archiver_internal_cleanup_int_fd))) int pipe_into_write =
-          pipe_into_cmd[1];
       __attribute__((cleanup(
           simple_archiver_internal_cleanup_int_fd))) int pipe_outof_read =
           pipe_outof_cmd[0];
+      __attribute__((cleanup(
+          simple_archiver_internal_cleanup_int_fd))) int pipe_into_write =
+          pipe_into_cmd[1];
 
       int decompressor_status;
       int decompressor_return_val;
@@ -3630,62 +3848,8 @@ int simple_archiver_parse_archive_version_1(FILE *in_f, int_fast8_t do_extract,
         return SDAS_INTERNAL_ERROR;
       }
 
-      // Write all of chunk into decompressor.
+      char hold_buf[1024];
-      uint64_t chunk_written = 0;
+      int has_hold = -1;
-      while (chunk_written < chunk_size) {
-        if (is_sig_pipe_occurred) {
-          fprintf(stderr,
-                  "WARNING: Failed to write to decompressor (SIGPIPE)! Invalid "
-                  "decompressor cmd?\n");
-          return SDAS_INTERNAL_ERROR;
-        } else if (chunk_size - chunk_written >= 1024) {
-          if (fread(buf, 1, 1024, in_f) != 1024) {
-            fprintf(stderr, "ERROR Failed to read chunk for decompressing!\n");
-            return SDAS_INTERNAL_ERROR;
-          }
-          ssize_t write_ret = write(pipe_into_cmd[1], buf, 1024);
-          if (write_ret > 0 && (size_t)write_ret == 1024) {
-            // Successful write.
-          } else if (write_ret == -1) {
-            fprintf(stderr,
-                    "WARNING: Failed to write chunk data into decompressor! "
-                    "Invalid decompressor cmd? (errno %d)\n",
-                    errno);
-            return SDAS_INTERNAL_ERROR;
-          } else {
-            fprintf(stderr,
-                    "WARNING: Failed to write chunk data into decompressor! "
-                    "Invalid decompressor cmd?\n");
-            return SDAS_INTERNAL_ERROR;
-          }
-          chunk_written += 1024;
-        } else {
-          if (fread(buf, 1, chunk_size - chunk_written, in_f) !=
-              chunk_size - chunk_written) {
-            fprintf(stderr, "ERROR Failed to read chunk for decompressing!\n");
-            return SDAS_INTERNAL_ERROR;
-          }
-          ssize_t write_ret =
-              write(pipe_into_cmd[1], buf, chunk_size - chunk_written);
-          if (write_ret > 0 &&
-              (size_t)write_ret == chunk_size - chunk_written) {
-            // Successful write.
-          } else if (write_ret == -1) {
-            fprintf(stderr,
-                    "WARNING: Failed to write chunk data into decompressor! "
-                    "Invalid decompressor cmd?\n");
-            return SDAS_INTERNAL_ERROR;
-          } else {
-            fprintf(stderr,
-                    "WARNING: Failed to write chunk data into decompressor! "
-                    "Invalid decompressor cmd?\n");
-            return SDAS_INTERNAL_ERROR;
-          }
-          chunk_written = chunk_size;
-        }
-      }
-
-      simple_archiver_internal_cleanup_int_fd(&pipe_into_write);
 
       while (node->next != file_info_list->tail) {
         node = node->next;
@@ -3719,16 +3883,19 @@ int simple_archiver_parse_archive_version_1(FILE *in_f, int_fast8_t do_extract,
               fprintf(stderr,
                       "  WARNING: File already exists and "
                       "\"--overwrite-extract\" is not specified, skipping!\n");
-              read_decomp_to_out_file(NULL, pipe_outof_cmd[0], (char *)buf,
+              read_decomp_to_out_file(NULL, pipe_outof_read, (char *)buf, 1024,
-                                      1024, file_info->file_size);
+                                      file_info->file_size, &pipe_into_write,
+                                      &chunk_remaining, in_f, hold_buf,
+                                      &has_hold);
               continue;
             }
           }
 
           simple_archiver_helper_make_dirs(file_info->filename);
-          int ret =
+          int ret = read_decomp_to_out_file(
-              read_decomp_to_out_file(file_info->filename, pipe_outof_cmd[0],
+              file_info->filename, pipe_outof_read, (char *)buf, 1024,
-                                      (char *)buf, 1024, file_info->file_size);
+              file_info->file_size, &pipe_into_write, &chunk_remaining, in_f,
+              hold_buf, &has_hold);
           if (ret != SDAS_SUCCESS) {
             return ret;
           }
@@ -3758,13 +3925,15 @@ int simple_archiver_parse_archive_version_1(FILE *in_f, int_fast8_t do_extract,
             fprintf(stderr, "    File size: %lu\n", file_info->file_size);
           }
           int ret = read_decomp_to_out_file(
-              NULL, pipe_outof_cmd[0], (char *)buf, 1024, file_info->file_size);
+              NULL, pipe_outof_read, (char *)buf, 1024, file_info->file_size,
+              &pipe_into_write, &chunk_remaining, in_f, hold_buf, &has_hold);
           if (ret != SDAS_SUCCESS) {
             return ret;
           }
         } else {
           int ret = read_decomp_to_out_file(
-              NULL, pipe_outof_cmd[0], (char *)buf, 1024, file_info->file_size);
+              NULL, pipe_outof_read, (char *)buf, 1024, file_info->file_size,
+              &pipe_into_write, &chunk_remaining, in_f, hold_buf, &has_hold);
           if (ret != SDAS_SUCCESS) {
             return ret;
           }
@@ -3772,8 +3941,8 @@ int simple_archiver_parse_archive_version_1(FILE *in_f, int_fast8_t do_extract,
       }
 
       // Ensure EOF is left from pipe.
-      ssize_t read_ret = read(pipe_outof_cmd[0], buf, 1024);
+      ssize_t read_ret = read(pipe_outof_read, buf, 1024);
-      if (read_ret != 0) {
+      if (read_ret > 0) {
         fprintf(stderr, "WARNING decompressor didn't reach EOF!\n");
       }
     } else {