Transport Endpoint Not Connected

[dlecocq]

I intermittently get errors like the following:

[ERROR] Failed to run task produce
Traceback (most recent call last):
  File "/usr/local/lib/python2.7/dist-packages/shovel/tasks.py", line 243, in __call__
    return self._obj(*args, **kwargs)
  File "/home/dan/git/proton-pack/shovel/production.py", line 17, in produce
    find_in_package(producers, producer).from_config(**options).produce()
  File "/usr/local/lib/python2.7/dist-packages/protonpack-0.1.0-py2.7.egg/protonpack/producers/__init__.py", line 41, in produce
    for item in self:
  File "/usr/local/lib/python2.7/dist-packages/protonpack-0.1.0-py2.7.egg/protonpack/streams/__init__.py", line 26, in __iter__
    for message in self._reader:
  File "/usr/local/lib/python2.7/dist-packages/nsq/reader.py", line 68, in __iter__
    for message in self.read():
  File "/usr/local/lib/python2.7/dist-packages/nsq/reader.py", line 56, in read
    found = Client.read(self)
  File "/usr/local/lib/python2.7/dist-packages/nsq/client.py", line 175, in read
    for res in conn.read():
  File "/usr/local/lib/python2.7/dist-packages/nsq/connection.py", line 127, in read
    packet = self._socket.recv(4096)
  File "/usr/local/lib/python2.7/dist-packages/gevent/socket.py", line 385, in recv
    return sock.recv(*args)
error: [Errno 107] Transport endpoint is not connected

Obviously, this should be remedied.

[dlecocq]

I also occasionally get this error:

[ERROR] Failed to run task produce
Traceback (most recent call last):
  File "/usr/local/lib/python2.7/dist-packages/shovel/tasks.py", line 243, in __call__
    return self._obj(*args, **kwargs)
  File "/home/dan/git/proton-pack/shovel/production.py", line 17, in produce
    find_in_package(producers, producer).from_config(**options).produce()
  File "/usr/local/lib/python2.7/dist-packages/protonpack-0.1.0-py2.7.egg/protonpack/producers/__init__.py", line 41, in produce
    for item in self:
  File "/usr/local/lib/python2.7/dist-packages/protonpack-0.1.0-py2.7.egg/protonpack/streams/__init__.py", line 26, in __iter__
    for message in self._reader:
  File "/usr/local/lib/python2.7/dist-packages/nsq_py-0.1.2-py2.7.egg/nsq/reader.py", line 68, in __iter__
    for message in self.read():
  File "/usr/local/lib/python2.7/dist-packages/nsq_py-0.1.2-py2.7.egg/nsq/reader.py", line 60, in read
    self.distribute_ready()
  File "/usr/local/lib/python2.7/dist-packages/nsq_py-0.1.2-py2.7.egg/nsq/reader.py", line 39, in distribute_ready
    conn.rdy(count)
  File "/usr/local/lib/python2.7/dist-packages/nsq_py-0.1.2-py2.7.egg/nsq/connection.py", line 257, in rdy
    return self.send(constants.RDY + ' ' + str(count))
  File "/usr/local/lib/python2.7/dist-packages/nsq_py-0.1.2-py2.7.egg/nsq/connection.py", line 235, in send
    self.flush()
  File "/usr/local/lib/python2.7/dist-packages/nsq_py-0.1.2-py2.7.egg/nsq/connection.py", line 208, in flush
    count = self._socket.send(self._pending[0])
  File "/usr/local/lib/python2.7/dist-packages/gevent/socket.py", line 441, in send
    self._wait(self._write_event)
  File "/usr/local/lib/python2.7/dist-packages/gevent/socket.py", line 292, in _wait
    assert watcher.callback is None, 'This socket is already used by another greenlet: %r' % (watcher.callback, )
AssertionError: This socket is already used by another greenlet: <bound method Waiter.switch of <gevent.hub.Waiter object at 0x26f0550>>
Traceback (most recent call last):
  File "/usr/local/bin/shovel", line 25, in <module>
    shovel.run()
  File "/usr/local/lib/python2.7/dist-packages/shovel/runner.py", line 119, in run
    task(*args, **kwargs)
  File "/usr/local/lib/python2.7/dist-packages/shovel/tasks.py", line 246, in __call__
    raise(exc)
AssertionError: This socket is already used by another greenlet: <bound method Waiter.switch of <gevent.hub.Waiter object at 0x26f0550>>

[dlecocq]

And sometimes this one:

dan@dalstgprotonpack02:~/git/proton-pack$ shovel production.consume hyena bitly
Traceback (most recent call last):
  File "/usr/local/lib/python2.7/dist-packages/gevent/greenlet.py", line 327, in run
    result = self._run(*self.args, **self.kwargs)
  File "/usr/local/lib/python2.7/dist-packages/gevent/pool.py", line 298, in _run
    for item in self.iterable:
  File "/usr/local/lib/python2.7/dist-packages/protonpack-0.1.0-py2.7.egg/protonpack/consumers/__init__.py", line 49, in groups
    for group in grouper(self.items(source), self._batch_size):
  File "/usr/local/lib/python2.7/dist-packages/protonpack-0.1.0-py2.7.egg/protonpack/util.py", line 9, in grouper
    for group in izip_longest(fillvalue=None, *args):
  File "/usr/local/lib/python2.7/dist-packages/protonpack-0.1.0-py2.7.egg/protonpack/streams/bitly.py", line 17, in urls
    for item in iter(self):
  File "/usr/local/lib/python2.7/dist-packages/protonpack-0.1.0-py2.7.egg/protonpack/streams/__init__.py", line 34, in __iter__
    logger.exception('Failed to parse a NSQ message')
  File "/usr/lib/python2.7/contextlib.py", line 24, in __exit__
    self.gen.next()
  File "/usr/local/lib/python2.7/dist-packages/nsq_py-0.1.2-py2.7.egg/nsq/response.py", line 114, in handle
    self.fin()
  File "/usr/local/lib/python2.7/dist-packages/nsq_py-0.1.2-py2.7.egg/nsq/response.py", line 86, in fin
    self.connection.fin(self.id)
  File "/usr/local/lib/python2.7/dist-packages/nsq_py-0.1.2-py2.7.egg/nsq/connection.py", line 282, in fin
    return self.send(constants.FIN + ' ' + message_id)
  File "/usr/local/lib/python2.7/dist-packages/nsq_py-0.1.2-py2.7.egg/nsq/connection.py", line 256, in send
    self.flush()
  File "/usr/local/lib/python2.7/dist-packages/nsq_py-0.1.2-py2.7.egg/nsq/connection.py", line 228, in flush
    count = sock.send(self._pending[0])
  File "/usr/local/lib/python2.7/dist-packages/gevent/socket.py", line 435, in send
    return sock.send(data, flags)
error: [Errno 32] Broken pipe
<IMap at 0x13d72d0> failed with error