Segmentation fault: Error in NdbTransaction

[Raghu DV]

I am running multiple instances of the below process. It was fine for a few hours then it crashes with a segmentation fault. gdb trace show error in NdbTransaction:

Core file:
==================================================================================================================================
gdb bin/NdbTransactions core.3245
GNU gdb (GDB) Red Hat Enterprise Linux 7.6.1-94.el7
Copyright (C) 2013 Free Software Foundation, Inc.
License GPLv3+: GNU GPL version 3 or later http://gnu.org/licenses/gpl.html
This is free software: you are free to change and redistribute it.
There is NO WARRANTY, to the extent permitted by law. Type "show copying"
and "show warranty" for details.
This GDB was configured as "x86_64-redhat-linux-gnu".
For bug reporting instructions, please see:
http://www.gnu.org/software/gdb/bugs/...
Reading symbols from /home/ndb/cvs/ndb/src/bin/NdbTransactions...(no debugging symbols found)...done.
[New LWP 3287]
[New LWP 3288]
[New LWP 3289]
[New LWP 3245]
[New LWP 3264]
[New LWP 3265]
[New LWP 3266]
[New LWP 3267]
[New LWP 3268]
[New LWP 3290]
[Thread debugging using libthread_db enabled]
Using host libthread_db library "/lib64/libthread_db.so.1".
Core was generated by `bin/NdbTransactions 10.0.2.66 1 5'.
Program terminated with signal 6, Aborted.
#0 0x00007f0b2624d1f7 in raise () from /lib64/libc.so.6
Missing separate debuginfos, use: debuginfo-install glibc-2.17-196.el7.x86_64 libgcc-4.8.5-16.el7.x86_64 libstdc++-4.8.5-16.el7.x86_64
(gdb) bt
#0 0x00007f0b2624d1f7 in raise () from /lib64/libc.so.6
#1 0x00007f0b2624e8e8 in abort () from /lib64/libc.so.6
#2 0x00007f0b2770ace2 in NdbTransaction::doSend (this=this@entry=0x1433460)
at /export/home/pb2/build/sb_0-23963488-1498205878.14/rpm/BUILD/mysql-cluster-com-7.5.7/mysql-cluster-com-7.5.7/storage/ndb/src/ndbapi/NdbTransaction.cpp:1132
#3 0x00007f0b27706ac1 in Ndb::sendPrepTrans (this=this@entry=0x7ffd5bf88650, forceSend=forceSend@entry=0)
at /export/home/pb2/build/sb_0-23963488-1498205878.14/rpm/BUILD/mysql-cluster-com-7.5.7/mysql-cluster-com-7.5.7/storage/ndb/src/ndbapi/Ndbif.cpp:1512
#4 0x00007f0b27706dd7 in Ndb::sendPollNdb (this=this@entry=0x7ffd5bf88650, aMillisecondNumber=aMillisecondNumber@entry=360000,
minNoOfEventsToWakeup=minNoOfEventsToWakeup@entry=1, forceSend=forceSend@entry=0)
at /export/home/pb2/build/sb_0-23963488-1498205878.14/rpm/BUILD/mysql-cluster-com-7.5.7/mysql-cluster-com-7.5.7/storage/ndb/src/ndbapi/Ndbif.cpp:1658
#5 0x00007f0b27709d83 in NdbTransaction::executeNoBlobs (this=0x7f0b00000900, aTypeOfExec=<optimized out>, abortOption=<optimized out>, forceSend=0)
at /export/home/pb2/build/sb_0-23963488-1498205878.14/rpm/BUILD/mysql-cluster-com-7.5.7/mysql-cluster-com-7.5.7/storage/ndb/src/ndbapi/NdbTransaction.cpp:585
#6 0x00007f0b2770a449 in NdbTransaction::execute (this=0x7f0b00000900, aTypeOfExec=NdbTransaction::Commit, abortOption=<optimized out>, forceSend=0)
at /export/home/pb2/build/sb_0-23963488-1498205878.14/rpm/BUILD/mysql-cluster-com-7.5.7/mysql-cluster-com-7.5.7/storage/ndb/src/ndbapi/NdbTransaction.cpp:306
#7 0x0000000000402603 in do_insert(Ndb&) ()
#8 0x0000000000404744 in std::thread::_Impl<std::_Bind_simple<void (*(std::reference_wrapper<Ndb>))(Ndb&)> >::_M_run() ()
#9 0x00007f0b26ba82b0 in ?? () from /lib64/libstdc++.so.6
#10 0x00007f0b2814ce25 in start_thread () from /lib64/libpthread.so.0
#11 0x00007f0b2631034d in clone () from /lib64/libc.so.6
(gdb) q
==================================================================================================================================

Final lines of trace before abort
==================================================================================================================================
Delete K:4521056
Update K:4521066 V:4521076
Insert K:4521071 V:4521071
Read K:4521061 V:4521071
Poped from Read Queue: 4521061
Pushed to Delete Queue: 4521056
Poped from Delete Queue: 4521056
Delete K:4521061
Poped from Delete Queue: 4521061
2017-09-20 16:49:04 [NdbApi] ERROR -- WARNING: Timeout in executeNoBlobs() waiting for response from NDB data nodes. This should NEVER occur. You have likely hit a NDB Bug. Please file a bug.
2017-09-20 16:49:04 [NdbApi] ERROR -- Forcibly trying to rollback txn (0x1402ac0) to try to clean up data node resources.
Error in src/NdbTransactions.cc, line: 188, code: 4012, msg: Request ndbd time-out, maybe due to high load or communication problems.
Poped from Update Queue: 4521066
Pushed to Read Queue: 4521066
2017-09-20 16:49:04 [NdbApi] ERROR -- WARNING: Timeout in executeNoBlobs() waiting for response from NDB data nodes. This should NEVER occur. You have likely hit a NDB Bug. Please file a bug.
2017-09-20 16:49:04 [NdbApi] ERROR -- Forcibly trying to rollback txn (0x7f0b080013d0) to try to clean up data node resources.
Error in src/NdbTransactions.cc, line: 136, code: 4012, msg: Request ndbd time-out, maybe due to high load or communication problems.
Pushed to Update Queue: 4521071
Insert K:4521076 V:4521076
Inconsistent theSendStatus = Update K:4521071 V:4521081
==================================================================================================================================
I was running multiple instances of the same process the last line from other traces is as follows
Inconsistent theSendStatus = 1


I have the cluster setup in the following configuration:
Cluster configuration is setup as below in Config.ini
================================================================================>
[ndb_mgmd default]
# Directory for MGM node log files
DataDir=/var/lib/mysql-cluster

[tcp default]

[ndb_mgmd]
#Management Node db1
NodeId=1
HostName=10.0.2.66

[ndbd default]
NoOfReplicas=2 # Number of replicas
DataMemory=256M # Memory allocate for data storage
IndexMemory=128M # Memory allocate for index storage
CrashOnCorruptedTuple=false
StopOnError=false
Diskless=false
#MaxNoOfConcurrentTransactions=10000
#MaxNoOfConcurrentOperations=32768
#Directory for Data Node
DataDir=/var/lib/mysql-cluster
MaxBufferedEpochs=100
TimeBetweenEpochsTimeout=4000

[ndbd]
NodeId=2
#Data Node db2
HostName=10.0.2.56

[ndbd]
NodeId=3
#Data Node db2
HostName=10.0.2.55

[mysqld]
#SQL Node db3
NodeId=10
[mysqld]
#SQL Node db3
NodeId=11
[mysqld]
#SQL Node db3
NodeId=12
[mysqld]
#SQL Node db3
NodeId=13
[mysqld]
#SQL Node db3
NodeId=14
[mysqld]
#SQL Node db3
NodeId=15
[mysqld]
#SQL Node db3
NodeId=16
[mysqld]
#SQL Node db3
NodeId=17
================================================================================>


Source Code
==================================================================================================================================
#include <mysql.h>
#include <mysqld_error.h>
#include <NdbApi.hpp>
// Used for cout
#include <stdio.h>
#include <iostream>
#include <cstdint>
#include <unistd.h>
#include <limits>
#include <climits>
#include <sys/syscall.h>
#include <cstring>
#include<thread>
#include <queue>
#define gettid() syscall(SYS_gettid)

#define PRINT_ERROR(code,msg) \
std::cout << "Error in " << __FILE__ << ", line: " << __LINE__ \
<< ", code: " << code \
<< ", msg: " << msg << "." << std::endl
#define MYSQLERROR(mysql) { \
PRINT_ERROR(mysql_errno(&mysql),mysql_error(&mysql)); }
#define APIERROR(error) { \
PRINT_ERROR(error.code,error.message); }

struct St
{
int key;
int val;
};

void do_insert(Ndb &);
void do_update(Ndb &);
void do_delete(Ndb &);
void do_read(Ndb &);

std::queue<St> UpdateQueue;
std::queue<St> ReadQueue;
std::queue<St> DeleteQueue;
int start=0,step=0, insert_counter=0, read_counter=0, update_counter=0, delete_counter=0;

int main(int argc, char** argv)
{
if (argc != 4)
{
std::cout << "Arguments are <connect_string cluster> <start> <step>.\n";
exit(-1);
}
// ndb_init must be called first
ndb_init();

//char * mysqld_sock = argv[1];
const char *connection_string = argv[1];
start = atoi(argv[2]);
step = atoi(argv[3]);

// Object representing the cluster
Ndb_cluster_connection cluster_connection(connection_string);

// Connect to cluster management server (ndb_mgmd)
if (cluster_connection.connect(4 /* retries */,
5 /* delay between retries */,
1 /* verbose */))
{
std::cout << "Cluster management server was not ready within 30 secs.\n";
exit(-1);
}

// Optionally connect and wait for the storage nodes (ndbd's)
if (cluster_connection.wait_until_ready(30,0) < 0)
{
std::cout << "Cluster was not ready within 30 secs.\n";
exit(-1);
}

Ndb myNdb( &cluster_connection, "ndbtest" );
if (myNdb.init()) APIERROR(myNdb.getNdbError());

St st;
memset(&st, 0, sizeof(St));

auto thread1 = std::thread(do_insert, std::ref(myNdb));
auto thread2 = std::thread(do_update, std::ref(myNdb));
auto thread3 = std::thread(do_read, std::ref(myNdb));
auto thread4 = std::thread(do_delete, std::ref(myNdb));
thread1.join();
thread2.join();
thread3.join();
thread4.join();

ndb_end(0);
return 0;
}
/**************************************************************************
**************************************************************************/
void do_insert(Ndb & myNdb)
{
St st;
st.key = start;
st.val = start;

std::cout << "Insert Thread Started" << std::endl;

double lStartime = time(NULL);
while(lStartime == time(NULL));
double lCurrentSec = time(NULL);

while(1)
{
st.val = st.key += step;
std::cout << "Insert K:" << st.key << " V:" << st.val << std::endl;

if((lCurrentSec + 1) <= time(NULL))
{
lCurrentSec = time(NULL);
std::cout << "INS - pid:" << getpid() << " tid:" << gettid() << " No of Records: " << insert_counter << std::endl;
insert_counter = 0;
}

const NdbDictionary::Dictionary* myDict= myNdb.getDictionary();
const NdbDictionary::Table *myTable= myDict->getTable("KV_STORE");

if (myTable == NULL)
APIERROR(myDict->getNdbError());
NdbTransaction *myTransaction= myNdb.startTransaction();
if (myTransaction == NULL) APIERROR(myNdb.getNdbError());

NdbOperation *myOperation= myTransaction->getNdbOperation(myTable);
if (myOperation == NULL) APIERROR(myTransaction->getNdbError());

myOperation->insertTuple();
myOperation->equal("ATTR1", st.key);
myOperation->setValue("ATTR2", st.val);

if(myTransaction->execute( NdbTransaction::Commit ) != 0)
APIERROR(myTransaction->getNdbError());

myNdb.closeTransaction(myTransaction);

insert_counter++;

UpdateQueue.push (st);
std::cout << "Pushed to Update Queue: " << UpdateQueue.front().key << std::endl;
usleep(10);
}
}
/*****************************************************************
*****************************************************************/
void do_update(Ndb & myNdb)
{
St st;
std::cout << "Update Thread Started" << std::endl;

double lStartime = time(NULL);
while(lStartime == time(NULL));
double lCurrentSec = time(NULL);

while(1)
{
//std::cout << "Update Queue Size: " << UpdateQueue.size() << std::endl;
while (!UpdateQueue.empty())
{
st = UpdateQueue.front();
std::cout << "Update K:" << st.key << " V:" << st.val+10 << std::endl;
if((lCurrentSec + 1) <= time(NULL))
{
lCurrentSec = time(NULL);
std::cout << "UPD - pid:" << getpid() << " tid:" << gettid() << " No of Records: " << update_counter << std::endl;
update_counter = 0;
}

const NdbDictionary::Dictionary* myDict= myNdb.getDictionary();
const NdbDictionary::Table *myTable= myDict->getTable("KV_STORE");

if (myTable == NULL)
APIERROR(myDict->getNdbError());
NdbTransaction *myTransaction= myNdb.startTransaction();
if (myTransaction == NULL) APIERROR(myNdb.getNdbError());

NdbOperation *myOperation= myTransaction->getNdbOperation(myTable);
if (myOperation == NULL) APIERROR(myTransaction->getNdbError());

myOperation->updateTuple();
myOperation->equal( "ATTR1", st.key);
myOperation->setValue( "ATTR2", st.val+10);

if( myTransaction->execute( NdbTransaction::Commit ) == -1 )
APIERROR(myTransaction->getNdbError());

myNdb.closeTransaction(myTransaction);
update_counter++;
std::cout << "Poped from Update Queue: " << UpdateQueue.front().key << std::endl;
UpdateQueue.pop();
ReadQueue.push(st);
std::cout << "Pushed to Read Queue: " << ReadQueue.front().key << std::endl;
}
//sleep(1);
usleep(10);
}
}
/*************************************************
*************************************************/
void do_delete(Ndb & myNdb)
{
St st;
std::cout << "Delete Thread Started" << std::endl;

double lStartime = time(NULL);
while(lStartime == time(NULL));
double lCurrentSec = time(NULL);

while(1)
{
//std::cout << "Delete Queue Size: " << DeleteQueue.size() << std::endl;
while(!DeleteQueue.empty())
{
st = DeleteQueue.front();
std::cout << "Delete K:" << st.key << std::endl;

if((lCurrentSec + 1) <= time(NULL))
{
lCurrentSec = time(NULL);
std::cout << "DEL - pid:" << getpid() << " tid:" << gettid() << " No of Records: " << delete_counter << std::endl;
delete_counter = 0;
}

const NdbDictionary::Dictionary* myDict= myNdb.getDictionary();
const NdbDictionary::Table *myTable= myDict->getTable("KV_STORE");

if (myTable == NULL)
APIERROR(myDict->getNdbError());

NdbTransaction *myTransaction= myNdb.startTransaction();
if (myTransaction == NULL) APIERROR(myNdb.getNdbError());

NdbOperation *myOperation= myTransaction->getNdbOperation(myTable);
if (myOperation == NULL) APIERROR(myTransaction->getNdbError());

myOperation->deleteTuple();
myOperation->equal( "ATTR1", st.key);

if (myTransaction->execute(NdbTransaction::Commit) == -1)
APIERROR(myTransaction->getNdbError());

myNdb.closeTransaction(myTransaction);
delete_counter++;
std::cout << "Poped from Delete Queue: " << DeleteQueue.front().key << std::endl;
DeleteQueue.pop();
}
//sleep(1);
usleep(10);
}
}
/*****************************
*****************************/
void do_read(Ndb & myNdb)
{
St st;
std::cout << "Read Thread Started" << std::endl;

double lStartime = time(NULL);
while(lStartime == time(NULL));
double lCurrentSec = time(NULL);

while(1)
{
//std::cout << "Read Queue Size: " << ReadQueue.size() << std::endl;
while(!ReadQueue.empty())
{
st = ReadQueue.front();
const NdbDictionary::Dictionary* myDict= myNdb.getDictionary();
const NdbDictionary::Table *myTable= myDict->getTable("KV_STORE");

if((lCurrentSec + 1) <= time(NULL))
{
lCurrentSec = time(NULL);
std::cout << "RED - pid:" << getpid() << " tid:" << gettid() << " No of Records: " << read_counter << std::endl;
read_counter = 0;
}

if (myTable == NULL)
APIERROR(myDict->getNdbError());

NdbTransaction *myTransaction= myNdb.startTransaction();
if (myTransaction == NULL) APIERROR(myNdb.getNdbError());

NdbOperation *myOperation= myTransaction->getNdbOperation(myTable);
if (myOperation == NULL) APIERROR(myTransaction->getNdbError());

myOperation->readTuple(NdbOperation::LM_Read);
myOperation->equal("ATTR1", st.key);

NdbRecAttr *myRecAttr= myOperation->getValue("ATTR2", NULL);
if (myRecAttr == NULL) APIERROR(myTransaction->getNdbError());

if(myTransaction->execute( NdbTransaction::Commit ) == -1)
APIERROR(myTransaction->getNdbError());

if (myTransaction->getNdbError().classification == NdbError::NoDataFound)
APIERROR(myTransaction->getNdbError());

myNdb.closeTransaction(myTransaction);

read_counter++;

std::cout << "Read K:" << st.key << " V:" << myRecAttr->u_32_value() << std::endl;
std::cout << "Poped from Read Queue: " << ReadQueue.front().key << std::endl;
ReadQueue.pop();
DeleteQueue.push(st);
std::cout << "Pushed to Delete Queue: " << DeleteQueue.front().key << std::endl;
}
//sleep(1);
usleep(10);
}
}