eitsupi commented on code in PR #975:
URL: https://github.com/apache/arrow-nanoarrow/pull/975#discussion_r4315916390
##########
r/src/array_stream.c:
##########
@@ -18,13 +18,209 @@
#define R_NO_REMAP
#include <R.h>
#include <Rinternals.h>
+#include <errno.h>
+#include <string.h>
#include "array.h"
#include "array_stream.h"
#include "nanoarrow.h"
#include "schema.h"
#include "util.h"
+struct FunctionArrayStreamData {
+ struct ArrowSchema schema;
+ SEXP fun;
+ SEXP schema_xptr;
+ int validate;
+ struct ArrowError error;
+ int error_code;
+ int finished;
+ int errored;
+ int evaluating;
+ int release_pending;
+};
+
+struct FunctionArrayStreamCall {
+ struct FunctionArrayStreamData* data;
+ struct ArrowArray* out;
+ int code;
+ int caught;
+};
+
+static void release_function_array_stream_data(struct FunctionArrayStreamData*
data) {
+ if (data->schema.release != NULL) {
+ data->schema.release(&data->schema);
+ }
+ nanoarrow_release_sexp(data->fun);
+ nanoarrow_release_sexp(data->schema_xptr);
+ ArrowFree(data);
+}
+
+static void function_array_stream_set_error(struct FunctionArrayStreamData*
data,
+ int code, const char* message) {
+ if (!data->errored) {
+ ArrowErrorSet(&data->error, "%s", message);
+ data->error_code = code;
+ data->errored = 1;
+ }
+}
+
+static void release_function_array_stream(struct ArrowArrayStream* stream) {
+ struct FunctionArrayStreamData* data =
+ (struct FunctionArrayStreamData*)stream->private_data;
+ stream->release = NULL;
+ stream->private_data = NULL;
+ if (data->evaluating) {
+ // The callback may release its own stream. Keep its state alive until it
returns.
+ data->release_pending = 1;
+ return;
+ }
+ release_function_array_stream_data(data);
+}
+
+static const char* function_array_stream_get_last_error(struct
ArrowArrayStream* stream) {
+ struct FunctionArrayStreamData* data =
+ (struct FunctionArrayStreamData*)stream->private_data;
+ return data == NULL ? "Array stream has been released"
+ : ArrowErrorMessage(&data->error);
+}
+
+static int function_array_stream_get_schema(struct ArrowArrayStream* stream,
+ struct ArrowSchema* out) {
+ struct FunctionArrayStreamData* data =
+ (struct FunctionArrayStreamData*)stream->private_data;
+ return ArrowSchemaDeepCopy(&data->schema, out);
+}
+
+static SEXP function_array_stream_condition_handler(SEXP condition, void*
opaque) {
+ struct FunctionArrayStreamCall* call = (struct
FunctionArrayStreamCall*)opaque;
+ call->caught = 1;
+ if (Rf_inherits(condition, "interrupt")) {
+ call->caught = 2;
+ }
+ SEXP message_call = PROTECT(Rf_lang2(Rf_install("conditionMessage"),
condition));
+ SEXP message = Rf_eval(message_call, R_BaseEnv);
+ UNPROTECT(1);
+ return message;
+}
+
+static SEXP function_array_stream_eval_call(void* opaque) {
+ return Rf_eval((SEXP)opaque, nanoarrow_ns_pkg);
+}
+
+static void function_array_stream_eval(void* opaque) {
+ struct FunctionArrayStreamCall* call = (struct
FunctionArrayStreamCall*)opaque;
+ struct FunctionArrayStreamData* data = call->data;
+ SEXP classes = PROTECT(Rf_allocVector(STRSXP, 2));
+ SET_STRING_ELT(classes, 0, Rf_mkChar("error"));
+ SET_STRING_ELT(classes, 1, Rf_mkChar("interrupt"));
+ SEXP validate = PROTECT(Rf_ScalarLogical(data->validate));
+ SEXP callback =
PROTECT(Rf_lang4(Rf_install("function_array_stream_get_next_wrapper"),
+ data->fun, data->schema_xptr, validate));
+ SEXP result =
+ PROTECT(R_tryCatch(&function_array_stream_eval_call, callback, classes,
+ &function_array_stream_condition_handler, call, NULL,
NULL));
+ if (data->errored) {
+ call->code = data->error_code;
+ } else if (call->caught) {
+ const char* message = Rf_translateCharUTF8(STRING_ELT(result, 0));
+ if (call->caught == 2) {
+ struct ArrowError error;
+ ArrowErrorSet(&error, "R callback interrupted: %s", message);
+ function_array_stream_set_error(data, EIO, ArrowErrorMessage(&error));
+ } else {
+ function_array_stream_set_error(data, EIO, message);
+ }
+ call->code = data->error_code;
+ } else if (result == R_NilValue) {
+ data->finished = 1;
+ call->code = NANOARROW_OK;
+ } else {
+ // Let the external pointer clean up a partial export if R jumps during
allocation.
+ SEXP array_xptr = PROTECT(nanoarrow_array_owning_xptr());
+ struct ArrowArray* array = nanoarrow_output_array_from_xptr(array_xptr);
+ array_export(result, array);
+ ArrowArrayMove(array, call->out);
+ UNPROTECT(1);
+ call->code = NANOARROW_OK;
+ }
+ UNPROTECT(4);
+}
+
+static int function_array_stream_get_next(struct ArrowArrayStream* stream,
+ struct ArrowArray* out) {
+ struct FunctionArrayStreamData* data =
+ (struct FunctionArrayStreamData*)stream->private_data;
+ out->release = NULL;
+ if (data == NULL || data->finished) {
+ return NANOARROW_OK;
+ }
+ if (data->errored) {
+ return data->error_code;
+ }
+ if (!nanoarrow_is_main_thread()) {
+ function_array_stream_set_error(data, EIO,
+ "Can't evaluate R callback on a non-R
thread");
+ return data->error_code;
+ }
+ if (data->evaluating) {
+ function_array_stream_set_error(data, EINVAL,
+ "Re-entrant call to function array
stream");
+ return data->error_code;
+ }
+
+ data->evaluating = 1;
+ struct FunctionArrayStreamCall call = {data, out, NANOARROW_OK, 0};
+ // R_tryCatch() does not catch restart jumps or failures in
conditionMessage().
+ // All R API calls must stay inside this outer boundary.
+ if (!R_ToplevelExec(&function_array_stream_eval, &call)) {
+ function_array_stream_set_error(
+ data, EIO, "R callback exited non-locally (e.g., via an abort
restart)");
+ call.code = data->error_code;
+ }
+ data->evaluating = 0;
+
+ if (data->release_pending) {
+ if (out->release != NULL) {
+ out->release(out);
+ }
+ out->release = NULL;
+ // The stream itself may already have been freed by a finalizer during
evaluation.
+ release_function_array_stream_data(data);
+ return EINVAL;
+ }
Review Comment:
I removed the deferred-release handling.
The docs now state that fun must not release the stream while it is being
evaluated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
