Farm.js 4.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207
  1. 'use strict';
  2. Object.defineProperty(exports, '__esModule', {
  3. value: true
  4. });
  5. exports.default = void 0;
  6. var _FifoQueue = _interopRequireDefault(require('./FifoQueue'));
  7. var _types = require('./types');
  8. function _interopRequireDefault(obj) {
  9. return obj && obj.__esModule ? obj : {default: obj};
  10. }
  11. function _defineProperty(obj, key, value) {
  12. if (key in obj) {
  13. Object.defineProperty(obj, key, {
  14. value: value,
  15. enumerable: true,
  16. configurable: true,
  17. writable: true
  18. });
  19. } else {
  20. obj[key] = value;
  21. }
  22. return obj;
  23. }
  24. class Farm {
  25. constructor(_numOfWorkers, _callback, options = {}) {
  26. var _options$workerSchedu, _options$taskQueue;
  27. this._numOfWorkers = _numOfWorkers;
  28. this._callback = _callback;
  29. _defineProperty(this, '_computeWorkerKey', void 0);
  30. _defineProperty(this, '_workerSchedulingPolicy', void 0);
  31. _defineProperty(this, '_cacheKeys', Object.create(null));
  32. _defineProperty(this, '_locks', []);
  33. _defineProperty(this, '_offset', 0);
  34. _defineProperty(this, '_taskQueue', void 0);
  35. this._computeWorkerKey = options.computeWorkerKey;
  36. this._workerSchedulingPolicy =
  37. (_options$workerSchedu = options.workerSchedulingPolicy) !== null &&
  38. _options$workerSchedu !== void 0
  39. ? _options$workerSchedu
  40. : 'round-robin';
  41. this._taskQueue =
  42. (_options$taskQueue = options.taskQueue) !== null &&
  43. _options$taskQueue !== void 0
  44. ? _options$taskQueue
  45. : new _FifoQueue.default();
  46. }
  47. doWork(method, ...args) {
  48. const customMessageListeners = new Set();
  49. const addCustomMessageListener = listener => {
  50. customMessageListeners.add(listener);
  51. return () => {
  52. customMessageListeners.delete(listener);
  53. };
  54. };
  55. const onCustomMessage = message => {
  56. customMessageListeners.forEach(listener => listener(message));
  57. };
  58. const promise = new Promise( // Bind args to this function so it won't reference to the parent scope.
  59. // This prevents a memory leak in v8, because otherwise the function will
  60. // retaine args for the closure.
  61. ((args, resolve, reject) => {
  62. const computeWorkerKey = this._computeWorkerKey;
  63. const request = [_types.CHILD_MESSAGE_CALL, false, method, args];
  64. let worker = null;
  65. let hash = null;
  66. if (computeWorkerKey) {
  67. hash = computeWorkerKey.call(this, method, ...args);
  68. worker = hash == null ? null : this._cacheKeys[hash];
  69. }
  70. const onStart = worker => {
  71. if (hash != null) {
  72. this._cacheKeys[hash] = worker;
  73. }
  74. };
  75. const onEnd = (error, result) => {
  76. customMessageListeners.clear();
  77. if (error) {
  78. reject(error);
  79. } else {
  80. resolve(result);
  81. }
  82. };
  83. const task = {
  84. onCustomMessage,
  85. onEnd,
  86. onStart,
  87. request
  88. };
  89. if (worker) {
  90. this._taskQueue.enqueue(task, worker.getWorkerId());
  91. this._process(worker.getWorkerId());
  92. } else {
  93. this._push(task);
  94. }
  95. }).bind(null, args)
  96. );
  97. promise.UNSTABLE_onCustomMessage = addCustomMessageListener;
  98. return promise;
  99. }
  100. _process(workerId) {
  101. if (this._isLocked(workerId)) {
  102. return this;
  103. }
  104. const task = this._taskQueue.dequeue(workerId);
  105. if (!task) {
  106. return this;
  107. }
  108. if (task.request[1]) {
  109. throw new Error('Queue implementation returned processed task');
  110. } // Reference the task object outside so it won't be retained by onEnd,
  111. // and other properties of the task object, such as task.request can be
  112. // garbage collected.
  113. const taskOnEnd = task.onEnd;
  114. const onEnd = (error, result) => {
  115. taskOnEnd(error, result);
  116. this._unlock(workerId);
  117. this._process(workerId);
  118. };
  119. task.request[1] = true;
  120. this._lock(workerId);
  121. this._callback(
  122. workerId,
  123. task.request,
  124. task.onStart,
  125. onEnd,
  126. task.onCustomMessage
  127. );
  128. return this;
  129. }
  130. _push(task) {
  131. this._taskQueue.enqueue(task);
  132. const offset = this._getNextWorkerOffset();
  133. for (let i = 0; i < this._numOfWorkers; i++) {
  134. this._process((offset + i) % this._numOfWorkers);
  135. if (task.request[1]) {
  136. break;
  137. }
  138. }
  139. return this;
  140. }
  141. _getNextWorkerOffset() {
  142. switch (this._workerSchedulingPolicy) {
  143. case 'in-order':
  144. return 0;
  145. case 'round-robin':
  146. return this._offset++;
  147. }
  148. }
  149. _lock(workerId) {
  150. this._locks[workerId] = true;
  151. }
  152. _unlock(workerId) {
  153. this._locks[workerId] = false;
  154. }
  155. _isLocked(workerId) {
  156. return this._locks[workerId];
  157. }
  158. }
  159. exports.default = Farm;