Implementing a lock-free queue in C++ is a complex task because lock-free programming involves intricate memory operations and synchronization mechanisms, requiring a deep understanding of concurrent programming and hardware-level atomic operations. The advantage of lock-free queues is that they can provide better performance in high-concurrency scenarios, as they avoid the thread contention and context-switching overhead associated with traditional locking mechanisms.
A C++ lock-free queue is a multithreading technique that achieves thread-safe queues without using locks. It can enhance the performance of multithreaded programs. The main idea behind lock-free queues is to allow multiple threads to access the queue simultaneously without needing locks to protect shared resources. This can avoid issues such as lock contention and deadlocks, thereby improving program efficiency. Implementing a lock-free queue typically requires using atomic operation libraries from C++11 or higher (such as<atomic>), along with some advanced concurrency control techniques.
1. Overview of Lock-Free Queues
A lock-free queue (Lock-Free Queue) is a concurrent data structure used to implement efficient data exchange in a multithreaded environment. Compared to traditional lock-based queues, lock-free queues utilize special algorithms and techniques to avoid mutual exclusion operations between threads, thus enhancing concurrency performance and responsiveness.
Lock-free queues are typically implemented based on atomic operations or other low-level synchronization primitives, and they employ clever methods to ensure the correctness of operations. The main idea is to achieve thread safety through atomic read-write operations or similar mechanisms without explicitly locking the entire queue.
Typical lock-free queue algorithms include circular buffers and linked lists. Circular buffers usually use two pointers (head and tail) to represent the start and end positions of the queue, utilizing spinning, CAS (Compare-and-Swap), and other atomic operations for enqueueing and dequeueing operations. Linked lists achieve concurrent access by inserting or deleting nodes using CAS operations.
Advantages:
-
Provides better concurrent performance, avoiding performance bottlenecks caused by mutual exclusion operations.
-
Offers better scalability in highly competitive situations.
Disadvantages:
-
Implementation is relatively complex, requiring consideration of concurrency safety and correctness issues.
-
In highly competitive scenarios, spinning may lead to performance loss.
2. Principles of Lock-Free Queues
The principle of lock-free queues is to achieve concurrent safety through atomic operations or other low-level synchronization primitives. They avoid mutual exclusion operations found in traditional locking mechanisms to improve concurrency performance and responsiveness.
Typical lock-free queue algorithms include circular buffers and linked lists.
In the implementation of circular buffers, two pointers are typically used to represent the start and end positions of the queue, namely the head pointer and the tail pointer. During enqueueing, the tail pointer is updated using spinning, CAS (Compare-and-Swap), and other atomic operations, and the element is placed in the corresponding position. During dequeueing, the head pointer is similarly updated using atomic operations, and the element at the corresponding position is returned.
When implementing a lock-free queue with linked lists, CAS operations are used during node insertion or deletion to ensure that only one thread successfully modifies the pointer value of the node. This avoids locking the entire linked list.
Whether using a circular buffer or linked list implementation, the key point is how to utilize atomic operations to ensure coordination and consistency between different threads. It is necessary to carefully handle potential race conditions that may arise in concurrent scenarios and design suitable algorithms to ensure correctness and performance.
2.1 Queue Operation Models
A queue is a very important data structure characterized by first-in-first-out (FIFO), suitable for pipeline business flows. Queues are often used as buffers in inter-process communication and network communication to alleviate data processing pressure. According to the scenarios of queue operations, they can be divided into four major models: single producer – single consumer, multiple producers – single consumer, single producer – multiple consumers, and multiple producers – multiple consumers. Based on the data in the queue, they can be classified into fixed-length data in the queue and variable-length data in the queue.
(1) Single Producer – Single Consumer
(2) Multiple Producers – Single Consumer
(3) Single Producer – Multiple Consumers
(4) Multiple Producers – Multiple Consumers
2.2 CAS Operations
CAS stands for Compare and Swap, which is an atomic operation supported by all CPU instructions (the CMPXCHG assembly instruction in X86) used to implement various lock-free data structures.
The C language implementation of CAS is as follows:
bool compare_and_swap ( int *memory_location, int expected_value, int new_value)
{
if (*memory_location == expected_value)
{
*memory_location = new_value;
return true;
}
return false;
}CAS checks whether a memory location contains the expected value; if it does, it assigns the new value to the memory location. It returns true on success and false on failure.
(1) GGC supports CAS, available in GCC 4.1+ versions.
bool __sync_bool_compare_and_swap (type *ptr, type oldval, type newval, ...);
type __sync_val_compare_and_swap (type *ptr, type oldval, type newval, ...);(2) Windows supports CAS, using Windows API to support CAS.
LONG InterlockedCompareExchange(
LONG volatile *Destination,
LONG ExChange,
LONG Comperand
);(3) C11 supports CAS, with atomic functions in C11 STL supporting CAS and being cross-platform.
template < class T >
bool atomic_compare_exchange_weak( std::atomic<T>* obj,T* expected, T desired );
template < class T >
bool atomic_compare_exchange_weak( volatile std::atomic<T>* obj,T* expected, T desired );Other atomic operations include:
-
Fetch-And-Add: Generally used for atomic operations of +1 on a variable
-
Test-and-set: Writes a value to a memory location and returns its old value
2.3 Fixed-Length and Variable-Length Queue Data
(1) Fixed-Length Queue Data
(2) Variable-Length Queue Data
3. Lock-Free Queue Solutions
3.1 Boost Solutions
Boost provides three lock-free solutions, each suitable for different usage scenarios.
-
boost::lockfree::queue is a lock-free queue that supports multiple producer and consumer threads.
-
boost::lockfree::stack is a lock-free stack that supports multiple producer and consumer threads.
-
boost::lockfree::spsc_queue is a lock-free queue that supports only a single producer and single consumer thread, providing better performance than boost::lockfree::queue.
The API for Boost’s lock-free data structures achieves lock-free status through lightweight atomic locks, which is not truly lock-free in the strictest sense.
The queue provided by Boost can set an initial capacity, and when adding new elements, if the capacity is insufficient, the total capacity will automatically increase; however, for lock-free data structures, when adding new elements, if the capacity is insufficient, the total capacity will not automatically increase.
3.2 Concurrent Queue
ConcurrentQueue employs a lock-free algorithm to implement concurrent operations. It is based on CAS (Compare-and-Swap) atomic operations and other low-level synchronization primitives to ensure thread safety. Specifically, it uses spin locks and atomic instructions to ensure that modifications to the queue are atomic and provide correctness guarantees when sharing data among multiple threads.
ConcurrentQueue is an industrial-level lock-free queue solution implemented in C++.
GitHub: https://github.com/cameron314/concurrentqueue
ReaderWriterQueue is a single producer single consumer lock-free queue solution implemented in C++.
GitHub: https://github.com/cameron314/readerwriterqueueConcurrentQueue has the following features:
-
Thread-safe: Multiple threads can operate on the queue simultaneously without additional locking.
-
Non-blocking: Enqueue and dequeue operations are generally non-blocking and have low overhead.
-
First-in-first-out (FIFO) order: Elements are arranged in the order they are inserted, returning the earliest enqueued element during dequeueing.
Using ConcurrentQueue can conveniently handle shared data among multiple threads and reduce performance overhead caused by locking. However, it should be noted that although ConcurrentQueue provides efficient, thread-safe concurrent operations, it may not be suitable for all application scenarios under certain specific conditions, so it is necessary to evaluate based on specific needs when selecting data structures.
3.3 Disruptor
Disruptor is a high-performance concurrent programming framework used to implement lock-free concurrent data structures. It was originally developed by LMAX Exchange and has become a key component of its core trading engine.
Disruptor aims to solve data sharing and communication issues in highly multithreaded environments. It is based on a ring buffer and event-driven model, providing a highly efficient messaging mechanism through optimized memory access and thread scheduling.
Disruptor is a high-performance queue developed in JAVA by the UK foreign exchange company LMAX.
GitHub: https://github.com/LMAX-Exchange/disruptorMain features include:
-
Lock-free design: Disruptor uses lock-free algorithms such as CAS (Compare-and-Swap) to avoid competition and blocking caused by traditional locks.
-
High throughput: Disruptor utilizes ring buffers and pre-allocated memory technologies to pursue the highest possible processing speed while ensuring correctness.
-
Low latency: Due to its lock-free design and compact memory layout, Disruptor can achieve very low message processing latency.
-
Thread coordination: Disruptor provides flexible and powerful event publishing, consumer waiting, and triggering mechanisms that can be used to implement complex inter-thread communication patterns.
Using Disruptor can effectively resolve performance bottlenecks in the data transmission process of the producer-consumer model, especially suitable for high-concurrency, low-latency application scenarios such as financial trading systems and message queues. However, since Disruptor requires a higher understanding of the programming model, careful consideration is needed when using it, and it should be evaluated based on specific needs.
4. Implementing Lock-Free Queues
4.1 Circular Buffer
RingBuffer is a commonly used data structure in the producer-consumer model, where the producer appends data to the end of the array; when it reaches the end of the array, the producer wraps back to the start of the array; the consumer removes data from the head of the array; when it reaches the end of the array, the consumer wraps back to the start of the array.
If there is only one producer and one consumer, the circular buffer can be accessed without locks; the writing index of the circular buffer is only allowed to be accessed and modified by the producer. As long as the producer saves the new value to the buffer before updating the index, the consumer will always see a consistent data structure; the reading index is only allowed to be accessed and modified by the consumer, and as long as the consumer updates the read index after removing data, the producer will always see a consistent data structure.
-
When the queue is empty, front equals rear; when elements are enqueued, rear moves forward; when elements are dequeued, front moves forward.
-
When the queue is empty, rear equals front; when the queue is full, one position is left empty at the tail of the queue, so when judging if the circular queue is full, it uses (rear – front + maxn) % maxn.
Enqueue operation:
data[rear] = x;
rear = (rear + 1) % maxn;Dequeue operation:
x = data[front];
front = (front + 1) % maxn;Single Producer Single Consumer
In the single producer and single consumer scenario, since the read_index and write_index will only be modified by one thread, there is no need to lock or use atomic operations; it can be modified directly, but when reading and writing data, it is necessary to consider the situation when encountering the end of the array.
The read and write operations of threads on write_index and read_index are as follows:
-
(1) Write operation. First, check if the queue is full; if the queue is not full, write the data first, and after writing the data, update the write_index.
-
(2) Read operation. First, check if the queue is empty; if the queue is not empty, read the data first, and then update the read_index.
Multiple Producers Single Consumer
In the scenario of multiple producers and a single consumer, since multiple producers will modify the write_index, atomic operations must be used without locks.
4.2 RingBuffer Implementation
RingBuffer.hpp file:
#pragma once
template <class T>
class RingBuffer
{
public:
RingBuffer(unsigned size): m_size(size), m_front(0), m_rear(0)
{
m_data = new T[size];
}
~RingBuffer()
{
delete [] m_data;
m_data = NULL;
}
inline bool isEmpty() const
{
return m_front == m_rear;
}
inline bool isFull() const
{
return m_front == (m_rear + 1) % m_size;
}
bool push(const T& value)
{
if(isFull())
{
return false;
}
m_data[m_rear] = value;
m_rear = (m_rear + 1) % m_size;
return true;
}
bool push(const T* value)
{
if(isFull())
{
return false;
}
m_data[m_rear] = *value;
m_rear = (m_rear + 1) % m_size;
return true;
}
inline bool pop(T& value)
{
if(isEmpty())
{
return false;
}
value = m_data[m_front];
m_front = (m_front + 1) % m_size;
return true;
}
inline unsigned int front()const
{
return m_front;
}
inline unsigned int rear()const
{
return m_rear;
}
inline unsigned int size()const
{
return m_size;
}
private:
unsigned int m_size;// Queue length
int m_front;// Head index
int m_rear;// Tail index
T* m_data;// Data buffer
};RingBufferTest.cpp test code:
#include <stdio.h>
#include <thread>
#include <unistd.h>
#include <sys/time.h>
#include "RingBuffer.hpp"
class Test
{
public:
Test(int id = 0, int value = 0)
{
this->id = id;
this->value = value;
sprintf(data, "id = %d, value = %d\n", this->id, this->value);
}
void display()
{
printf("%s", data);
}
private:
int id;
int value;
char data[128];
};
double getdetlatimeofday(struct timeval *begin, struct timeval *end)
{
return (end->tv_sec + end->tv_usec * 1.0 / 1000000) -
(begin->tv_sec + begin->tv_usec * 1.0 / 1000000);
}
RingBuffer<Test> queue(1 << 12);
#define N (10 * (1 << 20))
void produce()
{
struct timeval begin, end;
gettimeofday(&begin, NULL);
unsigned int i = 0;
while(i < N)
{
if(queue.push(Test(i % 1024, i)))
{
i++;
}
}
gettimeofday(&end, NULL);
double tm = getdetlatimeofday(&begin, &end);
printf("producer tid=%lu %f MB/s %f msg/s elapsed= %f size= %u\n", pthread_self(), N * sizeof(Test) * 1.0 / (tm * 1024 * 1024), N * 1.0 / tm, tm, i);
}
void consume()
{
sleep(1);
Test test;
struct timeval begin, end;
gettimeofday(&begin, NULL);
unsigned int i = 0;
while(i < N)
{
if(queue.pop(test))
{
// test.display();
i++;
}
}
gettimeofday(&end, NULL);
double tm = getdetlatimeofday(&begin, &end);
printf("consumer tid=%lu %f MB/s %f msg/s elapsed= %f, size=%u \n", pthread_self(), N * sizeof(Test) * 1.0 / (tm * 1024 * 1024), N * 1.0 / tm, tm, i);
}
int main(int argc, char const *argv[])
{
std::thread producer1(produce);
std::thread consumer(consume);
producer1.join();
consumer.join();
return 0;
}Compilation:
g++ --std=c++11 RingBufferTest.cpp -o test -pthreadIn the single producer single consumer scenario, the message throughput is about 3.5 million messages per second.
4.3 LockFreeQueue Implementation
LockFreeQueue.hpp:
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h>
#include <stdbool.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/time.h>
#include <sys/mman.h>
#define SHM_NAME_LEN 128
#define MIN(a, b) ((a) > (b) ? (b) : (a))
#define IS_POT(x) ((x) && !((x) & ((x)-1)))
#define MEMORY_BARRIER __sync_synchronize()
template <class T>
class LockFreeQueue
{
protected:
typedef struct
{
int m_lock;
inline void spinlock_init()
{
m_lock = 0;
}
inline void spinlock_lock()
{
while(!__sync_bool_compare_and_swap(&m_lock, 0, 1)) {}
}
inline void spinlock_unlock()
{
__sync_lock_release(&m_lock);
}
} spinlock_t;
public:
// size: Queue size
// name: Path name for the shared memory key, defaults to NULL, using an array as the underlying buffer.
LockFreeQueue(unsigned int size, const char* name = NULL)
{
memset(shm_name, 0, sizeof(shm_name));
createQueue(name, size);
}
~LockFreeQueue()
{
if(shm_name[0] == 0)
{
delete [] m_buffer;
m_buffer = NULL;
}
else
{
if (munmap(m_buffer, m_size * sizeof(T)) == -1) {
perror("munmap");
}
if (shm_unlink(shm_name) == -1) {
perror("shm_unlink");
}
}
}
bool isFull()const
{
#ifdef USE_POT
return m_head == (m_tail + 1) && (m_size - 1);
#else
return m_head == (m_tail + 1) % m_size;
#endif
}
bool isEmpty()const
{
return m_head == m_tail;
}
unsigned int front()const
{
return m_head;
}
unsigned int tail()const
{
return m_tail;
}
bool push(const T& value)
{
#ifdef USE_LOCK
m_spinLock.spinlock_lock();
#endif
if(isFull())
{
#ifdef USE_LOCK
m_spinLock.spinlock_unlock();
#endif
return false;
}
memcpy(m_buffer + m_tail, &value, sizeof(T));
#ifdef USE_MB
MEMORY_BARRIER;
#endif
#ifdef USE_POT
m_tail = (m_tail + 1) && (m_size - 1);
#else
m_tail = (m_tail + 1) % m_size;
#endif
#ifdef USE_LOCK
m_spinLock.spinlock_unlock();
#endif
return true;
}
bool pop(T& value)
{
#ifdef USE_LOCK
m_spinLock.spinlock_lock();
#endif
if (isEmpty())
{
#ifdef USE_LOCK
m_spinLock.spinlock_unlock();
#endif
return false;
}
memcpy(&value, m_buffer + m_head, sizeof(T));
#ifdef USE_MB
MEMORY_BARRIER;
#endif
#ifdef USE_POT
m_head = (m_head + 1) && (m_size - 1);
#else
m_head = (m_head + 1) % m_size;
#endif
#ifdef USE_LOCK
m_spinLock.spinlock_unlock();
#endif
return true;
}
protected:
virtual void createQueue(const char* name, unsigned int size)
{
#ifdef USE_POT
if (!IS_POT(size))
{
size = roundup_pow_of_two(size);
}
#endif
m_size = size;
m_head = m_tail = 0;
if(name == NULL)
{
m_buffer = new T[m_size];
}
else
{
int shm_fd = shm_open(name, O_CREAT | O_RDWR, 0666);
if (shm_fd < 0)
{
perror("shm_open");
}
if (ftruncate(shm_fd, m_size * sizeof(T)) < 0)
{
perror("ftruncate");
close(shm_fd);
}
void *addr = mmap(0, m_size * sizeof(T), PROT_READ | PROT_WRITE, MAP_SHARED, shm_fd, 0);
if (addr == MAP_FAILED)
{
perror("mmap");
close(shm_fd);
}
if (close(shm_fd) == -1)
{
perror("close");
exit(1);
}
m_buffer = static_cast<T*>(addr);
memcpy(shm_name, name, SHM_NAME_LEN - 1);
}
#ifdef USE_LOCK
spinlock_init(m_lock);
#endif
}
inline unsigned int roundup_pow_of_two(size_t size)
{
size |= size >>> 1;
size |= size >>> 2;
size |= size >>> 4;
size |= size >>> 8;
size |= size >>> 16;
size |= size >>> 32;
return size + 1;
}
protected:
char shm_name[SHM_NAME_LEN];
volatile unsigned int m_head;
volatile unsigned int m_tail;
unsigned int m_size;
#ifdef USE_LOCK
spinlock_t m_spinLock;
#endif
T* m_buffer;
};#define USE_LOCK
// Enable spinlock for multi-producer multi-consumer scenario
#define USE_MB
// Enable Memory Barrier
#define USE_POT
// Enable queue size alignment to the power of 2LockFreeQueueTest.cpp test file:
#include "LockFreeQueue.hpp"
#include <thread>
//#define USE_LOCK
class Test
{
public:
Test(int id = 0, int value = 0)
{
this->id = id;
this->value = value;
sprintf(data, "id = %d, value = %d\n", this->id, this->value);
}
void display()
{
printf("%s", data);
}
private:
int id;
int value;
char data[128];
};
double getdetlatimeofday(struct timeval *begin, struct timeval *end)
{
return (end->tv_sec + end->tv_usec * 1.0 / 1000000) -
(begin->tv_sec + begin->tv_usec * 1.0 / 1000000);
}
LockFreeQueue<Test> queue(1 << 10, "/shm");
#define N ((1 << 20))
void produce()
{
struct timeval begin, end;
gettimeofday(&begin, NULL);
unsigned int i = 0;
while(i < N)
{
if(queue.push(Test(i >>> 10, i)))
i++;
}
gettimeofday(&end, NULL);
double tm = getdetlatimeofday(&begin, &end);
printf("producer tid=%lu %f MB/s %f msg/s elapsed= %f size= %u\n", pthread_self(), N * sizeof(Test) * 1.0 / (tm * 1024 * 1024), N * 1.0 / tm, tm, i);
}
void consume()
{
Test test;
struct timeval begin, end;
gettimeofday(&begin, NULL);
unsigned int i = 0;
while(i < N)
{
if(queue.pop(test))
{
//test.display();
i++;
}
}
gettimeofday(&end, NULL);
double tm = getdetlatimeofday(&begin, &end);
printf("consumer tid=%lu %f MB/s %f msg/s elapsed= %f size=%u \n", pthread_self(), N * sizeof(Test) * 1.0 / (tm * 1024 * 1024), N * 1.0 / tm, tm, i);
}
int main(int argc, char const *argv[])
{
std::thread producer1(produce);
//std::thread producer2(produce);
std::thread consumer(consume);
producer1.join();
//producer2.join();
consumer.join();
return 0;
}In a multithreading scenario, the USE_LOCK macro needs to be defined to enable lock protection.
Compilation:
g++ --std=c++11 -O3 LockFreeQueueTest.cpp -o test -lrt -pthread5. Kernel Queue: kfifo
Computer scientists have proven that when there is only one reading thread and one writing thread operating concurrently, no additional locks are needed to ensure thread safety, which means kfifo uses lock-free programming techniques to enhance the kernel’s concurrency.
The Linux kernel has never lacked concise, elegant, and efficient code; rather, we lack the vision to discover and appreciate it. In the Linux kernel, simplicity does not mean the use of transcendental techniques that are elusive; instead, it uses basic data structures that everyone is very familiar with, but kernel developers can extract beautiful features from these basic data structures.
kfifo is one such elegant piece of code, extremely concise, with no extraneous lines of code, yet highly efficient.
Information about kfifo is as follows:
The original code version analyzed in this article: 2.6.24.4
kfifo's definition file: kernel/kfifo.c
kfifo's header file: include/linux/kfifo.hkfifo is a FIFO data structure in the Linux kernel that implements a circular queue data structure to provide an unbounded byte stream service and employs parallel lock-free programming techniques, allowing concurrent operations of a single producer and single consumer thread without any locking behavior to ensure kfifo’s thread safety.
Given that the kfifo code carries so many features, let’s take a look at its code:
struct kfifo {
unsigned char *buffer; /* the buffer holding the data */
unsigned int size; /* the size of the allocated buffer */
unsigned int in; /* data is added at offset (in % size) */
unsigned int out; /* data is extracted from off. (out % size) */
spinlock_t *lock; /* protects concurrent modifications */
};This is the data structure of kfifo. kfifo mainly provides two operations, __kfifo_put (enqueue operation) and __kfifo_get (dequeue operation). Its various data members are as follows:
-
buffer: Used to store data
-
size: The size of the buffer space, initialized by expanding it to the next power of 2 (e.g., 5, expanding to the nearest value that is a power of 2, which is 2^3, i.e., 8)
-
lock: If used, it cannot guarantee that at most one read thread and one write thread are active at any time, and this lock must be used to enforce synchronization.
-
in, out: Together with the buffer, they form a circular queue. in points to the queue head in the buffer, while out points to the queue tail in the buffer
Its structure is illustrated as follows:
+--------------------------------------------------------------+
| |<----------data---------->| |
+--------------------------------------------------------------+
^ ^ ^
| | |
out in sizeOf course, kernel developers have used a better technique to handle the relationship between in, out, and buffer, which we will analyze in detail below.
5.1 kfifo Function Description
kfifo provides the following external functional specifications:
-
Only supports concurrent operations of one reader and one writer
-
Non-blocking read and write operations; if space is insufficient, it returns the actual accessed space
(1) kfifo_alloc Allocates kfifo memory and initializes it
struct kfifo *kfifo_alloc(unsigned int size, gfp_t gfp_mask, spinlock_t *lock)
{
unsigned char *buffer;
struct kfifo *ret;
/*
* Round up to the next power of 2, since our 'let the indices
* wrap' technique works only in this case.
*/
if (size && (size - 1)) {
BUG_ON(size > 0x80000000);
size = roundup_pow_of_two(size);
}
buffer = kmalloc(size, gfp_mask);
if (!buffer)
return ERR_PTR(-ENOMEM);
ret = kfifo_init(buffer, size, gfp_mask, lock);
if (IS_ERR(ret))
kfree(buffer);
return ret;
}It is worth mentioning that the value of kfifo->size is always expanded to the next power of 2 based on the size parameter passed by the caller (roundup_pow_of_two, my own implementation at the end of the article). The advantage of this is obvious—taking the modulo of kfifo->size can be transformed into a bitwise operation, as follows:
kfifo->in % kfifo->size can be transformed to kfifo->in & (kfifo->size – 1)In the kfifo_alloc function, the use of size & (size – 1) determines whether the size is a power of 2; if true, it indicates that size is not a power of 2, then calls roundup_pow_of_two to expand it to a power of 2.
These are all common techniques; it’s just that everyone hasn’t thought to combine them. The following analysis of __kfifo_put and __kfifo_get will fully utilize the characteristics of kfifo->size.
(2) __kfifo_put and __kfifo_get Clever Enqueue and Dequeue
__kfifo_put is the enqueue operation; it first places data into the buffer, and only then modifies the in parameter; __kfifo_get is the dequeue operation; it first removes data from the buffer and only then modifies out (ensuring that even if in and out modifications fail, it can be retried).
You will find that in and out each have their responsibilities. Below are the codes for __kfifo_put and __kfifo_get:
unsigned int __kfifo_put(struct kfifo *fifo,
unsigned char *buffer, unsigned int len)
{
unsigned int l;
len = min(len, fifo->size - fifo->in + fifo->out);
/*
* Ensure that we sample the fifo->out index -before- we
* start putting bytes into the kfifo.
*/
smp_mb();
/* first put the data starting from fifo->in to buffer end */
l = min(len, fifo->size - (fifo->in && (fifo->size - 1)));
memcpy(fifo->buffer + (fifo->in && (fifo->size - 1)), buffer, l);
/* then put the rest (if any) at the beginning of the buffer */
memcpy(fifo->buffer, buffer + l, len - l);
/*
* Ensure that we add the bytes to the kfifo -before-
* we update the fifo->in index.
*/
smp_wmb();
fifo->in += len;
return len;
}Is it strange? The code is entirely linear, with no if-else branches to determine whether there is enough space to store data. The kernel code here is very concise, with no extraneous lines of code.
l = min(len, fifo->size - (fifo->in && (fifo->size - 1)));This expression calculates the current writable space, which can be understood in human language as:
l = the minimum of writable space in kfifo and expected writable space(3)Using the min macro instead of if-else branches
__kfifo_get also applies the same technique; the code is as follows:
unsigned int __kfifo_get(struct kfifo *fifo,
unsigned char *buffer, unsigned int len)
{
unsigned int l;
len = min(len, fifo->in - fifo->out);
/*
* Ensure that we sample the fifo->in index -before- we
* start removing bytes from the kfifo.
*/
smp_rmb();
/* first get the data from fifo->out until the end of the buffer */
l = min(len, fifo->size - (fifo->out && (fifo->size - 1)));
memcpy(buffer, fifo->buffer + (fifo->out && (fifo->size - 1)), l);
/* then get the rest (if any) from the beginning of the buffer */
memcpy(buffer + l, fifo->buffer, len - l);
/*
* Ensure that we remove the bytes from the kfifo -before-
* we update the fifo->out index.
*/
smp_mb();
fifo->out += len;
return len;
}Read it carefully twice; I have done so multiple times and always discover something new, because the relationship between in, out, and size is too clever, utilizing the properties of unsigned int wrapping.
It turns out that kfifo’s in and out simply keep increasing until the maximum value of unsigned int, at which point they wrap back to 0. However, they always satisfy:
kfifo->in - kfifo->out <= kfifo->size
Even if kfifo->in wraps back to the beginning, this property still holds.For a given kfifo:
The length of the data space is: kfifo->in - kfifo->out
The length of the remaining space (writable space) is: kfifo->size - (kfifo->in - kfifo->out)Although kfifo->in and kfifo->out keep increasing beyond kfifo->size, the actual indices in kfifo->buffer are as follows:
kfifo->in % kfifo->size (i.e., kfifo->in && (kfifo->size - 1))
kfifo->out % kfifo->size (i.e., kfifo->out && (kfifo->size - 1))When writing a block of data into kfifo, if the relationship between data space, writable space, and kfifo->size satisfies:
kfifo->in % size + len > sizeThen a write split is needed, as illustrated below:
kfifo_put (writing) space start address
|
\_/
|XXXXXXXXXX
XXXXXXXX|
+--------------------------------------------------------------+
| |<----------data---------->| |
+--------------------------------------------------------------+
^ ^ ^
| | |
out%size in%size size
^
|
Write space end address The first block is: [kfifo->in % kfifo->size, kfifo->size] and the second block is: [0, len – (kfifo->size – kfifo->in % kfifo->size)].
Below is the code; savor it:
/* first put the data starting from fifo->in to buffer end */
l = min(len, fifo->size - (fifo->in && (fifo->size - 1)));
memcpy(fifo->buffer + (fifo->in && (fifo->size - 1)), buffer, l);
/* then put the rest (if any) at the beginning of the buffer */
memcpy(fifo->buffer, buffer + l, len - l); The kfifo_get process is similar; please analyze it yourself.
(4) kfifo_get and kfifo_put Lock-Free Concurrent Operations
Computer scientists have proven that when there is only one reading thread and one writing thread operating concurrently, no additional locks are needed to ensure thread safety, which means kfifo uses lock-free programming techniques to enhance the kernel’s concurrency.
kfifo uses two pointers, in and out, to describe the writing and reading cursors; for writing operations, only the in pointer is updated, while for reading operations, only the out pointer is updated, which can be said to be non-interfering, as illustrated below:
|<--Writing-->|
+--------------------------------------------------------------+
| |<----------data----->| |
+--------------------------------------------------------------+
|<--Reading-->|
^ ^ ^
| | |
out in sizeTo avoid readers seeing that the writer has expected to write, but actually has not written data, the writer must ensure the following write order:
Write data into the space [kfifo->in, kfifo->in + len]
Update the kfifo->in pointer to kfifo->in + lenAt the completion of operation 1, the reader still does not see the written information, as kfifo->in has not changed, leading the reader to believe that the writer has not started the write operation; only after updating kfifo->in can the reader see it.
So how to ensure that operation 1 must complete before operation 2? The secret lies in using memory barriers: smp_mb(), smp_rmb(), smp_wmb(), to ensure the order of memory operations observed by the other party.
5.2 kfifo Kernel Queue Implementation
The kfifo data structure is defined as follows:
struct kfifo
{
unsigned char *buffer;
unsigned int size;
unsigned int in;
unsigned int out;
spinlock_t *lock;
};
// Create queue
struct kfifo *kfifo_init(unsigned char *buffer, unsigned int size, gfp_t gfp_mask, spinlock_t *lock)
{
struct kfifo *fifo;
// Check if it is a power of 2
BUG_ON(!is_power_of_2(size));
fifo = kmalloc(sizeof(struct kfifo), gfp_mask);
if (!fifo)
return ERR_PTR(-ENOMEM);
fifo->buffer = buffer;
fifo->size = size;
fifo->in = fifo->out = 0;
fifo->lock = lock;
return fifo;
}
// Allocate space
struct kfifo *kfifo_alloc(unsigned int size, gfp_t gfp_mask, spinlock_t *lock)
{
unsigned char *buffer;
struct kfifo *ret;
// Check if it is a power of 2
if (!is_power_of_2(size))
{
BUG_ON(size > 0x80000000);
// Round up to the next power of 2
size = roundup_pow_of_two(size);
}
buffer = kmalloc(size, gfp_mask);
if (!buffer)
return ERR_PTR(-ENOMEM);
ret = kfifo_init(buffer, size, gfp_mask, lock);
if (IS_ERR(ret))
kfree(buffer);
return ret;
}
void kfifo_free(struct kfifo *fifo)
{
kfree(fifo->buffer);
kfree(fifo);
}
// Enqueue operation
static inline unsigned int kfifo_put(struct kfifo *fifo, const unsigned char *buffer, unsigned int len)
{
unsigned long flags;
unsigned int ret;
spin_lock_irqsave(fifo->lock, flags);
ret = __kfifo_put(fifo, buffer, len);
spin_unlock_irqrestore(fifo->lock, flags);
return ret;
}
// Dequeue operation
static inline unsigned int kfifo_get(struct kfifo *fifo, unsigned char *buffer, unsigned int len)
{
unsigned long flags;
unsigned int ret;
spin_lock_irqsave(fifo->lock, flags);
ret = __kfifo_get(fifo, buffer, len);
// When fifo->in == fifo->out, buffer is empty
if (fifo->in == fifo->out)
fifo->in = fifo->out = 0;
spin_unlock_irqrestore(fifo->lock, flags);
return ret;
}
// Enqueue operation
unsigned int __kfifo_put(struct kfifo *fifo, const unsigned char *buffer, unsigned int len)
{
unsigned int l;
// Empty length in the buffer
len = min(len, fifo->size - fifo->in + fifo->out);
// Memory barrier: smp_mb(), smp_rmb(), smp_wmb() to ensure the order of memory operations observed by the other party
smp_mb();
// Append data to the end of the queue
l = min(len, fifo->size - (fifo->in && (fifo->size - 1)));
memcpy(fifo->buffer + (fifo->in && (fifo->size - 1)), buffer, l);
memcpy(fifo->buffer, buffer + l, len - l);
smp_wmb();
// Increment each time; when reaching the maximum value, it overflows and automatically resets to 0
fifo->in += len;
return len;
}
// Dequeue operation
unsigned int __kfifo_get(struct kfifo *fifo, unsigned char *buffer, unsigned int len)
{
unsigned int l;
// Length of data in the buffer
len = min(len, fifo->in - fifo->out);
smp_rmb();
l = min(len, fifo->size - (fifo->out && (fifo->size - 1)));
memcpy(buffer, fifo->buffer + (fifo->out && (fifo->size - 1)), l);
memcpy(buffer + l, fifo->buffer, len - l);
smp_mb();
fifo->out += len; // Increment each time; when reaching the maximum value, it overflows and automatically resets to 0
return len;
}
static inline void __kfifo_reset(struct kfifo *fifo)
{
fifo->in = fifo->out = 0;
}
static inline void kfifo_reset(struct kfifo *fifo)
{
unsigned long flags;
spin_lock_irqsave(fifo->lock, flags);
__kfifo_reset(fifo);
spin_unlock_irqrestore(fifo->lock, flags);
}
static inline unsigned int __kfifo_len(struct kfifo *fifo)
{
return fifo->in - fifo->out;
}
static inline unsigned int kfifo_len(struct kfifo *fifo)
{
unsigned long flags;
unsigned int ret;
spin_lock_irqsave(fifo->lock, flags);
ret = __kfifo_len(fifo);
spin_unlock_irqrestore(fifo->lock, flags);
return ret;
}5.3 kfifo Design Points
(1) Ensure buffer size is a power of 2
The value of kfifo->size is expanded to the next power of 2 based on the size parameter passed by the caller, aiming to make the modulo operation on kfifo->size convertible to a bitwise operation (improving runtime efficiency). Ensuring size is a power of 2 can greatly enhance efficiency in frequent queue operations.
(2) Implement synchronization using spin_lock_irqsave and spin_unlock_irqrestore.
In the Linux kernel, there are spin_lock, spin_lock_irq, and spin_lock_irqsave to guarantee synchronization.
static inline void __raw_spin_lock(raw_spinlock_t *lock)
{
preempt_disable();
spin_acquire(&lock->dep_map, 0, 0, _RET_IP_);
LOCK_CONTENDED(lock, do_raw_spin_trylock, do_raw_spin_lock);
}
static inline void __raw_spin_lock_irq(raw_spinlock_t *lock)
{
local_irq_disable();
preempt_disable();
spin_acquire(&lock->dep_map, 0, 0, _RET_IP_);
LOCK_CONTENDED(lock, do_raw_spin_trylock, do_raw_spin_lock);
}spin_lock is faster than spin_lock_irq but is not thread-safe. spin_lock_irq increases the call to local_irq_disable, which disables local interrupts and is thread-safe, preventing local interrupts and kernel preemption.
spin_lock_irqsave is an auxiliary interface based on spin_lock_irq, which does not change the state of interrupt enablement or disablement after entering and exiting the critical section.
If a spin lock is used in an interrupt handler, local interrupts must be disabled before acquiring the spin lock; spin_lock_irqsave is implemented as follows:
-
A. Save the local interrupt state;
-
B. Disable local interrupts;
-
C. Acquire the spin lock.
Unlocking is completed by spin_unlock_irqrestore, which releases the lock and restores the local interrupt to its original state.
(3) Linear code structure
The code does not contain any if-else branches to determine whether there is enough space to store data; kfifo’s enqueue or dequeue operations simply increment +len, and do not perform modulo operations on kfifo->size, so kfifo->in and kfifo->out continuously increase until the maximum value of unsigned int, at which point they wrap back to 0, but always satisfy:
kfifo->in - kfifo->out <= kfifo->size.(4) Use Memory Barrier
-
mb(): Suitable for memory barriers in both multiprocessor and single-processor environments.
-
rmb(): Suitable for read memory barriers in both multiprocessor and single-processor environments.
-
wmb(): Suitable for write memory barriers in both multiprocessor and single-processor environments.
-
smp_mb(): Suitable for multiprocessor memory barriers.
-
smp_rmb(): Suitable for multiprocessor read memory barriers.
-
smp_wmb(): Suitable for multiprocessor write memory barriers.
Memory Barrier usage scenarios include:
-
A. Implementing synchronization primitives
-
B. Implementing lock-free data structures
-
C. Driver programs
During program execution, the actual order of memory access may not align with the order written in the program code due to memory out-of-order access. Memory out-of-order access occurs to enhance runtime performance. This behavior primarily happens in two stages:
-
A. At compile time, compiler optimizations can cause memory out-of-order access (instruction reordering).
-
B. At runtime, interactions between multiple CPUs can lead to memory out-of-order access.
Memory barriers ensure that memory access operations before the barrier are completed before those after it. Memory barriers include two categories:
-
A. Compiler Memory Barriers.
-
B. CPU Memory Barriers.
Typically, memory out-of-order access caused by compilers and CPUs does not present issues; however, if the correctness of program logic depends on the order of memory access, memory out-of-order access can lead to logical errors.
During compilation, the compiler may alter the actual execution order of instructions with optimizations (e.g., GCC’s O2 or O3 can change the actual execution order).
At runtime, although the CPU may execute instructions out of order, on a single CPU, the hardware ensures that all memory access operations are executed in the order written by the programmer, making memory barriers unnecessary (not considering compiler optimizations). To execute instructions faster, the CPU adopts pipelined execution, and compilers optimize code to align instructions better with CPU pipelining and multiprocessor execution, which can lead to out-of-order situations.
2023 Past Reviews
The Development Direction of C/C++ (Highly Recommended!!)
Linux Kernel Source Code Analysis (Highly Recommended for Collection!)
From Rookie to Master, Creating Stunning World Applications with Qt
Full-Stack Storage Development: Building an Efficient Data Storage System
Breaking Performance Bottlenecks: Unlocking Endless Potential with DPDK
Embedded Audio and Video Technology: From Principles to Project Applications
C++ Game Development, Based on the Open Source Backend Framework of Warcraft