GCC Code Coverage Report
Directory: ../ Exec Total Coverage
File: /home/iojs/build/workspace/node-test-commit-linux-coverage-daily/nodes/benchmark/out/../src/stream_base.h Lines: 24 28 85.7 %
Date: 2020-07-19 22:14:24 Branches: 1 2 50.0 %

Line Branch Exec Source
1
#ifndef SRC_STREAM_BASE_H_
2
#define SRC_STREAM_BASE_H_
3
4
#if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
5
6
#include "env.h"
7
#include "allocated_buffer.h"
8
#include "async_wrap.h"
9
#include "node.h"
10
#include "util.h"
11
12
#include "v8.h"
13
14
namespace node {
15
16
// Forward declarations
17
class Environment;
18
class ShutdownWrap;
19
class WriteWrap;
20
class StreamBase;
21
class StreamResource;
22
23
struct StreamWriteResult {
24
  bool async;
25
  int err;
26
  WriteWrap* wrap;
27
  size_t bytes;
28
};
29
30
using JSMethodFunction = void(const v8::FunctionCallbackInfo<v8::Value>& args);
31
32
class StreamReq {
33
 public:
34
  // The kSlot internal field here mirrors BaseObject::InternalFields::kSlot
35
  // here because instances derived from StreamReq will also derive from
36
  // BaseObject, and the slots are used for the identical purpose.
37
  enum InternalFields {
38
    kSlot = BaseObject::kSlot,
39
    kStreamReqField = BaseObject::kInternalFieldCount,
40
    kInternalFieldCount
41
  };
42
43
  inline explicit StreamReq(
44
      StreamBase* stream,
45
      v8::Local<v8::Object> req_wrap_obj);
46
47
15154
  virtual ~StreamReq() = default;
48
  virtual AsyncWrap* GetAsyncWrap() = 0;
49
  inline v8::Local<v8::Object> object();
50
51
  inline void Done(int status, const char* error_str = nullptr);
52
  inline void Dispose();
53
54
15154
  StreamBase* stream() const { return stream_; }
55
56
  static inline StreamReq* FromObject(v8::Local<v8::Object> req_wrap_obj);
57
58
  // Sets all internal fields of `req_wrap_obj` to `nullptr`.
59
  // This is what the `WriteWrap` and `ShutdownWrap` JS constructors do,
60
  // and what we use in C++ after creating these objects from their
61
  // v8::ObjectTemplates, to avoid the overhead of calling the
62
  // constructor explicitly.
63
  static inline void ResetObject(v8::Local<v8::Object> req_wrap_obj);
64
65
 protected:
66
  virtual void OnDone(int status) = 0;
67
68
  inline void AttachToObject(v8::Local<v8::Object> req_wrap_obj);
69
70
 private:
71
  StreamBase* const stream_;
72
};
73
74
6654
class ShutdownWrap : public StreamReq {
75
 public:
76
  inline ShutdownWrap(
77
      StreamBase* stream,
78
      v8::Local<v8::Object> req_wrap_obj);
79
80
  // Call stream()->EmitAfterShutdown() and dispose of this request wrap.
81
  void OnDone(int status) override;
82
};
83
84
8500
class WriteWrap : public StreamReq {
85
 public:
86
  inline void SetAllocatedStorage(AllocatedBuffer&& storage);
87
88
  inline WriteWrap(
89
      StreamBase* stream,
90
      v8::Local<v8::Object> req_wrap_obj);
91
92
  // Call stream()->EmitAfterWrite() and dispose of this request wrap.
93
  void OnDone(int status) override;
94
95
 private:
96
  AllocatedBuffer storage_;
97
};
98
99
100
// This is the generic interface for objects that control Node.js' C++ streams.
101
// For example, the default `EmitToJSStreamListener` emits a stream's data
102
// as Buffers in JS, or `TLSWrap` reads and decrypts data from a stream.
103
105031
class StreamListener {
104
 public:
105
  virtual ~StreamListener();
106
107
  // This is called when a stream wants to allocate memory before
108
  // reading data into the freshly allocated buffer (i.e. it is always followed
109
  // by a `OnStreamRead()` call).
110
  // This memory may be statically or dynamically allocated; for example,
111
  // a protocol parser may want to read data into a static buffer if it knows
112
  // that all data is going to be fully handled during the next
113
  // `OnStreamRead()` call.
114
  // The returned buffer does not need to contain `suggested_size` bytes.
115
  // The default implementation of this method returns a buffer that has exactly
116
  // the suggested size and is allocated using malloc().
117
  // It is not valid to return a zero-length buffer from this method.
118
  // It is not guaranteed that the corresponding `OnStreamRead()` call
119
  // happens in the same event loop turn as this call.
120
  virtual uv_buf_t OnStreamAlloc(size_t suggested_size) = 0;
121
122
  // `OnStreamRead()` is called when data is available on the socket and has
123
  // been read into the buffer provided by `OnStreamAlloc()`.
124
  // The `buf` argument is the return value of `uv_buf_t`, or may be a buffer
125
  // with base nullptr in case of an error.
126
  // `nread` is the number of read bytes (which is at most the buffer length),
127
  // or, if negative, a libuv error code.
128
  virtual void OnStreamRead(ssize_t nread,
129
                            const uv_buf_t& buf) = 0;
130
131
  // This is called once a write has finished. `status` may be 0 or,
132
  // if negative, a libuv error code.
133
  // By default, this is simply passed on to the previous listener
134
  // (and raises an assertion if there is none).
135
  virtual void OnStreamAfterWrite(WriteWrap* w, int status);
136
137
  // This is called once a shutdown has finished. `status` may be 0 or,
138
  // if negative, a libuv error code.
139
  // By default, this is simply passed on to the previous listener
140
  // (and raises an assertion if there is none).
141
  virtual void OnStreamAfterShutdown(ShutdownWrap* w, int status);
142
143
  // This is called by the stream if it determines that it wants more data
144
  // to be written to it. Not all streams support this.
145
  // This callback will not be called as long as there are active writes.
146
  // It is not supported by all streams; `stream->HasWantsWrite()` returns
147
  // true if it is supported by a stream.
148
3559
  virtual void OnStreamWantsWrite(size_t suggested_size) {}
149
150
  // This is called immediately before the stream is destroyed.
151
560
  virtual void OnStreamDestroy() {}
152
153
  // The stream this is currently associated with, or nullptr if there is none.
154
1530
  StreamResource* stream() const { return stream_; }
155
156
 protected:
157
  // Pass along a read error to the `StreamListener` instance that was active
158
  // before this one. For example, a protocol parser does not care about read
159
  // errors and may instead want to let the original handler
160
  // (e.g. the JS handler) take care of the situation.
161
  inline void PassReadErrorToPreviousListener(ssize_t nread);
162
163
  StreamResource* stream_ = nullptr;
164
  StreamListener* previous_listener_ = nullptr;
165
166
  friend class StreamResource;
167
};
168
169
170
// An (incomplete) stream listener class that calls the `.oncomplete()`
171
// method of the JS objects associated with the wrap objects.
172
129402
class ReportWritesToJSStreamListener : public StreamListener {
173
 public:
174
  void OnStreamAfterWrite(WriteWrap* w, int status) override;
175
  void OnStreamAfterShutdown(ShutdownWrap* w, int status) override;
176
177
 private:
178
  void OnStreamAfterReqFinished(StreamReq* req_wrap, int status);
179
};
180
181
182
// A default emitter that just pushes data chunks as Buffer instances to
183
// JS land via the handle’s .ondata method.
184
129390
class EmitToJSStreamListener : public ReportWritesToJSStreamListener {
185
 public:
186
  uv_buf_t OnStreamAlloc(size_t suggested_size) override;
187
  void OnStreamRead(ssize_t nread, const uv_buf_t& buf) override;
188
};
189
190
191
// An alternative listener that uses a custom, user-provided buffer
192
// for reading data.
193
12
class CustomBufferJSListener : public ReportWritesToJSStreamListener {
194
 public:
195
  uv_buf_t OnStreamAlloc(size_t suggested_size) override;
196
  void OnStreamRead(ssize_t nread, const uv_buf_t& buf) override;
197
6
  void OnStreamDestroy() override { delete this; }
198
199
6
  explicit CustomBufferJSListener(uv_buf_t buffer) : buffer_(buffer) {}
200
201
 private:
202
  uv_buf_t buffer_;
203
};
204
205
206
// A generic stream, comparable to JS land’s `Duplex` streams.
207
// A stream is always controlled through one `StreamListener` instance.
208
64940
class StreamResource {
209
 public:
210
  virtual ~StreamResource();
211
212
  // These need to be implemented on the readable side of this stream:
213
214
  // Start reading from the underlying resource. This is called by the consumer
215
  // when more data is desired. Use `EmitAlloc()` and `EmitData()` to
216
  // pass data along to the consumer.
217
  virtual int ReadStart() = 0;
218
  // Stop reading from the underlying resource. This is called by the
219
  // consumer when its buffers are full and no more data can be handled.
220
  virtual int ReadStop() = 0;
221
222
  // These need to be implemented on the writable side of this stream:
223
  // All of these methods may return an error code synchronously.
224
  // In that case, the finish callback should *not* be called.
225
226
  // Perform a shutdown operation, and either call req_wrap->Done() when
227
  // finished and return 0, return 1 for synchronous success, or
228
  // a libuv error code for synchronous failures.
229
  virtual int DoShutdown(ShutdownWrap* req_wrap) = 0;
230
  // Try to write as much data as possible synchronously, and modify
231
  // `*bufs` and `*count` accordingly. This is a no-op by default.
232
  // Return 0 for success and a libuv error code for failures.
233
  virtual int DoTryWrite(uv_buf_t** bufs, size_t* count);
234
  // Initiate a write of data. If the write completes synchronously, return 0 on
235
  // success (with bufs modified to indicate how much data was consumed) or a
236
  // libuv error code on failure. If the write will complete asynchronously,
237
  // return 0. When the write completes asynchronously, call req_wrap->Done()
238
  // with 0 on success (with bufs modified to indicate how much data was
239
  // consumed) or a libuv error code on failure. Do not call req_wrap->Done() if
240
  // the write completes synchronously, that is, it should never be called
241
  // before DoWrite() has returned.
242
  virtual int DoWrite(WriteWrap* w,
243
                      uv_buf_t* bufs,
244
                      size_t count,
245
                      uv_stream_t* send_handle) = 0;
246
247
  // Returns true if the stream supports the `OnStreamWantsWrite()` interface.
248
  virtual bool HasWantsWrite() const { return false; }
249
250
  // Optionally, this may provide an error message to be used for
251
  // failing writes.
252
  virtual const char* Error() const;
253
  // Clear the current error (i.e. that would be returned by Error()).
254
  virtual void ClearError();
255
256
  // Transfer ownership of this stream to `listener`. The previous listener
257
  // will not receive any more callbacks while the new listener was active.
258
  inline void PushStreamListener(StreamListener* listener);
259
  // Remove a listener, and, if this was the currently active one,
260
  // transfer ownership back to the previous listener.
261
  inline void RemoveStreamListener(StreamListener* listener);
262
263
 protected:
264
  // Call the current listener's OnStreamAlloc() method.
265
  inline uv_buf_t EmitAlloc(size_t suggested_size);
266
  // Call the current listener's OnStreamRead() method and update the
267
  // stream's read byte counter.
268
  inline void EmitRead(
269
      ssize_t nread,
270
28014
      const uv_buf_t& buf = uv_buf_init(nullptr, 0));
271
  // Call the current listener's OnStreamAfterWrite() method.
272
  inline void EmitAfterWrite(WriteWrap* w, int status);
273
  // Call the current listener's OnStreamAfterShutdown() method.
274
  inline void EmitAfterShutdown(ShutdownWrap* w, int status);
275
  // Call the current listener's OnStreamWantsWrite() method.
276
  inline void EmitWantsWrite(size_t suggested_size);
277
278
  StreamListener* listener_ = nullptr;
279
  uint64_t bytes_read_ = 0;
280
  uint64_t bytes_written_ = 0;
281
282
  friend class StreamListener;
283
};
284
285
286
64450
class StreamBase : public StreamResource {
287
 public:
288
  // The kSlot field here mirrors that of BaseObject::InternalFields::kSlot
289
  // because instances deriving from StreamBase generally also derived from
290
  // BaseObject (it's possible for it not to, however).
291
  enum InternalFields {
292
    kSlot = BaseObject::kSlot,
293
    kStreamBaseField = BaseObject::kInternalFieldCount,
294
    kOnReadFunctionField,
295
    kInternalFieldCount
296
  };
297
298
  static void AddMethods(Environment* env,
299
                         v8::Local<v8::FunctionTemplate> target);
300
301
  virtual bool IsAlive() = 0;
302
  virtual bool IsClosing() = 0;
303
  virtual bool IsIPCPipe();
304
  virtual int GetFD();
305
306
  enum StreamBaseJSChecks { DONT_SKIP_NREAD_CHECKS, SKIP_NREAD_CHECKS };
307
308
  v8::MaybeLocal<v8::Value> CallJSOnreadMethod(
309
      ssize_t nread,
310
      v8::Local<v8::ArrayBuffer> ab,
311
      size_t offset = 0,
312
      StreamBaseJSChecks checks = DONT_SKIP_NREAD_CHECKS);
313
314
  // This is named `stream_env` to avoid name clashes, because a lot of
315
  // subclasses are also `BaseObject`s.
316
805540
  Environment* stream_env() const { return env_; }
317
318
  // Shut down the current stream. This request can use an existing
319
  // ShutdownWrap object (that was created in JS), or a new one will be created.
320
  // Returns 1 in case of a synchronous completion, 0 in case of asynchronous
321
  // completion, and a libuv error case in case of synchronous failure.
322
  inline int Shutdown(
323
      v8::Local<v8::Object> req_wrap_obj = v8::Local<v8::Object>());
324
325
  // Write data to the current stream. This request can use an existing
326
  // WriteWrap object (that was created in JS), or a new one will be created.
327
  // This will first try to write synchronously using `DoTryWrite()`, then
328
  // asynchronously using `DoWrite()`.
329
  // If the return value indicates a synchronous completion, no callback will
330
  // be invoked.
331
  inline StreamWriteResult Write(
332
      uv_buf_t* bufs,
333
      size_t count,
334
      uv_stream_t* send_handle = nullptr,
335
      v8::Local<v8::Object> req_wrap_obj = v8::Local<v8::Object>());
336
337
  // These can be overridden by subclasses to get more specific wrap instances.
338
  // For example, a subclass Foo could create a FooWriteWrap or FooShutdownWrap
339
  // (inheriting from ShutdownWrap/WriteWrap) that has extra fields, like
340
  // an associated libuv request.
341
  virtual ShutdownWrap* CreateShutdownWrap(v8::Local<v8::Object> object);
342
  virtual WriteWrap* CreateWriteWrap(v8::Local<v8::Object> object);
343
344
  // One of these must be implemented
345
  virtual AsyncWrap* GetAsyncWrap() = 0;
346
  virtual v8::Local<v8::Object> GetObject();
347
348
  static inline StreamBase* FromObject(v8::Local<v8::Object> obj);
349
350
 protected:
351
  inline explicit StreamBase(Environment* env);
352
353
  // JS Methods
354
  int ReadStartJS(const v8::FunctionCallbackInfo<v8::Value>& args);
355
  int ReadStopJS(const v8::FunctionCallbackInfo<v8::Value>& args);
356
  int Shutdown(const v8::FunctionCallbackInfo<v8::Value>& args);
357
  int Writev(const v8::FunctionCallbackInfo<v8::Value>& args);
358
  int WriteBuffer(const v8::FunctionCallbackInfo<v8::Value>& args);
359
  template <enum encoding enc>
360
  int WriteString(const v8::FunctionCallbackInfo<v8::Value>& args);
361
  int UseUserBuffer(const v8::FunctionCallbackInfo<v8::Value>& args);
362
363
  static void GetFD(const v8::FunctionCallbackInfo<v8::Value>& args);
364
  static void GetExternal(const v8::FunctionCallbackInfo<v8::Value>& args);
365
  static void GetBytesRead(const v8::FunctionCallbackInfo<v8::Value>& args);
366
  static void GetBytesWritten(const v8::FunctionCallbackInfo<v8::Value>& args);
367
  inline void AttachToObject(v8::Local<v8::Object> obj);
368
369
  template <int (StreamBase::*Method)(
370
      const v8::FunctionCallbackInfo<v8::Value>& args)>
371
  static void JSMethod(const v8::FunctionCallbackInfo<v8::Value>& args);
372
373
  // Internal, used only in StreamBase methods + env.cc.
374
  enum StreamBaseStateFields {
375
    kReadBytesOrError,
376
    kArrayBufferOffset,
377
    kBytesWritten,
378
    kLastWriteWasAsync,
379
    kNumStreamBaseStateFields
380
  };
381
382
 private:
383
  Environment* env_;
384
  EmitToJSStreamListener default_listener_;
385
386
  void SetWriteResult(const StreamWriteResult& res);
387
  static void AddMethod(Environment* env,
388
                        v8::Local<v8::Signature> sig,
389
                        enum v8::PropertyAttribute attributes,
390
                        v8::Local<v8::FunctionTemplate> t,
391
                        JSMethodFunction* stream_method,
392
                        v8::Local<v8::String> str);
393
394
  friend class WriteWrap;
395
  friend class ShutdownWrap;
396
  friend class Environment;  // For kNumStreamBaseStateFields.
397
};
398
399
400
// These are helpers for creating `ShutdownWrap`/`WriteWrap` instances.
401
// `OtherBase` must have a constructor that matches the `AsyncWrap`
402
// constructors’s (Environment*, Local<Object>, AsyncWrap::Provider) signature
403
// and be a subclass of `AsyncWrap`.
404
template <typename OtherBase>
405
13308
class SimpleShutdownWrap : public ShutdownWrap, public OtherBase {
406
 public:
407
  SimpleShutdownWrap(StreamBase* stream,
408
                     v8::Local<v8::Object> req_wrap_obj);
409
410
26616
  AsyncWrap* GetAsyncWrap() override { return this; }
411
412
3
  SET_NO_MEMORY_INFO()
413
3
  SET_MEMORY_INFO_NAME(SimpleShutdownWrap)
414
3
  SET_SELF_SIZE(SimpleShutdownWrap)
415
};
416
417
template <typename OtherBase>
418
17000
class SimpleWriteWrap : public WriteWrap, public OtherBase {
419
 public:
420
  SimpleWriteWrap(StreamBase* stream,
421
                  v8::Local<v8::Object> req_wrap_obj);
422
423
32362
  AsyncWrap* GetAsyncWrap() override { return this; }
424
425
  SET_NO_MEMORY_INFO()
426
  SET_MEMORY_INFO_NAME(SimpleWriteWrap)
427
  SET_SELF_SIZE(SimpleWriteWrap)
428
};
429
430
}  // namespace node
431
432
#endif  // defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
433
434
#endif  // SRC_STREAM_BASE_H_