# Copyright 2015 The TensorFlow Authors. All Rights Reserved.

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

# ==============================================================================

"""Tests for tensorflow.ops.data_flow_ops.FIFOQueue."""

from __future__ import absolute_import

from __future__ import division

from __future__ import print_function

import random

import time

import numpy as np

from six.moves import xrange # pylint: disable=redefined-builtin

from tensorflow.core.protobuf import config_pb2

from tensorflow.python.client import session as session_lib

from tensorflow.python.framework import constant_op

from tensorflow.python.framework import dtypes as dtypes_lib

from tensorflow.python.framework import errors_impl

from tensorflow.python.framework import ops

from tensorflow.python.framework import tensor_shape

from tensorflow.python.framework import test_util

from tensorflow.python.ops import array_ops

from tensorflow.python.ops import control_flow_ops

from tensorflow.python.ops import data_flow_ops

from tensorflow.python.platform import test

from tensorflow.python.util import compat

@test_util.run_v1_only("FIFOQueue removed from v2")

class FIFOQueueTest(test.TestCase):

def testConstructor(self):

with ops.Graph().as_default():

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32, name="Q")

self.assertTrue(isinstance(q.queue_ref, ops.Tensor))

self.assertProtoEquals("""

name:'Q' op:'FIFOQueueV2'

attr { key: 'component_types' value { list { type: DT_FLOAT } } }

attr { key: 'shapes' value { list {} } }

attr { key: 'capacity' value { i: 10 } }

attr { key: 'container' value { s: '' } }

attr { key: 'shared_name' value { s: '' } }

""", q.queue_ref.op.node_def)

def testMultiQueueConstructor(self):

with ops.Graph().as_default():

q = data_flow_ops.FIFOQueue(

5, (dtypes_lib.int32, dtypes_lib.float32),

shared_name="foo",

name="Q")

self.assertTrue(isinstance(q.queue_ref, ops.Tensor))

self.assertProtoEquals("""

name:'Q' op:'FIFOQueueV2'

attr { key: 'component_types' value { list {

type: DT_INT32 type : DT_FLOAT

} } }

attr { key: 'shapes' value { list {} } }

attr { key: 'capacity' value { i: 5 } }

attr { key: 'container' value { s: '' } }

attr { key: 'shared_name' value { s: 'foo' } }

""", q.queue_ref.op.node_def)

def testConstructorWithShapes(self):

with ops.Graph().as_default():

q = data_flow_ops.FIFOQueue(

5, (dtypes_lib.int32, dtypes_lib.float32),

shapes=(tensor_shape.TensorShape([1, 1, 2, 3]),

tensor_shape.TensorShape([5, 8])),

name="Q")

self.assertTrue(isinstance(q.queue_ref, ops.Tensor))

self.assertProtoEquals("""

name:'Q' op:'FIFOQueueV2'

attr { key: 'component_types' value { list {

type: DT_INT32 type : DT_FLOAT

} } }

attr { key: 'shapes' value { list {

shape { dim { size: 1 }

dim { size: 1 }

dim { size: 2 }

dim { size: 3 } }

shape { dim { size: 5 }

dim { size: 8 } }

} } }

attr { key: 'capacity' value { i: 5 } }

attr { key: 'container' value { s: '' } }

attr { key: 'shared_name' value { s: '' } }

""", q.queue_ref.op.node_def)

def testEnqueue(self):

with self.cached_session():

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32)

enqueue_op = q.enqueue((10.0,))

enqueue_op.run()

def testEnqueueHalf(self):

with self.cached_session():

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float16)

enqueue_op = q.enqueue((10.0,))

enqueue_op.run()

def testEnqueueWithShape(self):

with self.cached_session():

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32, shapes=(3, 2))

enqueue_correct_op = q.enqueue(([[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]],))

enqueue_correct_op.run()

with self.assertRaises(ValueError):

q.enqueue(([[1.0, 2.0, 3.0], [4.0, 5.0, 6.0]],))

self.assertEqual(1, q.size().eval())

def testEnqueueManyWithShape(self):

with self.cached_session():

q = data_flow_ops.FIFOQueue(

10, [dtypes_lib.int32, dtypes_lib.int32], shapes=[(), (2,)])

q.enqueue_many([[1, 2, 3, 4], [[1, 1], [2, 2], [3, 3], [4, 4]]]).run()

self.assertEqual(4, q.size().eval())

@test_util.run_in_graph_and_eager_modes

def testMultipleDequeues(self):

q = data_flow_ops.FIFOQueue(10, [dtypes_lib.int32], shapes=[()])

self.evaluate(q.enqueue_many([[1, 2, 3]]))

a, b, c = self.evaluate([q.dequeue(), q.dequeue(), q.dequeue()])

self.assertAllEqual(set([1, 2, 3]), set([a, b, c]))

@test_util.run_in_graph_and_eager_modes

def testQueuesDontShare(self):

q = data_flow_ops.FIFOQueue(10, [dtypes_lib.int32], shapes=[()])

self.evaluate(q.enqueue(1))

q2 = data_flow_ops.FIFOQueue(10, [dtypes_lib.int32], shapes=[()])

self.evaluate(q2.enqueue(2))

self.assertAllEqual(self.evaluate(q2.dequeue()), 2)

self.assertAllEqual(self.evaluate(q.dequeue()), 1)

def testEnqueueDictWithoutNames(self):

with self.cached_session():

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32)

with self.assertRaisesRegexp(ValueError, "must have names"):

q.enqueue({"a": 12.0})

with self.assertRaisesRegexp(ValueError, "must have names"):

q.enqueue_many({"a": [12.0, 13.0]})

def testParallelEnqueue(self):

# We need each thread to keep its own device stack or the device scopes

# won't be properly nested.

ops.get_default_graph().switch_to_thread_local()

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32)

elems = [10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0, 100.0]

enqueue_ops = [q.enqueue((x,)) for x in elems]

dequeued_t = q.dequeue()

# Run one producer thread for each element in elems.

def enqueue(enqueue_op):

self.evaluate(enqueue_op)

threads = [

self.checkedThread(

target=enqueue, args=(e,)) for e in enqueue_ops

]

for thread in threads:

thread.start()

for thread in threads:

thread.join()

# Dequeue every element using a single thread.

results = []

for _ in xrange(len(elems)):

results.append(dequeued_t.eval())

self.assertItemsEqual(elems, results)

def testParallelDequeue(self):

# We need each thread to keep its own device stack or the device scopes

# won't be properly nested.

ops.get_default_graph().switch_to_thread_local()

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32)

elems = [10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0, 100.0]

enqueue_ops = [q.enqueue((x,)) for x in elems]

dequeued_t = q.dequeue()

# Enqueue every element using a single thread.

for enqueue_op in enqueue_ops:

enqueue_op.run()

# Run one consumer thread for each element in elems.

results = []

def dequeue():

results.append(self.evaluate(dequeued_t))

threads = [self.checkedThread(target=dequeue) for _ in enqueue_ops]

for thread in threads:

thread.start()

for thread in threads:

thread.join()

self.assertItemsEqual(elems, results)

def testDequeue(self):

with self.cached_session():

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32)

elems = [10.0, 20.0, 30.0]

enqueue_ops = [q.enqueue((x,)) for x in elems]

dequeued_t = q.dequeue()

for enqueue_op in enqueue_ops:

enqueue_op.run()

for i in xrange(len(elems)):

vals = self.evaluate(dequeued_t)

self.assertEqual([elems[i]], vals)

def testDequeueHalf(self):

with self.cached_session():

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float16)

elems = [10.0, 20.0, 30.0]

enqueue_ops = [q.enqueue((x,)) for x in elems]

dequeued_t = q.dequeue()

for enqueue_op in enqueue_ops:

enqueue_op.run()

for i in xrange(len(elems)):

vals = self.evaluate(dequeued_t)

self.assertEqual([elems[i]], vals)

def testEnqueueAndBlockingDequeue(self):

# We need each thread to keep its own device stack or the device scopes

# won't be properly nested.

ops.get_default_graph().switch_to_thread_local()

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(3, dtypes_lib.float32)

elems = [10.0, 20.0, 30.0]

enqueue_ops = [q.enqueue((x,)) for x in elems]

dequeued_t = q.dequeue()

def enqueue():

# The enqueue_ops should run after the dequeue op has blocked.

# TODO(mrry): Figure out how to do this without sleeping.

time.sleep(0.1)

for enqueue_op in enqueue_ops:

self.evaluate(enqueue_op)

results = []

def dequeue():

for _ in xrange(len(elems)):

results.append(self.evaluate(dequeued_t))

enqueue_thread = self.checkedThread(target=enqueue)

dequeue_thread = self.checkedThread(target=dequeue)

enqueue_thread.start()

dequeue_thread.start()

enqueue_thread.join()

dequeue_thread.join()

for elem, result in zip(elems, results):

self.assertEqual([elem], result)

def testMultiEnqueueAndDequeue(self):

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(10, (dtypes_lib.int32, dtypes_lib.float32))

elems = [(5, 10.0), (10, 20.0), (15, 30.0)]

enqueue_ops = [q.enqueue((x, y)) for x, y in elems]

dequeued_t = q.dequeue()

for enqueue_op in enqueue_ops:

enqueue_op.run()

for i in xrange(len(elems)):

x_val, y_val = self.evaluate(dequeued_t)

x, y = elems[i]

self.assertEqual([x], x_val)

self.assertEqual([y], y_val)

def testQueueSizeEmpty(self):

with self.cached_session():

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32)

self.assertEqual([0], q.size().eval())

def testQueueSizeAfterEnqueueAndDequeue(self):

with self.cached_session():

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32)

enqueue_op = q.enqueue((10.0,))

dequeued_t = q.dequeue()

size = q.size()

self.assertEqual([], size.get_shape())

enqueue_op.run()

self.assertEqual(1, self.evaluate(size))

dequeued_t.op.run()

self.assertEqual(0, self.evaluate(size))

def testEnqueueMany(self):

with self.cached_session():

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32)

elems = [10.0, 20.0, 30.0, 40.0]

enqueue_op = q.enqueue_many((elems,))

dequeued_t = q.dequeue()

enqueue_op.run()

enqueue_op.run()

for i in range(8):

vals = self.evaluate(dequeued_t)

self.assertEqual([elems[i % 4]], vals)

def testEmptyEnqueueMany(self):

with self.cached_session():

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32)

empty_t = constant_op.constant(

[], dtype=dtypes_lib.float32, shape=[0, 2, 3])

enqueue_op = q.enqueue_many((empty_t,))

size_t = q.size()

self.assertEqual([0], self.evaluate(size_t))

enqueue_op.run()

self.assertEqual([0], self.evaluate(size_t))

def testEmptyDequeueMany(self):

with self.cached_session():

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32, shapes=())

enqueue_op = q.enqueue((10.0,))

dequeued_t = q.dequeue_many(0)

self.assertEqual([], self.evaluate(dequeued_t).tolist())

enqueue_op.run()

self.assertEqual([], self.evaluate(dequeued_t).tolist())

def testEmptyDequeueUpTo(self):

with self.cached_session():

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32, shapes=())

enqueue_op = q.enqueue((10.0,))

dequeued_t = q.dequeue_up_to(0)

self.assertEqual([], self.evaluate(dequeued_t).tolist())

enqueue_op.run()

self.assertEqual([], self.evaluate(dequeued_t).tolist())

def testEmptyDequeueManyWithNoShape(self):

with self.cached_session():

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32)

# Expect the operation to fail due to the shape not being constrained.

with self.assertRaisesOpError("specified shapes"):

q.dequeue_many(0).eval()

def testMultiEnqueueMany(self):

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(10, (dtypes_lib.float32, dtypes_lib.int32))

float_elems = [10.0, 20.0, 30.0, 40.0]

int_elems = [[1, 2], [3, 4], [5, 6], [7, 8]]

enqueue_op = q.enqueue_many((float_elems, int_elems))

dequeued_t = q.dequeue()

enqueue_op.run()

enqueue_op.run()

for i in range(8):

float_val, int_val = self.evaluate(dequeued_t)

self.assertEqual(float_elems[i % 4], float_val)

self.assertAllEqual(int_elems[i % 4], int_val)

def testDequeueMany(self):

with self.cached_session():

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32, ())

elems = [10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0, 100.0]

enqueue_op = q.enqueue_many((elems,))

dequeued_t = q.dequeue_many(4)

enqueue_op.run()

self.assertAllEqual(elems[0:4], self.evaluate(dequeued_t))

self.assertAllEqual(elems[4:8], self.evaluate(dequeued_t))

def testDequeueUpToNoBlocking(self):

with self.cached_session():

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32, ())

elems = [10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0, 100.0]

enqueue_op = q.enqueue_many((elems,))

dequeued_t = q.dequeue_up_to(4)

enqueue_op.run()

self.assertAllEqual(elems[0:4], self.evaluate(dequeued_t))

self.assertAllEqual(elems[4:8], self.evaluate(dequeued_t))

def testMultiDequeueMany(self):

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(

10, (dtypes_lib.float32, dtypes_lib.int32), shapes=((), (2,)))

float_elems = [

10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0, 100.0

]

int_elems = [[1, 2], [3, 4], [5, 6], [7, 8], [9, 10], [11, 12], [13, 14],

[15, 16], [17, 18], [19, 20]]

enqueue_op = q.enqueue_many((float_elems, int_elems))

dequeued_t = q.dequeue_many(4)

dequeued_single_t = q.dequeue()

enqueue_op.run()

float_val, int_val = self.evaluate(dequeued_t)

self.assertAllEqual(float_elems[0:4], float_val)

self.assertAllEqual(int_elems[0:4], int_val)

self.assertEqual(float_val.shape, dequeued_t[0].get_shape())

self.assertEqual(int_val.shape, dequeued_t[1].get_shape())

float_val, int_val = self.evaluate(dequeued_t)

self.assertAllEqual(float_elems[4:8], float_val)

self.assertAllEqual(int_elems[4:8], int_val)

float_val, int_val = self.evaluate(dequeued_single_t)

self.assertAllEqual(float_elems[8], float_val)

self.assertAllEqual(int_elems[8], int_val)

self.assertEqual(float_val.shape, dequeued_single_t[0].get_shape())

self.assertEqual(int_val.shape, dequeued_single_t[1].get_shape())

def testMultiDequeueUpToNoBlocking(self):

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(

10, (dtypes_lib.float32, dtypes_lib.int32), shapes=((), (2,)))

float_elems = [

10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0, 100.0

]

int_elems = [[1, 2], [3, 4], [5, 6], [7, 8], [9, 10], [11, 12], [13, 14],

[15, 16], [17, 18], [19, 20]]

enqueue_op = q.enqueue_many((float_elems, int_elems))

dequeued_t = q.dequeue_up_to(4)

enqueue_op.run()

float_val, int_val = self.evaluate(dequeued_t)

self.assertAllEqual(float_elems[0:4], float_val)

self.assertAllEqual(int_elems[0:4], int_val)

self.assertEqual([None], dequeued_t[0].get_shape().as_list())

self.assertEqual([None, 2], dequeued_t[1].get_shape().as_list())

float_val, int_val = self.evaluate(dequeued_t)

self.assertAllEqual(float_elems[4:8], float_val)

self.assertAllEqual(int_elems[4:8], int_val)

def testHighDimension(self):

with self.cached_session():

q = data_flow_ops.FIFOQueue(10, dtypes_lib.int32, (4, 4, 4, 4))

elems = np.array([[[[[x] * 4] * 4] * 4] * 4 for x in range(10)], np.int32)

enqueue_op = q.enqueue_many((elems,))

dequeued_t = q.dequeue_many(10)

enqueue_op.run()

self.assertAllEqual(dequeued_t.eval(), elems)

def testEnqueueWrongShape(self):

q = data_flow_ops.FIFOQueue(10, (dtypes_lib.int32, dtypes_lib.int32), ((),

(2)))

with self.assertRaises(ValueError):

q.enqueue(([1, 2], [2, 2]))

with self.assertRaises(ValueError):

q.enqueue_many((7, [[1, 2], [3, 4], [5, 6]]))

def testBatchSizeMismatch(self):

q = data_flow_ops.FIFOQueue(10, (dtypes_lib.int32, dtypes_lib.int32,

dtypes_lib.int32), ((), (), ()))

with self.assertRaises(ValueError):

q.enqueue_many(([1, 2, 3], [1, 2], [1, 2, 3]))

with self.assertRaises(ValueError):

q.enqueue_many(

([1, 2, 3], [1, 2], array_ops.placeholder(dtypes_lib.int32)))

with self.assertRaises(ValueError):

q.enqueue_many(

(array_ops.placeholder(dtypes_lib.int32), [1, 2], [1, 2, 3]))

def testEnqueueManyEmptyTypeConversion(self):

q = data_flow_ops.FIFOQueue(10, (dtypes_lib.int32, dtypes_lib.float32), (

(), ()))

enq = q.enqueue_many(([], []))

self.assertEqual(dtypes_lib.int32, enq.inputs[1].dtype)

self.assertEqual(dtypes_lib.float32, enq.inputs[2].dtype)

def testEnqueueWrongType(self):

q = data_flow_ops.FIFOQueue(10, (dtypes_lib.int32, dtypes_lib.float32), (

(), ()))

with self.assertRaises(ValueError):

q.enqueue((array_ops.placeholder(dtypes_lib.int32),

array_ops.placeholder(dtypes_lib.int32)))

with self.assertRaises(ValueError):

q.enqueue_many((array_ops.placeholder(dtypes_lib.int32),

array_ops.placeholder(dtypes_lib.int32)))

def testEnqueueWrongShapeAtRuntime(self):

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(10, (dtypes_lib.int32, dtypes_lib.int32), (

(2, 2), (3, 3)))

elems_ok = np.array([1] * 4).reshape((2, 2)).astype(np.int32)

elems_bad = array_ops.placeholder(dtypes_lib.int32)

enqueue_op = q.enqueue((elems_ok, elems_bad))

with self.assertRaisesRegexp(errors_impl.InvalidArgumentError,

r"Expected \[3,3\], got \[3,4\]"):

sess.run([enqueue_op],

feed_dict={elems_bad: np.array([1] * 12).reshape((3, 4))})

def testEnqueueDequeueManyWrongShape(self):

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(10, (dtypes_lib.int32, dtypes_lib.int32), (

(2, 2), (3, 3)))

elems_ok = np.array([1] * 8).reshape((2, 2, 2)).astype(np.int32)

elems_bad = array_ops.placeholder(dtypes_lib.int32)

enqueue_op = q.enqueue_many((elems_ok, elems_bad))

dequeued_t = q.dequeue_many(2)

with self.assertRaisesRegexp(errors_impl.InvalidArgumentError,

"Shape mismatch in tuple component 1. "

r"Expected \[2,3,3\], got \[2,3,4\]"):

sess.run([enqueue_op],

feed_dict={elems_bad: np.array([1] * 24).reshape((2, 3, 4))})

self.evaluate(dequeued_t)

def testParallelEnqueueMany(self):

# We need each thread to keep its own device stack or the device scopes

# won't be properly nested.

ops.get_default_graph().switch_to_thread_local()

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(1000, dtypes_lib.float32, shapes=())

elems = [10.0 * x for x in range(100)]

enqueue_op = q.enqueue_many((elems,))

dequeued_t = q.dequeue_many(1000)

# Enqueue 100 items in parallel on 10 threads.

def enqueue():

self.evaluate(enqueue_op)

threads = [self.checkedThread(target=enqueue) for _ in range(10)]

for thread in threads:

thread.start()

for thread in threads:

thread.join()

self.assertItemsEqual(dequeued_t.eval(), elems * 10)

def testParallelDequeueMany(self):

# We need each thread to keep its own device stack or the device scopes

# won't be properly nested.

ops.get_default_graph().switch_to_thread_local()

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(1000, dtypes_lib.float32, shapes=())

elems = [10.0 * x for x in range(1000)]

enqueue_op = q.enqueue_many((elems,))

dequeued_t = q.dequeue_many(100)

enqueue_op.run()

# Dequeue 100 items in parallel on 10 threads.

dequeued_elems = []

def dequeue():

dequeued_elems.extend(self.evaluate(dequeued_t))

threads = [self.checkedThread(target=dequeue) for _ in range(10)]

for thread in threads:

thread.start()

for thread in threads:

thread.join()

self.assertItemsEqual(elems, dequeued_elems)

def testParallelDequeueUpTo(self):

# We need each thread to keep its own device stack or the device scopes

# won't be properly nested.

ops.get_default_graph().switch_to_thread_local()

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(1000, dtypes_lib.float32, shapes=())

elems = [10.0 * x for x in range(1000)]

enqueue_op = q.enqueue_many((elems,))

close_op = q.close()

dequeued_t = q.dequeue_up_to(101)

enqueue_op.run()

close_op.run()

# Dequeue up to 101 items in parallel on 10 threads, from closed queue.

dequeued_elems = []

def dequeue():

dequeued_elems.extend(self.evaluate(dequeued_t))

threads = [self.checkedThread(target=dequeue) for _ in range(10)]

for thread in threads:

thread.start()

for thread in threads:

thread.join()

self.assertItemsEqual(elems, dequeued_elems)

def testParallelEnqueueAndDequeue(self):

# We need each thread to keep its own device stack or the device scopes

# won't be properly nested.

ops.get_default_graph().switch_to_thread_local()

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(50, dtypes_lib.float32, shapes=())

initial_elements = [10.0] * 49

q.enqueue_many((initial_elements,)).run()

enqueue_op = q.enqueue((20.0,))

dequeued_t = q.dequeue()

def enqueue():

for _ in xrange(100):

self.evaluate(enqueue_op)

def dequeue():

for _ in xrange(100):

self.assertTrue(self.evaluate(dequeued_t) in (10.0, 20.0))

enqueue_threads = [self.checkedThread(target=enqueue) for _ in range(10)]

dequeue_threads = [self.checkedThread(target=dequeue) for _ in range(10)]

for enqueue_thread in enqueue_threads:

enqueue_thread.start()

for dequeue_thread in dequeue_threads:

dequeue_thread.start()

for enqueue_thread in enqueue_threads:

enqueue_thread.join()

for dequeue_thread in dequeue_threads:

dequeue_thread.join()

# Dequeue the initial count of elements to clean up.

cleanup_elems = q.dequeue_many(49).eval()

for elem in cleanup_elems:

self.assertTrue(elem in (10.0, 20.0))

def testMixtureOfEnqueueAndEnqueueMany(self):

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(10, dtypes_lib.int32, shapes=())

enqueue_placeholder = array_ops.placeholder(dtypes_lib.int32, shape=())

enqueue_op = q.enqueue((enqueue_placeholder,))

enqueuemany_placeholder = array_ops.placeholder(

dtypes_lib.int32, shape=(None,))

enqueuemany_op = q.enqueue_many((enqueuemany_placeholder,))

dequeued_t = q.dequeue()

close_op = q.close()

def dequeue():

for i in xrange(250):

self.assertEqual(i, self.evaluate(dequeued_t))

dequeue_thread = self.checkedThread(target=dequeue)

dequeue_thread.start()

elements_enqueued = 0

while elements_enqueued < 250:

# With equal probability, run Enqueue or enqueue_many.

if random.random() > 0.5:

enqueue_op.run({enqueue_placeholder: elements_enqueued})

elements_enqueued += 1

else:

count = random.randint(0, min(20, 250 - elements_enqueued))

range_to_enqueue = np.arange(

elements_enqueued, elements_enqueued + count, dtype=np.int32)

enqueuemany_op.run({enqueuemany_placeholder: range_to_enqueue})

elements_enqueued += count

close_op.run()

dequeue_thread.join()

self.assertEqual(0, q.size().eval())

def testMixtureOfDequeueAndDequeueMany(self):

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(10, dtypes_lib.int32, shapes=())

enqueue_op = q.enqueue_many((np.arange(250, dtype=np.int32),))

dequeued_t = q.dequeue()

count_placeholder = array_ops.placeholder(dtypes_lib.int32, shape=())

dequeuemany_t = q.dequeue_many(count_placeholder)

def enqueue():

self.evaluate(enqueue_op)

enqueue_thread = self.checkedThread(target=enqueue)

enqueue_thread.start()

elements_dequeued = 0

while elements_dequeued < 250:

# With equal probability, run Dequeue or dequeue_many.

if random.random() > 0.5:

self.assertEqual(elements_dequeued, self.evaluate(dequeued_t))

elements_dequeued += 1

else:

count = random.randint(0, min(20, 250 - elements_dequeued))

expected_range = np.arange(

elements_dequeued, elements_dequeued + count, dtype=np.int32)

self.assertAllEqual(expected_range,

dequeuemany_t.eval({

count_placeholder: count

}))

elements_dequeued += count

q.close().run()

enqueue_thread.join()

self.assertEqual(0, q.size().eval())

def testBlockingDequeueMany(self):

# We need each thread to keep its own device stack or the device scopes

# won't be properly nested.

ops.get_default_graph().switch_to_thread_local()

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32, ())

elems = [10.0, 20.0, 30.0, 40.0]

enqueue_op = q.enqueue_many((elems,))

dequeued_t = q.dequeue_many(4)

dequeued_elems = []

def enqueue():

# The enqueue_op should run after the dequeue op has blocked.

# TODO(mrry): Figure out how to do this without sleeping.

time.sleep(0.1)

self.evaluate(enqueue_op)

def dequeue():

dequeued_elems.extend(self.evaluate(dequeued_t).tolist())

enqueue_thread = self.checkedThread(target=enqueue)

dequeue_thread = self.checkedThread(target=dequeue)

enqueue_thread.start()

dequeue_thread.start()

enqueue_thread.join()

dequeue_thread.join()

self.assertAllEqual(elems, dequeued_elems)

def testBlockingDequeueUpTo(self):

# We need each thread to keep its own device stack or the device scopes

# won't be properly nested.

ops.get_default_graph().switch_to_thread_local()

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32, ())

elems = [10.0, 20.0, 30.0, 40.0]

enqueue_op = q.enqueue_many((elems,))

dequeued_t = q.dequeue_up_to(4)

dequeued_elems = []

def enqueue():

# The enqueue_op should run after the dequeue op has blocked.

# TODO(mrry): Figure out how to do this without sleeping.

time.sleep(0.1)

self.evaluate(enqueue_op)

def dequeue():

dequeued_elems.extend(self.evaluate(dequeued_t).tolist())

enqueue_thread = self.checkedThread(target=enqueue)

dequeue_thread = self.checkedThread(target=dequeue)

enqueue_thread.start()

dequeue_thread.start()

enqueue_thread.join()

dequeue_thread.join()

self.assertAllEqual(elems, dequeued_elems)

def testDequeueManyWithTensorParameter(self):

with self.cached_session():

# Define a first queue that contains integer counts.

dequeue_counts = [random.randint(1, 10) for _ in range(100)]

count_q = data_flow_ops.FIFOQueue(100, dtypes_lib.int32, ())

enqueue_counts_op = count_q.enqueue_many((dequeue_counts,))

total_count = sum(dequeue_counts)

# Define a second queue that contains total_count elements.

elems = [random.randint(0, 100) for _ in range(total_count)]

q = data_flow_ops.FIFOQueue(total_count, dtypes_lib.int32, ())

enqueue_elems_op = q.enqueue_many((elems,))

# Define a subgraph that first dequeues a count, then DequeuesMany

# that number of elements.

dequeued_t = q.dequeue_many(count_q.dequeue())

enqueue_counts_op.run()

enqueue_elems_op.run()

dequeued_elems = []

for _ in dequeue_counts:

dequeued_elems.extend(dequeued_t.eval())

self.assertEqual(elems, dequeued_elems)

def testDequeueFromClosedQueue(self):

with self.cached_session():

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32)

elems = [10.0, 20.0, 30.0, 40.0]

enqueue_op = q.enqueue_many((elems,))

close_op = q.close()

dequeued_t = q.dequeue()

enqueue_op.run()

close_op.run()

for elem in elems:

self.assertEqual([elem], self.evaluate(dequeued_t))

# Expect the operation to fail due to the queue being closed.

with self.assertRaisesRegexp(errors_impl.OutOfRangeError,

"is closed and has insufficient"):

self.evaluate(dequeued_t)

def testBlockingDequeueFromClosedQueue(self):

# We need each thread to keep its own device stack or the device scopes

# won't be properly nested.

ops.get_default_graph().switch_to_thread_local()

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32)

elems = [10.0, 20.0, 30.0, 40.0]

enqueue_op = q.enqueue_many((elems,))

close_op = q.close()

dequeued_t = q.dequeue()

enqueue_op.run()

def dequeue():

for elem in elems:

self.assertEqual([elem], self.evaluate(dequeued_t))

# Expect the operation to fail due to the queue being closed.

with self.assertRaisesRegexp(errors_impl.OutOfRangeError,

"is closed and has insufficient"):

self.evaluate(dequeued_t)

dequeue_thread = self.checkedThread(target=dequeue)

dequeue_thread.start()

# The close_op should run after the dequeue_thread has blocked.

# TODO(mrry): Figure out how to do this without sleeping.

time.sleep(0.1)

close_op.run()

dequeue_thread.join()

def testBlockingDequeueFromClosedEmptyQueue(self):

# We need each thread to keep its own device stack or the device scopes

# won't be properly nested.

ops.get_default_graph().switch_to_thread_local()

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32)

close_op = q.close()

dequeued_t = q.dequeue()

def dequeue():

# Expect the operation to fail due to the queue being closed.

with self.assertRaisesRegexp(errors_impl.OutOfRangeError,

"is closed and has insufficient"):

self.evaluate(dequeued_t)

dequeue_thread = self.checkedThread(target=dequeue)

dequeue_thread.start()

# The close_op should run after the dequeue_thread has blocked.

# TODO(mrry): Figure out how to do this without sleeping.

time.sleep(0.1)

close_op.run()

dequeue_thread.join()

def testBlockingDequeueManyFromClosedQueue(self):

# We need each thread to keep its own device stack or the device scopes

# won't be properly nested.

ops.get_default_graph().switch_to_thread_local()

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32, ())

elems = [10.0, 20.0, 30.0, 40.0]

enqueue_op = q.enqueue_many((elems,))

close_op = q.close()

dequeued_t = q.dequeue_many(4)

enqueue_op.run()

def dequeue():

self.assertAllEqual(elems, self.evaluate(dequeued_t))

# Expect the operation to fail due to the queue being closed.

with self.assertRaisesRegexp(errors_impl.OutOfRangeError,

"is closed and has insufficient"):

self.evaluate(dequeued_t)

dequeue_thread = self.checkedThread(target=dequeue)

dequeue_thread.start()

# The close_op should run after the dequeue_thread has blocked.

# TODO(mrry): Figure out how to do this without sleeping.

time.sleep(0.1)

close_op.run()

dequeue_thread.join()

def testBlockingDequeueManyButNotAllFromClosedQueue(self):

# We need each thread to keep its own device stack or the device scopes

# won't be properly nested.

ops.get_default_graph().switch_to_thread_local()

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32, ())

elems = [10.0, 20.0, 30.0, 40.0]

enqueue_op = q.enqueue_many((elems,))

close_op = q.close()

dequeued_t = q.dequeue_many(3)

enqueue_op.run()

def dequeue():

self.assertAllEqual(elems[:3], self.evaluate(dequeued_t))

# Expect the operation to fail due to the queue being closed.

with self.assertRaisesRegexp(errors_impl.OutOfRangeError,

"is closed and has insufficient"):

self.evaluate(dequeued_t)

dequeue_thread = self.checkedThread(target=dequeue)

dequeue_thread.start()

# The close_op should run after the dequeue_thread has blocked.

# TODO(mrry): Figure out how to do this without sleeping.

time.sleep(0.1)

close_op.run()

dequeue_thread.join()

def testDequeueUpToFromClosedQueueReturnsRemainder(self):

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(10, dtypes_lib.float32, ())

elems = [10.0, 20.0, 30.0, 40.0]

enqueue_op = q.enqueue_many((elems,))

close_op = q.close()

dequeued_t = q.dequeue_up_to(3)

enqueue_op.run()

def dequeue():

self.assertAllEqual(elems[:3], self.evaluate(dequeued_t))

self.assertAllEqual(elems[3:], self.evaluate(dequeued_t))

dequeue_thread = self.checkedThread(target=dequeue)

dequeue_thread.start()

# The close_op should run after the dequeue_thread has blocked.

# TODO(mrry): Figure out how to do this without sleeping.

time.sleep(0.1)

close_op.run()

dequeue_thread.join()

def testEnqueueManyLargerThanCapacityWithConcurrentDequeueMany(self):

with ops.Graph().as_default(), self.session() as sess:

q = data_flow_ops.FIFOQueue(4, dtypes_lib.float32, ())

elems = [10.0, 20.0, 30.0, 40.0]

enqueue_op = q.enqueue_many((elems,))

close_op = q.close()

dequeued_t = q.dequeue_many(3)

cleanup_dequeue_t = q.dequeue()

def enqueue():

sess.run(enqueue_op)

def dequeue():

self.assertAllEqual(elems[0:3], sess.run(dequeued_t))

with self.assertRaises(errors_impl.OutOfRangeError):

sess.run(dequeued_t)

self.assertEqual(elems[3], sess.run(cleanup_dequeue_t))

def close():

sess.run(close_op)

enqueue_thread = self.checkedThread(target=enqueue)

enqueue_thread.start()

dequeue_thread = self.checkedThread(target=dequeue)

dequeue_thread.start()

# The close_op should run after the dequeue_thread has blocked.

# TODO(mrry): Figure out how to do this without sleeping.

time.sleep(0.1)

close_thread = self.checkedThread(target=close)

close_thread.start()

enqueue_thread.join()

dequeue_thread.join()

close_thread.join()

def testClosedBlockingDequeueManyRestoresPartialBatch(self):

with self.cached_session() as sess:

q = data_flow_ops.FIFOQueue(4, (dtypes_lib.float32, dtypes_lib.float32), (

(), ()))

elems_a = [1.0, 2.0, 3.0]

elems_b = [10.0, 20.0, 30.0]

enqueue_op = q.enqueue_many((elems_a, elems_b))

dequeued_a_t, dequeued_b_t = q.dequeue_many(4)

cleanup_dequeue_a_t, cleanup_dequeue_b_t = q.dequeue()

close_op = q.close()

enqueue_op.run()

def dequeue():

with self.assertRaises(errors_impl.OutOfRangeError):

self.evaluate([dequeued_a_t, dequeued_b_t])

dequeue_thread = self.checkedThread(target=dequeue)