control_flow.py 36.0 KB
Newer Older
Y
Yu Yang 已提交
1 2 3 4
from ..layer_helper import LayerHelper, unique_name
from ..framework import Program, Variable, Operator
from .. import core
from tensor import assign, fill_constant
D
dzhwinter 已提交
5 6
import contextlib

Q
QI JUN 已提交
7
__all__ = [
Y
Yang Yang 已提交
8 9 10 11 12 13
    'split_lod_tensor', 'merge_lod_tensor', 'BlockGuard',
    'BlockGuardWithCompletion', 'StaticRNNMemoryLink', 'WhileGuard', 'While',
    'lod_rank_table', 'max_sequence_len', 'topk', 'lod_tensor_to_array',
    'array_to_lod_tensor', 'increment', 'array_write', 'create_array',
    'less_than', 'array_read', 'shrink_memory', 'array_length', 'IfElse',
    'DynamicRNN', 'ConditionalBlock', 'StaticRNN', 'ParallelDo'
D
dzhwinter 已提交
14 15
]

Y
Yu Yang 已提交
16

17
def split_lod_tensor(input, mask, level=0):
18
    helper = LayerHelper('split_lod_tensor', **locals())
F
fengjiayi 已提交
19 20
    out_true = helper.create_tmp_variable(dtype=input.dtype)
    out_false = helper.create_tmp_variable(dtype=input.dtype)
21 22 23 24 25 26 27 28 29 30 31 32
    helper.append_op(
        type='split_lod_tensor',
        inputs={
            'X': input,
            'Mask': mask,
        },
        outputs={'OutTrue': out_true,
                 'OutFalse': out_false},
        attrs={'level': level})
    return out_true, out_false


33
def merge_lod_tensor(in_true, in_false, x, mask, level=0):
34
    helper = LayerHelper('merge_lod_tensor', **locals())
F
fengjiayi 已提交
35
    out = helper.create_tmp_variable(dtype=in_true.dtype)
36 37 38 39 40 41 42 43 44 45 46
    helper.append_op(
        type='merge_lod_tensor',
        inputs={'X': x,
                'Mask': mask,
                'InTrue': in_true,
                'InFalse': in_false},
        outputs={'Out': out},
        attrs={'level': level})
    return out


Y
Yu Yang 已提交
47 48
class BlockGuard(object):
    """
49 50 51 52
    BlockGuard class.

    BlockGuard class is used to create a sub-block in a program by
    using the Python `with` keyword.
Y
Yu Yang 已提交
53 54
    """

55 56
    def __init__(self, main_program):
        if not isinstance(main_program, Program):
Y
Yu Yang 已提交
57
            raise TypeError("BlockGuard takes a program")
58
        self.main_program = main_program
Y
Yu Yang 已提交
59 60

    def __enter__(self):
61
        self.main_program.create_block()
Y
Yu Yang 已提交
62 63

    def __exit__(self, exc_type, exc_val, exc_tb):
64
        self.main_program.rollback()
Y
Yu Yang 已提交
65 66 67 68 69
        if exc_type is not None:
            return False  # re-raise exception
        return True


Y
Yang Yang 已提交
70
class ParallelDo(object):
71
    """
Y
Yang Yang 已提交
72
    ParallelDo class.
73

Y
Yang Yang 已提交
74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105
    ParallelDo class is used to create a ParallelDo.
    """

    def __init__(self, places, name=None):
        self.helper = LayerHelper("parallel_do", name=name)
        self.inputs = []
        self.places = places
        self.outputs = []
        self.status = StaticRNN.BEFORE_RNN_BLOCK

    def do(self):
        return BlockGuardWithCompletion(self)

    def parent_block(self):
        prog = self.helper.main_program
        parent_idx = prog.current_block().parent_idx
        assert parent_idx >= 0
        parent_block = prog.block(parent_idx)
        return parent_block

    def __call__(self, *args, **kwargs):
        if self.status != StaticRNN.AFTER_RNN_BLOCK:
            raise ValueError("RNN output can only be retrieved after rnn block")
        if len(self.outputs) == 0:
            raise ValueError("RNN has no output")
        elif len(self.outputs) == 1:
            return self.outputs[0]
        else:
            return self.outputs

    def read_input(self, var):
        self.inputs.append(var)
Y
Yang Yang 已提交
106
        return var
Y
Yang Yang 已提交
107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152

    def write_output(self, var):
        self.outputs.append(var)

    def get_parameters(self):
        main_program = self.helper.main_program
        current_block = main_program.current_block()
        parent_block = self.parent_block()

        local_inputs = set()

        for op in current_block.ops:
            for oname in op.output_names:
                for out_var_name in op.output(oname):
                    local_inputs.add(out_var_name)

        for var in self.inputs:
            local_inputs.add(var.name)

        params = list()
        for op in current_block.ops:
            for iname in op.input_names:
                for in_var_name in op.input(iname):
                    if in_var_name not in local_inputs:
                        params.append(in_var_name)

        return [parent_block.var(name) for name in params]

    def complete_op(self):
        main_program = self.helper.main_program
        current_block = main_program.current_block()
        parent_block = self.parent_block()

        step_scope = parent_block.create_var(
            type=core.VarDesc.VarType.STEP_SCOPES)

        inputs = [parent_block.var(i.name) for i in self.inputs]

        parent_block.append_op(
            type='parallel_do',
            inputs={
                'inputs': inputs,
                'parameters': self.get_parameters(),
                'places': self.places
            },
            outputs={'outputs': self.outputs,
Y
Yang Yang 已提交
153
                     'parallel_scopes': [step_scope]},
Y
Yang Yang 已提交
154 155 156 157 158 159 160 161
            attrs={'sub_block': current_block})


class BlockGuardWithCompletion(BlockGuard):
    """
    BlockGuardWithCompletion class.

    BlockGuardWithCompletion class is used to create an op with a block in a program.
162 163
    """

Y
Yu Yang 已提交
164
    def __init__(self, rnn):
Y
Yang Yang 已提交
165 166 167 168
        if not (isinstance(rnn, StaticRNN) or isinstance(rnn, ParallelDo)):
            raise TypeError(
                "BlockGuardWithCompletion takes a StaticRNN or ParallelDo")
        super(BlockGuardWithCompletion, self).__init__(rnn.helper.main_program)
Y
Yu Yang 已提交
169 170 171 172
        self.rnn = rnn

    def __enter__(self):
        self.rnn.status = StaticRNN.IN_RNN_BLOCK
Y
Yang Yang 已提交
173
        return super(BlockGuardWithCompletion, self).__enter__()
Y
Yu Yang 已提交
174 175

    def __exit__(self, exc_type, exc_val, exc_tb):
Y
Yu Yang 已提交
176 177
        if exc_type is not None:
            return False
Y
Yu Yang 已提交
178
        self.rnn.status = StaticRNN.AFTER_RNN_BLOCK
Y
Yang Yang 已提交
179 180 181
        self.rnn.complete_op()
        return super(BlockGuardWithCompletion, self).__exit__(exc_type, exc_val,
                                                              exc_tb)
Y
Yu Yang 已提交
182 183 184 185


class StaticRNNMemoryLink(object):
    """
186 187 188 189 190 191 192 193 194 195 196 197
    StaticRNNMemoryLink class.

    Args:
        init: the initial variable for Memory
        init: Variable
        pre_mem: the memory variable in previous time step
        pre_mem: Variable
        mem: the memory variable in current time step
        mem: Variable

    StaticRNNMemoryLink class is used to create a link between two
    memory cells of a StaticRNN.
Y
Yu Yang 已提交
198 199 200 201 202 203 204 205 206
    """

    def __init__(self, init, pre_mem, mem=None):
        self.init = init
        self.pre_mem = pre_mem
        self.mem = mem


class StaticRNN(object):
207 208 209 210 211 212
    """
    StaticRNN class.

    StaticRNN class is used to create a StaticRNN. The RNN will have its
    own parameters like inputs, outputs, memories, status and length.
    """
Y
Yu Yang 已提交
213 214 215 216
    BEFORE_RNN_BLOCK = 0
    IN_RNN_BLOCK = 1
    AFTER_RNN_BLOCK = 2

217 218
    def __init__(self, name=None):
        self.helper = LayerHelper("static_rnn", name=name)
Y
Yu Yang 已提交
219 220 221 222 223 224 225 226
        self.memories = {}  # memory map, from pre_mem.name --> MemoryLink
        self.inputs = []  # input variable list in current block
        self.outputs = []  # output variable list in parent block
        self.status = StaticRNN.BEFORE_RNN_BLOCK  # status flag.
        # sequence length, since it is a static RNN, sequence length are fixed.
        self.seq_len = None

    def step(self):
Y
Yang Yang 已提交
227
        return BlockGuardWithCompletion(self)
Y
Yu Yang 已提交
228 229 230 231 232

    def _assert_in_rnn_block_(self, method):
        if self.status != StaticRNN.IN_RNN_BLOCK:
            raise ValueError("You must invoke {0} in rnn block".format(method))

233 234 235 236 237 238 239
    def memory(self,
               init=None,
               shape=None,
               batch_ref=None,
               init_value=0.0,
               init_batch_dim_idx=0,
               ref_batch_dim_idx=1):
240 241 242 243 244 245 246 247 248
        """
        Args:
            init: boot memory, if not set, a shape, batch_ref must be provided
            shape: shape of the boot memory
            batch_ref: batch size reference variable
            init_value: the init value of boot memory
            init_batch_dim_idx: the index of batch size in init's dimension
            ref_batch_dim_idx: the index of batch size in batch_ref's dimension
        """
Y
Yu Yang 已提交
249 250
        self._assert_in_rnn_block_('memory')
        if init is None:
251
            if shape is None or batch_ref is None:
Y
Yu Yang 已提交
252
                raise ValueError(
253
                    "if init is None, memory at least need shape and batch_ref")
Y
Yu Yang 已提交
254 255 256
            parent_block = self.parent_block()
            var_name = unique_name("@".join([self.helper.name, "memory_boot"]))
            boot_var = parent_block.create_var(
257 258
                name=var_name,
                shape=shape,
F
fengjiayi 已提交
259
                dtype=batch_ref.dtype,
260
                persistable=False)
Y
Yu Yang 已提交
261 262

            parent_block.append_op(
263 264
                type="fill_constant_batch_size_like",
                inputs={'Input': [batch_ref]},
Y
Yu Yang 已提交
265 266 267
                outputs={'Out': [boot_var]},
                attrs={
                    'value': init_value,
268
                    'shape': boot_var.shape,
F
fengjiayi 已提交
269
                    'dtype': boot_var.dtype,
270 271
                    'input_dim_idx': ref_batch_dim_idx,
                    'output_dim_idx': init_batch_dim_idx
Y
Yu Yang 已提交
272 273 274 275 276 277
                })

            return self.memory(init=boot_var)
        else:
            pre_mem = self.helper.create_variable(
                name=unique_name("@".join([self.helper.name, "mem"])),
F
fengjiayi 已提交
278
                dtype=init.dtype,
Y
Yu Yang 已提交
279 280 281 282 283 284 285 286 287 288
                shape=init.shape)
            self.memories[pre_mem.name] = StaticRNNMemoryLink(
                init=init, pre_mem=pre_mem)
            return pre_mem

    def step_input(self, x):
        self._assert_in_rnn_block_('step_input')
        if not isinstance(x, Variable):
            raise TypeError("step input takes a Variable")
        if self.seq_len is None:
Y
Yu Yang 已提交
289 290
            self.seq_len = x.shape[0]
        elif self.seq_len != x.shape[0]:
Y
Yu Yang 已提交
291 292 293
            raise ValueError("Static RNN only take fix seq_len input")

        ipt = self.helper.create_variable(
F
fengjiayi 已提交
294
            name=x.name, dtype=x.dtype, shape=list(x.shape[1:]), type=x.type)
Y
Yu Yang 已提交
295 296 297 298 299 300 301 302
        self.inputs.append(ipt)
        return ipt

    def step_output(self, o):
        self._assert_in_rnn_block_('step_output')
        if not isinstance(o, Variable):
            raise TypeError("step output takes a Variable")

F
fengjiayi 已提交
303
        tmp_o = self.helper.create_tmp_variable(dtype=o.dtype)
Y
Yu Yang 已提交
304 305 306 307
        self.helper.append_op(
            type='rnn_memory_helper',
            inputs={'X': [o]},
            outputs={'Out': tmp_o},
F
fengjiayi 已提交
308
            attrs={'dtype': o.dtype})
Y
Yu Yang 已提交
309

Y
Yu Yang 已提交
310
        out_var = self.parent_block().create_var(
Y
Yu Yang 已提交
311 312
            name=tmp_o.name,
            shape=[self.seq_len] + list(tmp_o.shape),
F
fengjiayi 已提交
313
            dtype=tmp_o.dtype)
Y
Yu Yang 已提交
314 315 316 317 318 319 320 321 322 323 324 325 326

        self.outputs.append(out_var)

    def output(self, *outputs):
        for each in outputs:
            self.step_output(each)

    def update_memory(self, mem, var):
        if not isinstance(mem, Variable) or not isinstance(var, Variable):
            raise TypeError("update memory should take variables")
        self.memories[mem.name].mem = var

    def parent_block(self):
327
        prog = self.helper.main_program
Y
Yu Yang 已提交
328 329 330 331 332 333 334 335 336 337 338 339 340 341 342
        parent_idx = prog.current_block().parent_idx
        assert parent_idx >= 0
        parent_block = prog.block(parent_idx)
        return parent_block

    def __call__(self, *args, **kwargs):
        if self.status != StaticRNN.AFTER_RNN_BLOCK:
            raise ValueError("RNN output can only be retrieved after rnn block")
        if len(self.outputs) == 0:
            raise ValueError("RNN has no output")
        elif len(self.outputs) == 1:
            return self.outputs[0]
        else:
            return self.outputs

Y
Yang Yang 已提交
343
    def complete_op(self):
344 345
        main_program = self.helper.main_program
        rnn_block = main_program.current_block()
Y
Yu Yang 已提交
346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384
        parent_block = self.parent_block()

        local_inputs = set()

        for op in rnn_block.ops:
            assert isinstance(op, Operator)
            for oname in op.output_names:
                for out_var_name in op.output(oname):
                    local_inputs.add(out_var_name)

        for var in self.inputs:
            local_inputs.add(var.name)
        for m in self.memories:
            local_inputs.add(m)

        params = list()
        for op in rnn_block.ops:
            assert isinstance(op, Operator)
            for iname in op.input_names:
                for in_var_name in op.input(iname):
                    if in_var_name not in local_inputs:
                        params.append(in_var_name)

        parameters = [parent_block.var(name) for name in params]

        step_scope = parent_block.create_var(
            type=core.VarDesc.VarType.STEP_SCOPES)

        inlinks = [parent_block.var(i.name) for i in self.inputs]
        outlinks = self.outputs

        boot_memories = []
        pre_memories = []
        memories = []
        for _, mem in self.memories.iteritems():
            boot_memories.append(mem.init)
            pre_memories.append(mem.pre_mem.name)
            mem_var = rnn_block.var(mem.mem.name)
            assert isinstance(mem_var, Variable)
F
fengjiayi 已提交
385
            new_mem = self.helper.create_tmp_variable(dtype=mem_var.dtype)
Y
Yu Yang 已提交
386 387 388 389 390

            rnn_block.append_op(
                type='rnn_memory_helper',
                inputs={'X': [mem_var]},
                outputs={'Out': [new_mem]},
F
fengjiayi 已提交
391
                attrs={'dtype': mem_var.dtype})
Y
Yu Yang 已提交
392 393 394 395 396 397 398 399 400 401 402 403 404 405 406

            memories.append(new_mem.name)

        parent_block.append_op(
            type='recurrent',
            inputs={
                'inputs': inlinks,
                'initial_states': boot_memories,
                'parameters': parameters
            },
            outputs={'outputs': outlinks,
                     'step_scopes': [step_scope]},
            attrs={
                'ex_states': pre_memories,
                'states': memories,
407
                'sub_block': rnn_block
Y
Yu Yang 已提交
408
            })
Y
Yu Yang 已提交
409 410


Y
Yang Yang(Tony) 已提交
411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434
class WhileGuard(BlockGuard):
    def __init__(self, while_op):
        if not isinstance(while_op, While):
            raise TypeError("WhileGuard takes a while op")
        super(WhileGuard, self).__init__(while_op.helper.main_program)
        self.while_op = while_op

    def __enter__(self):
        self.while_op.status = While.IN_WHILE_BLOCK
        return super(WhileGuard, self).__enter__()

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            return False
        self.while_op.status = While.AFTER_WHILE_BLOCK
        self.while_op.complete()
        return super(WhileGuard, self).__exit__(exc_type, exc_val, exc_tb)


class While(object):
    BEFORE_WHILE_BLOCK = 0
    IN_WHILE_BLOCK = 1
    AFTER_WHILE_BLOCK = 2

435 436
    def __init__(self, cond, name=None):
        self.helper = LayerHelper("while", name=name)
Y
Yang Yang(Tony) 已提交
437 438 439 440
        self.status = While.BEFORE_WHILE_BLOCK
        if not isinstance(cond, Variable):
            raise TypeError("condition should be a variable")
        assert isinstance(cond, Variable)
F
fengjiayi 已提交
441
        if cond.dtype != core.DataType.BOOL:
Y
Yang Yang(Tony) 已提交
442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483
            raise TypeError("condition should be a bool variable")
        if reduce(lambda a, b: a * b, cond.shape, 1) != 1:
            raise TypeError("condition should be a bool scalar")
        self.cond_var = cond

    def block(self):
        return WhileGuard(self)

    def complete(self):
        main_program = self.helper.main_program
        while_block = main_program.current_block()
        parent_block = main_program.block(main_program.current_block()
                                          .parent_idx)

        inner_outputs = {self.cond_var.name}
        x_name_list = set()
        for op in while_block.ops:
            for iname in op.input_names:
                for in_var_name in op.input(iname):
                    if in_var_name not in inner_outputs:
                        x_name_list.add(in_var_name)

            for oname in op.output_names:
                for out_var_name in op.output(oname):
                    inner_outputs.add(out_var_name)

        out_vars = []
        for inner_out_name in inner_outputs:
            if inner_out_name in parent_block.vars:
                out_vars.append(parent_block.var(inner_out_name))

        step_scope = parent_block.create_var(
            type=core.VarDesc.VarType.STEP_SCOPES)

        parent_block.append_op(
            type='while',
            inputs={
                'X': [parent_block.var(x_name) for x_name in x_name_list],
                'Condition': [self.cond_var]
            },
            outputs={'Out': out_vars,
                     'StepScopes': [step_scope]},
484
            attrs={'sub_block': while_block})
Y
Yang Yang(Tony) 已提交
485 486


487
def lod_rank_table(x, level=0):
488 489 490 491
    """
    This function creates an operator for creating a LOD_RANK_TABLE
    using the input x.
    """
Y
Yu Yang 已提交
492 493 494 495 496 497 498 499 500 501
    helper = LayerHelper("lod_rank_table", **locals())
    table = helper.create_variable(
        type=core.VarDesc.VarType.LOD_RANK_TABLE,
        name=unique_name("lod_rank_table"))
    helper.append_op(
        type='lod_rank_table',
        inputs={'X': x},
        outputs={'Out': table},
        attrs={'level': level})
    return table
Y
Yu Yang 已提交
502 503


504
def max_sequence_len(rank_table):
F
fengjiayi 已提交
505
    """
Y
Yu Yang 已提交
506
    This function creates an operator to calculate the length of
F
fengjiayi 已提交
507 508 509 510 511 512 513 514 515 516 517
    max seqence through input rank_table(should be a lod_rank_table)
    """
    helper = LayerHelper("max_seqence_len", **locals())
    res = helper.create_tmp_variable(dtype="int64")
    helper.append_op(
        type="max_sequence_len",
        inputs={"RankTable": rank_table},
        outputs={"Out": res})
    return res


518
def topk(input, k):
Y
Yu Yang 已提交
519 520 521 522 523 524 525 526 527 528 529 530
    helper = LayerHelper('topk', **locals())
    topk_out = helper.create_tmp_variable(dtype=input.data_type)
    topk_indices = helper.create_tmp_variable(dtype='int64')
    helper.append_op(
        type='top_k',
        inputs={'X': [input]},
        outputs={'Out': [topk_out],
                 'Indices': [topk_indices]},
        attrs={'k': k})
    return topk_out, topk_indices


531
def lod_tensor_to_array(x, table):
532 533 534 535
    """
    This function creates an operator to convert an LOD_Tensor to
    an array.
    """
536 537 538
    helper = LayerHelper("lod_tensor_to_array", **locals())
    array = helper.create_variable(
        name=unique_name("lod_tensor_to_array"),
539
        type=core.VarDesc.VarType.LOD_TENSOR_ARRAY,
F
fengjiayi 已提交
540
        dtype=x.dtype)
541 542 543 544 545 546 547 548
    helper.append_op(
        type='lod_tensor_to_array',
        inputs={'X': x,
                'RankTable': table},
        outputs={'Out': array})
    return array


549
def array_to_lod_tensor(x, table):
550 551 552 553
    """
    This function creates an operator to convert an array to a
    LOD_Tensor.
    """
554
    helper = LayerHelper("array_to_lod_tensor", **locals())
F
fengjiayi 已提交
555
    tmp = helper.create_tmp_variable(dtype=x.dtype)
556 557 558 559 560 561 562 563
    helper.append_op(
        type="array_to_lod_tensor",
        inputs={'X': x,
                'RankTable': table},
        outputs={'Out': tmp})
    return tmp


564
def increment(x, value=1.0, in_place=True):
565 566 567 568 569
    """
    This function creates an operator to increment each value in the input
    `x` by an amount: `value` as mentioned in the input parameter. This
    operation is performed in-place by default.
    """
Y
Yu Yang 已提交
570
    helper = LayerHelper("increment", **locals())
Y
Yang Yang(Tony) 已提交
571
    if not in_place:
F
fengjiayi 已提交
572
        out = helper.create_tmp_variable(dtype=x.dtype)
Y
Yang Yang(Tony) 已提交
573 574
    else:
        out = x
Y
Yu Yang 已提交
575 576 577
    helper.append_op(
        type='increment',
        inputs={'X': [x]},
Y
Yang Yu 已提交
578
        outputs={'Out': [out]},
579
        attrs={'step': float(value)})
Y
Yang Yu 已提交
580
    return out
Y
Yu Yang 已提交
581 582


583
def array_write(x, i, array=None):
584 585 586 587
    """
    This function creates an operator to write the data out as a
    LOD_TENSOR_ARRAY.
    """
Y
Yu Yang 已提交
588 589 590 591 592
    helper = LayerHelper('array_write', **locals())
    if array is None:
        array = helper.create_variable(
            name="{0}.out".format(helper.name),
            type=core.VarDesc.VarType.LOD_TENSOR_ARRAY,
F
fengjiayi 已提交
593
            dtype=x.dtype)
Y
Yu Yang 已提交
594 595 596 597 598 599 600 601
    helper.append_op(
        type='write_to_array',
        inputs={'X': [x],
                'I': [i]},
        outputs={'Out': [array]})
    return array


602
def create_array(dtype):
Y
Yang Yang(Tony) 已提交
603 604 605 606 607 608 609
    helper = LayerHelper("array", **locals())
    return helper.create_variable(
        name="{0}.out".format(helper.name),
        type=core.VarDesc.VarType.LOD_TENSOR_ARRAY,
        dtype=dtype)


610
def less_than(x, y, cond=None, **ignored):
Y
Yang Yang(Tony) 已提交
611 612 613 614 615 616 617 618 619 620 621
    helper = LayerHelper("less_than", **locals())
    if cond is None:
        cond = helper.create_tmp_variable(dtype='bool')
        cond.stop_gradient = True

    helper.append_op(
        type='less_than', inputs={'X': [x],
                                  'Y': [y]}, outputs={'Out': [cond]})
    return cond


622
def array_read(array, i):
623 624 625 626
    """
    This function creates an operator to read the data in as a
    LOD_TENSOR_ARRAY.
    """
Y
Yu Yang 已提交
627 628 629 630 631
    helper = LayerHelper('array_read', **locals())
    if not isinstance(
            array,
            Variable) or array.type != core.VarDesc.VarType.LOD_TENSOR_ARRAY:
        raise TypeError("array should be tensor array vairable")
F
fengjiayi 已提交
632
    out = helper.create_tmp_variable(dtype=array.dtype)
Y
Yu Yang 已提交
633 634 635 636 637 638
    helper.append_op(
        type='read_from_array',
        inputs={'X': [array],
                'I': [i]},
        outputs={'Out': [out]})
    return out
Y
Yang Yu 已提交
639 640


641
def shrink_memory(x, i, table):
642 643 644 645
    """
    This function creates an operator to shrink_rnn_memory using the RankTable
    as mentioned in the input parameter.
    """
Y
Yang Yu 已提交
646
    helper = LayerHelper('shrink_memory', **locals())
F
fengjiayi 已提交
647
    out = helper.create_tmp_variable(dtype=x.dtype)
Y
Yang Yu 已提交
648
    helper.append_op(
Y
Yang Yu 已提交
649
        type='shrink_rnn_memory',
Y
Yang Yu 已提交
650 651 652 653 654 655
        inputs={'X': [x],
                'I': [i],
                'RankTable': [table]},
        outputs={'Out': [out]},
        attrs={})
    return out
Y
Yang Yu 已提交
656 657


658
def array_length(array):
659 660 661 662
    """
    This function creates an operator to find the length of the
    LOD_TENSOR_ARRAY.
    """
Y
Yang Yu 已提交
663 664 665 666 667 668
    helper = LayerHelper('array_length', **locals())
    tmp = helper.create_tmp_variable(dtype='int64')
    tmp.stop_gradient = True
    helper.append_op(
        type='lod_array_length', inputs={'X': [array]}, outputs={'Out': [tmp]})
    return tmp
Y
Yu Yang 已提交
669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687


class ConditionalBlockGuard(BlockGuard):
    def __init__(self, block):
        if not isinstance(block, ConditionalBlock):
            raise TypeError("block should be conditional block")
        super(ConditionalBlockGuard, self).__init__(block.helper.main_program)
        self.block = block

    def __enter__(self):
        return super(ConditionalBlockGuard, self).__enter__()

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.block.complete()
        return super(ConditionalBlockGuard, self).__exit__(exc_type, exc_val,
                                                           exc_tb)


class ConditionalBlock(object):
688
    def __init__(self, inputs, name=None):
Y
Yu Yang 已提交
689 690 691 692
        for each_input in inputs:
            if not isinstance(each_input, Variable):
                raise TypeError("Each input should be variable")
        self.inputs = inputs
693
        self.helper = LayerHelper('conditional_block', name=name)
Y
Yu Yang 已提交
694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736

    def block(self):
        return ConditionalBlockGuard(self)

    def complete(self):
        inside_block = self.helper.main_program.current_block()
        parent_block = self.helper.main_program.block(inside_block.parent_idx)

        intermediate = set()
        params = set()

        for each_op in inside_block.ops:
            assert isinstance(each_op, Operator)
            for iname in each_op.input_names:
                for in_var_name in each_op.input(iname):
                    if in_var_name not in intermediate:
                        params.add(in_var_name)

            for oname in each_op.output_names:
                for out_var_name in each_op.output(oname):
                    intermediate.add(out_var_name)
        input_set = set([ipt.name for ipt in self.inputs])

        param_list = [
            parent_block.var(each_name) for each_name in params
            if each_name not in input_set
        ]

        out_list = [
            parent_block.var(var_name) for var_name in parent_block.vars
            if var_name not in intermediate
        ]

        step_scope = parent_block.create_var(
            type=core.VarDesc.VarType.STEP_SCOPES)
        parent_block.append_op(
            type='conditional_block',
            inputs={
                'X': self.inputs,
                'Params': param_list,
            },
            outputs={'Out': out_list,
                     'Scope': [step_scope]},
737
            attrs={'sub_block': inside_block})
Y
Yu Yang 已提交
738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777


class IfElseBlockGuard(object):
    def __init__(self, is_true, ifelse):
        if not isinstance(ifelse, IfElse):
            raise TypeError("ifelse must be an instance of IfElse class")

        if ifelse.status != IfElse.OUT_IF_ELSE_BLOCKS:
            raise ValueError("You cannot invoke IfElse.block() inside a block")

        self.is_true = is_true
        self.ie = ifelse
        if is_true:
            self.cond_block = ifelse.conditional_true_block
        else:
            self.cond_block = ifelse.conditional_false_block

        if not isinstance(self.cond_block, ConditionalBlock):
            raise TypeError("Unexpected situation")

        self.cond_block = self.cond_block.block()

    def __enter__(self):
        self.ie.status = IfElse.IN_IF_ELSE_TRUE_BLOCKS if self.is_true else IfElse.IN_IF_ELSE_FALSE_BLOCKS
        self.cond_block.__enter__()

    def __exit__(self, exc_type, exc_val, exc_tb):
        if not self.cond_block.__exit__(exc_type, exc_val, exc_tb):
            # re-raise inside exception
            return False
        if len(self.ie.output_table[1 if self.is_true else 0]) == 0:
            raise ValueError("Must set output inside block")
        self.ie.status = IfElse.OUT_IF_ELSE_BLOCKS


class IfElse(object):
    OUT_IF_ELSE_BLOCKS = 0
    IN_IF_ELSE_TRUE_BLOCKS = 1
    IN_IF_ELSE_FALSE_BLOCKS = 2

778
    def __init__(self, cond, name=None):
Y
Yu Yang 已提交
779 780
        if not isinstance(cond, Variable):
            raise TypeError("cond must be a Variable")
781
        self.helper = LayerHelper('ifelse', name=name)
Y
Yu Yang 已提交
782 783 784 785 786 787 788 789 790 791 792 793 794 795
        self.cond = cond
        self.input_table = {}
        self.status = IfElse.OUT_IF_ELSE_BLOCKS
        self.conditional_true_block = ConditionalBlock(inputs=[self.cond])
        self.conditional_false_block = ConditionalBlock(inputs=[self.cond])
        self.output_table = ([], [])  # (true_outs, false_outs)

    def input(self, x):
        if self.status == IfElse.OUT_IF_ELSE_BLOCKS:
            raise ValueError("input must in true/false blocks")
        if id(x) not in self.input_table:
            parent_block = self.parent_block()
            out_true = parent_block.create_var(
                name=unique_name('ifelse_input' + self.helper.name),
F
fengjiayi 已提交
796
                dtype=x.dtype)
Y
Yu Yang 已提交
797 798 799

            out_false = parent_block.create_var(
                name=unique_name('ifelse_input' + self.helper.name),
F
fengjiayi 已提交
800
                dtype=x.dtype)
Y
Yu Yang 已提交
801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841
            parent_block.append_op(
                type='split_lod_tensor',
                inputs={
                    'X': x,
                    'Mask': self.cond,
                },
                outputs={'OutTrue': out_true,
                         'OutFalse': out_false},
                attrs={'level': 0})
            self.input_table[id(x)] = (out_true, out_false)
        else:
            out_true, out_false = self.input_table[id(x)]

        if self.status == IfElse.IN_IF_ELSE_TRUE_BLOCKS:
            return out_true
        else:
            return out_false

    def parent_block(self):
        current_block = self.helper.main_program.current_block()
        return self.helper.main_program.block(current_block.parent_idx)

    def true_block(self):
        return IfElseBlockGuard(True, self)

    def false_block(self):
        return IfElseBlockGuard(False, self)

    def output(self, *outs):
        if self.status == self.OUT_IF_ELSE_BLOCKS:
            raise ValueError("output can only be invoked in the sub-block")

        out_table = self.output_table[1 if self.status ==
                                      self.IN_IF_ELSE_TRUE_BLOCKS else 0]
        parent_block = self.parent_block()
        for each_out in outs:
            if not isinstance(each_out, Variable):
                raise TypeError("Each output should be a variable")
            # create outside tensor
            outside_out = parent_block.create_var(
                name=unique_name("_".join([self.helper.name, 'output'])),
F
fengjiayi 已提交
842
                dtype=each_out.dtype)
Y
Yu Yang 已提交
843 844 845
            out_table.append(outside_out)

            # assign local var to outside
846
            assign(input=each_out, output=outside_out)
Y
Yu Yang 已提交
847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869

    def __call__(self):
        if self.status != self.OUT_IF_ELSE_BLOCKS:
            raise ValueError("IfElse::__call__ must be out of sub-block")
        false_len, true_len = map(len, self.output_table)
        if false_len == 0 and true_len == 0:
            raise ValueError("Must invoke true_block/false_block before "
                             "__call__")
        elif false_len != true_len and false_len != 0 and true_len != 0:
            raise ValueError("The output side must be same")
        elif false_len == 0 or true_len == 0:
            return self.output_table[0 if false_len != 0 else 1]

        # else none of false_len/true_len is zero
        # merge together
        rlist = []
        for false_var, true_var in zip(*self.output_table):
            rlist.append(
                merge_lod_tensor(
                    in_true=true_var,
                    in_false=false_var,
                    mask=self.cond,
                    x=self.cond,
870
                    level=0))
Y
Yu Yang 已提交
871
        return rlist
872 873 874 875 876 877 878


class DynamicRNN(object):
    BEFORE_RNN = 0
    IN_RNN = 1
    AFTER_RNN = 2

879 880
    def __init__(self, name=None):
        self.helper = LayerHelper('dynamic_rnn', name=name)
881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933
        self.status = DynamicRNN.BEFORE_RNN
        self.lod_rank_table = None
        self.max_seq_len = None
        self.step_idx = None
        self.zero_idx = fill_constant(shape=[1], value=0, dtype='int64')
        self.mem_dict = dict()
        self.output_array = []
        self.outputs = []
        self.cond = self.helper.create_tmp_variable(dtype='bool')
        self.cond.stop_gradient = False
        self.while_op = While(self.cond)
        self.input_array = []
        self.mem_link = []

    def step_input(self, x):
        self._assert_in_rnn_block_("step_input")
        if not isinstance(x, Variable):
            raise TypeError(
                "step_input() can only take a Variable as its input")
        parent_block = self._parent_block_()
        if self.lod_rank_table is None:
            self.lod_rank_table = parent_block.create_var(
                name=unique_name('lod_rank_table'),
                type=core.VarDesc.VarType.LOD_RANK_TABLE)
            self.lod_rank_table.stop_gradient = True
            parent_block.append_op(
                type='lod_rank_table',
                inputs={"X": x},
                outputs={"Out": self.lod_rank_table})
            self.max_seq_len = parent_block.create_var(
                name=unique_name('dynamic_rnn_max_seq_len'), dtype='int64')
            self.max_seq_len.stop_gradient = False
            parent_block.append_op(
                type='max_sequence_len',
                inputs={'RankTable': self.lod_rank_table},
                outputs={"Out": self.max_seq_len})
            self.cond.stop_gradient = True
            parent_block.append_op(
                type='less_than',
                inputs={'X': self.step_idx,
                        'Y': self.max_seq_len},
                outputs={'Out': self.cond})

        input_array = parent_block.create_var(
            name=unique_name('dynamic_rnn_input_array'),
            type=core.VarDesc.VarType.LOD_TENSOR_ARRAY,
            dtype=x.dtype)
        self.input_array.append((input_array, x.dtype))
        parent_block.append_op(
            type='lod_tensor_to_array',
            inputs={'X': x,
                    'RankTable': self.lod_rank_table},
            outputs={'Out': input_array})
934
        return array_read(array=input_array, i=self.step_idx)
935 936 937 938 939 940 941 942 943 944

    @contextlib.contextmanager
    def block(self):
        if self.status != DynamicRNN.BEFORE_RNN:
            raise ValueError("rnn.block() can only be invoke once")
        self.step_idx = fill_constant(shape=[1], dtype='int64', value=0)
        self.step_idx.stop_gradient = False
        self.status = DynamicRNN.IN_RNN
        with self.while_op.block():
            yield
945
            increment(x=self.step_idx, value=1.0, in_place=True)
946 947

            for new_mem, mem_array in self.mem_link:
948 949 950
                array_write(x=new_mem, i=self.step_idx, array=mem_array)

            less_than(x=self.step_idx, y=self.max_seq_len, cond=self.cond)
951 952 953 954 955

        self.status = DynamicRNN.AFTER_RNN
        for each_array in self.output_array:
            self.outputs.append(
                array_to_lod_tensor(
956
                    x=each_array, table=self.lod_rank_table))
957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982

    def __call__(self, *args, **kwargs):
        if self.status != DynamicRNN.AFTER_RNN:
            raise ValueError(
                "Dynamic RNN outputs can only be retrieved after rnn block")
        if len(self.outputs) == 1:
            return self.outputs[0]
        else:
            return self.outputs

    def memory(self, init=None, shape=None, value=0.0, dtype='float32'):
        self._assert_in_rnn_block_('memory')
        if init is not None:
            if not isinstance(init, Variable):
                raise TypeError(
                    "The input arg `init` of memory() must be a Variable")
            parent_block = self._parent_block_()
            mem_array = parent_block.create_var(
                name=unique_name('dynamic_rnn_mem_array'),
                type=core.VarDesc.VarType.LOD_TENSOR_ARRAY,
                dtype=init.dtype)
            parent_block.append_op(
                type='write_to_array',
                inputs={'X': init,
                        'I': self.zero_idx},
                outputs={'Out': mem_array})
983
            retv = array_read(array=mem_array, i=self.step_idx)
984
            retv = shrink_memory(
985
                x=retv, i=self.step_idx, table=self.lod_rank_table)
986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054
            self.mem_dict[retv.name] = mem_array
            return retv
        else:
            if len(self.input_array) == 0:
                raise ValueError(
                    "step_input should be invoked before memory(shape=..., value=...)"
                )
            parent_block = self._parent_block_()
            init = parent_block.create_var(
                name=unique_name('mem_init'), dtype=dtype)
            arr, dtype = self.input_array[0]
            in0 = parent_block.create_var(name=unique_name('in0'), dtype=dtype)
            parent_block.append_op(
                type='read_from_array',
                inputs={'X': [arr],
                        'I': [self.zero_idx]},
                outputs={'Out': [in0]})
            parent_block.append_op(
                type='fill_constant_batch_size_like',
                inputs={'Input': [in0]},
                outputs={'Out': [init]},
                attrs={
                    'shape': [-1] + shape,
                    'value': float(value),
                    'dtype': init.dtype
                })
            return self.memory(init=init)

    def update_memory(self, ex_mem, new_mem):
        self._assert_in_rnn_block_('update_memory')
        if not isinstance(ex_mem, Variable):
            raise TypeError("The input arg `ex_mem` of update_memory() must "
                            "be a Variable")
        if not isinstance(new_mem, Variable):
            raise TypeError("The input arg `new_mem` of update_memory() must "
                            "be a Variable")

        mem_array = self.mem_dict.get(ex_mem.name, None)
        if mem_array is None:
            raise ValueError("Please invoke memory before update_memory")
        if self.lod_rank_table is None:
            raise ValueError("Please invoke step_input before update_memory")

        self.mem_link.append((new_mem, mem_array))

    def output(self, *outputs):
        self._assert_in_rnn_block_('output')
        parent_block = self._parent_block_()
        for each in outputs:
            outside_array = parent_block.create_var(
                name=unique_name("_".join(
                    [self.helper.name, "output_array", each.name])),
                type=core.VarDesc.VarType.LOD_TENSOR_ARRAY,
                dtype=each.dtype)
            array_write(x=each, i=self.step_idx, array=outside_array)
            self.output_array.append(outside_array)

    def _parent_block_(self):
        prog = self.helper.main_program
        parent_idx = prog.current_block().parent_idx
        assert parent_idx >= 0
        parent_block = prog.block(parent_idx)

        return parent_block

    def _assert_in_rnn_block_(self, method):
        if self.status != DynamicRNN.IN_RNN:
            raise ValueError("{0} can only be invoked inside rnn block.".format(
                method))