xref/libmicro/pipe.c

/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms
 * of the Common Development and Distribution License
 * (the "License").  You may not use this file except
 * in compliance with the License.
 *
 * You can obtain a copy of the license at
 * src/OPENSOLARIS.LICENSE
 * or http://www.opensolaris.org/os/licensing.
 * See the License for the specific language governing
 * permissions and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL
 * HEADER in each file and include the License file at
 * usr/src/OPENSOLARIS.LICENSE.  If applicable,
 * add the following below this CDDL HEADER, with the
 * fields enclosed by brackets "[]" replaced with your
 * own identifying information: Portions Copyright [yyyy]
 * [name of copyright owner]
 *
 * CDDL HEADER END
 */

/*
 * Copyright 2007 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 */

#include <sys/types.h>
#include <sys/stat.h>
#include <sys/wait.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <pthread.h>
#include <signal.h>
#include <string.h>
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <fcntl.h>
#include <errno.h>

#include "libmicro.h"

typedef struct {
    int         ts_once;
    pid_t           ts_child;
    pthread_t       ts_thread;
    int         ts_in;
    int         ts_out;
    int         ts_in2;
    int         ts_out2;
    int         ts_lsn;
    struct sockaddr_in  ts_add;
} tsd_t;

#define FIRSTPORT       12345

static char         *modes[] = {"st", "mt", "mp", NULL};
#define MD_SINGLE       0
#define MD_MULTITHREAD      1
#define MD_MULTIPROCESS     2

static char         *xports[] = {"pipe", "fifo", "sock", "tcp",
                    NULL};
#define XP_PIPES        0
#define XP_FIFOS        1
#define XP_SOCKETPAIR       2
#define XP_LOCALTCP     3

#define DEFM            MD_SINGLE
#define DEFS            1024
#define DEFX            XP_PIPES

static int          optm = DEFM;
static size_t           opts = DEFS;
static int          optx = DEFX;
static void         *rbuf = NULL;
static void         *wbuf = NULL;

int readall(int s, void *buf, size_t len);
void *loopback(void *arg);
int prepare_pipes(tsd_t *tsd);
int prepare_fifos(tsd_t *tsd);
int cleanup_fifos(tsd_t *tsd);
int prepare_socketpair(tsd_t *tsd);
int prepare_localtcp(tsd_t *tsd);
int prepare_localtcp_once(tsd_t *tsd);
char *lookupa(int x, char *names[]);
int lookup(char *x, char *names[]);

int
benchmark_init()
{
    lm_tsdsize = sizeof (tsd_t);

    (void) sprintf(lm_optstr, "m:s:x:");

    (void) sprintf(lm_usage,
        "       [-m mode (st|mt|mp, default %s)]\n"
        "       [-s buffer-size (default %d)]\n"
        "       [-x transport (pipe|fifo|sock|tcp, default %s)]\n"
        "notes: measures write()/read() across various transports\n",
        lookupa(DEFM, modes), DEFS, lookupa(DEFX, xports));

    (void) sprintf(lm_header, "%2s %4s", "md", "xprt");

    return (0);
}

int
benchmark_optswitch(int opt, char *optarg)
{
    int         x;

    switch (opt) {
    case 'm':
        x = lookup(optarg, modes);
        if (x == -1)
            return (-1);
        optm = x;
        break;
    case 's':
        opts = sizetoll(optarg);
        break;
    case 'x':
        x = lookup(optarg, xports);
        if (x == -1)
            return (-1);
        optx = x;
        break;
    default:
        return (-1);
    }
    return (0);
}

int
benchmark_initrun()
{
    if (optx == XP_FIFOS) {
        if (geteuid() != 0) {
            (void) printf("sorry, must be root to create fifos\n");
            exit(1);
        }
    }

    (void) setfdlimit(4 * lm_optT + 10);

    rbuf = malloc(opts);
    wbuf = malloc(opts);

    return (0);
}

int
benchmark_initbatch(void *tsd)
{
    tsd_t           *ts = (tsd_t *)tsd;
    int         result;
    pid_t           pid;
    int         i;

    switch (optx) {
    case XP_SOCKETPAIR:
        result = prepare_socketpair(ts);
        break;
    case XP_LOCALTCP:
        result = prepare_localtcp(ts);
        break;
    case XP_FIFOS:
        result = prepare_fifos(ts);
        break;
    case XP_PIPES:
    default:
        result = prepare_pipes(ts);
        break;
    }
    if (result == -1) {
        return (1);
    }

    switch (optm) {
    case MD_MULTITHREAD:
        result = pthread_create(&ts->ts_thread, NULL, loopback, tsd);
        if (result == -1) {
            return (1);
        }
        break;
    case MD_MULTIPROCESS:
        pid = fork();
        switch (pid) {
        case 0:
            (void) loopback(tsd);
            exit(0);
            break;
        case -1:
            return (-1);
        default:
            ts->ts_child = pid;
            break;
        }
        break;
    case MD_SINGLE:
    default:
        break;
    }

    /* Prime the loopback */
    if (write(ts->ts_out, wbuf, opts) != opts) {
        return (1);
    }
    if (readall(ts->ts_in, rbuf, opts) != opts) {
        return (1);
    }

    return (0);
}

int
benchmark(void *tsd, result_t *res)
{
    tsd_t           *ts = (tsd_t *)tsd;
    int         i;
    int         n;

    for (i = 0; i < lm_optB; i++) {
        if (write(ts->ts_out, wbuf, opts) != opts) {
            res->re_errors++;
            continue;
        }

        n = readall(ts->ts_in, rbuf, opts);
        if (n == -1) {
            res->re_errors++;
            continue;
        }
    }
    res->re_count = i;

    return (0);
}

int
benchmark_finibatch(void *tsd)
{
    tsd_t           *ts = (tsd_t *)tsd;

    /* Terminate the loopback */
    (void) write(ts->ts_out, wbuf, opts);
    (void) readall(ts->ts_in, rbuf, opts);

    switch (optm) {
    case MD_MULTITHREAD:
        (void) close(ts->ts_in2);
        (void) close(ts->ts_out2);
        (void) pthread_join(ts->ts_thread, NULL);
        break;
    case MD_MULTIPROCESS:
        (void) close(ts->ts_in2);
        (void) close(ts->ts_out2);
        (void) waitpid(ts->ts_child, NULL, 0);
        break;
    case MD_SINGLE:
    default:
        break;
    }

    (void) close(ts->ts_in);
    (void) close(ts->ts_out);

    if (optx == XP_FIFOS) {
        (void) cleanup_fifos(ts);
    }

    return (0);
}

char *
benchmark_result()
{
    static char     result[256];

    (void) sprintf(result, "%2s %4s",
        lookupa(optm, modes), lookupa(optx, xports));

    return (result);
}

int
readall(int s, void *buf, size_t len)
{
    size_t          n;
    size_t          total = 0;

    for (;;) {
        n = read(s, (void *)((long)buf + total), len - total);
        if (n < 1) {
            return (-1);
        }
        total += n;
        if (total >= len) {
            return (total);
        }
    }
}

void *
loopback(void *arg)
{
    tsd_t           *ts = (tsd_t *)arg;
    int         i, n, m;

    /* Include priming and termination */
    m = lm_optB + 2;

    for (i = 0; i < m; i++) {
        n = readall(ts->ts_in2, rbuf, opts);
        if (n == -1) {
            break;
        }
        if (write(ts->ts_out2, wbuf, opts) != opts) {
            break;
        }
    }

    return (NULL);
}

int
prepare_localtcp_once(tsd_t *ts)
{
    int         j;
    int         opt = 1;
    struct hostent  *host;

    j = FIRSTPORT;

    ts->ts_lsn = socket(AF_INET, SOCK_STREAM, 0);
    if (ts->ts_lsn == -1) {
        return (-1);
    }

    if (setsockopt(ts->ts_lsn, SOL_SOCKET, SO_REUSEADDR,
        &opt, sizeof (int)) == -1) {
        return (-1);
    }

    if ((host = gethostbyname("localhost")) == NULL) {
        return (-1);
    }

    for (;;) {
        (void) memset(&ts->ts_add, 0,
            sizeof (struct sockaddr_in));
        ts->ts_add.sin_family = AF_INET;
        ts->ts_add.sin_port = htons(j++);
        (void) memcpy(&ts->ts_add.sin_addr.s_addr,
            host->h_addr_list[0], sizeof (struct in_addr));

        if (bind(ts->ts_lsn,
            (struct sockaddr *)&ts->ts_add,
            sizeof (struct sockaddr_in)) == 0) {
            break;
        }

        if (errno != EADDRINUSE) {
            return (-1);
        }
    }

    if (listen(ts->ts_lsn, 5) == -1) {
        return (-1);
    }

    return (0);
}

int
prepare_localtcp(tsd_t *ts)
{
    int         result;
    struct sockaddr_in  addr;
    int         opt = 1;
    socklen_t       size;

    if (ts->ts_once++ == 0) {
        if (prepare_localtcp_once(ts) == -1) {
            return (-1);
        }
    }

    ts->ts_out = socket(AF_INET, SOCK_STREAM, 0);
    if (ts->ts_out == -1) {
        return (-1);
    }

    if (fcntl(ts->ts_out, F_SETFL, O_NDELAY) == -1) {
        return (-1);
    }

    result = connect(ts->ts_out, (struct sockaddr *)&ts->ts_add,
        sizeof (struct sockaddr_in));
    if ((result == -1) && (errno != EINPROGRESS)) {
        return (-1);
    }

    if (fcntl(ts->ts_out, F_SETFL, 0) == -1) {
        return (-1);
    }

    size = sizeof (struct sockaddr);
    result = accept(ts->ts_lsn, (struct sockaddr *)&addr, &size);
    if (result == -1) {
        return (-1);
    }
    ts->ts_out2 = result;

    if (setsockopt(ts->ts_out, IPPROTO_TCP, TCP_NODELAY,
        &opt, sizeof (int)) == -1) {
        return (-1);
    }

    if (setsockopt(ts->ts_out2, IPPROTO_TCP, TCP_NODELAY,
        &opt, sizeof (int)) == -1) {
        return (-1);
    }

    if (optm == MD_SINGLE) {
        ts->ts_in = ts->ts_out2;
    } else {
        ts->ts_in = ts->ts_out;
        ts->ts_in2 = ts->ts_out2;
    }

    return (0);
}

int
prepare_socketpair(tsd_t *ts)
{
    int         s[2];

    if (socketpair(PF_UNIX, SOCK_STREAM, 0, s) == -1) {
        return (-1);
    }

    if (optm == MD_SINGLE) {
        ts->ts_in = s[0];
        ts->ts_out = s[1];
    } else {
        ts->ts_in = s[0];
        ts->ts_out = s[0];
        ts->ts_in2 = s[1];
        ts->ts_out2 = s[1];
    }

    return (0);
}

int
prepare_fifos(tsd_t *ts)
{
    char            path[64];

    (void) sprintf(path, "/tmp/pipe_%ld.%dA",
        getpid(), pthread_self());
    if (mknod(path, 0600, S_IFIFO) == -1) {
        return (-1);
    }

    if (optm == MD_SINGLE) {
        ts->ts_in = open(path, O_RDONLY);
        ts->ts_out = open(path, O_WRONLY);
    } else {
        ts->ts_in = open(path, O_RDONLY);
        ts->ts_out2 = open(path, O_WRONLY);

        (void) sprintf(path, "/tmp/pipe_%ld.%dB",
            getpid(), pthread_self());
        if (mknod(path, 0600, S_IFIFO) == -1) {
            return (-1);
        }

        ts->ts_in2 = open(path, O_RDONLY);
        ts->ts_out = open(path, O_WRONLY);
    }

    return (0);
}

/*ARGSUSED*/
int
cleanup_fifos(tsd_t *ts)
{
    char            path[64];

    (void) sprintf(path, "/tmp/pipe_%ld.%dA", getpid(), pthread_self());
    (void) unlink(path);
    (void) sprintf(path, "/tmp/pipe_%ld.%dB", getpid(), pthread_self());
    (void) unlink(path);

    return (0);
}

int
prepare_pipes(tsd_t *ts)
{
    int         p[2];

    if (optm == MD_SINGLE) {
        if (pipe(p) == -1) {
            return (-1);
        }
        ts->ts_in = p[0];
        ts->ts_out = p[1];

    } else {
        if (pipe(p) == -1) {
            return (-1);
        }
        ts->ts_in = p[0];
        ts->ts_out2 = p[1];

        if (pipe(p) == -1) {
            return (-1);
        }
        ts->ts_in2 = p[0];
        ts->ts_out = p[1];
    }

    return (0);
}

char *
lookupa(int x, char *names[])
{
    int         i = 0;

    while (names[i] != NULL) {
        if (x == i) {
            return (names[i]);
        }
        i++;
    }
    return (NULL);
}

int
lookup(char *x, char *names[])
{
    int         i = 0;

    while (names[i] != NULL) {
        if (strcmp(names[i], x) == 0) {
            return (i);
        }
        i++;
    }
    return (-1);
}