dtmbench deploy on several servers

Author: kelvich

tests/deploy/cluster.yml

@@ -43,3 +43,16 @@
 
   - name: restart postgrespro
     command: "{{pg_dst}}/bin/pg_ctl restart -w -D {{pg_datadir}} -l {{pg_datadir}}/pg.log"
+    environment:
+      LD_LIBRARY_PATH: "{{pg_dst}}/lib/"
+
+
+- hosts: clients
+  tasks:
+  - name: copy pg source
+    copy: src=./{{item}} dest=~/{{item}} mode=0755
+    with_items:
+      - "pg_cluster_install.tgz"
+
+  - name: extract postgres
+    command: "tar xzf pg_cluster_install.tgz"

tests/deploy/hosts

@@ -1,12 +1,12 @@
 [clients]
-158.250.29.4 ansible_ssh_user=cluster offset=6001
-158.250.29.7 ansible_ssh_user=cluster ansible_ssh_port=2299 offset=3001
+158.250.29.4 ansible_ssh_user=cluster offset=600001
+158.250.29.7 ansible_ssh_user=cluster ansible_ssh_port=2299 offset=300001
 
 [nodes]
-158.250.29.5 ansible_ssh_user=cluster offset=5001
-158.250.29.6 ansible_ssh_user=cluster offset=4001
-158.250.29.8 ansible_ssh_user=cluster offset=2001
-158.250.29.9 ansible_ssh_user=cluster offset=1001
+158.250.29.5 ansible_ssh_user=cluster offset=500001
+158.250.29.6 ansible_ssh_user=cluster offset=400001
+158.250.29.8 ansible_ssh_user=cluster offset=200001
+158.250.29.9 ansible_ssh_user=cluster offset=100001
 158.250.29.10 ansible_ssh_user=cluster offset=1
 
 [master]

tests/deploy/roles/postgres/tasks/main.yml

@@ -75,6 +75,8 @@
 
 - name: stop postgres if it was running
   command: "{{pg_dst}}/bin/pg_ctl stop -w -D {{pg_datadir}}"
+  environment:
+    LD_LIBRARY_PATH: "{{pg_dst}}/lib"
   when: pg_pidfile.stat.exists
 
 - name: remove datadirs on datanodes
@@ -83,6 +85,8 @@
 
 - name: create datadirs on datanodes
   command: "{{pg_dst}}/bin/initdb {{pg_datadir}}"
+  environment:
+    LD_LIBRARY_PATH: "{{pg_dst}}/lib/"
   args:
     creates: "{{pg_datadir}}"
 
@@ -106,7 +110,7 @@
     line: "host all all 0.0.0.0/0 trust"
 
 - name: start postgrespro
-  command: "{{pg_dst}}/bin/pg_ctl start -w -D {{pg_datadir}} -l {{pg_datadir}}/pg.log"
-
-
+  shell: "{{pg_dst}}/bin/pg_ctl start -w -D {{pg_datadir}} -l {{pg_datadir}}/pg.log"
+  environment:
+    LD_LIBRARY_PATH: "{{pg_dst}}/lib/"
 

tests/deploy/roles/postgres/vars/main.yml

@@ -8,8 +8,6 @@ pg_version_tag: master
 pg_destroy_and_init: false
 
 pg_port: 5432
-pg_dtm_enable: false
-pg_dtm_host: "127.0.0.1"
 pg_config:
   - line: "shared_buffers = 3GB"
   - line: "wal_keep_segments = 128"
@@ -26,4 +24,3 @@ pg_prefix: "{{ansible_env.HOME}}/pg_cluster"
 pg_src: "{{pg_prefix}}/src"
 pg_dst: "{{pg_prefix}}/install"
 pg_datadir: "{{pg_prefix}}/data_{{pg_port}}"
-

tests/dtmbench.cpp

@@ -59,13 +59,17 @@ struct config
     int nWriters;
     int nIterations;
     int nAccounts;
+    int startId;
+    int diapason;
     vector<string> connections;
 
     config() {
         nReaders = 1;
         nWriters = 10;
         nIterations = 1000;
-        nAccounts = 1000;        
+        nAccounts = 100000;  
+        startId = 1;
+        diapason = 100000;
     }
 };
 
@@ -141,36 +145,41 @@ void* reader(void* arg)
 void* writer(void* arg)
 {
     thread& t = *(thread*)arg;
-    vector< unique_ptr<connection> > conns(cfg.connections.size());
-    for (size_t i = 0; i < conns.size(); i++) {
-        conns[i] = new connection(cfg.connections[i]);
-    }
+    connection *srcCon, *dstCon;
+
+    srcCon = new connection(cfg.connections[t.id % cfg.connections.size()]);
+    dstCon = new connection(cfg.connections[(t.id + 1) % cfg.connections.size()]);
+
     for (int i = 0; i < cfg.nIterations; i++)
     { 
         char gtid[32];
-        int srcCon, dstCon;
-        int srcAcc = (random() % ((cfg.nAccounts-cfg.nWriters)/cfg.nWriters))*cfg.nWriters + t.id;
-        int dstAcc = (random() % ((cfg.nAccounts-cfg.nWriters)/cfg.nWriters))*cfg.nWriters + t.id;
 
-        sprintf(gtid, "%d.%d", t.id, i);
+        // int srcAcc = (random() % ((cfg.nAccounts-cfg.nWriters)/cfg.nWriters))*cfg.nWriters + t.id;
+        // int dstAcc = (random() % ((cfg.nAccounts-cfg.nWriters)/cfg.nWriters))*cfg.nWriters + t.id;
 
-        do { 
-            srcCon = random() % cfg.connections.size();
-            dstCon = random() % cfg.connections.size();
-        } while (srcCon == dstCon);
-        
-        nontransaction srcTx(*conns[srcCon]);
-        nontransaction dstTx(*conns[dstCon]);
+        int srcAcc = cfg.startId + random() % cfg.diapason;
+        int dstAcc = cfg.startId + random() % cfg.diapason;
+
+        if (srcAcc > dstAcc) {
+            int tmpAcc = dstAcc;
+            dstAcc = srcAcc;
+            srcAcc = tmpAcc;
+        }
+
+        sprintf(gtid, "%d.%d.%d", cfg.startId, t.id, i);
+
+        nontransaction srcTx(*srcCon);
+        nontransaction dstTx(*dstCon);
 
         exec(srcTx, "begin transaction");
         exec(dstTx, "begin transaction");
 
         csn_t snapshot = execQuery(srcTx, "select dtm_extend('%s')", gtid);
         snapshot = execQuery(dstTx, "select dtm_access(%ld, '%s')", snapshot, gtid);
-            
+
         exec(srcTx, "update t set v = v - 1 where u=%d", srcAcc);
         exec(dstTx, "update t set v = v + 1 where u=%d", dstAcc);
-        
+
         exec(srcTx, "prepare transaction '%s'", gtid);
         exec(dstTx, "prepare transaction '%s'", gtid);
         exec(srcTx, "select dtm_begin_prepare('%s')", gtid);
@@ -196,17 +205,20 @@ void initializeDatabase()
         exec(txn, "create extension pg_dtm");
         exec(txn, "drop table if exists t");
         exec(txn, "create table t(u int primary key, v int)");
-        exec(txn, "insert into t (select generate_series(0,%d), %d)", cfg.nAccounts-1, 0);
+        exec(txn, "insert into t (select generate_series(0,%d), %d)", cfg.nAccounts, 0);
         txn.commit();
-
-        // nontransaction vacTx(conn);
-        // exec(vacTx, "vacuum full");
     }        
 }
 
 int main (int argc, char* argv[])
 {
     bool initialize = false;
+
+    if (argc == 1){
+        printf("Use -h to show usage options\n");
+        return 1;
+    }
+
     for (int i = 1; i < argc; i++) { 
         if (argv[i][0] == '-') { 
             switch (argv[i][1]) { 
@@ -222,6 +234,12 @@ int main (int argc, char* argv[])
             case 'n':
                 cfg.nIterations = atoi(argv[++i]);
                 continue;
+            case 's':
+                cfg.startId = atoi(argv[++i]);
+                continue;
+            case 'd':
+                cfg.diapason = atoi(argv[++i]);
+                continue;
             case 'C':
                 cfg.connections.push_back(string(argv[++i]));
                 continue;
@@ -233,14 +251,24 @@ int main (int argc, char* argv[])
         printf("Options:\n"
                "\t-r N\tnumber of readers (1)\n"
                "\t-w N\tnumber of writers (10)\n"
-               "\t-a N\tnumber of accounts (1000)\n"
+               "\t-a N\tnumber of accounts (100000)\n"
+               "\t-s N\tperform updates starting from this id (1)\n"
+               "\t-d N\tperform updates in this diapason (100000)\n"
                "\t-n N\tnumber of iterations (1000)\n"
-               "\t-c STR\tdatabase connection string\n"
+               "\t-C STR\tdatabase connection string\n"
                "\t-i\tinitialize datanase\n");
         return 1;
     }
+
+    if (cfg.startId + cfg.diapason - 1 > cfg.nAccounts) {
+        printf("startId + diapason should be less that nAccounts. Exiting.\n");
+        return 1;
+    }
+
     if (initialize) { 
         initializeDatabase();
+        printf("%d account inserted\n", cfg.nAccounts);
+        return 0;
     }
 
     time_t start = getCurrentTime();
@@ -275,7 +303,9 @@ int main (int argc, char* argv[])
 
 
     printf(
-        "{\"update_tps\":%f, \"read_tps\":%f, \"readers\":%d, \"writers\":%d, \"accounts\":%d, \"iterations\":%d, \"hosts\":%d}\n", 
+        "{\"update_tps\":%f, \"read_tps\":%f,"
+        " \"readers\":%d, \"writers\":%d,"
+        " \"accounts\":%d, \"iterations\":%d, \"hosts\":%d}\n",
         (double)(nWrites*USEC)/elapsed,
         (double)(nReads*USEC)/elapsed,
         cfg.nReaders,

tests/perf.yml

@@ -6,7 +6,7 @@
 
   - name: generate connstrings
     set_fact:
-      connstr: "-C 'host={{item}} user=cluster port=15432 dbname=postgres' "
+      connstr: "-C 'host={{item}} user={{ansible_ssh_user}} port=15432 dbname=postgres' "
     with_items:
       groups['nodes'] | reverse | batch(nnodes | d(2) | int) | first
     register: connstrs
@@ -15,25 +15,35 @@
     set_fact:
       connections: "{{ connstrs.results | map(attribute='ansible_facts.connstr') | join }}"
 
-  # - name: copy transfers binary
+  # - name: copy transfers source
   #   copy: src=./{{item}} dest=~/{{item}} mode=0755
   #   with_items:
-  #     - "dtmbench"
-  #     - "libpqxx-4.0.so"
+  #     - "dtmbench.cpp"
 
-  - name: copy transfers source
-    copy: src=./{{item}} dest=~/{{item}} mode=0755
-    with_items:
-      - "dtmbench.cpp"
+  # - name: compile dtmbench
+  #   shell: "g++ -g -Wall -O2 -o dtmbench dtmbench.cpp -lpqxx -lpq -pthread"
+
+  # - name: compile dtmbench
+  #   shell: "mv dtmbench ~/pg_cluster/install/bin/dtmbench"
 
-  - name: compile dtmbench
-    shell: "g++ -g -Wall -O2 -o dtmbench dtmbench.cpp -lpqxx -pthread"
+- hosts: clients[0]
+  gather_facts: no
+  tasks:
+  - name: init database
+    shell: "~/pg_cluster/install/bin/dtmbench {{connections}} -a 1000000 -i"
+    register: init_result
+  - debug: var=init_result
 
+- hosts: clients
+  gather_facts: no
+  tasks:
   - name: run transfers
-    shell: "./dtmbench {{connections}} -w {{item}} -r 1 -n 100 -a 100000 -i "
+    shell: "~/pg_cluster/install/bin/dtmbench {{connections}} -w {{item}} -s {{offset}} -d 100000 -r 1 -n 1000 -a 1000000"
     register: transfers_result
+    environment:
+      LD_LIBRARY_PATH: "/home/{{ansible_ssh_user}}/pg_cluster/install/lib"
     with_sequence: start=100 end=200 stride=100
 
-  - local_action: shell echo "{{transfers_result.results | map(attribute='stdout') | join('\n')}}" >> x.out
+  - local_action: shell echo "{{transfers_result.results | map(attribute='stdout') | join('\n')}}" >> perf.results
 
   - debug: msg="{{ transfers_result.results | map(attribute='stdout') | join('\n') }}"