DEV Community: Andrey Matveyev

TicTacToe. Go Duel. AI vs Fate.

Andrey Matveyev — Tue, 01 Jul 2025 10:22:57 +0000

Neural Network vs Random Number Generator

Part #1. Developing a Neural Network in Golang.

"Knowledge itself is power" (с) -Francis Bacon

Creating a network is easy. Training it correctly is not an easy task. The result often does not match expectations. In reality, there is no magic here. The network does exactly what it is told to do. If the result is not what was intended, then the error is either in the training or in the interpretation of the results obtained. The creator's thoughts cannot yet be guessed by the network.

In our previous article, we delved into the fundamentals of neural networks, building a simple model in Golang and successfully solving the classic XOR problem. Now it's time to move on to a more exciting and complex area — Reinforcement Learning — and apply this knowledge to create an intelligent agent capable of playing Tic-Tac-Toe.

Unlike the XOR problem, where the network immediately received the "correct answer" and could adjust its weights, in games like Tic-Tac-Toe, a key difficulty arises: delayed reward. The agent makes moves, but the outcome of its actions (win, loss, or draw) is only known at the end of the game. This means we cannot immediately point out an "error" or "success" for each individual move to the network. The agent needs to learn to associate intermediate actions with future outcomes.

It is precisely to solve such problems that the Deep Q-Learning (DQN) algorithm was developed, which we will discuss in detail in this article. We will describe the game logic, the DQN agent's architecture, and analyze its training process as both the first and second player. The article is written in an accessible, popular style and will not delve deeply into the mathematical foundations, as there are many excellent resources on this topic available online (e.g., mathematics of reinforcement learning (RL) or video about DeepLearning).

Tic-Tac-Toe Game Logic

Tic-Tac-Toe is a simple deterministic game for two players on a 3x3 board. Players take turns placing their symbols (X and O) into empty cells. The goal of the game is to be the first to get three of your symbols in a row horizontally, vertically, or diagonally. If all cells are filled and no winner is determined, the game ends in a draw.

Key aspects of the game:

Game State: Determined by the arrangement of X and O symbols on the board.
Actions: Choosing an empty cell to place your symbol.
Game Outcome: A win for one of the players or a draw.
First Move Advantage: In Tic-Tac-Toe, the first player has a strategic advantage. With optimal play from both players, the game always ends in a draw or a win for the first player. According to my estimates, and confirmed by experiment (when the agent initially plays like a random opponent), the probability of winning for the player who makes the first move to the center is about 60% (600 out of 1000 games), a loss is about 30%, and a draw is 10%.

Board Representation and State Vector

The game board is represented by a Board struct, and its state is converted into a numerical vector for the neural network using the GetStateVector method.

// tictactoe.go

// Board represents the Tic-Tac-Toe game board.
type Board struct {
    Cells         [9]int // 0: empty, 1: X, -1: O
    CurrentPlayer int    // 1 for X, -1 for O
}

// NewBoard creates a new empty board.
func NewBoard() *Board {
    return &Board{
        Cells:         [9]int{Empty, Empty, Empty, Empty, Empty, Empty, Empty, Empty, Empty},
        CurrentPlayer: PlayerX, // X always starts
    }
}

// GetStateVector converts the board state into a vector for the neural network.
// Represents the 3x3 board as a flat 9-element vector.
// 1.0 for agent's cell, -1.0 for opponent's cell, 0.0 for empty.
func (item *Board) GetStateVector(agentPlayer int) []float64 {
    state := make([]float64, 9)
    for i, cell := range item.Cells {
        switch cell {
        case agentPlayer:
            state[i] = 1.0
        case -agentPlayer: // Opponent
            state[i] = -1.0
        default: // Empty
            state[i] = 0.0
        }
    }
    return state
}

Deep Q-Learning Agent (DQN)

Our agent is based on the Double DQN architecture, which combines Q-learning with deep neural networks. This is evident in how the action for the next state is selected using the main Q-network, and then its Q-value is evaluated using the target network. This helps to reduce the overestimation of Q-values characteristic of classic DQN.

State Representation

The board state is converted into a numerical vector that is fed into the neural network. For each of the 9 board cells:

1.0, if the cell is occupied by the agent's symbol.
-1.0, if the cell is occupied by the opponent's symbol.
0.0, if the cell is empty.

Neural Network Architecture

The agent uses a fully connected neural network.

Input Layer: 9 neurons (corresponding to the 9 board cells).
Hidden Layer: One hidden layer with 27 (or 45/72) neurons with a Tanh activation function. The minimum number of neurons in the hidden layer that yielded satisfactory results was 9.
Output Layer: 9 neurons (corresponding to 9 possible actions/cells), also with a Tanh activation function.

DQN Training Mechanism

The agent learns by interacting with the environment (the Tic-Tac-Toe game) and receiving rewards.

Experience Replay Buffer

The ReplayBuffer stores the agent's experiences, allowing for efficient training by sampling past interactions.

// agent.go

// Experience represents a single game experience.
type Experience struct {
    State     []float64
    Action    int
    Reward    float64
    NextState []float64
    Done      bool
}

// ReplayBuffer stores game experiences.
type ReplayBuffer struct {
    Experiences []Experience
    Capacity    int

    Index int
    Size  int
}

// NewReplayBuffer creates a new experience buffer.
func NewReplayBuffer(capacity int) *ReplayBuffer {
    return &ReplayBuffer{
        Experiences: make([]Experience, capacity),
        Capacity:    capacity,
    }
}

// Add adds a new experience to the buffer.
func (item *ReplayBuffer) Add(exp Experience) {
    item.Experiences[item.Index] = exp
    item.Index = (item.Index + 1) % item.Capacity
    if item.Size < item.Capacity {
        item.Size++
    }
}

// Sample selects a random batch of experiences from the buffer.
func (item *ReplayBuffer) Sample(batchSize int) []Experience {
    if item.Size < batchSize {
        return nil // Not enough experience to sample a batch
    }

    samples := make([]Experience, batchSize)
    for i := range batchSize {
        idx := rand.Intn(item.Size)
        samples[i] = item.Experiences[idx]
    }
    return samples
}

DQNAgent Structure and Action Selection

The DQNAgent struct holds the Q-network, target network, replay buffer, and training parameters. The ChooseAction method implements the epsilon-greedy strategy.

// DQNAgent represents a Deep Q-Learning agent.
type DQNAgent struct {
    QNetwork      *NeuralNetwork
    TargetNetwork *NeuralNetwork
    ReplayBuffer  *ReplayBuffer
    Gamma         float64 // Discount factor
    MaxEpsilon    float64 // For epsilon-greedy strategy
    MinEpsilon    float64 // Minimum epsilon value
    EpsilonDecay  float64 // Epsilon decay rate per episode
    LearningRate  float64
    UpdateTarget  int // Target network update interval (steps)
    PlayerSymbol  int // Symbol this agent plays (PlayerX or PlayerO)
}

// NewDQNAgent creates a new DQN agent.
func NewDQNAgent(inputSize, outputSize, bufferCapacity int, playerSymbol int) *DQNAgent {
    qNet := NewNeuralNetwork(inputSize, []int{hiddenLayerSize}, outputSize, "tanh") // Example architecture
    targetNet := qNet.Clone()                                                       // Clone for the target network

    return &DQNAgent{
        QNetwork:      qNet,
        TargetNetwork: targetNet,
        ReplayBuffer:  NewReplayBuffer(bufferCapacity),
        Gamma:         gamma,
        MaxEpsilon:    maxEpsilon,
        MinEpsilon:    minEpsilon,
        EpsilonDecay:  epsilonDecay,
        LearningRate:  learningRate,
        UpdateTarget:  updateTarget,
        PlayerSymbol:  playerSymbol,
    }
}

// ChooseAction selects an action using an epsilon-greedy strategy.
// board: current board state.
func (item *DQNAgent) ChooseAction(board *Board) int {
    emptyCells := board.GetEmptyCells()

    // Epsilon-greedy strategy: random move or best move according to Q-network
    if rand.Float64() < item.MaxEpsilon {
        return emptyCells[rand.Intn(len(emptyCells))] // Random move (Research process)
    }

    // Choose the best move according to the Q-network
    stateVec := board.GetStateVector(item.PlayerSymbol)
    qValues := item.QNetwork.Predict(stateVec)

    bestAction := -1
    maxQ := -math.MaxFloat64 // Initialize with a very small number
    for _, action := range emptyCells { // Iterate ONLY through empty cells
        if qValues[action] > maxQ {
            maxQ = qValues[action]
            bestAction = action // Found a new maximum
        }
    }
    return bestAction
}

Training the Agent

The Train method implements the core Double DQN update rule, using the replay buffer and target network.

// Train performs one training step for the agent.
// batchSize: batch size for training.
// step: current step (for target network update).
func (item *DQNAgent) Train(batchSize, step int) {
    batch := item.ReplayBuffer.Sample(batchSize)
    if batch == nil {
        return // Not enough experience
    }

    for _, exp := range batch {
        // Predicted Q-values for the current state from the Q-network
        currentQValues := item.QNetwork.Predict(exp.State)
        targetQValues := make([]float64, len(currentQValues))
        copy(targetQValues, currentQValues) // Copy to modify only one value

        // Calculate the target Q-value
        var targetQ float64
        if exp.Done {
            targetQ = exp.Reward // If the game is over, the target value is the immediate reward
        } else {
            // 1. Get Q-values for the next state from the Q-network (to choose the best action)
            qValuesNextStateFromQNetwork := item.QNetwork.Predict(exp.NextState)
            // Find the action that would be chosen by the Q-network in the next state.
            bestActionFromQNetwork := -1
            maxQValFromQNetwork := -math.MaxFloat64
            // Find the index of the best action from the Q-network's predictions.
            for i, qVal := range qValuesNextStateFromQNetwork {
                if qVal > maxQValFromQNetwork {
                    maxQValFromQNetwork = qVal
                    bestActionFromQNetwork = i
                }
            }
            // 2. Evaluate the Q-value of the chosen action using the Target Network
            qValueFromTargetNetwork := item.TargetNetwork.Predict(exp.NextState)[bestActionFromQNetwork]
            targetQ = exp.Reward + item.Gamma*qValueFromTargetNetwork // Bellman Equation (DDQN) !!!
        }
        // Update the target Q-value for the action taken in this experience
        targetQValues[exp.Action] = targetQ
        // Train the Q-network with the updated target Q-values
        item.QNetwork.Train(exp.State, targetQValues, item.LearningRate)
    }
    // Decay epsilon (applied per training step, not per episode)
    if item.MaxEpsilon > item.MinEpsilon {
        item.MaxEpsilon *= item.EpsilonDecay
    }
    // Update the target network
    if step%item.UpdateTarget == 0 {
        item.TargetNetwork = item.QNetwork.Clone()
        fmt.Printf("--- Target network updated at step %d (Epsilon: %.4f) ---\n", step, item.MaxEpsilon)
    }
}

Note the Bellman equation:

  ...
  targetQ = exp.Reward + item.Gamma*qValueFromTargetNetwork
  ...

Using this mechanism, the "reward" gradually "propagates" from the end of the game to its beginning.

Rewards

The GetReward function defines the reward structure for the agent:

// tictactoe.go

// GetReward returns the reward for the agent based on the game outcome.
// This function is called AFTER a move has been made and the game state potentially changed.
func (item *Board) GetReward(agentPlayer int) float64 {
    isOver, winner := item.GetGameOutcome()
    if isOver {
        switch winner {
        case agentPlayer:
            return winsReward // Agent wins
        case Empty:
            return drawReward // Draw
        default: // winner == -agentPlayer (opponent)
            return losesReward // Agent loses (opponent wins)
        }
    }
    return 0.0 // No negative reward for moves in Tic-Tac-Toe
}

Training Process and Results

So, we are all set for testing.
Let's briefly summarize what we have:

A neural network with a 9:27:9 architecture that knows nothing.
A game board and implementation of game logic (start, rule adherence, and end detector (win/loss/draw)).
An opponent who can make moves into free cells randomly. And that's all.
An agent that, from the start, plays like its opponent but has the ability to learn. It knows when the game ends. And it knows whether it finished the game well or poorly.

What can we observe and by what criteria can we determine the learning progress?

Firstly, it's the agent's win percentage (expected to increase).
Secondly, we can observe the decrease in Epsilon to understand what is happening – whether the agent is exploring (making random moves) or utilizing its accumulated experience.
Thirdly, we can look at the weight vector on the output layer to understand how the agent decides to make its first move on an empty board (it is expected that the center will have the largest weight, then the corners, and then the sides as the least promising).
And finally, we can track the maximum number of wins achieved throughout the entire experiment.

Let's see what came of this and whether our agent will show growth in its competence.

Training the Agent as the First Player

In this scenario, the agent (Player X) always makes the first move in the game. To accelerate convergence and ensure the learning of an optimal starting strategy, we can experimented with forcing the first move to the center of the board (default without this).

Training Parameters:

These are the settings that can be changed when conducting an experiment.
'Knobs' that can be 'turned' for fine-tuning.
The network implemented here usually forgives even gross errors.
The most you risk is falling into a local minimum instead of a global one.
Feel free to try it yourself.

// main.go
const (
    // Who makes the first move (first step)?
    agentsFirstStep bool = true // true = agent (PlayerX), false = opponent (PlayerO)
    // Training parameters
    episodes       int = 500000 // Number of game episodes for training
    batchSize      int = 8      // Batch size for DQN training
    bufferCapacity int = 50000  // Experience buffer capacity
    trainStartSize int = 1000   // Start training after accumulating enough experience
    // Learning parameters for DQNAgent
    gamma        float64 = 0.75     // Discount factor (how much the agent values future rewards)
    maxEpsilon   float64 = 1.0      // Start with exploration
    minEpsilon   float64 = 0.001    // Minimum epsilon value
    epsilonDecay float64 = 0.999996 // Epsilon decay rate per step (very slow)
    learningRate float64 = 0.0002   //
    updateTarget int     = 50000    // Update target network every 10000 steps (less frequently)
    // Reward parameters
    winsReward  float64 = 0.999
    drawReward  float64 = 0.001
    losesReward float64 = -1.000
    // Hidden layer size
    hiddenLayerSize int = 27
)

Results:

PS D:\go\go-sample-tictactoe> go run .
Starting DQN agent training (X) against a random opponent (O) for Tic-Tac-Toe...
Episode: 1000, Wins X: 571 (571), Losses X: 307, Draws: 122, Epsilon X: 0.9876, Q(start): 0.4501|0.5164|0.4117  0.5863[0.5449]0.4485  0.3473|0.4411|0.4166
Episode: 2000, Wins X: 590 (590), Losses X: 284, Draws: 126, Epsilon X: 0.9715, Q(start): 0.3683|0.4917|0.3963  0.2354[0.6179]0.3571  0.2806|0.3732|0.3737
Episode: 3000, Wins X: 585 (590), Losses X: 294, Draws: 121, Epsilon X: 0.9558, Q(start): 0.2797|0.4310|0.3559  0.1067[0.4802]0.2719  0.1742|0.2720|0.2669
Episode: 4000, Wins X: 588 (590), Losses X: 285, Draws: 127, Epsilon X: 0.9402, Q(start): 0.2361|0.4065|0.3263  0.1037[0.3945]0.2356  0.1445|0.2771|0.2186
...
Episode: 297000, Wins X: 952 (969), Losses X: 43, Draws: 5, Epsilon X: 0.0156, Q(start): 0.5193|0.3906|0.2095  0.5050[0.3286]0.4332  0.1040|0.3630|0.2807
Episode: 298000, Wins X: 957 (969), Losses X: 40, Draws: 3, Epsilon X: 0.0154, Q(start): 0.5189|0.3942|0.1822  0.4883[0.3528]0.4347  0.1214|0.3698|0.2528
Episode: 299000, Wins X: 977 (977), Losses X: 20, Draws: 3, Epsilon X: 0.0152, Q(start): 0.5201|0.4159|0.1651  0.4708[0.3775]0.4352  0.1291|0.3870|0.2078
--- Target network updated at step 1050000 (Epsilon: 0.0151) ---
Episode: 300000, Wins X: 968 (977), Losses X: 23, Draws: 9, Epsilon X: 0.0150, Q(start): 0.4733|0.4222|0.1718  0.4519[0.4072]0.4743  0.1526|0.4102|0.1889
...
Episode: 497000, Wins X: 952 (990), Losses X: 43, Draws: 5, Epsilon X: 0.0011, Q(start): 0.3910|-0.3152|-0.2335  -0.2994[0.4932]0.0485  0.0135|-0.4090|-0.2174
--- Target network updated at step 1700000 (Epsilon: 0.0011) ---
Episode: 498000, Wins X: 942 (990), Losses X: 55, Draws: 3, Epsilon X: 0.0011, Q(start): 0.3798|-0.3127|-0.2245  -0.3118[0.4557]0.0439  0.0072|-0.4120|-0.2115
Episode: 499000, Wins X: 936 (990), Losses X: 56, Draws: 8, Epsilon X: 0.0011, Q(start): 0.3651|-0.3107|-0.2292  -0.3250[0.3711]0.0254  -0.0033|-0.4216|-0.1881
Episode: 500000, Wins X: 954 (990), Losses X: 41, Draws: 5, Epsilon X: 0.0011, Q(start): 0.3561|-0.3119|-0.2014  -0.3267[0.3711]0.0196  -0.0191|-0.4155|-0.1827

Training complete.
Testing the agent (X against random O)...

Test Results (1000 games, Agent X vs random O):
Agent X Wins: 956
Agent X Losses (Random O Wins): 39
Draws: 5

When the agent was to make the first move to the center, it demonstrated outstanding results, achieving up to 992 wins out of 1000 (in some cases) test games against a random opponent, with a minimal number of losses and draws. This confirms that the agent successfully learned an optimal strategy for the first player.

"Win Growth (agent moves first)" graph:

Training the Agent as the Second Player

In this scenario, the opponent (Player O) always makes the first move randomly, and our agent (Player X) always responds second. This puts the agent in a less advantageous position, as the first move in Tic-Tac-Toe provides a strategic advantage. The goal of this experiment is to test how well the agent can adapt to the role of the second player and minimize the opponent's advantage.

Training Parameters:

The same hyperparameters as for the first scenario were used.
The only change is that the opponent always makes the first move.

// main.go
const (
    // Who makes the first move (first step)?
    agentsFirstStep bool = false // true = agent (PlayerX), false = opponent (PlayerO)
    ...
)

Results:

PS D:\go\go-sample-tictactoe> go run .
Starting DQN agent training (X) against a random opponent (O) for Tic-Tac-Toe...
Episode: 1000, Wins X: 296 (296), Losses X: 587, Draws: 117, Epsilon X: 0.9902, Q(start): 0.2536|0.3091|0.2323  0.3227[0.3963]0.3577  0.4702|0.4281|0.2465
Episode: 2000, Wins X: 298 (298), Losses X: 590, Draws: 112, Epsilon X: 0.9766, Q(start): 0.1909|0.3386|0.2124  0.3879[0.3856]0.3629  0.5409|0.4653|0.2537
Episode: 3000, Wins X: 295 (298), Losses X: 598, Draws: 107, Epsilon X: 0.9633, Q(start): 0.0990|0.3089|0.1477  0.3343[0.3218]0.2929  0.5055|0.4229|0.2093
Episode: 4000, Wins X: 261 (298), Losses X: 601, Draws: 138, Epsilon X: 0.9501, Q(start): 0.0718|0.2712|0.0945  0.3015[0.2998]0.2637  0.4218|0.3067|0.1649
...
Episode: 69000, Wins X: 610 (610), Losses X: 342, Draws: 48, Epsilon X: 0.3986, Q(start): 0.5987|0.5451|0.5798  0.5912[0.6872]0.5793  0.6331|0.5710|0.5508
Episode: 70000, Wins X: 610 (610), Losses X: 359, Draws: 31, Epsilon X: 0.3935, Q(start): 0.5962|0.5428|0.5695  0.5917[0.6848]0.5758  0.6282|0.5837|0.5531
Episode: 71000, Wins X: 606 (610), Losses X: 365, Draws: 29, Epsilon X: 0.3885, Q(start): 0.5914|0.5330|0.5650  0.5899[0.6844]0.5742  0.6268|0.5863|0.5423
Episode: 72000, Wins X: 570 (610), Losses X: 407, Draws: 23, Epsilon X: 0.3835, Q(start): 0.5867|0.5349|0.5650  0.5872[0.6871]0.5795  0.6202|0.5833|0.5385
Episode: 73000, Wins X: 564 (610), Losses X: 405, Draws: 31, Epsilon X: 0.3786, Q(start): 0.5912|0.5303|0.5606  0.5833[0.6815]0.5811  0.6198|0.5832|0.5418
Episode: 74000, Wins X: 612 (612), Losses X: 353, Draws: 35, Epsilon X: 0.3737, Q(start): 0.5958|0.5287|0.5575  0.5840[0.6816]0.5730  0.6146|0.5765|0.5359
--- Target network updated at step 250000 (Epsilon: 0.3694) ---
Episode: 75000, Wins X: 588 (612), Losses X: 373, Draws: 39, Epsilon X: 0.3689, Q(start): 0.6005|0.5305|0.5658  0.5903[0.6910]0.5730  0.6132|0.5845|0.5456
Episode: 76000, Wins X: 650 (650), Losses X: 311, Draws: 39, Epsilon X: 0.3642, Q(start): 0.6314|0.5703|0.5932  0.6218[0.7187]0.6036  0.6409|0.6085|0.5756
...
Episode: 497000, Wins X: 792 (822), Losses X: 185, Draws: 23, Epsilon X: 0.0020, Q(start): 0.5345|0.3504|0.2066  0.2787[0.5258]0.4991  0.1034|0.5461|0.5410
Episode: 498000, Wins X: 804 (822), Losses X: 168, Draws: 28, Epsilon X: 0.0020, Q(start): 0.5329|0.3472|0.2169  0.2769[0.5331]0.4969  0.1012|0.5451|0.5428
Episode: 499000, Wins X: 782 (822), Losses X: 180, Draws: 38, Epsilon X: 0.0019, Q(start): 0.5315|0.3456|0.2200  0.2724[0.5288]0.4962  0.1074|0.5430|0.5417
Episode: 500000, Wins X: 780 (822), Losses X: 188, Draws: 32, Epsilon X: 0.0019, Q(start): 0.5310|0.3443|0.2219  0.2718[0.5285]0.4971  0.1044|0.5442|0.5446

Training complete.
Testing the agent (X against random O)...

Test Results (1000 games, Agent X vs random O):
Agent X Wins: 783
Agent X Losses (Random O Wins): 191
Draws: 26

In the initial stages of training, the agent, as expected, showed a lower win percentage and a higher number of losses/draws due to the opponent's first-move advantage. However, as training progressed, the agent significantly improved its performance.

Example game after training:

Example game after training (X vs random O):
-------------
|   |   |   |
-------------
|   |   |   |
-------------
| O |   |   |
-------------
X's Turn:
-------------
|   |   |   |
-------------
|   |   | X |
-------------
| O |   |   |
-------------
O's Turn:
-------------
|   |   |   |
-------------
| O |   | X |
-------------
| O |   |   |
-------------
X's Turn:
-------------
|   |   |   |
-------------
| O |   | X |
-------------
| O |   | X |
-------------
O's Turn:
-------------
|   |   |   |
-------------
| O |   | X |
-------------
| O | O | X |
-------------
X's Turn:
-------------
|   |   | X |
-------------
| O |   | X |
-------------
| O | O | X |
-------------
Game Over! Player X won!

"Win Growth (agent moves second)" graph:

These results show that the agent successfully learned to optimally respond to various first moves by the opponent, significantly increasing its win rate despite the strategic disadvantage of moving second. The "first move selection" problem for the agent disappeared, as it focused on reactive tactics.

Conclusion

The project on training a DQN agent for Tic-Tac-Toe successfully demonstrated the effectiveness of deep reinforcement learning algorithms even for simple deterministic games. We saw how the agent can adapt to different roles (first/second player) and achieve near-optimal performance against a random opponent.

The full source code is available at the link:
https://github.com/andrey-matveyev/go-sample-tictactoe

Postscript

The most guaranteed way to make the agent learn "human" optimality (center, corners) is to train it against a stronger, strategic opponent (e.g., Minimax AI) or in self-play mode. These opponents will punish any suboptimal move, forcing the agent towards true optimality.

Write in the comments if you are interested, and I will arrange a battle (a real fight) between two agents. For now, my immediate plans include a final "move" to Linux and writing a small backend (e.g., a REST API) for a simple client to try playing with what has been developed.

Developing a Neural Network in Golang.

Andrey Matveyev — Sun, 22 Jun 2025 21:20:03 +0000

...from Scratch: From Basics to XOR Solution

"To know is to build". (c) -Anonymous

In the world of machine learning, neural networks are powerful tools for solving a wide range of tasks, from image recognition to natural language processing and games. While there are many high-level libraries that simplify the creation and training of neural networks (e.g., TensorFlow, PyTorch, Keras), understanding how these networks work "under the hood" is invaluable.

This article is dedicated to creating a simple yet functional feedforward neural network in Go, using only the standard library. This approach will allow us to delve deeply into the mechanisms of forward and backward propagation, weight initialization, and training.

Why Golang and "From Scratch"?

Go is a compiled language known for its performance, ease of concurrency, and strong typing. These qualities make it an excellent choice for low-level implementations where control over performance is crucial. Developing a neural network from scratch in Go allows you to:

Deeply understand algorithms: You implement each mathematical step, which strengthens your understanding of neural network principles.
Learn Go: This is excellent practice for learning matrix operations, data structures, and working with pointers in Go.
Control and optimization: Full control over every aspect of the implementation allows for deep optimization when necessary.

Alternatives: Existing ML Libraries in Go

Before diving into the "from scratch" implementation, it's worth mentioning that for more complex and "production-ready" projects, there are specialized libraries that provide pre-built and optimized components:

Gorgonia: A powerful library for building computational graphs, very similar to TensorFlow or PyTorch. It allows you to define and train complex deep learning models.
GoMind: A simpler library for creating and training basic neural networks.
GoLearn: While primarily a general machine learning library (including clustering, classification algorithms, etc.), it also offers some capabilities for working with neural networks.

These libraries significantly speed up development, but for understanding the fundamentals, it's best to start "from scratch."

Architecture of Our Neural Network

We will build a Feedforward Neural Network (FFNN). This means that neurons in one layer are fully connected to all neurons in the next layer, and information flows in only one direction – from input to output.

Layer Structure: `NeuralNetworkLayer`

In our implementation, we've combined the logic of linear transformation (weighted sums and biases) and activation functions into a single NeuralNetworkLayer structure. This aligns with the conceptual understanding of a "layer" in most modern frameworks.

// NeuralNetworkLayer represents one fully connected layer with an activation function.
type NeuralNetworkLayer struct {
    InputSize  int
    OutputSize int
    Weights    [][]float64 // Weights[output_neuron_idx][input_neuron_idx]
    Biases     []float64

    ActivationFunc func(float64) float64
    DerivativeFunc func(float64) float64

    // Temporal values for backpropagation
    InputVector  []float64 // Input values to layer (from previous layer)
    WeightedSums []float64 // Values after linear transformation (before activation)
    OutputVector []float64 // Output values after activation

    // Gradients for updating weights and biases
    WeightGradients [][]float64
    BiasGradients   []float64
    InputGradient   []float64 // Gradient passed to the previous layer
}

Weights and Biases: These are the trainable parameters of the layer. Weights is a matrix that multiplies the input vector, and Biases is a vector added to the result.
ActivationFunc and DerivativeFunc: Pointers to activation functions (e.g., ReLU, Sigmoid) and their derivatives, which are applied to the output of the linear transformation.

Weight Initialization: He Initialization

One of the most critical aspects is proper weight initialization. It helps prevent vanishing or exploding gradient problems that can slow down or halt training. We use He initialization for layers with ReLU activation, while for other types of activations (e.g., Sigmoid), a standard initialization with a smaller scale is used:

// In the NewNeuralNetworkLayer function
for i := range weights {
    weights[i] = make([]float64, inputSize)
    for j := range weights[i] {
        // Initialize weights based on the activation function
        if activationName == "relu" {
            weights[i][j] = rand.NormFloat64() * math.Sqrt(2.0/float64(inputSize)) // He initialization
        } else {
            weights[i][j] = rand.NormFloat64() * math.Sqrt(1.0/float64(inputSize)) // More general approach
        }
    }
    biases[i] = 0.0
}

Here, rand.NormFloat64() generates a random number from a standard normal distribution, and the multiplier math.Sqrt(2.0/float64(inputSize)) scales it to maintain a stable variance of activations when using ReLU. For other activations, a multiplier of math.Sqrt(1.0/float64(inputSize)) is used.

Activation Functions: ReLU and Sigmoid

Our implementation supports both ReLU and Sigmoid.

ReLU (f(x)=max(0,x)) is widely used in hidden layers to introduce non-linearity and combat the vanishing gradient problem.
Sigmoid (f(x)=1/(1+e^{−x})) squashes values into the range of 0 to 1, often used in output layers for classification tasks where probabilities are needed.

 // Example activation functions from tools.go
func Sigmoid(x float64) float64 {
    return 1.0 / (1.0 + math.Exp(-x))
}

func SigmoidDerivative(x float64) float64 {
    s := Sigmoid(x)
    return s * (1 - s)
}

func ReLU(x float64) float64 {
    return math.Max(0, x)
}

func ReLUDerivative(x float64) float64 {
    if x > 0 {
        return 1.0
    }
    return 0.0
}

For the output layer, a linear activation (denoted as "none" in our code) is typically used to obtain raw numerical values, such as Q-values in DQN or regression outputs.

Implementation Details: Forward and Backward Pass

Let's examine the key Forward and Backward methods from NeuralNetworkLayer, which form the core of neural network computations.

`Forward` Method (Forward Pass)

The Forward method is responsible for computing the output of a layer based on given input data.

// Forward performs a forward pass through the layer (linear part + activation).
func (item *NeuralNetworkLayer) Forward(input []float64) []float64 {
    item.InputVector = input // Save input for use in Backward
    // 1. Linear transformation: multiply by weights and add biases
    item.WeightedSums = MultiplyMatrixVector(item.Weights, input)
    item.WeightedSums = AddVectors(item.WeightedSums, item.Biases)

    // 2. Activation: apply the activation function to the result of the linear transformation
    item.OutputVector = make([]float64, len(item.WeightedSums))
    for i := range item.WeightedSums {
        item.OutputVector[i] = item.ActivationFunc(item.WeightedSums[i])
    }
    return item.OutputVector
}

At each step:

The input vector (input) is multiplied by the layer's weight matrix (item.Weights).
The bias vector (item.Biases) is added to the result. This gives WeightedSums (or z-values).
The activation function (item.ActivationFunc) is applied to each element of WeightedSums, yielding the final Output of the layer.

`Backward` Method (Backward Pass)

The Backward method computes the gradients of the loss with respect to the layer's weights, biases, and inputs. This is the foundation of the backpropagation algorithm.

// Backward performs a reverse pass through the layer.
func (item *NeuralNetworkLayer) Backward(outputGradient []float64) []float64 {
    // 1. Gradient via activation function:
    // Apply the derivative of the activation function to the saved WeightedSums
    activationGradient := make([]float64, len(item.WeightedSums))
    for i := range item.WeightedSums {
        activationGradient[i] = item.DerivativeFunc(item.WeightedSums[i])
    }
    // Combine the gradient from the next layer with the activation gradient (element-wise multiplication)
    gradientAfterActivation := MultiplyVectors(outputGradient, activationGradient)

    // 2. Gradient for biases: equals the gradient after activation
    item.BiasGradients = gradientAfterActivation

    // 3. Gradient for weights: computed as the outer product
    // (gradient_after_activation X InputVector)
    item.WeightGradients = OuterProduct(gradientAfterActivation, item.InputVector)

    // 4. Gradient for input: computed as the product of the transposed weight matrix
    // and the gradient after activation. This is the gradient passed to the previous layer.
    transposedWeights := TransposeMatrix(item.Weights)
    item.InputGradient = MultiplyMatrixVector(transposedWeights, gradientAfterActivation)

    return item.InputGradient
}

The main steps of Backward for NeuralNetworkLayer:

The gradient passing through the activation function is calculated. For this, outputGradient (the gradient received from the next layer) is element-wise multiplied by the derivative of the activation function, computed at the WeightedSums.
The gradients for Biases are equal to this gradientAfterActivation.
The gradients for Weights are computed as the outer product of gradientAfterActivation and the layer's Input.
The gradient to be passed to the previous layer (InputGradient) is calculated as the product of the transposed weight matrix of the layer and gradientAfterActivation.

Overall Network Structure: `NeuralNetwork`

The neural network itself (NeuralNetwork) is a collection of these layers:

// network.go
type NeuralNetwork struct {
    Layers []*NeuralNetworkLayer
}
...
// Predict performs a forward pass to obtain network predictions.
func (item *NeuralNetwork) Predict(input []float64) []float64 {
    output := input
    for _, layer := range item.Layers {
        output = layer.Forward(output)
    }
    return output
}

// Train performs one step of training the network.
// input: input data
// targetOutput: target output data (Q-values for training)
// learningRate: learning rate
func (item *NeuralNetwork) Train(input []float64, targetOutput []float64, learningRate float64) {
    // Forward pass (saving intermediate values)
    predictedOutput := item.Predict(input)

    // Calculate the gradient of the output (MSE loss derivative)
    // dLoss/dOutput = 2 * (predicted - target)
    outputGradient := make([]float64, len(predictedOutput))
    for i := range predictedOutput {
        outputGradient[i] = 2 * (predictedOutput[i] - targetOutput[i])
    }

    // Backward pass
    currentGradient := outputGradient
    for i := len(item.Layers) - 1; i >= 0; i-- {
        currentGradient = item.Layers[i].Backward(currentGradient)
    }

    // Updating weights
    for _, layer := range item.Layers {
        layer.Update(learningRate)
    }
}

Predict: The method for the forward pass. It simply passes the input data sequentially through each layer to get the final output.
Train: The method for training. It performs a forward pass, calculates the loss function gradient (MSE), then performs backpropagation by iterating through the layers in reverse order and computing gradients for each layer. Finally, it updates the weights and biases of the layers.

Application: Solving the XOR Problem

The XOR (exclusive OR) problem is a classic test for neural networks because it is not linearly separable. A simple network without hidden layers (a linear model) cannot solve it. However, a network with one or more hidden layers can successfully tackle XOR.

XOR Data

The XOR problem takes two binary inputs (0 or 1) and outputs 1 if the inputs are different, and 0 if they are the same.

Inputs: [0,0], [0,1], [1,0], [1,1]
Outputs: [0], [1], [1], [0]

Architecture for XOR

To solve XOR, according to the provided source code (main.go), we use a network with 2 input neurons, 1 hidden layer with 2 neurons (and Sigmoid activation), and 1 output neuron (with linear activation):

NeuralNetwork := NewNeuralNetwork(2, []int{2}, 1, "sigmoid")

This minimal architecture with one hidden layer is necessary for solving the XOR problem. Using a sigmoid in the hidden layer is a traditional approach for this task.

Training Process

Training occurs iteratively:

We define the learningRate and the number of epochs.
In each epoch, we iterate through all XOR examples.
For each example (input, target), the NeuralNetwork.Train method is called, which adjusts the network's weights.
We track the total loss for each epoch to ensure it decreases, indicating that the network is learning.

// main.go
unc main() {
    // Define XOR data
    // Inputs: [0,0], [0,1], [1,0], [1,1]
    // Outputs: [0], [1], [1], [0]
    xorInputs := [][]float64{
        {0.0, 0.0},
        {0.0, 1.0},
        {1.0, 0.0},
        {1.0, 1.0},
    }
    xorOutputs := [][]float64{
        {0.0},
        {1.0},
        {1.0},
        {0.0},
    }
    // Neural Network architecture for XOR
    // 2 inputs, 1 hidden layer with 2 neurons, 1 output, sigmoid activation
    NeuralNetwork := neural.NewNeuralNetwork(2, []int{2}, 1, "sigmoid")
    // Test the non-trained network
    fmt.Println("Testing the non-trained network:")
    for i, input := range xorInputs {
        predictedOutput := NeuralNetwork.Predict(input)
        fmt.Printf("Input: %v, Expected: %v, Predicted: %.4f\n", input, xorOutputs[i], predictedOutput)
    }

    fmt.Println("Starting XOR training...")
    learningRate := 0.01
    epochs := 20000 // Number of training epochs
    for i := range epochs {
        totalLoss := 0.0
        for j := range xorInputs {
            input := xorInputs[j]
            target := xorOutputs[j]

            // Train the network on one XOR example
            NeuralNetwork.Train(input, target, learningRate)

            // Calculate current loss for monitoring (optional, but good practice)
            predicted := NeuralNetwork.Predict(input)
            loss := 0.0
            for k := range predicted {
                diff := predicted[k] - target[k]
                loss += diff * diff // MSE
            }
            totalLoss += loss
        }

        if (i+1)%1000 == 0 {
            fmt.Printf("Epoch %d, Average Loss: %.6f\n", i+1, totalLoss/float64(len(xorInputs)))
        }
    }
    fmt.Println("XOR training finished.")
    fmt.Println("Testing the trained network:")
    // Test the trained network
    for i, input := range xorInputs {
        predictedOutput := NeuralNetwork.Predict(input)
        fmt.Printf("Input: %v, Expected: %v, Predicted: %.4f\n", input, xorOutputs[i], predictedOutput)
    }
}

Console Output Example

During training, you will see the average loss gradually decrease. Note that when using sigmoid and the specified training parameters, the results might not be perfectly 0.0 or 1.0, but they will be close enough for logical classification:

Testing the non-trained network:
Input: [0 0], Expected: [0], Predicted: [-0.6342]
Input: [0 1], Expected: [1], Predicted: [-0.6208]
Input: [1 0], Expected: [1], Predicted: [-0.0719]
Input: [1 1], Expected: [0], Predicted: [-0.0617]
Starting XOR training...
Epoch 1000, Average Loss: 0.210077
Epoch 2000, Average Loss: 0.153441
Epoch 3000, Average Loss: 0.056236
Epoch 4000, Average Loss: 0.003917
Epoch 5000, Average Loss: 0.000094
Epoch 6000, Average Loss: 0.000002
Epoch 7000, Average Loss: 0.000000
...
Epoch 19000, Average Loss: 0.000000
Epoch 20000, Average Loss: 0.000000
XOR training finished.
Testing the trained network:
Input: [0 0], Expected: [0], Predicted: [0.0000]
Input: [0 1], Expected: [1], Predicted: [1.0000]
Input: [1 0], Expected: [1], Predicted: [1.0000]
Input: [1 1], Expected: [0], Predicted: [0.0000]

As the example shows, the predicted values are very close to the expected 0 or 1, indicating successful training.

If you are interested in the weights and biases before and after training, here they are:

// Before training
Biases 0 -  [0 0]
Weights 0 -  [[-1.5470222607067037 -0.01696838240061016] [0.24909017310045511 0.06488626663290908]]
Biases 1 -  [0]
Weights 1 -  [[-1.6581899752969296 0.389885131323495]]

// Biases and Weights after training:
Biases 0 -  [0.3944580734523731 2.1331071307836806]
Weights 0 -  [[-3.3592486686297476 -3.275219318130251] [-1.6500618206834206 -1.631091932264212]]
Biases 1 -  [-0.7323529446320001]
Weights 1 -  [[-3.3658966295356247 3.0679477637291073]]

Conclusion

We've explored how to build a basic feedforward neural network in Golang from scratch, including its architecture, activation functions, and weight initialization principles. By applying this network to solve the classic XOR problem, we've confirmed the functionality of our implementation.

In the next part of this article series, we will use this architecture as a foundation for creating a more complex system – a Deep Q-Network (DQN) based agent that can learn to play Tic-Tac-Toe independently, tackling the challenge of delayed rewards. Stay tuned for updates!

The full source code is available at the link:
https://github.com/andrey-matveyev/go-sample-neural

File Duplicate Detector. Go implementation.

Andrey Matveyev — Mon, 16 Jun 2025 10:16:36 +0000

SearchEngine, Processing Pipeline, Usage and Result.

"Truth is not born pure from the earth; it requires refinement from the superfluous to shine in its essence."
— Ancient Wisdom

In previous parts of our series on the File Duplicate Detector, we thoroughly examined individual components: Workers that perform specific tasks (file discovery, size determination, hash calculation, byte-by-byte comparison); Queues that ensure reliable data transfer between workers; and Checkers that optimize the duplicate detection process by preventing redundant work.

Now it's time to put this puzzle together and understand how these independent, yet interconnected parts form a powerful system for detecting duplicate files. The SearchEngine component and the data processing pipeline are central to this process.

1. `SearchEngine`: The Heart of Orchestration.

The SearchEngine is the brain of the entire system. Its main task is to launch and coordinate all stages of the file processing pipeline. It does not perform direct file operations but acts as a conductor, ensuring the correct execution of all operations:

Initialization: Upon launch, the SearchEngine initializes internal structures, such as a sync.WaitGroup to track the completion of all workers, and metrics for collecting statistics.
Pipeline Launch: The Run method starts the main processing pipeline in a separate goroutine, allowing it to operate asynchronously.
Lifecycle Management: The SearchEngine manages a cancellation context (context.Context), allowing for the graceful shutdown of all workers upon receiving a cancellation signal (e.g., Ctrl+C).
Result Collection: After the pipeline completes its work, the SearchEngine collects and provides the final results via the GetResult() method, as well as current progress via GetProgress().

Here's the SearchEngine code, with inline comments for clarity::

// SearchEngine defines the interface for the search engine.
type SearchEngine interface {
    // Run starts the duplicate detection process.
    Run(ctx context.Context, rootPath string, callback func())
    // GetProgress returns the current progress statistics as a JSON byte slice.
    GetProgress() []byte
    // GetResult returns the final duplicate detection result.
    GetResult() *Result
}

// GetEngine returns a new SearchEngine instance.
func GetEngine() SearchEngine {
    return &searchEngine{}
}

// searchEngine is the concrete implementation of SearchEngine.
type searchEngine struct {
    rootPath  string
    callback  func()         // Callback function, called upon completion.
    poolCount sync.WaitGroup // WaitGroup for all worker pools.
    result    *Result        // Final duplicate detection result.
    metrics   *metrics       // Performance and progress metrics.
}

func (item *searchEngine) Run(ctx context.Context, rootPath string, callback func()) {
    item.rootPath = rootPath
    item.callback = callback
    item.metrics = &metrics{}
    item.metrics.StartTime = time.Now()
    go item.runPipeline(ctx)
}

// GetProgress returns the current progress statistics as a JSON byte slice.
func (item *searchEngine) GetProgress() []byte {
    item.metrics.Duration = time.Since(item.metrics.StartTime) // Update duration.
    jsonData, err := json.Marshal(item.metrics)                // Marshal metrics to JSON.
    if err != nil {
        slog.Info("Marshalling error",
            slog.String("method", "json.Marshal"),
            slog.String("error", err.Error()))
    }
    return jsonData // Return JSON data.
}

func (item *searchEngine) GetResult() *Result {
    return item.result
}

2. Building the Processing Pipeline: pipeline()

The most interesting part of the SearchEngine is the pipeline() method, which is responsible for constructing the entire data processing pipeline. This demonstrates the principles of pipeline processing and concurrent programming in Go.

The pipeline consists of several sequential stages, each represented by a pool of workers and its own queue. Data (represented by *task) is passed from one stage to the next via Go channels:

fileGenerator (File Collector):
- This is the first stage. It launches fetcher workers that traverse the file system, recursively scanning directories and sending tasks (*task) for each found file and directory.
- It's important to note that fileGenerator also contains a dispatcher that coordinates recursive directory traversal and completion signals for the fetchers.
- The output of fileGenerator is a channel that provides tasks containing the path and size of each file.
runPool(&sizer{}, N, ...) (Size Determiner):
- The output channel of fileGenerator becomes the input for the sizer pool.
- sizers filter files by size (e.g., excluding zero-byte files) and use a checker for initial screening of files with unique sizes. If multiple files of the same size are found, they are passed to the next stage.
runPool(&hasher{}, N, ...) (Hash Calculator):
- The output channel of the sizers becomes the input for the hasher pool.
- hashers calculate the CRC32 hash for a portion of the file. At this stage, a second, more precise, duplicate filtering occurs: files with different hashes are guaranteed not to be duplicates.
- A checker is used here to determine if the hash is already known (a potential duplicate).
runPool(&matcher{}, N, ...) (Byte-by-Byte Comparator):
- The output channel of the hashers becomes the input for the matcher pool.
- matchers perform the most resource-intensive action: byte-by-byte comparison of files whose size and partial hash have matched. Only at this stage is it definitively confirmed that two files are identical.
- A checker is used to manage groups of potential duplicates to avoid redundant comparisons and track already verified files.
...runPipeline(ctx context.Context) (Result Generator):
- The output channel of the matchers (containing only confirmed duplicates) is fed into the resultQueue. From this queue, the code in runPipeline() executes as the final stage, responsible for collecting and grouping the paths of all confirmed duplicate files, ultimately preparing the data for the final Result structure.

// Launching the pipeline and processing the results
func (item *searchEngine) runPipeline(ctx context.Context) {
    predResult := newPredResult()

    for task := range item.pipeline(ctx) {
        pathList, detected := predResult.List[task.key]
        if detected {
            predResult.List[task.key] = append(pathList, task.info.path)
        } else {
            pathList := make([]string, 1)
            pathList[0] = task.info.path
            predResult.List[task.key] = pathList
        }
    }
    item.result = result(predResult) // Converting predResult to final result
    item.poolCount.Wait()
    item.callback()
}
// Building pipeline
func (item *searchEngine) pipeline(ctx context.Context) chan *task {
    rec := make(chan *task)
    out := fileGenerator(item.rootPath, 4, rec, fetchQueue(ctx, rec, item.metrics), item)
    out = runPool(&sizer{}, 1, sizeQueue(ctx, out, item.metrics), newCheckList(), item)
    out = runPool(&hasher{}, 6, hashQueue(ctx, out, item.metrics), newCheckList(), item)
    out = runPool(&matcher{}, 8, matchQueue(ctx, out, item.metrics), newCheckList(), item)
    out = resultQueue(ctx, out, item.metrics)
    return out
}

All these stages are connected by channels, and each runPool launches a fixed number (amt) of worker goroutines that process incoming tasks in parallel. metrics collects statistics for each stage, allowing progress to be monitored.

3. Metrics and Monitoring (`metrics.go`, `monitor.go`)

One of the advantages of this implementation is its built-in mechanism for collecting metrics and monitoring progress. This allows not only tracking the process status in real-time but also analyzing the performance of each stage.

The metrics and queueStat structures (file metrics.go) collect data on the number of processed files (Count) and their total size (Size) at the input (Inp) and output (Out) of each queue (pipeline stage).

// metrics.go
type metrics struct {
    StartTime time.Time     `json:"start"`
    Duration  time.Duration `json:"duration"`
    Fetch     queueStat     `json:"fetch"`
    Size      queueStat     `json:"size"`
    Hash      queueStat     `json:"hash"`
    Match     queueStat     `json:"match"`
    Pack      queueStat     `json:"pack"`
}

type queueStat struct {
    Inp statistic `json:"inpQueue"`
    Out statistic `json:"outQueue"`
}

The counter function (also in metrics.go) is a channel wrapper that increments the corresponding metrics as a task passes through the channel.

The monitor.go file contains the logic for displaying progress. In the progressMonitor function, metric data is periodically retrieved via engine.GetProgress() (which returns a JSON representation of metrics) and printed to the console.

// monitor.go
func progressMonitor(ctx context.Context, engine fdd.SearchEngine) {
    // ...
    for {
        err := json.Unmarshal(engine.GetProgress(), &stat)
        // ...
        fmt.Println(
            runtime.NumGoroutine(),
            "folder",
            metrics(stat.Fetch),
            "fetch",
            // ...
            stat.Duration,
        )
        time.Sleep(10 * time.Second)
    }
}

This allows real-time viewing of:

The number of active goroutines.
Processing progress at each stage (number of files and their total size).
Total running duration.
Filtering performance at the sizer, hasher, and matcher stages – how many files were "discarded" at each stage.

4. Example Usage

Using the File Duplicate Detector from main.go is quite straightforward:

func main() {
    runtime.GOMAXPROCS(runtime.NumCPU())

    // Creating config application
    config := newConfig()

    // Saving default logger
    oldLogger := slog.Default()

    // Creating newLogger and set as default
    logFile := newLogFile(config.LoggerFileName)
    newLogger(
        withLevel(config.LoggerLevel),
        withAddSource(config.LoggerAddSource),
        withLogFile(logFile),
        withSetDefault(true),
    )

    // Creating context with Cancel
    ctx := context.Background()
    ctx, cancel := context.WithCancel(ctx)

    // Preparing cancel-mechanism (over signal <Ctrl+C>)
    ch := make(chan os.Signal, 1)
    signal.Notify(ch, os.Interrupt)
    go func() {
        <-ch
        fmt.Println("Canselling... ")
        fmt.Println("The application needs to close all resources and save the current result.")
        fmt.Println("Please wait...")
        cancel()
    }()

    engine := fdd.GetEngine()
    // Preparing callback function (event about all tasks completed)
    var wg sync.WaitGroup
    callback := func() {
        wg.Done()
    }

    // Running main work and progress-monitor
    wg.Add(1)
    engine.Run(ctx, config.RootPath, callback)
    go progressMonitor(ctx, engine)
    wg.Wait()

    // Saving result to file
    saveResult(engine, config.ResultFileName)

    // Saving logs
    logFile.Close()

    // Returning default logger
    slog.SetDefault(oldLogger)

    // Printing total statistic
    printStatistic(engine, config)
}

As seen in the example, you:

Create a context.Context with cancellation capability.
Set up OS signal handling for graceful termination.
Obtain a SearchEngine instance.
Call Run() with the root path for scanning and a callback function that will be invoked upon completion.
Launch progressMonitor to display progress.
Wait for all operations to complete using a sync.WaitGroup.
Retrieve and save (GetResult()) the found duplicates.

5. Obtained Results (`Result`)

The final result of the SearchEngine's operation is available via the GetResult() method and is returned as a Result structure:

type Result struct {
    List []element
}

type element struct {
    header
    body
}

type header struct {
    Size  int64
    Hash  uint32
    Group int
}

type body struct {
    Paths []string
}

Example of application execution from the console (scanning an SSD disk C:):

PS D:\go\go-sample-detector> go run .
---- CURRENT CONFIGURATION ----
Root Path:             c:\
File name for results:     fdd-result.txt
File name for logs:        fdd-output.log
Logging level (info/debug): debug
Adds source info in logs:  false
-------------------------------
Progress (every 10 seconds):
45 folder 1_0_1 fetch 0_0_0 size 0_0_0 hash 0_0_0 match 0_0_0 result 0s
50 folder 10849_6798_4051 fetch 41712_0_41712 size 29731_20019_9712 hash 1654_757_897 match 432_0_432 result 10.0514188s
...
50 folder 827791_5379_822412 fetch 2175225_1_2175224 size 2136075_1166018_970057 hash 737102_171_736931 match 725972_0_725972 result 11m30.1975563s
50 folder 838528_3506_835022 fetch 2249794_0_2249794 size 2208711_1225693_983018 hash 750085_965_749120 match 737973_0_737973 result 11m40.1993708s
31 folder 843222_0_843222 fetch 2270482_0_2270482 size 2229228_1234927_994301 hash 761939_5_761934 match 750026_6_750020 result 11m50.2018749s
31 folder 843222_0_843222 fetch 2270482_0_2270482 size 2229228_1219410_1009818 hash 773913_0_773913 match 761518_0_761518 result 12m0.2085391s
31 folder 843222_0_843222 fetch 2270482_0_2270482 size 2229228_1205527_1023701 hash 787439_42_787397 match 774824_0_774824 result 12m10.2143553s
...
31 folder 843222_0_843222 fetch 2270482_0_2270482 size 2229228_24236_2204992 hash 1722277_1662_1720615 match 1700204_1_1700203 result 22m0.3185755s
5 folder 843222_0_843222 fetch 2270482_0_2270482 size 2229228_0_2229228 hash 1735310_0_1735310 match 1714249_0_1714249 result 22m10.3194308s
4 folder 843222_0_843222 fetch 2270482_0_2270482 size 2229228_0_2229228 hash 1735310_0_1735310 match 1714249_0_1714249 result 22m20.320499s
4 folder 843222_0_843222 fetch 2270482_0_2270482 size 2229228_0_2229228 hash 1735310_0_1735310 match 1714249_0_1714249 result 22m30.3211656s
------- TOTAL STATISTIC -------
Time of start:       11:29:09
Time of ended:       11:51:41
Duration:        22m31.7249332s
Total processed <count (size Mb)>:
- folders:           843222
- files:             2270482 (178753.517 Mb)
Performance of filtration <inp-filtered-out (out %)>:
- sizer:             2270482     41254       2229228 (98.18 %)
- hasher:            2229228     493918      1735310 (77.84 %)
- matcher:           1735310     21061       1714249 (98.79 %)
Found duplicates <count (size Mb)>:
- groups of files:   311941
- files:             1714249 (63811.304 Mb)
File with result: fdd-result.txt
File with logs: fdd-output.log
-------------------------------

HDD Disk Scan Example:

PS D:\go\go-sample-detector> go run .
---- CURRENT CONFIGURATION ----
Root Path:             d:\
File name for results:     fdd-result.txt
File name for logs:        fdd-output.log
Logging level (info/debug): debug
Adds source info in logs:  false
-------------------------------
Progress (every 10 seconds):
17 folder 0_0_0 fetch 0_0_0 size 0_0_0 hash 0_0_0 match 0_0_0 result 520.5µs
50 folder 382_224_158 fetch 1910_0_1910 size 606_279_327 hash 277_155_122 match 108_0_108 result 10.0512721s
...
50 folder 6712_454_6258 fetch 54626_0_54626 size 32410_25710_6700 hash 3972_2552_1420 match 1334_0_1334 result 5m20.0859867s
50 folder 6952_73_6879 fetch 55794_1_55793 size 33647_26147_7500 hash 4658_3204_1454 match 1365_0_1365 result 5m30.0865653s
31 folder 7008_0_7008 fetch 56138_0_56138 size 34005_26153_7852 hash 4791_3313_1478 match 1389_0_1389 result 5m40.0886867s
31 folder 7008_0_7008 fetch 56138_0_56138 size 34005_25868_8137 hash 4922_3373_1549 match 1451_0_1451 result 5m50.0892709s
...
31 folder 7008_0_7008 fetch 56138_0_56138 size 34005_567_33438 hash 22755_20341_2414 match 2291_0_2291 result 12m30.1476568s
31 folder 7008_0_7008 fetch 56138_0_56138 size 34005_315_33690 hash 22961_20490_2471 match 2353_0_2353 result 12m40.1494179s
20 folder 7008_0_7008 fetch 56138_0_56138 size 34005_0_34005 hash 23323_20744_2579 match 2459_0_2459 result 12m50.1500104s
20 folder 7008_0_7008 fetch 56138_0_56138 size 34005_0_34005 hash 23323_20608_2715 match 2591_0_2591 result 13m0.1504921s
...
20 folder 7008_0_7008 fetch 56138_0_56138 size 34005_0_34005 hash 23323_423_22900 match 22122_0_22122 result 29m10.2776533s
12 folder 7008_0_7008 fetch 56138_0_56138 size 34005_0_34005 hash 23323_0_23323 match 22555_0_22555 result 29m20.2791754s
11 folder 7008_0_7008 fetch 56138_0_56138 size 34005_0_34005 hash 23323_0_23323 match 22557_0_22557 result 29m30.2803337s
11 folder 7008_0_7008 fetch 56138_0_56138 size 34005_0_34005 hash 23323_0_23323 match 22557_0_22557 result 29m40.2814158s
------- TOTAL STATISTIC -------
Time of start:       12:03:35
Time of ended:       12:33:22
Duration:        29m46.4419118s
Total processed <count (size Mb)>:
- folders:             7008
- files:               56138 (645181.041 Mb)
Performance of filtration <inp-filtered-out (out %)>:
- sizer:               56138     22133       34005 (60.57 %)
- hasher:              34005     10682       23323 (68.59 %)
- matcher:             23323       764       22559 (96.72 %)
Found duplicates <count (size Mb)>:
- groups of files:     7243
- files:               22559 (15674.642 Mb)
File with result: fdd-result.txt
File with logs: fdd-output.log
-------------------------------

Example Logging Output (from fdd-output.log)

time=2025-06-15T11:29:09.260+03:00 level=DEBUG msg="InpProcess of Queue - started." poolName=fetchers
time=2025-06-15T11:29:09.260+03:00 level=DEBUG msg="OutProcess of Queue - started." poolName=fetchers
time=2025-06-15T11:29:09.260+03:00 level=DEBUG msg="Worker-pool - started." workerType=fdd.fetcher
time=2025-06-15T11:29:09.306+03:00 level=DEBUG msg="Worker-pool - started." workerType=*fdd.sizer
time=2025-06-15T11:29:09.306+03:00 level=DEBUG msg="Worker-pool - started." workerType=*fdd.hasher
time=2025-06-15T11:29:09.306+03:00 level=DEBUG msg="Worker-pool - started." workerType=*fdd.matcher
time=2025-06-15T11:29:09.308+03:00 level=DEBUG msg="InpProcess of Queue - started." poolName=sizers
time=2025-06-15T11:29:09.308+03:00 level=DEBUG msg="OutProcess of Queue - started." poolName=sizers
time=2025-06-15T11:29:09.308+03:00 level=DEBUG msg="InpProcess of Queue - started." poolName=matchers
time=2025-06-15T11:29:09.308+03:00 level=DEBUG msg="InpProcess of Queue - started." poolName=hashers
time=2025-06-15T11:29:09.308+03:00 level=DEBUG msg="OutProcess of Queue - started." poolName=hashers
time=2025-06-15T11:29:09.308+03:00 level=DEBUG msg="OutProcess of Queue - started." poolName=matchers
time=2025-06-15T11:29:09.308+03:00 level=DEBUG msg="InpProcess of Queue - started." poolName=packer
time=2025-06-15T11:29:09.308+03:00 level=DEBUG msg="OutProcess of Queue - started." poolName=packer
time=2025-06-15T11:29:09.311+03:00 level=INFO msg="Objects read error." item=*fdd.fetcher method=readDir() error="open c:\\$Recycle.Bin\\S-1-5-18: Access is denied." path=c:\$Recycle.Bin\S-1-5-18
...
time=2025-06-15T11:39:52.958+03:00 level=INFO msg="File open error." item=*fdd.hasher method=os.Open() error="open c:\\Windows\\System32\\restore\\MachineGuid.txt: Access is denied." path=c:\Windows\System32\restore\MachineGuid.txt
time=2025-06-15T11:39:57.011+03:00 level=INFO msg="File open error." item=*fdd.hasher method=os.Open() error="open c:\\Windows\\System32\\wbem\\AutoRecover\\3FFDD473F026FB198DA9FA65EE71383C.mof: Access is denied." path=c:\Windows\System32\wbem\AutoRecover\3FFDD473F026FB198DA9FA65EE71383C.mof
time=2025-06-15T11:40:55.247+03:00 level=DEBUG msg="InpProcess of Queue - stoped." poolName=fetchers
time=2025-06-15T11:40:55.247+03:00 level=DEBUG msg="OutProcess of Queue - stopped because queue is done and empty." poolName=fetchers
time=2025-06-15T11:40:55.250+03:00 level=DEBUG msg="Worker-pool - stoped." workerType=fdd.fetcher
time=2025-06-15T11:40:55.250+03:00 level=DEBUG msg="InpProcess of Queue - stoped." poolName=sizers
time=2025-06-15T11:40:55.252+03:00 level=DEBUG msg="OutProcess of Queue - stopped because queue is done and empty." poolName=sizers
time=2025-06-15T11:40:55.252+03:00 level=DEBUG msg="Worker-pool - stoped." workerType=*fdd.sizer
time=2025-06-15T11:40:55.252+03:00 level=DEBUG msg="InpProcess of Queue - stoped." poolName=hashers
...
time=2025-06-15T11:50:53.101+03:00 level=INFO msg="File open error." item=*fdd.hasher method=os.Open() error="open c:\\Windows\\System32\\wbem\\AutoRecover\\DA736886F13A0E2EE2265319FB376753.mof: Access is denied." path=c:\Windows\System32\wbem\AutoRecover\DA736886F13A0E2EE2265319FB376753.mof
time=2025-06-15T11:51:19.521+03:00 level=DEBUG msg="OutProcess of Queue - stopped because queue is done and empty." poolName=hashers
time=2025-06-15T11:51:19.522+03:00 level=DEBUG msg="Worker-pool - stoped." workerType=*fdd.hasher
time=2025-06-15T11:51:19.522+03:00 level=DEBUG msg="InpProcess of Queue - stoped." poolName=matchers
time=2025-06-15T11:51:19.522+03:00 level=DEBUG msg="OutProcess of Queue - stopped because queue is done and empty." poolName=matchers
time=2025-06-15T11:51:19.522+03:00 level=DEBUG msg="Worker-pool - stoped." workerType=*fdd.matcher
time=2025-06-15T11:51:19.522+03:00 level=DEBUG msg="InpProcess of Queue - stoped." poolName=packer
time=2025-06-15T11:51:19.522+03:00 level=DEBUG msg="OutProcess of Queue - started." poolName=packer

DEBUG level logs show the start and stop of each worker pool (Worker-pool - started./stoped.) and queue processing (InpProcess/OutProcess of Queue - started./stoped.). INFO level logs often indicate file or folder access errors, for example, to system directories (Access is denied.). This is expected behavior, as the application attempts to access all files in the specified root directory.

Example Result File (fdd-result.txt)

After completion, the application saves the results to a text file specified in the configuration (fdd-result.txt). The output format groups duplicates by size, hash, and group ID, then lists the paths to the duplicate files.

      2  {32  3876609034  0}
d:\HP_Drivers_for_Win10\SWSetup\SP92183\Graphics\ocl_cpu_version.ini
d:\HP_Drivers_for_Win10\SWSetup\SP95347\Graphics\ocl_cpu_version.ini
      2  {33  554973275  0}
d:\HP_Drivers_for_Win10\SWSetup\SP92183\DisplayAudio\6.16\version.ini
d:\HP_Drivers_for_Win10\SWSetup\SP95347\DisplayAudio\6.16\version.ini
      3  {33  4084763797  0}
d:\HP_Drivers_for_Win10\SWSetup\SP57014\Driver1\silentsetup.bat
d:\HP_Drivers_for_Win10\SWSetup\SP57014\Driver2\silentsetup.bat
d:\HP_Drivers_for_Win10\SWSetup\SP57014\silentsetup.bat
      2  {41  388051727  0}
d:\go\go-sample-queue\.git\refs\heads\master
d:\go\go-sample-queue\.git\refs\remotes\origin\master
      2  {41  954715591  0}
d:\go\go-sample-detector\.git\ORIG_HEAD
d:\go\go-sample-detector\.git\refs\heads\master
      3  {41  1012172877  0}
d:\go\go-sample-recursion\.git\ORIG_HEAD
d:\go\go-sample-recursion\.git\refs\heads\master
d:\go\go-sample-recursion\.git\refs\remotes\origin\master

Each group of duplicates starts with a line containing:

The number of files in the group (e.g., 2 or 3).
The structure {size hash group}:
- size: File size in bytes.
- hash: CRC32 hash of the files.
- group: Group ID (in this case, 0). This field is used for grouping identical files that have the same size and hash. If files have the same size and hash but their content differs (which is determined at the matcher stage), they will be assigned a different Group ID, and they will not be included in the same group of duplicates in the final result.
Followed by the full paths to each duplicate file in that group.

6. Conclusion

The presented implementation of duplicate file detection in Golang demonstrates the power of the pipelined approach and parallel processing. The use of channels for data transfer between stages, worker pools for parallel execution of tasks, and a well-designed checker mechanism makes the solution efficient and scalable. Built-in metrics and monitoring significantly simplify debugging and performance analysis.

I was surprised by the performance, despite the fact that I almost didn't think about optimization. On the contrary - the overhead in the form of monitoring, which I used to observe the load in real time - slows down the work.

Of course, there is always room for improvement. Potential enhancements could include:

Support for very large files using streaming processing and more sophisticated hashing algorithms.
Or more thoughtful use of disk cache (for example, not allowing the cache to "cool down" after the hasher before the matcher).

Nevertheless, this solution serves as a starting point for understanding and implementing pipelined systems in Golang.

The full source code is available at the link:
https://github.com/andrey-matveyev/go-sample-detector

Go along the Pipeline: Sizer, Hasher, Matcher

Andrey Matveyev — Fri, 13 Jun 2025 06:31:15 +0000

Anatomy of Workers in a File Duplicate Detector.

Let's imagine we set ourselves the task of creating a file duplicate detector. How would we approach this problem? Most likely, we would need a pipeline of specialized components, each performing its unique part of the work. In our previous articles, we laid the groundwork for building a file processing pipeline capable of handling large volumes of data, and we discussed the advantages of a modular approach and concurrent programming. Today, it's time to dive deeper into the heart of our pipeline - its workers - specialized components that perform the core work of detecting duplicates.

These workers are organized into sequential stages, each performing its unique task, passing the results to the next, more detailed stage. This design allows for efficient filtering of unique files at early stages, minimizing the load on subsequent, more resource-intensive steps.

Unit of Work: The `task` Structure

Before we delve into the workers, it's important to understand the information they operate on. Not just files, but their descriptions, encapsulated in the task structure, are passed along the pipeline between workers. In previous versions of our system, the task structure was simpler, but as the duplicate detector's functionality evolved, it expanded to carry all the necessary information for decision-making at various stages.

The current form of the task structure in our package looks like this:

// key contains the main file attributes used for identification and grouping.
type key struct {
    size  int64 // File size
    hash  uint32 // File hash (e.g., CRC32)
    equal int // Comparison counter (how many times the file was part of a potential duplicate group)
}

// info contains file metadata not directly used for comparison but important for processing.
type info struct {
    checked bool // Flag indicating whether the file has already been checked
    path    string // Full path to the file
}

// task is the main unit of work passed along the pipeline.
// It combines key and info
type task struct {
    key
    info
}

// newTask creates a new task instance.
func newTask(key key, info info) *task {
    return &task{key: key, info: info}
}

Checker: Guardian of Uniqueness

Before we look at specific workers, it's important to mention the checker component. It is not a worker itself but plays a critically important role in the operation of sizer, hasher, and matcher. The checker is a mechanism that helps avoid reprocessing already checked files or comparing a file to itself. It stores information about files that have already been "registered" at previous stages and helps determine whether the current file is a potential duplicate requiring further verification, or if it has already been processed. This ensures efficient filtering and reduces unnecessary work.

The implementation of checker looks like this:

package fdd

import "sync"

// checker defines the interface for checking and reviewing tasks.
type checker interface {
    verify(task *task) (checkedTask *task, detected bool) // Checks if a task has already been processed or is part of duplicates.
    review(task *task) (checkedTask *task)               // Reviews tasks for comparison with the current one.
}

// checkList - a concrete implementation of checker using a map to store information.
type checkList struct {
    mtx  sync.Mutex      // Mutex for safe access to the map from multiple goroutines.
    list map[key]info // Map where the key is the task key, and the value is info.
}

// newCheckList creates a new checker instance.
func newCheckList() checker {
    return &checkList{list: make(map[key]info)}
}

// verify checks if a task has been previously added to the list.
// Returns (task, true) if a duplicate is found that has not yet been marked as checked,
// (nil, true) if a duplicate is found and already checked, (nil, false) if no duplicate is found.
func (item *checkList) verify(task *task) (checkedTask *task, detected bool) {
    item.mtx.Lock()
    defer item.mtx.Unlock()

    info, detected := item.list[task.key]
    if detected {
        if info.checked { // If already checked, return nil
            return nil, detected
        }
        checkedTask = newTask(task.key, info) // Create a task to return
        info.checked = true                   // Mark as checked
        item.list[task.key] = info
        return checkedTask, detected
    }
    item.list[task.key] = task.info // If not found, add to the list
    return nil, detected
}

// review returns a task from the list for comparison with the current one, if it exists.
// Used to find the next candidate for byte-by-byte comparison.
func (item *checkList) review(task *task) (checkedTask *task) {
    item.mtx.Lock()
    defer item.mtx.Unlock()

    info, detected := item.list[task.key]
    if detected {
        return newTask(task.key, info) // Return the found task
    }
    item.list[task.key] = task.info // If not found, add to the list
    return nil
}

Sizer: The First Sieve

Let's start with the simplest, yet highly effective worker—the Sizer.

What it does: Its task is to determine the size of each file.
Why it's important: This is the first and fastest filtering stage. Two files cannot be duplicates if they have different sizes. The Sizer quickly discards files with unique sizes, sending only those with matching sizes further down the pipeline. Size information is extracted directly from file metadata (e.g., os.FileInfo), making this process very fast and not requiring reading file content.

Implementation of the Sizer worker:

// sizer is a worker that processes tasks by determining file size.
type sizer struct{}

// run is the main execution loop for the sizer worker.
// It reads tasks from the inp channel, checks their size (which is already known at this stage),
// and sends them to the out channel if they meet the criteria (not filtered).
func (item *sizer) run(inp, out chan *task, checker checker) {
    for currentTask := range inp {
        // Sizer logic is simple: it already has the size from the previous stage
        // and passes it to the next stage. No additional IO or blocking for size.
        checkedTask, detected := checker.verify(currentTask)
        if detected {
            out <- currentTask
            if checkedTask != nil {
                out <- checkedTask
            }
        }
    }
}

Hasher: The Second Sieve (More Refined)

The next step in our pipeline is the Hasher worker.

What it does: The Hasher is responsible for calculating the checksum (hash) of the file content. In our case, this could be, for example, CRC32.
Why it's more complex: Unlike the Sizer, the Hasher must read part or all of the file content to calculate the hash. This makes it more resource-intensive. However, it only operates on files that have passed the Sizer stage (i.e., have the same size).
Role in the pipeline: The Hasher performs a more precise filtering. If two files have the same size but different hashes, they are definitely not duplicates. Only files with the same size and the same hash proceed to the next, most resource-intensive stage.

Implementation of the Hasher worker:

// hasher is a worker that calculates the hash of files.
type hasher struct{}

// run is the main execution loop for the hasher worker.
// It reads tasks from the inp channel, opens the file, reads its content
// (or part of it) to calculate the hash, and then sends the task further.
func (item *hasher) run(inp, out chan *task, checker checker) {
    buf := make([]byte, 512) // Buffer for reading part of the file

    for inpTask := range inp {
        func(currentTask *task) { // Anonymous function for using defer
            file, err := os.Open(currentTask.path)
            // checkError logs the error if it's not nil and not io.EOF.
            // nil in the last parameter, as it's not relevant here.
            if checkError(err, "File open error.", "os.Open()", item, currentTask, nil) {
                return // Skip the task on open error
            }
            defer func(f *os.File) {
                closeErr := f.Close()
                // checkError logs the file close error.
                checkError(closeErr, "File close error.", "file.Close()", item, currentTask, nil)
            }(file)

            n, err := file.Read(buf) // Read part of the file for hashing
            // file will be ignored, if size=0 or Read returns a non-EOF error
            // checkError logs the file read error.
            if checkError(err, "File read error.", "file.Read()", item, currentTask, nil) {
                return // Skip the task on read error (excluding EOF)
            }

            currentTask.key.hash = crc32.ChecksumIEEE(buf[:n]) // Calculate hash
            checkedTask, detected := checker.verify(currentTask)
            if detected {
                out <- currentTask // Send the current task
                if checkedTask != nil {
                    out <- checkedTask // Also send the previously found duplicate
                }
            }
        }(inpTask)
    }
}

Matcher: Final Verification

And finally, the most complex and resource-intensive worker—the Matcher.

What it does: The Matcher performs a byte-by-byte comparison of files. If the Sizer and Hasher have determined that two files have the same size and the same hash, the Matcher conducts a final, unambiguous check by comparing their content byte by byte.
Why it's the most complex stage: Byte-by-byte comparison requires reading the entire content of both files, which is the most expensive operation in terms of time and I/O resources.
Role in the pipeline: This stage is the ultimate verification. Only after successfully passing the Matcher are files recognized as true duplicates. Thanks to the previous filters (Sizer and Hasher), the Matcher processes significantly fewer files, which is critically important for performance.

Implementation of the Matcher worker:

// matcher is a worker that performs byte-by-byte comparison of files to confirm duplicates.
type matcher struct{}

// run is the main execution loop for the matcher worker.
// It reads tasks from the inp channel and compares them with duplicate candidates using the checker.
func (item *matcher) run(inp, out chan *task, checker checker) {
    buf1 := make([]byte, 2*1024) // Buffer for reading data from the first file
    buf2 := make([]byte, 2*1024) // Buffer for reading data from the second file

    // Main loop: reads tasks from the input channel inp.
    for inpTask := range inp {
        // Anonymous function to encapsulate defer and control break/continue in the inner loop.
        func(currentTask *task) {
            for { // Inner loop to compare currentTask with all its potential duplicates.
                // Get the next candidate for byte-by-byte comparison from the checker.
                reviewedTask := checker.review(currentTask) 
                if reviewedTask == nil {
                    // If there are no more candidates for the current task, exit the inner loop.
                    break 
                }

                // Opening the first file for comparison.
                file1, err := os.Open(currentTask.path)
                // checkError logs the file open error and returns true if the current task needs to be interrupted.
                // nil in the last parameter, as reviewedTask is not used here for logging the path.
                if checkError(err, "File1 open error.", "os.Open()", item, currentTask, nil) {
                    // If there's an error opening file1, skip the current task and exit the inner loop.
                    break 
                }
                defer func() {
                    // Closing file1 if it was successfully opened.
                    if file1 != nil {
                        closeErr := file1.Close()
                        // checkError logs the file1 close error.
                        checkError(closeErr, "File1 close error.", "file1.Close()", item, currentTask, nil)
                    }
                }()

                // Opening the second file for comparison.
                file2, err := os.Open(reviewedTask.path)
                // checkError logs the file2 open error and returns true if the current task needs to be skipped.
                // reviewedTask is passed as context for the error if it occurred with the second file.
                if checkError(err, "File2 open error.", "os.Open()", item, reviewedTask, nil) {
                    // Increment the comparison counter for currentTask, as comparison with this reviewedTask failed.
                    currentTask.key.equal++ 
                    // If file2 cannot be opened, we still want to continue comparing currentTask
                    // with other reviewedTask candidates, if any.
                    // Rewind file1 to the beginning to prepare for the next comparison.
                    file1.Seek(0, io.SeekStart)
                    continue // Proceed to the next reviewedTask candidate.
                }
                defer func() {
                    // Closing file2 if it was successfully opened.
                    if file2 != nil {
                        closeErr := file2.Close()
                        // checkError logs the file2 close error.
                        checkError(closeErr, "File2 close error.", "file2.Close()", item, reviewedTask, nil)
                    }
                }()

                // Perform byte-by-byte comparison of the two files.
                filesEqual, checkErr := checkEqual(file1, file2, buf1, buf2)
                // Special handling for checkEqual error. Logged at Info level, as requested.
                // This may indicate issues reading from files during comparison.
                // currentTask and reviewedTask are passed for maximum information in the log.
                checkError(checkErr, "Check equal error (file1.Read() or file2.Read()).", "checkEqual()", item, currentTask, reviewedTask)

                if filesEqual {
                    // If the files are found to be byte-for-byte equal.
                    verifiedTask, detected := checker.verify(currentTask)
                    if detected {
                        out <- currentTask // Send the current file as a confirmed duplicate.
                        if verifiedTask != nil {
                            out <- verifiedTask // Also send the previously found duplicate, which is now confirmed.
                        }
                        break // A match is found; no need to compare currentTask with other candidates.
                    }
                    currentTask.key.equal++ // Increment the comparison counter.
                    // Rewind file1 for the next comparison (if the current task is not yet fully verified,
                    // i.e., it still needs comparisons for final duplicate group determination).
                    file1.Seek(0, io.SeekStart)
                    continue // Proceed to the next candidate.
                } else {
                    // If the files are not byte-for-byte equal.
                    currentTask.key.equal++ // Increment the comparison counter.
                    // Rewind file1 for the next comparison (with another candidate).
                    file1.Seek(0, io.SeekStart)
                    continue // Proceed to the next candidate.
                }
            }
        }(inpTask)
    }
}

// Byte-to-byte compare two files
func checkEqual(file1, file2 io.Reader, buf1, buf2 []byte) (bool, error) {
    for {
        n1, err1 := file1.Read(buf1)
        n2, err2 := file2.Read(buf2)

        if err1 == io.EOF && err2 == io.EOF {
            return true, nil
        }

        if err1 == io.EOF || err2 == io.EOF {
            return false, nil
        }

        if err1 != nil {
            return false, err1
        }
        if err2 != nil {
            return false, err2
        }

        if n1 != n2 {
            return false, nil
        }

        if !bytes.Equal(buf1[:n1], buf2[:n2]) {
            return false, nil
        }
    }
}

Conclusion

Thus, each worker in our pipeline performs a specialized task, contributing to efficient and multi-stage file filtering. From simple size determination to complex byte-by-byte comparison, these components work in unison to quickly and accurately find duplicates. The checker, in turn, ensures the correctness and optimization of this process at every step.

In the next article, we will combine all these components, as well as other elements such as queues and the dispatcher, to assemble a complete File Duplicate Detector (FDD) and demonstrate its full functionality.

Parallel Traversal of Recursive Structures in Go.

Andrey Matveyev — Sun, 08 Jun 2025 16:25:26 +0000

A Flexible Approach Without Stack Issues

Introduction

In the world of programming, we often encounter the task of processing data represented as recursive or hierarchical structures – be it a file system, XML/JSON trees, graph data structures, or object hierarchies. Traditional recursive traversal, while intuitive, has its limitations, especially in high-load systems or very deep structures, where it can lead to stack overflow. Moreover, synchronous traversal often inefficiently utilizes computational resources, particularly with I/O-bound operations.

In Go, with its powerful concurrency model based on goroutines and channels, we can implement an elegant and scalable approach to parallel processing of recursive structures that completely eliminates stack depth issues and efficiently parallelizes tasks.

In this article, we will delve into the architecture of such a solution. To manage the flow of tasks between parallel processors, we will utilize a flexible queue, which you can read more about in my previous publication. As a practical example, we will implement a parallel file system traversal to output a list of files and their sizes from a specified directory of arbitrary depth.

The Challenge of Traditional Recursion and the Go Solution

Classic recursive calls, where a function calls itself to process subtasks, are well-suited for structures with shallow depth. However, if the recursion depth becomes too great (e.g., when traversing a file system with thousands of nested directories), the operating system might allocate insufficient memory for the call stack, leading to a stack overflow error and program termination.

Our Go solution offers an alternative: instead of each goroutine recursively calling another function that, in turn, would spawn a new goroutine, we transform "recursive calls" into messages passed through channels. These messages represent new tasks (e.g., paths to subdirectories) that are added to a central queue. A pool of "workers" (goroutines) then concurrently extracts tasks from this queue, processes them, and, if necessary, generates new "recursive" tasks for subsequent processing.

This approach provides:

Elimination of stack overflow: Since goroutines do not make deep recursive calls into the stack, the stack overflow problem disappears.
Arbitrary traversal depth: The system is capable of processing structures of any nesting level.
Scalability: The number of parallel workers is easily adjustable, allowing for efficient utilization of available CPUs and resources.
Efficient I/O management: I/O-bound tasks (e.g., reading directories) can be performed concurrently without blocking the entire process.

Solution Architecture

Our solution consists of three main components interacting with each other via Go channels and managed by context.Context for proper shutdown.

Producer/Initiator (worker.Start): Responsible for launching the initial task (root directory) and managing the overall count of pending "recursive" tasks (sync.WaitGroup). It also handles closing the input channel for the queue when all tasks are processed.
Queue (queue package): Acts as a buffer between task producers (in this case, the initiator itself and workers finding new directories) and consumers (the worker pool). It decouples the rate of task production from the rate of consumption, preventing busy-waiting and overloads. A detailed description of the queue implementation is available at Building a Queue for Go Pipelines.
Worker Pool (worker.RunPool and worker.Worker): A set of goroutines that concurrently extract tasks from the queue, process them (e.g., read directory contents), and:
- If a file is found — send its information to the output channel.
- If a subdirectory is found — add a new "recursive" task to a channel that is fed back into the queue.

The architectural diagram you can see in the main picture of the publication.

Implementation: File System Traversal

Let's look at the key code parts that implement this architecture using a file system traversal as an example.

Task Structure (`queue.Task`)

To pass information about files and directories between components, we use a simple Task structure:

type Task struct {
    Size int64  // File size (0 for directories)
    Path string // Full path to the file or directory
}

Initiator and Completion Management (`worker.Start`)

The worker.Start function kicks off the entire traversal process. It performs two key actions:

Initializes a sync.WaitGroup (recCount), adding a counter for the initial directory.
Starts a goroutine that waits until recCount becomes zero (i.e., all recursive tasks are processed) or wait signal sync.Cond(recClose) about all workers was stoped (if context ctx was canceled) worker.Start closes the channel rec that is fed into the queue. This signals the complete end of the traversal.

var recCount sync.WaitGroup
var recClose = sync.NewCond(&sync.Mutex{})

func Start(path string) chan *queue.Task {
    rec := make(chan *queue.Task)

    // send first task to "rec" Chan
    recCount.Add(1)
    go func() {
        rec <- &queue.Task{Size: 0, Path: path}
    }()

    // Function "Start" - owner of "rec" Chan
    // "Start" is responsible for closing this channel.
    // The channel should be closed if one of two events happens:
    // 1. All tasks are completed "recCount.Wait() - unlocked"
    // 2. The context is canceled "all workers was stoped"

    // 1.
    // We wait for the first event and send a signal about it
    go func() {
        recCount.Wait()

        recClose.L.Lock()
        defer recClose.L.Unlock()

        recClose.Signal()
    }()

    // 2.
    // The second event can occur in the worker pool.
    // Since workers are also senders of data to the "rec" channel,
    // they also send a signal about the completion of their work
    // (when the context is canceled, for example)

    // Here we wait until one of two events happens.
    go func() {
        recClose.L.Lock()
        defer recClose.L.Unlock()

        recClose.Wait()
        close(rec)
    }()

    return rec
}

"Workhorses" (`worker.Worker` and `worker.RunPool`)

The worker pool is a set of concurrently running goroutines, each executing the task processing logic.

// Worker defines the interface for a task handler
// (a small overhead for a future project)
type Worker interface {
    run(inp, out, rec chan *queue.Task)
}

// NewWorker creates a new worker instance
func NewWorker() Worker {
    return &fetcher{}
}

// fetcher - implementation of the worker for file system traversal
type fetcher struct {}

// run - the main logic of the worker
func (item *fetcher) run(inp, out, rec chan *queue.Task) {
    for currentTask := range inp { // Read tasks from the input channel
        func() { // Anonymous function for defer
            defer recCount.Done() // Decrement the counter upon completion of the current task

            objects, err := readDir(currentTask.Path) // Read directory contents
            if err != nil {
                fmt.Printf("Objects read error. Path:%s  error:%s\n", currentTask.Path, err.Error())
                return
            }

            for _, object := range objects {
                objectPath := filepath.Join(currentTask.Path, object.Name())

                if object.IsDir() { // If it's a directory
                    recCount.Add(1)                       // Increment counter for a new recursive task
                    rec <- &queue.Task{Size: 0, Path: objectPath} // Send to the recursive tasks channel
                    continue
                }

                // If it's a file
                objectInfo, err := object.Info()
                if err != nil {
                    fmt.Printf("Object-info read error. Path:%s  error:%s\n", currentTask.Path, err.Error())
                    continue
                }
                out <- &queue.Task{Size: objectInfo.Size(), Path: objectPath} // Send file info to the output channel
            }
        }()
    }
}

// RunPool starts a pool of workers
func RunPool(runWorker Worker, amt int, inp, out, rec chan *queue.Task) {
    var workers sync.WaitGroup // WaitGroup to track the worker goroutines themselves

    for range amt { // Create 'amt' workers
        workers.Add(1)
        go func() {
            defer workers.Done()
            runWorker.run(inp, out, rec)
        }()
    }

    // This goroutine waiting for all workers to complete and closing the main output channel
    go func(currentWorker Worker, outChan chan *queue.Task) {
        workers.Wait()
        close(outChan)

        // 2.
        // For function "Start" (owner of "rec" Chan) we send signal about second event
        // (when workers are complete or the context is canceled and "rec" Chan can be closed)
        recClose.L.Lock()
        defer recClose.L.Unlock()

        recClose.Signal()
    }(runWorker, out)
}

objects, err := readDir(currentTask.Path) in the code above - custom implementation of os.ReadDir without slices.SortFunc of entries (if you need, os.ReadDir can be used)

func readDir(name string) ([]os.DirEntry, error) {
    file, err := os.Open(name)
    if err != nil {
        return nil, err
    }
    defer file.Close()
    dirs, err := file.ReadDir(-1)
    return dirs, err
}

Tying Everything Together (`main.go`)

The main function acts as the orchestrator, connecting all components and launching the pipeline.

package main

import (
    "context"
    "fmt"
    "main/queue"   // Import your queue package
    "main/worker" // Import your worker package
)

func main() {
    // Initialize context to manage the program's lifecycle
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // rec - the channel through which worker.Start will provide initial and "recursive" tasks
    rec := worker.Start(".") // Start traversal from the current directory

    // inp - the channel from which workers will read. It is the output of your queue,
    // which, in turn, reads tasks from the 'rec' channel.
    inp := queue.OutQueue(ctx, queue.InpQueue(rec))

    // out - the final channel for receiving traversal results (file information)
    out := make(chan *queue.Task)

    // Launching the worker pool:
    // NewWorker() - creates a new instance of the task handler
    // 2 - the number of parallel workers
    // inp - input channel for workers
    // out - output channel for traversal results (files)
    // rec - channel for "recursive" tasks (subdirectories)
    worker.RunPool(worker.NewWorker(), 2, inp, out, rec)

    // Main loop: read and print results from the final 'out' channel.
    // The loop will terminate when the 'out' channel is closed.
    for task := range out {
        fmt.Printf("%d %s\n", task.Size, task.Path)
    }

    fmt.Println("Traversal completed.")
}

How It Works: Flow of Execution

main calls worker.Start("."), which:
- Increments recCount by 1.
- Starts a goroutine that sends the root directory (.) to the rec channel.
- Starts a goroutine that waits signal then rec must closes.
main initializes inp := queue.OutQueue(ctx, queue.InpQueue(rec)). Your queue begins reading from rec and providing tasks via inp.
main calls worker.RunPool, which launches 2 worker goroutines. These workers start reading tasks from inp.
When a worker receives a task (directory) from inp:
- It decrements recCount (defer recCount.Done()).
- Reads the directory contents.
- For each file: sends its information to the out channel (final output).
- For each subdirectory: increments recCount by 1 and sends the path to this subdirectory back to the rec channel. This new task enters the queue, then inp, to be processed by any available worker.
When all directories and subdirectories are processed, recCount eventually becomes zero.
The goroutine launched by worker.Start, which was waiting for recCount.Wait(), receives the signal and closes the rec channel.
The closing of rec signals queue.InpQueue to complete, which leads to the closing of the internal queue and then the closing of the inp channel (output from the queue).
The closing of inp leads to the termination of all workers in the pool (for range inp loop finishes).
The goroutine launched by worker.RunPool, which was waiting for workers.Wait(), receives the signal and closes the final out channel.
main stops reading from out and exits.

Running the example produces output similar to this:
(this is the content of the current directory :)

PS D:\go\go-sample-recursion> go run .
0 .gitignore
26 go.mod
8 .git\COMMIT_EDITMSG
1079 LICENSE.txt
371 .git\config
1361 main.go
73 .git\description
2 README.md
116 .git\FETCH_HEAD
23 .git\HEAD
1561 queue\queue.go
750 .git\index
8301 queue\queue_test.go
41 .git\ORIG_HEAD
1942 worker\worker.go
...
465 .git\logs\refs\remotes\origin\master
41 .git\refs\remotes\origin\master
Traversal completed.
PS D:\go\go-sample-recursion>

Conclusion

The presented approach to parallel traversal of recursive structures in Go demonstrates the language's power and flexibility. By utilizing channels and sync.WaitGroup to manage task flow and synchronization, we have created a solution that:

Efficiently parallelizes processing.
Eliminates stack depth issues, allowing the processing of structures with arbitrary nesting levels.
Ensures reliable shutdown of all pipeline components.
Scales through the use of a worker pool.

This pattern is applicable not only to file systems but also to any other hierarchical data, making it a valuable tool in a Go developer's arsenal.

The full source code is available at the link:
https://github.com/andrey-matveyev/go-sample-recursion

P.S. (Postscript)

It's worth noting that in this article, I intentionally did not conduct detailed benchmarks comparing the performance of the presented parallel traversal implementation against Go's standard filepath.Walk function. The primary focus of this publication was not to determine which method is faster, but rather to illustrate an architectural pattern for parallelizing recursive processes and mitigating stack overflow issues.

However, based on my personal observations and without formal measurements, filepath.Walk feels slower, even when comparing it to our solution running with a single worker (fetcher). And yeah, I remember about disk cache. This is an informal observation, and formal benchmarks would be necessary to confirm it.

If any readers are inspired to build such benchmarks and compare the performance metrics, I would be genuinely interested in seeing the results and insights. Feel free to share your findings!

Building a Queue for Go Pipelines

Andrey Matveyev — Fri, 06 Jun 2025 18:07:50 +0000

Hello everyone! This is my first post on this platform, and I'm kicking it off with a small publication. Happy reading!

In the Go world, pipelines built on channels are powerful tools for organizing streaming data processing. However, standard channels don't always provide all the necessary flexibility, especially when you need to manage data flow between producers and consumers operating at different speeds, and avoid busy-waiting.
In this article, we'll explore how to build such a queue in Go, leveraging its native concurrency primitives to create a non-blocking, efficient, and context-aware solution.

The Challenge: Rate Mismatches in Pipelines

Consider a typical data processing pipeline:
Producer (Fast) -> Stage 1 (Medium) -> Stage 2 (Slow) -> Consumer (Variable)
If Stage 1 is much faster than Stage 2, data can pile up, potentially leading to memory issues (or other resources) or blocking the entire pipeline. Conversely, if Stage 1 is slow and Stage 2 is fast, Stage 2 might spend too much time waiting for data (busy-waiting). A queue acts as a buffer, decoupling these stages and allowing them to operate at their own pace.

Designing Our Queue

Our Go queue will have the following characteristics:

Non-blocking Producer/Consumer: Adding to or taking from the queue should not block the respective goroutines unnecessarily.
Context-aware Consumer: The consumer (reading from the queue) should respect context.Context for cancellation, allowing graceful shutdown.
Completion Signaling: The queue should clearly signal when no more items will arrive from the producer, allowing the consumer to finish processing and shut down cleanly.
Simplicity & Flexibility & Efficiency: Leveraging Go's built-in container/list for the underlying data structure and channels for synchronization.

Let's imagine our queue as a friendly middleman, taking tasks from a fast sender and passing them to a slower receiver, ensuring everyone is happy and the process flows smoothly. You can see the visualization of the idea in the main picture of the publication.

The Code Implementation

queue struct:

mtx sync.Mutex: Protects queueTasks from concurrent access, ensuring thread safety for push and pop.
innerChan chan struct{}: A buffered channel used for signaling. inpProcess sends a signal to outProcess whenever a new task is added. The buffer of 1 prevents inpProcess from blocking if outProcess isn't immediately ready to consume the signal. When inpProcess finishes, it closes innerChan to notify outProcess of completion.
queueTasks *list.List: Go's standard library doubly linked list. It's a good fit for a queue as PushBack and Remove(Front()) are efficient.

type queue struct {
    mtx        sync.Mutex
    innerChan  chan struct{}
    queueTasks *list.List
}

func newQueue() *queue {
    item := queue{}
    item.innerChan = make(chan struct{}, 1)
    item.queueTasks = list.New()
    return &item
}

func (item *queue) push(task *Task) {
    item.mtx.Lock()
    item.queueTasks.PushBack(task)
    item.mtx.Unlock()
}

func (item *queue) pop() *Task {
    item.mtx.Lock()
    defer item.mtx.Unlock()

    if item.queueTasks.Len() == 0 {
        return nil
    }
    elem := item.queueTasks.Front()
    item.queueTasks.Remove(elem)
    return elem.Value.(*Task)
}

type Task struct {
    ID   int
    Data string
}

Key Components:

There are two main functions: InpQueue (for the producer side) and OutQueue (for the consumer side).

`inpProcess` Goroutine (Producer Side):

Reads tasks from its input channel (inp).
queue.push(value): Adds tasks to the internal queueTasks list, protected by mtx.
select { case queue.innerChan <- struct{}{}: default: }: This is a non-blocking send. It attempts to send a signal to innerChan. If innerChan's buffer is full (meaning outProcess already knows there are tasks to process and hasn't read the previous signal yet), the default case is taken, and inpProcess continues without blocking.
close(queue.innerChan): Crucially, when the inp channel is closed (meaning the producer has no more tasks), inpProcess closes innerChan. This acts as the final signal to outProcess that the stream of incoming tasks has ended.

func InpQueue(inp chan *Task) *queue {
    queue := newQueue()
    go inpProcess(inp, queue)
    return queue
}

func inpProcess(inp chan *Task, queue *queue) {
    for value := range inp {
        queue.push(value)

        select {
        case queue.innerChan <- struct{}{}:
        default:
        }
    }
    close(queue.innerChan)
}

`outProcess` Goroutine (Consumer Side):

defer close(out): Ensures the output channel out is closed when outProcess exits, signaling to its downstream consumer that no more data will arrive.
select { case <-ctx.Done(): ... case _, ok := <-queue.innerChan: ... }: This select statement is the heart of the consumer's behavior:
- <-ctx.Done(): Checking the context for cancellation. If the main context is cancelled, outProcess immediately returns, stopping consumption gracefully.
- _, ok := <-queue.innerChan: Waits for a signal that new tasks are available or for innerChan to be closed. The ok boolean indicates whether the channel was closed (false) or a value was received (true).
Drain Loop (for { task := queue.pop() ... }): After receiving a signal (or detecting that innerChan has been closed), outProcess enters a loop to pop all currently available tasks from queueTasks. This is important because a single signal might correspond to multiple tasks having been pushed (e.g., if the producer was very fast).
if !ok { return }: After exhausting the internal queue, if innerChan was found to be closed (ok is false), it means inpProcess has finished, and no more tasks will ever arrive. outProcess then safely returns.

func OutQueue(ctx context.Context, queue *queue) chan *Task {
    out := make(chan *Task)
    go outProcess(ctx, queue, out)
    return out
}

func outProcess(ctx context.Context, queue *queue, out chan *Task) {
    defer close(out)
    for {
        select {
        case <-ctx.Done():
            return
        case _, ok := <-queue.innerChan:
            for {
                task := queue.pop()
                if task != nil {
                    select {
                    case out <- task:
                    case <-ctx.Done():
                        return
                    }
                } else {
                    break
                }
            }
            if !ok {
                return
            }
        }
    }
}

Example Usage

Here's how you might integrate and use this queue in a main function to simulate a producer and a slower consumer:

func main() {
    startTime := time.Now()
    // Initialize context to manage the entire pipeline
    mainCtx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // 1. Create an initial input channel for tasks
    // In a real system, we get it from the previous pipeline element.
    inpChan := make(chan *queue.Task)

    // 2. Embed our queue into the pipeline:
    // inpChan -> inpQueue (transforms channel to queue) -> outQueue (transforms queue to channel) -> outChan
    // This stage simulates some processing and includes the queue.
    outChan := queue.OutQueue(mainCtx, queue.InpQueue(inpChan))

    // 3. Start a producer goroutine:
    // It will generate tasks and send them to inpChan.
    produced := 0
    go func() {
        fmt.Printf("Producer: started. (%dms)\n", time.Since(startTime).Milliseconds())
        for i := range 5 {
            task := &queue.Task{ID: i, Data: fmt.Sprintf("Task #%d", i)}
            fmt.Printf("Producer: Sending %s  (%dms)\n", task.Data, time.Since(startTime).Milliseconds())
            inpChan <- task
            produced++
            time.Sleep(200 * time.Millisecond) // Simulate producer work
        }
        close(inpChan) // Important: close the input channel when all tasks are sent
        fmt.Printf("Producer: All tasks sent, input channel closed. (%dms)\n", time.Since(startTime).Milliseconds())
    }()

    // 4. Start a consumer goroutine:
    // It will read tasks from outChan (the output of our queue).
    consumed := 0
    go func() {
        fmt.Printf("Consumer: started. (%dms)\n", time.Since(startTime).Milliseconds())
        for task := range outChan {
            consumed++
            fmt.Printf("Consumer: Received %s  (%dms)\n", task.Data, time.Since(startTime).Milliseconds())
            time.Sleep(400 * time.Millisecond) // Simulate a slower consumer
        }
        fmt.Printf("Consumer: All tasks processed, output channel closed. (%dms)\n", time.Since(startTime).Milliseconds())
    }()
    // The pipeline will finish when inpChan closes -> inpProcess finishes ->
    // queue.innerChan closes -> outProcess finishes -> outChan closes.

    /*
        // Uncomment this code to see how context manages the operation's lifecycle.
        time.Sleep(1 * time.Second) // Timeout in case of hang
        fmt.Printf("Main: Timeout reached, cancelling context. (%dms)\n", time.Since(startTime).Milliseconds())
        cancel()
    */
    // Small delay for all goroutines to finish after cancellation/completion.
    time.Sleep(3 * time.Second)
    fmt.Printf("-produced: %d tasks, -consumed: %d tasks.\n", produced, consumed)
    fmt.Printf("Main: Application finished. (%dms)\n", time.Since(startTime).Milliseconds())
}

Running the example produces output similar to this:

PS D:\go\queue> go run .
Producer: started. (0ms)
Producer: Sending Task #0  (0ms)
Consumer: started. (0ms)
Consumer: Received Task #0  (1ms)
Producer: Sending Task #1  (201ms)
Producer: Sending Task #2  (401ms)
Consumer: Received Task #1  (401ms)
Producer: Sending Task #3  (602ms)
Consumer: Received Task #2  (802ms)
Producer: Sending Task #4  (803ms)
Producer: All tasks sent, input channel closed. (1004ms)
Consumer: Received Task #3  (1203ms)
Consumer: Received Task #4  (1603ms)
Consumer: All tasks processed, output channel closed. (2004ms)
-produced: 5 tasks, -consumed: 5 tasks.
Main: Application finished. (3001ms)

Notice that Producer is not blocked and works successfully even if Consumer has not yet started.
Task #0 is produced at 0ms and consumed at 1ms. But Task #1 is produced at 201ms and consumed at 401ms — the queue is buffering. The producer finishes sending all tasks by 803ms, but the consumer continues processing until 1603ms, demonstrating the decoupling effect of the queue.

How context manages the operation's lifecycle (example):

PS D:\go\queue> go run .
Producer: started. (0ms)
Producer: Sending Task #0  (0ms)
Consumer: started. (0ms)
Consumer: Received Task #0  (1ms)
Producer: Sending Task #1  (201ms)
Producer: Sending Task #2  (403ms)
Consumer: Received Task #1  (403ms)
Producer: Sending Task #3  (603ms)
Consumer: Received Task #2  (804ms)
Producer: Sending Task #4  (804ms)
Main: Timeout reached, cancelling context. (1000ms)
Producer: All tasks sent, input channel closed. (1005ms)
Consumer: All tasks processed, output channel closed. (1204ms)
-produced: 5 tasks, -consumed: 3 tasks.
Main: Application finished. (4001ms)

Conclusion

By implementing this queue pattern using Go's sync.Mutex, container/list, and especially channels for inter-goroutine signaling and context.Context for cancellation, we've created a flexible and efficient buffering mechanism for pipelines. This approach allows different stages of your concurrent applications to operate independently, smoothing out variable processing rates and ensuring robust, graceful shutdowns.

The full source code (with testing more cases) is available at the link:
https://github.com/andrey-matveyev/go-sample-queue

DEV Community: Andrey Matveyev

TicTacToe. Go Duel. AI vs Fate.

Neural Network vs Random Number Generator

Tic-Tac-Toe Game Logic

Deep Q-Learning Agent (DQN)

DQN Training Mechanism

Training Process and Results

Conclusion

Postscript

Developing a Neural Network in Golang.

...from Scratch: From Basics to XOR Solution

Why Golang and "From Scratch"?

Alternatives: Existing ML Libraries in Go

Architecture of Our Neural Network

Layer Structure: NeuralNetworkLayer

Weight Initialization: He Initialization

Activation Functions: ReLU and Sigmoid

Implementation Details: Forward and Backward Pass

Forward Method (Forward Pass)

Backward Method (Backward Pass)

Overall Network Structure: NeuralNetwork

Application: Solving the XOR Problem

XOR Data

Architecture for XOR

Training Process

Console Output Example

Conclusion

File Duplicate Detector. Go implementation.

SearchEngine, Processing Pipeline, Usage and Result.

1. SearchEngine: The Heart of Orchestration.

2. Building the Processing Pipeline: pipeline()

3. Metrics and Monitoring (metrics.go, monitor.go)

4. Example Usage

5. Obtained Results (Result)

Example of application execution from the console (scanning an SSD disk C:):

HDD Disk Scan Example:

Example Logging Output (from fdd-output.log)

Example Result File (fdd-result.txt)

6. Conclusion

Go along the Pipeline: Sizer, Hasher, Matcher

Anatomy of Workers in a File Duplicate Detector.

Unit of Work: The task Structure

Checker: Guardian of Uniqueness

Sizer: The First Sieve

Hasher: The Second Sieve (More Refined)

Matcher: Final Verification

Conclusion

Parallel Traversal of Recursive Structures in Go.

A Flexible Approach Without Stack Issues

Introduction

The Challenge of Traditional Recursion and the Go Solution

Solution Architecture

Implementation: File System Traversal

Task Structure (queue.Task)

Initiator and Completion Management (worker.Start)

"Workhorses" (worker.Worker and worker.RunPool)

Tying Everything Together (main.go)

How It Works: Flow of Execution

Conclusion

P.S. (Postscript)

Building a Queue for Go Pipelines

The Challenge: Rate Mismatches in Pipelines

Designing Our Queue

The Code Implementation

Key Components:

inpProcess Goroutine (Producer Side):

outProcess Goroutine (Consumer Side):

Example Usage

Conclusion

Layer Structure: `NeuralNetworkLayer`

`Forward` Method (Forward Pass)

`Backward` Method (Backward Pass)

Overall Network Structure: `NeuralNetwork`

1. `SearchEngine`: The Heart of Orchestration.

3. Metrics and Monitoring (`metrics.go`, `monitor.go`)

5. Obtained Results (`Result`)

Unit of Work: The `task` Structure

Task Structure (`queue.Task`)

Initiator and Completion Management (`worker.Start`)

"Workhorses" (`worker.Worker` and `worker.RunPool`)

Tying Everything Together (`main.go`)

`inpProcess` Goroutine (Producer Side):

`outProcess` Goroutine (Consumer Side):