套接字发送有时仅在无限循环 (C) 中工作

Question

我正在为 Zynq FPGA 的 DAQ 系统 设置 原型。我通过以太网从服务器接收数据，使用 DMA 将其写入 FIFO，并使用两个不同的线程将其写入 viceversa。但是，发送操作挂起，经过一些迭代后，它正确执行，然后再次挂起，同时recv操作成功并填充 FIFO。

有时，只有当某些 printf 存在或不存在时，线程才会按预期工作，所以我暂时打印到 stderr。

很抱歉代码可耻，但是我现在尝试替换几乎每一行以发现问题。

void *sender(void *params) {
arguments *args = params;
if (args->head == NULL) {
fprintf(stderr, "[-] Head pointer not validn");
exit(0);
}
if (args->tail == NULL) {
fprintf(stderr, "[-] Tail pointer not validn");
exit(0);
}
if (args->virtual_address == NULL) {
fprintf(stderr, "[-] AXI DMA register pointer not validn");
exit(0);
}
if (args->virtual_buffer == NULL) {
fprintf(stderr, "[-] Send buffer pointer not validn");
exit(0);
}
unsigned long int units_sent = 0;
for (;;) {
pthread_mutex_lock(args->lock);
fprintf(stderr, "[*] Send   Head: %d    Tail: %dn", *(args->head), *(args->tail));
if (*(args->head) != *(args->tail)) {
pthread_mutex_unlock(args->lock);
int remaining = args->buffsize;
int sent = 0;
pthread_mutex_lock(args->lock);
int src = FIFO_ADDR + (*(args->tail)) * args->buffsize;
pthread_mutex_unlock(args->lock);
if (args->verbose > 2) {
fprintf(stderr, "[*] Sender: DMA is transferring data from 0x%x to 0x%xn", src, SEND_ADDR);
}
unsigned int length = args->buffsize;
unsigned int verb = args->verbose > 2 ? 1 : 0;
pthread_mutex_lock(args->lock);
while (remaining > 0) {
length = remaining < MAX_TRANSF ? remaining : remaining % MAX_TRANSF;
DMATransfer(args->virtual_address, src + sent, SEND_ADDR, length, verb);
remaining -= args->buffsize;
sent += remaining;
}
elements--;
units_sent++;
if (args->verbose > 2) {
fprintf(stderr, "[*] %f elements in FIFO: %lun", ((double)(clock() - start_time)) / CLOCKS_PER_SEC, elements);
fprintf(stderr, "[*] %f DMA tranfer to buffer: %dn", ((double)(clock() - start_time)) / CLOCKS_PER_SEC, units_sent);
}
pthread_mutex_unlock(args->lock);
remaining = args->buffsize;
sent = 0;
int result = 0;
pthread_mutex_lock(args->lock);
while (remaining > 0) {
result = send(args->sockfd, args->virtual_buffer + sent, remaining, 0);
if (result > 0) {
remaining -= result;
sent += remaining;
} else if (result < 0) {
fprintf(stderr, "[-] Error retrieving configuration from the servern");
exit(0);
}
}
//memset(args->virtual_buffer, 0, args->buffsize);
if (args->verbose > 2) {
fprintf(stderr, "[*] %f Unit sent: %dn", ((double)(clock() - start_time)) / CLOCKS_PER_SEC, units_sent);
}
if (args->verbose > 0) {
fprintf(stderr, "[*] Packet retrieved");
}
if (args->verbose > 1) {
fprintf(stderr, " content: ");
memdump(args->virtual_buffer, args->buffsize);
}
if (args->verbose > 0) {
fprintf(stderr, "n");
}
*(args->tail) = (*(args->tail) + 1) % args->fifosize;
pthread_mutex_unlock(args->lock);
if (args->verbose > 2) {
fprintf(stderr, "[*] %f CPU Usage: %dn", ((double)(clock() - start_time)) / CLOCKS_PER_SEC, GetCPULoad());
}
}
pthread_mutex_unlock(args->lock);
}
}

int main(int argc, char *argv[]) {
if (argc < 3 || argc > 5) {
fprintf(stderr, "nUsage: DAQTest [IP address] [fifo size]nExample: DAQTest 192.168.1.81 64nn");
fprintf(stderr, "Optional flags:    -v  Verbose (print operations)n");
fprintf(stderr, "       -vv Very verbose (also print data content)n");
fprintf(stderr, "       -vvv    Extremely verbose (also print DMA info)nn");
exit(0);
}
if (isValidIpAddress(argv[1]) == 1) {
fprintf(stderr, "[-] Invalid ip addressn");
exit(0);
}
int fifosize = atoi(argv[2]);
if (fifosize < 0 || fifosize > 8192) {
fprintf(stderr, "[-] Invalid fifo sizen");
exit(0);
}
char verbose = 0;
if (argc == 4) {
if (strcmp(argv[3], "-v") == 0) {
verbose = 1;
} else if (strcmp(argv[3], "-vv") == 0) {
verbose = 2;
} else if (strcmp(argv[3], "-vvv") == 0) {
verbose = 3;
} else {
fprintf(stderr, "[-] Unwanted parametern");
exit(0);
}
}
struct sockaddr_in servaddr, cli;
int sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (-1 == sockfd) {
fprintf(stderr, "[-] Socket creation failedn");
exit(0);
}
bzero(&servaddr, sizeof(servaddr));
servaddr.sin_family = AF_INET;
servaddr.sin_addr.s_addr = inet_addr(argv[1]);
servaddr.sin_port = htons(PORT);
/*
struct timeval tv;
tv.tv_sec = TIMEOUT;
tv.tv_usec = 0;
setsockopt(sockfd, SOL_SOCKET, SO_RCVTIMEO, (const char*)&tv, sizeof tv);
*/
if (connect(sockfd, (SA * ) & servaddr, sizeof(servaddr)) != 0) {
fprintf(stderr, "[-] Connection failedn");
exit(0);
}
fprintf(stderr, "[+] Connected to the servern");
int buffsize = 0;
char* recv_buffer = (char*)&buffsize;
int remaining = sizeof(int);
int received = 0;
int result = 0;
while (remaining > 0) {
result = recv(sockfd, recv_buffer + received, remaining, 0);
if (result > 0) {
remaining -= result;
received += result;
} else if (result == 0) {
fprintf(stderr, "[-] Remote side closed his end of the connection before all data was receivedn");
exit(0);
} else if (result < 0) {
fprintf(stderr, "[-] Error retrieving configuration from the servern");
exit(0);
}
}
//fprintf(stderr, "[*] Page size: %ldn", sysconf(_SC_PAGE_SIZE));
int dh = open("/dev/mem", O_RDWR | O_SYNC);
unsigned int *virtual_address = mmap(NULL, 65535, PROT_READ | PROT_WRITE, MAP_SHARED, dh, VIRT_ADDR);
unsigned int *virtual_sendbuff = mmap(NULL, buffsize, PROT_READ | PROT_WRITE, MAP_SHARED, dh, SEND_ADDR);
unsigned int *virtual_recvbuff = mmap(NULL, buffsize, PROT_READ | PROT_WRITE, MAP_SHARED, dh, RECV_ADDR);
unsigned int *virtual_fifo = mmap(NULL, (fifosize + 1) * buffsize, PROT_READ | PROT_WRITE, MAP_SHARED, dh, FIFO_ADDR);
if (virtual_address == MAP_FAILED) {
fprintf(stderr, "[-] AXI DMA registers mmap failedn");
}
if (virtual_sendbuff == MAP_FAILED) {
fprintf(stderr, "[-] Send buffer mmap failedn");
}
if (virtual_sendbuff == MAP_FAILED) {
fprintf(stderr, "[-] Send buffer mmap failedn");
}
if (virtual_recvbuff == MAP_FAILED) {
fprintf(stderr, "[-] Receiver buffer mmap failedn");
}
if (virtual_fifo == MAP_FAILED) {
fprintf(stderr, "[-] Fifo mmap failedn");
}
memset(virtual_address, 0, buffsize);
memset(virtual_sendbuff, 0, buffsize);
memset(virtual_recvbuff, 0, buffsize);
memset(virtual_fifo, 0, buffsize);
int head = 0, tail = 0;
pthread_t sendth, recvth;
pthread_mutex_t lock;
pthread_mutex_init(&lock, NULL);
arguments send_args;
send_args.virtual_address = virtual_address;
send_args.virtual_buffer = virtual_sendbuff;
send_args.head = &head;
send_args.tail = &tail;
send_args.buffsize = buffsize;
send_args.fifosize = fifosize;
send_args.sockfd = sockfd;
send_args.lock = &lock;
send_args.verbose = verbose;
arguments recv_args;
recv_args.virtual_address = virtual_address;
recv_args.virtual_buffer = virtual_recvbuff;
recv_args.head = &head;
recv_args.tail = &tail;
recv_args.buffsize = buffsize;
recv_args.fifosize = fifosize;
recv_args.sockfd = sockfd;
recv_args.lock = &lock;
recv_args.verbose = verbose;
start_time = clock();
if (pthread_create(&sendth, NULL, sender, &send_args)) {
fprintf(stderr, "[-] Error creating sender threadn");
exit(0);
}
if (pthread_create(&recvth, NULL, receiver, &recv_args)) {
fprintf(stderr, "[-] Error creating receiver threadn");
exit(0);
}
if (pthread_join(sendth, NULL)) {
fprintf(stderr, "[-] Error joining sender threadn");
exit(0);
}
if (pthread_join(recvth, NULL)) {
fprintf(stderr, "[-] Error joining receiver threadn");
exit(0);
}
pthread_mutex_destroy(&lock);
close(sockfd);
fprintf(stderr, "[+] Exitn");
return 0;
}

实际输出显示来自recv线程的打印，而发送线程打印

[*] 发送头： %d 尾部： %d

几乎总是，有时它必须做什么。

Answer 1

在sender函数中，你有这样的代码(缩短了很多(

if (*(args->head) != *(args->tail)) {
// ...
pthread_mutex_unlock(args->lock);
if (args->verbose > 2) {
fprintf(stderr, "[*] %f CPU Usage: %dn", ((double)(clock() - start_time)) / CLOCKS_PER_SEC, GetCPULoad());
}
}
pthread_mutex_unlock(args->lock);

这将导致pthread_mutex_unlock(args->lock)将在同一锁上被调用两次，而不会在两者之间锁定。

例如，从这个 POSIX 参考中可以看出，pthread_mutex_unlock尝试解锁不属于您的线程的互斥锁(例如解锁已解锁的互斥锁，或解锁另一个线程持有的互斥锁(会导致未定义的行为。

需要以某种方式避免这种情况，例如，通过将最后一个解锁调用放在else子句中：

if (*(args->head) != *(args->tail)) {
// ...
pthread_mutex_unlock(args->lock);
if (args->verbose > 2) {
fprintf(stderr, "[*] %f CPU Usage: %dn", ((double)(clock() - start_time)) / CLOCKS_PER_SEC, GetCPULoad());
}
}
else {
pthread_mutex_unlock(args->lock);
}