Commit 16b427d0 authored by Frederic Logier's avatar Frederic Logier
Browse files

use local z_message to remove a memory leak

parent 078eb23a
......@@ -46,7 +46,6 @@ Http_api::Http_api(QString basedirectory, QxtAbstractWebSessionManager * sm, QOb
mongodb_ = Mongodb::getInstance ();
zeromq_ = Zeromq::getInstance ();
z_message = new zmq::message_t(2);
//m_context = new zmq::context_t(1);
z_push_api = new zmq::socket_t(*zeromq_->m_context, ZMQ_PUSH);
......@@ -951,10 +950,10 @@ void Http_api::payload_post(QxtWebRequestEvent* event, QString action)
/****** PUSH API PAYLOAD *******/
qDebug() << "PUSH HTTP API PAYLOAD";
z_message->rebuild(l_payload.objsize());
memcpy(z_message->data(), (char*)l_payload.objdata(), l_payload.objsize());
zmq::message_t z_message(l_payload.objsize());
memcpy(z_message.data(), (char*)l_payload.objdata(), l_payload.objsize());
//z_push_api->send(*z_message, ZMQ_NOBLOCK);
z_push_api->send(*z_message);
z_push_api->send(z_message);
/************************/
}
......
......@@ -74,7 +74,6 @@ private:
QHash <QString, BSONObj> user_bson;
zmq::socket_t *z_push_api;
zmq::message_t *z_message;
Mongodb *mongodb_;
Zeromq *zeromq_;
......
......@@ -25,12 +25,6 @@ Worker_api::Worker_api(QString basedirectory, QObject *parent) : QObject(parent)
mongodb_ = Mongodb::getInstance ();
zeromq_ = Zeromq::getInstance ();
z_message_command = new zmq::message_t(2);
z_message = new zmq::message_t(2);
z_message_publish = new zmq::message_t(2);
z_message_publish_replay = new zmq::message_t(2);
z_receive_api = new zmq::socket_t (*zeromq_->m_context, ZMQ_PULL);
int hwm = 50000;
z_receive_api->setsockopt(ZMQ_SNDHWM, &hwm, sizeof (hwm));
......@@ -135,11 +129,9 @@ void Worker_api::pubsub_payload(bson::bo l_payload)
qDebug() << "Worker_api::publish_payload PUBLISH PAYLOAD";
QByteArray s_payload = payload.toAscii();
z_message_publish->rebuild(s_payload.size()+1);
memcpy(z_message_publish->data(), s_payload.constData(), s_payload.size()+1);
z_publish_api->send(*z_message_publish);
//delete(z_message_publish);
zmq::message_t z_message(s_payload.size()+1);
memcpy(z_message.data(), s_payload.constData(), s_payload.size()+1);
z_publish_api->send(z_message);
/************************/
}
......@@ -178,10 +170,9 @@ void Worker_api::replay_pubsub_payload(bson::bo a_payload)
qDebug() << "Worker_api::publish_payload REPLAY PAYLOAD";
QByteArray s_payload = payload.toAscii();
z_message_publish_replay->rebuild(s_payload.size()+1);
memcpy(z_message_publish_replay->data(), s_payload.constData(), s_payload.size()+1);
z_publish_api->send(*z_message_publish_replay);
zmq::message_t z_message(s_payload.size()+1);
memcpy(z_message.data(), s_payload.constData(), s_payload.size()+1);
z_publish_api->send(z_message);
//delete(z_message_publish_replay);
/************************/
}
......@@ -235,11 +226,9 @@ void Worker_api::get_ftp_users(bson::bo a_payload)
qDebug() << "Worker_api::get_ftp_users SEND PAYLOAD";
QByteArray s_payload = payload.toAscii();
z_message_publish_replay->rebuild(s_payload.size()+1);
memcpy(z_message_publish_replay->data(), s_payload.constData(), s_payload.size()+1);
z_publish_api->send(*z_message_publish_replay);
//delete(z_message_publish_replay);
zmq::message_t z_message(s_payload.size()+1);
memcpy(z_message.data(), s_payload.constData(), s_payload.size()+1);
z_publish_api->send(z_message);
/************************/
}
}
......@@ -276,11 +265,9 @@ void Worker_api::publish_command(QString dest, QString command)
qDebug() << "Worker_api::receive_command SEND COMMAND PAYLOAD";
QByteArray s_payload = payload.toAscii();
z_message_command->rebuild(s_payload.size()+1);
memcpy(z_message_command->data(), s_payload.constData(), s_payload.size()+1);
z_publish_api->send(*z_message_command);
//delete(z_message_publish_replay);
zmq::message_t z_message(s_payload.size()+1);
memcpy(z_message.data(), s_payload.constData(), s_payload.size()+1);
z_publish_api->send(z_message);
/************************/
}
......@@ -315,6 +302,7 @@ void Worker_api::receive_payload()
flush_socket:
zmq::message_t request;
zmq::message_t z_message(2);
bool res = z_receive_api->recv(&request, ZMQ_NOBLOCK);
if (!res && zmq_errno () == EAGAIN) break;
......@@ -554,11 +542,10 @@ void Worker_api::receive_payload()
/****** PUSH API PAYLOAD *******/
qDebug() << "PUSH WORKER CREATE PAYLOAD";
z_message->rebuild(l_payload.objsize());
memcpy(z_message->data(), (char*)l_payload.objdata(), l_payload.objsize());
z_message.rebuild(l_payload.objsize());
memcpy(z_message.data(), (char*)l_payload.objdata(), l_payload.objsize());
//z_push_api->send(*z_message, ZMQ_NOBLOCK);
z_push_api->send(*z_message);
//delete(z_message);
z_push_api->send(z_message);
/************************/
}
......@@ -588,10 +575,10 @@ void Worker_api::receive_payload()
/****** PUSH API PAYLOAD *******/
qDebug() << "PUSH WORKER NEXT PAYLOAD";
z_message->rebuild(l_payload.objsize());
memcpy(z_message->data(), (char*)l_payload.objdata(), l_payload.objsize());
z_message.rebuild(l_payload.objsize());
memcpy(z_message.data(), (char*)l_payload.objdata(), l_payload.objsize());
//z_push_api->send(*z_message, ZMQ_NOBLOCK);
z_push_api->send(*z_message);
z_push_api->send(z_message);
//delete(z_message);
/************************/
}
......
......@@ -41,10 +41,6 @@ private:
zmq::socket_t *z_receive_api;
zmq::socket_t *z_push_api;
zmq::socket_t *z_publish_api;
zmq::message_t *z_message;
zmq::message_t *z_message_publish;
zmq::message_t *z_message_publish_replay;
zmq::message_t *z_message_command;
Mongodb *mongodb_;
Zeromq *zeromq_;
......
......@@ -153,7 +153,6 @@ Zerogw::Zerogw(QString basedirectory, int port, QString ipc_name="", QObject *pa
void Zerogw::init()
{
// Socket to ZPULL class
z_message = new zmq::message_t(2);
z_push_api = new zmq::socket_t(*m_context, ZMQ_PUSH);
int hwm1 = 50000;
......@@ -165,7 +164,6 @@ void Zerogw::init()
// socket from ZEROGW
m_message = new zmq::message_t(2);
m_socket_zerogw = new zmq::socket_t (*m_context, ZMQ_REP);
int hwm = 50;
m_socket_zerogw->setsockopt(ZMQ_SNDHWM, &hwm, sizeof (hwm));
......@@ -216,9 +214,9 @@ void Zerogw::forward_payload_to_zpull(BSONObj payload)
{
/****** PUSH API PAYLOAD *******/
std::cout << "PUSH HTTP API PAYLOAD : " << payload << std::endl;
z_message->rebuild(payload.objsize());
memcpy(z_message->data(), (char*)payload.objdata(), payload.objsize());
z_push_api->send(*z_message, ZMQ_NOBLOCK);
zmq::message_t z_message(payload.objsize());
memcpy(z_message.data(), (char*)payload.objdata(), payload.objsize());
z_push_api->send(z_message, ZMQ_NOBLOCK);
//z_push_api->send(*z_message);
/************************/
}
......@@ -630,10 +628,10 @@ void Api_payload::receive_http_payload()
}
// send response to client
m_message->rebuild(bodyMessage.length() +1);
memcpy(m_message->data(), bodyMessage.toAscii(), bodyMessage.length() +1);
zmq::message_t m_message(bodyMessage.length() +1);
memcpy(m_message.data(), bodyMessage.toAscii(), bodyMessage.length() +1);
m_socket_zerogw->send(*m_message, 0);
m_socket_zerogw->send(m_message, 0);
qDebug() << "returning : " << bodyMessage;
bodyMessage="";
counter=-1;
......@@ -761,11 +759,10 @@ void Api_node::receive_http_payload()
}
else bodyMessage = "BAD REQUEST";
zmq::message_t m_message(bodyMessage.length());
memcpy(m_message.data(), bodyMessage.toAscii(), bodyMessage.length());
m_message->rebuild(bodyMessage.length());
memcpy(m_message->data(), bodyMessage.toAscii(), bodyMessage.length());
m_socket_zerogw->send(*m_message, 0);
m_socket_zerogw->send(m_message, 0);
qDebug() << "returning : " << bodyMessage;
}
counter++;
......@@ -889,10 +886,10 @@ void Api_workflow::receive_http_payload()
mongodb_->Insert("workflows", l_workflow);
}
m_message->rebuild(bodyMessage.length());
memcpy(m_message->data(), bodyMessage.toAscii(), bodyMessage.length());
zmq::message_t m_message(bodyMessage.length());
memcpy(m_message.data(), bodyMessage.toAscii(), bodyMessage.length());
m_socket_zerogw->send(*m_message, 0);
m_socket_zerogw->send(m_message, 0);
qDebug() << "returning : " << bodyMessage;
}
counter++;
......@@ -1062,11 +1059,10 @@ void Api_user::receive_http_payload()
}
bodyMessage = buildResponse("token", str_token);
}
zmq::message_t m_message(bodyMessage.length());
memcpy(m_message.data(), bodyMessage.toAscii(), bodyMessage.length());
m_message->rebuild(bodyMessage.length());
memcpy(m_message->data(), bodyMessage.toAscii(), bodyMessage.length());
m_socket_zerogw->send(*m_message, 0);
m_socket_zerogw->send(m_message, 0);
qDebug() << "returning : " << bodyMessage;
}
counter++;
......@@ -1239,10 +1235,10 @@ void Api_session::receive_http_payload()
}
m_message->rebuild(bodyMessage.length());
memcpy(m_message->data(), bodyMessage.toAscii(), bodyMessage.length());
zmq::message_t m_message(bodyMessage.length());
memcpy(m_message.data(), bodyMessage.toAscii(), bodyMessage.length());
m_socket_zerogw->send(*m_message, 0);
m_socket_zerogw->send(m_message, 0);
qDebug() << "returning : " << bodyMessage;
}
counter++;
......@@ -1383,10 +1379,10 @@ void Api_app::receive_http_payload()
}
m_message->rebuild(bodyMessage.length());
memcpy(m_message->data(), bodyMessage.toAscii(), bodyMessage.length());
zmq::message_t m_message(bodyMessage.length());
memcpy(m_message.data(), bodyMessage.toAscii(), bodyMessage.length());
m_socket_zerogw->send(*m_message, 0);
m_socket_zerogw->send(m_message, 0);
qDebug() << "returning : " << bodyMessage;
}
counter++;
......
......@@ -108,12 +108,7 @@ protected:
QSocketNotifier *check_http_data;
zmq::socket_t *m_socket_zerogw;
QMutex *m_mutex_http;
zmq::message_t *m_message;
zmq::socket_t *z_push_api;
zmq::message_t *z_message;
private slots:
void init();
......
......@@ -32,7 +32,7 @@ Ztracker::Ztracker(zmq::context_t *a_context, QObject *parent) : m_context(a_con
mongodb_ = Mongodb::getInstance();
/***************** SERVICE SOCKET *************/
m_message = new zmq::message_t(2);
//m_message = new zmq::message_t(2);
m_socket = new zmq::socket_t (*m_context, ZMQ_REP);
int hwm = 50000;
......@@ -106,6 +106,7 @@ void Ztracker::receive_payload()
while (true) {
flush_socket:
zmq::message_t request;
zmq::message_t m_message(2);
bool res = m_socket->recv (&request, ZMQ_NOBLOCK);
if (!res && zmq_errno () == EAGAIN) break;
......@@ -195,9 +196,8 @@ void Ztracker::receive_payload()
be worker_id = worker.getField("_id");
mongodb_->Addtoarray("workers", worker_id.wrap(), node);
m_message->rebuild(payload.objsize());
memcpy(m_message->data(), (char*)payload.objdata(), payload.objsize());
m_message.rebuild(payload.objsize());
memcpy(m_message.data(), (char*)payload.objdata(), payload.objsize());
}
else if (payload_action == "watchdog")
{
......@@ -211,15 +211,15 @@ void Ztracker::receive_payload()
mongodb_->Update("workers", bo_node_uuid, workers_node);
BSONObj payload = BSON("status" << "ACK");
m_message->rebuild(payload.objsize());
memcpy(m_message->data(), (char*)payload.objdata(), payload.objsize());
m_message.rebuild(payload.objsize());
memcpy(m_message.data(), (char*)payload.objdata(), payload.objsize());
}
else if (payload_action == "ping")
{
qDebug() << "RECEIVE INIT SOCKET";
BSONObj payload = BSON("status" << "pong");
m_message->rebuild(payload.objsize());
memcpy(m_message->data(), (char*)payload.objdata(), payload.objsize());
m_message.rebuild(payload.objsize());
memcpy(m_message.data(), (char*)payload.objdata(), payload.objsize());
}
/*else
{
......@@ -255,7 +255,7 @@ void Ztracker::receive_payload()
}*/
m_socket->send(*m_message);
m_socket->send(m_message);
//delete(m_message);
}
}
......@@ -1381,7 +1381,7 @@ Zworker_push::Zworker_push(zmq::context_t *a_context, string a_worker, string a_
std::cout << "ADDR : " << addr << std::endl;
z_sender->bind(addr.data());
z_message = new zmq::message_t(2);
//z_message = new zmq::message_t(2);
}
......@@ -1400,9 +1400,9 @@ void Zworker_push::push_payload(bson::bo a_payload)
/***************** PUSH ********************/
std::cout << "Zworker_push::push_payload Sending tasks to workers..." << std::endl;
z_message->rebuild(a_payload.objsize());
memcpy(z_message->data(), (char*)a_payload.objdata(), a_payload.objsize());
bool res = z_sender->send(*z_message, ZMQ_NOBLOCK);
zmq::message_t z_message(a_payload.objsize());
memcpy(z_message.data(), (char*)a_payload.objdata(), a_payload.objsize());
bool res = z_sender->send(z_message, ZMQ_NOBLOCK);
qint32 events = 0;
std::size_t eventsSize = sizeof(events);
......@@ -1452,10 +1452,7 @@ Zstream_push::Zstream_push(zmq::context_t *a_context, QObject *parent) : m_conte
int hwm = 50000;
z_stream->setsockopt(ZMQ_SNDHWM, &hwm, sizeof (hwm));
z_stream->setsockopt(ZMQ_RCVHWM, &hwm, sizeof (hwm));
z_stream->bind("tcp://*:5556");
//z_message = new zmq::message_t(2);
int socket_stream_fd;
size_t socket_size = sizeof(socket_stream_fd);
......@@ -1503,6 +1500,7 @@ void Zstream_push::stream_payload()
flush_socket:
zmq::message_t request;
zmq::message_t z_message(2);
bool res = z_stream->recv (&request, ZMQ_NOBLOCK);
if (!res && zmq_errno () == EAGAIN) break;
......@@ -1556,8 +1554,6 @@ void Zstream_push::stream_payload()
BSONObj b_filename = BSON("filename" << filename);
zmq::message_t z_message(2);
z_message.rebuild(b_filename.objsize());
memcpy ((void *) z_message.data(), (char*)b_filename.objdata(), b_filename.objsize());
z_stream->send (z_message, ZMQ_NOBLOCK);
......@@ -1642,16 +1638,10 @@ void Zstream_push::stream_payload()
//std::cout << "Zstream_push::stream_payload CHUNK toBase64 LEN : " << chunk_length << " size : " << s_chunk_data.size() << std::endl;
zmq::message_t z_message(2);
z_message.rebuild(chunk_data.size());
memcpy((void *) z_message.data(), chunk_data.constData(), chunk_data.size());
std::cout << "Zstream_push::stream_payload MESSAGE LEN : " << z_message.size() << std::endl;
// bool l_res = z_stream->send(&z_message, (chunk_index+1<num_chunck)? ZMQ_SNDMORE | ZMQ_NOBLOCK: 0);
bool l_res = z_stream->send(z_message, (chunk_index+1<num_chunck)? ZMQ_SNDMORE | ZMQ_NOBLOCK: 0);
......@@ -1671,7 +1661,6 @@ void Zstream_push::stream_payload()
{
gfsid = BSON("error" << "filename not found : " + t_payload.getField("filename").str());
zmq::message_t z_message(2);
z_message.rebuild(gfsid.objsize());
memcpy((void *) z_message.data(), gfsid.objdata(), gfsid.objsize());
......@@ -1690,7 +1679,6 @@ void Zstream_push::stream_payload()
else
{
BSONObj b_error = BSON("error" << "action unknown : " + payload_action.toStdString());
zmq::message_t z_message(2);
z_message.rebuild(b_error.objsize());
memcpy ((void *) z_message.data(), (char*)b_error.objdata(), b_error.objsize());
z_stream->send (z_message, ZMQ_NOBLOCK);
......
......@@ -56,7 +56,6 @@ private:
string m_port;
zmq::context_t *m_context;
zmq::socket_t *z_sender;
zmq::message_t *z_message;
QList <BSONObj> m_payload_list;
QMutex *m_mutex;
......@@ -80,7 +79,6 @@ private:
string m_port;
zmq::context_t *m_context;
zmq::socket_t *z_stream;
// zmq::message_t *z_message;
QMutex *m_mutex;
private slots:
......@@ -102,7 +100,7 @@ private:
QHash<QString, Zworker_push*> workers_push;
int get_available_port();
zmq::socket_t *m_socket;
zmq::message_t *m_message;
// zmq::message_t *m_message;
zmq::socket_t *m_data_socket;
zmq::message_t *m_data_message;
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment