GCC Code Coverage Report
Directory: ../ Exec Total Coverage
File: /home/node-core-coverage/node-core-coverage/workdir/node/out/../src/stream_wrap.cc Lines: 156 162 96.3 %
Date: 2016-09-05 Branches: 48 72 66.7 %

Line Branch Exec Source
1
#include "stream_wrap.h"
2
#include "stream_base.h"
3
#include "stream_base-inl.h"
4
5
#include "env-inl.h"
6
#include "env.h"
7
#include "handle_wrap.h"
8
#include "node_buffer.h"
9
#include "node_counters.h"
10
#include "pipe_wrap.h"
11
#include "req-wrap.h"
12
#include "req-wrap-inl.h"
13
#include "tcp_wrap.h"
14
#include "udp_wrap.h"
15
#include "util.h"
16
#include "util-inl.h"
17
18
#include <stdlib.h>  // abort()
19
#include <string.h>  // memcpy()
20
#include <limits.h>  // INT_MAX
21
22
23
namespace node {
24
25
using v8::Context;
26
using v8::EscapableHandleScope;
27
using v8::FunctionCallbackInfo;
28
using v8::FunctionTemplate;
29
using v8::HandleScope;
30
using v8::Integer;
31
using v8::Local;
32
using v8::Object;
33
using v8::Value;
34
35
36
1576
void StreamWrap::Initialize(Local<Object> target,
37
                            Local<Value> unused,
38
                            Local<Context> context) {
39
1576
  Environment* env = Environment::GetCurrent(context);
40
41
  Local<FunctionTemplate> sw =
42
3152
      FunctionTemplate::New(env->isolate(), ShutdownWrap::NewShutdownWrap);
43
3152
  sw->InstanceTemplate()->SetInternalFieldCount(1);
44
1576
  sw->SetClassName(FIXED_ONE_BYTE_STRING(env->isolate(), "ShutdownWrap"));
45
6304
  target->Set(FIXED_ONE_BYTE_STRING(env->isolate(), "ShutdownWrap"),
46
1576
              sw->GetFunction());
47
48
  Local<FunctionTemplate> ww =
49
3152
      FunctionTemplate::New(env->isolate(), WriteWrap::NewWriteWrap);
50
3152
  ww->InstanceTemplate()->SetInternalFieldCount(1);
51
1576
  ww->SetClassName(FIXED_ONE_BYTE_STRING(env->isolate(), "WriteWrap"));
52
6304
  target->Set(FIXED_ONE_BYTE_STRING(env->isolate(), "WriteWrap"),
53
1576
              ww->GetFunction());
54
1576
  env->set_write_wrap_constructor_function(ww->GetFunction());
55
1576
}
56
57
58
7889
StreamWrap::StreamWrap(Environment* env,
59
                       Local<Object> object,
60
                       uv_stream_t* stream,
61
                       AsyncWrap::ProviderType provider,
62
                       AsyncWrap* parent)
63
    : HandleWrap(env,
64
                 object,
65
                 reinterpret_cast<uv_handle_t*>(stream),
66
                 provider,
67
                 parent),
68
      StreamBase(env),
69
15778
      stream_(stream) {
70
15778
  set_after_write_cb({ OnAfterWriteImpl, this });
71
15778
  set_alloc_cb({ OnAllocImpl, this });
72
7889
  set_read_cb({ OnReadImpl, this });
73
7889
}
74
75
76
4737
void StreamWrap::AddMethods(Environment* env,
77
                            v8::Local<v8::FunctionTemplate> target,
78
                            int flags) {
79
4737
  env->SetProtoMethod(target, "setBlocking", SetBlocking);
80
4737
  StreamBase::AddMethods<StreamWrap>(env, target, flags);
81
4737
}
82
83
84
15
int StreamWrap::GetFD() {
85
15
  int fd = -1;
86
#if !defined(_WIN32)
87
15
  if (stream() != nullptr)
88
15
    uv_fileno(reinterpret_cast<uv_handle_t*>(stream()), &fd);
89
#endif
90
15
  return fd;
91
}
92
93
94
37615
bool StreamWrap::IsAlive() {
95
75230
  return HandleWrap::IsAlive(this);
96
}
97
98
99
2
bool StreamWrap::IsClosing() {
100
2
  return uv_is_closing(reinterpret_cast<uv_handle_t*>(stream()));
101
}
102
103
104
void* StreamWrap::Cast() {
105
  return reinterpret_cast<void*>(this);
106
}
107
108
109
34811
AsyncWrap* StreamWrap::GetAsyncWrap() {
110
34811
  return static_cast<AsyncWrap*>(this);
111
}
112
113
114
15074
bool StreamWrap::IsIPCPipe() {
115
15074
  return is_named_pipe_ipc();
116
}
117
118
119
34499
void StreamWrap::UpdateWriteQueueSize() {
120
68998
  HandleScope scope(env()->isolate());
121
  Local<Integer> write_queue_size =
122
34499
      Integer::NewFromUnsigned(env()->isolate(), stream()->write_queue_size);
123
137996
  object()->Set(env()->write_queue_size_string(), write_queue_size);
124
34499
}
125
126
127
6894
int StreamWrap::ReadStart() {
128
6894
  return uv_read_start(stream(), OnAlloc, OnRead);
129
}
130
131
132
225
int StreamWrap::ReadStop() {
133
225
  return uv_read_stop(stream());
134
}
135
136
137
25004
void StreamWrap::OnAlloc(uv_handle_t* handle,
138
                         size_t suggested_size,
139
                         uv_buf_t* buf) {
140
25004
  StreamWrap* wrap = static_cast<StreamWrap*>(handle->data);
141
50008
  HandleScope scope(wrap->env()->isolate());
142
100016
  Context::Scope context_scope(wrap->env()->context());
143
144
25004
  CHECK_EQ(wrap->stream(), reinterpret_cast<uv_stream_t*>(handle));
145
146
75012
  return static_cast<StreamBase*>(wrap)->OnAlloc(suggested_size, buf);
147
}
148
149
150
19606
void StreamWrap::OnAllocImpl(size_t size, uv_buf_t* buf, void* ctx) {
151
19606
  buf->base = static_cast<char*>(malloc(size));
152
19606
  buf->len = size;
153
154

19606
  if (buf->base == nullptr && size > 0) {
155
    FatalError(
156
        "node::StreamWrap::DoAlloc(size_t, uv_buf_t*, void*)",
157
        "Out Of Memory");
158
  }
159
19606
}
160
161
162
template <class WrapType, class UVType>
163
139
static Local<Object> AcceptHandle(Environment* env, StreamWrap* parent) {
164
278
  EscapableHandleScope scope(env->isolate());
165
139
  Local<Object> wrap_obj;
166
  UVType* handle;
167
168
139
  wrap_obj = WrapType::Instantiate(env, parent);
169

139
  if (wrap_obj.IsEmpty())
170
    return Local<Object>();
171
172
  WrapType* wrap;
173

139
  ASSIGN_OR_RETURN_UNWRAP(&wrap, wrap_obj, Local<Object>());
174
264
  handle = wrap->UVHandle();
175
176

139
  if (uv_accept(parent->stream(), reinterpret_cast<uv_stream_t*>(handle)))
177
    ABORT();
178
179
  return scope.Escape(wrap_obj);
180
}
181
182
183
19831
void StreamWrap::OnReadImpl(ssize_t nread,
184
                            const uv_buf_t* buf,
185
                            uv_handle_type pending,
186
                            void* ctx) {
187
19831
  StreamWrap* wrap = static_cast<StreamWrap*>(ctx);
188
19831
  Environment* env = wrap->env();
189
37422
  HandleScope handle_scope(env->isolate());
190
57253
  Context::Scope context_scope(env->context());
191
192
19831
  Local<Object> pending_obj;
193
194
19831
  if (nread < 0)  {
195
2189
    if (buf->base != nullptr)
196
1964
      free(buf->base);
197
2189
    wrap->EmitData(nread, Local<Object>(), pending_obj);
198
4357
    return;
199
  }
200
201
17642
  if (nread == 0) {
202
13
    if (buf->base != nullptr)
203
13
      free(buf->base);
204
    return;
205
  }
206
207
17629
  char* base = static_cast<char*>(realloc(buf->base, nread));
208
17629
  CHECK_LE(static_cast<size_t>(nread), buf->len);
209
210
17629
  if (pending == UV_TCP) {
211
124
    pending_obj = AcceptHandle<TCPWrap, uv_tcp_t>(env, wrap);
212
17505
  } else if (pending == UV_NAMED_PIPE) {
213
1
    pending_obj = AcceptHandle<PipeWrap, uv_pipe_t>(env, wrap);
214
17504
  } else if (pending == UV_UDP) {
215
14
    pending_obj = AcceptHandle<UDPWrap, uv_udp_t>(env, wrap);
216
  } else {
217
17490
    CHECK_EQ(pending, UV_UNKNOWN_HANDLE);
218
  }
219
220
35258
  Local<Object> obj = Buffer::New(env, base, nread).ToLocalChecked();
221
17629
  wrap->EmitData(nread, obj, pending_obj);
222
}
223
224
225
25145
void StreamWrap::OnReadCommon(uv_stream_t* handle,
226
                              ssize_t nread,
227
                              const uv_buf_t* buf,
228
                              uv_handle_type pending) {
229
25145
  StreamWrap* wrap = static_cast<StreamWrap*>(handle->data);
230
50230
  HandleScope scope(wrap->env()->isolate());
231
100520
  Context::Scope context_scope(wrap->env()->context());
232
233
  // We should not be getting this callback if someone as already called
234
  // uv_close() on the handle.
235
50290
  CHECK_EQ(wrap->persistent().IsEmpty(), false);
236
237
  if (nread > 0) {
238
    if (wrap->is_tcp()) {
239
      NODE_COUNT_NET_BYTES_RECV(nread);
240
    } else if (wrap->is_named_pipe()) {
241
      NODE_COUNT_PIPE_BYTES_RECV(nread);
242
    }
243
  }
244
245
50230
  static_cast<StreamBase*>(wrap)->OnRead(nread, buf, pending);
246
25085
}
247
248
249
25145
void StreamWrap::OnRead(uv_stream_t* handle,
250
                        ssize_t nread,
251
                        const uv_buf_t* buf) {
252
25145
  StreamWrap* wrap = static_cast<StreamWrap*>(handle->data);
253
25145
  uv_handle_type type = UV_UNKNOWN_HANDLE;
254
255

26201
  if (wrap->is_named_pipe_ipc() &&
256
1056
      uv_pipe_pending_count(reinterpret_cast<uv_pipe_t*>(handle)) > 0) {
257
139
    type = uv_pipe_pending_type(reinterpret_cast<uv_pipe_t*>(handle));
258
  }
259
260
25145
  OnReadCommon(handle, nread, buf, type);
261
25085
}
262
263
264
8
void StreamWrap::SetBlocking(const FunctionCallbackInfo<Value>& args) {
265
  StreamWrap* wrap;
266
8
  ASSIGN_OR_RETURN_UNWRAP(&wrap, args.Holder());
267
268
8
  CHECK_GT(args.Length(), 0);
269
8
  if (!wrap->IsAlive())
270
    return args.GetReturnValue().Set(UV_EINVAL);
271
272
8
  bool enable = args[0]->IsTrue();
273
24
  args.GetReturnValue().Set(uv_stream_set_blocking(wrap->stream(), enable));
274
}
275
276
277
3057
int StreamWrap::DoShutdown(ShutdownWrap* req_wrap) {
278
  int err;
279
3057
  err = uv_shutdown(&req_wrap->req_, stream(), AfterShutdown);
280
6114
  req_wrap->Dispatched();
281
3057
  return err;
282
}
283
284
285
3051
void StreamWrap::AfterShutdown(uv_shutdown_t* req, int status) {
286
3051
  ShutdownWrap* req_wrap = ContainerOf(&ShutdownWrap::req_, req);
287
6102
  HandleScope scope(req_wrap->env()->isolate());
288
12204
  Context::Scope context_scope(req_wrap->env()->context());
289
3051
  req_wrap->Done(status);
290
3051
}
291
292
293
// NOTE: Call to this function could change both `buf`'s and `count`'s
294
// values, shifting their base and decrementing their length. This is
295
// required in order to skip the data that was successfully written via
296
// uv_try_write().
297
14361
int StreamWrap::DoTryWrite(uv_buf_t** bufs, size_t* count) {
298
  int err;
299
  size_t written;
300
14361
  uv_buf_t* vbufs = *bufs;
301
14361
  size_t vcount = *count;
302
303
14361
  err = uv_try_write(stream(), vbufs, vcount);
304
14361
  if (err == UV_ENOSYS || err == UV_EAGAIN)
305
    return 0;
306
14354
  if (err < 0)
307
    return err;
308
309
  // Slice off the buffers: skip all written buffers and slice the one that
310
  // was partially written.
311
14354
  written = err;
312
42952
  for (; vcount > 0; vbufs++, vcount--) {
313
    // Slice
314
14354
    if (vbufs[0].len > written) {
315
55
      vbufs[0].base += written;
316
55
      vbufs[0].len -= written;
317
55
      written = 0;
318
55
      break;
319
320
    // Discard
321
    } else {
322
14299
      written -= vbufs[0].len;
323
    }
324
  }
325
326
14354
  *bufs = vbufs;
327
14354
  *count = vcount;
328
329
14354
  return 0;
330
}
331
332
333
14402
int StreamWrap::DoWrite(WriteWrap* w,
334
                        uv_buf_t* bufs,
335
                        size_t count,
336
                        uv_stream_t* send_handle) {
337
  int r;
338
14402
  if (send_handle == nullptr) {
339
14318
    r = uv_write(&w->req_, stream(), bufs, count, AfterWrite);
340
  } else {
341
84
    r = uv_write2(&w->req_, stream(), bufs, count, send_handle, AfterWrite);
342
  }
343
344
  if (!r) {
345
    size_t bytes = 0;
346
    for (size_t i = 0; i < count; i++)
347
      bytes += bufs[i].len;
348
    if (stream()->type == UV_TCP) {
349
      NODE_COUNT_NET_BYTES_SENT(bytes);
350
    } else if (stream()->type == UV_NAMED_PIPE) {
351
      NODE_COUNT_PIPE_BYTES_SENT(bytes);
352
    }
353
  }
354
355
28804
  w->Dispatched();
356
14402
  UpdateWriteQueueSize();
357
358
14402
  return r;
359
}
360
361
362
14384
void StreamWrap::AfterWrite(uv_write_t* req, int status) {
363
14384
  WriteWrap* req_wrap = ContainerOf(&WriteWrap::req_, req);
364
28764
  HandleScope scope(req_wrap->env()->isolate());
365
57532
  Context::Scope context_scope(req_wrap->env()->context());
366
14384
  req_wrap->Done(status);
367
14380
}
368
369
370
12208
void StreamWrap::OnAfterWriteImpl(WriteWrap* w, void* ctx) {
371
12208
  StreamWrap* wrap = static_cast<StreamWrap*>(ctx);
372
12208
  wrap->UpdateWriteQueueSize();
373
12208
}
374
375
}  // namespace node
376
377
1632
NODE_MODULE_CONTEXT_AWARE_BUILTIN(stream_wrap, node::StreamWrap::Initialize)