C中多线程的同步

Posted 2023-02-14

【中文标题】C中多线程的同步

【英文标题】：Synchronization of Multithreading in C

【发布时间】：2022-01-04 04:33:33

【问题描述】：

我是 C 中多线程的新手。我正在尝试编写使用多线程的代码。代码如下：

#include <stdio.h> 
#include <stdlib.h>
#include <unistd.h> 
#include <limits.h>
#include <windows.h> 
#include <time.h>
#include <math.h>

int finish = 0; 
int mess_size=15;

double exponGenerator(double myLamda)

    double pv = 0.0;
    while (1)
    
        pv = (double)rand() / (double)RAND_MAX;
        if (pv != 1)
            break;
    
    pv = (-1.0 / myLamda) * log(1 - pv);
    return pv;


struct packet
    int val;
    time_t start_time;
    double service_time;
;
struct queue
    struct packet arr_pac[10000];
    int start;
    int end;
    int size;
;

double lambda = 5;
double servicerate = 5;
struct queue q1 ;
struct queue q2; 

struct queue initialiseQueue()
   struct queue q ;
   q.start = 0;
   q.end = 0;
   q.size=0;
   return q;


struct process1
int method;
double lambda;
struct queue Q1;
struct queue Q2;
;

struct process2
struct queue q;
double u;
;

struct process1 queueenv(int method)

    struct process1 temp ;
    temp.method = method;
    temp.lambda = lambda;
    temp.Q1 = q1;
    temp.Q2 = q2;
return temp;


struct process2 serverenv(double u, struct queue q)

    struct process2 temp;
    temp.u = u;
    temp.q = q;
return temp;


int enque(struct queue q)
if (q.size < 10) 
 
  
  struct packet temp ;
  temp.start_time = time(NULL);

  temp.val = 1;
  q.arr_pac[q.end] = temp;

  q.end = q.end +1;

  q.size = q.end - q.start;
//printf("  %d",q.arr_pac[0].end);
  return q.size  ;

else
return -1;



struct packet deque(struct queue q) 
struct packet temp ;
printf("  %d ",q.end);
if(q.size >0)
  printf("entered");
  temp=q.arr_pac[q.start];
  temp.service_time = difftime(time(NULL),temp.start_time);
  q.start=q.start +1;
  q.size = q.end - q.start;
  return temp;

else
printf("entered 2");
temp.service_time=0 ;
return temp;



int randomSelection()
if(rand()%2 ==0)
return enque(q1);

else
return enque(q2);



int minQueue()
 if(q1.size > q2.size)
 return enque(q2);

else
return enque(q1);



void queueprocess(struct process1 params)
    double blockCounter = 0.0;
    double blockPro = 0.0;
    int queLenCounter = 0;
    int averageQueueLen = 0;
    int i = 0;
 
  if (params.method == 0) 
    
printf(" this is the %d =0",params.method);
        for (i = 0; i < mess_size ; i++)
        
            
            double interval = exponGenerator(params.lambda);
            sleep(interval);
            int res = randomSelection();
            
            if (res == -1) 
             
                blockCounter++;
                queLenCounter = queLenCounter +10;
            
        else

           queLenCounter = queLenCounter + res;
                   //printf(" %d ",queLenCounter);
        
        
    
    else if (params.method == 1)
     
printf(" this is the %d =1  ",params.method);
        for (i = 0; i < mess_size ; i++)
        
            double interval = exponGenerator(params.lambda);
            sleep(interval);
            int res = minQueue();
            printf("  %d mn",q1.end);
            if (res == -1) 
            
                blockCounter++;
                queLenCounter = queLenCounter +10;
            
        else
           queLenCounter = queLenCounter +res;
        
        
    
    blockPro = blockCounter/mess_size;
    averageQueueLen = queLenCounter/mess_size;
    printf("Blocking Probability is : %f",&blockPro);
    printf("Average Que Length is : %d",averageQueueLen);
    finish = 1;
    return;


void serverprocess(struct process2 serverparams) 

 struct packet processed_arr[10000];
 int i=0,j;
    while(1)
     
        if (finish == 1 && serverparams.q.size == 0) 
        
            break;
        
        else
        
            double interval = exponGenerator(serverparams.u);
 
            sleep(interval);
            
            struct packet k = deque(serverparams.q);
             
        if(!k.service_time)
 
            processed_arr[i]=k;

            i++;
        
        
    
float sourjanCounter=0;
float sourjan;
for(j=0;j<i;j++)
printf(" %f",processed_arr[j].service_time);
sourjanCounter = sourjanCounter+ processed_arr[j].service_time;

sourjan = sourjanCounter/(i-1);
printf("Average Sourjan Time is : %f ", &sourjan);


DWORD WINAPI threadone()
    printf(" thread one ");
    struct process1 queueparams = queueenv(1);
   
    queueprocess(queueparams);


DWORD WINAPI threadtwo()  
    struct process2 server1params = serverenv(servicerate, q1);
    serverprocess(server1params);


DWORD WINAPI threadthree()
    struct process2 server2params = serverenv(servicerate, q2);
    serverprocess(server2params);


int main(int argc, char * argv[]) 

    HANDLE hThrds[3];
    int i = 0;

    /*if (argc != 3)
    
        printf("Incorrect parameters!\n");
        return 0;
    
    double lambda = atof(argv[1]);
    double servicerate = atof(argv[2]);*/

    for (i = 0; i < 1; i++)
    
        DWORD thread1,thread2,thread3;
        hThrds[2]=CreateThread(NULL,0,  threadone, NULL,0,&thread1);
        hThrds[0]=CreateThread(NULL,0, threadtwo, NULL,0,&thread2);
        hThrds[1]=CreateThread(NULL,0, threadthree, NULL,0,&thread3); 
        WaitForMultipleObjects(3, hThrds,TRUE, INFINITE);
    

在我的代码中，我有一个名为enque 的函数，它是从一个名为queueprocess 的函数中调用的。 q1 是一个结构体，是全局定义的，可以被多个函数同时访问。我试图在函数enque 内更改q1 的值，当我在函数enque 内打印它的值时，它的值是正确的，但是如果我尝试在函数queueprocess 内打印它的值，它的值是0 .因此，我的代码没有前进。我读过关于mutex 的信息，但我不知道什么是相当于win32 的互斥锁。我不知道为什么我的程序也应该使用互斥锁，因为我只是从另一个函数访问q1，但我并不想改变它的值。

谁能解释一下为什么我会遇到上述问题以及可能的解决方案是什么？

【问题讨论】：

"我只是从另一个函数访问 q1，但我并不想改变它的值"。没关系。如果您在 any 线程中进行更改，则所有线程都需要同步。否则，当写入线程更新了它需要的部分但不是全部结构字段时，读取线程可能会读取结构。

@kaylum 在这种情况下足以使其成为原子。同步的意义远不止于此。

@0___________ 是的，你是对的。我的术语有点太松散了。

@Ram 您需要学习一些 IPC 基础知识：docs.microsoft.com/en-us/windows/win32/sync/… - 此链接用于 Microsoft 实施。但基本上所有系统都使用非常相似的机制。

是与否，@0__________。确实，从多个线程访问原子对象不需要同步，但是访问原子结构的成员具有未定义的行为（C17 6.5.2.3/5）。正如脚注 99 所阐明的，要获取原子结构的成员，必须通过结构分配制作非原子副本，并访问副本的成员。这不仅仅是使结构原子化。

【参考方案1】：

来自微软文档：

您可以使用互斥对象来保护共享资源 多个线程或进程同时访问。每个线程必须 等待互斥锁的所有权，然后才能执行 访问共享资源。例如，如果多个线程共享 访问数据库，线程可以使用互斥对象来允许 一次只能有一个线程写入数据库。

以下示例使用 CreateMutex 函数创建互斥锁 对象和 CreateThread 函数来创建工作线程。

当这个进程的一个线程写入数据库时​​，它首先 使用 WaitForSingleObject 请求互斥体的所有权 功能。如果线程获得互斥锁的所有权，它会写入 数据库，然后使用 释放互斥函数。

本示例使用结构化异常处理来确保 线程正确释放互斥对象。 __finally 代码块 无论 __try 块如何终止（除非 __try 块包括对 TerminateThread 函数的调用）。这可以防止 互斥对象不会被无意中遗弃。

如果一个互斥锁被放弃，拥有该互斥锁的线程没有 在终止之前正确释放它。在这种情况下，状态 共享资源不确定，并继续使用互斥锁 可以掩盖潜在的严重错误。一些应用程序可能 尝试将资源恢复到一致状态；这个例子 只是返回一个错误并停止使用互斥锁。更多 信息，请参阅互斥对象。

#include <windows.h>
#include <stdio.h>

#define THREADCOUNT 2

HANDLE ghMutex; 

DWORD WINAPI WriteToDatabase( LPVOID );

int main( void )

    HANDLE aThread[THREADCOUNT];
    DWORD ThreadID;
    int i;

    // Create a mutex with no initial owner

    ghMutex = CreateMutex( 
        NULL,              // default security attributes
        FALSE,             // initially not owned
        NULL);             // unnamed mutex

    if (ghMutex == NULL) 
    
        printf("CreateMutex error: %d\n", GetLastError());
        return 1;
    

    // Create worker threads

    for( i=0; i < THREADCOUNT; i++ )
    
        aThread[i] = CreateThread( 
                     NULL,       // default security attributes
                     0,          // default stack size
                     (LPTHREAD_START_ROUTINE) WriteToDatabase, 
                     NULL,       // no thread function arguments
                     0,          // default creation flags
                     &ThreadID); // receive thread identifier

        if( aThread[i] == NULL )
        
            printf("CreateThread error: %d\n", GetLastError());
            return 1;
        
    

    // Wait for all threads to terminate

    WaitForMultipleObjects(THREADCOUNT, aThread, TRUE, INFINITE);

    // Close thread and mutex handles

    for( i=0; i < THREADCOUNT; i++ )
        CloseHandle(aThread[i]);

    CloseHandle(ghMutex);

    return 0;


DWORD WINAPI WriteToDatabase( LPVOID lpParam )
 
    // lpParam not used in this example
    UNREFERENCED_PARAMETER(lpParam);

    DWORD dwCount=0, dwWaitResult; 

    // Request ownership of mutex.

    while( dwCount < 20 )
     
        dwWaitResult = WaitForSingleObject( 
            ghMutex,    // handle to mutex
            INFINITE);  // no time-out interval
 
        switch (dwWaitResult) 
        
            // The thread got ownership of the mutex
            case WAIT_OBJECT_0: 
                __try  
                    // TODO: Write to the database
                    printf("Thread %d writing to database...\n", 
                            GetCurrentThreadId());
                    dwCount++;
                 

                __finally  
                    // Release ownership of the mutex object
                    if (! ReleaseMutex(ghMutex)) 
                     
                        // Handle error.
                     
                 
                break; 

            // The thread got ownership of an abandoned mutex
            // The database is in an indeterminate state
            case WAIT_ABANDONED: 
                return FALSE; 
        
    
    return TRUE;