GCC Code Coverage Report
Directory: ../src/ Exec Total Coverage
File: /home/node-core-coverage/node-core-coverage/workdir/node/src/stream_wrap.cc Lines: 150 162 92.6 %
Date: 2016-07-18 Branches: 45 72 62.5 %

Line 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::True;
34
using v8::Value;
35
36
37
1506
void StreamWrap::Initialize(Local<Object> target,
38
                            Local<Value> unused,
39
                            Local<Context> context) {
40
1506
  Environment* env = Environment::GetCurrent(context);
41
42
  Local<FunctionTemplate> sw =
43
3012
      FunctionTemplate::New(env->isolate(), ShutdownWrap::NewShutdownWrap);
44
3012
  sw->InstanceTemplate()->SetInternalFieldCount(1);
45
1506
  sw->SetClassName(FIXED_ONE_BYTE_STRING(env->isolate(), "ShutdownWrap"));
46
6024
  target->Set(FIXED_ONE_BYTE_STRING(env->isolate(), "ShutdownWrap"),
47
1506
              sw->GetFunction());
48
49
  Local<FunctionTemplate> ww =
50
3012
      FunctionTemplate::New(env->isolate(), WriteWrap::NewWriteWrap);
51
3012
  ww->InstanceTemplate()->SetInternalFieldCount(1);
52
1506
  ww->SetClassName(FIXED_ONE_BYTE_STRING(env->isolate(), "WriteWrap"));
53
6024
  target->Set(FIXED_ONE_BYTE_STRING(env->isolate(), "WriteWrap"),
54
1506
              ww->GetFunction());
55
1506
  env->set_write_wrap_constructor_function(ww->GetFunction());
56
1506
}
57
58
59
7636
StreamWrap::StreamWrap(Environment* env,
60
                       Local<Object> object,
61
                       uv_stream_t* stream,
62
                       AsyncWrap::ProviderType provider,
63
                       AsyncWrap* parent)
64
    : HandleWrap(env,
65
                 object,
66
                 reinterpret_cast<uv_handle_t*>(stream),
67
                 provider,
68
                 parent),
69
      StreamBase(env),
70
15272
      stream_(stream) {
71
15272
  set_after_write_cb({ OnAfterWriteImpl, this });
72
15272
  set_alloc_cb({ OnAllocImpl, this });
73
7636
  set_read_cb({ OnReadImpl, this });
74
7636
}
75
76
77
4528
void StreamWrap::AddMethods(Environment* env,
78
                            v8::Local<v8::FunctionTemplate> target,
79
                            int flags) {
80
4528
  env->SetProtoMethod(target, "setBlocking", SetBlocking);
81
4528
  StreamBase::AddMethods<StreamWrap>(env, target, flags);
82
4528
}
83
84
85
15
int StreamWrap::GetFD() {
86
15
  int fd = -1;
87
#if !defined(_WIN32)
88
15
  if (stream() != nullptr)
89
15
    uv_fileno(reinterpret_cast<uv_handle_t*>(stream()), &fd);
90
#endif
91
15
  return fd;
92
}
93
94
95
36906
bool StreamWrap::IsAlive() {
96
73812
  return HandleWrap::IsAlive(this);
97
}
98
99
100
2
bool StreamWrap::IsClosing() {
101
2
  return uv_is_closing(reinterpret_cast<uv_handle_t*>(stream()));
102
}
103
104
105
void* StreamWrap::Cast() {
106
  return reinterpret_cast<void*>(this);
107
}
108
109
110
31087
AsyncWrap* StreamWrap::GetAsyncWrap() {
111
31087
  return static_cast<AsyncWrap*>(this);
112
}
113
114
115
14919
bool StreamWrap::IsIPCPipe() {
116
14919
  return is_named_pipe_ipc();
117
}
118
119
120
34088
void StreamWrap::UpdateWriteQueueSize() {
121
68176
  HandleScope scope(env()->isolate());
122
  Local<Integer> write_queue_size =
123
34088
      Integer::NewFromUnsigned(env()->isolate(), stream()->write_queue_size);
124
136352
  object()->Set(env()->write_queue_size_string(), write_queue_size);
125
34088
}
126
127
128
6638
int StreamWrap::ReadStart() {
129
6638
  return uv_read_start(stream(), OnAlloc, OnRead);
130
}
131
132
133
227
int StreamWrap::ReadStop() {
134
227
  return uv_read_stop(stream());
135
}
136
137
138
21316
void StreamWrap::OnAlloc(uv_handle_t* handle,
139
                         size_t suggested_size,
140
                         uv_buf_t* buf) {
141
21316
  StreamWrap* wrap = static_cast<StreamWrap*>(handle->data);
142
42632
  HandleScope scope(wrap->env()->isolate());
143
85264
  Context::Scope context_scope(wrap->env()->context());
144
145
21316
  CHECK_EQ(wrap->stream(), reinterpret_cast<uv_stream_t*>(handle));
146
147
63948
  return static_cast<StreamBase*>(wrap)->OnAlloc(suggested_size, buf);
148
}
149
150
151
15930
void StreamWrap::OnAllocImpl(size_t size, uv_buf_t* buf, void* ctx) {
152
15930
  buf->base = static_cast<char*>(malloc(size));
153
15930
  buf->len = size;
154
155
15930
  if (buf->base == nullptr && size > 0) {
156
    FatalError(
157
        "node::StreamWrap::DoAlloc(size_t, uv_buf_t*, void*)",
158
        "Out Of Memory");
159
  }
160
15930
}
161
162
163
template <class WrapType, class UVType>
164
61
static Local<Object> AcceptHandle(Environment* env, StreamWrap* parent) {
165
122
  EscapableHandleScope scope(env->isolate());
166
61
  Local<Object> wrap_obj;
167
  UVType* handle;
168
169
61
  wrap_obj = WrapType::Instantiate(env, parent);
170
61
  if (wrap_obj.IsEmpty())
171
    return Local<Object>();
172
173
  WrapType* wrap;
174
61
  ASSIGN_OR_RETURN_UNWRAP(&wrap, wrap_obj, Local<Object>());
175
61
  handle = wrap->UVHandle();
176
177
61
  if (uv_accept(parent->stream(), reinterpret_cast<uv_stream_t*>(handle)))
178
    ABORT();
179
180
  return scope.Escape(wrap_obj);
181
}
182
183
184
16157
void StreamWrap::OnReadImpl(ssize_t nread,
185
                            const uv_buf_t* buf,
186
                            uv_handle_type pending,
187
                            void* ctx) {
188
16157
  StreamWrap* wrap = static_cast<StreamWrap*>(ctx);
189
16157
  Environment* env = wrap->env();
190
30327
  HandleScope handle_scope(env->isolate());
191
46484
  Context::Scope context_scope(env->context());
192
193
16157
  Local<Object> pending_obj;
194
195
16157
  if (nread < 0)  {
196
1936
    if (buf->base != nullptr)
197
1709
      free(buf->base);
198
1936
    wrap->EmitData(nread, Local<Object>(), pending_obj);
199
3853
    return;
200
  }
201
202
14221
  if (nread == 0) {
203
13
    if (buf->base != nullptr)
204
13
      free(buf->base);
205
    return;
206
  }
207
208
14208
  char* base = static_cast<char*>(realloc(buf->base, nread));
209
14208
  CHECK_LE(static_cast<size_t>(nread), buf->len);
210
211
14208
  if (pending == UV_TCP) {
212
46
    pending_obj = AcceptHandle<TCPWrap, uv_tcp_t>(env, wrap);
213
14162
  } else if (pending == UV_NAMED_PIPE) {
214
1
    pending_obj = AcceptHandle<PipeWrap, uv_pipe_t>(env, wrap);
215
14161
  } else if (pending == UV_UDP) {
216
14
    pending_obj = AcceptHandle<UDPWrap, uv_udp_t>(env, wrap);
217
  } else {
218
14147
    CHECK_EQ(pending, UV_UNKNOWN_HANDLE);
219
  }
220
221
28416
  Local<Object> obj = Buffer::New(env, base, nread).ToLocalChecked();
222
14208
  wrap->EmitData(nread, obj, pending_obj);
223
}
224
225
226
21455
void StreamWrap::OnReadCommon(uv_stream_t* handle,
227
                              ssize_t nread,
228
                              const uv_buf_t* buf,
229
                              uv_handle_type pending) {
230
21455
  StreamWrap* wrap = static_cast<StreamWrap*>(handle->data);
231
42851
  HandleScope scope(wrap->env()->isolate());
232
85761
  Context::Scope context_scope(wrap->env()->context());
233
234
  // We should not be getting this callback if someone as already called
235
  // uv_close() on the handle.
236
42910
  CHECK_EQ(wrap->persistent().IsEmpty(), false);
237
238
  if (nread > 0) {
239
    if (wrap->is_tcp()) {
240
      NODE_COUNT_NET_BYTES_RECV(nread);
241
    } else if (wrap->is_named_pipe()) {
242
      NODE_COUNT_PIPE_BYTES_RECV(nread);
243
    }
244
  }
245
246
42851
  static_cast<StreamBase*>(wrap)->OnRead(nread, buf, pending);
247
21396
}
248
249
250
21455
void StreamWrap::OnRead(uv_stream_t* handle,
251
                        ssize_t nread,
252
                        const uv_buf_t* buf) {
253
21455
  StreamWrap* wrap = static_cast<StreamWrap*>(handle->data);
254
21455
  uv_handle_type type = UV_UNKNOWN_HANDLE;
255
256
22330
  if (wrap->is_named_pipe_ipc() &&
257
875
      uv_pipe_pending_count(reinterpret_cast<uv_pipe_t*>(handle)) > 0) {
258
61
    type = uv_pipe_pending_type(reinterpret_cast<uv_pipe_t*>(handle));
259
  }
260
261
21455
  OnReadCommon(handle, nread, buf, type);
262
21396
}
263
264
265
void StreamWrap::SetBlocking(const FunctionCallbackInfo<Value>& args) {
266
  StreamWrap* wrap;
267
  ASSIGN_OR_RETURN_UNWRAP(&wrap, args.Holder());
268
269
  CHECK_GT(args.Length(), 0);
270
  if (!wrap->IsAlive())
271
    return args.GetReturnValue().Set(UV_EINVAL);
272
273
  bool enable = args[0]->IsTrue();
274
  args.GetReturnValue().Set(uv_stream_set_blocking(wrap->stream(), enable));
275
}
276
277
278
3053
int StreamWrap::DoShutdown(ShutdownWrap* req_wrap) {
279
  int err;
280
3053
  err = uv_shutdown(&req_wrap->req_, stream(), AfterShutdown);
281
6106
  req_wrap->Dispatched();
282
3053
  return err;
283
}
284
285
286
3047
void StreamWrap::AfterShutdown(uv_shutdown_t* req, int status) {
287
3047
  ShutdownWrap* req_wrap = ContainerOf(&ShutdownWrap::req_, req);
288
6094
  HandleScope scope(req_wrap->env()->isolate());
289
12188
  Context::Scope context_scope(req_wrap->env()->context());
290
3047
  req_wrap->Done(status);
291
3047
}
292
293
294
// NOTE: Call to this function could change both `buf`'s and `count`'s
295
// values, shifting their base and decrementing their length. This is
296
// required in order to skip the data that was successfully written via
297
// uv_try_write().
298
13983
int StreamWrap::DoTryWrite(uv_buf_t** bufs, size_t* count) {
299
  int err;
300
  size_t written;
301
13983
  uv_buf_t* vbufs = *bufs;
302
13983
  size_t vcount = *count;
303
304
13983
  err = uv_try_write(stream(), vbufs, vcount);
305
13983
  if (err == UV_ENOSYS || err == UV_EAGAIN)
306
    return 0;
307
13971
  if (err < 0)
308
    return err;
309
310
  // Slice off the buffers: skip all written buffers and slice the one that
311
  // was partially written.
312
13971
  written = err;
313
41903
  for (; vcount > 0; vbufs++, vcount--) {
314
    // Slice
315
13971
    if (vbufs[0].len > written) {
316
5
      vbufs[0].base += written;
317
5
      vbufs[0].len -= written;
318
5
      written = 0;
319
5
      break;
320
321
    // Discard
322
    } else {
323
13966
      written -= vbufs[0].len;
324
    }
325
  }
326
327
13971
  *bufs = vbufs;
328
13971
  *count = vcount;
329
330
13971
  return 0;
331
}
332
333
334
14290
int StreamWrap::DoWrite(WriteWrap* w,
335
                        uv_buf_t* bufs,
336
                        size_t count,
337
                        uv_stream_t* send_handle) {
338
  int r;
339
14290
  if (send_handle == nullptr) {
340
14206
    r = uv_write(&w->req_, stream(), bufs, count, AfterWrite);
341
  } else {
342
84
    r = uv_write2(&w->req_, stream(), bufs, count, send_handle, AfterWrite);
343
  }
344
345
  if (!r) {
346
    size_t bytes = 0;
347
    for (size_t i = 0; i < count; i++)
348
      bytes += bufs[i].len;
349
    if (stream()->type == UV_TCP) {
350
      NODE_COUNT_NET_BYTES_SENT(bytes);
351
    } else if (stream()->type == UV_NAMED_PIPE) {
352
      NODE_COUNT_PIPE_BYTES_SENT(bytes);
353
    }
354
  }
355
356
28580
  w->Dispatched();
357
14290
  UpdateWriteQueueSize();
358
359
14290
  return r;
360
}
361
362
363
14272
void StreamWrap::AfterWrite(uv_write_t* req, int status) {
364
14272
  WriteWrap* req_wrap = ContainerOf(&WriteWrap::req_, req);
365
28540
  HandleScope scope(req_wrap->env()->isolate());
366
57084
  Context::Scope context_scope(req_wrap->env()->context());
367
14272
  req_wrap->Done(status);
368
14268
}
369
370
371
12162
void StreamWrap::OnAfterWriteImpl(WriteWrap* w, void* ctx) {
372
12162
  StreamWrap* wrap = static_cast<StreamWrap*>(ctx);
373
12162
  wrap->UpdateWriteQueueSize();
374
12162
}
375
376
}  // namespace node
377
378
1565
NODE_MODULE_CONTEXT_AWARE_BUILTIN(stream_wrap, node::StreamWrap::Initialize)