Tarantool Quick Test

## tar_test.c
// Tarantool quick test
// Copyright, Dennis Anikin 2016
//
// Quick disclaimer:
//
// This test shows 500K-1000K transactions per second on one CPU core
// and 600K-1600K queries per second on one CPU core.
//
// Based on the $6.57 per-month-price for the AWS t2.micro instance we can afford the tremendous number of 630bln queries for just $1
//
// A typical output of this test should look like this:
// https://cloclo18.datacloudmail.ru/weblink/view/28zo6HFUdUrh/img-2016-03-10-14-43-48.png?etag=E3D0131D9EFE2C928BD2CC819096535E09F4AFF1&key=98e4a8efef63f289046072a10c02a508fb204c82
//
// Here is a quick instruction how to run this test on Centos 6.x machines. All you need to run this
// test is a couple of such machines.
//
// Try to do the following things from point 1 to point 37
//
// Getting ready:
//
// 1. Launch two AWS t2.micro instances with AWS Linux: https://aws.amazon.com or
//    two Google 1v CPU instances on Centos 6.x: https://console.cloud.google.com. Otherwise you
//    can use any Centos 6.x instances on any cloud platform or in your own datacenter.
// 2. Connect via SSH terminal to both of them.
// 3. Head over to http://tarantool.org/download.html
// 4. Click Amazon Linux.
// 5. Copy the command set.
// 6. Open the server terminal
// 7. Paste everything to the console and press Enter.
// 8. Press “y” and then enter whenever the console asks you to do so.
// 9. sudo ln -sf /etc/tarantool/instances.available/example.lua /etc/tarantool/instances.enabled/example.lua
// 10. sudo sed "s/space:create_index('primary')/space:create_index('primary', {type='HASH'})/g" -i /etc/tarantool/instances.enabled/example.lua
// 11. sudo sed 's/wal_mode = ".*"/wal_mode = "write"/g' -i /etc/tarantool/instances.enabled/example.lua
// 11.5 sudo sed "s/listen = 'localhost:3301';/listen = '*:3301';/g" -i /etc/tarantool/instances.enabled/example.lua
// 12. sudo tarantoolctl start example
// 13. Open the client terminal
// 14. Head over to http://tarantool.org/download.html
// 15. Click Amazon Linux.
// 16. Copy the command set except the last line.
// 17. Paste everything to the console and press Enter.
// 18. Press “y” and then Enter whenever the console asks you to do so.
// 19. sudo yum -y install tarantool-c tarantool-c-devel
// 20. Press “y” and then Enter whenever the console asks you to do so.
// 20.5. Copy-paste the following command to the console and press Enter

/*

sudo tee /etc/yum.repos.d/tarantool_1_6.repo <<- EOF
[tarantool_1_6]
name=EnterpriseLinux-\$releasever - Tarantool
baseurl=http://download.tarantool.org/tarantool/1.6/el/6/\$basearch/
gpgkey=http://download.tarantool.org/tarantool/1.6/gpgkey
repo_gpgcheck=1
gpgcheck=0
enabled=1

[tarantool_1_6-source]
name=EnterpriseLinux-\$releasever - Tarantool Sources
baseurl=http://download.tarantool.org/tarantool/1.6/el/6/SRPMS
gpgkey=http://download.tarantool.org/tarantool/1.6/gpgkey
repo_gpgcheck=1
gpgcheck=0
EOF

*/

//
// The test
//
// in the client console
//
// 21. mkdir tar_test
// 22. cd tar_test
// 23. cat >tar_test.c // Then copy and paste the whole file tar_test.c that you're reading to the console and press CTRL+D
// 24. sudo yum install gcc
// 25. cc -g -lpthread -O3 -std=gnu99 tar_test.c -ltarantool -o tar_test
//
// In the server console
//
// 26. ifconfig
// 27. Copy IP address.
//
// In the client console
//
// 28. ./tar_test 0.0.0.0:3301 write 3 10000000 // Substitute 0.0.0.0 with the IP address pasted from the clipboard.
//
// 29. See one million transactions per second!!!
//
// In the server console
//
// 30. top // CPU is the bottleneck.
// 31. sudo ls -la /var/lib/tarantool/example // Number of xlogs is growing and xlogs size is growing.
// 32. sudo yum install sysstat
// 33. iostat -x 3 // Disk is 15-20% busy, 100-120K sectore per second
// 34. fdisk -l // The sector size is 512 bytes. Which means that totally we commit transactions at the speed of 50-60Mb per second.
//
// In the client console
//
// 35. ./tar_test 0.0.0.0:3301 read 5 10000000 // Substitute 0.0.0.0 with the server IP that you got above and you'll get
//     700K-1600K queries per second.
//
// In the server console
//
// 36. top // 30-40% CPU is used.
// 37. iostat -x 3 // No disk activity.
//
// The bottom line:
// One million transactions per second on one CPU core!
// 700K-1600K queries per second on 1/3 CPU core!

#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <sys/time.h>
#include <tarantool/tarantool.h>
#include <tarantool/tnt_net.h>
#include <tarantool/tnt_opt.h>
#include <pthread.h>
#include <errno.h>
#include <string.h>
#include <unistd.h>

// General parameters
typedef struct test_params
{
	char address[256];
	int is_read_test;
	int num_ops;
} test_params;

// Parameters per thread
typedef struct thread_data
{
	test_params *tp;
	long long num_requests, num_responses;

	struct tnt_stream *tnt;
} thread_data;

int NUM_THREADS = 30;

// The thread that reads responses
void *read_thread(void *ctx)
{
	thread_data *td = (thread_data*)ctx;
	struct tnt_stream *tnt = td->tnt;

	struct tnt_reply reply;
	tnt_reply_init(&reply);

	int num_errs = 0;

	// Iterate through all the operations
	for (int i = 0; i < td->tp->num_ops; ++i)
	{
		// Read a reply
                int r = tnt->read_reply(tnt, &reply);
                if (r == 1)
                {
                	// This means that we're reading responses way to fast and there is nothing in the socket
                	// Let's just wait a little bit
                	// It is better to wait 1 ms one in a while than to use mutex every time in terms of CPU  usage
                        --i;
                        usleep(1000);
                        continue;
                }
                else
                if (r == -1)
		{
			--i;
			printf("Read reply failed %s\n", reply.error);
			++num_errs;

			// To many errors
			if (num_errs >= 5)
				break;
		}
		else
		{
			num_errs = 0;
			++td->num_responses;
		}

 		tnt_reply_free(&reply);
	}
}

void *do_test(void *ctx)
{
	thread_data *td = (thread_data*)ctx;

	struct tnt_stream *tnt = tnt_net(NULL);

	td->tnt = tnt;

	tnt_set(tnt, TNT_OPT_URI, td->tp->address);

	// Connect to the Tarantool server
	if (tnt_connect(tnt) < 0)
	{
		printf("Connection refused\n");
       		exit(-1);
	}

	struct tnt_stream *tuple = tnt_object(NULL);

        int k = (int)time(NULL);

	struct timeval tv, prev_tv;

	// Create the thread that reads responses
	pthread_t read_pid;
        int r = pthread_create(&read_pid, NULL, &read_thread, td);
        if (r < 0)
        {
        	fprintf(stderr, "multithread_test: could not create thread, r=%d, errno='%s'\n", r, strerror(errno));
                fflush(stderr);
        }

	// Iterate through all the operations
	for (int i = 0; i < td->tp->num_ops; ++i)
        {
		// Form a request to a server
                if (td->tp->is_read_test)
                {
                        tnt_object_add_array(tuple, 1);
                        tnt_object_add_int(tuple, k + i);

                        tnt_select(tnt, 512, 0, UINT32_MAX, 0, 0, tuple);
               	}
                else
                {
                        tnt_object_add_array(tuple, 2);
                        tnt_object_add_int(tuple, k + i);

                   	tnt_object_add_int(tuple, k + i);

                        tnt_replace(tnt, 512, tuple);
                }

		// Send a request to the server
		tnt_flush(tnt);

		++td->num_requests;

                tnt_object_reset(tuple);
	}

	// Join the read thread before clear all the resources
	void *v;
	pthread_join(read_pid, &v);

        tnt_close(tnt);
        tnt_stream_free(tuple);
        tnt_stream_free(tnt);
}

// The thread that prints statistics one in a second
void *timer_thread(void *ctx)
{
	thread_data *tds = (thread_data*)ctx;

	long long prev_reqs = 0, prev_resps = 0;
	while (1)
	{
		sleep(1);

		long long reqs = 0, resps = 0;
		for (int i = 0;i < NUM_THREADS;++i)
		{
			reqs += tds[i].num_requests;
			resps += tds[i].num_responses;
		}

		long long rps = reqs - prev_reqs, rps2 = resps - prev_resps;

		// If nothing is happening and the pending queue is empty then stop
		if (!rps && !rps2 && reqs == resps)
			break;

		// Latency is very rough here. See RPS instead. The main target of this test was RPS
                printf("Requests per second: %d, Responses per second: %d, Pending requests: %d, Latency: %f ms\n",
                        (int)rps, (int)rps2, (int)(reqs-resps), 1000.0 * (reqs-resps) / rps2);
        	fflush(stdout);

		prev_reqs = reqs;
		prev_resps = resps;
	}
}

int main(int argc, char *argv[])
{
	if (argc < 5)
	{
		printf("Usage: tar_test address:port read|write num_threads num_ops\n");
		return 0;
	}

	test_params tp;
	if (strlen(argv[1]) < 255)
		strcpy(tp.address, argv[1]);
	else
	{
		fprintf(stderr, "Address is too long\n");
		return 1;
	}

	tp.is_read_test = !strcmp(argv[2], "read");
	NUM_THREADS = atoi(argv[3]);
	tp.num_ops = atoi(argv[4]);

        pthread_t pids[NUM_THREADS];

	thread_data tds[NUM_THREADS];

	// Start all the threads
        for (int i = 0;i < NUM_THREADS;++i)
        {
		tds[i].tp = &tp;
		tds[i].num_requests = tds[i].num_responses = 0;
                int r = pthread_create(pids + i, NULL, &do_test, tds + i);
                if (r < 0)
                {
                        fprintf(stderr, "multithread_test: could not create thread, i=%d, r=%d, errno='%s'\n", i, r, strerror(errno));
                        fflush(stderr);
                }
        }

	// Start a timer thread
	pthread_t timer_pid;
	int r = pthread_create(&timer_pid, NULL, &timer_thread, tds);
        if (r < 0)
        {
                fprintf(stderr, "multithread_test: could not create thread, r=%d, errno='%s'\n", r, strerror(errno));
                fflush(stderr);
        }

	// Join all the threads
        for (int i = 0;i < NUM_THREADS;++i)
        {
                void *v;
                int r = pthread_join(pids[i], &v);
                if (r < 0)
                {
                        fprintf(stderr, "multithread_test: could not join thread, i=%d, r=%d, errno='%s'\n", i, r, strerror(errno));
                        fflush(stderr);
                }
        }

        // Join the timer thread
	void *v;
	pthread_join(timer_pid, &v);

	return 0;
}