)]}'
{"oslo_messaging/_drivers/impl_zmq.py":[{"author":{"_account_id":13290,"name":"Oleksii Zamiatin","email":"ozamiatin@mirantis.com","username":"ozamiatin"},"change_message_id":"73934cc9d15dd9734301f41813cf20e2b16d6e14","unresolved":false,"context_lines":[{"line_number":108,"context_line":"                    \u0027serializing/deserializing outgoing/incoming messages.\u0027),"},{"line_number":109,"context_line":"]"},{"line_number":110,"context_line":""},{"line_number":111,"context_line":"zmq_driver_opts \u003d ["},{"line_number":112,"context_line":"    cfg.IntOpt(\u0027rpc_thread_pool_size\u0027, default\u003d100,"},{"line_number":113,"context_line":"               help\u003d\u0027Maximum number of (green) threads to work concurrently.\u0027),"},{"line_number":114,"context_line":""}],"source_content_type":"text/x-python","patch_set":29,"id":"bacf61ea_bff0b667","line":111,"updated":"2016-08-04 09:16:09.000000000","message":"Lets rename this list to zmq_ack_retry_opts, but still add them to oslo_messaging_zmq group when register.","commit_id":"8a02b3b6980288f4b06d074424254265dc089dd9"},{"author":{"_account_id":20864,"name":"Gevorg Davoian","email":"gdavoian@gmail.com","username":"gdavoian"},"change_message_id":"57463656ff16b6682da2681770825b913bdde6b6","unresolved":false,"context_lines":[{"line_number":108,"context_line":"                    \u0027serializing/deserializing outgoing/incoming messages.\u0027),"},{"line_number":109,"context_line":"]"},{"line_number":110,"context_line":""},{"line_number":111,"context_line":"zmq_driver_opts \u003d ["},{"line_number":112,"context_line":"    cfg.IntOpt(\u0027rpc_thread_pool_size\u0027, default\u003d100,"},{"line_number":113,"context_line":"               help\u003d\u0027Maximum number of (green) threads to work concurrently.\u0027),"},{"line_number":114,"context_line":""}],"source_content_type":"text/x-python","patch_set":29,"id":"bacf61ea_fbbb3b0d","line":111,"in_reply_to":"bacf61ea_bff0b667","updated":"2016-08-04 13:04:18.000000000","message":"Done","commit_id":"8a02b3b6980288f4b06d074424254265dc089dd9"}],"oslo_messaging/_drivers/zmq_driver/client/publishers/dealer/zmq_dealer_publisher_proxy.py":[{"author":{"_account_id":13290,"name":"Oleksii Zamiatin","email":"ozamiatin@mirantis.com","username":"ozamiatin"},"change_message_id":"2fb971a11b940e1116546aa68b6f88bea1bd98ab","unresolved":false,"context_lines":[{"line_number":49,"context_line":"            raise zmq_publisher_base.UnsupportedSendPattern(request.msg_type)"},{"line_number":50,"context_line":""},{"line_number":51,"context_line":"        if self.conf.use_pub_sub:"},{"line_number":52,"context_line":"            if request.msg_type \u003d\u003d zmq_names.CAST_TYPE:"},{"line_number":53,"context_line":"                request.routing_key \u003d \\"},{"line_number":54,"context_line":"                    self.routing_table.get_routable_host(request.target)"},{"line_number":55,"context_line":"                self.ack_and_reply_waiter.register_socket("}],"source_content_type":"text/x-python","patch_set":2,"id":"7aa08908_805469ca","line":52,"updated":"2016-06-16 17:42:56.000000000","message":"Does this mean we apply retries only for casts yet?","commit_id":"f77020d6e7abfe18f8b0c16f562d0410e183277b"},{"author":{"_account_id":13290,"name":"Oleksii Zamiatin","email":"ozamiatin@mirantis.com","username":"ozamiatin"},"change_message_id":"2fb971a11b940e1116546aa68b6f88bea1bd98ab","unresolved":false,"context_lines":[{"line_number":52,"context_line":"            if request.msg_type \u003d\u003d zmq_names.CAST_TYPE:"},{"line_number":53,"context_line":"                request.routing_key \u003d \\"},{"line_number":54,"context_line":"                    self.routing_table.get_routable_host(request.target)"},{"line_number":55,"context_line":"                self.ack_and_reply_waiter.register_socket("},{"line_number":56,"context_line":"                    self.socket, self._recv_ack_or_reply"},{"line_number":57,"context_line":"                )"},{"line_number":58,"context_line":"                self.ack_and_reply_waiter.track_request("}],"source_content_type":"text/x-python","patch_set":2,"id":"7aa08908_20d77550","line":55,"updated":"2016-06-16 17:42:56.000000000","message":"can we register socket in track_request? so we convert these 2 sequential calls into 1 call. Since track request also takes the socket as a parameter?","commit_id":"f77020d6e7abfe18f8b0c16f562d0410e183277b"},{"author":{"_account_id":13290,"name":"Oleksii Zamiatin","email":"ozamiatin@mirantis.com","username":"ozamiatin"},"change_message_id":"2fb971a11b940e1116546aa68b6f88bea1bd98ab","unresolved":false,"context_lines":[{"line_number":81,"context_line":"                   \"target\": request.target,"},{"line_number":82,"context_line":"                   \"addr\": list(self.socket.connections)})"},{"line_number":83,"context_line":""},{"line_number":84,"context_line":"    def _recv_ack_or_reply(self, socket):"},{"line_number":85,"context_line":"        empty \u003d socket.recv()"},{"line_number":86,"context_line":"        assert empty \u003d\u003d b\u0027\u0027, \"Empty expected!\""},{"line_number":87,"context_line":"        reply_id \u003d socket.recv()"}],"source_content_type":"text/x-python","patch_set":2,"id":"7aa08908_00b3590b","line":84,"updated":"2016-06-16 17:42:56.000000000","message":"Since it is not CALL why we expect reply here? I see that we have this method inherited, but I think handling acks for CALL will differ at least in case when we receive ack after reply, which really possible because we send all messages asynchronously. Or CALL timeout handling should also play the role here because after CALL timeout ack is also expired and useless as well as reply.\n\nBtw, the documentation item: for CALLs+retries lets recommend to set longer reply waiting timout e.g 3 min rather than 1.","commit_id":"f77020d6e7abfe18f8b0c16f562d0410e183277b"}],"oslo_messaging/_drivers/zmq_driver/client/zmq_ack_and_reply_waiter.py":[{"author":{"_account_id":13290,"name":"Oleksii Zamiatin","email":"ozamiatin@mirantis.com","username":"ozamiatin"},"change_message_id":"2fb971a11b940e1116546aa68b6f88bea1bd98ab","unresolved":false,"context_lines":[{"line_number":93,"context_line":"        (message_type, message_id, reply), socket \u003d self._poller.poll()"},{"line_number":94,"context_line":"        assert message_type in (zmq_names.ACK_TYPE, zmq_names.REPLY_TYPE), \\"},{"line_number":95,"context_line":"            \"Ack or reply expected!\""},{"line_number":96,"context_line":"        for_reply \u003d message_type \u003d\u003d zmq_names.REPLY_TYPE"},{"line_number":97,"context_line":"        message_type_str \u003d \"reply\" if for_reply else \"ack\""},{"line_number":98,"context_line":"        future \u003d self._get_future(message_id, for_reply)"},{"line_number":99,"context_line":"        if future is None or future.stopwatch.expired():"}],"source_content_type":"text/x-python","patch_set":2,"id":"7aa08908_6a8afe65","line":96,"updated":"2016-06-16 17:42:56.000000000","message":"Pease use zmq_names.message_type_str() to convert message type to a printable string.","commit_id":"f77020d6e7abfe18f8b0c16f562d0410e183277b"},{"author":{"_account_id":13290,"name":"Oleksii Zamiatin","email":"ozamiatin@mirantis.com","username":"ozamiatin"},"change_message_id":"2fb971a11b940e1116546aa68b6f88bea1bd98ab","unresolved":false,"context_lines":[{"line_number":132,"context_line":"            # NOTE(gdavoian): the smaller the queue of pending futures is,"},{"line_number":133,"context_line":"            # the longer the resend-loop sleep, allowing the recv-loop to"},{"line_number":134,"context_line":"            # work more, and vice versa."},{"line_number":135,"context_line":"            self._sleep(1.0 / (2.0 + self._futures.qsize()))"}],"source_content_type":"text/x-python","patch_set":2,"id":"7aa08908_056c0b07","line":135,"updated":"2016-06-16 17:42:56.000000000","message":"Am I correct that here sleep will always be less than 1 sec? If so, this timeout is too short for network. For reply waiting we use 60 seconds by default. For heartbeats implementation, for example, the recommended interval is not less than 10 seconds. Otherwise we will spam the network a lot. Take into account that with retries we resend the whole message while heartbeat has a minimal size (e.g. contains message type only).\n\nMy proposal for retries is:\nFor such values we definitely need to provide configuration options.\n\n1. We need to consider some kind of smart retries rather than periodic retries with fixed timeout. What I mean by \"smart\" is that time interval should increase for each retry twice or by some other formula (maybe randomly). So this is going to be a first option - something like zmq_base_retry_period\u003d10 sec. And all timeout calculations are going to include this option as a parameter.\n\nBy the way, lets create a zmq_driver section for these new options. Later in another patch we will move all the other driver\u0027s options to this section from DEFAULT where they are now.\n\n2. Number of retries should be limited by some value (e.g 3), -1 - means infinite retries but would be not recommended. 0 - by default (current behavior). So this is the second option zmq_driver.retries_limit\u003d0.\n\nBy the way we already have retry parameter in driver\u0027s send method. Please investigate could it be used instead of this option or not.","commit_id":"f77020d6e7abfe18f8b0c16f562d0410e183277b"}],"oslo_messaging/_drivers/zmq_driver/client/zmq_ack_manager.py":[{"author":{"_account_id":13290,"name":"Oleksii Zamiatin","email":"ozamiatin@mirantis.com","username":"ozamiatin"},"change_message_id":"67f9d816b99a17b6eacc8721d640f3b178c0f1c7","unresolved":false,"context_lines":[{"line_number":106,"context_line":""},{"line_number":107,"context_line":"    def send_cast(self, request):"},{"line_number":108,"context_line":"        self.publisher.send_cast(request)"},{"line_number":109,"context_line":"        self.pool.spawn_n(self._wait_for_ack, self._get_ack_future(request))"},{"line_number":110,"context_line":""},{"line_number":111,"context_line":"    def cleanup(self):"},{"line_number":112,"context_line":"        self.pool.waitall()"}],"source_content_type":"text/x-python","patch_set":34,"id":"9ad45d7e_d330570a","line":109,"updated":"2016-08-08 11:53:56.000000000","message":"Please remove direct dependency on eventlet","commit_id":"0155fae27ca4d6d76345fd4558d9a9c23d806659"}],"oslo_messaging/_drivers/zmq_driver/client/zmq_client.py":[{"author":{"_account_id":13290,"name":"Oleksii Zamiatin","email":"ozamiatin@mirantis.com","username":"ozamiatin"},"change_message_id":"2fb971a11b940e1116546aa68b6f88bea1bd98ab","unresolved":false,"context_lines":[{"line_number":119,"context_line":"            conf, matchmaker, zmq.ROUTER, zmq.DEALER"},{"line_number":120,"context_line":"        )"},{"line_number":121,"context_line":""},{"line_number":122,"context_line":"        self.ack_and_reply_waiter \u003d \\"},{"line_number":123,"context_line":"            zmq_ack_and_reply_waiter.AckAndReplyWaiter(conf)"},{"line_number":124,"context_line":""},{"line_number":125,"context_line":"        super(ZmqClientProxy, self).__init__("}],"source_content_type":"text/x-python","patch_set":2,"id":"7aa08908_4d03d9b8","line":122,"updated":"2016-06-16 17:42:56.000000000","message":"Lets switch between ack_and_reply_waiter and simple reply_waiter by option zmq_driver.retries_limit (if 0 then no acks).","commit_id":"f77020d6e7abfe18f8b0c16f562d0410e183277b"}],"oslo_messaging/_drivers/zmq_driver/server/consumers/zmq_dealer_consumer.py":[{"author":{"_account_id":13290,"name":"Oleksii Zamiatin","email":"ozamiatin@mirantis.com","username":"ozamiatin"},"change_message_id":"2fb971a11b940e1116546aa68b6f88bea1bd98ab","unresolved":false,"context_lines":[{"line_number":105,"context_line":"            message_type \u003d int(socket.recv())"},{"line_number":106,"context_line":"            message_id \u003d socket.recv()"},{"line_number":107,"context_line":"            context \u003d socket.recv_pyobj()"},{"line_number":108,"context_line":"            message \u003d socket.recv_pyobj()"},{"line_number":109,"context_line":"            LOG.debug(\"[%(host)s] Received message %(id)s\","},{"line_number":110,"context_line":"                      {\"host\": self.host, \"id\": message_id})"},{"line_number":111,"context_line":"            if message_type \u003d\u003d zmq_names.CALL_TYPE:"}],"source_content_type":"text/x-python","patch_set":2,"id":"7aa08908_4d5599c4","line":108,"updated":"2016-06-16 17:42:56.000000000","message":"Somewhere here we definitely need to handle the repeated message (because we do retries and we can make few of them before the first ack reached the client and we stop) to support at-most-once concept. We need to track the message id (some dict-table). We also need to consider some kind of record expiration logic here because this table will always grow. Records should expire by reply timeout or something.\n\nWe also need to handle the situation when the record being expired but we still receive retried message (Too OLD message). Something like TTL for message, but currently I have no clear idea how to implement it.\n\nAssuming that logic becomes more complicated I propose to extract DealerConsumerAck class inherited from DealerConsumer to support acked-messages table and all the other stuff there.","commit_id":"f77020d6e7abfe18f8b0c16f562d0410e183277b"},{"author":{"_account_id":13290,"name":"Oleksii Zamiatin","email":"ozamiatin@mirantis.com","username":"ozamiatin"},"change_message_id":"f94763790decd623e1db087db65b25b06ee0f618","unresolved":false,"context_lines":[{"line_number":36,"context_line":"    def __init__(self, conf, poller, server):"},{"line_number":37,"context_line":"        self.ack_sender \u003d zmq_senders.AckSenderProxy(conf)"},{"line_number":38,"context_line":"        self.reply_sender \u003d zmq_senders.ReplySenderProxy(conf)"},{"line_number":39,"context_line":"        self.received_messages \u003d zmq_queued_set.QueuedSet("},{"line_number":40,"context_line":"            maxsize\u003dconf.oslo_messaging_zmq.rpc_buffer_size"},{"line_number":41,"context_line":"        )"},{"line_number":42,"context_line":"        self.sockets_manager \u003d zmq_sockets_manager.SocketsManager("}],"source_content_type":"text/x-python","patch_set":20,"id":"dada55a8_b466a927","line":39,"updated":"2016-07-21 11:08:28.000000000","message":"As far as I understood (please correct me if I\u0027m wrong) this data structure is a kind of dictionary: \u003cid, bool\u003e and I didn\u0027t get the reason of the second value here (always True - never used?)\n\nWhat if we considered some more smart dictionary with timeout. So it would be like:\n\u003cid, timestamp\u003e - then we can have some background executor which time after time (update period from config e.g. \u0027rpc_response_timeout\u0027 - already exists) will run over the dictionary and will remove the expired  items?\n\nBtw I didn\u0027t find where the discard() method is used. Perhaps the only limit for now is buffer_size\u003d10k in such case this memory will be always consumed and we will never release it.","commit_id":"8589d095eaf8ac3f2a8d45b8209201aea680cba9"},{"author":{"_account_id":20864,"name":"Gevorg Davoian","email":"gdavoian@gmail.com","username":"gdavoian"},"change_message_id":"f1fda19d80f06d8613f2e7bd4d92e73cf5136b1d","unresolved":false,"context_lines":[{"line_number":36,"context_line":"    def __init__(self, conf, poller, server):"},{"line_number":37,"context_line":"        self.ack_sender \u003d zmq_senders.AckSenderProxy(conf)"},{"line_number":38,"context_line":"        self.reply_sender \u003d zmq_senders.ReplySenderProxy(conf)"},{"line_number":39,"context_line":"        self.received_messages \u003d zmq_queued_set.QueuedSet("},{"line_number":40,"context_line":"            maxsize\u003dconf.oslo_messaging_zmq.rpc_buffer_size"},{"line_number":41,"context_line":"        )"},{"line_number":42,"context_line":"        self.sockets_manager \u003d zmq_sockets_manager.SocketsManager("}],"source_content_type":"text/x-python","patch_set":20,"id":"dada55a8_d2288c9c","line":39,"in_reply_to":"dada55a8_b466a927","updated":"2016-07-21 12:01:18.000000000","message":"The second value is needed here since the current implementation (probably the easiest possible) is based on collections.OrderedDict (there is no collections.OrderedSet). I will try to add TTL to stored items.\n\nDiscard method is not used yet, but technically it is required because it is one of the abstract methods of collections.MutableSet (all the other methods are implemented via add, discard, __contains__, __iter__ and __len__). But I am not sure whether we need a fully functional set here.","commit_id":"8589d095eaf8ac3f2a8d45b8209201aea680cba9"},{"author":{"_account_id":13290,"name":"Oleksii Zamiatin","email":"ozamiatin@mirantis.com","username":"ozamiatin"},"change_message_id":"f94763790decd623e1db087db65b25b06ee0f618","unresolved":false,"context_lines":[{"line_number":74,"context_line":""},{"line_number":75,"context_line":"            # drop duplicate message"},{"line_number":76,"context_line":"            if message_id in self.received_messages:"},{"line_number":77,"context_line":"                return None"},{"line_number":78,"context_line":""},{"line_number":79,"context_line":"            self.received_messages.add(message_id)"},{"line_number":80,"context_line":"            LOG.debug(\"[%(host)s] Received %(msg_type)s message %(msg_id)s\","}],"source_content_type":"text/x-python","patch_set":20,"id":"dada55a8_d9e9d6b8","line":77,"updated":"2016-07-21 11:08:28.000000000","message":"Please put a LOG.warning here","commit_id":"8589d095eaf8ac3f2a8d45b8209201aea680cba9"},{"author":{"_account_id":13290,"name":"Oleksii Zamiatin","email":"ozamiatin@mirantis.com","username":"ozamiatin"},"change_message_id":"f94763790decd623e1db087db65b25b06ee0f618","unresolved":false,"context_lines":[{"line_number":76,"context_line":"            if message_id in self.received_messages:"},{"line_number":77,"context_line":"                return None"},{"line_number":78,"context_line":""},{"line_number":79,"context_line":"            self.received_messages.add(message_id)"},{"line_number":80,"context_line":"            LOG.debug(\"[%(host)s] Received %(msg_type)s message %(msg_id)s\","},{"line_number":81,"context_line":"                      {\"host\": self.host,"},{"line_number":82,"context_line":"                       \"msg_type\": zmq_names.message_type_str(msg_type),"}],"source_content_type":"text/x-python","patch_set":20,"id":"dada55a8_7949cacf","line":79,"updated":"2016-07-21 11:08:28.000000000","message":"How long will received message stay in the queue? Is there some expiration period?","commit_id":"8589d095eaf8ac3f2a8d45b8209201aea680cba9"}],"oslo_messaging/_drivers/zmq_driver/server/consumers/zmq_router_consumer.py":[{"author":{"_account_id":13290,"name":"Oleksii Zamiatin","email":"ozamiatin@mirantis.com","username":"ozamiatin"},"change_message_id":"f94763790decd623e1db087db65b25b06ee0f618","unresolved":false,"context_lines":[{"line_number":55,"context_line":""},{"line_number":56,"context_line":"            # drop duplicate message"},{"line_number":57,"context_line":"            if message_id in self.received_messages:"},{"line_number":58,"context_line":"                return None"},{"line_number":59,"context_line":""},{"line_number":60,"context_line":"            self.received_messages.add(message_id)"},{"line_number":61,"context_line":"            LOG.debug(\"[%(host)s] Received %(msg_type)s message %(msg_id)s\","}],"source_content_type":"text/x-python","patch_set":20,"id":"dada55a8_147695d6","line":58,"updated":"2016-07-21 11:08:28.000000000","message":"The same about warning here","commit_id":"8589d095eaf8ac3f2a8d45b8209201aea680cba9"}],"oslo_messaging/_drivers/zmq_driver/server/zmq_queued_set.py":[{"author":{"_account_id":13290,"name":"Oleksii Zamiatin","email":"ozamiatin@mirantis.com","username":"ozamiatin"},"change_message_id":"f94763790decd623e1db087db65b25b06ee0f618","unresolved":false,"context_lines":[{"line_number":29,"context_line":"        self._queued_map \u003d collections.OrderedDict()"},{"line_number":30,"context_line":""},{"line_number":31,"context_line":"    def add(self, item):"},{"line_number":32,"context_line":"        self._queued_map[item] \u003d True"},{"line_number":33,"context_line":"        if len(self._queued_map) \u003e self._maxsize:"},{"line_number":34,"context_line":"            self._queued_map.popitem(last\u003dFalse)"},{"line_number":35,"context_line":""}],"source_content_type":"text/x-python","patch_set":20,"id":"dada55a8_af9e50b6","line":32,"updated":"2016-07-21 11:08:28.000000000","message":"Timeout probably should be set up here and being updated on each retried message comes. Then we can discard message by timeout after the last retry.","commit_id":"8589d095eaf8ac3f2a8d45b8209201aea680cba9"},{"author":{"_account_id":13290,"name":"Oleksii Zamiatin","email":"ozamiatin@mirantis.com","username":"ozamiatin"},"change_message_id":"f94763790decd623e1db087db65b25b06ee0f618","unresolved":false,"context_lines":[{"line_number":33,"context_line":"        if len(self._queued_map) \u003e self._maxsize:"},{"line_number":34,"context_line":"            self._queued_map.popitem(last\u003dFalse)"},{"line_number":35,"context_line":""},{"line_number":36,"context_line":"    def discard(self, item):"},{"line_number":37,"context_line":"        self._queued_map.pop(item, None)"},{"line_number":38,"context_line":""},{"line_number":39,"context_line":"    def __contains__(self, item):"}],"source_content_type":"text/x-python","patch_set":20,"id":"dada55a8_7406f120","line":36,"updated":"2016-07-21 11:08:28.000000000","message":"Discard should happen automagically after expiration timeout on the record.","commit_id":"8589d095eaf8ac3f2a8d45b8209201aea680cba9"}],"oslo_messaging/_drivers/zmq_driver/zmq_async.py":[{"author":{"_account_id":13290,"name":"Oleksii Zamiatin","email":"ozamiatin@mirantis.com","username":"ozamiatin"},"change_message_id":"2fb971a11b940e1116546aa68b6f88bea1bd98ab","unresolved":false,"context_lines":[{"line_number":51,"context_line":"    return six.moves.queue.Queue(), six.moves.queue.Empty"},{"line_number":52,"context_line":""},{"line_number":53,"context_line":""},{"line_number":54,"context_line":"def get_future():"},{"line_number":55,"context_line":"    import futurist"},{"line_number":56,"context_line":"    if eventletutils.is_monkey_patched(\u0027thread\u0027):"},{"line_number":57,"context_line":"        return futurist.GreenFuture()"}],"source_content_type":"text/x-python","patch_set":2,"id":"7aa08908_6a14bee8","line":54,"updated":"2016-06-16 17:42:56.000000000","message":"Not sure about this one, but if I don\u0027t mistake Future is also monkey_patched. No need in this function.","commit_id":"f77020d6e7abfe18f8b0c16f562d0410e183277b"},{"author":{"_account_id":13290,"name":"Oleksii Zamiatin","email":"ozamiatin@mirantis.com","username":"ozamiatin"},"change_message_id":"2fb971a11b940e1116546aa68b6f88bea1bd98ab","unresolved":false,"context_lines":[{"line_number":58,"context_line":"    return futurist.Future()"},{"line_number":59,"context_line":""},{"line_number":60,"context_line":""},{"line_number":61,"context_line":"def get_sleep_function():"},{"line_number":62,"context_line":"    if eventletutils.is_monkey_patched(\u0027thread\u0027):"},{"line_number":63,"context_line":"        import eventlet"},{"line_number":64,"context_line":"        return eventlet.sleep"}],"source_content_type":"text/x-python","patch_set":2,"id":"7aa08908_ca0332bd","line":61,"updated":"2016-06-16 17:42:56.000000000","message":"No need in this just use time.sleep() everywhere, monkey_patch applies to it.","commit_id":"f77020d6e7abfe18f8b0c16f562d0410e183277b"}]}