GCC Code Coverage Report
Directory: ../ Exec Total Coverage
File: /home/iojs/build/workspace/node-test-commit-linux-coverage/nodes/benchmark/out/../src/stream_wrap.cc Lines: 176 181 97.2 %
Date: 2017-10-21 Branches: 65 90 72.2 %

Line Branch Exec Source
1
// Copyright Joyent, Inc. and other Node contributors.
2
//
3
// Permission is hereby granted, free of charge, to any person obtaining a
4
// copy of this software and associated documentation files (the
5
// "Software"), to deal in the Software without restriction, including
6
// without limitation the rights to use, copy, modify, merge, publish,
7
// distribute, sublicense, and/or sell copies of the Software, and to permit
8
// persons to whom the Software is furnished to do so, subject to the
9
// following conditions:
10
//
11
// The above copyright notice and this permission notice shall be included
12
// in all copies or substantial portions of the Software.
13
//
14
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
15
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
16
// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
17
// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
18
// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
19
// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
20
// USE OR OTHER DEALINGS IN THE SOFTWARE.
21
22
#include "stream_wrap.h"
23
#include "stream_base.h"
24
#include "stream_base-inl.h"
25
26
#include "env-inl.h"
27
#include "env.h"
28
#include "handle_wrap.h"
29
#include "node_buffer.h"
30
#include "node_counters.h"
31
#include "pipe_wrap.h"
32
#include "req-wrap.h"
33
#include "req-wrap-inl.h"
34
#include "tcp_wrap.h"
35
#include "udp_wrap.h"
36
#include "util.h"
37
#include "util-inl.h"
38
39
#include <stdlib.h>  // abort()
40
#include <string.h>  // memcpy()
41
#include <limits.h>  // INT_MAX
42
43
44
namespace node {
45
46
using v8::Context;
47
using v8::EscapableHandleScope;
48
using v8::FunctionCallbackInfo;
49
using v8::FunctionTemplate;
50
using v8::HandleScope;
51
using v8::Integer;
52
using v8::Local;
53
using v8::Object;
54
using v8::Value;
55
56
57
3169
void LibuvStreamWrap::Initialize(Local<Object> target,
58
                                 Local<Value> unused,
59
                                 Local<Context> context) {
60
3169
  Environment* env = Environment::GetCurrent(context);
61
62
  auto is_construct_call_callback =
63
151042
      [](const FunctionCallbackInfo<Value>& args) {
64
72352
    CHECK(args.IsConstructCall());
65
72352
    ClearWrap(args.This());
66
151042
  };
67
  Local<FunctionTemplate> sw =
68
3169
      FunctionTemplate::New(env->isolate(), is_construct_call_callback);
69
6338
  sw->InstanceTemplate()->SetInternalFieldCount(1);
70
  Local<String> wrapString =
71
3169
      FIXED_ONE_BYTE_STRING(env->isolate(), "ShutdownWrap");
72
3169
  sw->SetClassName(wrapString);
73
3169
  AsyncWrap::AddWrapMethods(env, sw);
74
6338
  target->Set(wrapString, sw->GetFunction());
75
76
  Local<FunctionTemplate> ww =
77
3169
      FunctionTemplate::New(env->isolate(), is_construct_call_callback);
78
6338
  ww->InstanceTemplate()->SetInternalFieldCount(1);
79
  Local<String> writeWrapString =
80
3169
      FIXED_ONE_BYTE_STRING(env->isolate(), "WriteWrap");
81
3169
  ww->SetClassName(writeWrapString);
82
3169
  AsyncWrap::AddWrapMethods(env, ww);
83
6338
  target->Set(writeWrapString, ww->GetFunction());
84
3169
  env->set_write_wrap_constructor_function(ww->GetFunction());
85
3169
}
86
87
88
10516
LibuvStreamWrap::LibuvStreamWrap(Environment* env,
89
                                 Local<Object> object,
90
                                 uv_stream_t* stream,
91
                                 AsyncWrap::ProviderType provider)
92
    : HandleWrap(env,
93
                 object,
94
                 reinterpret_cast<uv_handle_t*>(stream),
95
                 provider),
96
      StreamBase(env),
97
10516
      stream_(stream) {
98
10516
  set_after_write_cb({ OnAfterWriteImpl, this });
99
10516
  set_alloc_cb({ OnAllocImpl, this });
100
10516
  set_read_cb({ OnReadImpl, this });
101
10516
}
102
103
104
9538
void LibuvStreamWrap::AddMethods(Environment* env,
105
                                 v8::Local<v8::FunctionTemplate> target,
106
                                 int flags) {
107
9538
  env->SetProtoMethod(target, "updateWriteQueueSize", UpdateWriteQueueSize);
108
9538
  env->SetProtoMethod(target, "setBlocking", SetBlocking);
109
9538
  StreamBase::AddMethods<LibuvStreamWrap>(env, target, flags);
110
9538
}
111
112
113
104
int LibuvStreamWrap::GetFD() {
114
104
  int fd = -1;
115
#if !defined(_WIN32)
116
104
  if (stream() != nullptr)
117
104
    uv_fileno(reinterpret_cast<uv_handle_t*>(stream()), &fd);
118
#endif
119
104
  return fd;
120
}
121
122
123
76324
bool LibuvStreamWrap::IsAlive() {
124
76324
  return HandleWrap::IsAlive(this);
125
}
126
127
128
1105
bool LibuvStreamWrap::IsClosing() {
129
1105
  return uv_is_closing(reinterpret_cast<uv_handle_t*>(stream()));
130
}
131
132
133
void* LibuvStreamWrap::Cast() {
134
  return reinterpret_cast<void*>(this);
135
}
136
137
138
49724
AsyncWrap* LibuvStreamWrap::GetAsyncWrap() {
139
49724
  return static_cast<AsyncWrap*>(this);
140
}
141
142
143
34774
bool LibuvStreamWrap::IsIPCPipe() {
144
34774
  return is_named_pipe_ipc();
145
}
146
147
148
36796
uint32_t LibuvStreamWrap::UpdateWriteQueueSize() {
149
36796
  HandleScope scope(env()->isolate());
150
36796
  uint32_t write_queue_size = stream()->write_queue_size;
151
36796
  object()->Set(env()->context(),
152
                env()->write_queue_size_string(),
153
                Integer::NewFromUnsigned(env()->isolate(),
154
220776
                                         write_queue_size)).FromJust();
155
36796
  return write_queue_size;
156
}
157
158
159
8124
int LibuvStreamWrap::ReadStart() {
160
8124
  return uv_read_start(stream(), OnAlloc, OnRead);
161
}
162
163
164
243
int LibuvStreamWrap::ReadStop() {
165
243
  return uv_read_stop(stream());
166
}
167
168
169
29228
void LibuvStreamWrap::OnAlloc(uv_handle_t* handle,
170
                         size_t suggested_size,
171
                         uv_buf_t* buf) {
172
29228
  LibuvStreamWrap* wrap = static_cast<LibuvStreamWrap*>(handle->data);
173
29228
  HandleScope scope(wrap->env()->isolate());
174
29228
  Context::Scope context_scope(wrap->env()->context());
175
176
29228
  CHECK_EQ(wrap->stream(), reinterpret_cast<uv_stream_t*>(handle));
177
178
58456
  return static_cast<StreamBase*>(wrap)->OnAlloc(suggested_size, buf);
179
}
180
181
182
22987
void LibuvStreamWrap::OnAllocImpl(size_t size, uv_buf_t* buf, void* ctx) {
183
22987
  buf->base = node::Malloc(size);
184
22987
  buf->len = size;
185
22987
}
186
187
188
template <class WrapType, class UVType>
189
151
static Local<Object> AcceptHandle(Environment* env, LibuvStreamWrap* parent) {
190
151
  EscapableHandleScope scope(env->isolate());
191
  Local<Object> wrap_obj;
192
  UVType* handle;
193
194
151
  wrap_obj = WrapType::Instantiate(env, parent);
195

151
  if (wrap_obj.IsEmpty())
196
    return Local<Object>();
197
198
  WrapType* wrap;
199

151
  ASSIGN_OR_RETURN_UNWRAP(&wrap, wrap_obj, Local<Object>());
200
151
  handle = wrap->UVHandle();
201
202

151
  if (uv_accept(parent->stream(), reinterpret_cast<uv_stream_t*>(handle)))
203
    ABORT();
204
205
151
  return scope.Escape(wrap_obj);
206
}
207
208
209
23507
void LibuvStreamWrap::OnReadImpl(ssize_t nread,
210
                            const uv_buf_t* buf,
211
                            uv_handle_type pending,
212
                            void* ctx) {
213
23507
  LibuvStreamWrap* wrap = static_cast<LibuvStreamWrap*>(ctx);
214
23507
  Environment* env = wrap->env();
215
23507
  HandleScope handle_scope(env->isolate());
216
43413
  Context::Scope context_scope(env->context());
217
218
  Local<Object> pending_obj;
219
220
23507
  if (nread < 0)  {
221
3538
    if (buf->base != nullptr)
222
3018
      free(buf->base);
223
3538
    wrap->EmitData(nread, Local<Object>(), pending_obj);
224
3513
    return;
225
  }
226
227
19969
  if (nread == 0) {
228
13
    if (buf->base != nullptr)
229
13
      free(buf->base);
230
13
    return;
231
  }
232
233
19956
  CHECK_LE(static_cast<size_t>(nread), buf->len);
234
19956
  char* base = node::Realloc(buf->base, nread);
235
236
19956
  if (pending == UV_TCP) {
237
131
    pending_obj = AcceptHandle<TCPWrap, uv_tcp_t>(env, wrap);
238
19825
  } else if (pending == UV_NAMED_PIPE) {
239
1
    pending_obj = AcceptHandle<PipeWrap, uv_pipe_t>(env, wrap);
240
19824
  } else if (pending == UV_UDP) {
241
19
    pending_obj = AcceptHandle<UDPWrap, uv_udp_t>(env, wrap);
242
  } else {
243
19805
    CHECK_EQ(pending, UV_UNKNOWN_HANDLE);
244
  }
245
246
39912
  Local<Object> obj = Buffer::New(env, base, nread).ToLocalChecked();
247
39862
  wrap->EmitData(nread, obj, pending_obj);
248
}
249
250
251
29424
void LibuvStreamWrap::OnRead(uv_stream_t* handle,
252
                        ssize_t nread,
253
                        const uv_buf_t* buf) {
254
29424
  LibuvStreamWrap* wrap = static_cast<LibuvStreamWrap*>(handle->data);
255
29424
  HandleScope scope(wrap->env()->isolate());
256
29424
  Context::Scope context_scope(wrap->env()->context());
257
29424
  uv_handle_type type = UV_UNKNOWN_HANDLE;
258
259

31604
  if (wrap->is_named_pipe_ipc() &&
260
2180
      uv_pipe_pending_count(reinterpret_cast<uv_pipe_t*>(handle)) > 0) {
261
151
    type = uv_pipe_pending_type(reinterpret_cast<uv_pipe_t*>(handle));
262
  }
263
264
  // We should not be getting this callback if someone as already called
265
  // uv_close() on the handle.
266
58848
  CHECK_EQ(wrap->persistent().IsEmpty(), false);
267
268
29424
  if (nread > 0) {
269
25614
    if (wrap->is_tcp()) {
270
      NODE_COUNT_NET_BYTES_RECV(nread);
271
5465
    } else if (wrap->is_named_pipe()) {
272
      NODE_COUNT_PIPE_BYTES_RECV(nread);
273
    }
274
  }
275
276
58766
  static_cast<StreamBase*>(wrap)->OnRead(nread, buf, type);
277
29342
}
278
279
280
1
void LibuvStreamWrap::UpdateWriteQueueSize(
281
    const FunctionCallbackInfo<Value>& args) {
282
  LibuvStreamWrap* wrap;
283
2
  ASSIGN_OR_RETURN_UNWRAP(&wrap, args.Holder());
284
285
1
  uint32_t write_queue_size = wrap->UpdateWriteQueueSize();
286
2
  args.GetReturnValue().Set(write_queue_size);
287
}
288
289
290
16
void LibuvStreamWrap::SetBlocking(const FunctionCallbackInfo<Value>& args) {
291
  LibuvStreamWrap* wrap;
292
16
  ASSIGN_OR_RETURN_UNWRAP(&wrap, args.Holder());
293
294
16
  CHECK_GT(args.Length(), 0);
295
16
  if (!wrap->IsAlive())
296
    return args.GetReturnValue().Set(UV_EINVAL);
297
298
32
  bool enable = args[0]->IsTrue();
299
48
  args.GetReturnValue().Set(uv_stream_set_blocking(wrap->stream(), enable));
300
}
301
302
303
2123
int LibuvStreamWrap::DoShutdown(ShutdownWrap* req_wrap) {
304
  int err;
305
2123
  err = uv_shutdown(req_wrap->req(), stream(), AfterShutdown);
306
2123
  req_wrap->Dispatched();
307
2123
  return err;
308
}
309
310
311
2117
void LibuvStreamWrap::AfterShutdown(uv_shutdown_t* req, int status) {
312
2117
  ShutdownWrap* req_wrap = ShutdownWrap::from_req(req);
313
2117
  CHECK_NE(req_wrap, nullptr);
314
2117
  HandleScope scope(req_wrap->env()->isolate());
315
2117
  Context::Scope context_scope(req_wrap->env()->context());
316
4234
  req_wrap->Done(status);
317
2117
}
318
319
320
// NOTE: Call to this function could change both `buf`'s and `count`'s
321
// values, shifting their base and decrementing their length. This is
322
// required in order to skip the data that was successfully written via
323
// uv_try_write().
324
52494
int LibuvStreamWrap::DoTryWrite(uv_buf_t** bufs, size_t* count) {
325
  int err;
326
  size_t written;
327
52494
  uv_buf_t* vbufs = *bufs;
328
52494
  size_t vcount = *count;
329
330
52494
  err = uv_try_write(stream(), vbufs, vcount);
331

52494
  if (err == UV_ENOSYS || err == UV_EAGAIN)
332
15
    return 0;
333
52479
  if (err < 0)
334
1
    return err;
335
336
  // Slice off the buffers: skip all written buffers and slice the one that
337
  // was partially written.
338
52478
  written = err;
339
107480
  for (; vcount > 0; vbufs++, vcount--) {
340
    // Slice
341
55066
    if (vbufs[0].len > written) {
342
64
      vbufs[0].base += written;
343
64
      vbufs[0].len -= written;
344
64
      written = 0;
345
64
      break;
346
347
    // Discard
348
    } else {
349
55002
      written -= vbufs[0].len;
350
    }
351
  }
352
353
52478
  *bufs = vbufs;
354
52478
  *count = vcount;
355
356
52478
  return 0;
357
}
358
359
360
14965
int LibuvStreamWrap::DoWrite(WriteWrap* w,
361
                        uv_buf_t* bufs,
362
                        size_t count,
363
                        uv_stream_t* send_handle) {
364
  int r;
365
14965
  if (send_handle == nullptr) {
366
14872
    r = uv_write(w->req(), stream(), bufs, count, AfterWrite);
367
  } else {
368
93
    r = uv_write2(w->req(), stream(), bufs, count, send_handle, AfterWrite);
369
  }
370
371
14965
  if (!r) {
372
14949
    size_t bytes = 0;
373
70118
    for (size_t i = 0; i < count; i++)
374
55169
      bytes += bufs[i].len;
375
14949
    if (stream()->type == UV_TCP) {
376
      NODE_COUNT_NET_BYTES_SENT(bytes);
377
136
    } else if (stream()->type == UV_NAMED_PIPE) {
378
      NODE_COUNT_PIPE_BYTES_SENT(bytes);
379
    }
380
  }
381
382
14965
  w->Dispatched();
383
14965
  UpdateWriteQueueSize();
384
385
14965
  return r;
386
}
387
388
389
14939
void LibuvStreamWrap::AfterWrite(uv_write_t* req, int status) {
390
14939
  WriteWrap* req_wrap = WriteWrap::from_req(req);
391
14939
  CHECK_NE(req_wrap, nullptr);
392
14939
  HandleScope scope(req_wrap->env()->isolate());
393
14939
  Context::Scope context_scope(req_wrap->env()->context());
394
29876
  req_wrap->Done(status);
395
14937
}
396
397
398
11317
void LibuvStreamWrap::OnAfterWriteImpl(WriteWrap* w, void* ctx) {
399
11317
  LibuvStreamWrap* wrap = static_cast<LibuvStreamWrap*>(ctx);
400
11317
  wrap->UpdateWriteQueueSize();
401
11317
}
402
403
}  // namespace node
404
405
3245
NODE_MODULE_CONTEXT_AWARE_BUILTIN(stream_wrap,
406
                                  node::LibuvStreamWrap::Initialize)