<commit_before>import assert from 'power-assert'
import {arrayPool} from './pools'
import {TimeoutChan} from './special-chans'
import {CLOSED, FAILED, ERROR} from './constants'

const EMPTY = []


// TODO: create custom chan that knows how to calculate canTakeSync,
//       takeSync() and propagate maybeCanTakeSync() to srcs


export function mergeTo(dst, srcs, closeDst) {
  let timeoutSrcs = EMPTY
  let dataSrcs = arrayPool.take()

  for (let i = 0; i < srcs.length; ++i) {
    let chan = srcs[i]
    if (!chan.isClosed) {
      let src = withHandlers(onMaybeCanTakeSync, onSrcClosed, { chan,
        subscribed: false,
        onMaybeCanTakeSync: undefined,
        onClosed: undefined
      })
      if (chan instanceof TimeoutChan) {
        if (timeoutSrcs === EMPTY) timeoutSrcs = arrayPool.take()
        timeoutSrcs.push(src)
      } else {
        dataSrcs.push(src)
        chan.on('closed', src.onClosed)
      }
    }
  }

  let syncSrcs = arrayPool.take()
  let totalDataSrcs = dataSrcs.length
  let totalTimeoutSrcs = timeoutSrcs.length
  let inSyncRegion = false

  maybeSendNext(undefined)
  return dst

  function takeNextSync(clearState) {
    for (let i = 0; i < totalTimeoutSrcs; ++i) {
      let {chan} = timeoutSrcs[i]
      if (chan.canTakeSync) {
        return chan._takeSync()
      }
    }
    let totalSyncSrcs
    if (clearState) {
      // clearState equals true when takeNextSync is running the first time during
      // the current event loop tick
      syncSrcs.length = 0
      let i = 0; while (i < totalDataSrcs) {
        let src = dataSrcs[i], {chan} = src
        if (chan.isClosed) {
          dataSrcs.splice(i, 1)
          --totalDataSrcs
        } else {
          if (chan.canTakeSync) {
            syncSrcs.push(src)
          }
          ++i
        }
      }
      totalSyncSrcs = syncSrcs.length
    } else {
      // if we're running in the same tick as previous takeNextSync call, no channel
      // that was empty in the previous call can become non-empty; but the other way
      // is certainly possible
      totalSyncSrcs = syncSrcs.length
      let i = 0; while (i < totalSyncSrcs) {
        let src = syncSrcs[i], {chan} = src
        if (chan.canTakeSync) {
          ++i
        } else {
          syncSrcs.splice(i, 1); --totalSyncSrcs
        }
      }
    }
    if (totalSyncSrcs) {
      let i = (totalSyncSrcs == 1) ? 0 : Math.floor(totalSyncSrcs * Math.random())
      let ch = syncSrcs[i].chan
      let result = ch._takeSync()
      assert(result !== false)
      return result === ERROR ? result : ch.value
    }
    return FAILED
  }

  function maybeSendNext(syncSrc) {
    let canSendSync = dst.canSendSync
    let canTakeMore = true
    let clearSyncState = true
    inSyncRegion = true
    while (canSendSync && canTakeMore) {
      let syncResult; if (syncSrc) {
        syncResult = syncSrc.chan._takeSync()
        assert(syncResult !== false)
        if (syncResult === true) {
          syncResult = syncSrc.chan.value
        }
        syncSrc = undefined
      } else {
        syncResult = takeNextSync(clearSyncState)
        clearSyncState = false
      }
      if (syncResult === FAILED) {
        canTakeMore = false
      } else {
        if (syncResult === ERROR) {
          syncResult = dst._sendSync(ERROR.value, true)
          assert(syncResult === true)
        } else {
          syncResult = dst._sendSync(syncResult, false)
          assert(syncResult === true)
        }
        canSendSync = dst.canSendSync
      }
    }
    inSyncRegion = false
    assert(canSendSync || canTakeMore)
    if (!totalDataSrcs && !totalTimeoutSrcs) {
      // no srcs left alive => end merging
      end()
    } else if (canTakeMore) {
      // dst can't accept more data synchronously => wait until it can, then resume
      dst._maybeCanSendSync(onMaybeCanSendSync, false)
    } else {
      // dst can accept more data, but srcs can't provide it => wait until they can
      totalTimeoutSrcs && subscribeForSrcs(timeoutSrcs)
      totalDataSrcs && subscribeForSrcs(dataSrcs)
    }
  }

  function onMaybeCanSendSync(dstAlive) {
    if (dataSrcs === EMPTY) {
      return // already ended
    } else if (dstAlive) {
      maybeSendNext(undefined)
    } else {
      end()
    }
  }

  function onMaybeCanTakeSync(src, srcAlive) {
    if (dataSrcs === EMPTY || !srcAlive) {
      return // already ended
    }
    src.subscribed = false
    maybeSendNext(src.chan.canTakeSync ? src : undefined)
  }

  function onSrcClosed(src) {
    if (dataSrcs === EMPTY) {
      return // already ended
    }
    src.chan.removeListener('closed', src.onClosed)
    let index = dataSrcs.indexOf(src)
    assert(index >= 0)
    dataSrcs.splice(index, 1)
    if (!--totalDataSrcs && !totalTimeoutSrcs && !inSyncRegion) {
      end()
    }
  }

  function end() {
    assert(dataSrcs !== EMPTY)
    for (let i = 0; i < totalDataSrcs; ++i) {
      let src = dataSrcs[i]
      src.chan.removeListener('closed', src.onClosed)
    }
    arrayPool.put(dataSrcs)
    dataSrcs = EMPTY
    if (timeoutSrcs !== EMPTY) {
      arrayPool.put(timeoutSrcs)
    }
    arrayPool.put(syncSrcs)
    closeDst && dst.close()
  }
}

function subscribeForSrcs(srcs) {
  for (let i = 0; i < srcs.length; ++i) {
    let src = srcs[i]
    if (!src.subscribed) {
      src.subscribed = true
      src.chan._maybeCanTakeSync(src.onMaybeCanTakeSync, false)
    }
  }
}

function withHandlers(onMaybeCanTakeSync, onSrcClosed, src) {
  src.onMaybeCanTakeSync = isClosed => onMaybeCanTakeSync(src, isClosed)
  src.onClosed = () => onSrcClosed(src)
  return src
}
<commit_msg>Handle the case when merge output is closed at the time of call<commit_after>import assert from 'power-assert'
import {arrayPool} from './pools'
import {TimeoutChan} from './special-chans'
import {CLOSED, FAILED, ERROR} from './constants'

const EMPTY = []


// TODO: create custom chan that knows how to calculate canTakeSync,
//       takeSync() and propagate maybeCanTakeSync() to srcs


export function mergeTo(dst, srcs, closeDst) {
  if (dst.isClosed) {
    return
  }

  let timeoutSrcs = EMPTY
  let dataSrcs = arrayPool.take()

  for (let i = 0; i < srcs.length; ++i) {
    let chan = srcs[i]
    if (!chan.isClosed) {
      let src = withHandlers(onMaybeCanTakeSync, onSrcClosed, { chan,
        subscribed: false,
        onMaybeCanTakeSync: undefined,
        onClosed: undefined
      })
      if (chan instanceof TimeoutChan) {
        if (timeoutSrcs === EMPTY) timeoutSrcs = arrayPool.take()
        timeoutSrcs.push(src)
      } else {
        dataSrcs.push(src)
        chan.on('closed', src.onClosed)
      }
    }
  }

  let syncSrcs = arrayPool.take()
  let totalDataSrcs = dataSrcs.length
  let totalTimeoutSrcs = timeoutSrcs.length
  let inSyncRegion = false

  maybeSendNext(undefined)
  return dst

  function takeNextSync(clearState) {
    for (let i = 0; i < totalTimeoutSrcs; ++i) {
      let {chan} = timeoutSrcs[i]
      if (chan.canTakeSync) {
        return chan._takeSync()
      }
    }
    let totalSyncSrcs
    if (clearState) {
      // clearState equals true when takeNextSync is running the first time during
      // the current event loop tick
      syncSrcs.length = 0
      let i = 0; while (i < totalDataSrcs) {
        let src = dataSrcs[i], {chan} = src
        if (chan.isClosed) {
          dataSrcs.splice(i, 1)
          --totalDataSrcs
        } else {
          if (chan.canTakeSync) {
            syncSrcs.push(src)
          }
          ++i
        }
      }
      totalSyncSrcs = syncSrcs.length
    } else {
      // if we're running in the same tick as previous takeNextSync call, no channel
      // that was empty in the previous call can become non-empty; but the other way
      // is certainly possible
      totalSyncSrcs = syncSrcs.length
      let i = 0; while (i < totalSyncSrcs) {
        let src = syncSrcs[i], {chan} = src
        if (chan.canTakeSync) {
          ++i
        } else {
          syncSrcs.splice(i, 1); --totalSyncSrcs
        }
      }
    }
    if (totalSyncSrcs) {
      let i = (totalSyncSrcs == 1) ? 0 : Math.floor(totalSyncSrcs * Math.random())
      let ch = syncSrcs[i].chan
      let result = ch._takeSync()
      assert(result !== false)
      return result === ERROR ? result : ch.value
    }
    return FAILED
  }

  function maybeSendNext(syncSrc) {
    let canSendSync = dst.canSendSync
    let canTakeMore = true
    let clearSyncState = true
    inSyncRegion = true
    while (canSendSync && canTakeMore) {
      let syncResult; if (syncSrc) {
        syncResult = syncSrc.chan._takeSync()
        assert(syncResult !== false)
        if (syncResult === true) {
          syncResult = syncSrc.chan.value
        }
        syncSrc = undefined
      } else {
        syncResult = takeNextSync(clearSyncState)
        clearSyncState = false
      }
      if (syncResult === FAILED) {
        canTakeMore = false
      } else {
        if (syncResult === ERROR) {
          syncResult = dst._sendSync(ERROR.value, true)
          assert(syncResult === true)
        } else {
          syncResult = dst._sendSync(syncResult, false)
          assert(syncResult === true)
        }
        canSendSync = dst.canSendSync
      }
    }
    inSyncRegion = false
    assert(canSendSync || canTakeMore)
    if (!totalDataSrcs && !totalTimeoutSrcs) {
      // no srcs left alive => end merging
      end()
    } else if (canTakeMore) {
      // dst can't accept more data synchronously => wait until it can, then resume
      dst._maybeCanSendSync(onMaybeCanSendSync, false)
    } else {
      // dst can accept more data, but srcs can't provide it => wait until they can
      totalTimeoutSrcs && subscribeForSrcs(timeoutSrcs)
      totalDataSrcs && subscribeForSrcs(dataSrcs)
    }
  }

  function onMaybeCanSendSync(dstAlive) {
    if (dataSrcs === EMPTY) {
      return // already ended
    } else if (dstAlive) {
      maybeSendNext(undefined)
    } else {
      end()
    }
  }

  function onMaybeCanTakeSync(src, srcAlive) {
    if (dataSrcs === EMPTY || !srcAlive) {
      return // already ended
    }
    src.subscribed = false
    maybeSendNext(src.chan.canTakeSync ? src : undefined)
  }

  function onSrcClosed(src) {
    if (dataSrcs === EMPTY) {
      return // already ended
    }
    src.chan.removeListener('closed', src.onClosed)
    let index = dataSrcs.indexOf(src)
    assert(index >= 0)
    dataSrcs.splice(index, 1)
    if (!--totalDataSrcs && !totalTimeoutSrcs && !inSyncRegion) {
      end()
    }
  }

  function end() {
    assert(dataSrcs !== EMPTY)
    for (let i = 0; i < totalDataSrcs; ++i) {
      let src = dataSrcs[i]
      src.chan.removeListener('closed', src.onClosed)
    }
    arrayPool.put(dataSrcs)
    dataSrcs = EMPTY
    if (timeoutSrcs !== EMPTY) {
      arrayPool.put(timeoutSrcs)
    }
    arrayPool.put(syncSrcs)
    closeDst && dst.close()
  }
}

function subscribeForSrcs(srcs) {
  for (let i = 0; i < srcs.length; ++i) {
    let src = srcs[i]
    if (!src.subscribed) {
      src.subscribed = true
      src.chan._maybeCanTakeSync(src.onMaybeCanTakeSync, false)
    }
  }
}

function withHandlers(onMaybeCanTakeSync, onSrcClosed, src) {
  src.onMaybeCanTakeSync = isClosed => onMaybeCanTakeSync(src, isClosed)
  src.onClosed = () => onSrcClosed(src)
  return src
}

{'fraction_non_alphanumeric': 0.10978625684508038, 'fraction_numerical': 0.0022964140611199434, 'mean_word_length': 2.3302941176470586, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 15, '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': '172321', 'n_tokens_mistral': 3725, 'n_tokens_neox': 3622, 'n_words': 915}