<commit_before>import {slice} from "./array";
import noop from "./noop";

var noabort = {},
    success = [null];

function newQueue(concurrency) {
  if (!(concurrency >= 1)) throw new Error;

  var q,
      tasks = [],
      results = [],
      waiting = 0,
      active = 0,
      ended = 0,
      starting, // inside a synchronous task callback?
      error,
      callback = noop,
      callbackAll = true;

  function start() {
    if (starting) return; // let the current task complete
    while (starting = waiting && active < concurrency) {
      var i = ended + active,
          t = tasks[i],
          j = t.length - 1,
          c = t[j];
      t[j] = end(i);
      --waiting, ++active;
      try { t = c.apply(null, t); }
      catch (e) { if (tasks[i]) return abort(e); } // threw before finishing?
      if (!tasks[i]) continue; // task finished synchronously
      tasks[i] = t || noabort;
    }
  }

  function end(i) {
    return function(e, r) {
      if (!tasks[i]) return; // ignore multiple callbacks
      --active, ++ended;
      tasks[i] = null;
      if (error != null) return; // only report the first error
      if (e != null) {
        abort(e);
      } else {
        results[i] = r;
        if (waiting) start();
        else if (!active) notify();
      }
    };
  }

  function abort(e) {
    var i = ended + active, t;
    error = e; // ignore new tasks and squelch active callbacks
    waiting = active = NaN; // prevent starting and allow notification
    while (--i >= 0) if ((t = tasks[i]) && t.abort) t.abort();
    notify();
  }

  function notify() {
    if (error != null) callback(error);
    else if (callbackAll) callback(null, results);
    else callback.apply(null, success.concat(results));
  }

  return q = {
    defer: function(f) {
      if (callback !== noop) throw new Error;
      var t = slice.call(arguments, 1);
      t.push(f);
      ++waiting, tasks.push(t);
      start();
      return q;
    },
    abort: function() {
      if (error == null) abort(new Error("abort"));
      return q;
    },
    await: function(f) {
      if (callback !== noop) throw new Error;
      callback = f, callbackAll = false;
      if (!waiting && !active) notify();
      return q;
    },
    awaitAll: function(f) {
      if (callback !== noop) throw new Error;
      callback = f, callbackAll = true;
      if (!waiting && !active) notify();
      return q;
    }
  };
}

export default function(concurrency) {
  return newQueue(arguments.length ? +concurrency : Infinity);
}
<commit_msg>Ignore errors in task.abort.<commit_after>import {slice} from "./array";
import noop from "./noop";

var noabort = {},
    success = [null];

function newQueue(concurrency) {
  if (!(concurrency >= 1)) throw new Error;

  var q,
      tasks = [],
      results = [],
      waiting = 0,
      active = 0,
      ended = 0,
      starting, // inside a synchronous task callback?
      error,
      callback = noop,
      callbackAll = true;

  function start() {
    if (starting) return; // let the current task complete
    while (starting = waiting && active < concurrency) {
      var i = ended + active,
          t = tasks[i],
          j = t.length - 1,
          c = t[j];
      t[j] = end(i);
      --waiting, ++active;
      try { t = c.apply(null, t); }
      catch (e) { if (tasks[i]) return abort(e); } // threw before finishing?
      if (!tasks[i]) continue; // task finished synchronously
      tasks[i] = t || noabort;
    }
  }

  function end(i) {
    return function(e, r) {
      if (error != null || !tasks[i]) return; // ignore multiple callbacks
      if (e != null) {
        abort(e);
      } else {
        --active, ++ended;
        tasks[i] = null;
        results[i] = r;
        if (waiting) start();
        else if (!active) notify();
      }
    };
  }

  function abort(e) {
    var i = ended + active, t;
    error = e; // ignore active callbacks
    waiting = active = NaN; // prevent starting and allow notification

    while (--i >= 0) {
      if ((t = tasks[i]) && t.abort) {
        try { t.abort(); }
        catch (e) { /* ignore */ }
      }
    }

    notify();
  }

  function notify() {
    if (error != null) callback(error);
    else if (callbackAll) callback(null, results);
    else callback.apply(null, success.concat(results));
  }

  return q = {
    defer: function(f) {
      if (callback !== noop) throw new Error;
      var t = slice.call(arguments, 1);
      t.push(f);
      ++waiting, tasks.push(t);
      start();
      return q;
    },
    abort: function() {
      if (error == null) abort(new Error("abort"));
      return q;
    },
    await: function(f) {
      if (callback !== noop) throw new Error;
      callback = f, callbackAll = false;
      if (!waiting && !active) notify();
      return q;
    },
    awaitAll: function(f) {
      if (callback !== noop) throw new Error;
      callback = f, callbackAll = true;
      if (!waiting && !active) notify();
      return q;
    }
  };
}

export default function(concurrency) {
  return newQueue(arguments.length ? +concurrency : Infinity);
}

{'fraction_non_alphanumeric': 0.15250245338567223, 'fraction_numerical': 0.002747791952894995, 'mean_word_length': 2.101643335362142, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1245890', 'n_tokens_mistral': 1623, 'n_tokens_neox': 1605, 'n_words': 496}