首页 / 技术问答 / C++问答列表 / 问答详情

C++ 为什么在传输 Big Data 时共享内存段的运行时间比管道更长

我正在编写一个关于操作系统的实验室.您需要编写两个使用管道和共享内存段传输数据的程序.有必要比较传输时间并得出共享内存段在 Big Data 中工作得更快的结论.但事实恰恰相反.我的代码中有什么问题？我做错了什么？ 我计算工作时间为阅读结束减go 写作开始

UPD: I listened to the comments and changed the code, add new result

管道代号:如下:

#define _GNU_SOURCE

#include <sys/time.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <stdio.h>
#include <wait.h>
#include <math.h>
#include <stdlib.h>
#include <fcntl.h>

#define COUNT 6

int main()
{
    // fd[0]=READ, fd[1]=WRITE
    int fd[2], nread, pid, status, base = 1;
    for (int i = 0; i < COUNT; i++)
    {
        int size = 8*base; // bytes
        base *= 16;
        pipe(fd);
        fcntl(fd[1], F_SETPIPE_SZ, size);
        fcntl(fd[0], F_SETPIPE_SZ, size);
        struct timespec t1, t2;
        pid = fork();
        
        if (pid == 0)
        {
            // READ
            int tmp_size = 0;           
            close(fd[1]);               
            char *buf = (char *)malloc((size) * sizeof(char));
            while((nread = read(fd[0], buf, size)) != 0);

            free(buf);
            exit(1);    
        }
        else
        {
            // WRITE
            close(fd[0]);   
            clock_gettime(CLOCK_REALTIME, &t1);
        
            char *buf = (char *)malloc((size) * sizeof(char));
            for (int j = 0; j < size; j++)
            {
                *(buf+j) = '1';
            }
            write(fd[1], buf, size);
            
            close(fd[1]);
            free(buf);
        }
        wait(&status);
        clock_gettime(CLOCK_REALTIME, &t2);
        FILE *file = i == 0? fopen("program1_result", "w"): fopen("program1_result", "a+");
        char cur_time[150];
        sprintf(cur_time, "%ld nsec\n", 1000000000*(t2.tv_sec - t1.tv_sec) + (t2.tv_nsec-t1.tv_nsec));
        if (file != NULL) fputs(cur_time, file);
        fclose(file);

    }
}

共享内存代码:

#include <sys/ipc.h>
#include <sys/types.h>
#include <sys/shm.h>
#include <sys/time.h>
#include <unistd.h>
#include <stdio.h>
#include <sys/sem.h>
#include <math.h>
#include <string.h>
#include <stdlib.h>
#include <time.h>
#include <wait.h>

#define COUNT 6

int main()
{
    int pid, status, shmid, sem, base = 1;
    char *shmPtr;
    key_t shm_keys[COUNT], sem_keys[COUNT];
    for (int i = 0; i < COUNT; i++)
    {
        shm_keys[i] = ftok(".", (char)i);
        sem_keys[i] = ftok(".", 'a' + i);
    }
    
    for (int i = 0; i < COUNT; i++)
    {
        int size = 8*base;      
        base *= 16;
        if ((shmid = shmget(shm_keys[i], size + 1, IPC_CREAT | 0666)) == -1) exit(1);
        shmPtr = shmat(shmid, 0, 0);
        
        sem = semget(sem_keys[i], 1, IPC_CREAT | 0666);
        //struct timeval s, e;
        struct timespec t1, t2;
        pid = fork();

        if (pid == 0)
        {
            while(semctl(sem, 0, GETVAL) != 1);
            char *str = (char *)malloc(size * sizeof(char));
            strcpy(str, shmPtr);
            
            free(str);
            exit(1);
        }
        else
        {           
            clock_gettime(CLOCK_REALTIME, &t1);
            char *str = (char *)malloc((size+1) * sizeof(char));
            for (int j = 0; j < size; j++)
            {
                *(str+j) = '1';
            }
            *(str + size) = '\0';
            strcpy(shmPtr, str);
            
            semctl(sem, 0, SETVAL, 1);
            free(str);          
        }   
        wait(&status);
        clock_gettime(CLOCK_REALTIME, &t2);
        FILE *file = i == 0? fopen("program2_result", "w"): fopen("program2_result", "a+");
        char cur_time[150];
        sprintf(cur_time, "%ld nsec\n", 1000000000*(t2.tv_sec - t1.tv_sec) + (t2.tv_nsec-t1.tv_nsec));
        if (file != NULL) fputs(cur_time, file);
        fclose(file);
        
        shmdt(shmPtr);
        shmctl(shmid, IPC_RMID, NULL);
        semctl(sem, 0, IPC_RMID);
    }
    
    return 0;
}

结果管道:

  • 510134 ns(8字节)
  • 751695 ns(128字节)
  • 317859 ns(2048字节)
  • 372219 ns(32768字节)
  • 2158510 ns(524288字节)
  • 19722241 ns(8388608字节)

结果共享记忆:

  • 463168 ns(8字节)
  • 369917 ns(128字节)
  • 307799 ns(2048字节)
  • 361162 ns(32768字节)
  • 2019749 ns(524288字节)
  • 24943196 ns(8388608字节)
New result:
|bytes    | pipe (nsec) | shm (nsec)|
|---------|-------------|-----------|
| 8       |   517516    |  489703   |
| 128     |   205485    |  440699   |
| 2048    |   374170    |  1162227  |
| 32768   |   584830    |  548490   |
| 524288  |   3162177   |  4010005  |
| 8388608 |   50293808  |  67142116 |

请帮助理解我做错了什么.

推荐答案

我的代码中有什么问题？我做错了什么？

共享内存(particularly用于大量数据)几乎总是更快.

这两个程序的测量值相当于not，这会导致错误的结果.

更正后，结果clearly显示共享内存更快(快2倍-3倍).而且，就接收器的延迟而言，通过共享内存，接收器可以更快地开始处理数据.

考虑接收器可以处理数据的逻辑步骤until:

  1. 收件箱在缓冲区中创建数据
  2. 对于管道，发送者执行write次操作，这是数据的复制.接收器执行read，这是数据的[另一次]复制.
  3. 对于共享内存，需要将no个副本复制到内核中或从内核中取出.就内核操作而言，共享内存是"零副本".

换句话说，管道版本执行shm版本没有的两个[不必要]复制操作.

此外，还要考虑"延迟".从发送者填满缓冲区到接收者[能够]处理数据的时间.管道版本still必须将write个数据传输到管道.但是，对于shm版本，接收器可以在收到"通知"消息后立即启动(下面的更新中有更多信息). 代码中的问题.

  1. 我们想要测量从一个过程到另一个过程的just和transfer时间.
  2. 在共享内存版本中，strcpy是无关的.它违背了使用共享内存的目的.
  3. 我们需要在shared缓冲区上操作directly.
  4. strcpy正在做管道版本没有做的额外工作.也就是说，管道版本是only对数据进行read - no处理.
  5. 为了减轻时间切片和系统加载的影响，我们应该多次执行相同的测试并花费minimum次时间.
  6. 你的时机有点不对.
  7. 因为我们正在执行fork，所以我们可以使用IPC_PRIVATE而不是ftok(更干净一点).

这是一个重构的版本.它带有错误/修复程序注释.

  1. 我改进了基准代码.
  2. 我添加了内部基准测试来显示启动和延迟.
  3. 我已将这两个程序合并到一个.c文件中.
  4. 您的fopen个电话可以被清理.
#define _GNU_SOURCE

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <time.h>

#include <sys/time.h>
#include <wait.h>
#include <math.h>
#include <fcntl.h>

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/sem.h>

#define sysfault(_fmt...) \
    do { \
        fprintf(stderr,_fmt); \
        exit(1); \
    } while (0)

#define Tmark       (tscgetf() - tscbeg)

double tsczero;

double
_tscgetf(void)
{
    struct timespec ts;
    double sec;

    clock_gettime(CLOCK_MONOTONIC,&ts);

    sec = ts.tv_nsec;
    sec /= 1e9;
    sec += ts.tv_sec;

    sec -= tsczero;

    return sec;
}

#define tscgetf() \
    ({ \
        __asm__ __volatile__("nop\n" ::: "memory"); \
        _tscgetf(); \
    })

int base;
size_t size;
double t1;
double t2;
int shmid, sem;
char *shmPtr;
double tscbeg;
double tscsend[3];
double tscbest[3];
double tscrcv[3];

struct hdr {
    const char *sym;
    const char *reason;
};

#define HDRMAX  (sizeof(hdrs) / sizeof(hdrs[0]))

#define _HDRPUT(_fmt,_data...) \
    do { \
        totlen = fprintf(file,_fmt,_data); \
        for (;  totlen < lim;  ++totlen) \
            fputc(' ',file); \
    } while (0)
#define HDRPUT(_fmt,_data...) \
    do { \
        _HDRPUT(_fmt,_data); \
        ++ihdr; \
    } while (0)

void
result(int i,const char *pre,double tbest)
{
    int totlen;
    int lim = 12;

    char name[100];
    sprintf(name,"%s_result",pre);

    if (i == 0)
        unlink(name);

    FILE *file = fopen(name,"a");

    static struct hdr hdrs[] = {
        { "ELAPSED", "total elapsed time" },
        { "Tsend", "sender unlocks receiver" },
        { "Twait", "receiver becomes ready for input" },
        { "Trcv", "receiver released by sender" },
        { "Latency", "Trcv - Twait (time receiver is delayed)" },
        { "size", "buffer size" },
    };

    int ihdr;
    if (i == 0) {
        fprintf(file,"Legend:\n");
        for (ihdr = 0;  ihdr < HDRMAX;  ++ihdr) {
            _HDRPUT("%s",hdrs[ihdr].sym);
            fprintf(file,"-- %s\n",hdrs[ihdr].reason);
        }

        fprintf(file,"\n");
        for (ihdr = 0;  ihdr < HDRMAX;  ++ihdr)
            _HDRPUT("%s",hdrs[ihdr].sym);
        fprintf(file,"\n");
    }

    ihdr = 0;
    HDRPUT("%.9f",tbest);
    HDRPUT("%.9f",tscsend[0]);
    HDRPUT("%.9f",tscrcv[0]);

    HDRPUT("%.9f",tscrcv[1]);
    HDRPUT("%.9f",tscrcv[1] - tscrcv[0]);
#if 0
    HDRPUT("%d/%d",base,size);
#else
    fprintf(file,"%zu",size);
#endif
    fprintf(file,"\n");

    fclose(file);
}

#define COUNT 16

void
dobest(const char *pre,void (*fnc)(int i,int dtflg))
{
    // fd[0]=READ, fd[1]=WRITE

    base = 1;
    for (int i = 0; i < COUNT; i++) {
        double tbest = 1e9;

        size = base;
        if ((shmid = shmget(IPC_PRIVATE, size + 1, IPC_CREAT | 0666)) == -1)
            sysfault("dobest: shmget failure -- %s\n",strerror(errno));
        shmPtr = shmat(shmid, 0, 0);
        sem = semget(IPC_PRIVATE, 1, IPC_CREAT | 0666);

        // do multiple trials to minimize the effects of timeslicing and system
        // loading
        for (int iter = 3;  iter > 0;  --iter) {
            tscbeg = tscgetf();
            fnc(i,iter == 1);
            double tdif = t2 - t1;
            if (tdif < tbest) {
                tbest = tdif;
                memcpy(tscrcv,shmPtr,sizeof(tscrcv));
                memcpy(tscbest,tscsend,sizeof(tscbest));
            }
        }

        // record the best result
        result(i,pre,tbest);

        shmdt(shmPtr);
        shmctl(shmid, IPC_RMID, NULL);
        semctl(sem, 0, IPC_RMID);

        base *= 4;
    }
}

void
pipeone(int i,int dtflg)
{
    int fd[2];
    int nread, pid, status;

    pipe(fd);
    fcntl(fd[1], F_SETPIPE_SZ, size);
    fcntl(fd[0], F_SETPIPE_SZ, size);

    char *buf = malloc(size);
    pid = fork();

    if (pid == 0) {
        // READ

        close(fd[1]);

        tscrcv[0] = Tmark;
        int mark = 1;
        while ((nread = read(fd[0], buf, size)) != 0) {
            if (mark)
                tscrcv[1] = Tmark;
            mark = 0;
        }

        memcpy(shmPtr,tscrcv,sizeof(tscrcv));

        exit(0);
    }

    // WRITE
    close(fd[0]);

    t1 = tscgetf();

    for (size_t j = 0; j < size; j++)
        buf[j] = '1';

    tscsend[0] = Tmark;
    write(fd[1], buf, size);
    close(fd[1]);

    wait(&status);

    t2 = tscgetf();

    free(buf);
}

key_t shm_keys[COUNT], sem_keys[COUNT];

void
shmone(int i,int dtflg)
{
    int pid, status;

    pid = fork();

    if (pid == 0) {
        tscrcv[0] = Tmark;
        while (semctl(sem, 0, GETVAL) != 1);
        tscrcv[1] = Tmark;

// NOTE/BUG: not valid to copy from the shared buffer -- the whole point is to
// operate on the shared buffer directly!!!
#if 0
        char *str = (char *) malloc(size);
        strcpy(str, shmPtr);
        free(str);
        exit(1);

// NOTE/FIX: do _not_ copy out the buffer -- for testing purposes, we only want
// to measure the delivery time
#else
        size_t len = 0;
#if 0
        len = strlen(shmPtr);
#endif
        memcpy(shmPtr,tscrcv,sizeof(tscrcv));
        exit(len % 0x7f);
#endif
    }

    t1 = tscgetf();
#if 0
    char *str = (char *) malloc((size + 1) * sizeof(char));
#else
    char *str = shmPtr;
#endif

    for (size_t j = 0; j < size; j++)
        str[j] = '1';
    str[size] = '\0';

#if 0
    strcpy(shmPtr, str);
#endif

    tscsend[0] = Tmark;
    semctl(sem, 0, SETVAL, 1);
#if 0
    free(str);
#endif

    wait(&status);

    t2 = tscgetf();
}

void
pipeall(void)
{
    // fd[0]=READ, fd[1]=WRITE

    dobest("pipe",pipeone);
}

void
shmall(void)
{
    // fd[0]=READ, fd[1]=WRITE

    for (int i = 0; i < COUNT; i++) {
        shm_keys[i] = ftok(".", (char) i);
        sem_keys[i] = ftok(".", 'a' + i);
    }

    dobest("shm",shmone);
}

int
main(int argc,char **argv)
{

    --argc;
    ++argv;

    tsczero = tscgetf();

    if (argc < 1)
        sysfault("main: specify type: pipe or shm\n");

    do {
        char *cp = *argv;

        if (strcmp(cp,"pipe") == 0) {
            pipeall();
            break;
        }

        if (strcmp(cp,"shm") == 0) {
            shmall();
            break;
        }

        sysfault("main: unknown type -- '%s'\n",cp);
    } while (0);

    return 0;
}

在上面的代码中，我使用了cpp个条件来表示旧代码和新代码:

#if 0
// old code
#else
// new code
#endif

#if 1
// new code
#endif

注意:这可以通过运行文件到unifdef -k

这是./program pipe的输出:

Legend:
ELAPSED     -- total elapsed time
Tsend       -- sender unlocks receiver
Twait       -- receiver becomes ready for input
Trcv        -- receiver released by sender
Latency     -- Trcv - Twait (time receiver is delayed)
size        -- buffer size

ELAPSED     Tsend       Twait       Trcv        Latency     size
0.000212793 0.000097429 0.000125559 0.000162328 0.000036769 1
0.000167312 0.000104965 0.000125442 0.000147394 0.000021952 4
0.000177647 0.000096102 0.000093215 0.000128481 0.000035266 16
0.000106641 0.000080761 0.000077737 0.000096395 0.000018658 64
0.000204464 0.000163753 0.000112410 0.000148289 0.000035879 256
0.000147881 0.000060119 0.000144019 0.000166102 0.000022083 1024
0.000108481 0.000072081 0.000080240 0.000099838 0.000019598 4096
0.000156202 0.000130236 0.000145147 0.000173962 0.000028815 16384
0.000233014 0.000144926 0.000076950 0.000188236 0.000111286 65536
0.000605337 0.000425224 0.000079762 0.000537632 0.000457870 262144
0.002080624 0.001375300 0.000150789 0.001993695 0.001842906 1048576
0.009131487 0.005655973 0.000243503 0.008914248 0.008670745 4194304
0.036384772 0.021980127 0.000213079 0.035238290 0.035025211 16777216
0.142985735 0.072771127 0.000424021 0.137922121 0.137498100 67108864
0.572007428 0.292893437 0.000403933 0.553239097 0.552835164 268435456
2.276978512 1.176618786 0.000574158 2.220540762 2.219966604 1073741824

这是./program shm的输出:

Legend:
ELAPSED     -- total elapsed time
Tsend       -- sender unlocks receiver
Twait       -- receiver becomes ready for input
Trcv        -- receiver released by sender
Latency     -- Trcv - Twait (time receiver is delayed)
size        -- buffer size

ELAPSED     Tsend       Twait       Trcv        Latency     size
0.000181048 0.000081319 0.000126865 0.000152599 0.000025734 1
0.000277881 0.000079515 0.000165932 0.000177756 0.000011824 4
0.000191934 0.000081238 0.000126623 0.000134039 0.000007416 16
0.000201862 0.000080135 0.000124149 0.000135244 0.000011095 64
0.000183210 0.000080799 0.000107105 0.000113932 0.000006827 256
0.000146235 0.000085935 0.000112573 0.000119606 0.000007033 1024
0.000144176 0.000085737 0.000000000 0.000000000 0.000000000 4096
0.000235418 0.000100743 0.000136712 0.000148129 0.000011417 16384
0.000172821 0.000160697 0.000123424 0.000134592 0.000011168 65536
0.000201004 0.000241614 0.000122755 0.000129961 0.000007206 262144
0.000748101 0.000788737 0.000091531 0.000098554 0.000007023 1048576
0.002963725 0.003005338 0.000150246 0.000157523 0.000007277 4194304
0.011919764 0.011992419 0.000192575 0.000201370 0.000008795 16777216
0.047795073 0.047874544 0.000200046 0.000209144 0.000009098 67108864
0.190522428 0.190601263 0.000178723 0.000187669 0.000008946 268435456
0.757310083 0.757391226 0.000177408 0.000186831 0.000009423 1073741824

以下是清理后的来源:

#define _GNU_SOURCE

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <time.h>

#include <sys/time.h>
#include <wait.h>
#include <math.h>
#include <fcntl.h>

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/sem.h>

#define sysfault(_fmt...) \
    do { \
        fprintf(stderr,_fmt); \
        exit(1); \
    } while (0)

#define Tmark       (tscgetf() - tscbeg)

double tsczero;

double
_tscgetf(void)
{
    struct timespec ts;
    double sec;

    clock_gettime(CLOCK_MONOTONIC,&ts);

    sec = ts.tv_nsec;
    sec /= 1e9;
    sec += ts.tv_sec;

    sec -= tsczero;

    return sec;
}

#define tscgetf() \
    ({ \
        __asm__ __volatile__("nop\n" ::: "memory"); \
        _tscgetf(); \
    })

int base;
size_t size;
double t1;
double t2;
int shmid, sem;
char *shmPtr;
double tscbeg;
double tscsend[3];
double tscbest[3];
double tscrcv[3];

struct hdr {
    const char *sym;
    const char *reason;
};

#define HDRMAX  (sizeof(hdrs) / sizeof(hdrs[0]))

#define _HDRPUT(_fmt,_data...) \
    do { \
        totlen = fprintf(file,_fmt,_data); \
        for (;  totlen < lim;  ++totlen) \
            fputc(' ',file); \
    } while (0)
#define HDRPUT(_fmt,_data...) \
    do { \
        _HDRPUT(_fmt,_data); \
        ++ihdr; \
    } while (0)

void
result(int i,const char *pre,double tbest)
{
    int totlen;
    int lim = 12;

    char name[100];
    sprintf(name,"%s_result",pre);

    if (i == 0)
        unlink(name);

    FILE *file = fopen(name,"a");

    static struct hdr hdrs[] = {
        { "ELAPSED", "total elapsed time" },
        { "Tsend", "sender unlocks receiver" },
        { "Twait", "receiver becomes ready for input" },
        { "Trcv", "receiver released by sender" },
        { "Latency", "Trcv - Twait (time receiver is delayed)" },
        { "size", "buffer size" },
    };

    int ihdr;
    if (i == 0) {
        fprintf(file,"Legend:\n");
        for (ihdr = 0;  ihdr < HDRMAX;  ++ihdr) {
            _HDRPUT("%s",hdrs[ihdr].sym);
            fprintf(file,"-- %s\n",hdrs[ihdr].reason);
        }

        fprintf(file,"\n");
        for (ihdr = 0;  ihdr < HDRMAX;  ++ihdr)
            _HDRPUT("%s",hdrs[ihdr].sym);
        fprintf(file,"\n");
    }

    ihdr = 0;
    HDRPUT("%.9f",tbest);
    HDRPUT("%.9f",tscsend[0]);
    HDRPUT("%.9f",tscrcv[0]);

    HDRPUT("%.9f",tscrcv[1]);
    HDRPUT("%.9f",tscrcv[1] - tscrcv[0]);
    fprintf(file,"%zu",size);
    fprintf(file,"\n");

    fclose(file);
}

#define COUNT 16

void
dobest(const char *pre,void (*fnc)(int i,int dtflg))
{
    // fd[0]=READ, fd[1]=WRITE

    base = 1;
    for (int i = 0; i < COUNT; i++) {
        double tbest = 1e9;

        size = base;
        if ((shmid = shmget(IPC_PRIVATE, size + 1, IPC_CREAT | 0666)) == -1)
            sysfault("dobest: shmget failure -- %s\n",strerror(errno));
        shmPtr = shmat(shmid, 0, 0);
        sem = semget(IPC_PRIVATE, 1, IPC_CREAT | 0666);

        // do multiple trials to minimize the effects of timeslicing and system
        // loading
        for (int iter = 3;  iter > 0;  --iter) {
            tscbeg = tscgetf();
            fnc(i,iter == 1);
            double tdif = t2 - t1;
            if (tdif < tbest) {
                tbest = tdif;
                memcpy(tscrcv,shmPtr,sizeof(tscrcv));
                memcpy(tscbest,tscsend,sizeof(tscbest));
            }
        }

        // record the best result
        result(i,pre,tbest);

        shmdt(shmPtr);
        shmctl(shmid, IPC_RMID, NULL);
        semctl(sem, 0, IPC_RMID);

        base *= 4;
    }
}

void
pipeone(int i,int dtflg)
{
    int fd[2];
    int nread, pid, status;

    pipe(fd);
    fcntl(fd[1], F_SETPIPE_SZ, size);
    fcntl(fd[0], F_SETPIPE_SZ, size);

    char *buf = malloc(size);
    pid = fork();

    if (pid == 0) {
        // READ

        close(fd[1]);

        tscrcv[0] = Tmark;
        int mark = 1;
        while ((nread = read(fd[0], buf, size)) != 0) {
            if (mark)
                tscrcv[1] = Tmark;
            mark = 0;
        }

        memcpy(shmPtr,tscrcv,sizeof(tscrcv));

        exit(0);
    }

    // WRITE
    close(fd[0]);

    t1 = tscgetf();

    for (size_t j = 0; j < size; j++)
        buf[j] = '1';

    tscsend[0] = Tmark;
    write(fd[1], buf, size);
    close(fd[1]);

    wait(&status);

    t2 = tscgetf();

    free(buf);
}

key_t shm_keys[COUNT], sem_keys[COUNT];

void
shmone(int i,int dtflg)
{
    int pid, status;

    pid = fork();

    if (pid == 0) {
        tscrcv[0] = Tmark;
        while (semctl(sem, 0, GETVAL) != 1);
        tscrcv[1] = Tmark;

// NOTE/BUG: not valid to copy from the shared buffer -- the whole point is to
// operate on the shared buffer directly!!!
        size_t len = 0;
        memcpy(shmPtr,tscrcv,sizeof(tscrcv));
        exit(len % 0x7f);
    }

    t1 = tscgetf();
    char *str = shmPtr;

    for (size_t j = 0; j < size; j++)
        str[j] = '1';
    str[size] = '\0';

    tscsend[0] = Tmark;
    semctl(sem, 0, SETVAL, 1);

    wait(&status);

    t2 = tscgetf();
}

void
pipeall(void)
{
    // fd[0]=READ, fd[1]=WRITE

    dobest("pipe",pipeone);
}

void
shmall(void)
{
    // fd[0]=READ, fd[1]=WRITE

    for (int i = 0; i < COUNT; i++) {
        shm_keys[i] = ftok(".", (char) i);
        sem_keys[i] = ftok(".", 'a' + i);
    }

    dobest("shm",shmone);
}

int
main(int argc,char **argv)
{

    --argc;
    ++argv;

    tsczero = tscgetf();

    if (argc < 1)
        sysfault("main: specify type: pipe or shm\n");

    do {
        char *cp = *argv;

        if (strcmp(cp,"pipe") == 0) {
            pipeall();
            break;
        }

        if (strcmp(cp,"shm") == 0) {
            shmall();
            break;
        }

        sysfault("main: unknown type -- '%s'\n",cp);
    } while (0);

    return 0;
}

UPDATE:

我对测试程序做了一些相当大的清理工作.由于一些反对意见，我添加了一些选项:

-C<n> -- test count
-I<n> -- iteration count
-m<0/1> -- notification (1=msgsnd/msgrcv, 0=sem*)
-S<0/1> -- 1=pipe set maximum size (F_SETPIPE_SZ)
-x<0/1> -- 1=child scan/process data
-v<0/1> -- 1=copy results to stdout

事实证明，使用sem*作为通知(正如这里发布的那样)与msgsnd/msgrcv(这是我传统上使用的)一样可靠.使用sem*可能会为某些测试产生负延迟值.这表明该机制[出于某种原因]并不稳健.所以，我把它换成了msgsnd/msgrcv

我添加了-x选项，用于强制接收器在缓冲区上"工作".这产生了轻微的影响.正如我提到的，我们想要测量transfer时间(和not测量transfer时间+ processing时间)，所以我会省略-x，但它是为了完整性.

此外，如果没有给出"测试"参数，为了方便，程序将顺序(在子进程中)运行"管道"测试，然后运行"shm"测试.

有关选项的更多详细信息请参阅代码:

#define _GNU_SOURCE

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <time.h>

#include <sys/time.h>
#include <wait.h>
#include <math.h>
#include <fcntl.h>

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/sem.h>
#include <sys/msg.h>

int opt_C = 16;
int opt_I = 3;
int opt_m = 1;
int opt_S = 1;
int opt_x;
int opt_v;

#define sysfault(_fmt...) \
    do { \
        fprintf(stderr,_fmt); \
        exit(1); \
    } while (0)

#define Tmark       (tscgetf() - tscbeg)

double tsczero;

double
_tscgetf(void)
{
    struct timespec ts;
    double sec;

    clock_gettime(CLOCK_MONOTONIC,&ts);

    sec = ts.tv_nsec;
    sec /= 1e9;
    sec += ts.tv_sec;

    sec -= tsczero;

    return sec;
}

#define barrier \
    __asm__ __volatile__("nop\n" ::: "memory"); \

#define tscgetf() \
    ({ \
        barrier; \
        _tscgetf(); \
    })

int base;
size_t size;
double t1;
double t2;
int shmid;
int sem;
char *shmPtr;
char *fake;
int forkflg;
double tscbeg;

double tscsend[3];
double tscbest[3];
double tscrcv[3];

struct msg {
    long msg_type;
    int msg_payload;
};
#define MSG_TYPE        17

enum {
    Tnotify
};

enum {
    Tready,
    Trunning,
};

char result_file[100];

struct hdr {
    const char *sym;
    const char *reason;
};

#define HDRMAX  (sizeof(hdrs) / sizeof(hdrs[0]))

#define _HDRPUT(_fmt,_data...) \
    do { \
        totlen = fprintf(xf,_fmt,_data); \
        for (;  totlen < lim;  ++totlen) \
            fputc(' ',xf); \
    } while (0)
#define HDRPUT(_fmt,_data...) \
    do { \
        _HDRPUT(_fmt,_data); \
        ++ihdr; \
    } while (0)

void
optshow(FILE *xf)
{
    fprintf(xf,"options:\n");
    fprintf(xf,"  C=%d -- test count\n",opt_C);
    fprintf(xf,"  I=%d -- iteration count\n",opt_I);
    fprintf(xf,"  m=%d -- notification (1=msgsnd/msgrcv, 0=sem*)\n",opt_m);
    fprintf(xf,"  S=%d -- 1=pipe set maximum size (F_SETPIPE_SZ)\n",opt_S);
    fprintf(xf,"  x=%d -- 1=child scan/process data\n",opt_x);
}

void
result(int i,const char *pre,double tbest)
{
    int totlen;
    int lim = opt_m ? 12 : 13;

    if (i == 0)
        unlink(result_file);

    FILE *xf = fopen(result_file,"a");

    static struct hdr hdrs[] = {
        { "ELAPSED", "total elapsed time" },
        { "Tnotify", "sender buffer filled -- receiver notified" },
        { "Tready", "receiver becomes ready for input" },
        { "Trunning", "receiver released by sender" },
        { "Latency", "Trunning - Tnotify (time receiver is delayed)" },
        { "size", "buffer size" },
    };

    int ihdr;
    if (i == 0) {
        fprintf(xf,"TEST: %s ...\n",pre);
        optshow(xf);
        fprintf(xf,"\n");

        fprintf(xf,"Legend:\n");
        for (ihdr = 0;  ihdr < HDRMAX;  ++ihdr) {
            _HDRPUT("%s",hdrs[ihdr].sym);
            fprintf(xf,"-- %s\n",hdrs[ihdr].reason);
        }

        fprintf(xf,"\n");
        for (ihdr = 0;  ihdr < HDRMAX;  ++ihdr)
            _HDRPUT("%s",hdrs[ihdr].sym);
        fprintf(xf,"\n");
    }

    ihdr = 0;
    HDRPUT("%.9f",tbest);
    HDRPUT("%.9f",tscbest[Tnotify]);
    HDRPUT("%.9f",tscrcv[Tready]);

    HDRPUT("%.9f",tscrcv[Trunning]);
    HDRPUT("%.9f",tscrcv[Trunning] - tscbest[Tnotify]);
    fprintf(xf,"%zu",size);
    fprintf(xf,"\n");

    fclose(xf);
}

void
result_show(void)
{
    char *av[3];
    pid_t pid;

    av[0] = "cat";
    av[1] = result_file;
    av[2] = NULL;

    if (opt_v) {
        if (result_file[0] == 0)
            sysfault("result_show: null result_file\n");

        pid = fork();

        if (pid == 0) {
            printf("\n");
            execvp(av[0],av);
            sysfault("result_show: exec fault -- %s\n",strerror(errno));
        }

        waitpid(pid,NULL,0);
    }
}

void
scan(char *buf,size_t len)
{

    if (opt_x)
        fake = memchr(buf,'2',len);
}

void
notify_open(void)
{

    if (opt_m)
        sem = msgget(IPC_PRIVATE,IPC_CREAT | 0666);
    else
        sem = semget(IPC_PRIVATE, 1, IPC_CREAT | 0666);
}

void
notify_close(void)
{

    if (opt_m)
        msgctl(sem, 0, IPC_RMID);
    else
        semctl(sem, 0, IPC_RMID);
}

void
notify_send(void)
{
    struct msg msg;

    tscsend[Tnotify] = Tmark;

    if (opt_m) {
        msg.msg_type = MSG_TYPE;
        msgsnd(sem,&msg,sizeof(msg) - sizeof(msg.msg_type),0);
    }
    else
        semctl(sem, 0, SETVAL, 1);
}

void
notify_wait(void)
{
    struct msg msg;

    tscrcv[Tready] = Tmark;

    if (opt_m) {
        msgrcv(sem,&msg,sizeof(msg) - sizeof(msg.msg_type),MSG_TYPE,0);
    }
    else {
        while (semctl(sem, 0, GETVAL) != 1);
    }

    tscrcv[Trunning] = Tmark;
}

void
dobest(const char *pre,void (*fnc)(int i,int dtflg))
{
    // fd[0]=READ, fd[1]=WRITE

    if (! opt_v) {
        printf("\n");
        printf("dobest: %s ...\n",pre);
    }
    fflush(stdout);

    sprintf(result_file,"%s_result",pre);

    if (forkflg) {
        pid_t pid = fork();
        if (pid != 0) {
            waitpid(pid,NULL,0);
            result_show();
            return;
        }
    }

    base = 1;
    for (int i = 0; i < opt_C; i++) {
        double tbest = 1e9;

        size = base;
        if ((shmid = shmget(IPC_PRIVATE, size + 1, IPC_CREAT | 0666)) == -1)
            sysfault("dobest: shmget failure -- %s\n",strerror(errno));
        shmPtr = shmat(shmid, 0, 0);
        notify_open();

        // do multiple trials to minimize the effects of timeslicing and system
        // loading
        for (int iter = opt_I;  iter > 0;  --iter) {
            tscbeg = tscgetf();
            barrier;

            fnc(i,iter == 1);

            double tdif = t2 - t1;
            if (tdif < tbest) {
                tbest = tdif;
                memcpy(tscrcv,shmPtr,sizeof(tscrcv));
                memcpy(tscbest,tscsend,sizeof(tscbest));
            }
        }

        // record the best result
        result(i,pre,tbest);

        shmdt(shmPtr);
        shmctl(shmid, IPC_RMID, NULL);
        notify_close();

        base *= 4;
    }

    if (forkflg)
        exit(0);

    result_show();
}

void
pipeone(int i,int dtflg)
{
    int fd[2];
    ssize_t nread;
    int pid, status;

    pipe(fd);
    if (opt_S) {
        fcntl(fd[1], F_SETPIPE_SZ, size);
        fcntl(fd[0], F_SETPIPE_SZ, size);
    }

    char *buf = malloc(size);
    pid = fork();

    if (pid == 0) {
        // READ

        close(fd[1]);

        tscrcv[0] = Tmark;
        int mark = 1;
        while (1) {
            nread = read(fd[0], buf, size);
            int sverr = errno;

            if (nread == 0)
                break;

            if (mark)
                tscrcv[1] = Tmark;
            mark = 0;

            if (nread < 0) {
                printf("pipeone: read error -- %s\n",strerror(sverr));
                break;
            }

            scan(buf,nread);
        }

        memcpy(shmPtr,tscrcv,sizeof(tscrcv));

        exit(0);
    }

    // WRITE
    close(fd[0]);

    t1 = tscgetf();

    for (size_t j = 0; j < size; j++)
        buf[j] = '1';

    tscsend[0] = Tmark;
    size_t off = 0;
    ssize_t xlen;
    for (;  off < size;  off += xlen) {
        xlen = write(fd[1], &buf[off], size - off);
        if (xlen < 0) {
            printf("pipeone: write error off=%zu size=%zu -- %s\n",
                off,size,strerror(errno));
            break;
        }
    }
    close(fd[1]);

    waitpid(pid,&status,0);

    t2 = tscgetf();

    free(buf);
}

void
shmone(int i,int dtflg)
{
    int pid, status;

    pid = fork();

    if (pid == 0) {
        notify_wait();

// NOTE/BUG: not valid to copy from the shared buffer -- the whole point is to
// operate on the shared buffer directly!!!
        scan(shmPtr,size);
        memcpy(shmPtr,tscrcv,sizeof(tscrcv));
        exit(0);
    }

    t1 = tscgetf();
    char *str = shmPtr;

    for (size_t j = 0; j < size; j++)
        str[j] = '1';
    str[size] = '\0';

    notify_send();

    waitpid(pid,&status,0);

    t2 = tscgetf();
}

int
main(int argc,char **argv)
{
    char *cp;

    --argc;
    ++argv;

    setlinebuf(stdout);
    tsczero = tscgetf();
    printf("main (%d)\n",getpid());

    for (;  argc > 0;  --argc, ++argv) {
        cp = *argv;
        if (*cp != '-')
            break;

        cp += 2;
        switch (cp[-1]) {
        case 'C':  // number of trials
            opt_C = (*cp != 0) ? atoi(cp) : 1;
            break;

        case 'I':  // number of iterations
            opt_I = (*cp != 0) ? atoi(cp) : 1;
            break;

        case 'm':  // notifcation message type
            opt_m = (*cp != 0) ? atoi(cp) : 1;
            break;

        case 'S':  // pipe sender should issue F_SETPIPE_SZ
            opt_S = (*cp != 0) ? atoi(cp) : 1;
            break;

        case 'v':  // dump log files to stdout
            opt_v = (*cp != 0) ? atoi(cp) : 1;
            break;

        case 'x':  // child should scan data
            opt_x = (*cp != 0) ? atoi(cp) : 1;
            break;

        default:
            sysfault("main: unknown option -- '%s'\n",cp - 2);
            break;
        }
    }

    if (! opt_v)
        optshow(stdout);

    do {
        cp = *argv;

        if (argc < 1) {
            forkflg = 1;
            dobest("pipe",pipeone);
            dobest("shm",shmone);
            break;
        }

        if (strcmp(cp,"pipe") == 0) {
            dobest("pipe",pipeone);
            break;
        }

        if (strcmp(cp,"shm") == 0) {
            dobest("shm",shmone);
            break;
        }

        sysfault("main: unknown type -- '%s'\n",cp);
    } while (0);

    return 0;
}

C++相关问答推荐

如何用C(使用两个s补数算术的32位程序)计算

VS代码C/C++扩展intellisense无法检测环境特定函数'

如何判断宏参数是否为C语言中的整型文字

如何使用Python C API实现多线程程序？

SDL 2.0-从数组渲染纹理

如何调试LD_PRELOAD库中的构造函数？

ARM64 ASIMD固有的加载uint8_t* 到uint16x8(x3)？

X86/x64上的SIGSEGV，由于原始内存访问和C中的DS寄存器之间的冲突，在Linux上用TCC编译为JIT引擎

在Rust和C之间使用ffi时如何通过 struct 中的[U8；1]成员传递指针

理解C版宏(看起来像未声明的变量？)

获取每个循环迭代结束时的当前时间

用gcc-msse 2编译的C程序包含AVX 1指令

我的C函数起作用了，但我不确定为什么

C I/O：在Windows控制台上处理键盘输入

当b是无符号字符时，int a=(b<；<；2)>；>；2；和int a=b&；0x3F；之间有什么区别？

C"；中的ANN运行时判断失败#2-变量outputLayer；周围的堆栈已损坏.运行后出错

Realloc()：中止的下一个大小无效(核心转储)

从另一个宏函数调用C宏

try 判断长整数是否为素数

分支预测和UB(未定义的行为)

实用课程推荐
相关教程推荐
休闲君评测 |  TOOLFK工具网 |  古诗文网 |  良许Linux教程网 |  Python之禅 |  红色石头 |  入党申请书 | 

© 2020 Hi LearnFk 意见反馈 免责声明 关于我们 桂ICP备11002319号-3 友链&广告位+QQ: 1963612630