use R_ExecWithCleanup to clean up on errors

Author: ben-schwen

src/freadR.c

@@ -741,83 +741,104 @@ void progress(int p, int eta)
 }
 // # nocov end
 
-// Spill connection contents to a tempfile so R-level fread can treat it like a filename
-SEXP spillConnectionToFile(SEXP connection, SEXP tempfile_path, SEXP nrows_limit) {
-#if R_CONNECTIONS_VERSION != 1
-INTERNAL_STOP(_("spillConnectionToFile: unexpected R_CONNECTIONS_VERSION = %d", R_CONNECTIONS_VERSION)); // # nocov
-#else
-  if (!isString(tempfile_path) || LENGTH(tempfile_path) != 1) {
-    INTERNAL_STOP(_("spillConnectionToFile: tempfile_path must be a single string")); // # nocov
-  }
-
-  if (!isReal(nrows_limit) || LENGTH(nrows_limit) != 1) {
-    INTERNAL_STOP(_("spillConnectionToFile: nrows_limit must be a single numeric value")); // # nocov
+typedef struct {
+  Rconnection con;
+  const char *filepath;
+  size_t row_limit;
+  FILE *outfile;
+  char *buffer;
+} SpillState;
+
+static void spill_cleanup(void *data)
+{
+  SpillState *state = (SpillState *)data;
+  if (!state) return;
+  free(state->buffer); // free(NULL) is safe no-op
+  if (state->outfile) {
+    fclose(state->outfile);
   }
+}
 
-  Rconnection con = R_GetConnection(connection);
-  const char *filepath = CHAR(STRING_ELT(tempfile_path, 0));
-  const double nrows_max = REAL_RO(nrows_limit)[0];
-  const bool limit_rows = R_FINITE(nrows_max) && nrows_max >= 0.0;
-  size_t row_limit = 0;
-  if (limit_rows) {
-    row_limit = (size_t)nrows_max;
-    if (row_limit == 0) row_limit = 100;  // read at least 100 rows if nrows==0
-    row_limit++; // cater for potential header row
-  }
+static SEXP do_spill(void *data)
+{
+  SpillState *state = (SpillState *)data;
+  const size_t chunk_size = 256 * 1024; // TODO tune chunk size
 
-  FILE *outfile = fopen(filepath, "wb");
-  if (outfile == NULL) {
-    STOP(_("spillConnectionToFile: failed to open temp file '%s' for writing: %s"), filepath, strerror(errno)); // # nocov
+  state->outfile = fopen(state->filepath, "wb");
+  if (state->outfile == NULL) {
+    STOP(_("spillConnectionToFile: failed to open temp file '%s' for writing: %s"), state->filepath, strerror(errno)); // # nocov
   }
 
-  // Read and write in chunks // TODO tune chunk size
-  size_t chunk_size = 256 * 1024;
-  char *buffer = malloc(chunk_size);
-  if (!buffer) {
-    fclose(outfile);                                             // # nocov
+  state->buffer = malloc(chunk_size);
+  if (!state->buffer) {
     STOP(_("spillConnectionToFile: failed to allocate buffer")); // # nocov
   }
-
+  const bool limit_rows = R_FINITE(state->row_limit) && (state->row_limit > 0);
   size_t total_read = 0;
   size_t nrows_seen = 0;
 
   while (true) {
-    size_t nread = R_ReadConnection(con, buffer, chunk_size);
+    size_t nread = R_ReadConnection(state->con, state->buffer, chunk_size);
     if (nread == 0) {
       break; // EOF
     }
 
     size_t bytes_to_write = nread;
-    if (limit_rows && nrows_seen < row_limit) {
+    if (limit_rows && nrows_seen < state->row_limit) {
       for (size_t i = 0; i < nread; i++) {
-        if (buffer[i] == '\n') {
+        if (state->buffer[i] == '\n') {
           nrows_seen++;
-          if (nrows_seen >= row_limit) {
+          if (nrows_seen >= state->row_limit) {
             bytes_to_write = i + 1;
             break;
           }
         }
       }
     }
 
-    size_t nwritten = fwrite(buffer, 1, bytes_to_write, outfile);
+    size_t nwritten = fwrite(state->buffer, 1, bytes_to_write, state->outfile);
     if (nwritten != bytes_to_write) {
-      // # nocov start
-      free(buffer);
-      fclose(outfile);
-      STOP(_("spillConnectionToFile: write error %s (wrote %zu of %zu bytes)"), strerror(errno), nwritten, bytes_to_write);
-      // # nocov end
+      STOP(_("spillConnectionToFile: write error %s (wrote %zu of %zu bytes)"), strerror(errno), nwritten, bytes_to_write); // # nocov
     }
     total_read += bytes_to_write;
 
-    if (limit_rows && nrows_seen >= row_limit) {
+    if (limit_rows && nrows_seen >= state->row_limit) {
       break;
     }
   }
 
-  free(buffer);
-  fclose(outfile);
   return ScalarReal((double)total_read);
+}
+
+// Spill connection contents to a tempfile so R-level fread can treat it like a filename
+SEXP spillConnectionToFile(SEXP connection, SEXP tempfile_path, SEXP nrows_limit) {
+#if R_CONNECTIONS_VERSION != 1
+INTERNAL_STOP(_("spillConnectionToFile: unexpected R_CONNECTIONS_VERSION = %d", R_CONNECTIONS_VERSION)); // # nocov
+#else
+  if (!isString(tempfile_path) || LENGTH(tempfile_path) != 1) {
+    INTERNAL_STOP(_("spillConnectionToFile: tempfile_path must be a single string")); // # nocov
+  }
+
+  if (!isReal(nrows_limit) || LENGTH(nrows_limit) != 1) {
+    INTERNAL_STOP(_("spillConnectionToFile: nrows_limit must be a single numeric value")); // # nocov
+  }
+
+  SpillState state = {
+    .con = R_GetConnection(connection),
+    .filepath = CHAR(STRING_ELT(tempfile_path, 0)),
+    .row_limit = 0,
+    .outfile = NULL,
+    .buffer = NULL
+  };
+
+  const double nrows_max = REAL_RO(nrows_limit)[0];
+  if (R_FINITE(nrows_max) && nrows_max >= 0.0) {
+    state.row_limit = (size_t)nrows_max;
+    if (state.row_limit == 0) state.row_limit = 100;  // read at least 100 rows if nrows==0
+    state.row_limit++; // cater for potential header row
+  }
+
+  return R_ExecWithCleanup(do_spill, &state, spill_cleanup, &state);
 #endif // was R_CONNECTIONS_VERSION not != 1?
 }