def sample(self, sentence, phrase, h):
        # use dynamic programming to find the probability of a given context C
        # and generated multiset A:
        #
        # p(a | C, A) = ( p(a|C) + Z_{A-{a} | trunc(C, N-1)} ) / Z_{A | C}
        #
        sequence = self.depgen.conditioning_fn(sentence, phrase, h)
        h_index = cond.head_index(sequence)
        head, (l_deps, r_deps) = pop(sequence, h_index)
        deps = l_deps + r_deps

        def gen_left_probs(context, xs_bag):
            left_score = self.depgen.left_distro.rfs[head].score
            right_score = self.depgen.right_distro.rfs[head].score
            n = self.depgen.n
            yield HALT, exp(left_score(HALT, context)
                            + ngram_perm_norm_from((right_score, n, context, xs_bag)))
            for x, rest in rfutils.thing_and_rest(xs_bag): # potential inefficiency
                next_context = truncate(context + (x,), n - 1)
                A_rest = double_ngram_left_perm_norm_from((left_score,
                                                          right_score,
                                                          n,
                                                          next_context,
                                                          pbag(rest))) 
                yield x, exp(left_score(x, context) + A_rest)

        def gen_right_probs(context, xs_bag):
            score = self.depgen.right_distro.rfs[head].score
            n = self.depgen.n
            if not xs_bag:
                yield HALT, exp(score(HALT, context))
            else:
                for x, rest in rfutils.thing_and_rest(xs_bag): # potential inefficiency
                    next_context = truncate(context + (x,), n - 1)
                    A_rest = ngram_perm_norm_from((score, n, next_context, pbag(rest))) 
                    yield x, exp(score(x, context) + A_rest)

        def sample_part(get_probs, context, xs_bag):
            while xs_bag:
                probs = get_probs(context, xs_bag)
                x = weighted_choice(list(probs))
                if x is HALT:
                    break
                else:
                    yield x
                    context = truncate(context + (x,), len(context))
                    xs_bag = xs_bag.remove(x)

        deps_bag = pbag(deps)
        start = (HALT,) * (self.depgen.n - 1)
        left = list(sample_part(gen_left_probs, start, deps_bag))
        remainder = deps_bag - pbag(left)
        right = sample_part(gen_right_probs, start, remainder)

        new_sequence = list(it.chain(reversed(left), [head], right))

        rfutils.get_cache(double_ngram_left_perm_norm_from).clear()
        rfutils.get_cache(ngram_perm_norm_from).clear()
        
        return o.sample_indices_in(sequence, new_sequence)

    def test_optimize_clb_adds_maintain_unique_ports(self):
        """
        Multiple ports can be specified for the same address and LB ID when
        adding to a CLB.
        """
        steps = pbag([
            AddNodesToCLB(
                lb_id='5',
                address_configs=s(('1.1.1.1',
                                   CLBDescription(lb_id='5', port=80)))),
            AddNodesToCLB(
                lb_id='5',
                address_configs=s(('1.1.1.1',
                                   CLBDescription(lb_id='5', port=8080))))])

        self.assertEqual(
            optimize_steps(steps),
            pbag([
                AddNodesToCLB(
                    lb_id='5',
                    address_configs=s(
                        ('1.1.1.1',
                         CLBDescription(lb_id='5', port=80)),
                        ('1.1.1.1',
                         CLBDescription(lb_id='5', port=8080))))]))

 def test_filters_clb_types(self):
     """
     Only one CLB step is returned per CLB
     """
     steps = pbag([
         AddNodesToCLB(
             lb_id='5',
             address_configs=s(('1.1.1.1',
                                CLBDescription(lb_id='5', port=80)))),
         RemoveNodesFromCLB(lb_id='5', node_ids=s('1')),
         # Unoptimizable step
         CreateServer(server_config=pmap({})),
     ])
     # returned steps could be pbag of any of the 2 lists below depending
     # on how `one_clb_step` iterates over the steps. Since it is pbag the
     # order of elements is not guaranteed
     list1 = [
         AddNodesToCLB(
             lb_id='5',
             address_configs=s(
                 ('1.1.1.1', CLBDescription(lb_id='5', port=80)))),
         CreateServer(server_config=pmap({}))
     ]
     list2 = [
         RemoveNodesFromCLB(lb_id='5', node_ids=s('1')),
         CreateServer(server_config=pmap({}))
     ]
     self.assertEqual(
         matches(MatchesAny(Equals(pbag(list1)), Equals(pbag(list2)))),
         optimize_steps(steps)
     )

    def test_mixed_optimization(self):
        """
        Mixes of optimizable and unoptimizable steps still get optimized
        correctly.
        """
        steps = pbag([
            # CLB adds
            AddNodesToCLB(
                lb_id='5',
                address_configs=s(('1.1.1.1',
                                   CLBDescription(lb_id='5', port=80)))),
            AddNodesToCLB(
                lb_id='5',
                address_configs=s(('1.1.1.2',
                                   CLBDescription(lb_id='5', port=80)))),
            AddNodesToCLB(
                lb_id='6',
                address_configs=s(('1.1.1.1',
                                   CLBDescription(lb_id='6', port=80)))),
            AddNodesToCLB(
                lb_id='6',
                address_configs=s(('1.1.1.2',
                                   CLBDescription(lb_id='6', port=80)))),
            RemoveNodesFromCLB(lb_id='5', node_ids=s('1')),
            RemoveNodesFromCLB(lb_id='5', node_ids=s('2')),
            RemoveNodesFromCLB(lb_id='6', node_ids=s('3')),
            RemoveNodesFromCLB(lb_id='6', node_ids=s('4')),

            # Unoptimizable steps
            CreateServer(server_config=pmap({})),
        ])

        self.assertEqual(
            optimize_steps(steps),
            pbag([
                # Optimized CLB adds
                AddNodesToCLB(
                    lb_id='5',
                    address_configs=s(('1.1.1.1',
                                       CLBDescription(lb_id='5', port=80)),
                                      ('1.1.1.2',
                                       CLBDescription(lb_id='5', port=80)))),
                AddNodesToCLB(
                    lb_id='6',
                    address_configs=s(('1.1.1.1',
                                       CLBDescription(lb_id='6', port=80)),
                                      ('1.1.1.2',
                                       CLBDescription(lb_id='6', port=80)))),
                RemoveNodesFromCLB(lb_id='5', node_ids=s('1', '2')),
                RemoveNodesFromCLB(lb_id='6', node_ids=s('3', '4')),

                # Unoptimizable steps
                CreateServer(server_config=pmap({}))
            ]))

    def test_draining_server_has_all_enabled_lb_set_to_draining(self):
        """
        If a draining server is associated with any load balancers, those
        load balancer nodes will be set to draining and the server is not
        deleted.  The metadata on the server is not re-set to draining.

        This can happen for instance if the server was supposed to be deleted
        in a previous convergence run, and the server metadata was set but
        the load balancers update failed.

        Or if the server is set to be manually deleted via the API.
        """
        self.assertEqual(
            converge(
                DesiredGroupState(server_config={}, capacity=0,
                                  draining_timeout=10.0),
                set([server('abc', state=ServerState.ACTIVE,
                            metadata=dict([DRAINING_METADATA]),
                            servicenet_address='1.1.1.1',
                            desired_lbs=s(self.clb_desc, self.rcv3_desc))]),
                set([CLBNode(node_id='1', address='1.1.1.1',
                             description=self.clb_desc)]),
                1),
            pbag([
                ChangeCLBNode(lb_id='1', node_id='1', weight=1,
                              condition=CLBNodeCondition.DRAINING,
                              type=CLBNodeType.PRIMARY)
            ]))

    def test_active_server_is_drained_even_if_all_already_in_draining(self):
        """
        If an active server is attached to load balancers, and all those load
        balancer nodes are already in draining but it cannot be removed yet,
        the server is set to draining state even though no load balancer
        actions need to be performed.

        This can happen for instance if the server was supposed to be deleted
        in a previous convergence run, and the load balancers were set to
        draining but setting the server metadata failed.
        """
        self.assertEqual(
            converge(
                DesiredGroupState(server_config={}, capacity=0,
                                  draining_timeout=10.0),
                set([server('abc', state=ServerState.ACTIVE,
                            servicenet_address='1.1.1.1',
                            desired_lbs=s(self.clb_desc, self.rcv3_desc))]),
                set([CLBNode(node_id='1', address='1.1.1.1',
                             description=copy_clb_desc(
                                 self.clb_desc,
                                 condition=CLBNodeCondition.DRAINING),
                             connections=1, drained_at=0.0)]),
                1),
            pbag([
                SetMetadataItemOnServer(server_id='abc',
                                        key=DRAINING_METADATA[0],
                                        value=DRAINING_METADATA[1]),
                self.clstep
            ]))

 def test_active_server_is_drained_if_not_all_lbs_can_be_removed(self):
     """
     If an active server to be deleted cannot be removed from all the load
     balancers, it is set to draining state and all the nodes are set to
     draining condition.
     """
     self.assertEqual(
         converge(
             DesiredGroupState(server_config={}, capacity=0,
                               draining_timeout=10.0),
             set([server('abc', state=ServerState.ACTIVE,
                         servicenet_address='1.1.1.1',
                         desired_lbs=s(self.clb_desc, self.rcv3_desc))]),
             set([CLBNode(node_id='1', address='1.1.1.1',
                          description=self.clb_desc),
                  RCv3Node(node_id='2', cloud_server_id='abc',
                           description=self.rcv3_desc)]),
             0),
         pbag([
             ChangeCLBNode(lb_id='1', node_id='1', weight=1,
                           condition=CLBNodeCondition.DRAINING,
                           type=CLBNodeType.PRIMARY),
             SetMetadataItemOnServer(server_id='abc',
                                     key=DRAINING_METADATA[0],
                                     value=DRAINING_METADATA[1]),
             BulkRemoveFromRCv3(lb_node_pairs=s(
                 (self.rcv3_desc.lb_id, 'abc')))
         ]))

    def wait_for_stack_list(self, expected_states, timeout=180, period=10):
        def check(content):
            states = pbag([s['stack_status'] for s in content['stacks']])
            if not (states == expected_states):
                msg("Waiting for group {} to reach desired group state.\n"
                    "{} (actual) {} (expected)"
                    .format(self.group.group_id, states, expected_states))
                raise TransientRetryError(
                    "Group states of {} did not match expected {})"
                    .format(states, expected_states))

            msg("Success: desired group state reached:\n{}"
                .format(expected_states))
            return self.rcs

        def poll():
            return self.get_stack_list().addCallback(check)

        expected_states = pbag(expected_states)

        return retry_and_timeout(
            poll, timeout,
            can_retry=terminal_errors_except(TransientRetryError),
            next_interval=repeating_interval(period),
            clock=reactor,
            deferred_description=(
                "Waiting for group {} to reach state {}".format(
                    self.group.group_id, str(expected_states))))

    def test_change_lb_node(self):
        """
        If a desired CLB mapping is in the set of current configs,
        but the configuration is wrong, `converge_lb_state` returns a
        :class:`ChangeCLBNode` object.  RCv3 nodes cannot be changed - they are
        either right or wrong.
        """
        clb_desc = CLBDescription(lb_id='5', port=80)
        rcv3_desc = RCv3Description(
            lb_id='c6fe49fa-114a-4ea4-9425-0af8b30ff1e7')

        current = [CLBNode(node_id='123', address='1.1.1.1',
                           description=copy_clb_desc(clb_desc, weight=5)),
                   RCv3Node(node_id='234', cloud_server_id='abc',
                            description=rcv3_desc)]
        self.assertEqual(
            converge(
                DesiredGroupState(server_config={}, capacity=1),
                set([server('abc', ServerState.ACTIVE,
                            servicenet_address='1.1.1.1',
                            desired_lbs=s(clb_desc, rcv3_desc))]),
                set(current),
                0),
            pbag([
                ChangeCLBNode(lb_id='5', node_id='123', weight=1,
                              condition=CLBNodeCondition.ENABLED,
                              type=CLBNodeType.PRIMARY)]))

    def test_add_to_lb(self):
        """
        If a desired LB config is not in the set of current configs,
        `converge_lb_state` returns the relevant adding-to-load-balancer
        steps (:class:`AddNodesToCLB` in the case of CLB,
        :class:`BulkAddToRCv3` in the case of RCv3).
        """
        clb_desc = CLBDescription(lb_id='5', port=80)
        rcv3_desc = RCv3Description(
            lb_id='c6fe49fa-114a-4ea4-9425-0af8b30ff1e7')

        self.assertEqual(
            converge(
                DesiredGroupState(server_config={}, capacity=1),
                set([server('abc', ServerState.ACTIVE,
                            servicenet_address='1.1.1.1',
                            desired_lbs=s(clb_desc, rcv3_desc))]),
                set(),
                0),
            pbag([
                AddNodesToCLB(
                    lb_id='5',
                    address_configs=s(('1.1.1.1', clb_desc))),
                BulkAddToRCv3(
                    lb_node_pairs=s(
                        ('c6fe49fa-114a-4ea4-9425-0af8b30ff1e7', 'abc')))
            ]))

    def test_plan(self):
        """An optimized plan is returned. Steps are limited."""
        desc = CLBDescription(lb_id='5', port=80)
        desired_lbs = s(desc)
        desired_group_state = DesiredGroupState(
            server_config={}, capacity=20, desired_lbs=desired_lbs)

        result = plan(
            desired_group_state,
            set([server('server1', state=ServerState.ACTIVE,
                        servicenet_address='1.1.1.1',
                        desired_lbs=desired_lbs),
                 server('server2', state=ServerState.ACTIVE,
                        servicenet_address='1.2.3.4',
                        desired_lbs=desired_lbs)]),
            set(),
            0,
            build_timeout=3600)

        self.assertEqual(
            result,
            pbag([
                AddNodesToCLB(
                    lb_id='5',
                    address_configs=s(('1.1.1.1', desc), ('1.2.3.4', desc))
                )] + [CreateServer(server_config=pmap({}))] * 10))

 def test_converge_active_servers_ignores_servers_to_be_deleted(self):
     """
     Only servers in active that are not being deleted will have their
     load balancers converged.
     """
     desc = CLBDescription(lb_id='5', port=80)
     desired_lbs = s(desc)
     self.assertEqual(
         converge(
             DesiredGroupState(server_config={}, capacity=1,
                               desired_lbs=desired_lbs),
             set([server('abc', ServerState.ACTIVE,
                         servicenet_address='1.1.1.1', created=0,
                         desired_lbs=desired_lbs),
                  server('bcd', ServerState.ACTIVE,
                         servicenet_address='2.2.2.2', created=1,
                         desired_lbs=desired_lbs)]),
             set(),
             0),
         pbag([
             DeleteServer(server_id='abc'),
             AddNodesToCLB(
                 lb_id='5',
                 address_configs=s(('2.2.2.2', desc)))
         ]))

 def _filter_only_lb_steps(self, steps):
     """
     Converge may do other things to a server depending on its draining
     state.  This suite of tests is only testing how it handles the load
     balancer, so ignore steps that are not load-balancer related.
     """
     return pbag([step for step in steps if type(step) in self.LB_STEPS])

 def test_bulk_remove_from_rcv3(self):
     """Logs :obj:`BulkRemoveFromRCv3`."""
     adds = pbag([
         BulkRemoveFromRCv3(lb_node_pairs=pset([
             ('lb1', 'node1'), ('lb1', 'node2'),
             ('lb2', 'node2'), ('lb2', 'node3'),
             ('lb3', 'node4')])),
         BulkRemoveFromRCv3(lb_node_pairs=pset([
             ('lba', 'nodea'), ('lba', 'nodeb'),
             ('lb1', 'nodea')]))
     ])
     self.assert_logs(adds, [
         Log('convergence-remove-rcv3-nodes',
             fields={'lb_id': 'lb1', 'servers': ['node1', 'node2', 'nodea'],
                     'cloud_feed': True}),
         Log('convergence-remove-rcv3-nodes',
             fields={'lb_id': 'lb2', 'servers': ['node2', 'node3'],
                     'cloud_feed': True}),
         Log('convergence-remove-rcv3-nodes',
             fields={'lb_id': 'lb3', 'servers': ['node4'],
                     'cloud_feed': True}),
         Log('convergence-remove-rcv3-nodes',
             fields={'lb_id': 'lba', 'servers': ['nodea', 'nodeb'],
                     'cloud_feed': True})
     ])

    def test_same_clb_multiple_ports(self):
        """
        It's possible to have the same cloud load balancer using multiple ports
        on the host.

        (use case: running multiple single-threaded server processes on a
        machine)
        """
        desired = s(CLBDescription(lb_id='5', port=8080),
                    CLBDescription(lb_id='5', port=8081))
        current = []
        self.assertEqual(
            converge(
                DesiredGroupState(server_config={}, capacity=1),
                set([server('abc', ServerState.ACTIVE,
                            servicenet_address='1.1.1.1',
                            desired_lbs=desired)]),
                set(current),
                0),
            pbag([
                AddNodesToCLB(
                    lb_id='5',
                    address_configs=s(('1.1.1.1',
                                       CLBDescription(lb_id='5', port=8080)))),
                AddNodesToCLB(
                    lb_id='5',
                    address_configs=s(('1.1.1.1',
                                       CLBDescription(lb_id='5', port=8081))))
                ]))

    def test_scale_down_order(self):
        """Preferred order of servers to delete when scaling down:

        - WAIT_WITH_TIMEOUT
        - WAIT
        - AVOID_REPLACING
        - CONSIDER_ACTIVE
        """
        order = (Destiny.WAIT_WITH_TIMEOUT, Destiny.WAIT,
                 Destiny.AVOID_REPLACING, Destiny.CONSIDER_AVAILABLE)
        examples = {Destiny.WAIT_WITH_TIMEOUT: ServerState.BUILD,
                    Destiny.WAIT: ServerState.HARD_REBOOT,
                    Destiny.AVOID_REPLACING: ServerState.RESCUE,
                    Destiny.CONSIDER_AVAILABLE: ServerState.ACTIVE}
        for combo in combinations(order, 2):
            before, after = combo
            also = []
            if after == Destiny.WAIT:
                # If we're waiting for some other servers we need to also
                # expect a ConvergeLater
                also = [ConvergeLater(reasons=[
                    ErrorReason.String(
                        'waiting for temporarily unavailable server to become '
                        'ACTIVE')],
                    limited=True)]

            self.assertEqual(
                converge(
                    DesiredGroupState(server_config={}, capacity=2),
                    set([server('abc', examples[after], created=0),
                         server('def', examples[before], created=1),
                         server('ghi', examples[after], created=2)]),
                    set(),
                    0),
                pbag([DeleteServer(server_id='def')] + also))

 def test_draining_server_can_be_deleted_if_all_lbs_can_be_removed(self):
     """
     If draining server can be removed from all the load balancers, the
     server can be deleted.
     """
     self.assertEqual(
         converge(
             DesiredGroupState(server_config={}, capacity=0),
             set([server('abc', state=ServerState.ACTIVE,
                         metadata=dict([DRAINING_METADATA]),
                         servicenet_address='1.1.1.1',
                         desired_lbs=s(self.clb_desc, self.rcv3_desc))]),
             set([CLBNode(node_id='1', address='1.1.1.1',
                          description=copy_clb_desc(
                              self.clb_desc,
                              condition=CLBNodeCondition.DRAINING)),
                  RCv3Node(node_id='2', cloud_server_id='abc',
                           description=self.rcv3_desc)]),
             0),
         pbag([
             DeleteServer(server_id='abc'),
             RemoveNodesFromCLB(lb_id='1', node_ids=s('1')),
             BulkRemoveFromRCv3(lb_node_pairs=s(
                 (self.rcv3_desc.lb_id, 'abc')))
         ]))

 def test_rcv3_mixed(self):
     """
     Multiple BulkAddToRCv3 and BulkRemoveFromRCv3 steps are combined
     into one BulkAddToRCv3 step and one BulkRemoveFromRCv3 step
     """
     steps = [
         BulkAddToRCv3(
             lb_node_pairs=pset([("l1", "s1"), ("l1", "s2")])),
         # Same pair for different class does not conflict
         BulkRemoveFromRCv3(lb_node_pairs=pset([("l1", "s1")])),
         BulkAddToRCv3(lb_node_pairs=pset([("l1", "s3")])),
         BulkRemoveFromRCv3(
             lb_node_pairs=pset([("l3", "s3"), ("l2", "s3")]))
     ]
     self.assertEqual(
         optimize_steps(steps),
         pbag([
             BulkAddToRCv3(
                 lb_node_pairs=pset([
                     ("l1", "s1"), ("l1", "s2"), ("l1", "s3")])),
             BulkRemoveFromRCv3(
                 lb_node_pairs=pset([
                     ("l1", "s1"), ("l3", "s3"), ("l2", "s3")]))
         ])
     )

 def test_delete_error_state_servers_with_lb_nodes(self):
     """
     If a server we created enters error state and it is attached to one
     or more load balancers, it will be removed from its load balancers
     as well as get deleted.  (Tests that error state servers are not
     excluded from converging load balancer state.)
     """
     self.assertEqual(
         converge(
             DesiredGroupState(server_config={}, capacity=1),
             set([server('abc', ServerState.ERROR,
                         servicenet_address='1.1.1.1',
                         desired_lbs=s(CLBDescription(lb_id='5', port=80),
                                       CLBDescription(lb_id='5', port=8080),
                                       RCv3Description(lb_id='6')))]),
             set([CLBNode(address='1.1.1.1', node_id='3',
                          description=CLBDescription(lb_id='5',
                                                     port=80)),
                  CLBNode(address='1.1.1.1', node_id='5',
                          description=CLBDescription(lb_id='5',
                                                     port=8080)),
                  RCv3Node(node_id='123', cloud_server_id='abc',
                           description=RCv3Description(lb_id='6'))]),
             0),
         pbag([
             DeleteServer(server_id='abc'),
             RemoveNodesFromCLB(lb_id='5', node_ids=s('3')),
             RemoveNodesFromCLB(lb_id='5', node_ids=s('5')),
             BulkRemoveFromRCv3(lb_node_pairs=s(('6', 'abc'))),
             CreateServer(server_config=pmap()),
         ]))

def decode(obj):
    if isinstance(obj, ExtType):
        if obj.code == TYPE_PSET:
            unpacked_data = unpackb(obj.data,
                                    use_list=False,
                                    encoding='utf-8')
            return pset(decode(item) for item in unpacked_data)
        if obj.code == TYPE_PLIST:
            unpacked_data = unpackb(obj.data,
                                    use_list=False,
                                    encoding='utf-8')
            return plist(decode(item) for item in unpacked_data)
        if obj.code == TYPE_PBAG:
            unpacked_data = unpackb(obj.data,
                                    use_list=False,
                                    encoding='utf-8')
            return pbag(decode(item) for item in unpacked_data)
        module_name, class_name, *data = unpackb(obj.data,
                                                 use_list=False,
                                                 encoding='utf-8')
        cls = getattr(sys.modules[module_name],
                      class_name)
        return cls(*(decode(item) for item in data))
    if isinstance(obj, tuple):
        return pvector(decode(item) for item in obj)
    if isinstance(obj, dict):
        new_dict = dict()
        for key in obj.keys():
            new_dict[decode(key)] = decode(obj[key])
        return pmap(new_dict)
    return obj