【发布时间】：2021-12-19 09:38:41
【问题描述】：
我正在尝试构建一个使用线程打开并行 TCP 套接字的系统。
我的线程是使用消息队列 IPC 触发的，因此每次数据包到达消息队列时，一个线程“唤醒”，打开与远程服务器的 TCP 连接并发送数据包。
我的问题是，在 Wireshark 中，我可以看到使用线程而不是一个连接发送文件所需的时间更短，但吞吐量没有改变。
我的问题是：
- 如何验证我的线程并行工作？
- 如何改进此代码？ 3.如何用一个线程打开多个socket？
我正在使用虚拟机来运行多线程客户端。 我使用的 IDE 是 Clion ，语言是 C。 我的代码：
#include<stdio.h>
#include<stdlib.h>
#include<sys/socket.h>
#include<string.h>
#include <arpa/inet.h>
#include <unistd.h> // for close
#include<pthread.h>
#include <math.h>
#include<malloc.h>
#include<signal.h>
#include<stdbool.h>
#include<sys/types.h>
#include<linux/if_packet.h>
#include<netinet/in.h>
#include<netinet/if_ether.h> // for ethernet header
#include<netinet/ip.h> // for ip header
#include<netinet/udp.h> // for udp header
#include<netinet/tcp.h>
#include <byteswap.h>
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <mqueue.h>
#include <assert.h>
#include <time.h>
#define QUEUE_NAME "/ServerDan_Queue"
#define QUEUE_PERM 0660
#define MAX_MESSAGES 10 //Max size = 10
#define MAX_MSG_SIZE 4105 //Max size = 8192B
#define MSG_BUFFER_SIZE MAX_MSG_SIZE+10
#define BSIZE 1024
#define Nbytes 4096
#define ElorServer_addr "192.168.1.54"
///params:
pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
struct sockaddr_in server;
struct stat obj;
int sock;
int k, size, status;
int i = 0;
typedef struct frag
{
int packet_number;
int seq;
uint8_t data[4096];
bool lastfrag;
} fragma;
void * middlemanThread(void *arg)
{
///========================================///
///**** Waiting for message queue trigger!:///
///=======================================///
long id = (long)arg;
id+=1;
mqd_t qd; //queue descriptor
//open the queue for reading//
qd= mq_open(QUEUE_NAME,O_RDONLY);
assert(qd != -1);
struct mq_attr attr;
assert(mq_getattr(qd,&attr) != -1);
uint8_t *income_buf = calloc(attr.mq_msgsize,1);
uint8_t *cast_buf = calloc(attr.mq_msgsize,1);
assert(income_buf);
fragma frag;
struct timespec timeout;
clock_gettime(CLOCK_REALTIME,&timeout);
timeout.tv_sec+=50;
//bool closesoc =false;
printf("Waiting for messages ..... \n\n");
while(1){
///========================================///
///**** Open message queue fo receive:///
///=======================================///
if((mq_timedreceive(qd,income_buf,attr.mq_msgsize,0,&timeout))<0){
printf("Failed to receive message for 50 sec \n");
//closesoc =true;
pthread_exit(NULL);
}
else{
cast_buf = income_buf;
printf("Received successfully , your msg :\n");
frag.packet_number = *cast_buf;
cast_buf = (cast_buf + sizeof(int));
frag.seq = *cast_buf;
cast_buf = (cast_buf + sizeof(int));
memccpy(frag.data,((fragma*)cast_buf)->data,0,Nbytes);
cast_buf = cast_buf + Nbytes;
frag.lastfrag = *cast_buf;
uint8_t * data = frag.data;
}
pthread_mutex_lock(&lock);
///========================================///
///**** Connecting to Server and send Frament:///
///=======================================///
int size = sizeof(( fragma *)income_buf)->packet_number + sizeof(( fragma *)income_buf)->seq + sizeof(( fragma *)income_buf)->data + sizeof(( fragma *)income_buf)->lastfrag;
printf("In thread\n");
int clientSocket;
struct sockaddr_in serverAddr;
socklen_t addr_size;
// Create the socket.
clientSocket = socket(PF_INET, SOCK_STREAM, 0);
//Configure settings of the server address
// Address family is Internet
serverAddr.sin_family = AF_INET;
//Set port number, using htons function
serverAddr.sin_port = htons(8081);
//Set IP address to localhost
serverAddr.sin_addr.s_addr = inet_addr("192.168.14.149");
memset(serverAddr.sin_zero, '\0', sizeof serverAddr.sin_zero);
//Connect the socket to the server using the address
addr_size = sizeof serverAddr;
connect(clientSocket, (struct sockaddr *) &serverAddr, addr_size);
if(send(clientSocket , income_buf , size,0) < 0)
{
printf("Send failed\n");
}
printf("Trhead Id : %ld \n" , id);
printf("Packet number : %d \n Seq = %d \n lasfrag = %d\n\n",frag.packet_number,frag.seq,(int)frag.lastfrag);
pthread_mutex_unlock(&lock);
//if(closesoc)
close(clientSocket);
usleep(20000);
}
}
int main(){
int i = 0;
pthread_t tid[5];
while(i< 5)
{
if( pthread_create(&tid[i], NULL, middlemanThread, (void*)i) != 0 )
printf("Failed to create thread\n");
i++;
}
sleep(2);
i = 0;
while(i< 5)
{
pthread_join(tid[i++],NULL);
printf("Thread ID : %d:\n",i);
}
return 0;
}
【问题讨论】：
-
我不知道从哪里开始。在“线程化”时，代码既没有利用线程化（即，与多进程实现相比，它不会增加吞吐量/减少延迟），甚至没有共享 any 线程之间的状态（我看不到它使用互斥锁来做什么——“关键路径”代码都没有使用全局状态，所以为什么要锁定？）。其次，每个请求创建/使用/销毁套接字通常是一个坏主意（为了速度）。相反，可以管理一个连接池，并让现有连接至少停留一段时间。
-
更重要的是：TCP是一个stream协议。没有数据包。
-
所以你建议尝试多处理？
-
thoughput 没有增加是什么意思？如果时间更小，文件相同，不是自动变大吗？
-
'usleep(20000);'......为什么？
标签： c linux multithreading sockets tcp