Segfault while calling redisasynccommand inside an epoll loop

[vaishalijhalani]

Running with GDB i am getting error:
Program received signal SIGSEGV, Segmentation fault.
sdscatlen (s=0x0, t=t@entry=0x609950, len=len@entry=40)
at sds.c:239
239 size_t curlen = sdslen(s);

My code is:

#include "redis_nb.h"

#define MAXCON 8192
#define MAXEVENTS 1024
#define DATASTORE_IP "10.129.28.54"

int pkt_sent = 0;
int port_nos = 6440;

//unordered_map <UeContext, string> ue_ctx;
//unordered_map <uint64_t, string> guti_ueid;

void SignalHandler(int signum) {
fflush(stdout);
}

struct Getcontext {
string key;
int send_sockid;
UeContext Context;
};

struct Getguti {
string key;
int send_sockid;
uint64_t value;
};

int write_stream(int conn_fd, uint8_t *buf, int len) {
int ptr;
int retval;
int written_bytes;
int remaining_bytes;

ptr = 0;
remaining_bytes = len;
if (conn_fd < 0 || len <= 0) {
    return -1;
}
while (1) {
    written_bytes = write(conn_fd, buf + ptr, remaining_bytes);
    if (written_bytes <= 0) {
        retval = written_bytes;
        break;
    }
    ptr += written_bytes;
    remaining_bytes -= written_bytes;
    if (remaining_bytes == 0) {
        retval = len;
        break;
    }
}
return retval;

}

int make_socket_nb(int sfd) {
int flags, s;

flags = fcntl(sfd, F_GETFL, 0);
if (flags == -1) {
    printf("Error: NBS fcntl\n");
    return -1;
}

flags |= O_NONBLOCK;
s = fcntl(sfd, F_SETFL, flags);
if (s == -1) {
    printf("Error: NBS fcntl flags\n");
    return -1;
}

return 0;

}

void getCallbackContext(redisAsyncContext *c, void *r, void *privdata) {
redisAsyncCommand(c, (redisCallbackFn *) NULL, NULL, "EXEC");
redisReply *reply = (redisReply *) r;
int j, ret;
if (reply == NULL) return;
Getcontext *val1 = (Getcontext *) privdata;
string rep = reply->str;
Packet pkt1;
string cmd = "get_context";
pkt1.clear_pkt();
pkt1.append_item(cmd);
//pkt1.append_item(val1->send_sockid);
pkt1.append_item(val1->key);
pkt1.append_item(rep);
//pkt1.prepend_len();
//ue_ctx.erase((UeContext)rep);
ret = write_stream(val1->send_sockid, pkt1.data, pkt1.len); //sep25 //uncomment oct25
if (ret < 0) {
cout << "Error: getcallbackcontext_write" << endl;
exit(-1);
}
//pkt_sent++;
redisAsyncDisconnect(c); //sep20 //oct4
}

void getCallbackGuti(redisAsyncContext *c, void *r, void *privdata) {
redisAsyncCommand(c, (redisCallbackFn *) NULL, NULL, "EXEC");
redisReply *reply = (redisReply *) r;
int j, ret;
if (reply == NULL) return;
Getguti *val1 = (Getguti *) privdata;
string rep = reply->str;
Packet pkt1;
string cmd = "get_guti";
pkt1.clear_pkt();
pkt1.append_item(cmd);
//pkt1.append_item(val1->send_sockid);
pkt1.append_item(val1->key);
pkt1.append_item(rep);
//pkt1.prepend_len();
//guti_ueid.erase((uint64_t)rep);
ret = write_stream(val1->send_sockid, pkt1.data, pkt1.len); //sep25 //uncomment oct25
if (ret < 0) {
cout << "Error: getCallbackGuti_write" << endl;
exit(-1);
}
//pkt_sent++;
redisAsyncDisconnect(c); //sep20 //oct4
}

void setCallback1(redisAsyncContext *c, void *r, void *privdata) {
redisAsyncCommand(c, (redisCallbackFn *) NULL, NULL, "EXEC");
redisAsyncDisconnect(c);
}

void connectCallback(const redisAsyncContext *c, int status) {
if (status != REDIS_OK) {
printf("Error: %s\n", c->errstr);
return;
}
printf("Connected...\n");
}

void disconnectCallback(const redisAsyncContext *c, int status) {
if (status != REDIS_OK) {
printf("Error: %s\n", c->errstr);
return;
}
printf("Disconnected...\n");
}

int read_stream(int conn_fd, uint8_t *buf, int len) {
int ptr;
int retval;
int read_bytes;
int remaining_bytes;

ptr = 0;
remaining_bytes = len;
if (conn_fd < 0 || len <= 0) {
    return -1;
}
while (1) {
    read_bytes = read(conn_fd, buf + ptr, remaining_bytes);
    if (read_bytes <= 0) {
        retval = read_bytes;
        break;
    }
    ptr += read_bytes;
    remaining_bytes -= read_bytes;
    if (remaining_bytes == 0) {
        retval = len;
        break;
    }
}
return retval;

}

int main(int argc, char **argv)
{

signal(SIGPIPE, SIG_IGN);
signal(SIGINT, SignalHandler);
struct event_base *base = event_base_new();

redisAsyncContext *c = redisAsyncConnect("10.129.28.54", 6379);
if (c->err) {
    /* Let *c leak for now... */
    printf("Error: %s\n", c->errstr);
    return 1;
}

redisLibeventAttach(c, base);
redisAsyncSetConnectCallback(c, (redisConnectCallback *) connectCallback);
redisAsyncSetDisconnectCallback(c, (redisConnectCallback *) disconnectCallback);

printf("callback to be called\n");

event_base_dispatch(base);
int lsfd, acfd, data, n, numev, i, ccfd, cafd, cret, trf, loop_count = 0, redis_ret;
char buf[110], buf1[110];
//int buf_sockid, buf_key;
string buf_cmd;
char *buf2;
redisReply *reply;
long long transactions = 0;

int count, tcount;

struct sockaddr_in server, c_addr;
struct hostent *c_ip;
int g_reuse = 1;


lsfd = socket(AF_INET, SOCK_STREAM, 0);
if (lsfd < 0) {
    printf("ERROR : opening socket\n");
    exit(-1);
}

setsockopt(lsfd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &g_reuse, sizeof(int));

bzero((char *) &server, sizeof(server));

server.sin_family = AF_INET;
server.sin_addr.s_addr = inet_addr(DATASTORE_IP);
server.sin_port = htons(port_nos);

if (bind(lsfd, (struct sockaddr *) &server, sizeof(server)) < 0) {
    printf("ERROR: BIND ERROR\n");
    exit(-1);
}

make_socket_nb(lsfd);

listen(lsfd, MAXCON);

int epfd = epoll_create(MAXEVENTS + 5);
if (epfd == -1) {
    printf("Error: epoll create\n");
    exit(-1);
}

int retval;
struct epoll_event ev, *rev;

ev.data.fd = lsfd;
ev.events = EPOLLIN | EPOLLET;

retval = epoll_ctl(epfd, EPOLL_CTL_ADD, lsfd, &ev);
if (retval == -1) {
    printf("Error: epoll ctl lsfd add\n");
    exit(-1);
}
rev = (struct epoll_event *) calloc(MAXEVENTS, sizeof(struct epoll_event));   //sep20
printf("Entering Loop\n");
count = 0;
tcount = 0;

trf = 0;
transactions = 0;
while (1) 
{

    numev = epoll_wait(epfd, rev, MAXEVENTS, -1);
    if (numev < 0) {
        printf("Error: EPOLL wait!\n");
        exit(-1);
    }
    for (i = 0; i < numev; i++) {
        trf = 1;
        if ((rev[i].events & EPOLLERR) || (rev[i].events & EPOLLHUP)) 
        {
                printf("ERROR: epoll monitoring failed, closing fd\n");
                if (rev[i].data.fd == lsfd) {
                    printf("lsfd error\n");
                    exit(-1);
                }
                
                close(rev[i].data.fd);
                continue;
        } 

        else if (rev[i].data.fd == lsfd) 
        {
            
                acfd = accept(lsfd, NULL, NULL);

                if (acfd < 0) {

                    if ((errno == EAGAIN) || (errno == EWOULDBLOCK)) {
                    } else
                        printf("Error on accept\n");
                    break;
                }

                make_socket_nb(acfd);
                ev.data.fd = acfd;
                ev.events = EPOLLIN | EPOLLET;
                retval = epoll_ctl(epfd, EPOLL_CTL_ADD, acfd, &ev);
                if (retval == -1) {
                    printf("Error: epoll ctl lsfd add\n");
                    exit(-1);
                }

                printf("client accepted\n");

        } 

        else if (rev[i].events & EPOLLIN) 
        {

                    Packet pkt;
                    int pkt_len, retval;

                    pkt.clear_pkt();
                    retval = read_stream(rev[i].data.fd, pkt.data, sizeof(int));
                    if (retval < 0) {
                        if (errno == EAGAIN) {
                            break;
                        }
                    } 
                    else 
                    {
                        memmove(&pkt_len, pkt.data, sizeof(int) * sizeof(uint8_t));
                        cout << "pkt_len " << pkt_len << endl;
                        pkt.clear_pkt();
                        retval = read_stream(rev[i].data.fd, pkt.data, pkt_len);
                        pkt.data_ptr = 0;
                        pkt.len = retval;
                        if (retval < 0) {
                            TRACE(cout << "Error: in epoll" << endl;)
                            break;
                        }
                        cout << "retval " << retval << endl;

                    }

                    cout << "buf " << strlen(c->obuf) << endl;
                    int x = redisAsyncCommand(c, NULL, NULL, "SET %s %s", "foo", "hello world");
        }
    }
}

return 0;

}

Getting error on redisAsyncCommand function.