perftest.c

#include "test.h"
#include "../oio.h"

#include <stdio.h>

#define BUFSIZE 4096
#define CONNS 10000
#define ACTIVE_CONNECTS 500

int connections = 0;
int connections_made = 0;
DWORD start;
__int64 read_total = 0;
__int64 written_total = 0;

typedef struct {
  union {
    struct {
  oio_handle handle;
  oio_req req1;
  oio_req req2;
  oio_buf vec1;
  oio_buf vec2;
    };

    char dummy[BUFSIZE];
  };

  char buf1[BUFSIZE];
  char buf2[BUFSIZE];
} pump_t;

typedef struct {
  oio_handle handle;
  oio_req reqs[ACTIVE_CONNECTS];
} server_t;

struct sockaddr_in addr_listen; 
struct sockaddr_in addr_connect;
struct sockaddr_in addr_client;

void pump_connect(oio_req *req);
void pump_read(oio_req *req);
void pump_write(oio_req *req);

static int mbit(__int64 bytes, DWORD passed_milis) {
  return (int)(bytes / (125 * passed_milis));
}

static void show_stats() {
  DWORD now = GetTickCount();
  DWORD passed_milis = (now - start);
  if (passed_milis >= 1000) {
    fprintf(stderr, "read: %d mbit/s, write: %d mbit/s\n", mbit(read_total, passed_milis), mbit(written_total, passed_milis));
    start = now;
    written_total=read_total=0;
  }
}

void pump_on_close(oio_handle* handle, oio_err err) {
  if (err) {
    fprintf(stdout, "Pump socket error\n");
  } else {
    fprintf(stdout, "Pump closed");
  }
}

static void on_server_close(oio_handle* handle, oio_err err) {
  if (err) {
    fprintf(stdout, "Server socket error\n");
  } else {
    fprintf(stdout, "Server closed");
  }
}

void pump_read_done(oio_req *req, int bytes) {
  if (bytes == 0)
    assert(0);

  read_total += bytes;
  pump_read(req);
}

void pump_read(oio_req *req) {
  oio_read(req, (oio_buf*)req->data, 1);
}

void pump_write_done(oio_req *req) {
  oio_buf *vec = (oio_buf*)req->data;
  written_total += vec->len;
  pump_write(req);
  show_stats();
}

void pump_write(oio_req *req) {
  if (oio_write(req, (oio_buf*)req->data, 1))
    abort();
}

void pump_connect_done(oio_req *req, oio_err e) {
  pump_t* pump = (pump_t*)req->data;

  if (e) {
    connections--;
    pump_connect(req);
    return;
  }

  oio_req_init(&pump->req1, &pump->handle, pump_write_done);
  oio_req_init(&pump->req2, &pump->handle, pump_write_done);

  pump->req1.data = (void*)&pump->vec1;
  pump->vec1.base = (char*)&pump->buf1;
  pump->req2.data = (void*)&pump->vec2;
  pump->vec1.len = sizeof(pump->buf1);
  pump->vec2.base = (char*)&pump->buf2;
  pump->vec2.len = sizeof(pump->buf2);

  pump_write(&pump->req1);
  pump_write(&pump->req2);

  connections_made++;
  fprintf(stderr, "%d connections\n", connections_made);

  if (connections < CONNS)
    pump_connect(req);
}

void pump_connect(oio_req *req) {
 pump_t* pump = (pump_t*)VirtualAlloc(NULL, sizeof(*pump), MEM_COMMIT | MEM_RESERVE, PAGE_READWRITE);

 if (oio_tcp_handle_init(&pump->handle, pump_on_close, NULL))
   assert(0);
 if (oio_bind(&pump->handle, (struct sockaddr*) &addr_client))
   assert(0);

 oio_req_init(req, &pump->handle, pump_connect_done);
 req->data = (void*)pump;
 
 if (oio_connect(req, (struct sockaddr*) &addr_connect))
   assert(0);

 connections++;
}

void on_server_accept(oio_handle* server) {
  pump_t* pump = malloc(sizeof(*pump));

  if (oio_tcp_handle_accept(server, &pump->handle, pump_on_close, (void*)pump))
    assert(0);

  oio_req_init(&pump->req1, &pump->handle, pump_read_done);
  oio_req_init(&pump->req2, &pump->handle, pump_read_done);

  pump->req1.data = (void*)&pump->vec1;
  pump->vec1.base = (char*)&pump->buf1;
  pump->req2.data = (void*)&pump->vec2;
  pump->vec1.len = sizeof(pump->buf1);
  pump->vec2.base = (char*)&pump->buf2;
  pump->vec2.len = sizeof(pump->buf2);

  pump_read(&pump->req1);
  pump_read(&pump->req2);
}

void start_server() {
  server_t *server = (server_t*)malloc(sizeof(*server));
  int i;

  if (oio_tcp_handle_init(&server->handle, on_server_close, NULL))
    assert(0);
  if (oio_bind(&server->handle, (struct sockaddr*) &addr_listen))
    assert(0);
  if (oio_listen(&server->handle, CONNS, on_server_accept))
    assert(0);

  for (i = 0; i < ACTIVE_CONNECTS && connections < CONNS; i++)
    pump_connect(&server->reqs[i]);
}

int main() {
  addr_listen = oio_ip4_addr("0.0.0.0", TEST_PORT);
  addr_connect = oio_ip4_addr("127.0.0.1", TEST_PORT);
  addr_client = oio_ip4_addr("0.0.0.0", 0);

  oio_init();
  start_server();
  start = GetTickCount();
  oio_run();

  return 0;
}