Lwt_unix.writev performance measurements

writev.c

#include <stdlib.h>
#include <stdio.h>
#include <string.h>

#include <unistd.h>
#include <sys/uio.h>
#include <fcntl.h>
#include <sys/time.h>
#include <sys/times.h>

#include <limits.h>

#define COUNT       1024
#define SIZE        1024
#define FILENAME    "foo"

double measure(void (*f)(int, char**), int file, char **buffers, char *tag)
{
    struct tms      start, end;
    struct timeval  start_real, end_real;

    size_t          repetitions = (1 << 28) / SIZE / COUNT;

    if (times(&start) == -1)
        exit(EXIT_FAILURE);
    if (gettimeofday(&start_real, NULL) == -1)
        exit(EXIT_FAILURE);

    for (int repetition = 0; repetition < repetitions; ++repetition)
        f(file, buffers);

    if (times(&end) == -1)
        exit(EXIT_FAILURE);
    if (gettimeofday(&end_real, NULL) == -1)
        exit(EXIT_FAILURE);

    long        rate = sysconf(_SC_CLK_TCK);
    double      user_ticks = end.tms_utime - start.tms_utime;
    double      system_ticks = end.tms_stime - start.tms_stime;
    double      total_time = (user_ticks + system_ticks) / rate;

    double      throughput =
        (double)COUNT * SIZE / 1024 * repetitions / 1024 / total_time;

    double      wall_sec_delta = end_real.tv_sec - start_real.tv_sec;
    double      wall_usec_delta = end_real.tv_usec - start_real.tv_usec;
    double      wall_time = wall_sec_delta + wall_usec_delta / 1000000;

    double      wall_throughput =
        (double)COUNT * SIZE / 1024 * repetitions / 1024 / wall_time;

    printf(
        " %-32s   %6.1lf MB/s  (%4.2lf s)  %6.1lf MB/s  (%4.2lf s)\n",
        tag, throughput, total_time, wall_throughput, wall_time);

    return throughput;
}

void use_write(int file, char **buffers)
{
    for (int index = 0; index < COUNT; ++index) {
        if (write(file, buffers[index], SIZE) != SIZE)
            exit(EXIT_FAILURE);
    }
}

void use_writev(int file, char **buffers)
{
    struct iovec    io_vectors[COUNT];

    for (int index = 0; index < COUNT; ++index) {
        io_vectors[index].iov_base = buffers[index];
        io_vectors[index].iov_len = SIZE;
    }

    if (writev(file, io_vectors, COUNT) != SIZE * COUNT)
        exit(EXIT_FAILURE);
}

void use_write_coalesce(int file, char **buffers)
{
    char    *coalesced = malloc(SIZE * COUNT);
    if (coalesced == NULL)
        exit(EXIT_FAILURE);

    for (int index = 0; index < COUNT; ++index)
        memcpy(coalesced + index * SIZE, buffers[index], SIZE);

    if (write(file, coalesced, SIZE * COUNT) != SIZE * COUNT)
        exit(EXIT_FAILURE);

    free(coalesced);
}

int main()
{
    char    *buffers[COUNT];
    char    symbol = 'a';

    for (int index = 0; index < COUNT; ++index) {
        buffers[index] = malloc(SIZE);
        memset(buffers[index], symbol, SIZE);
        buffers[index][SIZE - 1] = '\n';

        ++symbol;
        if (symbol > 'z')
            symbol = 'a';
    }

    int     file = open(FILENAME, O_WRONLY | O_CREAT | O_TRUNC, 0644);

    printf(
        "function                             process                wall\n");

    measure(use_write, file, buffers, "write (C)");

    double  writev =
        measure(use_writev, file, buffers, "writev (C)");
    double  coalescing =
        measure(use_write_coalesce, file, buffers, "write (C, coalesced)");

    printf("writev:coalsecing %lf\n", writev / coalescing);

    return EXIT_SUCCESS;
}

writev.ml

open Lwt.Infix

let count = 1024
let size = 1024
let repetitions = (1 lsl 28) / count / size
let filename = "bar"

let measure_general tag f =
  let open Unix in

  let start_times = times () in
  let start_wall = gettimeofday () in

  f ();

  let end_times = times () in
  let end_wall = gettimeofday () in

  let total_times =
    end_times.tms_utime -. start_times.tms_utime +.
    end_times.tms_stime -. start_times.tms_stime
  in

  let total_wall = end_wall -. start_wall in

  let count = float_of_int count in
  let size = float_of_int size in
  let repetitions = float_of_int repetitions in

  let times_throughput =
    count *. size /. 1024. *. repetitions /. 1024. /. total_times in

  let wall_throughput =
    count *. size /. 1024. *. repetitions /. 1024. /. total_wall in

  Printf.printf
    " %-40s   %6.1f MB/s  (%4.2f s)  %6.1f MB/s  (%4.2f s)\n"
    tag times_throughput total_times wall_throughput total_wall;
  flush Pervasives.stdout

let measure_lwt tag f =
  measure_general tag (fun () ->
    let rec repeat count =
      if count = 0 then Lwt.return_unit
      else f () >>= fun () -> repeat (count - 1)
    in
    Lwt_main.run (repeat repetitions))

let measure_unix tag f =
  measure_general tag (fun () -> for _ = 1 to repetitions do f () done)

let () =
  let buffers =
    Array.init count (fun index ->
      let character = Char.chr ((Char.code 'a') + (index mod 26)) in
      let bytes = Bytes.make size character in
      Bytes.set bytes (size - 1) '\n';
      bytes)
  in

  let bigarray_buffers =
    Array.init count (fun index -> Lwt_bytes.of_bytes buffers.(index)) in

  let file =
    Lwt_main.run
      (Lwt_unix.openfile filename Unix.[O_WRONLY; O_CREAT; O_TRUNC] 0o644)
  in
  let unix_file = Lwt_unix.unix_file_descr file in

  Printf.printf
    "function                             process                wall\n";
  flush stdout;

  measure_unix "Unix.write" (fun () ->
    for index = 0 to count - 1 do
      if Unix.single_write unix_file buffers.(index) 0 size <> size then
        raise Exit
    done);

  let lwt_measurements blocking =
    measure_lwt (Printf.sprintf "Lwt_unix.write, %s" blocking) (fun () ->
      let rec loop index =
        if index >= count then Lwt.return_unit
        else
          Lwt_unix.write file buffers.(index) 0 size >>= fun bytes_written ->
          if bytes_written <> size then raise Exit
          else loop (index + 1)
      in
      loop 0);

    measure_lwt (Printf.sprintf "Lwt_bytes.write, %s" blocking) (fun () ->
      let rec loop index =
        if index >= count then Lwt.return_unit
        else
          Lwt_bytes.write file bigarray_buffers.(index) 0 size
            >>= fun bytes_written ->
          if bytes_written <> size then raise Exit
          else loop (index + 1)
      in
      loop 0);

    measure_lwt (Printf.sprintf "writev, bytes, %s" blocking) (fun () ->
      let io_vectors =
        let count_ = count in
        let open Lwt_unix.IO_vectors in
        let io_vectors = create () in
        for index = 0 to count_ - 1 do
          append_bytes io_vectors buffers.(index) 0 size
        done;
        io_vectors
      in
      Lwt_unix.writev file io_vectors >>= fun bytes_written ->
      if bytes_written <> size * count then raise Exit
      else Lwt.return_unit);

    measure_lwt (Printf.sprintf "writev, bigarrays, %s" blocking) (fun () ->
      let io_vectors =
        let count_ = count in
        let open Lwt_unix.IO_vectors in
        let io_vectors = create () in
        for index = 0 to count_ - 1 do
          append_bigarray io_vectors bigarray_buffers.(index) 0 size
        done;
        io_vectors
      in
      Lwt_unix.writev file io_vectors >>= fun bytes_written ->
      if bytes_written <> size * count then raise Exit
      else Lwt.return_unit);

    measure_lwt (Printf.sprintf "write, bytes, coalesced, %s" blocking)
        (fun () ->
      let coalesced = Bytes.create (size * count) in
      for index = 0 to count - 1 do
        Bytes.blit buffers.(index) 0 coalesced (index * size) size
      done;
      Lwt_unix.write file coalesced 0 (size * count) >>= fun bytes_written ->
      if bytes_written <> size * count then raise Exit
      else Lwt.return_unit);
  in

  lwt_measurements "blocking";

  Lwt_unix.set_blocking file false;
  lwt_measurements "non-blocking"