GCC Code Coverage Report
Directory: ./ Exec Total Coverage
File: stream_wrap.cc Lines: 180 186 96.8 %
Date: 2022-08-29 04:21:03 Branches: 51 76 67.1 %

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

8492
  CHECK(!sw.IsEmpty() && sw->HasInstance(object));
167
4246
  return Unwrap<LibuvStreamWrap>(object);
168
}
169
170
171
120
int LibuvStreamWrap::GetFD() {
172
#ifdef _WIN32
173
  return fd_;
174
#else
175
120
  int fd = -1;
176
120
  if (stream() != nullptr)
177
120
    uv_fileno(reinterpret_cast<uv_handle_t*>(stream()), &fd);
178
120
  return fd;
179
#endif
180
}
181
182
183
118770
bool LibuvStreamWrap::IsAlive() {
184
118770
  return HandleWrap::IsAlive(this);
185
}
186
187
188
3
bool LibuvStreamWrap::IsClosing() {
189
3
  return uv_is_closing(reinterpret_cast<uv_handle_t*>(stream()));
190
}
191
192
193
162101
AsyncWrap* LibuvStreamWrap::GetAsyncWrap() {
194
162101
  return static_cast<AsyncWrap*>(this);
195
}
196
197
198
26390
bool LibuvStreamWrap::IsIPCPipe() {
199
26390
  return is_named_pipe_ipc();
200
}
201
202
21936
int LibuvStreamWrap::ReadStart() {
203
43872
  return uv_read_start(
204
      stream(),
205
72794
      [](uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
206
72794
        static_cast<LibuvStreamWrap*>(handle->data)
207
72794
            ->OnUvAlloc(suggested_size, buf);
208
72794
      },
209
73020
      [](uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf) {
210
73020
        LibuvStreamWrap* wrap = static_cast<LibuvStreamWrap*>(stream->data);
211
145828
        TryCatchScope try_catch(wrap->env());
212
73020
        try_catch.SetVerbose(true);
213
73020
        wrap->OnUvRead(nread, buf);
214
21936
      });
215
}
216
217
218
9220
int LibuvStreamWrap::ReadStop() {
219
9220
  return uv_read_stop(stream());
220
}
221
222
223
72794
void LibuvStreamWrap::OnUvAlloc(size_t suggested_size, uv_buf_t* buf) {
224
145588
  HandleScope scope(env()->isolate());
225
72794
  Context::Scope context_scope(env()->context());
226
227
72794
  *buf = EmitAlloc(suggested_size);
228
72794
}
229
230
template <class WrapType>
231
408
static MaybeLocal<Object> AcceptHandle(Environment* env,
232
                                       LibuvStreamWrap* parent) {
233
  static_assert(std::is_base_of<LibuvStreamWrap, WrapType>::value ||
234
                std::is_base_of<UDPWrap, WrapType>::value,
235
                "Can only accept stream handles");
236
237
408
  EscapableHandleScope scope(env->isolate());
238
  Local<Object> wrap_obj;
239
240
816
  if (!WrapType::Instantiate(env, parent, WrapType::SOCKET).ToLocal(&wrap_obj))
241
    return Local<Object>();
242
243
408
  HandleWrap* wrap = Unwrap<HandleWrap>(wrap_obj);
244
408
  CHECK_NOT_NULL(wrap);
245
408
  uv_stream_t* stream = reinterpret_cast<uv_stream_t*>(wrap->GetHandle());
246
408
  CHECK_NOT_NULL(stream);
247
248
408
  if (uv_accept(parent->stream(), stream))
249
    ABORT();
250
251
408
  return scope.Escape(wrap_obj);
252
}
253
254
73020
Maybe<void> LibuvStreamWrap::OnUvRead(ssize_t nread, const uv_buf_t* buf) {
255
145828
  HandleScope scope(env()->isolate());
256
73020
  Context::Scope context_scope(env()->context());
257
73020
  uv_handle_type type = UV_UNKNOWN_HANDLE;
258
259

77337
  if (is_named_pipe_ipc() &&
260
4317
      uv_pipe_pending_count(reinterpret_cast<uv_pipe_t*>(stream())) > 0) {
261
204
    type = uv_pipe_pending_type(reinterpret_cast<uv_pipe_t*>(stream()));
262
  }
263
264
  // We should not be getting this callback if someone has already called
265
  // uv_close() on the handle.
266
73020
  CHECK_EQ(persistent().IsEmpty(), false);
267
268
73020
  if (nread > 0) {
269
    MaybeLocal<Object> pending_obj;
270
271
66993
    if (type == UV_TCP) {
272
130
      pending_obj = AcceptHandle<TCPWrap>(env(), this);
273
66863
    } else if (type == UV_NAMED_PIPE) {
274
41
      pending_obj = AcceptHandle<PipeWrap>(env(), this);
275
66822
    } else if (type == UV_UDP) {
276
33
      pending_obj = AcceptHandle<UDPWrap>(env(), this);
277
    } else {
278
66789
      CHECK_EQ(type, UV_UNKNOWN_HANDLE);
279
    }
280
281
    Local<Object> local_pending_obj;
282
67197
    if (type != UV_UNKNOWN_HANDLE &&
283
204
        (!pending_obj.ToLocal(&local_pending_obj) ||
284
204
         object()
285
66993
             ->Set(env()->context(),
286
                   env()->pending_handle_string(),
287
67401
                   local_pending_obj)
288
204
             .IsNothing())) {
289
      return Nothing<void>();
290
    }
291
  }
292
293
73020
  EmitRead(nread, *buf);
294
72808
  return JustVoid();
295
}
296
297
5585
void LibuvStreamWrap::GetWriteQueueSize(
298
    const FunctionCallbackInfo<Value>& info) {
299
  LibuvStreamWrap* wrap;
300
5585
  ASSIGN_OR_RETURN_UNWRAP(&wrap, info.This());
301
302
5585
  if (wrap->stream() == nullptr) {
303
    info.GetReturnValue().Set(0);
304
    return;
305
  }
306
307
5585
  uint32_t write_queue_size = wrap->stream()->write_queue_size;
308
11170
  info.GetReturnValue().Set(write_queue_size);
309
}
310
311
312
32
void LibuvStreamWrap::SetBlocking(const FunctionCallbackInfo<Value>& args) {
313
  LibuvStreamWrap* wrap;
314
32
  ASSIGN_OR_RETURN_UNWRAP(&wrap, args.Holder());
315
316
32
  CHECK_GT(args.Length(), 0);
317
32
  if (!wrap->IsAlive())
318
    return args.GetReturnValue().Set(UV_EINVAL);
319
320
32
  bool enable = args[0]->IsTrue();
321
64
  args.GetReturnValue().Set(uv_stream_set_blocking(wrap->stream(), enable));
322
}
323
324
typedef SimpleShutdownWrap<ReqWrap<uv_shutdown_t>> LibuvShutdownWrap;
325
typedef SimpleWriteWrap<ReqWrap<uv_write_t>> LibuvWriteWrap;
326
327
7512
ShutdownWrap* LibuvStreamWrap::CreateShutdownWrap(Local<Object> object) {
328
7512
  return new LibuvShutdownWrap(this, object);
329
}
330
331
325
WriteWrap* LibuvStreamWrap::CreateWriteWrap(Local<Object> object) {
332
325
  return new LibuvWriteWrap(this, object);
333
}
334
335
336
7512
int LibuvStreamWrap::DoShutdown(ShutdownWrap* req_wrap_) {
337
7512
  LibuvShutdownWrap* req_wrap = static_cast<LibuvShutdownWrap*>(req_wrap_);
338
7512
  return req_wrap->Dispatch(uv_shutdown, stream(), AfterUvShutdown);
339
}
340
341
342
7509
void LibuvStreamWrap::AfterUvShutdown(uv_shutdown_t* req, int status) {
343
7509
  LibuvShutdownWrap* req_wrap = static_cast<LibuvShutdownWrap*>(
344
7509
      LibuvShutdownWrap::from_req(req));
345
7509
  CHECK_NOT_NULL(req_wrap);
346
15018
  HandleScope scope(req_wrap->env()->isolate());
347
7509
  Context::Scope context_scope(req_wrap->env()->context());
348
7509
  req_wrap->Done(status);
349
7509
}
350
351
352
// NOTE: Call to this function could change both `buf`'s and `count`'s
353
// values, shifting their base and decrementing their length. This is
354
// required in order to skip the data that was successfully written via
355
// uv_try_write().
356
118768
int LibuvStreamWrap::DoTryWrite(uv_buf_t** bufs, size_t* count) {
357
  int err;
358
  size_t written;
359
118768
  uv_buf_t* vbufs = *bufs;
360
118768
  size_t vcount = *count;
361
362
118768
  err = uv_try_write(stream(), vbufs, vcount);
363

118768
  if (err == UV_ENOSYS || err == UV_EAGAIN)
364
38
    return 0;
365
118730
  if (err < 0)
366
2346
    return err;
367
368
  // Slice off the buffers: skip all written buffers and slice the one that
369
  // was partially written.
370
116384
  written = err;
371
310552
  for (; vcount > 0; vbufs++, vcount--) {
372
    // Slice
373
194305
    if (vbufs[0].len > written) {
374
137
      vbufs[0].base += written;
375
137
      vbufs[0].len -= written;
376
137
      written = 0;
377
137
      break;
378
379
    // Discard
380
    } else {
381
194168
      written -= vbufs[0].len;
382
    }
383
  }
384
385
116384
  *bufs = vbufs;
386
116384
  *count = vcount;
387
388
116384
  return 0;
389
}
390
391
392
325
int LibuvStreamWrap::DoWrite(WriteWrap* req_wrap,
393
                             uv_buf_t* bufs,
394
                             size_t count,
395
                             uv_stream_t* send_handle) {
396
325
  LibuvWriteWrap* w = static_cast<LibuvWriteWrap*>(req_wrap);
397
325
  return w->Dispatch(uv_write2,
398
                     stream(),
399
                     bufs,
400
                     count,
401
                     send_handle,
402
325
                     AfterUvWrite);
403
}
404
405
406
407
323
void LibuvStreamWrap::AfterUvWrite(uv_write_t* req, int status) {
408
323
  LibuvWriteWrap* req_wrap = static_cast<LibuvWriteWrap*>(
409
323
      LibuvWriteWrap::from_req(req));
410
323
  CHECK_NOT_NULL(req_wrap);
411
645
  HandleScope scope(req_wrap->env()->isolate());
412
323
  Context::Scope context_scope(req_wrap->env()->context());
413
323
  req_wrap->Done(status);
414
322
}
415
416
}  // namespace node
417
418
5487
NODE_MODULE_CONTEXT_AWARE_INTERNAL(stream_wrap,
419
                                   node::LibuvStreamWrap::Initialize)
420
5415
NODE_MODULE_EXTERNAL_REFERENCE(
421
    stream_wrap, node::LibuvStreamWrap::RegisterExternalReferences)