GCC Code Coverage Report
Directory: ./ Exec Total Coverage
File: stream_wrap.cc Lines: 168 174 96.6 %
Date: 2022-03-31 04:14:52 Branches: 50 74 67.6 %

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-inl.h"
24
25
#include "env-inl.h"
26
#include "handle_wrap.h"
27
#include "node_buffer.h"
28
#include "node_external_reference.h"
29
#include "pipe_wrap.h"
30
#include "req_wrap-inl.h"
31
#include "tcp_wrap.h"
32
#include "udp_wrap.h"
33
#include "util-inl.h"
34
35
#include <cstring>  // memcpy()
36
#include <climits>  // INT_MAX
37
38
39
namespace node {
40
41
using v8::Context;
42
using v8::DontDelete;
43
using v8::EscapableHandleScope;
44
using v8::FunctionCallbackInfo;
45
using v8::FunctionTemplate;
46
using v8::HandleScope;
47
using v8::Local;
48
using v8::MaybeLocal;
49
using v8::Object;
50
using v8::PropertyAttribute;
51
using v8::ReadOnly;
52
using v8::Signature;
53
using v8::Value;
54
55
119859
void IsConstructCallCallback(const FunctionCallbackInfo<Value>& args) {
56
119859
  CHECK(args.IsConstructCall());
57
119859
  StreamReq::ResetObject(args.This());
58
119859
}
59
60
628
void LibuvStreamWrap::Initialize(Local<Object> target,
61
                                 Local<Value> unused,
62
                                 Local<Context> context,
63
                                 void* priv) {
64
628
  Environment* env = Environment::GetCurrent(context);
65
66
  Local<FunctionTemplate> sw =
67
628
      FunctionTemplate::New(env->isolate(), IsConstructCallCallback);
68
1256
  sw->InstanceTemplate()->SetInternalFieldCount(StreamReq::kInternalFieldCount);
69
70
  // we need to set handle and callback to null,
71
  // so that those fields are created and functions
72
  // do not become megamorphic
73
  // Fields:
74
  // - oncomplete
75
  // - callback
76
  // - handle
77
2512
  sw->InstanceTemplate()->Set(
78
      env->oncomplete_string(),
79
      v8::Null(env->isolate()));
80
2512
  sw->InstanceTemplate()->Set(FIXED_ONE_BYTE_STRING(env->isolate(), "callback"),
81
      v8::Null(env->isolate()));
82
2512
  sw->InstanceTemplate()->Set(FIXED_ONE_BYTE_STRING(env->isolate(), "handle"),
83
      v8::Null(env->isolate()));
84
85
628
  sw->Inherit(AsyncWrap::GetConstructorTemplate(env));
86
87
628
  env->SetConstructorFunction(target, "ShutdownWrap", sw);
88
628
  env->set_shutdown_wrap_template(sw->InstanceTemplate());
89
90
  Local<FunctionTemplate> ww =
91
628
      FunctionTemplate::New(env->isolate(), IsConstructCallCallback);
92
1256
  ww->InstanceTemplate()->SetInternalFieldCount(
93
      StreamReq::kInternalFieldCount);
94
628
  ww->Inherit(AsyncWrap::GetConstructorTemplate(env));
95
628
  env->SetConstructorFunction(target, "WriteWrap", ww);
96
628
  env->set_write_wrap_template(ww->InstanceTemplate());
97
98
1884
  NODE_DEFINE_CONSTANT(target, kReadBytesOrError);
99
1884
  NODE_DEFINE_CONSTANT(target, kArrayBufferOffset);
100
1884
  NODE_DEFINE_CONSTANT(target, kBytesWritten);
101
1884
  NODE_DEFINE_CONSTANT(target, kLastWriteWasAsync);
102
628
  target->Set(context, FIXED_ONE_BYTE_STRING(env->isolate(), "streamBaseState"),
103
1884
              env->stream_base_state().GetJSArray()).Check();
104
628
}
105
106
4972
void LibuvStreamWrap::RegisterExternalReferences(
107
    ExternalReferenceRegistry* registry) {
108
4972
  registry->Register(IsConstructCallCallback);
109
4972
  registry->Register(GetWriteQueueSize);
110
4972
  registry->Register(SetBlocking);
111
  // TODO(joyee): StreamBase::RegisterExternalReferences() is called somewhere
112
  // else but we may want to do it here too and guard it with a static flag.
113
4972
}
114
115
19119
LibuvStreamWrap::LibuvStreamWrap(Environment* env,
116
                                 Local<Object> object,
117
                                 uv_stream_t* stream,
118
19119
                                 AsyncWrap::ProviderType provider)
119
    : HandleWrap(env,
120
                 object,
121
                 reinterpret_cast<uv_handle_t*>(stream),
122
                 provider),
123
      StreamBase(env),
124
19119
      stream_(stream) {
125
19119
  StreamBase::AttachToObject(object);
126
19119
}
127
128
129
15582
Local<FunctionTemplate> LibuvStreamWrap::GetConstructorTemplate(
130
    Environment* env) {
131
15582
  Local<FunctionTemplate> tmpl = env->libuv_stream_wrap_ctor_template();
132
15582
  if (tmpl.IsEmpty()) {
133
5554
    tmpl = env->NewFunctionTemplate(nullptr);
134
5554
    tmpl->SetClassName(
135
        FIXED_ONE_BYTE_STRING(env->isolate(), "LibuvStreamWrap"));
136
5554
    tmpl->Inherit(HandleWrap::GetConstructorTemplate(env));
137
11108
    tmpl->InstanceTemplate()->SetInternalFieldCount(
138
        StreamBase::kInternalFieldCount);
139
    Local<FunctionTemplate> get_write_queue_size =
140
        FunctionTemplate::New(env->isolate(),
141
                              GetWriteQueueSize,
142
                              Local<Value>(),
143
5554
                              Signature::New(env->isolate(), tmpl));
144
22216
    tmpl->PrototypeTemplate()->SetAccessorProperty(
145
        env->write_queue_size_string(),
146
        get_write_queue_size,
147
        Local<FunctionTemplate>(),
148
        static_cast<PropertyAttribute>(ReadOnly | DontDelete));
149
5554
    env->SetProtoMethod(tmpl, "setBlocking", SetBlocking);
150
5554
    StreamBase::AddMethods(env, tmpl);
151
5554
    env->set_libuv_stream_wrap_ctor_template(tmpl);
152
  }
153
15582
  return tmpl;
154
}
155
156
157
3834
LibuvStreamWrap* LibuvStreamWrap::From(Environment* env, Local<Object> object) {
158
3834
  Local<FunctionTemplate> sw = env->libuv_stream_wrap_ctor_template();
159

7668
  CHECK(!sw.IsEmpty() && sw->HasInstance(object));
160
3834
  return Unwrap<LibuvStreamWrap>(object);
161
}
162
163
164
118
int LibuvStreamWrap::GetFD() {
165
#ifdef _WIN32
166
  return fd_;
167
#else
168
118
  int fd = -1;
169
118
  if (stream() != nullptr)
170
118
    uv_fileno(reinterpret_cast<uv_handle_t*>(stream()), &fd);
171
118
  return fd;
172
#endif
173
}
174
175
176
120896
bool LibuvStreamWrap::IsAlive() {
177
120896
  return HandleWrap::IsAlive(this);
178
}
179
180
181
3
bool LibuvStreamWrap::IsClosing() {
182
3
  return uv_is_closing(reinterpret_cast<uv_handle_t*>(stream()));
183
}
184
185
186
167788
AsyncWrap* LibuvStreamWrap::GetAsyncWrap() {
187
167788
  return static_cast<AsyncWrap*>(this);
188
}
189
190
191
25796
bool LibuvStreamWrap::IsIPCPipe() {
192
25796
  return is_named_pipe_ipc();
193
}
194
195
196
22451
int LibuvStreamWrap::ReadStart() {
197
44902
  return uv_read_start(stream(), [](uv_handle_t* handle,
198
                                    size_t suggested_size,
199
71649
                                    uv_buf_t* buf) {
200
71649
    static_cast<LibuvStreamWrap*>(handle->data)->OnUvAlloc(suggested_size, buf);
201
143507
  }, [](uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf) {
202
71858
    static_cast<LibuvStreamWrap*>(stream->data)->OnUvRead(nread, buf);
203
22451
  });
204
}
205
206
207
9185
int LibuvStreamWrap::ReadStop() {
208
9185
  return uv_read_stop(stream());
209
}
210
211
212
71649
void LibuvStreamWrap::OnUvAlloc(size_t suggested_size, uv_buf_t* buf) {
213
143298
  HandleScope scope(env()->isolate());
214
71649
  Context::Scope context_scope(env()->context());
215
216
71649
  *buf = EmitAlloc(suggested_size);
217
71649
}
218
219
template <class WrapType>
220
316
static MaybeLocal<Object> AcceptHandle(Environment* env,
221
                                       LibuvStreamWrap* parent) {
222
  static_assert(std::is_base_of<LibuvStreamWrap, WrapType>::value ||
223
                std::is_base_of<UDPWrap, WrapType>::value,
224
                "Can only accept stream handles");
225
226
316
  EscapableHandleScope scope(env->isolate());
227
  Local<Object> wrap_obj;
228
229
632
  if (!WrapType::Instantiate(env, parent, WrapType::SOCKET).ToLocal(&wrap_obj))
230
    return Local<Object>();
231
232
316
  HandleWrap* wrap = Unwrap<HandleWrap>(wrap_obj);
233
316
  CHECK_NOT_NULL(wrap);
234
316
  uv_stream_t* stream = reinterpret_cast<uv_stream_t*>(wrap->GetHandle());
235
316
  CHECK_NOT_NULL(stream);
236
237
316
  if (uv_accept(parent->stream(), stream))
238
    ABORT();
239
240
316
  return scope.Escape(wrap_obj);
241
}
242
243
244
71858
void LibuvStreamWrap::OnUvRead(ssize_t nread, const uv_buf_t* buf) {
245
71858
  HandleScope scope(env()->isolate());
246
71858
  Context::Scope context_scope(env()->context());
247
71858
  uv_handle_type type = UV_UNKNOWN_HANDLE;
248
249

74872
  if (is_named_pipe_ipc() &&
250
3014
      uv_pipe_pending_count(reinterpret_cast<uv_pipe_t*>(stream())) > 0) {
251
158
    type = uv_pipe_pending_type(reinterpret_cast<uv_pipe_t*>(stream()));
252
  }
253
254
  // We should not be getting this callback if someone has already called
255
  // uv_close() on the handle.
256
71858
  CHECK_EQ(persistent().IsEmpty(), false);
257
258
71858
  if (nread > 0) {
259
    MaybeLocal<Object> pending_obj;
260
261
64650
    if (type == UV_TCP) {
262
127
      pending_obj = AcceptHandle<TCPWrap>(env(), this);
263
64523
    } else if (type == UV_NAMED_PIPE) {
264
1
      pending_obj = AcceptHandle<PipeWrap>(env(), this);
265
64522
    } else if (type == UV_UDP) {
266
30
      pending_obj = AcceptHandle<UDPWrap>(env(), this);
267
    } else {
268
64492
      CHECK_EQ(type, UV_UNKNOWN_HANDLE);
269
    }
270
271
    Local<Object> local_pending_obj;
272
64808
    if (pending_obj.ToLocal(&local_pending_obj) &&
273
158
          object()->Set(env()->context(),
274
                        env()->pending_handle_string(),
275

65124
                        local_pending_obj).IsNothing()) {
276
      return;
277
    }
278
  }
279
280
71858
  EmitRead(nread, *buf);
281
}
282
283
284
5406
void LibuvStreamWrap::GetWriteQueueSize(
285
    const FunctionCallbackInfo<Value>& info) {
286
  LibuvStreamWrap* wrap;
287
5406
  ASSIGN_OR_RETURN_UNWRAP(&wrap, info.This());
288
289
5406
  if (wrap->stream() == nullptr) {
290
    info.GetReturnValue().Set(0);
291
    return;
292
  }
293
294
5406
  uint32_t write_queue_size = wrap->stream()->write_queue_size;
295
10812
  info.GetReturnValue().Set(write_queue_size);
296
}
297
298
299
32
void LibuvStreamWrap::SetBlocking(const FunctionCallbackInfo<Value>& args) {
300
  LibuvStreamWrap* wrap;
301
32
  ASSIGN_OR_RETURN_UNWRAP(&wrap, args.Holder());
302
303
32
  CHECK_GT(args.Length(), 0);
304
32
  if (!wrap->IsAlive())
305
    return args.GetReturnValue().Set(UV_EINVAL);
306
307
32
  bool enable = args[0]->IsTrue();
308
64
  args.GetReturnValue().Set(uv_stream_set_blocking(wrap->stream(), enable));
309
}
310
311
typedef SimpleShutdownWrap<ReqWrap<uv_shutdown_t>> LibuvShutdownWrap;
312
typedef SimpleWriteWrap<ReqWrap<uv_write_t>> LibuvWriteWrap;
313
314
9902
ShutdownWrap* LibuvStreamWrap::CreateShutdownWrap(Local<Object> object) {
315
9902
  return new LibuvShutdownWrap(this, object);
316
}
317
318
244
WriteWrap* LibuvStreamWrap::CreateWriteWrap(Local<Object> object) {
319
244
  return new LibuvWriteWrap(this, object);
320
}
321
322
323
9902
int LibuvStreamWrap::DoShutdown(ShutdownWrap* req_wrap_) {
324
9902
  LibuvShutdownWrap* req_wrap = static_cast<LibuvShutdownWrap*>(req_wrap_);
325
9902
  return req_wrap->Dispatch(uv_shutdown, stream(), AfterUvShutdown);
326
}
327
328
329
9898
void LibuvStreamWrap::AfterUvShutdown(uv_shutdown_t* req, int status) {
330
9898
  LibuvShutdownWrap* req_wrap = static_cast<LibuvShutdownWrap*>(
331
9898
      LibuvShutdownWrap::from_req(req));
332
9898
  CHECK_NOT_NULL(req_wrap);
333
19796
  HandleScope scope(req_wrap->env()->isolate());
334
9898
  Context::Scope context_scope(req_wrap->env()->context());
335
9898
  req_wrap->Done(status);
336
9898
}
337
338
339
// NOTE: Call to this function could change both `buf`'s and `count`'s
340
// values, shifting their base and decrementing their length. This is
341
// required in order to skip the data that was successfully written via
342
// uv_try_write().
343
118353
int LibuvStreamWrap::DoTryWrite(uv_buf_t** bufs, size_t* count) {
344
  int err;
345
  size_t written;
346
118353
  uv_buf_t* vbufs = *bufs;
347
118353
  size_t vcount = *count;
348
349
118353
  err = uv_try_write(stream(), vbufs, vcount);
350

118353
  if (err == UV_ENOSYS || err == UV_EAGAIN)
351
5
    return 0;
352
118348
  if (err < 0)
353
3519
    return err;
354
355
  // Slice off the buffers: skip all written buffers and slice the one that
356
  // was partially written.
357
114829
  written = err;
358
303918
  for (; vcount > 0; vbufs++, vcount--) {
359
    // Slice
360
189224
    if (vbufs[0].len > written) {
361
135
      vbufs[0].base += written;
362
135
      vbufs[0].len -= written;
363
135
      written = 0;
364
135
      break;
365
366
    // Discard
367
    } else {
368
189089
      written -= vbufs[0].len;
369
    }
370
  }
371
372
114829
  *bufs = vbufs;
373
114829
  *count = vcount;
374
375
114829
  return 0;
376
}
377
378
379
244
int LibuvStreamWrap::DoWrite(WriteWrap* req_wrap,
380
                             uv_buf_t* bufs,
381
                             size_t count,
382
                             uv_stream_t* send_handle) {
383
244
  LibuvWriteWrap* w = static_cast<LibuvWriteWrap*>(req_wrap);
384
244
  return w->Dispatch(uv_write2,
385
                     stream(),
386
                     bufs,
387
                     count,
388
                     send_handle,
389
244
                     AfterUvWrite);
390
}
391
392
393
394
243
void LibuvStreamWrap::AfterUvWrite(uv_write_t* req, int status) {
395
243
  LibuvWriteWrap* req_wrap = static_cast<LibuvWriteWrap*>(
396
243
      LibuvWriteWrap::from_req(req));
397
243
  CHECK_NOT_NULL(req_wrap);
398
485
  HandleScope scope(req_wrap->env()->isolate());
399
243
  Context::Scope context_scope(req_wrap->env()->context());
400
243
  req_wrap->Done(status);
401
242
}
402
403
}  // namespace node
404
405
5033
NODE_MODULE_CONTEXT_AWARE_INTERNAL(stream_wrap,
406
                                   node::LibuvStreamWrap::Initialize)
407
4972
NODE_MODULE_EXTERNAL_REFERENCE(
408
    stream_wrap, node::LibuvStreamWrap::RegisterExternalReferences)