c语言环形队列

一位数组队列

#include 
#include 
#include 
#include 
#include 
#include 
#include 

#define BUFFER_SIZE 256 * 1024 * 1024 // 256MB
#define PACKET_SIZE 8192
#define HEADER_SIZE 16

typedef struct {
    char* buffer;
    int read_idx;
    int write_idx;
    int count;
    pthread_mutex_t lock;
    pthread_cond_t not_full;
    pthread_cond_t not_empty;
} CircularBuffer;

CircularBuffer* circular_buffer_init();
void circular_buffer_write(CircularBuffer* buffer, const char* data, int length);
void circular_buffer_read(CircularBuffer* buffer, char* data, int length);
void circular_buffer_destroy(CircularBuffer* buffer);
void* read_thread(void* arg);
void* send_thread(void* arg);

int main() {
    CircularBuffer* buffer = circular_buffer_init();

    pthread_t thread1, thread2;
    pthread_create(&thread1, NULL, read_thread, buffer);
    pthread_create(&thread2, NULL, send_thread, buffer);

    pthread_join(thread1, NULL);
    pthread_join(thread2, NULL);

    circular_buffer_destroy(buffer);

    return 0;
}

CircularBuffer* circular_buffer_init() {
    CircularBuffer* buffer = (CircularBuffer*)malloc(sizeof(CircularBuffer));
    buffer->buffer = (char*)malloc(BUFFER_SIZE);
    buffer->read_idx = 0;
    buffer->write_idx = 0;
    buffer->count = 0;
    pthread_mutex_init(&buffer->lock, NULL);
    pthread_cond_init(&buffer->not_full, NULL);
    pthread_cond_init(&buffer->not_empty, NULL);
    return buffer;
}

void circular_buffer_write(CircularBuffer* buffer, const char* data, int length) {
    pthread_mutex_lock(&buffer->lock);
    while (buffer->count == BUFFER_SIZE) {
        pthread_cond_wait(&buffer->not_full, &buffer->lock);
    }

    int remaining = PACKET_SIZE - HEADER_SIZE - length;
    if (remaining > 0) {
        memset(buffer->buffer + buffer->write_idx + HEADER_SIZE + length, 0, remaining);
    }

    memcpy(buffer->buffer + buffer->write_idx, data, length + HEADER_SIZE);
    buffer->write_idx = (buffer->write_idx + PACKET_SIZE) % BUFFER_SIZE;
    buffer->count++;

    pthread_cond_signal(&buffer->not_empty);
    pthread_mutex_unlock(&buffer->lock);
}

void circular_buffer_read(CircularBuffer* buffer, char* data, int length) {
    pthread_mutex_lock(&buffer->lock);
    while (buffer->count == 0) {
        pthread_cond_wait(&buffer->not_empty, &buffer->lock);
    }

    memcpy(data, buffer->buffer + buffer->read_idx, length);
    buffer->read_idx = (buffer->read_idx + PACKET_SIZE) % BUFFER_SIZE;
    buffer->count--;

    pthread_cond_signal(&buffer->not_full);
    pthread_mutex_unlock(&buffer->lock);
}

void circular_buffer_destroy(CircularBuffer* buffer) {
    pthread_mutex_destroy(&buffer->lock);
    pthread_cond_destroy(&buffer->not_full);
    pthread_cond_destroy(&buffer->not_empty);
    free(buffer->buffer);
    free(buffer);
}

void* read_thread(void* arg) {
    CircularBuffer* buffer = (CircularBuffer*)arg;

    FILE* file = fopen("input.txt", "r");
    if (file == NULL) {
        printf("Failed to open file.\n");
        pthread_exit(NULL);
    }

    char* data = (char*)malloc(PACKET_SIZE);
    memset(data, 0, PACKET_SIZE);
    char header[HEADER_SIZE] = "DATA HEADER";
    memcpy(data, header, HEADER_SIZE);

    ssize_t bytesRead;
    while ((bytesRead = fread(data + HEADER_SIZE, 1, PACKET_SIZE - HEADER_SIZE, file)) > 0) {
        circular_buffer_write(buffer, data, PACKET_SIZE);
    }

    free(data);
    fclose(file);
    pthread_exit(NULL);
}

void* send_thread(void* arg) {
    CircularBuffer* buffer = (CircularBuffer*)arg;

    int udp_socket = socket(AF_INET, SOCK_DGRAM, 0);
    if (udp_socket < 0) {
        printf("Failed to create UDP socket.\n");
        pthread_exit(NULL);
    }

    struct sockaddr_in server_address;
    server_address.sin_family = AF_INET;
    server_address.sin_port = htons(1234);
    server_address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);

    char* data = (char*)malloc(PACKET_SIZE);

    while (1) {
        circular_buffer_read(buffer, data, PACKET_SIZE);

        ssize_t bytesSent = sendto(udp_socket, data, PACKET_SIZE, 0, (struct sockaddr*)&server_address, sizeof(server_address));
        if (bytesSent < 0) {
            printf("Failed to send UDP packet.\n");
            pthread_exit(NULL);
        }
    }

    free(data);
    close(udp_socket);
    pthread_exit(NULL);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151

全部打印

#include 
#include 
#include 
#include 
#include 
#include 
#include 

#define BUFFER_SIZE 256 * 1024 * 1024 // 256MB
#define PACKET_SIZE 8192
#define HEADER_SIZE 16

typedef struct {
    char* buffer;
    int read_idx;
    int write_idx;
    int count;
    pthread_mutex_t lock;
    pthread_cond_t not_full;
    pthread_cond_t not_empty;
} CircularBuffer;

CircularBuffer* circular_buffer_init();
void circular_buffer_write(CircularBuffer* buffer, const char* data, int length);
void circular_buffer_read(CircularBuffer* buffer, char* data, int length);
void circular_buffer_destroy(CircularBuffer* buffer);
void print_hex(const char* data, int length);
void* read_thread(void* arg);
void* send_thread(void* arg);

int main() {
    CircularBuffer* buffer = circular_buffer_init();

    pthread_t thread1, thread2;
    pthread_create(&thread1, NULL, read_thread, buffer);
    pthread_create(&thread2, NULL, send_thread, buffer);

    pthread_join(thread1, NULL);
    pthread_join(thread2, NULL);

    circular_buffer_destroy(buffer);

    return 0;
}

CircularBuffer* circular_buffer_init() {
    CircularBuffer* buffer = (CircularBuffer*)malloc(sizeof(CircularBuffer));
    buffer->buffer = (char*)malloc(BUFFER_SIZE);
    buffer->read_idx = 0;
    buffer->write_idx = 0;
    buffer->count = 0;
    pthread_mutex_init(&buffer->lock, NULL);
    pthread_cond_init(&buffer->not_full, NULL);
    pthread_cond_init(&buffer->not_empty, NULL);
    return buffer;
}

void circular_buffer_write(CircularBuffer* buffer, const char* data, int length) {
    pthread_mutex_lock(&buffer->lock);
    while (buffer->count == BUFFER_SIZE) {
        pthread_cond_wait(&buffer->not_full, &buffer->lock);
    }

    int remaining = PACKET_SIZE - HEADER_SIZE - length;
    if (remaining > 0) {
        memset(buffer->buffer + buffer->write_idx + HEADER_SIZE + length, 0, remaining);
    }

    memcpy(buffer->buffer + buffer->write_idx, data, length + HEADER_SIZE);
    buffer->write_idx = (buffer->write_idx + PACKET_SIZE) % BUFFER_SIZE;
    buffer->count++;

    pthread_cond_signal(&buffer->not_empty);
    pthread_mutex_unlock(&buffer->lock);

    // Print the written data in hexadecimal format
    print_hex(data, length);
}

void circular_buffer_read(CircularBuffer* buffer, char* data, int length) {
    pthread_mutex_lock(&buffer->lock);
    while (buffer->count == 0) {
        pthread_cond_wait(&buffer->not_empty, &buffer->lock);
    }

    memcpy(data, buffer->buffer + buffer->read_idx, length);
    buffer->read_idx = (buffer->read_idx + PACKET_SIZE) % BUFFER_SIZE;
    buffer->count--;

    pthread_cond_signal(&buffer->not_full);
    pthread_mutex_unlock(&buffer->lock);

    // Print the read data in hexadecimal format
    print_hex(data, length);
}

void circular_buffer_destroy(CircularBuffer* buffer) {
    pthread_mutex_destroy(&buffer->lock);
    pthread_cond_destroy(&buffer->not_full);
    pthread_cond_destroy(&buffer->not_empty);
    free(buffer->buffer);
    free(buffer);
}

void print_hex(const char* data, int length) {
    printf("Data: ");
    for (int i = 0; i < length; i++) {
        printf("%02X ", (unsigned char)data[i]);
    }
    printf("\n");
}

void* read_thread(void* arg) {
    CircularBuffer* buffer = (CircularBuffer*)arg;

    FILE* file = fopen("./123.txt", "r");
    if (file == NULL) {
        printf("Failed to open file.\n");
        pthread_exit(NULL);
    }

    char* data = (char*)malloc(PACKET_SIZE);
    memset(data, 0, PACKET_SIZE);
    char header[HEADER_SIZE] = "DATA HEADER";
    memcpy(data, header, HEADER_SIZE);

    ssize_t bytesRead;
    while ((bytesRead = fread(data + HEADER_SIZE, 1, PACKET_SIZE - HEADER_SIZE, file)) > 0) {
        circular_buffer_write(buffer, data, PACKET_SIZE);
    }

    free(data);
    fclose(file);
    pthread_exit(NULL);
}

void* send_thread(void* arg) {
    CircularBuffer* buffer = (CircularBuffer*)arg;

    int udp_socket = socket(AF_INET, SOCK_DGRAM, 0);
    if (udp_socket < 0) {
        printf("Failed to create UDP socket.\n");
        pthread_exit(NULL);
    }

    struct sockaddr_in server_address;
    server_address.sin_family = AF_INET;
    server_address.sin_port = htons(1234);
    server_address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);

    char* data = (char*)malloc(PACKET_SIZE);

    while (1) {
        circular_buffer_read(buffer, data, PACKET_SIZE);

        ssize_t bytesSent = sendto(udp_socket, data, PACKET_SIZE, 0, (struct sockaddr*)&server_address, sizeof(server_address));
        if (bytesSent < 0) {
            printf("Failed to send UDP packet.\n");
            pthread_exit(NULL);
        }

        // Print the UDP sent data in hexadecimal format
        print_hex(data, PACKET_SIZE);
    }

    free(data);
    close(udp_socket);
    pthread_exit(NULL);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169

只打印发送

#include 
#include 
#include 
#include 
#include 
#include 
#include 

#define BUFFER_SIZE 256 * 1024 * 1024 // 256MB
#define PACKET_SIZE 8192
#define HEADER_SIZE 16

typedef struct {
    char* buffer;
    int read_idx;
    int write_idx;
    int count;
    pthread_mutex_t lock;
    pthread_cond_t not_full;
    pthread_cond_t not_empty;
} CircularBuffer;

CircularBuffer* circular_buffer_init();
void circular_buffer_write(CircularBuffer* buffer, const char* data, int length);
void circular_buffer_read(CircularBuffer* buffer, char* data, int length);
void circular_buffer_destroy(CircularBuffer* buffer);
void print_hex(const char* data, int length);
void* read_thread(void* arg);
void* send_thread(void* arg);

int main() {
    CircularBuffer* buffer = circular_buffer_init();

    pthread_t thread1, thread2;
    pthread_create(&thread1, NULL, read_thread, buffer);
    pthread_create(&thread2, NULL, send_thread, buffer);

    pthread_join(thread1, NULL);
    pthread_join(thread2, NULL);

    circular_buffer_destroy(buffer);

    return 0;
}

CircularBuffer* circular_buffer_init() {
    CircularBuffer* buffer = (CircularBuffer*)malloc(sizeof(CircularBuffer));
    buffer->buffer = (char*)malloc(BUFFER_SIZE);
    buffer->read_idx = 0;
    buffer->write_idx = 0;
    buffer->count = 0;
    pthread_mutex_init(&buffer->lock, NULL);
    pthread_cond_init(&buffer->not_full, NULL);
    pthread_cond_init(&buffer->not_empty, NULL);
    return buffer;
}

void circular_buffer_write(CircularBuffer* buffer, const char* data, int length) {
    pthread_mutex_lock(&buffer->lock);
    while (buffer->count == BUFFER_SIZE) {
        pthread_cond_wait(&buffer->not_full, &buffer->lock);
    }

    int remaining = PACKET_SIZE - HEADER_SIZE - length;
    if (remaining > 0) {
        memset(buffer->buffer + buffer->write_idx + HEADER_SIZE + length, 0, remaining);
    }

    memcpy(buffer->buffer + buffer->write_idx, data, length + HEADER_SIZE);
    buffer->write_idx = (buffer->write_idx + PACKET_SIZE) % BUFFER_SIZE;
    buffer->count++;

    pthread_cond_signal(&buffer->not_empty);
    pthread_mutex_unlock(&buffer->lock);
}

void circular_buffer_read(CircularBuffer* buffer, char* data, int length) {
    pthread_mutex_lock(&buffer->lock);
    while (buffer->count == 0) {
        pthread_cond_wait(&buffer->not_empty, &buffer->lock);
    }

    memcpy(data, buffer->buffer + buffer->read_idx, length);
    buffer->read_idx = (buffer->read_idx + PACKET_SIZE) % BUFFER_SIZE;
    buffer->count--;

    pthread_cond_signal(&buffer->not_full);
    pthread_mutex_unlock(&buffer->lock);
}

void circular_buffer_destroy(CircularBuffer* buffer) {
    pthread_mutex_destroy(&buffer->lock);
    pthread_cond_destroy(&buffer->not_full);
    pthread_cond_destroy(&buffer->not_empty);
    free(buffer->buffer);
    free(buffer);
}

void print_hex(const char* data, int length) {
    printf("Data: ");
    for (int i = 0; i < length; i++) {
        printf("%02X ", (unsigned char)data[i]);
    }
    printf("\n");
}

void* read_thread(void* arg) {
    CircularBuffer* buffer = (CircularBuffer*)arg;

    FILE* file = fopen("input.txt", "r");
    if (file == NULL) {
        printf("Failed to open file.\n");
        pthread_exit(NULL);
    }

    char* data = (char*)malloc(PACKET_SIZE);
    memset(data, 0, PACKET_SIZE);
    char header[HEADER_SIZE] = "DATA HEADER";
    memcpy(data, header, HEADER_SIZE);

    ssize_t bytesRead;
    while ((bytesRead = fread(data + HEADER_SIZE, 1, PACKET_SIZE - HEADER_SIZE, file)) > 0) {
        circular_buffer_write(buffer, data, HEADER_SIZE + bytesRead);
    }

    free(data);
    fclose(file);
    pthread_exit(NULL);
}

void* send_thread(void* arg) {
    CircularBuffer* buffer = (CircularBuffer*)arg;

    int udp_socket = socket(AF_INET, SOCK_DGRAM, 0);
    if (udp_socket < 0) {
        printf("Failed to create UDP socket.\n");
        pthread_exit(NULL);
    }

    struct sockaddr_in server_address;
    server_address.sin_family = AF_INET;
    server_address.sin_port = htons(1234);
    server_address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);

    char* data = (char*)malloc(PACKET_SIZE);

    while (1) {
        circular_buffer_read(buffer, data, PACKET_SIZE);

        int valid_data_length = PACKET_SIZE - HEADER_SIZE;
        int padded_data_length = valid_data_length;
        for (int i = valid_data_length - 1; i >= 0; i--) {
            if (data[HEADER_SIZE + i] != 0) {
                break;
            }
            padded_data_length--;
        }

        // Print the UDP sent data in hexadecimal format
        print_hex(data + HEADER_SIZE, padded_data_length);

        ssize_t bytesSent = sendto(udp_socket, data + HEADER_SIZE, padded_data_length, 0, (struct sockaddr*)&server_address, sizeof(server_address));
        if (bytesSent < 0) {
            printf("Failed to send UDP packet.\n");
            pthread_exit(NULL);
        }
    }

    free(data);
    close(udp_socket);
    pthread_exit(NULL);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172

inux c 建立一个256M大小的一位数组环形队列，建立两个线程一个用于读取指定文件内容存入环形队列，一个用UDP每次读取环形队列中8192字节发送到指定地址，不足8192仅发送剩余数据。读取数据时最大以1024字节为一包数据读取，读取时每包数据自定义16字节的数据包头组委有效数据存入环形队列，如果环形队列数据满了，等待UDP发送数据有空闲空间读取线程再继续读。环形队列读写、初始化环形缓冲区、销毁环形缓冲区分别写成独立函数，UDP将在fread读取读取的数据发送完成后结束所有线程。分别用16进制printf写入缓冲区的数据和udp发送的数据

#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 

#define BUFFER_SIZE 256 * 1024 * 1024  // 256M字节大小的环形队列
#define PACKET_SIZE 8192
#define MAX_PACKET_SIZE 1024
#define HEADER_SIZE 16

typedef struct {
    char buffer[BUFFER_SIZE];
    int read_index;
    int write_index;
    int count;
    pthread_mutex_t lock;
    pthread_cond_t full;
    pthread_cond_t empty;
} CircularBuffer;

typedef struct {
    CircularBuffer* buffer;
    FILE* file;
} ReaderArgs;

typedef struct {
    CircularBuffer* buffer;
    struct sockaddr_in* addr;
} UdpArgs;

void initCircularBuffer(CircularBuffer* buffer) {
    buffer->read_index = 0;
    buffer->write_index = 0;
    buffer->count = 0;
    pthread_mutex_init(&buffer->lock, NULL);
    pthread_cond_init(&buffer->full, NULL);
    pthread_cond_init(&buffer->empty, NULL);
}

void destroyCircularBuffer(CircularBuffer* buffer) {
    pthread_mutex_destroy(&buffer->lock);
    pthread_cond_destroy(&buffer->full);
    pthread_cond_destroy(&buffer->empty);
}

void writeToCircularBuffer(CircularBuffer* buffer, const char* data, int size) {
    pthread_mutex_lock(&buffer->lock);
    while (buffer->count == BUFFER_SIZE) {
        pthread_cond_wait(&buffer->empty, &buffer->lock);
    }

    int remainingBytes = size;
    while (remainingBytes > 0) {
        int writeSize = (remainingBytes > BUFFER_SIZE - buffer->count) ? BUFFER_SIZE - buffer->count : remainingBytes;
        memcpy(buffer->buffer + buffer->write_index, data + size - remainingBytes, writeSize);
        buffer->write_index = (buffer->write_index + writeSize) % BUFFER_SIZE;
        buffer->count += writeSize;
        remainingBytes -= writeSize;
    }

    pthread_cond_signal(&buffer->full);
    pthread_mutex_unlock(&buffer->lock);
}

void readFromCircularBuffer(CircularBuffer* buffer, char* dest, int size) {
    pthread_mutex_lock(&buffer->lock);
    while (buffer->count == 0) {
        pthread_cond_wait(&buffer->full, &buffer->lock);
    }

    int remainingBytes = size;
    while (remainingBytes > 0 && buffer->count > 0) {
        int readSize = (remainingBytes > buffer->count) ? buffer->count : remainingBytes;
        memcpy(dest + size - remainingBytes, buffer->buffer + buffer->read_index, readSize);
        buffer->read_index = (buffer->read_index + readSize) % BUFFER_SIZE;
        buffer->count -= readSize;
        remainingBytes -= readSize;
    }

    pthread_cond_signal(&buffer->empty);
    pthread_mutex_unlock(&buffer->lock);
}

void* readerThread(void* arg) {
    ReaderArgs* args = (ReaderArgs*)arg;
    CircularBuffer* buffer = args->buffer;
    FILE* file = args->file;
    char packet[MAX_PACKET_SIZE];
    int bytesRead;

    while ((bytesRead = fread(packet + HEADER_SIZE, sizeof(char), MAX_PACKET_SIZE - HEADER_SIZE, file)) > 0) {
        // Write header
        memcpy(packet, &bytesRead, sizeof(int));

        // Write to circular buffer
        writeToCircularBuffer(buffer, packet, bytesRead + HEADER_SIZE);
    }

    fclose(file);
    pthread_exit(NULL);
}

void* udpThread(void* arg) {
    UdpArgs* args = (UdpArgs*)arg;
    CircularBuffer* buffer = args->buffer;
    struct sockaddr_in* addr = args->addr;
    int sockfd = socket(AF_INET, SOCK_DGRAM, 0);
    char packet[PACKET_SIZE];
    int bytesSent, dataLength;

    while (1) {
        // Read from the circular buffer
        readFromCircularBuffer(buffer, packet, PACKET_SIZE);

        // Extract data length from header
        memcpy(&dataLength, packet, sizeof(int));

        // Send UDP packet
        bytesSent = sendto(sockfd, packet + HEADER_SIZE, dataLength, 0, (struct sockaddr*)addr, sizeof(struct sockaddr));
        if (bytesSent < 0) {
            perror("Failed to send UDP packet");
            break;
        }
    }

    close(sockfd);
    pthread_exit(NULL);
}

void printBufferAsHex(const char* buffer, int size) {
    int i;
    for (i = 0; i < size; i++) {
        printf("%02X ", (unsigned char)buffer[i]);
        if ((i + 1) % 16 == 0)
            printf("\n");
    }
    printf("\n");
}

int main() {
    CircularBuffer buffer;
    initCircularBuffer(&buffer);

    FILE* file = fopen("path/to/file", "rb");
    if (!file) {
        perror("Failed to open file");
        return EXIT_FAILURE;
    }

    pthread_t readerThreadID, udpThreadID;

    ReaderArgs readerArgs = { .buffer = &buffer, .file = file };
    if (pthread_create(&readerThreadID, NULL, readerThread, (void*)&readerArgs) != 0) {
        perror("Failed to create reader thread");
        return EXIT_FAILURE;
    }

    struct sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_port = htons(UDP_PORT);  // 替换为所需的UDP端口
    inet_aton("127.0.0.1", &addr.sin_addr);  // 替换为所需的目标IP地址

    UdpArgs udpArgs = { .buffer = &buffer, .addr = &addr };
    if (pthread_create(&udpThreadID, NULL, udpThread, (void*)&udpArgs) != 0) {
        perror("Failed to create UDP thread");
        return EXIT_FAILURE;
    }

    pthread_join(readerThreadID, NULL);
    pthread_cond_signal(&buffer.full);  // 通知UDP线程结束
    pthread_join(udpThreadID, NULL);

    destroyCircularBuffer(&buffer);

    // 打印缓冲区内容（以16进制形式）
    printf("缓冲区内容：\n");
    printBufferAsHex(buffer.buffer, BUFFER_SIZE);

    return EXIT_SUCCESS;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185

二维数字队列

#include 
#include 
#include 
#include 
#include 
#include 
#include 

#define BUFFER_SIZE 256 * 1024
#define PACKET_SIZE 8192
#define HEADER_SIZE 16

typedef struct {
    char **buffer;
    int *length;
    int size;
    int read_idx;
    int write_idx;
    int count;
    pthread_mutex_t lock;
    pthread_cond_t not_full;
    pthread_cond_t not_empty;
} CircularBuffer;

typedef struct {
    CircularBuffer *buffer;
    const char *filename;
} ReaderParams;

typedef struct {
    CircularBuffer *buffer;
    const char *dest_ip;
    int dest_port;
} SenderParams;

void circularBufferInit(CircularBuffer *buffer, int size) {
    buffer->buffer = (char**)malloc(sizeof(char*) * size);
    buffer->length = (int*)malloc(sizeof(int) * size);
    for (int i = 0; i < size; i++) {
        buffer->buffer[i] = (char*)malloc(PACKET_SIZE);
        buffer->length[i] = 0;
    }
    buffer->size = size;
    buffer->read_idx = 0;
    buffer->write_idx = 0;
    buffer->count = 0;
    pthread_mutex_init(&buffer->lock, NULL);
    pthread_cond_init(&buffer->not_full, NULL);
    pthread_cond_init(&buffer->not_empty, NULL);
}

void circularBufferWrite(CircularBuffer *buffer, const void *data, size_t size) {
    pthread_mutex_lock(&buffer->lock);

    while (buffer->count == buffer->size) {
        pthread_cond_wait(&buffer->not_full, &buffer->lock);
    }

    char *packet = buffer->buffer[buffer->write_idx];
    int padding_size = PACKET_SIZE - HEADER_SIZE - size;

    // 添加包头
    memcpy(packet, "Custom Header", HEADER_SIZE);

    // 添加数据
    memcpy(packet + HEADER_SIZE, data, size);

    // 补充0
    memset(packet + HEADER_SIZE + size, 0, padding_size);

    // 记录数据包长度
    buffer->length[buffer->write_idx] = size;

    buffer->write_idx = (buffer->write_idx + 1) % buffer->size;
    buffer->count++;

    pthread_cond_signal(&buffer->not_empty);
    pthread_mutex_unlock(&buffer->lock);
}

void circularBufferRead(CircularBuffer *buffer, void *data, size_t size) {
    pthread_mutex_lock(&buffer->lock);

    while (buffer->count == 0) {
        pthread_cond_wait(&buffer->not_empty, &buffer->lock);
    }

    char *packet = buffer->buffer[buffer->read_idx];
    int packet_size = buffer->length[buffer->read_idx];
    int padding_size = PACKET_SIZE - HEADER_SIZE - packet_size;

    // 读取数据
    memcpy(data, packet + HEADER_SIZE, packet_size);

    // 填补的0不需要被读取，直接跳过

    buffer->read_idx = (buffer->read_idx + 1) % buffer->size;
    buffer->count--;

    pthread_cond_signal(&buffer->not_full);
    pthread_mutex_unlock(&buffer->lock);
}

void circularBufferDestroy(CircularBuffer *buffer) {
    pthread_mutex_destroy(&buffer->lock);
    pthread_cond_destroy(&buffer->not_full);
    pthread_cond_destroy(&buffer->not_empty);

    for (int i = 0; i < buffer->size; i++) {
        free(buffer->buffer[i]);
    }
    free(buffer->buffer);
    free(buffer->length);

    buffer->size = 0;
    buffer->read_idx = 0;
    buffer->write_idx = 0;
    buffer->count = 0;
}

void *readerThread(void *params) {
    ReaderParams *readerParams = (ReaderParams*)params;

    FILE *file = fopen(readerParams->filename, "rb");
    if (file == NULL) {
        perror("Failed to open file");
        return NULL;
    }

    char data[PACKET_SIZE];
    size_t bytesRead = 0;

    while ((bytesRead = fread(data, 1, PACKET_SIZE - HEADER_SIZE, file)) > 0) {
        circularBufferWrite(readerParams->buffer, data, bytesRead);
        usleep(1000);
    }

    fclose(file);

    return NULL;
}

int createUdpSocket(const char *dest_ip, int dest_port) {
    int sock = socket(AF_INET, SOCK_DGRAM, 0);

    struct sockaddr_in server_addr;
    memset(&server_addr, 0, sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = inet_addr(dest_ip);
    server_addr.sin_port = htons(dest_port);

    if (connect(sock, (struct sockaddr *)&server_addr, sizeof(server_addr)) < 0) {
        perror("Failed to connect to UDP server");
        close(sock);
        return -1;
    }

    return sock;
}

void *senderThread(void *params) {
    SenderParams *senderParams = (SenderParams*)params;

    int sock = createUdpSocket(senderParams->dest_ip, senderParams->dest_port);

    char data[PACKET_SIZE];
    while (1) {
        circularBufferRead(senderParams->buffer, data, PACKET_SIZE - HEADER_SIZE);
        send(sock, data, PACKET_SIZE - HEADER_SIZE, 0);
        usleep(1000);
    }

    close(sock);

    return NULL;
}

int main() {
    CircularBuffer buffer;
    circularBufferInit(&buffer, BUFFER_SIZE);

    ReaderParams readerParams;
    readerParams.buffer = &buffer;
    readerParams.filename = "input.txt";

    SenderParams senderParams;
    senderParams.buffer = &buffer;
    senderParams.dest_ip = "127.0.0.1";  // 目标地址
    senderParams.dest_port = 12345;  // 目标端口

    pthread_t readerThreadId;
    pthread_t senderThreadId;

    pthread_create(&readerThreadId, NULL, readerThread, &readerParams);
    pthread_create(&senderThreadId, NULL, senderThread, &senderParams);

    pthread_join(readerThreadId, NULL);
    pthread_join(senderThreadId, NULL);

    circularBufferDestroy(&buffer);

    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203

备注

#include 
#include 
#include 
#include 
#include 
#include 
#include 

#define BUFFER_SIZE 256 * 1024
#define PACKET_SIZE 8192
#define HEADER_SIZE 16

typedef struct {
    char **buffer;
    int *length;
    int size;
    int read_idx;
    int write_idx;
    int count;
    pthread_mutex_t lock;
    pthread_cond_t not_full;
    pthread_cond_t not_empty;
} CircularBuffer;

typedef struct {
    CircularBuffer *buffer;
    const char *filename;
} ReaderParams;

typedef struct {
    CircularBuffer *buffer;
    const char *dest_ip;
    int dest_port;
} SenderParams;

void circularBufferInit(CircularBuffer *buffer, int size) {
    buffer->buffer = (char**)malloc(sizeof(char*) * size);
    buffer->length = (int*)malloc(sizeof(int) * size);
    for (int i = 0; i < size; i++) {
        buffer->buffer[i] = (char*)malloc(PACKET_SIZE);
        buffer->length[i] = 0;
    }
    buffer->size = size;
    buffer->read_idx = 0;
    buffer->write_idx = 0;
    buffer->count = 0;
    pthread_mutex_init(&buffer->lock, NULL);
    pthread_cond_init(&buffer->not_full, NULL);
    pthread_cond_init(&buffer->not_empty, NULL);
}

void circularBufferWrite(CircularBuffer *writbuffer, const void *data, size_t size) {
    pthread_mutex_lock(&writbuffer->lock);

    while (writbuffer->count == writbuffer->size) {
        pthread_cond_wait(&writbuffer->not_full, &writbuffer->lock);
    }

    char *packet = writbuffer->buffer[writbuffer->write_idx];
    int padding_size = PACKET_SIZE - HEADER_SIZE - size;

    // 添加包头
    memcpy(packet, "Custom Header", HEADER_SIZE);

    // 添加数据
    memcpy(packet + HEADER_SIZE, data, size);

    // 补充0
    memset(packet + HEADER_SIZE + size, 0, padding_size);

    // 记录数据包长度
    writbuffer->length[writbuffer->write_idx] = size;

    writbuffer->write_idx = (writbuffer->write_idx + 1) % writbuffer->size;
    writbuffer->count++;

    pthread_cond_signal(&writbuffer->not_empty);
    pthread_mutex_unlock(&writbuffer->lock);
}

void circularBufferWrite(CircularBuffer *buffer, const void *data, size_t size) {
    pthread_mutex_lock(&buffer->lock);

    while (buffer->count == buffer->size) {
        pthread_cond_wait(&buffer->not_full, &buffer->lock);
    }
    /*
	用于判断环形队列是否已满的逻辑，并在队列已满时进行阻塞等待的操作。
buffer->count表示当前环形队列中已经存储的数据包数量。
buffer->size表示环形队列的最大容量，即可以存储的最大数据包数量。
在以上代码中，通过条件判断 buffer->count == buffer->size 来判断环形队列是否已满。如果队列已满，即存储的数据包数量等于最大容量，进入 while 循环。
	*/
    char *packet = buffer->buffer[buffer->write_idx];/*将环形缓冲区中写入位置（buffer->write_idx）索引所指向的字符指针（buffer->buffer[buffer->write_idx]）赋值给 packet。

这行代码假设 buffer->buffer 是一个 char** 类型的指针，指向存储数据的字符指针数组。通过访问 buffer->buffer[buffer->write_idx]，可以获得当前待写入的位置在环形缓冲区中指向的字符数组的指针。*/
    int padding_size = PACKET_SIZE - HEADER_SIZE - size;/*运算得出填充大小,大于零表示本包数据不足*/

    // 添加包头
    memcpy(packet, "Custom Header", HEADER_SIZE);

    // 添加数据
    memcpy(packet + HEADER_SIZE, data, size);

    // 补充0
    memset(packet + HEADER_SIZE + size, 0, padding_size);

    // 记录数据包长度
    buffer->length[buffer->write_idx] = size;

    buffer->write_idx = (buffer->write_idx + 1) % buffer->size;
    buffer->count++;

    pthread_cond_signal(&buffer->not_empty);//发送信号通知其他等待中的线程，表示缓冲区不再为空，这可能是用于唤醒一个等待从缓冲区中读取数据的消费者线程。
    pthread_mutex_unlock(&buffer->lock);//解锁缓冲区的互斥锁，表示写入操作已完成，
}

void circularBufferRead(CircularBuffer *buffer, void *data, size_t size) {
    pthread_mutex_lock(&buffer->lock);

    while (buffer->count == 0) {
        pthread_cond_wait(&buffer->not_empty, &buffer->lock);
    }

    char *packet = buffer->buffer[buffer->read_idx];
    int packet_size = buffer->length[buffer->read_idx];
    int padding_size = PACKET_SIZE - HEADER_SIZE - packet_size;

    // 读取数据
    memcpy(data, packet + HEADER_SIZE, packet_size);

    // 填补的0不需要被读取，直接跳过

    buffer->read_idx = (buffer->read_idx + 1) % buffer->size;
    buffer->count--;

    pthread_cond_signal(&buffer->not_full);
    pthread_mutex_unlock(&buffer->lock);
}

void circularBufferDestroy(CircularBuffer *buffer) {
    pthread_mutex_destroy(&buffer->lock);
    pthread_cond_destroy(&buffer->not_full);
    pthread_cond_destroy(&buffer->not_empty);

    for (int i = 0; i < buffer->size; i++) {
        free(buffer->buffer[i]);
    }
    free(buffer->buffer);
    free(buffer->length);

    buffer->size = 0;
    buffer->read_idx = 0;
    buffer->write_idx = 0;
    buffer->count = 0;
}

void *readerThread(void *params) {
    ReaderParams *readerParams = (ReaderParams*)params;

    FILE *file = fopen(readerParams->filename, "rb");
    if (file == NULL) {
        perror("Failed to open file");
        return NULL;
    }

    char data[PACKET_SIZE];
    size_t bytesRead = 0;

    while ((bytesRead = fread(data, 1, PACKET_SIZE - HEADER_SIZE, file)) > 0) {
        circularBufferWrite(readerParams->buffer, data, bytesRead);
        usleep(1000);
    }

    fclose(file);

    return NULL;
}

int createUdpSocket(const char *dest_ip, int dest_port) {
    int sock = socket(AF_INET, SOCK_DGRAM, 0);

    struct sockaddr_in server_addr;
    memset(&server_addr, 0, sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = inet_addr(dest_ip);
    server_addr.sin_port = htons(dest_port);

    if (connect(sock, (struct sockaddr *)&server_addr, sizeof(server_addr)) < 0) {
        perror("Failed to connect to UDP server");
        close(sock);
        return -1;
    }

    return sock;
}

void *senderThread(void *params) {
    SenderParams *senderParams = (SenderParams*)params;

    int sock = createUdpSocket(senderParams->dest_ip, senderParams->dest_port);

    char data[PACKET_SIZE];
    while (1) {
        circularBufferRead(senderParams->buffer, data, PACKET_SIZE - HEADER_SIZE);
        send(sock, data, PACKET_SIZE - HEADER_SIZE, 0);
        usleep(1000);
    }

    close(sock);

    return NULL;
}

int main() {
    CircularBuffer buffer;
    circularBufferInit(&buffer, BUFFER_SIZE);

    ReaderParams readerParams;
    readerParams.buffer = &buffer;
    readerParams.filename = "input.txt";

    SenderParams senderParams;
    senderParams.buffer = &buffer;
    senderParams.dest_ip = "127.0.0.1";  // 目标地址
    senderParams.dest_port = 12345;  // 目标端口

    pthread_t readerThreadId;
    pthread_t senderThreadId;

    pthread_create(&readerThreadId, NULL, readerThread, &readerParams);
    pthread_create(&senderThreadId, NULL, senderThread, &senderParams);

    pthread_join(readerThreadId, NULL);
    pthread_join(senderThreadId, NULL);

    circularBufferDestroy(&buffer);

    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239

相关阅读:
贪心算法之过河问题
 浅谈Oauth2.0授权
 EM@解三角形@正弦定理@余弦定理
 Java Stream流对List集合进行分页
 【windows】网络设置了代理，怎么关闭
 mongodb常用命令（3）
【Greenhills】MULTI IDE加密狗无法识别
 智能合约漏洞案例，NeverFall 漏洞复现
 java-php-net-python-绥化市北林区房屋拆迁管理信息管理系统计算机毕业设计程序
 Docker仓库harbor
原文地址：https://blog.csdn.net/weixin_38849487/article/details/133697564