grpc
/
grpc-node
mirror of https://github.com/grpc/grpc-node.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into cq_lockfree

This commit is contained in:
Sree Kuchibhotla 2017-04-13 14:26:12 -07:00
parent ee8068490b 48ba0d0c4a
commit dbaf97f0a1
7 changed files with 85 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
ext/call.cc
View File

@ -99,7 +99,6 @@ Local<Value> nanErrorWithCode(const char *msg, grpc_call_error code) {
bool CreateMetadataArray(Local<Object> metadata, grpc_metadata_array *array) {
HandleScope scope;
grpc_metadata_array_init(array);
Local<Array> keys = Nan::GetOwnPropertyNames(metadata).ToLocalChecked();
for (unsigned int i = 0; i < keys->Length(); i++) {
Local<String> current_key = Nan::To<String>(
@ -111,18 +110,20 @@ bool CreateMetadataArray(Local<Object> metadata, grpc_metadata_array *array) {
array->capacity += Local<Array>::Cast(value_array)->Length();
}
array->metadata = reinterpret_cast<grpc_metadata*>(
gpr_malloc(array->capacity * sizeof(grpc_metadata)));
gpr_zalloc(array->capacity * sizeof(grpc_metadata)));
for (unsigned int i = 0; i < keys->Length(); i++) {
Local<String> current_key(Nan::To<String>(keys->Get(i)).ToLocalChecked());
Local<Array> values = Local<Array>::Cast(
Nan::Get(metadata, current_key).ToLocalChecked());
grpc_slice key_slice = grpc_slice_intern(CreateSliceFromString(current_key));
grpc_slice key_slice = CreateSliceFromString(current_key);
grpc_slice key_intern_slice = grpc_slice_intern(key_slice);
grpc_slice_unref(key_slice);
for (unsigned int j = 0; j < values->Length(); j++) {
Local<Value> value = Nan::Get(values, j).ToLocalChecked();
grpc_metadata *current = &array->metadata[array->count];
current->key = key_slice;
current->key = key_intern_slice;
// Only allow binary headers for "-bin" keys
if (grpc_is_binary_header(key_slice)) {
if (grpc_is_binary_header(key_intern_slice)) {
if (::node::Buffer::HasInstance(value)) {
current->value = CreateSliceFromBuffer(value);
} else {
@ -142,6 +143,14 @@ bool CreateMetadataArray(Local<Object> metadata, grpc_metadata_array *array) {
return true;
}
void DestroyMetadataArray(grpc_metadata_array *array) {
for (size_t i = 0; i < array->count; i++) {
// Don't unref keys because they are interned
grpc_slice_unref(array->metadata[i].value);
}
grpc_metadata_array_destroy(array);
}
Local<Value> ParseMetadata(const grpc_metadata_array *metadata_array) {
EscapableHandleScope scope;
grpc_metadata *metadata_elements = metadata_array->metadata;
@ -179,6 +188,12 @@ Op::~Op() {
class SendMetadataOp : public Op {
public:
SendMetadataOp() {
grpc_metadata_array_init(&send_metadata);
}
~SendMetadataOp() {
DestroyMetadataArray(&send_metadata);
}
Local<Value> GetNodeValue() const {
EscapableHandleScope scope;
return scope.Escape(Nan::True());
@ -187,17 +202,16 @@ class SendMetadataOp : public Op {
if (!value->IsObject()) {
return false;
}
grpc_metadata_array array;
MaybeLocal<Object> maybe_metadata = Nan::To<Object>(value);
if (maybe_metadata.IsEmpty()) {
return false;
}
if (!CreateMetadataArray(maybe_metadata.ToLocalChecked(),
&array)) {
&send_metadata)) {
return false;
}
out->data.send_initial_metadata.count = array.count;
out->data.send_initial_metadata.metadata = array.metadata;
out->data.send_initial_metadata.count = send_metadata.count;
out->data.send_initial_metadata.metadata = send_metadata.metadata;
return true;
}
bool IsFinalOp() {
@ -207,6 +221,8 @@ class SendMetadataOp : public Op {
std::string GetTypeString() const {
return "send_metadata";
}
private:
grpc_metadata_array send_metadata;
};
class SendMessageOp : public Op {
@ -272,8 +288,12 @@ class SendClientCloseOp : public Op {
class SendServerStatusOp : public Op {
public:
SendServerStatusOp() {
grpc_metadata_array_init(&status_metadata);
}
~SendServerStatusOp() {
grpc_slice_unref(details);
DestroyMetadataArray(&status_metadata);
}
Local<Value> GetNodeValue() const {
EscapableHandleScope scope;
@ -313,12 +333,13 @@ class SendServerStatusOp : public Op {
}
Local<String> details = Nan::To<String>(
maybe_details.ToLocalChecked()).ToLocalChecked();
grpc_metadata_array array;
if (!CreateMetadataArray(metadata, &array)) {
if (!CreateMetadataArray(metadata, &status_metadata)) {
return false;
}
out->data.send_status_from_server.trailing_metadata_count = array.count;
out->data.send_status_from_server.trailing_metadata = array.metadata;
out->data.send_status_from_server.trailing_metadata_count =
status_metadata.count;
out->data.send_status_from_server.trailing_metadata =
status_metadata.metadata;
out->data.send_status_from_server.status =
static_cast<grpc_status_code>(code);
this->details = CreateSliceFromString(details);
@ -335,6 +356,7 @@ class SendServerStatusOp : public Op {
private:
grpc_slice details;
grpc_metadata_array status_metadata;
};
class GetMetadataOp : public Op {
@ -466,8 +488,10 @@ class ServerCloseResponseOp : public Op {
int cancelled;
};
tag::tag(Callback *callback, OpVec *ops, Call *call) :
tag::tag(Callback *callback, OpVec *ops, Call *call, Local<Value> call_value) :
callback(callback), ops(ops), call(call){
HandleScope scope;
call_persist.Reset(call_value);
}
tag::~tag() {
@ -513,15 +537,20 @@ void DestroyTag(void *tag) {
delete tag_struct;
}
void Call::DestroyCall() {
if (this->wrapped_call != NULL) {
grpc_call_destroy(this->wrapped_call);
this->wrapped_call = NULL;
}
}
Call::Call(grpc_call *call) : wrapped_call(call),
pending_batches(0),
has_final_op_completed(false) {
}
Call::~Call() {
if (wrapped_call != NULL) {
grpc_call_destroy(wrapped_call);
}
DestroyCall();
}
void Call::Init(Local<Object> exports) {
@ -568,12 +597,19 @@ void Call::CompleteBatch(bool is_final_op) {
}
this->pending_batches--;
if (this->has_final_op_completed && this->pending_batches == 0) {
grpc_call_destroy(this->wrapped_call);
this->wrapped_call = NULL;
this->DestroyCall();
}
}
NAN_METHOD(Call::New) {
/* Arguments:
* 0: Channel to make the call on
* 1: Method
* 2: Deadline
* 3: host
* 4: parent Call
* 5: propagation flags
*/
if (info.IsConstructCall()) {
Call *call;
if (info[0]->IsExternal()) {
@ -618,25 +654,26 @@ NAN_METHOD(Call::New) {
double deadline = Nan::To<double>(info[2]).FromJust();
grpc_channel *wrapped_channel = channel->GetWrappedChannel();
grpc_call *wrapped_call;
grpc_slice method = CreateSliceFromString(
Nan::To<String>(info[1]).ToLocalChecked());
if (info[3]->IsString()) {
grpc_slice *host = new grpc_slice;
*host = CreateSliceFromString(
Nan::To<String>(info[3]).ToLocalChecked());
wrapped_call = grpc_channel_create_call(
wrapped_channel, parent_call, propagate_flags,
GetCompletionQueue(), CreateSliceFromString(
Nan::To<String>(info[1]).ToLocalChecked()),
GetCompletionQueue(), method,
host, MillisecondsToTimespec(deadline), NULL);
delete host;
} else if (info[3]->IsUndefined() || info[3]->IsNull()) {
wrapped_call = grpc_channel_create_call(
wrapped_channel, parent_call, propagate_flags,
GetCompletionQueue(), CreateSliceFromString(
Nan::To<String>(info[1]).ToLocalChecked()),
GetCompletionQueue(), method,
NULL, MillisecondsToTimespec(deadline), NULL);
} else {
return Nan::ThrowTypeError("Call's fourth argument must be a string");
}
grpc_slice_unref(method);
call = new Call(wrapped_call);
Nan::Set(info.This(), Nan::New("channel_").ToLocalChecked(),
channel_object);
@ -721,7 +758,7 @@ NAN_METHOD(Call::StartBatch) {
Callback *callback = new Callback(callback_func);
grpc_call_error error = grpc_call_start_batch(
call->wrapped_call, &ops[0], nops, new struct tag(
callback, op_vector.release(), call), NULL);
callback, op_vector.release(), call, info.This()), NULL);
if (error != GRPC_CALL_OK) {
return Nan::ThrowError(nanErrorWithCode("startBatch failed", error));
}

9
ext/call.h
View File

@ -58,6 +58,8 @@ v8::Local<v8::Value> ParseMetadata(const grpc_metadata_array *metadata_array);
bool CreateMetadataArray(v8::Local<v8::Object> metadata,
grpc_metadata_array *array);
void DestroyMetadataArray(grpc_metadata_array *array);
/* Wrapper class for grpc_call structs. */
class Call : public Nan::ObjectWrap {
public:
@ -76,6 +78,8 @@ class Call : public Nan::ObjectWrap {
Call(const Call &);
Call &operator=(const Call &);
void DestroyCall();
static NAN_METHOD(New);
static NAN_METHOD(StartBatch);
static NAN_METHOD(Cancel);
@ -109,11 +113,14 @@ class Op {
typedef std::vector<unique_ptr<Op>> OpVec;
struct tag {
tag(Nan::Callback *callback, OpVec *ops, Call *call);
tag(Nan::Callback *callback, OpVec *ops, Call *call,
v8::Local<v8::Value> call_value);
~tag();
Nan::Callback *callback;
OpVec *ops;
Call *call;
Nan::Persistent<v8::Value, Nan::CopyablePersistentTraits<v8::Value>>
call_persist;
};
v8::Local<v8::Value> GetTagNodeValue(void *tag);

2
ext/call_credentials.cc
View File

@ -211,6 +211,7 @@ NAN_METHOD(PluginCallback) {
Utf8String details_utf8_str(info[1]);
char *details = *details_utf8_str;
grpc_metadata_array array;
grpc_metadata_array_init(&array);
Local<Object> callback_data = Nan::To<Object>(info[3]).ToLocalChecked();
if (!CreateMetadataArray(Nan::To<Object>(info[2]).ToLocalChecked(),
&array)){
@ -226,6 +227,7 @@ NAN_METHOD(PluginCallback) {
Nan::New("user_data").ToLocalChecked()
).ToLocalChecked().As<External>()->Value();
cb(user_data, array.metadata, array.count, code, details);
DestroyMetadataArray(&array);
}
NAUV_WORK_CB(SendPluginCallback) {

2
ext/channel.cc
View File

@ -280,7 +280,7 @@ NAN_METHOD(Channel::WatchConnectivityState) {
channel->wrapped_channel, last_state, MillisecondsToTimespec(deadline),
GetCompletionQueue(),
new struct tag(callback,
ops.release(), NULL));
ops.release(), NULL, Nan::Null()));
CompletionQueueNext();
}

12
ext/node_grpc.cc
View File

@ -286,8 +286,10 @@ NAN_METHOD(MetadataKeyIsLegal) {
"headerKeyIsLegal's argument must be a string");
}
Local<String> key = Nan::To<String>(info[0]).ToLocalChecked();
grpc_slice slice = CreateSliceFromString(key);
info.GetReturnValue().Set(static_cast<bool>(
grpc_header_key_is_legal(CreateSliceFromString(key))));
grpc_header_key_is_legal(slice)));
grpc_slice_unref(slice);
}
NAN_METHOD(MetadataNonbinValueIsLegal) {
@ -296,8 +298,10 @@ NAN_METHOD(MetadataNonbinValueIsLegal) {
"metadataNonbinValueIsLegal's argument must be a string");
}
Local<String> value = Nan::To<String>(info[0]).ToLocalChecked();
grpc_slice slice = CreateSliceFromString(value);
info.GetReturnValue().Set(static_cast<bool>(
grpc_header_nonbin_value_is_legal(CreateSliceFromString(value))));
grpc_header_nonbin_value_is_legal(slice)));
grpc_slice_unref(slice);
}
NAN_METHOD(MetadataKeyIsBinary) {
@ -306,8 +310,10 @@ NAN_METHOD(MetadataKeyIsBinary) {
"metadataKeyIsLegal's argument must be a string");
}
Local<String> key = Nan::To<String>(info[0]).ToLocalChecked();
grpc_slice slice = CreateSliceFromString(key);
info.GetReturnValue().Set(static_cast<bool>(
grpc_is_binary_header(CreateSliceFromString(key))));
grpc_is_binary_header(slice)));
grpc_slice_unref(slice);
}
static grpc_ssl_roots_override_result get_ssl_roots_override(

4
ext/server.cc
View File

@ -193,7 +193,7 @@ NAN_METHOD(Server::RequestCall) {
GetCompletionQueue(),
GetCompletionQueue(),
new struct tag(new Callback(info[0].As<Function>()), ops.release(),
NULL));
NULL, Nan::Null()));
if (error != GRPC_CALL_OK) {
return Nan::ThrowError(nanErrorWithCode("requestCall failed", error));
}
@ -246,7 +246,7 @@ NAN_METHOD(Server::TryShutdown) {
grpc_server_shutdown_and_notify(
server->wrapped_server, GetCompletionQueue(),
new struct tag(new Nan::Callback(info[0].As<Function>()), ops.release(),
NULL));
NULL, Nan::Null()));
CompletionQueueNext();
}

3
ext/server_uv.cc
View File

@ -118,7 +118,8 @@ void Server::ShutdownServer() {
grpc_server_shutdown_and_notify(
this->wrapped_server, GetCompletionQueue(),
new struct tag(new Callback(**shutdown_callback), ops.release(), NULL));
new struct tag(new Callback(**shutdown_callback), ops.release(), NULL,
Nan::Null()));
grpc_server_cancel_all_calls(this->wrapped_server);
CompletionQueueNext();
this->wrapped_server = NULL;