[select] break the select loop if interrupt_on_signal flag is set (#624)

Author: stepanblyschak

common/Makefile.am

@@ -45,7 +45,6 @@ libswsscommon_la_SOURCES = \
     select.cpp                \
     selectableevent.cpp       \
     selectabletimer.cpp       \
-    signalhandlerhelper.cpp   \
     consumertable.cpp         \
     consumertablebase.cpp     \
     consumerstatetable.cpp    \

common/configdb.h

@@ -76,7 +76,7 @@ class ConfigDBConnector_Native : public SonicV2Connector_Native
             ## Start listen Redis keyspace event. Pass a callback function to `init` to handle initial table data.
             self.pubsub = self.get_redis_client(self.db_name).pubsub()
             self.pubsub.psubscribe("__keyspace@{}__:*".format(self.get_dbid(self.db_name)))
-        
+
             # Build a cache of data for all subscribed tables that will recieve the initial table data so we dont send duplicate event notifications
             init_data = {tbl: self.get_table(tbl) for tbl in self.handlers if init_data_handler or self.fire_init_data[tbl]}
 
@@ -88,16 +88,16 @@ class ConfigDBConnector_Native : public SonicV2Connector_Native
                     return False
                 return True
 
-            init_callback_data = {tbl: data for tbl, data in init_data.items() if load_data(tbl, data)}	
+            init_callback_data = {tbl: data for tbl, data in init_data.items() if load_data(tbl, data)}
 
             # Pass all initial data that we DID NOT send as updates to handlers through the init callback if provided by caller
             if init_data_handler:
                 init_data_handler(init_callback_data)
 
-            while not SignalHandlerHelper.checkSignal(SIGNAL_INT):
-                item = self.pubsub.listen_message()
+            while True:
+                item = self.pubsub.listen_message(interrupt_on_signal=True)
                 if 'type' not in item:
-                    # When timeout or cancelled, item will not contains 'type' 
+                    # When timeout or interrupted, item will not contains 'type'
                     continue
 
                 if item['type'] == 'pmessage':

common/pubsub.cpp

@@ -77,12 +77,12 @@ bool PubSub::hasCachedData()
     return m_keyspace_event_buffer.size() > 1;
 }
 
-map<string, string> PubSub::get_message(double timeout)
+map<string, string> PubSub::get_message(double timeout, bool interrupt_on_signal)
 {
-    return get_message_internal(timeout).second;
+    return get_message_internal(timeout, interrupt_on_signal).second;
 }
 
-MessageResultPair PubSub::get_message_internal(double timeout)
+MessageResultPair PubSub::get_message_internal(double timeout, bool interrupt_on_signal)
 {
     MessageResultPair ret;
 
@@ -93,7 +93,7 @@ MessageResultPair PubSub::get_message_internal(double timeout)
     }
 
     Selectable *selected;
-    int rc = m_select.select(&selected, int(timeout));
+    int rc = m_select.select(&selected, int(timeout), interrupt_on_signal);
     ret.first = rc;
     switch (rc)
     {
@@ -128,13 +128,13 @@ MessageResultPair PubSub::get_message_internal(double timeout)
 
 // Note: it is not straightforward to implement redis-py PubSub.listen() directly in c++
 // due to the `yield` syntax, so we implement this function for blocking listen one message
-std::map<std::string, std::string> PubSub::listen_message()
+std::map<std::string, std::string> PubSub::listen_message(bool interrupt_on_signal)
 {
     const double GET_MESSAGE_INTERVAL = 600.0; // in seconds
     MessageResultPair ret;
     for (;;)
     {
-        ret = get_message_internal(GET_MESSAGE_INTERVAL);
+        ret = get_message_internal(GET_MESSAGE_INTERVAL, interrupt_on_signal);
         if (!ret.second.empty() || ret.first == Select::SIGNALINT)
         {
             break;

common/pubsub.h

@@ -18,8 +18,8 @@ class PubSub : protected RedisSelect
 public:
     explicit PubSub(DBConnector *other);
 
-    std::map<std::string, std::string> get_message(double timeout = 0.0);
-    std::map<std::string, std::string> listen_message();
+    std::map<std::string, std::string> get_message(double timeout = 0.0, bool interrupt_on_signal = false);
+    std::map<std::string, std::string> listen_message(bool interrupt_on_signal = false);
 
     void psubscribe(const std::string &pattern);
     void punsubscribe(const std::string &pattern);
@@ -32,7 +32,7 @@ class PubSub : protected RedisSelect
 private:
     /* Pop keyspace event from event buffer. Caller should free resources. */
     std::shared_ptr<RedisReply> popEventBuffer();
-    MessageResultPair get_message_internal(double timeout = 0.0);
+    MessageResultPair get_message_internal(double timeout = 0.0, bool interrupt_on_signal = false);
 
     DBConnector *m_parentConnector;
     Select m_select;

common/select.cpp

@@ -1,7 +1,6 @@
 #include "common/selectable.h"
 #include "common/logger.h"
 #include "common/select.h"
-#include "common/signalhandlerhelper.h"
 #include <algorithm>
 #include <stdio.h>
 #include <sys/time.h>
@@ -89,26 +88,34 @@ void Select::addSelectables(vector<Selectable *> selectables)
     }
 }
 
-int Select::poll_descriptors(Selectable **c, unsigned int timeout)
+int Select::poll_descriptors(Selectable **c, unsigned int timeout, bool interrupt_on_signal = false)
 {
     int sz_selectables = static_cast<int>(m_objects.size());
     std::vector<struct epoll_event> events(sz_selectables);
     int ret;
 
-    do
+    while(true)
     {
         ret = ::epoll_wait(m_epoll_fd, events.data(), sz_selectables, timeout);
-    }
-    while(ret == -1 && errno == EINTR && !SignalHandlerHelper::checkSignal(Signals::SIGNAL_INT)); // Retry the select if the process was interrupted by a signal
-
-    if (SignalHandlerHelper::checkSignal(Signals::SIGNAL_INT))
-    {
-        // Return if the epoll_wait was interrupted by SIGTERM
-        return Select::SIGNALINT;
+        // on signal interrupt check if we need to return
+        if (ret == -1 && errno == EINTR)
+        {
+            if (interrupt_on_signal)
+            {
+                return Select::SIGNALINT;
+            }
+        }
+        // on all other errors break the loop
+        else
+        {
+            break;
+        }
     }
 
     if (ret < 0)
+    {
         return Select::ERROR;
+    }
 
     for (int i = 0; i < ret; ++i)
     {
@@ -156,7 +163,7 @@ int Select::poll_descriptors(Selectable **c, unsigned int timeout)
     return Select::TIMEOUT;
 }
 
-int Select::select(Selectable **c, int timeout)
+int Select::select(Selectable **c, int timeout, bool interrupt_on_signal)
 {
     SWSS_LOG_ENTER();
 
@@ -172,7 +179,7 @@ int Select::select(Selectable **c, int timeout)
         return ret;
 
     /* wait for data */
-    ret = poll_descriptors(c, timeout);
+    ret = poll_descriptors(c, timeout, interrupt_on_signal);
 
     return ret;
 

common/select.h

@@ -30,10 +30,10 @@ class Select
         OBJECT = 0,
         ERROR = 1,
         TIMEOUT = 2,
-        SIGNALINT = 3,// Read operation interrupted by SIGINT
+        SIGNALINT = 3,// Read operation interrupted by a signal
     };
 
-    int select(Selectable **c, int timeout = -1);
+    int select(Selectable **c, int timeout = -1, bool interrupt_on_signal = false);
     bool isQueueEmpty();
 
     /**
@@ -66,7 +66,7 @@ class Select
         }
     };
 
-    int poll_descriptors(Selectable **c, unsigned int timeout);
+    int poll_descriptors(Selectable **c, unsigned int timeout, bool interrupt_on_signal);
 
     int m_epoll_fd;
     std::unordered_map<int, Selectable *> m_objects;

common/signalhandlerhelper.cpp

@@ -19,7 +19,6 @@
 #include "pubsub.h"
 #include "select.h"
 #include "selectable.h"
-#include "signalhandlerhelper.h"
 #include "rediscommand.h"
 #include "table.h"
 #include "countertable.h"
@@ -156,7 +155,6 @@ T castSelectableObj(swss::Selectable *temp)
 %include "pubsub.h"
 %include "selectable.h"
 %include "select.h"
-%include "signalhandlerhelper.h"
 %include "rediscommand.h"
 %include "redispipeline.h"
 %include "redisselect.h"

common/signalhandlerhelper.h

@@ -1,33 +1,34 @@
 import signal
 import os
 import pytest
+import multiprocessing
+import time
 from swsscommon import swsscommon
-from swsscommon.swsscommon import SignalHandlerHelper
-
-def dummy_signal_handler(signum, stack):
-    # ignore signal so UT will not break
-    pass
-
-def test_SignalHandler():
-    signal.signal(signal.SIGUSR1, dummy_signal_handler)
-
-    # Register SIGUSER1
-    SignalHandlerHelper.registerSignalHandler(signal.SIGUSR1)
-    happened = SignalHandlerHelper.checkSignal(signal.SIGUSR1)
-    assert happened == False
-
-    # trigger SIGUSER manually
-    os.kill(os.getpid(), signal.SIGUSR1)
-    happened = SignalHandlerHelper.checkSignal(signal.SIGUSR1)
-    assert happened == True
-    
-    # Reset signal
-    SignalHandlerHelper.resetSignal(signal.SIGUSR1)
-    happened = SignalHandlerHelper.checkSignal(signal.SIGUSR1)
-    assert happened == False
-    
-    # un-register signal handler
-    SignalHandlerHelper.restoreSignalHandler(signal.SIGUSR1)
-    os.kill(os.getpid(), signal.SIGUSR1)
-    happened = SignalHandlerHelper.checkSignal(signal.SIGUSR1)
-    assert happened == False
+
+def test_config_db_listen_while_signal_received():
+    """ Test performs ConfigDBConnector.listen() while signal is received,
+    checks that the listen() call is interrupted and the regular KeyboardInterrupt is raised.
+    """
+    c=swsscommon.ConfigDBConnector()
+    c.subscribe('A', lambda a: None)
+    c.connect(wait_for_init=False)
+    event = multiprocessing.Event()
+
+    def signal_handler(signum, frame):
+        event.set()
+        sys.exit(0)
+
+    signal.signal(signal.SIGUSR1, signal_handler)
+
+    def listen():
+        c.listen()
+
+    thr = multiprocessing.Process(target=listen)
+    thr.start()
+
+    time.sleep(5)
+    os.kill(thr.pid, signal.SIGUSR1)
+
+    thr.join()
+
+    assert event.is_set()