pb0.c

#include <stdlib.h>
#include <stdio.h>
#include <stdbool.h>
#include <unistd.h>
#include <string.h>
#include <thread.h>
#include <synch.h>
#include <errno.h>
#include <err.h>

#define KILOBYTE        1024
#define MEGABYTE        (1024 * KILOBYTE)

#define QUANTUM         (128 * KILOBYTE)

typedef struct {
        int ctx_fd_write;
        int ctx_fd_read;
        size_t ctx_quantum;

        mutex_t ctx_lock;
        hrtime_t ctx_last_reset;
        size_t ctx_sz_read;

} context_t;

static void *
thread_write(void *arg)
{
        context_t *ctx = arg;
        int fd = ctx->ctx_fd_write;
        char buf[QUANTUM];

        for (size_t i = 0; i < sizeof (buf); i++) {
                buf[i] = 'A' + (i % 26);
        }

        for (;;) {
                errno = 0;
                if (write(fd, buf, QUANTUM) != QUANTUM) {
                        err(1, "write");
                }
        }
}

static void *
thread_read(void *arg)
{
        context_t *ctx = arg;
        int fd = ctx->ctx_fd_read;
        static char buf[10 * MEGABYTE];

        for (;;) {
                ssize_t rsz;

                if ((rsz = read(fd, buf, sizeof (buf))) < 0) {
                        err(1, "read");
                } else if (rsz == 0) {
                        errx(1, "unexpected 0 size read");
                }

                mutex_enter(&ctx->ctx_lock);
                ctx->ctx_sz_read += rsz;
                mutex_exit(&ctx->ctx_lock);
        }
}

int
main(int argc, char *argv[])
{
        context_t ctx;

        setbuf(stdout, NULL);

        unsigned runforsecs = 0;
        if (argc > 1) {
                int a = atoi(argv[1]);

                if (a < 1 || a > 1000) {
                        errx(1, "invalid seconds argument");
                }
                runforsecs = (unsigned)a;
        }

        int fds[2];
        if (pipe(fds) != 0) {
                err(1, "pipe");
        }

        (void) memset(&ctx, 0, sizeof (ctx));
        ctx.ctx_fd_read = fds[0];
        ctx.ctx_fd_write = fds[1];
        ctx.ctx_quantum = QUANTUM;

        if (mutex_init(&ctx.ctx_lock, USYNC_THREAD | LOCK_ERRORCHECK,
            NULL) != 0) {
                err(1, "mutex_init");
        }

        hrtime_t start = gethrtime();

        int e;
        if ((e = thr_create(NULL, 0, thread_read, &ctx, 0, NULL) != 0) ||
            (e = thr_create(NULL, 0, thread_write, &ctx, 0, NULL) != 0)) {
                errno = e;
                err(1, "thr_create");
        }

        unsigned long long total = 0;

        for (;;) {
                bool display = false;
                size_t sz;
                hrtime_t delta = 0;
                hrtime_t now;

                mutex_enter(&ctx.ctx_lock);
                now = gethrtime();
                total += ctx.ctx_sz_read;
                if (ctx.ctx_last_reset != 0) {
                        delta = now - ctx.ctx_last_reset;
                        sz = ctx.ctx_sz_read;
                        display = true;
                }
                ctx.ctx_last_reset = now;
                ctx.ctx_sz_read = 0;
                mutex_exit(&ctx.ctx_lock);

                if (display) {
                        double bps = sz * 1000000000.0 / (double)delta;
                        double mbps = bps / 1024.0 / 1024.0;

                        printf("%llu msec %llu bytes read (%.2f bytes/sec) "
                            "(%.2f MB/sec)\n",
                            (long long unsigned)(delta / 1000000),
                            (long long unsigned)sz,
                            bps,
                            mbps);
                }

                if (runforsecs != 0) {
                        delta = gethrtime() - start;

                        if ((delta / 1000000000ULL) > runforsecs) {
                                double mbps = total *
                                    1000000000.0 / (double)delta /
                                    1024.0 / 1024.0;

                                printf("TOTAL %llu msec %llu bytes "
                                    "read @ %.2f MB/sec\n",
                                    (long long unsigned)(delta / 1000000),
                                    (long long unsigned)total,
                                    mbps);
                                exit(0);
                        }
                }

                sleep(1);
        }
}