aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/lib/transport
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/lib/transport')
-rw-r--r--src/core/lib/transport/connectivity_state.c48
-rw-r--r--src/core/lib/transport/connectivity_state.h5
-rw-r--r--src/core/lib/transport/transport.c19
-rw-r--r--src/core/lib/transport/transport.h5
4 files changed, 53 insertions, 24 deletions
diff --git a/src/core/lib/transport/connectivity_state.c b/src/core/lib/transport/connectivity_state.c
index e24ee638fd..3286af9fb2 100644
--- a/src/core/lib/transport/connectivity_state.c
+++ b/src/core/lib/transport/connectivity_state.c
@@ -61,35 +61,40 @@ void grpc_connectivity_state_init(grpc_connectivity_state_tracker *tracker,
grpc_connectivity_state init_state,
const char *name) {
tracker->current_state = init_state;
+ tracker->current_error = GRPC_ERROR_NONE;
tracker->watchers = NULL;
tracker->name = gpr_strdup(name);
}
void grpc_connectivity_state_destroy(grpc_exec_ctx *exec_ctx,
grpc_connectivity_state_tracker *tracker) {
- int success;
+ grpc_error *error;
grpc_connectivity_state_watcher *w;
while ((w = tracker->watchers)) {
tracker->watchers = w->next;
if (GRPC_CHANNEL_FATAL_FAILURE != *w->current) {
*w->current = GRPC_CHANNEL_FATAL_FAILURE;
- success = 1;
+ error = GRPC_ERROR_NONE;
} else {
- success = 0;
+ error = GRPC_ERROR_CREATE("Shutdown connectivity owner");
}
- grpc_exec_ctx_enqueue(exec_ctx, w->notify, success, NULL);
+ grpc_exec_ctx_push(exec_ctx, w->notify, error, NULL);
gpr_free(w);
}
+ GRPC_ERROR_UNREF(tracker->current_error);
gpr_free(tracker->name);
}
grpc_connectivity_state grpc_connectivity_state_check(
- grpc_connectivity_state_tracker *tracker) {
+ grpc_connectivity_state_tracker *tracker, grpc_error **error) {
if (grpc_connectivity_state_trace) {
gpr_log(GPR_DEBUG, "CONWATCH: %p %s: get %s", tracker, tracker->name,
grpc_connectivity_state_name(tracker->current_state));
}
+ if (error != NULL) {
+ *error = GRPC_ERROR_REF(tracker->current_error);
+ }
return tracker->current_state;
}
@@ -109,7 +114,7 @@ int grpc_connectivity_state_notify_on_state_change(
if (current == NULL) {
grpc_connectivity_state_watcher *w = tracker->watchers;
if (w != NULL && w->notify == notify) {
- grpc_exec_ctx_enqueue(exec_ctx, notify, false, NULL);
+ grpc_exec_ctx_push(exec_ctx, notify, GRPC_ERROR_CANCELLED, NULL);
tracker->watchers = w->next;
gpr_free(w);
return 0;
@@ -117,7 +122,7 @@ int grpc_connectivity_state_notify_on_state_change(
while (w != NULL) {
grpc_connectivity_state_watcher *rm_candidate = w->next;
if (rm_candidate != NULL && rm_candidate->notify == notify) {
- grpc_exec_ctx_enqueue(exec_ctx, notify, false, NULL);
+ grpc_exec_ctx_push(exec_ctx, notify, GRPC_ERROR_CANCELLED, NULL);
w->next = w->next->next;
gpr_free(rm_candidate);
return 0;
@@ -128,7 +133,8 @@ int grpc_connectivity_state_notify_on_state_change(
} else {
if (tracker->current_state != *current) {
*current = tracker->current_state;
- grpc_exec_ctx_enqueue(exec_ctx, notify, true, NULL);
+ grpc_exec_ctx_push(exec_ctx, notify,
+ GRPC_ERROR_REF(tracker->current_error), NULL);
} else {
grpc_connectivity_state_watcher *w = gpr_malloc(sizeof(*w));
w->current = current;
@@ -143,13 +149,28 @@ int grpc_connectivity_state_notify_on_state_change(
void grpc_connectivity_state_set(grpc_exec_ctx *exec_ctx,
grpc_connectivity_state_tracker *tracker,
grpc_connectivity_state state,
- const char *reason) {
+ grpc_error *error, const char *reason) {
grpc_connectivity_state_watcher *w;
if (grpc_connectivity_state_trace) {
- gpr_log(GPR_DEBUG, "SET: %p %s: %s --> %s [%s]", tracker, tracker->name,
- grpc_connectivity_state_name(tracker->current_state),
- grpc_connectivity_state_name(state), reason);
+ const char *error_string = grpc_error_string(error);
+ gpr_log(GPR_DEBUG, "SET: %p %s: %s --> %s [%s] error=%p %s", tracker,
+ tracker->name, grpc_connectivity_state_name(tracker->current_state),
+ grpc_connectivity_state_name(state), reason, error, error_string);
+ grpc_error_free_string(error_string);
+ }
+ switch (state) {
+ case GRPC_CHANNEL_CONNECTING:
+ case GRPC_CHANNEL_IDLE:
+ case GRPC_CHANNEL_READY:
+ GPR_ASSERT(error == GRPC_ERROR_NONE);
+ break;
+ case GRPC_CHANNEL_FATAL_FAILURE:
+ case GRPC_CHANNEL_TRANSIENT_FAILURE:
+ GPR_ASSERT(error != GRPC_ERROR_NONE);
+ break;
}
+ GRPC_ERROR_UNREF(tracker->current_error);
+ tracker->current_error = error;
if (tracker->current_state == state) {
return;
}
@@ -158,7 +179,8 @@ void grpc_connectivity_state_set(grpc_exec_ctx *exec_ctx,
while ((w = tracker->watchers) != NULL) {
*w->current = tracker->current_state;
tracker->watchers = w->next;
- grpc_exec_ctx_enqueue(exec_ctx, w->notify, true, NULL);
+ grpc_exec_ctx_push(exec_ctx, w->notify,
+ GRPC_ERROR_REF(tracker->current_error), NULL);
gpr_free(w);
}
}
diff --git a/src/core/lib/transport/connectivity_state.h b/src/core/lib/transport/connectivity_state.h
index 2eb7e09124..7a2fa52c10 100644
--- a/src/core/lib/transport/connectivity_state.h
+++ b/src/core/lib/transport/connectivity_state.h
@@ -49,6 +49,8 @@ typedef struct grpc_connectivity_state_watcher {
typedef struct {
/** current connectivity state */
grpc_connectivity_state current_state;
+ /** error associated with state */
+ grpc_error *current_error;
/** all our watchers */
grpc_connectivity_state_watcher *watchers;
/** a name to help debugging */
@@ -70,10 +72,11 @@ void grpc_connectivity_state_destroy(grpc_exec_ctx *exec_ctx,
void grpc_connectivity_state_set(grpc_exec_ctx *exec_ctx,
grpc_connectivity_state_tracker *tracker,
grpc_connectivity_state state,
+ grpc_error *associated_error,
const char *reason);
grpc_connectivity_state grpc_connectivity_state_check(
- grpc_connectivity_state_tracker *tracker);
+ grpc_connectivity_state_tracker *tracker, grpc_error **current_error);
/** Return 1 if the channel should start connecting, 0 otherwise.
If current==NULL cancel notify if it is already queued (success==0 in that
diff --git a/src/core/lib/transport/transport.c b/src/core/lib/transport/transport.c
index e6d524abe6..fdf0f4b2aa 100644
--- a/src/core/lib/transport/transport.c
+++ b/src/core/lib/transport/transport.c
@@ -60,7 +60,7 @@ void grpc_stream_unref(grpc_exec_ctx *exec_ctx,
grpc_stream_refcount *refcount) {
#endif
if (gpr_unref(&refcount->refs)) {
- grpc_exec_ctx_enqueue(exec_ctx, &refcount->destroy, true, NULL);
+ grpc_exec_ctx_push(exec_ctx, &refcount->destroy, GRPC_ERROR_NONE, NULL);
}
}
@@ -143,11 +143,14 @@ char *grpc_transport_get_peer(grpc_exec_ctx *exec_ctx,
return transport->vtable->get_peer(exec_ctx, transport);
}
-void grpc_transport_stream_op_finish_with_failure(
- grpc_exec_ctx *exec_ctx, grpc_transport_stream_op *op) {
- grpc_exec_ctx_enqueue(exec_ctx, op->recv_message_ready, false, NULL);
- grpc_exec_ctx_enqueue(exec_ctx, op->recv_initial_metadata_ready, false, NULL);
- grpc_exec_ctx_enqueue(exec_ctx, op->on_complete, false, NULL);
+void grpc_transport_stream_op_finish_with_failure(grpc_exec_ctx *exec_ctx,
+ grpc_transport_stream_op *op,
+ grpc_error *error) {
+ grpc_exec_ctx_push(exec_ctx, op->recv_message_ready, GRPC_ERROR_REF(error),
+ NULL);
+ grpc_exec_ctx_push(exec_ctx, op->recv_initial_metadata_ready,
+ GRPC_ERROR_REF(error), NULL);
+ grpc_exec_ctx_push(exec_ctx, op->on_complete, error, NULL);
}
void grpc_transport_stream_op_add_cancellation(grpc_transport_stream_op *op,
@@ -171,11 +174,11 @@ typedef struct {
grpc_closure closure;
} close_message_data;
-static void free_message(grpc_exec_ctx *exec_ctx, void *p, bool iomgr_success) {
+static void free_message(grpc_exec_ctx *exec_ctx, void *p, grpc_error *error) {
close_message_data *cmd = p;
gpr_slice_unref(cmd->message);
if (cmd->then_call != NULL) {
- cmd->then_call->cb(exec_ctx, cmd->then_call->cb_arg, iomgr_success);
+ cmd->then_call->cb(exec_ctx, cmd->then_call->cb_arg, GRPC_ERROR_REF(error));
}
gpr_free(cmd);
}
diff --git a/src/core/lib/transport/transport.h b/src/core/lib/transport/transport.h
index 482a9d1791..f2d750e870 100644
--- a/src/core/lib/transport/transport.h
+++ b/src/core/lib/transport/transport.h
@@ -154,7 +154,7 @@ typedef struct grpc_transport_op {
grpc_closure *on_connectivity_state_change;
grpc_connectivity_state *connectivity_state;
/** should the transport be disconnected */
- int disconnect;
+ grpc_error *disconnect_with_error;
/** should we send a goaway?
after a goaway is sent, once there are no more active calls on
the transport, the transport should disconnect */
@@ -216,7 +216,8 @@ void grpc_transport_destroy_stream(grpc_exec_ctx *exec_ctx,
grpc_stream *stream, void *and_free_memory);
void grpc_transport_stream_op_finish_with_failure(grpc_exec_ctx *exec_ctx,
- grpc_transport_stream_op *op);
+ grpc_transport_stream_op *op,
+ grpc_error *error);
void grpc_transport_stream_op_add_cancellation(grpc_transport_stream_op *op,
grpc_status_code status);
generated by cgit v1.2.3 (git 2.39.1) at 2024-07-07 07:14:55 +0000