Merge pull request #347 from xmidt-org/4.1_pb

4.1 pb

CHANGELOG.md

@@ -17,6 +17,13 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 - Add pause/resume heartBeatTimer
 - parodus event handler to listen to interface_down and interface_up event
 - Pause connection retry during interface_down event
+- Add callback handler for ping status change event
+- Fixed nopoll_conn_unref crash
+- Update retry timestamp in connection-health-file
+- fix so that parodus can be killed, even if in a connection fail loop
+- provide signal handlers so we shut down properly when INCLUDE_BREAKPAD active
+- send status code and reason in websocket close message
+- dont try to install handler for signal 9 
 
 ## [1.0.2] - 2019-02-08
 - Refactored connection.c and updated corresponding unit tests

src/ParodusInternal.h

@@ -49,6 +49,7 @@
 /*----------------------------------------------------------------------------*/
 /*                                   Macros                                   */
 /*----------------------------------------------------------------------------*/
+
 #define UNUSED(x) (void )(x)
 #define NANO_SOCKET_SEND_TIMEOUT_MS                     2000
 #define NANO_SOCKET_RCV_TIMEOUT_MS			500
@@ -107,6 +108,7 @@ typedef struct {
 
 //--- Used in connection.c for backoff delay timer
 typedef struct {
+  struct timespec ts;
   int count;
   int max_count;
   int delay;

src/conn_interface.c

@@ -168,6 +168,10 @@ void createSocketConnection(void (* initKeypress)())
         if( false == seshat_registered ) {
             seshat_registered = __registerWithSeshat();
         }
+        
+        if (get_interface_down_event ())
+          if (0 != wait_while_interface_down ())
+            break;
 
         if(get_close_retry())
         {
@@ -214,13 +218,18 @@ void createSocketConnection(void (* initKeypress)())
 
     deleteAllClients ();
 
+    ParodusInfo ("reconnect reason at close %s\n", get_global_reconnect_reason());
+    ParodusInfo ("shutdown reason at close %s\n", get_global_shutdown_reason()); 
     close_and_unref_connection(get_global_conn());
     nopoll_ctx_unref(ctx);
     nopoll_cleanup_library();
     curl_global_cleanup();
 }
 
-void shutdownSocketConnection(void) {
+void shutdownSocketConnection(char *reason) {
+   set_global_shutdown_reason (reason);
    g_shutdown = true;
+   reset_interface_down_event ();
+   terminate_backoff_delay ();
 }
 

src/conn_interface.h

@@ -45,7 +45,7 @@ extern UpStreamMsg *UpStreamMsgQ;
  *  and creates the intial connection and manages the connection wait, close mechanisms.
  */
 void createSocketConnection(void (* initKeypress)());
-void shutdownSocketConnection(void);
+void shutdownSocketConnection(char *reason);
    
 #ifdef __cplusplus
 }

src/connection.c

@@ -38,11 +38,34 @@
 
 #define HTTP_CUSTOM_HEADER_COUNT                    	5
 #define INITIAL_CJWT_RETRY                    	-2
+#define UPDATE_HEALTH_FILE_INTERVAL_SECS		450
+
+/* Close codes defined in RFC 6455, section 11.7. */
+enum {
+	CloseNormalClosure           = 1000,
+	CloseGoingAway               = 1001,
+	CloseProtocolError           = 1002,
+	CloseUnsupportedData         = 1003,
+	CloseNoStatus		     = 1005,
+	CloseAbnormalClosure         = 1006,
+	CloseInvalidFramePayloadData = 1007,
+	ClosePolicyViolation         = 1008,
+	CloseMessageTooBig           = 1009,
+	CloseMandatoryExtension      = 1010,
+	CloseInternalServerErr       = 1011,
+	CloseServiceRestart          = 1012,
+	CloseTryAgainLater           = 1013,
+	CloseTLSHandshake            = 1015
+};
 
 /*----------------------------------------------------------------------------*/
 /*                            File Scoped Variables                           */
 /*----------------------------------------------------------------------------*/
 
+pthread_mutex_t backoff_delay_mut=PTHREAD_MUTEX_INITIALIZER;
+pthread_cond_t backoff_delay_con=PTHREAD_COND_INITIALIZER;
+
+static char *shutdown_reason = SHUTDOWN_REASON_PARODUS_STOP;  /* goes in the close message */
 static char *reconnect_reason = "webpa_process_starts";
 static int cloud_disconnect_max_time = 5;
 static noPollConn *g_conn = NULL;
@@ -65,6 +88,16 @@ void set_global_conn(noPollConn *conn)
     g_conn = conn;
 }
 
+char *get_global_shutdown_reason()
+{
+    return shutdown_reason;
+}
+
+void set_global_shutdown_reason(char *reason)
+{
+    shutdown_reason = reason;
+}
+
 char *get_global_reconnect_reason()
 {
     return reconnect_reason;
@@ -199,6 +232,14 @@ void init_backoff_timer (backoff_timer_t *timer, int max_count)
   timer->count = 1;
   timer->max_count = max_count;
   timer->delay = 1;
+  clock_gettime (CLOCK_REALTIME, &timer->ts);
+}
+
+void terminate_backoff_delay (void)
+{
+	pthread_mutex_lock (&backoff_delay_mut);
+	pthread_cond_signal(&backoff_delay_con);
+	pthread_mutex_unlock (&backoff_delay_mut);
 }
 
 int update_backoff_delay (backoff_timer_t *timer)
@@ -211,18 +252,50 @@ int update_backoff_delay (backoff_timer_t *timer)
   return timer->delay;
 }  
 
-static void backoff_delay (backoff_timer_t *timer)
-{
-  update_backoff_delay (timer);
+#define BACKOFF_ERR -1
+#define BACKOFF_SHUTDOWN   1
+#define BACKOFF_DELAY_TAKEN 0
 
-  // Update retry time for conn progress
-  if(timer->count == timer->max_count) 
-  {
-	start_conn_in_progress();
+void start_conn_in_progress (void);
+
+/* backoff_delay
+ * 
+ * delays for the number of seconds specified in parameter timer
+ * g_shutdown can break out of the delay.
+ *
+ * returns -1 pthread_cond_timedwait error
+ *  1   shutdown
+ *  0    delay taken
+*/  
+static int backoff_delay (backoff_timer_t *timer)
+{
+  struct timespec ts;
+  int rtn;
+
+  // periodically update the health file.
+  clock_gettime (CLOCK_REALTIME, &ts);
+  if ((ts.tv_sec - timer->ts.tv_sec) >= UPDATE_HEALTH_FILE_INTERVAL_SECS) {
+    start_conn_in_progress ();
+    timer->ts.tv_sec += UPDATE_HEALTH_FILE_INTERVAL_SECS;
+  }	  
+
+  update_backoff_delay (timer);
+  ParodusInfo("Waiting with backoffRetryTime %d seconds\n", timer->delay);
+  ts.tv_sec += timer->delay;
+
+  pthread_mutex_lock (&backoff_delay_mut);
+  // The condition variable will only be set if we shut down.
+  rtn = pthread_cond_timedwait (&backoff_delay_con, &backoff_delay_mut, &ts);
+  pthread_mutex_unlock (&backoff_delay_mut);
+
+  if (g_shutdown)
+    return BACKOFF_SHUTDOWN;
+  if ((rtn != 0) && (rtn != ETIMEDOUT)) {
+    ParodusError ("pthread_cond_timedwait error (%d) in backoff_delay.\n", rtn);
+    return BACKOFF_ERR;
   }
 
-  ParodusInfo("Waiting with backoffRetryTime %d seconds\n", timer->delay);
-  sleep (timer->delay);
+  return BACKOFF_DELAY_TAKEN;
 }  
 
 //--------------------------------------------------------------------
@@ -436,7 +509,7 @@ int wait_connection_ready (create_connection_ctx_t *ctx)
   if(wait_status == 307 || wait_status == 302 || wait_status == 303)    // only when there is a http redirect
   {
 	char *redirect_ptr = redirectURL;
-	ParodusError("Received temporary redirection response message %s\n", redirectURL);
+	ParodusInfo("Received temporary redirection response message %s\n", redirectURL);
 	// Extract server Address and port from the redirectURL
 	if (strncmp (redirect_ptr, "Redirect:", 9) == 0)
 	    redirect_ptr += 9;
@@ -527,7 +600,7 @@ int keep_trying_to_connect (create_connection_ctx_t *ctx,
 {
     int rtn;
     
-    while (true)
+    while (!g_shutdown)
     {
       set_extra_headers (ctx);
 
@@ -537,13 +610,50 @@ int keep_trying_to_connect (create_connection_ctx_t *ctx,
 
       if (rtn == CONN_WAIT_ACTION_RETRY) // if redirected or build_headers
         continue;
-      backoff_delay (backoff_timer); // 3,7,15,31 ..
+      // If interface down event is set, stop retry
+      // and wait till interface is up again.
+      if(get_interface_down_event()) {
+	    if (0 != wait_while_interface_down())
+	      return false;
+	    start_conn_in_progress();
+	    ParodusInfo("Interface is back up, re-initializing the convey header\n");
+	    // Reset the reconnect reason by initializing the convey header again
+	    ((header_info_t *)(&ctx->header_info))->conveyHeader = getWebpaConveyHeader();
+	    ParodusInfo("Received reconnect_reason as:%s\n", reconnect_reason);  
+      } else { 
+        if (backoff_delay (backoff_timer) // 3,7,15,31 ..
+              != BACKOFF_DELAY_TAKEN) // shutdown or cond wait error
+          return false;
+      }
       if (rtn == CONN_WAIT_RETRY_DNS)
         return false;  //find_server again
       // else retry
     }
+    return false;
 }
 
+int wait_while_interface_down()
+
+{
+	int rtn;
+	
+	ParodusError("Interface is down, hence waiting until its up\n");
+	close_and_unref_connection (get_global_conn());
+	set_global_conn(NULL);
+
+	while (get_interface_down_event ()) {
+	  pthread_mutex_lock(get_interface_down_mut());
+  	  rtn = pthread_cond_wait(get_interface_down_con(), get_interface_down_mut());
+      pthread_mutex_unlock (get_interface_down_mut());
+  	  if (rtn != 0)
+  	    ParodusError 
+  	      ("Error on pthread_cond_wait (%d) in wait_while_interface_down\n", rtn);
+  	  if ((rtn != 0) || g_shutdown) {
+	    return -1;
+	  }
+	}
+    return 0;
+}
 
 //--------------------------------------------------------------------
 
@@ -587,19 +697,6 @@ int createNopollConnection(noPollCtx *ctx)
 	  if (keep_trying_to_connect (&conn_ctx, &backoff_timer))
 		break;
 	  // retry dns query
-
-	  // If interface down event is set, stop retry
-	  // and wait till interface is up again.
-	  if(get_interface_down_event()) {
-		ParodusError("Interface is down, hence pausing retry and waiting until its up\n");
-		pthread_mutex_lock(get_interface_down_mut());
-		pthread_cond_wait(get_interface_down_con(), get_interface_down_mut());
-		pthread_mutex_unlock (get_interface_down_mut());
-		ParodusInfo("Interface is back up, re-initializing the convey header\n");
-		// Reset the reconnect reason by initializing the convey header again
-		((header_info_t *)(&conn_ctx.header_info))->conveyHeader = getWebpaConveyHeader();
-		ParodusInfo("Received reconnect_reason as:%s\n", reconnect_reason);  
-	  }
 	}
       
 	if(conn_ctx.current_server->allow_insecure <= 0)
@@ -694,10 +791,16 @@ static noPollConnOpts * createConnOpts (char * extra_headers, bool secure)
 void close_and_unref_connection(noPollConn *conn)
 {
     if (conn) {
-        nopoll_conn_close(conn);
-
-	get_parodus_cfg()->cloud_status = CLOUD_STATUS_OFFLINE;
-      	ParodusInfo("cloud_status set as %s after connection close\n", get_parodus_cfg()->cloud_status);
+      const char *reason = get_global_shutdown_reason();
+      int reason_len = 0;
+      int status = CloseNoStatus;
+      if (NULL != reason) {
+	reason_len = (int) strlen (reason);
+	status = CloseNormalClosure;
+      }
+      nopoll_conn_close_ext(conn, status, reason, reason_len);
+      get_parodus_cfg()->cloud_status = CLOUD_STATUS_OFFLINE;
+      ParodusInfo("cloud_status set as %s after connection close\n", get_parodus_cfg()->cloud_status);
     }
 }
 

src/connection.h

@@ -33,6 +33,8 @@ extern "C" {
 /*----------------------------------------------------------------------------*/
 /*                            File Scoped Variables                           */
 /*----------------------------------------------------------------------------*/
+#define SHUTDOWN_REASON_PARODUS_STOP    "parodus_stopping"
+#define SHUTDOWN_REASON_SYSTEM_RESTART  "system_restarting"
 
 /**
 * parodusOnPingStatusChangeHandler - Function pointer
@@ -57,6 +59,9 @@ void close_and_unref_connection(noPollConn *);
 noPollConn *get_global_conn(void);
 void set_global_conn(noPollConn *);
 
+char *get_global_shutdown_reason();
+void set_global_shutdown_reason(char *reason);
+
 char *get_global_reconnect_reason();
 void set_global_reconnect_reason(char *reason);
 
@@ -75,6 +80,11 @@ void stop_conn_in_progress (void);
 // To Register parodusOnPingStatusChangeHandler Callback function
 void registerParodusOnPingStatusChangeHandler(parodusOnPingStatusChangeHandler on_ping_status_change);
 
+// To stop connection and wait duing interface down state
+int wait_while_interface_down (void);
+
+void terminate_backoff_delay (void);
+
 #ifdef __cplusplus
 }
 #endif

src/main.c

@@ -19,14 +19,14 @@
 #include "stdlib.h"
 #include "config.h"
 #include "auth_token.h"
+#include "connection.h"
 #include "conn_interface.h"
 #include "parodus_log.h"
 #include <curl/curl.h>
 #ifdef INCLUDE_BREAKPAD
 #include "breakpad_wrapper.h"
-#else
-#include "signal.h"
 #endif
+#include "signal.h"
 
 /*----------------------------------------------------------------------------*/
 /*                                   Macros                                   */
@@ -36,7 +36,7 @@
 /*----------------------------------------------------------------------------*/
 /*                               Data Structures                              */
 /*----------------------------------------------------------------------------*/
-/* none */
+typedef    void    Sigfunc(int);
 
 /*----------------------------------------------------------------------------*/
 /*                            File Scoped Variables                           */
@@ -46,30 +46,53 @@
 /*----------------------------------------------------------------------------*/
 /*                             Function Prototypes                            */
 /*----------------------------------------------------------------------------*/
-#ifndef INCLUDE_BREAKPAD
 static void sig_handler(int sig);
+
+Sigfunc *
+signal (int signo, Sigfunc *func)
+{
+    struct sigaction act, oact;
+
+    act.sa_handler = func;
+    sigemptyset (&act.sa_mask);
+    act.sa_flags = 0;
+    if (signo == SIGALRM) {
+#ifdef  SA_INTERRUPT
+        act.sa_flags |= SA_INTERRUPT;     /* SunOS 4.x */
 #endif
+    } else {
+#ifdef  SA_RESTART
+        act.sa_flags |= SA_RESTART; /* SVR4, 4.4BSD */
+#endif
+    }
+    if (sigaction (signo, &act, &oact) < 0) {
+	ParodusError ("Signal Handler for signal %d not installed!\n", signo);
+        return (SIG_ERR);
+    }
+    return (oact.sa_handler);
+}
 
 /*----------------------------------------------------------------------------*/
 /*                             External Functions                             */
 /*----------------------------------------------------------------------------*/
 int main( int argc, char **argv)
 {
-#ifdef INCLUDE_BREAKPAD
-    breakpad_ExceptionHandler();
-#else
+    set_global_shutdown_reason (SHUTDOWN_REASON_PARODUS_STOP);
     signal(SIGTERM, sig_handler);
-	signal(SIGINT, sig_handler);
+    signal(SIGINT, sig_handler);
 	signal(SIGUSR1, sig_handler);
 	signal(SIGUSR2, sig_handler);
+	signal(SIGQUIT, sig_handler);
+	signal(SIGHUP, sig_handler);   
+	signal(SIGALRM, sig_handler);
+#ifdef INCLUDE_BREAKPAD
+    /* breakpad handles the signals SIGSEGV, SIGBUS, SIGFPE, and SIGILL */
+    breakpad_ExceptionHandler();
+#else
 	signal(SIGSEGV, sig_handler);
 	signal(SIGBUS, sig_handler);
-	signal(SIGKILL, sig_handler);
 	signal(SIGFPE, sig_handler);
 	signal(SIGILL, sig_handler);
-	signal(SIGQUIT, sig_handler);
-	signal(SIGHUP, sig_handler);
-	signal(SIGALRM, sig_handler);
 #endif	
     ParodusCfg *cfg;
 
@@ -97,7 +120,6 @@ const char *rdk_logger_module_fetch(void)
 /*----------------------------------------------------------------------------*/
 /*                             Internal functions                             */
 /*----------------------------------------------------------------------------*/
-#ifndef INCLUDE_BREAKPAD
 static void sig_handler(int sig)
 {
 
@@ -105,12 +127,13 @@ static void sig_handler(int sig)
 	{
 		signal(SIGINT, sig_handler); /* reset it to this function */
 		ParodusInfo("SIGINT received!\n");
-		shutdownSocketConnection();
+		shutdownSocketConnection(SHUTDOWN_REASON_PARODUS_STOP);
 	}
 	else if ( sig == SIGUSR1 ) 
 	{
 		signal(SIGUSR1, sig_handler); /* reset it to this function */
 		ParodusInfo("SIGUSR1 received!\n");
+		shutdownSocketConnection(SHUTDOWN_REASON_SYSTEM_RESTART);
 	}
 	else if ( sig == SIGUSR2 ) 
 	{
@@ -134,8 +157,7 @@ static void sig_handler(int sig)
 	else 
 	{
 		ParodusInfo("Signal %d received!\n", sig);
-		shutdownSocketConnection();
+		shutdownSocketConnection(SHUTDOWN_REASON_PARODUS_STOP);
 	}
 	
 }
-#endif

src/nopoll_helpers.c

@@ -146,7 +146,7 @@ void __report_log (noPollCtx * ctx, noPollDebugLevel level, const char * log_msg
     
     if (level == NOPOLL_LEVEL_DEBUG) 
     {
-        //ParodusPrint("%s\n", log_msg);
+        ParodusInfo("%s\n", log_msg);
     }
     if (level == NOPOLL_LEVEL_INFO) 
     {
@@ -154,7 +154,7 @@ void __report_log (noPollCtx * ctx, noPollDebugLevel level, const char * log_msg
     }
     if (level == NOPOLL_LEVEL_WARNING) 
     {
-        ParodusPrint("%s\n", log_msg);
+        ParodusInfo("%s\n", log_msg);
     }
     if (level == NOPOLL_LEVEL_CRITICAL) 
     {

tests/test_conn_interface.c

@@ -33,6 +33,7 @@
 /*----------------------------------------------------------------------------*/
 /*                            File Scoped Variables                           */
 /*----------------------------------------------------------------------------*/
+static char *reconnect_reason = "webpa_process_starts";
 UpStreamMsg *UpStreamMsgQ;
 ParodusMsg *ParodusMsgQ;
 pthread_mutex_t g_mutex=PTHREAD_MUTEX_INITIALIZER;
@@ -75,6 +76,21 @@ noPollMutexUnlock 	mutex_unlock
     function_called();
 }
 
+char *get_global_reconnect_reason()
+{
+    return reconnect_reason;
+}
+
+char *get_global_shutdown_reason()
+{
+    return SHUTDOWN_REASON_PARODUS_STOP;
+}
+
+void set_global_shutdown_reason(char *reason)
+{
+    UNUSED(reason);
+}
+
 void start_conn_in_progress (void)
 {
 }   
@@ -83,6 +99,24 @@ void stop_conn_in_progress (void)
 {
 }   
 
+void reset_interface_down_event (void)
+{
+}
+
+bool get_interface_down_event (void)
+{
+	return false;
+}
+
+int wait_while_interface_down (void)
+{
+	return 0;
+}
+
+void terminate_backoff_delay (void)
+{
+}
+
 void packMetaData()
 {
     function_called();

tests/test_connection.c

@@ -181,6 +181,11 @@ void nopoll_conn_close (noPollConn *conn)
     UNUSED(conn);
 }
 
+void nopoll_conn_close_ext (noPollConn *conn, int status, const char *reason, int reason_size)
+{
+    UNUSED(conn); UNUSED(status); UNUSED(reason); UNUSED(reason_size);
+}
+
 int checkHostIp(char * serverIP)
 {
     UNUSED(serverIP);
@@ -1035,7 +1040,6 @@ void test_create_nopoll_connection()
 
 }
 
-
 void test_get_interface_down_event()
 {
     assert_false(get_interface_down_event());
@@ -1180,8 +1184,6 @@ void test_interface_down_retry()
    pthread_join(thread_a, NULL);
 } 
 
-
-
 /*----------------------------------------------------------------------------*/
 /*                             External Functions                             */
 /*----------------------------------------------------------------------------*/