error with pynsq

[cheng zhang]

when I use pynsq client ，I got the errors as follow, How can I deal with this?

WARNING:nsq.reader:[hg014:4150:3_pages:content_parser] connection closed

ERROR:tornado.application:Exception in callback <functools.partial object at 0x30ebaf8>

Traceback (most recent call last):

  File "/root/escp_core/build/content_parser/out00-PYZ.pyz/tornado.ioloop", line 565, in _run_callback

  File "/root/escp_core/build/content_parser/out00-PYZ.pyz/tornado.stack_context", line 275, in null_wrapper

  File "/root/escp_core/build/content_parser/out00-PYZ.pyz/nsq.async", line 186, in _start_read

  File "/root/escp_core/build/content_parser/out00-PYZ.pyz/tornado.iostream", line 284, in read_bytes

  File "/root/escp_core/build/content_parser/out00-PYZ.pyz/tornado.iostream", line 654, in _try_inline_read

  File "/root/escp_core/build/content_parser/out00-PYZ.pyz/tornado.iostream", line 834, in _check_closed

StreamClosedError: Stream is closed

[Matt Reiferson]

That looks like the underlying connection was closed.  It doesn't *look* like a pynsq bug.

Is it repeatable/continuing to occur?

[cheng zhang]

when I use the writer of pynsq， it is repeatable。It's no problem for reader.

在 2014年7月17日星期四UTC+8下午5时19分23秒，Matt Reiferson写道：

[cheng zhang]

I have read the code of pynsq writer, find that:

    def _on_connection_close(self, conn, **kwargs):

        if conn.id in self.conns:

            del self.conns[conn.id]

        for callback in conn.callback_queue:

            try:

                callback(conn, nsq.ConnectionClosedError())

            except Exception:

                logging.exception('[%s] uncaught exception in callback', conn.id)

        logging.warning('[%s] connection closed', conn.id)

        logging.info('[%s] attempting to reconnect in %0.2fs', conn.id, self.reconnect_interval)

        reconnect_callback = functools.partial(self.connect_to_nsqd,

                                               host=conn.host, port=conn.port)

        self.io_loop.add_timeout(time.time() + self.reconnect_interval, reconnect_callback)

the pynsq will reconnect after the self.reconnect_interval, I can't  understand the meanings. I want to know why the connect closed?

在 2014年7月17日星期四UTC+8下午5时19分23秒，Matt Reiferson写道：

[cheng zhang]

And I find the follow code :

    def _check_last_recv_timestamps(self):

        now = time.time()

        def is_stale(conn):

            timestamp = conn.last_recv_timestamp

            return (now - timestamp) > ((conn.heartbeat_interval * 2) / 1000.0)

        # first get the list of stale connections, then close

        # (`conn.close()` may modify the list of connections while we're iterating)

        stale_connections = [conn for conn in self.conns.values() if is_stale(conn)]

        for conn in stale_connections:

            timestamp = conn.last_recv_timestamp

            # this connection hasnt received data for more than

            # the configured heartbeat interval, close it

            logging.warning('[%s:%s] connection is stale (%.02fs), closing',

                            conn.id, self.name, (now - timestamp))

            conn.close()

when there is no recv data more than two heartbreak interval, the connection will be close. why do this?

在 2014年7月17日星期四UTC+8下午5时10分40秒，cheng zhang写道：

[Matt Reiferson]

Can you show the code you’re using in your writer?

Writer manages connections for you, so it reconnects if a connection is closed.  It also closes connections over which it has not seen data for some time (2x the heartbeat interval).

[cheng zhang]

nsq_writer = nsq.Writer(['127.0.0.1:4150'])

@timing

def parse_page(task_id, message):

    xpath_json = task_xpath[task_id]

    xf = XpathConfig()

    xf.load_config(xpath_json)

    fe = XpathFetch(xf)

    html = message.body

    url = ''

    try:

        idx = html.find('\t', 5)

        if idx < 0:

            return True

        url = html[html.find('\t') + 1:idx]

        rawdata = html[idx + 1:]

        result = fe.parse_content('content', rawdata)

        if 'title' not in result or not result['title']:

            print 'not content page'

            return True

        msg_data = {}

        for key, value in result.items():

            if not value:

                continue

            msg_data[key] = value[0].strip()

        msg_data['url'] = url

        article = goose.extract(raw_html=rawdata)

        msg_data['title'] = article.title

        msg_data['meta_keys'] = article.meta_keywords

        msg_data['meta_desc'] = article.meta_description

        msg_data['content'] = article.cleaned_text

        msg_data['time'] = parse_time(msg_data['pub_time'])

        topic = '%s_content' % task_id

        msg = json.dumps(msg_data)

        print msg_data['title'], msg_data['pub_time'], msg_data['time']

        callback = functools.partial(finish_pub, topic=topic, msg=msg)

        nsq_writer.pub(topic, msg, callback=callback)

    except:

        print('%s error: %s' % (task_id, traceback.format_exc()))

        #print url

        pass

    return True

在 2014年7月17日星期四UTC+8下午5时43分35秒，Matt Reiferson写道：

[Matt Reiferson]

can you paste more of the logs?

[cheng zhang]

WARNING:nsq.reader:[74_pages:content_parser] lookupd http://hg007:4161/lookup?topic=74_pages query error: HTTP 599: Timeout

WARNING:nsq.reader:[38_pages:content_parser] lookupd http://hg007:4161/lookup?topic=38_pages query error: HTTP 599: Timeout

WARNING:nsq.reader:[3_pages:content_parser] lookupd http://hg007:4161/lookup?topic=3_pages query error: HTTP 599: Timeout

There are some logs like this, does It matter?

在 2014年7月17日星期四UTC+8下午5时57分13秒，Matt Reiferson写道：

[cheng zhang]

more logs 

WARNING:nsq.client:[hg011:4150:74_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg006:4150:74_pages:content_parser] connection is stale (65.46s), closing

WARNING:nsq.client:[hg004:4150:74_pages:content_parser] connection is stale (65.47s), closing

WARNING:nsq.client:[hg010:4150:74_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg015:4150:38_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg009:4150:38_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg003:4150:38_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg011:4150:38_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg010:4150:38_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg005:4150:38_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg014:4150:38_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg013:4150:38_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg007:4150:38_pages:content_parser] connection is stale (65.93s), closing

WARNING:nsq.client:[hg004:4150:38_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg006:4150:38_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg008:4150:38_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg002:4150:38_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg006:4150:3_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg015:4150:3_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg009:4150:3_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg003:4150:3_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg005:4150:3_pages:content_parser] connection is stale (65.93s), closing

WARNING:nsq.client:[hg011:4150:3_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg010:4150:3_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg012:4150:3_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg014:4150:3_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg004:4150:3_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg007:4150:3_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg013:4150:3_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg002:4150:3_pages:content_parser] connection is stale (70.17s), closing

WARNING:nsq.client:[hg008:4150:3_pages:content_parser] connection is stale (70.17s), closing

在 2014年7月17日星期四UTC+8下午5时57分13秒，Matt Reiferson写道：

[Matt Reiferson]

How do you have nsqd configured?

[cheng zhang]

-max-bytes-per-file=104857600 -max-body-size=5123840 -max-msg-size=10247680

the others are default!

在 2014年7月18日星期五UTC+8下午4时37分22秒，Matt Reiferson写道：

[Matt Reiferson]

Sorry for all these questions, but (if possible) can you post the full source of your reader and writer?

I believe something is blocking the IOLoop and causing timeouts.

[cheng zhang]

Thanks for you reply! I got that.

the message handler of reader maybe blocking the ioloop, I have tried to improve this. Thank you very much.

在 2014年7月21日星期一UTC+8下午4时45分17秒，Matt Reiferson写道：