-name:enable dtm extension on datanodes

lineinfile:

dest:"{{pg_datadir}}/postgresql.conf"

regexp:"^shared_preload_libraries"

line:"shared_preload_libraries = 'pg_dtm'"

line:"{{item}}"

state:present

with_items:

-"shared_preload_libraries = 'pg_dtm'"

-"dtm.host = '{{ groups['nodes'][0] }}'"

-name:restart postgrespro

command:"{{pg_dst}}/bin/pg_ctl restart -w -D {{pg_datadir}} -l {{pg_datadir}}/pg.log"

int nWriters;

int nIterations;

int nAccounts;

int startId;

int diapason;

charconst* isolationLevel;

vector<string> connections;

config() {

nReaders =1;

nWriters =10;

nIterations =1000;

nAccounts =1000;

nAccounts =100000;

isolationLevel ="read committed";

startId =1;

diapason =100000;

}

};

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] =newconnection(cfg.connections[i]);

}

connection *srcCon, *dstCon;

srcCon =newconnection(cfg.connections[t.id % cfg.connections.size()]);

dstCon =newconnection(cfg.connections[(t.id +1) % cfg.connections.size()]);

for (int i =0; i < cfg.nIterations; i++)

{

intsrcCon, dstCon;

intsrcAcc =(random() %((cfg.nAccounts-cfg.nWriters)/cfg.nWriters))*cfg.nWriters + t.id;

srcCon =random() % cfg.connections.size();

dstCon =random() % cfg.connections.size();

}while (srcCon == dstCon);

nontransactionsrcTx(*conns[srcCon]);

nontransactiondstTx(*conns[dstCon]);

intsrcAcc = cfg.startId +random() % cfg.diapason;

intdstAcc =cfg.startId +random() % cfg.diapason;

dstAcc =srcAcc;

srcAcc =tmpAcc;

}

nontransactionsrcTx(*srcCon);

nontransactiondstTx(*dstCon);

xid_t xid =execQuery(srcTx,"select dtm_begin_transaction()");

exec(dstTx,"select dtm_join_transaction(%u)", xid);

i -=1;

continue;

}

pipelinesrcPipe(srcTx);

pipelinedstPipe(dstTx);

srcPipe.insert("commit transaction");

dstPipe.insert("commit transaction");

srcPipe.complete();

dstPipe.complete();

t.proceeded +=1;

}

returnNULL;

intmain (int argc,char* argv[])

{

bool initialize =false;

if (argc ==1){

printf("Use -h to show usage options\n");

return1;

}

for (int i =1; i < argc; i++) {

if (argv[i][0] =='-') {

switch (argv[i][1]) {

case'n':

cfg.nIterations =atoi(argv[++i]);

continue;

case'c':

cfg.connections.push_back(string(argv[++i]));

case's':

cfg.startId =atoi(argv[++i]);

continue;

case'd':

cfg.diapason =atoi(argv[++i]);

continue;

case'l':

cfg.isolationLevel = argv[++i];

case'C':

cfg.connections.push_back(string(argv[++i]));

continue;

case'i':

initialize =true;

printf("Options:\n"

"\t-i\tinitialize datanase\n");

return1;

}

if (cfg.startId + cfg.diapason -1 > cfg.nAccounts) {

printf("startId + diapason should be less that nAccounts. Exiting.\n");

return1;

}

if (initialize) {

initializeDatabase();

printf("%d account inserted\n", cfg.nAccounts);

return0;

}

time_t start =getCurrentTime();

}

time_t elapsed =getCurrentTime() - start;

printf("TPS(updates)=%f, TPS(selects)=%f, aborts=%ld\n", (double)(nWrites*USEC)/elapsed, (double)(nReads*USEC)/elapsed, nAborts);

printf(

"\"accounts\":%d,\"iterations\":%d,\"hosts\":%d}\n",

(double)(nWrites*USEC)/elapsed,

(double)(nReads*USEC)/elapsed,

cfg.nReaders,

cfg.nWriters,

cfg.nAccounts,

cfg.nIterations,

cfg.connections.size()

);

return0;

}

-name:generate connstrings

set_fact:

connstr:"-d 'host={{item}} user=cluster port=15432 dbname=postgres'"

connstr:"-C 'host={{item}} user={{ansible_ssh_user}} port=15432 dbname=postgres'"

with_items:

register:connstrs

set_fact:

connections:"{{ connstrs.results | map(attribute='ansible_facts.connstr') | join }}"

-name:copy transfers binary

copy:src=transfers.linux dest=~/transfers mode=a+x

-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"

-hosts:clients[0]

gather_facts:no

tasks:

-name:fill the databases

shell:"./transfers {{connections}} -f -g"

register:transfers_result

-debug:"var=transfers_result"

-name:init database

shell:"~/pg_cluster/install/bin/dtmbench {{connections}} -a 2000000 -i"

register:init_result

-debug:var=init_result

-hosts:clients

gather_facts:no

tasks:

-local_action:shell echo "Bench started at `date`" >> perf.results

-name:run transfers

shell:"./transfers {{connections}} -s {{runkeys | d('-g -w 60 -m -o -u 10000 -r 0')}} -k {{hostvars[inventory_hostname]['offset']}}"

shell:>

register:transfers_result

environment:

LD_LIBRARY_PATH:"/home/{{ansible_ssh_user}}/pg_cluster/install/lib"

-debug:var=transfers_result

-local_action:'shell echo "{{transfers_result.stdout }}" >> perf.results'

-local_action:shell echo "Bench finished at `date`" >> perf.results