<commit_before>local socket = require'socket'
assert(socket._VERSION:match('^LuaSocket 3%.'))
local emitter = require'emitter'
local ev = require'ev'

local isip = function(ip)
  local addrinfo,err = socket.dns.getaddrinfo(ip)
  if err then
    return false
  end
  return true
end

local isipv6 = function(ip)
  local addrinfo,err = socket.dns.getaddrinfo(ip)
  if addrinfo then
    assert(#addrinfo > 0)
    if addrinfo[1].family == 'inet6' then
      return true
    end
  end
  return false
end

local isipv4 = function(ip)
  return isip(ip) and not isipv6(ip)
end

local new = function()
  local self = emitter.new()
  local sock
  local loop = ev.Loop.default
  local connecting = false
  local connected = false
  local closing = false
  local watchers = {}
  
  local on_error = function(err)
    if err ~= 'closed' then
      self:emit('error',err)
    end
    self:emit('close')
    self:destroy()
  end
  
  local read_io = function()
    assert(sock)
    return ev.IO.new(function()
        local data,err,part = sock:receive(8192)
        data = part or data
        if data then
          if watchers.timer then
            watchers.timer:again(loop)
          end
          self:emit('data',data or part)
        end
        if err and err ~= 'timeout' then
          on_error(err)
        end
      end,sock:getfd(),ev.READ)
  end
  
  local pending
  local pos
  
  local write_io = function()
    assert(sock)
    return ev.IO.new(function(loop,io)
        local sent,err,so_far = sock:send(pending,pos)
        if not sent and err ~= 'timeout' then
          io:stop(loop)
          on_error(err)
        elseif sent then
          pos = nil
          pending = nil
          io:stop(loop)
          self:emit('_drain')
          self:emit('drain')
        else
          pos = so_far + 1
        end
        if watchers.timer then
          watchers.timer:again(loop)
        end
      end,sock:getfd(),ev.WRITE)
  end
  
  local on_connect = function()
    connecting = false
    connected = true
    watchers.read = read_io()
    watchers.write = write_io()
    self:resume()
    self:emit('connect',self)
  end
  
  self.connect = function(_,port,ip)
    ip = ip or '127.0.0.1'
    if not isip(ip) then
      on_error(err)
    end
    if sock and closing then
      self:once('close',function(self)
          self:_connect(port,ip)
        end)
      
    elseif not connecting then
      self:_connect(port,ip)
    end
  end
  
  self._connect = function(_,port,ip)
    assert(not sock)
    if isipv6(ip) then
      sock = socket.tcp6()
    else
      sock = socket.tcp()
    end
    sock:settimeout(0)
    connecting = true
    closing = false
    local ok,err = sock:connect(ip,port)
    if ok or err == 'already connected' then
      on_connect()
    elseif err == 'timeout' or err == 'Operation already in progress' then
      watchers.connect = ev.IO.new(function(loop,io)
          local ok,err = sock:connect(ip,port)
          if ok or err == 'already connected' then
            io:stop(loop)
            watchers.connect = nil
            on_connect()
          else
            on_error(err)
          end
        end,sock:getfd(),ev.WRITE)
      watchers.connect:start(loop)
    else
      on_error(err)
    end
  end
  
  self._transfer = function(_,s)
    sock = s
    on_connect()
  end
  
  self.write = function(_,data)
    if pending then
      pending = pending..data
    else
      pending = data
      if connecting then
        self:once('connect',function()
            watchers.write:start(loop)
          end)
      elseif connected then
        watchers.write:start(loop)
      else
        self:emit('error',err)<commit_msg>add make sock non-blocking in _transfer<commit_after>local socket = require'socket'
assert(socket._VERSION:match('^LuaSocket 3%.'))
local emitter = require'emitter'
local ev = require'ev'

local isip = function(ip)
  local addrinfo,err = socket.dns.getaddrinfo(ip)
  if err then
    return false
  end
  return true
end

local isipv6 = function(ip)
  local addrinfo,err = socket.dns.getaddrinfo(ip)
  if addrinfo then
    assert(#addrinfo > 0)
    if addrinfo[1].family == 'inet6' then
      return true
    end
  end
  return false
end

local isipv4 = function(ip)
  return isip(ip) and not isipv6(ip)
end

local new = function()
  local self = emitter.new()
  local sock
  local loop = ev.Loop.default
  local connecting = false
  local connected = false
  local closing = false
  local watchers = {}
  
  local on_error = function(err)
    if err ~= 'closed' then
      self:emit('error',err)
    end
    self:emit('close')
    self:destroy()
  end
  
  local read_io = function()
    assert(sock)
    return ev.IO.new(function()
        local data,err,part = sock:receive(8192)
        data = part or data
        if data then
          if watchers.timer then
            watchers.timer:again(loop)
          end
          self:emit('data',data or part)
        end
        if err and err ~= 'timeout' then
          on_error(err)
        end
      end,sock:getfd(),ev.READ)
  end
  
  local pending
  local pos
  
  local write_io = function()
    assert(sock)
    return ev.IO.new(function(loop,io)
        local sent,err,so_far = sock:send(pending,pos)
        if not sent and err ~= 'timeout' then
          io:stop(loop)
          on_error(err)
        elseif sent then
          pos = nil
          pending = nil
          io:stop(loop)
          self:emit('_drain')
          self:emit('drain')
        else
          pos = so_far + 1
        end
        if watchers.timer then
          watchers.timer:again(loop)
        end
      end,sock:getfd(),ev.WRITE)
  end
  
  local on_connect = function()
    connecting = false
    connected = true
    watchers.read = read_io()
    watchers.write = write_io()
    self:resume()
    self:emit('connect',self)
  end
  
  self.connect = function(_,port,ip)
    ip = ip or '127.0.0.1'
    if not isip(ip) then
      on_error(err)
    end
    if sock and closing then
      self:once('close',function(self)
          self:_connect(port,ip)
        end)
      
    elseif not connecting then
      self:_connect(port,ip)
    end
  end
  
  self._connect = function(_,port,ip)
    assert(not sock)
    if isipv6(ip) then
      sock = socket.tcp6()
    else
      sock = socket.tcp()
    end
    sock:settimeout(0)
    connecting = true
    closing = false
    local ok,err = sock:connect(ip,port)
    if ok or err == 'already connected' then
      on_connect()
    elseif err == 'timeout' or err == 'Operation already in progress' then
      watchers.connect = ev.IO.new(function(loop,io)
          local ok,err = sock:connect(ip,port)
          if ok or err == 'already connected' then
            io:stop(loop)
            watchers.connect = nil
            on_connect()
          else
            on_error(err)
          end
        end,sock:getfd(),ev.WRITE)
      watchers.connect:start(loop)
    else
      on_error(err)
    end
  end
  
  self._transfer = function(_,s)
    sock = s
    sock:settimeout(0)
    on_connect()
  end
  
  self.write = function(_,data)
    if pending then
      pending = pending..data
    else
      pending = data
      if connecting then
        self:once('connect',function()
            watchers.write:start(loop)
          end)
      elseif connected then
        watchers.write:start(loop)
      else
        self:emit('error',err)
{'fraction_non_alphanumeric': 0.10028382213812677, 'fraction_numerical': 0.005811596161643466, 'mean_word_length': 2.1276415891800506, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, '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': '4509675', 'n_tokens_mistral': 2365, 'n_tokens_neox': 2362, 'n_words': 611}