Branch data Line data Source code
1 : : #include <sys/mman.h>
2 : : #include <sys/stat.h>
3 : : #include <fcntl.h>
4 : : #include <unistd.h>
5 : : #include <sys/types.h>
6 : : #include <memory>
7 : : #include <napi.h>
8 : : #include <memory>
9 : : #include <vector>
10 : : #include <unordered_set>
11 : : #include <algorithm>
12 : : #include <limits>
13 : :
14 : : typedef uint64_t sequence_t;
15 : : typedef int32_t status_t;
16 : :
17 : : const sequence_t sequence_max = std::numeric_limits<sequence_t>::max();
18 : :
19 : : // Needs to be heap allocated because we access it from finalizers which can be called
20 : : // on process exit
21 : : static std::unordered_set<uint8_t*> *buffers;
22 : : static std::mutex buffers_mutex;
23 : :
24 : : const int memorder = __ATOMIC_SEQ_CST;
25 : :
26 : : class Disruptor : public Napi::ObjectWrap<Disruptor>
27 : : {
28 : : public:
29 : : Disruptor(const Napi::CallbackInfo& info);
30 : : ~Disruptor();
31 : :
32 : : static Napi::Object Initialize(Napi::Env env, Napi::Object exports);
33 : :
34 : : // Unmap the shared memory. Don't access it again from this Disruptor!
35 : : void Release(const Napi::CallbackInfo& info);
36 : :
37 : : // Return unconsumed slots for a consumer
38 : : Napi::Value ConsumeNew(const Napi::CallbackInfo& info);
39 : : Napi::Value ConsumeNewSync(const Napi::CallbackInfo& info);
40 : :
41 : : // Commit consumed slots
42 : : Napi::Value ConsumeCommit(const Napi::CallbackInfo&);
43 : :
44 : : // Claim a slot for writing a value
45 : : Napi::Value ProduceClaim(const Napi::CallbackInfo& info);
46 : : Napi::Value ProduceClaimSync(const Napi::CallbackInfo& info);
47 : :
48 : : // Claim multiple slots for writing values
49 : : Napi::Value ProduceClaimMany(const Napi::CallbackInfo& info);
50 : : Napi::Value ProduceClaimManySync(const Napi::CallbackInfo& info);
51 : :
52 : : // Claim all available slots for writing values
53 : : Napi::Value ProduceClaimAvail(const Napi::CallbackInfo& info);
54 : : Napi::Value ProduceClaimAvailSync(const Napi::CallbackInfo& info);
55 : :
56 : : // Commit a claimed slot
57 : : Napi::Value ProduceCommit(const Napi::CallbackInfo& info);
58 : : Napi::Value ProduceCommitSync(const Napi::CallbackInfo& info);
59 : :
60 : : // Get slots previously claimed but not committed
61 : : Napi::Value ProduceRecover(const Napi::CallbackInfo& info);
62 : :
63 : : // Get size of each element in bytes
64 : : Napi::Value GetElementSize(const Napi::CallbackInfo& info);
65 : :
66 : : // Get whether not to return while not ready or full
67 : : Napi::Value GetSpin(const Napi::CallbackInfo& info);
68 : :
69 : : // Get status value
70 : : Napi::Value GetStatus(const Napi::CallbackInfo& info);
71 : :
72 : : // Set status value
73 : : void SetStatus(const Napi::CallbackInfo& info, const Napi::Value& value);
74 : :
75 : 3596153 : inline bool Spin()
76 : : {
77 : 3596153 : return spin;
78 : : }
79 : :
80 : : private:
81 : : friend class ConsumeNewAsyncWorker;
82 : : friend class ProduceClaimAsyncWorker;
83 : : friend class ProduceClaimManyAsyncWorker;
84 : : friend class ProduceClaimAvailAsyncWorker;
85 : : friend class ProduceCommitAsyncWorker;
86 : : friend class SyncBuffer;
87 : : friend class AsyncBuffer;
88 : :
89 : : int Release();
90 : :
91 : : void UpdatePending(sequence_t seq_consumer, sequence_t seq_cursor);
92 : : void UpdateSeqNext(const sequence_t seq_next,
93 : : const sequence_t seq_next_end,
94 : : const bool all_ignored);
95 : : uint32_t GetSeqNext(const Napi::CallbackInfo& info,
96 : : sequence_t& seq_next,
97 : : sequence_t& seq_next_end);
98 : :
99 : : template<typename Array, typename DisruptorBuffer>
100 : : void ProduceGetBuffers(const Napi::Env& env,
101 : : const sequence_t seq_next,
102 : : const sequence_t seq_next_end,
103 : : const bool all_ignored,
104 : : Array& r);
105 : :
106 : : template<typename Array, typename DisruptorBuffer>
107 : : Array ConsumeNewSync(const Napi::Env& env, const bool retry, sequence_t& start);
108 : : void ConsumeNewAsync(const Napi::CallbackInfo& info);
109 : :
110 : : bool ConsumeCommit();
111 : :
112 : : template<typename DisruptorBuffer>
113 : : typename DisruptorBuffer::Buffer ProduceClaimSync(const Napi::Env& env,
114 : : const bool retry,
115 : : sequence_t& out_next,
116 : : sequence_t& out_next_end,
117 : : bool& out_all_ignored);
118 : : void ProduceClaimAsync(const Napi::CallbackInfo& info);
119 : :
120 : : template<typename Array, typename DisruptorBuffer>
121 : : Array ProduceClaimManySync(const Napi::Env& env,
122 : : const uint32_t n,
123 : : const bool retry,
124 : : sequence_t& out_next,
125 : : sequence_t& out_next_end,
126 : : bool& out_all_ignored);
127 : : void ProduceClaimManyAsync(const Napi::CallbackInfo& info);
128 : : void ProduceClaimManyAsync(const Napi::CallbackInfo& info,
129 : : const uint32_t n);
130 : :
131 : : template<typename Array, typename DisruptorBuffer>
132 : : Array ProduceClaimAvailSync(const Napi::Env& env,
133 : : const uint32_t max,
134 : : const bool retry,
135 : : sequence_t& out_next,
136 : : sequence_t& out_next_end,
137 : : bool& out_all_ignored);
138 : : void ProduceClaimAvailAsync(const Napi::CallbackInfo& info);
139 : : void ProduceClaimAvailAsync(const Napi::CallbackInfo& info,
140 : : const uint32_t max);
141 : :
142 : : template<typename Boolean>
143 : : Boolean ProduceCommitSync(const Napi::Env& env,
144 : : sequence_t seq_next,
145 : : sequence_t seq_next_end,
146 : : const bool retry);
147 : : void ProduceCommitAsync(const Napi::CallbackInfo& info);
148 : : void ProduceCommitAsync(const Napi::CallbackInfo& info,
149 : : sequence_t seq_next,
150 : : sequence_t seq_next_end,
151 : : const uint32_t cb_arg);
152 : :
153 : : uint32_t num_elements;
154 : : uint32_t element_size;
155 : : uint32_t num_consumers;
156 : : uint32_t consumer;
157 : : bool init;
158 : : bool spin;
159 : :
160 : : size_t shm_size;
161 : : void* shm_buf;
162 : :
163 : : sequence_t *consumers; // for each consumer, next slot to read
164 : : sequence_t *cursor; // next slot to be filled
165 : : sequence_t *next; // next slot to claim
166 : : status_t *status; // status code (app-specific)
167 : : uint8_t* elements;
168 : : sequence_t *ptr_consumer;
169 : :
170 : : sequence_t pending_seq_consumer;
171 : : sequence_t pending_seq_cursor;
172 : :
173 : : sequence_t pending_seq_next;
174 : : sequence_t pending_seq_next_end;
175 : :
176 : : bool all_consumers_ignoring;
177 : :
178 : : Napi::Reference<Napi::Buffer<uint8_t>> shm_buffer_ref;
179 : : Napi::Reference<Napi::Buffer<uint8_t>> elements_buffer_ref;
180 : : Napi::Reference<Napi::Buffer<uint8_t>> consumers_buffer_ref;
181 : : Napi::FunctionReference slice_ref;
182 : :
183 : : Napi::Value GetConsumers(const Napi::CallbackInfo& info);
184 : : Napi::Value GetCursor(const Napi::CallbackInfo& info);
185 : : Napi::Value GetNext(const Napi::CallbackInfo& info);
186 : : Napi::Value GetElements(const Napi::CallbackInfo& info);
187 : : Napi::Value GetConsumer(const Napi::CallbackInfo& info);
188 : : Napi::Value GetPendingSeqConsumer(const Napi::CallbackInfo& info);
189 : : Napi::Value GetPendingSeqCursor(const Napi::CallbackInfo& info);
190 : : Napi::Value GetPendingSeqNext(const Napi::CallbackInfo& info);
191 : : Napi::Value GetPendingSeqNextEnd(const Napi::CallbackInfo& info);
192 : : Napi::Value GetAllConsumersIgnoring(const Napi::CallbackInfo& info);
193 : :
194 : : void ThrowErrnoError(const Napi::CallbackInfo& info,
195 : : const char *msg);
196 : : };
197 : :
198 : : //LCOV_EXCL_START
199 : : void NullCallback(const Napi::CallbackInfo&)
200 : : {
201 : : }
202 : : //LCOV_EXCL_STOP
203 : :
204 : 488217 : Napi::Function GetCallback(const Napi::CallbackInfo& info, const uint32_t cb_arg)
205 : : {
206 : 488217 : if (info.Length() > cb_arg)
207 : : {
208 : 488217 : Napi::Value cb = info[cb_arg];
209 : 488217 : if (cb.IsFunction())
210 : : {
211 : 488217 : return cb.As<Napi::Function>();
212 : : }
213 : : }
214 : :
215 : : return Napi::Function::New<&NullCallback>(info.Env()); //LCOV_EXCL_LINE
216 : : }
217 : :
218 : : class SyncBuffer
219 : : {
220 : : public:
221 : : typedef Napi::Buffer<uint8_t> Buffer;
222 : :
223 : 1582385 : static Buffer New(Napi::Env env, Disruptor *d, sequence_t start, sequence_t end)
224 : : {
225 : : #ifdef COVERAGE
226 : 4747155 : return d->slice_ref.Value().Call(d->elements_buffer_ref.Value(),
227 : : #else
228 : : return d->slice_ref.Call(d->elements_buffer_ref.Value(),
229 : : #endif
230 : : {
231 : 1582385 : Napi::Number::New(env, start * d->element_size),
232 : 3164770 : Napi::Number::New(env, end * d->element_size)
233 : 3164770 : }).As<Buffer>();
234 : : }
235 : : };
236 : :
237 : : class AsyncBuffer
238 : : {
239 : : public:
240 : : typedef AsyncBuffer Buffer;
241 : :
242 : 526365 : AsyncBuffer() : AsyncBuffer(0, 0, 0)
243 : : {
244 : 526365 : }
245 : :
246 : 526365 : static Buffer New(Napi::Env, Disruptor *d, sequence_t start, sequence_t end)
247 : : {
248 : 526365 : return Buffer(start, end, d->element_size);
249 : : }
250 : :
251 : 360539 : Napi::Value ToValue(Napi::Env env, Disruptor *d)
252 : : {
253 : 360539 : return SyncBuffer::New(env, d, start, end);
254 : : }
255 : :
256 : 170845 : size_t Length()
257 : : {
258 : 170845 : return length;
259 : : }
260 : :
261 : : private:
262 : 1052730 : AsyncBuffer(sequence_t start, sequence_t end, uint32_t element_size) :
263 : 1052730 : start(start),
264 : 1052730 : end(end),
265 : 1052730 : length((end - start) * element_size)
266 : : {
267 : 1052730 : }
268 : :
269 : : sequence_t start;
270 : : sequence_t end;
271 : : size_t length;
272 : : };
273 : :
274 : : template<typename T>
275 : : class AsyncArray
276 : : {
277 : : public:
278 : 2589014 : AsyncArray() :
279 : 2589014 : elements(std::make_unique<std::vector<T>>())
280 : : {
281 : 2589014 : }
282 : :
283 : 1294507 : static AsyncArray<T> New(Napi::Env)
284 : : {
285 : 1294507 : return AsyncArray<T>();
286 : : }
287 : :
288 : 355520 : void Set(uint32_t index, T&& el)
289 : : {
290 : 355520 : if (elements->size() <= index)
291 : : {
292 : 355520 : elements->resize(index + 1);
293 : : }
294 : :
295 : 355520 : (*elements)[index] = std::move(el);
296 : 355520 : }
297 : :
298 : 355475 : Napi::Value ToValue(Napi::Env env, Disruptor *d)
299 : : {
300 : 355475 : size_t length = elements->size();
301 : 355475 : Napi::Array r = Napi::Array::New(env);
302 : 710995 : for (size_t i = 0; i < length; ++i)
303 : : {
304 : 355520 : r[i] = (*elements)[i].ToValue(env, d);
305 : : }
306 : 355475 : return r;
307 : : }
308 : :
309 : 1294507 : size_t Length()
310 : : {
311 : 1294507 : return elements->size();
312 : : }
313 : :
314 : : private:
315 : : std::unique_ptr<std::vector<T>> elements;
316 : : };
317 : :
318 : : class AsyncBoolean
319 : : {
320 : : public:
321 : 2130801 : AsyncBoolean() :
322 : 2130801 : b(false)
323 : : {
324 : 2130801 : }
325 : :
326 : 2130801 : static AsyncBoolean New(Napi::Env, bool b)
327 : : {
328 : 2130801 : return AsyncBoolean(b);
329 : : }
330 : :
331 : 127723 : Napi::Value ToValue(Napi::Env env, Disruptor*)
332 : : {
333 : 127723 : return Napi::Boolean::New(env, b);
334 : : }
335 : :
336 : 2130801 : operator bool() const
337 : : {
338 : 2130801 : return b;
339 : : }
340 : :
341 : : private:
342 : 2130801 : AsyncBoolean(bool b) :
343 : 2130801 : b(b)
344 : : {
345 : 2130801 : }
346 : :
347 : : bool b;
348 : : };
349 : :
350 : : class AsyncUndefined
351 : : {
352 : : };
353 : :
354 : 370562 : Napi::Number ToValue(const Napi::Env& env, sequence_t n)
355 : : {
356 : 370562 : return Napi::Number::New(env, n);
357 : : }
358 : :
359 : 1094089 : Napi::Value ToValue(const Napi::Env& env, const AsyncUndefined&)
360 : : {
361 : 1094089 : return env.Undefined();
362 : : }
363 : :
364 : : template <typename Result,
365 : : typename Arg1 = AsyncUndefined,
366 : : typename Arg2 = AsyncUndefined,
367 : : typename Arg3 = AsyncUndefined>
368 : : class DisruptorAsyncWorker : public Napi::AsyncWorker
369 : : {
370 : : public:
371 : 3596153 : DisruptorAsyncWorker(Disruptor *disruptor,
372 : : const Napi::Function& callback) :
373 : : Napi::AsyncWorker(callback),
374 : 3596153 : retry(false),
375 : 3596153 : disruptor(disruptor), // disruptor_ref keeps this around
376 : 3596153 : disruptor_ref(Napi::Persistent(disruptor->Value()))
377 : : {
378 : 3596153 : }
379 : :
380 : : protected:
381 : : virtual void Retry() = 0;
382 : :
383 : 3596153 : void OnOK() override
384 : : {
385 : 3596153 : if (disruptor->Spin() && retry)
386 : : {
387 : 3107936 : return Retry();
388 : : }
389 : :
390 : 488217 : Napi::Env env = Env();
391 : :
392 : 3417519 : Callback().MakeCallback(
393 : 488217 : Receiver().Value(),
394 : : std::initializer_list<napi_value>
395 : : {
396 : 488217 : env.Null(),
397 : 488217 : result.ToValue(env, disruptor),
398 : 488217 : ToValue(env, arg1),
399 : 488217 : ToValue(env, arg2),
400 : 976434 : ToValue(env, arg3)
401 : : });
402 : : }
403 : :
404 : : Result result;
405 : : Arg1 arg1;
406 : : Arg2 arg2;
407 : : Arg3 arg3;
408 : : bool retry;
409 : : Disruptor *disruptor;
410 : :
411 : : private:
412 : : Napi::ObjectReference disruptor_ref;
413 : : };
414 : :
415 : : class CloseFD
416 : : {
417 : : public:
418 : 5763 : void operator()(int *fd)
419 : : {
420 : 5763 : close(*fd);
421 : 5763 : delete fd;
422 : 5763 : }
423 : : };
424 : :
425 : 8 : void Disruptor::ThrowErrnoError(const Napi::CallbackInfo& info,
426 : : const char *msg)
427 : : {
428 : 8 : int errnum = errno;
429 : 8 : char buf[1024] = {0};
430 : : #ifdef __APPLE__
431 : : auto err = strerror_r(errnum, buf, sizeof(buf));
432 : : static_assert(std::is_same<decltype(err), int>::value,
433 : : "strerror_r must return int");
434 : : char *errmsg = err == 0 ? buf : nullptr;
435 : : #else
436 : 8 : auto errmsg = strerror_r(errnum, buf, sizeof(buf));
437 : : static_assert(std::is_same<decltype(errmsg), char*>::value,
438 : : "strerror_r must return char*");
439 : : #endif
440 : 8 : throw Napi::Error::New(info.Env(),
441 : 16 : std::string(msg) + ": " + (errmsg ? errmsg : std::to_string(errnum)));
442 : : }
443 : :
444 : 5771 : Disruptor::Disruptor(const Napi::CallbackInfo& info) :
445 : : Napi::ObjectWrap<Disruptor>(info),
446 : 5771 : shm_buf(MAP_FAILED)
447 : : {
448 : : // Arguments
449 : 5771 : Napi::String shm_name = info[0].As<Napi::String>();
450 : 5771 : num_elements = info[1].As<Napi::Number>();
451 : 5771 : element_size = info[2].As<Napi::Number>();
452 : 5771 : num_consumers = info[3].As<Napi::Number>();
453 : 5771 : consumer = info[4].As<Napi::Number>();
454 : 5771 : init = info[5].As<Napi::Boolean>();
455 : 5771 : spin = info[6].As<Napi::Boolean>();
456 : :
457 : : // Open shared memory object
458 : : // OS X does not allow using O_TRUNC with shm_open.
459 : : // If this item exists, and init flag is true, delete it and recreate.
460 : 5771 : const auto name = shm_name.Utf8Value();
461 : 5771 : int shm_fd_tmp = shm_open(name.c_str(),
462 : 5771 : (init ? O_CREAT | O_EXCL : 0) | O_RDWR,
463 : : S_IRUSR | S_IWUSR);
464 : :
465 : 5771 : if (init && shm_fd_tmp < 0 && errno == EEXIST)
466 : : {
467 : 282 : shm_unlink(name.c_str());
468 : 282 : shm_fd_tmp = shm_open(name.c_str(),
469 : : O_CREAT | O_EXCL | O_RDWR,
470 : : S_IRUSR | S_IWUSR);
471 : : }
472 : :
473 : 5771 : if (shm_fd_tmp < 0)
474 : : {
475 : 8 : ThrowErrnoError(info, "Failed to open shared memory object");
476 : : }
477 : :
478 : 5763 : std::unique_ptr<int, CloseFD> shm_fd(new int(shm_fd_tmp));
479 : :
480 : : // Allow space for:
481 : : // - a sequence number for each consumer
482 : : // - the cursor sequence number (last filled slot)
483 : : // - the next sequence number (first free slot)
484 : : // - a status code
485 : : // - all the elements
486 : 5763 : shm_size = (num_consumers + 2) * sizeof(sequence_t) +
487 : 5763 : sizeof(status_t) +
488 : 5763 : num_elements * element_size;
489 : :
490 : : // Resize the shared memory if we're initializing it.
491 : : // Note: ftruncate initializes to null bytes.
492 : 5763 : if (init && (ftruncate(*shm_fd, shm_size) < 0))
493 : : {
494 : : ThrowErrnoError(info, "Failed to size shared memory"); //LCOV_EXCL_LINE
495 : : }
496 : :
497 : : // Map the shared memory
498 : 5763 : shm_buf = mmap(NULL,
499 : : shm_size,
500 : : PROT_READ | PROT_WRITE, MAP_SHARED,
501 : 5763 : *shm_fd,
502 : : 0);
503 : 5763 : if (shm_buf == MAP_FAILED)
504 : : {
505 : : ThrowErrnoError(info, "Failed to map shared memory"); //LCOV_EXCL_LINE
506 : : }
507 : :
508 : 5763 : consumers = static_cast<sequence_t*>(shm_buf);
509 : 5763 : cursor = &consumers[num_consumers];
510 : 5763 : next = &cursor[1];
511 : 5763 : status = reinterpret_cast<status_t*>(&next[1]);
512 : 5763 : elements = reinterpret_cast<uint8_t*>(&status[1]);
513 : 5763 : ptr_consumer = &consumers[consumer];
514 : :
515 : 5763 : pending_seq_consumer = 0;
516 : 5763 : pending_seq_cursor = 0;
517 : :
518 : 5763 : pending_seq_next = 1;
519 : 5763 : pending_seq_next_end = 0;
520 : :
521 : 5763 : all_consumers_ignoring = false;
522 : :
523 : : // From Node 14, V8 doesn't allow buffers pointing to the same memory:
524 : : //
525 : : // https://monorail-prod.appspot.com/p/v8/issues/detail?id=9908
526 : : // https://github.com/nodejs/node/issues/32463
527 : : //
528 : : // We work around this by having a single Buffer over all of shm_buf and using
529 : : // Buffer#slice() to return data from it.
530 : : //
531 : : // This leaves the issue of shm_buf getting the same address as a previous call -
532 : : // whether because the same shared memory is mapped or because it's been
533 : : // unmapped in Release() and then remapped again.
534 : : //
535 : : // Since Node 14.3.0, we can be sure that the finalizer is run after the memory
536 : : // pointer is removed from the BackingStore:
537 : : //
538 : : // https://github.com/nodejs/node/pull/33321
539 : : //
540 : : // However, if the memory is Release()d before it's removed from the BackingStore
541 : : // then mmap may return it again and we won't be able to pass it to V8 without
542 : : // it exiting.
543 : : //
544 : : // Further, even if we remember the Buffers we create in an unordered_map, they
545 : : // become invalid before the finalizer is called. The finalizer is only guaranteed
546 : : // to be called sometime after the object is collected:
547 : : //
548 : : // https://github.com/nodejs/node-addon-api/issues/702#issuecomment-625897608
549 : : //
550 : : // This also applies to weak N-API references to the object.
551 : : //
552 : : // We can detect this using napi_get_reference_value but there's nothing we can
553 : : // do about it since the value will still be in the BackingStore until the finalizer
554 : : // is run. It's this window of time that's the problem - between the Buffer being
555 : : // collected and the finalizer being called.
556 : : //
557 : : // The solution implemented below is to maintain an unordered set of shm_bufs that
558 : : // are still alive in the BackingStore. If mmap returns one of these, we search
559 : : // downwards for the next address not in the set, adjusting the length of the
560 : : // Buffer we need to create accordingly. Since we're slicing Buffer views over it,
561 : : // where it starts from doesn't matter.
562 : :
563 : 5763 : Napi::Env env = info.Env();
564 : 5763 : const auto JSBuffer = env.Global().Get("Buffer").As<Napi::Function>();
565 : 5763 : const auto proto = JSBuffer.Get("prototype").As<Napi::Object>();
566 : 5763 : slice_ref = Napi::Persistent(proto.Get("slice").As<Napi::Function>());
567 : :
568 : 5763 : auto shm_buf8 = static_cast<uint8_t*>(shm_buf);
569 : 5763 : auto shm_size8 = shm_size;
570 : 5763 : Napi::Buffer<uint8_t> shm_buffer;
571 : :
572 : : {
573 : 5763 : std::lock_guard<std::mutex> lock(buffers_mutex);
574 : :
575 : 9818 : while (shm_buf8) {
576 : 9818 : if (buffers->find(shm_buf8) == buffers->end()) {
577 : 5763 : break;
578 : : }
579 : 4055 : --shm_buf8;
580 : 4055 : ++shm_size8;
581 : : }
582 : 5763 : }
583 : :
584 : 5763 : if (!shm_buf8) {
585 : : //LCOV_EXCL_START
586 : : Release();
587 : : throw Napi::Error::New(env, "No space for buffer due to due to https://github.com/nodejs/node/issues/32463");
588 : : //LCOV_EXCL_STOP
589 : : }
590 : :
591 : 5763 : shm_buffer = Napi::Buffer<uint8_t>::New(
592 : : env,
593 : : shm_buf8,
594 : : shm_size8,
595 : 4758 : [](Napi::Env, uint8_t* shm_buf8) {
596 : 4758 : std::lock_guard<std::mutex> lock(buffers_mutex);
597 : 4758 : buffers->erase(shm_buf8);
598 : 4758 : });
599 : :
600 : : {
601 : 5763 : std::lock_guard<std::mutex> lock(buffers_mutex);
602 : 5763 : buffers->emplace(shm_buf8);
603 : 5763 : }
604 : :
605 : 5763 : shm_buffer_ref = Napi::Persistent(shm_buffer);
606 : :
607 : 5763 : const auto elements_start = elements - shm_buf8;
608 : 17289 : elements_buffer_ref = Napi::Persistent(slice_ref.Call(shm_buffer_ref.Value(),
609 : : {
610 : 5763 : Napi::Number::New(env, elements_start),
611 : 5763 : Napi::Number::New(env, elements_start + num_elements * element_size)
612 : 5763 : }).As<Napi::Buffer<uint8_t>>());
613 : :
614 : 5763 : const auto consumers_start = reinterpret_cast<uint8_t*>(consumers) - shm_buf8;
615 : 17289 : consumers_buffer_ref = Napi::Persistent(slice_ref.Call(shm_buffer_ref.Value(),
616 : : {
617 : 5763 : Napi::Number::New(env, consumers_start),
618 : 5763 : Napi::Number::New(env, consumers_start + num_consumers * sizeof(sequence_t))
619 : 5763 : }).As<Napi::Buffer<uint8_t>>());
620 : 5811 : }
621 : :
622 : 9516 : Disruptor::~Disruptor()
623 : : {
624 : 4758 : Release();
625 : 9516 : }
626 : :
627 : 10015 : int Disruptor::Release()
628 : : {
629 : 10015 : shm_buffer_ref.Reset();
630 : 10015 : elements_buffer_ref.Reset();
631 : 10015 : consumers_buffer_ref.Reset();
632 : 10015 : slice_ref.Reset();
633 : :
634 : 10015 : if (shm_buf != MAP_FAILED)
635 : : {
636 : 5762 : int r = munmap(shm_buf, shm_size);
637 : :
638 : 5762 : if (r < 0)
639 : : {
640 : : return r; //LCOV_EXCL_LINE
641 : : }
642 : :
643 : 5762 : shm_buf = MAP_FAILED;
644 : : }
645 : :
646 : 10015 : return 0;
647 : : }
648 : :
649 : 5257 : void Disruptor::Release(const Napi::CallbackInfo& info)
650 : : {
651 : 15771 : if ((shm_buf != MAP_FAILED) &&
652 : 6265 : (info.Length() >= 1) &&
653 : 6265 : info[0].As<Napi::Boolean>())
654 : : {
655 : 1008 : __atomic_store_n(ptr_consumer, sequence_max, memorder);
656 : : }
657 : :
658 : 5257 : if (Release() < 0)
659 : : {
660 : : ThrowErrnoError(info, "Failed to unmap shared memory"); //LCOV_EXCL_LINE
661 : : }
662 : 5257 : }
663 : :
664 : : #include <iostream>
665 : :
666 : : template<typename Array, typename DisruptorBuffer>
667 : 8149199 : Array Disruptor::ConsumeNewSync(const Napi::Env& env,
668 : : const bool retry,
669 : : sequence_t &start)
670 : : {
671 : : // Return all elements [&consumers[consumer], cursor)
672 : :
673 : : // Commit previous consume
674 : 8149199 : ConsumeCommit();
675 : :
676 : : do
677 : : {
678 : 8149199 : sequence_t seq_consumer = __atomic_load_n(ptr_consumer, memorder);
679 : 8149199 : sequence_t seq_cursor = __atomic_load_n(cursor, memorder);
680 : 8149199 : sequence_t pos_consumer = seq_consumer % num_elements;
681 : 8149199 : sequence_t pos_cursor = seq_cursor % num_elements;
682 : :
683 : 8149199 : if (pos_cursor > pos_consumer)
684 : : {
685 : 948060 : Array r = Array::New(env);
686 : 948060 : r.Set(0U, DisruptorBuffer::New(env, this, pos_consumer, pos_cursor));
687 : 948060 : UpdatePending(seq_consumer, seq_cursor);
688 : 948060 : start = seq_consumer;
689 : 948060 : return r;
690 : 355318 : }
691 : :
692 : 7201139 : if (seq_cursor != seq_consumer)
693 : : {
694 : 106157 : Array r = Array::New(env);
695 : 106157 : r.Set(0U, DisruptorBuffer::New(env, this, pos_consumer, num_elements));
696 : 106157 : if (pos_cursor > 0)
697 : : {
698 : 653 : r.Set(1U, DisruptorBuffer::New(env, this, 0, pos_cursor));
699 : : }
700 : 106157 : UpdatePending(seq_consumer, seq_cursor);
701 : 106157 : start = seq_consumer;
702 : 106157 : return r;
703 : 140 : }
704 : : }
705 : 7094982 : while (retry);
706 : :
707 : 7094982 : start = 0;
708 : : // ConsumeCommit() above already set pending_set_cursor to 0
709 : 7094982 : return Array::New(env);
710 : : }
711 : :
712 : 13 : Napi::Value Disruptor::ConsumeNewSync(const Napi::CallbackInfo& info)
713 : : {
714 : : sequence_t start;
715 : 13 : return ConsumeNewSync<Napi::Array, SyncBuffer>(info.Env(), spin, start);
716 : : }
717 : :
718 : : class ConsumeNewAsyncWorker :
719 : : public DisruptorAsyncWorker<AsyncArray<AsyncBuffer>, sequence_t>
720 : : {
721 : : public:
722 : 1074324 : ConsumeNewAsyncWorker(Disruptor *disruptor,
723 : 1074324 : const Napi::Function& callback) :
724 : : DisruptorAsyncWorker<AsyncArray<AsyncBuffer>, sequence_t>(
725 : 1074324 : disruptor, callback)
726 : : {
727 : 1074324 : arg1 = 0;
728 : 1074324 : }
729 : :
730 : : protected:
731 : 1074324 : void Execute() override
732 : : {
733 : : // Remember: don't access any V8 stuff in worker thread
734 : 1074324 : result = disruptor->ConsumeNewSync<AsyncArray<AsyncBuffer>, AsyncBuffer>(Env(), false, arg1);
735 : 1074324 : retry = result.Length() == 0;
736 : 1074324 : }
737 : :
738 : 718864 : void Retry() override
739 : : {
740 : 718864 : (new ConsumeNewAsyncWorker(disruptor, Callback().Value()))->Queue();
741 : 718864 : }
742 : : };
743 : :
744 : 355460 : void Disruptor::ConsumeNewAsync(const Napi::CallbackInfo& info)
745 : : {
746 : 355460 : (new ConsumeNewAsyncWorker(this, GetCallback(info, 0)))->Queue();
747 : 355460 : }
748 : :
749 : 7074862 : Napi::Value Disruptor::ConsumeNew(const Napi::CallbackInfo& info)
750 : : {
751 : : sequence_t start;
752 : 7074862 : Napi::Array r = ConsumeNewSync<Napi::Array, SyncBuffer>(
753 : 7074862 : info.Env(), false, start);
754 : :
755 : 7074862 : if ((r.Length() > 0) || !spin)
756 : : {
757 : 6719415 : return r;
758 : : }
759 : :
760 : 355447 : ConsumeNewAsync(info);
761 : 355447 : return info.Env().Undefined();
762 : : }
763 : :
764 : 7074880 : Napi::Value Disruptor::ConsumeCommit(const Napi::CallbackInfo& info)
765 : : {
766 : 7074880 : return Napi::Boolean::New(info.Env(), ConsumeCommit());
767 : : }
768 : :
769 : 1054217 : void Disruptor::UpdatePending(sequence_t seq_consumer, sequence_t seq_cursor)
770 : : {
771 : 1054217 : pending_seq_consumer = seq_consumer;
772 : 1054217 : pending_seq_cursor = seq_cursor;
773 : 1054217 : }
774 : :
775 : 15224079 : bool Disruptor::ConsumeCommit()
776 : : {
777 : 15224079 : bool r = true;
778 : :
779 : 15224079 : if (pending_seq_cursor)
780 : : {
781 : 1054216 : sequence_t expected = pending_seq_consumer;
782 : 1054216 : r = __atomic_compare_exchange_n(ptr_consumer,
783 : : &expected,
784 : : pending_seq_cursor,
785 : : false,
786 : : memorder,
787 : : memorder);
788 : 1054216 : pending_seq_cursor = 0;
789 : : }
790 : :
791 : 15224079 : return r;
792 : : }
793 : :
794 : : template<typename DisruptorBuffer>
795 : 691097 : typename DisruptorBuffer::Buffer Disruptor::ProduceClaimSync(const Napi::Env& env,
796 : : const bool retry,
797 : : sequence_t& out_next,
798 : : sequence_t& out_next_end,
799 : : bool& out_all_ignored)
800 : : {
801 : : bool all_ignored;
802 : :
803 : : do
804 : : {
805 : 691097 : sequence_t seq_next = __atomic_load_n(next, memorder);
806 : :
807 : 691097 : bool can_claim = true;
808 : 691097 : all_ignored = true;
809 : :
810 : 3236761 : for (uint32_t i = 0; i < num_consumers; ++i)
811 : : {
812 : 2711502 : sequence_t seq_consumer = __atomic_load_n(&consumers[i], memorder);
813 : :
814 : 2711502 : if (seq_consumer != sequence_max)
815 : : {
816 : 2711494 : all_ignored = false;
817 : :
818 : 2711494 : if ((seq_next - seq_consumer) >= num_elements)
819 : : {
820 : 165838 : can_claim = false;
821 : 165838 : break;
822 : : }
823 : : }
824 : : }
825 : :
826 : 691097 : if (all_ignored)
827 : : {
828 : 4 : break;
829 : : }
830 : :
831 : 1216348 : if (can_claim &&
832 : 525255 : __atomic_compare_exchange_n(next, &seq_next, seq_next + 1, false, memorder, memorder))
833 : : {
834 : 521039 : sequence_t start = seq_next % num_elements;
835 : 521039 : auto r = DisruptorBuffer::New(env, this, start, start + 1);
836 : 521039 : UpdateSeqNext(seq_next, seq_next, all_ignored);
837 : 521039 : out_next = seq_next;
838 : 521039 : out_next_end = seq_next;
839 : 521039 : out_all_ignored = all_ignored;
840 : 521039 : return r;
841 : : }
842 : : }
843 : 170054 : while (retry);
844 : :
845 : 170058 : UpdateSeqNext(1, 0, all_ignored);
846 : 170058 : out_next = 1;
847 : 170058 : out_next_end = 0;
848 : 170058 : out_all_ignored = all_ignored;
849 : 170058 : return DisruptorBuffer::New(env, this, 0, 0);
850 : : }
851 : :
852 : 807 : Napi::Value Disruptor::ProduceClaimSync(const Napi::CallbackInfo& info)
853 : : {
854 : : sequence_t seq_next, seq_next_end;
855 : : bool all_ignored;
856 : 807 : return ProduceClaimSync<SyncBuffer>(info.Env(), spin, seq_next, seq_next_end, all_ignored);
857 : : }
858 : :
859 : : class ProduceClaimAsyncWorker :
860 : : public DisruptorAsyncWorker<AsyncBuffer, sequence_t, sequence_t, bool>
861 : : {
862 : : public:
863 : 170845 : ProduceClaimAsyncWorker(Disruptor *disruptor,
864 : 170845 : const Napi::Function& callback) :
865 : : DisruptorAsyncWorker<AsyncBuffer, sequence_t, sequence_t, bool>(
866 : 170845 : disruptor, callback)
867 : : {
868 : 170845 : arg1 = 1;
869 : 170845 : arg2 = 0;
870 : 170845 : arg3 = false;
871 : 170845 : }
872 : :
873 : : protected:
874 : 170845 : void Execute() override
875 : : {
876 : : // Remember: don't access any V8 stuff in worker thread
877 : 170845 : result = disruptor->ProduceClaimSync<AsyncBuffer>(
878 : 170845 : Env(), false, arg1, arg2, arg3);
879 : 170845 : retry = result.Length() == 0;
880 : 170845 : }
881 : :
882 : 165826 : void Retry() override
883 : : {
884 : 165826 : (new ProduceClaimAsyncWorker(disruptor, Callback().Value()))->Queue();
885 : 165826 : }
886 : : };
887 : :
888 : 5019 : void Disruptor::ProduceClaimAsync(const Napi::CallbackInfo& info)
889 : : {
890 : 5019 : (new ProduceClaimAsyncWorker(this, GetCallback(info, 0)))->Queue();
891 : 5019 : }
892 : :
893 : 519445 : Napi::Value Disruptor::ProduceClaim(const Napi::CallbackInfo& info)
894 : : {
895 : : sequence_t seq_next, seq_next_end;
896 : : bool all_ignored;
897 : 519445 : Napi::Buffer<uint8_t> r = ProduceClaimSync<SyncBuffer>(
898 : 519445 : info.Env(), false, seq_next, seq_next_end, all_ignored);
899 : :
900 : 519445 : if ((r.Length() > 0) || all_ignored || !spin)
901 : : {
902 : 515233 : return r;
903 : : }
904 : :
905 : 4212 : ProduceClaimAsync(info);
906 : 4212 : return info.Env().Undefined();
907 : : }
908 : :
909 : : template<typename Array, typename DisruptorBuffer>
910 : 2236 : void Disruptor::ProduceGetBuffers(const Napi::Env& env,
911 : : const sequence_t seq_next,
912 : : const sequence_t seq_next_end,
913 : : const bool all_ignored,
914 : : Array& r)
915 : : {
916 : 2236 : sequence_t pos_next = seq_next % num_elements;
917 : 2236 : sequence_t pos_next_end = seq_next_end % num_elements;
918 : :
919 : 2236 : if (pos_next_end < pos_next)
920 : : {
921 : 8 : r.Set(0U, DisruptorBuffer::New(env, this, pos_next, num_elements));
922 : 8 : r.Set(1U, DisruptorBuffer::New(env, this, 0, pos_next_end + 1));
923 : : }
924 : : else
925 : : {
926 : 2228 : r.Set(0U, DisruptorBuffer::New(env, this, pos_next, pos_next_end + 1));
927 : : }
928 : :
929 : 2236 : UpdateSeqNext(seq_next, seq_next_end, all_ignored);
930 : 2236 : }
931 : :
932 : : template<typename Array, typename DisruptorBuffer>
933 : 147547 : Array Disruptor::ProduceClaimManySync(const Napi::Env& env,
934 : : const uint32_t n,
935 : : const bool retry,
936 : : sequence_t& out_next,
937 : : sequence_t& out_next_end,
938 : : bool& out_all_ignored)
939 : : {
940 : : bool all_ignored;
941 : :
942 : : do
943 : : {
944 : 147547 : sequence_t seq_next = __atomic_load_n(next, memorder);
945 : 147547 : sequence_t seq_next_end = seq_next + std::min(n, num_elements) - 1;
946 : :
947 : 147547 : bool can_claim = true;
948 : 147547 : all_ignored = true;
949 : :
950 : 147609 : for (uint32_t i = 0; i < num_consumers; ++i)
951 : : {
952 : 147563 : sequence_t seq_consumer = __atomic_load_n(&consumers[i], memorder);
953 : :
954 : 147563 : if (seq_consumer != sequence_max)
955 : : {
956 : 147551 : all_ignored = false;
957 : :
958 : 147551 : if ((seq_next_end - seq_consumer) >= num_elements)
959 : : {
960 : 147501 : can_claim = false;
961 : 147501 : break;
962 : : }
963 : : }
964 : : }
965 : :
966 : 147547 : if (all_ignored)
967 : : {
968 : 4 : break;
969 : : }
970 : :
971 : 147585 : if (can_claim &&
972 : 42 : __atomic_compare_exchange_n(next, &seq_next, seq_next_end + 1, false, memorder, memorder))
973 : : {
974 : 42 : Array r = Array::New(env);
975 : 42 : ProduceGetBuffers<Array, DisruptorBuffer>(env, seq_next, seq_next_end, all_ignored, r);
976 : 42 : out_next = seq_next;
977 : 42 : out_next_end = seq_next_end;
978 : 42 : out_all_ignored = all_ignored;
979 : 42 : return r;
980 : 9 : }
981 : : }
982 : 147501 : while (retry);
983 : :
984 : 147505 : UpdateSeqNext(1, 0, all_ignored);
985 : 147505 : out_next = 1;
986 : 147505 : out_next_end = 0;
987 : 147505 : out_all_ignored = all_ignored;
988 : 147505 : return Array::New(env);
989 : : }
990 : :
991 : 11 : Napi::Value Disruptor::ProduceClaimManySync(const Napi::CallbackInfo& info)
992 : : {
993 : : sequence_t seq_next, seq_next_end;
994 : : bool all_ignored;
995 : 22 : return ProduceClaimManySync<Napi::Array, SyncBuffer>(
996 : 33 : info.Env(), info[0].As<Napi::Number>(), spin, seq_next, seq_next_end, all_ignored);
997 : : }
998 : :
999 : : class ProduceClaimManyAsyncWorker :
1000 : : public DisruptorAsyncWorker<AsyncArray<AsyncBuffer>,
1001 : : sequence_t,
1002 : : sequence_t,
1003 : : bool>
1004 : : {
1005 : : public:
1006 : 147496 : ProduceClaimManyAsyncWorker(Disruptor *disruptor,
1007 : : const Napi::Function& callback,
1008 : 147496 : uint32_t n) :
1009 : : DisruptorAsyncWorker<AsyncArray<AsyncBuffer>,
1010 : : sequence_t,
1011 : : sequence_t,
1012 : : bool>(
1013 : : disruptor, callback),
1014 : 147496 : n(n)
1015 : : {
1016 : 147496 : arg1 = 1;
1017 : 147496 : arg2 = 0;
1018 : 147496 : arg3 = false;
1019 : 147496 : }
1020 : :
1021 : : protected:
1022 : 147496 : void Execute() override
1023 : : {
1024 : : // Remember: don't access any V8 stuff in worker thread
1025 : 294992 : result = disruptor->ProduceClaimManySync<AsyncArray<AsyncBuffer>, AsyncBuffer>(
1026 : 294992 : Env(), n, false, arg1, arg2, arg3);
1027 : 147496 : retry = result.Length() == 0;
1028 : 147496 : }
1029 : :
1030 : 147484 : void Retry() override
1031 : : {
1032 : 147484 : (new ProduceClaimManyAsyncWorker(disruptor, Callback().Value(), n))->Queue();
1033 : 147484 : }
1034 : :
1035 : : private:
1036 : : uint32_t n;
1037 : : };
1038 : :
1039 : 12 : void Disruptor::ProduceClaimManyAsync(const Napi::CallbackInfo& info,
1040 : : uint32_t n)
1041 : : {
1042 : 12 : (new ProduceClaimManyAsyncWorker(this, GetCallback(info, 1), n))->Queue();
1043 : 12 : }
1044 : :
1045 : 11 : void Disruptor::ProduceClaimManyAsync(const Napi::CallbackInfo& info)
1046 : : {
1047 : 11 : ProduceClaimManyAsync(info, info[0].As<Napi::Number>());
1048 : 11 : }
1049 : :
1050 : 40 : Napi::Value Disruptor::ProduceClaimMany(const Napi::CallbackInfo& info)
1051 : : {
1052 : 40 : uint32_t n = info[0].As<Napi::Number>();
1053 : : sequence_t seq_next, seq_next_end;
1054 : : bool all_ignored;
1055 : 40 : Napi::Array r = ProduceClaimManySync<Napi::Array, SyncBuffer>(
1056 : 40 : info.Env(), n, false, seq_next, seq_next_end, all_ignored);
1057 : :
1058 : 40 : if ((r.Length() > 0) || all_ignored || !spin)
1059 : : {
1060 : 39 : return r;
1061 : : }
1062 : :
1063 : 1 : ProduceClaimManyAsync(info, n);
1064 : 1 : return info.Env().Undefined();
1065 : : }
1066 : :
1067 : : template<typename Array, typename DisruptorBuffer>
1068 : 157001 : Array Disruptor::ProduceClaimAvailSync(const Napi::Env& env,
1069 : : const uint32_t max,
1070 : : const bool retry,
1071 : : sequence_t& out_next,
1072 : : sequence_t& out_next_end,
1073 : : bool& out_all_ignored)
1074 : : {
1075 : : bool all_ignored;
1076 : :
1077 : : do
1078 : : {
1079 : 157001 : sequence_t seq_next = __atomic_load_n(next, memorder);
1080 : 157001 : auto n = std::min(max, num_elements);
1081 : 157001 : all_ignored = true;
1082 : :
1083 : 420899 : for (uint32_t i = 0; i < num_consumers; ++i)
1084 : : {
1085 : 263898 : sequence_t seq_consumer = __atomic_load_n(&consumers[i], memorder);
1086 : 263898 : if (seq_consumer != sequence_max)
1087 : : {
1088 : 262890 : all_ignored = false;
1089 : 262890 : n = std::min(static_cast<sequence_t>(n), num_elements - (seq_next - seq_consumer));
1090 : : }
1091 : : }
1092 : :
1093 : 157001 : if (all_ignored)
1094 : : {
1095 : 5 : break;
1096 : : }
1097 : :
1098 : 159170 : if ((n > 0) &&
1099 : 2174 : __atomic_compare_exchange_n(next, &seq_next, seq_next + n, false, memorder, memorder))
1100 : : {
1101 : 2174 : Array r = Array::New(env);
1102 : 2174 : ProduceGetBuffers<Array, DisruptorBuffer>(env, seq_next, seq_next + n - 1, all_ignored, r);
1103 : 2174 : out_next = seq_next;
1104 : 2174 : out_next_end = seq_next + n - 1;
1105 : 2174 : out_all_ignored = all_ignored;
1106 : 2174 : return r;
1107 : 3 : }
1108 : : }
1109 : 154822 : while (retry);
1110 : :
1111 : 154827 : UpdateSeqNext(1, 0, all_ignored);
1112 : 154827 : out_next = 1;
1113 : 154827 : out_next_end = 0;
1114 : 154827 : out_all_ignored = all_ignored;
1115 : 154827 : return Array::New(env);
1116 : : }
1117 : :
1118 : 2 : Napi::Value Disruptor::ProduceClaimAvailSync(const Napi::CallbackInfo& info)
1119 : : {
1120 : : sequence_t seq_next, seq_next_end;
1121 : : bool all_ignored;
1122 : 4 : return ProduceClaimAvailSync<Napi::Array, SyncBuffer>(
1123 : 6 : info.Env(), info[0].As<Napi::Number>(), spin, seq_next, seq_next_end, all_ignored);
1124 : : }
1125 : :
1126 : : class ProduceClaimAvailAsyncWorker :
1127 : : public DisruptorAsyncWorker<AsyncArray<AsyncBuffer>,
1128 : : sequence_t,
1129 : : sequence_t,
1130 : : bool>
1131 : : {
1132 : : public:
1133 : 72687 : ProduceClaimAvailAsyncWorker(Disruptor *disruptor,
1134 : : const Napi::Function& callback,
1135 : 72687 : uint32_t max) :
1136 : : DisruptorAsyncWorker<AsyncArray<AsyncBuffer>,
1137 : : sequence_t,
1138 : : sequence_t,
1139 : : bool>(
1140 : : disruptor, callback),
1141 : 72687 : max(max)
1142 : : {
1143 : 72687 : arg1 = 1;
1144 : 72687 : arg2 = 0;
1145 : 72687 : arg3 = false;
1146 : 72687 : }
1147 : :
1148 : : protected:
1149 : 72687 : void Execute() override
1150 : : {
1151 : : // Remember: don't access any V8 stuff in worker thread
1152 : 145374 : result = disruptor->ProduceClaimAvailSync<AsyncArray<AsyncBuffer>, AsyncBuffer>(
1153 : 145374 : Env(), max, false, arg1, arg2, arg3);
1154 : 72687 : retry = result.Length() == 0;
1155 : 72687 : }
1156 : :
1157 : 72684 : void Retry() override
1158 : : {
1159 : 72684 : (new ProduceClaimAvailAsyncWorker(disruptor, Callback().Value(), max))->Queue();
1160 : 72684 : }
1161 : :
1162 : : private:
1163 : : uint32_t max;
1164 : : };
1165 : :
1166 : 3 : void Disruptor::ProduceClaimAvailAsync(const Napi::CallbackInfo& info,
1167 : : uint32_t max)
1168 : : {
1169 : 3 : (new ProduceClaimAvailAsyncWorker(this, GetCallback(info, 1), max))->Queue();
1170 : 3 : }
1171 : :
1172 : :
1173 : 2 : void Disruptor::ProduceClaimAvailAsync(const Napi::CallbackInfo& info)
1174 : : {
1175 : 2 : ProduceClaimAvailAsync(info, info[0].As<Napi::Number>());
1176 : 2 : }
1177 : :
1178 : 84312 : Napi::Value Disruptor::ProduceClaimAvail(const Napi::CallbackInfo& info)
1179 : : {
1180 : 84312 : uint32_t max = info[0].As<Napi::Number>();
1181 : : sequence_t seq_next, seq_next_end;
1182 : : bool all_ignored;
1183 : 84312 : Napi::Array r = ProduceClaimAvailSync<Napi::Array, SyncBuffer>(
1184 : 84312 : info.Env(), max, false, seq_next, seq_next_end, all_ignored);
1185 : :
1186 : 84312 : if ((r.Length() > 0) || all_ignored || !spin)
1187 : : {
1188 : 84311 : return r;
1189 : : }
1190 : :
1191 : 1 : ProduceClaimAvailAsync(info, max);
1192 : 1 : return info.Env().Undefined();
1193 : : }
1194 : :
1195 : 76 : Napi::Value Disruptor::ProduceRecover(const Napi::CallbackInfo& info)
1196 : : {
1197 : 76 : sequence_t seq_next = info[0].As<Napi::Number>().Int64Value();
1198 : 76 : sequence_t seq_next_end = info[1].As<Napi::Number>().Int64Value();
1199 : :
1200 : 76 : Napi::Array r = Napi::Array::New(info.Env());
1201 : :
1202 : 140 : if ((seq_next <= seq_next_end) &&
1203 : 116 : (__atomic_load_n(cursor, memorder) <= seq_next) &&
1204 : 40 : (__atomic_load_n(next, memorder) > seq_next_end))
1205 : : {
1206 : 20 : ProduceGetBuffers<Napi::Array, SyncBuffer>(
1207 : 40 : info.Env(), seq_next, seq_next_end, false, r);
1208 : : }
1209 : :
1210 : 76 : return r;
1211 : : }
1212 : :
1213 : : template<typename Boolean>
1214 : 2653230 : Boolean Disruptor::ProduceCommitSync(const Napi::Env& env,
1215 : : sequence_t seq_next,
1216 : : sequence_t seq_next_end,
1217 : : const bool retry)
1218 : : {
1219 : 2653230 : if (seq_next <= seq_next_end)
1220 : : {
1221 : : do
1222 : : {
1223 : 2653222 : sequence_t expected = seq_next;
1224 : 2653222 : if (__atomic_compare_exchange_n(cursor, &expected, seq_next_end + 1, false, memorder, memorder))
1225 : : {
1226 : 523228 : return Boolean::New(env, true);
1227 : : }
1228 : : }
1229 : 2129994 : while (retry);
1230 : : }
1231 : :
1232 : 2130002 : return Boolean::New(env, false);
1233 : : }
1234 : :
1235 : 995665 : void Disruptor::UpdateSeqNext(const sequence_t seq_next,
1236 : : const sequence_t seq_next_end,
1237 : : const bool all_ignored)
1238 : : {
1239 : 995665 : pending_seq_next = seq_next;
1240 : 995665 : pending_seq_next_end = seq_next_end;
1241 : 995665 : all_consumers_ignoring = all_ignored;
1242 : 995665 : }
1243 : :
1244 : 523240 : uint32_t Disruptor::GetSeqNext(const Napi::CallbackInfo& info,
1245 : : sequence_t& seq_next,
1246 : : sequence_t& seq_next_end)
1247 : : {
1248 : 523240 : if (info.Length() >= 2)
1249 : : {
1250 : 2177 : seq_next = info[0].As<Napi::Number>().Int64Value();
1251 : 2177 : seq_next_end = info[1].As<Napi::Number>().Int64Value();
1252 : 2177 : return 2;
1253 : : }
1254 : :
1255 : 521063 : seq_next = pending_seq_next;
1256 : 521063 : seq_next_end = pending_seq_next_end;
1257 : 521063 : return 0;
1258 : : }
1259 : :
1260 : 811 : Napi::Value Disruptor::ProduceCommitSync(const Napi::CallbackInfo& info)
1261 : : {
1262 : : sequence_t seq_next, seq_next_end;
1263 : 811 : GetSeqNext(info, seq_next, seq_next_end);
1264 : 811 : return ProduceCommitSync<Napi::Boolean>(info.Env(), seq_next, seq_next_end, spin);
1265 : : }
1266 : :
1267 : : class ProduceCommitAsyncWorker :
1268 : : public DisruptorAsyncWorker<AsyncBoolean>
1269 : : {
1270 : : public:
1271 : 2130801 : ProduceCommitAsyncWorker(Disruptor *disruptor,
1272 : : const Napi::Function& callback,
1273 : : sequence_t seq_next,
1274 : 2130801 : sequence_t seq_next_end) :
1275 : : DisruptorAsyncWorker<AsyncBoolean>(disruptor, callback),
1276 : 2130801 : seq_next(seq_next),
1277 : 2130801 : seq_next_end(seq_next_end)
1278 : : {
1279 : 2130801 : }
1280 : :
1281 : : protected:
1282 : 2130801 : void Execute() override
1283 : : {
1284 : : // Remember: don't access any V8 stuff in worker thread
1285 : 2130801 : result = disruptor->ProduceCommitSync<AsyncBoolean>(Env(), seq_next, seq_next_end, false);
1286 : 2130801 : retry = !result;
1287 : 2130801 : }
1288 : :
1289 : 2003078 : void Retry() override
1290 : : {
1291 : 2003078 : (new ProduceCommitAsyncWorker(disruptor, Callback().Value(), seq_next, seq_next_end))->Queue();
1292 : 2003078 : }
1293 : :
1294 : : private:
1295 : : sequence_t seq_next, seq_next_end;
1296 : : };
1297 : :
1298 : 127723 : void Disruptor::ProduceCommitAsync(const Napi::CallbackInfo& info,
1299 : : sequence_t seq_next,
1300 : : sequence_t seq_next_end,
1301 : : uint32_t cb_arg)
1302 : : {
1303 : : (new ProduceCommitAsyncWorker(
1304 : 127723 : this, GetCallback(info, cb_arg), seq_next, seq_next_end))->Queue();
1305 : 127723 : }
1306 : :
1307 : 811 : void Disruptor::ProduceCommitAsync(const Napi::CallbackInfo& info)
1308 : : {
1309 : : sequence_t seq_next, seq_next_end;
1310 : 811 : uint32_t cb_arg = GetSeqNext(info, seq_next, seq_next_end);
1311 : 811 : ProduceCommitAsync(info, seq_next, seq_next_end, cb_arg);
1312 : 811 : }
1313 : :
1314 : 521618 : Napi::Value Disruptor::ProduceCommit(const Napi::CallbackInfo& info)
1315 : : {
1316 : : sequence_t seq_next, seq_next_end;
1317 : 521618 : uint32_t cb_arg = GetSeqNext(info, seq_next, seq_next_end);
1318 : :
1319 : 521618 : Napi::Boolean r = ProduceCommitSync<Napi::Boolean>(info.Env(), seq_next, seq_next_end, false);
1320 : 521618 : if (r || !spin)
1321 : : {
1322 : 394706 : return r;
1323 : : }
1324 : :
1325 : 126912 : ProduceCommitAsync(info, seq_next, seq_next_end, cb_arg);
1326 : 126912 : return info.Env().Undefined();
1327 : : }
1328 : :
1329 : 12 : Napi::Value Disruptor::GetConsumers(const Napi::CallbackInfo&)
1330 : : {
1331 : 12 : return consumers_buffer_ref.Value();
1332 : : }
1333 : :
1334 : 222 : Napi::Value Disruptor::GetCursor(const Napi::CallbackInfo& info)
1335 : : {
1336 : 222 : return Napi::Number::New(info.Env(), __atomic_load_n(cursor, memorder));
1337 : : }
1338 : :
1339 : 222 : Napi::Value Disruptor::GetNext(const Napi::CallbackInfo& info)
1340 : : {
1341 : 222 : return Napi::Number::New(info.Env(), __atomic_load_n(next, memorder));
1342 : : }
1343 : :
1344 : 1556 : Napi::Value Disruptor::GetElements(const Napi::CallbackInfo&)
1345 : : {
1346 : 1556 : return elements_buffer_ref.Value();
1347 : : }
1348 : :
1349 : 222 : Napi::Value Disruptor::GetConsumer(const Napi::CallbackInfo& info)
1350 : : {
1351 : 222 : return Napi::Number::New(info.Env(), __atomic_load_n(ptr_consumer, memorder));
1352 : : }
1353 : :
1354 : 7074962 : Napi::Value Disruptor::GetPendingSeqConsumer(const Napi::CallbackInfo& info)
1355 : : {
1356 : 7074962 : return Napi::Number::New(info.Env(), pending_seq_consumer);
1357 : : }
1358 : :
1359 : 87 : Napi::Value Disruptor::GetPendingSeqCursor(const Napi::CallbackInfo& info)
1360 : : {
1361 : 87 : return Napi::Number::New(info.Env(), pending_seq_cursor);
1362 : : }
1363 : :
1364 : 604663 : Napi::Value Disruptor::GetPendingSeqNext(const Napi::CallbackInfo& info)
1365 : : {
1366 : 604663 : return Napi::Number::New(info.Env(), pending_seq_next);
1367 : : }
1368 : :
1369 : 604663 : Napi::Value Disruptor::GetPendingSeqNextEnd(const Napi::CallbackInfo& info)
1370 : : {
1371 : 604663 : return Napi::Number::New(info.Env(), pending_seq_next_end);
1372 : : }
1373 : :
1374 : 604649 : Napi::Value Disruptor::GetAllConsumersIgnoring(const Napi::CallbackInfo& info)
1375 : : {
1376 : 604649 : return Napi::Boolean::New(info.Env(), all_consumers_ignoring);
1377 : : }
1378 : :
1379 : 2104 : Napi::Value Disruptor::GetElementSize(const Napi::CallbackInfo& info)
1380 : : {
1381 : 2104 : return Napi::Number::New(info.Env(), element_size);
1382 : : }
1383 : :
1384 : 2026 : Napi::Value Disruptor::GetSpin(const Napi::CallbackInfo& info)
1385 : : {
1386 : 2026 : return Napi::Boolean::New(info.Env(), spin);
1387 : : }
1388 : :
1389 : 12036314 : Napi::Value Disruptor::GetStatus(const Napi::CallbackInfo& info)
1390 : : {
1391 : 12036314 : return Napi::Number::New(info.Env(), __atomic_load_n(status, memorder));
1392 : : }
1393 : :
1394 : 11 : void Disruptor::SetStatus(const Napi::CallbackInfo&, const Napi::Value& value)
1395 : : {
1396 : 11 : __atomic_store_n(status, value.As<Napi::Number>(), memorder);
1397 : 11 : }
1398 : :
1399 : 506 : Napi::Object Disruptor::Initialize(Napi::Env env, Napi::Object exports)
1400 : : {
1401 : : {
1402 : 506 : std::lock_guard<std::mutex> lock(buffers_mutex);
1403 : 506 : if (!buffers) {
1404 : 254 : buffers = new std::unordered_set<uint8_t*>();
1405 : : }
1406 : 506 : }
1407 : :
1408 : 506 : exports.Set("Disruptor", DefineClass(env, "Disruptor",
1409 : : {
1410 : 506 : InstanceMethod<&Disruptor::ProduceClaim>("produceClaim"),
1411 : 506 : InstanceMethod<&Disruptor::ProduceClaimSync>("produceClaimSync"),
1412 : 506 : InstanceMethod<&Disruptor::ProduceClaimMany>("produceClaimMany"),
1413 : 506 : InstanceMethod<&Disruptor::ProduceClaimManySync>("produceClaimManySync"),
1414 : 506 : InstanceMethod<&Disruptor::ProduceClaimAvail>("produceClaimAvail"),
1415 : 506 : InstanceMethod<&Disruptor::ProduceClaimAvailSync>("produceClaimAvailSync"),
1416 : 506 : InstanceMethod<&Disruptor::ProduceCommit>("produceCommit"),
1417 : 506 : InstanceMethod<&Disruptor::ProduceCommitSync>("produceCommitSync"),
1418 : 506 : InstanceMethod<&Disruptor::ProduceRecover>("produceRecover"),
1419 : 506 : InstanceMethod<&Disruptor::ConsumeNew>("consumeNew"),
1420 : 506 : InstanceMethod<&Disruptor::ConsumeNewSync>("consumeNewSync"),
1421 : 506 : InstanceMethod<&Disruptor::ConsumeCommit>("consumeCommit"),
1422 : 506 : InstanceMethod<&Disruptor::Release>("release"),
1423 : 506 : InstanceAccessor<&Disruptor::GetPendingSeqConsumer>("prevConsumeStart"),
1424 : 506 : InstanceAccessor<&Disruptor::GetPendingSeqNext>("prevClaimStart"),
1425 : 506 : InstanceAccessor<&Disruptor::GetPendingSeqNextEnd>("prevClaimEnd"),
1426 : 506 : InstanceAccessor<&Disruptor::GetAllConsumersIgnoring>("allConsumersIgnoring"),
1427 : 506 : InstanceAccessor<&Disruptor::GetElementSize>("elementSize"),
1428 : 506 : InstanceAccessor<&Disruptor::GetSpin>("spin"),
1429 : 506 : InstanceAccessor<&Disruptor::GetStatus, &Disruptor::SetStatus>("status"),
1430 : :
1431 : : // For testing only
1432 : 506 : InstanceAccessor<&Disruptor::GetConsumers>("consumers"),
1433 : 506 : InstanceAccessor<&Disruptor::GetCursor>("cursor"),
1434 : 506 : InstanceAccessor<&Disruptor::GetNext>("next"),
1435 : 506 : InstanceAccessor<&Disruptor::GetElements>("elements"),
1436 : 506 : InstanceAccessor<&Disruptor::GetConsumer>("consumer"),
1437 : 506 : InstanceAccessor<&Disruptor::GetPendingSeqCursor>("prevConsumeNext"),
1438 : 506 : InstanceMethod<&Disruptor::ConsumeNewAsync>("consumeNewAsync"),
1439 : 506 : InstanceMethod<&Disruptor::ProduceClaimAsync>("produceClaimAsync"),
1440 : 506 : InstanceMethod<&Disruptor::ProduceClaimManyAsync>("produceClaimManyAsync"),
1441 : 506 : InstanceMethod<&Disruptor::ProduceClaimAvailAsync>("produceClaimAvailAsync"),
1442 : 506 : InstanceMethod<&Disruptor::ProduceCommitAsync>("produceCommitAsync")
1443 : : }));
1444 : :
1445 : 506 : return exports;
1446 : : }
1447 : :
1448 : 506 : Napi::Object Initialize(Napi::Env env, Napi::Object exports)
1449 : : {
1450 : 506 : return Disruptor::Initialize(env, exports);
1451 : : }
1452 : :
1453 : 760 : NODE_API_MODULE(disruptor, Initialize)
|
