WIP

1 files changed, 356 insertions, 241 deletions

diff --git a/src/core/ext/transport/cronet/transport/cronet_transport.c b/src/core/ext/transport/cronet/transport/cronet_transport.c
index 7e65def4de..1d75603980 100644
--- a/src/core/ext/transport/cronet/transport/cronet_transport.c
+++ b/src/core/ext/transport/cronet/transport/cronet_transport.c
@@ -50,8 +50,17 @@
 #include "third_party/objective_c/Cronet/cronet_c_for_grpc.h"
 
 #define GRPC_HEADER_SIZE_IN_BYTES 5
+// maximum ops in a batch. There is not much thinking behind this limit, except
+// that it seems to be enough for most use cases.
+#define MAX_PENDING_OPS 100
+
+#define CRONET_LOG(...)                          \
+  {                                              \
+    if (grpc_cronet_trace) gpr_log(__VA_ARGS__); \
+  }
 
-#define CRONET_LOG(...) {if (1) gpr_log(__VA_ARGS__);}
+// TODO (makdharma): Hook up into the wider tracing mechanism
+int grpc_cronet_trace = 1;
 
 enum OP_RESULT {
   ACTION_TAKEN_WITH_CALLBACK,
@@ -59,11 +68,10 @@ enum OP_RESULT {
   NO_ACTION_POSSIBLE
 };
 
-const char *OP_RESULT_STRING[] = {
-  "ACTION_TAKEN_WITH_CALLBACK",
-  "ACTION_TAKEN_NO_CALLBACK",
-  "NO_ACTION_POSSIBLE"
-};
+// Used for printing debug
+const char *op_result_string[] = {"ACTION_TAKEN_WITH_CALLBACK",
+                                  "ACTION_TAKEN_NO_CALLBACK",
+                                  "NO_ACTION_POSSIBLE"};
 
 enum OP_ID {
   OP_SEND_INITIAL_METADATA = 0,
@@ -82,32 +90,31 @@ enum OP_ID {
   OP_NUM_OPS
 };
 
-const char *op_id_string[] = {
-  "OP_SEND_INITIAL_METADATA",
-  "OP_SEND_MESSAGE",
-  "OP_SEND_TRAILING_METADATA",
-  "OP_RECV_MESSAGE",
-  "OP_RECV_INITIAL_METADATA",
-  "OP_RECV_TRAILING_METADATA",
-  "OP_CANCEL_ERROR",
-  "OP_ON_COMPLETE",
-  "OP_FAILED",
-  "OP_SUCCEEDED",
-  "OP_CANCELED",
-  "OP_RECV_MESSAGE_AND_ON_COMPLETE",
-  "OP_READ_REQ_MADE",
-  "OP_NUM_OPS"
-};
+const char *op_id_string[] = {"OP_SEND_INITIAL_METADATA",
+                              "OP_SEND_MESSAGE",
+                              "OP_SEND_TRAILING_METADATA",
+                              "OP_RECV_MESSAGE",
+                              "OP_RECV_INITIAL_METADATA",
+                              "OP_RECV_TRAILING_METADATA",
+                              "OP_CANCEL_ERROR",
+                              "OP_ON_COMPLETE",
+                              "OP_FAILED",
+                              "OP_SUCCEEDED",
+                              "OP_CANCELED",
+                              "OP_RECV_MESSAGE_AND_ON_COMPLETE",
+                              "OP_READ_REQ_MADE",
+                              "OP_NUM_OPS"};
 
 /* Cronet callbacks */
 
 static void on_request_headers_sent(cronet_bidirectional_stream *);
-static void on_response_headers_received(cronet_bidirectional_stream *,
-    const cronet_bidirectional_stream_header_array *,
-    const char *);
+static void on_response_headers_received(
+    cronet_bidirectional_stream *,
+    const cronet_bidirectional_stream_header_array *, const char *);
 static void on_write_completed(cronet_bidirectional_stream *, const char *);
 static void on_read_completed(cronet_bidirectional_stream *, char *, int);
-static void on_response_trailers_received(cronet_bidirectional_stream *,
+static void on_response_trailers_received(
+    cronet_bidirectional_stream *,
     const cronet_bidirectional_stream_header_array *);
 static void on_succeeded(cronet_bidirectional_stream *);
 static void on_failed(cronet_bidirectional_stream *, int);
@@ -120,8 +127,7 @@ static cronet_bidirectional_stream_callback cronet_callbacks = {
     on_response_trailers_received,
     on_succeeded,
     on_failed,
-    on_canceled
-};
+    on_canceled};
 
 // Cronet transport object
 struct grpc_cronet_transport {
@@ -132,7 +138,7 @@ struct grpc_cronet_transport {
 typedef struct grpc_cronet_transport grpc_cronet_transport;
 
 struct read_state {
-  // vars to store data coming from cronet
+  /* vars to store data coming from server */
   char *read_buffer;
   bool length_field_received;
   int received_bytes;
@@ -141,15 +147,15 @@ struct read_state {
   char grpc_header_bytes[GRPC_HEADER_SIZE_IN_BYTES];
   char *payload_field;
 
-  // vars for holding data destined for the application
+  /* vars for holding data destined for the application */
   struct grpc_slice_buffer_stream sbs;
   gpr_slice_buffer read_slice_buffer;
 
-  // vars for trailing metadata
+  /* vars for trailing metadata */
   grpc_chttp2_incoming_metadata_buffer trailing_metadata;
   bool trailing_metadata_valid;
 
-  // vars for initial metadata
+  /* vars for initial metadata */
   grpc_chttp2_incoming_metadata_buffer initial_metadata;
 };
 
@@ -157,16 +163,13 @@ struct write_state {
   char *write_buffer;
 };
 
-// maximum ops in a batch.. There is not much thinking behind this limit, except
-// that it seems to be enough for most use cases.
-#define MAX_PENDING_OPS 100
-
+/* track state of one stream op */
 struct op_state {
   bool state_op_done[OP_NUM_OPS];
   bool state_callback_received[OP_NUM_OPS];
-  // data structure for storing data coming from server
+  /* data structure for storing data coming from server */
   struct read_state rs;
-  // data structure for storing data going to the server
+  /* data structure for storing data going to the server */
   struct write_state ws;
 };
 
@@ -174,7 +177,7 @@ struct op_and_state {
   grpc_transport_stream_op op;
   struct op_state state;
   bool done;
-  struct stream_obj *s; // Pointer back to the stream object
+  struct stream_obj *s; /* Pointer back to the stream object */
 };
 
 struct op_storage {
@@ -190,73 +193,74 @@ struct stream_obj {
   grpc_cronet_transport curr_ct;
   grpc_stream *curr_gs;
   cronet_bidirectional_stream *cbs;
+  cronet_bidirectional_stream_header_array header_array;
 
-  // Used for executing callbacks for ops
+  /* Used for executing callbacks for ops */
   grpc_exec_ctx exec_ctx;
-  // This holds the state that is at stream level (response and req metadata)
+  /* Stream level state. Some state will be tracked both at stream and stream_op
+   * level */
   struct op_state state;
 
-  //struct op_state state;
-  // OP storage
+  /* OP storage */
   struct op_storage storage;
 
-  // Mutex to protect execute_curr_streaming_op
+  /* Mutex to protect storage */
   gpr_mu mu;
 };
 typedef struct stream_obj stream_obj;
 
-/* Globals */
-cronet_bidirectional_stream_header_array header_array;
-
-//
 static enum OP_RESULT execute_stream_op(struct op_and_state *oas);
 
-/*************************************************************
-  Op Storage
+/*
+  Add a new stream op to op storage.
 */
-
 static void add_to_storage(struct stream_obj *s, grpc_transport_stream_op *op) {
+  gpr_mu_lock(&s->mu);
   struct op_storage *storage = &s->storage;
   GPR_ASSERT(storage->num_pending_ops < MAX_PENDING_OPS);
   storage->num_pending_ops++;
   CRONET_LOG(GPR_DEBUG, "adding new op @wrptr=%d. %d in the queue.",
-      storage->wrptr, storage->num_pending_ops);
-  memcpy(&storage->pending_ops[storage->wrptr].op, op, sizeof(grpc_transport_stream_op));
-  memset(&storage->pending_ops[storage->wrptr].state, 0, sizeof(storage->pending_ops[storage->wrptr].state));
+             storage->wrptr, storage->num_pending_ops);
+  memcpy(&storage->pending_ops[storage->wrptr].op, op,
+         sizeof(grpc_transport_stream_op));
+  memset(&storage->pending_ops[storage->wrptr].state, 0,
+         sizeof(storage->pending_ops[storage->wrptr].state));
   storage->pending_ops[storage->wrptr].done = false;
   storage->pending_ops[storage->wrptr].s = s;
-  storage->wrptr = (storage->wrptr + 1) % MAX_PENDING_OPS; 
+  storage->wrptr = (storage->wrptr + 1) % MAX_PENDING_OPS;
+  gpr_mu_unlock(&s->mu);
 }
 
+/*
+  Cycle through ops and try to take next action. Break when either
+  an action with callback is taken, or no action is possible.
+  This can be executed from the Cronet network thread via cronet callback
+  or on the application supplied thread via the perform_stream_op function.
+*/
 static void execute_from_storage(stream_obj *s) {
-  // Cycle through ops and try to take next action. Break when either
-  // an action with callback is taken, or no action is possible.
-  // This can be executed from the Cronet network thread via cronet callback
-  // or on the application supplied thread via the perform_stream_op function.
-  if (1) {//gpr_mu_lock(&s->mu) == 0) {
-    gpr_mu_lock(&s->mu);
-    for (int i = 0; i < s->storage.wrptr; ) {
-      CRONET_LOG(GPR_DEBUG, "calling execute_stream_op[%d]. done = %d", i, s->storage.pending_ops[i].done);
-      if (s->storage.pending_ops[i].done) {
-        i++;
-        continue;
-      }
-      enum OP_RESULT result = execute_stream_op(&s->storage.pending_ops[i]);
-      CRONET_LOG(GPR_DEBUG, "%s = execute_stream_op[%d]", OP_RESULT_STRING[result], i);
-      if (result == NO_ACTION_POSSIBLE) {
-        i++;
-      } else if (result == ACTION_TAKEN_WITH_CALLBACK) {
-        break;
-      }
+  gpr_mu_lock(&s->mu);
+  for (int i = 0; i < s->storage.wrptr;) {
+    CRONET_LOG(GPR_DEBUG, "calling execute_stream_op[%d]. done = %d", i,
+               s->storage.pending_ops[i].done);
+    if (s->storage.pending_ops[i].done) {
+      i++;
+      continue;
+    }
+    enum OP_RESULT result = execute_stream_op(&s->storage.pending_ops[i]);
+    CRONET_LOG(GPR_DEBUG, "%s = execute_stream_op[%d]",
+               op_result_string[result], i);
+    if (result == NO_ACTION_POSSIBLE) {
+      i++;
+    } else if (result == ACTION_TAKEN_WITH_CALLBACK) {
+      break;
     }
-    gpr_mu_unlock(&s->mu);
   }
+  gpr_mu_unlock(&s->mu);
   grpc_exec_ctx_finish(&s->exec_ctx);
 }
 
-
-/*************************************************************
-Cronet Callback Ipmlementation
+/*
+  Cronet callback
 */
 static void on_failed(cronet_bidirectional_stream *stream, int net_error) {
   CRONET_LOG(GPR_DEBUG, "on_failed(%p, %d)", stream, net_error);
@@ -267,6 +271,9 @@ static void on_failed(cronet_bidirectional_stream *stream, int net_error) {
   execute_from_storage(s);
 }
 
+/*
+  Cronet callback
+*/
 static void on_canceled(cronet_bidirectional_stream *stream) {
   CRONET_LOG(GPR_DEBUG, "on_canceled(%p)", stream);
   stream_obj *s = (stream_obj *)stream->annotation;
@@ -276,6 +283,9 @@ static void on_canceled(cronet_bidirectional_stream *stream) {
   execute_from_storage(s);
 }
 
+/*
+  Cronet callback
+*/
 static void on_succeeded(cronet_bidirectional_stream *stream) {
   CRONET_LOG(GPR_DEBUG, "on_succeeded(%p)", stream);
   stream_obj *s = (stream_obj *)stream->annotation;
@@ -285,22 +295,33 @@ static void on_succeeded(cronet_bidirectional_stream *stream) {
   execute_from_storage(s);
 }
 
+/*
+  Cronet callback
+*/
 static void on_request_headers_sent(cronet_bidirectional_stream *stream) {
   CRONET_LOG(GPR_DEBUG, "W: on_request_headers_sent(%p)", stream);
   stream_obj *s = (stream_obj *)stream->annotation;
   s->state.state_op_done[OP_SEND_INITIAL_METADATA] = true;
   s->state.state_callback_received[OP_SEND_INITIAL_METADATA] = true;
+  /* Free the memory allocated for headers */
+  if (s->header_array.headers) {
+    gpr_free(s->header_array.headers);
+  }
   execute_from_storage(s);
 }
 
+/*
+  Cronet callback
+*/
 static void on_response_headers_received(
     cronet_bidirectional_stream *stream,
     const cronet_bidirectional_stream_header_array *headers,
     const char *negotiated_protocol) {
   CRONET_LOG(GPR_DEBUG, "R: on_response_headers_received(%p, %p, %s)", stream,
-            headers, negotiated_protocol);
+             headers, negotiated_protocol);
   stream_obj *s = (stream_obj *)stream->annotation;
-  memset(&s->state.rs.initial_metadata, 0, sizeof(s->state.rs.initial_metadata));
+  memset(&s->state.rs.initial_metadata, 0,
+         sizeof(s->state.rs.initial_metadata));
   grpc_chttp2_incoming_metadata_buffer_init(&s->state.rs.initial_metadata);
   unsigned int i = 0;
   for (i = 0; i < headers->count; i++) {
@@ -314,6 +335,9 @@ static void on_response_headers_received(
   execute_from_storage(s);
 }
 
+/*
+  Cronet callback
+*/
 static void on_write_completed(cronet_bidirectional_stream *stream,
                                const char *data) {
   stream_obj *s = (stream_obj *)stream->annotation;
@@ -326,17 +350,23 @@ static void on_write_completed(cronet_bidirectional_stream *stream,
   execute_from_storage(s);
 }
 
+/*
+  Cronet callback
+*/
 static void on_read_completed(cronet_bidirectional_stream *stream, char *data,
                               int count) {
   stream_obj *s = (stream_obj *)stream->annotation;
-  CRONET_LOG(GPR_DEBUG, "R: on_read_completed(%p, %p, %d)", stream, data, count);
+  CRONET_LOG(GPR_DEBUG, "R: on_read_completed(%p, %p, %d)", stream, data,
+             count);
   if (count > 0) {
     s->state.rs.received_bytes += count;
     s->state.rs.remaining_bytes -= count;
     if (s->state.rs.remaining_bytes > 0) {
-      //CRONET_LOG(GPR_DEBUG, "cronet_bidirectional_stream_read");
-      s->state.state_op_done[OP_READ_REQ_MADE] = true; // If at least one read request has been made
-      cronet_bidirectional_stream_read(s->cbs, s->state.rs.read_buffer + s->state.rs.received_bytes, s->state.rs.remaining_bytes);
+      CRONET_LOG(GPR_DEBUG, "cronet_bidirectional_stream_read");
+      s->state.state_op_done[OP_READ_REQ_MADE] = true;
+      cronet_bidirectional_stream_read(
+          s->cbs, s->state.rs.read_buffer + s->state.rs.received_bytes,
+          s->state.rs.remaining_bytes);
     } else {
       execute_from_storage(s);
     }
@@ -344,76 +374,82 @@ static void on_read_completed(cronet_bidirectional_stream *stream, char *data,
   }
 }
 
+/*
+  Cronet callback
+*/
 static void on_response_trailers_received(
     cronet_bidirectional_stream *stream,
     const cronet_bidirectional_stream_header_array *trailers) {
   CRONET_LOG(GPR_DEBUG, "R: on_response_trailers_received(%p,%p)", stream,
-          trailers);
+             trailers);
   stream_obj *s = (stream_obj *)stream->annotation;
-  memset(&s->state.rs.trailing_metadata, 0, sizeof(s->state.rs.trailing_metadata));
+  memset(&s->state.rs.trailing_metadata, 0,
+         sizeof(s->state.rs.trailing_metadata));
   s->state.rs.trailing_metadata_valid = false;
   grpc_chttp2_incoming_metadata_buffer_init(&s->state.rs.trailing_metadata);
   unsigned int i = 0;
   for (i = 0; i < trailers->count; i++) {
     CRONET_LOG(GPR_DEBUG, "trailer key=%s, value=%s", trailers->headers[i].key,
-            trailers->headers[i].value);
+               trailers->headers[i].value);
     grpc_chttp2_incoming_metadata_buffer_add(
-        &s->state.rs.trailing_metadata, grpc_mdelem_from_metadata_strings(
-                     grpc_mdstr_from_string(trailers->headers[i].key),
-                     grpc_mdstr_from_string(trailers->headers[i].value)));
+        &s->state.rs.trailing_metadata,
+        grpc_mdelem_from_metadata_strings(
+            grpc_mdstr_from_string(trailers->headers[i].key),
+            grpc_mdstr_from_string(trailers->headers[i].value)));
     s->state.rs.trailing_metadata_valid = true;
   }
   s->state.state_callback_received[OP_RECV_TRAILING_METADATA] = true;
   execute_from_storage(s);
 }
 
-/*************************************************************
-Utility functions. Can be in their own file
+/*
+ Utility function that takes the data from s->write_slice_buffer and assembles
+ into a contiguous byte stream with 5 byte gRPC header prepended.
 */
-// This function takes the data from s->write_slice_buffer and assembles into
-// a contiguous byte stream with 5 byte gRPC header prepended.
 static void create_grpc_frame(gpr_slice_buffer *write_slice_buffer,
-                              char **pp_write_buffer, int *p_write_buffer_size) {
+                              char **pp_write_buffer,
+                              int *p_write_buffer_size) {
   gpr_slice slice = gpr_slice_buffer_take_first(write_slice_buffer);
   size_t length = GPR_SLICE_LENGTH(slice);
-  // TODO (makdharma): FREE THIS!! HACK!
   *p_write_buffer_size = (int)length + GRPC_HEADER_SIZE_IN_BYTES;
+  /* This is freed in the on_write_completed callback */
   char *write_buffer = gpr_malloc(length + GRPC_HEADER_SIZE_IN_BYTES);
   *pp_write_buffer = write_buffer;
   uint8_t *p = (uint8_t *)write_buffer;
-  // Append 5 byte header
+  /* Append 5 byte header */
   *p++ = 0;
   *p++ = (uint8_t)(length >> 24);
   *p++ = (uint8_t)(length >> 16);
   *p++ = (uint8_t)(length >> 8);
   *p++ = (uint8_t)(length);
-  // append actual data
+  /* append actual data */
   memcpy(p, GPR_SLICE_START_PTR(slice), length);
 }
 
+/*
+ Convert metadata in a format that Cronet can consume
+*/
 static void convert_metadata_to_cronet_headers(
-                              grpc_linked_mdelem *head,
-                              const char *host,
-                              char **pp_url,
-                              cronet_bidirectional_stream_header **pp_headers,
-                              size_t *p_num_headers) {
+    grpc_linked_mdelem *head, const char *host, char **pp_url,
+    cronet_bidirectional_stream_header **pp_headers, size_t *p_num_headers) {
   grpc_linked_mdelem *curr = head;
-  // Walk the linked list and get number of header fields
+  /* Walk the linked list and get number of header fields */
   uint32_t num_headers_available = 0;
   while (curr != NULL) {
     curr = curr->next;
     num_headers_available++;
   }
-  // Allocate enough memory. TODO (makdharma): FREE MEMORY! HACK HACK
-  cronet_bidirectional_stream_header *headers = 
-                          (cronet_bidirectional_stream_header *)gpr_malloc(
-      sizeof(cronet_bidirectional_stream_header) * num_headers_available);
+  /* Allocate enough memory. It is freed in the on_request_headers_sent callback
+   */
+  cronet_bidirectional_stream_header *headers =
+      (cronet_bidirectional_stream_header *)gpr_malloc(
+          sizeof(cronet_bidirectional_stream_header) * num_headers_available);
   *pp_headers = headers;
 
-  // Walk the linked list again, this time copying the header fields.
-  // s->num_headers
-  // can be less than num_headers_available, as some headers are not used for
-  // cronet
+  /* Walk the linked list again, this time copying the header fields.
+    s->num_headers can be less than num_headers_available, as some headers
+    are not used for cronet
+   */
   curr = head;
   int num_headers = 0;
   while (num_headers < num_headers_available) {
@@ -423,15 +459,15 @@ static void convert_metadata_to_cronet_headers(
     const char *value = grpc_mdstr_as_c_string(mdelem->value);
     if (strcmp(key, ":scheme") == 0 || strcmp(key, ":method") == 0 ||
         strcmp(key, ":authority") == 0) {
-      // Cronet populates these fields on its own.
+      /* Cronet populates these fields on its own */
       continue;
     }
     if (strcmp(key, ":path") == 0) {
-      // Create URL by appending :path value to the hostname
+      /* Create URL by appending :path value to the hostname */
       gpr_asprintf(pp_url, "https://%s%s", host, value);
       continue;
     }
-    gpr_log(GPR_DEBUG, "header %s = %s", key, value);
+    CRONET_LOG(GPR_DEBUG, "header %s = %s", key, value);
     headers[num_headers].key = key;
     headers[num_headers].value = value;
     num_headers++;
@@ -453,261 +489,342 @@ static int parse_grpc_header(const uint8_t *data) {
 }
 
 /*
-Op Execution
+  Op Execution: Decide if one of the actions contained in the stream op can be
+  executed. This is the heart of the state machine.
 */
-static bool op_can_be_run(grpc_transport_stream_op *curr_op, struct op_state *stream_state, struct op_state *op_state, enum OP_ID op_id) {
-  // We use op's own state for most state, except for metadata related callbacks, which
-  // are at the stream level. TODO: WTF does this comment mean?
+static bool op_can_be_run(grpc_transport_stream_op *curr_op,
+                          struct op_state *stream_state,
+                          struct op_state *op_state, enum OP_ID op_id) {
   bool result = true;
-  // When call is canceled, every op can be run
-  if (stream_state->state_op_done[OP_CANCEL_ERROR] || stream_state->state_callback_received[OP_FAILED]) {
+  /* When call is canceled, every op can be run, except under following
+  conditions
+  */
+  if (stream_state->state_op_done[OP_CANCEL_ERROR] ||
+      stream_state->state_callback_received[OP_FAILED]) {
     if (op_id == OP_SEND_INITIAL_METADATA) result = false;
     if (op_id == OP_SEND_MESSAGE) result = false;
     if (op_id == OP_SEND_TRAILING_METADATA) result = false;
     if (op_id == OP_CANCEL_ERROR) result = false;
-    // already executed
-    if (op_id == OP_RECV_INITIAL_METADATA && stream_state->state_op_done[OP_RECV_INITIAL_METADATA]) result = false;
-    if (op_id == OP_RECV_MESSAGE && stream_state->state_op_done[OP_RECV_MESSAGE]) result = false;
-    if (op_id == OP_RECV_TRAILING_METADATA && stream_state->state_op_done[OP_RECV_TRAILING_METADATA]) result = false;
+    /* already executed */
+    if (op_id == OP_RECV_INITIAL_METADATA &&
+        stream_state->state_op_done[OP_RECV_INITIAL_METADATA])
+      result = false;
+    if (op_id == OP_RECV_MESSAGE &&
+        stream_state->state_op_done[OP_RECV_MESSAGE])
+      result = false;
+    if (op_id == OP_RECV_TRAILING_METADATA &&
+        stream_state->state_op_done[OP_RECV_TRAILING_METADATA])
+      result = false;
   } else if (op_id == OP_SEND_INITIAL_METADATA) {
-    // already executed
+    /* already executed */
     if (stream_state->state_op_done[OP_SEND_INITIAL_METADATA]) result = false;
   } else if (op_id == OP_RECV_INITIAL_METADATA) {
-    // already executed
+    /* already executed */
     if (stream_state->state_op_done[OP_RECV_INITIAL_METADATA]) result = false;
-    // we haven't sent headers yet.
-    else if (!stream_state->state_callback_received[OP_SEND_INITIAL_METADATA]) result = false;
-    // we haven't received headers yet.
-    else if (!stream_state->state_callback_received[OP_RECV_INITIAL_METADATA]) result = false;
+    /* we haven't sent headers yet. */
+    else if (!stream_state->state_callback_received[OP_SEND_INITIAL_METADATA])
+      result = false;
+    /* we haven't received headers yet. */
+    else if (!stream_state->state_callback_received[OP_RECV_INITIAL_METADATA])
+      result = false;
   } else if (op_id == OP_SEND_MESSAGE) {
-    // already executed (note we're checking op specific state, not stream state)
+    /* already executed (note we're checking op specific state, not stream
+     state) */
     if (op_state->state_op_done[OP_SEND_MESSAGE]) result = false;
-    // we haven't sent headers yet.
-    else if (!stream_state->state_callback_received[OP_SEND_INITIAL_METADATA]) result = false;
+    /* we haven't sent headers yet. */
+    else if (!stream_state->state_callback_received[OP_SEND_INITIAL_METADATA])
+      result = false;
   } else if (op_id == OP_RECV_MESSAGE) {
-    // already executed
+    /* already executed */
     if (op_state->state_op_done[OP_RECV_MESSAGE]) result = false;
-    // we haven't received headers yet.
-    else if (!stream_state->state_callback_received[OP_RECV_INITIAL_METADATA]) result = false;
+    /* we haven't received headers yet. */
+    else if (!stream_state->state_callback_received[OP_RECV_INITIAL_METADATA])
+      result = false;
   } else if (op_id == OP_RECV_TRAILING_METADATA) {
-    // already executed
+    /* already executed */
     if (stream_state->state_op_done[OP_RECV_TRAILING_METADATA]) result = false;
-    // we have asked for but haven't received message yet.
-    else if (stream_state->state_op_done[OP_READ_REQ_MADE] && !stream_state->state_op_done[OP_RECV_MESSAGE]) result = false;
-    // we haven't received trailers  yet.
-    else if (!stream_state->state_callback_received[OP_RECV_TRAILING_METADATA]) result = false;
+    /* we have asked for but haven't received message yet. */
+    else if (stream_state->state_op_done[OP_READ_REQ_MADE] &&
+             !stream_state->state_op_done[OP_RECV_MESSAGE])
+      result = false;
+    /* we haven't received trailers  yet. */
+    else if (!stream_state->state_callback_received[OP_RECV_TRAILING_METADATA])
+      result = false;
   } else if (op_id == OP_SEND_TRAILING_METADATA) {
-    // already executed
+    /* already executed */
     if (stream_state->state_op_done[OP_SEND_TRAILING_METADATA]) result = false;
-    // we haven't sent initial metadata yet
-    else if (!stream_state->state_callback_received[OP_SEND_INITIAL_METADATA]) result = false;
-    // we haven't sent message yet
-    // TODO: Streaming Write case is a problem. What if there is an outstanding write (2nd, 3rd,..) present.
-    else if (curr_op->send_message && !stream_state->state_op_done[OP_SEND_MESSAGE]) result = false;
-    // we haven't got on_write_completed for the send yet
-    else if (stream_state->state_op_done[OP_SEND_MESSAGE] && !stream_state->state_callback_received[OP_SEND_MESSAGE]) result = false;
+    /* we haven't sent initial metadata yet */
+    else if (!stream_state->state_callback_received[OP_SEND_INITIAL_METADATA])
+      result = false;
+    /* we haven't sent message yet */
+    else if (curr_op->send_message &&
+             !stream_state->state_op_done[OP_SEND_MESSAGE])
+      result = false;
+    /* we haven't got on_write_completed for the send yet */
+    else if (stream_state->state_op_done[OP_SEND_MESSAGE] &&
+             !stream_state->state_callback_received[OP_SEND_MESSAGE])
+      result = false;
   } else if (op_id == OP_CANCEL_ERROR) {
-    // already executed
+    /* already executed */
     if (stream_state->state_op_done[OP_CANCEL_ERROR]) result = false;
   } else if (op_id == OP_ON_COMPLETE) {
-    // already executed (note we're checking op specific state, not stream state)
+    /* already executed (note we're checking op specific state, not stream
+    state) */
     if (op_state->state_op_done[OP_ON_COMPLETE]) result = false;
-    // Check if every op that was asked for is done.
-    else if (curr_op->send_initial_metadata && !stream_state->state_callback_received[OP_SEND_INITIAL_METADATA]) result = false;
-    else if (curr_op->send_message && !stream_state->state_op_done[OP_SEND_MESSAGE]) result = false;
-    else if (curr_op->send_message && !stream_state->state_callback_received[OP_SEND_MESSAGE]) result = false;
-    else if (curr_op->send_trailing_metadata && !stream_state->state_op_done[OP_SEND_TRAILING_METADATA]) result = false;
-    else if (curr_op->recv_initial_metadata && !stream_state->state_op_done[OP_RECV_INITIAL_METADATA]) result = false;
-    else if (curr_op->recv_message && !stream_state->state_op_done[OP_RECV_MESSAGE]) result = false;
+    /* Check if every op that was asked for is done. */
+    else if (curr_op->send_initial_metadata &&
+             !stream_state->state_callback_received[OP_SEND_INITIAL_METADATA])
+      result = false;
+    else if (curr_op->send_message &&
+             !stream_state->state_op_done[OP_SEND_MESSAGE])
+      result = false;
+    else if (curr_op->send_message &&
+             !stream_state->state_callback_received[OP_SEND_MESSAGE])
+      result = false;
+    else if (curr_op->send_trailing_metadata &&
+             !stream_state->state_op_done[OP_SEND_TRAILING_METADATA])
+      result = false;
+    else if (curr_op->recv_initial_metadata &&
+             !stream_state->state_op_done[OP_RECV_INITIAL_METADATA])
+      result = false;
+    else if (curr_op->recv_message &&
+             !stream_state->state_op_done[OP_RECV_MESSAGE])
+      result = false;
     else if (curr_op->recv_trailing_metadata) {
-      //if (!stream_state->state_op_done[OP_SUCCEEDED]) result = false; gpr_log(GPR_DEBUG, "HACK!!");
-      // We aren't done with trailing metadata yet
-      if (!stream_state->state_op_done[OP_RECV_TRAILING_METADATA]) result = false;
-      // We've asked for actual message in an earlier op, and it hasn't been delivered yet.
-      // (TODO: What happens when multiple messages are asked for? How do you know when last message arrived?)
+      /* We aren't done with trailing metadata yet */
+      if (!stream_state->state_op_done[OP_RECV_TRAILING_METADATA])
+        result = false;
+      /* We've asked for actual message in an earlier op, and it hasn't been
+        delivered yet. */
       else if (stream_state->state_op_done[OP_READ_REQ_MADE]) {
-        // If this op is not the one asking for read, (which means some earlier op has asked), and the
-        // read hasn't been delivered.
-        if(!curr_op->recv_message && !stream_state->state_op_done[OP_SUCCEEDED]) result = false;
+        /* If this op is not the one asking for read, (which means some earlier
+          op has asked), and the read hasn't been delivered. */
+        if (!curr_op->recv_message &&
+            !stream_state->state_op_done[OP_SUCCEEDED])
+          result = false;
       }
     }
-    // We should see at least one on_write_completed for the trailers that we sent
-    else if (curr_op->send_trailing_metadata && !stream_state->state_callback_received[OP_SEND_MESSAGE]) result = false;
+    /* We should see at least one on_write_completed for the trailers that we
+      sent */
+    else if (curr_op->send_trailing_metadata &&
+             !stream_state->state_callback_received[OP_SEND_MESSAGE])
+      result = false;
   }
-  CRONET_LOG(GPR_DEBUG, "op_can_be_run %s : %s", op_id_string[op_id], result? "YES":"NO");
+  CRONET_LOG(GPR_DEBUG, "op_can_be_run %s : %s", op_id_string[op_id],
+             result ? "YES" : "NO");
   return result;
 }
 
 static enum OP_RESULT execute_stream_op(struct op_and_state *oas) {
-  // TODO TODO : This can be called from network thread and main thread. add a mutex.
   grpc_transport_stream_op *stream_op = &oas->op;
   struct stream_obj *s = oas->s;
   struct op_state *stream_state = &s->state;
-  //CRONET_LOG(GPR_DEBUG, "execute_stream_op");
   enum OP_RESULT result = NO_ACTION_POSSIBLE;
-  if (stream_op->send_initial_metadata && op_can_be_run(stream_op, stream_state, &oas->state, OP_SEND_INITIAL_METADATA)) {
+  if (stream_op->send_initial_metadata &&
+      op_can_be_run(stream_op, stream_state, &oas->state,
+                    OP_SEND_INITIAL_METADATA)) {
     CRONET_LOG(GPR_DEBUG, "running: %p OP_SEND_INITIAL_METADATA", oas);
-    // This OP is the beginning. Reset various states
+    /* This OP is the beginning. Reset various states */
     memset(&stream_state->rs, 0, sizeof(stream_state->rs));
     memset(&stream_state->ws, 0, sizeof(stream_state->ws));
     memset(stream_state->state_op_done, 0, sizeof(stream_state->state_op_done));
-    memset(stream_state->state_callback_received, 0, sizeof(stream_state->state_callback_received));
-    // Start new cronet stream
+    memset(stream_state->state_callback_received, 0,
+           sizeof(stream_state->state_callback_received));
+    /* Start new cronet stream. It is destroyed in on_succeeded, on_canceled,
+     * on_failed */
     GPR_ASSERT(s->cbs == NULL);
     CRONET_LOG(GPR_DEBUG, "cronet_bidirectional_stream_create");
-    s->cbs = cronet_bidirectional_stream_create(s->curr_ct.engine, s->curr_gs, &cronet_callbacks);
+    s->cbs = cronet_bidirectional_stream_create(s->curr_ct.engine, s->curr_gs,
+                                                &cronet_callbacks);
     char *url;
-    convert_metadata_to_cronet_headers(stream_op->send_initial_metadata->list.head,
-                               s->curr_ct.host, &url, &header_array.headers, &header_array.count);
-    header_array.capacity = header_array.count;
+    convert_metadata_to_cronet_headers(
+        stream_op->send_initial_metadata->list.head, s->curr_ct.host, &url,
+        &s->header_array.headers, &s->header_array.count);
+    s->header_array.capacity = s->header_array.count;
     CRONET_LOG(GPR_DEBUG, "cronet_bidirectional_stream_start %s", url);
-    cronet_bidirectional_stream_start(s->cbs, url, 0, "POST", &header_array, false);
+    cronet_bidirectional_stream_start(s->cbs, url, 0, "POST", &s->header_array,
+                                      false);
     stream_state->state_op_done[OP_SEND_INITIAL_METADATA] = true;
     result = ACTION_TAKEN_WITH_CALLBACK;
   } else if (stream_op->recv_initial_metadata &&
-                      op_can_be_run(stream_op, stream_state, &oas->state, OP_RECV_INITIAL_METADATA)) {
+             op_can_be_run(stream_op, stream_state, &oas->state,
+                           OP_RECV_INITIAL_METADATA)) {
     CRONET_LOG(GPR_DEBUG, "running: %p  OP_RECV_INITIAL_METADATA", oas);
     if (!stream_state->state_op_done[OP_CANCEL_ERROR]) {
-      grpc_chttp2_incoming_metadata_buffer_publish(&oas->s->state.rs.initial_metadata,
-                                               stream_op->recv_initial_metadata);
-      grpc_exec_ctx_sched(&s->exec_ctx, stream_op->recv_initial_metadata_ready, GRPC_ERROR_NONE, NULL);
+      grpc_chttp2_incoming_metadata_buffer_publish(
+          &oas->s->state.rs.initial_metadata, stream_op->recv_initial_metadata);
+      grpc_exec_ctx_sched(&s->exec_ctx, stream_op->recv_initial_metadata_ready,
+                          GRPC_ERROR_NONE, NULL);
     } else {
-      grpc_exec_ctx_sched(&s->exec_ctx, stream_op->recv_initial_metadata_ready, GRPC_ERROR_CANCELLED, NULL);
+      grpc_exec_ctx_sched(&s->exec_ctx, stream_op->recv_initial_metadata_ready,
+                          GRPC_ERROR_CANCELLED, NULL);
     }
     stream_state->state_op_done[OP_RECV_INITIAL_METADATA] = true;
     result = ACTION_TAKEN_NO_CALLBACK;
-  } else if (stream_op->send_message && op_can_be_run(stream_op, stream_state, &oas->state, OP_SEND_MESSAGE)) {
+  } else if (stream_op->send_message &&
+             op_can_be_run(stream_op, stream_state, &oas->state,
+                           OP_SEND_MESSAGE)) {
     CRONET_LOG(GPR_DEBUG, "running: %p  OP_SEND_MESSAGE", oas);
-    // TODO (makdharma): Make into a standalone function
     gpr_slice_buffer write_slice_buffer;
     gpr_slice slice;
     gpr_slice_buffer_init(&write_slice_buffer);
     grpc_byte_stream_next(NULL, stream_op->send_message, &slice,
                           stream_op->send_message->length, NULL);
-    // Check that compression flag is not ON. We don't support compression yet.
-    // TODO (makdharma): add compression support
+    /* Check that compression flag is OFF. We don't support compression yet. */
     GPR_ASSERT(stream_op->send_message->flags == 0);
     gpr_slice_buffer_add(&write_slice_buffer, slice);
-    GPR_ASSERT(write_slice_buffer.count == 1); // Empty request not handled yet
+    GPR_ASSERT(write_slice_buffer.count ==
+               1); /* Empty request not handled yet */
     if (write_slice_buffer.count > 0) {
       int write_buffer_size;
-      create_grpc_frame(&write_slice_buffer, &stream_state->ws.write_buffer, &write_buffer_size);
-      CRONET_LOG(GPR_DEBUG, "cronet_bidirectional_stream_write (%p)", stream_state->ws.write_buffer);
+      create_grpc_frame(&write_slice_buffer, &stream_state->ws.write_buffer,
+                        &write_buffer_size);
+      CRONET_LOG(GPR_DEBUG, "cronet_bidirectional_stream_write (%p)",
+                 stream_state->ws.write_buffer);
       stream_state->state_callback_received[OP_SEND_MESSAGE] = false;
       cronet_bidirectional_stream_write(s->cbs, stream_state->ws.write_buffer,
-                                        write_buffer_size, false); // TODO: What if this is not the last write?
+                                        write_buffer_size, false);
       result = ACTION_TAKEN_WITH_CALLBACK;
     }
     stream_state->state_op_done[OP_SEND_MESSAGE] = true;
     oas->state.state_op_done[OP_SEND_MESSAGE] = true;
-  } else if (stream_op->recv_message && op_can_be_run(stream_op, stream_state, &oas->state, OP_RECV_MESSAGE)) {
+  } else if (stream_op->recv_message &&
+             op_can_be_run(stream_op, stream_state, &oas->state,
+                           OP_RECV_MESSAGE)) {
     CRONET_LOG(GPR_DEBUG, "running: %p  OP_RECV_MESSAGE", oas);
     if (stream_state->state_op_done[OP_CANCEL_ERROR]) {
-      grpc_exec_ctx_sched(&s->exec_ctx, stream_op->recv_message_ready, GRPC_ERROR_CANCELLED, NULL);
+      grpc_exec_ctx_sched(&s->exec_ctx, stream_op->recv_message_ready,
+                          GRPC_ERROR_CANCELLED, NULL);
       stream_state->state_op_done[OP_RECV_MESSAGE] = true;
     } else if (stream_state->rs.length_field_received == false) {
-      if (stream_state->rs.received_bytes == GRPC_HEADER_SIZE_IN_BYTES && stream_state->rs.remaining_bytes == 0) {
-        // Start a read operation for data
+      if (stream_state->rs.received_bytes == GRPC_HEADER_SIZE_IN_BYTES &&
+          stream_state->rs.remaining_bytes == 0) {
+        /* Start a read operation for data */
         stream_state->rs.length_field_received = true;
         stream_state->rs.length_field = stream_state->rs.remaining_bytes =
-                             parse_grpc_header((const uint8_t *)stream_state->rs.read_buffer);
-        CRONET_LOG(GPR_DEBUG, "length field = %d", stream_state->rs.length_field);
+            parse_grpc_header((const uint8_t *)stream_state->rs.read_buffer);
+        CRONET_LOG(GPR_DEBUG, "length field = %d",
+                   stream_state->rs.length_field);
         if (stream_state->rs.length_field > 0) {
-          stream_state->rs.read_buffer = gpr_malloc(stream_state->rs.length_field);
+          stream_state->rs.read_buffer =
+              gpr_malloc(stream_state->rs.length_field);
           GPR_ASSERT(stream_state->rs.read_buffer);
           stream_state->rs.remaining_bytes = stream_state->rs.length_field;
           stream_state->rs.received_bytes = 0;
           CRONET_LOG(GPR_DEBUG, "cronet_bidirectional_stream_read");
-          stream_state->state_op_done[OP_READ_REQ_MADE] = true; // If at least one read request has been made
+          stream_state->state_op_done[OP_READ_REQ_MADE] =
+              true; /* Indicates that at least one read request has been made */
           cronet_bidirectional_stream_read(s->cbs, stream_state->rs.read_buffer,
-                                         stream_state->rs.remaining_bytes);
+                                           stream_state->rs.remaining_bytes);
           result = ACTION_TAKEN_WITH_CALLBACK;
         } else {
           stream_state->rs.remaining_bytes = 0;
           CRONET_LOG(GPR_DEBUG, "read operation complete. Empty response.");
           gpr_slice_buffer_init(&stream_state->rs.read_slice_buffer);
-          grpc_slice_buffer_stream_init(&stream_state->rs.sbs, &stream_state->rs.read_slice_buffer, 0);
-          *((grpc_byte_buffer **)stream_op->recv_message) = (grpc_byte_buffer *)&stream_state->rs.sbs;
-          grpc_exec_ctx_sched(&s->exec_ctx, stream_op->recv_message_ready, GRPC_ERROR_NONE, NULL);
+          grpc_slice_buffer_stream_init(&stream_state->rs.sbs,
+                                        &stream_state->rs.read_slice_buffer, 0);
+          *((grpc_byte_buffer **)stream_op->recv_message) =
+              (grpc_byte_buffer *)&stream_state->rs.sbs;
+          grpc_exec_ctx_sched(&s->exec_ctx, stream_op->recv_message_ready,
+                              GRPC_ERROR_NONE, NULL);
           stream_state->state_op_done[OP_RECV_MESSAGE] = true;
-          oas->state.state_op_done[OP_RECV_MESSAGE] = true; // Also set per op state.
+          oas->state.state_op_done[OP_RECV_MESSAGE] = true;
           result = ACTION_TAKEN_NO_CALLBACK;
         }
       } else if (stream_state->rs.remaining_bytes == 0) {
-        // Start a read operation for first 5 bytes (GRPC header)
+        /* Start a read operation for first 5 bytes (GRPC header) */
         stream_state->rs.read_buffer = stream_state->rs.grpc_header_bytes;
         stream_state->rs.remaining_bytes = GRPC_HEADER_SIZE_IN_BYTES;
         stream_state->rs.received_bytes = 0;
         CRONET_LOG(GPR_DEBUG, "cronet_bidirectional_stream_read");
-        stream_state->state_op_done[OP_READ_REQ_MADE] = true; // If at least one read request has been made
+        stream_state->state_op_done[OP_READ_REQ_MADE] =
+            true; /* Indicates that at least one read request has been made */
         cronet_bidirectional_stream_read(s->cbs, stream_state->rs.read_buffer,
                                          stream_state->rs.remaining_bytes);
       }
       result = ACTION_TAKEN_WITH_CALLBACK;
     } else if (stream_state->rs.remaining_bytes == 0) {
       CRONET_LOG(GPR_DEBUG, "read operation complete");
-      gpr_slice read_data_slice = gpr_slice_malloc((uint32_t)stream_state->rs.length_field);
+      gpr_slice read_data_slice =
+          gpr_slice_malloc((uint32_t)stream_state->rs.length_field);
       uint8_t *dst_p = GPR_SLICE_START_PTR(read_data_slice);
-      memcpy(dst_p, stream_state->rs.read_buffer, (size_t)stream_state->rs.length_field);
+      memcpy(dst_p, stream_state->rs.read_buffer,
+             (size_t)stream_state->rs.length_field);
       gpr_slice_buffer_init(&stream_state->rs.read_slice_buffer);
-      gpr_slice_buffer_add(&stream_state->rs.read_slice_buffer, read_data_slice);
-      grpc_slice_buffer_stream_init(&stream_state->rs.sbs, &stream_state->rs.read_slice_buffer, 0);
-      *((grpc_byte_buffer **)stream_op->recv_message) = (grpc_byte_buffer *)&stream_state->rs.sbs;
-      grpc_exec_ctx_sched(&s->exec_ctx, stream_op->recv_message_ready, GRPC_ERROR_NONE, NULL);
+      gpr_slice_buffer_add(&stream_state->rs.read_slice_buffer,
+                           read_data_slice);
+      grpc_slice_buffer_stream_init(&stream_state->rs.sbs,
+                                    &stream_state->rs.read_slice_buffer, 0);
+      *((grpc_byte_buffer **)stream_op->recv_message) =
+          (grpc_byte_buffer *)&stream_state->rs.sbs;
+      grpc_exec_ctx_sched(&s->exec_ctx, stream_op->recv_message_ready,
+                          GRPC_ERROR_NONE, NULL);
       stream_state->state_op_done[OP_RECV_MESSAGE] = true;
-      oas->state.state_op_done[OP_RECV_MESSAGE] = true; // Also set per op state.
-      // Clear read state of the stream, so next read op (if it were to come) will work
-      stream_state->rs.received_bytes = stream_state->rs.remaining_bytes = stream_state->rs.length_field_received = 0;
+      oas->state.state_op_done[OP_RECV_MESSAGE] = true;
+      /* Clear read state of the stream, so next read op (if it were to come)
+       * will work */
+      stream_state->rs.received_bytes = stream_state->rs.remaining_bytes =
+          stream_state->rs.length_field_received = 0;
       result = ACTION_TAKEN_NO_CALLBACK;
     }
   } else if (stream_op->recv_trailing_metadata &&
-               op_can_be_run(stream_op, stream_state, &oas->state, OP_RECV_TRAILING_METADATA)) {
+             op_can_be_run(stream_op, stream_state, &oas->state,
+                           OP_RECV_TRAILING_METADATA)) {
     CRONET_LOG(GPR_DEBUG, "running: %p  OP_RECV_TRAILING_METADATA", oas);
     if (oas->s->state.rs.trailing_metadata_valid) {
       grpc_chttp2_incoming_metadata_buffer_publish(
-          &oas->s->state.rs.trailing_metadata, stream_op->recv_trailing_metadata);
+          &oas->s->state.rs.trailing_metadata,
+          stream_op->recv_trailing_metadata);
       stream_state->rs.trailing_metadata_valid = false;
     }
     stream_state->state_op_done[OP_RECV_TRAILING_METADATA] = true;
     result = ACTION_TAKEN_NO_CALLBACK;
-  } else if (stream_op->send_trailing_metadata && op_can_be_run(stream_op, stream_state, &oas->state, OP_SEND_TRAILING_METADATA)) {
+  } else if (stream_op->send_trailing_metadata &&
+             op_can_be_run(stream_op, stream_state, &oas->state,
+                           OP_SEND_TRAILING_METADATA)) {
     CRONET_LOG(GPR_DEBUG, "running: %p  OP_SEND_TRAILING_METADATA", oas);
     CRONET_LOG(GPR_DEBUG, "cronet_bidirectional_stream_write (0)");
     stream_state->state_callback_received[OP_SEND_MESSAGE] = false;
     cronet_bidirectional_stream_write(s->cbs, "", 0, true);
     stream_state->state_op_done[OP_SEND_TRAILING_METADATA] = true;
     result = ACTION_TAKEN_WITH_CALLBACK;
-  } else if (stream_op->cancel_error && op_can_be_run(stream_op, stream_state, &oas->state, OP_CANCEL_ERROR)) {
+  } else if (stream_op->cancel_error &&
+             op_can_be_run(stream_op, stream_state, &oas->state,
+                           OP_CANCEL_ERROR)) {
     CRONET_LOG(GPR_DEBUG, "running: %p  OP_CANCEL_ERROR", oas);
     CRONET_LOG(GPR_DEBUG, "W: cronet_bidirectional_stream_cancel(%p)", s->cbs);
-    // Cancel might have come before creation of stream
     if (s->cbs) {
       cronet_bidirectional_stream_cancel(s->cbs);
     }
     stream_state->state_op_done[OP_CANCEL_ERROR] = true;
     result = ACTION_TAKEN_WITH_CALLBACK;
-  } else if (stream_op->on_complete && op_can_be_run(stream_op, stream_state, &oas->state, OP_ON_COMPLETE)) {
-    // All ops are complete. Call the on_complete callback
+  } else if (stream_op->on_complete &&
+             op_can_be_run(stream_op, stream_state, &oas->state,
+                           OP_ON_COMPLETE)) {
+    /* All actions in this stream_op are complete. Call the on_complete callback
+     */
     CRONET_LOG(GPR_DEBUG, "running: %p  OP_ON_COMPLETE", oas);
-    //CRONET_LOG(GPR_DEBUG, "calling on_complete");
-    grpc_exec_ctx_sched(&s->exec_ctx, stream_op->on_complete, GRPC_ERROR_NONE, NULL);
-    // Instead of setting stream state, use the op state as on_complete is on per op basis
+    grpc_exec_ctx_sched(&s->exec_ctx, stream_op->on_complete, GRPC_ERROR_NONE,
+                        NULL);
     oas->state.state_op_done[OP_ON_COMPLETE] = true;
-    oas->done = true; // Mark this op as completed
-    // reset any send or receive message state.
+    oas->done = true;
+    /* reset any send or receive message state. */
     stream_state->state_callback_received[OP_SEND_MESSAGE] = false;
     stream_state->state_op_done[OP_SEND_MESSAGE] = false;
     result = ACTION_TAKEN_NO_CALLBACK;
-    // If this is the on_complete callback being called for a received message - make a note
-    if (stream_op->recv_message) stream_state->state_op_done[OP_RECV_MESSAGE_AND_ON_COMPLETE] = true;
+    /* If this is the on_complete callback being called for a received message -
+      make a note */
+    if (stream_op->recv_message)
+      stream_state->state_op_done[OP_RECV_MESSAGE_AND_ON_COMPLETE] = true;
   } else {
-    //CRONET_LOG(GPR_DEBUG, "No op ready to run");
     result = NO_ACTION_POSSIBLE;
   }
   return result;
 }
 
-////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+/*
+  Functions used by upper layers to access transport functionality.
+*/
 
 static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
                        grpc_stream *gs, grpc_stream_refcount *refcount,
@@ -720,7 +837,8 @@ static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
   memset(&s->state.rs, 0, sizeof(s->state.rs));
   memset(&s->state.ws, 0, sizeof(s->state.ws));
   memset(s->state.state_op_done, 0, sizeof(s->state.state_op_done));
-  memset(s->state.state_callback_received, 0, sizeof(s->state.state_callback_received));
+  memset(s->state.state_callback_received, 0,
+         sizeof(s->state.state_callback_received));
   gpr_mu_init(&s->mu);
   s->exec_ctx = *exec_ctx;
   return 0;
@@ -744,19 +862,16 @@ static void perform_stream_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
 }
 
 static void destroy_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
-                           grpc_stream *gs, void *and_free_memory) {
-}
+                           grpc_stream *gs, void *and_free_memory) {}
 
-static void destroy_transport(grpc_exec_ctx *exec_ctx, grpc_transport *gt) {
-}
+static void destroy_transport(grpc_exec_ctx *exec_ctx, grpc_transport *gt) {}
 
 static char *get_peer(grpc_exec_ctx *exec_ctx, grpc_transport *gt) {
   return NULL;
 }
 
 static void perform_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
-                       grpc_transport_op *op) {
-}
+                       grpc_transport_op *op) {}
 
 const grpc_transport_vtable grpc_cronet_vtable = {sizeof(stream_obj),
                                                   "cronet_http",