Enqueue and increment variable in Tensor Flow

TensorFlow

machine learning

enqueue operation

variable manipulation

programming tutorial

Enqueue and increment variable in Tensor Flow

ML System Design practice on Codemia

Design recommenders, ranking systems and training pipelines the way ML interviews actually ask for them, with worked solutions.

Practice ML system design

TensorFlow & Keras Machine Learning

Introduction

In TensorFlow 1.x, queues and variable increment operations were core building blocks for data pipelines and training loops. Queues (FIFOQueue, RandomShuffleQueue) managed asynchronous data feeding, while tf.Variable with assign_add handled counters like global step. TensorFlow 2.x replaced queues with tf.data.Dataset and simplified variable operations with eager execution. This article covers both the legacy queue-based approach and the modern tf.data equivalent.

TF 1.x Queue Operations

python

1import tensorflow as tf
2
3# TensorFlow 1.x style (graph mode)
4tf.compat.v1.disable_eager_execution()
5
6# Create a FIFO queue that holds integer tensors
7queue = tf.queue.FIFOQueue(capacity=10, dtypes=[tf.int32])
8
9# Enqueue values
10enqueue_op = queue.enqueue([1])
11enqueue_many_op = queue.enqueue_many([[2, 3, 4, 5]])
12
13# Dequeue a single value
14dequeue_op = queue.dequeue()
15
16# Get queue size
17size_op = queue.size()
18
19with tf.compat.v1.Session() as sess:
20    # Enqueue one element
21    sess.run(enqueue_op)
22    print(f"Size after enqueue: {sess.run(size_op)}")  # 1
23
24    # Enqueue multiple elements
25    sess.run(enqueue_many_op)
26    print(f"Size after enqueue_many: {sess.run(size_op)}")  # 5
27
28    # Dequeue elements one at a time
29    for _ in range(5):
30        value = sess.run(dequeue_op)
31        print(f"Dequeued: {value}")  # 1, 2, 3, 4, 5

FIFOQueue returns elements in the order they were added. RandomShuffleQueue returns elements in random order, which was commonly used for shuffling training data.

Incrementing Variables in TF 1.x

python

1import tensorflow as tf
2tf.compat.v1.disable_eager_execution()
3
4# Create a variable to use as a counter
5counter = tf.compat.v1.Variable(0, name="counter")
6
7# Increment operation
8increment_op = counter.assign_add(1)
9
10# Combined: dequeue and increment
11queue = tf.queue.FIFOQueue(capacity=100, dtypes=[tf.float32])
12enqueue_op = queue.enqueue([42.0])
13dequeue_op = queue.dequeue()
14
15with tf.compat.v1.Session() as sess:
16    sess.run(tf.compat.v1.global_variables_initializer())
17
18    # Enqueue some data
19    for _ in range(5):
20        sess.run(enqueue_op)
21
22    # Process queue: dequeue and increment counter
23    for _ in range(5):
24        value, _, count = sess.run([dequeue_op, increment_op, counter])
25        print(f"Value: {value}, Count: {count}")
26    # Value: 42.0, Count: 1
27    # Value: 42.0, Count: 2
28    # ...

Queue with Coordinator and Threads (TF 1.x Pattern)

python

1import tensorflow as tf
2import threading
3tf.compat.v1.disable_eager_execution()
4
5queue = tf.queue.FIFOQueue(capacity=5, dtypes=[tf.float32])
6enqueue_op = queue.enqueue([tf.random.uniform([])])
7dequeue_op = queue.dequeue()
8close_op = queue.close()
9
10global_step = tf.compat.v1.Variable(0, trainable=False)
11increment_step = global_step.assign_add(1)
12
13with tf.compat.v1.Session() as sess:
14    sess.run(tf.compat.v1.global_variables_initializer())
15
16    # Use a coordinator to manage threads
17    coord = tf.train.Coordinator()
18
19    def enqueue_thread():
20        while not coord.should_stop():
21            try:
22                sess.run(enqueue_op)
23            except tf.errors.CancelledError:
24                break
25
26    # Start enqueue thread
27    thread = threading.Thread(target=enqueue_thread)
28    thread.start()
29
30    # Dequeue and process in main thread
31    for _ in range(10):
32        val, step = sess.run([dequeue_op, increment_step])
33        print(f"Step {step}: {val:.4f}")
34
35    coord.request_stop()
36    sess.run(close_op)
37    thread.join()

The Coordinator pattern managed background threads that fed data into queues while the main thread consumed data for training.

Modern Approach: tf.data.Dataset (TF 2.x)

TensorFlow 2.x replaces queues with tf.data.Dataset, which is simpler, faster, and integrates with eager execution:

python

1import tensorflow as tf
2
3# Create a dataset from a list
4dataset = tf.data.Dataset.from_tensor_slices([1.0, 2.0, 3.0, 4.0, 5.0])
5
6# Apply transformations (replaces queue shuffling and batching)
7dataset = dataset.shuffle(buffer_size=5)
8dataset = dataset.batch(2)
9dataset = dataset.prefetch(tf.data.AUTOTUNE)  # Async prefetching
10
11# Iterate with a simple loop (no session, no coordinator)
12step = tf.Variable(0, trainable=False)
13for batch in dataset:
14    step.assign_add(1)
15    print(f"Step {step.numpy()}: {batch.numpy()}")

Variable Increment in TF 2.x (Eager Mode)

python

1import tensorflow as tf
2
3# Variables work like normal Python objects in eager mode
4counter = tf.Variable(0, dtype=tf.int32)
5
6# Increment directly
7counter.assign_add(1)
8print(counter.numpy())  # 1
9
10# In a training loop
11global_step = tf.Variable(0, trainable=False)
12for epoch in range(3):
13    for batch_idx in range(5):
14        global_step.assign_add(1)
15        # Training logic here
16    print(f"Epoch {epoch}, global step: {global_step.numpy()}")
17
18# Using tf.summary for logging
19summary_writer = tf.summary.create_file_writer("/tmp/logs")
20step = tf.Variable(0, dtype=tf.int64)
21with summary_writer.as_default():
22    for i in range(10):
23        step.assign_add(1)
24        tf.summary.scalar("loss", 1.0 / (i + 1), step=step)

Complete Training Loop Example (Modern)

python

1import tensorflow as tf
2
3# Dataset pipeline (replaces all queue operations)
4(x_train, y_train), _ = tf.keras.datasets.mnist.load_data()
5x_train = x_train.reshape(-1, 784).astype("float32") / 255.0
6
7dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train))
8dataset = dataset.shuffle(10000).batch(32).prefetch(tf.data.AUTOTUNE)
9
10# Simple model
11model = tf.keras.Sequential([
12    tf.keras.layers.Dense(128, activation="relu"),
13    tf.keras.layers.Dense(10)
14])
15
16optimizer = tf.keras.optimizers.Adam()
17loss_fn = tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True)
18global_step = tf.Variable(0, trainable=False)
19
20for epoch in range(3):
21    for x_batch, y_batch in dataset:
22        with tf.GradientTape() as tape:
23            logits = model(x_batch, training=True)
24            loss = loss_fn(y_batch, logits)
25        grads = tape.gradient(loss, model.trainable_variables)
26        optimizer.apply_gradients(zip(grads, model.trainable_variables))
27        global_step.assign_add(1)
28
29    print(f"Epoch {epoch + 1}, step {global_step.numpy()}, loss: {loss.numpy():.4f}")

Common Pitfalls

Using TF 1.x queues in TF 2.x code: Queues (FIFOQueue, RandomShuffleQueue) are legacy APIs. In TensorFlow 2.x, use tf.data.Dataset for all data pipeline needs — it handles shuffling, batching, prefetching, and parallel loading.
Forgetting to initialize variables in TF 1.x: In graph mode, tf.Variable must be initialized with sess.run(tf.global_variables_initializer()) before use. Without initialization, assign_add raises FailedPreconditionError. TF 2.x handles initialization automatically.
Deadlocking with queue operations: If you enqueue fewer elements than you dequeue, the dequeue operation blocks indefinitely waiting for data. Always close the queue when done and use a Coordinator to manage threads and detect this condition.
Using assign_add without capturing the result in TF 1.x: In graph mode, counter.assign_add(1) returns an operation that must be executed with sess.run(). Simply calling it does not increment the variable. In TF 2.x eager mode, assign_add executes immediately.
Not using prefetch in tf.data pipelines: Without prefetch(tf.data.AUTOTUNE), the GPU sits idle while the CPU prepares the next batch. Prefetching overlaps data preparation with model execution, significantly improving throughput.

Summary

TF 1.x used FIFOQueue/RandomShuffleQueue with enqueue/dequeue operations for async data feeding — these are now legacy
tf.Variable.assign_add() increments a variable — works in both graph and eager mode
TF 2.x replaces queues with tf.data.Dataset — use shuffle(), batch(), and prefetch() for data pipelines
In eager mode (TF 2.x default), variable operations execute immediately without Session.run()
Always use prefetch(tf.data.AUTOTUNE) in tf.data pipelines to overlap data loading with training

Tackling System Design Interview Problems

A short course that equips you with the skills to approach system design interviews methodically.

Start the free course

Track what you have practised

A free account saves your progress, solutions and study plan across every problem on Codemia.

ML System Design practice on Codemia

Design recommenders, ranking systems and training pipelines the way ML interviews actually ask for them, with worked solutions.

Practice ML system design

Enqueue and increment variable in Tensor Flow

ML System Design practice on Codemia

Introduction

TF 1.x Queue Operations

Incrementing Variables in TF 1.x

Queue with Coordinator and Threads (TF 1.x Pattern)

Modern Approach: tf.data.Dataset (TF 2.x)

Variable Increment in TF 2.x (Eager Mode)

Complete Training Loop Example (Modern)

Common Pitfalls

Summary

Related reading

Tackling System Design Interview Problems

Track what you have practised

ML System Design practice on Codemia