Commit 8ecca1d7 authored by fred's avatar fred
Browse files

improved threads shutdown with deleteLater, should be now stop each threads properly

parent 2db5aad0
......@@ -93,4 +93,7 @@ void Alert::success()
Alert::~Alert()
{}
{
if (m_smtp) delete(m_smtp);
if (m_message) delete(m_message);
}
No preview for this file type
docs/ncs_dev.png

454 KB | W: | H:

docs/ncs_dev.png

498 KB | W: | H:

docs/ncs_dev.png
docs/ncs_dev.png
docs/ncs_dev.png
docs/ncs_dev.png
  • 2-up
  • Swipe
  • Onion skin
......@@ -90,11 +90,14 @@ Dispatcher::Dispatcher(params ncs_params)
zeromq->init();
connect(zeromq->dispatch, SIGNAL(emit_pubsub(bson::bo)), service->worker_api, SLOT(pubsub_payload(bson::bo)), Qt::QueuedConnection);
QThread *thread_alert = new QThread;
thread_alert = new QThread;
alert = new Alert(ncs_params.alert_email);
alert->moveToThread(thread_alert);
thread_alert->start();
connect(alert, SIGNAL(destroyed()), thread_alert, SLOT(quit()), Qt::DirectConnection);
connect(zeromq->ztracker, SIGNAL(sendAlert(QString)), alert, SLOT(sendEmail(QString)), Qt::QueuedConnection);
alert->moveToThread(thread_alert);
thread_alert->start();
}
Dispatcher::~Dispatcher()
......@@ -136,9 +139,16 @@ void Dispatcher::handleSigHup()
// do Qt stuff
std::cout << "Received SIGHUP" << std::endl;
delete(alert);
/* delete(alert);
delete(service);
delete(zeromq);
delete(zeromq);*/
alert->deleteLater();
thread_alert->wait();
service->deleteLater();
zeromq->deleteLater();
mongodb_->kill ();
snHup->setEnabled(true);
......
......@@ -62,6 +62,8 @@ private:
QSocketNotifier *snHup;
QSocketNotifier *snTerm;
QThread *thread_alert;
};
......
......@@ -105,7 +105,7 @@ Mongodb::Mongodb(QString a_server, QString a_database) : m_server(a_server), m_d
}
connect(this, SIGNAL(forward_payload(BSONObj)), this, SLOT(push_payload(BSONObj)), Qt::DirectConnection);
//connect(this, SIGNAL(forward_payload(BSONObj)), this, SLOT(push_payload(BSONObj)), Qt::DirectConnection);
/*
......
......@@ -23,11 +23,17 @@
ZerogwProxy::ZerogwProxy(params a_ncs_params, int port, QObject *parent) : m_ncs_params(a_ncs_params), m_port(port), QObject(parent)
{
thread = new QThread;
/* thread = new QThread;
connect(thread, SIGNAL(started()), this, SLOT(init()));
connect(thread, SIGNAL(finished()), this, SLOT(end()));
this->moveToThread(thread);
thread->start();
thread->connect(thread, SIGNAL(started()), this, SLOT(init()));
*/
// QObject::connect(this, SIGNAL(destroyed()), this, SLOT(end()), Qt::DirectConnection);
qDebug() << "ZerogwProxy::ZerogwProxy";
zeromq_ = Zeromq::getInstance ();
......@@ -43,8 +49,24 @@ ZerogwProxy::~ZerogwProxy()
zerogw->close();
worker_payload->close();
}
void ZerogwProxy::end()
{
qDebug() << "!!!!!!! ZerogwProxy::clear 0 !!!!!!!!!!!!!!!!!!!!!!!!!!!!!";
sleep(1);
api_payload_thread[0].clear();
qDebug() << "!!!!!!! ZerogwProxy::wait 0 !!!!!!!!!!!!!!!!!!!!!!!!!!!!!";
thread_payload[0]->wait();
qDebug() << "!!!!!!! ZerogwProxy::clear 1 !!!!!!!!!!!!!!!!!!!!!!!!!!!!!";
api_payload_thread[1].clear();
thread_payload[1]->wait();
emit shutdown();
}
void ZerogwProxy::init()
......@@ -87,10 +109,27 @@ void ZerogwProxy::init()
// should be enough for everybody :)
for(int i=0; i<2; i++)
{
api_payload_thread[i] = QSharedPointer<Api_payload> (new Api_payload(m_ncs_params.base_directory, 0, "/payloads_" + QString::number(m_port)));
connect(this, SIGNAL(shutdown()), api_payload_thread[i].data(), SLOT(destructor()), Qt::BlockingQueuedConnection);
thread_payload[i] = new QThread(this);
api_payload_thread[i] = QSharedPointer<Api_payload> (new Api_payload(m_ncs_params.base_directory, 0, "/payloads_" + QString::number(m_port)), &QObject::deleteLater);
connect(thread_payload[i], SIGNAL(started()), api_payload_thread[i].data(), SLOT(init()));
connect(api_payload_thread[i].data(), SIGNAL(destroyed()), thread_payload[i], SLOT(quit()), Qt::DirectConnection);
api_payload_thread[i]->moveToThread(thread_payload[i]);
thread_payload[i]->start();
}
/* thread_api_payload = new QThread(this);
api_payload = new Api_payload(m_ncs_params.base_directory, 0, "/payloads_" + QString::number(m_port));
connect(thread_api_payload, SIGNAL(started()), api_payload, SLOT(init()));
connect(api_payload, SIGNAL(destroyed()), thread_api_payload, SLOT(quit()), Qt::DirectConnection);
api_payload->moveToThread(thread_api_payload);
thread_api_payload->start();
*/
/* qDebug() << "ZerogwProxy::init BEFORE PROXY";
zmq::proxy (*zerogw, *worker_payload, NULL);
qDebug() << "ZerogwProxy::init AFTER PROXY";
......@@ -189,36 +228,84 @@ Service::~Service()
if (m_nodetrack) delete(m_nodetrack);
qDebug() << "delete ftp";
if (m_nodeftp) delete(m_nodeftp);
if (m_nodeftp) {
//m_nodeftp->deleteLater();
delete(m_nodeftp);
node_thread_ftp->quit();
}
qDebug() << "delete xmpp server";
if (m_xmpp_server) delete(m_xmpp_server);
if (m_xmpp_server)
{
m_xmpp_server->deleteLater();
thread_xmpp_server->wait();
}
//if (m_xmpp_server) delete(m_xmpp_server);
qDebug() << "delete xmpp client";
if (m_xmpp_client) delete(m_xmpp_client);
if (m_xmpp_client)
{
m_xmpp_client->deleteLater();
thread_xmpp_client->wait();
}
//if (m_xmpp_client) delete(m_xmpp_client);
qDebug() << "delete worker api";
emit shutdown();
delete(worker_api);
//emit shutdown();
// delete(worker_api);
worker_api->deleteLater();
worker_thread_api->wait();
if (zerogwToPayload[0])
{
delete(zerogwToPayload[0]);
zerogwToPayload[0]->thread->quit();
// delete(zerogwToPayload[0]);
// zerogwToPayload[0]->thread->quit();
//while(!zerogwToPayload[0]->thread->isFinished ()){};
qDebug() << "zerogwToPayload[0]->deleteLater();";
zerogwToPayload[0]->deleteLater();
while(!zerogwToPayload[0]->thread->isFinished ()){};
}
if (zerogwToPayload[1])
{
delete(zerogwToPayload[1]);
/*delete(zerogwToPayload[1]);
zerogwToPayload[1]->thread->quit();
while(!zerogwToPayload[1]->thread->isFinished ()){};
*/
qDebug() << "zerogwToPayload[1]->deleteLater();";
zerogwToPayload[1]->deleteLater();
}
qDebug() << "zerogw_payload->deleteLater();";
zerogw_payload->deleteLater();
thread_ZerogwProxy->wait();
qDebug() << "zerogw_payload2->deleteLater();";
zerogw_payload2->deleteLater();
thread_ZerogwProxy2->wait();
delete(api_node);
qDebug() << "api_node->deleteLater();";
api_node->deleteLater();
thread_api_node->wait();
qDebug() << "api_workflow->deleteLater();";
api_workflow->deleteLater();
thread_api_workflow->wait();
qDebug() << "api_user->deleteLater();";
api_user->deleteLater();
thread_api_user->wait();
/* delete(api_node);
delete(api_workflow);
delete(api_user);
delete(api_user);*/
}
......@@ -246,49 +333,62 @@ void Service::Http_api_init()
m_ncs_params.api_port == 0 ? port = 2502 : port = m_ncs_params.api_port;
zerogwToPayload[0] = new ZerogwProxy(m_ncs_params, port);
zerogwToPayload[1] = new ZerogwProxy(m_ncs_params, port+1);
//zerogwToPayload[0] = new ZerogwProxy(m_ncs_params, port);
// zerogwToPayload[1] = new ZerogwProxy(m_ncs_params, port+1);
thread_ZerogwProxy = new QThread(this);
zerogw_payload = new ZerogwProxy(m_ncs_params, port);
/*
zmq::socket_t zerogw (*zeromq_->m_context, ZMQ_ROUTER);
zerogw.bind ("tcp://*:2503");
zmq::socket_t worker_payload (*zeromq_->m_context, ZMQ_DEALER);
worker_payload.bind ("inproc://workers");
qDebug() << "Service::Http_api_init BEFORE PROXY";
connect(thread_ZerogwProxy, SIGNAL(started()), zerogw_payload, SLOT(init()));
connect(zerogw_payload, SIGNAL(destroyed()), thread_ZerogwProxy, SLOT(quit()), Qt::DirectConnection);
zmq::proxy (zerogw, worker_payload, NULL);
qDebug() << "Service::Http_api_init AFTER PROXY";
zerogw_payload->moveToThread(thread_ZerogwProxy);
thread_ZerogwProxy->start();
thread_ZerogwProxy2 = new QThread(this);
zerogw_payload2 = new ZerogwProxy(m_ncs_params, port+1);
QThread *thread_api_payload = new QThread;
api_payload = new Api_payload(m_ncs_params.base_directory, port);
api_payload->moveToThread(thread_api_payload);
thread_api_payload->start();
connect(thread_ZerogwProxy2, SIGNAL(started()), zerogw_payload2, SLOT(init()));
connect(zerogw_payload2, SIGNAL(destroyed()), thread_ZerogwProxy2, SLOT(quit()), Qt::DirectConnection);
QThread *thread_api_payload2 = new QThread;
api_payload2 = new Api_payload(m_ncs_params.base_directory, port + 101);
api_payload2->moveToThread(thread_api_payload2);
thread_api_payload2->start();
*/
zerogw_payload2->moveToThread(thread_ZerogwProxy2);
thread_ZerogwProxy2->start();
//connect(this, SIGNAL(shutdown()), api_payload, SLOT(destructor()), Qt::BlockingQueuedConnection);
//connect(api_payload, SIGNAL(forward_payload(BSONObj)), dispatch, SLOT(push_payload(BSONObj)), Qt::QueuedConnection);
thread_api_node = new QThread(this);
api_node = new Api_node(m_ncs_params.base_directory, port + 100);
//connect(this, SIGNAL(shutdown()), api_node, SLOT(destructor()), Qt::BlockingQueuedConnection);
//connect(api_node, SIGNAL(forward_payload(BSONObj)), dispatch, SLOT(push_payload(BSONObj)), Qt::QueuedConnection);
connect(thread_api_node, SIGNAL(started()), api_node, SLOT(init()));
connect(api_node, SIGNAL(destroyed()), thread_api_node, SLOT(quit()), Qt::DirectConnection);
api_node->moveToThread(thread_api_node);
thread_api_node->start();
thread_api_workflow = new QThread(this);
api_workflow = new Api_workflow(m_ncs_params.base_directory, port + 101);
//connect(this, SIGNAL(shutdown()), api_workflow, SLOT(destructor()), Qt::BlockingQueuedConnection);
//connect(api_workflow, SIGNAL(forward_payload(BSONObj)), dispatch, SLOT(push_payload(BSONObj)), Qt::QueuedConnection);
connect(thread_api_workflow, SIGNAL(started()), api_workflow, SLOT(init()));
connect(api_workflow, SIGNAL(destroyed()), thread_api_workflow, SLOT(quit()), Qt::DirectConnection);
api_workflow->moveToThread(thread_api_workflow);
thread_api_workflow->start();
thread_api_user = new QThread(this);
api_user = new Api_user(m_ncs_params.base_directory, port + 102);
connect(thread_api_user, SIGNAL(started()), api_user, SLOT(init()));
connect(api_user, SIGNAL(destroyed()), thread_api_user, SLOT(quit()), Qt::DirectConnection);
api_user->moveToThread(thread_api_user);
thread_api_user->start();
}
......@@ -330,12 +430,13 @@ void Service::Nodeftp_init()
int port;
m_ncs_params.ftp_server_port == 0 ? port = 2121 : port = m_ncs_params.ftp_server_port;
QThread *node_ftp = new QThread;
m_nodeftp = new Nodeftp(m_ncs_params.base_directory, port);
m_nodeftp->moveToThread(node_ftp);
node_ftp->start();
node_thread_ftp = new QThread(this);
m_nodeftp = new Nodeftp(m_ncs_params.base_directory, port);
connect(m_nodeftp, SIGNAL(destroyed()), node_thread_ftp, SLOT(quit()), Qt::DirectConnection);
m_nodeftp->connect(node_thread_ftp, SIGNAL(started()), SLOT(ftp_init()));
m_nodeftp->connect(node_ftp, SIGNAL(started()), SLOT(ftp_init()));
m_nodeftp->moveToThread(node_thread_ftp);
node_thread_ftp->start();
}
void Service::Tracker_init()
......@@ -354,19 +455,32 @@ void Service::Xmpp_init()
{
qRegisterMetaType<QXmppLogger::MessageType>("QXmppLogger::MessageType");
thread_xmpp_server = new QThread;
m_xmpp_server = new Xmpp_server(m_ncs_params.domain_name, m_ncs_params.xmpp_client_port, m_ncs_params.xmpp_server_port);
connect(m_xmpp_server, SIGNAL(destroyed()), thread_xmpp_server, SLOT(quit()), Qt::DirectConnection);
m_xmpp_server->moveToThread(thread_xmpp_server);
thread_xmpp_server->start();
thread_xmpp_client = new QThread;
m_xmpp_client = new Xmpp_client(m_ncs_params.base_directory, m_ncs_params.domain_name, m_ncs_params.xmpp_client_port);
connect(m_xmpp_client, SIGNAL(destroyed()), thread_xmpp_client, SLOT(quit()), Qt::DirectConnection);
m_xmpp_client->moveToThread(thread_xmpp_client);
thread_xmpp_client->start();
}
void Service::Worker_init()
{
QThread *worker_pull = new QThread;
worker_thread_api = new QThread;
worker_api = new Worker_api(m_ncs_params.base_directory);
worker_api->moveToThread(worker_pull);
worker_pull->start();
connect(this, SIGNAL(shutdown()), worker_api, SLOT(destructor()), Qt::BlockingQueuedConnection);
connect(worker_api, SIGNAL(destroyed()), worker_thread_api, SLOT(quit()), Qt::DirectConnection);
worker_api->moveToThread(worker_thread_api);
worker_thread_api->start();
//connect(this, SIGNAL(shutdown()), worker_api, SLOT(destructor()), Qt::BlockingQueuedConnection);
}
......
......@@ -46,15 +46,17 @@ class ZerogwProxy : public QObject
public:
ZerogwProxy(params a_ncs_params, int port, QObject *parent = 0);
~ZerogwProxy();
QThread *thread;
private slots:
void init();
void end();
signals:
void shutdown();
private:
QThread *thread;
QThread *thread_payload[2];
QSocketNotifier *check_zerogw;
QSocketNotifier *check_reply;
int m_port;
......@@ -88,14 +90,33 @@ public:
Worker_api *worker_api;
ZerogwProxy *zerogwToPayload[2];
ZerogwProxy *zerogw_payload;
ZerogwProxy *zerogw_payload2;
Api_node *api_node;
Api_workflow *api_workflow;
Api_user *api_user;
private:
QThread *thread_ZerogwProxy;
QThread *thread_ZerogwProxy2;
Http_admin *m_http_admin;
QThread *thread_api_node;
QThread *thread_api_workflow;
QThread *thread_api_user;
QThread *thread_xmpp_server;
QThread *thread_xmpp_client;
QThread *node_thread_ftp;
Nodeftp *m_nodeftp;
QThread *worker_thread_api;
Http_admin *m_http_admin;
Http_api *m_http_api;
Xmpp_server *m_xmpp_server;
Xmpp_client *m_xmpp_client;
......
......@@ -511,12 +511,30 @@ void Worker_api::receive_payload()
Worker_api::~Worker_api()
{}
{
qDebug() << "Worker_api destructor";
check_payload->setEnabled(false);
z_receive_api->close ();
z_push_api->close();
z_publish_api->close ();
qDebug() << "Worker_api delete z_receive_api socket";
delete(z_receive_api);
qDebug() << "Worker_api delete z_push_api socket";
delete(z_push_api);
qDebug() << "Worker_api delete z_publish_api socket";
delete(z_publish_api);
}
void Worker_api::destructor()
{
qDebug() << "Worker_api destructor";
/* qDebug() << "Worker_api destructor";
check_payload->setEnabled(false);
......@@ -532,5 +550,5 @@ void Worker_api::destructor()
delete(z_push_api);
qDebug() << "Worker_api delete z_publish_api socket";
delete(z_publish_api);
delete(z_publish_api);*/
}
......@@ -21,7 +21,7 @@
#include "xmpp_client.h"
Xmpp_client::Xmpp_client(QString basedirectory, QString a_domain, int a_xmpp_client_port, QObject *parent) : m_basedirectory(basedirectory), QXmppClient(parent), m_domain(a_domain), m_xmpp_client_port(a_xmpp_client_port)
Xmpp_client::Xmpp_client(QString basedirectory, QString a_domain, int a_xmpp_client_port, QObject *parent) : m_basedirectory(basedirectory), m_domain(a_domain), m_xmpp_client_port(a_xmpp_client_port), QXmppClient(parent)
{
qDebug() << "Xmpp_client::Xmpp_client !!!";
......
......@@ -26,11 +26,14 @@ Zerogw::Zerogw(QString basedirectory, int port, QString ipc_name="", QObject *pa
{
std::cout << "Zerogw::Zerogw construct" << std::endl;
thread = new QThread;
// thread = new QThread;
// connect(thread, SIGNAL(started()), this, SLOT(init()));
// connect(thread, SIGNAL(finished()), thread, SLOT(deleteLater()));
// moveToThread(thread);
// thread->start();
this->moveToThread(thread);
thread->start();
thread->connect(thread, SIGNAL(started()), this, SLOT(init()));
mongodb_ = Mongodb::getInstance();
zeromq_ = Zeromq::getInstance ();
......@@ -87,6 +90,14 @@ void Zerogw::init()
Zerogw::~Zerogw()
{
qDebug() << "Zerogw destruct";
check_http_data->setEnabled(false);
z_push_api->close();
m_socket_zerogw->close();
qDebug() << "Zerogw DELETE destruct";
delete(m_socket_zerogw);
delete(z_push_api);
}
......@@ -94,9 +105,7 @@ Zerogw::~Zerogw()
void Zerogw::destructor()
{
qDebug() << "Zerogw destructor";
check_http_data->setEnabled(false);
z_push_api->close();
m_socket_zerogw->close();
}
......
......@@ -22,7 +22,7 @@
#include <fstream>
Ztracker::Ztracker(zmq::context_t *a_context) : m_context(a_context)
Ztracker::Ztracker(zmq::context_t *a_context, QObject *parent) : m_context(a_context), QObject(parent)
{
std::cout << "Ztracker::Ztracker construct" << std::endl;
......@@ -372,16 +372,7 @@ void Ztracker::worker_update_ticker()
Ztracker::~Ztracker()
{
qDebug() << "Ztracker deleted socket";
delete(m_socket);
}
void Ztracker::destructor()
{
m_mutex->lock();
qDebug() << "Ztracker destructor";
qDebug() << "Ztracker destruct";
check_tracker->setEnabled(false);
qDebug() << "Ztracker worker_timer stop";
......@@ -389,207 +380,15 @@ void Ztracker::destructor()
qDebug() << "Ztracker close socket";
m_socket->close ();
}
Zapi::Zapi(zmq::context_t *a_context) : m_context(a_context)
{
std::cout << "Zapi::Zapi constructeur" << std::endl;
m_mutex = new QMutex();
mongodb_ = Mongodb::getInstance();
m_message = new zmq::message_t(2);
m_socket_http = new zmq::socket_t (*m_context, ZMQ_REP);
int hwm = 50000;
m_socket_http->setsockopt(ZMQ_SNDHWM, &hwm, sizeof (hwm));
m_socket_http->setsockopt(ZMQ_RCVHWM, &hwm, sizeof (hwm));
m_socket_http->bind("tcp://*:2503");
int http_socket_fd;
size_t socket_size = sizeof(http_socket_fd);
m_socket_http->getsockopt(ZMQ_FD, &http_socket_fd, &socket_size);
qDebug() << "RES getsockopt : " << "res" << " FD : " << http_socket_fd << " errno : " << zmq_strerror (errno);
check_http_data = new QSocketNotifier(http_socket_fd, QSocketNotifier::Read, this);
connect(check_http_data, SIGNAL(activated(int)), this, SLOT(receive_http_payload()), Qt::DirectConnection);
}
Zapi::~Zapi()
{
qDebug() << "Zapi::~Zapi";
delete(m_socket_http);
}
void Zapi::receive_http_payload()
{
m_mutex->lock();
check_http_data->setEnabled(false);
std::cout << "Zapi::receive_payload" << std::endl;
QHash <QString, QString> zerogw;
QString key;
int counter=0;
while (true)
{
zmq::message_t request;
bool res = m_socket_http->recv(&request, ZMQ_NOBLOCK);
if (!res && zmq_errno () == EAGAIN) break;
qint32 events = 0;
std::size_t eventsSize = sizeof(events);
m_socket_http->getsockopt(ZMQ_RCVMORE, &events, &eventsSize);
std::cout << "Zapi::receive_payload received request: [" << (char*) request.data() << "]" << std::endl;
BSONObj data;
try {
data = mongo::fromjson(QString::fromAscii((char*)request.data()).toStdString());
if (data.isValid() && !data.isEmpty())
{
std::cout << "Zapi received : " << res << " data : " << <