pull/16445/head
Yash Tibrewal 7 years ago
parent b473c72854
commit 1e21f68149
  1. 8
      src/core/ext/transport/chttp2/transport/chttp2_transport.cc
  2. 42
      src/core/ext/transport/chttp2/transport/context_list.cc
  3. 44
      src/core/ext/transport/chttp2/transport/context_list.h
  4. 2
      src/core/ext/transport/chttp2/transport/internal.h
  5. 1
      src/core/ext/transport/chttp2/transport/writing.cc

@ -236,6 +236,7 @@ static void init_transport(grpc_chttp2_transport* t,
size_t i;
int j;
grpc_tcp_set_write_timestamps_callback(ContextList::Execute);
GPR_ASSERT(strlen(GRPC_CHTTP2_CLIENT_CONNECT_STRING) ==
GRPC_CHTTP2_CLIENT_CONNECT_STRLEN);
@ -1026,11 +1027,13 @@ static void write_action_begin_locked(void* gt, grpc_error* error_ignored) {
static void write_action(void* gt, grpc_error* error) {
GPR_TIMER_SCOPE("write_action", 0);
grpc_chttp2_transport* t = static_cast<grpc_chttp2_transport*>(gt);
void *cl = t->cl;
t->cl = nullptr;
grpc_endpoint_write(
t->ep, &t->outbuf,
GRPC_CLOSURE_INIT(&t->write_action_end_locked, write_action_end_locked, t,
grpc_combiner_scheduler(t->combiner)),
nullptr);
grpc_combiner_scheduler(t->combiner)), cl
);
}
/* Callback from the grpc_endpoint after bytes have been written by calling
@ -1354,6 +1357,7 @@ static void perform_stream_op_locked(void* stream_op,
GRPC_STATS_INC_HTTP2_OP_BATCHES();
s->context = op->context;
if (grpc_http_trace.enabled()) {
char* str = grpc_transport_stream_op_batch_string(op);
gpr_log(GPR_INFO, "perform_stream_op_locked: %s; on_complete = %p", str,

@ -0,0 +1,42 @@
/*
*
* Copyright 2018 gRPC authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
#include <grpc/support/port_platform.h>
#include "src/core/ext/transport/chttp2/transport/context_list.h"
namespace {
void (*cb)(void *, grpc_core::Timestamps*);
}
void ContextList::Execute(ContextList *head, grpc_core::Timestamps *ts, grpc_error* error) {
ContextList *ptr;
while(head != nullptr) {
if(error == GRPC_ERROR_NONE) {
cb(head->context, ts);
}
ptr = head;
head_ = head->next;
gpr_free(ptr);
}
}
grpc_http2_set_write_timestamps_callback(void (*fn)(void *, grpc_core::Timestamps *)) {
cb = fn;
}

@ -22,26 +22,26 @@
#include <grpc/support/port_platform.h>
/** A list of RPC Contexts */
class ContextList {
struct ContextListElement {
void *context;
ContextListElement *next;
};
class ContextList {\
public:
ContextList() : head_(nullptr);
/* Creates a new element with \a context as the value and appends it to the
/* Creates a new element with \a context as the value and appends it to the
* list. */
void Append(void *context) {
ContextListElement *elem = static_cast<ContextListElement *>(gpr_malloc(sizeof(ContextlistELement)));
void Append(ContextList **head, void *context) {
/* Make sure context is not already present */
ContextList *ptr = *head;
while(ptr != nullptr) {
if(ptr->context == context) {
GPR_ASSERT(false);
}
}
ContextList *elem = static_cast<ContextListElement *>(gpr_malloc(sizeof(ContextList)));
elem->context = context;
elem->next = nullptr;
if(head_ == nullptr) {
head = elem;
if(*head_ == nullptr) {
*head = elem;
return;
}
ContextListElement *ptr = head_;
ptr = *head;
while(ptr->next != nullptr) {
ptr = ptr->next;
}
@ -50,17 +50,13 @@ class ContextList {
/* Executes a function \a fn with each context in the list and \a arg. It also
* frees up the entire list after this operation. */
void Execute(void (*fn)(void *context, void *arg)) {
ContextListElement *ptr;
while(head_ != nullptr){
fn(head->context, arg);
ptr = head_;
head_ = head_->next;
gpr_free(ptr);
}
}
void Execute(ContextList *head, grpc_core::Timestamps *ts, grpc_error* error);
private:
ContextListElement *head_;
void *context;
ContextListElement *next;
};
grpc_http2_set_write_timestamps_callback(void (*fn)(void *, grpc_core::Timestamps*));
#endif /* GRPC_CORE_EXT_TRANSPORT_CONTEXT_LIST_H */

@ -469,6 +469,7 @@ struct grpc_chttp2_transport {
bool keepalive_permit_without_calls;
/** keep-alive state machine state */
grpc_chttp2_keepalive_state keepalive_state;
ContextList *cl;
};
typedef enum {
@ -479,6 +480,7 @@ typedef enum {
} grpc_published_metadata_method;
struct grpc_chttp2_stream {
void *context;
grpc_chttp2_transport* t;
grpc_stream_refcount* refcount;

@ -487,6 +487,7 @@ class StreamWriteContext {
return; // early out: nothing to do
}
ContextList::Append(&t_->cl, s_->context);
while ((s_->flow_controlled_buffer.length > 0 ||
s_->compressed_data_buffer.length > 0) &&
data_send_context.max_outgoing() > 0) {

Loading…
Cancel
Save